From 44afe2db3e16adcafa31dd705aeb5f5f6142032e Mon Sep 17 00:00:00 2001 From: Sergio Garcia <38561120+sergargar@users.noreply.github.com> Date: Thu, 18 Jul 2024 02:36:57 -0400 Subject: [PATCH] chore(compliance): simplify ComplianceOutput class (#4467) --- .../aws_well_architected.py | 33 ----------------- prowler/lib/outputs/compliance/cis/cis_aws.py | 33 ----------------- .../lib/outputs/compliance/cis/cis_azure.py | 27 -------------- prowler/lib/outputs/compliance/cis/cis_gcp.py | 27 -------------- .../outputs/compliance/cis/cis_kubernetes.py | 26 -------------- .../outputs/compliance/compliance_output.py | 31 ++++++++++++++++ prowler/lib/outputs/compliance/ens/ens_aws.py | 33 ----------------- .../lib/outputs/compliance/generic/generic.py | 36 ------------------- .../compliance/iso27001/iso27001_aws.py | 33 ----------------- .../mitre_attack/mitre_attack_aws.py | 33 ----------------- .../mitre_attack/mitre_attack_azure.py | 33 ----------------- .../mitre_attack/mitre_attack_gcp.py | 33 ----------------- 12 files changed, 31 insertions(+), 347 deletions(-) diff --git a/prowler/lib/outputs/compliance/aws_well_architected/aws_well_architected.py b/prowler/lib/outputs/compliance/aws_well_architected/aws_well_architected.py index 134d5440ec..ebd2f144f2 100644 --- a/prowler/lib/outputs/compliance/aws_well_architected/aws_well_architected.py +++ b/prowler/lib/outputs/compliance/aws_well_architected/aws_well_architected.py @@ -1,6 +1,3 @@ -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 ( AWSWellArchitectedModel, @@ -19,7 +16,6 @@ class AWSWellArchitected(ComplianceOutput): 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( @@ -99,32 +95,3 @@ class AWSWellArchitected(ComplianceOutput): Muted=False, ) 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}" - ) diff --git a/prowler/lib/outputs/compliance/cis/cis_aws.py b/prowler/lib/outputs/compliance/cis/cis_aws.py index b80e7267a1..0ce3d70650 100644 --- a/prowler/lib/outputs/compliance/cis/cis_aws.py +++ b/prowler/lib/outputs/compliance/cis/cis_aws.py @@ -1,6 +1,3 @@ -from csv import DictWriter -from venv import logger - from prowler.lib.check.compliance_models import ComplianceBaseModel from prowler.lib.outputs.compliance.cis.models import AWSCISModel from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput @@ -17,7 +14,6 @@ class AWSCIS(ComplianceOutput): Methods: - transform: Transforms findings into AWS CIS compliance format. - - batch_write_data_to_file: Writes the findings data to a CSV file in AWS CIS compliance format. """ def transform( @@ -99,32 +95,3 @@ class AWSCIS(ComplianceOutput): Muted=False, ) self._data.append(compliance_row) - - def batch_write_data_to_file(self) -> None: - """ - Writes the findings data to a CSV file in AWS CIS 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/cis/cis_azure.py b/prowler/lib/outputs/compliance/cis/cis_azure.py index afe518bb72..eebcf365d5 100644 --- a/prowler/lib/outputs/compliance/cis/cis_azure.py +++ b/prowler/lib/outputs/compliance/cis/cis_azure.py @@ -1,6 +1,3 @@ -from csv import DictWriter -from venv import logger - from prowler.lib.check.compliance_models import ComplianceBaseModel from prowler.lib.outputs.compliance.cis.models import AzureCISModel from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput @@ -17,7 +14,6 @@ class AzureCIS(ComplianceOutput): Methods: - transform: Transforms findings into Azure CIS compliance format. - - batch_write_data_to_file: Writes the findings data to a CSV file in Azure CIS compliance format. """ def transform( @@ -101,26 +97,3 @@ class AzureCIS(ComplianceOutput): Muted=False, ) self._data.append(compliance_row) - - def batch_write_data_to_file(self) -> 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/cis/cis_gcp.py b/prowler/lib/outputs/compliance/cis/cis_gcp.py index d69be18420..3488e6b4ed 100644 --- a/prowler/lib/outputs/compliance/cis/cis_gcp.py +++ b/prowler/lib/outputs/compliance/cis/cis_gcp.py @@ -1,6 +1,3 @@ -from csv import DictWriter -from venv import logger - from prowler.lib.check.compliance_models import ComplianceBaseModel from prowler.lib.outputs.compliance.cis.models import GCPCISModel from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput @@ -17,7 +14,6 @@ class GCPCIS(ComplianceOutput): Methods: - transform: Transforms findings into GCP CIS compliance format. - - batch_write_data_to_file: Writes the findings data to a CSV file in GCP CIS compliance format. """ def transform( @@ -99,26 +95,3 @@ class GCPCIS(ComplianceOutput): Muted=False, ) self._data.append(compliance_row) - - def batch_write_data_to_file(self) -> 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/cis/cis_kubernetes.py b/prowler/lib/outputs/compliance/cis/cis_kubernetes.py index 67b957d135..ac19840d42 100644 --- a/prowler/lib/outputs/compliance/cis/cis_kubernetes.py +++ b/prowler/lib/outputs/compliance/cis/cis_kubernetes.py @@ -1,6 +1,4 @@ -from csv import DictWriter from datetime import datetime -from venv import logger from prowler.lib.check.compliance_models import ComplianceBaseModel from prowler.lib.outputs.compliance.cis.models import KubernetesCISModel @@ -18,7 +16,6 @@ class KubernetesCIS(ComplianceOutput): Methods: - transform: Transforms findings into Kubernetes CIS compliance format. - - batch_write_data_to_file: Writes the findings data to a CSV file in Kubernetes CIS compliance format. """ def transform( @@ -102,26 +99,3 @@ class KubernetesCIS(ComplianceOutput): Muted=False, ) self._data.append(compliance_row) - - def batch_write_data_to_file(self) -> 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/compliance_output.py b/prowler/lib/outputs/compliance/compliance_output.py index 338309b166..bea5e8948d 100644 --- a/prowler/lib/outputs/compliance/compliance_output.py +++ b/prowler/lib/outputs/compliance/compliance_output.py @@ -1,7 +1,9 @@ +from csv import DictWriter from io import TextIOWrapper from typing import List from prowler.lib.check.compliance_models import ComplianceBaseModel +from prowler.lib.logger import logger from prowler.lib.outputs.finding import Finding from prowler.lib.outputs.output import Output @@ -44,3 +46,32 @@ class ComplianceOutput(Output): self.transform(findings, compliance, compliance_name) if create_file_descriptor: self.create_file_descriptor(file_path) + + def batch_write_data_to_file(self) -> None: + """ + Writes the findings data to a CSV file in the specific 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/ens_aws.py b/prowler/lib/outputs/compliance/ens/ens_aws.py index da8a3ad684..7e63805e38 100644 --- a/prowler/lib/outputs/compliance/ens/ens_aws.py +++ b/prowler/lib/outputs/compliance/ens/ens_aws.py @@ -1,6 +1,3 @@ -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 AWSENSModel @@ -17,7 +14,6 @@ class AWSENS(ComplianceOutput): 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( @@ -105,32 +101,3 @@ class AWSENS(ComplianceOutput): Muted=False, ) 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/generic/generic.py b/prowler/lib/outputs/compliance/generic/generic.py index 92bd4a7fb5..47d9494b99 100644 --- a/prowler/lib/outputs/compliance/generic/generic.py +++ b/prowler/lib/outputs/compliance/generic/generic.py @@ -1,6 +1,3 @@ -from csv import DictWriter -from venv import logger - from prowler.lib.check.compliance_models import ComplianceBaseModel from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput from prowler.lib.outputs.compliance.generic.models import GenericComplianceModel @@ -17,7 +14,6 @@ class GenericCompliance(ComplianceOutput): Methods: - transform: Transforms findings into Generic compliance format. - - batch_write_data_to_file: Writes the findings data to a CSV file in Generic compliance format. """ def transform( @@ -89,35 +85,3 @@ class GenericCompliance(ComplianceOutput): Muted=False, ) self._data.append(compliance_row) - - def batch_write_data_to_file(self) -> None: - """ - Writes the findings data to a CSV file in Generic compliance format. - - Parameters: - - file_descriptor (Dict[str, DictWriter]): A dictionary of file descriptors. - - 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/iso27001_aws.py b/prowler/lib/outputs/compliance/iso27001/iso27001_aws.py index ce2724de8f..cc732ebd57 100644 --- a/prowler/lib/outputs/compliance/iso27001/iso27001_aws.py +++ b/prowler/lib/outputs/compliance/iso27001/iso27001_aws.py @@ -1,6 +1,3 @@ -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 AWSISO27001Model @@ -17,7 +14,6 @@ class AWSISO27001(ComplianceOutput): 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( @@ -83,32 +79,3 @@ class AWSISO27001(ComplianceOutput): Muted=False, ) 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/mitre_attack/mitre_attack_aws.py b/prowler/lib/outputs/compliance/mitre_attack/mitre_attack_aws.py index 3f4a9c23c3..0241a0cba9 100644 --- a/prowler/lib/outputs/compliance/mitre_attack/mitre_attack_aws.py +++ b/prowler/lib/outputs/compliance/mitre_attack/mitre_attack_aws.py @@ -1,6 +1,3 @@ -from csv import DictWriter -from venv import logger - from prowler.lib.check.compliance_models import ComplianceBaseModel from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput from prowler.lib.outputs.compliance.mitre_attack.models import AWSMitreAttackModel @@ -18,7 +15,6 @@ class AWSMitreAttack(ComplianceOutput): Methods: - transform: Transforms findings into AWS MITRE ATT&CK compliance format. - - batch_write_data_to_file: Writes the findings data to a CSV file in AWS MITRE ATT&CK compliance format. """ def transform( @@ -117,32 +113,3 @@ class AWSMitreAttack(ComplianceOutput): Muted=False, ) self._data.append(compliance_row) - - def batch_write_data_to_file(self) -> None: - """ - Writes the findings data to a CSV file in AWS MITRE ATT&CK 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/mitre_attack/mitre_attack_azure.py b/prowler/lib/outputs/compliance/mitre_attack/mitre_attack_azure.py index 949437ebc1..fb10bb650b 100644 --- a/prowler/lib/outputs/compliance/mitre_attack/mitre_attack_azure.py +++ b/prowler/lib/outputs/compliance/mitre_attack/mitre_attack_azure.py @@ -1,6 +1,3 @@ -from csv import DictWriter -from venv import logger - from prowler.lib.check.compliance_models import ComplianceBaseModel from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput from prowler.lib.outputs.compliance.mitre_attack.models import AzureMitreAttackModel @@ -18,7 +15,6 @@ class AzureMitreAttack(ComplianceOutput): Methods: - transform: Transforms findings into Azure MITRE ATT&CK compliance format. - - batch_write_data_to_file: Writes the findings data to a CSV file in Azure MITRE ATT&CK compliance format. """ def transform( @@ -119,32 +115,3 @@ class AzureMitreAttack(ComplianceOutput): Muted=False, ) self._data.append(compliance_row) - - def batch_write_data_to_file(self) -> None: - """ - Writes the findings data to a CSV file in Azure MITRE ATT&CK 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/mitre_attack/mitre_attack_gcp.py b/prowler/lib/outputs/compliance/mitre_attack/mitre_attack_gcp.py index f99004059a..d3ec4fd2b5 100644 --- a/prowler/lib/outputs/compliance/mitre_attack/mitre_attack_gcp.py +++ b/prowler/lib/outputs/compliance/mitre_attack/mitre_attack_gcp.py @@ -1,6 +1,3 @@ -from csv import DictWriter -from venv import logger - from prowler.lib.check.compliance_models import ComplianceBaseModel from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput from prowler.lib.outputs.compliance.mitre_attack.models import GCPMitreAttackModel @@ -18,7 +15,6 @@ class GCPMitreAttack(ComplianceOutput): Methods: - transform: Transforms findings into GCP MITRE ATT&CK compliance format. - - batch_write_data_to_file: Writes the findings data to a CSV file in GCP MITRE ATT&CK compliance format. """ def transform( @@ -117,32 +113,3 @@ class GCPMitreAttack(ComplianceOutput): Muted=False, ) self._data.append(compliance_row) - - def batch_write_data_to_file(self) -> None: - """ - Writes the findings data to a CSV file in GCP MITRE ATT&CK 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}" - )