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

alembic/versions/0002_integrity_status.py

@@ -0,0 +1,43 @@
+"""add integrity_status to audit_events
+
+PR2 of the audit:events integrity remediation (see audit/signing.py, PR1;
+consumers/processor.py::classify_event_integrity, this PR). Adds one
+column recording whether the worker was able to cryptographically verify
+each event at ingest time: "valid", "invalid", or "unsigned".
+
+server_default="unsigned" back-fills every existing row automatically on
+this ALTER -- no manual UPDATE, no backfill script. That default is
+correct for pre-existing rows: they predate signing entirely, so
+"unsigned" is not a guess, it's simply true.
+
+Revision ID: 0002_integrity_status
+Revises: 0001_audit_events
+Create Date: 2026-08-14
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = "0002_integrity_status"
+down_revision: Union[str, None] = "0001_audit_events"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "audit_events",
+        sa.Column(
+            "integrity_status",
+            sa.String(length=16),
+            nullable=False,
+            server_default="unsigned",
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("audit_events", "integrity_status")

audit/config.py

@@ -17,4 +17,18 @@ class AuditConfig:
         "mysql+pymysql://root:root@localhost:3306/omnibioai_audit",
     )
     CONSUMER_GROUP = os.getenv("AUDIT_CONSUMER_GROUP", "audit-workers")
-    CONSUMER_NAME = os.getenv("AUDIT_CONSUMER_NAME", f"worker-{os.getpid()}")
+    CONSUMER_NAME = os.getenv("AUDIT_CONSUMER_NAME", f"worker-{os.getpid()}")
+
+    # PR2 of the audit:events integrity remediation (see audit/signing.py,
+    # PR1): reuses the same JWT_SECRET every other platform service (and
+    # this repo's own audit/jwt_verify.py) already reads -- not a new
+    # secret, not a new convention. Falls back to "change-me" the same way
+    # jwt_verify.JWT_SECRET does; a deployment that never set the real
+    # JWT_SECRET already has a forgeable HS256 token secret, so this adds
+    # no new exposure. See PR2's own report: the dev-compose worker
+    # container currently has JWT_SECRET unset entirely (falls back here
+    # too) -- until that's fixed, this worker cannot correctly verify a
+    # signature made with the platform's real secret. Harmless today since
+    # no producer signs yet (every event classifies as "unsigned"), but
+    # this default must not be trusted once a producer starts signing.
+    EVENT_SIGNING_SECRET = os.getenv("JWT_SECRET", "change-me")

consumers/processor.py

@@ -1,6 +1,9 @@
+from __future__ import annotations
+
 import json
 
 from audit.models import AuditEvent
+from audit.signing import verify_audit_event
 
 
 def process_event(raw_event: str):
@@ -28,4 +31,31 @@ def parse_audit_event(raw_event: str) -> AuditEvent:
     payload -- the caller (worker/main.py) decides what to do with that.
     """
     payload = json.loads(raw_event)
-    return AuditEvent(**payload)
+    return AuditEvent(**payload)
+
+
+def classify_event_integrity(
+    service: str, signature: str | None, data: str, secret: str
+) -> str:
+    """PR2: the worker's answer to "should this event be trusted as
+    signed?" -- deliberately a separate function from parse_audit_event()
+    (which has other callers, see this PR's report) rather than a change
+    to its signature.
+
+    `data` must be the exact raw stream string (fields["data"]), never a
+    re-serialization -- audit.signing's own MAC covers the exact
+    transmitted bytes, see its module docstring. `signature` is
+    fields.get("sig"), which is None for every event today (no producer
+    signs yet); this is not folded into verify_audit_event() itself
+    because that function already collapses "missing" and "invalid" into
+    the same False (correctly -- that distinction is this consumer's
+    business, not the signing module's), so the emptiness check has to
+    happen here, before delegating.
+
+    Returns exactly one of "valid" / "invalid" / "unsigned" -- never
+    raises (verify_audit_event() already never raises; the emptiness
+    check above it can't either).
+    """
+    if not signature:
+        return "unsigned"
+    return "valid" if verify_audit_event(service, data, signature, secret) else "invalid"

