From 97a8a5683105bdd36cb7c45e8dba7a6778cf1b6b Mon Sep 17 00:00:00 2001 From: manuelsommer <47991713+manuel-sommer@users.noreply.github.com> Date: Thu, 28 Mar 2024 02:28:09 +0100 Subject: [PATCH] Refactor awssecurityhub and add endpoint (#9814) * rework_awssecurityhub * add endpoints for inspector * add endpoints for guardduty * :lipstick: --- dojo/tools/awssecurityhub/compliance.py | 72 +++++++++ dojo/tools/awssecurityhub/guardduty.py | 82 ++++++++++ dojo/tools/awssecurityhub/inspector.py | 94 +++++++++++ dojo/tools/awssecurityhub/parser.py | 152 ++---------------- unittests/tools/test_awssecurityhub_parser.py | 8 + 5 files changed, 266 insertions(+), 142 deletions(-) create mode 100644 dojo/tools/awssecurityhub/compliance.py create mode 100644 dojo/tools/awssecurityhub/guardduty.py create mode 100644 dojo/tools/awssecurityhub/inspector.py diff --git a/dojo/tools/awssecurityhub/compliance.py b/dojo/tools/awssecurityhub/compliance.py new file mode 100644 index 00000000000..3898442d69e --- /dev/null +++ b/dojo/tools/awssecurityhub/compliance.py @@ -0,0 +1,72 @@ +from datetime import datetime +from dojo.models import Finding + + +class Compliance(object): + 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 + mitigation = finding.get("Remediation", {}).get("Recommendation", {}).get("Text", "") + description = "This is a Security Hub Finding \n" + finding.get("Description", "") + if finding.get("Compliance", {}).get("Status", "PASSED") == "PASSED": + is_Mitigated = True + active = False + if finding.get("LastObservedAt", None): + try: + mitigated = datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%S.%fZ") + except Exception: + mitigated = datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%fZ") + else: + mitigated = datetime.utcnow() + else: + mitigated = None + is_Mitigated = False + active = True + title_suffix = "" + for resource in finding.get("Resources", []): + component_name = resource.get("Type") + if component_name == "AwsEcrContainerImage": + details = resource.get("Details", {}).get("AwsEcrContainerImage") + arn = resource.get("Id") + if details: + impact.append(f"Image ARN: {arn}") + impact.append(f"Registry: {details.get('RegistryId')}") + impact.append(f"Repository: {details.get('RepositoryName')}") + impact.append(f"Image digest: {details.get('ImageDigest')}") + title_suffix = f" - Image: {arn.split('/', 1)[1]}" # repo-name/sha256:digest + else: # generic implementation + 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, + ) + 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 diff --git a/dojo/tools/awssecurityhub/guardduty.py b/dojo/tools/awssecurityhub/guardduty.py new file mode 100644 index 00000000000..3b22498ddc3 --- /dev/null +++ b/dojo/tools/awssecurityhub/guardduty.py @@ -0,0 +1,82 @@ +from datetime import datetime +from dojo.models import Finding, Endpoint + + +class GuardDuty(object): + 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" + mitigation += "https://docs.aws.amazon.com/guardduty/latest/ug/guardduty_finding-types-active.html" + active = True + if finding.get("RecordState") == "ACTIVE": + is_Mitigated = False + mitigated = None + else: + is_Mitigated = True + if finding.get("LastObservedAt", None): + try: + mitigated = datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%S.%fZ") + except Exception: + mitigated = datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%fZ") + else: + mitigated = datetime.utcnow() + description = f"This is a GuardDuty Finding\n{finding.get('Description', '')}" + description += f"SourceURL: {finding.get('SourceUrl', '')}\n" + description += f"AwsAccountId: {finding.get('AwsAccountId', '')}\n" + description += f"Region: {finding.get('Region', '')}\n" + title_suffix = "" + hosts = list() + for resource in finding.get("Resources", []): + component_name = resource.get("Type") + if component_name in ("AwsEcrContainerImage", "AwsEc2Instance"): + hosts.append(Endpoint(host=f"{component_name} {resource.get('Id')}")) + if component_name == "AwsEcrContainerImage": + details = resource.get("Details", {}).get("AwsEcrContainerImage") + arn = resource.get("Id") + if details: + impact.append(f"Image ARN: {arn}") + impact.append(f"Registry: {details.get('RegistryId')}") + impact.append(f"Repository: {details.get('RepositoryName')}") + impact.append(f"Image digest: {details.get('ImageDigest')}") + title_suffix = f" - Image: {arn.split('/', 1)[1]}" # repo-name/sha256:digest + else: # generic implementation + 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 = list() + 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 diff --git a/dojo/tools/awssecurityhub/inspector.py b/dojo/tools/awssecurityhub/inspector.py new file mode 100644 index 00000000000..2c4c79db4ed --- /dev/null +++ b/dojo/tools/awssecurityhub/inspector.py @@ -0,0 +1,94 @@ +from datetime import datetime +from dojo.models import Finding, Endpoint + + +class Inspector(object): + 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 + description = f"This is an Inspector Finding\n{finding.get('Description', '')}" + vulnerabilities = finding.get("Vulnerabilities", []) + for vulnerability in vulnerabilities: + # Save the CVE if it is present + if cve := vulnerability.get("Id"): + unsaved_vulnerability_ids.append(cve) + for alias in vulnerability.get("RelatedVulnerabilities", []): + if alias != cve: + unsaved_vulnerability_ids.append(alias) + # Add information about the vulnerable packages to the description and mitigation + vulnerable_packages = vulnerability.get("VulnerablePackages", []) + for package in vulnerable_packages: + mitigation += f"- Update {package.get('Name', '')}-{package.get('Version', '')}\n" + if remediation := package.get("Remediation"): + mitigation += f"\t- {remediation}\n" + if vendor := vulnerability.get("Vendor"): + if vendor_url := vendor.get("Url"): + references.append(vendor_url) + if vulnerability.get("EpssScore") is not None: + epss_score = vulnerability.get("EpssScore") + if finding.get("ProductFields", {}).get("aws/inspector/FindingStatus", "ACTIVE") == "ACTIVE": + mitigated = None + is_Mitigated = False + active = True + else: + is_Mitigated = True + active = False + if finding.get("LastObservedAt", None): + try: + mitigated = datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%S.%fZ") + except Exception: + mitigated = datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%fZ") + else: + mitigated = datetime.utcnow() + title_suffix = "" + hosts = list() + for resource in finding.get("Resources", []): + component_name = resource.get("Type") + hosts.append(Endpoint(host=f"{component_name} {resource.get('Id')}")) + if component_name == "AwsEcrContainerImage": + details = resource.get("Details", {}).get("AwsEcrContainerImage") + arn = resource.get("Id") + if details: + impact.append(f"Image ARN: {arn}") + impact.append(f"Registry: {details.get('RegistryId')}") + impact.append(f"Repository: {details.get('RepositoryName')}") + impact.append(f"Image digest: {details.get('ImageDigest')}") + title_suffix = f" - Image: {arn.split('/', 1)[1]}" # repo-name/sha256:digest + else: # generic implementation + 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 = list() + 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 diff --git a/dojo/tools/awssecurityhub/parser.py b/dojo/tools/awssecurityhub/parser.py index d697ad6bfba..7380ece6954 100644 --- a/dojo/tools/awssecurityhub/parser.py +++ b/dojo/tools/awssecurityhub/parser.py @@ -1,6 +1,7 @@ import json -from datetime import datetime -from dojo.models import Finding +from dojo.tools.awssecurityhub.inspector import Inspector +from dojo.tools.awssecurityhub.guardduty import GuardDuty +from dojo.tools.awssecurityhub.compliance import Compliance class AwsSecurityHubParser(object): @@ -22,152 +23,19 @@ def get_findings(self, filehandle, test): def get_items(self, tree: dict, test): items = {} - # DefectDojo/django-DefectDojo/issues/2780 findings = tree.get("Findings", tree.get("findings", None)) - if not isinstance(findings, list): raise TypeError("Incorrect Security Hub report format") - for node in findings: - item = get_item(node, test) + aws_scanner_type = node.get("ProductFields", {}).get("aws/securityhub/ProductName", "") + if aws_scanner_type == "Inspector": + item = Inspector().get_item(node, test) + elif aws_scanner_type == "GuardDuty": + item = GuardDuty().get_item(node, test) + else: + item = Compliance().get_item(node, test) key = node["Id"] if not isinstance(key, str): raise TypeError("Incorrect Security Hub report format") items[key] = item - return list(items.values()) - - -def get_item(finding: dict, test): - aws_scanner_type = finding.get("ProductFields", {}).get("aws/securityhub/ProductName", "") - 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 - if aws_scanner_type == "Inspector": - description = f"This is an Inspector Finding\n{finding.get('Description', '')}" - vulnerabilities = finding.get("Vulnerabilities", []) - for vulnerability in vulnerabilities: - # Save the CVE if it is present - if cve := vulnerability.get("Id"): - unsaved_vulnerability_ids.append(cve) - for alias in vulnerability.get("RelatedVulnerabilities", []): - if alias != cve: - unsaved_vulnerability_ids.append(alias) - # Add information about the vulnerable packages to the description and mitigation - vulnerable_packages = vulnerability.get("VulnerablePackages", []) - for package in vulnerable_packages: - mitigation += f"- Update {package.get('Name', '')}-{package.get('Version', '')}\n" - if remediation := package.get("Remediation"): - mitigation += f"\t- {remediation}\n" - if vendor := vulnerability.get("Vendor"): - if vendor_url := vendor.get("Url"): - references.append(vendor_url) - if vulnerability.get("EpssScore") is not None: - epss_score = vulnerability.get("EpssScore") - - if finding.get("ProductFields", {}).get("aws/inspector/FindingStatus", "ACTIVE") == "ACTIVE": - mitigated = None - is_Mitigated = False - active = True - else: - is_Mitigated = True - active = False - if finding.get("LastObservedAt", None): - try: - mitigated = datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%S.%fZ") - except Exception: - mitigated = datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%fZ") - else: - mitigated = datetime.utcnow() - elif aws_scanner_type == "GuardDuty": - mitigations = finding.get("FindingProviderFields", {}).get("Types") - for mitigate in mitigations: - mitigation += mitigate + "\n" - mitigation += "https://docs.aws.amazon.com/guardduty/latest/ug/guardduty_finding-types-active.html" - active = True - if finding.get("RecordState") == "ACTIVE": - is_Mitigated = False - mitigated = None - else: - is_Mitigated = True - if finding.get("LastObservedAt", None): - try: - mitigated = datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%S.%fZ") - except Exception: - mitigated = datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%fZ") - else: - mitigated = datetime.utcnow() - description = f"This is a GuardDuty Finding\n{finding.get('Description', '')}" - description += f"SourceURL: {finding.get('SourceUrl', '')}\n" - description += f"AwsAccountId: {finding.get('AwsAccountId', '')}\n" - description += f"Region: {finding.get('Region', '')}\n" - else: - mitigation = finding.get("Remediation", {}).get("Recommendation", {}).get("Text", "") - description = "This is a Security Hub Finding \n" + finding.get("Description", "") - if finding.get("Compliance", {}).get("Status", "PASSED") == "PASSED": - is_Mitigated = True - active = False - if finding.get("LastObservedAt", None): - try: - mitigated = datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%S.%fZ") - except Exception: - mitigated = datetime.strptime(finding.get("LastObservedAt"), "%Y-%m-%dT%H:%M:%fZ") - else: - mitigated = datetime.utcnow() - else: - mitigated = None - is_Mitigated = False - active = True - - title_suffix = "" - for resource in finding.get("Resources", []): - component_name = resource.get("Type") - if component_name == "AwsEcrContainerImage": - details = resource.get("Details", {}).get("AwsEcrContainerImage") - arn = resource.get("Id") - if details: - impact.append(f"Image ARN: {arn}") - impact.append(f"Registry: {details.get('RegistryId')}") - impact.append(f"Repository: {details.get('RepositoryName')}") - impact.append(f"Image digest: {details.get('ImageDigest')}") - title_suffix = f" - Image: {arn.split('/', 1)[1]}" # repo-name/sha256:digest - else: # generic implementation - 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, - ) - - 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 diff --git a/unittests/tools/test_awssecurityhub_parser.py b/unittests/tools/test_awssecurityhub_parser.py index 93c0ab8a46b..5f3621a609c 100644 --- a/unittests/tools/test_awssecurityhub_parser.py +++ b/unittests/tools/test_awssecurityhub_parser.py @@ -66,6 +66,8 @@ def test_inspector_ec2(self): self.assertEqual(1, len(finding.unsaved_vulnerability_ids)) self.assertEqual("CVE-2022-3643", finding.unsaved_vulnerability_ids[0]) self.assertEqual("- Update kernel-4.14.301\n\t- yum update kernel\n", finding.mitigation) + endpoint = finding.unsaved_endpoints[0] + self.assertEqual('AwsEc2Instance arn:aws:ec2:us-east-1:XXXXXXXXXXXX:i-11111111111111111', endpoint.host) def test_inspector_ec2_with_no_vulnerabilities(self): with open(get_unit_tests_path() + sample_path("inspector_ec2_cve_no_vulnerabilities.json")) as test_file: @@ -87,6 +89,8 @@ def test_inspector_ec2_ghsa(self): self.assertIn("GHSA-p98r-538v-jgw5", finding.title) self.assertSetEqual({"CVE-2023-34256", "GHSA-p98r-538v-jgw5"}, set(finding.unsaved_vulnerability_ids)) self.assertEqual("https://github.com/bottlerocket-os/bottlerocket/security/advisories/GHSA-p98r-538v-jgw5", finding.references) + endpoint = finding.unsaved_endpoints[0] + self.assertEqual('AwsEc2Instance arn:aws:ec2:eu-central-1:012345678912:instance/i-07c11cc535d830123', endpoint.host) def test_inspector_ecr(self): with open(get_unit_tests_path() + sample_path("inspector_ecr.json")) as test_file: @@ -102,6 +106,8 @@ def test_inspector_ecr(self): self.assertIn("repo-os/sha256:af965ef68c78374a5f987fce98c0ddfa45801df2395bf012c50b863e65978d74", finding.impact) self.assertIn("Repository: repo-os", finding.impact) self.assertEqual(0.0014, finding.epss_score) + endpoint = finding.unsaved_endpoints[0] + self.assertEqual('AwsEcrContainerImage arn:aws:ecr:eu-central-1:123456789012:repository/repo-os/sha256:af965ef68c78374a5f987fce98c0ddfa45801df2395bf012c50b863e65978d74', endpoint.host) def test_guardduty(self): with open(get_unit_tests_path() + sample_path("guardduty.json")) as test_file: @@ -116,3 +122,5 @@ def test_guardduty(self): self.assertTrue(finding.active) self.assertEqual("User AssumedRole : 123123123 is anomalously invoking APIs commonly used in Discovery tactics. - Resource: 123123123", finding.title) self.assertEqual("TTPs/Discovery/IAMUser-AnomalousBehavior\nhttps://docs.aws.amazon.com/guardduty/latest/ug/guardduty_finding-types-active.html", finding.mitigation) + endpoint = findings[0].unsaved_endpoints[0] + self.assertEqual('AwsEc2Instance arn:aws:ec2:us-east-1:123456789012:instance/i-1234567890', endpoint.host)