Automatisiertes Cloud-Infrastruktur-Testtool mit Terraform und PyTest erstellen

Vor Kurzem erhielt ich den Auftrag, ein automatisiertes Test-Tool für Fugue zu entwickeln. Fugue überwacht Cloud-Ressourcen auf Compliance und Sicherheit. Wir benötigten eine Möglichkeit, zu überprüfen, ob die vollständigen Ergebnisse eines Fugue-Scans korrekt waren. Mein Ziel war ein automatisiertes System, das lokal oder in CI ausgeführt wird, konfigurierbare Infrastruktur bereitstellt, sie mit Fugue scannt und die Ergebnisse überprüft. In diesem Blogbeitrag führe ich durch den Design- und Implementierungsprozess für autotest, unser internes automatisiertes Test-Tool.

Dieses Beispiel verwendet zwar Terraform und die Fugue API, doch das Design der Fixtures lässt sich auf beliebige CLI-Befehle und APIs übertragen. Die hier vorgestellten Ideen können daher leicht an andere Anwendungsfälle angepasst werden.

Design

Bevor ich richtig mit dem Projekt begann, nahm ich mir ein paar Tage Zeit, um zu recherchieren und das Problem zu verstehen. Nach einigen Google-Suchen beschloss ich, mein Tool auf dem Framework pytest aufzubauen. Ich hatte bereits mit pytest für Unit-Tests gearbeitet, doch dies war die perfekte Gelegenheit, mich intensiv damit auseinanderzusetzen und herauszufinden, wie sich das Framework als Grundlage für einen Funktionstest nutzen lässt.

Der Testablauf lässt sich in drei Schritte unterteilen:

1. Mit Terraform eine modulare Infrastruktur bereitstellen

2. Eine Fugue-Umgebung erstellen, die auf diese Infrastruktur verweist, und sie scannen

3. Die Scan-Ergebnisse abrufen und überprüfen

Für Schritt 1 wollte ich unbedingt Terraform verwenden, da wir bereits über eine Reihe von Terraform-Konfigurationen für Infrastrukturtests verfügten und Terraform AWS, Azure und viele weitere Cloud-Service-Anbieter unterstützt. Für Schritt 2 wusste ich, dass ich mit der Fugue API die Umgebung erstellen und einen Scan starten konnte. Schließlich konnte ich in Schritt 3 erneut die Fugue API nutzen, um die vollständigen Scan-Ergebnisse abzurufen und zu überprüfen.

Nachdem das grobe Design feststand, konnte ich mir pytest genauer ansehen und herausfinden, wie ich es für die Umsetzung meines Designs einsetzen konnte.

PyTest

pytest ist ein Test-Framework für Python mit automatischer Testerkennung, Assertions mithilfe der integrierten Python-Anweisung assert und Fixtures für das Abhängigkeitsmanagement.

Fixtures sind das Geheimrezept von pytest. Dabei handelt es sich um dekorierte Funktionen, die den Lebenszyklus von Testabhängigkeiten verwalten. Einrichtungs- und Aufräumcode, der traditionell in den Methoden setUp() und tearDown() stünde, wird stattdessen in Fixtures platziert. Der Gültigkeitsbereich einer Fixture lässt sich so festlegen, dass ihre Ergebnisse für mehrere Klassen, Module oder Sitzungen wiederverwendet werden.

Sehen wir uns ein Beispiel an. Hier definiere und verwende ich eine Fixture für den Zugriff auf eine API. Während ich in einem Unit-Test ein Mock-API-Objekt verwenden würde, muss ich in einem Funktionstest die echte API nutzen.

@pytest.fixture(scope="session")
def api_client(creds):
  api_client = api.initialize_client(creds)
  return api_client

def test_api_client_username(api_client):
  assert api_client.username == "mike"

def test_api_client_host(api_client):
  assert api_client.host == "https://api.fugue.co"

