feat(audit): consume signed events, classify integrity_status (PR2)

[man4ish]

PR2 of the audit:events integrity remediation (PR1 #4, merged — the sign/verify helper, previously unwired anywhere). Wires classification into the worker without requiring any producer to sign yet.

What changes

• AuditConfig.EVENT_SIGNING_SECRET — reuses the existing JWT_SECRET env var and "change-me" fallback, same convention audit/jwt_verify.py already uses. No new secret mechanism. audit/jwt_verify.py itself is untouched.
• consumers/processor.py::classify_event_integrity(service, signature, data, secret) -> str — new, separate function. parse_audit_event()'s signature and its 3 existing callers are unchanged.
• worker/main.py::handle_message — classifies after a successful parse, using the exact raw fields["data"] string (never a re-serialization), persists via a purely additive consumers/sink.py lookup, then ACKs.
• db/models.py + alembic/versions/0002_integrity_status.py — exactly one new column, integrity_status (String(16), NOT NULL, server_default="unsigned"). No index — nothing queries it yet, and the table has none on any column today.

Classification behavior

if not signature: return "unsigned"
return "valid" if verify_audit_event(service, data, signature, secret) else "invalid"

The missing-signature check happens before verify_audit_event() — verified by a dedicated ordering test, not just asserted.

ACK/retry behavior

Case	Persisted?	ACK?
Valid	yes, "valid"	yes
Unsigned (100% of current traffic)	yes, "unsigned"	yes
Invalid/malformed sig	yes, "invalid"	yes + distinct SIGNATURE INVALID print
Malformed JSON/schema	no	unchanged — still unacked/retried
DB failure	no	unchanged — still unacked/retried

Invalid signatures are persisted and ACKed rather than left pending: a signature verdict can't change on retry, so leaving it pending would only accumulate a permanent poison-pill entry, whereas persisting it preserves the forged attempt as durable evidence. A dedicated test proves the distinct log line never includes the secret, the signature, or the raw event data.

Migration

0002_integrity_status, down_revision="0001_audit_events" (21-char revision id, under the installed alembic's 32-char alembic_version.version_num limit — verified against source). Tests prove: a row inserted before this migration existed backfills to "unsigned" on upgrade (real backfill via server_default, not just schema presence), and downgrading exactly one revision removes only this column, distinct from the existing full-downgrade-to-base test.

Tests — 32 new

12 classification (valid/invalid/malformed/missing/empty, service+data+signature mismatches, exact-raw-string preservation via a reordered-JSON-keys case, missing-check ordering), 3 migration, 4 sink, 10 worker (including the secret/signature/data non-leakage security test), 3 real Redis+MySQL integration tests extending the existing PR-B0 fixture — signing with whatever AuditConfig.EVENT_SIGNING_SECRET actually resolves to in the running test process, never assuming which value it holds.

Full suite: 423 passed (391 baseline + 32 new), 0 regressions
ruff:       16 findings, all pre-existing or matching 0001's own Union/Sequence migration style — 0 new avoidable findings
git diff --check: clean

Compatibility

Every existing producer keeps working unmodified — none write sig today; the real integration test proves today's exact traffic shape ({"data": ...}, no sig) classifies as "unsigned" and persists/ACKs against real Redis+MySQL. No API Gateway, Control Center, or other-repo changes. audit/signing.py — zero-line diff.

Follow-up (deliberately not fixed here)

The dev-compose security-audit-worker container currently has JWT_SECRET unset (falls back to "change-me") while other platform services have the real shared secret configured. Harmless today since no producer signs yet, but must be corrected before any producer-signing rollout, or valid producer signatures will be classified as invalid.

🤖 Generated with Claude Code