chore(compliance): simplify ComplianceOutput class (#4467)

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}"
-            )

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}"
-            )

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}"
-            )

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}"
-            )

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}"
-            )

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}"
+            )

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}"
-            )

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}"
-            )

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}"
-            )

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}"
-            )

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}"
-            )

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}"
-            )