Prowler Scan Parser (#13831)

* Draft of prowler parser

* prowler parser lint changes

* Ruff fixes and formatting

* D203 fix

* D300 fix

* added unittests for all cloud types in both file types

* added docs for prowler parser

* added docs for prowler parser

* security and bug resistance for prowler parser

* Update prowler unittests

Author: Jino-T

docs/content/supported_tools/parsers/file/prowler.md

@@ -0,0 +1,14 @@
+---
+title: "Prowler Scan"
+toc_hide: true
+---
+This parser imports Prowler Scan files in JSON and CSV format. The AWS, GCP, Azure, and Kubernetes could types are supported by this parser.
+
+### Sample Scan Data
+Sample Prowler scans can be found [here](https://github.com/DefectDojo/django-DefectDojo/tree/master/unittests/scans/prowler).
+
+### Default Deduplication Hashcode Fields
+By default, DefectDojo identifies duplicate Findings using these [hashcode fields](https://docs.defectdojo.com/en/working_with_findings/finding_deduplication/about_deduplication/):
+
+- title
+- description

dojo/fixtures/test_type.json

@@ -47,5 +47,12 @@
         },
         "model": "dojo.test_type",
         "pk": 7
+    },
+    {
+        "fields": {
+            "name": "Prowler"
+        },
+        "model": "dojo.test_type",
+        "pk": 8
     }
 ]

dojo/tools/prowler/__init__.py

@@ -0,0 +1,24 @@
+from dojo.tools.prowler.parser_csv import ProwlerParserCSV
+from dojo.tools.prowler.parser_json import ProwlerParserJSON
+
+
+class ProwlerParser:
+
+    """Prowler is an Open Cloud Security that automates security and compliance in cloud environments. This parser is for Prowler JSON files and Prowler CSV files."""
+
+    def get_scan_types(self):
+        return ["Prowler Scan"]
+
+    def get_label_for_scan_types(self, scan_type):
+        return "Prowler Scan"
+
+    def get_description_for_scan_types(self, scan_type):
+        return "Prowler report file can be imported in JSON format or in CSV format."
+
+    def get_findings(self, filename, test):
+        name = getattr(filename, "name", str(filename)).lower()
+        if name.endswith(".csv"):
+            return ProwlerParserCSV().get_findings(filename, test)
+        if name.endswith(".json"):
+            return ProwlerParserJSON().get_findings(filename, test)
+        return []

dojo/tools/prowler/parser.py

