feat(security-audit): add first-class audit tenant contract

alembic/versions/0003_tenant_contract.py

@@ -0,0 +1,35 @@
+"""add first-class tenant provenance to audit_events
+
+SAT-1: organization_id is nullable for backward compatibility. Existing
+rows are explicitly classified as unknown, never global. Producers may mark
+an event organization-scoped or explicitly global in tenant_scope.
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "0003_tenant_contract"
+down_revision: str | None = "0002_integrity_status"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column("audit_events", sa.Column("organization_id", sa.String(length=255), nullable=True))
+    op.add_column(
+        "audit_events",
+        sa.Column("tenant_scope", sa.String(length=16), nullable=False, server_default="unknown"),
+    )
+    op.create_index(
+        "ix_audit_events_org_timestamp_event",
+        "audit_events",
+        ["organization_id", "timestamp", "event_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_audit_events_org_timestamp_event", table_name="audit_events")
+    op.drop_column("audit_events", "tenant_scope")
+    op.drop_column("audit_events", "organization_id")

audit/decorators.py

@@ -1,9 +1,9 @@
 import functools
-import asyncio
+
+from audit.config import AuditConfig
+from audit.context import get_identity, get_trace_id, get_user_id
 from audit.logger import AuditLogger
 from audit.models import AuditEvent
-from audit.config import AuditConfig
-from audit.context import get_trace_id, get_user_id, get_identity
 
 logger = AuditLogger()
 
@@ -26,6 +26,8 @@ async def async_inner(*args, **kwargs):
                     service=AuditConfig.SERVICE_NAME,
                     event_type=event_type,
                     user_id=get_user_id(),
+                    organization_id=str(identity.org_id) if identity and identity.org_id is not None else None,
+                    tenant_scope="organization" if identity and identity.org_id is not None else "unknown",
                     action=action,
                     decision="success",
                     trace_id=get_trace_id(),
@@ -37,4 +39,4 @@ async def async_inner(*args, **kwargs):
 
         return async_inner
 
-    return wrapper
+    return wrapper

audit/models.py

@@ -1,7 +1,8 @@
-from pydantic import BaseModel, Field
-from typing import Optional, Dict, Any
-from datetime import datetime
 import uuid
+from datetime import datetime
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field, model_validator
 
 
 class AuditEvent(BaseModel):
@@ -14,12 +15,30 @@ class AuditEvent(BaseModel):
     service: str
     event_type: str   # auth, iam, policy, tes
 
-    user_id: Optional[str] = None
+    user_id: str | None = None
+    # First-class tenant provenance. This is intentionally separate from the
+    # arbitrary context object: context is never an authority for tenant
+    # isolation. Legacy payloads default to UNKNOWN, not GLOBAL.
+    organization_id: str | None = None
+    tenant_scope: Literal["organization", "global", "unknown"] = "unknown"
     action: str = ""
-    resource: Optional[str] = None
+    resource: str | None = None
+
+    decision: str | None = None  # allow / deny / success / fail
+    reason: str | None = None
 
-    decision: Optional[str] = None  # allow / deny / success / fail
-    reason: Optional[str] = None
+    trace_id: str | None = None
+    context: dict[str, Any] = Field(default_factory=dict)
 
-    trace_id: Optional[str] = None
-    context: Dict[str, Any] = {}
+    @model_validator(mode="after")
+    def validate_tenant_scope(self) -> "AuditEvent":
+        # Supplying a first-class organization ID is itself an authoritative
+        # organization-scoped declaration. The explicit discriminator is
+        # still emitted so readers can distinguish legacy UNKNOWN from GLOBAL.
+        if self.organization_id is not None and self.tenant_scope == "unknown":
+            self.tenant_scope = "organization"
+        if self.tenant_scope == "organization" and self.organization_id is None:
+            raise ValueError("organization tenant scope requires organization_id")
+        if self.tenant_scope != "organization" and self.organization_id is not None:
+            raise ValueError("organization_id requires organization tenant scope")
+        return self

consumers/sink.py

@@ -28,6 +28,8 @@ def write(self, event: dict) -> bool:
             service=event["service"],
             event_type=event["event_type"],
             user_id=event.get("user_id"),
+            organization_id=event.get("organization_id"),
+            tenant_scope=event.get("tenant_scope", "unknown"),
             action=event.get("action", ""),
             resource=event.get("resource"),
             decision=event.get("decision"),

db/models.py

@@ -1,4 +1,4 @@
-from sqlalchemy import JSON, Column, DateTime, String, Text
+from sqlalchemy import JSON, Column, DateTime, Index, String, Text
 from sqlalchemy.sql import func
 
 from db.base import Base
@@ -22,6 +22,10 @@ class AuditEventRecord(Base):
     service = Column(String(255), nullable=False)
     event_type = Column(String(255), nullable=False)
     user_id = Column(String(255), nullable=True)
+    organization_id = Column(String(255), nullable=True)
+    # Legacy rows and events without an authoritative tenant are UNKNOWN.
+    # GLOBAL is only valid when a producer explicitly declares it.
+    tenant_scope = Column(String(16), nullable=False, server_default="unknown")
     action = Column(String(255), nullable=False, default="")
     resource = Column(String(255), nullable=True)
     decision = Column(String(64), nullable=True)
@@ -37,3 +41,7 @@ class AuditEventRecord(Base):
     # 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)
+
+    __table_args__ = (
+        Index("ix_audit_events_org_timestamp_event", "organization_id", "timestamp", "event_id"),
+    )

docs/AUDIT_TENANT_CONTRACT_SAT1.md

@@ -0,0 +1,81 @@
+# SAT-1 — Security Audit Tenant Contract
+
+Status: **implemented; ready for AE-2 authorization work**
+
+## Current lifecycle
+
+Producers create an `AuditEvent`, serialize it once, sign that exact JSON
+string, and publish it to Redis `audit:events`. The worker parses the payload,
+classifies integrity, and persists it to SQL `audit_events`. `GET /audit/events`
+reads the durable table. Redis remains ingestion/backlog, not query history.
+
+## Canonical tenant contract
+
+`organization_id: string | null` is now a first-class event-envelope and
+durable column. It is separate from `context` and is covered by existing
+signing because the logger signs the complete serialized envelope.
+
+`tenant_scope` distinguishes:
+
+| Value | Meaning |
+|---|---|
+| `organization` | `organization_id` is authoritative and present |
+| `global` | Producer explicitly declares a platform/global event |
+| `unknown` | Tenant is unavailable or event predates SAT-1 |
+
+Null `organization_id` never means global. Legacy payloads default to
+`unknown`; the consumer does not infer tenant identity from context, resources,
+URLs, users, traces, or service names.
+
+## Producer coverage matrix
+
+| Producer | Authoritative tenant | Current propagation | Required future change | Risk |
+|---|---|---|---|---|
+| Security Audit native logger/test route | No for platform smoke events; field supported | Envelope model/logger | Explicitly mark global only when deliberate | Low |
+| API Gateway | Verified user/org context exists | Builder omits field | Pass verified org ID | Medium |
+| RAG | Verified IAM `UserContext.org_id` | Context in several paths | Pass top-level field | Medium |
+| TES | Organization context on authenticated runs | Older producer contract | Add verified field | Medium |
+| Workflow Bundles | Varies by path | Producer-specific context | Establish verified propagation | High |
+| LIMS | Request identity varies | Producer-specific context | Add only when verified | High |
+| Model Registry | Verified `UserContext.org_id` | Producer-specific clients | Pass explicit field | Medium |
+| Control Center/other producers | Varies | Producer-specific payloads | Adopt selectively | Medium |
+
+No upstream repositories were modified. Old producers remain readable and
+become `tenant_scope=unknown`.
+
+## Storage and query preparation
+
+Migration `0003_tenant_contract` adds nullable `organization_id` and non-null
+`tenant_scope` with an `unknown` server default. Existing rows are not
+rewritten. Index `ix_audit_events_org_timestamp_event` supports future bounded
+organization/time/ID queries. The query service accepts an internal
+`organization_id` filter, but the existing HTTP route remains platform-admin
+only; SAT-1 does not expose organization-scoped access or trust a query-string
+tenant ID.
+
+## Authorization boundary
+
+Platform-wide access remains verified JWT plus `platform_admin`. Organization-
+scoped access is deferred until the authoritative permission and verified claim
+used to derive caller organization are specified and tested. A browser-supplied
+organization ID must never widen scope.
+
+## Integrity, metadata, retention, and freshness
+
+The tenant fields are included in signed JSON bytes; verification is unchanged.
+Raw signatures and keys remain absent from responses. Legacy arbitrary `context`
+is retained for compatibility but is not tenant identity and needs an explicit
+allowlist before Audit Explorer exposes it. Credentials, tokens, cookies, JWTs,
+authorization headers, API keys, signing material, environment values, private
+paths, and unrestricted context are excluded from the future safe response.
+
+Redis max length is a backlog cap, not retention. Durable retention,
+freshness/as-of timestamps, source-unavailable responses, and safe metadata
+redaction remain undefined and are not invented here.
+
+## AE-2 prerequisites
+
+Expand producer coverage, then define verified caller scope, platform-wide
+override rules, unknown/global handling, bounded filters, safe metadata output,
+and unavailable/error behavior. Do not broaden the existing route based on this
+migration alone.

schemas/audit.py

@@ -16,6 +16,8 @@ class AuditEventOut(BaseModel):
     service: str
     event_type: str
     user_id: str | None = None
+    organization_id: str | None = None
+    tenant_scope: str
     action: str
     resource: str | None = None
     decision: str | None = None

services/audit_query_service.py

@@ -10,6 +10,7 @@ def list_audit_events(
     page: int,
     page_size: int,
     user_id: str | None = None,
+    organization_id: str | None = None,
     service: str | None = None,
     event_type: str | None = None,
     decision: str | None = None,
@@ -32,6 +33,8 @@ def list_audit_events(
 
     if user_id is not None:
         query = query.filter(AuditEventRecord.user_id == user_id)
+    if organization_id is not None:
+        query = query.filter(AuditEventRecord.organization_id == organization_id)
     if service is not None:
         query = query.filter(AuditEventRecord.service == service)
     if event_type is not None:

tests/test_audit_query_service.py

@@ -14,6 +14,7 @@ def _add(db_session, event_id, minutes_offset=0, **overrides):
         service=overrides.get("service", "auth"),
         event_type=overrides.get("event_type", "auth_login"),
         user_id=overrides.get("user_id", "u1"),
+        organization_id=overrides.get("organization_id"),
         action=overrides.get("action", "login"),
         resource=overrides.get("resource"),
         decision=overrides.get("decision", "success"),
@@ -43,6 +44,17 @@ def test_filters_by_user_id(db_session):
     assert [r.event_id for r in rows] == ["e1"]
 
 
+def test_filters_by_organization_id_without_authorizing_scope(db_session):
+    _add(db_session, "e1", organization_id="org-1")
+    _add(db_session, "e2", organization_id="org-2")
+    db_session.commit()
+    rows, total = audit_query_service.list_audit_events(
+        db_session, page=1, page_size=20, organization_id="org-1"
+    )
+    assert total == 1
+    assert [r.event_id for r in rows] == ["e1"]
+
+
 def test_filters_by_service(db_session):
     _add(db_session, "e1", service="auth")
     _add(db_session, "e2", service="policy")

tests/test_decorators.py

@@ -1,6 +1,8 @@
-import pytest
 from unittest.mock import AsyncMock, MagicMock, patch
-from audit.context import trace_id_var, user_id_var, identity_var
+
+import pytest
+
+from audit.context import identity_var, trace_id_var, user_id_var
 
 
 @pytest.fixture(autouse=True)
@@ -224,6 +226,8 @@ async def my_func():
             "verified": True,
         }
     }
+    assert event.organization_id == "7"
+    assert event.tenant_scope == "organization"
 
 
 @pytest.mark.asyncio

tests/test_logger.py

@@ -9,9 +9,11 @@
 exercise the real serializer construct a real AuditEvent.
 """
 import json
