mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-09 21:32:10 +02:00
d5f3b6d9f6
Add a complete audit event pipeline that emits structured, machine- parseable events for every gateway request and IAM decision. Schema and publisher: - AuditEvent dataclass and notify-class queue (audit_events_queue) - AuditPublisher utility: fire-and-forget emission with envelope (schema_version, event_id, event_type, timestamp, producer) - New request_id and client_ip fields on IamRequest for correlation Gateway (gateway.request events): - aiohttp middleware assigns request_id, captures timing/status/sizes and emits an event after every HTTP request completes - IamAuth.authenticate annotates the request with identity - Main endpoint handlers annotate capability and workspace - request_id and client_ip forwarded to IAM on authenticate/authorise IAM service (iam.authenticate, iam.authorise, iam.management events): - Emits iam.authenticate for resolve-api-key, login, anonymous auth - Emits iam.authorise for authorise and authorise-many decisions - Emits iam.management for user/workspace/key mutations - All events include request_id for correlation with gateway events Design: events land on a pub/sub notify topic — non-persistent, per-subscriber delivery. If no audit consumer is deployed, events are silently discarded. Storage, retention, and alerting are consumer-side concerns outside this boundary. Added unit tests for the publisher and gateway middleware, unit tests for IAM audit emission, and a contract test for the AuditEvent schema. Tech spec: docs/tech-specs/audit-events.md
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""
|
|
Contract tests for schema dataclass field sets.
|
|
|
|
These pin the *field names* of small, widely-constructed schema dataclasses
|
|
so that any rename, removal, or accidental addition fails CI loudly instead
|
|
of waiting for a runtime TypeError on the next websocket message.
|
|
|
|
Background: in v2.2 the `Metadata` dataclass dropped a `metadata: list[Triple]`
|
|
field but several call sites kept passing `Metadata(metadata=...)`. The bug
|
|
was only discovered when a websocket import dispatcher received its first
|
|
real message in production. A trivial structural assertion of the kind
|
|
below would have caught it at unit-test time.
|
|
|
|
Add to this file whenever a schema rename burns you. The cost of a frozen
|
|
field set is a one-line update when you intentionally evolve the schema; the
|
|
benefit is that every call site is forced to come along for the ride.
|
|
"""
|
|
|
|
import dataclasses
|
|
import pytest
|
|
|
|
from trustgraph.schema import (
|
|
Metadata,
|
|
EntityContext,
|
|
EntityEmbeddings,
|
|
ChunkEmbeddings,
|
|
AuditEvent,
|
|
IamRequest,
|
|
)
|
|
|
|
|
|
def _field_names(dc):
|
|
return {f.name for f in dataclasses.fields(dc)}
|
|
|
|
|
|
@pytest.mark.contract
|
|
class TestSchemaFieldContracts:
|
|
"""Pin the field set of dataclasses that get constructed all over the
|
|
codebase. If you intentionally change one of these, update the
|
|
expected set in the same commit — that diff will surface every call
|
|
site that needs to come along."""
|
|
|
|
def test_metadata_fields(self):
|
|
# NOTE: there is no `metadata` field. A previous regression
|
|
# constructed Metadata(metadata=...) and crashed at runtime.
|
|
# `user` was also dropped in the workspace refactor — workspace
|
|
# now flows via flow.workspace, not via message payload.
|
|
assert _field_names(Metadata) == {
|
|
"id",
|
|
"root",
|
|
"collection",
|
|
}
|
|
|
|
def test_entity_embeddings_fields(self):
|
|
# NOTE: the embedding field is `vector` (singular, list[float]).
|
|
# There is no `vectors` field. Several call sites historically
|
|
# passed `vectors=` and crashed at runtime.
|
|
assert _field_names(EntityEmbeddings) == {
|
|
"entity",
|
|
"vector",
|
|
"chunk_id",
|
|
}
|
|
|
|
def test_chunk_embeddings_fields(self):
|
|
# Same `vector` (singular) convention as EntityEmbeddings.
|
|
assert _field_names(ChunkEmbeddings) == {
|
|
"chunk_id",
|
|
"vector",
|
|
}
|
|
|
|
def test_entity_context_fields(self):
|
|
assert _field_names(EntityContext) == {
|
|
"entity",
|
|
"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
|