HIRS/.ci/system-tests/system_test_Driver.py

# System Test Driver to help with debugging.

import binascii
from ConfigParser import SafeConfigParser
import datetime
import json
import os
import shlex
import subprocess
import unittest
import re
import requests
import logging
import pprint
import hashlib
import random
import uuid
import time
import sys
import argparse

from system_test_core import HIRSPortal, AttestationCAPortal, collectors, \
    send_command, send_command_sha1sum, run_hirs_report, run_hirs_provisioner_tpm_1_2, \
    run_hirs_provisioner_tpm_2_0, parse_xml_with_stripped_namespaces, get_current_timestamp, \
    get_all_nodes_recursively, touch_random_file_and_remove, get_random_pcr_hex_value, \
    is_ubuntu_client, is_tpm_2_0, is_tpm_1_2, \
    DEFAULT_IMA_POLICY, DEFAULT_TPM_POLICY

NUMBER_OF_PCRS = 24

suffix = os.environ.get('RANDOM_SYS_TEST_ID')
if suffix != None:
    print("Configuring with suffix: %s" % suffix)
    suffix = "-" + suffix
else:
    suffix = ""

# Change to point to your HIRS directory
HOME_DIR = "/HIRS/"
HIRS_ACA_PORTAL_IP="172.17.0.2"
TPM_VERSION="2.0"
#TPM_VERSION="1.2"
# Change accordingly
#COLLECTOR_LIST = None
#COLLECTOR_LIST = ["IMA"]
COLLECTOR_LIST = ["TPM"]
#COLLECTOR_LIST = ["IMA", "TPM"]
#COLLECTOR_LIST = ["BASE_DELTA_GOOD"]
#COLLECTOR_LIST = ["BASE_DELTA_BAD"]

FORMAT = "%(asctime)-15s %(message)s"
provisioner_out = None

HIRS_ACA_PROVISIONER_IP="172.19.0.3"
HIRS_ACA_PROVISIONER_TPM2_IP="172.19.0.4"
TPM_ENABLED=True
IMA_ENABLED=False

HIRS_ACA_PORTAL_PORT="8443"
HIRS_BROKER_PORT="61616"
HIRS_ACA_PORTAL_CONTAINER_PORT="80"
HIRS_ACA_HOSTNAME="hirsaca"
HIRS_SUBNET="172.19.0.0/16"
CLIENT_OS="centos7"
CLIENT_HOSTNAME="hirs-client-"+ CLIENT_OS + "-tpm2"
CLIENT=CLIENT_HOSTNAME
SERVER_OS="$CLIENT_OS"
SERVER_HOSTNAME="hirs-appraiser-$SERVER_OS"

HIRS_ATTESTATION_CA_PORTAL_URL = "https://" + \
    HIRS_ACA_PORTAL_IP + ":" + \
    HIRS_ACA_PORTAL_PORT + \
    "/HIRS_AttestationCAPortal/"

CA_CERT_LOCATION = HOME_DIR + ".ci/setup/certs/ca.crt"
EK_CA_CERT_LOCATION = HOME_DIR + ".ci/setup/certs/ek_cert.der"
PBaseCertA_LOCATION = HOME_DIR + "PBaseCertA.der"
PBaseCertB_LOCATION = HOME_DIR + "PBaseCertB.der"
SIDeltaCertA1_LOCATION = HOME_DIR + "SIDeltaCertA1.der"
SIDeltaCertA2_resolved_LOCATION = HOME_DIR + "SIDeltaCertA2_resolved.der"
SIDeltaCertA2_LOCATION = HOME_DIR + "SIDeltaCertA2.der"
SIDeltaCertA3_LOCATION = HOME_DIR + "SIDeltaCertA3.der"
VARDeltaCertA1_LOCATION = HOME_DIR + "VARDeltaCertA1.der"
VARDeltaCertA2_LOCATION = HOME_DIR + "VARDeltaCertA2.der"
VARDeltaCertA2_resolved_LOCATION = HOME_DIR + "VARDeltaCertA2_resolved.der"
SIDeltaCertB1_LOCATION = HOME_DIR + "SIDeltaCertB1.der"
VARDeltaCertB1_LOCATION = HOME_DIR + "VARDeltaCertB1.der"

