chore(GenericCompliance): add Generic Compliance class (#4447)

Co-authored-by: Sergio <sergio@prowler.com>
Co-authored-by: Pepe Fagoaga <pepe@prowler.com>

prowler/__main__.py

@@ -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.generic.generic import GenericCompliance
 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 (
@@ -435,6 +436,18 @@ def prowler():
                     file_path=filename,
                 )
                 iso27001_finding.batch_write_data_to_file()
+            else:
+                filename = (
+                    f"{global_provider.output_options.output_directory}/compliance/"
+                    f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
+                )
+                generic_compliance = GenericCompliance(
+                    findings=finding_outputs,
+                    compliance=bulk_compliance_frameworks[compliance_name],
+                    create_file_descriptor=True,
+                    file_path=filename,
+                )
+                generic_compliance.batch_write_data_to_file()
 
     elif provider == "azure":
         for compliance_name in input_compliance_frameworks:
@@ -464,6 +477,18 @@ def prowler():
                     file_path=filename,
                 )
                 mitre_attack_finding.batch_write_data_to_file()
+            else:
+                filename = (
+                    f"{global_provider.output_options.output_directory}/compliance/"
+                    f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
+                )
+                generic_compliance = GenericCompliance(
+                    findings=finding_outputs,
+                    compliance=bulk_compliance_frameworks[compliance_name],
+                    create_file_descriptor=True,
+                    file_path=filename,
+                )
+                generic_compliance.batch_write_data_to_file()
 
     elif provider == "gcp":
         for compliance_name in input_compliance_frameworks:
@@ -493,6 +518,18 @@ def prowler():
                     file_path=filename,
                 )
                 mitre_attack_finding.batch_write_data_to_file()
+            else:
+                filename = (
+                    f"{global_provider.output_options.output_directory}/compliance/"
+                    f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
+                )
+                generic_compliance = GenericCompliance(
+                    findings=finding_outputs,
+                    compliance=bulk_compliance_frameworks[compliance_name],
+                    create_file_descriptor=True,
+                    file_path=filename,
+                )
+                generic_compliance.batch_write_data_to_file()
 
     elif provider == "kubernetes":
         for compliance_name in input_compliance_frameworks:
@@ -509,6 +546,18 @@ def prowler():
                     file_path=filename,
                 )
                 cis_finding.batch_write_data_to_file()
+            else:
+                filename = (
+                    f"{global_provider.output_options.output_directory}/compliance/"
+                    f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
+                )
+                generic_compliance = GenericCompliance(
+                    findings=finding_outputs,
+                    compliance=bulk_compliance_frameworks[compliance_name],
+                    create_file_descriptor=True,
+                    file_path=filename,
+                )
+                generic_compliance.batch_write_data_to_file()
 
     # AWS Security Hub Integration
     if provider == "aws" and args.security_hub:

prowler/lib/outputs/compliance/compliance.py