Die Fixture api_client wird mit dem Dekorator @pytest.fixture deklariert. Den scope der Fixture kann ich mithilfe des Scope-Arguments im Dekorator festlegen. Der API-Client kann bedenkenlos in allen Tests wiederverwendet werden. Deshalb habe ich für ihn den Scope session festgelegt. Fixtures lassen sich im selben Modul wie Testfunktionen oder -klassen definieren oder in conftest.py ablegen, wodurch sie allen Testmodulen zur Verfügung stehen.

Ich kann die Fixture verwenden, indem ich ihren Namen einfach als Argument in eine Testfunktion aufnehme. In diesem Beispiel habe ich die Fixture api_client in zwei Tests verwendet. Da ich für die Fixture den Scope session festgelegt habe, stellt pytest sicher, dass sie während jeder Testsitzung nur einmal aufgerufen wird.

Vielleicht ist Ihnen aufgefallen, dass die Fixture api_client einen Eingabeparameter creds entgegennimmt. Wir rufen die Funktion fugue_api nie direkt auf. Woher erhält creds also einen Wert?

Wenn Sie vermutet haben, dass creds selbst eine Fixture ist, haben Sie sich einen goldenen Stern verdient: ⭐️. Fixtures können andere Fixtures aufrufen. So lassen sich komplexe Abhängigkeiten aus einer Kette von Fixtures aufbauen. Sehen wir uns die Definition von creds an:

def pytest_addoption(parser):
         parser.addoption(
             "--api-secret", action="store",
             help="name of credstash secret to use when 
initializing the API"
        )

@pytest.fixture(scope="session")
def creds(request):
  """returns API credentials"""
  secret_name = request.config.getoption("--api-secret")
  return credstash.getSecret(secret_name)

Ich achte auf gute Sicherheitspraktiken. Daher verwendet creds credstash, um geheime Werte zu speichern und abzurufen. Der Name des Secrets wird als Befehlszeilenoption an pytest übergeben. Diese Optionen werden in pytest_addoption definiert und über die Fixture request abgerufen. Die Fixture creds erleichtert es mir, die Implementierung für den Abruf von Anmeldedaten zu abstrahieren und mich darauf zu konzentrieren, welche Eingaben für einen bestimmten Test oder eine Fixture erforderlich sind. Ändere ich später die Art und Weise, wie ich auf Anmeldedaten zugreife, muss ich nur creds anpassen.

Nachdem wir nun die Grundlagen der pytest-Fixtures kennen, sehen wir uns an, wie wir sie für die Umsetzung unseres Designs nutzen können.

Schritt 1: Infrastruktur mit Terraform bereitstellen

Im ersten Schritt erstellt unser Tool die Infrastruktur, die gescannt und in den Schritten 2 und 3 als Grundlage für die Überprüfung verwendet wird. Ich wusste, dass ich Terraform für die Bereitstellung dieser Infrastruktur verwenden würde. Doch wie genau sollte das funktionieren?

Design

├── aws
| ├── api_gateway
│ │ ├── aws_api_gateway.tf
│ │ ├── aws_api_gateway_client_certificate.tf
│ ├── autoscaling
│ │ ├── aws_autoscaling_group.tf
│ │ ├── aws_autoscaling_lifecycle_hook.tf
│ │ ├── ...

Dieser Ausschnitt zeigt die Struktur unserer bestehenden Testkonfigurationen. Sie lassen sich manuell bereitstellen, indem Sie terraform apply im Stammverzeichnis aws/ oder in einem Dienst-Unterverzeichnis wie aws/api_gateway ausführen. Diese Modularität wollte ich beibehalten, damit wir auswählen können, welche Dienste getestet werden. Ich könnte Terraform in der bestehenden Struktur aufrufen, doch das wäre eine schlechte Praxis: Alle Arbeiten im Dateisystem sollten in eindeutigen temporären Verzeichnissen erfolgen, damit Testläufe in kontrollierten Umgebungen ausgeführt werden. Daher benötige ich eine Möglichkeit, temporäre Arbeitsbereiche zu verwalten. Außerdem muss ich Terraform aufrufen, seine Ausgabe protokollieren und auf Fehler prüfen können. Terraform erstellt echte (und teure) Infrastruktur. Deshalb ist eine Fehlerbehandlung unerlässlich, die bei Problemen die Infrastruktur wieder entfernt.