consumers/sink.py

@@ -34,6 +34,12 @@ def write(self, event: dict) -> bool:
             reason=event.get("reason"),
             trace_id=event.get("trace_id"),
             context=event.get("context", {}),
+            # PR2: additive -- callers that never pass this key (every
+            # existing caller as of this PR) get the same "unsigned"
+            # default the column's own server_default applies at the DB
+            # level, so omitting it is indistinguishable from a genuinely
+            # unsigned event, not an error.
+            integrity_status=event.get("integrity_status", "unsigned"),
         )
         self.db_session.add(record)
         try:

db/models.py

@@ -28,4 +28,12 @@ class AuditEventRecord(Base):
     reason = Column(Text, nullable=True)
     trace_id = Column(String(255), nullable=True)
     context = Column(JSON, nullable=False, default=dict)
+    # PR2 of the audit:events integrity remediation: one of "valid" /
+    # "invalid" / "unsigned", set by the worker (consumers/processor.py::
+    # classify_event_integrity) at ingest time -- never supplied by a
+    # producer. server_default="unsigned" back-fills every pre-existing
+    # row on migration with no manual UPDATE, and matches what an event
+    # missing a signature already classifies as going forward, so old and
+    # new "no signature" rows read identically.
+    integrity_status = Column(String(16), nullable=False, server_default="unsigned")
     created_at = Column(DateTime, server_default=func.now(), nullable=False)

tests/test_classify_event_integrity.py

