From a5057762275d344002ee36d6971dada080d27013 Mon Sep 17 00:00:00 2001 From: Sergio Garcia <38561120+sergargar@users.noreply.github.com> Date: Fri, 12 Jul 2024 10:50:41 -0400 Subject: [PATCH] chore(ens): add ENS output class (#4435) --- prowler/__main__.py | 14 +++ prowler/lib/cli/parser.py | 2 +- prowler/lib/outputs/compliance/compliance.py | 20 ++-- .../{ens_rd2022_aws.py => ens/ens.py} | 58 +--------- prowler/lib/outputs/compliance/ens/ens_aws.py | 103 ++++++++++++++++++ prowler/lib/outputs/compliance/ens/models.py | 30 +++++ prowler/lib/outputs/compliance/models.py | 27 ----- prowler/lib/outputs/file_descriptors.py | 12 +- .../outputs/compliance/ens/ens_aws_test.py | 86 +++++++++++++++ tests/lib/outputs/compliance/fixtures.py | 31 ++++++ 10 files changed, 279 insertions(+), 104 deletions(-) rename prowler/lib/outputs/compliance/{ens_rd2022_aws.py => ens/ens.py} (67%) create mode 100644 prowler/lib/outputs/compliance/ens/ens_aws.py create mode 100644 prowler/lib/outputs/compliance/ens/models.py create mode 100644 tests/lib/outputs/compliance/ens/ens_aws_test.py diff --git a/prowler/__main__.py b/prowler/__main__.py index aeade8688c..4a8ccef055 100644 --- a/prowler/__main__.py +++ b/prowler/__main__.py @@ -48,6 +48,7 @@ from prowler.lib.outputs.compliance.cis.cis_azure import AzureCIS 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.mitre_attack.mitre_attack_aws import AWSMitreAttack from prowler.lib.outputs.compliance.mitre_attack.mitre_attack_azure import ( AzureMitreAttack, @@ -391,6 +392,19 @@ def prowler(): file_path=filename, ) mitre_attack_finding.batch_write_data_to_file() + elif compliance_name.startswith("ens_"): + # Generate ENS Finding Object + filename = ( + f"{global_provider.output_options.output_directory}/compliance/" + f"{global_provider.output_options.output_filename}_{compliance_name}.csv" + ) + ens_finding = AWSENS( + findings=finding_outputs, + compliance=bulk_compliance_frameworks[compliance_name], + create_file_descriptor=True, + file_path=filename, + ) + ens_finding.batch_write_data_to_file() elif provider == "azure": for compliance_name in input_compliance_frameworks: diff --git a/prowler/lib/cli/parser.py b/prowler/lib/cli/parser.py index bb3af1de2f..2309d8742c 100644 --- a/prowler/lib/cli/parser.py +++ b/prowler/lib/cli/parser.py @@ -263,7 +263,7 @@ Detailed documentation at https://docs.prowler.com group.add_argument( "--compliance", nargs="+", - help="Compliance Framework to check against for. The format should be the following: framework_version_provider (e.g.: ens_rd2022_aws)", + help="Compliance Framework to check against for. The format should be the following: framework_version_provider (e.g.: cis_3.0_aws)", choices=available_compliance_frameworks, ) group.add_argument( diff --git a/prowler/lib/outputs/compliance/compliance.py b/prowler/lib/outputs/compliance/compliance.py index 924224decb..077de85473 100644 --- a/prowler/lib/outputs/compliance/compliance.py +++ b/prowler/lib/outputs/compliance/compliance.py @@ -6,10 +6,7 @@ from prowler.lib.outputs.compliance.aws_well_architected_framework import ( write_compliance_row_aws_well_architected_framework, ) from prowler.lib.outputs.compliance.cis.cis import get_cis_table -from prowler.lib.outputs.compliance.ens_rd2022_aws import ( - get_ens_rd2022_aws_table, - write_compliance_row_ens_rd2022_aws, -) +from prowler.lib.outputs.compliance.ens.ens import get_ens_table from prowler.lib.outputs.compliance.generic import ( get_generic_compliance_table, write_compliance_row_generic, @@ -96,13 +93,12 @@ def fill_compliance( ) for compliance in check_compliances: - if compliance.Framework == "ENS" and compliance.Version == "RD2022": - write_compliance_row_ens_rd2022_aws( - file_descriptors, finding, compliance, output_options, provider - ) - # FIXME: Remove this once we merge all the compliance frameworks - elif compliance.Framework == "CIS": + if compliance.Framework == "CIS": + continue + elif compliance.Framework == "MITRE-ATTACK" and compliance.Version == "": + continue + elif compliance.Framework == "ENS": continue elif compliance.Framework == "MITRE-ATTACK" and compliance.Version == "": continue @@ -144,8 +140,8 @@ def display_compliance_table( compliance_overview: bool, ): try: - if "ens_rd2022_aws" == compliance_framework: - get_ens_rd2022_aws_table( + if "ens_" in compliance_framework: + get_ens_table( findings, bulk_checks_metadata, compliance_framework, diff --git a/prowler/lib/outputs/compliance/ens_rd2022_aws.py b/prowler/lib/outputs/compliance/ens/ens.py similarity index 67% rename from prowler/lib/outputs/compliance/ens_rd2022_aws.py rename to prowler/lib/outputs/compliance/ens/ens.py index 1dfc1164db..923638350f 100644 --- a/prowler/lib/outputs/compliance/ens_rd2022_aws.py +++ b/prowler/lib/outputs/compliance/ens/ens.py @@ -1,58 +1,10 @@ -from csv import DictWriter - from colorama import Fore, Style from tabulate import tabulate -from prowler.config.config import orange_color, timestamp -from prowler.lib.outputs.compliance.models import Check_Output_CSV_ENS_RD2022 -from prowler.lib.outputs.csv.csv import generate_csv_fields -from prowler.lib.utils.utils import outputs_unix_timestamp +from prowler.config.config import orange_color -def write_compliance_row_ens_rd2022_aws( - file_descriptors, finding, compliance, output_options, provider -): - compliance_output = "ens_rd2022_aws" - csv_header = generate_csv_fields(Check_Output_CSV_ENS_RD2022) - csv_writer = DictWriter( - file_descriptors[compliance_output], - fieldnames=csv_header, - delimiter=";", - ) - for requirement in compliance.Requirements: - requirement_description = requirement.Description - requirement_id = requirement.Id - for attribute in requirement.Attributes: - compliance_row = Check_Output_CSV_ENS_RD2022( - 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_Description=requirement_description, - Requirements_Attributes_IdGrupoControl=attribute.IdGrupoControl, - Requirements_Attributes_Marco=attribute.Marco, - Requirements_Attributes_Categoria=attribute.Categoria, - Requirements_Attributes_DescripcionControl=attribute.DescripcionControl, - Requirements_Attributes_Nivel=attribute.Nivel, - Requirements_Attributes_Tipo=attribute.Tipo, - Requirements_Attributes_Dimensiones=",".join(attribute.Dimensiones), - Requirements_Attributes_ModoEjecucion=attribute.ModoEjecucion, - Requirements_Attributes_Dependencias=",".join(attribute.Dependencias), - 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__) - - -def get_ens_rd2022_aws_table( +def get_ens_table( findings: list, bulk_checks_metadata: dict, compliance_framework: str, @@ -78,11 +30,7 @@ def get_ens_rd2022_aws_table( check = bulk_checks_metadata[finding.check_metadata.CheckID] check_compliances = check.Compliance for compliance in check_compliances: - if ( - compliance.Framework == "ENS" - and compliance.Provider == "AWS" - and compliance.Version == "RD2022" - ): + if compliance.Framework == "ENS" and compliance.Provider == "AWS": for requirement in compliance.Requirements: for attribute in requirement.Attributes: marco_categoria = f"{attribute.Marco}/{attribute.Categoria}" diff --git a/prowler/lib/outputs/compliance/ens/ens_aws.py b/prowler/lib/outputs/compliance/ens/ens_aws.py new file mode 100644 index 0000000000..ed96576455 --- /dev/null +++ b/prowler/lib/outputs/compliance/ens/ens_aws.py @@ -0,0 +1,103 @@ +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.ens.models import ENSAWS +from prowler.lib.outputs.finding import Finding + + +class AWSENS(ComplianceOutput): + """ + This class represents the AWS ENS 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 = ENSAWS( + Provider=finding.provider, + Description=compliance.Description, + AccountId=finding.account_uid, + Region=finding.region, + AssessmentDate=str(finding.timestamp), + Requirements_Id=requirement.Id, + Requirements_Description=requirement.Description, + Requirements_Attributes_IdGrupoControl=attribute.IdGrupoControl, + Requirements_Attributes_Marco=attribute.Marco, + Requirements_Attributes_Categoria=attribute.Categoria, + Requirements_Attributes_DescripcionControl=attribute.DescripcionControl, + Requirements_Attributes_Nivel=attribute.Nivel, + Requirements_Attributes_Tipo=attribute.Tipo, + Requirements_Attributes_Dimensiones=",".join( + attribute.Dimensiones + ), + Requirements_Attributes_ModoEjecucion=attribute.ModoEjecucion, + Requirements_Attributes_Dependencias=",".join( + attribute.Dependencias + ), + 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/ens/models.py b/prowler/lib/outputs/compliance/ens/models.py new file mode 100644 index 0000000000..12b20ed84f --- /dev/null +++ b/prowler/lib/outputs/compliance/ens/models.py @@ -0,0 +1,30 @@ +from pydantic import BaseModel + + +class ENSAWS(BaseModel): + """ + ENSAWS generates a finding's output in CSV ENS format for AWS. + """ + + Provider: str + Description: str + AccountId: str + Region: str + AssessmentDate: str + Requirements_Id: str + Requirements_Description: str + Requirements_Attributes_IdGrupoControl: str + Requirements_Attributes_Marco: str + Requirements_Attributes_Categoria: str + Requirements_Attributes_DescripcionControl: str + Requirements_Attributes_Nivel: str + Requirements_Attributes_Tipo: str + Requirements_Attributes_Dimensiones: str + Requirements_Attributes_ModoEjecucion: str + Requirements_Attributes_Dependencias: str + Status: str + StatusExtended: str + ResourceId: str + CheckId: str + Muted: bool + ResourceName: str diff --git a/prowler/lib/outputs/compliance/models.py b/prowler/lib/outputs/compliance/models.py index c651f092b0..68a26ea982 100644 --- a/prowler/lib/outputs/compliance/models.py +++ b/prowler/lib/outputs/compliance/models.py @@ -2,34 +2,7 @@ from typing import Optional from pydantic import BaseModel - # TODO: move this to outputs//models.py -class Check_Output_CSV_ENS_RD2022(BaseModel): - """ - Check_Output_CSV_ENS_RD2022 generates a finding's output in CSV ENS RD2022 format. - """ - - Provider: str - Description: str - AccountId: str - Region: str - AssessmentDate: str - Requirements_Id: str - Requirements_Description: str - Requirements_Attributes_IdGrupoControl: str - Requirements_Attributes_Marco: str - Requirements_Attributes_Categoria: str - Requirements_Attributes_DescripcionControl: str - Requirements_Attributes_Nivel: str - Requirements_Attributes_Tipo: str - Requirements_Attributes_Dimensiones: str - Requirements_Attributes_ModoEjecucion: str - Requirements_Attributes_Dependencias: str - Status: str - StatusExtended: str - ResourceId: str - CheckId: str - Muted: bool class Check_Output_CSV_Generic_Compliance(BaseModel): diff --git a/prowler/lib/outputs/file_descriptors.py b/prowler/lib/outputs/file_descriptors.py index 4f6406285b..e33a268590 100644 --- a/prowler/lib/outputs/file_descriptors.py +++ b/prowler/lib/outputs/file_descriptors.py @@ -7,7 +7,6 @@ from prowler.lib.logger import logger from prowler.lib.outputs.compliance.models import ( Check_Output_CSV_AWS_ISO27001_2013, Check_Output_CSV_AWS_Well_Architected, - Check_Output_CSV_ENS_RD2022, Check_Output_CSV_Generic_Compliance, ) from prowler.lib.outputs.csv.csv import generate_csv_fields @@ -63,6 +62,8 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi continue elif "mitre_attack_" in output_mode: continue + elif "ens_" in output_mode: + continue elif provider.type == "gcp": filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}" @@ -91,14 +92,7 @@ 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 == "ens_rd2022_aws": - file_descriptor = initialize_file_descriptor( - filename, - Check_Output_CSV_ENS_RD2022, - ) - file_descriptors.update({output_mode: file_descriptor}) - - elif "aws_well_architected_framework" in output_mode: + if "aws_well_architected_framework" in output_mode: file_descriptor = initialize_file_descriptor( filename, Check_Output_CSV_AWS_Well_Architected, diff --git a/tests/lib/outputs/compliance/ens/ens_aws_test.py b/tests/lib/outputs/compliance/ens/ens_aws_test.py new file mode 100644 index 0000000000..e6dcbb2a35 --- /dev/null +++ b/tests/lib/outputs/compliance/ens/ens_aws_test.py @@ -0,0 +1,86 @@ +from datetime import datetime +from io import StringIO + +from freezegun import freeze_time +from mock import patch + +from prowler.lib.outputs.compliance.ens.ens_aws import AWSENS +from prowler.lib.outputs.compliance.ens.models import ENSAWS +from tests.lib.outputs.compliance.fixtures import ENS_RD2022_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 TestAWSENS: + def test_output_transform(self): + findings = [ + generate_finding_output(compliance={"ENS-RD2022": "op.exp.8.aws.ct.3"}) + ] + + output = AWSENS(findings, ENS_RD2022_AWS) + output_data = output.data[0] + assert isinstance(output_data, ENSAWS) + 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 == ENS_RD2022_AWS.Description + assert output_data.Requirements_Id == ENS_RD2022_AWS.Requirements[0].Id + assert ( + output_data.Requirements_Description + == ENS_RD2022_AWS.Requirements[0].Description + ) + assert ( + output_data.Requirements_Attributes_IdGrupoControl + == ENS_RD2022_AWS.Requirements[0].Attributes[0].IdGrupoControl + ) + assert ( + output_data.Requirements_Attributes_Marco + == ENS_RD2022_AWS.Requirements[0].Attributes[0].Marco + ) + assert ( + output_data.Requirements_Attributes_Categoria + == ENS_RD2022_AWS.Requirements[0].Attributes[0].Categoria + ) + assert ( + output_data.Requirements_Attributes_DescripcionControl + == ENS_RD2022_AWS.Requirements[0].Attributes[0].DescripcionControl + ) + assert ( + output_data.Requirements_Attributes_Nivel + == ENS_RD2022_AWS.Requirements[0].Attributes[0].Nivel + ) + assert ( + output_data.Requirements_Attributes_Tipo + == ENS_RD2022_AWS.Requirements[0].Attributes[0].Tipo + ) + assert [ + output_data.Requirements_Attributes_Dimensiones + ] == ENS_RD2022_AWS.Requirements[0].Attributes[0].Dimensiones + assert ( + output_data.Requirements_Attributes_ModoEjecucion + == ENS_RD2022_AWS.Requirements[0].Attributes[0].ModoEjecucion + ) + assert output_data.Requirements_Attributes_Dependencias == "" + 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={"ENS-RD2022": "op.exp.8.aws.ct.3"}) + ] + output = AWSENS(findings, ENS_RD2022_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_ID;REQUIREMENTS_DESCRIPTION;REQUIREMENTS_ATTRIBUTES_IDGRUPOCONTROL;REQUIREMENTS_ATTRIBUTES_MARCO;REQUIREMENTS_ATTRIBUTES_CATEGORIA;REQUIREMENTS_ATTRIBUTES_DESCRIPCIONCONTROL;REQUIREMENTS_ATTRIBUTES_NIVEL;REQUIREMENTS_ATTRIBUTES_TIPO;REQUIREMENTS_ATTRIBUTES_DIMENSIONES;REQUIREMENTS_ATTRIBUTES_MODOEJECUCION;REQUIREMENTS_ATTRIBUTES_DEPENDENCIAS;STATUS;STATUSEXTENDED;RESOURCEID;CHECKID;MUTED;RESOURCENAME\r\naws;The accreditation scheme of the ENS (National Security Scheme) has been developed by the Ministry of Finance and Public Administrations and the CCN (National Cryptological Center). This includes the basic principles and minimum requirements necessary for the adequate protection of information.;123456789012;eu-west-1;{datetime.now()};op.exp.8.aws.ct.3;Registro de actividad;op.exp.8;operacional;explotación;Habilitar la validación de archivos en todos los trails, evitando así que estos se vean modificados o eliminados.;alto;requisito;trazabilidad;automático;;PASS;;;test-check-id;False;\r\n""" + assert content == expected_csv diff --git a/tests/lib/outputs/compliance/fixtures.py b/tests/lib/outputs/compliance/fixtures.py index eb2ccbe3f8..3aa1fa6500 100644 --- a/tests/lib/outputs/compliance/fixtures.py +++ b/tests/lib/outputs/compliance/fixtures.py @@ -2,6 +2,9 @@ from prowler.lib.check.compliance_models import ( CIS_Requirement_Attribute, Compliance_Requirement, ComplianceBaseModel, + ENS_Requirement_Attribute, + ENS_Requirement_Attribute_Nivel, + ENS_Requirement_Attribute_Tipos, Mitre_Requirement, Mitre_Requirement_Attribute_AWS, Mitre_Requirement_Attribute_Azure, @@ -283,6 +286,34 @@ MITRE_ATTACK_GCP = ComplianceBaseModel( ) ], ) +ENS_RD2022_AWS_NAME = "ens_rd2022_aws" +ENS_RD2022_AWS = ComplianceBaseModel( + Framework="ENS", + Provider="AWS", + Version="RD2022", + Description="The accreditation scheme of the ENS (National Security Scheme) has been developed by the Ministry of Finance and Public Administrations and the CCN (National Cryptological Center). This includes the basic principles and minimum requirements necessary for the adequate protection of information.", + Requirements=[ + Compliance_Requirement( + Id="op.exp.8.aws.ct.3", + Description="Registro de actividad", + Name=None, + Attributes=[ + ENS_Requirement_Attribute( + IdGrupoControl="op.exp.8", + Marco="operacional", + Categoria="explotación", + DescripcionControl="Habilitar la validación de archivos en todos los trails, evitando así que estos se vean modificados o eliminados.", + Tipo=ENS_Requirement_Attribute_Tipos.requisito, + Nivel=ENS_Requirement_Attribute_Nivel.alto, + Dimensiones=["trazabilidad"], + ModoEjecucion="automático", + Dependencias=[], + ) + ], + Checks=["cloudtrail_log_file_validation_enabled"], + ) + ], +) NOT_PRESENT_COMPLIANCE_NAME = "not_present_compliance_name" NOT_PRESENT_COMPLIANCE = ComplianceBaseModel( Framework="NOT_EXISTENT",