TEST_LOG_FILE= HOME_DIR + ".ci/system-tests/test_logs/system_test_" + CLIENT_OS + ".log"
LOG_LEVEL="logging.INFO"

print("Start of Log file: " + TEST_LOG_FILE)
logging.basicConfig(filename=TEST_LOG_FILE,level=eval(LOG_LEVEL), format=FORMAT)
logging.info("*****************beginning of system_test.py*****************")
logging.info("The Collector list is: " + ' '.join(COLLECTOR_LIST))
logging.info("The ACA Portal is: " + HIRS_ATTESTATION_CA_PORTAL_URL)

#Portal = HIRSPortal(HIRS_SERVER_URL)
AcaPortal = AttestationCAPortal(HIRS_ATTESTATION_CA_PORTAL_URL)

requests.packages.urllib3.disable_warnings()

class SystemTest(unittest.TestCase):

    @classmethod
    def setUpClass(self):
        """Set the class up"""

    @classmethod
    def tearDownClass(self):
        """Tears down the class"""

    def setUp(self):
        """Set the systems tests state up for testing"""
        #AcaPortal.disable_supply_chain_validations()

    def tearDown(self):
        """Tears down the state for testing"""

    def test_01_attestation_ca_portal_online(self):
      """Test that the Attestation CA Portal is online and accessible by making a GET request.
          If not online, an exception will be raised since the response code is non-200"""
      logging.info("***************** Beginning of attestation ca portal online test *****************")
      AcaPortal.check_is_online()

    @collectors(['IMA', 'TPM'], COLLECTOR_LIST)
    def test_02_empty_baselines(self):
        """Test that appraisal succeeds with empty IMA and TPM baselines"""
        logging.info("***************** Beginning of empty baseline test *****************")

    @collectors(['IMA'], COLLECTOR_LIST)
    def test_03_small_ima_appraisal(self):
        """Test that appraisal works with a small hard-coded IMA baseline"""
        logging.info("***************** Beginning of small IMA appraisal test *****************")

    @collectors(['IMA'], COLLECTOR_LIST)
    def test_04_large_ima_appraisal(self):
          """Test that appraisal works with a full-size IMA baseline"""
          logging.info("***************** Beginning of large IMA appraisal test *****************")

    @collectors(['IMA'], COLLECTOR_LIST)
    def test_05_small_ima_appraisal_required_set_missing(self):
        """Test that appraisal results in an appropriate alert generation when a required set file is missing

            steps:
              - upload a small hard-coded required set (two records)
              - add a fictitious file to the baseline
              - make a policy that points to that baseline as its required set
              - set the default device group to point to that policy
              - run a report from the client machine using vagrant ssh
              - make sure it failed and that one appropriate alert was thrown
        """
        logging.info("***************** Beginning of small IMA appraisal test with required set missing *****************")

    @collectors(['TPM', 'IMA'], COLLECTOR_LIST)
    def test_06_tpm_white_list_appraisal(self):
        """Test that appraisal works with a TPM white list baseline

            steps:
              - run hirs report to generate an XML report for baseline creation
              - download the latest report in XML format
              - convert the TPM part of the report into a json baseline
              - make a policy that points to that json TPM white list baseline
              - set the default device group to point to that policy
              - run a report from the client machine
         """
        logging.info("***************** Beginning of TPM white list appraisal test *****************")

    @collectors(['IMA'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_07_ima_blacklist_appraisal(self):
        """Test that appraisal works with a small IMA blacklist baseline

        steps:
          - upload a policy with a small hard-coded blacklist baseline
          - set the default device group to point to that policy
          - run a report from the client machine and ensure the appraisal passes
          - touch a file on the client that is contained in the blacklist
          - run another report from the client machine and ensure the appraisal fails
        """
        logging.info("***************** Beginning of blacklist IMA appraisal test *****************")

    @collectors(['IMA'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_08_delta_reports_required_set(self):
        """Test that appraisal works with delta reports and required sets.

            steps:
            - Run hirs report with an empty required set and delta reports
              enabled
            - Check first report for success and to make sure the test files
              are not there
            - Add the two test files (foo-file and foo-bar-file) to the required
              set with a hashes that indicates the files are empty
            - create foo-file and read it as root so it is measured by IMA
            - Run second hirs report
            - Check for failed appraisal (foo-bar-file hasn't been created yet)
            - Check that the report includes foo-file, but not foo-bar-file
            - Create foo-bar-file and read it as root
            - Run third hirs report
            - Check for failed appraisal (foo-file was in the previous report,
              so it won't be included in this one.
            - Check that foo-bar-file is in this report, but not foo-file
        """
        logging.info("***************** Beginning of Delta Reports required set appraisal test *****************")

    @collectors(['IMA'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_09_delta_reports_whitelist(self):
        """Test that appraisal works with delta reports. Each report should be
           appraised individually. Checks that a failed appraisal can be followed
           by a successful appraisal if there are no errors in the second delta
           report.

            steps:
            - Run hirs report with an empty required set and delta reports
              enabled
            - Check first report for success and to make sure the test files
              are not there
            - Add a test file (foo-file) to the whitelist with a hash that
              indicates the file is empty
            - Create foo-file with contents and read it as root so it is
              measured by IMA
            - Run second hirs report
            - Check for failed appraisal (foo-file should be a whitelist
              mismatch because the file isn't empty)
            - Check that the report includes foo-file
            - Run third hirs report
            - Check for successful appraisal (the mismatch was in the previous
              report so it won't be included in this one.
            - Check that foo-file is not in this report
        """
        logging.info("***************** Beginning of Delta Reports whitelist appraisal test *****************")

    @collectors(['IMA', 'TPM'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_10_on_demand(self):
        """Test that on-demand (server-initiated) appraisal works.

            steps:
            - push a simple ima baseline
            - set the policy
            - touch a random file, take the hash, then remove it
            - kick off an on-demand report on the server for the default device group
            - sleep to let the appraisal finish
            - pull the generated report
                - check that it passed appraisal
                - check that it has the random filename and hash
                - check that it contains a TPM Report
            """
        logging.info("***************** Beginning of on-demand test *****************")

    @collectors(['IMA'], COLLECTOR_LIST)
    @unittest.skip("SELinux issues are preventing repo sync from working")
    def test_11_failing_ima_appraisal_broad_repo_baseline(self):
        """Test that an appraisal not containing expected packages in a broad repo IMA baseline fails.

            steps:
            - Create a Yum repository with a local file URL and sync it
            - Create a broad baseline using the Yum repository
            - Add the baseline to the required set for the default IMA policy
            - Run a HIRS report and ensure it fails
            - Ensure that at least one of the expected alerts has been generated
            """
        logging.info("***************** Beginning of broad repo failing appraisal test *****************")

    @collectors(['IMA'], COLLECTOR_LIST)
    @unittest.skip("SELinux issues are preventing repo sync from working")
    @unittest.skipIf(is_ubuntu_client(CLIENT_OS), "Skipping this test due to client OS " + CLIENT_OS)
    def test_12_successful_ima_appraisal_broad_repo_baseline(self):
        """Test that an appraisal containing expected packages in a broad repo IMA baseline passes.
           This test only works on CentOS 6 and 7.

            steps:
            - Create a Yum repository with a local file URL and sync it
            - Create a broad baseline using the Yum repository
            - Add the baseline to the required set for the default IMA policy
            - Install RPMs in repository to client machine and read them with root to ensure their placement in the IMA log
            - Run a HIRS report and ensure it passes
            - Ensure that there are no new alerts
            """
        logging.info("***************** Beginning of broad repo successful appraisal test *****************")

    @collectors(['TPM'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_1_2(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_13_tpm_1_2_initial_provision(self):
      """Test that running the TPM 1.2 hirs provisioner works"""
      logging.info("***************** Beginning of initial TPM 1.2 provisioner run *****************")

      # Run the provisioner to ensure that it provisions successfully
      provisioner_out = run_hirs_provisioner_tpm_1_2(CLIENT)
      print("Initial TPM 1.2 provisioner run output: {0}".format(provisioner_out))

    @collectors(['TPM'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_14_tpm_2_0_initial_provision(self):
        """Test that running the TPM 2.0 hirs provisioner works"""
        logging.info("***************** Beginning of initial TPM 2.0 provisioner run *****************")

        # Run the provisioner to ensure that it provisions successfully
        provisioner_out = run_hirs_provisioner_tpm2(CLIENT)
        print("Initial provisioner run output: {0}".format(provisioner_out))

    @collectors(['TPM'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_15_device_info_report_stored_after_provisioning(self):
        """Test that running the hirs provisioner results in storing a device info report for
        the device in the DB"""
        logging.info("***************** Beginning of device info report test *****************")

        logging.info("Getting devices from ACA portal...")
        aca_portal_devices = AcaPortal.get_devices()
        self.assertEqual(aca_portal_devices['recordsTotal'], 1)

    @collectors(['TPM'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_16_supply_chain_validation_summary_stored_after_second_provisioning(self):
        """Test that running the hirs provisioner, a second time, results in storing a supply chain validation
           record in the database"""
        logging.info("***************** Beginning of supply chain validation summary test *****************")

        logging.info("Uploading CA cert: " + CA_CERT_LOCATION)
        AcaPortal.upload_ca_cert(CA_CERT_LOCATION)
        AcaPortal.enable_supply_chain_validations()

        provisioner_out = run_hirs_provisioner_tpm_2_0(CLIENT)
        print("Second provisioner run output: {0}".format(provisioner_out))

        supply_chain_validation_summaries = AcaPortal.get_supply_chain_validation_summaries()
        # verify this is one SCVS record indicating PASS
        self.assertEqual(supply_chain_validation_summaries['recordsTotal'], 2)
        self.assertEqual(supply_chain_validation_summaries['data'][0]['overallValidationResult'], "PASS")
        self.assertEqual(supply_chain_validation_summaries['data'][1]['overallValidationResult'], "PASS")

        # verify device has been updated with supply chain appraisal result
        devices = AcaPortal.get_devices()
        self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "PASS")

    @collectors(['TPM'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_17_ek_info_report(self):
      """Test that running the hirs provisioner results in storing EK certs info report for
         the device in the DB"""
      logging.info("***************** Beginning of Endorsement Certs info report test *****************")

      logging.info("Getting EK Certs from ACA portal...")
      cert_list = AcaPortal.get_ek_certs()
      self.assertEqual(cert_list['recordsTotal'], 1)
      self.assertEqual(cert_list['data'][0]['credentialType'], "TCPA Trusted Platform Module Endorsement")

    @collectors(['TPM'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_18_pk_info_report(self):
      """Test that running the hirs provisioner results in storing PK certs info report for
         the device in the DB"""
      logging.info("***************** Beginning Platform Certs info report test *****************")

      logging.info("Getting PK Certs from ACA portal...")
      cert_list = AcaPortal.get_pk_certs()
      self.assertEqual(cert_list['recordsTotal'], 1)
      self.assertEqual(cert_list['data'][0]['credentialType'], "TCG Trusted Platform Endorsement")

    @collectors(['TPM'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_19_trust_chain_info_report(self):
      """Test that running the hirs provisioner results in storing trust chains info report for
         the device in the DB"""
      logging.info("***************** Beginning of Trust Chain info report test *****************")

      logging.info("Getting Trust Chains from ACA portal...")
      trust_chain_list = AcaPortal.get_trust_chains()
      self.assertEqual(trust_chain_list['recordsTotal'], 1)

    @collectors(['BASE_DELTA_GOOD'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_20_A1_base_delta(self):
      """Test Delta Certificates A1 - Provisioning with Good Base Platform Cert (via Platform Cert on TPM Emulator)"""
      logging.info("***************** test_20_A1 - Beginning of delta certificate test *****************")
      logging.info("Provisioning with Good Base Platform Cert (via Platform Cert on TPM Emulator)")

      logging.info("Check if ACA is online...")
      AcaPortal.check_is_online()

      logging.info("Uploading CA Cert: " + CA_CERT_LOCATION)
      AcaPortal.upload_ca_cert(CA_CERT_LOCATION)
      AcaPortal.enable_supply_chain_validations()
      provisioner_out = run_hirs_provisioner_tpm_2_0(CLIENT)

      print("test_20_A1_base_delta run output: {0}".format(provisioner_out))

      # Verify device supply chain appraisal result is PASS
      devices = AcaPortal.get_devices()
      self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "PASS")

    @collectors(['BASE_DELTA_GOOD'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_20_A2_base_delta(self):
      """Test Delta Certificates A2 - Attempt to upload Base cert with holder already having a Base Platform Cert associated with it"""
      logging.info("***************** test_20_A2 - Beginning of delta certificate test *****************")
      logging.info("Attempt to upload PBaseCertB, with PBaseCertA already loaded in the ACA.")

      print("test_20_A2_base_delta. PBaseCertA has already been loaded. Attempting to upload second Platform Cert: %s" % (PBaseCertB_LOCATION))

      # Confirm there is one Platform Base Cert already loaded
      cert_list = AcaPortal.get_pk_certs()
      self.assertEqual(cert_list['recordsTotal'], 1)
      print("Number of Platform Certs: %d" % (cert_list['recordsTotal']))
      self.assertEqual(cert_list['data'][0]['credentialType'], "TCG Trusted Platform Endorsement")
      self.assertEqual(cert_list['data'][0]['platformType'], "Base")

      # Try uploading a second Platform Base Cert
      print("Attempting to upload a second Platform Base Cert...")
      AcaPortal.upload_pk_cert(PBaseCertB_LOCATION)

      # Confirm Platform Base Cert has not been loaded
      cert_list = AcaPortal.get_pk_certs()
      self.assertEqual(cert_list['recordsTotal'], 1)
      print("Number of Platform Certs: %d" % (cert_list['recordsTotal']))
      self.assertEqual(cert_list['data'][0]['credentialType'], "TCG Trusted Platform Endorsement")
      self.assertEqual(cert_list['data'][0]['platformType'], "Base")

      if (cert_list['recordsTotal'] == 1):
         print ("SUCCESS.\n")
      else:
         print ("FAILED.\n")

    @collectors(['BASE_DELTA_GOOD'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_20_A3_base_delta(self):
      """Test Delta Certificates A3 - Provisioning with Good Base Platform Cert Base and 1 Delta Cert"""
      logging.info("***************** test_20_A3 - Beginning of delta certificate test *****************")
      logging.info("Provisioning with Good Base Platform Cert Base and 1 Delta Cert")

      # Verify device supply chain appraisal result is PASS
      devices = AcaPortal.get_devices()
      self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "PASS")

      # Upload the SIDeltaCertA1 and provision
      AcaPortal.upload_pk_cert(SIDeltaCertA1_LOCATION)
      AcaPortal.enable_supply_chain_validations()
      provisioner_out = run_hirs_provisioner_tpm_2_0(CLIENT)
      print("test_20_A3_base_delta run output: {0}".format(provisioner_out))

      supply_chain_validation_summaries = AcaPortal.get_supply_chain_validation_summaries()
      # Verify this is one SCVS record indicating PASS
      self.assertEqual(supply_chain_validation_summaries['recordsTotal'], 2)
      self.assertEqual(supply_chain_validation_summaries['data'][0]['overallValidationResult'], "PASS")
      self.assertEqual(supply_chain_validation_summaries['data'][1]['overallValidationResult'], "PASS")

      # Verify device has been updated with supply chain appraisal result
      devices = AcaPortal.get_devices()
      self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "PASS")

    @collectors(['BASE_DELTA_GOOD'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_20_A4_base_delta(self):
      """Test Delta Certificates A4 - Provisioning with Good Base Platform Cert Base and 2 Delta Certs"""
      logging.info("***************** test_20_A4 - Beginning of delta certificate test *****************")
      logging.info("Provisioning with Good Base Platform Cert Base and 2 Delta Certs")

      # Verify device supply chain appraisal result is PASS
      devices = AcaPortal.get_devices()
      self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "PASS")

      # Upload the VARDeltaCertA1 and provision
      AcaPortal.upload_pk_cert(VARDeltaCertA1_LOCATION)
      AcaPortal.enable_supply_chain_validations()
      provisioner_out = run_hirs_provisioner_tpm_2_0(CLIENT)

      print("test_20_A4_base_delta run output: {0}".format(provisioner_out))
      supply_chain_validation_summaries = AcaPortal.get_supply_chain_validation_summaries()

      # Verify this is one SCVS record indicating PASS
      self.assertEqual(supply_chain_validation_summaries['recordsTotal'], 3)
      self.assertEqual(supply_chain_validation_summaries['data'][0]['overallValidationResult'], "PASS")
      self.assertEqual(supply_chain_validation_summaries['data'][1]['overallValidationResult'], "PASS")
      self.assertEqual(supply_chain_validation_summaries['data'][2]['overallValidationResult'], "PASS")

      # Verify device has been updated with supply chain appraisal result
      devices = AcaPortal.get_devices()
      self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "PASS")

    @collectors(['BASE_DELTA_GOOD'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_20_A5_base_delta(self):
      """Test Delta Certificates A5 - Provisioning with Good Base Platform Cert and 1 Bad Delta Cert"""
      logging.info("***************** test_20_A5 - Beginning of delta certificate test *****************")
      logging.info("Provisioning with Good Base Platform Cert and 1 Bad Delta Cert")

        # TODO: Determine if we need this test

    @collectors(['BASE_DELTA_GOOD'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_20_A6_base_delta(self):
      """Test Delta Certificates A6 - Provisioning with Good Base Platform, 2 Good Delta Certs and 1 Bad Delta Cert"""
      logging.info("***************** test_20_A6 - Beginning of delta certificate test *****************")
      logging.info("Provisioning with Good Base Platform, 2 Good Delta Certs and 1 Bad Delta Cert")

      # Verify device supply chain appraisal result is PASS
      devices = AcaPortal.get_devices()
      self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "PASS")

      # Upload the SIDeltaCertA2 and provision
      AcaPortal.upload_pk_cert(SIDeltaCertA2_LOCATION)
      AcaPortal.enable_supply_chain_validations()
      provisioner_out = run_hirs_provisioner_tpm_2_0(CLIENT)

      print("test_20_A6_base_delta SHOULD FAIL provisioning using: %s" % (SIDeltaCertA2_LOCATION))
      print("test_20_A6_base_delta run output: {0}".format(provisioner_out))

      # Provisioning should fail since the Delta contains a bad component.
      self.assertIn("Provisioning failed", format(provisioner_out))

      # Upload the SIDeltaCertA2_resolved and provision
      AcaPortal.upload_pk_cert(SIDeltaCertA2_resolved_LOCATION)
      AcaPortal.enable_supply_chain_validations()
      provisioner_out = run_hirs_provisioner_tpm_2_0(CLIENT)

      print("test_20_A6_base_delta SHOULD PASS provisioning using: %s" % (SIDeltaCertA2_resolved_LOCATION))
      print("test_20_A6_base_delta run output: {0}".format(provisioner_out))

       # Verify device has been updated with supply chain appraisal result
      devices = AcaPortal.get_devices()
      self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "PASS")

    @collectors(['BASE_DELTA_GOOD'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_20_A7_base_delta(self):
      """Test Delta Certificates A7 - Provisioning with Good Base Platform, 2 Good Delta Certs and
      1 Bad Delta Cert with non present component"""
      logging.info("***************** test_20_A7 - Beginning of delta certificate test *****************")
      logging.info("Provisioning with Good Base Platform, 2 Good Delta Certs and 1 Bad Delta Cert with non present component")

      # Upload the VARDeltaCertA2 and provision
      AcaPortal.upload_pk_cert(VARDeltaCertA2_LOCATION)
      AcaPortal.enable_supply_chain_validations()
      provisioner_out = run_hirs_provisioner_tpm_2_0(CLIENT)

      print("test_20_A7_base_delta SHOULD FAIL provisioning using: %s" % (VARDeltaCertA2_LOCATION))
      print("test_20_A7_base_delta run output: {0}".format(provisioner_out))

      # Provisioning should fail since the Delta contains a component thats not in the Base
      self.assertIn("Provisioning failed", format(provisioner_out))

      # Upload the VARDeltaCertA2_resolved and provision
      AcaPortal.upload_pk_cert(VARDeltaCertA2_resolved_LOCATION)
      AcaPortal.enable_supply_chain_validations()
      provisioner_out = run_hirs_provisioner_tpm_2_0(CLIENT)

      print("test_20_A7_base_delta SHOULD PASS provisioning using: %s" % (VARDeltaCertA2_resolved_LOCATION))
      print("test_20_A7_base_delta run output: {0}".format(provisioner_out))

       # Verify device has been updated with supply chain appraisal result
      devices = AcaPortal.get_devices()
      self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "PASS")

    @collectors(['BASE_DELTA_GOOD'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_20_A8_base_delta(self):
      """Test Delta Certificates A8 - Provisioning with Good Base Platform, 2 Good Delta Certs with 1 Delta cert
         replacing component from previous, using the Delta as a base certificate"""
      logging.info("***************** test_20_A8 - Beginning of delta certificate test *****************")
      logging.info("Provisioning with Good Base Platform, 2 Good Delta Certs with 1 Delta cert replacing component from previous, using the Delta as a base certificate")

      # Upload the SIDeltaCertA3 and provision
      AcaPortal.upload_pk_cert(SIDeltaCertA3_LOCATION)
      AcaPortal.enable_supply_chain_validations()
      provisioner_out = run_hirs_provisioner_tpm_2_0(CLIENT)

      print("test_20_A8_base_delta run output: {0}".format(provisioner_out))

      # Verify device has been updated with supply chain appraisal result
      devices = AcaPortal.get_devices()
      self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "PASS")

    @collectors(['BASE_DELTA_BAD'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_20_B1_base_delta(self):
      """Test Base/Delta Certificates B1 - Provisioning with Bad Platform Cert Base """
      logging.info("***************** test_20_B1 - Beginning of delta certificate test *****************")
      logging.info("Provisioning with Bad Platform Cert Base")

      logging.info("Check if ACA is online...")
      AcaPortal.check_is_online()

      logging.info("Uploading CA cert: " + CA_CERT_LOCATION)
      AcaPortal.upload_ca_cert(CA_CERT_LOCATION)
      AcaPortal.enable_supply_chain_validations()
      provisioner_out = run_hirs_provisioner_tpm_2_0(CLIENT)

      print("test_20_B1_base_delta SHOULD FAIL provisioning using: %s" % (PBaseCertB_LOCATION))
      print("test_20_B1_base_delta run output: {0}".format(provisioner_out))

      # Provisioning should fail since the PC contains FAULTY components.
      self.assertIn("Provisioning failed", format(provisioner_out))

    @collectors(['BASE_DELTA_BAD'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_20_B2_base_delta(self):
      """Test Base/Delta Certificates B2 - Provisioning with Bad Platform Cert Base and 1 Good delta with 1 bad component unresolved"""
      logging.info("***************** test_20_B2 - Beginning of delta certificate test *****************")
      logging.info("Provisioning with Bad Platform Cert Base and 1 Good delta with 1 bad component unresolved")

      # Verify device supply chain appraisal result is FAIL
      devices = AcaPortal.get_devices()
      self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "FAIL")

      # Upload the SIDeltaCertB1 and provision
      AcaPortal.upload_pk_cert(SIDeltaCertB1_LOCATION)
      AcaPortal.enable_supply_chain_validations()
      provisioner_out = run_hirs_provisioner_tpm_2_0(CLIENT)

      print("test_20_B2_base_delta SHOULD FAIL provisioning using: %s" % (SIDeltaCertB1_LOCATION))
      print("test_20_B2_base_delta run output: {0}".format(provisioner_out))

      # Provisioning should fail since the delta contains FAULTY component.
      self.assertIn("Provisioning failed", format(provisioner_out))

    @collectors(['BASE_DELTA_BAD'], COLLECTOR_LIST)
    @unittest.skipIf(not is_tpm_2_0(TPM_VERSION), "Skipping this test due to TPM Version " + TPM_VERSION)
    def test_20_B3_base_delta(self):
      """Test Base/Delta Certificates B3 - Provisioning with Bad Platform Cert Base and 2 Good delta with all component resolved"""
      logging.info("***************** test_20_B3 - Beginning of delta certificate test *****************")
      logging.info("Provisioning with Bad Platform Cert Base and 2 Good delta with all component resolved")

      # Verify device supply chain appraisal result is FAIL
      devices = AcaPortal.get_devices()
      self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "FAIL")

      # Upload the VARDeltaCertB1 and provision
      AcaPortal.upload_pk_cert(VARDeltaCertB1_LOCATION)
      AcaPortal.enable_supply_chain_validations()
      provisioner_out = run_hirs_provisioner_tpm_2_0(CLIENT)

      print("test_20_B3_base_delta run output: {0}".format(provisioner_out))

      # Verify device has been updated with supply chain appraisal of PASS
      devices = AcaPortal.get_devices()
      self.assertEqual(devices['data'][0]['device']['supplyChainStatus'], "PASS")

if __name__ == '__main__':
    suite = unittest.TestLoader().loadTestsFromTestCase(SystemTest)
    ret = not unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful()
    sys.exit(ret)