@@ -0,0 +1,116 @@
+"""PR2 of the audit:events integrity remediation: classify_event_integrity()
+(consumers/processor.py) -- the worker-side glue between PR1's sign/verify
+helper (audit/signing.py, untouched by this PR) and the new integrity_status
+persisted on each AuditEventRecord.
+
+Synthetic secrets only, matching tests/test_signing.py's own convention --
+never a real deployment JWT_SECRET.
+"""
+from audit.signing import sign_audit_event
+from consumers.processor import classify_event_integrity
+
+SECRET = "synthetic-test-secret-do-not-use-in-prod"
+OTHER_SECRET = "a-different-synthetic-secret"
+SERVICE = "tes"
+DATA = '{"event_id":"e1","service":"tes","action":"submit"}'
+
+
+# ---------------------------------------------------------------------------
+# valid / wrong / malformed / missing / empty
+# ---------------------------------------------------------------------------
+
+def test_valid_signature_classifies_as_valid():
+    sig = sign_audit_event(SERVICE, DATA, SECRET)
+    assert classify_event_integrity(SERVICE, sig, DATA, SECRET) == "valid"
+
+
+def test_wrong_secret_classifies_as_invalid():
+    sig = sign_audit_event(SERVICE, DATA, SECRET)
+    assert classify_event_integrity(SERVICE, sig, DATA, OTHER_SECRET) == "invalid"
+
+
+def test_tampered_data_classifies_as_invalid():
+    sig = sign_audit_event(SERVICE, DATA, SECRET)
+    tampered = DATA.replace("submit", "delete_all")
+    assert classify_event_integrity(SERVICE, sig, tampered, SECRET) == "invalid"
+
+
+def test_malformed_signature_classifies_as_invalid():
+    assert classify_event_integrity(SERVICE, "not-a-real-signature", DATA, SECRET) == "invalid"
+
+
+def test_missing_signature_classifies_as_unsigned():
+    assert classify_event_integrity(SERVICE, None, DATA, SECRET) == "unsigned"
+
+
+def test_empty_signature_classifies_as_unsigned():
+    assert classify_event_integrity(SERVICE, "", DATA, SECRET) == "unsigned"
+
+
+# ---------------------------------------------------------------------------
+# The missing-signature check happens BEFORE verify_audit_event() -- proven,
+# not just asserted, by using a secret that would make ANY real verification
+# fail (including of a well-formed-but-absent signature check), yet the
+# missing/empty cases above still return "unsigned", never "invalid".
+# Explicit ordering check: a None/"" signature must short-circuit even when
+# it could theoretically have been run through verify_audit_event() and
+# returned False (which would have produced the wrong classification,
+# "invalid" instead of "unsigned").
+# ---------------------------------------------------------------------------
+
+def test_missing_signature_is_unsigned_not_invalid_even_with_wrong_secret():
+    assert classify_event_integrity(SERVICE, None, DATA, OTHER_SECRET) == "unsigned"
+
+
+# ---------------------------------------------------------------------------
+# service/signature/data mismatch combinations
+# ---------------------------------------------------------------------------
+
+def test_signature_for_different_service_classifies_as_invalid():
+    sig = sign_audit_event("workflow-bundles", DATA, SECRET)
+    assert classify_event_integrity("auth-service", sig, DATA, SECRET) == "invalid"
+
+
+def test_signature_for_different_data_classifies_as_invalid():
+    other_data = '{"event_id":"e2","service":"tes","action":"delete"}'
+    sig = sign_audit_event(SERVICE, other_data, SECRET)
+    assert classify_event_integrity(SERVICE, sig, DATA, SECRET) == "invalid"
+
+
+# ---------------------------------------------------------------------------
+# exact raw JSON string passed through without reserialization -- reordered
+# keys (same logical content, different on-the-wire bytes) must invalidate
+# a signature made for the original ordering. This is the property that
+# makes "verify fields['data'] directly, never event.model_dump()" load-
+# bearing rather than incidental.
+# ---------------------------------------------------------------------------
+
+def test_reordered_json_keys_invalidate_the_original_signature():
+    original = '{"a": 1, "b": 2}'
+    reordered = '{"b": 2, "a": 1}'
+    sig = sign_audit_event(SERVICE, original, SECRET)
+    assert classify_event_integrity(SERVICE, sig, reordered, SECRET) == "invalid"
+
+
+def test_exact_original_string_still_classifies_as_valid():
+    """Sanity companion to the reordering test above -- confirms the
+    invalidation is specifically about the byte-level change, not some
+    unrelated break."""
+    sig = sign_audit_event(SERVICE, DATA, SECRET)
+    assert classify_event_integrity(SERVICE, sig, DATA, SECRET) == "valid"
+
+
+# ---------------------------------------------------------------------------
+# Return type/value contract: exactly one of the three strings, nothing else
+# ---------------------------------------------------------------------------
+
+def test_return_value_is_always_one_of_the_three_literal_strings():
+    sig = sign_audit_event(SERVICE, DATA, SECRET)
+    for signature, secret, expected in [
+        (sig, SECRET, "valid"),
+        (sig, OTHER_SECRET, "invalid"),
+        (None, SECRET, "unsigned"),
+    ]:
+        result = classify_event_integrity(SERVICE, signature, DATA, secret)
+        assert result in {"valid", "invalid", "unsigned"}
+        assert result == expected

tests/test_migrations.py

@@ -13,6 +13,7 @@
 EXPECTED_COLUMNS = {
     "event_id", "timestamp", "service", "event_type", "user_id", "action",
     "resource", "decision", "reason", "trace_id", "context", "created_at",
+    "integrity_status",
 }
 
 
@@ -65,3 +66,76 @@ def test_downgrade_drops_audit_events_table(tmp_path):
     engine = create_engine(db_url)
     inspector = inspect(engine)
     assert "audit_events" not in inspector.get_table_names()
