fix(outputs): address OSCAL review — unique result UUID + real Finding model

Use uuid4() for result UUID so each export is unique, type Finding and
read fields directly (OCSF-style), skip muted FAILs as findings, omit
empty related-observations, and normalize naive timestamps for schema
timezone requirements. Tests now use generate_finding_output.

Signed-off-by: Ahmed Hassan <th3reality72@gmail.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: AAH20

prowler/lib/outputs/oscal/models.py

@@ -109,8 +109,12 @@ def to_dict(self) -> dict[str, Any]:
                 "target-id": self.target_id,
                 "status": target_status,
             },
-            "related-observations": [ro.to_dict() for ro in self.related_observations],
         }
+        # related-observations has minItems: 1 when present — omit if empty.
+        if self.related_observations:
+            res["related-observations"] = [
+                ro.to_dict() for ro in self.related_observations
+            ]
         if self.props:
             res["props"] = [p.to_dict() for p in self.props]
         if self.remarks:

prowler/lib/outputs/oscal/oscal.py

@@ -1,8 +1,10 @@
 import json
 import uuid
 from datetime import datetime, timezone
-from typing import Any, List, Optional
+from typing import List, Optional
 
+from prowler.lib.outputs.common import Status
+from prowler.lib.outputs.finding import Finding
 from prowler.lib.outputs.oscal.models import (
     DEFAULT_IMPORT_AP_HREF,
     AssessmentResults,
@@ -18,20 +20,18 @@
 
 
 class OSCAL:
-    """
-    Transforms Prowler findings into NIST OSCAL 1.2.3 Assessment Results JSON format.
-    """
+    """Transforms Prowler findings into NIST OSCAL 1.2.3 Assessment Results JSON."""
 
     def __init__(
         self,
-        findings: List[Any],
+        findings: List[Finding],
         file_path: Optional[str] = None,
         import_ap_href: str = DEFAULT_IMPORT_AP_HREF,
     ) -> None:
         """Build and hold the OSCAL document(s) for the given findings.
 
         Args:
-            findings: Prowler finding objects to transform.
+            findings: Prowler ``Finding`` objects to transform.
             file_path: Optional output path (used by callers that write
                 straight to disk rather than via ``batch_write_data_to_file``).
             import_ap_href: The assessment-plan reference this run is
@@ -49,23 +49,21 @@ def data(self) -> List[OscalDocument]:
         """The transformed OSCAL document(s), one per ``transform()`` call."""
         return self._data
 
-    def transform(self, findings: List[Any]) -> None:
-        """
-        Transforms a list of Prowler findings into an OSCAL AssessmentResults document.
+    def transform(self, findings: List[Finding]) -> None:
+        """Transform Prowler findings into an OSCAL AssessmentResults document.
+
+        Args:
+            findings: Non-empty list of ``Finding`` instances for one export.
         """
         if not findings:
             return
 
-        now_iso = datetime.now(timezone.utc).isoformat()
+        now = datetime.now(timezone.utc)
+        now_iso = now.isoformat()
         first_finding = findings[0]
-        account_uid = getattr(first_finding, "account_uid", "default-account")
-        provider_name = (
-            first_finding.metadata.Provider
-            if hasattr(first_finding, "metadata")
-            and hasattr(first_finding.metadata, "Provider")
-            else "aws"
-        )
-        prowler_version = getattr(first_finding, "prowler_version", "4.0.0")
+        account_uid = first_finding.account_uid
+        provider_name = first_finding.metadata.Provider
+        prowler_version = first_finding.prowler_version
 
         metadata = Metadata(
             title=f"Prowler Security Assessment — {provider_name.upper()} ({account_uid})",
@@ -85,9 +83,9 @@ def transform(self, findings: List[Any]) -> None:
             ],
         )
 
-        result_uuid = str(
-            uuid.uuid5(uuid.NAMESPACE_DNS, f"prowler.result.{account_uid}")
-        )
+        # Result UUID must be unique per export (one assessment run), while
+        # observation/finding UUIDs stay reproducible from finding.uid.
+        result_uuid = str(uuid.uuid4())
         result = Result(
             result_uuid=result_uuid,
             title=f"Assessment Results for {provider_name.upper()} Account {account_uid}",
@@ -103,98 +101,83 @@ def transform(self, findings: List[Any]) -> None:
         )
 
         for finding in findings:
-            finding_timestamp = (
-                finding.timestamp.isoformat()
-                if hasattr(finding, "timestamp")
-                and isinstance(finding.timestamp, datetime)
-                else now_iso
-            )
-            resource_uid = getattr(finding, "resource_uid", "resource-default")
-            resource_name = getattr(finding, "resource_name", resource_uid)
-            finding_uid = getattr(finding, "uid", str(uuid.uuid4()))
-            region = getattr(finding, "region", "global")
-            muted = getattr(finding, "muted", False)
+            # OSCAL date-time-with-timezone requires a Z / offset; naive
+            # finding timestamps (common in tests and some providers) must
+            # be treated as UTC rather than emitted without a zone.
+            if isinstance(finding.timestamp, datetime):
+                ts = finding.timestamp
+                if ts.tzinfo is None:
+                    ts = ts.replace(tzinfo=timezone.utc)
+                finding_timestamp = ts.isoformat()
+            else:
+                finding_timestamp = now_iso
             status_val = (
                 finding.status.value
                 if hasattr(finding.status, "value")
                 else str(finding.status)
             )
