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

chore(aws): add AWS Well-Architected output class (#4439)

Browse Source
This commit is contained in:
Sergio Garcia
2024-07-12 11:27:21 -04:00
committed by GitHub
parent a505776227
commit 027aa9796d
14 changed files with 292 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/__main__.py
+16
View File
@@ -43,6 +43,9 @@ 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.aws_well_architected.aws_well_architected import (
AWSWellArchitected,
)
from prowler.lib.outputs.compliance.cis.cis_aws import AWSCIS
from prowler.lib.outputs.compliance.cis.cis_azure import AzureCIS
from prowler.lib.outputs.compliance.cis.cis_gcp import GCPCIS
@@ -405,6 +408,19 @@ def prowler():
file_path=filename,
)
ens_finding.batch_write_data_to_file()
elif compliance_name.startswith("aws_well_architected_framework"):
# Generate AWS Well-Architected Finding Object
filename = (
f"{global_provider.output_options.output_directory}/compliance/"
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
)
aws_well_architected_finding = AWSWellArchitected(
findings=finding_outputs,
compliance=bulk_compliance_frameworks[compliance_name],
create_file_descriptor=True,
file_path=filename,
)
aws_well_architected_finding.batch_write_data_to_file()
elif provider == "azure":
for compliance_name in input_compliance_frameworks:
prowler/lib/outputs/compliance/aws_well_architected/__init__.py
View File
prowler/lib/outputs/compliance/aws_well_architected/aws_well_architected.py
+101
View File
@@ -0,0 +1,101 @@
from csv import DictWriter
from venv import logger
from prowler.lib.check.compliance_models import ComplianceBaseModel
from prowler.lib.outputs.compliance.aws_well_architected.models import (
AWSWellArchitected as AWSWellArchitectedModel,
)
from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput
from prowler.lib.outputs.finding import Finding
class AWSWellArchitected(ComplianceOutput):
"""
This class represents the AWS Well-Architected 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 Well-Architected compliance format.
- batch_write_data_to_file: Writes the findings data to a CSV file in AWS Well-Architected compliance format.
"""
def transform(
self,
findings: list[Finding],
compliance: ComplianceBaseModel,
compliance_name: str,
) -> None:
"""
Transforms a list of findings into AWS Well-Architected 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 = AWSWellArchitectedModel(
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_Name=attribute.Name,
Requirements_Attributes_WellArchitectedQuestionId=attribute.WellArchitectedQuestionId,
Requirements_Attributes_WellArchitectedPracticeId=attribute.WellArchitectedPracticeId,
Requirements_Attributes_Section=attribute.Section,
Requirements_Attributes_SubSection=attribute.SubSection,
Requirements_Attributes_LevelOfRisk=attribute.LevelOfRisk,
Requirements_Attributes_AssessmentMethod=attribute.AssessmentMethod,
Requirements_Attributes_Description=attribute.Description,
Requirements_Attributes_ImplementationGuidanceUrl=attribute.ImplementationGuidanceUrl,
Status=finding.status,
StatusExtended=finding.status_extended,
ResourceId=finding.resource_uid,
ResourceName=finding.resource_name,
CheckId=finding.check_id,
Muted=finding.muted,
)
self._data.append(compliance_row)
def batch_write_data_to_file(self) -> None:
"""
Writes the findings data to a CSV file in AWS Well-Architected 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/aws_well_architected/models.py
+32
View File
@@ -0,0 +1,32 @@
from typing import Optional
from pydantic import BaseModel
class AWSWellArchitected(BaseModel):
"""
AWSWellArchitected generates a finding's output in AWS Well-Architected Framework format.
"""
Provider: str
Description: str
AccountId: str
Region: str
AssessmentDate: str
Requirements_Id: str
Requirements_Description: str
Requirements_Attributes_Name: str
Requirements_Attributes_WellArchitectedQuestionId: str
Requirements_Attributes_WellArchitectedPracticeId: str
Requirements_Attributes_Section: str
Requirements_Attributes_SubSection: Optional[str]
Requirements_Attributes_LevelOfRisk: str
Requirements_Attributes_AssessmentMethod: str
Requirements_Attributes_Description: str
Requirements_Attributes_ImplementationGuidanceUrl: str
Status: str
StatusExtended: str
ResourceId: str
CheckId: str
Muted: bool
ResourceName: str
prowler/lib/outputs/compliance/aws_well_architected_framework.py
-60
View File
@@ -1,60 +0,0 @@
from csv import DictWriter
from prowler.config.config import timestamp
from prowler.lib.logger import logger
from prowler.lib.outputs.compliance.models import Check_Output_CSV_AWS_Well_Architected
from prowler.lib.outputs.csv.csv import generate_csv_fields
from prowler.lib.utils.utils import outputs_unix_timestamp
def write_compliance_row_aws_well_architected_framework(
file_descriptors, finding, compliance, output_options, provider
):
try:
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_Well_Architected)
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_AWS_Well_Architected(
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_Name=attribute.Name,
Requirements_Attributes_WellArchitectedQuestionId=attribute.WellArchitectedQuestionId,
Requirements_Attributes_WellArchitectedPracticeId=attribute.WellArchitectedPracticeId,
Requirements_Attributes_Section=attribute.Section,
Requirements_Attributes_SubSection=attribute.SubSection,
Requirements_Attributes_LevelOfRisk=attribute.LevelOfRisk,
Requirements_Attributes_AssessmentMethod=attribute.AssessmentMethod,
Requirements_Attributes_Description=attribute.Description,
Requirements_Attributes_ImplementationGuidanceUrl=attribute.ImplementationGuidanceUrl,
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__)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
prowler/lib/outputs/compliance/compliance.py
+1 -13
View File
@@ -2,9 +2,6 @@ import sys
from prowler.lib.check.models import Check_Report
from prowler.lib.logger import logger
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.ens import get_ens_table
from prowler.lib.outputs.compliance.generic import (
@@ -100,17 +97,8 @@ def fill_compliance(
continue
elif compliance.Framework == "ENS":
continue
elif compliance.Framework == "MITRE-ATTACK" and compliance.Version == "":
elif "AWS-Well-Architected-Framework" in compliance.Framework:
continue
elif (
"AWS-Well-Architected-Framework" in compliance.Framework
and compliance.Provider == "AWS"
):
write_compliance_row_aws_well_architected_framework(
file_descriptors, finding, compliance, output_options, provider
)
elif (
compliance.Framework == "ISO27001"
and compliance.Version == "2013"
prowler/lib/outputs/compliance/ens/__init__.py
View File
prowler/lib/outputs/compliance/models.py
-26
View File
@@ -29,32 +29,6 @@ class Check_Output_CSV_Generic_Compliance(BaseModel):
Muted: bool
class Check_Output_CSV_AWS_Well_Architected(BaseModel):
"""
Check_Output_CSV_AWS_Well_Architected generates a finding's output in CSV AWS Well Architected Compliance format.
"""
Provider: str
Description: str
AccountId: str
Region: str
AssessmentDate: str
Requirements_Attributes_Name: str
Requirements_Attributes_WellArchitectedQuestionId: str
Requirements_Attributes_WellArchitectedPracticeId: str
Requirements_Attributes_Section: str
Requirements_Attributes_SubSection: Optional[str]
Requirements_Attributes_LevelOfRisk: str
Requirements_Attributes_AssessmentMethod: str
Requirements_Attributes_Description: str
Requirements_Attributes_ImplementationGuidanceUrl: str
Status: str
StatusExtended: str
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.
prowler/lib/outputs/file_descriptors.py
+3 -9
View File
@@ -6,7 +6,6 @@ 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_AWS_Well_Architected,
Check_Output_CSV_Generic_Compliance,
)
from prowler.lib.outputs.csv.csv import generate_csv_fields
@@ -64,6 +63,8 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
continue
elif "ens_" in output_mode:
continue
elif "aws_well_architected_framework" in output_mode:
continue
elif provider.type == "gcp":
filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}"
@@ -92,14 +93,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 "aws_well_architected_framework" in output_mode:
file_descriptor = initialize_file_descriptor(
filename,
Check_Output_CSV_AWS_Well_Architected,
)
file_descriptors.update({output_mode: file_descriptor})
elif output_mode == "iso27001_2013_aws":
if output_mode == "iso27001_2013_aws":
file_descriptor = initialize_file_descriptor(
filename,
Check_Output_CSV_AWS_ISO27001_2013,
tests/lib/outputs/compliance/aws_well_architected/aws_well_architected_test.py
+107
View File
@@ -0,0 +1,107 @@
from datetime import datetime
from io import StringIO
from freezegun import freeze_time
from mock import patch
from prowler.lib.outputs.compliance.aws_well_architected.aws_well_architected import (
AWSWellArchitected,
)
from prowler.lib.outputs.compliance.aws_well_architected.models import (
AWSWellArchitected as AWSWellArchitectedModel,
)
from tests.lib.outputs.compliance.fixtures import AWS_WELL_ARCHITECTED
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 TestAWSWellArchitected:
def test_output_transform(self):
findings = [
generate_finding_output(
compliance={
"AWS-Well-Architected-Framework-Security-Pillar": "SEC01-BP01"
}
)
]
output = AWSWellArchitected(findings, AWS_WELL_ARCHITECTED)
output_data = output.data[0]
assert isinstance(output_data, AWSWellArchitectedModel)
assert output_data.Provider == "aws"
assert output_data.AccountId == AWS_ACCOUNT_NUMBER
assert output_data.Region == AWS_REGION_EU_WEST_1
assert output_data.Requirements_Id == AWS_WELL_ARCHITECTED.Requirements[0].Id
assert (
output_data.Requirements_Description
== AWS_WELL_ARCHITECTED.Requirements[0].Description
)
assert (
output_data.Requirements_Attributes_Name
== AWS_WELL_ARCHITECTED.Requirements[0].Attributes[0].Name
)
assert (
output_data.Requirements_Attributes_WellArchitectedQuestionId
== AWS_WELL_ARCHITECTED.Requirements[0]
.Attributes[0]
.WellArchitectedQuestionId
)
assert (
output_data.Requirements_Attributes_WellArchitectedPracticeId
== AWS_WELL_ARCHITECTED.Requirements[0]
.Attributes[0]
.WellArchitectedPracticeId
)
assert (
output_data.Requirements_Attributes_Section
== AWS_WELL_ARCHITECTED.Requirements[0].Attributes[0].Section
)
assert (
output_data.Requirements_Attributes_SubSection
== AWS_WELL_ARCHITECTED.Requirements[0].Attributes[0].SubSection
)
assert (
output_data.Requirements_Attributes_LevelOfRisk
== AWS_WELL_ARCHITECTED.Requirements[0].Attributes[0].LevelOfRisk
)
assert (
output_data.Requirements_Attributes_AssessmentMethod
== AWS_WELL_ARCHITECTED.Requirements[0].Attributes[0].AssessmentMethod
)
assert (
output_data.Requirements_Attributes_Description
== AWS_WELL_ARCHITECTED.Requirements[0].Attributes[0].Description
)
assert (
output_data.Requirements_Attributes_ImplementationGuidanceUrl
== AWS_WELL_ARCHITECTED.Requirements[0]
.Attributes[0]
.ImplementationGuidanceUrl
)
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={
"AWS-Well-Architected-Framework-Security-Pillar": "SEC01-BP01"
}
)
]
output = AWSWellArchitected(findings, AWS_WELL_ARCHITECTED)
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_NAME;REQUIREMENTS_ATTRIBUTES_WELLARCHITECTEDQUESTIONID;REQUIREMENTS_ATTRIBUTES_WELLARCHITECTEDPRACTICEID;REQUIREMENTS_ATTRIBUTES_SECTION;REQUIREMENTS_ATTRIBUTES_SUBSECTION;REQUIREMENTS_ATTRIBUTES_LEVELOFRISK;REQUIREMENTS_ATTRIBUTES_ASSESSMENTMETHOD;REQUIREMENTS_ATTRIBUTES_DESCRIPTION;REQUIREMENTS_ATTRIBUTES_IMPLEMENTATIONGUIDANCEURL;STATUS;STATUSEXTENDED;RESOURCEID;CHECKID;MUTED;RESOURCENAME\r\naws;Best Practices for AWS Well-Architected Framework Security Pillar. The focus of this framework is the security pillar of the AWS Well-Architected Framework. It provides guidance to help you apply best practices, current recommendations in the design, delivery, and maintenance of secure AWS workloads.;123456789012;eu-west-1;{datetime.now()};SEC01-BP01;Establish common guardrails and isolation between environments (such as production, development, and test) and workloads through a multi-account strategy. Account-level separation is strongly recommended, as it provides a strong isolation boundary for security, billing, and access.;SEC01-BP01 Separate workloads using accounts;securely-operate;sec_securely_operate_multi_accounts;Security foundations;AWS account management and separation;High;Automated;Establish common guardrails and isolation between environments (such as production, development, and test) and workloads through a multi-account strategy. Account-level separation is strongly recommended, as it provides a strong isolation boundary for security, billing, and access.;https://docs.aws.amazon.com/wellarchitected/latest/security-pillar/sec_securely_operate_multi_accounts.html#implementation-guidance.;PASS;;;test-check-id;False;\r\n"""
assert content == expected_csv
tests/lib/outputs/compliance/fixtures.py
+29
View File
@@ -1,4 +1,5 @@
from prowler.lib.check.compliance_models import (
AWS_Well_Architected_Requirement_Attribute,
CIS_Requirement_Attribute,
Compliance_Requirement,
ComplianceBaseModel,
@@ -322,3 +323,31 @@ NOT_PRESENT_COMPLIANCE = ComplianceBaseModel(
Description="NOT_EXISTENT",
Requirements=[],
)
AWS_WELL_ARCHITECTED_NAME = "aws_well_architected_framework_security_pillar_aws"
AWS_WELL_ARCHITECTED = ComplianceBaseModel(
Framework="AWS-Well-Architected-Framework-Security-Pillar",
Provider="AWS",
Version="",
Description="Best Practices for AWS Well-Architected Framework Security Pillar. The focus of this framework is the security pillar of the AWS Well-Architected Framework. It provides guidance to help you apply best practices, current recommendations in the design, delivery, and maintenance of secure AWS workloads.",
Requirements=[
Compliance_Requirement(
Id="SEC01-BP01",
Description="Establish common guardrails and isolation between environments (such as production, development, and test) and workloads through a multi-account strategy. Account-level separation is strongly recommended, as it provides a strong isolation boundary for security, billing, and access.",
Name=None,
Attributes=[
AWS_Well_Architected_Requirement_Attribute(
Name="SEC01-BP01 Separate workloads using accounts",
WellArchitectedQuestionId="securely-operate",
WellArchitectedPracticeId="sec_securely_operate_multi_accounts",
Section="Security foundations",
SubSection="AWS account management and separation",
LevelOfRisk="High",
AssessmentMethod="Automated",
Description="Establish common guardrails and isolation between environments (such as production, development, and test) and workloads through a multi-account strategy. Account-level separation is strongly recommended, as it provides a strong isolation boundary for security, billing, and access.",
ImplementationGuidanceUrl="https://docs.aws.amazon.com/wellarchitected/latest/security-pillar/sec_securely_operate_multi_accounts.html#implementation-guidance.",
)
],
Checks=["organizations_account_part_of_organizations"],
),
],
)
tests/lib/outputs/compliance/mitre_attack/mitre_attack_aws_test.py
+1 -1
View File
@@ -12,7 +12,7 @@ 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 TestAWSCIS:
class TestAWSMITREAttack:
def test_output_transform(self):
findings = [generate_finding_output(compliance={"MITRE-ATTACK": "T1190"})]
tests/lib/outputs/compliance/mitre_attack/mitre_attack_azure_test.py
+1 -1
View File
@@ -17,7 +17,7 @@ from tests.providers.azure.azure_fixtures import (
)
class TestAzureCIS:
class TestAzureMITREAttack:
def test_output_transform(self):
findings = [
generate_finding_output(
tests/lib/outputs/compliance/mitre_attack/mitre_attack_gcp_test.py
+1 -1
View File
@@ -12,7 +12,7 @@ from tests.lib.outputs.fixtures.fixtures import generate_finding_output
from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID
class TestGCPCIS:
class TestGCPMITREAttack:
def test_output_transform(self):
findings = [
generate_finding_output(