diff --git a/alembic/versions/0002_integrity_status.py b/alembic/versions/0002_integrity_status.py new file mode 100644 index 0000000..d6870f7 --- /dev/null +++ b/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") diff --git a/audit/config.py b/audit/config.py index 0f11f06..efccffd 100644 --- a/audit/config.py +++ b/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()}") \ No newline at end of file + 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") \ No newline at end of file diff --git a/consumers/processor.py b/consumers/processor.py index 57ec4fc..128ebac 100644 --- a/consumers/processor.py +++ b/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) \ No newline at end of file + 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" \ No newline at end of file diff --git a/consumers/sink.py b/consumers/sink.py index 2b9b9c5..7818aab 100644 --- a/consumers/sink.py +++ b/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: diff --git a/db/models.py b/db/models.py index b7d582a..9c436ee 100644 --- a/db/models.py +++ b/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) diff --git a/tests/test_classify_event_integrity.py b/tests/test_classify_event_integrity.py new file mode 100644 index 0000000..2706e7a --- /dev/null +++ b/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 diff --git a/tests/test_migrations.py b/tests/test_migrations.py index fd39d4e..1f3d1be 100644 --- a/tests/test_migrations.py +++ b/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 diff --git a/tests/test_sink.py b/tests/test_sink.py index a1cc444..950a019 100644 --- a/tests/test_sink.py +++ b/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" diff --git a/tests/test_worker.py b/tests/test_worker.py index c166561..be7c79a 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -1,10 +1,15 @@ """PR4.2 regression tests: worker/main.py -- the Redis Streams consumer-group loop that reads audit:events, parses/persists each message, and only ACKs -after a successful DB write.""" +after a successful DB write. + +PR2 additions (bottom of file): integrity_status classification -- signed +valid/invalid events and unsigned (today's only real traffic shape) all +persist and ACK; only "invalid" gets the distinct observability print.""" import json from unittest.mock import MagicMock, patch import worker.main as worker +from audit.signing import sign_audit_event def _raw(event_id="evt-1"): @@ -227,3 +232,149 @@ def test_run_does_not_swallow_keyboard_interrupt(): raised = True assert raised is True + + +# --------------------------------------------------------------------------- +# PR2: integrity_status classification (consumers/processor.py:: +# classify_event_integrity), wired into handle_message() +# --------------------------------------------------------------------------- + +SECRET = "synthetic-worker-test-secret" + + +def _handle_with_status(fields, secret=SECRET): + """Runs handle_message() with EVENT_SIGNING_SECRET fixed to a known + synthetic value and SessionLocal/Sink mocked, returning + (handle_result, integrity_status_written_to_sink).""" + reader = MagicMock() + mock_sink_instance = MagicMock() + mock_sink_instance.write.return_value = True + + with patch("worker.main.SessionLocal") as mock_session_local, \ + patch("worker.main.Sink", return_value=mock_sink_instance), \ + patch.object(worker.AuditConfig, "EVENT_SIGNING_SECRET", secret): + mock_session_local.return_value = MagicMock() + result = worker.handle_message(reader, "1-0", fields) + + written = mock_sink_instance.write.call_args[0][0] if mock_sink_instance.write.called else None + status = written["integrity_status"] if written else None + return result, status, reader + + +def test_valid_signed_event_persists_as_valid_and_acks(): + raw = _raw(event_id="evt-valid") + sig = sign_audit_event("auth", raw, SECRET) + + result, status, reader = _handle_with_status({"data": raw, "sig": sig}) + + assert result is True + assert status == "valid" + reader.ack.assert_called_once_with("1-0") + + +def test_unsigned_event_persists_as_unsigned_and_acks(): + """Today's only real traffic shape -- no sig field at all.""" + raw = _raw(event_id="evt-unsigned") + + result, status, reader = _handle_with_status({"data": raw}) + + assert result is True + assert status == "unsigned" + reader.ack.assert_called_once_with("1-0") + + +def test_invalid_signature_persists_as_invalid_and_acks(): + raw = _raw(event_id="evt-invalid") + sig = sign_audit_event("auth", raw, "a-different-secret-entirely") + + result, status, reader = _handle_with_status({"data": raw, "sig": sig}) + + assert result is True + assert status == "invalid" + reader.ack.assert_called_once_with("1-0") + + +def test_malformed_signature_persists_as_invalid_and_acks(): + raw = _raw(event_id="evt-malformed-sig") + + result, status, reader = _handle_with_status({"data": raw, "sig": "not-a-real-signature"}) + + assert result is True + assert status == "invalid" + reader.ack.assert_called_once_with("1-0") + + +def test_invalid_signature_emits_distinct_observability_message(capsys): + raw = _raw(event_id="evt-loud-invalid") + sig = sign_audit_event("auth", raw, "wrong-secret") + + _handle_with_status({"data": raw, "sig": sig}) + + captured = capsys.readouterr() + assert "SIGNATURE INVALID" in captured.out + assert "evt-loud-invalid" in captured.out + + +def test_valid_signature_does_not_emit_the_invalid_message(capsys): + raw = _raw(event_id="evt-quiet-valid") + sig = sign_audit_event("auth", raw, SECRET) + + _handle_with_status({"data": raw, "sig": sig}) + + captured = capsys.readouterr() + assert "SIGNATURE INVALID" not in captured.out + + +def test_unsigned_event_does_not_emit_the_invalid_message(capsys): + raw = _raw(event_id="evt-quiet-unsigned") + + _handle_with_status({"data": raw}) + + captured = capsys.readouterr() + assert "SIGNATURE INVALID" not in captured.out + + +def test_invalid_signature_message_does_not_print_the_secret_or_signature(): + """Security check: the observability print must never leak the secret + or the signature/data payload -- only identifying metadata.""" + raw = _raw(event_id="evt-secret-check") + sig = sign_audit_event("auth", raw, "wrong-secret") + + import io + from contextlib import redirect_stdout + + buf = io.StringIO() + with redirect_stdout(buf): + _handle_with_status({"data": raw, "sig": sig}) + + output = buf.getvalue() + assert SECRET not in output + assert "wrong-secret" not in output + assert sig not in output + + +def test_malformed_json_still_does_not_ack_unchanged_behavior(): + """Parse/schema failures remain retriable -- integrity classification + must never run on data that failed to parse at all.""" + result, status, reader = _handle_with_status({"data": "not-json"}) + + assert result is False + assert status is None + reader.ack.assert_not_called() + + +def test_db_failure_still_does_not_ack_unchanged_behavior(): + reader = MagicMock() + mock_sink_instance = MagicMock() + mock_sink_instance.write.side_effect = Exception("db connection lost") + raw = _raw(event_id="evt-db-fail") + + with patch("worker.main.SessionLocal") as mock_session_local, \ + patch("worker.main.Sink", return_value=mock_sink_instance), \ + patch.object(worker.AuditConfig, "EVENT_SIGNING_SECRET", SECRET): + mock_db = MagicMock() + mock_session_local.return_value = mock_db + result = worker.handle_message(reader, "1-0", {"data": raw}) + + assert result is False + reader.ack.assert_not_called() diff --git a/tests/test_worker_integration_real_backends.py b/tests/test_worker_integration_real_backends.py index 59e3829..9ad2307 100644 --- a/tests/test_worker_integration_real_backends.py +++ b/tests/test_worker_integration_real_backends.py @@ -20,6 +20,7 @@ PR -- see the B0 report for why that's a deliberate, separately-flagged follow-up rather than bundled into this change. """ +import json import os import uuid @@ -187,6 +188,124 @@ def test_real_produce_consume_persist_ack_round_trip(real_redis_stream, real_mys assert pending["pending"] == 0 +def _integration_payload(event_id, service="b0-integration-test", **overrides): + from datetime import datetime, timezone + + payload = { + "event_id": event_id, + "timestamp": datetime.now(timezone.utc).isoformat(), + "service": service, + "event_type": "test", + "user_id": "test-user", + "action": "pr2_integration_smoke", + "resource": None, + "decision": "success", + "reason": None, + "trace_id": "pr2-trace-1", + "context": {}, + } + payload.update(overrides) + return payload + + +def _run_real_round_trip(real_redis_stream, real_mysql_url, monkeypatch, event_id, build_fields): + """Shared plumbing for the three PR2 integration tests below -- + identical chain to test_real_produce_consume_persist_ack_round_trip + above (real XADD -> real consumer-group read -> worker.handle_message + -> real MySQL -> XACK). + + `build_fields(raw_data) -> dict` decides the extra Redis fields (a + `sig`, or none) for this specific raw_data string -- taking a callback + rather than a precomputed `sig` guarantees any signature is always + computed against the exact bytes that get XADDed, not a separately + (and therefore mismatched) constructed copy. + """ + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + + import worker.main as worker_module + + engine = create_engine(real_mysql_url) + TestSessionLocal = sessionmaker(bind=engine) + monkeypatch.setattr(worker_module, "SessionLocal", TestSessionLocal) + + raw_data = json.dumps(_integration_payload(event_id)) + fields = {"data": raw_data, **build_fields(raw_data)} + real_redis_stream.redis.xadd(TEST_STREAM, fields) + + response = real_redis_stream.read_group(consumer_name="pr2-test-consumer", block=3000) + assert response, "expected the real event just XADDed to be delivered" + + acked = False + for _stream_name, messages in response: + for message_id, delivered_fields in messages: + result = worker_module.handle_message(real_redis_stream, message_id, delivered_fields) + assert result is True + acked = True + assert acked + + with TestSessionLocal() as session: + from db.models import AuditEventRecord + + row = session.get(AuditEventRecord, event_id) + assert row is not None + return row + + +# --------------------------------------------------------------------------- +# PR2: integrity_status through the real Redis -> worker -> MySQL chain. +# +# Signs with whatever AuditConfig.EVENT_SIGNING_SECRET actually resolves to +# in *this* process/environment -- worker.main.handle_message() (called +# inside _run_real_round_trip, same process) reads the same AuditConfig, +# so this is correct regardless of whether this environment has the real +# platform JWT_SECRET set or is falling back to "change-me". Never assumes +# which -- that is the entire point of reading it at test time rather than +# hard-coding a value. +# --------------------------------------------------------------------------- + +def test_real_valid_signed_event_persists_as_valid(real_redis_stream, real_mysql_url, monkeypatch): + from audit.config import AuditConfig + from audit.signing import sign_audit_event + + event_id = f"pr2-valid-{uuid.uuid4()}" + row = _run_real_round_trip( + real_redis_stream, real_mysql_url, monkeypatch, + event_id=event_id, + build_fields=lambda raw_data: { + "sig": sign_audit_event("b0-integration-test", raw_data, AuditConfig.EVENT_SIGNING_SECRET) + }, + ) + assert row.integrity_status == "valid" + + +def test_real_unsigned_event_persists_as_unsigned(real_redis_stream, real_mysql_url, monkeypatch): + event_id = f"pr2-unsigned-{uuid.uuid4()}" + row = _run_real_round_trip( + real_redis_stream, real_mysql_url, monkeypatch, + event_id=event_id, + build_fields=lambda raw_data: {}, # no sig field -- today's actual production traffic shape + ) + assert row.integrity_status == "unsigned" + + +def test_real_invalid_signed_event_persists_as_invalid(real_redis_stream, real_mysql_url, monkeypatch): + from audit.config import AuditConfig + from audit.signing import sign_audit_event + + event_id = f"pr2-invalid-{uuid.uuid4()}" + row = _run_real_round_trip( + real_redis_stream, real_mysql_url, monkeypatch, + event_id=event_id, + # Well-formed v1 signature, correct secret, WRONG service -- signed + # for a different producer identity than the event actually claims. + build_fields=lambda raw_data: { + "sig": sign_audit_event("some-other-service", raw_data, AuditConfig.EVENT_SIGNING_SECRET) + }, + ) + assert row.integrity_status == "invalid" + + def test_real_duplicate_delivery_does_not_duplicate_row(real_mysql_url): """Sink.write()'s IntegrityError-swallow behavior (consumers/sink.py) against a real MySQL unique-constraint violation, not SQLite's -- the diff --git a/worker/main.py b/worker/main.py index 9659762..e9aefeb 100644 --- a/worker/main.py +++ b/worker/main.py @@ -12,7 +12,7 @@ from redis.exceptions import TimeoutError as RedisTimeoutError from audit.config import AuditConfig -from consumers.processor import parse_audit_event +from consumers.processor import classify_event_integrity, parse_audit_event from consumers.sink import Sink from consumers.stream_reader import StreamReader from db.session import SessionLocal @@ -21,20 +21,47 @@ def handle_message(reader: StreamReader, message_id: str, fields: dict) -> bool: """Process one Redis Streams message. Returns True if it was acked. - Deserialize -> validate -> persist -> ack, in that order, and only ack + Deserialize -> classify -> persist -> ack, in that order, and only ack after a successful DB commit. Any failure along the way leaves the message unacknowledged in the consumer group's pending entries list so it can be retried -- it is never dropped. + + PR2: classification (valid/invalid/unsigned -- see consumers/ + processor.py::classify_event_integrity) is deliberately NOT another + reason to leave a message unacked. Parse/DB failures above are retried + because they may be transient or fixable; a signature is not -- it is + either right or wrong forever, so retrying it accomplishes nothing and + would only accumulate a permanent poison-pill pending entry. Every + classification is persisted (as durable evidence, including forged + attempts) and acked, exactly like today's fully-unsigned traffic. + Uses fields["data"] itself, not a re-serialization of `event` -- the + signature covers the exact transmitted bytes (see audit/signing.py). """ + raw_data = fields["data"] try: - event = parse_audit_event(fields["data"]) + event = parse_audit_event(raw_data) except Exception as e: print(f"[WORKER] failed to parse message {message_id}: {e}") return False + integrity_status = classify_event_integrity( + event.service, fields.get("sig"), raw_data, AuditConfig.EVENT_SIGNING_SECRET + ) + if integrity_status == "invalid": + # Distinct from the plain print() convention elsewhere in this + # file: this is not an infra blip, it's a signature that failed + # verification -- worth being loudly, separately visible. + print( + f"[WORKER] SIGNATURE INVALID for message {message_id} " + f"(service={event.service!r}, event_id={event.event_id!r}) -- " + f"persisting as durable evidence, not trusting the payload" + ) + db = SessionLocal() try: - Sink(db).write(event.model_dump()) + payload = event.model_dump() + payload["integrity_status"] = integrity_status + Sink(db).write(payload) except Exception as e: print(f"[WORKER] failed to persist message {message_id}: {e}") return False