-
-            subject_uuid = str(
-                uuid.uuid5(uuid.NAMESPACE_DNS, f"prowler.resource.{resource_uid}")
+            severity_val = (
+                finding.metadata.Severity.value
+                if hasattr(finding.metadata.Severity, "value")
+                else str(finding.metadata.Severity)
             )
-            obs_uuid = str(uuid.uuid5(uuid.NAMESPACE_DNS, f"prowler.obs.{finding_uid}"))
 
-            check_id = (
-                finding.metadata.CheckID
-                if hasattr(finding, "metadata") and hasattr(finding.metadata, "CheckID")
-                else "check_id"
+            subject_uuid = str(
+                uuid.uuid5(
+                    uuid.NAMESPACE_DNS, f"prowler.resource.{finding.resource_uid}"
+                )
             )
-            check_title = (
-                finding.metadata.CheckTitle
-                if hasattr(finding, "metadata")
-                and hasattr(finding.metadata, "CheckTitle")
-                else check_id
+            obs_uuid = str(
+                uuid.uuid5(uuid.NAMESPACE_DNS, f"prowler.obs.{finding.uid}")
             )
-            status_extended = getattr(finding, "status_extended", check_title)
-
-            severity_val = "medium"
-            if hasattr(finding, "metadata") and hasattr(finding.metadata, "Severity"):
-                sev = finding.metadata.Severity
-                severity_val = sev.value if hasattr(sev, "value") else str(sev)
 
             subject = Subject(
                 subject_uuid=subject_uuid,
                 type="component",
-                title=resource_name or resource_uid,
-                description=f"Resource evaluated in region {region}",
+                title=finding.resource_name or finding.resource_uid,
                 props=[
-                    Property(name="resource-uid", value=resource_uid),
-                    Property(name="region", value=region),
+                    Property(name="resource-uid", value=finding.resource_uid),
+                    Property(name="region", value=finding.region),
                 ],
             )
 
             observation = Observation(
                 observation_uuid=obs_uuid,
-                title=f"{check_id}: {status_val}",
-                description=status_extended,
+                title=f"{finding.metadata.CheckID}: {status_val}",
+                description=finding.status_extended,
                 collected=finding_timestamp,
                 subjects=[subject],
                 props=[
-                    Property(name="check-id", value=check_id),
+                    Property(name="check-id", value=finding.metadata.CheckID),
                     Property(name="status", value=status_val),
                     Property(name="severity", value=severity_val),
-                    Property(name="muted", value=str(muted).lower()),
-                    Property(name="region", value=region),
+                    Property(name="muted", value=str(finding.muted).lower()),
+                    Property(name="region", value=finding.region),
                 ],
             )
             result.observations.append(observation)
 
-            # Emit OSCAL finding only for FAIL status
-            if status_val == "FAIL":
+            # FAIL only, and never muted — muted FAILs stay as observations
+            # with muted=true so suppressed results do not assert remediation.
+            if status_val == Status.FAIL and not finding.muted:
                 find_uuid = str(
-                    uuid.uuid5(uuid.NAMESPACE_DNS, f"prowler.finding.{finding_uid}")
+                    uuid.uuid5(uuid.NAMESPACE_DNS, f"prowler.finding.{finding.uid}")
                 )
                 remediation_rec = ""
-                if hasattr(finding, "metadata") and hasattr(
-                    finding.metadata, "Remediation"
-                ):
-                    rem = finding.metadata.Remediation
-                    if hasattr(rem, "Recommendation") and hasattr(
-                        rem.Recommendation, "Text"
-                    ):
-                        remediation_rec = rem.Recommendation.Text
+                rem = finding.metadata.Remediation
+                if rem and rem.Recommendation and rem.Recommendation.Text:
+                    remediation_rec = rem.Recommendation.Text
 
                 oscal_finding = OscalFinding(
                     finding_uuid=find_uuid,
-                    title=f"Non-compliant check: {check_id}",
-                    description=status_extended,
-                    target_id=check_id,
+                    title=f"Non-compliant check: {finding.metadata.CheckID}",
+                    description=finding.status_extended,
+                    target_id=finding.metadata.CheckID,
                     related_observations=[
                         RelatedObservation(observation_uuid=obs_uuid)
                     ],
                     props=[
-                        Property(name="check-id", value=check_id),
+                        Property(name="check-id", value=finding.metadata.CheckID),
                         Property(name="status", value="unsatisfied"),
                         Property(name="severity", value=severity_val),
                     ],
@@ -206,10 +189,8 @@ def transform(self, findings: List[Any]) -> None:
                         )
                     )
 
-                # Extract NIST compliance mappings if present
-                compliance = getattr(finding, "compliance", {})
-                if isinstance(compliance, dict):
-                    for framework, controls in compliance.items():
+                if isinstance(finding.compliance, dict):
+                    for framework, controls in finding.compliance.items():
                         if "nist" in framework.lower() and isinstance(controls, list):
                             for ctrl in controls:
                                 oscal_finding.props.append(
@@ -233,9 +214,7 @@ def transform(self, findings: List[Any]) -> None:
         self._data.append(doc)
 
     def batch_write_data_to_file(self) -> None:
-        """
-        Serializes and writes the OSCAL document to file.
-        """
+        """Serialize and write the OSCAL document to ``file_descriptor``."""
         if self._data and self.file_descriptor:
             for doc in self._data:
                 payload = doc.to_dict()