Zusammengefasst lauten die Anforderungen für den ersten Schritt:

1. Temporäre Arbeitsbereiche erstellen und verwalten

2. Testkonfigurationen in einen Arbeitsbereich kopieren

3. Terraform aufrufen und bei Fehlern bereits erstellte Infrastruktur wieder entfernen

Punkt 1 und 2 lassen sich offensichtlich mit einer Fixture umsetzen. Doch kann auch etwas so Komplexes und Fehleranfälliges wie das Erstellen von Infrastruktur mit Terraform als Fixture funktionieren?

Wenn Sie JA gesagt haben, gibt es einen weiteren goldenen Stern für Sie: ⭐️. Schauen wir uns genauer an, wie wir unsere Workspace- und Terraform-Fixtures definieren.

Die Workspace-Fixture

@pytest.yield_fixture(scope="class")
def workspace(tmpdir_factory, provider, services):
    """returns an initialized workspace that closes after 
it's finished"""
    working_dir = tmpdir_factory.mktemp("autotest_tf")

    with Workspace(working_dir, provider, services) as w: 
         yield w

class Workspace:
    def __init__(self, base_dir: str, provider: str, services: List[str]) -> 
 'Workspace':
          self.provider = provider
          self.base_dir = base_dir
          self.working_dir = os.path.join(base_dir, provider)
          self.services = services

    def __enter__(self):
       return self._initialize_workspace(self.base_dir, self.provider, self.services)

    def __exit__(self, typ, value, traceback):
        pass

In den sechs Codezeilen von workspace steckt eine Menge. Zunächst fällt auf, dass workspace als yield_fixture definiert ist. Das bedeutet, dass die Funktion die Kontrolle über yield statt über return an ihren Aufrufer zurückgibt. Die Ausführung der Fixture wird währenddessen angehalten, sodass Context Manager geöffnet bleiben. Sobald die Kontrolle zur Fixture zurückkehrt, wird der Code nach der Anweisung yield ausgeführt. In diesem Fall wird der Context Manager Workspace beendet. pytest ruft den zurückgegebenen Wert einer yield_fixture ab. Deshalb können wir sie wie gewöhnliche Fixtures behandeln. workspace ist eine Fixture mit Klassen-Scope. Daher wird für jede Testklasse ein neuer Arbeitsbereich erstellt. So kann ich AWS- und Azure-Tests nach Klassen aufteilen und sicherstellen, dass für jeden Anbieter ein neuer Arbeitsbereich erstellt wird.

Als Nächstes sehen wir, dass workspace von drei Fixtures abhängt: tmpdir_factory, provider und services. tmpdir_factory ist eine integrierte Fixture, die ein eindeutiges temporäres Verzeichnis zurückgibt. Das Basisverzeichnis lässt sich mithilfe der Option --basetemp=DIRECTORY manuell festlegen, was beim Debuggen hilfreich ist. provider und services sind von mir definierte Fixtures. Sie geben den Anbieter zurück, gegen den der aktuelle Test ausgeführt wird, sowie die Liste der Dienste aus der Eingabe.

Besonders gut an Fixtures gefällt mir, dass sie zu klaren Abstraktionen anregen. Die Eingaben eines Tests oder einer Fixture werden über eine Schnittstelle – die Funktionsdefinition – festgelegt, während die Implementierung in der Fixture-Definition verborgen bleibt. Dank der Modularität von Fixtures kann ich schnell und einfach für jede Eingabe von workspace eine eigene Fixture erstellen. So komme ich nicht in Versuchung, die Werte in workspace zu erstellen.

