feat: add authenticated streaming group export with self-service read tokens

AGENTS.md

@@ -34,6 +34,25 @@ Upload bearer tokens are long-lived, reusable credentials, not one-time codes:
 - Treat a leaked token as a standing credential: deactivate or delete it in the
   admin to revoke access immediately.
 
+## 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/<slug>/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.
+- For a service account (e.g. the CGKA pipeline), mint one bound to a user with
+  `manage.py create_access_token "name" --user <username>` (optionally
+  `--expires-in-days N`).
+- A token is only as live as its owner: it stops authenticating when revoked
+  (`is_active = False`), when `expires_at` passes, or when the owning user is
+  deactivated.
+- The shared hashing/verification/expiry mechanics live in `forensics/token_crypto.py`
+  and are single-sourced across both token models.
+
 ## Guardrails
 
 - Keep upload and forensic behavior grounded in the JSONL schema and existing

Dockerfile

@@ -25,4 +25,11 @@ USER goggles
 
 EXPOSE 8000
 
-CMD ["gunicorn", "config.wsgi:application", "--bind", "0.0.0.0:8000", "--workers", "3"]
+# Threaded workers so a long-running streaming export (the group export endpoint)
+# occupies a thread rather than blocking a whole worker; --timeout is raised well
+# past a multi-minute stream so gunicorn does not reap it. --max-requests recycles
+# workers periodically (gracefully, after in-flight streams finish) for leak hygiene
+# now that workers are long-lived. See docs/deployment.md for the capacity model.
+CMD ["gunicorn", "config.wsgi:application", "--bind", "0.0.0.0:8000", \
+     "--workers", "3", "--threads", "4", "--timeout", "300", \
+     "--max-requests", "500", "--max-requests-jitter", "50"]

config/settings.py

@@ -247,6 +247,12 @@ def scrub_glitchtip_event(event, _hint):
 }
 if not DEBUG and DATABASES["default"]["ENGINE"] == "django.db.backends.sqlite3":
     raise ImproperlyConfigured("Production DATABASE_URL must not use SQLite.")