@@ -0,0 +1,122 @@
+import csv
+import hashlib
+import io
+
+from dojo.models import Finding
+
+
+class ProwlerParserCSV:
+
+    """Parser for Prowler CSV (semicolon-separated)."""
+
+    def get_findings(self, filename, test):
+        if filename is None:
+            return []
+
+        content = filename.read()
+        if isinstance(content, bytes):
+            content = content.decode("utf-8")
+        reader = csv.DictReader(io.StringIO(content), delimiter=";")
+        csvarray = []
+        for row in reader:
+            csvarray.append(row)
+
+        dupes = {}
+        for row in csvarray:
+            # Skip vulnerability if the status is "PASS", continue parsing is status is "FAIL" or "MANUAL"
+            if row.get("STATUS") == "PASS":
+                continue
+
+            provider = row.get("PROVIDER", "N/A").upper()
+
+            description = (
+                "**Cloud Type** : "
+                + provider
+                + "\n\n"
+                + "**Description** : "
+                + row.get("DESCRIPTION", "N/A")
+                + "\n\n"
+                + "**Service Name** : "
+                + row.get("SERVICE_NAME", "N/A")
+                + "\n\n"
+                + "**Status Detail** : "
+                + row.get("STATUS_EXTENDED", "N/A")
+                + "\n\n"
+                + "**Finding Created Time** : "
+                + row.get("TIMESTAMP", "N/A")
+                + "\n\n"
+                + "**Region** : "
+                + row.get("REGION", "N/A")
+                + "\n\n"
+                + "**Notes** : "
+                + row.get("NOTES", "N/A")
+            )
+
+            related = row.get("RELATED_URL", "")
+            additional = row.get("ADDITIONAL_URLS", "")
+            if related:
+                description += "\n\n**Related URL** : " + related
+            if additional:
+                description += "\n\n**Additional URLs** : " + additional
+
+            mitigation = (
+                "**Remediation Recommendation** : "
+                + row.get("REMEDIATION_RECOMMENDATION_TEXT", "N/A")
+                + "\n\n"
+                + "**Remediation Recommendation URL** : "
+                + row.get("REMEDIATION_RECOMMENDATION_URL", "N/A")
+                + "\n\n"
+                + "**Remediation Code Native IaC** : "
+                + row.get("REMEDIATION_CODE_NATIVEIAC", "N/A")
+                + "\n\n"
+                + "**Remediation Code Terraform** : "
+                + row.get("REMEDIATION_CODE_TERRAFORM", "N/A")
+                + "\n\n"
+                + "**Remediation Code CLI** : "
+                + row.get("REMEDIATION_CODE_CLI", "N/A")
+                + "\n\n"
+                + "**Other Remediation Info** : "
+                + row.get("REMEDIATION_CODE_OTHER", "N/A")
+            )
+
+            title = row.get("CHECK_TITLE", "")
+            severity = self.convert_severity(row.get("SEVERITY"))
+            impact = row.get("RISK", "")
+            compliance = row.get("COMPLIANCE", "N/A")
+
+            # If compliance is not 'N/A', break info into multiple lines
+            references = "\n".join(part.strip() for part in compliance.split("|")) if compliance != "N/A" else "N/A"
+
+            finding = Finding(
+                title=title,
+                test=test,
+                description=description,
+                severity=severity,
+                references=references,
+                mitigation=mitigation,
+                impact=impact,
+                static_finding=False,
+                dynamic_finding=True,
+            )
+
+            key = hashlib.sha256((finding.title + "|" + finding.description).encode("utf-8")).hexdigest()
+            if key not in dupes:
+                dupes[key] = finding
+
+        return list(dupes.values())
+
+    def convert_severity(self, severity: str) -> str:
+        """Convert severity value"""
+        if not severity:
+            return "Info"
+
+        s = severity.lower()
+        if s == "critical":
+            return "Critical"
+        if s == "high":
+            return "High"
+        if s == "medium":
+            return "Medium"
+        if s == "low":
+            return "Low"
+        return "Info"

dojo/tools/prowler/parser_csv.py

