fix(sdk): align compliance CSV row emission with framework JSON (#11370)

prowler/CHANGELOG.md

@@ -9,6 +9,10 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - AWS AI Security Framework compliance for AWS provider [(#11353)](https://github.com/prowler-cloud/prowler/pull/11353)
 - `storage_account_public_network_access_disabled` check for Azure provider and remapped the Azure CIS "Public Network Access is Disabled" requirements to it [(#11334)](https://github.com/prowler-cloud/prowler/pull/11334)
 
+### 🐞 Fixed
+
+- Compliance CSV row count now matches the UI per requirement by sourcing rows from the framework JSON's `requirement.Checks` instead of the stale `finding.compliance` snapshot [(#11370)](https://github.com/prowler-cloud/prowler/pull/11370)
+
 ---
 
 ## [5.28.1] (Prowler 5.28.1)

prowler/lib/outputs/compliance/asd_essential_eight/asd_essential_eight_aws.py

@@ -37,9 +37,9 @@ class ASDEssentialEightAWS(ComplianceOutput):
             - None
         """
         for finding in findings:
-            finding_requirements = finding.compliance.get(compliance_name, [])
             for requirement in compliance.Requirements:
-                if requirement.Id in finding_requirements:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = ASDEssentialEightAWSModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/aws_well_architected/aws_well_architected.py

@@ -37,10 +37,9 @@ class AWSWellArchitected(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AWSWellArchitectedModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/c5/c5_aws.py

@@ -35,10 +35,9 @@ class AWSC5(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AWSC5Model(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/c5/c5_azure.py

@@ -35,10 +35,9 @@ class AzureC5(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AzureC5Model(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/c5/c5_gcp.py

@@ -35,10 +35,9 @@ class GCPC5(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = GCPC5Model(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/ccc/ccc_aws.py

@@ -35,10 +35,9 @@ class CCC_AWS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = CCC_AWSModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/ccc/ccc_azure.py

@@ -35,10 +35,9 @@ class CCC_Azure(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = CCC_AzureModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/ccc/ccc_gcp.py

@@ -35,10 +35,9 @@ class CCC_GCP(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = CCC_GCPModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/cis/cis_alibabacloud.py

@@ -35,10 +35,9 @@ class AlibabaCloudCIS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AlibabaCloudCISModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/cis/cis_aws.py

@@ -35,10 +35,9 @@ class AWSCIS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AWSCISModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/cis/cis_azure.py

@@ -35,10 +35,9 @@ class AzureCIS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AzureCISModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/cis/cis_gcp.py

@@ -35,10 +35,9 @@ class GCPCIS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = GCPCISModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/cis/cis_github.py

@@ -35,10 +35,9 @@ class GithubCIS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = GithubCISModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/cis/cis_googleworkspace.py

@@ -35,10 +35,9 @@ class GoogleWorkspaceCIS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = GoogleWorkspaceCISModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/cis/cis_kubernetes.py

@@ -35,10 +35,9 @@ class KubernetesCIS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = KubernetesCISModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/cis/cis_m365.py

@@ -35,10 +35,9 @@ class M365CIS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = M365CISModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/cis/cis_oraclecloud.py

@@ -35,10 +35,9 @@ class OracleCloudCIS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = OracleCloudCISModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/cisa_scuba/cisa_scuba_googleworkspace.py

@@ -37,10 +37,9 @@ class GoogleWorkspaceCISASCuBA(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = GoogleWorkspaceCISASCuBAModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/csa/csa_alibabacloud.py

@@ -35,10 +35,9 @@ class AlibabaCloudCSA(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AlibabaCloudCSAModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/csa/csa_aws.py

@@ -35,10 +35,9 @@ class AWSCSA(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AWSCSAModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/csa/csa_azure.py

@@ -35,10 +35,9 @@ class AzureCSA(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AzureCSAModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/csa/csa_gcp.py

@@ -35,10 +35,9 @@ class GCPCSA(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = GCPCSAModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/csa/csa_oraclecloud.py

@@ -35,10 +35,9 @@ class OracleCloudCSA(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = OracleCloudCSAModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/ens/ens_aws.py

@@ -35,10 +35,9 @@ class AWSENS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AWSENSModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/ens/ens_azure.py

@@ -35,10 +35,9 @@ class AzureENS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AzureENSModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/ens/ens_gcp.py

@@ -35,10 +35,9 @@ class GCPENS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = GCPENSModel(
                             Provider=finding.provider,

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

@@ -35,10 +35,9 @@ class GenericCompliance(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = GenericComplianceModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/iso27001/iso27001_aws.py

@@ -35,10 +35,9 @@ class AWSISO27001(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AWSISO27001Model(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/iso27001/iso27001_azure.py

@@ -35,10 +35,9 @@ class AzureISO27001(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AzureISO27001Model(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/iso27001/iso27001_gcp.py

@@ -35,10 +35,9 @@ class GCPISO27001(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = GCPISO27001Model(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/iso27001/iso27001_kubernetes.py

@@ -35,10 +35,9 @@ class KubernetesISO27001(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = KubernetesISO27001Model(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/iso27001/iso27001_m365.py

@@ -35,9 +35,9 @@ class M365ISO27001(ComplianceOutput):
             - None
         """
         for finding in findings:
-            finding_requirements = finding.compliance.get(compliance_name, [])
             for requirement in compliance.Requirements:
-                if requirement.Id in finding_requirements:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = M365ISO27001Model(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/iso27001/iso27001_nhn.py

@@ -35,9 +35,9 @@ class NHNISO27001(ComplianceOutput):
             - None
         """
         for finding in findings:
-            finding_requirements = finding.compliance.get(compliance_name, [])
             for requirement in compliance.Requirements:
-                if requirement.Id in finding_requirements:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = NHNISO27001Model(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/kisa_ismsp/kisa_ismsp_aws.py

@@ -35,10 +35,9 @@ class AWSKISAISMSP(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = AWSKISAISMSPModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/mitre_attack/mitre_attack_aws.py

@@ -36,10 +36,9 @@ class AWSMitreAttack(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     compliance_row = AWSMitreAttackModel(
                         Provider=finding.provider,
                         Description=compliance.Description,

prowler/lib/outputs/compliance/mitre_attack/mitre_attack_azure.py

@@ -36,10 +36,9 @@ class AzureMitreAttack(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     compliance_row = AzureMitreAttackModel(
                         Provider=finding.provider,
                         Description=compliance.Description,

prowler/lib/outputs/compliance/mitre_attack/mitre_attack_gcp.py

@@ -36,10 +36,9 @@ class GCPMitreAttack(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     compliance_row = GCPMitreAttackModel(
                         Provider=finding.provider,
                         Description=compliance.Description,

prowler/lib/outputs/compliance/prowler_threatscore/prowler_threatscore_alibaba.py

@@ -37,10 +37,9 @@ class ProwlerThreatScoreAlibaba(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = ProwlerThreatScoreAlibabaModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/prowler_threatscore/prowler_threatscore_aws.py

@@ -37,10 +37,9 @@ class ProwlerThreatScoreAWS(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = ProwlerThreatScoreAWSModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/prowler_threatscore/prowler_threatscore_azure.py

@@ -37,10 +37,9 @@ class ProwlerThreatScoreAzure(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = ProwlerThreatScoreAzureModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/prowler_threatscore/prowler_threatscore_gcp.py

@@ -37,10 +37,9 @@ class ProwlerThreatScoreGCP(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = ProwlerThreatScoreGCPModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/prowler_threatscore/prowler_threatscore_kubernetes.py

@@ -37,10 +37,9 @@ class ProwlerThreatScoreKubernetes(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = ProwlerThreatScoreKubernetesModel(
                             Provider=finding.provider,

prowler/lib/outputs/compliance/prowler_threatscore/prowler_threatscore_m365.py

@@ -37,10 +37,9 @@ class ProwlerThreatScoreM365(ComplianceOutput):
             - 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:
+                # Source of truth: framework JSON, not finding.compliance snapshot (avoids CSV/UI count drift).
+                if finding.check_id in requirement.Checks:
                     for attribute in requirement.Attributes:
                         compliance_row = ProwlerThreatScoreM365Model(
                             Provider=finding.provider,

tests/lib/outputs/compliance/fixtures.py

@@ -128,7 +128,7 @@ CIS_2_0_GCP = Compliance(
     Description="This CIS Benchmark is the product of a community consensus process and consists of secure configuration guidelines developed for Google Cloud Computing Platform",
     Requirements=[
         Compliance_Requirement(
-            Checks=["apikeys_key_exits"],
+            Checks=["apikeys_key_exits", "service_test_check_id"],
             Id="2.13",
             Description="Ensure That Microsoft Defender for Databases Is Set To 'On'",
             Attributes=[
@@ -177,7 +177,7 @@ CIS_1_8_KUBERNETES = Compliance(
     Description="This CIS Kubernetes Benchmark provides prescriptive guidance for establishing a secure configuration posture for Kubernetes v1.27.",
     Requirements=[
         Compliance_Requirement(
-            Checks=["apiserver_always_pull_images_plugin"],
+            Checks=["apiserver_always_pull_images_plugin", "service_test_check_id"],
             Id="1.1.3",
             Description="Ensure that the controller manager pod specification file permissions are set to 600 or more restrictive",
             Attributes=[
@@ -259,6 +259,7 @@ CIS_4_0_M365 = Compliance(
         Compliance_Requirement(
             Checks=[
                 "mfa_delete_enabled",
+                "service_test_check_id",
             ],
             Id="2.1.3",
             Description="Ensure MFA Delete is enabled on S3 buckets",
@@ -336,6 +337,7 @@ MITRE_ATTACK_AWS = Compliance(
                 "inspector2_active_findings_exist",
                 "awslambda_function_not_publicly_accessible",
                 "ec2_instance_public_ip",
+                "service_test_check_id",
             ],
         ),
         Mitre_Requirement(
@@ -412,6 +414,7 @@ MITRE_ATTACK_AZURE = Compliance(
                 "defender_ensure_notify_emails_to_owners",
                 "defender_ensure_system_updates_are_applied",
                 "defender_ensure_wdatp_is_enabled",
+                "service_test_check_id",
             ],
         ),
         Mitre_Requirement(
@@ -467,6 +470,7 @@ MITRE_ATTACK_GCP = Compliance(
                 "compute_instance_public_ip",
                 "compute_public_address_shodan",
                 "kms_key_not_publicly_accessible",
+                "service_test_check_id",
             ],
         ),
         Mitre_Requirement(
@@ -514,7 +518,7 @@ ENS_RD2022_AWS = Compliance(
                     Dependencias=[],
                 )
             ],
-            Checks=["cloudtrail_log_file_validation_enabled"],
+            Checks=["cloudtrail_log_file_validation_enabled", "service_test_check_id"],
         ),
         Compliance_Requirement(
             Id="op.exp.8.aws.ct.4",
@@ -562,7 +566,7 @@ ENS_RD2022_AZURE = Compliance(
                     Dependencias=[],
                 )
             ],
-            Checks=["cloudtrail_log_file_validation_enabled"],
+            Checks=["cloudtrail_log_file_validation_enabled", "service_test_check_id"],
         ),
         Compliance_Requirement(
             Id="op.exp.8.azure.ct.4",
@@ -609,7 +613,7 @@ ENS_RD2022_GCP = Compliance(
                     Dependencias=[],
                 )
             ],
-            Checks=["cloudtrail_log_file_validation_enabled"],
+            Checks=["cloudtrail_log_file_validation_enabled", "service_test_check_id"],
         ),
         Compliance_Requirement(
             Id="op.exp.8.gcp.ct.4",
@@ -666,7 +670,10 @@ AWS_WELL_ARCHITECTED = Compliance(
                     ImplementationGuidanceUrl="https://docs.aws.amazon.com/wellarchitected/latest/security-pillar/sec_securely_operate_multi_accounts.html#implementation-guidance.",
                 )
             ],
-            Checks=["organizations_account_part_of_organizations"],
+            Checks=[
+                "organizations_account_part_of_organizations",
+                "service_test_check_id",
+            ],
         ),
         Compliance_Requirement(
             Id="SEC01-BP02",
@@ -709,7 +716,7 @@ ISO27001_2013_AWS = Compliance(
                     Check_Summary="Setup Encryption at rest for RDS instances",
                 )
             ],
-            Checks=["rds_instance_storage_encrypted"],
+            Checks=["rds_instance_storage_encrypted", "service_test_check_id"],
         ),
         Compliance_Requirement(
             Id="A.10.2",
@@ -759,6 +766,7 @@ NIST_800_53_REVISION_4_AWS = Compliance(
                 "rds_instance_integration_cloudwatch_logs",
                 "redshift_cluster_audit_logging",
                 "securityhub_enabled",
+                "service_test_check_id",
             ],
         ),
         Compliance_Requirement(
@@ -815,6 +823,7 @@ KISA_ISMSP_AWS = Compliance(
             Checks=[
                 "cloudwatch_log_metric_filter_authentication_failures",
                 "cognito_user_pool_mfa_enabled",
+                "service_test_check_id",
             ],
         ),
         Compliance_Requirement(
@@ -872,6 +881,7 @@ PROWLER_THREATSCORE_AWS = Compliance(
             ],
             Checks=[
                 "iam_root_mfa_enabled",
+                "service_test_check_id",
             ],
         ),
         Compliance_Requirement(
@@ -916,6 +926,7 @@ PROWLER_THREATSCORE_AZURE = Compliance(
             ],
             Checks=[
                 "iam_root_mfa_enabled",
+                "service_test_check_id",
             ],
         ),
         Compliance_Requirement(
@@ -960,6 +971,7 @@ PROWLER_THREATSCORE_GCP = Compliance(
             ],
             Checks=[
                 "iam_root_mfa_enabled",
+                "service_test_check_id",
             ],
         ),
         Compliance_Requirement(
@@ -1004,6 +1016,7 @@ PROWLER_THREATSCORE_M365 = Compliance(
             ],
             Checks=[
                 "iam_root_mfa_enabled",
+                "service_test_check_id",
             ],
         ),
         Compliance_Requirement(

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

@@ -5,6 +5,11 @@ from unittest import mock
 from freezegun import freeze_time
 from mock import patch
 
+from prowler.lib.check.compliance_models import (
+    Compliance,
+    Compliance_Requirement,
+    Generic_Compliance_Requirement_Attribute,
+)
 from prowler.lib.outputs.compliance.generic.generic import GenericCompliance
 from prowler.lib.outputs.compliance.generic.models import GenericComplianceModel
 from tests.lib.outputs.compliance.fixtures import NIST_800_53_REVISION_4_AWS
@@ -129,3 +134,67 @@ class TestAWSGenericCompliance:
         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;FRAMEWORK;NAME;REQUIREMENTS_ATTRIBUTES_COMMENT\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;;;service_test_check_id;False;;NIST-800-53-Revision-4;National Institute of Standards and Technology (NIST) 800-53 Revision 4;\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.;;;{datetime.now()};ac_2_5;Account Management;Access Control (AC);Account Management (AC-2);;aws;;MANUAL;Manual check;manual_check;manual;False;Manual check;NIST-800-53-Revision-4;National Institute of Standards and Technology (NIST) 800-53 Revision 4;\r\n"
 
         assert content == expected_csv
+
+    def test_csv_row_count_matches_framework_checks_not_stored_compliance(self):
+        """Regression test for PROWLER-1763.
+
+        Ensures CSV emission is driven by the framework JSON's Requirements[].Checks
+        (the same source the UI uses) and not by the per-finding `finding.compliance`
+        snapshot stored at scan time. If `finding.check_id` is in a requirement's
+        Checks list, the row must be emitted regardless of what was stored in
+        `finding.compliance`. Conversely, a stale `finding.compliance` entry pointing
+        to a requirement whose Checks list no longer contains the finding's check_id
+        must not produce a row.
+        """
+        framework_name = "Test-Framework-1763"
+        compliance = Compliance(
+            Framework=framework_name,
+            Name=framework_name,
+            Provider="AWS",
+            Version="",
+            Description="Regression fixture for PROWLER-1763",
+            Requirements=[
+                Compliance_Requirement(
+                    Id="req_in_framework",
+                    Description="Requirement currently in framework",
+                    Attributes=[
+                        Generic_Compliance_Requirement_Attribute(
+                            Section="Section A", Service="aws"
+                        )
+                    ],
+                    Checks=["service_check_in_framework"],
+                ),
+                Compliance_Requirement(
+                    Id="req_no_longer_in_framework",
+                    Description="Requirement whose Checks list no longer includes the finding's check_id",
+                    Attributes=[
+                        Generic_Compliance_Requirement_Attribute(
+                            Section="Section B", Service="aws"
+                        )
+                    ],
+                    Checks=["service_different_check"],
+                ),
+            ],
+        )
+
+        # Snapshot drift case: finding.compliance maps to a requirement whose
+        # current Checks list no longer includes the finding's check_id, AND
+        # the finding belongs to a requirement that is NOT in the snapshot.
+        findings = [
+            generate_finding_output(
+                check_id="service_check_in_framework",
+                compliance={framework_name: ["req_no_longer_in_framework"]},
+            )
+        ]
+
+        output = GenericCompliance(findings, compliance)
+        rows = [
+            row
+            for row in output.data
+            if row.Status != "MANUAL" and row.ResourceName != "Manual check"
+        ]
+        assert (
+            len(rows) == 1
+        ), f"Expected 1 row driven by framework JSON, got {len(rows)}"
+        assert rows[0].Requirements_Id == "req_in_framework"
+        assert rows[0].CheckId == "service_check_in_framework"