Fix mapping to Detection Finding OCSF class (#220)

wazuh · Sep 9, 2024 · 6c9ab83 · 6c9ab83
1 parent 8aa9d52
commit 6c9ab83
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 14 deletions.
diff --git a/integrations/amazon-security-lake/src/models/ocsf.py b/integrations/amazon-security-lake/src/models/ocsf.py
@@ -22,7 +22,7 @@ class AttackInfo(pydantic.BaseModel):
 
 class FindingInfo(pydantic.BaseModel):
     analytic: AnalyticInfo
-    attacks: AttackInfo
+    attacks: typing.List[AttackInfo]
     title: str
     types: typing.List[str]
     uid: str
@@ -61,6 +61,6 @@ class DetectionFinding(pydantic.BaseModel):
     risk_score: int
     severity_id: int
     status_id: int = 99
-    time: str
+    time: int
     type_uid: int = 200401
     unmapped: typing.Dict[str, typing.List[str]] = pydantic.Field()
diff --git a/integrations/amazon-security-lake/src/wazuh_ocsf_converter.py b/integrations/amazon-security-lake/src/wazuh_ocsf_converter.py
@@ -1,7 +1,9 @@
 import pydantic
 import models
 import logging
+from datetime import datetime
 
+timestamp_pattern = "%Y-%m-%dT%H:%M:%S.%f%z"
 
 def normalize(level: int) -> int:
     """
@@ -40,17 +42,19 @@ def to_detection_finding(event: models.wazuh.Event) -> models.ocsf.DetectionFind
                 type_id=1,
                 uid=event.rule.id
             ),
-            attacks=models.ocsf.AttackInfo(
-                tactic=models.ocsf.TechniqueInfo(
-                    name=", ".join(event.rule.mitre.tactic),
-                    uid=", ".join(event.rule.mitre.id)
-                ),
-                technique=models.ocsf.TechniqueInfo(
-                    name=", ".join(event.rule.mitre.technique),
-                    uid=", ".join(event.rule.mitre.id)
-                ),
-                version="v13.1"
-            ),
+            attacks=[
+                models.ocsf.AttackInfo(
+                    tactic=models.ocsf.TechniqueInfo(
+                        name=", ".join(event.rule.mitre.tactic),
+                        uid=", ".join(event.rule.mitre.id)
+                    ),
+                    technique=models.ocsf.TechniqueInfo(
+                        name=", ".join(event.rule.mitre.technique),
+                        uid=", ".join(event.rule.mitre.id)
+                    ),
+                    version="v13.1"
+                )
+            ],
             title=event.rule.description,
             types=[event.input.type],
             uid=event.id
@@ -89,13 +93,16 @@ def to_detection_finding(event: models.wazuh.Event) -> models.ocsf.DetectionFind
             resources=resources,
             risk_score=event.rule.level,
             severity_id=severity_id,
-            time=event.timestamp,
+            time=to_epoch(event.timestamp),
             unmapped=unmapped
         )
     except AttributeError as e:
         logging.error(f"Error transforming event: {e}")
         return {}
 
+def to_epoch(timestamp: str) -> int:
+    return int(datetime.strptime(timestamp, timestamp_pattern).timestamp())
+
 
 def from_json(json_line: str) -> models.wazuh.Event:
     """