Redact secret-shaped fields by default, countable per field path

AUDIT-WP-0004-T04, closing the workplan.

Decision (Bernd): default to redaction, allow rejection per sender. Losing an
audit record over one field is worse than storing it masked, but a
higher-assurance channel must be able to refuse rather than mask. secret_policy
is set per sender identity in AUDIT_CORE_SENDERS and defaults to redact.

Detection now covers the whole payload at any depth, including lists, rather
than only the top level of data. Under redaction the value is masked and the
key is preserved: dropping the key would hide that the sender transmitted the
field at all, which is exactly what an operator needs in order to stop it. The
stored record carries details.redaction with policy and affected paths, so a
reader never has to infer whether what they see is what was sent.

Idempotency is unaffected - the payload hash is taken over the original request
body, so redaction is deterministic and a resubmission still reconciles as a
duplicate.

Both outcomes are counted durably by sender, source, action and field path,
exposed at GET /v1/secret-findings. Per-path aggregation is the point: the
actionable unit is "stop emitting data.auth.token on membership.added", not
"there were 47 redactions". Counters survive restart because the fix they drive
lives in another service.

Contract doc updated to match. Tests 46 -> 50. WP-0004 is finished.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

audit_core/ingestion.py

@@ -38,11 +38,16 @@ from audit_core.interface import (
     EventValidationError,
     IdempotentAuditBackend,
 )
+from audit_core.redaction import (
+    POLICY_REDACT,
+    SecretFieldRejection,
+    apply_policy,
+    finding_from_path,
+)
 from audit_core.senders import SenderIdentity, SenderRegistry, development_registry
 from audit_core.sqlite_backend import SQLiteAuditBackend
 
 MAX_BODY_BYTES = 256 * 1024
-_SECRET_FRAGMENTS = ("password", "secret", "token", "credential", "private_key")
 
 log = logging.getLogger("audit_core.ingestion")
 
@@ -99,7 +104,10 @@ class IngestionApplication:
                 start_response, HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"}
             )
 
-        if method == "GET" and (path.startswith("/v1/events") or path == "/v1/dead-letters"):
+        if method == "GET" and (
+            path.startswith("/v1/events")
+            or path in ("/v1/dead-letters", "/v1/secret-findings")
+        ):
             return self._read(start_response, environ, path, identity)
 
         if path != "/v1/events" or method != "POST":
@@ -109,15 +117,26 @@ class IngestionApplication:
             return self._json(start_response, HTTPStatus.FORBIDDEN, {"error": "write_forbidden"})
 
         raw = b""
+        payload: Any = {}
         try:
             raw = self._read_body(environ)
-            event = normalize(
-                json.loads(raw), environ.get("HTTP_IDEMPOTENCY_KEY"), identity
-            )
+            payload = json.loads(raw)
+            event = normalize(payload, environ.get("HTTP_IDEMPOTENCY_KEY"), identity)
+        except SecretFieldRejection as exc:
+            self._count_secrets(payload, identity, "rejected", exc.findings)
+            self._dead_letter(raw, str(exc), identity)
+            return self._json(start_response, HTTPStatus.BAD_REQUEST, {"error": str(exc)})
         except (ValueError, TypeError, KeyError, json.JSONDecodeError) as exc:
             self._dead_letter(raw, str(exc), identity)
             return self._json(start_response, HTTPStatus.BAD_REQUEST, {"error": str(exc)})
 
+        redaction = event.details.get("redaction")
+        if redaction:
+            self._count_secrets(
+                payload, identity, "redacted",
+                [finding_from_path(p) for p in redaction["paths"]],
+            )
+
         try:
             result = self.backend.accept(event, hashlib.sha256(raw).hexdigest())
         except EventConflictError as exc:
@@ -156,6 +175,11 @@ class IngestionApplication:
                     start_response, HTTPStatus.OK,
                     {"dead_letters": self.backend.dead_letters(_limit(query))},
                 )
+            if path == "/v1/secret-findings":
+                return self._json(
+                    start_response, HTTPStatus.OK,
+                    {"secret_findings": self.backend.secret_findings(_limit(query))},
+                )
             if path == "/v1/events":
                 correlation = (query.get("correlation_id") or [""])[0]
                 if not correlation:
@@ -178,6 +202,28 @@ class IngestionApplication:
                 start_response, HTTPStatus.SERVICE_UNAVAILABLE, {"error": "backend_unavailable"}
             )
 
+    def _count_secrets(self, payload: Any, identity, outcome: str, findings) -> None:
+        """Count secret-shaped fields by path, sender, source and action.
+
+        Counted so the sending service can be fixed. Aggregation is by field
+        path rather than by event, because the actionable unit is "stop
+        emitting ``data.auth.token`` on ``membership.added``", not "there were
+        47 redactions".
+        """
+        counter = getattr(self.backend, "count_secret_findings", None)
+        if not callable(counter) or not findings:
+            return
+        source = str(payload.get("source") or "") if isinstance(payload, dict) else ""
+        action = str(payload.get("type") or "") if isinstance(payload, dict) else ""
+        try:
+            counter(
+                sender=identity.name, source=source, action=action,
+                outcome=outcome, findings=findings,
+            )
+        except BackendUnavailableError as exc:
+            # A missed counter must never change the event's outcome.
+            log.error("could not count secret findings: %s", exc)
+
     def _dead_letter(self, raw: bytes, reason: str, identity) -> None:
         """Record a rejection so an operator can see what the sender dropped."""
         recorder = getattr(self.backend, "record_rejection", None)
@@ -268,8 +314,21 @@ def normalize(
     elif source != "user-engine":
         raise ValueError("source_not_allowed")
     observed_at = _normalize_timestamp(payload["occurred_at"])
-    if _contains_secret(payload["data"]):
-        raise ValueError("secret_shaped_field")
+    policy = identity.secret_policy if identity is not None else POLICY_REDACT
+    # Raises SecretFieldRejection under the reject policy; that exception
+    # carries the findings so the caller can count them.
+    data, findings = apply_policy(payload, policy)
+    details: dict[str, Any] = {
+        "correlation_id": str(payload["correlation_id"]),
+        "data": data,
+    }
+    if findings:
+        # The stored record differs from what the sender transmitted. Say so in
+        # the record itself rather than leaving it to be inferred.
+        details["redaction"] = {
+            "policy": policy,
+            "paths": [f.path for f in findings],
+        }
     return AuditEvent(
         event_id=str(payload["id"]),
         observed_at=observed_at,
@@ -280,7 +339,7 @@ def normalize(
         resource=str(payload["subject"]),
         outcome="recorded",
         actor=None,
-        details={"correlation_id": str(payload["correlation_id"]), "data": payload["data"]},
+        details=details,
     )
 
 
@@ -299,18 +358,6 @@ def _normalize_timestamp(value: Any) -> str:
     return parsed.astimezone(timezone.utc).isoformat()
 
 
-def _contains_secret(value: Any) -> bool:
-    if isinstance(value, dict):
-        return any(
-            any(fragment in str(key).lower() for fragment in _SECRET_FRAGMENTS)
-            or _contains_secret(item)
-            for key, item in value.items()
-        )
-    if isinstance(value, list):
-        return any(_contains_secret(item) for item in value)
-    return False
-
-
 def _limit(query: dict[str, list[str]], default: int = 100, ceiling: int = 1000) -> int:
     try:
         return max(1, min(int((query.get("limit") or [default])[0]), ceiling))