diff --git a/docs/tech-specs/audit-events.md b/docs/tech-specs/audit-events.md new file mode 100644 index 00000000..d988d7b0 --- /dev/null +++ b/docs/tech-specs/audit-events.md @@ -0,0 +1,402 @@ +--- +layout: default +title: "Audit Events Technical Specification" +parent: "Tech Specs" +--- + +# Audit Events Technical Specification + +## Overview + +This specification defines the audit event system for TrustGraph. +Audit events provide a structured, complete record of security- +relevant operations: API gateway invocations and IAM decisions. + +The design principle is: **emit everything, let consumers decide.** +Audit events are cheap to produce (a pub/sub message per operation) +and rich enough to support any downstream consumer — compliance +dashboards, SIEM integration, anomaly detection, billing metering, +or simple grep-based debugging. This spec covers event production +only. Storage, retention, alerting, and presentation are +deployment-specific concerns handled by consumers outside this +boundary. + +## Motivation + +TrustGraph currently has operational logging (Python `logging` to +stdout/Loki) but no structured audit trail. Operational logs are +unstructured, filtered by level, and designed for debugging — not +for answering "who did what, when, and was it allowed?" + +Enterprise deployments need: + +- **Compliance evidence** — demonstrable record of access for + auditors. +- **Incident investigation** — reconstruct what happened around a + security event. +- **Anomaly detection** — feed structured events into monitoring + systems. +- **Accountability** — attribute actions to identities across + workspaces. + +The current logging infrastructure cannot serve these needs because +it is unstructured, inconsistently formatted, and interleaves +debug noise with security-relevant signals. + +## Design Principles + +- **Complete.** Every gateway request and every IAM decision emits + an event. No sampling, no level-gating. The pub/sub cost is + negligible; consumers filter what they need. + +- **Structured.** Events are typed, versioned, machine-parseable + JSON objects with a fixed envelope and operation-specific payloads. + No free-text messages. + +- **Cheap to produce.** Events land on a pub/sub topic. No + synchronous writes, no blocking on consumer availability. If no + consumer is subscribed, events are discarded by the broker — that + is acceptable. + +- **Rich.** Events carry enough context to reconstruct the full + security narrative without correlating against operational logs. + Identity, workspace, capability, resource, outcome, timing, + client metadata. + +- **Immutable.** Once emitted, an event is a fact. Consumers may + filter, aggregate, or discard events, but never mutate them. + +- **Decoupled.** Producers (gateway, IAM service) have no knowledge + of consumers. The topic is fire-and-forget. This keeps the + critical path fast and allows diverse consumer deployments. + +## Architecture + +### Event transport + +Audit events are published to a dedicated pub/sub topic, declared +in the schema layer following the project's queue naming convention: + +```python +audit_events_queue = queue('audit-events', cls='notify') +``` + +This produces the queue identifier `notify:tg:audit-events`, which +each backend maps to its native topic format (e.g. Pulsar maps +`notify` to `non-persistent://tg/notify/audit-events`). + +The `notify` class is the right fit: non-persistent, per-subscriber +delivery, no competing-consumer semantics. Audit event production +must never block the gateway or IAM service. Consumers that need +durability persist events themselves on receipt. + +A single topic carries all event types, distinguished by the +`event_type` field in the envelope. This simplifies producer +logic and allows consumers to subscribe once and filter client-side. + +### Producers + +Two components emit audit events: + +1. **API Gateway** — emits a `gateway.request` event for every + inbound HTTP/WebSocket request after the request completes + (or fails). + +2. **IAM Service** — emits `iam.authenticate` and `iam.authorise` + events for every authentication and authorisation decision. + +Both producers emit asynchronously — the event is published after +the response is sent (gateway) or after the decision is returned +(IAM). Audit emission is never on the critical path. + +### Consumers + +Not defined by this spec. Example consumers that deployments +may wire up: + +- Append to an immutable log store (S3, Cassandra, ClickHouse). +- Forward to a SIEM (Splunk, Elastic, Sentinel). +- Aggregate for billing/metering. +- Feed an anomaly detection model. +- Write to stdout for development debugging. + +## Event Envelope + +Every audit event shares a common envelope: + +```json +{ + "schema_version": 1, + "event_id": "uuid-v4", + "event_type": "gateway.request", + "timestamp": "2026-07-05T14:23:01.123Z", + "producer": "api-gateway", + "payload": { ... } +} +``` + +| Field | Type | Description | +|---|---|---| +| `schema_version` | int | Envelope schema version. Consumers must ignore events with versions they don't understand. | +| `event_id` | string | Globally unique event identifier (UUID v4). | +| `event_type` | string | Dot-separated event type from the vocabulary below. | +| `timestamp` | string | ISO 8601 UTC timestamp at event emission. | +| `producer` | string | Component identity that emitted the event. | +| `payload` | object | Event-type-specific structured data. | + +## Event Types + +### `gateway.request` + +Emitted by the API gateway for every completed request. + +```json +{ + "request_id": "uuid-v4", + "method": "POST", + "path": "/api/v1/flow/default/graph-rag", + "capability": "graph-rag:query", + "workspace": "production", + "identity": "user:mark", + "client_ip": "192.168.1.42", + "user_agent": "trustgraph-cli/2.6.11", + "status_code": 200, + "outcome": "success", + "duration_ms": 1423, + "request_size_bytes": 256, + "response_size_bytes": 4096, + "parameters": { + "collection": "default", + "entity_limit": 50 + } +} +``` + +| Field | Type | Description | +|---|---|---| +| `request_id` | string | Unique ID for this request, propagated to IAM events for correlation. | +| `method` | string | HTTP method. | +| `path` | string | Request path (no query string). | +| `capability` | string | The capability required for this endpoint (from the capability vocabulary). | +| `workspace` | string | Resolved workspace for this request. | +| `identity` | string | Authenticated identity handle, or `"anonymous"` if unauthenticated. | +| `client_ip` | string | Client IP address (may be from X-Forwarded-For). | +| `user_agent` | string | Client User-Agent header. | +| `status_code` | int | HTTP response status code. | +| `outcome` | string | One of `success`, `denied`, `error`, `unauthenticated`. | +| `duration_ms` | int | Request duration in milliseconds. | +| `request_size_bytes` | int | Request body size. | +| `response_size_bytes` | int | Response body size. | +| `error` | string | Error category. Present only when outcome is not `success`. | +| `parameters` | object | Operation-specific parameters extracted from the request (not the full body — only semantically relevant fields). | + +### `iam.authenticate` + +Emitted by the IAM service for every authentication attempt. + +```json +{ + "request_id": "uuid-v4", + "credential_type": "api-key", + "identity": "user:mark", + "outcome": "success", + "client_ip": "192.168.1.42", + "key_id": "key-abc123" +} +``` + +| Field | Type | Description | +|---|---|---| +| `request_id` | string | Correlates with the gateway request that triggered this authentication. | +| `credential_type` | string | One of `api-key`, `jwt`, `login-password`. | +| `identity` | string | Resolved identity on success, or `"unknown"` on failure. | +| `outcome` | string | One of `success`, `failure`. | +| `failure_reason` | string | Internal failure category (not exposed to clients): `invalid-key`, `expired-jwt`, `bad-signature`, `user-disabled`, `unknown-user`. Present only on failure. | +| `client_ip` | string | Forwarded from the gateway request. | +| `key_id` | string | API key identifier (not the secret). Present only on key-based auth. | + +**Note:** `failure_reason` is for the audit log only. The client +response is always the same masked error per the IAM contract's +security rule. The audit consumer sees the real reason; the +attacker does not. + +### `iam.authorise` + +Emitted by the IAM service for every authorisation decision. + +```json +{ + "request_id": "uuid-v4", + "identity": "user:mark", + "capability": "graph-rag:query", + "workspace": "production", + "resource": "flow:default", + "outcome": "allow", + "evaluated_roles": ["workspace-user"], + "evaluation_time_us": 42 +} +``` + +| Field | Type | Description | +|---|---|---| +| `request_id` | string | Correlates with the gateway request. | +| `identity` | string | Identity being authorised. | +| `capability` | string | Capability being checked. | +| `workspace` | string | Workspace scope of the resource. | +| `resource` | string | Structured resource identifier. | +| `outcome` | string | One of `allow`, `deny`. | +| `denial_reason` | string | Why denied: `no-matching-role`, `capability-not-in-role`, `workspace-not-accessible`, `user-disabled`. Present only on denial. | +| `evaluated_roles` | list of string | Roles evaluated during the decision (OSS regime specific — other regimes may populate differently). | +| `evaluation_time_us` | int | Time to evaluate the decision in microseconds. | + +### `iam.management` + +Emitted by the IAM service for administrative mutations. + +```json +{ + "request_id": "uuid-v4", + "actor": "user:admin", + "operation": "create-user", + "target_identity": "user:new-hire", + "target_workspace": "engineering", + "outcome": "success", + "details": { + "roles_assigned": ["workspace-user"] + } +} +``` + +| Field | Type | Description | +|---|---|---| +| `request_id` | string | Correlates with the gateway request. | +| `actor` | string | Identity performing the action. | +| `operation` | string | IAM operation name (`create-user`, `delete-api-key`, `assign-role`, `create-workspace`, etc.). | +| `target_identity` | string | Identity being acted upon. Present only when applicable. | +| `target_workspace` | string | Workspace being acted upon. Present only when applicable. | +| `outcome` | string | One of `success`, `error`. | +| `details` | object | Operation-specific details (roles assigned, key created, etc.). | + +## Correlation + +All events from a single gateway request share the same +`request_id`. A typical request produces: + +1. One `gateway.request` event (after completion). +2. One `iam.authenticate` event (credential validation). +3. One or more `iam.authorise` events (capability checks). + +Consumers can reconstruct the full request lifecycle by grouping +on `request_id`. + +## Implementation + +### Gateway changes + +The gateway emits `gateway.request` events. Implementation: + +- Assign a UUID `request_id` at request entry. +- Pass `request_id` and `client_ip` to the IAM service in the + `IamRequest` (new fields on the dataclass). +- After the response is sent, publish the audit event to the + audit topic. This is a non-blocking fire-and-forget publish. + +### IAM service changes + +The IAM service emits `iam.authenticate`, `iam.authorise`, and +`iam.management` events. Implementation: + +- Accept `request_id` and `client_ip` from the gateway on each + `IamRequest`. +- After each decision or mutation, publish the corresponding audit + event. Non-blocking. + +### Schema additions + +New queue declaration in `trustgraph-base/trustgraph/schema/`: + +```python +from trustgraph.schema.core.topic import queue + +audit_events_queue = queue('audit-events', cls='notify') +``` + +New fields on `IamRequest`: + +```python +@dataclass +class IamRequest: + ... + request_id: str = "" + client_ip: str = "" +``` + +These are informational — the IAM service does not act on them +beyond echoing them into audit events. + +### Pub/sub producer + +A lightweight audit publisher utility in `trustgraph-base`: + +```python +class AuditPublisher: + def __init__(self, producer): + self.producer = producer + + async def emit(self, event_type, payload): + event = { + "schema_version": 1, + "event_id": str(uuid4()), + "event_type": event_type, + "timestamp": datetime.utcnow().isoformat() + "Z", + "producer": self.component_name, + "payload": payload, + } + await self.producer.send(json.dumps(event).encode()) +``` + +The publisher is instantiated once per component and shared across +request handlers. + +## What This Spec Does Not Cover + +- **Storage.** Where audit events are persisted, for how long, + and in what format. Deployment-specific. +- **Alerting.** What conditions trigger alerts. Consumer logic. +- **Retention policy.** How long events are kept. Compliance- + dependent. +- **UI.** Audit log viewers, dashboards, search interfaces. +- **Filtering/routing.** Topic partitioning, consumer-side + filtering, event routing to different backends. +- **Redaction.** PII handling in audit events (may be needed for + GDPR — a future concern for enterprise consumers). + +These are all consumer-side concerns. The value of this boundary +is that producers remain simple and fast while consumers can be +as sophisticated as the deployment requires. + +## Open Questions + +- **Should WebSocket upgrade events emit separately from per-message + events?** Current proposal: one `gateway.request` per WebSocket + session (on close), with `duration_ms` covering the full session. + Per-message audit for long-lived sockets (e.g. streaming RAG) may + be needed for metering but adds volume. + +- **Should `parameters` in `gateway.request` be standardised per + endpoint, or free-form?** Standardised is more useful for + consumers but requires maintenance as endpoints evolve. + +- **Event ordering guarantees.** Pub/sub does not guarantee + ordering across partitions. Consumers that need strict ordering + must sort by `timestamp` or `request_id` sequence. + +## References + +- [IAM Contract](iam-contract.md) — the authentication/authorisation + abstraction. +- [IAM Protocol](iam-protocol.md) — the OSS regime wire protocol. +- [Capability Vocabulary](capabilities.md) — the capability strings + used in authorisation and audit events. +- [Logging Strategy](logging-strategy.md) — operational logging + (complementary, not overlapping). diff --git a/tests/contract/test_schema_field_contracts.py b/tests/contract/test_schema_field_contracts.py index 5be745b8..0c31231c 100644 --- a/tests/contract/test_schema_field_contracts.py +++ b/tests/contract/test_schema_field_contracts.py @@ -24,6 +24,8 @@ from trustgraph.schema import ( EntityContext, EntityEmbeddings, ChunkEmbeddings, + AuditEvent, + IamRequest, ) @@ -72,3 +74,21 @@ class TestSchemaFieldContracts: "context", "chunk_id", } + + def test_audit_event_fields(self): + assert _field_names(AuditEvent) == { + "schema_version", + "event_id", + "event_type", + "timestamp", + "producer", + "payload_json", + } + + def test_iam_request_has_audit_fields(self): + """IamRequest must carry request_id and client_ip for audit + correlation. Removing these breaks the gateway→IAM audit + chain.""" + names = _field_names(IamRequest) + assert "request_id" in names + assert "client_ip" in names diff --git a/tests/unit/test_base/test_audit_publisher.py b/tests/unit/test_base/test_audit_publisher.py new file mode 100644 index 00000000..19854093 --- /dev/null +++ b/tests/unit/test_base/test_audit_publisher.py @@ -0,0 +1,90 @@ +""" +Tests for the AuditPublisher utility. + +Verifies envelope construction, fire-and-forget semantics, and +failure suppression. +""" + +import json +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from trustgraph.base.audit_publisher import AuditPublisher +from trustgraph.schema import AuditEvent, audit_events_queue + + +class TestAuditPublisherInit: + + def test_queue_is_notify_class(self): + assert audit_events_queue == "notify:tg:audit-events" + + def test_creates_producer_with_audit_queue(self): + backend = MagicMock() + pub = AuditPublisher( + backend=backend, + component_name="test-component", + ) + assert pub.producer.topic == audit_events_queue + assert pub.producer.schema == AuditEvent + assert pub.component_name == "test-component" + + +class TestAuditPublisherEmit: + + @pytest.fixture + def publisher(self): + backend = MagicMock() + pub = AuditPublisher( + backend=backend, + component_name="test-svc", + processor_id="proc-1", + ) + pub.producer = AsyncMock() + return pub + + @pytest.mark.asyncio + async def test_emit_sends_structured_envelope(self, publisher): + await publisher.emit("gateway.request", {"path": "/test"}) + + publisher.producer.send.assert_called_once() + event = publisher.producer.send.call_args[0][0] + + assert isinstance(event, AuditEvent) + assert event.schema_version == 1 + assert event.event_type == "gateway.request" + assert event.producer == "test-svc" + assert event.event_id != "" + assert event.timestamp != "" + + @pytest.mark.asyncio + async def test_emit_serializes_payload_as_json(self, publisher): + payload = {"method": "POST", "status_code": 200} + await publisher.emit("gateway.request", payload) + + event = publisher.producer.send.call_args[0][0] + decoded = json.loads(event.payload_json) + assert decoded == payload + + @pytest.mark.asyncio + async def test_emit_generates_unique_event_ids(self, publisher): + await publisher.emit("test.a", {}) + await publisher.emit("test.b", {}) + + ids = [ + call[0][0].event_id + for call in publisher.producer.send.call_args_list + ] + assert ids[0] != ids[1] + + @pytest.mark.asyncio + async def test_emit_swallows_send_failure(self, publisher): + publisher.producer.send.side_effect = RuntimeError("pub/sub down") + await publisher.emit("test.event", {"key": "value"}) + + @pytest.mark.asyncio + async def test_emit_timestamp_is_utc_iso(self, publisher): + await publisher.emit("test.event", {}) + + event = publisher.producer.send.call_args[0][0] + assert "T" in event.timestamp + assert "+" in event.timestamp or "Z" in event.timestamp diff --git a/tests/unit/test_gateway/test_audit_middleware.py b/tests/unit/test_gateway/test_audit_middleware.py new file mode 100644 index 00000000..5c2e32c3 --- /dev/null +++ b/tests/unit/test_gateway/test_audit_middleware.py @@ -0,0 +1,198 @@ +""" +Tests for the gateway audit middleware. + +Verifies that gateway.request events are emitted with correct +metadata for success, error, and auth failure paths. +""" + +import pytest +from unittest.mock import AsyncMock, MagicMock +from aiohttp import web +from aiohttp.test_utils import make_mocked_request + +from trustgraph.gateway.audit import make_audit_middleware, _client_ip, _outcome_from_status + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_request(method="POST", path="/api/v1/config", headers=None, + remote="10.0.0.1"): + hdrs = headers or {} + req = make_mocked_request(method, path, headers=hdrs) + req._transport_peername = (remote, 0) + return req + + +# --------------------------------------------------------------------------- +# Pure helpers +# --------------------------------------------------------------------------- + +class TestClientIp: + + def test_uses_x_forwarded_for(self): + req = _make_request(headers={"X-Forwarded-For": "1.2.3.4, 10.0.0.1"}) + assert _client_ip(req) == "1.2.3.4" + + def test_falls_back_to_remote(self): + req = _make_request(remote="192.168.1.1") + assert _client_ip(req) == "192.168.1.1" + + +class TestOutcomeFromStatus: + + def test_success_range(self): + assert _outcome_from_status(200) == "success" + assert _outcome_from_status(201) == "success" + assert _outcome_from_status(204) == "success" + assert _outcome_from_status(301) == "success" + + def test_unauthenticated(self): + assert _outcome_from_status(401) == "unauthenticated" + + def test_denied(self): + assert _outcome_from_status(403) == "denied" + + def test_error(self): + assert _outcome_from_status(400) == "error" + assert _outcome_from_status(404) == "error" + assert _outcome_from_status(500) == "error" + + +# --------------------------------------------------------------------------- +# Middleware integration +# --------------------------------------------------------------------------- + +class TestAuditMiddleware: + + @pytest.fixture + def audit_publisher(self): + pub = AsyncMock() + pub.emit = AsyncMock() + return pub + + @pytest.fixture + def middleware(self, audit_publisher): + return make_audit_middleware(audit_publisher) + + @pytest.mark.asyncio + async def test_emits_event_on_success(self, middleware, audit_publisher): + async def handler(request): + return web.json_response({"ok": True}) + + req = _make_request() + await middleware(req, handler) + + audit_publisher.emit.assert_called_once() + event_type, payload = audit_publisher.emit.call_args[0] + assert event_type == "gateway.request" + assert payload["method"] == "POST" + assert payload["path"] == "/api/v1/config" + assert payload["status_code"] == 200 + assert payload["outcome"] == "success" + assert "request_id" in payload + assert "duration_ms" in payload + assert "error" not in payload + + @pytest.mark.asyncio + async def test_emits_event_on_http_exception(self, middleware, + audit_publisher): + async def handler(request): + raise web.HTTPForbidden(reason="access denied") + + req = _make_request() + with pytest.raises(web.HTTPForbidden): + await middleware(req, handler) + + audit_publisher.emit.assert_called_once() + _, payload = audit_publisher.emit.call_args[0] + assert payload["status_code"] == 403 + assert payload["outcome"] == "denied" + assert payload["error"] == "access denied" + + @pytest.mark.asyncio + async def test_emits_event_on_unhandled_exception(self, middleware, + audit_publisher): + async def handler(request): + raise ValueError("boom") + + req = _make_request() + with pytest.raises(ValueError): + await middleware(req, handler) + + audit_publisher.emit.assert_called_once() + _, payload = audit_publisher.emit.call_args[0] + assert payload["status_code"] == 500 + assert payload["outcome"] == "error" + assert payload["error"] == "ValueError" + + @pytest.mark.asyncio + async def test_includes_identity_when_annotated(self, middleware, + audit_publisher): + async def handler(request): + request['audit_identity'] = "user:mark" + return web.json_response({"ok": True}) + + req = _make_request() + await middleware(req, handler) + + _, payload = audit_publisher.emit.call_args[0] + assert payload["identity"] == "user:mark" + + @pytest.mark.asyncio + async def test_includes_capability_when_annotated(self, middleware, + audit_publisher): + async def handler(request): + request['audit_capability'] = "config:read" + request['audit_workspace'] = "production" + return web.json_response({"ok": True}) + + req = _make_request() + await middleware(req, handler) + + _, payload = audit_publisher.emit.call_args[0] + assert payload["capability"] == "config:read" + assert payload["workspace"] == "production" + + @pytest.mark.asyncio + async def test_omits_unannotated_fields(self, middleware, + audit_publisher): + async def handler(request): + return web.json_response({"ok": True}) + + req = _make_request() + await middleware(req, handler) + + _, payload = audit_publisher.emit.call_args[0] + assert "identity" not in payload + assert "capability" not in payload + assert "workspace" not in payload + + @pytest.mark.asyncio + async def test_assigns_request_id(self, middleware, audit_publisher): + captured_id = None + + async def handler(request): + nonlocal captured_id + captured_id = request.get('audit_request_id') + return web.json_response({"ok": True}) + + req = _make_request() + await middleware(req, handler) + + assert captured_id is not None + _, payload = audit_publisher.emit.call_args[0] + assert payload["request_id"] == captured_id + + @pytest.mark.asyncio + async def test_still_returns_response_if_emit_fails(self, middleware, + audit_publisher): + audit_publisher.emit.side_effect = RuntimeError("pub/sub down") + + async def handler(request): + return web.json_response({"ok": True}) + + req = _make_request() + resp = await middleware(req, handler) + assert resp.status == 200 diff --git a/tests/unit/test_gateway/test_auth.py b/tests/unit/test_gateway/test_auth.py index 2775b0b0..33d3d6f9 100644 --- a/tests/unit/test_gateway/test_auth.py +++ b/tests/unit/test_gateway/test_auth.py @@ -244,7 +244,7 @@ class TestIamAuthDispatch: async def test_api_key_path(self): auth = IamAuth(backend=Mock()) - async def fake_resolve(api_key): + async def fake_resolve(api_key, **kwargs): assert api_key == "tg_testkey" # Roles are returned by the regime as a hint but the # gateway ignores them — kept here so the resolve @@ -309,7 +309,7 @@ class TestApiKeyCache: seen = [] async def fake_with_client(op): - async def resolve(plaintext): + async def resolve(plaintext, **kwargs): seen.append(plaintext) return ("u-" + plaintext, "default", ["reader"]) return await op(Mock(resolve_api_key=resolve)) diff --git a/tests/unit/test_iam/test_iam_audit_events.py b/tests/unit/test_iam/test_iam_audit_events.py new file mode 100644 index 00000000..cef20bac --- /dev/null +++ b/tests/unit/test_iam/test_iam_audit_events.py @@ -0,0 +1,345 @@ +""" +Tests for IAM service audit event emission. + +Verifies that the IAM Processor emits correctly categorised audit +events (iam.authenticate, iam.authorise, iam.management) after +handling each request type. +""" + +import asyncio +import pytest +from unittest.mock import AsyncMock, MagicMock, Mock + +from trustgraph.schema import IamRequest, IamResponse, Error +from trustgraph.iam.service.service import Processor + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_msg(request, msg_id="test-msg-1"): + msg = MagicMock() + msg.value.return_value = request + msg.properties.return_value = {"id": msg_id} + return msg + + +def _make_processor(): + """Create a Processor with stubbed dependencies.""" + proc = object.__new__(Processor) + proc.id = "test-iam-svc" + proc.iam_response_producer = AsyncMock() + proc.audit = AsyncMock() + return proc + + +# --------------------------------------------------------------------------- +# Authentication events +# --------------------------------------------------------------------------- + +class TestAuthenticateAuditEvents: + + @pytest.mark.asyncio + async def test_resolve_api_key_success(self): + proc = _make_processor() + resp = IamResponse( + resolved_user_id="user-123", + resolved_default_workspace="default", + ) + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest( + operation="resolve-api-key", + api_key="tg_test", + request_id="req-1", + client_ip="10.0.0.1", + ) + await proc.on_iam_request(_make_msg(req), None, None) + + proc.audit.emit.assert_called_once() + event_type, payload = proc.audit.emit.call_args[0] + assert event_type == "iam.authenticate" + assert payload["credential_type"] == "api-key" + assert payload["identity"] == "user-123" + assert payload["outcome"] == "success" + assert payload["request_id"] == "req-1" + assert payload["client_ip"] == "10.0.0.1" + assert "failure_reason" not in payload + + @pytest.mark.asyncio + async def test_resolve_api_key_failure(self): + proc = _make_processor() + resp = IamResponse(error=Error(type="auth-failed", message="unknown")) + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest( + operation="resolve-api-key", + api_key="tg_bad", + request_id="req-2", + client_ip="10.0.0.2", + ) + await proc.on_iam_request(_make_msg(req), None, None) + + _, payload = proc.audit.emit.call_args[0] + assert payload["outcome"] == "failure" + assert payload["identity"] == "unknown" + assert payload["failure_reason"] == "auth-failed" + + @pytest.mark.asyncio + async def test_login_emits_authenticate(self): + proc = _make_processor() + resp = IamResponse( + resolved_user_id="user-456", + jwt="eyJ...", + jwt_expires="2026-07-06T10:00:00Z", + ) + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest( + operation="login", + username="mark", + password="secret", + request_id="req-3", + ) + await proc.on_iam_request(_make_msg(req), None, None) + + event_type, payload = proc.audit.emit.call_args[0] + assert event_type == "iam.authenticate" + assert payload["credential_type"] == "login-password" + + @pytest.mark.asyncio + async def test_anonymous_emits_authenticate(self): + proc = _make_processor() + resp = IamResponse(error=Error(type="auth-failed", message="no")) + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest(operation="authenticate-anonymous") + await proc.on_iam_request(_make_msg(req), None, None) + + event_type, payload = proc.audit.emit.call_args[0] + assert event_type == "iam.authenticate" + assert payload["credential_type"] == "anonymous" + assert payload["outcome"] == "failure" + + +# --------------------------------------------------------------------------- +# Authorise events +# --------------------------------------------------------------------------- + +class TestAuthoriseAuditEvents: + + @pytest.mark.asyncio + async def test_authorise_allow(self): + proc = _make_processor() + resp = IamResponse(decision_allow=True, decision_ttl_seconds=60) + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest( + operation="authorise", + user_id="user-1", + capability="config:read", + resource_json='{"workspace": "production"}', + request_id="req-4", + ) + await proc.on_iam_request(_make_msg(req), None, None) + + event_type, payload = proc.audit.emit.call_args[0] + assert event_type == "iam.authorise" + assert payload["outcome"] == "allow" + assert payload["capability"] == "config:read" + assert payload["identity"] == "user-1" + assert payload["workspace"] == "production" + assert "denial_reason" not in payload + + @pytest.mark.asyncio + async def test_authorise_deny(self): + proc = _make_processor() + resp = IamResponse(decision_allow=False, decision_ttl_seconds=5) + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest( + operation="authorise", + user_id="user-2", + capability="users:admin", + resource_json='{}', + request_id="req-5", + ) + await proc.on_iam_request(_make_msg(req), None, None) + + _, payload = proc.audit.emit.call_args[0] + assert payload["outcome"] == "deny" + assert payload["denial_reason"] == "capability-not-in-role" + + @pytest.mark.asyncio + async def test_authorise_extracts_workspace_from_resource_json(self): + proc = _make_processor() + resp = IamResponse(decision_allow=True, decision_ttl_seconds=60) + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest( + operation="authorise", + user_id="user-1", + capability="graph-rag:query", + resource_json='{"workspace": "engineering"}', + ) + await proc.on_iam_request(_make_msg(req), None, None) + + _, payload = proc.audit.emit.call_args[0] + assert payload["workspace"] == "engineering" + + @pytest.mark.asyncio + async def test_authorise_omits_empty_workspace(self): + proc = _make_processor() + resp = IamResponse(decision_allow=True, decision_ttl_seconds=60) + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest( + operation="authorise", + user_id="user-1", + capability="metrics:read", + resource_json='{}', + ) + await proc.on_iam_request(_make_msg(req), None, None) + + _, payload = proc.audit.emit.call_args[0] + assert "workspace" not in payload + + +# --------------------------------------------------------------------------- +# Management events +# --------------------------------------------------------------------------- + +class TestManagementAuditEvents: + + @pytest.mark.asyncio + async def test_create_user_emits_management(self): + proc = _make_processor() + resp = IamResponse() + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest( + operation="create-user", + actor="admin-1", + user_id="new-user", + workspace="default", + request_id="req-6", + ) + await proc.on_iam_request(_make_msg(req), None, None) + + event_type, payload = proc.audit.emit.call_args[0] + assert event_type == "iam.management" + assert payload["operation"] == "create-user" + assert payload["actor"] == "admin-1" + assert payload["target_identity"] == "new-user" + assert payload["target_workspace"] == "default" + assert payload["outcome"] == "success" + + @pytest.mark.asyncio + async def test_revoke_api_key_emits_management(self): + proc = _make_processor() + resp = IamResponse() + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest( + operation="revoke-api-key", + actor="admin-1", + key_id="key-abc", + workspace="production", + request_id="req-7", + ) + await proc.on_iam_request(_make_msg(req), None, None) + + event_type, payload = proc.audit.emit.call_args[0] + assert event_type == "iam.management" + assert payload["operation"] == "revoke-api-key" + assert payload["outcome"] == "success" + + @pytest.mark.asyncio + async def test_management_error_outcome(self): + proc = _make_processor() + resp = IamResponse( + error=Error(type="not-found", message="user not found"), + ) + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest( + operation="delete-user", + actor="admin-1", + user_id="ghost", + ) + await proc.on_iam_request(_make_msg(req), None, None) + + _, payload = proc.audit.emit.call_args[0] + assert payload["outcome"] == "error" + + @pytest.mark.asyncio + async def test_management_omits_empty_target_fields(self): + proc = _make_processor() + resp = IamResponse() + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest( + operation="rotate-signing-key", + actor="admin-1", + ) + await proc.on_iam_request(_make_msg(req), None, None) + + _, payload = proc.audit.emit.call_args[0] + assert "target_identity" not in payload + assert "target_workspace" not in payload + + +# --------------------------------------------------------------------------- +# Non-audited operations +# --------------------------------------------------------------------------- + +class TestNonAuditedOperations: + + @pytest.mark.asyncio + async def test_whoami_does_not_emit(self): + proc = _make_processor() + resp = IamResponse() + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest(operation="whoami", actor="user-1") + await proc.on_iam_request(_make_msg(req), None, None) + + proc.audit.emit.assert_not_called() + + @pytest.mark.asyncio + async def test_list_users_does_not_emit(self): + proc = _make_processor() + resp = IamResponse() + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest(operation="list-users", workspace="default") + await proc.on_iam_request(_make_msg(req), None, None) + + proc.audit.emit.assert_not_called() + + @pytest.mark.asyncio + async def test_get_signing_key_does_not_emit(self): + proc = _make_processor() + resp = IamResponse(signing_key_public="PEM...") + proc.iam = MagicMock() + proc.iam.handle = AsyncMock(return_value=resp) + + req = IamRequest(operation="get-signing-key-public") + await proc.on_iam_request(_make_msg(req), None, None) + + proc.audit.emit.assert_not_called() diff --git a/trustgraph-base/trustgraph/base/__init__.py b/trustgraph-base/trustgraph/base/__init__.py index be905116..0e77f8ac 100644 --- a/trustgraph-base/trustgraph/base/__init__.py +++ b/trustgraph-base/trustgraph/base/__init__.py @@ -46,4 +46,5 @@ from . reranker_client import RerankerClientSpec from . reranker_service import RerankerService from . row_embeddings_query_client import RowEmbeddingsQueryClientSpec from . collection_config_handler import CollectionConfigHandler +from . audit_publisher import AuditPublisher diff --git a/trustgraph-base/trustgraph/base/audit_publisher.py b/trustgraph-base/trustgraph/base/audit_publisher.py new file mode 100644 index 00000000..83d41357 --- /dev/null +++ b/trustgraph-base/trustgraph/base/audit_publisher.py @@ -0,0 +1,42 @@ + +import json +import logging +from datetime import datetime, timezone +from uuid import uuid4 + +from . producer import Producer +from . metrics import ProducerMetrics +from trustgraph.schema import AuditEvent, audit_events_queue + +logger = logging.getLogger(__name__) + + +class AuditPublisher: + + def __init__(self, backend, component_name, processor_id=None): + self.component_name = component_name + self.producer = Producer( + backend=backend, + topic=audit_events_queue, + schema=AuditEvent, + metrics=ProducerMetrics( + processor=processor_id or component_name, + flow=None, + name="audit-events", + ), + ) + + async def emit(self, event_type, payload): + event = AuditEvent( + schema_version=1, + event_id=str(uuid4()), + event_type=event_type, + timestamp=datetime.now(timezone.utc).isoformat(), + producer=self.component_name, + payload_json=json.dumps(payload), + ) + + try: + await self.producer.send(event) + except Exception as e: + logger.warning(f"Failed to emit audit event: {e}") diff --git a/trustgraph-base/trustgraph/base/iam_client.py b/trustgraph-base/trustgraph/base/iam_client.py index 423abb8e..3cb14901 100644 --- a/trustgraph-base/trustgraph/base/iam_client.py +++ b/trustgraph-base/trustgraph/base/iam_client.py @@ -62,7 +62,8 @@ class IamClient(RequestResponse): ) return resp.user - async def authenticate_anonymous(self, timeout=IAM_TIMEOUT): + async def authenticate_anonymous(self, timeout=IAM_TIMEOUT, + request_id="", client_ip=""): """Request anonymous access from the IAM regime. Returns ``(user_id, default_workspace, roles)`` if the regime @@ -70,6 +71,8 @@ class IamClient(RequestResponse): error type ``auth-failed`` if it does not.""" resp = await self._request( operation="authenticate-anonymous", + request_id=request_id, + client_ip=client_ip, timeout=timeout, ) return ( @@ -78,7 +81,8 @@ class IamClient(RequestResponse): list(resp.resolved_roles), ) - async def resolve_api_key(self, api_key, timeout=IAM_TIMEOUT): + async def resolve_api_key(self, api_key, timeout=IAM_TIMEOUT, + request_id="", client_ip=""): """Resolve a plaintext API key to its identity triple. Returns ``(user_id, default_workspace, roles)`` or raises @@ -87,6 +91,8 @@ class IamClient(RequestResponse): resp = await self._request( operation="resolve-api-key", api_key=api_key, + request_id=request_id, + client_ip=client_ip, timeout=timeout, ) return ( @@ -96,7 +102,8 @@ class IamClient(RequestResponse): ) async def authorise(self, identity_handle, capability, - resource, parameters, timeout=IAM_TIMEOUT): + resource, parameters, timeout=IAM_TIMEOUT, + request_id="", client_ip=""): """Ask the IAM regime whether ``identity_handle`` may perform ``capability`` on ``resource`` given ``parameters``. @@ -111,6 +118,8 @@ class IamClient(RequestResponse): capability=capability, resource_json=json.dumps(resource or {}, sort_keys=True), parameters_json=json.dumps(parameters or {}, sort_keys=True), + request_id=request_id, + client_ip=client_ip, timeout=timeout, ) return resp.decision_allow, resp.decision_ttl_seconds @@ -186,7 +195,7 @@ class IamClient(RequestResponse): ) async def login(self, username, password, workspace="", - timeout=IAM_TIMEOUT): + timeout=IAM_TIMEOUT, request_id="", client_ip=""): """Validate credentials and return ``(jwt, expires_iso)``. ``workspace`` is optional; defaults at the server to the OSS default workspace.""" @@ -195,6 +204,8 @@ class IamClient(RequestResponse): workspace=workspace, username=username, password=password, + request_id=request_id, + client_ip=client_ip, timeout=timeout, ) return resp.jwt, resp.jwt_expires diff --git a/trustgraph-base/trustgraph/schema/services/__init__.py b/trustgraph-base/trustgraph/schema/services/__init__.py index 63dc05fd..e7a215db 100644 --- a/trustgraph-base/trustgraph/schema/services/__init__.py +++ b/trustgraph-base/trustgraph/schema/services/__init__.py @@ -16,4 +16,5 @@ from .collection import * from .storage import * from .tool_service import * from .sparql_query import * -from .reranker import * \ No newline at end of file +from .reranker import * +from .audit import * \ No newline at end of file diff --git a/trustgraph-base/trustgraph/schema/services/audit.py b/trustgraph-base/trustgraph/schema/services/audit.py new file mode 100644 index 00000000..797c0bc7 --- /dev/null +++ b/trustgraph-base/trustgraph/schema/services/audit.py @@ -0,0 +1,27 @@ + +from dataclasses import dataclass, field + +from ..core.topic import queue + +############################################################################ + +# Audit events — see docs/tech-specs/audit-events.md for the full spec. +# +# Transport: notify-class pub/sub (fire-and-forget, per-subscriber +# delivery). Producers are the API gateway and the IAM service. +# Consumers are optional enterprise components. + + +@dataclass +class AuditEvent: + schema_version: int = 1 + event_id: str = "" + event_type: str = "" + timestamp: str = "" + producer: str = "" + payload_json: str = "" + + +audit_events_queue = queue('audit-events', cls='notify') + +############################################################################ diff --git a/trustgraph-base/trustgraph/schema/services/iam.py b/trustgraph-base/trustgraph/schema/services/iam.py index 6b2599e0..9f84872b 100644 --- a/trustgraph-base/trustgraph/schema/services/iam.py +++ b/trustgraph-base/trustgraph/schema/services/iam.py @@ -121,6 +121,10 @@ class IamRequest: group: GroupInput | None = None grant: GrantInput | None = None + # ---- Audit context (informational, echoed into audit events) ---- + request_id: str = "" + client_ip: str = "" + # ---- authorise / authorise-many inputs ---- # Capability string from the vocabulary in capabilities.md. capability: str = "" diff --git a/trustgraph-flow/trustgraph/gateway/audit.py b/trustgraph-flow/trustgraph/gateway/audit.py new file mode 100644 index 00000000..12365fa2 --- /dev/null +++ b/trustgraph-flow/trustgraph/gateway/audit.py @@ -0,0 +1,105 @@ +""" +Audit middleware for the API gateway. + +Wraps every HTTP request with timing and metadata collection, then +emits a ``gateway.request`` audit event after the response is sent. +Handlers can enrich the event by annotating the request dict: + + request['audit_identity'] = identity.principal_id + request['audit_capability'] = capability + request['audit_workspace'] = workspace +""" + +import time +import logging +from uuid import uuid4 + +from aiohttp import web + +from trustgraph.base import AuditPublisher + +logger = logging.getLogger("audit") + + +def _client_ip(request): + forwarded = request.headers.get("X-Forwarded-For") + if forwarded: + return forwarded.split(",")[0].strip() + peer = request.remote + return peer or "" + + +def _outcome_from_status(status): + if 200 <= status < 400: + return "success" + if status == 401: + return "unauthenticated" + if status == 403: + return "denied" + return "error" + + +def make_audit_middleware(audit_publisher): + + @web.middleware + async def audit_middleware(request, handler): + request_id = str(uuid4()) + request['audit_request_id'] = request_id + start = time.monotonic() + + status_code = 500 + error = None + response_size = 0 + + try: + response = await handler(request) + status_code = response.status + response_size = response.content_length or 0 + return response + except web.HTTPException as exc: + status_code = exc.status + error = exc.reason + raise + except Exception as exc: + status_code = 500 + error = type(exc).__name__ + raise + finally: + duration_ms = int((time.monotonic() - start) * 1000) + outcome = _outcome_from_status(status_code) + + payload = { + "request_id": request_id, + "method": request.method, + "path": request.path, + "client_ip": _client_ip(request), + "user_agent": request.headers.get("User-Agent", ""), + "status_code": status_code, + "outcome": outcome, + "duration_ms": duration_ms, + "request_size_bytes": request.content_length or 0, + "response_size_bytes": response_size, + } + + identity = request.get('audit_identity') + if identity: + payload["identity"] = identity + + capability = request.get('audit_capability') + if capability: + payload["capability"] = capability + + workspace = request.get('audit_workspace') + if workspace: + payload["workspace"] = workspace + + if error and outcome != "success": + payload["error"] = error + + try: + await audit_publisher.emit("gateway.request", payload) + except Exception: + logger.debug("Failed to emit gateway audit event", + exc_info=True) + + return audit_middleware diff --git a/trustgraph-flow/trustgraph/gateway/auth.py b/trustgraph-flow/trustgraph/gateway/auth.py index b08bf650..e81b6908 100644 --- a/trustgraph-flow/trustgraph/gateway/auth.py +++ b/trustgraph-flow/trustgraph/gateway/auth.py @@ -232,21 +232,55 @@ class IamAuth: cannot distinguish missing / malformed / invalid / expired / revoked credentials.""" + request_id = request.get('audit_request_id', '') if hasattr(request, 'get') else '' + client_ip = self._extract_client_ip(request) + header = request.headers.get("Authorization", "") if not header.startswith("Bearer "): - return await self._authenticate_anonymous() + identity = await self._authenticate_anonymous( + request_id=request_id, client_ip=client_ip, + ) + self._annotate_request(request, identity) + return identity token = header[len("Bearer "):].strip() if not token: - return await self._authenticate_anonymous() + identity = await self._authenticate_anonymous( + request_id=request_id, client_ip=client_ip, + ) + self._annotate_request(request, identity) + return identity # API keys always start with "tg_". JWTs have two dots and # no "tg_" prefix. Discriminate cheaply. if token.startswith("tg_"): - return await self._resolve_api_key(token) + identity = await self._resolve_api_key( + token, request_id=request_id, client_ip=client_ip, + ) + self._annotate_request(request, identity) + return identity if token.count(".") == 2: - return self._verify_jwt(token) + identity = self._verify_jwt(token) + self._annotate_request(request, identity) + return identity raise _auth_failure() + @staticmethod + def _extract_client_ip(request): + try: + forwarded = request.headers.get("X-Forwarded-For") + if forwarded: + return forwarded.split(",")[0].strip() + return request.remote or "" + except Exception: + return "" + + @staticmethod + def _annotate_request(request, identity): + try: + request['audit_identity'] = identity.principal_id + except Exception: + pass + def _verify_jwt(self, token): if not self._signing_public_pem: raise _auth_failure() @@ -266,10 +300,12 @@ class IamAuth: principal_id=sub, source="jwt", ) - async def _authenticate_anonymous(self): + async def _authenticate_anonymous(self, request_id="", client_ip=""): try: async def _call(client): - return await client.authenticate_anonymous() + return await client.authenticate_anonymous( + request_id=request_id, client_ip=client_ip, + ) user_id, default_workspace, _roles = await self._with_client( _call, ) @@ -288,7 +324,7 @@ class IamAuth: principal_id=user_id, source="anonymous", ) - async def _resolve_api_key(self, plaintext): + async def _resolve_api_key(self, plaintext, request_id="", client_ip=""): h = hashlib.sha256(plaintext.encode("utf-8")).hexdigest() cached = self._key_cache.get(h) @@ -303,7 +339,10 @@ class IamAuth: try: async def _call(client): - return await client.resolve_api_key(plaintext) + return await client.resolve_api_key( + plaintext, request_id=request_id, + client_ip=client_ip, + ) # ``roles`` is returned by the OSS regime as a hint # but is not consulted by the gateway; all policy # decisions go through ``authorise``. @@ -345,7 +384,8 @@ class IamAuth: ) return hashlib.sha256(payload.encode("utf-8")).hexdigest() - async def authorise(self, identity, capability, resource, parameters): + async def authorise(self, identity, capability, resource, parameters, + request_id="", client_ip=""): """Ask the IAM regime whether ``identity`` may perform ``capability`` on ``resource`` given ``parameters``. @@ -383,6 +423,7 @@ class IamAuth: return await client.authorise( identity.handle, capability, resource or {}, parameters or {}, + request_id=request_id, client_ip=client_ip, ) allow, ttl = await self._with_client(_call) except Exception as e: diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/iam_endpoint.py b/trustgraph-flow/trustgraph/gateway/endpoint/iam_endpoint.py index 749eacd3..9ad2cc26 100644 --- a/trustgraph-flow/trustgraph/gateway/endpoint/iam_endpoint.py +++ b/trustgraph-flow/trustgraph/gateway/endpoint/iam_endpoint.py @@ -90,7 +90,13 @@ class IamEndpoint: await self.auth.authorise( identity, op.capability, resource, parameters, + request_id=request.get('audit_request_id', ''), + client_ip=self.auth._extract_client_ip(request), ) + request['audit_capability'] = op.capability + ws = resource.get('workspace', '') + if ws: + request['audit_workspace'] = ws # Plumb the authenticated caller's handle through as ``actor`` # so iam-svc handlers (e.g. whoami, future actor-scoped diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/manager.py b/trustgraph-flow/trustgraph/gateway/endpoint/manager.py index 2b3bb83b..d80baa19 100644 --- a/trustgraph-flow/trustgraph/gateway/endpoint/manager.py +++ b/trustgraph-flow/trustgraph/gateway/endpoint/manager.py @@ -75,9 +75,14 @@ class _RoutedVariableEndpoint: parameters = op.extract_parameters(ctx) await self.auth.authorise( identity, op.capability, resource, parameters, + request_id=request.get('audit_request_id', ''), + client_ip=self.auth._extract_client_ip(request), ) + request['audit_capability'] = op.capability ws = resource.get("workspace", "") + if ws: + request['audit_workspace'] = ws if ws and ws not in self.auth.known_workspaces: raise workspace_not_found() @@ -143,9 +148,14 @@ class _RoutedSocketEndpoint: parameters = op.extract_parameters(ctx) await self.auth.authorise( identity, op.capability, resource, parameters, + request_id=request.get('audit_request_id', ''), + client_ip=self.auth._extract_client_ip(request), ) + request['audit_capability'] = op.capability ws = resource.get("workspace", "") + if ws: + request['audit_workspace'] = ws if ws and ws not in self.auth.known_workspaces: raise workspace_not_found() diff --git a/trustgraph-flow/trustgraph/gateway/endpoint/registry_endpoint.py b/trustgraph-flow/trustgraph/gateway/endpoint/registry_endpoint.py index 0861bcd3..aae9604d 100644 --- a/trustgraph-flow/trustgraph/gateway/endpoint/registry_endpoint.py +++ b/trustgraph-flow/trustgraph/gateway/endpoint/registry_endpoint.py @@ -98,7 +98,13 @@ class RegistryRoutedVariableEndpoint: await self.auth.authorise( identity, op.capability, resource, parameters, + request_id=request.get('audit_request_id', ''), + client_ip=self.auth._extract_client_ip(request), ) + request['audit_capability'] = op.capability + ws = resource.get('workspace', '') + if ws: + request['audit_workspace'] = ws # Default-fill workspace into the body so downstream # dispatchers see the canonical resolved value. The diff --git a/trustgraph-flow/trustgraph/gateway/service.py b/trustgraph-flow/trustgraph/gateway/service.py index fb51e1a2..098ce0b5 100755 --- a/trustgraph-flow/trustgraph/gateway/service.py +++ b/trustgraph-flow/trustgraph/gateway/service.py @@ -11,8 +11,10 @@ import os from trustgraph.base.logging import setup_logging, add_logging_args from trustgraph.base.pubsub import get_pubsub, add_pubsub_args +from trustgraph.base import AuditPublisher from . auth import IamAuth +from . audit import make_audit_middleware from . config.receiver import ConfigReceiver from . dispatch.manager import DispatcherManager @@ -122,12 +124,18 @@ class Api: timeout = self.timeout, ) + self.audit_publisher = AuditPublisher( + backend=self.pubsub_backend, + component_name="api-gateway", + processor_id=config.get("id", "api-gateway"), + ) + self.endpoint_manager = EndpointManager( dispatcher_manager = self.dispatcher_manager, auth = self.auth, prometheus_url = self.prometheus_url, timeout = self.timeout, - + ) self.endpoints = [ @@ -136,7 +144,7 @@ class Api: async def app_factory(self): self.app = web.Application( - middlewares=[], + middlewares=[make_audit_middleware(self.audit_publisher)], client_max_size=256 * 1024 * 1024 ) diff --git a/trustgraph-flow/trustgraph/iam/service/service.py b/trustgraph-flow/trustgraph/iam/service/service.py index b2f3976d..52498230 100644 --- a/trustgraph-flow/trustgraph/iam/service/service.py +++ b/trustgraph-flow/trustgraph/iam/service/service.py @@ -16,6 +16,7 @@ from trustgraph.schema import ConfigRequest, ConfigResponse, ConfigValue from trustgraph.schema import config_request_queue, config_response_queue from trustgraph.base import AsyncProcessor, Consumer, Producer +from trustgraph.base import AuditPublisher from trustgraph.base import ConsumerMetrics, ProducerMetrics from trustgraph.base.metrics import SubscriberMetrics from trustgraph.base.request_response_spec import RequestResponse @@ -145,6 +146,12 @@ class Processor(AsyncProcessor): metrics=iam_response_metrics, ) + self.audit = AuditPublisher( + backend=self.pubsub, + component_name="iam-service", + processor_id=self.id, + ) + self.iam = IamService( host=self.cassandra_host, username=self.cassandra_username, @@ -243,6 +250,19 @@ class Processor(AsyncProcessor): f"{workspace_id}: {e}", exc_info=True, ) + AUTHENTICATE_OPS = frozenset({ + "resolve-api-key", "login", "authenticate-anonymous", + }) + AUTHORISE_OPS = frozenset({ + "authorise", "authorise-many", + }) + MANAGEMENT_OPS = frozenset({ + "create-user", "update-user", "disable-user", "enable-user", + "delete-user", "create-api-key", "revoke-api-key", + "create-workspace", "update-workspace", "disable-workspace", + "reset-password", "rotate-signing-key", "bootstrap", + }) + async def on_iam_request(self, msg, consumer, flow): id = None @@ -256,6 +276,7 @@ class Processor(AsyncProcessor): await self.iam_response_producer.send( resp, properties={"id": id}, ) + await self._emit_audit(v, resp) except Exception as e: logger.error( f"IAM request failed: {type(e).__name__}: {e}", @@ -269,6 +290,76 @@ class Processor(AsyncProcessor): resp, properties={"id": id}, ) + async def _emit_audit(self, v, resp): + try: + op = v.operation + if op in self.AUTHENTICATE_OPS: + await self._emit_authenticate(v, resp) + elif op in self.AUTHORISE_OPS: + await self._emit_authorise(v, resp) + elif op in self.MANAGEMENT_OPS: + await self._emit_management(v, resp) + except Exception: + logger.debug("Failed to emit IAM audit event", exc_info=True) + + async def _emit_authenticate(self, v, resp): + has_error = resp.error is not None + payload = { + "request_id": v.request_id, + "credential_type": self._credential_type(v.operation), + "identity": resp.resolved_user_id if not has_error else "unknown", + "outcome": "failure" if has_error else "success", + "client_ip": v.client_ip, + } + if has_error: + payload["failure_reason"] = resp.error.type + if v.key_id: + payload["key_id"] = v.key_id + await self.audit.emit("iam.authenticate", payload) + + async def _emit_authorise(self, v, resp): + import json as _json + workspace = v.workspace + if not workspace: + try: + resource = _json.loads(v.resource_json or "{}") + workspace = resource.get("workspace", "") + except Exception: + pass + payload = { + "request_id": v.request_id, + "identity": v.user_id, + "capability": v.capability, + "outcome": "allow" if resp.decision_allow else "deny", + } + if workspace: + payload["workspace"] = workspace + if not resp.decision_allow: + payload["denial_reason"] = "capability-not-in-role" + await self.audit.emit("iam.authorise", payload) + + async def _emit_management(self, v, resp): + has_error = resp.error is not None + payload = { + "request_id": v.request_id, + "actor": v.actor, + "operation": v.operation, + "outcome": "error" if has_error else "success", + } + if v.user_id: + payload["target_identity"] = v.user_id + if v.workspace: + payload["target_workspace"] = v.workspace + await self.audit.emit("iam.management", payload) + + @staticmethod + def _credential_type(operation): + if operation == "resolve-api-key": + return "api-key" + if operation == "login": + return "login-password" + return "anonymous" + @staticmethod def add_args(parser): AsyncProcessor.add_args(parser)