diff --git a/prowler/__main__.py b/prowler/__main__.py index b56b0ff7da..ca7f457f06 100644 --- a/prowler/__main__.py +++ b/prowler/__main__.py @@ -52,6 +52,7 @@ from prowler.lib.outputs.compliance.cis.cis_gcp import GCPCIS from prowler.lib.outputs.compliance.cis.cis_kubernetes import KubernetesCIS from prowler.lib.outputs.compliance.compliance import display_compliance_table from prowler.lib.outputs.compliance.ens.ens_aws import AWSENS +from prowler.lib.outputs.compliance.iso27001.iso27001_aws import AWSISO27001 from prowler.lib.outputs.compliance.mitre_attack.mitre_attack_aws import AWSMitreAttack from prowler.lib.outputs.compliance.mitre_attack.mitre_attack_azure import ( AzureMitreAttack, @@ -421,6 +422,19 @@ def prowler(): file_path=filename, ) aws_well_architected_finding.batch_write_data_to_file() + elif compliance_name.startswith("iso27001_"): + # Generate ISO27001 Finding Object + filename = ( + f"{global_provider.output_options.output_directory}/compliance/" + f"{global_provider.output_options.output_filename}_{compliance_name}.csv" + ) + iso27001_finding = AWSISO27001( + findings=finding_outputs, + compliance=bulk_compliance_frameworks[compliance_name], + create_file_descriptor=True, + file_path=filename, + ) + iso27001_finding.batch_write_data_to_file() elif provider == "azure": for compliance_name in input_compliance_frameworks: diff --git a/prowler/lib/outputs/compliance/compliance.py b/prowler/lib/outputs/compliance/compliance.py index cbeb4e8afc..47669d33b2 100644 --- a/prowler/lib/outputs/compliance/compliance.py +++ b/prowler/lib/outputs/compliance/compliance.py @@ -8,9 +8,6 @@ from prowler.lib.outputs.compliance.generic import ( get_generic_compliance_table, write_compliance_row_generic, ) -from prowler.lib.outputs.compliance.iso27001_2013_aws import ( - write_compliance_row_iso27001_2013_aws, -) from prowler.lib.outputs.compliance.mitre_attack.mitre_attack import ( get_mitre_attack_table, ) @@ -104,9 +101,7 @@ def fill_compliance( and compliance.Version == "2013" and compliance.Provider == "AWS" ): - write_compliance_row_iso27001_2013_aws( - file_descriptors, finding, compliance, output_options, provider - ) + continue else: write_compliance_row_generic( diff --git a/prowler/lib/outputs/compliance/iso27001/__init__.py b/prowler/lib/outputs/compliance/iso27001/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/lib/outputs/compliance/iso27001/iso27001_aws.py b/prowler/lib/outputs/compliance/iso27001/iso27001_aws.py new file mode 100644 index 0000000000..f16829cb10 --- /dev/null +++ b/prowler/lib/outputs/compliance/iso27001/iso27001_aws.py @@ -0,0 +1,92 @@ +from csv import DictWriter + +from prowler.lib import logger +from prowler.lib.check.compliance_models import ComplianceBaseModel +from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput +from prowler.lib.outputs.compliance.iso27001.models import ISO27001AWS +from prowler.lib.outputs.finding import Finding + + +class AWSISO27001(ComplianceOutput): + """ + This class represents the AWS ISO 27001 compliance output. + + Attributes: + - _data (list): A list to store transformed data from findings. + - _file_descriptor (TextIOWrapper): A file descriptor to write data to a file. + + Methods: + - transform: Transforms findings into AWS ENS compliance format. + - batch_write_data_to_file: Writes the findings data to a CSV file in AWS ENS compliance format. + """ + + def transform( + self, + findings: list[Finding], + compliance: ComplianceBaseModel, + compliance_name: str, + ) -> None: + """ + Transforms a list of findings into AWS ENS compliance format. + + Parameters: + - findings (list): A list of findings. + - compliance (ComplianceBaseModel): A compliance model. + - compliance_name (str): The name of the compliance model. + + Returns: + - None + """ + for finding in findings: + # Get the compliance requirements for the finding + finding_requirements = finding.compliance.get(compliance_name, []) + for requirement in compliance.Requirements: + if requirement.Id in finding_requirements: + for attribute in requirement.Attributes: + compliance_row = ISO27001AWS( + Provider=finding.provider, + Description=compliance.Description, + AccountId=finding.account_uid, + Region=finding.region, + AssessmentDate=str(finding.timestamp), + Requirements_Attributes_Category=attribute.Category, + Requirements_Attributes_Objetive_ID=attribute.Objetive_ID, + Requirements_Attributes_Objetive_Name=attribute.Objetive_Name, + Requirements_Attributes_Check_Summary=attribute.Check_Summary, + Status=finding.status, + StatusExtended=finding.status_extended, + ResourceId=finding.resource_uid, + CheckId=finding.check_id, + Muted=finding.muted, + ResourceName=finding.resource_name, + ) + self._data.append(compliance_row) + + def batch_write_data_to_file(self) -> None: + """ + Writes the findings data to a CSV file in AWS ENS compliance format. + + Returns: + - None + """ + try: + if ( + getattr(self, "_file_descriptor", None) + and not self._file_descriptor.closed + and self._data + ): + csv_writer = DictWriter( + self._file_descriptor, + fieldnames=[field.upper() for field in self._data[0].dict().keys()], + delimiter=";", + ) + csv_writer.writeheader() + for finding in self._data: + csv_writer.writerow( + {k.upper(): v for k, v in finding.dict().items()} + ) + 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/compliance/iso27001/models.py b/prowler/lib/outputs/compliance/iso27001/models.py new file mode 100644 index 0000000000..880ff77fff --- /dev/null +++ b/prowler/lib/outputs/compliance/iso27001/models.py @@ -0,0 +1,23 @@ +from pydantic import BaseModel + + +class ISO27001AWS(BaseModel): + """ + ISO27001AWS generates a finding's output in CSV AWS ISO27001 format. + """ + + Provider: str + Description: str + AccountId: str + Region: str + AssessmentDate: str + Requirements_Attributes_Category: str + Requirements_Attributes_Objetive_ID: str + Requirements_Attributes_Objetive_Name: str + Requirements_Attributes_Check_Summary: str + Status: str + StatusExtended: str + ResourceId: str + CheckId: str + Muted: bool + ResourceName: str diff --git a/prowler/lib/outputs/compliance/iso27001_2013_aws.py b/prowler/lib/outputs/compliance/iso27001_2013_aws.py deleted file mode 100644 index 6f31769b2a..0000000000 --- a/prowler/lib/outputs/compliance/iso27001_2013_aws.py +++ /dev/null @@ -1,52 +0,0 @@ -from csv import DictWriter - -from prowler.config.config import timestamp -from prowler.lib.outputs.compliance.models import Check_Output_CSV_AWS_ISO27001_2013 -from prowler.lib.outputs.csv.csv import generate_csv_fields -from prowler.lib.utils.utils import outputs_unix_timestamp - - -def write_compliance_row_iso27001_2013_aws( - file_descriptors, finding, compliance, output_options, provider -): - compliance_output = compliance.Framework - if compliance.Version != "": - compliance_output += "_" + compliance.Version - if compliance.Provider != "": - compliance_output += "_" + compliance.Provider - - compliance_output = compliance_output.lower().replace("-", "_") - csv_header = generate_csv_fields(Check_Output_CSV_AWS_ISO27001_2013) - csv_writer = DictWriter( - file_descriptors[compliance_output], - fieldnames=csv_header, - delimiter=";", - ) - for requirement in compliance.Requirements: - requirement_description = requirement.Description - requirement_id = requirement.Id - requirement_name = requirement.Name - for attribute in requirement.Attributes: - compliance_row = Check_Output_CSV_AWS_ISO27001_2013( - Provider=finding.check_metadata.Provider, - Description=compliance.Description, - AccountId=provider.identity.account, - Region=finding.region, - AssessmentDate=outputs_unix_timestamp( - output_options.unix_timestamp, timestamp - ), - Requirements_Id=requirement_id, - Requirements_Name=requirement_name, - Requirements_Description=requirement_description, - Requirements_Attributes_Category=attribute.Category, - Requirements_Attributes_Objetive_ID=attribute.Objetive_ID, - Requirements_Attributes_Objetive_Name=attribute.Objetive_Name, - Requirements_Attributes_Check_Summary=attribute.Check_Summary, - Status=finding.status, - StatusExtended=finding.status_extended, - ResourceId=finding.resource_id, - CheckId=finding.check_metadata.CheckID, - Muted=finding.muted, - ) - - csv_writer.writerow(compliance_row.__dict__) diff --git a/prowler/lib/outputs/compliance/models.py b/prowler/lib/outputs/compliance/models.py index f36556fc8e..1600a4c99a 100644 --- a/prowler/lib/outputs/compliance/models.py +++ b/prowler/lib/outputs/compliance/models.py @@ -27,24 +27,3 @@ class Check_Output_CSV_Generic_Compliance(BaseModel): ResourceId: str CheckId: str Muted: bool - - -class Check_Output_CSV_AWS_ISO27001_2013(BaseModel): - """ - Check_Output_CSV_AWS_ISO27001_2013 generates a finding's output in CSV AWS ISO27001 Compliance format. - """ - - Provider: str - Description: str - AccountId: str - Region: str - AssessmentDate: str - Requirements_Attributes_Category: str - Requirements_Attributes_Objetive_ID: str - Requirements_Attributes_Objetive_Name: str - Requirements_Attributes_Check_Summary: str - Status: str - StatusExtended: str - ResourceId: str - CheckId: str - Muted: bool diff --git a/prowler/lib/outputs/file_descriptors.py b/prowler/lib/outputs/file_descriptors.py index 63a8ad7693..b424b2d477 100644 --- a/prowler/lib/outputs/file_descriptors.py +++ b/prowler/lib/outputs/file_descriptors.py @@ -4,10 +4,7 @@ from typing import Any from prowler.config.config import csv_file_suffix from prowler.lib.logger import logger -from prowler.lib.outputs.compliance.models import ( - Check_Output_CSV_AWS_ISO27001_2013, - Check_Output_CSV_Generic_Compliance, -) +from prowler.lib.outputs.compliance.models import Check_Output_CSV_Generic_Compliance from prowler.lib.outputs.csv.csv import generate_csv_fields from prowler.lib.outputs.output import Finding from prowler.lib.utils.utils import file_exists, open_file @@ -65,6 +62,8 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi continue elif "aws_well_architected_framework" in output_mode: continue + elif "iso27001_" in output_mode: + continue elif provider.type == "gcp": filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}" @@ -93,20 +92,12 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi elif provider.type == "aws": # Compliance frameworks filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}" - if output_mode == "iso27001_2013_aws": - file_descriptor = initialize_file_descriptor( - filename, - Check_Output_CSV_AWS_ISO27001_2013, - ) - file_descriptors.update({output_mode: file_descriptor}) - - else: - # Generic Compliance framework - file_descriptor = initialize_file_descriptor( - filename, - Check_Output_CSV_Generic_Compliance, - ) - file_descriptors.update({output_mode: file_descriptor}) + # Generic Compliance framework + file_descriptor = initialize_file_descriptor( + filename, + Check_Output_CSV_Generic_Compliance, + ) + file_descriptors.update({output_mode: file_descriptor}) except Exception as error: logger.error( diff --git a/tests/lib/outputs/compliance/fixtures.py b/tests/lib/outputs/compliance/fixtures.py index f481d3e6b0..539c5a1d25 100644 --- a/tests/lib/outputs/compliance/fixtures.py +++ b/tests/lib/outputs/compliance/fixtures.py @@ -6,6 +6,7 @@ from prowler.lib.check.compliance_models import ( ENS_Requirement_Attribute, ENS_Requirement_Attribute_Nivel, ENS_Requirement_Attribute_Tipos, + ISO27001_2013_Requirement_Attribute, Mitre_Requirement, Mitre_Requirement_Attribute_AWS, Mitre_Requirement_Attribute_Azure, @@ -351,3 +352,26 @@ AWS_WELL_ARCHITECTED = ComplianceBaseModel( ), ], ) +AWISO27001_2013_AWS_NAME = "iso27001_2013_aws" +ISO27001_2013_AWS = ComplianceBaseModel( + Framework="ISO27001", + Provider="AWS", + Version="2013", + Description="ISO (the International Organization for Standardization) and IEC (the International Electrotechnical Commission) form the specialized system for worldwide standardization. National bodies that are members of ISO or IEC participate in the development of International Standards through technical committees established by the respective organization to deal with particular fields of technical activity. ISO and IEC technical committees collaborate in fields of mutual interest. Other international organizations, governmental and non-governmental, in liaison with ISO and IEC, also take part in the work.", + Requirements=[ + Compliance_Requirement( + Id="A.10.1", + Description="Setup Encryption at rest for RDS instances", + Name="Cryptographic Controls", + Attributes=[ + ISO27001_2013_Requirement_Attribute( + Category="A.10 Cryptography", + Objetive_ID="A.10.1", + Objetive_Name="Cryptographic Controls", + Check_Summary="Setup Encryption at rest for RDS instances", + ) + ], + Checks=["rds_instance_storage_encrypted"], + ), + ], +) diff --git a/tests/lib/outputs/compliance/iso27001/iso27001_aws_test.py b/tests/lib/outputs/compliance/iso27001/iso27001_aws_test.py new file mode 100644 index 0000000000..e915831d43 --- /dev/null +++ b/tests/lib/outputs/compliance/iso27001/iso27001_aws_test.py @@ -0,0 +1,61 @@ +from datetime import datetime +from io import StringIO + +from freezegun import freeze_time +from mock import patch + +from prowler.lib.outputs.compliance.iso27001.iso27001_aws import AWSISO27001 +from prowler.lib.outputs.compliance.iso27001.models import ISO27001AWS +from tests.lib.outputs.compliance.fixtures import ISO27001_2013_AWS +from tests.lib.outputs.fixtures.fixtures import generate_finding_output +from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1 + + +class TestAWSISO27001: + def test_output_transform(self): + findings = [generate_finding_output(compliance={"ISO27001-2013": "A.10.1"})] + + output = AWSISO27001(findings, ISO27001_2013_AWS) + output_data = output.data[0] + assert isinstance(output_data, ISO27001AWS) + assert output_data.Provider == "aws" + assert output_data.AccountId == AWS_ACCOUNT_NUMBER + assert output_data.Region == AWS_REGION_EU_WEST_1 + assert output_data.Description == ISO27001_2013_AWS.Description + assert ( + output_data.Requirements_Attributes_Category + == ISO27001_2013_AWS.Requirements[0].Attributes[0].Category + ) + assert ( + output_data.Requirements_Attributes_Objetive_ID + == ISO27001_2013_AWS.Requirements[0].Attributes[0].Objetive_ID + ) + assert ( + output_data.Requirements_Attributes_Objetive_Name + == ISO27001_2013_AWS.Requirements[0].Attributes[0].Objetive_Name + ) + assert ( + output_data.Requirements_Attributes_Check_Summary + == ISO27001_2013_AWS.Requirements[0].Attributes[0].Check_Summary + ) + assert output_data.Status == "PASS" + assert output_data.StatusExtended == "" + assert output_data.ResourceId == "" + assert output_data.ResourceName == "" + assert output_data.CheckId == "test-check-id" + assert output_data.Muted is False + + @freeze_time(datetime.now()) + def test_batch_write_data_to_file(self): + mock_file = StringIO() + findings = [generate_finding_output(compliance={"ISO27001-2013": "A.10.1"})] + output = AWSISO27001(findings, ISO27001_2013_AWS) + output._file_descriptor = mock_file + + with patch.object(mock_file, "close", return_value=None): + output.batch_write_data_to_file() + + mock_file.seek(0) + content = mock_file.read() + expected_csv = f"""PROVIDER;DESCRIPTION;ACCOUNTID;REGION;ASSESSMENTDATE;REQUIREMENTS_ATTRIBUTES_CATEGORY;REQUIREMENTS_ATTRIBUTES_OBJETIVE_ID;REQUIREMENTS_ATTRIBUTES_OBJETIVE_NAME;REQUIREMENTS_ATTRIBUTES_CHECK_SUMMARY;STATUS;STATUSEXTENDED;RESOURCEID;CHECKID;MUTED;RESOURCENAME\r\naws;ISO (the International Organization for Standardization) and IEC (the International Electrotechnical Commission) form the specialized system for worldwide standardization. National bodies that are members of ISO or IEC participate in the development of International Standards through technical committees established by the respective organization to deal with particular fields of technical activity. ISO and IEC technical committees collaborate in fields of mutual interest. Other international organizations, governmental and non-governmental, in liaison with ISO and IEC, also take part in the work.;123456789012;eu-west-1;{datetime.now()};A.10 Cryptography;A.10.1;Cryptographic Controls;Setup Encryption at rest for RDS instances;PASS;;;test-check-id;False;\r\n""" + assert content == expected_csv