awssecurityhub: introduce support for 'Security Hub', 'IAM Access Analyzer', 'Health', 'Config' products. fail if unsupported one is imported

Author: pna-nca

dojo/tools/awssecurityhub/config.py

@@ -0,0 +1,45 @@
+from dojo.models import Finding
+
+
+class Config:
+    def get_item(self, finding: dict, test):
+        finding_id = finding.get("Id", "")
+        title = finding.get("Title", "")
+        severity = finding.get("Severity", {}).get("Label", "INFORMATIONAL").title()
+        resource_arns = [arn for resource in finding.get("Resources", [])
+                                if (arn := resource.get("Id"))]
+        impact = []
+        unsaved_vulnerability_ids = []
+        epss_score = None
+        description = "This is a Config Finding \n" + finding.get("Description", "") + "\n"
+        description += f"**AWS Finding ARN:** {finding_id}\n"
+        description += f"**Resource IDs:** {', '.join(set(resource_arns))}\n"
+        description += f"**AwsAccountId:** {finding.get('AwsAccountId', '')}\n"
+        if finding.get("Region"):
+            description += f"**Region:** {finding.get('Region', '')}\n"
+        description += f"**Generator ID:** {finding.get('GeneratorId', '')}\n"
+        title_suffix = ""
+        for resource in finding.get("Resources", []):
+            component_name = resource.get("Type")
+            resource_id = resource["Id"].split(":")[-1]
+            impact.append(f"Resource: {resource_id}")
+            title_suffix = f" - Resource: {resource_id}"
+        false_p = False
+        result = Finding(
+            title=f"{title}{title_suffix}",
+            test=test,
+            description=description,
+            severity=severity,
+            impact="\n".join(impact),
+            verified=False,
+            false_p=false_p,
+            unique_id_from_tool=finding_id,
+            static_finding=True,
+            dynamic_finding=False,
+            component_name=component_name,
+        )
+        if epss_score is not None:
+            result.epss_score = epss_score
+        # Add the unsaved vulnerability ids
+        result.unsaved_vulnerability_ids = unsaved_vulnerability_ids
+        return result

dojo/tools/awssecurityhub/health.py

@@ -0,0 +1,50 @@
+import datetime
+
+from dojo.models import Finding
+
+
+class Health:
+    def get_item(self, finding: dict, test):
+        finding_id = finding.get("Id", "")
+        title = finding.get("Title", "")
+        severity = finding.get("Severity", {}).get("Label", "INFORMATIONAL").title()
+        resource_arns = [arn for resource in finding.get("Resources", [])
+                                if (arn := resource.get("Id"))]
+        impact = []
+        references = []
+        unsaved_vulnerability_ids = []
+        epss_score = None
+        description = "This is a Health Finding \n" + finding.get("Description", "") + "\n"
+        description += f"**AWS Finding ARN:** {finding_id}\n"
+        description += f"**Resource IDs:** {', '.join(set(resource_arns))}\n"
+        description += f"**AwsAccountId:** {finding.get('AwsAccountId', '')}\n"
+        if finding.get("Region"):
+            description += f"**Region:** {finding.get('Region', '')}\n"
+        description += f"**Generator ID:** {finding.get('GeneratorId', '')}\n"
+        title_suffix = ""
+        for resource in finding.get("Resources", []):
+            component_name = resource.get("Type")
+            resource_id = resource["Id"].split(":")[-1]
+            impact.append(f"Resource: {resource_id}")
+            title_suffix = f" - Resource: {resource_id}"
+        references.append((finding.get("SourceUrl", "") or ""))
+        false_p = False
+        result = Finding(
+            title=f"{title}{title_suffix}",
+            test=test,
+            description=description,
+            references="\n".join(references),
+            severity=severity,
+            impact="\n".join(impact),
+            verified=False,
+            false_p=false_p,
+            unique_id_from_tool=finding_id,
+            static_finding=True,
+            dynamic_finding=False,
+            component_name=component_name,
+        )
+        if epss_score is not None:
+            result.epss_score = epss_score
+        # Add the unsaved vulnerability ids
+        result.unsaved_vulnerability_ids = unsaved_vulnerability_ids
+        return result

dojo/tools/awssecurityhub/iam_access_analyzer.py

@@ -0,0 +1,51 @@
+from dojo.models import Finding
+
+
+class IamAccessAnalyzer:
+    def get_item(self, finding: dict, test):
+        finding_id = finding.get("Id", "")
+        title = finding.get("Title", "")
+        severity = finding.get("Severity", {}).get("Label", "INFORMATIONAL").title()
+        resource_arns = [arn for resource in finding.get("Resources", [])
+                         if (arn := resource.get("Id"))]
+        impact = []
+        references = []
+        unsaved_vulnerability_ids = []
+        epss_score = None
+        mitigation = finding.get("Remediation", {}).get("Recommendation", {}).get("Text", "")
+        mitigation += "\n" + (finding.get("Remediation", {}).get("Recommendation", {}).get("Url", "") or "")
+        description = "This is an IAM Access Analyzer Finding \n" + finding.get("Description", "") + "\n"
+        description += f"**AWS Finding ARN:** {finding_id}\n"
+        description += f"**Resource IDs:** {', '.join(set(resource_arns))}\n"
+        description += f"**AwsAccountId:** {finding.get('AwsAccountId', '')}\n"
+        if finding.get("Region"):
+            description += f"**Region:** {finding.get('Region', '')}\n"
+        description += f"**Generator ID:** {finding.get('GeneratorId', '')}\n"
+        title_suffix = ""
+        for resource in finding.get("Resources", []):
+            resource_id = resource["Id"].split(":")[-1]
+            impact.append(f"Resource: {resource_id}")
+            title_suffix = f" - Resource: {resource_id}"
+        if remediation_rec_url := finding.get("Remediation", {}).get("Recommendation", {}).get("Url"):
+            references.append(remediation_rec_url)
+        result = Finding(
+            title=f"{title}{title_suffix}",
+            test=test,
+            description=description,
+            mitigation=mitigation,
+            references="\n".join(references),
+            severity=severity,
+            impact="\n".join(impact),
+            active=True,
+            verified=False,
+            false_p=False,
+            unique_id_from_tool=finding_id,
+            is_mitigated=False,
+            static_finding=True,
+            dynamic_finding=False,
+        )
+        if epss_score is not None:
+            result.epss_score = epss_score
+        # Add the unsaved vulnerability ids
+        result.unsaved_vulnerability_ids = unsaved_vulnerability_ids
+        return result

