fix(OCSF): fix OCSF output when timestamp is UNIX format (#6626)

Co-authored-by: Rubén De la Torre Vico <[email protected]>
prowler-cloud · Jan 20, 2025 · 0a9ea21 · 0a9ea21
1 parent a591beb
commit 0a9ea21
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 5 deletions.
diff --git a/prowler/lib/outputs/ocsf/ocsf.py b/prowler/lib/outputs/ocsf/ocsf.py
@@ -1,4 +1,5 @@
 import os
+from datetime import datetime
 from typing import List
 
 from py_ocsf_models.events.base_event import SeverityID, StatusID
@@ -68,7 +69,11 @@ def transform(self, findings: List[Finding]) -> None:
                     activity_name=finding_activity.name,
                     finding_info=FindingInformation(
                         created_time_dt=finding.timestamp,
-                        created_time=int(finding.timestamp.timestamp()),
+                        created_time=(
+                            int(finding.timestamp.timestamp())
+                            if isinstance(finding.timestamp, datetime)
+                            else finding.timestamp
+                        ),
                         desc=finding.metadata.Description,
                         title=finding.metadata.CheckTitle,
                         uid=finding.uid,
@@ -77,7 +82,11 @@ def transform(self, findings: List[Finding]) -> None:
                         types=finding.metadata.CheckType,
                     ),
                     time_dt=finding.timestamp,
-                    time=int(finding.timestamp.timestamp()),
+                    time=(
+                        int(finding.timestamp.timestamp())
+                        if isinstance(finding.timestamp, datetime)
+                        else finding.timestamp
+                    ),
                     remediation=Remediation(
                         desc=finding.metadata.Remediation.Recommendation.Text,
                         references=list(

diff --git a/tests/lib/outputs/fixtures/fixtures.py b/tests/lib/outputs/fixtures/fixtures.py
@@ -1,4 +1,5 @@
 from datetime import datetime
+from typing import Union
 
 from prowler.config.config import prowler_version
 from prowler.lib.check.models import CheckMetadata, Code, Recommendation, Remediation
@@ -19,7 +20,7 @@ def generate_finding_output(
     resource_name: str = "",
     resource_tags: dict = {},
     compliance: dict = {"test-compliance": "test-compliance"},
-    timestamp: datetime = None,
+    timestamp: Union[int, datetime] = None,
     provider: str = "aws",
     partition: str = "aws",
     description: str = "check description",

diff --git a/tests/lib/outputs/ocsf/ocsf_test.py b/tests/lib/outputs/ocsf/ocsf_test.py
@@ -1,5 +1,5 @@
 import json
-from datetime import datetime
+from datetime import datetime, timezone
 from io import StringIO
 
 import requests
@@ -36,7 +36,15 @@ def test_transform(self):
                 muted=False,
                 region=AWS_REGION_EU_WEST_1,
                 resource_tags={"Name": "test", "Environment": "dev"},
-            )
+            ),
+            # Test with int timestamp (UNIX timestamp)
+            generate_finding_output(
+                status="FAIL",
+                severity="medium",
+                muted=False,
+                region=AWS_REGION_EU_WEST_1,
+                timestamp=1619600000,
+            ),
         ]
 
         ocsf = OCSF(findings)
@@ -100,6 +108,14 @@ def test_transform(self):
             "compliance": findings[0].compliance,
         }
 
+        # Test with int timestamp (UNIX timestamp)
+        output_data = ocsf.data[1]
+
+        assert output_data.time == 1619600000
+        assert output_data.time_dt == datetime.fromtimestamp(
+            1619600000, tz=timezone.utc
+        )
+
     def test_validate_ocsf(self):
         mock_file = StringIO()
         findings = [