diff --git a/AGENTS.md b/AGENTS.md index 5029922..2c4c881 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -37,10 +37,10 @@ Upload bearer tokens are long-lived, reusable credentials, not one-time codes: ## Personal access token lifecycle Personal access tokens (`PersonalAccessToken`, raw prefix `gpat_`) are the -**read-only** counterpart to upload tokens. They authenticate reads — currently the -streaming group export (`GET /api/v1/groups//export/`) — never uploads. They -are a distinct model and credential from `UploadToken`; the two are never -interchangeable. +**read-only** counterpart to upload tokens. They authenticate the group index and +streaming group export (`GET /api/v1/groups/` and +`GET /api/v1/groups//export/`) — never uploads. They are a distinct model and +credential from `UploadToken`; the two are never interchangeable. - A user mints and revokes their own from the profile page; the raw token is shown exactly once and only its hash is stored. diff --git a/docs/api-v1.md b/docs/api-v1.md index 55c8276..3211106 100644 --- a/docs/api-v1.md +++ b/docs/api-v1.md @@ -7,24 +7,30 @@ JSONL evidence. ## Authentication -All read APIs require a logged-in Goggles user. Upload APIs use reusable bearer -tokens and are documented separately in the app workflow notes. - -The streaming group export additionally accepts a **personal access token** — a -read-only bearer credential a user mints from their profile page (or an operator -mints for a service account with `manage.py create_access_token "" --user -`). Send it as `Authorization: Bearer gpat_…`. A personal access token -authorizes only the streaming group export (below) — not uploads, and not the -session-authenticated read APIs; it is revoked by the owner, by an admin, by expiry, -or by deactivating the owning user. Upload tokens (`goggles_…`) and personal access -tokens (`gpat_…`) are distinct credentials and never interchangeable. +Most read APIs require a logged-in Goggles user session. The group index and +streaming group export additionally accept a **personal access token** bearer +credential. Send it as `Authorization: Bearer gpat_…`. On those bearer-enabled +endpoints, missing, malformed, invalid, inactive, expired, or owner-deactivated +credentials return `401` JSON — never an HTML login redirect. + +Upload APIs use reusable upload bearer tokens (`goggles_…`) and are documented +separately in the app workflow notes. Upload tokens never authorize read APIs. + +Personal access tokens are read-only credentials a user mints from their profile +page (or an operator mints for a service account with +`manage.py create_access_token "" --user `). They authorize the +group list (`GET /api/v1/groups/`) and the streaming group export +(`GET /api/v1/groups/{slug}/export/`) — not uploads or the other +session-authenticated projection APIs. A token is revoked by the owner, by an +admin, by expiry, or by deactivating the owning user. Upload tokens and +personal access tokens are distinct credentials and never interchangeable. The current internal deployment treats authenticated Goggles users as one shared -internal tenant. Even so, endpoint implementations should route through -object-level scope checks before returning group, account, engine, message, -report, or evidence data. If Goggles later adds tenant or account isolation, -unauthorized resources should use the same not-found style behavior as unknown -resources so callers cannot enumerate data outside their scope. +internal tenant. Even so, endpoint implementations route through a shared +object-level readable-group scope before returning group data. If Goggles later +adds tenant or account isolation, unauthorized groups are omitted from the list +and exports for unknown slugs return `404`, indistinguishable from a missing +group, so callers cannot enumerate data outside their scope. Responses must not expose bearer tokens, upload secrets, source IPs, or user agents. Derived projection responses carry pointer-only evidence refs. Raw event @@ -109,6 +115,71 @@ metadata is the field-level hook for enforcing stricter access. - `GET /api/v1/accounts/{account_ref}/groups/` - `GET /api/v1/engines/{engine_id}/groups/` +`GET /api/v1/groups/` accepts a logged-in session or a personal access token. +It returns metadata for every group the reader may export. Newly-created groups +appear on the next poll without manual slug configuration. + +Query parameters: + +- `limit`: defaults to `100`, capped at `500` +- `cursor`: optional opaque continuation token from a previous page's + `pagination.next_cursor`. Omit on the first page of a poll. +- `updated_since`: optional ISO-8601 timestamp; when set, only groups with + `updated_at` strictly after this value are returned. The response echoes the + applied filter as `updated_since`. This is a **best-effort** change hint only: + it does not guarantee every group that became visible since your last poll (see + [Polling contract](#polling-contract)). + +Results are ordered by `updated_at desc` with a stable internal tie-breaker. The +first page of each poll fixes a server `polling_watermark` timestamp; every page +in that traversal reuses the same watermark and only includes groups with +`updated_at` at or before it. Groups that are still uncommitted, or that commit +after the watermark is captured, can require a later full index poll. The +watermark is **not** a +commit-safe upper bound on `updated_at`: a group can commit after page 1 with +`updated_at` at or before the watermark and still be invisible to both the +remaining pages and a subsequent `updated_since=polling_watermark` poll. + +Paginated responses include: + +```json +{ + "pagination": { + "limit": 100, + "returned": 10, + "has_more": true, + "next_cursor": "…" + }, + "polling_watermark": "2026-07-29T09:00:00+00:00" +} +``` + +#### Polling contract + +1. Start each poll without `cursor`. Read `polling_watermark` from the first + response and keep it for the whole traversal. Use it only to bound that + traversal; do not treat it as a commit-safe cursor for change detection. +2. Follow `pagination.next_cursor` until `has_more` is `false`. Each cursor is + bound to that poll's watermark and original `updated_since` filter, so only + `cursor` and the desired `limit` need to be sent on continuation requests. + Tampered or foreign cursors return `400` `{"error":"invalid cursor"}`. +3. After completing a traversal, you may set `updated_since` to the maximum + `updated_at` among groups you actually received to skip unchanged groups on + the next incremental poll. This is an optimization only: uploads assign + `updated_at` before commit, so a group can appear after your traversal with + `updated_at` at or before your `updated_since` bound and be omitted from + incremental polls. +4. For **eventual completeness**, periodically run a full index poll (omit + `updated_since` and `cursor`) and deduplicate by `slug`. A finite overlap + window alone does not guarantee discovery of arbitrarily delayed commits. + An empty `groups` array on an incremental poll means no group has + `updated_at` strictly after your `updated_since` bound; it does **not** prove + the index is unchanged. + +Projection endpoints still use numeric `offset` pagination (see +[Common Query Parameters](#common-query-parameters)); only the group index uses +cursor pagination. + Group responses include `schema_version`, group summary fields, tab counts, and classification metadata indicating whether full-data audit content may be present. diff --git a/forensics/analysis.py b/forensics/analysis.py index f9df396..2ee9c11 100644 --- a/forensics/analysis.py +++ b/forensics/analysis.py @@ -239,8 +239,11 @@ def file_rows_for_group(audit_files, group): # --------------------------------------------------------------------------- -def group_list_rows(): - groups = list(AuditGroup.objects.all()) +def group_list_rows(groups=None): + if groups is None: + groups = list(AuditGroup.objects.all()) + else: + groups = list(groups) group_file_counts = audit_file_counts_for_groups(groups) group_event_stats, fork_group_ids = event_stats_for_groups(groups) for group in groups: diff --git a/forensics/group_list_cursor.py b/forensics/group_list_cursor.py new file mode 100644 index 0000000..45409ad --- /dev/null +++ b/forensics/group_list_cursor.py @@ -0,0 +1,90 @@ +"""Opaque keyset cursor helpers for the group index API.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime + +from django.core import signing +from django.core.signing import BadSignature +from django.db.models import Q +from django.utils import timezone +from django.utils.dateparse import parse_datetime + +CURSOR_SALT = "goggles-groups-v1-cursor" + + +class InvalidGroupListCursor(ValueError): + pass + + +@dataclass(frozen=True) +class GroupListCursor: + watermark: datetime + updated_at: datetime + group_id: int + updated_since: datetime | None + + def keyset_filter(self) -> Q: + return Q(updated_at__lt=self.updated_at) | Q( + updated_at=self.updated_at, + pk__gt=self.group_id, + ) + + +def encode_group_list_cursor( + *, + watermark: datetime, + updated_at: datetime, + group_id: int, + updated_since: datetime | None, +) -> str: + payload = { + "w": watermark.isoformat(), + "u": updated_at.isoformat(), + "i": group_id, + "a": updated_since.isoformat() if updated_since is not None else None, + } + return signing.dumps(payload, salt=CURSOR_SALT, compress=True) + + +def decode_group_list_cursor(raw_cursor: str) -> GroupListCursor: + if not raw_cursor: + raise InvalidGroupListCursor("cursor is required") + try: + payload = signing.loads(raw_cursor, salt=CURSOR_SALT) + if not isinstance(payload, dict): + raise InvalidGroupListCursor("cursor payload is invalid") + watermark = _parse_cursor_timestamp(payload.get("w")) + updated_at = _parse_cursor_timestamp(payload.get("u")) + group_id = payload.get("i") + updated_since = _parse_optional_cursor_timestamp(payload.get("a")) + if not isinstance(group_id, int) or group_id <= 0: + raise InvalidGroupListCursor("cursor group id is invalid") + return GroupListCursor( + watermark=watermark, + updated_at=updated_at, + group_id=group_id, + updated_since=updated_since, + ) + except InvalidGroupListCursor: + raise + except (BadSignature, TypeError, ValueError) as exc: + raise InvalidGroupListCursor("cursor is invalid") from exc + + +def _parse_cursor_timestamp(value) -> datetime: + if not isinstance(value, str) or not value: + raise InvalidGroupListCursor("cursor timestamp is invalid") + parsed = parse_datetime(value) + if parsed is None: + raise InvalidGroupListCursor("cursor timestamp is invalid") + if timezone.is_naive(parsed): + parsed = timezone.make_aware(parsed, timezone.get_current_timezone()) + return parsed + + +def _parse_optional_cursor_timestamp(value) -> datetime | None: + if value is None: + return None + return _parse_cursor_timestamp(value) diff --git a/forensics/migrations/0014_auditgroup_updated_at_id_index.py b/forensics/migrations/0014_auditgroup_updated_at_id_index.py new file mode 100644 index 0000000..5f015f7 --- /dev/null +++ b/forensics/migrations/0014_auditgroup_updated_at_id_index.py @@ -0,0 +1,17 @@ +# Generated by Django 6.0.7 on 2026-07-29 08:43 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('forensics', '0013_personalaccesstoken'), + ] + + operations = [ + migrations.AddIndex( + model_name='auditgroup', + index=models.Index(fields=['-updated_at', 'id'], name='forensics_a_updated_8f04c0_idx'), + ), + ] diff --git a/forensics/models.py b/forensics/models.py index 830a0bf..7b8691a 100644 --- a/forensics/models.py +++ b/forensics/models.py @@ -21,6 +21,9 @@ class AuditGroup(models.Model): class Meta: ordering = ["-updated_at", "-created_at"] + indexes = [ + models.Index(fields=["-updated_at", "id"]), + ] def __str__(self) -> str: return self.name diff --git a/forensics/tests.py b/forensics/tests.py index f89ff91..5df8a06 100644 --- a/forensics/tests.py +++ b/forensics/tests.py @@ -1,8 +1,11 @@ +import base64 import contextlib import hashlib import json import os -from datetime import timedelta +import unittest +import zlib +from datetime import datetime, timedelta from io import StringIO from pathlib import Path from tempfile import TemporaryDirectory @@ -15,7 +18,13 @@ from django.core.management import call_command from django.core.management.base import CommandError from django.db import connection -from django.test import RequestFactory, SimpleTestCase, TestCase, override_settings +from django.test import ( + RequestFactory, + SimpleTestCase, + TestCase, + TransactionTestCase, + override_settings, +) from django.test.utils import CaptureQueriesContext from django.urls import reverse from django.utils import timezone @@ -90,11 +99,42 @@ ACCOUNT_BOB = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" GROUP_REF = "11" * 32 OTHER_GROUP_REF = "44" * 32 +DELAYED_COMMIT_GROUP_REF = "55" * 32 MSG_ID = "22" * 32 OTHER_MSG_ID = "33" * 32 DIGEST_A = "aa" * 32 DIGEST_B = "bb" * 32 + +def create_ordered_pagination_groups(count: int = 4): + base = timezone.now() + AuditGroup.objects.all().delete() + slugs = [] + for index in range(count): + slug = f"{index:02d}" * 32 + group = AuditGroup.objects.create( + name=f"Group {index}", + slug=slug, + group_ref=slug, + ) + AuditGroup.objects.filter(pk=group.pk).update(updated_at=base - timedelta(hours=index)) + slugs.append(slug) + return slugs + + +def decode_group_list_cursor_payload_without_secret(cursor: str) -> dict: + """Decode the client-visible signed cursor payload without SECRET_KEY.""" + encoded = cursor.split(":")[0] + compressed = encoded.startswith(".") + if compressed: + encoded = encoded[1:] + pad = b"=" * (-len(encoded) % 4) + data = base64.urlsafe_b64decode(encoded.encode("ascii") + pad) + if compressed: + data = zlib.decompress(data) + return json.loads(data.decode("latin-1")) + + HEAVY_EVENT_SELECT_COLUMNS = { field: f'"forensics_auditevent"."{field}"' for field in ( @@ -5774,6 +5814,480 @@ def test_mid_stream_error_yields_error_line_and_no_eof_within_a_200(self): self.assertNotIn("eof", {record["t"] for record in records}) +class GroupListApiTests(TestCase): + """GET /api/v1/groups/ — readable group enumeration for sessions and PATs.""" + + def setUp(self): + ingest_audit_log_bytes(dump_bytes=representative_audit_log().encode("utf-8")) + self.group = AuditGroup.objects.get(slug=GROUP_REF) + self.user = User.objects.create_user( + username="reader", password="correct horse battery staple" + ) + self.url = reverse("api-group-list") + + def test_lists_readable_groups_with_personal_access_token(self): + raw_token, token = PersonalAccessToken.issue("watchdog", user=self.user) + + response = self.client.get(self.url, HTTP_AUTHORIZATION=f"Bearer {raw_token}") + + self.assertEqual(response.status_code, 200) + payload = response.json() + self.assertEqual(payload["schema_version"], "goggles-groups/v1") + slugs = [group["slug"] for group in payload["groups"]] + self.assertIn(self.group.slug, slugs) + token.refresh_from_db() + self.assertIsNotNone(token.last_used_at) + + def test_requires_authentication_returns_401_json_not_redirect(self): + response = self.client.get(self.url) + + self.assertEqual(response.status_code, 401) + self.assertEqual(response.json(), {"error": "authentication required"}) + + def test_rejects_invalid_bearer_token(self): + response = self.client.get(self.url, HTTP_AUTHORIZATION="Bearer gpat_deadbeef_nope") + + self.assertEqual(response.status_code, 401) + self.assertEqual(response.json(), {"error": "authentication required"}) + + def test_rejects_expired_personal_access_token(self): + raw_token, _token = PersonalAccessToken.issue( + "stale", user=self.user, expires_at=timezone.now() - timedelta(seconds=1) + ) + + response = self.client.get(self.url, HTTP_AUTHORIZATION=f"Bearer {raw_token}") + + self.assertEqual(response.status_code, 401) + self.assertEqual(response.json(), {"error": "authentication required"}) + + def test_rejects_malformed_authorization_scheme(self): + response = self.client.get(self.url, HTTP_AUTHORIZATION="Token gpat_not_bearer") + + self.assertEqual(response.status_code, 401) + self.assertEqual(response.json(), {"error": "authentication required"}) + + def test_rejects_inactive_personal_access_token(self): + raw_token, token = PersonalAccessToken.issue("revoked", user=self.user) + token.is_active = False + token.save(update_fields=["is_active"]) + + response = self.client.get(self.url, HTTP_AUTHORIZATION=f"Bearer {raw_token}") + + self.assertEqual(response.status_code, 401) + self.assertEqual(response.json(), {"error": "authentication required"}) + + def test_rejects_personal_access_token_of_deactivated_owner(self): + raw_token, _token = PersonalAccessToken.issue("orphaned", user=self.user) + self.user.is_active = False + self.user.save(update_fields=["is_active"]) + + response = self.client.get(self.url, HTTP_AUTHORIZATION=f"Bearer {raw_token}") + + self.assertEqual(response.status_code, 401) + self.assertEqual(response.json(), {"error": "authentication required"}) + + def test_upload_token_cannot_list_groups(self): + raw_token, _token = UploadToken.issue("device") + + response = self.client.get(self.url, HTTP_AUTHORIZATION=f"Bearer {raw_token}") + + self.assertEqual(response.status_code, 401) + + def test_lists_with_logged_in_session(self): + self.client.login(username="reader", password="correct horse battery staple") + + response = self.client.get(self.url) + + self.assertEqual(response.status_code, 200) + slugs = [group["slug"] for group in response.json()["groups"]] + self.assertIn(self.group.slug, slugs) + + def test_denied_group_is_omitted_from_list_and_export_returns_404(self): + other = AuditGroup.objects.create( + name="Out of scope", + slug=OTHER_GROUP_REF, + group_ref=OTHER_GROUP_REF, + ) + raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + export_url = reverse("api-group-export-stream", kwargs={"slug": other.slug}) + + def scoped_queryset(reader): + self.assertEqual(reader.user, self.user) + return AuditGroup.objects.filter(slug=self.group.slug) + + with mock.patch("forensics.views.readable_groups_queryset", side_effect=scoped_queryset): + list_response = self.client.get(self.url, HTTP_AUTHORIZATION=f"Bearer {raw_token}") + export_response = self.client.get(export_url, HTTP_AUTHORIZATION=f"Bearer {raw_token}") + + self.assertEqual(list_response.status_code, 200) + slugs = [group["slug"] for group in list_response.json()["groups"]] + self.assertEqual(slugs, [self.group.slug]) + self.assertEqual(export_response.status_code, 404) + + def test_newly_created_group_appears_without_manual_slug_configuration(self): + raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + new_group = AuditGroup.objects.create( + name="Fresh group", + slug=OTHER_GROUP_REF, + group_ref=OTHER_GROUP_REF, + ) + + response = self.client.get(self.url, HTTP_AUTHORIZATION=f"Bearer {raw_token}") + + slugs = [group["slug"] for group in response.json()["groups"]] + self.assertIn(new_group.slug, slugs) + + def test_response_includes_bounded_pagination_metadata(self): + raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + + response = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"limit": "1"}, + ) + + payload = response.json() + self.assertEqual(payload["pagination"]["limit"], 1) + self.assertEqual(payload["pagination"]["returned"], 1) + self.assertIn("has_more", payload["pagination"]) + self.assertIn("next_cursor", payload["pagination"]) + self.assertIn("polling_watermark", payload) + + def test_cursor_does_not_expose_group_slug(self): + raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + slugs = create_ordered_pagination_groups() + exposed_slug = slugs[0] + + response = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"limit": "1"}, + ) + + cursor = response.json()["pagination"]["next_cursor"] + payload = decode_group_list_cursor_payload_without_secret(cursor) + self.assertNotIn(exposed_slug, cursor) + self.assertNotIn(exposed_slug, json.dumps(payload)) + self.assertNotIn(exposed_slug, payload.values()) + + def test_updated_since_filters_groups_for_polling(self): + raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + stale = AuditGroup.objects.create( + name="Stale", + slug="aa" * 32, + group_ref="aa" * 32, + ) + AuditGroup.objects.filter(pk=stale.pk).update(updated_at=timezone.now() - timedelta(days=2)) + stale.refresh_from_db() + fresh = AuditGroup.objects.create( + name="Fresh", + slug=OTHER_GROUP_REF, + group_ref=OTHER_GROUP_REF, + ) + since = (timezone.now() - timedelta(hours=1)).isoformat() + + response = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"updated_since": since}, + ) + + slugs = [group["slug"] for group in response.json()["groups"]] + self.assertIn(fresh.slug, slugs) + self.assertNotIn(stale.slug, slugs) + self.assertIn(self.group.slug, slugs) + + def test_updated_since_returns_empty_index_when_nothing_changed(self): + raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + after_all_updates = (timezone.now() + timedelta(seconds=1)).isoformat() + + response = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"updated_since": after_all_updates}, + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["groups"], []) + self.assertFalse(response.json()["pagination"]["has_more"]) + + def test_group_list_enrichment_is_bounded_to_the_requested_page(self): + raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + for index in range(3): + AuditGroup.objects.create( + name=f"Extra {index}", + slug=f"{index:02d}" * 32, + group_ref=f"{index:02d}" * 32, + ) + with mock.patch("forensics.views.group_list_rows", wraps=group_list_rows) as wrapped_rows: + self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"limit": "2"}, + ) + + self.assertEqual(len(wrapped_rows.call_args.args[0]), 2) + + def test_deleting_earlier_group_between_pages_does_not_skip_later_groups(self): + raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + slugs = create_ordered_pagination_groups() + + page1 = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"limit": "2"}, + ) + self.assertEqual(page1.status_code, 200) + self.assertEqual([group["slug"] for group in page1.json()["groups"]], slugs[:2]) + + AuditGroup.objects.filter(slug=slugs[0]).delete() + + page2 = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={ + "limit": "2", + "cursor": page1.json()["pagination"]["next_cursor"], + }, + ) + self.assertEqual(page2.status_code, 200) + self.assertEqual( + [group["slug"] for group in page2.json()["groups"]], + slugs[2:], + ) + + def test_cursor_keeps_updated_since_filter_across_pages(self): + raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + slugs = create_ordered_pagination_groups() + newest_group = AuditGroup.objects.get(slug=slugs[0]) + updated_since = newest_group.updated_at - timedelta(hours=2, minutes=30) + + page1 = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"limit": "2", "updated_since": updated_since.isoformat()}, + ) + self.assertEqual(page1.status_code, 200) + self.assertEqual([group["slug"] for group in page1.json()["groups"]], slugs[:2]) + + page2 = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={ + "limit": "2", + "cursor": page1.json()["pagination"]["next_cursor"], + }, + ) + self.assertEqual(page2.status_code, 200) + self.assertEqual( + [group["slug"] for group in page2.json()["groups"]], + [slugs[2]], + ) + self.assertEqual(page2.json()["updated_since"], updated_since.isoformat()) + + def test_group_updated_after_polling_watermark_appears_on_next_poll(self): + raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + slugs = create_ordered_pagination_groups() + + page1 = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"limit": "2"}, + ) + self.assertEqual(page1.status_code, 200) + page1_payload = page1.json() + polling_watermark = page1_payload["polling_watermark"] + next_cursor = page1_payload["pagination"]["next_cursor"] + + AuditGroup.objects.filter(slug=slugs[2]).update(updated_at=timezone.now()) + + page2 = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"limit": "2", "cursor": next_cursor}, + ) + self.assertEqual(page2.status_code, 200) + page2_slugs = [group["slug"] for group in page2.json()["groups"]] + self.assertNotIn(slugs[2], page2_slugs) + self.assertEqual(page2_slugs, [slugs[3]]) + self.assertFalse(page2.json()["pagination"]["has_more"]) + + next_poll = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"updated_since": polling_watermark}, + ) + self.assertEqual(next_poll.status_code, 200) + self.assertIn(slugs[2], [group["slug"] for group in next_poll.json()["groups"]]) + + def test_rejects_invalid_or_tampered_cursor(self): + raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + + for cursor in ("not-a-cursor", "tampered:cursor:value"): + response = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"cursor": cursor}, + ) + self.assertEqual(response.status_code, 400, cursor) + self.assertEqual(response.json(), {"error": "invalid cursor"}) + + def test_pre_watermark_group_missed_by_incremental_poll_needs_full_rescan(self): + """Groups that become visible after page 1 with updated_at <= watermark are + excluded from page 2 and from updated_since=polling_watermark; only a full + index rescan discovers them.""" + raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + slugs = create_ordered_pagination_groups() + base = AuditGroup.objects.get(slug=slugs[0]).updated_at + + page1 = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"limit": "2"}, + ) + self.assertEqual(page1.status_code, 200) + page1_payload = page1.json() + polling_watermark = page1_payload["polling_watermark"] + next_cursor = page1_payload["pagination"]["next_cursor"] + + delayed = AuditGroup.objects.create( + name="Delayed commit", + slug=DELAYED_COMMIT_GROUP_REF, + group_ref=DELAYED_COMMIT_GROUP_REF, + ) + AuditGroup.objects.filter(pk=delayed.pk).update( + updated_at=base - timedelta(minutes=30), + ) + + page2 = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"limit": "2", "cursor": next_cursor}, + ) + self.assertEqual(page2.status_code, 200) + page2_slugs = [group["slug"] for group in page2.json()["groups"]] + self.assertNotIn(DELAYED_COMMIT_GROUP_REF, page2_slugs) + self.assertEqual(page2_slugs, slugs[2:]) + + incremental = self.client.get( + self.url, + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + data={"updated_since": polling_watermark}, + ) + self.assertEqual(incremental.status_code, 200) + incremental_slugs = [group["slug"] for group in incremental.json()["groups"]] + self.assertNotIn(DELAYED_COMMIT_GROUP_REF, incremental_slugs) + + full_rescan = self.client.get(self.url, HTTP_AUTHORIZATION=f"Bearer {raw_token}") + self.assertEqual(full_rescan.status_code, 200) + rescan_slugs = [group["slug"] for group in full_rescan.json()["groups"]] + self.assertIn(DELAYED_COMMIT_GROUP_REF, rescan_slugs) + + +@unittest.skipUnless(connection.vendor == "postgresql", "requires PostgreSQL") +class GroupListPollingCommitRaceTests(TransactionTestCase): + """Real commit-order race: updated_at is assigned before upload commit.""" + + def setUp(self): + self.user = User.objects.create_user( + username="reader", password="correct horse battery staple" + ) + self.raw_token, _token = PersonalAccessToken.issue("watchdog", user=self.user) + self.url = reverse("api-group-list") + self.slugs = create_ordered_pagination_groups() + self.base = AuditGroup.objects.get(slug=self.slugs[0]).updated_at + self.delayed_updated_at = self.base - timedelta(minutes=30) + + def _postgres_connection(self): + import psycopg + + params = connection.get_connection_params() + params.setdefault("connect_timeout", 10) + return psycopg.connect(**params) + + def test_delayed_upload_commit_is_discovered_by_full_rescan_not_incremental_watermark(self): + auth = {"HTTP_AUTHORIZATION": f"Bearer {self.raw_token}"} + delayed_slug = DELAYED_COMMIT_GROUP_REF + delayed_name = "Delayed upload commit" + now = timezone.now() + + pg_conn = self._postgres_connection() + try: + with pg_conn.transaction(): + with pg_conn.cursor() as cursor: + self.assertEqual( + [ + field.column + for field in AuditGroup._meta.concrete_fields + if not field.primary_key + ], + [ + "name", + "slug", + "group_ref", + "divergent_message_count", + "notes", + "created_at", + "updated_at", + ], + ) + cursor.execute( + """ + INSERT INTO forensics_auditgroup ( + name, slug, group_ref, divergent_message_count, notes, + created_at, updated_at + ) + VALUES (%s, %s, %s, 0, '', %s, %s) + """, + ( + delayed_name, + delayed_slug, + delayed_slug, + now, + self.delayed_updated_at, + ), + ) + + page1 = self.client.get(self.url, data={"limit": "2"}, **auth) + self.assertEqual(page1.status_code, 200) + page1_payload = page1.json() + polling_watermark = page1_payload["polling_watermark"] + next_cursor = page1_payload["pagination"]["next_cursor"] + self.assertLess( + self.delayed_updated_at, + datetime.fromisoformat(polling_watermark), + ) + page1_slugs = [group["slug"] for group in page1_payload["groups"]] + self.assertEqual(page1_slugs, self.slugs[:2]) + self.assertNotIn(delayed_slug, page1_slugs) + + page2 = self.client.get( + self.url, + data={"limit": "2", "cursor": next_cursor}, + **auth, + ) + self.assertEqual(page2.status_code, 200) + page2_slugs = [group["slug"] for group in page2.json()["groups"]] + self.assertNotIn(delayed_slug, page2_slugs) + self.assertEqual(page2_slugs, self.slugs[2:]) + + incremental = self.client.get( + self.url, + data={"updated_since": polling_watermark}, + **auth, + ) + self.assertEqual(incremental.status_code, 200) + incremental_slugs = [group["slug"] for group in incremental.json()["groups"]] + self.assertNotIn(delayed_slug, incremental_slugs) + + full_rescan = self.client.get(self.url, **auth) + self.assertEqual(full_rescan.status_code, 200) + rescan_slugs = [group["slug"] for group in full_rescan.json()["groups"]] + self.assertIn(delayed_slug, rescan_slugs) + finally: + pg_conn.close() + + class ProfileAccessTokenTests(TestCase): """Self-service personal access tokens on the profile page — strictly owner-scoped: a user can only see, mint, and revoke their own.""" diff --git a/forensics/views.py b/forensics/views.py index 0006030..072af4e 100644 --- a/forensics/views.py +++ b/forensics/views.py @@ -2,6 +2,7 @@ import ipaddress from datetime import datetime +from typing import NamedTuple from django.conf import settings from django.contrib import messages @@ -25,6 +26,7 @@ from django.template.defaultfilters import slugify from django.urls import reverse from django.utils import timezone +from django.utils.dateparse import parse_datetime from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_GET, require_POST @@ -43,6 +45,11 @@ structural_quarantine_exclusion, valid_events_for_group, ) +from .group_list_cursor import ( + InvalidGroupListCursor, + decode_group_list_cursor, + encode_group_list_cursor, +) from .ingest import ingest_audit_log_bytes from .models import ( AnalysisRun, @@ -69,6 +76,8 @@ GROUP_DETAIL_TAB_EVENT_LIMIT = 100 GROUP_PROJECTION_API_DEFAULT_LIMIT = 100 GROUP_PROJECTION_API_MAX_LIMIT = 500 +GROUP_LIST_API_DEFAULT_LIMIT = 100 +GROUP_LIST_API_MAX_LIMIT = 500 GROUP_EXPORT_SCHEMA_VERSION = "goggles-group-export/v1" FULL_DATA_AUDIT_MODE = "full_data" ERROR_SEVERITY_TOKENS = ( @@ -860,13 +869,54 @@ def nested_json_value(value: dict, path: tuple[str, ...]): return current -@login_required +@require_GET def api_group_list(request: HttpRequest): - groups = group_list_rows() + reader = authenticate_reader(request) + if reader is None: + return JsonResponse({"error": "authentication required"}, status=401) + + filters = group_list_api_filters(request) + limit = filters["limit"] + cursor_raw = filters["cursor"] + try: + if cursor_raw: + cursor = decode_group_list_cursor(cursor_raw) + polling_watermark = cursor.watermark + keyset_filter = cursor.keyset_filter() + updated_since = cursor.updated_since + else: + polling_watermark = timezone.now() + keyset_filter = Q() + updated_since = filters["updated_since"] + except InvalidGroupListCursor: + return JsonResponse({"error": "invalid cursor"}, status=400) + + queryset = readable_groups_queryset(reader).order_by("-updated_at", "pk") + queryset = queryset.filter(updated_at__lte=polling_watermark) + if updated_since is not None: + queryset = queryset.filter(updated_at__gt=updated_since) + queryset = queryset.filter(keyset_filter) + + page_candidates = list(queryset[: limit + 1]) + page_groups = page_candidates[:limit] + groups = group_list_rows(page_groups) + has_more = len(page_candidates) > limit + next_cursor = None + if has_more and page_groups: + last_group = page_groups[-1] + next_cursor = encode_group_list_cursor( + watermark=polling_watermark, + updated_at=last_group.updated_at, + group_id=last_group.pk, + updated_since=updated_since, + ) return JsonResponse( { "schema_version": "goggles-groups/v1", "groups": [group_list_api_payload(group) for group in groups], + "pagination": group_list_pagination_payload(limit, len(groups), has_more, next_cursor), + "polling_watermark": polling_watermark.isoformat(), + **group_list_change_detection_payload(filters, updated_since), }, json_dumps_params={"separators": (",", ":")}, ) @@ -1400,6 +1450,39 @@ def group_list_api_payload(group) -> dict: } +def group_list_api_filters(request: HttpRequest) -> dict: + limit = GROUP_LIST_API_DEFAULT_LIMIT + limit_raw = request.GET.get("limit") + if limit_raw not in (None, ""): + try: + limit = int(limit_raw) + except (TypeError, ValueError): + limit = GROUP_LIST_API_DEFAULT_LIMIT + limit = min(max(limit, 1), GROUP_LIST_API_MAX_LIMIT) + return { + "limit": limit, + "cursor": (request.GET.get("cursor") or "").strip(), + "updated_since": parse_group_list_updated_since(request.GET.get("updated_since")), + } + + +def parse_group_list_updated_since(value: str | None): + if value in (None, ""): + return None + parsed = parse_datetime(value) + if parsed is None: + return None + if timezone.is_naive(parsed): + parsed = timezone.make_aware(parsed, timezone.get_current_timezone()) + return parsed + + +def group_list_change_detection_payload(filters: dict, updated_since) -> dict: + if updated_since is None: + return {} + return {"updated_since": updated_since.isoformat()} + + def group_api_payload(group: AuditGroup) -> dict: shell = group_summary_header_context(group) return { @@ -1693,6 +1776,20 @@ def pagination_payload(limit: int, offset: int, returned: int, has_more: bool) - } +def group_list_pagination_payload( + limit: int, + returned: int, + has_more: bool, + next_cursor: str | None, +) -> dict: + return { + "limit": limit, + "returned": returned, + "has_more": has_more, + "next_cursor": next_cursor, + } + + def severity_from_values(*values) -> str: text = " ".join(str(value).lower() for value in values if value not in (None, "", [])) if any(token in text for token in ERROR_SEVERITY_TOKENS): @@ -3182,21 +3279,36 @@ def authenticate_request(request: HttpRequest) -> UploadToken | None: return UploadToken.authenticate(bearer_value(request.headers.get("Authorization"))) -def authenticate_reader(request: HttpRequest) -> bool: - """Whether the request may read forensic data: a logged-in session or a valid - personal access token. +class ReaderPrincipal(NamedTuple): + user: object + access_token: PersonalAccessToken | None = None + - Deliberately not group-scoped — Goggles has no per-user data authorization, so - this only answers "is this an authenticated reader?". On the token path it - records ``last_used_at`` (a real write, not a pure predicate). +def readable_groups_queryset(reader: ReaderPrincipal): + """Groups the reader may export. Shared by the list and export endpoints.""" + del reader # reserved for future per-user scope + return AuditGroup.objects.all() + + +def get_readable_group_or_404(reader: ReaderPrincipal, slug: str) -> AuditGroup: + return get_object_or_404(readable_groups_queryset(reader), slug=slug) + + +def authenticate_reader(request: HttpRequest) -> ReaderPrincipal | None: + """Authenticate a forensic reader from a logged-in session or valid personal + access token. + + Returns the authenticated reader principal so list/export can apply a shared + object-level readable-group scope. On the token path it records + ``last_used_at`` (a real write, not a pure predicate). """ if request.user.is_authenticated: - return True + return ReaderPrincipal(user=request.user) token = PersonalAccessToken.authenticate(bearer_value(request.headers.get("Authorization"))) if token is None: - return False + return None token.mark_used() - return True + return ReaderPrincipal(user=token.user, access_token=token) @require_GET @@ -3212,10 +3324,11 @@ def api_group_export_stream(request: HttpRequest, slug: str): """ if not settings.GOGGLES_EXPORTS_ENABLED: return JsonResponse({"error": "exports are temporarily disabled"}, status=503) - if not authenticate_reader(request): + reader = authenticate_reader(request) + if reader is None: return JsonResponse({"error": "authentication required"}, status=401) - group = get_object_or_404(AuditGroup, slug=slug) + group = get_readable_group_or_404(reader, slug) # The export is unconditionally the complete group; it takes no query filters. # Honoring them would mean filtering raw ``events`` too — undermining the # "complete aggregate" contract the CGKA consumer depends on (every