@@ -4,9 +4,8 @@ from prowler.lib.check.models import Check_Report
 from prowler.lib.logger import logger
 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 (
+from prowler.lib.outputs.compliance.generic.generic_table import (
     get_generic_compliance_table,
-    write_compliance_row_generic,
 )
 from prowler.lib.outputs.compliance.mitre_attack.mitre_attack import (
     get_mitre_attack_table,
@@ -103,11 +102,6 @@ def fill_compliance(
             ):
                 continue
 
-            else:
-                write_compliance_row_generic(
-                    file_descriptors, finding, compliance, output_options, provider
-                )
-
     except Exception as error:
         logger.error(
             f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"

prowler/lib/outputs/compliance/generic.py

@@ -1,105 +0,0 @@
-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_Generic_Compliance
-from prowler.lib.outputs.csv.csv import generate_csv_fields
-from prowler.lib.utils.utils import outputs_unix_timestamp
-
-
-def write_compliance_row_generic(
-    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_Generic_Compliance)
-    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_Generic_Compliance(
-                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_Section=attribute.Section,
-                Requirements_Attributes_SubSection=attribute.SubSection,
-                Requirements_Attributes_SubGroup=attribute.SubGroup,
-                Requirements_Attributes_Service=attribute.Service,
-                Requirements_Attributes_Type=attribute.Type,
-                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_generic_compliance_table(
-    findings: list,
-    bulk_checks_metadata: dict,
-    compliance_framework: str,
-    output_filename: str,
-    output_directory: str,
-    compliance_overview: bool,
-):
-    pass_count = []
-    fail_count = []
-    muted_count = []
-    for index, finding in enumerate(findings):
-        check = bulk_checks_metadata[finding.check_metadata.CheckID]
-        check_compliances = check.Compliance
-        for compliance in check_compliances:
-            if (
-                compliance.Framework.upper()
-                in compliance_framework.upper().replace("_", "-")
-                and compliance.Version in compliance_framework.upper()
-                and compliance.Provider in compliance_framework.upper()
-            ):
-                for requirement in compliance.Requirements:
-                    for attribute in requirement.Attributes:
-                        if finding.muted:
-                            if index not in muted_count:
-                                muted_count.append(index)
-                        else:
-                            if finding.status == "FAIL" and index not in fail_count:
-                                fail_count.append(index)
-                            elif finding.status == "PASS" and index not in pass_count:
-                                pass_count.append(index)
-    if (
-        len(fail_count) + len(pass_count) + len(muted_count) > 1
-    ):  # If there are no resources, don't print the compliance table
-        print(
-            f"\nCompliance Status of {Fore.YELLOW}{compliance_framework.upper()}{Style.RESET_ALL} Framework:"
-        )
-        overview_table = [
-            [
-                f"{Fore.RED}{round(len(fail_count) / len(findings) * 100, 2)}% ({len(fail_count)}) FAIL{Style.RESET_ALL}",
-                f"{Fore.GREEN}{round(len(pass_count) / len(findings) * 100, 2)}% ({len(pass_count)}) PASS{Style.RESET_ALL}",
-                f"{orange_color}{round(len(muted_count) / len(findings) * 100, 2)}% ({len(muted_count)}) MUTED{Style.RESET_ALL}",
-            ]
-        ]
-        print(tabulate(overview_table, tablefmt="rounded_grid"))
-        if not compliance_overview:
-            print(f"\nDetailed results of {compliance_framework.upper()} are in:")
-            print(
-                f" - CSV: {output_directory}/compliance/{output_filename}_{compliance_framework}.csv\n"
-            )

prowler/lib/outputs/compliance/generic/generic.py

@@ -0,0 +1,98 @@
+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 Generic
+from prowler.lib.outputs.finding import Finding
+
+
+class GenericCompliance(ComplianceOutput):
+    """
+    This class represents the Generic 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 Generic compliance format.
+        - batch_write_data_to_file: Writes the findings data to a CSV file in Generic compliance format.
+    """
+
+    def transform(
+        self,
+        findings: list[Finding],
+        compliance: ComplianceBaseModel,
+        compliance_name: str,
+    ) -> None:
+        """
+        Transforms a list of findings into Generic 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 = Generic(
+                            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_Section=attribute.Section,
+                            Requirements_Attributes_SubSection=attribute.SubSection,
+                            Requirements_Attributes_SubGroup=attribute.SubGroup,
+                            Requirements_Attributes_Service=attribute.Service,
+                            Requirements_Attributes_Type=attribute.Type,
+                            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 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/generic/generic_table.py

@@ -0,0 +1,54 @@
+from colorama import Fore, Style
+from tabulate import tabulate
+
+from prowler.config.config import orange_color
+
+
+def get_generic_compliance_table(
+    findings: list,
+    bulk_checks_metadata: dict,
+    compliance_framework: str,
+    output_filename: str,
+    output_directory: str,
+    compliance_overview: bool,
+):
+    pass_count = []
+    fail_count = []
+    muted_count = []
+    for index, finding in enumerate(findings):
+        check = bulk_checks_metadata[finding.check_metadata.CheckID]
+        check_compliances = check.Compliance
+        for compliance in check_compliances:
+            if (
+                compliance.Framework.upper()
+                in compliance_framework.upper().replace("_", "-")
+                and compliance.Version in compliance_framework.upper()
+                and compliance.Provider in compliance_framework.upper()
+            ):
+                if finding.muted:
+                    if index not in muted_count:
+                        muted_count.append(index)
+                else:
+                    if finding.status == "FAIL" and index not in fail_count:
+                        fail_count.append(index)
+                    elif finding.status == "PASS" and index not in pass_count:
+                        pass_count.append(index)
+    if (
+        len(fail_count) + len(pass_count) + len(muted_count) > 1
+    ):  # If there are no resources, don't print the compliance table
+        print(
+            f"\nCompliance Status of {Fore.YELLOW}{compliance_framework.upper()}{Style.RESET_ALL} Framework:"
+        )
+        overview_table = [
+            [
+                f"{Fore.RED}{round(len(fail_count) / len(findings) * 100, 2)}% ({len(fail_count)}) FAIL{Style.RESET_ALL}",
+                f"{Fore.GREEN}{round(len(pass_count) / len(findings) * 100, 2)}% ({len(pass_count)}) PASS{Style.RESET_ALL}",
+                f"{orange_color}{round(len(muted_count) / len(findings) * 100, 2)}% ({len(muted_count)}) MUTED{Style.RESET_ALL}",
+            ]
+        ]
+        print(tabulate(overview_table, tablefmt="rounded_grid"))
+        if not compliance_overview:
+            print(f"\nDetailed results of {compliance_framework.upper()} are in:")
+            print(
+                f" - CSV: {output_directory}/compliance/{output_filename}_{compliance_framework}.csv\n"
+            )

prowler/lib/outputs/compliance/generic/models.py

@@ -2,12 +2,10 @@ from typing import Optional
 
 from pydantic import BaseModel
 
-# TODO: move this to outputs/<compliance>/models.py
 
-
-class Check_Output_CSV_Generic_Compliance(BaseModel):
+class Generic(BaseModel):
     """
-    Check_Output_CSV_Generic_Compliance generates a finding's output in CSV Generic Compliance format.
+    Generic generates a finding's output in Generic Compliance format.
     """
 
     Provider: str
@@ -27,3 +25,4 @@ class Check_Output_CSV_Generic_Compliance(BaseModel):
     ResourceId: str
     CheckId: str
     Muted: bool
+    ResourceName: str

prowler/lib/outputs/file_descriptors.py

@@ -1,107 +0,0 @@
-from csv import DictWriter
-from io import TextIOWrapper
-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_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
-
-
-def initialize_file_descriptor(
-    filename: str,
-    format: Any = Finding,
-) -> TextIOWrapper:
-    """Open/Create the output file. If needed include headers or the required format, by default will use the FindingOutput"""
-    try:
-        if file_exists(filename):
-            file_descriptor = open_file(
-                filename,
-                "a",
-            )
-        else:
-            file_descriptor = open_file(
-                filename,
-                "a",
-            )
-            # Format is the class model of the CSV format to print the headers
-            csv_header = [x.upper() for x in generate_csv_fields(format)]
-            csv_writer = DictWriter(
-                file_descriptor, fieldnames=csv_header, delimiter=";"
-            )
-            csv_writer.writeheader()
-        return file_descriptor
-    except Exception as error:
-        logger.error(
-            f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
-        )
-
-
-def fill_file_descriptors(output_modes, output_directory, output_filename, provider):
-    try:
-        file_descriptors = {}
-        if output_modes:
-            for output_mode in output_modes:
-                # FIXME: Remove this once we always use the new CSV(Output)
-                if output_mode == "csv":
-                    continue
-                elif output_mode == "json-ocsf":
-                    continue
-                elif output_mode == "json-asff":
-                    continue
-                elif output_mode == "html":
-                    continue
-                # FIXME: Remove this once we merge all the compliance frameworks
-                if "cis_" in output_mode:
-                    continue
-                elif "mitre_attack_" in output_mode:
-                    continue
-                elif "ens_" in output_mode:
-                    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}"
-                    file_descriptor = initialize_file_descriptor(
-                        filename,
-                        Check_Output_CSV_Generic_Compliance,
-                    )
-                    file_descriptors.update({output_mode: file_descriptor})
-
-                elif provider.type == "kubernetes":
-                    filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}"
-                    file_descriptor = initialize_file_descriptor(
-                        filename,
-                        Check_Output_CSV_Generic_Compliance,
-                    )
-                    file_descriptors.update({output_mode: file_descriptor})
-
-                elif provider.type == "azure":
-                    filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}"
-                    file_descriptor = initialize_file_descriptor(
-                        filename,
-                        Check_Output_CSV_Generic_Compliance,
-                    )
-                    file_descriptors.update({output_mode: file_descriptor})
-
-                elif provider.type == "aws":
-                    # Compliance frameworks
-                    filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}"
-                    # 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(
-            f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
-        )
-
-    return file_descriptors

prowler/lib/outputs/outputs.py

@@ -6,7 +6,6 @@ from prowler.lib.outputs.compliance.compliance import (
     add_manual_controls,
     fill_compliance,
 )
-from prowler.lib.outputs.file_descriptors import fill_file_descriptors
 
 
 def stdout_report(finding, color, verbose, status, fix):
@@ -42,16 +41,6 @@ def report(check_findings, provider):
             if provider.type == "azure":
                 check_findings.sort(key=lambda x: x.subscription)
 
-            # Generate the required output files
-            if output_options.output_modes and not output_options.fixer:
-                # We have to create the required output files
-                file_descriptors = fill_file_descriptors(
-                    output_options.output_modes,
-                    output_options.output_directory,
-                    output_options.output_filename,
-                    provider,
-                )
-
             for finding in check_findings:
                 # Print findings by stdout
                 color = set_report_color(finding.status, finding.muted)

tests/lib/outputs/compliance/fixtures.py

@@ -6,6 +6,7 @@ from prowler.lib.check.compliance_models import (
     ENS_Requirement_Attribute,
     ENS_Requirement_Attribute_Nivel,
     ENS_Requirement_Attribute_Tipos,
+    Generic_Compliance_Requirement_Attribute,
     ISO27001_2013_Requirement_Attribute,
     Mitre_Requirement,
     Mitre_Requirement_Attribute_AWS,
@@ -375,3 +376,61 @@ ISO27001_2013_AWS = ComplianceBaseModel(
         ),
     ],
 )
+ISO27001_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"],
+        ),
+    ],
+)
+NIST_800_53_REVISION_4_AWS_NAME = "nist_800_53_revision_4_aws"
+NIST_800_53_REVISION_4_AWS = ComplianceBaseModel(
+    Framework="NIST-800-53-Revision-4",
+    Provider="AWS",
+    Version="",
+    Description="NIST 800-53 is a regulatory standard that defines the minimum baseline of security controls for all U.S. federal information systems except those related to national security. The controls defined in this standard are customizable and address a diverse set of security and privacy requirements.",
+    Requirements=[
+        Compliance_Requirement(
+            Id="ac_2_4",
+            Description="Account Management",
+            Name="The information system automatically audits account creation, modification, enabling, disabling, and removal actions, and notifies [Assignment: organization-defined personnel or roles].",
+            Attributes=[
+                Generic_Compliance_Requirement_Attribute(
+                    ItemId="ac_2_4",
+                    Section="Access Control (AC)",
+                    SubSection="Account Management (AC-2)",
+                    Service="aws",
+                )
+            ],
+            Checks=[
+                "cloudtrail_multi_region_enabled",
+                "cloudtrail_multi_region_enabled",
+                "cloudtrail_cloudwatch_logging_enabled",
+                "cloudwatch_changes_to_network_acls_alarm_configured",
+                "cloudwatch_changes_to_network_gateways_alarm_configured",
+                "cloudwatch_changes_to_network_route_tables_alarm_configured",
+                "cloudwatch_changes_to_vpcs_alarm_configured",
+                "guardduty_is_enabled",
+                "rds_instance_integration_cloudwatch_logs",
+                "redshift_cluster_audit_logging",
+                "securityhub_enabled",
+            ],
+        )
+    ],
+)

tests/lib/outputs/compliance/generic/generic_aws_test.py

@@ -0,0 +1,76 @@
+from datetime import datetime
+from io import StringIO
+
+from freezegun import freeze_time
+from mock import patch
+
+from prowler.lib.outputs.compliance.generic.generic import GenericCompliance
+from prowler.lib.outputs.compliance.generic.models import Generic
+from tests.lib.outputs.compliance.fixtures import NIST_800_53_REVISION_4_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 TestAWSGenericCompliance:
+    def test_output_transform(self):
+        findings = [
+            generate_finding_output(compliance={"NIST-800-53-Revision-4": "ac_2_4"})
+        ]
+
+        output = GenericCompliance(findings, NIST_800_53_REVISION_4_AWS)
+        output_data = output.data[0]
+        assert isinstance(output_data, Generic)
+        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 == NIST_800_53_REVISION_4_AWS.Description
+        assert (
+            output_data.Requirements_Id == NIST_800_53_REVISION_4_AWS.Requirements[0].Id
+        )
+        assert (
+            output_data.Requirements_Description
+            == NIST_800_53_REVISION_4_AWS.Requirements[0].Description
+        )
+        assert (
+            output_data.Requirements_Attributes_Section
+            == NIST_800_53_REVISION_4_AWS.Requirements[0].Attributes[0].Section
+        )
+        assert (
+            output_data.Requirements_Attributes_SubSection
+            == NIST_800_53_REVISION_4_AWS.Requirements[0].Attributes[0].SubSection
+        )
+        assert (
+            output_data.Requirements_Attributes_SubGroup
+            == NIST_800_53_REVISION_4_AWS.Requirements[0].Attributes[0].SubGroup
+        )
+        assert (
+            output_data.Requirements_Attributes_Service
+            == NIST_800_53_REVISION_4_AWS.Requirements[0].Attributes[0].Service
+        )
+        assert (
+            output_data.Requirements_Attributes_Type
+            == NIST_800_53_REVISION_4_AWS.Requirements[0].Attributes[0].Type
+        )
+        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={"NIST-800-53-Revision-4": "ac_2_4"})
+        ]
+        output = GenericCompliance(findings, NIST_800_53_REVISION_4_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_SECTION;REQUIREMENTS_ATTRIBUTES_SUBSECTION;REQUIREMENTS_ATTRIBUTES_SUBGROUP;REQUIREMENTS_ATTRIBUTES_SERVICE;REQUIREMENTS_ATTRIBUTES_TYPE;STATUS;STATUSEXTENDED;RESOURCEID;CHECKID;MUTED;RESOURCENAME\r\naws;NIST 800-53 is a regulatory standard that defines the minimum baseline of security controls for all U.S. federal information systems except those related to national security. The controls defined in this standard are customizable and address a diverse set of security and privacy requirements.;123456789012;eu-west-1;{datetime.now()};ac_2_4;Account Management;Access Control (AC);Account Management (AC-2);;aws;;PASS;;;test-check-id;False;\r\n"""
+        assert content == expected_csv