@@ -0,0 +1,137 @@
+import hashlib
+import json
+
+from dojo.models import Finding
+
+
+class ProwlerParserJSON:
+
+    """This parser is for Prowler JSON files."""
+
+    def get_findings(self, file, test):
+        data = json.load(file)
+
+        dupes = {}
+        for node in data:
+            # Skip vulnerability if the status is "PASS", continue parsing is status is "FAIL" or "MANUAL"
+            if node.get("status_code") == "PASS":
+                continue
+
+            cloudtype = self.get_cloud_type(node)
+            description = (
+                "**Cloud Type** : "
+                + cloudtype
+                + "\n\n"
+                + "**Finding Description** : "
+                + node.get("finding_info", {}).get("desc", "N/A")
+                + "\n\n"
+                + "**Product Name** : "
+                + node.get("metadata", {}).get("product", {}).get("name", "N/A")
+                + "\n\n"
+                + "**Status Detail** : "
+                + node.get("status_detail", "N/A")
+                + "\n\n"
+                + "**Finding Created Time** : "
+                + node.get("finding_info", {}).get("created_time_dt", "N/A")
+            )
+            # Add cloud type sepecific information to description
+            description = self.add_cloud_type_metadata(node, cloudtype, description)
+
+            title = node.get("message", "")
+            severity = self.convert_severity(node.get("severity"))
+            mitigation = (
+                "**Remediation Description** : "
+                + node.get("remediation", {}).get("desc", "N/A")
+                + "\n\n"
+                + "**Remediation References** : "
+                + ", ".join(node.get("remediation", {}).get("references", []))
+            )
+            impact = node.get("risk_details", "")
+            compliance = node.get("unmapped", {}).get("compliance", {})
+            references = "**Related URL** : " + node.get("unmapped", {}).get("related_url", "")
+            # Add data presnet in scan to References
+            for key, values in compliance.items():
+                joined = ", ".join(values)
+                # Ex: CIS-1.10 : 1.2.16
+                references += f"\n\n**{key}** : {joined}"
+
+            finding = Finding(
+                title=title,
+                test=test,
+                description=description,
+                severity=severity,
+                references=references,
+                mitigation=mitigation,
+                impact=impact,
+                static_finding=False,
+                dynamic_finding=True,
+            )
+
+            # internal de-duplication
+            dupe_key = hashlib.sha256(str(description + title).encode("utf-8")).hexdigest()
+            if dupe_key in dupes:
+                find = dupes[dupe_key]
+                if finding.description:
+                    find.description += "\n" + finding.description
+                # find.unsaved_endpoints.extend(finding.unsaved_endpoints)
+                dupes[dupe_key] = find
+            else:
+                dupes[dupe_key] = finding
+
+        return list(dupes.values())
+
+    def convert_severity(self, severity: str) -> str:
+        """Convert severity value"""
+        if not severity:
+            return "Info"
+
+        s = severity.lower()
+        if s == "critical":
+            return "Critical"
+        if s == "high":
+            return "High"
+        if s == "medium":
+            return "Medium"
+        if s == "low":
+            return "Low"
+        return "Info"
+
+    def get_cloud_type(self, node: dict) -> str:
+        """Determine the cloud type of a Prowler JSON finding. Returns one of: AWS, Azure, Kubernetes, GCP, or N/A"""
+        # Check for GCP, AWS, or Azure
+        account_type = node.get("cloud", {}).get("provider")
+        if account_type:
+            account_type.lower()
+            if account_type == "gcp":
+                return "GCP"
+            if account_type == "aws":
+                return "AWS"
+            if account_type == "azure":
+                return "AZURE"
+
+        # Check for Kubernetes
+        for resource in node.get("resources", []):
+            namespace = resource.get("data", {}).get("metadata", {}).get("namespace")
+            if namespace is not None:
+                return "KUBERNETES"
+
+        # No Cloud Type information was found
+        return "N/A"
+
+    def add_cloud_type_metadata(self, node: dict, cloudtype: str, description: str) -> str:
+        # Add metadata for GCP, AWS, and Azure
+        if cloudtype in {"GCP", "AWS", "AZURE"}:
+            description += "\n\n" + "**" + cloudtype + " Region** : " + node.get("cloud", {}).get("region", "N/A")
+            return description
+
+        # Add metadata for Kubernetes
+        if cloudtype == "KUBERNETES":
+            for resource in node.get("resources", []):
+                pod = resource.get("data", {}).get("metadata", {}).get("name")
+                namespace = resource.get("data", {}).get("metadata", {}).get("namespace")
+                if pod is not None:
+                    description += "\n\n" + "**Pod Name** : " + pod
+                if namespace is not None:
+                    description += "\n\n" + "**Namespace** : " + namespace
+                return description
+        return description

dojo/tools/prowler/parser_json.py