Beachten Sie schließlich, dass Workspace eine Klasse ist, die als context manager fungiert. Dadurch wird der Arbeitsbereich automatisch geschlossen, sobald die Fixture selbst abgeschlossen ist. Wie bereits erwähnt, wird der Context Manager durch die yield-Anweisung erst dann abgeschlossen, wenn die Kontrolle beim Abschließen der Fixture an diese zurückgegeben wird. Dieses leistungsstarke Muster habe ich im gesamten Design von autotest verwendet. Das Kopieren der gewünschten Terraform-Konfigurationen in den Arbeitsbereich wird in initialize_workspace gehandhabt und bleibt eine Übung für die Leserinnen und Leser.

Die Terraform-Fixture

@pytest.fixture(scope="class") 
def terraform(workspace): 
     """initialize Terraform in workspace""" 
     return initialize_terraform(workspace.working_dir) 

@pytest.yield_fixture(scope="class") 
def infra(request, terraform, services):
    """returns a Terraform instance associated with created resources""" 
    with build_infra(terraform, services): 
         yield terraform

@pytest.yield_fixture(scope="class") 
def terraform_resources(infra): 
    """returns the Terraform resources map""" 
    yield infra.tfstate.modules[0]['resources']

Die Terraform-Fixture besteht eigentlich aus drei Fixtures: terraform, infra und terraform_resources. Das Ergebnis dieser Fixture-Kette ist ein Dictionary mit Terraform-Ressourcen, die aus der Terraform-State-Datei ausgelesen werden. terraform initialisiert Terraform in einem Arbeitsbereich, infra führt terraform plan und terraform apply aus, um die Infrastruktur zu erstellen, und terraform_resources ruft die Ressourcenübersicht ab.

Die Initialisierung und das Extrahieren der Ressourcenübersicht sind unkompliziert. Ich möchte jedoch darauf hinweisen, dass terraform_resources eine yield_fixture sein muss, weil infra eine yield_fixture ist. Würde terraform_resources einen Wert zurückgeben, würde dadurch der von infra offengehaltene Kontext geschlossen und der Aufräumcode ausgelöst.

Anstelle einer Context-Manager-Klasse verwendet infra einen funktionsbasierten Context Manager namens build_infra, der wie folgt definiert ist:

 @contextmanager
 def build_infra(t: Terraform, services: List[str]):
    try:
       code, out, err = t.plan(out='temp.plan')
       if code == 1:
             raise TerraformPlanError(out, err)
       code, out, err = t.apply(dir_or_plan='temp.plan')
       if code == 1:
            raise TerraformApplyError(out, err)
    except TerraformApplyError:
         code, out, err = t.destroy()
         if code != 0:
             raise TerraformDestroyError(out, err)
         raise

 try:
    yield t
 finally:
    code, out, err = t.destroy()
    if code != 0:
        raise TerraformDestroyError(out, err)

build_infra verwendet das Paket python-terraform, um Terraform aufzurufen und seine Ausgabe zu erfassen. Bei allen Vorgängen prüfen wir den Fehlercode und lösen bei Fehlern benutzerdefinierte Ausnahmen aus. t.plan und t.apply rufen terraform plan bzw. terraform apply auf, um die Infrastruktur zu planen und zu erstellen. terraform apply kann fehlschlagen, nachdem die Infrastruktur bereits teilweise erstellt wurde. Da terraform bereits erstellte Infrastruktur nicht automatisch entfernt, fangen wir TerraformApplyError ab und rufen manuell t.destroy auf, bevor wir die ursprüngliche Ausnahme erneut auslösen.

build_infra ist ein Generator: Statt return aufzurufen, verwendet er yield. Dadurch kann ich den Dekorator @contextmanager verwenden, um daraus einen Context Manager zu machen. Der try...finally-Block um yield stellt sicher, dass der Aufräumcode auch dann ausgeführt wird, wenn im aufrufenden Kontext Ausnahmen auftreten.