+
+
+# ---------------------------------------------------------------------------
+# PR2: 0002_integrity_status
+# ---------------------------------------------------------------------------
+
+def test_integrity_status_column_exists_after_upgrade(tmp_path):
+    db_file = tmp_path / "migration_test.db"
+    db_url = f"sqlite:///{db_file}"
+
+    cfg = _alembic_config(db_url)
+    command.upgrade(cfg, "head")
+
+    engine = create_engine(db_url)
+    inspector = inspect(engine)
+    columns = {c["name"]: c for c in inspector.get_columns("audit_events")}
+    assert "integrity_status" in columns
+    assert columns["integrity_status"]["nullable"] is False
+
+
+def test_existing_rows_backfill_to_unsigned_on_upgrade(tmp_path):
+    """The exact PR2 migration guarantee: a row written under 0001 (before
+    integrity_status existed at all) must read back as "unsigned" after
+    upgrading to head -- via the column's own server_default, not a
+    manual backfill script."""
+    db_file = tmp_path / "migration_test.db"
+    db_url = f"sqlite:///{db_file}"
+
+    cfg = _alembic_config(db_url)
+    command.upgrade(cfg, "0001_audit_events")
+
+    engine = create_engine(db_url)
+    with engine.begin() as conn:
+        from sqlalchemy import text
+
+        conn.execute(
+            text(
+                "INSERT INTO audit_events "
+                "(event_id, timestamp, service, event_type, action, context) "
+                "VALUES ('pre-0002-evt', '2026-01-01 00:00:00', 'svc', 'test', '', '{}')"
+            )
+        )
+
+    command.upgrade(cfg, "head")
+
+    with engine.begin() as conn:
+        from sqlalchemy import text
+
+        row = conn.execute(
+            text("SELECT integrity_status FROM audit_events WHERE event_id = 'pre-0002-evt'")
+        ).fetchone()
+    assert row is not None
+    assert row[0] == "unsigned"
+
+
+def test_downgrade_to_0001_removes_integrity_status_column_only(tmp_path):
+    """Distinct from test_downgrade_drops_audit_events_table above (which
+    downgrades all the way to "base", dropping the whole table) -- this
+    downgrades exactly one revision, proving 0002's own downgrade() drops
+    only the column it added, leaving the table and 0001's columns intact."""
+    db_file = tmp_path / "migration_test.db"
+    db_url = f"sqlite:///{db_file}"
+
+    cfg = _alembic_config(db_url)
+    command.upgrade(cfg, "head")
+    command.downgrade(cfg, "0001_audit_events")
+
+    engine = create_engine(db_url)
+    inspector = inspect(engine)
+    assert "audit_events" in inspector.get_table_names()
+    columns = {c["name"] for c in inspector.get_columns("audit_events")}
+    assert "integrity_status" not in columns
+    assert "event_id" in columns  # 0001's own columns untouched

tests/test_sink.py

@@ -75,3 +75,43 @@ def test_sink_write_handles_optional_fields_missing(db_session):
     fetched = db_session.get(AuditEventRecord, "evt-minimal")
     assert fetched.user_id is None
     assert fetched.context == {}
+
+
+# ---------------------------------------------------------------------------
+# PR2: integrity_status -- additive, existing callers above are unaffected
+# and unmodified (none of them pass this key).
+# ---------------------------------------------------------------------------
+
+def test_sink_write_persists_explicit_valid_status(db_session):
+    sink = Sink(db_session)
+    sink.write(_event(integrity_status="valid"))
+
+    fetched = db_session.get(AuditEventRecord, "evt-1")
+    assert fetched.integrity_status == "valid"
+
+
+def test_sink_write_persists_explicit_invalid_status(db_session):
+    sink = Sink(db_session)
+    sink.write(_event(integrity_status="invalid"))
+
+    fetched = db_session.get(AuditEventRecord, "evt-1")
+    assert fetched.integrity_status == "invalid"
+
+
+def test_sink_write_persists_explicit_unsigned_status(db_session):
+    sink = Sink(db_session)
+    sink.write(_event(integrity_status="unsigned"))
+
+    fetched = db_session.get(AuditEventRecord, "evt-1")
+    assert fetched.integrity_status == "unsigned"
+
+
+def test_sink_write_omitted_status_defaults_to_unsigned(db_session):
+    """No existing caller (this file's own earlier tests, worker/main.py
+    before PR2, test_producer_contract_reconciliation.py) passes this key
+    -- must default safely rather than KeyError or persist None."""
+    sink = Sink(db_session)
+    sink.write(_event())  # no integrity_status key at all
+
+    fetched = db_session.get(AuditEventRecord, "evt-1")
+    assert fetched.integrity_status == "unsigned"