@@ -0,0 +1,5 @@
+AUTH_METHOD;TIMESTAMP;ACCOUNT_UID;ACCOUNT_NAME;ACCOUNT_EMAIL;ACCOUNT_ORGANIZATION_UID;ACCOUNT_ORGANIZATION_NAME;ACCOUNT_TAGS;FINDING_UID;PROVIDER;CHECK_ID;CHECK_TITLE;CHECK_TYPE;STATUS;STATUS_EXTENDED;MUTED;SERVICE_NAME;SUBSERVICE_NAME;SEVERITY;RESOURCE_TYPE;RESOURCE_UID;RESOURCE_NAME;RESOURCE_DETAILS;RESOURCE_TAGS;PARTITION;REGION;DESCRIPTION;RISK;RELATED_URL;REMEDIATION_RECOMMENDATION_TEXT;REMEDIATION_RECOMMENDATION_URL;REMEDIATION_CODE_NATIVEIAC;REMEDIATION_CODE_TERRAFORM;REMEDIATION_CODE_CLI;REMEDIATION_CODE_OTHER;COMPLIANCE;CATEGORIES;DEPENDS_ON;RELATED_TO;NOTES;PROWLER_VERSION;ADDITIONAL_URLS
+<auth_method>;2025-02-14 14:27:03.913874;<account_uid>;;;;;;<finding_uid>;aws;accessanalyzer_enabled;Check if IAM Access Analyzer is enabled;IAM;FAIL;IAM Access Analyzer in account <account_uid> is not enabled.;False;accessanalyzer;;low;Other;<resource_uid>;<resource_name>;;;aws;<region>;Check if IAM Access Analyzer is enabled;AWS IAM Access Analyzer helps you identify the resources in your organization and accounts, such as Amazon S3 buckets or IAM roles, that are shared with an external entity. This lets you identify unintended access to your resources and data, which is a security risk. IAM Access Analyzer uses a form of mathematical analysis called automated reasoning, which applies logic and mathematical inference to determine all possible access paths allowed by a resource policy.;https://docs.aws.amazon.com/IAM/latest/UserGuide/what-is-access-analyzer.html;Enable IAM Access Analyzer for all accounts, create analyzer and take action over it is recommendations (IAM Access Analyzer is available at no additional cost).;https://docs.aws.amazon.com/IAM/latest/UserGuide/what-is-access-analyzer.html;;;aws accessanalyzer create-analyzer --analyzer-name <NAME> --type <ACCOUNT|ORGANIZATION>;;CIS-1.4: 1.20 | CIS-1.5: 1.20 | KISA-ISMS-P-2023: 2.5.6, 2.6.4, 2.8.1, 2.8.2 | CIS-2.0: 1.20 | KISA-ISMS-P-2023-korean: 2.5.6, 2.6.4, 2.8.1, 2.8.2 | AWS-Account-Security-Onboarding: Enabled security services, Create analyzers in each active regions, Verify that events are present in SecurityHub aggregated view | CIS-3.0: 1.20;;;;;<prowler_version>;https://docs.aws.amazon.com/IAM/latest/UserGuide/access-analyzer-getting-started.html | https://aws.amazon.com/iam/features/analyze-access/
+<auth_method>;2025-02-14 14:27:03.913874;<account_uid>;;;;;;<finding_uid>;aws;account_maintain_current_contact_details;Maintain current contact details.;IAM;MANUAL;Login to the AWS Console. Choose your account name on the top right of the window -> My Account -> Contact Information.;False;account;;medium;Other;<resource_uid>;<account_uid>;;;aws;<region>;Maintain current contact details.;Ensure contact email and telephone details for AWS accounts are current and map to more than one individual in your organization. An AWS account supports a number of contact details, and AWS will use these to contact the account owner if activity judged to be in breach of Acceptable Use Policy. If an AWS account is observed to be behaving in a prohibited or suspicious manner, AWS will attempt to contact the account owner by email and phone using the contact details listed. If this is unsuccessful and the account behavior needs urgent mitigation, proactive measures may be taken, including throttling of traffic between the account exhibiting suspicious behavior and the AWS API endpoints and the Internet. This will result in impaired service to and from the account in question.;;Using the Billing and Cost Management console complete contact details.;https://docs.aws.amazon.com/accounts/latest/reference/manage-acct-update-contact.html;;;No command available.;https://docs.prowler.com/checks/aws/iam-policies/iam_18-maintain-contact-details#aws-console;CIS-1.4: 1.1 | CIS-1.5: 1.1 | KISA-ISMS-P-2023: 2.1.3 | CIS-2.0: 1.1 | KISA-ISMS-P-2023-korean: 2.1.3 | AWS-Well-Architected-Framework-Security-Pillar: SEC03-BP03, SEC10-BP01 | AWS-Account-Security-Onboarding: Billing, emergency, security contacts | CIS-3.0: 1.1 | ENS-RD2022: op.ext.7.aws.am.1;;;;;<prowler_version>;https://docs.aws.amazon.com/IAM/latest/UserGuide/access-analyzer-getting-started.html | https://aws.amazon.com/iam/features/analyze-access/
+<auth_method>;2025-02-14 14:27:03.913874;<account_uid>;;;;;;<finding_uid>;aws;account_maintain_different_contact_details_to_security_billing_and_operations;Maintain different contact details to security, billing and operations.;IAM;FAIL;SECURITY, BILLING and OPERATIONS contacts not found or they are not different between each other and between ROOT contact.;False;account;;medium;Other;<resource_uid>;<account_uid>;;;aws;<region>;Maintain different contact details to security, billing and operations.;Ensure contact email and telephone details for AWS accounts are current and map to more than one individual in your organization. An AWS account supports a number of contact details, and AWS will use these to contact the account owner if activity judged to be in breach of Acceptable Use Policy. If an AWS account is observed to be behaving in a prohibited or suspicious manner, AWS will attempt to contact the account owner by email and phone using the contact details listed. If this is unsuccessful and the account behavior needs urgent mitigation, proactive measures may be taken, including throttling of traffic between the account exhibiting suspicious behavior and the AWS API endpoints and the Internet. This will result in impaired service to and from the account in question.;https://docs.aws.amazon.com/accounts/latest/reference/manage-acct-update-contact.html;Using the Billing and Cost Management console complete contact details.;https://docs.aws.amazon.com/accounts/latest/reference/manage-acct-update-contact.html;;;;https://docs.prowler.com/checks/aws/iam-policies/iam_18-maintain-contact-details#aws-console;KISA-ISMS-P-2023: 2.1.3 | KISA-ISMS-P-2023-korean: 2.1.3;;;;;<prowler_version>;https://docs.aws.amazon.com/IAM/latest/UserGuide/access-analyzer-getting-started.html | https://aws.amazon.com/iam/features/analyze-access/
+<auth_method>;2025-02-14 14:27:03.913874;<account_uid>;;;;;;<finding_uid>;aws;account_security_contact_information_is_registered;Ensure security contact information is registered.;IAM;MANUAL;Login to the AWS Console. Choose your account name on the top right of the window -> My Account -> Alternate Contacts -> Security Section.;False;account;;medium;Other;<resource_uid>:root;<account_uid>;;;aws;<region>;Ensure security contact information is registered.;AWS provides customers with the option of specifying the contact information for accounts security team. It is recommended that this information be provided. Specifying security-specific contact information will help ensure that security advisories sent by AWS reach the team in your organization that is best equipped to respond to them.;;Go to the My Account section and complete alternate contacts.;https://docs.aws.amazon.com/accounts/latest/reference/manage-acct-update-contact.html;;;No command available.;https://docs.prowler.com/checks/aws/iam-policies/iam_19#aws-console;CIS-1.4: 1.2 | CIS-1.5: 1.2 | AWS-Foundational-Security-Best-Practices: account, acm | KISA-ISMS-P-2023: 2.1.3, 2.2.1 | CIS-2.0: 1.2 | KISA-ISMS-P-2023-korean: 2.1.3, 2.2.1 | AWS-Well-Architected-Framework-Security-Pillar: SEC03-BP03, SEC10-BP01 | AWS-Account-Security-Onboarding: Billing, emergency, security contacts | CIS-3.0: 1.2 | ENS-RD2022: op.ext.7.aws.am.1;;;;;<prowler_version>;https://docs.aws.amazon.com/IAM/latest/UserGuide/access-analyzer-getting-started.html | https://aws.amazon.com/iam/features/analyze-access/