diff --git a/conftest.py b/conftest.py index 0266c3c0594..fee9c0c46e6 100644 --- a/conftest.py +++ b/conftest.py @@ -348,6 +348,20 @@ async def mock_reencrypt_kiq(mocker) -> AsyncGenerator[Any, Any]: yield mock +@pytest.fixture(autouse=True) +def mock_audit_event_kiq(mocker: MockerFixture): + """Prevent audit events from being persisted during tests. + + ``audit.log()`` enqueues ``send_audit_event``, which the in-memory test + broker runs synchronously. That task opens its own DB session and commits + to ``audit_logs`` outside the test's transaction, leaking rows across tests. + Tests that assert audit behaviour mock ``log`` directly, and the audit + query/export tests seed ``audit_logs`` explicitly, so disabling the enqueue + here is safe. + """ + return mocker.patch("apps.audit.service.send_audit_event.kiq") + + @pytest.fixture(scope="session") def uuid_zero() -> uuid.UUID: return uuid.UUID("00000000-0000-0000-0000-000000000000") diff --git a/docker-compose.yaml b/docker-compose.yaml index 11c3bdcef0b..07ba6cbc1ee 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -70,9 +70,6 @@ services: - mailhog ports: - "8000:80" - networks: - - opensearch-net - - default develop: watch: - action: sync @@ -186,43 +183,7 @@ services: - './compose/minio:/etc/minio' entrypoint: /etc/minio/create_bucket.sh - opensearch: - image: opensearchproject/opensearch:latest - container_name: opensearch-node - environment: - discovery.type: single-node - OPENSEARCH_INITIAL_ADMIN_PASSWORD: "${OPENSEARCH__PASSWORD:-MyStrongPassword123!}" - node.name: opensearch - ports: - - "9200:9200" - - "9600:9600" - volumes: - - opensearch-data:/usr/share/opensearch/data - networks: - - opensearch-net - - opensearch-dashboards: - image: opensearchproject/opensearch-dashboards:latest - container_name: opensearch-dashboards - environment: - OPENSEARCH_HOSTS: '["https://opensearch:9200"]' - OPENSEARCH_USERNAME: admin - OPENSEARCH_PASSWORD: "${OPENSEARCH__PASSWORD:-MyStrongPassword123!}" - OPENSEARCH_SSL_VERIFICATIONMODE: none - explore.enabled: true - depends_on: - - opensearch - ports: - - "5601:5601" - networks: - - opensearch-net - - -networks: - opensearch-net: - volumes: pg_data: {} pg_arb_data: {} datastore: {} - opensearch-data: {} diff --git a/docs/audit-pipeline.md b/docs/audit-pipeline.md index 34d69534093..bc4b7c8cfa2 100644 --- a/docs/audit-pipeline.md +++ b/docs/audit-pipeline.md @@ -33,14 +33,16 @@ API process Worker process │ └──────── RabbitMQ queue ──────────► 3. send_audit_event(payload) │ - │ OpenSearchClient().index_document() - │ + │ AuditLogCRUD(session).save(...) + │ INSERT ... ON CONFLICT DO NOTHING ▼ - POST /audit-logs/_doc - → document stored in OpenSearch + row stored in the `audit_logs` + table (primary Postgres database) ``` -**The API never waits for OpenSearch.** As soon as the message is on the queue (step 2), the HTTP response can return. The write to OpenSearch happens in a separate worker process. +**The API never waits for the audit write.** As soon as the message is on the queue (step 2), the HTTP response can return. The write to Postgres happens in a separate worker process. + +Audit events are stored in the primary Postgres database. A few fields are denormalized into typed columns for filtering/sorting; the full ECS document is kept in a JSONB `payload` column and is what the export endpoint returns. --- @@ -51,11 +53,10 @@ API process Worker process | `apps/audit/domain.py` | `AuditEvent` — the Pydantic model that defines every field an audit event can have | | `apps/audit/enums.py` | `EventAction`, `EventOutcome` — the vocabulary of valid actions and outcomes | | `apps/audit/service.py` | `audit.log()` — the one function callers use. Serializes and enqueues. | -| `apps/audit/tasks.py` | `send_audit_event` — the Taskiq worker task. Writes to OpenSearch, handles retries. | -| `apps/audit/index_mapping.py` | OpenSearch field type definitions for the `audit-logs` index | -| `infrastructure/utility/opensearch_client.py` | Singleton `AsyncOpenSearch` wrapper used by the worker | -| `infrastructure/lifespan.py` | Creates the `audit-logs` index on app startup (once, idempotent) | -| `config/opensearch.py` | `OpenSearchSettings` — host, port, credentials, index name | +| `apps/audit/tasks.py` | `send_audit_event` — the Taskiq worker task. Writes to Postgres, handles retries. | +| `apps/audit/db/schemas.py` | `AuditLogSchema` — the `audit_logs` table (typed columns + JSONB `payload`) | +| `apps/audit/crud.py` | `AuditLogCRUD` — idempotent insert and the applet-scoped export query | +| `apps/audit/query_service.py` | `AuditQueryService` — rebuilds `AuditEvent`s from stored rows for the export endpoint | --- @@ -83,39 +84,34 @@ Only `user_id` and `event_action` are required. Everything else is optional and ## Failure handling -If OpenSearch is unavailable or returns an error: +If the database write fails: 1. The worker logs a warning and re-enqueues the task with a **5-second delay**. 2. This repeats up to **3 times** (configurable via `retries`). 3. After all retries are exhausted, the full event payload is logged at **ERROR level**, which captures it in Datadog. The event is not silently dropped. +Each event carries a unique `event_id`, stored in a unique column. The insert uses `ON CONFLICT (event_id) DO NOTHING`, so a retried task that re-delivers the same event never creates a duplicate row. + --- -## OpenSearch index +## Storage -The `audit-logs` index is created once on app startup by `startup_opensearch()` in `lifespan.py`. The mapping is defined explicitly in `index_mapping.py` — field types are chosen for the queries we expect: +Audit events live in the `audit_logs` table (`apps/audit/db/schemas.py`), created by an Alembic migration like any other table. -- `keyword` — IDs, enums, statuses (exact match, aggregation) -- `ip` — `client.ip` (CIDR range queries) -- `date` — `@timestamp` (time-range queries) -- `text` — URL paths, user agent strings (full-text search) +- Typed columns — `event_id`, `event_timestamp`, `event_action`, `event_outcome`, `user_id`, `applet_ids` — back the export query's filter and sort. +- `payload` (JSONB) holds the full event document; the export endpoint returns it verbatim. -Adding new **event actions** (new values of `EventAction`) requires no mapping changes. Adding new **fields** to `AuditEvent` requires a corresponding entry in `index_mapping.py`. +Indexes: +- unique on `event_id` (idempotency) +- `(event_timestamp, event_id)` — the date-range filter and sort order +- GIN on `applet_ids` — `:applet_id = ANY(applet_ids)` membership lookups ---- +Adding new **event actions** (new values of `EventAction`) requires no schema change. Adding new **fields** to `AuditEvent` requires no schema change either — they are stored in the JSONB `payload` automatically; only promote a field to its own column (with a migration) if you need to filter or sort on it. -## Configuration +> Note: the legacy OpenSearch modules (`infrastructure/utility/opensearch_client.py`, `apps/audit/index_mapping.py`, `config/opensearch.py`) remain in the tree but are no longer wired into the audit pipeline. -All OpenSearch settings are nested under `OPENSEARCH__` in the environment: +--- -```bash -OPENSEARCH__HOST=opensearch # default: opensearch (Docker service name) -OPENSEARCH__PORT=9200 # default: 9200 -OPENSEARCH__USER=admin # default: admin -OPENSEARCH__PASSWORD=admin # default: admin -OPENSEARCH__USE_SSL=True # default: True (required by the Docker image) -OPENSEARCH__VERIFY_CERTS=False # default: False (self-signed cert in local/dev) -OPENSEARCH__AUDIT_INDEX=audit-logs # default: audit-logs -``` +## Retention -For local development, if running the backend outside Docker, set `OPENSEARCH__HOST=localhost`. +There is currently no retention policy — the `audit_logs` table grows unbounded (the previous OpenSearch setup had no retention either). When retention becomes necessary, the recommended approach is native Postgres range partitioning by `event_timestamp` (e.g. monthly) with a scheduled job that drops partitions older than a configurable window. diff --git a/src/apps/audit/api.py b/src/apps/audit/api.py index 0fda99d786b..bd38b643e2f 100644 --- a/src/apps/audit/api.py +++ b/src/apps/audit/api.py @@ -41,7 +41,7 @@ async def applet_audit_export( if from_datetime and to_datetime and from_datetime > to_datetime: raise InvalidAuditDateRangeError(path=["fromDatetime"]) - events, total = await AuditQueryService().search_applet_events( + events, total = await AuditQueryService(session).search_applet_events( applet_id, from_datetime=from_datetime, to_datetime=to_datetime, diff --git a/src/apps/audit/crud.py b/src/apps/audit/crud.py new file mode 100644 index 00000000000..f89778405fa --- /dev/null +++ b/src/apps/audit/crud.py @@ -0,0 +1,74 @@ +import datetime +import uuid + +from sqlalchemy import and_, func, select +from sqlalchemy.dialects.postgresql import insert + +from apps.audit.db.schemas import AuditLogSchema +from infrastructure.database.crud import BaseCRUD + +DEFAULT_PAGE_SIZE = 1000 + + +def _to_naive_utc(value: datetime.datetime) -> datetime.datetime: + """Normalise a datetime to naive UTC to match the stored ``event_timestamp``.""" + if value.tzinfo is not None: + value = value.astimezone(datetime.timezone.utc).replace(tzinfo=None) + return value + + +class AuditLogCRUD(BaseCRUD[AuditLogSchema]): + schema_class = AuditLogSchema + + async def save(self, schema: AuditLogSchema) -> None: + """Insert an audit log row. + + Uses ``ON CONFLICT (event_id) DO NOTHING`` so that a retried worker + task (which re-delivers the same event) does not create a duplicate + row — preserving the idempotency OpenSearch gave us via ``id=event.id``. + """ + query = ( + insert(AuditLogSchema) + .values( + event_id=schema.event_id, + event_timestamp=schema.event_timestamp, + event_action=schema.event_action, + event_outcome=schema.event_outcome, + user_id=schema.user_id, + applet_ids=schema.applet_ids, + payload=schema.payload, + ) + .on_conflict_do_nothing(index_elements=[AuditLogSchema.event_id]) + ) + await self._execute(query) + + async def search_applet_events( + self, + applet_id: uuid.UUID, + *, + from_datetime: datetime.datetime | None = None, + to_datetime: datetime.datetime | None = None, + page: int = 1, + limit: int = DEFAULT_PAGE_SIZE, + ) -> tuple[list[AuditLogSchema], int]: + conditions = [AuditLogSchema.applet_ids.contains([applet_id])] + if from_datetime is not None: + conditions.append(AuditLogSchema.event_timestamp >= _to_naive_utc(from_datetime)) + if to_datetime is not None: + conditions.append(AuditLogSchema.event_timestamp < _to_naive_utc(to_datetime)) + + where = and_(*conditions) + + query = ( + select(AuditLogSchema) + .where(where) + .order_by(AuditLogSchema.event_timestamp.asc(), AuditLogSchema.event_id.asc()) + .limit(limit) + .offset((page - 1) * limit) + ) + rows = (await self._execute(query)).scalars().all() + + count_query = select(func.count()).select_from(AuditLogSchema).where(where) + total = (await self._execute(count_query)).scalar() or 0 + + return list(rows), total diff --git a/src/apps/audit/db/__init__.py b/src/apps/audit/db/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/src/apps/audit/db/schemas.py b/src/apps/audit/db/schemas.py new file mode 100644 index 00000000000..e88bace6ed2 --- /dev/null +++ b/src/apps/audit/db/schemas.py @@ -0,0 +1,30 @@ +from sqlalchemy import Column, DateTime, Index, String +from sqlalchemy.dialects.postgresql import ARRAY, JSONB, UUID + +from infrastructure.database.base import Base + + +class AuditLogSchema(Base): + """Audit event stored in Postgres. + + Append-only. A few columns are denormalized from the ECS document for + filtering/sorting; the full event is kept in ``payload`` as the same + dotted-alias JSON that is produced by ``AuditEvent.model_dump(mode="json")``. + """ + + __tablename__ = "audit_logs" + + event_id = Column(UUID(as_uuid=True), nullable=False, unique=True) + event_timestamp = Column(DateTime(), nullable=False) + event_action = Column(String(), nullable=False) + event_outcome = Column(String(), nullable=True) + user_id = Column(UUID(as_uuid=True), nullable=True, index=True) + applet_ids = Column(ARRAY(UUID(as_uuid=True)), nullable=True) + payload = Column(JSONB(), nullable=False) + + __table_args__ = ( + # Covers the date-range filter and the (timestamp, event_id) sort order. + Index("ix_audit_logs_event_timestamp_event_id", "event_timestamp", "event_id"), + # Covers `applet_ids @> ARRAY[:applet_id]` membership lookups. + Index("ix_audit_logs_applet_ids", "applet_ids", postgresql_using="gin"), + ) diff --git a/src/apps/audit/query_service.py b/src/apps/audit/query_service.py index 3f78bf3a08e..bab3face1d8 100644 --- a/src/apps/audit/query_service.py +++ b/src/apps/audit/query_service.py @@ -1,17 +1,15 @@ import datetime import uuid -from apps.audit.domain import AuditEvent -from config import settings -from infrastructure.utility.opensearch_client import DEFAULT_PAGE_SIZE, OpenSearchClient +from sqlalchemy.ext.asyncio import AsyncSession -SORT: list[dict] = [{"@timestamp": "asc"}, {"event.id": "asc"}] +from apps.audit.crud import DEFAULT_PAGE_SIZE, AuditLogCRUD +from apps.audit.domain import AuditEvent class AuditQueryService: - def __init__(self, client: OpenSearchClient | None = None) -> None: - self._client = client or OpenSearchClient() - self._index = settings.opensearch.audit_index + def __init__(self, session: AsyncSession) -> None: + self._crud = AuditLogCRUD(session) async def search_applet_events( self, @@ -22,37 +20,12 @@ async def search_applet_events( page: int = 1, limit: int = DEFAULT_PAGE_SIZE, ) -> tuple[list[AuditEvent], int]: - query = self._build_query(applet_id, from_datetime, to_datetime) - response = await self._client.search( - self._index, - query=query, - sort=SORT, - size=limit, - from_=(page - 1) * limit, + rows, total = await self._crud.search_applet_events( + applet_id, + from_datetime=from_datetime, + to_datetime=to_datetime, + page=page, + limit=limit, ) - hits = response.get("hits", {}) - total = hits.get("total", {}).get("value", 0) - events = [AuditEvent.model_validate(hit["_source"]) for hit in hits.get("hits", [])] + events = [AuditEvent.model_validate(row.payload) for row in rows] return events, total - - @staticmethod - def _build_query( - applet_id: uuid.UUID, - from_datetime: datetime.datetime | None, - to_datetime: datetime.datetime | None, - ) -> dict: - filters: list[dict] = [{"term": {"curious.applet_id": str(applet_id)}}] - - timestamp_range: dict = {} - if from_datetime is not None: - if from_datetime.tzinfo is None: - from_datetime = from_datetime.replace(tzinfo=datetime.timezone.utc) - timestamp_range["gte"] = from_datetime.isoformat() - if to_datetime is not None: - if to_datetime.tzinfo is None: - to_datetime = to_datetime.replace(tzinfo=datetime.timezone.utc) - timestamp_range["lt"] = to_datetime.isoformat() - if timestamp_range: - filters.append({"range": {"@timestamp": timestamp_range}}) - - return {"bool": {"filter": filters}} diff --git a/src/apps/audit/tasks.py b/src/apps/audit/tasks.py index 57b569e7ef5..0a4e2933a37 100644 --- a/src/apps/audit/tasks.py +++ b/src/apps/audit/tasks.py @@ -1,20 +1,50 @@ +import datetime +import uuid + +from apps.audit.crud import AuditLogCRUD +from apps.audit.db.schemas import AuditLogSchema from broker import broker -from config import settings +from infrastructure.database import atomic, session_manager from infrastructure.logger import logger -from infrastructure.utility.opensearch_client import OpenSearchClient + + +def _build_schema(payload: dict) -> AuditLogSchema: + """Map the dotted-alias ECS payload to an ``AuditLogSchema`` row. + + The full payload is stored as-is in the JSONB ``payload`` column; a few + fields are denormalized into typed columns for filtering and sorting. + """ + timestamp = datetime.datetime.fromisoformat(payload["@timestamp"]) + if timestamp.tzinfo is not None: + timestamp = timestamp.astimezone(datetime.timezone.utc).replace(tzinfo=None) + + user_id = payload.get("user.id") + applet_ids = payload.get("curious.applet_id") + + return AuditLogSchema( + event_id=uuid.UUID(payload["event.id"]), + event_timestamp=timestamp, + event_action=payload["event.action"], + event_outcome=payload.get("event.outcome"), + user_id=uuid.UUID(user_id) if user_id else None, + applet_ids=[uuid.UUID(applet_id) for applet_id in applet_ids] if applet_ids else None, + payload=payload, + ) @broker.task() async def send_audit_event(payload: dict, retries: int = 3) -> None: - """Index an audit event into OpenSearch. + """Persist an audit event to Postgres. Retries on failure with a 5s delay; on final failure logs the payload so it is captured by the structured log pipeline (Datadog) instead of being silently dropped. """ try: - doc_id = str(payload["event.id"]) if payload.get("event.id") else None - await OpenSearchClient().index_document(settings.opensearch.audit_index, payload, id=doc_id) + session_maker = session_manager.get_session() + async with session_maker() as session: + async with atomic(session): + await AuditLogCRUD(session).save(_build_schema(payload)) except Exception as e: if retries > 0: logger.warning("audit_event_retry", retries_left=retries, error=str(e)) diff --git a/src/apps/audit/tests/test_api.py b/src/apps/audit/tests/test_api.py index aaa01085bec..f1f8c217b72 100644 --- a/src/apps/audit/tests/test_api.py +++ b/src/apps/audit/tests/test_api.py @@ -1,22 +1,52 @@ +import datetime import http import uuid import pytest +from pytest_mock import MockerFixture +from sqlalchemy.ext.asyncio import AsyncSession from apps.applets.domain.applet_full import AppletFull +from apps.audit.crud import AuditLogCRUD +from apps.audit.domain import AuditEvent from apps.audit.enums import EventAction +from apps.audit.tasks import _build_schema from apps.shared.test.client import TestClient from apps.users.domain import User -from infrastructure.utility.opensearch_client import OpenSearchClientTest URL = "/audit/applets/{applet_id}/events" @pytest.fixture(autouse=True) -def reset_opensearch(): - OpenSearchClientTest._storage = {} - OpenSearchClientTest._indices = set() - OpenSearchClientTest.last_search_body = {} +def mock_audit_log(mocker: MockerFixture): + """Stub the export endpoint's self-logging. + + The endpoint emits an ``applet:audit:export`` event via ``log()``. Under the + in-memory test broker this runs the worker task synchronously, which would + open its own DB session and write outside the test transaction. Mocking + ``log`` keeps the test isolated and lets us assert the self-log separately. + """ + return mocker.patch("apps.audit.api.log") + + +async def _seed_event( + session: AsyncSession, + applet_id: uuid.UUID, + *, + user_id: uuid.UUID, + timestamp: datetime.datetime, + event_id: uuid.UUID | None = None, + action: EventAction = EventAction.APPLET_ANSWER_VIEW, +) -> AuditEvent: + event = AuditEvent( + event_action=action, + user_id=user_id, + curious_applet_id=[applet_id], + timestamp=timestamp, + event_id=event_id or uuid.uuid4(), + ) + await AuditLogCRUD(session).save(_build_schema(event.model_dump(mode="json"))) + return event async def test_unauthenticated_request_returns_401(client: TestClient, applet_one: AppletFull): @@ -60,24 +90,21 @@ async def test_invalid_date_range_returns_422(client: TestClient, tom: User, app assert response.status_code == http.HTTPStatus.UNPROCESSABLE_ENTITY -async def test_returns_seeded_audit_event(client: TestClient, tom: User, applet_one: AppletFull): - client.login(tom) - - OpenSearchClientTest._storage["audit-logs"] = [ - { - "@timestamp": "2026-05-01T10:00:00+00:00", - "event.id": "11111111-1111-1111-1111-111111111111", - "event.action": "applet:answer:view", - "user.id": str(tom.id), - "curious.applet_id": [str(applet_one.id)], - } - ] +async def test_returns_seeded_audit_event(client: TestClient, session: AsyncSession, tom: User, applet_one: AppletFull): + event = await _seed_event( + session, + applet_one.id, + user_id=tom.id, + timestamp=datetime.datetime(2026, 5, 1, 10, 0, 0), + ) + client.login(tom) response = await client.get(URL.format(applet_id=applet_one.id)) assert response.status_code == http.HTTPStatus.OK body = response.json() assert body["count"] == 1 - assert body["result"][0]["event.action"] == "applet:answer:view" + assert body["result"][0]["event.id"] == str(event.event_id) + assert body["result"][0]["event.action"] == EventAction.APPLET_ANSWER_VIEW.value assert body["result"][0]["user.id"] == str(tom.id) @@ -87,38 +114,33 @@ async def test_returns_404_when_applet_missing(client: TestClient, tom: User): assert response.status_code == http.HTTPStatus.NOT_FOUND -async def test_date_range_reaches_opensearch_query(client: TestClient, tom: User, applet_one: AppletFull): +async def test_date_range_filters_results(client: TestClient, session: AsyncSession, tom: User, applet_one: AppletFull): + await _seed_event(session, applet_one.id, user_id=tom.id, timestamp=datetime.datetime(2026, 5, 1, 10, 0, 0)) + inside = await _seed_event( + session, applet_one.id, user_id=tom.id, timestamp=datetime.datetime(2026, 5, 5, 10, 0, 0) + ) + await _seed_event(session, applet_one.id, user_id=tom.id, timestamp=datetime.datetime(2026, 5, 10, 10, 0, 0)) + client.login(tom) response = await client.get( URL.format(applet_id=applet_one.id), - dict(fromDatetime="2026-05-01T14:30:00", toDatetime="2026-05-07T18:00:00"), + dict(fromDatetime="2026-05-03T00:00:00", toDatetime="2026-05-07T00:00:00"), ) assert response.status_code == http.HTTPStatus.OK - - filters = OpenSearchClientTest.last_search_body["query"]["bool"]["filter"] - range_clause = next(f for f in filters if "range" in f) - assert range_clause["range"]["@timestamp"] == { - "gte": "2026-05-01T14:30:00+00:00", - "lt": "2026-05-07T18:00:00+00:00", - } + body = response.json() + assert body["count"] == 1 + assert body["result"][0]["event.id"] == str(inside.event_id) async def test_export_self_logs_applet_audit_export( - client: TestClient, tom: User, applet_one: AppletFull, monkeypatch: pytest.MonkeyPatch + client: TestClient, tom: User, applet_one: AppletFull, mock_audit_log ): - captured = [] - - async def fake_log(event): - captured.append(event) - - monkeypatch.setattr("apps.audit.api.log", fake_log) - client.login(tom) response = await client.get(URL.format(applet_id=applet_one.id)) assert response.status_code == http.HTTPStatus.OK - assert len(captured) == 1 - event = captured[0] + mock_audit_log.assert_awaited_once() + event = mock_audit_log.call_args[0][0] assert event.event_action == EventAction.APPLET_AUDIT_EXPORT assert event.user_id == tom.id assert event.curious_applet_id == [applet_one.id] diff --git a/src/apps/audit/tests/test_query_service.py b/src/apps/audit/tests/test_query_service.py index e38407e5cc7..d92c92407d7 100644 --- a/src/apps/audit/tests/test_query_service.py +++ b/src/apps/audit/tests/test_query_service.py @@ -1,82 +1,114 @@ import datetime import uuid -import pytest +from sqlalchemy.ext.asyncio import AsyncSession +from apps.audit.crud import AuditLogCRUD from apps.audit.domain import AuditEvent +from apps.audit.enums import EventAction from apps.audit.query_service import AuditQueryService -from infrastructure.utility.opensearch_client import OpenSearchClient, OpenSearchClientTest +from apps.audit.tasks import _build_schema + + +def _make_event( + applet_id: uuid.UUID, + *, + timestamp: datetime.datetime, + event_id: uuid.UUID | None = None, + user_id: uuid.UUID | None = None, +) -> AuditEvent: + return AuditEvent( + event_action=EventAction.APPLET_ANSWER_VIEW, + user_id=user_id or uuid.uuid4(), + curious_applet_id=[applet_id], + timestamp=timestamp, + event_id=event_id or uuid.uuid4(), + ) + -INDEX = "audit-logs" +async def _seed(session: AsyncSession, event: AuditEvent) -> None: + await AuditLogCRUD(session).save(_build_schema(event.model_dump(mode="json"))) -@pytest.fixture -def fresh_service() -> AuditQueryService: - OpenSearchClientTest._storage = {} - OpenSearchClientTest._indices = set() - OpenSearchClientTest.last_search_body = {} - OpenSearchClient._initialized = False - OpenSearchClient._instance = None - return AuditQueryService() +async def test_filters_by_applet(session: AsyncSession): + applet_id = uuid.uuid4() + other_applet_id = uuid.uuid4() + ts = datetime.datetime(2026, 5, 1, 10, 0, 0) + await _seed(session, _make_event(applet_id, timestamp=ts)) + await _seed(session, _make_event(other_applet_id, timestamp=ts)) -def _seed_doc(**overrides: object) -> dict: - base: dict = { - "@timestamp": "2026-05-01T10:00:00+00:00", - "event.id": str(uuid.uuid4()), - "event.action": "applet:answer:view", - "user.id": str(uuid.uuid4()), - "curious.applet_id": [str(uuid.uuid4())], - } - base.update(overrides) - return base + events, total = await AuditQueryService(session).search_applet_events(applet_id) + assert total == 1 + assert len(events) == 1 + assert events[0].curious_applet_id == [applet_id] -async def test_query_filters_by_applet_and_dates(fresh_service: AuditQueryService): - applet_id = uuid.uuid4() - from_dt = datetime.datetime(2026, 5, 1, 14, 30, 0, tzinfo=datetime.timezone.utc) - to_dt = datetime.datetime(2026, 5, 7, 18, 0, 0, tzinfo=datetime.timezone.utc) +async def test_filters_by_date_range(session: AsyncSession): + applet_id = uuid.uuid4() + before = _make_event(applet_id, timestamp=datetime.datetime(2026, 5, 1, 10, 0, 0)) + inside = _make_event(applet_id, timestamp=datetime.datetime(2026, 5, 5, 10, 0, 0)) + after = _make_event(applet_id, timestamp=datetime.datetime(2026, 5, 10, 10, 0, 0)) + for event in (before, inside, after): + await _seed(session, event) - await fresh_service.search_applet_events( + events, total = await AuditQueryService(session).search_applet_events( applet_id, - from_datetime=from_dt, - to_datetime=to_dt, + from_datetime=datetime.datetime(2026, 5, 3, 0, 0, 0, tzinfo=datetime.timezone.utc), + to_datetime=datetime.datetime(2026, 5, 7, 0, 0, 0, tzinfo=datetime.timezone.utc), ) - body = OpenSearchClientTest.last_search_body - filters = body["query"]["bool"]["filter"] - assert filters[0] == {"term": {"curious.applet_id": str(applet_id)}} - assert filters[1] == { - "range": { - "@timestamp": { - "gte": "2026-05-01T14:30:00+00:00", - "lt": "2026-05-07T18:00:00+00:00", - } - } - } - assert body["sort"] == [{"@timestamp": "asc"}, {"event.id": "asc"}] + assert total == 1 + assert [e.event_id for e in events] == [inside.event_id] + + +async def test_results_sorted_by_timestamp_ascending(session: AsyncSession): + applet_id = uuid.uuid4() + third = _make_event(applet_id, timestamp=datetime.datetime(2026, 5, 9, 10, 0, 0)) + first = _make_event(applet_id, timestamp=datetime.datetime(2026, 5, 1, 10, 0, 0)) + second = _make_event(applet_id, timestamp=datetime.datetime(2026, 5, 5, 10, 0, 0)) + # Seed out of chronological order. + for event in (third, first, second): + await _seed(session, event) + + events, _ = await AuditQueryService(session).search_applet_events(applet_id) + + assert [e.event_id for e in events] == [first.event_id, second.event_id, third.event_id] + + +async def test_pagination(session: AsyncSession): + applet_id = uuid.uuid4() + seeded = [] + for day in range(1, 6): + event = _make_event(applet_id, timestamp=datetime.datetime(2026, 5, day, 10, 0, 0)) + seeded.append(event) + await _seed(session, event) + + page1, total = await AuditQueryService(session).search_applet_events(applet_id, page=1, limit=2) + page2, _ = await AuditQueryService(session).search_applet_events(applet_id, page=2, limit=2) + page3, _ = await AuditQueryService(session).search_applet_events(applet_id, page=3, limit=2) + assert total == 5 + assert [e.event_id for e in page1] == [seeded[0].event_id, seeded[1].event_id] + assert [e.event_id for e in page2] == [seeded[2].event_id, seeded[3].event_id] + assert [e.event_id for e in page3] == [seeded[4].event_id] -async def test_query_omits_range_when_dates_missing(fresh_service: AuditQueryService): - await fresh_service.search_applet_events(uuid.uuid4()) - filters = OpenSearchClientTest.last_search_body["query"]["bool"]["filter"] - assert len(filters) == 1 - assert "range" not in filters[0] +async def test_returns_empty_when_no_events(session: AsyncSession): + events, total = await AuditQueryService(session).search_applet_events(uuid.uuid4()) + assert events == [] + assert total == 0 -async def test_yields_audit_events_in_storage_order(fresh_service: AuditQueryService): +async def test_save_is_idempotent_on_event_id(session: AsyncSession): applet_id = uuid.uuid4() - seeded_ids = [] - for _ in range(3): - doc = _seed_doc(**{"curious.applet_id": [str(applet_id)]}) - seeded_ids.append(doc["event.id"]) - await OpenSearchClient().index_document(INDEX, doc) - - out, total = await fresh_service.search_applet_events(applet_id) - - assert total == 3 - assert len(out) == 3 - assert all(isinstance(e, AuditEvent) for e in out) - assert [str(e.event_id) for e in out] == seeded_ids + event = _make_event(applet_id, timestamp=datetime.datetime(2026, 5, 1, 10, 0, 0)) + + await _seed(session, event) + # Re-deliver the same event (simulating a worker retry). + await _seed(session, event) + + events, total = await AuditQueryService(session).search_applet_events(applet_id) + assert total == 1 + assert len(events) == 1 diff --git a/src/config/__init__.py b/src/config/__init__.py index 763a6fe4aef..f82f06fd9f2 100644 --- a/src/config/__init__.py +++ b/src/config/__init__.py @@ -140,6 +140,7 @@ def uploads_dir(self): "authentication", "job", "subjects", + "audit", "integrations", "integrations.loris", ], diff --git a/src/infrastructure/database/migrations/versions/2026_06_17_15_47-create_audit_logs_table.py b/src/infrastructure/database/migrations/versions/2026_06_17_15_47-create_audit_logs_table.py new file mode 100644 index 00000000000..d7c369d5c98 --- /dev/null +++ b/src/infrastructure/database/migrations/versions/2026_06_17_15_47-create_audit_logs_table.py @@ -0,0 +1,59 @@ +"""create audit_logs table + +Revision ID: f475633a2836 +Revises: 8c88d334aba6 +Create Date: 2026-06-17 15:47:36.654683 + +""" + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision = "f475633a2836" +down_revision = "8c88d334aba6" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "audit_logs", + sa.Column("id", postgresql.UUID(as_uuid=True), server_default=sa.text("gen_random_uuid()"), nullable=False), + sa.Column("created_at", sa.DateTime(), server_default=sa.text("timezone('utc', now())"), nullable=True), + sa.Column("updated_at", sa.DateTime(), server_default=sa.text("timezone('utc', now())"), nullable=True), + sa.Column("migrated_date", sa.DateTime(), nullable=True), + sa.Column("migrated_updated", sa.DateTime(), nullable=True), + sa.Column("is_deleted", sa.Boolean(), nullable=True), + sa.Column("event_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("event_timestamp", sa.DateTime(), nullable=False), + sa.Column("event_action", sa.String(), nullable=False), + sa.Column("event_outcome", sa.String(), nullable=True), + sa.Column("user_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("applet_ids", postgresql.ARRAY(postgresql.UUID(as_uuid=True)), nullable=True), + sa.Column("payload", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.PrimaryKeyConstraint("id", name=op.f("pk_audit_logs")), + sa.UniqueConstraint("event_id", name=op.f("uq_audit_logs_event_id")), + ) + op.create_index(op.f("ix_audit_logs_user_id"), "audit_logs", ["user_id"], unique=False) + op.create_index( + "ix_audit_logs_event_timestamp_event_id", + "audit_logs", + ["event_timestamp", "event_id"], + unique=False, + ) + op.create_index( + "ix_audit_logs_applet_ids", + "audit_logs", + ["applet_ids"], + unique=False, + postgresql_using="gin", + ) + + +def downgrade() -> None: + op.drop_index("ix_audit_logs_applet_ids", table_name="audit_logs", postgresql_using="gin") + op.drop_index("ix_audit_logs_event_timestamp_event_id", table_name="audit_logs") + op.drop_index(op.f("ix_audit_logs_user_id"), table_name="audit_logs") + op.drop_table("audit_logs") diff --git a/src/infrastructure/lifespan.py b/src/infrastructure/lifespan.py index 7bdf69f84d3..a99b8bb7dfa 100644 --- a/src/infrastructure/lifespan.py +++ b/src/infrastructure/lifespan.py @@ -1,10 +1,7 @@ from fastapi import FastAPI -from apps.audit.index_mapping import AUDIT_LOG_MAPPING from broker import broker -from config import settings from infrastructure.logger import logger -from infrastructure.utility.opensearch_client import OpenSearchClient async def startup_taskiq() -> None: @@ -19,19 +16,9 @@ async def shutdown_taskiq() -> None: await broker.shutdown() -async def startup_opensearch() -> None: - logger.info("OpenSearch index ensure", index=settings.opensearch.audit_index) - await OpenSearchClient().ensure_index(settings.opensearch.audit_index, AUDIT_LOG_MAPPING) - - -async def shutdown_opensearch() -> None: - await OpenSearchClient().close() - - def startup(app: FastAPI): async def _startup(): await startup_taskiq() - await startup_opensearch() return _startup @@ -39,6 +26,5 @@ async def _startup(): def shutdown(app: FastAPI): async def _shutdown(): await shutdown_taskiq() - await shutdown_opensearch() return _shutdown