Beachten Sie schließlich die Ähnlichkeit zwischen dem Code im except-Block und im finally-Block. Beide Blöcke tun dasselbe: Sie stellen sicher, dass die Infrastruktur unabhängig vom Verlauf der Ausführung entfernt wird. Da der Code in einem finally-Block sowohl bei einer Ausnahme als auch ohne Ausnahme ausgeführt wird, können wir die try-Blöcke zusammenfassen.

Außerdem wissen wir, dass t.plan keine Ressourcen erstellt. Daher können wir den Aufruf aus dem try-Block herausnehmen. Generell ist es empfehlenswert, try-Blöcke so eng wie möglich zu halten. Der fertige Context Manager build_infra sieht so aus:

 @contextmanager
 def build_infra(t: Terraform, services: List[str]):
    code, out, err = t.plan(out='temp.plan')
    if code == 1:
       raise TerraformPlanError(out, err)
    try:
        code, out, err = t.apply(dir_or_plan='temp.plan')
        if code == 1:
            raise TerraformApplyError(out, err)
        yield t
    finally:
         code, out, err = t.destroy()
         if code != 0:
              raise TerraformDestroyError(out, err)

Schritt 2: Eine Fugue-Umgebung erstellen und scannen

Im nächsten Schritt unseres übergeordneten Designs verwenden wir die Fugue API, um in dem Konto, in dem die Infrastruktur in Schritt 1 erstellt wurde, eine Umgebung anzulegen und zu scannen.

Design

Die Fugue API ist eine REST API, die mit Swagger definiert wurde. So lässt sich leicht ein Client generieren, mit dem ich über native Funktionen und Typen auf die API zugreifen kann, statt HTTP-Aufrufe zu verwenden. Sobald ein funktionsfähiger Client bereitsteht, kann ich damit eine Umgebung erstellen und scannen. Der Ablauf lässt sich in drei Schritte unterteilen:

1. Einen Client für die Fugue API generieren und konfigurieren

2. Mit dem Client eine Umgebung erstellen

3. Mit dem Client einen Scan starten und warten, bis er abgeschlossen ist

Der Swagger-API-Client

Mit swagger-codegen habe ich einen Client auf Grundlage der Fugue API-Definition generiert:

swagger-codegen generate -i swagger.yaml -l python -o .

Dadurch wurde ein vollständiges Python-Paket für den API-Client generiert, das ich nicht benötigte. Daher kopierte ich den Client-Code direkt in autotest. Anschließend initialisierte ich den Client wie folgt:

 def initialize_client(creds: dict) - ApiClient:
    c = Configuration()
    c.host = os.environ["FUGUE_API_URL"] + os.environ.get("FUGUE_API_VERSION", "v0")
    c.username = creds["username"]
    c.password = creds["password"]

    client = ApiClient(configuration=c)
    return client

Hier sind creds lediglich ein Eingabeparameter und keine Fixture, da initialize_client weder eine Fixture noch ein Test ist. Zur Erinnerung: initialize_client wird von einer Fixture aufgerufen, die von der Fixture creds abhängt:

 @pytest.fixture(scope="session")
 def api_client(creds):
     """returns a Fugue API client"""
     return api.initialize_client(creds)

@pytest.fixture(scope="session")
def environments_api(api_client):
     """returns a Fugue Environment API instance"""
     return api.environments_api(api_client)

 @pytest.fixture(scope="session")
 def scans_api(api_client):
     """returns a Fugue Scans API instance"""
     return api.scans_api(api_client)

Die Fixture für die Environments-API lässt sich unkompliziert definieren, da API-Clients nicht abgeschlossen werden müssen. Der von Swagger generierte Client verfügt über einen grundlegenden API-Client, der die Verbindung zur API verwaltet, sowie über separate Klassen für die einzelnen Hauptbereiche der API. Die API-Aktionen stehen dann als Methoden der jeweiligen Klassen zur Verfügung.

Ein Hinweis zu Abstraktionen: Über das Modul api abstrahiere ich die Details des Swagger-Clients. Das entspricht dem Prinzip, beide Seiten einer Schnittstelle zu abstrahieren, und gibt mir die Flexibilität, auf Änderungen an der Client-Schnittstelle zu reagieren. Darauf gehe ich näher ein, wenn ich erläutere, wie sich mithilfe der API eine Environment erstellen lässt.

