From 92c05089c742f25d31a0056ffdb8d26eea89ab2b Mon Sep 17 00:00:00 2001 From: arc53-machine <232052973+arc53-machine@users.noreply.github.com> Date: Tue, 22 Sep 2026 10:23:39 +0100 Subject: [PATCH] feat(admin): merged activity feed with filters and export DocsGPT keeps three append-only journals -- auth_events, device_audit_log and guardrail_events -- each readable only on its own terms, and the second of those had an admin endpoint that no UI ever called. An operator reviewing an incident wants one timeline, not three. Adds GET /api/admin/activity: all three journals projected onto a common row shape and merged, ordered newest first, with facets for journal, category, event name, actor, affected user, a time window and free-text search over the detail payload. A category filter that only one journal can satisfy drops the others from the union rather than scanning and discarding them. GET /api/admin/activity/events returns the distinct (event, category) pairs the instance has actually recorded, so the filter UI can offer real choices instead of asking an operator to type "oidc_login_denied" from memory. GET /api/admin/activity/export streams the same filtered feed as CSV or NDJSON. Streamed rather than buffered, and capped, so a compliance export of a busy instance is bounded work. guardrail_events.api_key (a raw agent key) and matched_value (unredacted source text) are excluded from the projection, mirroring the exclusion the per-agent guardrail view already makes. Admin-gating is not a reason to widen what a list response carries. --- docsgpt/api/admin/__init__.py | 1 + docsgpt/api/admin/activity.py | 234 ++++++++++++ docsgpt/api/audit.py | 56 +-- docsgpt/audit_events.py | 104 ++++++ docsgpt/storage/db/repositories/activity.py | 332 ++++++++++++++++++ tests/api/test_admin_activity.py | 236 +++++++++++++ tests/api/test_audit_helper.py | 23 +- .../storage/db/repositories/test_activity.py | 165 +++++++++ 8 files changed, 1096 insertions(+), 55 deletions(-) create mode 100644 docsgpt/api/admin/activity.py create mode 100644 docsgpt/audit_events.py create mode 100644 docsgpt/storage/db/repositories/activity.py create mode 100644 tests/api/test_admin_activity.py create mode 100644 tests/storage/db/repositories/test_activity.py diff --git a/docsgpt/api/admin/__init__.py b/docsgpt/api/admin/__init__.py index 7bfc3059..1dcdd686 100644 --- a/docsgpt/api/admin/__init__.py +++ b/docsgpt/api/admin/__init__.py @@ -1,4 +1,5 @@ from .routes import admin_ns from . import quotas # noqa: F401 (registers the quota resources on admin_ns) +from . import activity # noqa: F401 (registers the activity resources on admin_ns) __all__ = ["admin_ns"] diff --git a/docsgpt/api/admin/activity.py b/docsgpt/api/admin/activity.py new file mode 100644 index 00000000..832b2686 --- /dev/null +++ b/docsgpt/api/admin/activity.py @@ -0,0 +1,234 @@ +"""Admin activity feed: one timeline over all three audit journals. + +``GET /api/admin/audit`` remains the ``auth_events``-only feed. These endpoints +add the merged view an incident review actually wants — identity, access, +configuration, data-plane, remote-device and guardrail activity in one ordered +list — plus the filters and the export that make a review possible without +shelling into the database. + +Every resource here is ``@admin_required``, like the rest of the namespace. +""" + +from __future__ import annotations + +import csv +import io +import json +import logging +from datetime import datetime, timezone +from typing import Iterator, Optional + +from flask import Response, jsonify, make_response, request, stream_with_context +from flask_restx import Resource + +from docsgpt.api.admin.routes import _int_arg, _page, admin_ns +from docsgpt.api.user.authz import admin_required +from docsgpt.audit_events import ACTIVITY_CATEGORIES +from docsgpt.storage.db.repositories.activity import ( + ACTIVITY_COLUMNS, + ActivityRepository, +) +from docsgpt.storage.db.session import db_readonly + +logger = logging.getLogger(__name__) + +_FEEDS = ("auth", "device", "guardrail") +_EXPORT_FORMATS = ("csv", "ndjson") + +# An export runs against a live instance, so it is bounded. Operators needing +# the full history take a database dump, not an HTTP request. +_EXPORT_MAX_ROWS = 100_000 +_EXPORT_CHUNK = 1_000 + +_MAX_SEARCH_LENGTH = 200 + + +def _csv_list(name: str, allowed: tuple[str, ...]) -> Optional[list[str]]: + """Parse a repeated/comma-separated query arg, dropping unknown values. + + Args: + name: Query parameter name. + allowed: The permitted values; anything else is ignored rather than + rejected, so an old bookmark degrades instead of erroring. + + Returns: + The distinct valid values in the order given, or None when absent. + """ + raw: list[str] = [] + for value in request.args.getlist(name): + raw.extend(part.strip() for part in value.split(",")) + seen: list[str] = [] + for value in raw: + if value in allowed and value not in seen: + seen.append(value) + return seen or None + + +def _event_list() -> Optional[list[str]]: + """Event names to filter on. Unconstrained — the catalogue is data, not an enum.""" + names: list[str] = [] + for value in request.args.getlist("event"): + for part in value.split(","): + part = part.strip() + if part and part not in names: + names.append(part) + return names or None + + +def _time_arg(name: str) -> Optional[datetime]: + """Parse an ISO-8601 timestamp query arg, or None when absent/unparseable.""" + raw = (request.args.get(name) or "").strip() + if not raw: + return None + try: + parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except ValueError: + return None + # A naive bound is read as UTC, matching how the feed stores timestamps. + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + + +def _search_arg() -> Optional[str]: + value = (request.args.get("search") or "").strip() + return value[:_MAX_SEARCH_LENGTH] or None + + +def _filters() -> dict: + """The filter set shared by the feed and the export.""" + return { + "feeds": _csv_list("feed", _FEEDS), + "categories": _csv_list("category", ACTIVITY_CATEGORIES), + "events": _event_list(), + "actor_id": request.args.get("actor_id") or None, + "user_id": request.args.get("user_id") or None, + "since": _time_arg("since"), + "until": _time_arg("until"), + "search": _search_arg(), + } + + +def _serialize(row: dict) -> dict: + """Make one feed row JSON-safe (timestamps to ISO-8601).""" + created = row.get("created_at") + return { + **row, + "created_at": created.isoformat() if hasattr(created, "isoformat") else created, + } + + +@admin_ns.route("/admin/activity") +class AdminActivityResource(Resource): + @admin_required + def get(self): + """Merged audit feed, newest first, with filters and a total.""" + page, page_size, offset = _page() + filters = _filters() + with db_readonly() as conn: + repo = ActivityRepository(conn) + total = repo.count(**filters) + rows = repo.list(limit=page_size, offset=offset, **filters) + return make_response( + jsonify( + { + "success": True, + "activity": [_serialize(row) for row in rows], + "page": page, + "page_size": page_size, + "total": total, + "has_more": offset + len(rows) < total, + } + ), + 200, + ) + + +@admin_ns.route("/admin/activity/events") +class AdminActivityCatalogueResource(Resource): + @admin_required + def get(self): + """Distinct ``(event, category)`` pairs this instance has recorded. + + The filter UI offers these instead of a free-text box, so an operator + never has to guess that denied logins are ``oidc_login_denied``. + """ + with db_readonly() as conn: + catalogue = ActivityRepository(conn).event_names() + return make_response( + jsonify( + { + "success": True, + "events": catalogue, + "categories": list(ACTIVITY_CATEGORIES), + "feeds": list(_FEEDS), + } + ), + 200, + ) + + +@admin_ns.route("/admin/activity/export") +class AdminActivityExportResource(Resource): + @admin_required + def get(self): + """Stream the filtered feed as CSV or NDJSON. + + Streamed, not buffered: a compliance export of a busy instance should + not be built in memory first. Capped at ``_EXPORT_MAX_ROWS``; the + response header says whether the cap was reached. + """ + fmt = (request.args.get("format") or "csv").lower() + if fmt not in _EXPORT_FORMATS: + return make_response( + jsonify({"success": False, "message": "Invalid format"}), 400 + ) + limit = max(1, min(_EXPORT_MAX_ROWS, _int_arg("max_rows", _EXPORT_MAX_ROWS))) + filters = _filters() + stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") + generator = _csv_rows if fmt == "csv" else _ndjson_rows + return Response( + stream_with_context(generator(filters, limit)), + mimetype="text/csv" if fmt == "csv" else "application/x-ndjson", + headers={ + "Content-Disposition": ( + f'attachment; filename="docsgpt-activity-{stamp}.{fmt}"' + ), + # Partial content would otherwise be indistinguishable from a + # complete export. + "X-Export-Max-Rows": str(limit), + "Cache-Control": "no-store", + }, + ) + + +def _export_rows(filters: dict, limit: int) -> Iterator[dict]: + """Walk the filtered feed, holding the connection open for the whole stream.""" + with db_readonly() as conn: + yield from ActivityRepository(conn).iter_all( + chunk_size=_EXPORT_CHUNK, max_rows=limit, **filters + ) + + +def _csv_rows(filters: dict, limit: int) -> Iterator[str]: + buffer = io.StringIO() + writer = csv.DictWriter(buffer, fieldnames=list(ACTIVITY_COLUMNS)) + writer.writeheader() + yield _drain(buffer) + for row in _export_rows(filters, limit): + serialized = _serialize(row) + # ``detail`` is a JSON object; a CSV cell holds its compact encoding. + serialized["detail"] = json.dumps(serialized.get("detail") or {}) + writer.writerow(serialized) + yield _drain(buffer) + + +def _ndjson_rows(filters: dict, limit: int) -> Iterator[str]: + for row in _export_rows(filters, limit): + yield json.dumps(_serialize(row), default=str) + "\n" + + +def _drain(buffer: io.StringIO) -> str: + """Return and clear the writer's buffer, so each row streams as written.""" + value = buffer.getvalue() + buffer.seek(0) + buffer.truncate(0) + return value diff --git a/docsgpt/api/audit.py b/docsgpt/api/audit.py index ed6c22d2..0ca37827 100644 --- a/docsgpt/api/audit.py +++ b/docsgpt/api/audit.py @@ -1,4 +1,4 @@ -"""Shared audit-trail helper and the event → category map. +"""Shared audit-trail helper for recording data-plane and identity events. Identity and access events (login, role grants, provisioning) have been audited since ``auth_events`` was introduced. Data-plane actions — creating and @@ -23,61 +23,13 @@ from typing import Any, Optional from flask import has_request_context, request +from docsgpt.audit_events import ACTIVITY_CATEGORIES, category_for from docsgpt.storage.db.repositories.auth_events import AuthEventsRepository logger = logging.getLogger(__name__) - -# Coarse grouping for the admin activity feed's facet filter. Keyed by exact -# event name or by a ``prefix.`` namespace; see :func:`category_for`. -EVENT_CATEGORIES: dict[str, str] = { - # Who you are: authentication, provisioning, personal credentials. - "oidc_login": "identity", - "oidc_login_denied": "identity", - "oidc_refresh": "identity", - "backchannel_logout": "identity", - "scim_created": "identity", - "scim_deactivated": "identity", - "scim_reactivated": "identity", - "pat_created": "identity", - "pat_revoked": "identity", - "pat_regenerated": "identity", - # What you may do: roles, account state, team membership and sharing. - "role_granted": "access", - "role_revoked": "access", - "admin_user_activated": "access", - "admin_user_deactivated": "access", - "admin_sessions_revoked": "access", - "team.": "access", - # How the instance is configured. - "quota_policy_set": "config", - "quota_policy_deleted": "config", - # What the data looks like. - "source.": "data", - "agent.": "data", - "conversation.": "data", - # Everything an older release wrote that this one does not know about. - "other": "other", -} - - -def category_for(event: str) -> str: - """Return the activity category for ``event``. - - Args: - event: The audit event name. - - Returns: - One of ``identity``, ``access``, ``config``, ``data``, or ``other`` - for names this release does not recognise. - """ - exact = EVENT_CATEGORIES.get(event) - if exact is not None and not event.endswith("."): - return exact - namespace, dot, _ = event.partition(".") - if dot: - return EVENT_CATEGORIES.get(f"{namespace}.", "other") - return "other" +# Re-exported so route modules have one import for recording and classifying. +__all__ = ["ACTIVITY_CATEGORIES", "category_for", "record_event"] def record_event( diff --git a/docsgpt/audit_events.py b/docsgpt/audit_events.py new file mode 100644 index 00000000..db6bf2e3 --- /dev/null +++ b/docsgpt/audit_events.py @@ -0,0 +1,104 @@ +"""Audit event taxonomy: the categories the admin activity feed filters on. + +Lives at the top level (like ``docsgpt/pricing.py``) because both the API +layer — which records events — and the storage layer — which has to filter a +merged feed in SQL — need the same map, and neither should import the other. + +Two representations of one table: + +* :func:`category_for` classifies a single event name in Python. +* :func:`category_case_sql` emits the equivalent ``CASE`` expression so a + category filter can be pushed into the query instead of enumerating every + event name a release might have written. +""" + +from __future__ import annotations + +# Categories in the order the UI shows them. ``other`` catches event names an +# older release wrote that this one does not know about — the feed must never +# hide a row just because its name is unfamiliar. +ACTIVITY_CATEGORIES = ("identity", "access", "config", "data", "device", "safety", "other") + +# Exact event names. +_EXACT: dict[str, str] = { + # Who you are: authentication, provisioning, personal credentials. + "oidc_login": "identity", + "oidc_login_denied": "identity", + "oidc_refresh": "identity", + "backchannel_logout": "identity", + "scim_created": "identity", + "scim_deactivated": "identity", + "scim_reactivated": "identity", + "pat_created": "identity", + "pat_revoked": "identity", + "pat_regenerated": "identity", + # What you may do: roles, account state. + "role_granted": "access", + "role_revoked": "access", + "admin_user_activated": "access", + "admin_user_deactivated": "access", + "admin_sessions_revoked": "access", + # How the instance is configured. + "quota_policy_set": "config", + "quota_policy_deleted": "config", +} + +# Dotted namespaces, matched on the part before the first ``.``. +_NAMESPACES: dict[str, str] = { + "team": "access", + "source": "data", + "agent": "data", + "conversation": "data", + "device": "device", + "guardrail": "safety", +} + + +def category_for(event: str) -> str: + """Return the activity category for ``event``. + + Args: + event: The audit event name (``oidc_login``, ``source.deleted``). + + Returns: + One of :data:`ACTIVITY_CATEGORIES`; ``"other"`` for names this release + does not recognise. + """ + namespace, dot, _ = event.partition(".") + if dot: + return _NAMESPACES.get(namespace, "other") + return _EXACT.get(event, "other") + + +def category_case_sql(column: str) -> str: + """Build the SQL ``CASE`` that classifies ``column`` the way :func:`category_for` does. + + Uses ``split_part`` rather than ``LIKE`` so the expression carries no ``%`` + — which a pyformat DBAPI would otherwise read as a parameter marker. + + Args: + column: The SQL expression holding an event name. + + Returns: + A parenthesised ``CASE`` expression. Every literal in it comes from the + maps above, never from user input, so it is safe to interpolate. + """ + namespace_branches = " ".join( + f"WHEN '{namespace}' THEN '{category}'" + for namespace, category in _NAMESPACES.items() + ) + by_category: dict[str, list[str]] = {} + for event, category in _EXACT.items(): + by_category.setdefault(category, []).append(event) + exact_branches = " ".join( + f"WHEN {column} IN ({', '.join(repr(event) for event in sorted(events))}) " + f"THEN '{category}'" + for category, events in sorted(by_category.items()) + ) + return ( + "(CASE " + f"WHEN position('.' in {column}) > 0 " + f"THEN (CASE split_part({column}, '.', 1) {namespace_branches} ELSE 'other' END) " + f"{exact_branches} " + "ELSE 'other' END)" + ) diff --git a/docsgpt/storage/db/repositories/activity.py b/docsgpt/storage/db/repositories/activity.py new file mode 100644 index 00000000..be49b1e4 --- /dev/null +++ b/docsgpt/storage/db/repositories/activity.py @@ -0,0 +1,332 @@ +"""One merged, normalized feed over the instance's three audit journals. + +DocsGPT keeps three append-only journals, each written by a different part of +the system and each previously readable only on its own terms: + +* ``auth_events`` — identity, access, configuration and (since the data-plane + hooks) resource mutations. +* ``device_audit_log`` — every remote-device command dispatch and its outcome. +* ``guardrail_events`` — every guardrail decision on a request. + +An operator reviewing an incident wants one timeline, not three. This +repository projects all three onto a common row shape and merges them with +``UNION ALL``, so a single ordered, filtered, paginated feed spans the lot. + +Two exclusions are deliberate and mirror the per-agent guardrail view: the +``guardrail_events.api_key`` column is a raw agent key, and ``matched_value`` +is unredacted source text. Neither is projected here, admin-gated or not. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Iterator, Optional, Sequence + +from sqlalchemy import Connection, text + +from docsgpt.audit_events import category_case_sql + + +# Columns every branch of the union produces, in order. +ACTIVITY_COLUMNS = ( + "feed", + "id", + "event", + "category", + "actor_id", + "target_id", + "ip", + "user_agent", + "outcome", + "detail", + "created_at", +) + +# Which journal each feed name reads from. Used to skip a branch entirely when +# the category filter cannot match it. +_FEED_CATEGORIES = {"auth": None, "device": "device", "guardrail": "safety"} + +_AUTH_BRANCH = f""" + SELECT 'auth' AS feed, + id::text AS id, + event, + {category_case_sql("event")} AS category, + actor_id, + target_id, + ip, + user_agent, + NULL::text AS outcome, + metadata AS detail, + created_at + FROM auth_events +""" + +_DEVICE_BRANCH = """ + SELECT 'device' AS feed, + id::text AS id, + 'device.' || action AS event, + 'device' AS category, + user_id AS actor_id, + NULL::text AS target_id, + NULL::text AS ip, + NULL::text AS user_agent, + decision AS outcome, + jsonb_strip_nulls(jsonb_build_object( + 'device_id', device_id, + 'invocation_id', invocation_id, + 'command', command, + 'working_dir', working_dir, + 'approval_mode', approval_mode, + 'decision_reason', decision_reason, + 'exit_code', exit_code, + 'duration_ms', duration_ms, + 'agent_id', agent_id, + 'conversation_id', conversation_id + )) AS detail, + created_at + FROM device_audit_log +""" + +_GUARDRAIL_BRANCH = """ + SELECT 'guardrail' AS feed, + g.id::text AS id, + 'guardrail.' || g.stage AS event, + 'safety' AS category, + g.user_id AS actor_id, + NULL::text AS target_id, + NULL::text AS ip, + NULL::text AS user_agent, + g.outcome, + jsonb_strip_nulls(jsonb_build_object( + 'check_name', g.check_name, + 'detector_type', g.detector_type, + 'action', g.action, + 'category', g.category, + 'score', g.score, + 'match_count', g.match_count, + 'agent_id', g.agent_id, + 'message_id', g.message_id, + 'request_id', g.request_id, + 'detail', g.detail + )) AS detail, + g.created_at + FROM guardrail_events g +""" + +_BRANCHES = { + "auth": _AUTH_BRANCH, + "device": _DEVICE_BRANCH, + "guardrail": _GUARDRAIL_BRANCH, +} + + +class ActivityRepository: + """Read-only merged feed over ``auth_events`` + the two side journals.""" + + def __init__(self, conn: Connection) -> None: + self._conn = conn + + @staticmethod + def _feeds_for( + feeds: Optional[Sequence[str]], categories: Optional[Sequence[str]] + ) -> list[str]: + """Which union branches can still produce a matching row. + + A ``category=device`` filter cannot be satisfied by ``auth_events``, so + that branch is dropped from the union rather than scanned and + discarded. ``auth`` spans several categories and is only dropped when + the filter names none of them. + """ + selected = list(feeds) if feeds else list(_BRANCHES) + selected = [feed for feed in selected if feed in _BRANCHES] + if not categories: + return selected + wanted = set(categories) + fixed_only = {category for category in _FEED_CATEGORIES.values() if category} + keep = [] + for feed in selected: + fixed = _FEED_CATEGORIES[feed] + if fixed is None: + # ``auth`` spans several categories, so keep it unless the + # filter names only categories another journal owns outright. + if wanted - fixed_only: + keep.append(feed) + elif fixed in wanted: + keep.append(feed) + return keep + + def _query( + self, + *, + feeds: Optional[Sequence[str]], + categories: Optional[Sequence[str]], + events: Optional[Sequence[str]], + actor_id: Optional[str], + user_id: Optional[str], + since: Optional[datetime], + until: Optional[datetime], + search: Optional[str], + ) -> tuple[str, dict]: + selected = self._feeds_for(feeds, categories) + if not selected: + return "", {} + union = " UNION ALL ".join(_BRANCHES[feed] for feed in selected) + + clauses: list[str] = [] + params: dict[str, Any] = {} + if categories: + clauses.append("category = ANY(:categories)") + params["categories"] = list(categories) + if events: + clauses.append("event = ANY(:events)") + params["events"] = list(events) + if actor_id: + clauses.append("actor_id = :actor_id") + params["actor_id"] = actor_id + if user_id: + clauses.append("(actor_id = :user_id OR target_id = :user_id)") + params["user_id"] = user_id + if since is not None: + clauses.append("created_at >= :since") + params["since"] = since + if until is not None: + clauses.append("created_at <= :until") + params["until"] = until + if search: + clauses.append( + "(actor_id ILIKE :search OR target_id ILIKE :search " + "OR event ILIKE :search OR ip ILIKE :search " + "OR COALESCE(outcome, '') ILIKE :search OR detail::text ILIKE :search)" + ) + params["search"] = f"%{search}%" + where = ("WHERE " + " AND ".join(clauses)) if clauses else "" + return f"SELECT * FROM ({union}) AS activity {where}", params + + def list( + self, + *, + feeds: Optional[Sequence[str]] = None, + categories: Optional[Sequence[str]] = None, + events: Optional[Sequence[str]] = None, + actor_id: Optional[str] = None, + user_id: Optional[str] = None, + since: Optional[datetime] = None, + until: Optional[datetime] = None, + search: Optional[str] = None, + limit: int = 50, + offset: int = 0, + ) -> list[dict]: + """Return one page of the merged feed, newest first. + + Args: + feeds: Restrict to named journals (``auth``, ``device``, ``guardrail``). + categories: Restrict to activity categories. + events: Restrict to exact event names. + actor_id: Only rows this user performed. + user_id: Rows where this user is the actor or the target. + since: Inclusive lower bound on ``created_at``. + until: Inclusive upper bound on ``created_at``. + search: Case-insensitive substring over ids, event, IP and detail. + limit: Page size. + offset: Rows to skip. + + Returns: + Row dicts carrying :data:`ACTIVITY_COLUMNS`. + """ + sql, params = self._query( + feeds=feeds, + categories=categories, + events=events, + actor_id=actor_id, + user_id=user_id, + since=since, + until=until, + search=search, + ) + if not sql: + return [] + params.update({"limit": int(limit), "offset": int(offset)}) + result = self._conn.execute( + text(f"{sql} ORDER BY created_at DESC, feed, id DESC LIMIT :limit OFFSET :offset"), + params, + ) + return [dict(row._mapping) for row in result.fetchall()] + + def count( + self, + *, + feeds: Optional[Sequence[str]] = None, + categories: Optional[Sequence[str]] = None, + events: Optional[Sequence[str]] = None, + actor_id: Optional[str] = None, + user_id: Optional[str] = None, + since: Optional[datetime] = None, + until: Optional[datetime] = None, + search: Optional[str] = None, + ) -> int: + """Total rows matching the same filters as :meth:`list`.""" + sql, params = self._query( + feeds=feeds, + categories=categories, + events=events, + actor_id=actor_id, + user_id=user_id, + since=since, + until=until, + search=search, + ) + if not sql: + return 0 + scalar = self._conn.execute( + text(f"SELECT count(*) FROM ({sql}) AS matched"), params + ).scalar() + return int(scalar or 0) + + def iter_all( + self, + *, + chunk_size: int = 1000, + max_rows: int = 100_000, + **filters: Any, + ) -> Iterator[dict]: + """Yield every matching row, newest first, for the export endpoint. + + Pages through the feed rather than materializing it, so an export of a + large instance streams instead of building the whole result in memory. + + Args: + chunk_size: Rows fetched per round trip. + max_rows: Hard ceiling; an export stops here rather than running + unbounded against a busy instance. + **filters: As :meth:`list`, minus ``limit`` / ``offset``. + + Yields: + Row dicts carrying :data:`ACTIVITY_COLUMNS`. + """ + emitted = 0 + offset = 0 + while emitted < max_rows: + page = self.list( + limit=min(chunk_size, max_rows - emitted), offset=offset, **filters + ) + if not page: + return + for row in page: + yield row + emitted += len(page) + offset += len(page) + + def event_names(self) -> list[dict]: + """Distinct ``(event, category)`` pairs across all three journals. + + Feeds the filter's event picker, so operators choose from what the + instance actually recorded rather than typing a name from memory. + """ + union = " UNION ALL ".join(_BRANCHES.values()) + result = self._conn.execute( + text( + f"SELECT DISTINCT event, category FROM ({union}) AS activity " + "ORDER BY category, event" + ) + ) + return [{"event": row[0], "category": row[1]} for row in result.fetchall()] diff --git a/tests/api/test_admin_activity.py b/tests/api/test_admin_activity.py new file mode 100644 index 00000000..38ba0c79 --- /dev/null +++ b/tests/api/test_admin_activity.py @@ -0,0 +1,236 @@ +"""Endpoint tests for the merged admin activity feed, its filters and export. + +The repository's SQL is covered by +tests/storage/db/repositories/test_activity.py; these pin the route wiring, +the @admin_required boundary, query-argument parsing and the export framing. +""" + +from __future__ import annotations + +import csv +import io +import json +from contextlib import ExitStack, contextmanager +from datetime import datetime, timezone +from unittest.mock import Mock, patch + +import pytest + + +@pytest.fixture +def client(): + from docsgpt.app import app as flask_app + + flask_app.config["TESTING"] = True + return flask_app.test_client() + + +@contextmanager +def _fake_conn(): + yield Mock() + + +@contextmanager +def _admin(repo: Mock): + with ExitStack() as stack: + stack.enter_context( + patch("docsgpt.app.handle_auth", return_value={"sub": "admin1"}) + ) + stack.enter_context( + patch("docsgpt.app.resolve_roles", return_value=["admin", "user"]) + ) + stack.enter_context(patch("docsgpt.api.admin.activity.db_readonly", _fake_conn)) + stack.enter_context( + patch( + "docsgpt.api.admin.activity.ActivityRepository", return_value=repo + ) + ) + yield + + +def _row(**overrides): + base = { + "feed": "auth", + "id": "1", + "event": "oidc_login", + "category": "identity", + "actor_id": "u1", + "target_id": "u1", + "ip": "203.0.113.1", + "user_agent": "curl/8", + "outcome": None, + "detail": {"via": "oidc"}, + "created_at": datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc), + } + base.update(overrides) + return base + + +def _repo(rows=None, total=None): + repo = Mock() + rows = [_row()] if rows is None else rows + repo.list.return_value = rows + repo.count.return_value = len(rows) if total is None else total + repo.iter_all.return_value = iter(rows) + repo.event_names.return_value = [ + {"event": "oidc_login", "category": "identity"}, + ] + return repo + + +@pytest.mark.unit +class TestGuard: + def test_non_admin_forbidden(self, client): + with patch("docsgpt.app.handle_auth", return_value={"sub": "u"}), patch( + "docsgpt.app.resolve_roles", return_value=["user"] + ): + assert client.get("/api/admin/activity").status_code == 403 + assert client.get("/api/admin/activity/export").status_code == 403 + assert client.get("/api/admin/activity/events").status_code == 403 + + def test_unauthenticated(self, client): + with patch("docsgpt.app.handle_auth", return_value=None): + assert client.get("/api/admin/activity").status_code == 401 + + +@pytest.mark.unit +class TestFeed: + def test_returns_rows_with_iso_timestamps(self, client): + repo = _repo() + with _admin(repo): + body = json.loads(client.get("/api/admin/activity").data) + assert body["success"] is True + assert body["activity"][0]["created_at"] == "2026-01-02T03:04:05+00:00" + assert body["total"] == 1 + assert body["has_more"] is False + + def test_has_more_when_the_page_is_short_of_the_total(self, client): + repo = _repo(total=500) + with _admin(repo): + body = json.loads(client.get("/api/admin/activity").data) + assert body["has_more"] is True + + def test_filters_reach_the_repository(self, client): + repo = _repo() + with _admin(repo): + client.get( + "/api/admin/activity?feed=auth,device&category=data" + "&event=source.deleted&actor_id=admin1&user_id=u2" + "&since=2026-01-01T00:00:00Z&until=2026-02-01&search=handbook" + ) + kwargs = repo.list.call_args.kwargs + assert kwargs["feeds"] == ["auth", "device"] + assert kwargs["categories"] == ["data"] + assert kwargs["events"] == ["source.deleted"] + assert kwargs["actor_id"] == "admin1" + assert kwargs["user_id"] == "u2" + assert kwargs["since"] == datetime(2026, 1, 1, tzinfo=timezone.utc) + # A naive bound is read as UTC. + assert kwargs["until"] == datetime(2026, 2, 1, tzinfo=timezone.utc) + assert kwargs["search"] == "handbook" + + def test_unknown_facet_values_are_ignored_not_rejected(self, client): + repo = _repo() + with _admin(repo): + resp = client.get("/api/admin/activity?feed=bogus&category=bogus") + assert resp.status_code == 200 + kwargs = repo.list.call_args.kwargs + assert kwargs["feeds"] is None and kwargs["categories"] is None + + def test_unparseable_time_is_ignored(self, client): + repo = _repo() + with _admin(repo): + assert client.get("/api/admin/activity?since=nonsense").status_code == 200 + assert repo.list.call_args.kwargs["since"] is None + + def test_repeated_params_accumulate(self, client): + repo = _repo() + with _admin(repo): + client.get("/api/admin/activity?category=data&category=access") + assert repo.list.call_args.kwargs["categories"] == ["data", "access"] + + def test_search_is_length_capped(self, client): + repo = _repo() + with _admin(repo): + client.get("/api/admin/activity?search=" + "x" * 5000) + assert len(repo.list.call_args.kwargs["search"]) == 200 + + +@pytest.mark.unit +class TestCatalogue: + def test_lists_events_categories_and_feeds(self, client): + repo = _repo() + with _admin(repo): + body = json.loads(client.get("/api/admin/activity/events").data) + assert body["events"] == [{"event": "oidc_login", "category": "identity"}] + assert "identity" in body["categories"] + assert body["feeds"] == ["auth", "device", "guardrail"] + + +@pytest.mark.unit +class TestExport: + """The export streams lazily, so every test consumes it inside the patch.""" + + def test_csv_has_a_header_and_one_row_per_event(self, client): + repo = _repo(rows=[_row(), _row(id="2", event="pat_created")]) + with _admin(repo): + resp = client.get("/api/admin/activity/export?format=csv") + body = resp.get_data(as_text=True) + assert resp.status_code == 200 + assert resp.mimetype == "text/csv" + assert "attachment; filename=" in resp.headers["Content-Disposition"] + rows = list(csv.DictReader(io.StringIO(body))) + assert [row["event"] for row in rows] == ["oidc_login", "pat_created"] + # ``detail`` is a JSON object, encoded into the cell. + assert json.loads(rows[0]["detail"]) == {"via": "oidc"} + + def test_ndjson_emits_one_object_per_line(self, client): + repo = _repo(rows=[_row(), _row(id="2")]) + with _admin(repo): + resp = client.get("/api/admin/activity/export?format=ndjson") + body = resp.get_data(as_text=True) + assert resp.mimetype == "application/x-ndjson" + lines = body.strip().split("\n") + assert [json.loads(line)["id"] for line in lines] == ["1", "2"] + + def test_empty_csv_export_still_has_its_header(self, client): + with _admin(_repo(rows=[])): + body = client.get("/api/admin/activity/export").get_data(as_text=True) + assert body.strip().startswith("feed,id,event,category") + + def test_rejects_an_unknown_format(self, client): + with _admin(_repo()): + assert ( + client.get("/api/admin/activity/export?format=xlsx").status_code == 400 + ) + + def test_export_applies_the_same_filters(self, client): + repo = _repo() + with _admin(repo): + client.get( + "/api/admin/activity/export?category=data&search=q" + ).get_data() + kwargs = repo.iter_all.call_args.kwargs + assert kwargs["categories"] == ["data"] + assert kwargs["search"] == "q" + + def test_row_cap_is_bounded_and_reported(self, client): + repo = _repo() + with _admin(repo): + resp = client.get("/api/admin/activity/export?max_rows=10") + resp.get_data() + assert resp.headers["X-Export-Max-Rows"] == "10" + assert repo.iter_all.call_args.kwargs["max_rows"] == 10 + + def test_an_absurd_cap_clamps_to_the_ceiling(self, client): + repo = _repo() + with _admin(repo): + resp = client.get("/api/admin/activity/export?max_rows=999999999") + resp.get_data() + assert resp.headers["X-Export-Max-Rows"] == "100000" + + def test_export_is_not_cached(self, client): + with _admin(_repo()): + resp = client.get("/api/admin/activity/export") + resp.get_data() + assert resp.headers["Cache-Control"] == "no-store" diff --git a/tests/api/test_audit_helper.py b/tests/api/test_audit_helper.py index ec064330..f017a7ac 100644 --- a/tests/api/test_audit_helper.py +++ b/tests/api/test_audit_helper.py @@ -6,7 +6,7 @@ from unittest.mock import MagicMock, Mock, patch import pytest -from docsgpt.api.audit import EVENT_CATEGORIES, category_for, record_event +from docsgpt.api.audit import ACTIVITY_CATEGORIES, category_for, record_event @pytest.fixture @@ -102,5 +102,22 @@ class TestCategories: def test_unknown_event_is_other(self): assert category_for("something_new") == "other" - def test_every_category_is_declared(self): - assert "other" in EVENT_CATEGORIES + @pytest.mark.parametrize( + "event,expected", + [("device.run_command", "device"), ("guardrail.input", "safety")], + ) + def test_merged_feed_namespaces(self, event, expected): + assert category_for(event) == expected + + def test_every_produced_category_is_declared(self): + assert "other" in ACTIVITY_CATEGORIES + for event in ( + "oidc_login", + "role_granted", + "quota_policy_set", + "source.deleted", + "device.run_command", + "guardrail.input", + "mystery", + ): + assert category_for(event) in ACTIVITY_CATEGORIES diff --git a/tests/storage/db/repositories/test_activity.py b/tests/storage/db/repositories/test_activity.py new file mode 100644 index 00000000..a22b01f3 --- /dev/null +++ b/tests/storage/db/repositories/test_activity.py @@ -0,0 +1,165 @@ +"""Repository tests for the merged admin activity feed.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import text + +from docsgpt.storage.db.repositories.activity import ( + ACTIVITY_COLUMNS, + ActivityRepository, +) +from docsgpt.storage.db.repositories.auth_events import AuthEventsRepository + + +pytestmark = pytest.mark.integration + + +@pytest.fixture() +def seeded(pg_conn): + """One row in each journal, oldest to newest: guardrail, device, auth.""" + now = datetime.now(timezone.utc) + pg_conn.execute( + text( + "INSERT INTO guardrail_events " + "(user_id, api_key, stage, check_name, detector_type, action, " + " outcome, matched_value, created_at) VALUES " + "('u-guard', 'sk-secret', 'input', 'pii', 'regex', 'block', " + " 'blocked', 'alex@example.com', :ts)" + ), + {"ts": now - timedelta(minutes=10)}, + ) + pg_conn.execute( + text( + "INSERT INTO devices " + "(id, user_id, name, machine_pubkey_fingerprint, token_hash, status) " + "VALUES ('dev-1', 'u-dev', 'laptop', 'fp-1', 'hash-1', 'active')" + ) + ) + pg_conn.execute( + text( + "INSERT INTO device_audit_log " + "(device_id, user_id, invocation_id, action, command, " + " approval_mode, decision, issued_at, created_at) VALUES " + "('dev-1', 'u-dev', 'inv-1', 'run_command', 'ls -la', " + " 'ask', 'allowed', :ts, :ts)" + ), + {"ts": now - timedelta(minutes=5)}, + ) + AuthEventsRepository(pg_conn).insert( + "victim", "admin_user_deactivated", ip="203.0.113.9", + actor_id="admin-1", target_id="victim", + ) + return pg_conn + + +@pytest.fixture() +def repo(seeded): + return ActivityRepository(seeded) + + +class TestMergedFeed: + def test_spans_all_three_journals_newest_first(self, repo): + rows = repo.list() + assert [row["feed"] for row in rows] == ["auth", "device", "guardrail"] + assert repo.count() == 3 + + def test_every_row_has_the_common_shape(self, repo): + for row in repo.list(): + assert set(row) == set(ACTIVITY_COLUMNS) + + def test_side_journals_get_namespaced_event_names(self, repo): + events = {row["feed"]: row["event"] for row in repo.list()} + assert events["device"] == "device.run_command" + assert events["guardrail"] == "guardrail.input" + + def test_categories_are_derived(self, repo): + rows = {row["feed"]: row["category"] for row in repo.list()} + assert rows == { + "auth": "access", + "device": "device", + "guardrail": "safety", + } + + def test_outcome_carries_the_side_journal_verdict(self, repo): + rows = {row["feed"]: row["outcome"] for row in repo.list()} + assert rows["device"] == "allowed" + assert rows["guardrail"] == "blocked" + assert rows["auth"] is None + + +class TestSecretExclusion: + def test_guardrail_secrets_never_reach_the_feed(self, repo): + """``api_key`` is a raw agent key and ``matched_value`` is source text.""" + row = next(r for r in repo.list() if r["feed"] == "guardrail") + serialized = str(row) + assert "sk-secret" not in serialized + assert "alex@example.com" not in serialized + + +class TestFilters: + def test_by_feed(self, repo): + assert [r["feed"] for r in repo.list(feeds=["device"])] == ["device"] + + def test_by_category(self, repo): + rows = repo.list(categories=["safety"]) + assert [r["feed"] for r in rows] == ["guardrail"] + assert repo.count(categories=["safety"]) == 1 + + def test_by_category_spanning_auth_and_a_side_journal(self, repo): + rows = repo.list(categories=["access", "device"]) + assert {r["feed"] for r in rows} == {"auth", "device"} + + def test_by_actor(self, repo): + assert [r["actor_id"] for r in repo.list(actor_id="admin-1")] == ["admin-1"] + + def test_by_user_matches_actor_or_target(self, repo): + assert repo.count(user_id="victim") == 1 + assert repo.count(user_id="admin-1") == 1 + + def test_by_event(self, repo): + assert repo.count(events=["device.run_command"]) == 1 + + def test_by_time_window(self, repo): + now = datetime.now(timezone.utc) + assert repo.count(since=now - timedelta(minutes=7)) == 2 + assert repo.count(until=now - timedelta(minutes=7)) == 1 + + def test_search_spans_detail(self, repo): + assert repo.count(search="ls -la") == 1 + assert repo.count(search="203.0.113") == 1 + assert repo.count(search="nope") == 0 + + def test_unknown_feed_yields_nothing(self, repo): + assert repo.list(feeds=["made-up"]) == [] + assert repo.count(feeds=["made-up"]) == 0 + + def test_category_no_journal_can_produce(self, repo): + assert repo.count(feeds=["device"], categories=["data"]) == 0 + + +class TestPagination: + def test_limit_and_offset(self, repo): + assert len(repo.list(limit=2)) == 2 + assert [r["feed"] for r in repo.list(limit=2, offset=2)] == ["guardrail"] + + +class TestExportStream: + def test_iter_all_walks_every_row(self, repo): + assert len(list(repo.iter_all(chunk_size=1))) == 3 + + def test_iter_all_honours_max_rows(self, repo): + assert len(list(repo.iter_all(chunk_size=1, max_rows=2))) == 2 + + def test_iter_all_applies_filters(self, repo): + rows = list(repo.iter_all(chunk_size=1, feeds=["auth"])) + assert [r["feed"] for r in rows] == ["auth"] + + +class TestCatalogue: + def test_event_names_are_paired_with_categories(self, repo): + catalogue = repo.event_names() + assert {"event": "device.run_command", "category": "device"} in catalogue + assert {"event": "admin_user_deactivated", "category": "access"} in catalogue