dojo/tools/awssecurityhub/parser.py

@@ -1,8 +1,12 @@
 import json
 
 from dojo.tools.awssecurityhub.compliance import Compliance
+from dojo.tools.awssecurityhub.config import Config
 from dojo.tools.awssecurityhub.guardduty import GuardDuty
+from dojo.tools.awssecurityhub.iam_access_analyzer import IamAccessAnalyzer
 from dojo.tools.awssecurityhub.inspector import Inspector
+from dojo.tools.awssecurityhub.health import Health
+from dojo.tools.awssecurityhub.securityhub import SecurityHub
 from dojo.tools.parser_test import ParserTest
 
 
@@ -57,8 +61,19 @@ def get_items(self, tree: dict, test):
                 item = Inspector().get_item(node, test)
             elif aws_scanner_type == "GuardDuty":
                 item = GuardDuty().get_item(node, test)
-            else:
+            elif aws_scanner_type == "Compliance":
                 item = Compliance().get_item(node, test)
+            elif aws_scanner_type == "IAM Access Analyzer":
+                item = IamAccessAnalyzer().get_item(node, test)
+            elif aws_scanner_type == "Health":
+                item = Health().get_item(node, test)
+            elif aws_scanner_type == "Config":
+                item = Config().get_item(node, test)
+            elif aws_scanner_type == "Security Hub":
+                item = SecurityHub().get_item(node, test)
+            else:
+                msg = "Unsupported Security Hub report format"
+                raise TypeError(msg)
             key = node["Id"]
             if not isinstance(key, str):
                 msg = "Incorrect Security Hub report format"

dojo/tools/awssecurityhub/securityhub.py

@@ -0,0 +1,71 @@
+import datetime
+
+from dojo.models import Finding
+
+
+class SecurityHub:
+    def get_item(self, finding: dict, test):
+        finding_id = finding.get("Id", "")
+        title = finding.get("Title", "")
+        severity = finding.get("Severity", {}).get("Label", "INFORMATIONAL").title()
+        mitigation = ""
+        impact = []
+        references = []
+        unsaved_vulnerability_ids = []
+        epss_score = None
+        mitigations = finding.get("FindingProviderFields", {}).get("Types")
+        for mitigate in mitigations:
+            mitigation += mitigate + "\n"
+        active = True
+        if finding.get("RecordState") == "ACTIVE":
+            is_Mitigated = False
+            mitigated = None
+        else:
+            is_Mitigated = True
+            if finding.get("LastObservedAt"):
+                try:
+                    mitigated = datetime.datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%S.%fZ")
+                except Exception:
+                    mitigated = datetime.datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%fZ")
+            else:
+                mitigated = datetime.datetime.now(datetime.UTC)
+        description = f"This is a Security Hub Finding\n{finding.get('Description', '')}" + "\n"
+        description += f"**AWS Finding ARN:** {finding_id}\n"
+        description += f"**AwsAccountId:** {finding.get('AwsAccountId', '')}\n"
+        description += f"**Region:** {finding.get('Region', '')}\n"
+        description += f"**Generator ID:** {finding.get('GeneratorId', '')}\n"
+        title_suffix = ""
+        hosts = []
+        for resource in finding.get("Resources", []):
+            component_name = resource.get("Type")
+            resource_id = resource["Id"].split(":")[-1]
+            impact.append(f"Resource: {resource_id}")
+            title_suffix = f" - Resource: {resource_id}"
+        if remediation_rec_url := finding.get("Remediation", {}).get("Recommendation", {}).get("Url"):
+            references.append(remediation_rec_url)
+        false_p = False
+        result = Finding(
+            title=f"{title}{title_suffix}",
+            test=test,
+            description=description,
+            mitigation=mitigation,
+            references="\n".join(references),
+            severity=severity,
+            impact="\n".join(impact),
+            active=active,
+            verified=False,
+            false_p=false_p,
+            unique_id_from_tool=finding_id,
+            mitigated=mitigated,
+            is_mitigated=is_Mitigated,
+            static_finding=True,
+            dynamic_finding=False,
+            component_name=component_name,
+        )
+        result.unsaved_endpoints = []
+        result.unsaved_endpoints.extend(hosts)
+        if epss_score is not None:
+            result.epss_score = epss_score
+        # Add the unsaved vulnerability ids
+        result.unsaved_vulnerability_ids = unsaved_vulnerability_ids
+        return result