-import pytest
 from datetime import datetime
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
 from audit.config import AuditConfig
 from audit.models import AuditEvent
 
@@ -45,7 +47,7 @@ async def test_log_writes_to_redis_stream(audit_logger):
     logger, mock_redis = audit_logger
     mock_redis.xadd = AsyncMock()
 
-    event, payload = _serializable_event(service="auth", event_type="auth_login",
+    event, _payload = _serializable_event(service="auth", event_type="auth_login",
                                          user_id="u1", action="login", decision="success")
     await logger.log(event)
 
@@ -88,7 +90,7 @@ async def test_log_event_includes_all_fields(audit_logger):
     logger, mock_redis = audit_logger
     mock_redis.xadd = AsyncMock()
 
-    event, expected = _serializable_event(
+    event, _expected = _serializable_event(
         service="iam",
         event_type="iam_cache_hit",
         user_id="u2",
@@ -172,6 +174,36 @@ async def test_log_real_event_id_survives_serialization(audit_logger):
     assert stored["event_id"] == "fixed-id-123"
 
 
+@pytest.mark.asyncio
+async def test_log_serializes_first_class_tenant_in_signed_payload(audit_logger):
+    logger, mock_redis = audit_logger
+    mock_redis.xadd = AsyncMock()
+    await logger.log(AuditEvent(
+        service="tes", event_type="run", organization_id="org-7",
+        tenant_scope="organization",
+    ))
+    stored = json.loads(mock_redis.xadd.call_args[0][1]["data"])
+    assert stored["organization_id"] == "org-7"
+    assert stored["tenant_scope"] == "organization"
+
+
+@pytest.mark.asyncio
+async def test_tenant_field_is_covered_by_signature(audit_logger):
+    logger, mock_redis = audit_logger
+    mock_redis.xadd = AsyncMock()
+    await logger.log(AuditEvent(
+        service="tes", event_type="run", organization_id="org-7",
+        tenant_scope="organization",
+    ))
+    fields = mock_redis.xadd.call_args[0][1]
+    from audit.signing import verify_audit_event
+    assert verify_audit_event("tes", fields["data"], fields["sig"], AuditConfig.EVENT_SIGNING_SECRET)
+    assert not verify_audit_event(
+        "tes", fields["data"].replace('"org-7"', '"org-8"'), fields["sig"],
+        AuditConfig.EVENT_SIGNING_SECRET,
+    )
+
+
 # ---------------------------------------------------------------------------
 # Error path: failures must never propagate
 # ---------------------------------------------------------------------------