Rebase to primary domain-protect/main branch

.github/workflows/codeql.yml

@@ -0,0 +1,34 @@
+name: "CodeQL"
+
+on:
+  push:
+
+jobs:
+  CodeQL-Build:
+    runs-on: ubuntu-latest
+    permissions:
+      actions: read
+      contents: read
+      security-events: write
+
+    strategy:
+      fail-fast: false
+      matrix:
+        language: [ 'python' ]
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Initialize CodeQL
+        uses: github/codeql-action/init@v3
+        with:
+          languages: ${{ matrix.language }}
+
+      - name: Autobuild
+        uses: github/codeql-action/autobuild@v3
+
+      - name: Perform CodeQL Analysis
+        uses: github/codeql-action/analyze@v3
+        with:
+          category: "/language:${{matrix.language}}"

.github/workflows/integration_tests.yml

@@ -0,0 +1,35 @@
+name: Integration Tests
+on:
+  push:
+
+jobs:
+  integration_tests:
+    name: Integration Test
+    runs-on: ubuntu-latest
+    permissions:
+      checks: write # Required for Publish Test Results
+      pull-requests: write # Required for Publish Test Results
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Python 3.11
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Install dependencies
+        run: |
+          pip install -r requirements-dev.txt
+
+      - name: test
+        run: |
+          pytest --cov=manual_scans --cov=utils --cov-branch --cov-fail-under=19 --junitxml=test_reports/junit-integration.xml integration_tests
+
+      - name: Publish Test Results
+        uses: EnricoMi/publish-unit-test-result-action@v2
+        if: always()
+        with:
+          files: "test_reports/junit-integration.xml"
+          check_name: "Integration Test Results"
+          comment_title: "Integration Test Results"

.github/workflows/unit_tests.yml

@@ -0,0 +1,36 @@
+name: Formatting and Unit Tests
+on:
+  push:
+
+jobs:
+  unit_tests:
+    name: Unit Test
+    runs-on: ubuntu-latest
+    permissions:
+      checks: write # Required for Publish Test Results
+      pull-requests: write # Required for Publish Test Results
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Python 3.11
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: 'pip' # caching pip dependencies
+      - run : pip install -r requirements-dev.txt
+
+      - name: Linting and formatting
+        uses: pre-commit/[email protected]
+        with:
+          extra_args: --all-files --show-diff-on-failure
+
+      - name: Unit tests
+        run: |
+          pytest --cov=manual_scans --cov=utils --cov-branch --cov-fail-under=15 unittests
+
+      - name: Publish Test Results
+        uses: EnricoMi/publish-unit-test-result-action@v2
+        if: always()
+        with:
+          files: "test_reports/*.xml"

lambda_code/scan/scan.py

@@ -1,6 +1,5 @@
 #!/usr/bin/env python
 import json
-import logging
 import os
 
 from utils.utils_aws import eb_susceptible
@@ -27,9 +26,6 @@
 env_name = os.environ["ENVIRONMENT"]
 production_env = os.environ["PRODUCTION_ENVIRONMENT"]
 
-BC_ACCT_ID_BLACKLIST = [
-    "876504563909"
-]
 
 def process_vulnerability(domain, account_name, resource_type, vulnerability_type, takeover=""):
 
@@ -295,6 +291,7 @@ def lambda_handler(event, context):  # pylint:disable=unused-argument
     account_name = event["Name"]
 
     hosted_zones = list_hosted_zones(event)
+
     for hosted_zone in hosted_zones:
         print(f"Searching for vulnerable domain records in hosted zone {hosted_zone['Name']}")
 
@@ -304,10 +301,10 @@ def lambda_handler(event, context):  # pylint:disable=unused-argument
         alias_cloudfront_s3(account_name, record_sets, account_id)
         alias_eb(account_name, record_sets)
         alias_s3(account_name, record_sets)
