feat(webhooks): SIEM enrichment fields on MemoryDefenseEventData

hindsight-api-slim/hindsight_api/engine/retain/orchestrator.py

@@ -89,7 +89,29 @@ async def _fire_memory_defense_webhook(
     if webhook_manager is None:
         return
     try:
-        from ...webhooks import MemoryDefenseEventData, WebhookEvent, WebhookEventType
+        from ...webhooks import (
+            MemoryDefenseEventData,
+            MemoryDefenseHit,
+            WebhookEvent,
+            WebhookEventType,
+        )
+
+        # Translate per-match raw dicts on the decision into MemoryDefenseHit
+        # entries on the wire. The decision's hits list is already fingerprinted
+        # by apply_redaction (the raw value never lands in hits, by contract),
+        # so this is purely a shape conversion. None when no per-hit data is
+        # available so receivers can distinguish "no preview info" from
+        # "scanned, nothing matched" (the latter wouldn't be a webhook delivery
+        # in the first place).
+        decision_hits = getattr(decision, "hits", None) or []
+        hits: list[MemoryDefenseHit] | None = [
+            MemoryDefenseHit(
+                detector=str(h.get("detector") or ""),
+                preview=str(h.get("preview") or ""),
+            )
+            for h in decision_hits
+            if h.get("detector") and h.get("preview")
+        ] or None
 
         event = WebhookEvent(
             event=WebhookEventType.MEMORY_DEFENSE_TRIGGERED,
@@ -103,6 +125,7 @@ async def _fire_memory_defense_webhook(
                 document_id=document_id,
                 matched_types=decision.matched_types or None,
                 message=decision.message or None,
+                hits=hits,
             ),
         )
         await webhook_manager.fire_event_with_conn(event, conn, schema=schema)

hindsight-api-slim/hindsight_api/extensions/builtin/memory_defense_regex.py

@@ -52,4 +52,5 @@ async def screen(
             message=f"Sensitive data pattern matched: {', '.join(result.matched_types)}",
             redacted_content=result.content if rule.action is DefenseAction.REDACT else None,
             matched_types=result.matched_types,
+            hits=result.hits,
         )

hindsight-api-slim/hindsight_api/extensions/memory_defense.py

@@ -58,12 +58,51 @@ class DefenseDecision:
     message: str = ""
     redacted_content: str | None = None
     matched_types: list[str] = field(default_factory=list)
+    # Per-match fingerprinted previews. Each entry is
+    # ``{"detector": <pattern label>, "preview": <fingerprinted value>}``.
+    # The preview is *never* the raw value — see :func:`_fingerprint_value`.
+    # OSS populates this from ``apply_redaction``; downstream extensions
+    # populate it from their own detectors. Optional: empty when the
+    # match path didn't capture per-hit values.
+    hits: list[dict] = field(default_factory=list)
 
 
 @dataclass
 class RedactionResult:
     content: str
     matched_types: list[str]
+    # Same shape as ``DefenseDecision.hits`` — one entry per matched value
+    # (so a single content with two GitHub tokens produces two entries).
+    hits: list[dict] = field(default_factory=list)
+
+
+def _fingerprint_value(value: str) -> str:
+    """Return a redaction-identifiable preview of a matched value.
+
+    The preview keeps the prefix and a short suffix so a SIEM operator can
+    correlate against their credential inventory (the prefix names the
+    provider; the suffix disambiguates specific instances) without the raw
+    secret crossing the wire. Length-aware so short values don't accidentally
+    leak material:
+
+    - Length < 6:  redact entirely (return a fixed-length mask). Catches
+      noise like a single ``-----BEGIN...`` marker line.
+    - Length 6-15: keep the first 2 + last 2 around an ellipsis.
+    - Length > 15: keep the first 4 + last 4 around an ellipsis.
+
+    Examples::
+
+        _fingerprint_value("ghp_AAAA...AAAA" + "A" * 36)  -> "ghp_...AAAA"
+        _fingerprint_value("AKIA" + "B" * 16)              -> "AKIA...BBBB"
+        _fingerprint_value("123-45-6789")                  -> "12...89"
+        _fingerprint_value("abc")                          -> "[redacted]"
+    """
+    n = len(value)
+    if n < 6:
+        return "[redacted]"
+    if n <= 15:
+        return f"{value[:2]}...{value[-2:]}"
+    return f"{value[:4]}...{value[-4:]}"
 
 
 def parse_policy(raw: dict | None) -> DefensePolicy:
@@ -171,16 +210,42 @@ def parse_policy(raw: dict | None) -> DefensePolicy:
 def apply_redaction(content: str) -> RedactionResult:
     """Scrub known secret/PII patterns from content with [REDACTED:type] markers.
 
-    Returns the (possibly unchanged) content alongside the list of pattern
-    labels that matched (empty when nothing matched).
+    Returns the (possibly unchanged) content alongside:
+      - ``matched_types``: pattern labels that matched (deduplicated, in
+        first-occurrence order). Empty when nothing matched.
+      - ``hits``: per-match fingerprinted previews — one entry per matched
+        substring (so two GitHub tokens in the same content produce two
+        entries). Each entry is ``{"detector": label, "preview": fingerprint}``
+        where ``preview`` is a length-aware redaction of the original value.
+        The raw secret never appears in ``hits``.
+
+    The two-pass shape (find matches first, then substitute) lets us capture
+    raw values for fingerprinting before they're replaced by ``[REDACTED:type]``
+    markers. A single-pass approach would lose the originals.
     """
     matched: list[str] = []
+    hits: list[dict] = []
     for label, pattern in _COMPILED_REDACTIONS:
-        new_content = pattern.sub(f"[REDACTED:{label}]", content)
-        if new_content != content:
+        raw_hits = pattern.findall(content)
+        if not raw_hits:
+            continue
+        if label not in matched:
             matched.append(label)
-            content = new_content
-    return RedactionResult(content=content, matched_types=matched)
+        for raw in raw_hits:
+            # findall returns either a string or a tuple of capture groups
+            # depending on the pattern. The redaction-pattern catalog uses a
+            # mix; coerce to the matched substring as best we can.
+            if isinstance(raw, tuple):
+                # Pick the longest non-empty group as the canonical match.
+                non_empty = [g for g in raw if g]
+                raw_str = max(non_empty, key=len) if non_empty else ""
+            else:
+                raw_str = raw
+            if not raw_str:
+                continue
+            hits.append({"detector": label, "preview": _fingerprint_value(raw_str)})
+        content = pattern.sub(f"[REDACTED:{label}]", content)
+    return RedactionResult(content=content, matched_types=matched, hits=hits)
 
 
 class MemoryDefenseExtension(Extension, ABC):

hindsight-api-slim/hindsight_api/webhooks/__init__.py

@@ -4,6 +4,7 @@
 from .models import (
     ConsolidationEventData,
     MemoryDefenseEventData,
+    MemoryDefenseHit,
     RetainEventData,
     WebhookConfig,
     WebhookEvent,
@@ -17,5 +18,6 @@
     "WebhookEventType",
     "ConsolidationEventData",
     "MemoryDefenseEventData",
+    "MemoryDefenseHit",
     "RetainEventData",
 ]

hindsight-api-slim/hindsight_api/webhooks/models.py

@@ -24,14 +24,43 @@ class RetainEventData(BaseModel):
     tags: list[str] | None = None
 
 
+class MemoryDefenseHit(BaseModel):
+    """A single secret match inside a non-allow decision.
+
+    ``preview`` is a fingerprinted, redaction-identifiable rendering of the
+    matched value (e.g. ``ghp_AAAA...BBBB``) so SIEM operators can correlate
+    against their credential inventory WITHOUT the raw secret crossing the
+    network. Implementations must never put the raw value here.
+    """
+
+    detector: str  # the inner detector that matched (e.g. "GitHub Token")
+    preview: str  # fingerprinted value, never the raw secret
+
+
 class MemoryDefenseEventData(BaseModel):
-    """Payload for a memory_defense.triggered event (one item, one non-allow decision)."""
+    """Payload for a memory_defense.triggered event (one item, one non-allow decision).
+
+    The four base fields (``action``/``detector``/``document_id``/``message``)
+    plus ``matched_types`` are populated by every implementation including OSS's
+    built-in regex defense. The remaining fields are optional SIEM-enrichment
+    surfaces that downstream extensions (e.g. hindsight-cloud) populate when
+    they have richer per-decision context — severity classification, the API
+    key that submitted the retain, fingerprinted hit previews for SIEM
+    correlation, and pointers into the audit trail. OSS leaves them ``None``;
+    receivers should treat absence as "not provided" rather than "no match".
+    """
 
     action: str  # "redact" or "block"
     detector: str | None = None  # e.g. "sensitive_data"
     document_id: str | None = None
     matched_types: list[str] | None = None  # redaction pattern labels that fired
     message: str | None = None
+    # --- Optional SIEM enrichment (populated by extensions, not OSS) ---
+    severity: str | None = None  # "low" / "medium" / "high" / "critical"
+    api_key_name: str | None = None  # human-readable name of the submitting API key
+    hits: list[MemoryDefenseHit] | None = None  # per-match fingerprints for correlation
+    memory_unit_id: str | None = None  # drill-down pointer (when the decision was REDACT)
+    receipt_uri: str | None = None  # storage pointer for the audit trail entry
 
 
 class WebhookEvent(BaseModel):

hindsight-api-slim/tests/test_memory_defense.py

@@ -19,6 +19,8 @@
 from hindsight_api.extensions.memory_defense import (
     DefenseAction,
     MemoryDefenseExtension,
+    _fingerprint_value,
+    apply_redaction,
     parse_policy,
 )
 
@@ -92,6 +94,73 @@ def test_defense_action_string_round_trip() -> None:
     assert DefenseAction.BLOCK.value == "block"
 
 
+# ---------------------------------------------------------------------------
+# Fingerprinting (unit)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        # Length > 15 → first-4 + ellipsis + last-4.
+        ("ghp_" + "A" * 36, "ghp_...AAAA"),
+        ("AKIA" + "B" * 16, "AKIA...BBBB"),
+        ("sk-ant-" + "Z" * 40, "sk-a...ZZZZ"),
+        # Length 6–15 → first-2 + ellipsis + last-2.
+        ("123-45-6789", "12...89"),
+        ("xoxb-12345", "xo...45"),
+        # Length < 6 → fully masked; we don't preview anything.
+        ("abcde", "[redacted]"),
+        ("", "[redacted]"),
+    ],
+)
+def test_fingerprint_value_shape(value: str, expected: str) -> None:
+    """_fingerprint_value never returns the raw value and uses length-aware
+    bracketing so short matches don't leak material."""
+    out = _fingerprint_value(value)
+    assert out == expected
+    if value:
+        assert value not in out, f"raw value leaked into fingerprint: {out!r}"
+
+
+def test_apply_redaction_hits_carry_fingerprinted_previews() -> None:
+    """apply_redaction returns per-match fingerprinted previews — one entry
+    per matched substring — with the raw secret nowhere present in the hits."""
+    s1 = "ghp_" + "A" * 36
+    s2 = "AKIA" + "B" * 16
+    s3 = "123-45-6789"
+    content = f"rotate {s1}, drop {s2}, also ssn {s3}"
+
+    result = apply_redaction(content)
+
+    # Same-shape labels still flow to matched_types (deduplicated).
+    assert set(result.matched_types) >= {"github_token", "aws_access_key", "ssn_us"}
+
+    # One hit per matched substring; raw secret never appears.
+    by_detector = {h["detector"]: h["preview"] for h in result.hits}
+    assert by_detector["github_token"] == "ghp_...AAAA"
+    assert by_detector["aws_access_key"] == "AKIA...BBBB"
+    assert by_detector["ssn_us"] == "12...89"
+    for h in result.hits:
+        assert s1 not in h["preview"]
+        assert s2 not in h["preview"]
+        assert s3 not in h["preview"]
+
+
+def test_apply_redaction_multiple_hits_per_pattern() -> None:
+    """Two matches of the same pattern produce two hits — receivers can count
+    occurrences, not just types."""
+    a = "ghp_" + "A" * 36
+    b = "ghp_" + "B" * 36
+    content = f"old {a} new {b}"
+    result = apply_redaction(content)
+
+    gh_hits = [h for h in result.hits if h["detector"] == "github_token"]
+    assert len(gh_hits) == 2
+    previews = {h["preview"] for h in gh_hits}
+    assert previews == {"ghp_...AAAA", "ghp_...BBBB"}
+
+
 # ---------------------------------------------------------------------------
 # Regex screening (unit)
 # ---------------------------------------------------------------------------