Eine Environment erstellen und scannen

Mein erster Entwurf für eine Environment-Fixture sah ungefähr so aus:

 @pytest.yield_fixture(scope="class")
  def environment(environments_api, aws_creds,
 survey_resource_types, services):
     environment = 
 environments_api.create_environment(CreateEnvironmentInput
 (
          name="autotest-123",
          provider="aws",
          provider_options=ProviderOptions(
              aws=ProviderOptionsAws(
                  region="us-east-2",
                  role_arn=aws_creds
              )
         ),
        survey_resource_types=survey_resource_types,
        compliance_families=["FBP"]
    ))
   yield environment
   environments_api.delete_environment(environment.id)

Hier kommt die Methode create_environment der Environments-API ganz direkt zum Einsatz. Zugangsdaten und Eingabeparameter beziehe ich aus den Fixtures aws_creds, survey_resource_types und services. Die Fixture gilt für Klassen, da ich Tests für verschiedene Cloud-Anbieter nach Klassen organisiere. Schließlich gebe ich die Environment mit yield zurück, damit sie nach Abschluss des Tests automatisch gelöscht werden kann.

Der Entwurf dieser Fixture ist einfach und erfüllt seinen Zweck. Ich könnte ihn unverändert lassen und weitermachen. Dieser einfache Ansatz hat jedoch Nachteile. Die Eingabe für create_environment ist umständlich, weil sie viele Felder umfasst, von denen einige vom Anbieter abhängen. Wenn ich die Environments-API direkt in der Fixture verwende, kann ich die Eingabeparameter für create_environment nur eingeschränkt vorbereiten, ohne eine riesige, schwer lesbare Funktion zu schreiben. Der schwierige Kompromiss zwischen Lesbarkeit und Wiederverwendbarkeit führte dazu, dass ich auf Wiederverwendbarkeit verzichtete, den Anbieter fest codierte und die Fixture environment für Azure-Environments duplizierte – keine gute Lösung. Was ich wirklich brauche, ist eine zusätzliche Ebene zwischen dem Fixture-Code und der Environments-API, die die umständliche Vorbereitung der Eingaben kapselt und die Beziehung zwischen Fixture und API verwaltet. Zum Glück bietet mir das Modul api genau das.

Ich habe die Eingabevorbereitung und den API-Aufruf in eine Funktion im Modul api ausgelagert:

 # api.py
 def create_environment(
     environments_api: EnvironmentsApi,
     provider: str,
     creds,
     services: List[str],
     compliance_families: List[str] = None,
     survey_resource_types: List[str] = None,
     survey_resource_groups: List[str] = None
 ) -> Environment:
   if provider == "aws":
       provider_options = ProviderOptions(
           aws=ProviderOptionsAws(
               region="us-east-2",
               role_arn=creds
           )
       )
   elif provider == "azure":
      provider_options = ProviderOptions(
          azure=ProviderOptionsAzure(
              tenant_id=creds["ARM_TENANT_ID"],
              subscription_id=creds["ARM_SUBSCRIPTION_ID"],
              application_id=creds["ARM_CLIENT_ID"],
              client_secret=creds["ARM_CLIENT_SECRET"],
              survey_resource_groups=survey_resource_groups,
           )
     )
     survey_resource_types = []
  else:
      raise AttributeError("invalid provider")

 environments_api.create_environment(CreateEnvironmentInput( 
      name=f"autotest-{provider}", 
      provider=provider, 
      provider_options=provider_options, 
      survey_resource_types=survey_resource_types, 
      compliance_families=compliance_families 
  ))  
  return environment 