-        # cname_azure(account_name, record_sets)
+        cname_azure(account_name, record_sets)
         cname_cloudfront_s3(account_name, record_sets, account_id)
         cname_eb(account_name, record_sets)
-        # cname_google(account_name, record_sets)
+        cname_google(account_name, record_sets)
         cname_s3(account_name, record_sets)
         ns_subdomain(account_name, hosted_zone, record_sets)
 

utils/utils_aws.py

@@ -48,10 +48,7 @@ def assume_role(account, region_override="None"):
     try:
         assumed_role_object = generate_temporary_credentials(account, security_audit_role_name, external_id, project)
 
-        if assumed_role_object:
-            credentials = assumed_role_object["Credentials"]
-        else:
-            raise RuntimeError(f"could not generate STS credentials for {security_audit_role_name} role in AWS account {account}")
+        credentials = assumed_role_object["Credentials"]
 
         return create_session(credentials, region_override)
 
@@ -123,43 +120,48 @@ def list_hosted_zones(account):
 
 
 def list_resource_record_sets(account_id, account_name, hosted_zone_id):
-    record_set_list = []
 
-    boto3_session = assume_role(account_id)
-    route53 = boto3_session.client("route53")
-
     try:
-        paginator_records = route53.get_paginator("list_resource_record_sets")
-        pages_records = paginator_records.paginate(
-            HostedZoneId=hosted_zone_id,
-            StartRecordName="_",
-            StartRecordType="NS",
-        )
+        boto3_session = assume_role(account_id)
+        route53 = boto3_session.client("route53")
 
-        for page_records in pages_records:
-            record_sets = page_records["ResourceRecordSets"]
+        record_set_list = []
+
+        try:
+            paginator_records = route53.get_paginator("list_resource_record_sets")
+            pages_records = paginator_records.paginate(
+                HostedZoneId=hosted_zone_id,
+                StartRecordName="_",
+                StartRecordType="NS",
+            )
 
-            record_set_list = record_set_list + record_sets
+            for page_records in pages_records:
+                record_sets = page_records["ResourceRecordSets"]
 
-        return record_set_list
+                record_set_list = record_set_list + record_sets
+
+            return record_set_list
+
+        except Exception:
+            logging.exception(
+                "ERROR: Lambda execution role requires route53:ListResourceRecordSets permission in %a account",
+                account_name,
+            )
 
     except Exception:
-        logging.exception(
-            "ERROR: Lambda execution role requires route53:ListResourceRecordSets permission in %a account",
-            account_name,
-        )
+        logging.error("ERROR: unable to assume role in %a account %s", account_name, account_id)
 
     return []
 
 
 def list_domains(account_id, account_name):
 
-    domain_list = []
-
     try:
         boto3_session = assume_role(account_id, "us-east-1")
         route53domains = boto3_session.client("route53domains")
 
+        domain_list = []
+
         try:
             paginator_domains = route53domains.get_paginator("list_domains")
             pages_domains = paginator_domains.paginate()
@@ -181,7 +183,7 @@ def list_domains(account_id, account_name):
     except Exception:
         logging.error("ERROR: unable to assume role in %a account %s", account_name, account_id)
 
-    return domain_list
+    return []
 
 
 def publish_to_sns(json_data, subject):
@@ -223,12 +225,14 @@ def get_cloudfront_s3_origin_url(account_id, account_name, domain):
                             return distribution["Origins"]["Items"][0]["DomainName"]
 
         except exceptions.ClientError as e:
+            print(e.response["Error"]["Code"])
             logging.error(
-                "ERROR: error when fetching CloudFront S3 origins URLs in %a account; MSG: %s",
-                account_name, e.msg
+                "ERROR: Lambda execution role requires cloudfront:ListDistributions permission in %a account",
+                account_name,
             )
 
     except exceptions.ClientError as e:
+        print(e.response["Error"]["Code"])
         logging.error("ERROR: unable to assume role in %a account %s", account_name, account_id)
 
     return None