fix(api): ensure hit statuses are not mismatched on ingest

api/howler/odm/models/howler_data.py

@@ -131,17 +131,17 @@ def __init__(self, data: dict = None, *args, **kwargs):
 
 @odm.model(index=True, store=True, description="Hit outline header.")
 class Header(odm.Model):
-    threat: Optional[str] = odm.Optional(odm.Keyword(description="The IP of the threat."))
-    target: Optional[str] = odm.Optional(odm.Keyword(description="The target of the hit."))
+    threat: str | None = odm.Optional(odm.Keyword(description="The IP of the threat."))
+    target: str | None = odm.Optional(odm.Keyword(description="The target of the hit."))
     indicators: list[str] = odm.List(odm.Keyword(description="Indicators of the hit."), default=[])
-    summary: Optional[str] = odm.Optional(odm.Keyword(description="Summary of the hit."))
+    summary: str | None = odm.Optional(odm.Keyword(description="Summary of the hit."))
 
 
 @odm.model(index=True, store=True, description="Fields describing the location where this alert has been retained.")
 class Incident(odm.Model):
     platform: str = odm.Keyword(description="The name of the platform for this incident.")
-    incident_id: Optional[str] = odm.Keyword(description="The ID of the incident.", optional=True)
-    url: Optional[str] = odm.Keyword(description="The url where the incident can be found.", optional=True)
+    incident_id: str | None = odm.Keyword(description="The ID of the incident.", optional=True)
+    url: str | None = odm.Keyword(description="The url where the incident can be found.", optional=True)
 
 
 @odm.model(index=True, store=True, description="Labels for the hit")
@@ -177,7 +177,7 @@ class Votes(odm.Model):
 class HowlerData(odm.Model):
     id: str = odm.UUID(description="A UUID for this hit.")
     analytic: str = odm.CaseInsensitiveKeyword(description="Title of the analytic.")
-    assignment: Optional[str] = odm.Keyword(
+    assignment: str = odm.Keyword(
         description="Unique identifier of the assigned user.",
         default=DEFAULT_ASSIGNMENT,
     )
@@ -191,7 +191,7 @@ class HowlerData(odm.Model):
         default=[],
         description="A list of links associated with this hit.",
     )
-    detection: Optional[str] = odm.Optional(
+    detection: str | None = odm.Optional(
         odm.CaseInsensitiveKeyword(description="The detection that produced this hit.")
     )
     hash: str = odm.HowlerHash(
@@ -219,13 +219,13 @@ class HowlerData(odm.Model):
     score: Optional[float] = odm.Optional(
         odm.Float(description="A score assigned by an enrichment to help prioritize triage.", default=0)
     )
-    status = odm.Enum(values=Status, default=Status.OPEN, description="Status of the hit.")
-    scrutiny = odm.Enum(
+    status: str = odm.Enum(values=Status, default=Status.OPEN, description="Status of the hit.")
+    scrutiny: str = odm.Enum(
         values=Scrutiny,
         default=Scrutiny.UNSEEN,
         description="Level of scrutiny done to this hit.",
     )
-    escalation = odm.Enum(
+    escalation: str = odm.Enum(
         values=Escalation,
         default=Escalation.HIT,
         description="Level of escalation of this hit.",
@@ -235,8 +235,8 @@ class HowlerData(odm.Model):
             description="User selected time for hit expiry",
         )
     )
-    assessment: Optional[str] = odm.Optional(odm.Enum(values=Assessment, description="Assessment of the hit."))
-    rationale: Optional[str] = odm.Optional(
+    assessment: str | None = odm.Optional(odm.Enum(values=Assessment, description="Assessment of the hit."))
+    rationale: str | None = odm.Optional(
         odm.Keyword(
             description=(
                 "The rationale behind the hit assessment. Allows it to be understood and verified by other analysts."
@@ -254,9 +254,9 @@ class HowlerData(odm.Model):
         default=[],
         description="A list of changes to the hit with timestamps and attribution.",
     )
-    monitored: Optional[str] = odm.Optional(odm.Keyword(description="Link to the incident monitoring dashboard."))
-    reported: Optional[str] = odm.Optional(odm.Keyword(description="Link to the incident report."))
-    mitigated: Optional[str] = odm.Optional(odm.Keyword(description="Link to the mitigation record (tool dependent)."))
+    monitored: str | None = odm.Optional(odm.Keyword(description="Link to the incident monitoring dashboard."))
+    reported: str | None = odm.Optional(odm.Keyword(description="Link to the incident report."))
+    mitigated: str | None = odm.Optional(odm.Keyword(description="Link to the mitigation record (tool dependent)."))
     outline: Optional[Header] = odm.Optional(odm.Compound(Header), description="The user specified header of the hit")
     incidents: list[Incident] = odm.List(
         odm.Compound(Incident), default=[], description="Fields describing an incident associated with this alert."

api/howler/services/hit_service.py

@@ -317,6 +317,18 @@ def convert_hit(  # noqa: C901
             "See howler's documentation for more information."
         )
 
+    if odm.howler.assessment:
+        target_escalation = AssessmentEscalationMap[odm.howler.assessment]
+        if odm.howler.escalation != target_escalation:
+            warnings.append(
+                f"Hits with assessment {odm.howler.assessment} must also have escalation set to {target_escalation}."
+            )
+            odm.howler.escalation = str(target_escalation)
+
+    if odm.howler.escalation in [Escalation.MISS, Escalation.EVIDENCE] and odm.howler.status != Status.RESOLVED:
+        warnings.append("Hits with escalation miss or evidence must also have their status set to resolved.")
+        odm.howler.status = Status.RESOLVED.value
+
     if odm.event:
         odm.event.id = odm.howler.id
         if not odm.event.created:

api/test/unit/odm/test_hit_fields.py

@@ -111,3 +111,51 @@ def test_howler_hashing():
 
     for _hash in EXAMPLE_HASHES.split("\n"):
         Hit({"howler.analytic": "Test Hash Analytic", "howler.hash": _hash})
+
+
+def test_convert_hit_assessment_normalizes_escalation_with_warning():
+    hit, warnings = hit_service.convert_hit(
+        {
+            "howler.analytic": "Assessment Test",
+            "howler.assessment": "security",
+            "howler.escalation": "alert",
+        },
+        unique=False,
+    )
+
+    assert hit.howler.assessment == "security"
+    assert hit.howler.escalation == "miss"
+    assert "Hits with assessment security must also have escalation set to miss." in warnings
+
+
+def test_convert_hit_escalation_normalizes_status_with_warning():
+    hit, warnings = hit_service.convert_hit(
+        {
+            "howler.analytic": "Escalation Test",
+            "howler.escalation": "evidence",
+            "howler.status": "open",
+        },
+        unique=False,
+    )
+
+    assert hit.howler.escalation == "evidence"
+    assert hit.howler.status == "resolved"
+    assert "Hits with escalation miss or evidence must also have their status set to resolved." in warnings
+
+
+def test_convert_hit_assessment_and_escalation_normalize_together():
+    hit, warnings = hit_service.convert_hit(
+        {
+            "howler.analytic": "Combined Assessment Test",
+            "howler.assessment": "security",
+            "howler.escalation": "evidence",
+            "howler.status": "open",
+        },
+        unique=False,
+    )
+
+    assert hit.howler.assessment == "security"
+    assert hit.howler.escalation == "miss"
+    assert hit.howler.status == "resolved"
+    assert "Hits with assessment security must also have escalation set to miss." in warnings
+    assert "Hits with escalation miss or evidence must also have their status set to resolved." in warnings