security: sign this service's own native audit:events payloads (HIPAA PR3b)

audit/logger.py

@@ -1,8 +1,10 @@
 import json
+
 import redis.asyncio as redis
-from typing import Optional
+
 from audit.config import AuditConfig
 from audit.models import AuditEvent
+from audit.signing import sign_audit_event
 
 
 class AuditLogger:
@@ -16,12 +18,30 @@ async def log(self, event: AuditEvent):
             # so json.dumps below never sees a raw datetime.
             payload = event.model_dump(mode="json")
 
+            # HIPAA PR3b: this is the one producer that lives in the same
+            # repo as the consumer/audit.signing itself, so unlike every
+            # other producer (api-gateway, tes, workflow-bundles,
+            # control-center, rag -- all separate deployables that
+            # hand-port sign_audit_event), this one imports it directly --
+            # no drift risk, no parallel copy to keep in sync.
+            #
+            # data is computed exactly once and that exact string is both
+            # signed and published, matching every other producer's
+            # contract. service comes from the already-serialized payload
+            # (not event.service) so a signature always covers the
+            # identity actually present in the signed bytes.
+            data = json.dumps(payload)
+            fields = {"data": data}
+            service = payload.get("service")
+            if service:
+                fields["sig"] = sign_audit_event(service, data, AuditConfig.EVENT_SIGNING_SECRET)
+
             await self.redis.xadd(
                 AuditConfig.STREAM_NAME,
-                {"data": json.dumps(payload)},
+                fields,
                 maxlen=AuditConfig.MAX_STREAM_LENGTH,
                 approximate=True,
             )
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001 -- intentional fire-and-forget: audit logging must NEVER break core system
             # NEVER break core system
             print(f"[AUDIT ERROR] {e}")

tests/test_logger_signing.py

@@ -0,0 +1,126 @@
+"""HIPAA PR3b: producer-side signing for AuditLogger.log() -- this
+repo's own native audit:events producer, used by api/routes_audit.py.
+
+Unlike every other producer in the platform (api-gateway, tes,
+workflow-bundles, control-center, rag -- all separate deployables that
+hand-port sign_audit_event()), this one lives in the same repo as
+audit.signing itself and imports it directly: no drift risk, no parallel
+copy to keep in sync. These tests exercise AuditLogger.log()'s new
+signing behavior specifically; tests/test_signing.py already covers
+sign_audit_event/verify_audit_event in isolation and is untouched.
+"""
+import json
+from unittest.mock import AsyncMock
+
+import pytest
+
+from audit.config import AuditConfig
+from audit.models import AuditEvent
+from audit.signing import verify_audit_event
+
+
+@pytest.mark.asyncio
+async def test_log_signs_the_exact_data_string_it_publishes(audit_logger, monkeypatch):
+    logger, mock_redis = audit_logger
+    mock_redis.xadd = AsyncMock()
+    monkeypatch.setattr(AuditConfig, "EVENT_SIGNING_SECRET", "s3cr3t")
+
+    event = AuditEvent(service="security-audit", event_type="platform_admin_action")
+    await logger.log(event)
+
+    fields = mock_redis.xadd.call_args[0][1]
+    data = fields["data"]
+    sig = fields["sig"]
+    assert verify_audit_event("security-audit", data, sig, "s3cr3t") is True
+    assert json.loads(data)["event_id"] == event.event_id
+
+
+@pytest.mark.asyncio
+async def test_log_includes_both_data_and_sig_fields(audit_logger, monkeypatch):
+    logger, mock_redis = audit_logger
+    mock_redis.xadd = AsyncMock()
+    monkeypatch.setattr(AuditConfig, "EVENT_SIGNING_SECRET", "s3cr3t")
+
+    await logger.log(AuditEvent(service="security-audit", event_type="platform_admin_action"))
+
+    fields = mock_redis.xadd.call_args[0][1]
+    assert "data" in fields
+    assert fields["sig"].startswith("v1:")
+
+
+@pytest.mark.asyncio
+async def test_log_signature_does_not_verify_under_a_different_secret(audit_logger, monkeypatch):
+    logger, mock_redis = audit_logger
+    mock_redis.xadd = AsyncMock()
+    monkeypatch.setattr(AuditConfig, "EVENT_SIGNING_SECRET", "s3cr3t")
+
+    await logger.log(AuditEvent(service="security-audit", event_type="platform_admin_action"))
+
+    fields = mock_redis.xadd.call_args[0][1]
+    assert verify_audit_event("security-audit", fields["data"], fields["sig"], "wrong-secret") is False
+
+
+@pytest.mark.asyncio
+async def test_log_tampered_data_fails_verification(audit_logger, monkeypatch):
+    """Proves the signature is bound to this exact payload -- modifying
+    even one field after signing must invalidate it."""
+    logger, mock_redis = audit_logger
+    mock_redis.xadd = AsyncMock()
+    monkeypatch.setattr(AuditConfig, "EVENT_SIGNING_SECRET", "s3cr3t")
+
+    await logger.log(AuditEvent(service="security-audit", event_type="platform_admin_action", decision="allow"))
+
+    fields = mock_redis.xadd.call_args[0][1]
+    tampered = fields["data"].replace('"allow"', '"deny"')
+    assert verify_audit_event("security-audit", tampered, fields["sig"], "s3cr3t") is False
+    # the untampered original still verifies -- proves the failure above
+    # is specifically about the tampering, not a broken signature.
+    assert verify_audit_event("security-audit", fields["data"], fields["sig"], "s3cr3t") is True
+
+
+@pytest.mark.asyncio
+async def test_log_without_a_service_still_publishes_unsigned(audit_logger, monkeypatch):
+    """AuditEvent.service is a required, non-empty field on the real
+    model, so this can only happen via a MagicMock stand-in whose
+    model_dump() omits it -- still must not silently drop the event."""
+    from unittest.mock import MagicMock
+
+    logger, mock_redis = audit_logger
+    mock_redis.xadd = AsyncMock()
+    monkeypatch.setattr(AuditConfig, "EVENT_SIGNING_SECRET", "s3cr3t")
+
+    fake_event = MagicMock()
+    fake_event.model_dump.return_value = {
+        "event_id": "e1", "timestamp": "2024-01-01T00:00:00", "event_type": "x",
+        "user_id": None, "action": "", "resource": None, "decision": None,
+        "reason": None, "trace_id": None, "context": {},
+    }  # no "service" key
+    await logger.log(fake_event)
+
+    fields = mock_redis.xadd.call_args[0][1]
+    assert "data" in fields
+    assert "sig" not in fields
+
+
+@pytest.mark.asyncio
+async def test_log_exception_never_leaks_the_secret(audit_logger, monkeypatch, capsys):
+    logger, mock_redis = audit_logger
+    mock_redis.xadd = AsyncMock(side_effect=RuntimeError("boom"))
+    monkeypatch.setattr(AuditConfig, "EVENT_SIGNING_SECRET", "super-secret-value")
+
+    await logger.log(
+        AuditEvent(service="security-audit", event_type="platform_admin_action")
+    )  # must not raise
+
+    captured = capsys.readouterr()
+    assert "super-secret-value" not in captured.out
+    assert "super-secret-value" not in captured.err
+
+
+def test_config_signing_secret_is_the_one_source_of_truth():
+    """AuditConfig.EVENT_SIGNING_SECRET (audit/config.py) was already
+    defined by PR2 specifically for this eventual purpose -- signing
+    imports it directly rather than reading JWT_SECRET a second time."""
+    import os
+
+    assert AuditConfig.EVENT_SIGNING_SECRET == os.getenv("JWT_SECRET", "change-me")