ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 12:31:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(ens): add ENS output class (#4435)

Browse Source
This commit is contained in:
Sergio Garcia
2024-07-12 10:50:41 -04:00
committed by GitHub
parent 3be9de376a
commit a505776227
10 changed files with 279 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/__main__.py
+14
View File
@@ -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:
prowler/lib/cli/parser.py
+1 -1
View File
@@ -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(
prowler/lib/outputs/compliance/compliance.py
+8 -12
View File
@@ -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,
prowler/lib/outputs/compliance/ens_rd2022_aws.py → prowler/lib/outputs/compliance/ens/ens.py
+3 -55
View File
@@ -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}"
prowler/lib/outputs/compliance/ens/ens_aws.py
+103
View File
@@ -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}"
)
prowler/lib/outputs/compliance/ens/models.py
+30
View File
@@ -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
prowler/lib/outputs/compliance/models.py
-27
View File
@@ -2,34 +2,7 @@ from typing import Optional
from pydantic import BaseModel
# TODO: move this to outputs/<compliance>/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):
prowler/lib/outputs/file_descriptors.py
+3 -9
View File
@@ -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,
tests/lib/outputs/compliance/ens/ens_aws_test.py
+86
View File
@@ -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
tests/lib/outputs/compliance/fixtures.py
+31
View File
@@ -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",