Die Funktion ist zwar umfangreich, unterstützt aber sowohl AWS als auch Azure. Außerdem lässt sich die Fixture environment dadurch auf das Wesentliche reduzieren:

 @pytest.yield_fixture(scope="class")
 def environment(
     environments_api,
     provider: str,
     aws_creds: str,
     azure_creds: dict,
     survey_resource_types: List[str],
     survey_resource_groups: List[str],
     services: List[str],
 ):

     """creates a Fugue Environment and deletes it automatically"""
     if provider == "aws":
         creds = aws_creds
         compliance_families = ["FBP"]
     elif provider == "azure":
         creds = azure_creds
         compliance_families = ["CISAZURE"]

     environment = api.create_environment(
         environments_api,
         provider,
         creds,
         services,
         compliance_families,
         survey_resource_types,
         survey_resource_groups
  ) 
  yield environment  
  api.delete_environment(environments_api, environment)

Dieselbe Vorgehensweise kann ich auch zum Starten von Scans in einer Environment verwenden:

 @pytest.fixture(scope="class")
 def scan
    terraform_resources: dict, #terraform resources must exist before scan
    environment: Environment,
    scans_api: ScansApi
 )-> Scan:
    """creates a Fugue Scan and waits for it to finish"""
    return api.create_scan(
         scans_api,
         environment
  ) 

Auch wenn ich die Ausgabe nicht verwende: Durch das Einbinden der Fixture terraform_resources ist sichergestellt, dass die Ressourcen erstellt wurden, bevor ich einen Scan starte. Im Modul api sieht das dann so aus:

  def create_scan(scans_api: ScansApi, environment: Environment) -> Scan:
      scan: Scan = scans_api.create_scan(environment.id)

      while scan.status not in ["ERROR", "SUCCESS", "CANCELLED"]:
          time.sleep(1)
          scan = scans_api.get_scan(scan.id)
      if scan.status != "SUCCESS":
          print(scan.message)
      assert scan.status == "SUCCESS"

      return scan

Jetzt habe ich Fixtures, um mit Terraform Ressourcen zu erstellen, eine Fugue-Environment anzulegen und sie zu scannen. Damit kann ich fast mit dem Schreiben von Tests beginnen.

Schritt 3: Scan-Ergebnisse abrufen und überprüfen

Bevor ich Tests schreibe, muss ich die Scan-Ergebnisse über die Fugue-API abrufen können. In der API sind die Scan-Ergebnisse über den Endpunkt Eingabe für den Test benutzerdefinierter Regeln verfügbar. Diese Fixture zu erstellen, ist eine einfache Anwendung der oben beschriebenen Konzepte:

 @pytest.fixture(scope="class")
 def scan_resources(scan: Scan, custom_rules_api: CustomRulesApi) -> dict:
      """returns the results from scan Scan"""
      return api.get_scan_resources(custom_rules_api, scan.id)

get_scan_resources leitet den Aufruf einfach weiter:

 def get_scan_resources(custom_rules_api: CustomRulesApi, scan_id: str) -> dict:
     scan_resources: ScanResources = custom_rules_api.test_custom_rule_input(scan_id)
     return scan_resources.resources

Bei so einfachen Funktionen liegt es nahe, die Abstraktionsebene zu überspringen und den API-Aufruf direkt in die Fixture zu schreiben. Genau das habe ich in der ersten Version dieser Funktion getan. Dadurch entsteht jedoch eine undichte Abstraktion – eine Abstraktionsebene, die einen Teil der darunterliegenden Funktionalität offenlegt, die sie eigentlich verbergen soll. Undichte Abstraktionen führen dazu, dass die kognitive Belastung bei der Verwendung einer Abstraktion steigt, statt zu sinken.

Mit einer scan_resources-Fixture habe ich nun alle Eingaben, die ich für meinen Test benötige. Nach all den Vorbereitungen und Abstraktionen aus den vorherigen Abschnitten besteht der Test selbst nur aus wenigen Codezeilen:

  # test_aws.py
  class TestAWS:
     def test_verify_resource_ids(self, terraform_resources, scan_resources, check):
          for tf_id, tf_resource in terraform_resources.items():
              check.is_true(verify.verify_resource_id(tf_id, tf_resource, scan_resources))

