From 673619c8a1ca8b9e651002e7f986d0f9969bf4ff Mon Sep 17 00:00:00 2001 From: Pepe Fagoaga Date: Thu, 4 Jul 2024 18:04:36 +0200 Subject: [PATCH] refactor(ASFF): create class (#4368) Co-authored-by: pedrooot --- prowler/__main__.py | 31 +- .../outputs/{json_asff => asff}/__init__.py | 0 prowler/lib/outputs/asff/asff.py | 426 ++++++++++++++++ prowler/lib/outputs/csv/models.py | 3 +- prowler/lib/outputs/file_descriptors.py | 115 ++--- prowler/lib/outputs/json_asff/json_asff.py | 137 ----- prowler/lib/outputs/json_asff/models.py | 83 --- prowler/lib/outputs/ocsf/ocsf.py | 4 +- prowler/lib/outputs/output.py | 33 +- prowler/lib/outputs/outputs.py | 16 - .../aws/lib/security_hub/security_hub.py | 122 +++-- tests/lib/outputs/asff/asff_test.py | 448 +++++++++++++++++ tests/lib/outputs/csv/csv_test.py | 5 +- tests/lib/outputs/fixtures/fixtures.py | 33 +- tests/lib/outputs/json_asff/json_asff_test.py | 472 ------------------ tests/lib/outputs/ocsf/ocsf_test.py | 30 +- tests/lib/outputs/outputs_test.py | 24 +- .../aws/lib/security_hub/security_hub_test.py | 374 ++++++-------- 18 files changed, 1255 insertions(+), 1101 deletions(-) rename prowler/lib/outputs/{json_asff => asff}/__init__.py (100%) create mode 100644 prowler/lib/outputs/asff/asff.py delete mode 100644 prowler/lib/outputs/json_asff/json_asff.py delete mode 100644 prowler/lib/outputs/json_asff/models.py create mode 100644 tests/lib/outputs/asff/asff_test.py delete mode 100644 tests/lib/outputs/json_asff/json_asff_test.py diff --git a/prowler/__main__.py b/prowler/__main__.py index bc5c832a4d..d7f1436719 100644 --- a/prowler/__main__.py +++ b/prowler/__main__.py @@ -9,6 +9,7 @@ from colorama import Fore, Style from prowler.config.config import ( csv_file_suffix, get_available_compliance_frameworks, + json_asff_file_suffix, json_ocsf_file_suffix, ) from prowler.lib.banner import print_banner @@ -40,6 +41,7 @@ from prowler.lib.check.custom_checks_metadata import ( ) from prowler.lib.cli.parser import ProwlerArgumentParser from prowler.lib.logger import logger, set_logging_config +from prowler.lib.outputs.asff.asff import ASFF from prowler.lib.outputs.compliance.compliance import display_compliance_table from prowler.lib.outputs.csv.models import CSV from prowler.lib.outputs.finding import Finding @@ -51,7 +53,7 @@ from prowler.lib.outputs.summary_table import display_summary_table from prowler.providers.aws.lib.s3.s3 import send_to_s3_bucket from prowler.providers.aws.lib.security_hub.security_hub import ( batch_send_to_security_hub, - prepare_security_hub_findings, + filter_security_hub_findings_per_region, resolve_security_hub_previous_findings, verify_security_hub_integration_enabled_per_region, ) @@ -295,20 +297,32 @@ def prowler(): if args.output_formats: for mode in args.output_formats: - if "csv" in mode: # Generate CSV Finding Object filename = ( f"{global_provider.output_options.output_directory}/" f"{global_provider.output_options.output_filename}{csv_file_suffix}" ) - csv_finding = CSV( + csv_output = CSV( findings=finding_outputs, create_file_descriptor=True, file_path=filename, ) # Write CSV Finding Object to file - csv_finding.batch_write_data_to_file() + csv_output.batch_write_data_to_file() + + if "json-asff" in mode: + filename = ( + f"{global_provider.output_options.output_directory}/" + f"{global_provider.output_options.output_filename}{json_asff_file_suffix}" + ) + asff_output = ASFF( + findings=finding_outputs, + create_file_descriptor=True, + file_path=filename, + ) + # Write ASFF Finding Object to file + asff_output.batch_write_data_to_file() # Close json file if exists # TODO: generate JSON here @@ -376,13 +390,12 @@ def prowler(): aws_security_enabled_regions.append(region) # Prepare the findings to be sent to Security Hub - security_hub_findings_per_region = prepare_security_hub_findings( - findings, - global_provider, - global_provider.output_options, + security_hub_findings_per_region = filter_security_hub_findings_per_region( + asff_output.data, + global_provider.output_options.send_sh_only_fails, + global_provider.output_options.status, aws_security_enabled_regions, ) - # Send the findings to Security Hub findings_sent_to_security_hub = batch_send_to_security_hub( security_hub_findings_per_region, global_provider.session.current_session diff --git a/prowler/lib/outputs/json_asff/__init__.py b/prowler/lib/outputs/asff/__init__.py similarity index 100% rename from prowler/lib/outputs/json_asff/__init__.py rename to prowler/lib/outputs/asff/__init__.py diff --git a/prowler/lib/outputs/asff/asff.py b/prowler/lib/outputs/asff/asff.py new file mode 100644 index 0000000000..2f6e1f657f --- /dev/null +++ b/prowler/lib/outputs/asff/asff.py @@ -0,0 +1,426 @@ +from json import dump +from os import SEEK_SET +from typing import Optional + +from pydantic import BaseModel, validator + +from prowler.config.config import prowler_version, timestamp_utc +from prowler.lib.logger import logger +from prowler.lib.outputs.finding import Finding +from prowler.lib.outputs.output import Output +from prowler.lib.utils.utils import hash_sha512 + + +class ASFF(Output): + """ + ASFF class represents a transformation of findings into AWS Security Finding Format (ASFF). + + This class provides methods to transform a list of findings into the ASFF format required by AWS Security Hub. It includes operations such as generating unique identifiers, formatting timestamps, handling compliance frameworks, and ensuring the status values match the allowed values in ASFF. + + Attributes: + - _data: A list to store the transformed findings. + - _file_descriptor: A file descriptor to write to file. + + Methods: + - transform(findings: list[Finding]) -> None: Transforms a list of findings into ASFF format. + - batch_write_data_to_file() -> None: Writes the findings data to a file in JSON ASFF format. + - generate_status(status: str, muted: bool = False) -> str: Generates the ASFF status based on the provided status and muted flag. + - format_resource_tags(tags: str) -> dict: Transforms a string of tags into a dictionary format. + + References: + - AWS Security Hub API Reference: https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html + - AWS Security Finding Format Syntax: https://docs.aws.amazon.com/securityhub/latest/userguide/securityhub-findings-format-syntax.html + """ + + def transform(self, findings: list[Finding]) -> None: + """ + Transforms a list of findings into AWS Security Finding Format (ASFF). + + This method iterates over the list of findings provided as input and transforms each finding into the ASFF format required by AWS Security Hub. It performs several operations for each finding, including generating unique identifiers, formatting timestamps, handling compliance frameworks, and ensuring the status values match the allowed values in ASFF. + + Parameters: + - findings (list[Finding]): A list of Finding objects representing the findings to be transformed. + + Returns: + - None + + Notes: + - The method skips findings with a status of "MANUAL" as it is not valid in SecurityHub. + - It generates unique identifiers for each finding based on specific attributes. + - It formats timestamps in the required ASFF format. + - It handles compliance frameworks and associated standards for each finding. + - It ensures that the finding status matches the allowed values in ASFF. + + References: + - AWS Security Hub API Reference: https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html + - AWS Security Finding Format Syntax: https://docs.aws.amazon.com/securityhub/latest/userguide/securityhub-findings-format-syntax.html + """ + try: + for finding in findings: + # MANUAL status is not valid in SecurityHub + # https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html + if finding.status == "MANUAL": + continue + timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ") + resource_tags = ASFF.format_resource_tags(finding.resource_tags) + + associated_standards, compliance_summary = ASFF.format_compliance( + finding.compliance + ) + + # Ensures finding_status matches allowed values in ASFF + finding_status = ASFF.generate_status(finding.status, finding.muted) + + self._data.append( + AWSSecurityFindingFormat( + # The following line cannot be changed because it is the format we use to generate unique findings for AWS Security Hub + # If changed some findings could be lost because the unique identifier will be different + Id=f"prowler-{finding.check_id}-{finding.account_uid}-{finding.region}-{hash_sha512(finding.resource_uid)}", + ProductArn=f"arn:{finding.partition}:securityhub:{finding.region}::product/prowler/prowler", + ProductFields=ProductFields( + ProwlerResourceName=finding.resource_uid, + ), + GeneratorId="prowler-" + finding.check_id, + AwsAccountId=finding.account_uid, + Types=finding.check_type.split(","), + FirstObservedAt=timestamp, + UpdatedAt=timestamp, + CreatedAt=timestamp, + Severity=Severity(Label=finding.severity.value), + Title=finding.check_title, + Description=finding.description, + Resources=[ + Resource( + Id=finding.resource_uid, + Type=finding.resource_type, + Partition=finding.partition, + Region=finding.region, + Tags=resource_tags, + ) + ], + Compliance=Compliance( + Status=finding_status, + AssociatedStandards=associated_standards, + RelatedRequirements=compliance_summary, + ), + Remediation=Remediation( + Recommendation=Recommendation( + Text=finding.remediation_recommendation_text, + Url=finding.remediation_recommendation_url, + ) + ), + ) + ) + + except Exception as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + + def batch_write_data_to_file(self) -> None: + """ + Writes the findings data to a file in JSON ASFF format. + + This method iterates over the findings data stored in the '_data' attribute and writes it to the file descriptor '_file_descriptor' in JSON format. It starts by writing the JSON opening/header '[', then iterates over each finding, dumping it to the file with an indent of 4 spaces. After writing all findings, it writes the closing ']' to complete the JSON array structure. Finally, it closes the file descriptor. + + Returns: + None + """ + try: + if self._file_descriptor and not self._file_descriptor.closed: + # Write JSON opening/header [ + self._file_descriptor.write("[") + + # Write findings + for finding in self._data: + dump( + finding.dict(exclude_none=True), + self._file_descriptor, + indent=4, + ) + self._file_descriptor.write(",") + + # Write footer/closing ] + if self._file_descriptor.tell() > 0: + if self._file_descriptor.tell() != 1: + self._file_descriptor.seek( + self._file_descriptor.tell() - 1, SEEK_SET + ) + self._file_descriptor.truncate() + self._file_descriptor.write("]") + + # Close file descriptor + self._file_descriptor.close() + except Exception as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + + @staticmethod + def generate_status(status: str, muted: bool = False) -> str: + """ + Generates the ASFF status based on the provided status and muted flag. + + Parameters: + - status (str): The status of the finding. + - muted (bool): Flag indicating if the finding is muted. + + Returns: + - str: The ASFF status corresponding to the provided status and muted flag. + + References: + - AWS Security Hub API Reference: https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html + """ + json_asff_status = "" + if muted: + # Per AWS Security Hub "MUTED" is not a valid status + # https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html + json_asff_status = "WARNING" + else: + if status == "PASS": + json_asff_status = "PASSED" + elif status == "FAIL": + json_asff_status = "FAILED" + else: + # MANUAL is set to NOT_AVAILABLE + json_asff_status = "NOT_AVAILABLE" + + return json_asff_status + + @staticmethod + def format_resource_tags(tags: str) -> dict: + """ + Transforms a string of tags into a dictionary format. + + Parameters: + - tags (str): A string containing tags separated by ' | ' and key-value pairs separated by '='. + + Returns: + - dict: A dictionary where keys are tag names and values are tag values. + + Notes: + - If the input string is empty or None, it returns None. + - Each tag in the input string should be in the format 'key=value'. + - If the input string is not formatted correctly, it logs an error and returns None. + """ + try: + tags_dict = None + if tags: + tags = tags.split(" | ") + tags_dict = {} + for tag in tags: + value = tag.split("=") + tags_dict[value[0]] = value[1] + return tags_dict + except IndexError as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return None + except AttributeError as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return None + + @staticmethod + def format_compliance(compliance: dict) -> tuple[list[dict], list[str]]: + """ + Transforms a dictionary of compliance data into a tuple of associated standards and compliance summaries. + + Parameters: + - compliance (dict): A dictionary containing compliance data where keys are standards and values are lists of compliance details. + + Returns: + - tuple[list[dict], list[str]]: A tuple containing a list of associated standards (each as a dictionary with 'StandardsId') and a list of compliance summaries. + + Notes: + - The method limits the number of associated standards to 20. + - Each compliance summary is a concatenation of the standard key and its associated compliance details. + - If the concatenated summary exceeds 64 characters, it is truncated to 63 characters. + + Example: + format_compliance({"standard1": ["detail1", "detail2"], "standard2": ["detail3"]}) -> ([{"StandardsId": "standard1"}, {"StandardsId": "standard2"}], ["standard1 detail1 detail2", "standard2 detail3"]) + """ + compliance_summary = [] + associated_standards = [] + for key, value in compliance.items(): + if ( + len(associated_standards) < 20 + ): # AssociatedStandards should NOT have more than 20 items + associated_standards.append({"StandardsId": key}) + item = f"{key} {' '.join(value)}" + if len(item) > 64: + item = item[0:63] + compliance_summary.append(item) + return associated_standards, compliance_summary + + +class ProductFields(BaseModel): + """ + Class representing the Product Fields of a finding in the AWS Security Finding Format. + + Attributes: + - ProviderName (str): The name of the provider, default value is "Prowler". + - ProviderVersion (str): The version of the provider, fetched from the prowler_version in config.py. + - ProwlerResourceName (str): The name of the Prowler resource. + """ + + ProviderName: str = "Prowler" + ProviderVersion: str = prowler_version + ProwlerResourceName: str + + +class Severity(BaseModel): + """ + Class representing the severity of a finding in the AWS Security Finding Format. + + Attributes: + - Label (str): A string representing the severity label of the finding. + + This class is used to define the severity level of a finding in the AWS Security Finding Format. + """ + + Label: str + + @validator("Label", pre=True, always=True) + def severity_uppercase(severity): + return severity.upper() + + +class Resource(BaseModel): + """ + Class representing a resource in the AWS Security Finding Format. + + Attributes: + - Type (str): The type of the resource. + - Id (str): The unique identifier of the resource. + - Partition (str): The partition where the resource resides. + - Region (str): The region where the resource is located. + - Tags (Optional[dict]): Optional dictionary of tags associated with the resource. + + This class defines the structure of a resource within the AWS Security Finding Format. It includes attributes to specify the type, unique identifier, partition, region, and optional tags of the resource. + """ + + Type: str + Id: str + Partition: str + Region: str + Tags: Optional[dict] + + +class Compliance(BaseModel): + """ + Class representing the compliance details of a finding in the AWS Security Finding Format. + + Attributes: + - Status (str): The compliance status of the finding. + - RelatedRequirements (list[str]): A list of related compliance requirements for the finding. + - AssociatedStandards (list[dict]): A list of associated standards with the finding, where each item is a dictionary containing the 'StandardsId'. + + This class defines the structure of compliance information within the AWS Security Finding Format. It includes attributes to specify the compliance status, related requirements, and associated standards of a finding. + """ + + Status: str + RelatedRequirements: list[str] + AssociatedStandards: list[dict] + + @validator("Status", pre=True, always=True) + def status(status): + if status not in ["PASSED", "WARNING", "FAILED", "NOT_AVAILABLE"]: + raise ValueError("must contain a space") + return status + + +class Recommendation(BaseModel): + """ + Class representing a recommendation for remediation in the AWS Security Finding Format. + + Attributes: + - Text (str): The text description of the recommendation. + - Url (str): The URL link for additional information related to the recommendation. + + This class defines the structure of a recommendation within the AWS Security Finding Format. It includes attributes to specify the text description and URL link for further details regarding the recommendation. + """ + + Text: str = "" + Url: str = "" + + @validator("Text", pre=True, always=True) + def text_must_not_exceed_512_chars(text): + text_validated = text + if len(text) > 512: + text_validated = text[:509] + "..." + return text_validated + + @validator("Url", pre=True, always=True) + def set_default_url_if_empty(url): + default_url = "https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html" + if url: + default_url = url + return default_url + + +class Remediation(BaseModel): + """ + Class representing a remediation action in the AWS Security Finding Format. + + Attributes: + - Recommendation (Recommendation): An instance of the Recommendation class providing details for remediation. + + This class defines the structure of a remediation action within the AWS Security Finding Format. It includes an attribute to specify the recommendation for remediation, which is an instance of the Recommendation class. + """ + + Recommendation: Recommendation + + +class AWSSecurityFindingFormat(BaseModel): + """ + AWSSecurityFindingFormat generates a finding's output in JSON ASFF format: https://docs.aws.amazon.com/securityhub/latest/userguide/securityhub-findings-format-syntax.html + + Attributes: + - SchemaVersion (str): The version of the ASFF schema being used, default value is "2018-10-08". + - Id (str): The unique identifier of the finding. + - ProductArn (str): The ARN of the product generating the finding. + - RecordState (str): The state of the finding record, default value is "ACTIVE". + - ProductFields (ProductFields): An instance of the ProductFields class representing the product fields of the finding. + - GeneratorId (str): The ID of the generator. + - AwsAccountId (str): The AWS account ID associated with the finding. + - Types (list[str]): A list of types associated with the finding, default value is None. + - FirstObservedAt (str): The timestamp when the finding was first observed. + - UpdatedAt (str): The timestamp when the finding was last updated. + - CreatedAt (str): The timestamp when the finding was created. + - Severity (Severity): An instance of the Severity class representing the severity of the finding. + - Title (str): The title of the finding. + - Description (str): The description of the finding, truncated to 1024 characters if longer. + - Resources (list[Resource]): A list of resources associated with the finding, default value is None. + - Compliance (Compliance): An instance of the Compliance class representing the compliance details of the finding. + - Remediation (Remediation): An instance of the Remediation class providing details for remediation. + + This class defines the structure of a finding in the AWS Security Finding Format, including various attributes such as schema version, identifiers, timestamps, severity, title, description, resources, compliance details, and remediation information. + """ + + SchemaVersion: str = "2018-10-08" + Id: str + ProductArn: str + RecordState: str = "ACTIVE" + ProductFields: ProductFields + GeneratorId: str + AwsAccountId: str + Types: list[str] = None + FirstObservedAt: str + UpdatedAt: str + CreatedAt: str + Severity: Severity + Title: str + Description: str + Resources: list[Resource] = None + Compliance: Compliance + Remediation: Remediation + + @validator("Description", pre=True, always=True) + def description_must_not_exceed_1024_chars(description): + description_validated = description + if len(description) > 1024: + description_validated = description[:1021] + "..." + return description_validated diff --git a/prowler/lib/outputs/csv/models.py b/prowler/lib/outputs/csv/models.py index 110b00243a..d2720a53af 100644 --- a/prowler/lib/outputs/csv/models.py +++ b/prowler/lib/outputs/csv/models.py @@ -30,7 +30,7 @@ class CSV(Output): def batch_write_data_to_file(self) -> None: """Writes the findings to a file using the CSV format using the `Output._file_descriptor`.""" try: - if self._file_descriptor: + if self._file_descriptor and not self._file_descriptor.closed: csv_writer = DictWriter( self._file_descriptor, fieldnames=self._data[0].keys(), @@ -39,6 +39,7 @@ class CSV(Output): csv_writer.writeheader() for finding in self._data: csv_writer.writerow(finding) + self._file_descriptor.close() except Exception as error: logger.error( f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" diff --git a/prowler/lib/outputs/file_descriptors.py b/prowler/lib/outputs/file_descriptors.py index ce86662c66..0db4ed65c2 100644 --- a/prowler/lib/outputs/file_descriptors.py +++ b/prowler/lib/outputs/file_descriptors.py @@ -2,11 +2,7 @@ from csv import DictWriter from io import TextIOWrapper from typing import Any -from prowler.config.config import ( - csv_file_suffix, - html_file_suffix, - json_asff_file_suffix, -) +from prowler.config.config import csv_file_suffix, html_file_suffix from prowler.lib.logger import logger from prowler.lib.outputs.compliance.mitre_attack.models import ( MitreAttackAWS, @@ -48,10 +44,7 @@ def initialize_file_descriptor( filename, "a", ) - - if output_mode == "json-asff": - file_descriptor.write("[") - elif "html" in output_mode: + if "html" in output_mode: add_html_header(file_descriptor, provider) else: # Format is the class model of the CSV format to print the headers @@ -76,6 +69,8 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi # FIXME: Remove this once we always use the new CSV(Output) if output_mode == "csv": continue + elif output_mode == "json-asff": + continue elif output_mode == "html": filename = f"{output_directory}/{output_filename}{html_file_suffix}" file_descriptor = initialize_file_descriptor( @@ -157,68 +152,62 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi file_descriptors.update({output_mode: file_descriptor}) elif provider.type == "aws": - if output_mode == "json-asff": - filename = f"{output_directory}/{output_filename}{json_asff_file_suffix}" + # Compliance frameworks + filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}" + if output_mode == "ens_rd2022_aws": file_descriptor = initialize_file_descriptor( - filename, output_mode + filename, + output_mode, + provider.type, + Check_Output_CSV_ENS_RD2022, ) file_descriptors.update({output_mode: file_descriptor}) - else: # Compliance frameworks - filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}" - if output_mode == "ens_rd2022_aws": - file_descriptor = initialize_file_descriptor( - filename, - output_mode, - provider.type, - Check_Output_CSV_ENS_RD2022, - ) - file_descriptors.update({output_mode: file_descriptor}) - elif "cis_" in output_mode: - file_descriptor = initialize_file_descriptor( - filename, - output_mode, - provider.type, - Check_Output_CSV_AWS_CIS, - ) - file_descriptors.update({output_mode: file_descriptor}) + elif "cis_" in output_mode: + file_descriptor = initialize_file_descriptor( + filename, + output_mode, + provider.type, + Check_Output_CSV_AWS_CIS, + ) + file_descriptors.update({output_mode: file_descriptor}) - elif "aws_well_architected_framework" in output_mode: - file_descriptor = initialize_file_descriptor( - filename, - output_mode, - provider.type, - Check_Output_CSV_AWS_Well_Architected, - ) - file_descriptors.update({output_mode: file_descriptor}) + elif "aws_well_architected_framework" in output_mode: + file_descriptor = initialize_file_descriptor( + filename, + output_mode, + provider.type, + Check_Output_CSV_AWS_Well_Architected, + ) + file_descriptors.update({output_mode: file_descriptor}) - elif output_mode == "iso27001_2013_aws": - file_descriptor = initialize_file_descriptor( - filename, - output_mode, - provider.type, - Check_Output_CSV_AWS_ISO27001_2013, - ) - file_descriptors.update({output_mode: file_descriptor}) + elif output_mode == "iso27001_2013_aws": + file_descriptor = initialize_file_descriptor( + filename, + output_mode, + provider.type, + Check_Output_CSV_AWS_ISO27001_2013, + ) + file_descriptors.update({output_mode: file_descriptor}) - elif output_mode == "mitre_attack_aws": - file_descriptor = initialize_file_descriptor( - filename, - output_mode, - provider.type, - MitreAttackAWS, - ) - file_descriptors.update({output_mode: file_descriptor}) + elif output_mode == "mitre_attack_aws": + file_descriptor = initialize_file_descriptor( + filename, + output_mode, + provider.type, + MitreAttackAWS, + ) + file_descriptors.update({output_mode: file_descriptor}) - else: - # Generic Compliance framework - file_descriptor = initialize_file_descriptor( - filename, - output_mode, - provider.type, - Check_Output_CSV_Generic_Compliance, - ) - file_descriptors.update({output_mode: file_descriptor}) + else: + # Generic Compliance framework + file_descriptor = initialize_file_descriptor( + filename, + output_mode, + provider.type, + Check_Output_CSV_Generic_Compliance, + ) + file_descriptors.update({output_mode: file_descriptor}) except Exception as error: logger.error( diff --git a/prowler/lib/outputs/json_asff/json_asff.py b/prowler/lib/outputs/json_asff/json_asff.py deleted file mode 100644 index 39143f47f1..0000000000 --- a/prowler/lib/outputs/json_asff/json_asff.py +++ /dev/null @@ -1,137 +0,0 @@ -from prowler.config.config import timestamp_utc -from prowler.lib.logger import logger -from prowler.lib.outputs.compliance.compliance import get_check_compliance -from prowler.lib.outputs.json_asff.models import ( - Check_Output_JSON_ASFF, - Compliance, - ProductFields, - Recommendation, - Remediation, - Resource, - Severity, -) -from prowler.lib.utils.utils import hash_sha512 - - -def generate_json_asff_status(status: str, muted: bool = False) -> str: - json_asff_status = "" - if muted: - # Per AWS Security Hub "MUTED" is not a valid status - # https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html - json_asff_status = "WARNING" - else: - if status == "PASS": - json_asff_status = "PASSED" - elif status == "FAIL": - json_asff_status = "FAILED" - else: - json_asff_status = "NOT_AVAILABLE" - - return json_asff_status - - -def generate_json_asff_resource_tags(tags): - try: - resource_tags = {} - if tags and tags != [None]: - for tag in tags: - if "Key" in tag and "Value" in tag: - resource_tags[tag["Key"]] = tag["Value"] - else: - resource_tags.update(tag) - if len(resource_tags) == 0: - return None - else: - return None - return resource_tags - except Exception as error: - logger.error( - f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" - ) - - -def fill_json_asff(provider, finding): - """ - Fill the finding's output in JSON ASFF format. - - Parameters: - - provider: The provider object containing information about the provider (e.g., AWS) and the output options object containing information about the desired output format. - - finding: The finding object containing information about the specific finding. - - Returns: - - finding_output: The filled finding's output in JSON ASFF format. - """ - - try: - # Check if there are no resources in the finding - if finding.resource_arn == "": - if finding.resource_id == "": - finding.resource_id = "NONE_PROVIDED" - finding.resource_arn = finding.resource_id - - timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ") - resource_tags = generate_json_asff_resource_tags(finding.resource_tags) - - # Iterate for each compliance framework - compliance_summary = [] - associated_standards = [] - check_compliance = get_check_compliance( - finding, provider.type, provider.output_options - ) - for key, value in check_compliance.items(): - if ( - len(associated_standards) < 20 - ): # AssociatedStandards should NOT have more than 20 items - associated_standards.append({"StandardsId": key}) - item = f"{key} {' '.join(value)}" - if len(item) > 64: - item = item[0:63] - compliance_summary.append(item) - - # Ensures finding_status matches allowed values in ASFF - finding_status = generate_json_asff_status(finding.status, finding.muted) - - json_asff_output = Check_Output_JSON_ASFF( - # The following line cannot be changed because it is the format we use to generate unique findings for AWS Security Hub - # If changed some findings could be lost because the unique identifier will be different - # TODO: get this from the provider output - Id=f"prowler-{finding.check_metadata.CheckID}-{provider.identity.account}-{finding.region}-{hash_sha512(finding.resource_id)}", - ProductArn=f"arn:{provider.identity.partition}:securityhub:{finding.region}::product/prowler/prowler", - ProductFields=ProductFields( - ProwlerResourceName=finding.resource_arn, - ), - GeneratorId="prowler-" + finding.check_metadata.CheckID, - AwsAccountId=provider.identity.account, - Types=finding.check_metadata.CheckType, - FirstObservedAt=timestamp, - UpdatedAt=timestamp, - CreatedAt=timestamp, - Severity=Severity(Label=finding.check_metadata.Severity.upper()), - Title=finding.check_metadata.CheckTitle, - Description=finding.status_extended, - Resources=[ - Resource( - Id=finding.resource_arn, - Type=finding.check_metadata.ResourceType, - Partition=provider.identity.partition, - Region=finding.region, - Tags=resource_tags, - ) - ], - Compliance=Compliance( - Status=finding_status, - AssociatedStandards=associated_standards, - RelatedRequirements=compliance_summary, - ), - Remediation=Remediation( - Recommendation=Recommendation( - Text=finding.check_metadata.Remediation.Recommendation.Text, - Url=finding.check_metadata.Remediation.Recommendation.Url, - ) - ), - ) - return json_asff_output - except Exception as error: - logger.error( - f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" - ) diff --git a/prowler/lib/outputs/json_asff/models.py b/prowler/lib/outputs/json_asff/models.py deleted file mode 100644 index 336cb0c138..0000000000 --- a/prowler/lib/outputs/json_asff/models.py +++ /dev/null @@ -1,83 +0,0 @@ -from typing import Optional - -from pydantic import BaseModel, validator - -from prowler.config.config import prowler_version - - -class ProductFields(BaseModel): - ProviderName: str = "Prowler" - ProviderVersion: str = prowler_version - ProwlerResourceName: str - - -class Severity(BaseModel): - Label: str - - -class Resource(BaseModel): - Type: str - Id: str - Partition: str - Region: str - Tags: Optional[dict] - - -class Compliance(BaseModel): - Status: str - RelatedRequirements: list[str] - AssociatedStandards: list[dict] - - -class Recommendation(BaseModel): - Text: str = "" - Url: str = "" - - @validator("Text", pre=True, always=True) - def text_must_not_exceed_512_chars(text): - text_validated = text - if len(text) > 512: - text_validated = text[:509] + "..." - return text_validated - - @validator("Url", pre=True, always=True) - def set_default_url_if_empty(url): - default_url = "https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html" - if url: - default_url = url - return default_url - - -class Remediation(BaseModel): - Recommendation: Recommendation - - -class Check_Output_JSON_ASFF(BaseModel): - """ - Check_Output_JSON_ASFF generates a finding's output in JSON ASFF format: https://docs.aws.amazon.com/securityhub/latest/userguide/securityhub-findings-format-syntax.html - """ - - SchemaVersion: str = "2018-10-08" - Id: str - ProductArn: str - RecordState: str = "ACTIVE" - ProductFields: ProductFields - GeneratorId: str - AwsAccountId: str - Types: list[str] = None - FirstObservedAt: str - UpdatedAt: str - CreatedAt: str - Severity: Severity - Title: str - Description: str - Resources: list[Resource] = None - Compliance: Compliance - Remediation: Remediation - - @validator("Description", pre=True, always=True) - def description_must_not_exceed_1024_chars(description): - description_validated = description - if len(description) > 1024: - description_validated = description[:1021] + "..." - return description_validated diff --git a/prowler/lib/outputs/ocsf/ocsf.py b/prowler/lib/outputs/ocsf/ocsf.py index d753fc17ed..5fc5663aea 100644 --- a/prowler/lib/outputs/ocsf/ocsf.py +++ b/prowler/lib/outputs/ocsf/ocsf.py @@ -171,7 +171,9 @@ class OCSF(Output): if self._file_descriptor and not self._file_descriptor.closed: self._file_descriptor.write("[") for finding in self._data: - self._file_descriptor.write(finding.json(exclude_none=True)) + self._file_descriptor.write( + finding.json(exclude_none=True, indent=4) + ) self._file_descriptor.write(",") if self._file_descriptor.tell() > 0: if self._file_descriptor.tell() != 1: diff --git a/prowler/lib/outputs/output.py b/prowler/lib/outputs/output.py index 43159c5061..f9dfde3752 100644 --- a/prowler/lib/outputs/output.py +++ b/prowler/lib/outputs/output.py @@ -8,7 +8,23 @@ from prowler.lib.utils.utils import open_file class Output(ABC): - _data: list[Finding] + """ + This class represents an abstract base class for defining different types of outputs for findings. + + Attributes: + _data (list): A list to store transformed data from findings. + _file_descriptor (TextIOWrapper): A file descriptor to write data to a file. + + Methods: + __init__: Initializes the Output class with findings, optionally creates a file descriptor. + data: Property to access the transformed data. + file_descriptor: Property to access the file descriptor. + transform: Abstract method to transform findings into a specific format. + batch_write_data_to_file: Abstract method to write data to a file in batches. + create_file_descriptor: Method to create a file descriptor for writing data to a file. + """ + + _data: list _file_descriptor: TextIOWrapper def __init__( @@ -39,6 +55,21 @@ class Output(ABC): raise NotImplementedError def create_file_descriptor(self, file_path) -> None: + """ + Creates a file descriptor for writing data to a file. + + Parameters: + file_path (str): The path to the file where the data will be written. + + Returns: + None + + Raises: + Any exception that occurs during the file creation process will be caught and logged using the logger. + + Note: + The file is opened in append mode ("a") to ensure data is written at the end of the file without overwriting existing content. + """ try: mode = "a" self._file_descriptor = open_file( diff --git a/prowler/lib/outputs/outputs.py b/prowler/lib/outputs/outputs.py index 09e9c1bc46..252e6d730b 100644 --- a/prowler/lib/outputs/outputs.py +++ b/prowler/lib/outputs/outputs.py @@ -1,5 +1,3 @@ -import json - from colorama import Fore, Style from prowler.config.config import available_compliance_frameworks, orange_color @@ -11,7 +9,6 @@ from prowler.lib.outputs.compliance.compliance import ( from prowler.lib.outputs.file_descriptors import fill_file_descriptors from prowler.lib.outputs.finding import Finding from prowler.lib.outputs.html.html import fill_html -from prowler.lib.outputs.json_asff.json_asff import fill_json_asff def stdout_report(finding, color, verbose, status, fix): @@ -95,19 +92,6 @@ def report(check_findings, provider): input_compliance_frameworks, ) - # AWS specific outputs - if finding.check_metadata.Provider == "aws": - if "json-asff" in file_descriptors: - # Initialize this field using the class within fill_json_asff not here - json_asff_finding = fill_json_asff(provider, finding) - - json.dump( - json_asff_finding.dict(exclude_none=True), - file_descriptors["json-asff"], - indent=4, - ) - file_descriptors["json-asff"].write(",") - # Common Output Data finding_output = Finding.generate_output(provider, finding) diff --git a/prowler/providers/aws/lib/security_hub/security_hub.py b/prowler/providers/aws/lib/security_hub/security_hub.py index 65e2ac1240..6fc0517ee4 100644 --- a/prowler/providers/aws/lib/security_hub/security_hub.py +++ b/prowler/providers/aws/lib/security_hub/security_hub.py @@ -3,52 +3,60 @@ from botocore.client import ClientError from prowler.config.config import timestamp_utc from prowler.lib.logger import logger -from prowler.lib.outputs.json_asff.json_asff import fill_json_asff +from prowler.lib.outputs.asff.asff import AWSSecurityFindingFormat SECURITY_HUB_INTEGRATION_NAME = "prowler/prowler" SECURITY_HUB_MAX_BATCH = 100 -def prepare_security_hub_findings( - findings: list, provider, output_options, enabled_regions: list +def filter_security_hub_findings_per_region( + findings: list[AWSSecurityFindingFormat], + send_only_fails: bool, + status: list, + enabled_regions: list, ) -> dict: - security_hub_findings_per_region = {} + """filter_security_hub_findings_per_region filters the findings by region and status. It returns a dictionary with the findings per region. + Args: + findings (list[AWSSecurityFindingFormat]): List of findings + send_only_fails (bool): Send only the findings that have failed + status (list): List of statuses to filter the findings + enabled_regions (list): List of enabled regions + + Returns: + dict: Dictionary containing the findings per region + """ + security_hub_findings_per_region = {} # Create a key per audited region for region in enabled_regions: security_hub_findings_per_region[region] = [] - for finding in findings: - # We don't send the MANUAL findings to AWS Security Hub - if finding.status == "MANUAL": - continue - # We don't send findings to not enabled regions - if finding.region not in enabled_regions: + if finding.Resources[0].Region not in enabled_regions: continue if ( - finding.status != "FAIL" or finding.muted - ) and output_options.send_sh_only_fails: + finding.Compliance.Status != "FAILED" + or finding.Compliance.Status == "WARNING" + ) and send_only_fails: continue - if output_options.status: - if finding.status not in output_options.status: + # SecurityHub valid statuses are: PASSED, FAILED, WARNING + if status: + if finding.Compliance.Status == "PASSED" and "PASS" not in status: continue - - if finding.muted: + if finding.Compliance.Status == "FAILED" and "FAIL" not in status: + continue + # Check muted finding + if finding.Compliance.Status == "WARNING": continue # Get the finding region - region = finding.region + # We can do that since the finding always stores just one finding + region = finding.Resources[0].Region - # Format the finding in the JSON ASFF format - finding_json_asff = fill_json_asff(provider, finding) - - # Include that finding within their region in the JSON format - security_hub_findings_per_region[region].append( - finding_json_asff.dict(exclude_none=True) - ) + # Include that finding within their region + security_hub_findings_per_region[region].append(finding) return security_hub_findings_per_region @@ -59,6 +67,18 @@ def verify_security_hub_integration_enabled_per_region( session: session.Session, aws_account_number: str, ) -> bool: + """ + verify_security_hub_integration_enabled_per_region returns True if the Prowler integration is enabled for the given region. Otherwise returns false. + + Args: + partition (str): AWS partition + region (str): AWS region + session (session.Session): AWS session + aws_account_number (str): AWS account number + + Returns: + bool: True if the Prowler integration is enabled for the given region. Otherwise returns false. + """ f"""verify_security_hub_integration_enabled returns True if the {SECURITY_HUB_INTEGRATION_NAME} is enabled for the given region. Otherwise returns false.""" prowler_integration_enabled = False @@ -112,7 +132,14 @@ def batch_send_to_security_hub( session: session.Session, ) -> int: """ - send_to_security_hub sends findings to Security Hub and returns the number of findings that were successfully sent. + batch_send_to_security_hub sends findings to Security Hub and returns the number of findings that were successfully sent. + + Args: + security_hub_findings_per_region (dict): Dictionary containing the findings per region + session (session.Session): AWS session + + Returns: + int: Number of sent findings """ success_count = 0 @@ -120,11 +147,14 @@ def batch_send_to_security_hub( # Iterate findings by region for region, findings in security_hub_findings_per_region.items(): # Send findings to Security Hub - logger.info(f"Sending findings to Security Hub in the region {region}") + logger.info( + f"Sending {len(findings)} findings to Security Hub in the region {region}" + ) security_hub_client = session.client("securityhub", region_name=region) - - success_count = __send_findings_to_security_hub__( + # Convert findings to dict + findings = [finding.dict(exclude_none=True) for finding in findings] + success_count += _send_findings_to_security_hub( findings, region, security_hub_client ) @@ -138,9 +168,16 @@ def batch_send_to_security_hub( # Move previous Security Hub check findings to ARCHIVED (as prowler didn't re-detect them) def resolve_security_hub_previous_findings( security_hub_findings_per_region: dict, provider -) -> list: +) -> int: """ resolve_security_hub_previous_findings archives all the findings that does not appear in the current execution + + Args: + security_hub_findings_per_region (dict): Dictionary containing the findings per region + provider: Provider object + + Returns: + int: Number of archived findings """ logger.info("Checking previous findings in Security Hub to archive them.") success_count = 0 @@ -150,7 +187,7 @@ def resolve_security_hub_previous_findings( # Get current findings IDs current_findings_ids = [] for finding in current_findings: - current_findings_ids.append(finding["Id"]) + current_findings_ids.append(finding.Id) # Get findings of that region security_hub_client = provider.session.current_session.client( "securityhub", region_name=region @@ -165,7 +202,9 @@ def resolve_security_hub_previous_findings( } get_findings_paginator = security_hub_client.get_paginator("get_findings") findings_to_archive = [] - for page in get_findings_paginator.paginate(Filters=findings_filter): + for page in get_findings_paginator.paginate( + Filters=findings_filter, PaginationConfig={"PageSize": 100} + ): # Archive findings that have not appear in this execution for finding in page["Findings"]: if finding["Id"] not in current_findings_ids: @@ -178,7 +217,7 @@ def resolve_security_hub_previous_findings( logger.info(f"Archiving {len(findings_to_archive)} findings.") # Send archive findings to SHub - success_count += __send_findings_to_security_hub__( + success_count += _send_findings_to_security_hub( findings_to_archive, region, security_hub_client ) except Exception as error: @@ -188,17 +227,25 @@ def resolve_security_hub_previous_findings( return success_count -def __send_findings_to_security_hub__( +def _send_findings_to_security_hub( findings: list[dict], region: str, security_hub_client -): - """Private function send_findings_to_security_hub chunks the findings in groups of 100 findings and send them to AWS Security Hub. It returns the number of sent findings.""" +) -> int: + """Private function send_findings_to_security_hub chunks the findings in groups of 100 findings and send them to AWS Security Hub. It returns the number of sent findings. + + Args: + findings (list[dict]): List of findings to send to AWS Security Hub + region (str): AWS region to send the findings + security_hub_client: AWS Security Hub client + + Returns: + int: Number of sent findings + """ success_count = 0 try: list_chunked = [ findings[i : i + SECURITY_HUB_MAX_BATCH] for i in range(0, len(findings), SECURITY_HUB_MAX_BATCH) ] - for findings in list_chunked: batch_import = security_hub_client.batch_import_findings(Findings=findings) if batch_import["FailedCount"] > 0: @@ -207,10 +254,9 @@ def __send_findings_to_security_hub__( f"Failed to send findings to AWS Security Hub -- {failed_import['ErrorCode']} -- {failed_import['ErrorMessage']}" ) success_count += batch_import["SuccessCount"] - + return success_count except Exception as error: logger.error( f"{error.__class__.__name__} -- [{error.__traceback__.tb_lineno}]:{error} in region {region}" ) - finally: return success_count diff --git a/tests/lib/outputs/asff/asff_test.py b/tests/lib/outputs/asff/asff_test.py new file mode 100644 index 0000000000..9a7e1ab0c5 --- /dev/null +++ b/tests/lib/outputs/asff/asff_test.py @@ -0,0 +1,448 @@ +from os import path + +from prowler.config.config import prowler_version, timestamp_utc +from prowler.lib.outputs.asff.asff import ( + ASFF, + AWSSecurityFindingFormat, + Compliance, + ProductFields, + Recommendation, + Remediation, + Resource, + Severity, +) +from prowler.lib.utils.utils import hash_sha512 +from tests.lib.outputs.fixtures.fixtures import generate_finding_output +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_COMMERCIAL_PARTITION, + AWS_REGION_EU_WEST_1, +) + +METADATA_FIXTURE_PATH = ( + f"{path.dirname(path.realpath(__file__))}/../fixtures/metadata.json" +) + + +class TestASFF: + def test_asff(self): + status = "PASS" + finding = generate_finding_output( + status=status, + status_extended="This is a test", + region=AWS_REGION_EU_WEST_1, + resource_details="Test resource details", + resource_name="test-resource", + resource_uid="test-arn", + resource_tags="key1=value1", + ) + + timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ") + + associated_standards, compliance_summary = ASFF.format_compliance( + finding.compliance + ) + + expected = AWSSecurityFindingFormat( + Id=f"prowler-{finding.check_id}-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-{hash_sha512(finding.resource_uid)}", + ProductArn=f"arn:{AWS_COMMERCIAL_PARTITION}:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler", + ProductFields=ProductFields( + ProviderVersion=prowler_version, + ProwlerResourceName=finding.resource_uid, + ), + GeneratorId="prowler-" + finding.check_id, + AwsAccountId=AWS_ACCOUNT_NUMBER, + Types=finding.check_type.split(","), + FirstObservedAt=timestamp, + UpdatedAt=timestamp, + CreatedAt=timestamp, + Severity=Severity(Label=finding.severity), + Title=finding.check_title, + Resources=[ + Resource( + Id=finding.resource_uid, + Type=finding.resource_type, + Partition=AWS_COMMERCIAL_PARTITION, + Region=AWS_REGION_EU_WEST_1, + Tags=ASFF.format_resource_tags(finding.resource_tags), + ) + ], + Compliance=Compliance( + Status=ASFF.generate_status(status), + RelatedRequirements=compliance_summary, + AssociatedStandards=associated_standards, + ), + Remediation=Remediation( + Recommendation=Recommendation( + Text=finding.remediation_recommendation_text, + Url=finding.remediation_recommendation_url, + ) + ), + Description=finding.description, + ) + + asff = ASFF(findings=[finding]) + + assert len(asff.data) == 1 + asff_finding = asff.data[0] + + assert asff_finding == expected + + def test_asff_without_remediation_recommendation_url(self): + status = "PASS" + finding = generate_finding_output( + status=status, + status_extended="This is a test", + region=AWS_REGION_EU_WEST_1, + resource_details="Test resource details", + resource_name="test-resource", + resource_uid="test-arn", + resource_tags="key1=value1", + ) + finding.remediation_recommendation_url = "" + + timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ") + + associated_standards, compliance_summary = ASFF.format_compliance( + finding.compliance + ) + + timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ") + + expected = AWSSecurityFindingFormat( + Id=f"prowler-{finding.check_id}-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-{hash_sha512(finding.resource_uid)}", + ProductArn=f"arn:{AWS_COMMERCIAL_PARTITION}:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler", + ProductFields=ProductFields( + ProviderVersion=prowler_version, + ProwlerResourceName=finding.resource_uid, + ), + GeneratorId="prowler-" + finding.check_id, + AwsAccountId=AWS_ACCOUNT_NUMBER, + Types=finding.check_type.split(","), + FirstObservedAt=timestamp, + UpdatedAt=timestamp, + CreatedAt=timestamp, + Severity=Severity(Label=finding.severity), + Title=finding.check_title, + Resources=[ + Resource( + Id=finding.resource_uid, + Type=finding.resource_type, + Partition=AWS_COMMERCIAL_PARTITION, + Region=AWS_REGION_EU_WEST_1, + Tags=ASFF.format_resource_tags(finding.resource_tags), + ) + ], + Compliance=Compliance( + Status=ASFF.generate_status(status), + RelatedRequirements=compliance_summary, + AssociatedStandards=associated_standards, + ), + Remediation=Remediation( + Recommendation=Recommendation( + Text=finding.remediation_recommendation_text, + Url="https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html", + ) + ), + Description=finding.description, + ) + + asff = ASFF(findings=[finding]) + + assert len(asff.data) == 1 + asff_finding = asff.data[0] + + assert asff_finding == expected + + def test_asff_with_long_description_and_remediation_recommendation_text( + self, + ): + status = "PASS" + finding = generate_finding_output( + status=status, + status_extended="This is a test", + region=AWS_REGION_EU_WEST_1, + resource_details="Test resource details", + resource_name="test-resource", + resource_uid="test-arn", + resource_tags="key1=value1", + ) + finding.remediation_recommendation_url = "" + finding.remediation_recommendation_text = "x" * 513 + + timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ") + + associated_standards, compliance_summary = ASFF.format_compliance( + finding.compliance + ) + + timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ") + + expected = AWSSecurityFindingFormat( + Id=f"prowler-{finding.check_id}-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-{hash_sha512(finding.resource_uid)}", + ProductArn=f"arn:{AWS_COMMERCIAL_PARTITION}:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler", + ProductFields=ProductFields( + ProviderVersion=prowler_version, + ProwlerResourceName=finding.resource_uid, + ), + GeneratorId="prowler-" + finding.check_id, + AwsAccountId=AWS_ACCOUNT_NUMBER, + Types=finding.check_type.split(","), + FirstObservedAt=timestamp, + UpdatedAt=timestamp, + CreatedAt=timestamp, + Severity=Severity(Label=finding.severity), + Title=finding.check_title, + Resources=[ + Resource( + Id=finding.resource_uid, + Type=finding.resource_type, + Partition=AWS_COMMERCIAL_PARTITION, + Region=AWS_REGION_EU_WEST_1, + Tags=ASFF.format_resource_tags(finding.resource_tags), + ) + ], + Compliance=Compliance( + Status=ASFF.generate_status(status), + RelatedRequirements=compliance_summary, + AssociatedStandards=associated_standards, + ), + Remediation=Remediation( + Recommendation=Recommendation( + Text=f"{'x' * 509}...", + Url="https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html", + ) + ), + Description=finding.description, + ) + + asff = ASFF(findings=[finding]) + + assert len(asff.data) == 1 + asff_finding = asff.data[0] + + assert asff_finding == expected + + def test_fill_json_asff_with_long_associated_standards(self): + + status = "PASS" + finding = generate_finding_output( + status=status, + status_extended="This is a test", + region=AWS_REGION_EU_WEST_1, + resource_details="Test resource details", + resource_name="test-resource", + resource_uid="test-arn", + resource_tags="key1=value1", + compliance={ + "CISA": ["your-systems-3", "your-data-2"], + "SOC2": ["cc_2_1", "cc_7_2", "cc_a_1_2"], + "CIS-1.4": ["3.1"], + "CIS-1.5": ["3.1"], + "GDPR": ["article_25", "article_30"], + "AWS-Foundational-Security-Best-Practices": ["cloudtrail"], + "HIPAA": [ + "164_308_a_1_ii_d", + "164_308_a_3_ii_a", + "164_308_a_6_ii", + "164_312_b", + "164_312_e_2_i", + ], + "ISO27001": ["A.12.4"], + "GxP-21-CFR-Part-11": ["11.10-e", "11.10-k", "11.300-d"], + "AWS-Well-Architected-Framework-Security-Pillar": [ + "SEC04-BP02", + "SEC04-BP03", + ], + "GxP-EU-Annex-11": [ + "1-risk-management", + "4.2-validation-documentation-change-control", + ], + "NIST-800-171-Revision-2": [ + "3_1_12", + "3_3_1", + "3_3_2", + "3_3_3", + "3_4_1", + "3_6_1", + "3_6_2", + "3_13_1", + "3_13_2", + "3_14_6", + "3_14_7", + ], + "NIST-800-53-Revision-4": [ + "ac_2_4", + "ac_2", + "au_2", + "au_3", + "au_12", + "cm_2", + ], + "NIST-800-53-Revision-5": [ + "ac_2_4", + "ac_3_1", + "ac_3_10", + "ac_4_26", + "ac_6_9", + "au_2_b", + "au_3_1", + "au_3_a", + "au_3_b", + "au_3_c", + "au_3_d", + "au_3_e", + "au_3_f", + "au_6_3", + "au_6_4", + "au_6_6", + "au_6_9", + "au_8_b", + "au_10", + "au_12_a", + "au_12_c", + "au_12_1", + "au_12_2", + "au_12_3", + "au_12_4", + "au_14_a", + "au_14_b", + "au_14_3", + "ca_7_b", + "cm_5_1_b", + "cm_6_a", + "cm_9_b", + "ia_3_3_b", + "ma_4_1_a", + "pm_14_a_1", + "pm_14_b", + "pm_31", + "sc_7_9_b", + "si_1_1_c", + "si_3_8_b", + "si_4_2", + "si_4_17", + "si_4_20", + "si_7_8", + "si_10_1_c", + ], + "ENS-RD2022": [ + "op.acc.6.r5.aws.iam.1", + "op.exp.5.aws.ct.1", + "op.exp.8.aws.ct.1", + "op.exp.8.aws.ct.6", + "op.exp.9.aws.ct.1", + "op.mon.1.aws.ct.1", + ], + "NIST-CSF-1.1": [ + "ae_1", + "ae_3", + "ae_4", + "cm_1", + "cm_3", + "cm_6", + "cm_7", + "am_3", + "ac_6", + "ds_5", + "ma_2", + "pt_1", + ], + "RBI-Cyber-Security-Framework": ["annex_i_7_4"], + "FFIEC": [ + "d2-ma-ma-b-1", + "d2-ma-ma-b-2", + "d3-dc-an-b-3", + "d3-dc-an-b-4", + "d3-dc-an-b-5", + "d3-dc-ev-b-1", + "d3-dc-ev-b-3", + "d3-pc-im-b-3", + "d3-pc-im-b-7", + "d5-dr-de-b-3", + ], + "PCI-3.2.1": ["cloudtrail"], + "FedRamp-Moderate-Revision-4": [ + "ac-2-4", + "ac-2-g", + "au-2-a-d", + "au-3", + "au-6-1-3", + "au-12-a-c", + "ca-7-a-b", + "si-4-16", + "si-4-2", + "si-4-4", + "si-4-5", + ], + "FedRAMP-Low-Revision-4": ["ac-2", "au-2", "ca-7"], + }, + ) + + timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ") + + associated_standards, compliance_summary = ASFF.format_compliance( + finding.compliance + ) + + expected = AWSSecurityFindingFormat( + Id=f"prowler-{finding.check_id}-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-{hash_sha512(finding.resource_uid)}", + ProductArn=f"arn:{AWS_COMMERCIAL_PARTITION}:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler", + ProductFields=ProductFields( + ProviderVersion=prowler_version, + ProwlerResourceName=finding.resource_uid, + ), + GeneratorId="prowler-" + finding.check_id, + AwsAccountId=AWS_ACCOUNT_NUMBER, + Types=finding.check_type.split(","), + FirstObservedAt=timestamp, + UpdatedAt=timestamp, + CreatedAt=timestamp, + Severity=Severity(Label=finding.severity), + Title=finding.check_title, + Resources=[ + Resource( + Id=finding.resource_uid, + Type=finding.resource_type, + Partition=AWS_COMMERCIAL_PARTITION, + Region=AWS_REGION_EU_WEST_1, + Tags=ASFF.format_resource_tags(finding.resource_tags), + ) + ], + Compliance=Compliance( + Status=ASFF.generate_status(status), + RelatedRequirements=compliance_summary, + AssociatedStandards=associated_standards, + ), + Remediation=Remediation( + Recommendation=Recommendation( + Text=finding.remediation_recommendation_text, + Url=finding.remediation_recommendation_url, + ) + ), + Description=finding.description, + ) + + asff = ASFF(findings=[finding]) + + assert len(asff.data) == 1 + asff_finding = asff.data[0] + + assert asff_finding == expected + + def test_asff_generate_status(self): + assert ASFF.generate_status("PASS") == "PASSED" + assert ASFF.generate_status("FAIL") == "FAILED" + assert ASFF.generate_status("FAIL", True) == "WARNING" + assert ASFF.generate_status("SOMETHING ELSE") == "NOT_AVAILABLE" + + def test_asff_format_resource_tags(self): + assert ASFF.format_resource_tags(None) is None + assert ASFF.format_resource_tags("") is None + assert ASFF.format_resource_tags([]) is None + assert ASFF.format_resource_tags([{}]) is None + assert ASFF.format_resource_tags("key1=value1") == {"key1": "value1"} + assert ASFF.format_resource_tags("key1=value1 | key2=value2") == { + "key1": "value1", + "key2": "value2", + } diff --git a/tests/lib/outputs/csv/csv_test.py b/tests/lib/outputs/csv/csv_test.py index 035dff9dae..9c8fe53e50 100644 --- a/tests/lib/outputs/csv/csv_test.py +++ b/tests/lib/outputs/csv/csv_test.py @@ -5,6 +5,7 @@ from typing import List from unittest.mock import MagicMock import pytest +from mock import patch from prowler.lib.outputs.csv.csv import write_csv from prowler.lib.outputs.csv.models import CSV @@ -60,7 +61,6 @@ def generate_finding(): class TestCSV: - def test_output_transform(self, generate_finding): findings = [generate_finding] @@ -131,7 +131,8 @@ class TestCSV: output = CSV(findings) output._file_descriptor = mock_file - output.batch_write_data_to_file() + with patch.object(mock_file, "close", return_value=None): + output.batch_write_data_to_file() mock_file.seek(0) content = mock_file.read() diff --git a/tests/lib/outputs/fixtures/fixtures.py b/tests/lib/outputs/fixtures/fixtures.py index 35414b3019..62ca5b6acf 100644 --- a/tests/lib/outputs/fixtures/fixtures.py +++ b/tests/lib/outputs/fixtures/fixtures.py @@ -2,20 +2,23 @@ from datetime import datetime from prowler.config.config import prowler_version from prowler.lib.outputs.finding import Finding -from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER +from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1 -# TODO: customize it per provider def generate_finding_output( - status, - severity, - muted, - region, + status: str = "PASS", + status_extended: str = "", + severity: str = "high", + muted: bool = False, + region: str = AWS_REGION_EU_WEST_1, + resource_details: str = "", + resource_uid: str = "", + resource_name: str = "", + resource_tags: str = "", + compliance: dict = {"test-compliance": "test-compliance"}, timestamp: datetime = datetime.now(), provider: str = "aws", ) -> Finding: - # TODO: Include metadata from a valid file - return Finding( auth_method="profile: default", timestamp=timestamp, @@ -31,17 +34,17 @@ def generate_finding_output( check_title="test-check-id", check_type="test-type", status=status, - status_extended="status extended", + status_extended=status_extended, muted=muted, service_name="test-service", subservice_name="", severity=severity, resource_type="test-resource", - resource_uid="resource-id", - resource_name="resource_name", - resource_details="resource_details", - resource_tags="", - partition="aws", + resource_uid=resource_uid, + resource_name=resource_name, + resource_details=resource_details, + resource_tags=resource_tags, + partition=provider, region=region, description="check description", risk="test-risk", @@ -52,7 +55,7 @@ def generate_finding_output( remediation_code_terraform="", remediation_code_cli="", remediation_code_other="", - compliance={"test-compliance": "test-compliance"}, + compliance=compliance, categories="test-category", depends_on="test-dependency", related_to="test-related-to", diff --git a/tests/lib/outputs/json_asff/json_asff_test.py b/tests/lib/outputs/json_asff/json_asff_test.py deleted file mode 100644 index 99eeabdf6f..0000000000 --- a/tests/lib/outputs/json_asff/json_asff_test.py +++ /dev/null @@ -1,472 +0,0 @@ -from os import path - -import mock - -from prowler.config.config import prowler_version, timestamp_utc -from prowler.lib.check.models import Check_Report, load_check_metadata -from prowler.lib.outputs.json_asff.json_asff import ( - fill_json_asff, - generate_json_asff_resource_tags, - generate_json_asff_status, -) -from prowler.lib.outputs.json_asff.models import ( - Check_Output_JSON_ASFF, - Compliance, - ProductFields, - Recommendation, - Remediation, - Resource, - Severity, -) -from prowler.lib.utils.utils import hash_sha512 -from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, set_mocked_aws_provider - -METADATA_FIXTURE_PATH = ( - f"{path.dirname(path.realpath(__file__))}/../fixtures/metadata.json" -) - - -class TestOutputJSONASFF: - def test_fill_json_asff(self): - aws_provider = set_mocked_aws_provider() - finding = Check_Report(load_check_metadata(METADATA_FIXTURE_PATH).json()) - finding.resource_details = "Test resource details" - finding.resource_id = "test-resource" - finding.resource_arn = "test-arn" - finding.region = "eu-west-1" - finding.status = "PASS" - finding.status_extended = "This is a test" - - timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ") - - expected = Check_Output_JSON_ASFF( - Id=f"prowler-{finding.check_metadata.CheckID}-123456789012-eu-west-1-{hash_sha512('test-resource')}", - ProductArn="arn:aws:securityhub:eu-west-1::product/prowler/prowler", - ProductFields=ProductFields( - ProviderVersion=prowler_version, ProwlerResourceName="test-arn" - ), - GeneratorId="prowler-" + finding.check_metadata.CheckID, - AwsAccountId=AWS_ACCOUNT_NUMBER, - Types=finding.check_metadata.CheckType, - FirstObservedAt=timestamp, - UpdatedAt=timestamp, - CreatedAt=timestamp, - Severity=Severity(Label=finding.check_metadata.Severity.upper()), - Title=finding.check_metadata.CheckTitle, - Description=finding.status_extended, - Resources=[ - Resource( - Id="test-arn", - Type=finding.check_metadata.ResourceType, - Partition="aws", - Region="eu-west-1", - ) - ], - Compliance=Compliance( - Status="PASS" + "ED", - RelatedRequirements=[], - AssociatedStandards=[], - ), - Remediation={ - "Recommendation": finding.check_metadata.Remediation.Recommendation - }, - ) - - assert fill_json_asff(aws_provider, finding) == expected - - def test_fill_json_asff_without_remediation_recommendation_url(self): - aws_provider = set_mocked_aws_provider() - finding = Check_Report(load_check_metadata(METADATA_FIXTURE_PATH).json()) - - # Empty the Remediation.Recomendation.URL - finding.check_metadata.Remediation.Recommendation.Url = "" - - finding.resource_details = "Test resource details" - finding.resource_id = "test-resource" - finding.resource_arn = "test-arn" - finding.region = "eu-west-1" - finding.status = "PASS" - finding.status_extended = "This is a test" - - timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ") - - expected = Check_Output_JSON_ASFF( - Id=f"prowler-{finding.check_metadata.CheckID}-123456789012-eu-west-1-{hash_sha512('test-resource')}", - ProductArn="arn:aws:securityhub:eu-west-1::product/prowler/prowler", - ProductFields=ProductFields( - ProviderVersion=prowler_version, ProwlerResourceName="test-arn" - ), - GeneratorId="prowler-" + finding.check_metadata.CheckID, - AwsAccountId=AWS_ACCOUNT_NUMBER, - Types=finding.check_metadata.CheckType, - FirstObservedAt=timestamp, - UpdatedAt=timestamp, - CreatedAt=timestamp, - Severity=Severity(Label=finding.check_metadata.Severity.upper()), - Title=finding.check_metadata.CheckTitle, - Description=finding.status_extended, - Resources=[ - Resource( - Id="test-arn", - Type=finding.check_metadata.ResourceType, - Partition="aws", - Region="eu-west-1", - ) - ], - Compliance=Compliance( - Status="PASS" + "ED", - RelatedRequirements=[], - AssociatedStandards=[], - ), - Remediation=Remediation( - Recommendation=Recommendation( - Text=finding.check_metadata.Remediation.Recommendation.Text, - Url="https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html", - ) - ), - ) - - assert fill_json_asff(aws_provider, finding) == expected - - def test_fill_json_asff_with_long_description_and_remediation_recommendation_text( - self, - ): - aws_provider = set_mocked_aws_provider() - finding = Check_Report(load_check_metadata(METADATA_FIXTURE_PATH).json()) - - # Empty the Remediation.Recomendation.URL - finding.check_metadata.Remediation.Recommendation.Url = "" - finding.check_metadata.Remediation.Recommendation.Text = "x" * 513 - - finding.resource_details = "Test resource details" - finding.resource_id = "test-resource" - finding.resource_arn = "test-arn" - finding.region = "eu-west-1" - finding.status = "PASS" - finding.status_extended = "x" * 2000 # it has to be limited to 1000+... - - timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ") - - expected = Check_Output_JSON_ASFF( - Id=f"prowler-{finding.check_metadata.CheckID}-123456789012-eu-west-1-{hash_sha512('test-resource')}", - ProductArn="arn:aws:securityhub:eu-west-1::product/prowler/prowler", - ProductFields=ProductFields( - ProviderVersion=prowler_version, ProwlerResourceName="test-arn" - ), - GeneratorId="prowler-" + finding.check_metadata.CheckID, - AwsAccountId=AWS_ACCOUNT_NUMBER, - Types=finding.check_metadata.CheckType, - FirstObservedAt=timestamp, - UpdatedAt=timestamp, - CreatedAt=timestamp, - Severity=Severity(Label=finding.check_metadata.Severity.upper()), - Title=finding.check_metadata.CheckTitle, - Description=f"{finding.status_extended[:1021]}...", - Resources=[ - Resource( - Id="test-arn", - Type=finding.check_metadata.ResourceType, - Partition="aws", - Region="eu-west-1", - ) - ], - Compliance=Compliance( - Status="PASS" + "ED", - RelatedRequirements=[], - AssociatedStandards=[], - ), - Remediation=Remediation( - Recommendation=Recommendation( - Text=f"{'x' * 509}...", - Url="https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html", - ) - ), - ) - output_json_asff = fill_json_asff(aws_provider, finding) - - assert isinstance(output_json_asff, Check_Output_JSON_ASFF) - assert output_json_asff.Id == expected.Id - assert output_json_asff.ProductArn == expected.ProductArn - assert output_json_asff.ProductFields == expected.ProductFields - assert output_json_asff.GeneratorId == expected.GeneratorId - assert output_json_asff.AwsAccountId == expected.AwsAccountId - assert output_json_asff.Types == expected.Types - assert output_json_asff.FirstObservedAt == expected.FirstObservedAt - assert output_json_asff.UpdatedAt == expected.UpdatedAt - assert output_json_asff.CreatedAt == expected.CreatedAt - assert output_json_asff.Severity == expected.Severity - assert output_json_asff.Title == expected.Title - assert output_json_asff.Description == expected.Description - assert output_json_asff.Resources == expected.Resources - assert output_json_asff.Compliance == expected.Compliance - assert isinstance(output_json_asff.Remediation, Remediation) - assert isinstance(output_json_asff.Remediation.Recommendation, Recommendation) - assert ( - output_json_asff.Remediation.Recommendation.Text - == expected.Remediation.Recommendation.Text - ) - assert ( - output_json_asff.Remediation.Recommendation.Url - == expected.Remediation.Recommendation.Url - ) - - def test_fill_json_asff_with_long_associated_standards(self): - aws_provider = set_mocked_aws_provider() - with mock.patch( - "prowler.lib.outputs.json_asff.json_asff.get_check_compliance", - return_value={ - "CISA": ["your-systems-3", "your-data-2"], - "SOC2": ["cc_2_1", "cc_7_2", "cc_a_1_2"], - "CIS-1.4": ["3.1"], - "CIS-1.5": ["3.1"], - "GDPR": ["article_25", "article_30"], - "AWS-Foundational-Security-Best-Practices": ["cloudtrail"], - "HIPAA": [ - "164_308_a_1_ii_d", - "164_308_a_3_ii_a", - "164_308_a_6_ii", - "164_312_b", - "164_312_e_2_i", - ], - "ISO27001": ["A.12.4"], - "GxP-21-CFR-Part-11": ["11.10-e", "11.10-k", "11.300-d"], - "AWS-Well-Architected-Framework-Security-Pillar": [ - "SEC04-BP02", - "SEC04-BP03", - ], - "GxP-EU-Annex-11": [ - "1-risk-management", - "4.2-validation-documentation-change-control", - ], - "NIST-800-171-Revision-2": [ - "3_1_12", - "3_3_1", - "3_3_2", - "3_3_3", - "3_4_1", - "3_6_1", - "3_6_2", - "3_13_1", - "3_13_2", - "3_14_6", - "3_14_7", - ], - "NIST-800-53-Revision-4": [ - "ac_2_4", - "ac_2", - "au_2", - "au_3", - "au_12", - "cm_2", - ], - "NIST-800-53-Revision-5": [ - "ac_2_4", - "ac_3_1", - "ac_3_10", - "ac_4_26", - "ac_6_9", - "au_2_b", - "au_3_1", - "au_3_a", - "au_3_b", - "au_3_c", - "au_3_d", - "au_3_e", - "au_3_f", - "au_6_3", - "au_6_4", - "au_6_6", - "au_6_9", - "au_8_b", - "au_10", - "au_12_a", - "au_12_c", - "au_12_1", - "au_12_2", - "au_12_3", - "au_12_4", - "au_14_a", - "au_14_b", - "au_14_3", - "ca_7_b", - "cm_5_1_b", - "cm_6_a", - "cm_9_b", - "ia_3_3_b", - "ma_4_1_a", - "pm_14_a_1", - "pm_14_b", - "pm_31", - "sc_7_9_b", - "si_1_1_c", - "si_3_8_b", - "si_4_2", - "si_4_17", - "si_4_20", - "si_7_8", - "si_10_1_c", - ], - "ENS-RD2022": [ - "op.acc.6.r5.aws.iam.1", - "op.exp.5.aws.ct.1", - "op.exp.8.aws.ct.1", - "op.exp.8.aws.ct.6", - "op.exp.9.aws.ct.1", - "op.mon.1.aws.ct.1", - ], - "NIST-CSF-1.1": [ - "ae_1", - "ae_3", - "ae_4", - "cm_1", - "cm_3", - "cm_6", - "cm_7", - "am_3", - "ac_6", - "ds_5", - "ma_2", - "pt_1", - ], - "RBI-Cyber-Security-Framework": ["annex_i_7_4"], - "FFIEC": [ - "d2-ma-ma-b-1", - "d2-ma-ma-b-2", - "d3-dc-an-b-3", - "d3-dc-an-b-4", - "d3-dc-an-b-5", - "d3-dc-ev-b-1", - "d3-dc-ev-b-3", - "d3-pc-im-b-3", - "d3-pc-im-b-7", - "d5-dr-de-b-3", - ], - "PCI-3.2.1": ["cloudtrail"], - "FedRamp-Moderate-Revision-4": [ - "ac-2-4", - "ac-2-g", - "au-2-a-d", - "au-3", - "au-6-1-3", - "au-12-a-c", - "ca-7-a-b", - "si-4-16", - "si-4-2", - "si-4-4", - "si-4-5", - ], - "FedRAMP-Low-Revision-4": ["ac-2", "au-2", "ca-7"], - }, - ): - finding = Check_Report(load_check_metadata(METADATA_FIXTURE_PATH).json()) - - # Empty the Remediation.Recomendation.URL - finding.check_metadata.Remediation.Recommendation.Url = "" - - finding.resource_details = "Test resource details" - finding.resource_id = "test-resource" - finding.resource_arn = "test-arn" - finding.region = "eu-west-1" - finding.status = "PASS" - finding.status_extended = "This is a test" - - timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ") - - expected = Check_Output_JSON_ASFF( - Id=f"prowler-{finding.check_metadata.CheckID}-123456789012-eu-west-1-{hash_sha512('test-resource')}", - ProductArn="arn:aws:securityhub:eu-west-1::product/prowler/prowler", - ProductFields=ProductFields( - ProviderVersion=prowler_version, ProwlerResourceName="test-arn" - ), - GeneratorId="prowler-" + finding.check_metadata.CheckID, - AwsAccountId=AWS_ACCOUNT_NUMBER, - Types=finding.check_metadata.CheckType, - FirstObservedAt=timestamp, - UpdatedAt=timestamp, - CreatedAt=timestamp, - Severity=Severity(Label=finding.check_metadata.Severity.upper()), - Title=finding.check_metadata.CheckTitle, - Description=finding.status_extended, - Resources=[ - Resource( - Id="test-arn", - Type=finding.check_metadata.ResourceType, - Partition="aws", - Region="eu-west-1", - ) - ], - Compliance=Compliance( - Status="PASS" + "ED", - RelatedRequirements=[ - "CISA your-systems-3 your-data-2", - "SOC2 cc_2_1 cc_7_2 cc_a_1_2", - "CIS-1.4 3.1", - "CIS-1.5 3.1", - "GDPR article_25 article_30", - "AWS-Foundational-Security-Best-Practices cloudtrail", - "HIPAA 164_308_a_1_ii_d 164_308_a_3_ii_a 164_308_a_6_ii 164_312_", - "ISO27001 A.12.4", - "GxP-21-CFR-Part-11 11.10-e 11.10-k 11.300-d", - "AWS-Well-Architected-Framework-Security-Pillar SEC04-BP02 SEC04", - "GxP-EU-Annex-11 1-risk-management 4.2-validation-documentation-", - "NIST-800-171-Revision-2 3_1_12 3_3_1 3_3_2 3_3_3 3_4_1 3_6_1 3_", - "NIST-800-53-Revision-4 ac_2_4 ac_2 au_2 au_3 au_12 cm_2", - "NIST-800-53-Revision-5 ac_2_4 ac_3_1 ac_3_10 ac_4_26 ac_6_9 au_", - "ENS-RD2022 op.acc.6.r5.aws.iam.1 op.exp.5.aws.ct.1 op.exp.8.aws", - "NIST-CSF-1.1 ae_1 ae_3 ae_4 cm_1 cm_3 cm_6 cm_7 am_3 ac_6 ds_5 ", - "RBI-Cyber-Security-Framework annex_i_7_4", - "FFIEC d2-ma-ma-b-1 d2-ma-ma-b-2 d3-dc-an-b-3 d3-dc-an-b-4 d3-dc", - "PCI-3.2.1 cloudtrail", - "FedRamp-Moderate-Revision-4 ac-2-4 ac-2-g au-2-a-d au-3 au-6-1-", - ], - AssociatedStandards=[ - {"StandardsId": "CISA"}, - {"StandardsId": "SOC2"}, - {"StandardsId": "CIS-1.4"}, - {"StandardsId": "CIS-1.5"}, - {"StandardsId": "GDPR"}, - {"StandardsId": "AWS-Foundational-Security-Best-Practices"}, - {"StandardsId": "HIPAA"}, - {"StandardsId": "ISO27001"}, - {"StandardsId": "GxP-21-CFR-Part-11"}, - { - "StandardsId": "AWS-Well-Architected-Framework-Security-Pillar" - }, - {"StandardsId": "GxP-EU-Annex-11"}, - {"StandardsId": "NIST-800-171-Revision-2"}, - {"StandardsId": "NIST-800-53-Revision-4"}, - {"StandardsId": "NIST-800-53-Revision-5"}, - {"StandardsId": "ENS-RD2022"}, - {"StandardsId": "NIST-CSF-1.1"}, - {"StandardsId": "RBI-Cyber-Security-Framework"}, - {"StandardsId": "FFIEC"}, - {"StandardsId": "PCI-3.2.1"}, - {"StandardsId": "FedRamp-Moderate-Revision-4"}, - ], - ), - Remediation=Remediation( - Recommendation=Recommendation( - Text=finding.check_metadata.Remediation.Recommendation.Text, - Url="https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html", - ) - ), - ) - - assert fill_json_asff(aws_provider, finding) == expected - - def test_generate_json_asff_status(self): - assert generate_json_asff_status("PASS") == "PASSED" - assert generate_json_asff_status("FAIL") == "FAILED" - assert generate_json_asff_status("FAIL", True) == "WARNING" - assert generate_json_asff_status("SOMETHING ELSE") == "NOT_AVAILABLE" - - def test_generate_json_asff_resource_tags(self): - assert generate_json_asff_resource_tags(None) is None - assert generate_json_asff_resource_tags([]) is None - assert generate_json_asff_resource_tags([{}]) is None - assert generate_json_asff_resource_tags([{"key1": "value1"}]) == { - "key1": "value1" - } - assert generate_json_asff_resource_tags( - [{"Key": "key1", "Value": "value1"}] - ) == {"key1": "value1"} diff --git a/tests/lib/outputs/ocsf/ocsf_test.py b/tests/lib/outputs/ocsf/ocsf_test.py index 3263b449e2..9d491628e0 100644 --- a/tests/lib/outputs/ocsf/ocsf_test.py +++ b/tests/lib/outputs/ocsf/ocsf_test.py @@ -100,7 +100,11 @@ expected_json_output = json.dumps( class TestOCSF: def test_transform(self): - findings = [generate_finding_output("FAIL", "low", False, AWS_REGION_EU_WEST_1)] + findings = [ + generate_finding_output( + status="FAIL", severity="low", muted=False, region=AWS_REGION_EU_WEST_1 + ) + ] ocsf = OCSF(findings) @@ -110,7 +114,17 @@ class TestOCSF: def test_batch_write_data_to_file(self): mock_file = StringIO() findings = [ - generate_finding_output("FAIL", "low", False, AWS_REGION_EU_WEST_1, now) + generate_finding_output( + status="FAIL", + severity="low", + muted=False, + region=AWS_REGION_EU_WEST_1, + timestamp=now, + resource_details="resource_details", + resource_name="resource_name", + resource_uid="resource-id", + status_extended="status extended", + ) ] output = OCSF(findings) @@ -126,7 +140,7 @@ class TestOCSF: def test_finding_output_cloud_pass_low_muted(self): finding_output = generate_finding_output( - "PASS", "low", True, AWS_REGION_EU_WEST_1 + status="PASS", severity="low", muted=True, region=AWS_REGION_EU_WEST_1 ) finding_ocsf = OCSF([finding_output]) @@ -232,7 +246,11 @@ class TestOCSF: def test_finding_output_kubernetes(self): finding_output = generate_finding_output( - "PASS", "low", True, AWS_REGION_EU_WEST_1, provider="kubernetes" + status="PASS", + severity="low", + muted=True, + region=AWS_REGION_EU_WEST_1, + provider="kubernetes", ) finding_ocsf = OCSF([finding_output]) @@ -243,7 +261,7 @@ class TestOCSF: def test_finding_output_cloud_fail_low_not_muted(self): finding_output = generate_finding_output( - "FAIL", "low", False, AWS_REGION_EU_WEST_1 + status="FAIL", severity="low", muted=False, region=AWS_REGION_EU_WEST_1 ) finding_ocsf = OCSF([finding_output]) @@ -257,7 +275,7 @@ class TestOCSF: def test_finding_output_cloud_pass_low_not_muted(self): finding_output = generate_finding_output( - "PASS", "low", False, AWS_REGION_EU_WEST_1 + status="PASS", severity="low", muted=False, region=AWS_REGION_EU_WEST_1 ) finding_ocsf = OCSF([finding_output]) diff --git a/tests/lib/outputs/outputs_test.py b/tests/lib/outputs/outputs_test.py index b72f4e13b8..58a9ec510b 100644 --- a/tests/lib/outputs/outputs_test.py +++ b/tests/lib/outputs/outputs_test.py @@ -5,11 +5,7 @@ from unittest import mock import pytest from colorama import Fore -from prowler.config.config import ( - html_file_suffix, - json_asff_file_suffix, - output_file_timestamp, -) +from prowler.config.config import html_file_suffix, output_file_timestamp from prowler.lib.check.compliance_models import ( CIS_Requirement_Attribute, Compliance_Base_Model, @@ -39,34 +35,16 @@ class TestOutputs: output_directory = f"{os.path.dirname(os.path.realpath(__file__))}" aws_provider = set_mocked_aws_provider() test_output_modes = [ - ["json-asff"], ["html"], - ["json-asff", "html"], ] output_filename = f"prowler-output-{audited_account}-{output_file_timestamp}" expected = [ - { - "json-asff": open_file( - f"{output_directory}/{output_filename}{json_asff_file_suffix}", - "a", - ) - }, { "html": open_file( f"{output_directory}/{output_filename}{html_file_suffix}", "a", ) }, - { - "json-asff": open_file( - f"{output_directory}/{output_filename}{json_asff_file_suffix}", - "a", - ), - "html": open_file( - f"{output_directory}/{output_filename}{html_file_suffix}", - "a", - ), - }, ] for index, output_mode_list in enumerate(test_output_modes): diff --git a/tests/providers/aws/lib/security_hub/security_hub_test.py b/tests/providers/aws/lib/security_hub/security_hub_test.py index ad14dba694..38ea57accc 100644 --- a/tests/providers/aws/lib/security_hub/security_hub_test.py +++ b/tests/providers/aws/lib/security_hub/security_hub_test.py @@ -1,69 +1,24 @@ from logging import ERROR, WARNING -from os import path import botocore from boto3 import session from botocore.client import ClientError -from mock import MagicMock, patch +from mock import patch -from prowler.config.config import prowler_version, timestamp_utc -from prowler.lib.check.models import Check_Report, load_check_metadata +from prowler.lib.outputs.asff.asff import ASFF from prowler.providers.aws.lib.security_hub.security_hub import ( batch_send_to_security_hub, - prepare_security_hub_findings, + filter_security_hub_findings_per_region, verify_security_hub_integration_enabled_per_region, ) +from tests.lib.outputs.fixtures.fixtures import generate_finding_output from tests.providers.aws.utils import ( AWS_ACCOUNT_NUMBER, AWS_COMMERCIAL_PARTITION, AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2, - set_mocked_aws_provider, ) - -def get_security_hub_finding(status: str): - return { - "SchemaVersion": "2018-10-08", - "Id": f"prowler-iam_user_accesskey_unused-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-ee26b0dd4", - "ProductArn": f"arn:aws:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler", - "RecordState": "ACTIVE", - "ProductFields": { - "ProviderName": "Prowler", - "ProviderVersion": prowler_version, - "ProwlerResourceName": "test", - }, - "GeneratorId": "prowler-iam_user_accesskey_unused", - "AwsAccountId": f"{AWS_ACCOUNT_NUMBER}", - "Types": ["Software and Configuration Checks"], - "FirstObservedAt": timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ"), - "UpdatedAt": timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ"), - "CreatedAt": timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ"), - "Severity": {"Label": "LOW"}, - "Title": "Ensure Access Keys unused are disabled", - "Description": "test", - "Resources": [ - { - "Type": "AwsIamAccessAnalyzer", - "Id": "test", - "Partition": "aws", - "Region": f"{AWS_REGION_EU_WEST_1}", - } - ], - "Compliance": { - "Status": status, - "RelatedRequirements": [], - "AssociatedStandards": [], - }, - "Remediation": { - "Recommendation": { - "Text": "Run sudo yum update and cross your fingers and toes.", - "Url": "https://myfp.com/recommendations/dangerous_things_and_how_to_fix_them.html", - } - }, - } - - # Mocking Security Hub Get Findings make_api_call = botocore.client.BaseClient._make_api_call @@ -92,41 +47,18 @@ def mock_make_api_call(self, operation_name, kwarg): return make_api_call(self, operation_name, kwarg) -class Test_SecurityHub: - def generate_finding(self, status, region, muted=False): - finding = Check_Report( - load_check_metadata( - f"{path.dirname(path.realpath(__file__))}/fixtures/metadata.json" - ).json() - ) - finding.status = status - finding.status_extended = "test" - finding.resource_id = "test" - finding.resource_arn = "test" - finding.region = region - finding.muted = muted +def set_mocked_session(region=None): + # Create mock session + return session.Session( + region_name=region, + ) - return finding - def set_mocked_output_options( - self, status: list[str] = [], send_sh_only_fails: bool = False - ): - output_options = MagicMock - output_options.bulk_checks_metadata = {} - output_options.status = status - output_options.send_sh_only_fails = send_sh_only_fails - - return output_options - - def set_mocked_session(self, region): - # Create mock session - return session.Session( - region_name=region, - ) +class TestSecurityHub: @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) def test_verify_security_hub_integration_enabled_per_region(self): - session = self.set_mocked_session(AWS_REGION_EU_WEST_1) + session = set_mocked_session(AWS_REGION_EU_WEST_1) assert verify_security_hub_integration_enabled_per_region( AWS_COMMERCIAL_PARTITION, AWS_REGION_EU_WEST_1, session, AWS_ACCOUNT_NUMBER ) @@ -135,7 +67,7 @@ class Test_SecurityHub: self, caplog ): caplog.set_level(WARNING) - session = self.set_mocked_session(AWS_REGION_EU_WEST_1) + session = set_mocked_session(AWS_REGION_EU_WEST_1) with patch( "prowler.providers.aws.lib.security_hub.security_hub.session.Session.client", @@ -161,7 +93,7 @@ class Test_SecurityHub: ( "root", WARNING, - f"ClientError -- [70]: An error occurred ({error_code}) when calling the {operation_name} operation: {error_message}", + f"ClientError -- [90]: An error occurred ({error_code}) when calling the {operation_name} operation: {error_message}", ) ] @@ -169,7 +101,7 @@ class Test_SecurityHub: self, caplog ): caplog.set_level(WARNING) - session = self.set_mocked_session(AWS_REGION_EU_WEST_1) + session = set_mocked_session(AWS_REGION_EU_WEST_1) with patch( "prowler.providers.aws.lib.security_hub.security_hub.session.Session.client", @@ -195,7 +127,7 @@ class Test_SecurityHub: self, caplog ): caplog.set_level(WARNING) - session = self.set_mocked_session(AWS_REGION_EU_WEST_1) + session = set_mocked_session(AWS_REGION_EU_WEST_1) with patch( "prowler.providers.aws.lib.security_hub.security_hub.session.Session.client", @@ -221,7 +153,7 @@ class Test_SecurityHub: ( "root", ERROR, - f"ClientError -- [70]: An error occurred ({error_code}) when calling the {operation_name} operation: {error_message}", + f"ClientError -- [90]: An error occurred ({error_code}) when calling the {operation_name} operation: {error_message}", ) ] @@ -229,7 +161,7 @@ class Test_SecurityHub: self, caplog ): caplog.set_level(WARNING) - session = self.set_mocked_session(AWS_REGION_EU_WEST_1) + session = set_mocked_session(AWS_REGION_EU_WEST_1) with patch( "prowler.providers.aws.lib.security_hub.security_hub.session.Session.client", @@ -247,167 +179,141 @@ class Test_SecurityHub: ( "root", ERROR, - f"Exception -- [70]: {error_message}", + f"Exception -- [90]: {error_message}", ) ] - def test_prepare_security_hub_findings_enabled_region_all_statuses(self): + def test_filter_security_hub_findings_per_region_enabled_region_all_statuses(self): enabled_regions = [AWS_REGION_EU_WEST_1] - output_options = self.set_mocked_output_options() - findings = [self.generate_finding("PASS", AWS_REGION_EU_WEST_1)] - aws_provider = set_mocked_aws_provider( - audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2] - ) + findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1)] + asff = ASFF(findings=findings) - assert prepare_security_hub_findings( - findings, - aws_provider, - output_options, + assert filter_security_hub_findings_per_region( + asff.data, + False, + [], enabled_regions, - ) == { - AWS_REGION_EU_WEST_1: [get_security_hub_finding("PASSED")], - } + ) == {AWS_REGION_EU_WEST_1: [asff.data[0]]} - def test_prepare_security_hub_findings_all_statuses_MANUAL_finding(self): + def test_filter_security_hub_findings_per_region_all_statuses_MANUAL_finding(self): enabled_regions = [AWS_REGION_EU_WEST_1] - output_options = self.set_mocked_output_options() - findings = [self.generate_finding("MANUAL", AWS_REGION_EU_WEST_1)] - aws_provider = set_mocked_aws_provider( - audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2] - ) - - assert prepare_security_hub_findings( - findings, - aws_provider, - output_options, - enabled_regions, - ) == {AWS_REGION_EU_WEST_1: []} - - def test_prepare_security_hub_findings_disabled_region(self): - enabled_regions = [AWS_REGION_EU_WEST_1] - output_options = self.set_mocked_output_options() - findings = [self.generate_finding("PASS", AWS_REGION_EU_WEST_2)] - aws_provider = set_mocked_aws_provider( - audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2] - ) - - assert prepare_security_hub_findings( - findings, - aws_provider, - output_options, - enabled_regions, - ) == {AWS_REGION_EU_WEST_1: []} - - def test_prepare_security_hub_findings_PASS_and_FAIL_statuses(self): - enabled_regions = [AWS_REGION_EU_WEST_1] - output_options = self.set_mocked_output_options(status=["FAIL"]) - findings = [self.generate_finding("PASS", AWS_REGION_EU_WEST_1)] - aws_provider = set_mocked_aws_provider( - audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2] - ) - - assert prepare_security_hub_findings( - findings, - aws_provider, - output_options, - enabled_regions, - ) == {AWS_REGION_EU_WEST_1: []} - - def test_prepare_security_hub_findings_FAIL_and_FAIL_statuses(self): - enabled_regions = [AWS_REGION_EU_WEST_1] - output_options = self.set_mocked_output_options(status=["FAIL"]) - findings = [self.generate_finding("FAIL", AWS_REGION_EU_WEST_1)] - aws_provider = set_mocked_aws_provider( - audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2] - ) - - assert prepare_security_hub_findings( - findings, - aws_provider, - output_options, - enabled_regions, - ) == {AWS_REGION_EU_WEST_1: [get_security_hub_finding("FAILED")]} - - def test_prepare_security_hub_findings_send_sh_only_fails_PASS(self): - enabled_regions = [AWS_REGION_EU_WEST_1] - output_options = self.set_mocked_output_options(send_sh_only_fails=True) - findings = [self.generate_finding("PASS", AWS_REGION_EU_WEST_1)] - aws_provider = set_mocked_aws_provider( - audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2] - ) - - assert prepare_security_hub_findings( - findings, - aws_provider, - output_options, - enabled_regions, - ) == {AWS_REGION_EU_WEST_1: []} - - def test_prepare_security_hub_findings_send_sh_only_fails_FAIL(self): - enabled_regions = [AWS_REGION_EU_WEST_1] - output_options = self.set_mocked_output_options(send_sh_only_fails=True) - findings = [self.generate_finding("FAIL", AWS_REGION_EU_WEST_1)] - aws_provider = set_mocked_aws_provider( - audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2] - ) - - assert prepare_security_hub_findings( - findings, - aws_provider, - output_options, - enabled_regions, - ) == {AWS_REGION_EU_WEST_1: [get_security_hub_finding("FAILED")]} - - def test_prepare_security_hub_findings_no_audited_regions(self): - enabled_regions = [AWS_REGION_EU_WEST_1] - output_options = self.set_mocked_output_options() - findings = [self.generate_finding("PASS", AWS_REGION_EU_WEST_1)] - aws_provider = set_mocked_aws_provider() - - assert prepare_security_hub_findings( - findings, - aws_provider, - output_options, - enabled_regions, - ) == { - AWS_REGION_EU_WEST_1: [get_security_hub_finding("PASSED")], - } - - def test_prepare_security_hub_findings_muted_fail_with_send_sh_only_fails(self): - enabled_regions = [AWS_REGION_EU_WEST_1] - output_options = self.set_mocked_output_options( - send_sh_only_fails=True, - ) findings = [ - self.generate_finding( + generate_finding_output(status="MANUAL", region=AWS_REGION_EU_WEST_1) + ] + asff = ASFF(findings=findings) + + assert filter_security_hub_findings_per_region( + asff.data, + False, + [], + enabled_regions, + ) == {AWS_REGION_EU_WEST_1: []} + + def test_filter_security_hub_findings_per_region_disabled_region(self): + enabled_regions = [AWS_REGION_EU_WEST_1] + findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_2)] + asff = ASFF(findings=findings) + + assert filter_security_hub_findings_per_region( + asff.data, + False, + [], + enabled_regions, + ) == {AWS_REGION_EU_WEST_1: []} + + def test_filter_security_hub_findings_per_region_PASS_and_FAIL_statuses(self): + enabled_regions = [AWS_REGION_EU_WEST_1] + findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1)] + asff = ASFF(findings=findings) + + assert filter_security_hub_findings_per_region( + asff.data, + False, + ["FAIL"], + enabled_regions, + ) == {AWS_REGION_EU_WEST_1: []} + + def test_filter_security_hub_findings_per_region_FAIL_and_FAIL_statuses(self): + enabled_regions = [AWS_REGION_EU_WEST_1] + findings = [generate_finding_output(status="FAIL", region=AWS_REGION_EU_WEST_1)] + asff = ASFF(findings=findings) + + assert filter_security_hub_findings_per_region( + asff.data, + False, + ["FAIL"], + enabled_regions, + ) == {AWS_REGION_EU_WEST_1: [asff.data[0]]} + + def test_filter_security_hub_findings_per_region_send_sh_only_fails_PASS(self): + enabled_regions = [AWS_REGION_EU_WEST_1] + findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1)] + asff = ASFF(findings=findings) + + assert filter_security_hub_findings_per_region( + asff.data, + True, + [], + enabled_regions, + ) == {AWS_REGION_EU_WEST_1: []} + + def test_filter_security_hub_findings_per_region_send_sh_only_fails_FAIL(self): + enabled_regions = [AWS_REGION_EU_WEST_1] + findings = [generate_finding_output(status="FAIL", region=AWS_REGION_EU_WEST_1)] + asff = ASFF(findings=findings) + + assert filter_security_hub_findings_per_region( + asff.data, + True, + [], + enabled_regions, + ) == {AWS_REGION_EU_WEST_1: [asff.data[0]]} + + def test_filter_security_hub_findings_per_region_no_audited_regions(self): + enabled_regions = [AWS_REGION_EU_WEST_1] + findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1)] + asff = ASFF(findings=findings) + + assert filter_security_hub_findings_per_region( + asff.data, + False, + [], + enabled_regions, + ) == {AWS_REGION_EU_WEST_1: [asff.data[0]]} + + def test_filter_security_hub_findings_per_region_muted_fail_with_send_sh_only_fails( + self, + ): + enabled_regions = [AWS_REGION_EU_WEST_1] + findings = [ + generate_finding_output( status="FAIL", region=AWS_REGION_EU_WEST_1, muted=True ) ] - aws_provider = set_mocked_aws_provider() + asff = ASFF(findings=findings) - assert prepare_security_hub_findings( - findings, - aws_provider, - output_options, + assert filter_security_hub_findings_per_region( + asff.data, + True, + [], enabled_regions, ) == { AWS_REGION_EU_WEST_1: [], } - def test_prepare_security_hub_findings_muted_fail_with_status_FAIL(self): + def test_filter_security_hub_findings_per_region_muted_fail_with_status_FAIL(self): enabled_regions = [AWS_REGION_EU_WEST_1] - output_options = self.set_mocked_output_options(status=["FAIL"]) findings = [ - self.generate_finding( + generate_finding_output( status="FAIL", region=AWS_REGION_EU_WEST_1, muted=True ) ] - aws_provider = set_mocked_aws_provider() + asff = ASFF(findings=findings) - assert prepare_security_hub_findings( - findings, - aws_provider, - output_options, + assert filter_security_hub_findings_per_region( + asff.data, + False, + ["FAIL"], enabled_regions, ) == { AWS_REGION_EU_WEST_1: [], @@ -415,18 +321,18 @@ class Test_SecurityHub: @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) def test_batch_send_to_security_hub_one_finding(self): - enabled_regions = [AWS_REGION_EU_WEST_1] - output_options = self.set_mocked_output_options() - findings = [self.generate_finding("PASS", AWS_REGION_EU_WEST_1)] - aws_provider = set_mocked_aws_provider( - audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2] - ) - session = self.set_mocked_session(AWS_REGION_EU_WEST_1) + enabled_regions = [AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2] + findings = [ + generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1), + generate_finding_output(status="FAIL", region=AWS_REGION_EU_WEST_2), + ] + asff = ASFF(findings=findings) + session = set_mocked_session(AWS_REGION_EU_WEST_1) - security_hub_findings = prepare_security_hub_findings( - findings, - aws_provider, - output_options, + security_hub_findings = filter_security_hub_findings_per_region( + asff.data, + False, + [], enabled_regions, ) @@ -435,5 +341,5 @@ class Test_SecurityHub: security_hub_findings, session, ) - == 1 + == 2 )