+# Transaction-pooling connection poolers (e.g. PgBouncer) are incompatible with
+# Postgres server-side cursors, which the streaming export relies on. Set this behind
+# such a pooler; streaming reads then fall back to client-side chunked fetches, still
+# bounded by the query chunk_size. See docs/deployment.md.
+if env_bool("GOGGLES_DISABLE_SERVER_SIDE_CURSORS", False):
+    DATABASES["default"]["DISABLE_SERVER_SIDE_CURSORS"] = True
 
 AUTH_PASSWORD_VALIDATORS = [
     {"NAME": "django.contrib.auth.password_validation.UserAttributeSimilarityValidator"},
@@ -271,6 +277,10 @@ def scrub_glitchtip_event(event, _hint):
 
 GOGGLES_MAX_DUMP_BYTES = int(os.environ.get("GOGGLES_MAX_DUMP_BYTES", 50 * 1024 * 1024))
 GOGGLES_UPLOADS_ENABLED = env_bool("GOGGLES_UPLOADS_ENABLED", True)
+# Operational kill-switch for the streaming group-export endpoint, mirroring the
+# upload toggle. Lets an operator shed a resource-intensive read surface without a
+# redeploy.
+GOGGLES_EXPORTS_ENABLED = env_bool("GOGGLES_EXPORTS_ENABLED", True)
 DATA_UPLOAD_MAX_MEMORY_SIZE = GOGGLES_MAX_DUMP_BYTES
 FILE_UPLOAD_MAX_MEMORY_SIZE = GOGGLES_MAX_DUMP_BYTES
 # The upload endpoint only ever ingests a single file part, and every part at

docker-compose.yml

@@ -45,7 +45,9 @@ services:
     command: >
       sh -c "python manage.py migrate --noinput &&
              python manage.py collectstatic --noinput &&
-             gunicorn config.wsgi:application --bind 0.0.0.0:8000 --workers 3"
+             gunicorn config.wsgi:application --bind 0.0.0.0:8000
+             --workers 3 --threads 4 --timeout 300
+             --max-requests 500 --max-requests-jitter 50"
     healthcheck:
       test:
         [

docs/api-v1.md

@@ -10,6 +10,15 @@ JSONL evidence.
 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 "<name>" --user
+<username>`). 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.
+
 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,
@@ -96,6 +105,63 @@ Group responses include `schema_version`, group summary fields, tab counts, and
 classification metadata indicating whether full-data audit content may be
 present.
 
+### Group Export (streaming)
+
+- `GET /api/v1/groups/{slug}/export/`
+
+Streams the complete forensic aggregate for one group as a single **NDJSON**
+download (`Content-Type: application/x-ndjson`) — one JSON object per line. Rows are
+read from the database with server-side cursors, so the response is bounded in
+server memory regardless of group size and is **not paginated**. Authenticate with a
+logged-in session or a personal access token (`Authorization: Bearer gpat_…`).
+
+The export is unconditionally the complete group: it takes **no query filters** (the
+[common filters](#common-query-parameters) do not apply — applying them to only some
+sections would misrepresent the payload, and filtering raw events would break the
+completeness the export exists to provide). Filter client-side, or use the paginated
+projection endpoints. Disabled via `GOGGLES_EXPORTS_ENABLED=0` (returns `503`).
+
+Line schema:
+
+```text
+{"t":"manifest","schema_version":"goggles-group-export/v1","generated_at":"…","group":{…},"classification":{…},"sensitivity":{…},"sections":[…]}
+{"t":"source", …}                  # one per audit file
+{"t":"event", …}                   # every valid event, uncapped
+{"t":"delivery_artifact", …}
+{"t":"network_observation", …}
+{"t":"convergence_run", …}
+{"t":"state_delta", …}
+{"t":"epoch_state_transition", …}
+{"t":"audit_data_mode_change", …}
+{"t":"eof","complete":true,"counts":{"event":N, …}}
+```
+
+`event` records use the agent-state export shape; projection records use the
+projection-API shape — the export is a tagged union of the two, discriminated by the
+leading `t` on every line. Derived aggregates (`timeline`, `messages`, `actions`,
+`action_attribution`) are intentionally excluded; reconstruct them from the raw
+records (e.g. fork resolutions are `event` records with
+`event_type == "fork_resolution"`).
+
+**Fail-closed contract.** Two distinct failure surfaces:
+
+- *Before the first byte* (authentication, unknown group, kill-switch): a
+  conventional non-`200` response (`401`/`404`/`503`). Check the HTTP status first.
+- *Mid-stream*: the status is already committed at `200`, so a later error is
+  reported in-band as a final `{"t":"error","complete":false}` line with **no**
+  `eof`. A response whose last line is not `{"t":"eof",…}` is incomplete and must be
+  discarded.
+
+**Consistency.** The export is append-time-consistent, not a single atomic snapshot:
+each section is read in its own transaction, so a record appended mid-export may be
+referenced by a later section but missing from an earlier one. Re-export if a
+cross-section-consistent view matters.
+
+**Revocation is point-in-time.** Authentication is checked once, before streaming
+begins. Revoking a token (or deactivating its owner) stops the *next* request; an
+export already in flight continues to completion. Incident response should not assume
+revoke is an immediate cut-off for a stream already underway.
+
 ### Delivery
 
 - `GET /api/v1/groups/{group_slug}/delivery/`

docs/deployment.md

@@ -25,3 +25,32 @@ projections, and saved reports. It preserves user accounts and upload tokens.
 
 On large Postgres databases, run `VACUUM ANALYZE` after the purge if reclaiming
 space or refreshing planner statistics matters for the deployment window.
+
+## Streaming Group Export
+
+`GET /api/v1/groups/{slug}/export/` streams a group's full forensic aggregate as
+NDJSON (see `docs/api-v1.md`). It is a long-lived, resource-intensive response, so
+the gunicorn command runs threaded workers with a raised timeout:
+`--workers 3 --threads 4 --timeout 300 --max-requests 500 --max-requests-jitter 50`.
+
+Capacity model — size the database for it:
+
+- **Connections.** Each in-flight request holds one database connection for its full
+  duration. With `--workers 3 --threads 4`, up to **12** connections may be live at
+  once, and an export can hold one for minutes. Provision Postgres `max_connections`
+  (or pooler slots) for at least `workers × threads` plus headroom for background
+  tasks. If a transaction-mode pooler (e.g. PgBouncer) fronts the database it breaks
+  server-side cursors; set `GOGGLES_DISABLE_SERVER_SIDE_CURSORS=1` in that case (reads
+  fall back to client-side chunked fetches, still bounded by the query `chunk_size`).
+- **CPU / GIL.** Serializing rows to JSON is CPU-bound and holds the GIL, so
+  concurrent exports within one worker do not run in parallel — throughput is roughly
+  one export per worker at a time. Scale workers (and DB connections) if concurrent
+  large exports are expected.
+- **Timeout scope.** `--timeout 300` is process-wide: it also relaxes gunicorn's
+  liveness guard for uploads and every other request, not just exports.
+- **Kill-switch.** Set `GOGGLES_EXPORTS_ENABLED=0` and restart to shed the export
+  surface without affecting uploads or the rest of the API.
+
+The edge proxy (Caddy) streams `reverse_proxy` responses by default, so no proxy
+change is required; `nginx` serves only static assets and is not in the export
+request path.

forensics/admin.py

@@ -15,6 +15,7 @@
     DeliveryObservation,
     EpochStateTransition,
     NetworkObservation,
+    PersonalAccessToken,
     RecipientExpectation,
     StateDelta,
     UploadToken,
@@ -50,6 +51,23 @@ class UploadTokenAdmin(admin.ModelAdmin):
     readonly_fields = ("token_prefix", "token_hash", "created_at", "last_used_at")
 
 
+@admin.register(PersonalAccessToken)
+class PersonalAccessTokenAdmin(admin.ModelAdmin):
+    list_display = (
+        "name",
+        "user",
+        "token_prefix",
+        "is_active",
+        "created_at",
+        "expires_at",
+        "last_used_at",
+    )
+    list_filter = ("is_active",)
+    search_fields = ("name", "token_prefix", "user__username")
+    autocomplete_fields = ("user",)
+    readonly_fields = ("token_prefix", "token_hash", "created_at", "last_used_at")
+
+
 @admin.register(AuditFile)
 class AuditFileAdmin(admin.ModelAdmin):
     list_display = (

forensics/analysis.py

@@ -59,6 +59,27 @@
 AGENT_EXPORT_SCHEMA_VERSION = "goggles-agent-group-state/v1"
 AGENT_EXPORT_NORMALIZED_FIELDS = normalized_field_config.agent_export_normalized_fields()
 
+# What a forensic export carries versus deliberately omits. Shared verbatim by the
+# agent-state export and the streaming group export so the two never disagree about
+# the sensitivity contract of the data they emit.
+EXPORT_SENSITIVITY = {
+    "classification": "sensitive_forensic_export",
+    "contains": [
+        "engine_ids",
+        "account_refs",
+        "group_refs",
+        "message_ids",
+        "payload_digests",
+        "relay_urls",
+    ],
+    "omits": [
+        "raw_upload_bodies",
+        "bearer_tokens",
+        "source_ips",
+        "user_agents",
+    ],
+}
+
 VIZ_PALETTE_SIZE = 8
 GROUP_REF_FULL_DISPLAY_MAX = 80
 GROUP_REF_EDGE_DISPLAY_CHARS = 32
@@ -1033,23 +1054,7 @@ def agent_state_export_for_group(group, events, audit_files):
     return {
         "schema_version": AGENT_EXPORT_SCHEMA_VERSION,
         "generated_at": timezone.now().isoformat(),
-        "sensitivity": {
-            "classification": "sensitive_forensic_export",
-            "contains": [
-                "engine_ids",
-                "account_refs",
-                "group_refs",
-                "message_ids",
-                "payload_digests",
-                "relay_urls",
-            ],
-            "omits": [
-                "raw_upload_bodies",
-                "bearer_tokens",
-                "source_ips",
-                "user_agents",
-            ],
-        },
+        "sensitivity": EXPORT_SENSITIVITY,
         "group": timeline["group"],
         "summary": group_summary(group, audit_files, events=ordered),
         "sources": [agent_source_row(audit_file) for audit_file in audit_files],

forensics/management/commands/create_access_token.py

@@ -0,0 +1,55 @@
+from django.contrib.auth import get_user_model
+from django.core.management.base import BaseCommand, CommandError, CommandParser
+
+from forensics.models import PersonalAccessToken
+from forensics.token_crypto import expiry_from_days
+
+
+class Command(BaseCommand):
+    help = "Create a read-only personal access token owned by a user (e.g. a service account)."
+
+    def add_arguments(self, parser: CommandParser) -> None:
+        parser.add_argument("name", help="Human-friendly token name, e.g. 'cgka pipeline'")
+        parser.add_argument(
+            "--user",
+            required=True,
+            help="Username that will own the token. Deactivating this user revokes it.",
+        )
+        parser.add_argument(
+            "--expires-in-days",
+            type=int,
+            default=None,
+            metavar="N",
+            help=(
+                "Optional lifetime in days. Omit for a token that never expires. "
+                "Revoke any token early from the profile page or the admin."
+            ),
+        )
+
+    def handle(self, *args, **options):
+        expires_in_days = options["expires_in_days"]
+        try:
+            expires_at = None if expires_in_days is None else expiry_from_days(expires_in_days)
+        except ValueError as exc:
+            raise CommandError(str(exc)) from exc
+
+        user_model = get_user_model()
+        try:
+            user = user_model.objects.get(username=options["user"])
+        except user_model.DoesNotExist as exc:
+            raise CommandError(f"No user named {options['user']!r}.") from exc
+
+        raw_token, token = PersonalAccessToken.issue(
+            options["name"], user=user, expires_at=expires_at
+        )
+        self.stdout.write(
+            f"Created personal access token {token.name} ({token.token_prefix}) for {user.username}"
+        )
+        if expires_at is None:
+            self.stdout.write("This token is read-only and does not expire.")
+        else:
+            self.stdout.write(
+                f"This token is read-only and expires at {expires_at:%Y-%m-%d %H:%M:%S %Z}."
+            )
+        self.stdout.write("Store this token now; it will not be shown again:")
+        self.stdout.write(raw_token)

forensics/management/commands/create_upload_token.py

@@ -1,9 +1,7 @@
-from datetime import timedelta
-
 from django.core.management.base import BaseCommand, CommandError, CommandParser
-from django.utils import timezone
 
 from forensics.models import UploadToken
+from forensics.token_crypto import expiry_from_days
 
 
 class Command(BaseCommand):
@@ -24,11 +22,10 @@ def add_arguments(self, parser: CommandParser) -> None:
 
     def handle(self, *args, **options):
         expires_in_days = options["expires_in_days"]
-        expires_at = None
-        if expires_in_days is not None:
-            if expires_in_days <= 0:
-                raise CommandError("--expires-in-days must be a positive integer.")
-            expires_at = timezone.now() + timedelta(days=expires_in_days)
+        try:
+            expires_at = None if expires_in_days is None else expiry_from_days(expires_in_days)
+        except ValueError as exc:
+            raise CommandError(str(exc)) from exc
 
         raw_token, token = UploadToken.issue(options["name"], expires_at=expires_at)
         self.stdout.write(f"Created upload token {token.name} ({token.token_prefix})")

forensics/migrations/0013_personalaccesstoken.py

@@ -0,0 +1,33 @@
+# Generated by Django 6.0.6 on 2026-07-06 20:05
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('forensics', '0012_analysisrun_created_by_analysisrun_notes_and_more'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='PersonalAccessToken',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('name', models.CharField(max_length=120)),
+                ('token_prefix', models.CharField(max_length=16, unique=True)),
+                ('token_hash', models.CharField(max_length=128)),
+                ('is_active', models.BooleanField(default=True)),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('expires_at', models.DateTimeField(blank=True, help_text='Optional expiry. Null means the token never expires.', null=True)),
+                ('last_used_at', models.DateTimeField(blank=True, null=True)),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='access_tokens', to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'ordering': ['user', 'name', 'token_prefix'],
+            },
+        ),
+    ]