check ist eine Fixture des pytest-Plugins, die Assertions bereitstellt, bei denen der Test nicht fehlschlägt. verify_resource_id ist eine einfache Verifizierungsfunktion, die zwei Ressourcen vergleicht und true zurückgibt, wenn ihre Ressourcen-IDs übereinstimmen:

  def verify_resource_id(tf_id: str, tf_resource: dict, scan_resources: dict) -> bool:
      """verify that a resource in the terraform state file is in the scan output"""
      # skip data sources
      if tf_id.startswith('data.'):
          return True
      print(f"verifying {tf_resource['type']} {tf_resource['primary']['id']}...", end="")
      for resource in scan_resources.values():
          if tf_resource['primary']['id'] == resource['_skeleton']['primary'] ['id']:
               print("✅")
               return True
       print("❌")
       return False 

Bei Bedarf kann ich auf alle Details der Ressourcen zugreifen und umfangreichere Tests schreiben. Für den Moment reicht es aber, die Infrastruktur bereitzustellen und zu überprüfen, ob sie korrekt gescannt wird. Bei der Ausführung des Tests erscheint folgende Ausgabe:

 ) pytest autotest --services sns
 ============================ test session starts ============================
 platform darwin -- Python 3.6.5, pytest-5.3.5, py-1.7.0, pluggy-0.13.1
 plugins: cov-2.6.1, check-0.3.7
 collected 1 item / 1 selected

 autotest/test_aws.py building infrastructure in sns
 creating environment
 environment abcdef12-abcdef12-abcdef12-abcdef12-abcdef12 created
 creating scan
 scan abcdef12-abcdef12-abcdef12-abcdef12-abcdef12 created
 scan started...
 ...
 scan 588f8ded-c219-468b-821e-9b11ee06ef17 successful!
 verifying aws_default_security_group sg-abcdef12...✅
 verifying aws_default_vpc vpc-abcdef12...✅
 verifying aws_kms_key abcdef12-abcdef12-abcdef12-abcdef12-abcdef12...✅
 verifying aws_sns_topic arn:aws:sns:us-east-2:123456789012:example-topic...✅
 .
 deleting environment
 environment deleted
 tearing down infrastructure
 infrastructure successfully torn down

 =========== 1 passed in 280.80s (0:04:40) ===========

Fazit

In diesem Blogbeitrag habe ich beschrieben, wie ich mit pytest autotest, ein automatisiertes Infrastruktur-Testtool, erstellt habe. Ich habe erläutert, wie leistungsfähig die modularen Fixtures von pytest sind und wie einfach sich damit übersichtliche, leistungsstarke Abstraktionen erstellen lassen. Außerdem habe ich gezeigt, wie sich mithilfe der Fugue-API Environments und Scans erstellen und diese Aktionen in Fixtures kapseln lassen. Anschließend habe ich erklärt, wie man über den API-Endpunkt für Eingaben in Tests benutzerdefinierter Regeln eine vollständige Beschreibung der Ressourcen in einem Scan herunterlädt. Zum Schluss habe ich alles in einem Test zusammengeführt.

Noch etwas …

Fugue steht für Cloud-Sicherheit und Compliance. Von Ingenieuren für Ingenieure entwickelt. Mit Fugue können Sie:

1. Mit dynamischen Visualisierungen und Compliance-Berichten erhalten Sie vollständige Transparenz über Ihre Cloud-Infrastruktur und deren Sicherheitsstatus.

2. Überprüfen Sie die Compliance in jeder Phase des Softwareentwicklungslebenszyklus – hinsichtlich CIS Foundations Benchmark, HIPAA, PCI, SOC 2, NIST 800-53, ISO 27001, DSGVO und Ihrer eigenen Richtlinien.

3. Schützen Sie sich vor Fehlkonfigurationen in der Cloud: Mit der Durchsetzung von Baselines kann sich geschäftskritische Cloud-Infrastruktur selbst heilen.