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