@@ -134,6 +203,14 @@ async def test_screen_redacts_secret(regex_defense, redact_policy) -> None:
     assert secret not in decision.redacted_content
     assert "[REDACTED:github_token]" in decision.redacted_content
     assert "github_token" in decision.matched_types
+    # The decision carries a per-match fingerprinted preview — never the raw
+    # value — so SIEM receivers can correlate without the secret crossing
+    # the wire.
+    assert decision.hits, "OSS should populate at least one hit"
+    hit = decision.hits[0]
+    assert hit["detector"] == "github_token"
+    assert hit["preview"] == "ghp_...AAAA"
+    assert secret not in hit["preview"]
 
 
 @pytest.mark.asyncio
@@ -417,8 +494,13 @@ async def _memory_defense_webhook_events(memory, bank: str) -> list[dict]:
     deliveries queued for ``bank``. The webhook_delivery task_payload nests the
     serialized event under ``payload`` (a JSON string)."""
     async with memory._pool.acquire() as conn:
+        # Order most-recent-first so callers using ``events[0]`` always see
+        # the latest queued delivery — otherwise pollution from earlier test
+        # runs against the same bank surfaces stale payloads.
         rows = await conn.fetch(
-            "SELECT task_payload FROM async_operations WHERE operation_type = 'webhook_delivery' AND bank_id = $1",
+            "SELECT task_payload FROM async_operations "
+            "WHERE operation_type = 'webhook_delivery' AND bank_id = $1 "
+            "ORDER BY created_at DESC",
             bank,
         )
     events: list[dict] = []
@@ -462,6 +544,14 @@ async def test_retain_fires_webhook_on_redact(api_client, memory) -> None:
     assert data["detector"] == "sensitive_data"
     assert "github_token" in data["matched_types"]
     assert data["message"]
+    # The webhook payload carries a per-match fingerprinted preview — the raw
+    # secret never crosses the wire, but a SIEM can still correlate against
+    # its credential inventory using the leading provider prefix + trailing
+    # discriminator (e.g. `ghp_...AAAA`). Populated by OSS as of #2157.
+    hits = data.get("hits") or []
+    assert any(h.get("detector") == "github_token" and h.get("preview") == "ghp_...AAAA" for h in hits), hits
+    for h in hits:
+        assert secret not in (h.get("preview") or ""), "raw secret leaked into preview"
 
 
 @pytest.mark.asyncio