mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 13:01:56 +00:00
chore(iso27001): add ISO27001 output class (#4441)
Co-authored-by: Pepe Fagoaga <pepe@prowler.com>
This commit is contained in:
Sergio Garcia
@@ -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:
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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}"
|
||||
)
|
||||
@@ -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
|
||||
@@ -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__)
|
||||
@@ -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
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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"],
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user