From 441a15e7fa9ae185b5958d00f310306b79db6f6f Mon Sep 17 00:00:00 2001 From: "Javier G. Montoya S" Date: Mon, 6 Jul 2026 19:39:35 -0400 Subject: [PATCH 1/2] feat: add authenticated streaming group export with self-service read tokens --- AGENTS.md | 19 + Dockerfile | 9 +- config/settings.py | 4 + docker-compose.yml | 4 +- docs/api-v1.md | 61 ++ docs/deployment.md | 29 + forensics/admin.py | 18 + forensics/analysis.py | 39 +- .../commands/create_access_token.py | 55 ++ .../commands/create_upload_token.py | 13 +- .../migrations/0013_personalaccesstoken.py | 33 + forensics/models.py | 129 ++-- forensics/streaming.py | 70 ++ forensics/templates/forensics/profile.html | 87 +++ forensics/tests.py | 603 ++++++++++++++++++ forensics/token_crypto.py | 129 ++++ forensics/urls.py | 11 + forensics/views.py | 251 +++++++- 18 files changed, 1443 insertions(+), 121 deletions(-) create mode 100644 forensics/management/commands/create_access_token.py create mode 100644 forensics/migrations/0013_personalaccesstoken.py create mode 100644 forensics/streaming.py create mode 100644 forensics/token_crypto.py diff --git a/AGENTS.md b/AGENTS.md index 7f1a250..5029922 100644 --- a/AGENTS.md +++ b/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//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 ` (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 diff --git a/Dockerfile b/Dockerfile index eee3ff8..8d1b0fd 100644 --- a/Dockerfile +++ b/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"] diff --git a/config/settings.py b/config/settings.py index 5949b93..00a52af 100644 --- a/config/settings.py +++ b/config/settings.py @@ -271,6 +271,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 diff --git a/docker-compose.yml b/docker-compose.yml index 4a0ff09..00cf924 100644 --- a/docker-compose.yml +++ b/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: [ diff --git a/docs/api-v1.md b/docs/api-v1.md index 1682d23..7c3ce54 100644 --- a/docs/api-v1.md +++ b/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 "" --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. + 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,58 @@ 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. + ### Delivery - `GET /api/v1/groups/{group_slug}/delivery/` diff --git a/docs/deployment.md b/docs/deployment.md index d6d13cd..83614fa 100644 --- a/docs/deployment.md +++ b/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 `DISABLE_SERVER_SIDE_CURSORS=1` in that case (memory stays + bounded via 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. diff --git a/forensics/admin.py b/forensics/admin.py index e73c617..b99e23b 100644 --- a/forensics/admin.py +++ b/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 = ( diff --git a/forensics/analysis.py b/forensics/analysis.py index 07f8baf..5c2e9d0 100644 --- a/forensics/analysis.py +++ b/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], diff --git a/forensics/management/commands/create_access_token.py b/forensics/management/commands/create_access_token.py new file mode 100644 index 0000000..6648aad --- /dev/null +++ b/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) diff --git a/forensics/management/commands/create_upload_token.py b/forensics/management/commands/create_upload_token.py index b87a8b9..b502612 100644 --- a/forensics/management/commands/create_upload_token.py +++ b/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})") diff --git a/forensics/migrations/0013_personalaccesstoken.py b/forensics/migrations/0013_personalaccesstoken.py new file mode 100644 index 0000000..c7d2a30 --- /dev/null +++ b/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'], + }, + ), + ] diff --git a/forensics/models.py b/forensics/models.py index 8628724..830a0bf 100644 --- a/forensics/models.py +++ b/forensics/models.py @@ -1,8 +1,5 @@ from __future__ import annotations -import hashlib -import hmac -import secrets from datetime import datetime from django.conf import settings @@ -10,6 +7,8 @@ from django.db import models from django.utils import timezone +from . import token_crypto + class AuditGroup(models.Model): name = models.CharField(max_length=200) @@ -51,78 +50,90 @@ def __str__(self) -> str: @classmethod def issue(cls, name: str, expires_at: datetime | None = None) -> tuple[str, UploadToken]: - prefix = secrets.token_hex(4) - secret = secrets.token_urlsafe(32) - raw_token = f"{cls.TOKEN_PREFIX}_{prefix}_{secret}" + raw_token, lookup_prefix, secret = token_crypto.generate_raw_token(cls.TOKEN_PREFIX) token = cls.objects.create( name=name, - token_prefix=prefix, - token_hash=cls.hash_secret(secret), + token_prefix=lookup_prefix, + token_hash=token_crypto.hash_secret(secret), expires_at=expires_at, ) return raw_token, token @classmethod def hash_secret(cls, secret: str, *, key: str | None = None) -> str: - # Keyed on GOGGLES_TOKEN_HASH_KEY (a dedicated, independently - # rotatable secret) rather than SECRET_KEY, so rotating the Django - # signing key does not invalidate every issued upload token. The - # setting falls back to SECRET_KEY when unset, preserving existing - # hashes for deployments that have not provisioned a dedicated key. - hash_key = key or settings.GOGGLES_TOKEN_HASH_KEY - return hmac.new( - hash_key.encode("utf-8"), secret.encode("utf-8"), hashlib.sha256 - ).hexdigest() - - @classmethod - def legacy_hash_keys(cls) -> tuple[str, ...]: - """Return fallback keys for one-way lazy migration of legacy hashes. - - Before GOGGLES_TOKEN_HASH_KEY existed, token hashes were keyed on - SECRET_KEY. During the first dedicated-key cutover, an active token can - be authenticated against that legacy hash and then rekeyed to the - dedicated setting. - """ - if settings.SECRET_KEY and settings.SECRET_KEY != settings.GOGGLES_TOKEN_HASH_KEY: - return (settings.SECRET_KEY,) - return () + return token_crypto.hash_secret(secret, key=key) def is_expired(self, *, at: datetime | None = None) -> bool: - if self.expires_at is None: - return False - return (at or timezone.now()) >= self.expires_at + return token_crypto.is_expired(self, at=at) @classmethod def authenticate(cls, raw_token: str | None) -> UploadToken | None: - if not raw_token: - return None - parts = raw_token.split("_", 2) - if len(parts) != 3 or parts[0] != cls.TOKEN_PREFIX: - return None - _, prefix, secret = parts - try: - token = cls.objects.get(token_prefix=prefix, is_active=True) - except cls.DoesNotExist: - return None + return token_crypto.authenticate(cls, raw_token, cls.TOKEN_PREFIX) - current_hash = cls.hash_secret(secret) - matched_legacy_key = False - if not hmac.compare_digest(token.token_hash, current_hash): - for legacy_key in cls.legacy_hash_keys(): - legacy_hash = cls.hash_secret(secret, key=legacy_key) - if hmac.compare_digest(token.token_hash, legacy_hash): - matched_legacy_key = True - break - else: - return None - - if token.is_expired(): - return None + def mark_used(self) -> None: + self.last_used_at = timezone.now() + self.save(update_fields=["last_used_at"]) + + +class PersonalAccessToken(models.Model): + """A user-owned, read-only API credential, self-service from the profile page. + + Distinct from :class:`UploadToken`: that authorizes a device to *upload*; this + authorizes a person to *read/export*. The two never overlap — separate tables + and separate raw-token type prefixes, enforced at ``authenticate``. + """ + + TOKEN_PREFIX = "gpat" + + user = models.ForeignKey( + settings.AUTH_USER_MODEL, + on_delete=models.CASCADE, + related_name="access_tokens", + ) + 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( + null=True, + blank=True, + help_text="Optional expiry. Null means the token never expires.", + ) + last_used_at = models.DateTimeField(null=True, blank=True) + + class Meta: + ordering = ["user", "name", "token_prefix"] + + def __str__(self) -> str: + state = "active" if self.is_active else "disabled" + return f"{self.name} ({self.token_prefix}, {state})" - if matched_legacy_key: - token.token_hash = current_hash - token.save(update_fields=["token_hash"]) + @classmethod + def issue( + cls, name: str, *, user, expires_at: datetime | None = None + ) -> tuple[str, PersonalAccessToken]: + raw_token, lookup_prefix, secret = token_crypto.generate_raw_token(cls.TOKEN_PREFIX) + token = cls.objects.create( + user=user, + name=name, + token_prefix=lookup_prefix, + token_hash=token_crypto.hash_secret(secret), + expires_at=expires_at, + ) + return raw_token, token + + def is_expired(self, *, at: datetime | None = None) -> bool: + return token_crypto.is_expired(self, at=at) + @classmethod + def authenticate(cls, raw_token: str | None) -> PersonalAccessToken | None: + token = token_crypto.authenticate(cls, raw_token, cls.TOKEN_PREFIX) + # A token is only as live as its owner: deactivating a user immediately + # revokes their tokens. This check is model-specific (UploadToken has no + # owner), so it lives here rather than in the shared helper. + if token is not None and not token.user.is_active: + return None return token def mark_used(self) -> None: diff --git a/forensics/streaming.py b/forensics/streaming.py new file mode 100644 index 0000000..64c684d --- /dev/null +++ b/forensics/streaming.py @@ -0,0 +1,70 @@ +"""A domain-blind NDJSON streamer for bulk exports. + +Emits one JSON object per line: a ``manifest`` line, then every row of each +:class:`ExportSection` tagged with its ``record_type``, then a terminal ``eof`` +line carrying per-section counts. Rows are pulled through ``QuerySet.iterator()`` +so peak memory is one chunk regardless of dataset size. + +This module knows nothing about the forensic domain — the caller injects the +querysets and their payload factories — so it stays a pure, reusable seam that +tests can drive with fakes. +""" + +from __future__ import annotations + +import json +import logging +from collections.abc import Callable, Iterable, Iterator +from dataclasses import dataclass + +from django.core.serializers.json import DjangoJSONEncoder +from django.db.models import QuerySet + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class ExportSection: + """One typed stream of records: a queryset and how to serialize each row.""" + + record_type: str + rows: QuerySet + to_payload: Callable[[object], dict] + + +def _line(obj: dict) -> str: + # DjangoJSONEncoder matches the project's JsonResponse idiom and keeps the + # serializer total over datetimes/Decimals — important because the broad + # except below would otherwise turn a stray un-encodable value into a silent + # truncated export. No sort_keys: keys keep insertion order, so the "t" + # discriminator leads every line (part of the wire contract). + return json.dumps(obj, cls=DjangoJSONEncoder, separators=(",", ":")) + "\n" + + +def stream_ndjson( + manifest: dict, + sections: Iterable[ExportSection], + *, + chunk_size: int = 2000, +) -> Iterator[str]: + """Yield the export as NDJSON lines: manifest, records, then eof (or error). + + Fail-closed: once the HTTP status is committed the body cannot signal failure + out of band, so a mid-stream error yields a terminal ``{"t":"error"}`` line + and **no** ``eof``. Consumers treat "last line is not eof" as incomplete. + ``chunk_size`` is forwarded to ``.iterator()`` (required for querysets that + carry ``prefetch_related``). + """ + try: + yield _line({"t": "manifest", **manifest}) + counts: dict[str, int] = {} + for section in sections: + count = 0 + for row in section.rows.iterator(chunk_size=chunk_size): + yield _line({"t": section.record_type, **section.to_payload(row)}) + count += 1 + counts[section.record_type] = count + yield _line({"t": "eof", "complete": True, "counts": counts}) + except Exception: # noqa: BLE001 — fail closed: report in-band, never half-claim success + logger.exception("group export stream failed") + yield _line({"t": "error", "complete": False}) diff --git a/forensics/templates/forensics/profile.html b/forensics/templates/forensics/profile.html index 90d6e9c..7a090b8 100644 --- a/forensics/templates/forensics/profile.html +++ b/forensics/templates/forensics/profile.html @@ -73,4 +73,91 @@

Profile

+ +
+
+ Personal access tokens +
+
+

+ Read-only API credentials — send as Authorization: Bearer <token>, + e.g. to stream a group export. They can read forensic data but never upload, and stop + working the moment you revoke them or your account is disabled. +

+ + {% if new_token %} +
+ + +

This is the only time the token is shown. Store it securely.

+
+ {% endif %} + +
+ {% csrf_token %} +
+ + +
+
+ + +
+
+ +
+
+
+ +
+ + + + + + + + + + + + + + {% for token in access_tokens %} + + + + + + + + + + {% empty %} + + {% endfor %} + +
NamePrefixCreatedLast usedExpiresStatus
{{ token.name }}{{ token.token_prefix }}{{ token.created_at|date:"Y-m-d" }} + {% if token.last_used_at %}{{ token.last_used_at|date:"Y-m-d H:i" }}{% else %}—{% endif %} + + {% if token.expires_at %}{{ token.expires_at|date:"Y-m-d" }}{% else %}Never{% endif %} + + {% if token.is_active %} + Active + {% else %} + Revoked + {% endif %} + + {% if token.is_active %} +
+ {% csrf_token %} + +
+ {% endif %} +
No personal access tokens yet.
+
+
{% endblock %} diff --git a/forensics/tests.py b/forensics/tests.py index cafd4c7..8fbe2a6 100644 --- a/forensics/tests.py +++ b/forensics/tests.py @@ -26,6 +26,7 @@ from . import ingest as ingest_module from . import projections as projections_module from .analysis import ( + EXPORT_SENSITIVITY, audit_files_for_group, display_group_ref, group_list_rows, @@ -51,18 +52,24 @@ DeliveryObservation, EpochStateTransition, NetworkObservation, + PersonalAccessToken, RecipientExpectation, StateDelta, UploadToken, ) from .seed_data import SCENARIO_GROUPS, build_dev_scenario, group_ref_for +from .streaming import ExportSection, stream_ndjson +from .token_crypto import MAX_TOKEN_EXPIRY_DAYS, expiry_from_days from .views import ( AUDIT_FILE_EVENT_PAGE_SIZE, GROUP_DETAIL_TAB_EVENT_LIMIT, GROUP_EPOCH_FIELDS, + GROUP_EXPORT_SCHEMA_VERSION, + GROUP_PROJECTION_API_DEFAULT_LIMIT, RAW_TEXT_PREVIEW_CHARS, audit_bytes_from_request, client_ip, + delivery_identity_index, group_api_payload, group_detail_shell_context, group_engine_rows, @@ -5097,6 +5104,15 @@ def test_create_upload_token_command_defaults_to_no_expiry(self): self.assertIsNone(token.expires_at) self.assertIn("does not expire", out.getvalue()) + def test_create_upload_token_command_rejects_over_max_expiry_cleanly(self): + # Shares the bounded expiry helper with the export tokens: an overflow-scale + # day count is a clean CommandError, not an uncaught OverflowError (PR #199). + with self.assertRaises(CommandError): + call_command( + "create_upload_token", "ios qa", "--expires-in-days", str(MAX_TOKEN_EXPIRY_DAYS + 1) + ) + self.assertFalse(UploadToken.objects.filter(name="ios qa").exists()) + @override_settings(GOGGLES_UPLOADS_ENABLED=False) def test_global_upload_pause_rejects_authenticated_upload_without_ingesting(self): raw_token, token = UploadToken.issue("ios qa") @@ -5110,6 +5126,593 @@ def test_global_upload_pause_rejects_authenticated_upload_without_ingesting(self self.assertIsNone(token.last_used_at) +class PersonalAccessTokenTests(TestCase): + """The user-owned, read-only credential the streaming export accepts. + + Distinct from UploadToken: self-service, read-only, and only as live as its + owner. See authenticated-group-export.md. The shared hashing/rekey/expiry + mechanics live in token_crypto and are exercised via UploadToken elsewhere; + these tests cover the PersonalAccessToken-specific contract. + """ + + def setUp(self): + self.user = User.objects.create_user( + username="reader", + password="correct horse battery staple", + ) + + def test_issue_returns_gpat_prefixed_token_and_stores_only_the_hash(self): + raw_token, token = PersonalAccessToken.issue("cgka laptop", user=self.user) + + self.assertTrue(raw_token.startswith("gpat_")) + self.assertEqual(token.user, self.user) + # The raw secret is shown once and never persisted; only its hash is stored. + secret = raw_token.split("_", 2)[2] + self.assertNotIn(secret, token.token_hash) + self.assertEqual(token.token_hash, PersonalAccessToken.objects.get(pk=token.pk).token_hash) + + def test_issued_lookup_prefix_fills_the_field_without_exceeding_it(self): + # PR #199: the lookup prefix is 16 hex chars (64 bits), filling + # token_prefix's max_length=16 — wide enough that an accidental + # unique-collision on issuance (which does not retry) is negligible. + _raw_token, token = PersonalAccessToken.issue("prefix width", user=self.user) + field_max = PersonalAccessToken._meta.get_field("token_prefix").max_length + self.assertEqual(len(token.token_prefix), 16) + self.assertLessEqual(len(token.token_prefix), field_max) + + def test_authenticate_round_trips_a_freshly_issued_token(self): + raw_token, token = PersonalAccessToken.issue("cgka laptop", user=self.user) + self.assertEqual(PersonalAccessToken.authenticate(raw_token).pk, token.pk) + + def test_inactive_token_is_rejected(self): + raw_token, token = PersonalAccessToken.issue("revoked", user=self.user) + token.is_active = False + token.save(update_fields=["is_active"]) + self.assertIsNone(PersonalAccessToken.authenticate(raw_token)) + + def test_expired_token_is_rejected(self): + raw_token, _token = PersonalAccessToken.issue( + "stale", user=self.user, expires_at=timezone.now() - timedelta(seconds=1) + ) + self.assertIsNone(PersonalAccessToken.authenticate(raw_token)) + + def test_token_of_deactivated_user_is_rejected(self): + # A token is only as live as its owner — unique to PersonalAccessToken. + raw_token, _token = PersonalAccessToken.issue("cgka laptop", user=self.user) + self.user.is_active = False + self.user.save(update_fields=["is_active"]) + self.assertIsNone(PersonalAccessToken.authenticate(raw_token)) + + def test_upload_and_personal_tokens_never_cross_authenticate(self): + # Distinct type prefixes (goggles_ vs gpat_) make the two families + # structurally un-confusable at authenticate(). + upload_raw, _token = UploadToken.issue("device") + pat_raw, _pat = PersonalAccessToken.issue("laptop", user=self.user) + + self.assertIsNone(PersonalAccessToken.authenticate(upload_raw)) + self.assertIsNone(UploadToken.authenticate(pat_raw)) + + def test_mark_used_records_last_used_at_without_revoking(self): + _raw_token, token = PersonalAccessToken.issue("cgka laptop", user=self.user) + self.assertIsNone(token.last_used_at) + + token.mark_used() + + token.refresh_from_db() + self.assertIsNotNone(token.last_used_at) + self.assertTrue(token.is_active) + + +class StreamNdjsonTests(SimpleTestCase): + """The domain-blind NDJSON serializer: manifest-first, typed rows, and a + fail-closed eof/error terminator. See forensics/streaming.py.""" + + @staticmethod + def _rows(items, *, raise_at=None): + """A minimal QuerySet stand-in exposing just ``.iterator(chunk_size=)``.""" + + class _FakeRows: + def iterator(self, chunk_size): + # The serializer must always pass a chunk_size — prefetch_related + # querysets require it under .iterator(). + assert chunk_size, "stream_ndjson must forward a chunk_size" + for index, item in enumerate(items): + if raise_at is not None and index == raise_at: + raise RuntimeError("db cursor died mid-stream") + yield item + + return _FakeRows() + + def _records(self, manifest, sections): + return [json.loads(line) for line in stream_ndjson(manifest, sections)] + + def test_manifest_first_then_typed_rows_then_eof_with_counts(self): + sections = [ + ExportSection("event", self._rows([{"id": 1}, {"id": 2}]), lambda row: row), + ExportSection("source", self._rows([{"id": 9}]), lambda row: row), + ] + + records = self._records({"schema_version": "goggles-group-export/v1"}, sections) + + self.assertEqual(records[0]["t"], "manifest") + self.assertEqual(records[0]["schema_version"], "goggles-group-export/v1") + self.assertEqual([r["t"] for r in records[1:4]], ["event", "event", "source"]) + self.assertEqual( + records[-1], + {"t": "eof", "complete": True, "counts": {"event": 2, "source": 1}}, + ) + + def test_empty_section_still_reports_complete_with_zero_count(self): + records = self._records({}, [ExportSection("event", self._rows([]), lambda row: row)]) + self.assertEqual(records[-1], {"t": "eof", "complete": True, "counts": {"event": 0}}) + + def test_mid_stream_error_yields_error_line_and_no_eof(self): + sections = [ + ExportSection("event", self._rows([{"id": 1}, {"id": 2}], raise_at=1), lambda row: row), + ] + + with self.assertLogs("forensics.streaming", level="ERROR"): + records = self._records({}, sections) + + self.assertEqual(records[1], {"t": "event", "id": 1}) # the row before the failure streamed + self.assertEqual(records[-1], {"t": "error", "complete": False}) + self.assertNotIn("eof", [record["t"] for record in records]) + + def test_t_discriminator_leads_every_line(self): + # No sort_keys: a consumer may route by the line's leading key. + lines = list( + stream_ndjson( + {"schema_version": "goggles-group-export/v1"}, + [ExportSection("event", self._rows([{"z": 1}]), lambda row: row)], + ) + ) + for line in lines: + self.assertTrue(line.startswith('{"t":'), line) + + +class TokenExpiryFromDaysTests(SimpleTestCase): + """The shared bounded expiry helper (PR #199): a positive, sane day count → + future datetime; non-positive, over-ceiling, or overflow-scale → ValueError + raised *before* any timedelta arithmetic, so it never leaks an OverflowError.""" + + def test_positive_days_within_bound_returns_future_datetime(self): + before = timezone.now() + self.assertGreaterEqual(expiry_from_days(7), before + timedelta(days=7)) + + def test_ceiling_is_allowed(self): + self.assertIsNotNone(expiry_from_days(MAX_TOKEN_EXPIRY_DAYS)) + + def test_rejects_non_positive(self): + for days in (0, -1): + with self.subTest(days=days), self.assertRaises(ValueError): + expiry_from_days(days) + + def test_rejects_over_ceiling_and_overflow_scale_as_valueerror(self): + # 99_999_999_999 would raise OverflowError at timedelta(); the bound check + # must catch it as a ValueError first. + for days in (MAX_TOKEN_EXPIRY_DAYS + 1, 99_999_999_999): + with self.subTest(days=days), self.assertRaises(ValueError): + expiry_from_days(days) + + +class DeliveryIdentityIndexTests(TestCase): + """delivery_identity_index collects distinct identities via bounded SQL, not a + Python scan of every event (B1 in authenticated-group-export.md). Output must + match the previous implementation exactly — it is shared with the delivery tab. + """ + + def _valid_event(self, group, audit_file, seq, **fields): + return AuditEvent.objects.create( + audit_file=audit_file, + group=group, + line_number=seq, + line_hash=f"{seq:064d}", + raw_line="{}", + parse_status=AuditEvent.STATUS_VALID, + event_type="transport_received", + group_ref=GROUP_REF, + seq=seq, + wall_time_ms=1_700_000_000_000 + seq, + **fields, + ) + + def test_collects_distinct_identities_from_events_files_and_context(self): + group = AuditGroup.objects.create(name="G", slug="g", group_ref=GROUP_REF) + file_with_pubkey = AuditFile.objects.create( + file_sha256="a" * 64, + byte_size=1, + raw_text="{}", + validation_status=AuditFile.STATUS_VALID, + source_account_pubkey_hex="ff" * 32, + ) + file_with_pubkey.groups.add(group) + plain_file = AuditFile.objects.create( + file_sha256="b" * 64, + byte_size=1, + raw_text="{}", + validation_status=AuditFile.STATUS_VALID, + ) + plain_file.groups.add(group) + + # Duplicates across events collapse; the account pubkey is drawn from both + # the backing file and the event's context_source JSON. + self._valid_event( + group, file_with_pubkey, 1, account_ref=ACCOUNT_ALICE, engine_id=ENGINE_ALICE + ) + self._valid_event( + group, file_with_pubkey, 2, account_ref=ACCOUNT_ALICE, engine_id=ENGINE_ALICE + ) + self._valid_event( + group, + plain_file, + 3, + account_ref=ACCOUNT_BOB, + engine_id=ENGINE_BOB, + context_source={"account_pubkey_hex": "cc" * 32}, + ) + # An event with blank identity fields contributes nothing. + self._valid_event(group, plain_file, 4) + + index = delivery_identity_index(group) + + self.assertEqual(index["account_refs"], {ACCOUNT_ALICE, ACCOUNT_BOB}) + self.assertEqual(index["engine_ids"], {ENGINE_ALICE, ENGINE_BOB}) + self.assertEqual(index["pubkeys_hex"], {"ff" * 32, "cc" * 32}) + + def test_ignores_invalid_events(self): + group = AuditGroup.objects.create(name="G", slug="g", group_ref=GROUP_REF) + audit_file = AuditFile.objects.create( + file_sha256="a" * 64, + byte_size=1, + raw_text="{}", + validation_status=AuditFile.STATUS_VALID, + source_account_pubkey_hex="ff" * 32, + ) + audit_file.groups.add(group) + AuditEvent.objects.create( + audit_file=audit_file, + group=group, + line_number=1, + line_hash="1".ljust(64, "0"), + raw_line="{}", + parse_status=AuditEvent.STATUS_INVALID, + event_type="transport_received", + account_ref=ACCOUNT_ALICE, + engine_id=ENGINE_ALICE, + group_ref=GROUP_REF, + seq=1, + wall_time_ms=1, + ) + + self.assertEqual( + delivery_identity_index(group), + {"account_refs": set(), "engine_ids": set(), "pubkeys_hex": set()}, + ) + + +class GroupExportStreamTests(TestCase): + """The authenticated NDJSON streaming export. See authenticated-group-export.md.""" + + 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-export-stream", kwargs={"slug": self.group.slug}) + + @staticmethod + def _records(response): + body = b"".join(response.streaming_content).decode("utf-8") + return [json.loads(line) for line in body.splitlines() if line] + + def test_requires_authentication(self): + response = self.client.get(self.url) + self.assertEqual(response.status_code, 401) + + 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) + + 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) + + def test_upload_token_cannot_read_the_export(self): + # Least privilege: an upload credential can never read forensic data. + 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_streams_with_personal_access_token_and_records_last_used(self): + raw_token, token = PersonalAccessToken.issue("cgka", user=self.user) + + response = self.client.get(self.url, HTTP_AUTHORIZATION=f"Bearer {raw_token}") + + self.assertEqual(response.status_code, 200) + self.assertEqual(response["Content-Type"], "application/x-ndjson") + self.assertIn("attachment", response["Content-Disposition"]) + self.assertEqual(response["Cache-Control"], "no-store") + records = self._records(response) + self.assertEqual(records[0]["t"], "manifest") + self.assertEqual(records[-1]["t"], "eof") + self.assertTrue(records[-1]["complete"]) + token.refresh_from_db() + self.assertIsNotNone(token.last_used_at) + + def test_streams_with_logged_in_session(self): + self.client.login(username="reader", password="correct horse battery staple") + + records = self._records(self.client.get(self.url)) + + self.assertEqual(records[0]["t"], "manifest") + self.assertEqual(records[-1]["t"], "eof") + + def test_manifest_advertises_sections_and_eof_counts_are_accurate(self): + self.client.login(username="reader", password="correct horse battery staple") + + records = self._records(self.client.get(self.url)) + + manifest, eof = records[0], records[-1] + self.assertEqual(manifest["schema_version"], GROUP_EXPORT_SCHEMA_VERSION) + self.assertEqual(manifest["sensitivity"], EXPORT_SENSITIVITY) + # The export is unconditionally complete: it advertises no filter contract. + self.assertNotIn("filters", manifest) + + # Advertised sections are exactly the record types the stream can emit, + # and eof.counts match the rows actually streamed. + body_records = records[1:-1] + for record_type, count in eof["counts"].items(): + self.assertIn(record_type, manifest["sections"]) + self.assertEqual(sum(1 for record in body_records if record["t"] == record_type), count) + # The representative log yields at least its source file and its events. + self.assertGreaterEqual(eof["counts"]["source"], 1) + self.assertGreaterEqual(eof["counts"]["event"], 1) + + def test_streams_every_row_past_the_projection_page_cap(self): + # The paginated projections API caps a section at GROUP_PROJECTION_API_DEFAULT_LIMIT + # (500). The export must stream every row — this is its reason to exist. + over_cap = GROUP_PROJECTION_API_DEFAULT_LIMIT + 1 + evidence_event = AuditEvent.objects.filter( + group=self.group, parse_status=AuditEvent.STATUS_VALID + ).first() + StateDelta.objects.bulk_create( + [ + StateDelta( + group=self.group, + audit_event=evidence_event, + epoch=i, + change_kind=f"state-marker-{i:04d}", + wall_time_ms=1_700_000_300_000 + i, + ) + for i in range(over_cap) + ] + ) + self.client.login(username="reader", password="correct horse battery staple") + + records = self._records(self.client.get(self.url)) + + self.assertGreaterEqual(records[-1]["counts"]["state_delta"], over_cap) + + def test_query_filters_are_ignored_export_is_always_complete(self): + # Regression (PR #199): the manifest advertised filters that only some + # sections honored — projections were filtered while events/sources were not, + # and severity/convergence message_id were dropped entirely. The export now + # ignores query filters: the same complete group whether or not they are sent. + alice_event = AuditEvent.objects.filter( + group=self.group, parse_status=AuditEvent.STATUS_VALID, engine_id=ENGINE_ALICE + ).first() + StateDelta.objects.create( + group=self.group, + audit_event=alice_event, + epoch=0, + change_kind="group_renamed", + wall_time_ms=1_700_000_300_000, + ) + self.client.login(username="reader", password="correct horse battery staple") + + unfiltered = self._records(self.client.get(self.url)) + # A real engine filter that only projections honored would drop the state_delta + # above (its event is ENGINE_ALICE), leaving events untouched — the old split. + filtered = self._records( + self.client.get(self.url, {"engine_id": "ff" * 16, "severity": "error"}) + ) + + self.assertNotIn("filters", unfiltered[0]) + self.assertEqual(unfiltered[-1]["counts"], filtered[-1]["counts"]) + self.assertGreaterEqual(unfiltered[-1]["counts"]["state_delta"], 1) + + @override_settings(GOGGLES_EXPORTS_ENABLED=False) + def test_disabled_export_returns_503(self): + raw_token, _token = PersonalAccessToken.issue("cgka", user=self.user) + response = self.client.get(self.url, HTTP_AUTHORIZATION=f"Bearer {raw_token}") + self.assertEqual(response.status_code, 503) + + def test_mid_stream_error_yields_error_line_and_no_eof_within_a_200(self): + # Once the manifest is sent the status is committed at 200, so a later + # failure can only be signalled in-band: an error line and no eof (M4). + self.client.login(username="reader", password="correct horse battery staple") + + with mock.patch( + "forensics.views.agent_event_row", + side_effect=RuntimeError("db cursor died mid-stream"), + ): + with self.assertLogs("forensics.streaming", level="ERROR"): + response = self.client.get(self.url) + records = self._records(response) + + self.assertEqual(response.status_code, 200) + self.assertEqual(records[0]["t"], "manifest") + # Rows emitted before the failure still streamed (sources precede events). + self.assertTrue(any(record["t"] == "source" for record in records)) + self.assertEqual(records[-1], {"t": "error", "complete": False}) + self.assertNotIn("eof", {record["t"] for record in records}) + + +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.""" + + def setUp(self): + self.user = User.objects.create_user( + username="reader", password="correct horse battery staple" + ) + self.other = User.objects.create_user( + username="other", password="correct horse battery staple" + ) + self.client.login(username="reader", password="correct horse battery staple") + + def test_profile_lists_only_the_callers_tokens(self): + _raw, mine = PersonalAccessToken.issue("mine-laptop", user=self.user) + PersonalAccessToken.issue("their-laptop", user=self.other) + + response = self.client.get(reverse("profile")) + + self.assertEqual(response.status_code, 200) + self.assertContains(response, "mine-laptop") + self.assertNotContains(response, "their-laptop") + self.assertEqual(list(response.context["access_tokens"]), [mine]) + + def test_create_token_shows_secret_once_and_owns_it(self): + response = self.client.post(reverse("create-access-token"), {"name": "cgka pipeline"}) + + self.assertEqual(response.status_code, 200) + # The page shows the raw secret once, so it must not be cached anywhere. + self.assertEqual(response["Cache-Control"], "no-store") + token = PersonalAccessToken.objects.get(user=self.user, name="cgka pipeline") + raw_token = response.context["new_token"] + self.assertTrue(raw_token.startswith("gpat_")) + self.assertContains(response, raw_token) # shown exactly once, in the page + self.assertEqual(PersonalAccessToken.authenticate(raw_token).pk, token.pk) + + def test_create_token_honors_optional_expiry(self): + before = timezone.now() + self.client.post(reverse("create-access-token"), {"name": "temp", "expires_in_days": "7"}) + token = PersonalAccessToken.objects.get(user=self.user, name="temp") + self.assertIsNotNone(token.expires_at) + self.assertGreaterEqual(token.expires_at, before + timedelta(days=7)) + + def test_create_token_defaults_to_no_expiry(self): + self.client.post(reverse("create-access-token"), {"name": "forever"}) + token = PersonalAccessToken.objects.get(user=self.user, name="forever") + self.assertIsNone(token.expires_at) + + def test_create_token_rejects_invalid_expiry_and_mints_nothing(self): + # Regression (PR #199): a crafted POST bypasses the input's min=1, so a + # non-positive, unparseable, or absurdly large expiry must be rejected — never + # silently minting a permanent read credential, and never crashing on the + # OverflowError a huge day count would otherwise raise. Blank still means "never + # expires" (tested above). + for bad_value in ("0", "-5", "abc", "99999999999"): + with self.subTest(expires_in_days=bad_value): + response = self.client.post( + reverse("create-access-token"), + {"name": "sneaky", "expires_in_days": bad_value}, + ) + self.assertEqual(response.status_code, 200) + self.assertFalse( + PersonalAccessToken.objects.filter(user=self.user, name="sneaky").exists() + ) + + def test_revoke_deactivates_own_token(self): + _raw, token = PersonalAccessToken.issue("mine", user=self.user) + + response = self.client.post(reverse("revoke-access-token", kwargs={"pk": token.pk})) + + self.assertRedirects(response, reverse("profile")) + token.refresh_from_db() + self.assertFalse(token.is_active) + + def test_cannot_revoke_another_users_token(self): + _raw, victim = PersonalAccessToken.issue("theirs", user=self.other) + + response = self.client.post(reverse("revoke-access-token", kwargs={"pk": victim.pk})) + + self.assertEqual(response.status_code, 404) + victim.refresh_from_db() + self.assertTrue(victim.is_active) + + def test_token_endpoints_require_login(self): + self.client.logout() + + response = self.client.post(reverse("create-access-token"), {"name": "x"}) + + self.assertEqual(response.status_code, 302) # redirect to login + self.assertEqual(PersonalAccessToken.objects.count(), 0) + + def test_password_change_still_works_after_refactor(self): + response = self.client.post( + reverse("profile"), + { + "old_password": "correct horse battery staple", + "new_password1": "a-fresh-passphrase-9271", + "new_password2": "a-fresh-passphrase-9271", + }, + ) + + self.assertRedirects(response, reverse("profile")) + self.user.refresh_from_db() + self.assertTrue(self.user.check_password("a-fresh-passphrase-9271")) + + +class CreateAccessTokenCommandTests(TestCase): + def setUp(self): + self.user = User.objects.create_user( + username="svc-cgka", password="correct horse battery staple" + ) + + def test_creates_read_token_owned_by_named_user(self): + out = StringIO() + call_command("create_access_token", "cgka pipeline", "--user", "svc-cgka", stdout=out) + + token = PersonalAccessToken.objects.get(name="cgka pipeline") + self.assertEqual(token.user, self.user) + self.assertIsNone(token.expires_at) + raw_token = out.getvalue().strip().splitlines()[-1] + self.assertTrue(raw_token.startswith("gpat_")) + self.assertEqual(PersonalAccessToken.authenticate(raw_token).pk, token.pk) + + def test_expiry_flag_sets_expires_at(self): + out = StringIO() + before = timezone.now() + call_command( + "create_access_token", + "temp", + "--user", + "svc-cgka", + "--expires-in-days", + "7", + stdout=out, + ) + token = PersonalAccessToken.objects.get(name="temp") + self.assertGreaterEqual(token.expires_at, before + timedelta(days=7)) + + def test_unknown_user_errors(self): + with self.assertRaises(CommandError): + call_command("create_access_token", "x", "--user", "nobody") + + def test_non_positive_expiry_errors(self): + with self.assertRaises(CommandError): + call_command("create_access_token", "x", "--user", "svc-cgka", "--expires-in-days", "0") + + def test_over_max_expiry_errors_cleanly(self): + # A day count that would overflow timedelta/datetime is rejected as a clean + # CommandError, not an uncaught OverflowError (PR #199). + with self.assertRaises(CommandError): + call_command( + "create_access_token", + "x", + "--user", + "svc-cgka", + "--expires-in-days", + str(MAX_TOKEN_EXPIRY_DAYS + 1), + ) + self.assertFalse(PersonalAccessToken.objects.filter(name="x").exists()) + + class PurgeAuditDataCommandTests(TestCase): def seed_audit_data(self): user = User.objects.create_user( diff --git a/forensics/token_crypto.py b/forensics/token_crypto.py new file mode 100644 index 0000000..f6c0afd --- /dev/null +++ b/forensics/token_crypto.py @@ -0,0 +1,129 @@ +"""Shared secret-token crypto for the app's bearer-token models. + +Single-sources the security-critical mechanics — HMAC-SHA256 hashing keyed on a +dedicated, rotatable secret, constant-time verification, one-way lazy rekeying of +legacy ``SECRET_KEY``-derived hashes, and expiry enforcement — so that +``UploadToken`` and ``PersonalAccessToken`` cannot drift apart on how a credential +is minted or checked. + +Raw token format: ``__``. The type prefix +(``goggles`` for uploads, ``gpat`` for personal access tokens) makes the two +credential families structurally un-confusable: :func:`authenticate` rejects a raw +token whose type prefix is not the caller's own. + +Duck-typed token protocol (both models satisfy it): a ``token_prefix`` / +``token_hash`` / ``is_active`` / ``expires_at`` field set, an ``objects`` manager, +and an ``is_expired()`` method. Model-specific checks (e.g. "owner is active") are +layered by the caller on the returned row, keeping this module model-agnostic. +""" + +from __future__ import annotations + +import hashlib +import hmac +import secrets +from datetime import datetime, timedelta + +from django.conf import settings +from django.utils import timezone + +# A sanity ceiling on token lifetime (~100 years). It rejects absurd input and keeps +# ``now() + timedelta(days=...)`` well clear of the OverflowError that extreme day +# counts would otherwise raise. +MAX_TOKEN_EXPIRY_DAYS = 36500 + + +def generate_raw_token(type_prefix: str) -> tuple[str, str, str]: + """Mint a fresh credential, returning ``(raw_token, lookup_prefix, secret)``. + + Only the hash of ``secret`` is ever persisted; the raw token is shown once to + the caller and never stored. + """ + # 8 bytes → 16 hex chars, filling the token_prefix max_length=16. 64 bits keeps an + # accidental collision on the unique lookup key negligible; issuance does not retry. + lookup_prefix = secrets.token_hex(8) + secret = secrets.token_urlsafe(32) + raw_token = f"{type_prefix}_{lookup_prefix}_{secret}" + return raw_token, lookup_prefix, secret + + +def hash_secret(secret: str, *, key: str | None = None) -> str: + # Keyed on GOGGLES_TOKEN_HASH_KEY (a dedicated, independently rotatable + # secret) rather than SECRET_KEY, so rotating the Django signing key does not + # invalidate every issued token. The setting falls back to SECRET_KEY when + # unset, preserving existing hashes for deployments that have not provisioned + # a dedicated key. + hash_key = key or settings.GOGGLES_TOKEN_HASH_KEY + return hmac.new(hash_key.encode("utf-8"), secret.encode("utf-8"), hashlib.sha256).hexdigest() + + +def legacy_hash_keys() -> tuple[str, ...]: + """Return fallback keys for one-way lazy migration of legacy hashes. + + Before GOGGLES_TOKEN_HASH_KEY existed, token hashes were keyed on SECRET_KEY. + During the first dedicated-key cutover, an active token can be authenticated + against that legacy hash and then rekeyed to the dedicated setting. + """ + if settings.SECRET_KEY and settings.SECRET_KEY != settings.GOGGLES_TOKEN_HASH_KEY: + return (settings.SECRET_KEY,) + return () + + +def is_expired(token, *, at: datetime | None = None) -> bool: + if token.expires_at is None: + return False + return (at or timezone.now()) >= token.expires_at + + +def expiry_from_days(days: int) -> datetime: + """Return a bounded expiry datetime for a positive day count. + + Rejects non-positive and absurdly large values with ``ValueError`` *before* any + ``timedelta`` arithmetic, so a crafted day count is a clean rejection rather than + an ``OverflowError`` from ``timedelta`` / ``datetime`` overflow. + """ + if days <= 0: + raise ValueError("Token expiry must be a positive number of days.") + if days > MAX_TOKEN_EXPIRY_DAYS: + raise ValueError(f"Token expiry must be at most {MAX_TOKEN_EXPIRY_DAYS} days.") + return timezone.now() + timedelta(days=days) + + +def authenticate(model, raw_token: str | None, type_prefix: str): + """Return the active, unexpired ``model`` row matching ``raw_token``, or None. + + Validates the type prefix, looks the token up by its lookup prefix, verifies + the secret in constant time (with legacy-key fallback and lazy rekey to the + dedicated key), and enforces expiry. Callers layer any model-specific checks + on the result. + """ + if not raw_token: + return None + parts = raw_token.split("_", 2) + if len(parts) != 3 or parts[0] != type_prefix: + return None + _, lookup_prefix, secret = parts + try: + token = model.objects.get(token_prefix=lookup_prefix, is_active=True) + except model.DoesNotExist: + return None + + current_hash = hash_secret(secret) + matched_legacy_key = False + if not hmac.compare_digest(token.token_hash, current_hash): + for legacy_key in legacy_hash_keys(): + legacy = hash_secret(secret, key=legacy_key) + if hmac.compare_digest(token.token_hash, legacy): + matched_legacy_key = True + break + else: + return None + + if is_expired(token): + return None + + if matched_legacy_key: + token.token_hash = current_hash + token.save(update_fields=["token_hash"]) + + return token diff --git a/forensics/urls.py b/forensics/urls.py index 5d6f1fa..94567ce 100644 --- a/forensics/urls.py +++ b/forensics/urls.py @@ -6,6 +6,12 @@ path("healthz/", views.healthz, name="healthz"), path("", views.group_list, name="group-list"), path("profile/", views.profile, name="profile"), + path("profile/tokens/", views.create_access_token, name="create-access-token"), + path( + "profile/tokens//revoke/", + views.revoke_access_token, + name="revoke-access-token", + ), path("uploads/", views.upload_log_list, name="upload-log-list"), path( "investigations/accounts//", @@ -52,6 +58,11 @@ name="api-group-audit-log-upload", ), path("api/v1/groups/", views.api_group_list, name="api-group-list"), + path( + "api/v1/groups//export/", + views.api_group_export_stream, + name="api-group-export-stream", + ), path("api/v1/groups//", views.api_group_detail, name="api-group-detail"), path( "api/v1/groups//delivery/", diff --git a/forensics/views.py b/forensics/views.py index 1ac5312..4ad4675 100644 --- a/forensics/views.py +++ b/forensics/views.py @@ -1,6 +1,7 @@ from __future__ import annotations import ipaddress +from datetime import datetime from django.conf import settings from django.contrib import messages @@ -11,16 +12,26 @@ from django.core.files.uploadhandler import FileUploadHandler from django.core.paginator import Paginator from django.db.models import Count, Max, Min, Prefetch, Q +from django.db.models.fields.json import KeyTextTransform from django.db.models.functions import Length, Substr -from django.http import Http404, HttpRequest, HttpResponse, JsonResponse +from django.http import ( + Http404, + HttpRequest, + HttpResponse, + JsonResponse, + StreamingHttpResponse, +) from django.shortcuts import get_object_or_404, redirect, render from django.template.defaultfilters import slugify from django.urls import reverse from django.utils import timezone from django.views.decorators.csrf import csrf_exempt -from django.views.decorators.http import require_POST +from django.views.decorators.http import require_GET, require_POST from .analysis import ( + EXPORT_SENSITIVITY, + agent_event_row, + agent_source_row, agent_state_export_for_group, audit_files_for_group, color_index, @@ -43,10 +54,13 @@ DeliveryObservation, EpochStateTransition, NetworkObservation, + PersonalAccessToken, RecipientExpectation, StateDelta, UploadToken, ) +from .streaming import ExportSection, stream_ndjson +from .token_crypto import expiry_from_days UPLOAD_TOO_LARGE_ERROR = "audit log exceeds maximum upload size" AUDIT_FILE_EVENT_PAGE_SIZE = 100 @@ -55,6 +69,7 @@ GROUP_DETAIL_TAB_EVENT_LIMIT = 100 GROUP_PROJECTION_API_DEFAULT_LIMIT = 500 GROUP_PROJECTION_API_MAX_LIMIT = 5_000 +GROUP_EXPORT_SCHEMA_VERSION = "goggles-group-export/v1" FULL_DATA_AUDIT_MODE = "full_data" ERROR_SEVERITY_TOKENS = ( "error", @@ -2156,29 +2171,29 @@ def evidence_refs_payload(events) -> list[dict]: def delivery_identity_index(group: AuditGroup) -> dict[str, set[str]]: - index = {"account_refs": set(), "pubkeys_hex": set(), "engine_ids": set()} - events = ( - AuditEvent.objects.filter(group=group, parse_status=AuditEvent.STATUS_VALID) - .select_related("audit_file") - .only( - "account_ref", - "engine_id", - "context_source", - "audit_file__source_account_pubkey_hex", - ) + """Distinct delivery identities (account refs, engine ids, account pubkeys) + across a group's valid events. + + Computed with SQL ``DISTINCT`` pulls rather than a Python scan of every event: + the result is bounded by identity cardinality, not event count. This matters on + the streaming export's hot path, where materializing every event would break the + constant-memory guarantee. Account pubkeys come from two places — the backing + audit file and each event's ``context_source`` JSON — unioned here. + """ + valid_events = AuditEvent.objects.filter(group=group, parse_status=AuditEvent.STATUS_VALID) + + def distinct_values(queryset, field: str) -> set[str]: + return {value for value in queryset.values_list(field, flat=True).distinct() if value} + + context_pubkeys = valid_events.annotate( + context_account_pubkey_hex=KeyTextTransform("account_pubkey_hex", "context_source") ) - for event in events: - if event.account_ref: - index["account_refs"].add(event.account_ref) - if event.engine_id: - index["engine_ids"].add(event.engine_id) - if event.audit_file.source_account_pubkey_hex: - index["pubkeys_hex"].add(event.audit_file.source_account_pubkey_hex) - if isinstance(event.context_source, dict) and event.context_source.get( - "account_pubkey_hex" - ): - index["pubkeys_hex"].add(event.context_source["account_pubkey_hex"]) - return index + return { + "account_refs": distinct_values(valid_events, "account_ref"), + "engine_ids": distinct_values(valid_events, "engine_id"), + "pubkeys_hex": distinct_values(valid_events, "audit_file__source_account_pubkey_hex") + | distinct_values(context_pubkeys, "context_account_pubkey_hex"), + } def attach_delivery_matrices( @@ -2843,13 +2858,41 @@ def raw_text_download_filename(audit_file: AuditFile) -> str: return f"{stem or f'audit-file-{audit_file.pk}'}.jsonl" +def parse_access_token_expiry(raw: str | None) -> datetime | None: + """Resolve an optional token expiry from user input, failing closed. + + Distinguishes "not provided" (blank → no expiry, the documented default) from + "provided but invalid" (non-positive, too large, or unparseable → ``ValueError``). + A personal access token is a standing read credential, so a malformed expiry must + be rejected rather than silently minting a never-expiring one. + """ + value = (raw or "").strip() + if not value: + return None + try: + days = int(value) + except ValueError: + raise ValueError("Token expiry must be a whole number of days.") from None + return expiry_from_days(days) + + +def profile_context(request: HttpRequest, *, form=None, new_token: str | None = None) -> dict: + """Context for the profile page: the password form, the caller's own access + tokens, and — only immediately after minting — the one-time raw token.""" + return { + "form": form if form is not None else PasswordChangeForm(request.user), + "access_tokens": request.user.access_tokens.order_by("-created_at"), + "new_token": new_token, + } + + @login_required def profile(request: HttpRequest): - """Let a signed-in user change their own password. + """Let a signed-in user change their password and manage personal access tokens. - Usernames are intentionally not editable here: the form only ever touches - the password of ``request.user``, so a user can never modify another - account or rename their own. + Every action here only ever touches ``request.user``'s own credentials: the + password form binds to ``request.user`` and the token views are owner-scoped, so + a user can never read, mint, or revoke another account's tokens. """ if request.method == "POST": form = PasswordChangeForm(request.user, request.POST) @@ -2859,9 +2902,41 @@ def profile(request: HttpRequest): update_session_auth_hash(request, user) messages.success(request, "Your password has been updated.") return redirect("profile") - else: - form = PasswordChangeForm(request.user) - return render(request, "forensics/profile.html", {"form": form}) + return render(request, "forensics/profile.html", profile_context(request, form=form)) + return render(request, "forensics/profile.html", profile_context(request)) + + +@login_required +@require_POST +def create_access_token(request: HttpRequest): + name = (request.POST.get("name") or "").strip()[:120] or "Personal access token" + try: + expires_at = parse_access_token_expiry(request.POST.get("expires_in_days")) + except ValueError as error: + messages.error(request, str(error)) + return render(request, "forensics/profile.html", profile_context(request)) + raw_token, _token = PersonalAccessToken.issue(name, user=request.user, expires_at=expires_at) + messages.success( + request, + f"Created access token “{name}”. Copy it now — it will not be shown again.", + ) + # Render directly rather than redirecting so the one-time secret is never written + # to the session store, and mark it no-store so no cache retains the raw token. + response = render( + request, "forensics/profile.html", profile_context(request, new_token=raw_token) + ) + response["Cache-Control"] = "no-store" + return response + + +@login_required +@require_POST +def revoke_access_token(request: HttpRequest, pk: int): + token = get_object_or_404(PersonalAccessToken, pk=pk, user=request.user) + token.is_active = False + token.save(update_fields=["is_active"]) + messages.success(request, f"Revoked access token “{token.name}”.") + return redirect("profile") class MaxDumpSizeUploadHandler(FileUploadHandler): @@ -2959,12 +3034,118 @@ def api_audit_log_upload(request: HttpRequest, group_slug: str | None = None): return JsonResponse(body, status=response_status) -def authenticate_request(request: HttpRequest) -> UploadToken | None: - authorization = request.headers.get("Authorization", "") - scheme, _, value = authorization.partition(" ") +def bearer_value(authorization: str | None) -> str | None: + """Return the credential from an ``Authorization: Bearer `` header.""" + scheme, _, value = (authorization or "").partition(" ") if scheme.lower() != "bearer" or not value: return None - return UploadToken.authenticate(value.strip()) + return value.strip() + + +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. + + 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). + """ + if request.user.is_authenticated: + return True + token = PersonalAccessToken.authenticate(bearer_value(request.headers.get("Authorization"))) + if token is None: + return False + token.mark_used() + return True + + +@require_GET +def api_group_export_stream(request: HttpRequest, slug: str): + """Stream the complete forensic aggregate for one group as NDJSON. + + One request, one self-contained download: a ``manifest`` line, then every + ``source``, ``event``, and projection row, then a terminal ``eof``. Rows stream + through ``QuerySet.iterator()`` so peak memory is one chunk regardless of group + size. Derived aggregates (timeline/messages/actions/attribution) are excluded by + design — see authenticated-group-export.md — and reconstructable by the consumer + from the raw records. + """ + if not settings.GOGGLES_EXPORTS_ENABLED: + return JsonResponse({"error": "exports are temporarily disabled"}, status=503) + if not authenticate_reader(request): + return JsonResponse({"error": "authentication required"}, status=401) + + group = get_object_or_404(AuditGroup, slug=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 + # ``fork_resolution`` must be present) — and reimplementing the post-serialization + # severity/message-id filters that live in ``paginated_payloads``. A filter that + # only some sections applied is worse than none, so every section streams in full. + filters = default_api_filters() + identity_index = delivery_identity_index(group) + + sections = [ + ExportSection("source", audit_files_for_group(group), agent_source_row), + ExportSection( + "event", + valid_events_for_group(group, include_export_fields=True), + agent_event_row, + ), + ExportSection( + "delivery_artifact", + filtered_delivery_artifacts(group, filters), + lambda artifact: delivery_artifact_payload(artifact, identity_index), + ), + ExportSection( + "network_observation", + filtered_network_observations(group, filters), + network_observation_payload, + ), + ExportSection( + "convergence_run", + filtered_convergence_runs(group, filters), + convergence_run_payload, + ), + ExportSection( + "state_delta", + filtered_state_deltas(group, filters), + state_delta_payload, + ), + ExportSection( + "epoch_state_transition", + filtered_epoch_transitions(group, filters), + epoch_transition_payload, + ), + ExportSection( + "audit_data_mode_change", + filtered_audit_data_mode_changes(group, filters), + audit_data_mode_change_payload, + ), + ] + manifest = { + "schema_version": GROUP_EXPORT_SCHEMA_VERSION, + "generated_at": timezone.now().isoformat(), + "group": group_api_payload(group), + "classification": group_classification(group), + "sensitivity": EXPORT_SENSITIVITY, + # Derived from the sections themselves so the advertised list can never + # drift from what is actually streamed. + "sections": [section.record_type for section in sections], + } + + response = StreamingHttpResponse( + stream_ndjson(manifest, sections), + content_type="application/x-ndjson", + ) + response["Content-Disposition"] = f'attachment; filename="{group.slug}-export.ndjson"' + response["Cache-Control"] = "no-store" # sensitive payload + response["X-Accel-Buffering"] = "no" # harmless; guards a future nginx front + return response def install_max_dump_size_upload_handler(request: HttpRequest) -> None: From 209bf45e2eeda4911c37c45062102ed4c9eb9fd5 Mon Sep 17 00:00:00 2001 From: "Javier G. Montoya S" Date: Wed, 8 Jul 2026 09:56:07 -0400 Subject: [PATCH 2/2] fix: bound the group export's identity lookup and wire the server-side-cursor toggle --- config/settings.py | 6 ++++ docs/api-v1.md | 5 +++ docs/deployment.md | 4 +-- forensics/tests.py | 83 ++++++++++++++++++++++++++++++++++++++++++++++ forensics/views.py | 8 ++++- 5 files changed, 103 insertions(+), 3 deletions(-) diff --git a/config/settings.py b/config/settings.py index 00a52af..8dab1b9 100644 --- a/config/settings.py +++ b/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"}, diff --git a/docs/api-v1.md b/docs/api-v1.md index 7c3ce54..eb4e100 100644 --- a/docs/api-v1.md +++ b/docs/api-v1.md @@ -157,6 +157,11 @@ each section is read in its own transaction, so a record appended mid-export may 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/` diff --git a/docs/deployment.md b/docs/deployment.md index 83614fa..4e0516c 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -40,8 +40,8 @@ Capacity model — size the database for it: 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 `DISABLE_SERVER_SIDE_CURSORS=1` in that case (memory stays - bounded via the query `chunk_size`). + 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 diff --git a/forensics/tests.py b/forensics/tests.py index 8fbe2a6..d3ee21c 100644 --- a/forensics/tests.py +++ b/forensics/tests.py @@ -5389,6 +5389,32 @@ def test_ignores_invalid_events(self): {"account_refs": set(), "engine_ids": set(), "pubkeys_hex": set()}, ) + def test_distinct_queries_do_not_leak_auditevent_ordering(self): + # AuditEvent has Meta.ordering; if it leaks into SELECT DISTINCT the query + # dedupes per row (id is unique) and scans every event, defeating the + # bounded-by-cardinality guarantee. order_by() must keep it out. + group = AuditGroup.objects.create(name="G", slug="g", group_ref=GROUP_REF) + audit_file = AuditFile.objects.create( + file_sha256="a" * 64, + byte_size=1, + raw_text="{}", + validation_status=AuditFile.STATUS_VALID, + source_account_pubkey_hex="ff" * 32, + ) + audit_file.groups.add(group) + for seq in range(5): # many events, a single identity + self._valid_event( + group, audit_file, seq, account_ref=ACCOUNT_ALICE, engine_id=ENGINE_ALICE + ) + + with CaptureQueriesContext(connection) as queries: + index = delivery_identity_index(group) + + self.assertEqual(index["account_refs"], {ACCOUNT_ALICE}) + self.assertTrue(queries.captured_queries) + for query in queries.captured_queries: + self.assertNotIn("ORDER BY", query["sql"].upper()) + class GroupExportStreamTests(TestCase): """The authenticated NDJSON streaming export. See authenticated-group-export.md.""" @@ -5531,6 +5557,63 @@ def test_disabled_export_returns_503(self): response = self.client.get(self.url, HTTP_AUTHORIZATION=f"Bearer {raw_token}") self.assertEqual(response.status_code, 503) + def test_unknown_group_slug_returns_404(self): + raw_token, _token = PersonalAccessToken.issue("cgka", user=self.user) + response = self.client.get( + reverse("api-group-export-stream", kwargs={"slug": "not-a-real-slug"}), + HTTP_AUTHORIZATION=f"Bearer {raw_token}", + ) + self.assertEqual(response.status_code, 404) + + def _seed_prefetch_heavy_rows(self, count, evidence_event): + """Add `count` delivery artifacts and convergence runs, each with the related + rows their payloads iterate (expectations/observations, candidates/rules).""" + start = DeliveryArtifact.objects.filter(group=self.group).count() + for i in range(start, start + count): + artifact = DeliveryArtifact.objects.create( + group=self.group, + artifact_id=f"{i:064x}", + artifact_kind="application_message", + first_seen_ms=1_700_000_000_000 + i, + ) + RecipientExpectation.objects.create( + artifact=artifact, recipient_scope="group", evidence_event=evidence_event + ) + DeliveryObservation.objects.create( + artifact=artifact, engine_id=ENGINE_ALICE, latest_state="transport_received" + ) + run = ConvergenceRun.objects.create( + group=self.group, + run_id=f"run-{i:04x}", + engine_id=ENGINE_ALICE, + started_at_ms=1_700_000_200_000 + i, + ) + ConvergenceCandidate.objects.create( + run=run, branch_id=f"branch-{i}", fork_epoch=i, tip_epoch=i + 1 + ) + ConvergenceRuleEvaluation.objects.create(run=run, rule_name="highest_weight") + + def _export_query_count(self): + self.client.login(username="reader", password="correct horse battery staple") + with CaptureQueriesContext(connection) as queries: + response = self.client.get(self.url) + list(response.streaming_content) # force the generator to run + return len(queries.captured_queries) + + def test_export_query_count_is_flat_in_row_count(self): + # Adversarial finding #1 claimed iterator() drops prefetch_related → N+1. That + # is false on Django 6 when chunk_size is passed (it is): prefetch is honored, + # so multiplying the prefetch-heavy rows must not add queries. If a future edit + # drops chunk_size, this turns linear and fails. + event = AuditEvent.objects.filter( + group=self.group, parse_status=AuditEvent.STATUS_VALID + ).first() + self._seed_prefetch_heavy_rows(3, event) + few = self._export_query_count() + self._seed_prefetch_heavy_rows(9, event) # 12 total — 4× the artifacts and runs + many = self._export_query_count() + self.assertEqual(few, many) + def test_mid_stream_error_yields_error_line_and_no_eof_within_a_200(self): # Once the manifest is sent the status is committed at 200, so a later # failure can only be signalled in-band: an error line and no eof (M4). diff --git a/forensics/views.py b/forensics/views.py index 4ad4675..1af0983 100644 --- a/forensics/views.py +++ b/forensics/views.py @@ -2183,7 +2183,13 @@ def delivery_identity_index(group: AuditGroup) -> dict[str, set[str]]: valid_events = AuditEvent.objects.filter(group=group, parse_status=AuditEvent.STATUS_VALID) def distinct_values(queryset, field: str) -> set[str]: - return {value for value in queryset.values_list(field, flat=True).distinct() if value} + # order_by() clears AuditEvent's Meta.ordering; without it those columns leak + # into SELECT DISTINCT (id is unique), so it dedupes per row and scans every + # event instead of the handful of distinct identities — the exact O(all-events) + # cost this function exists to avoid. + return { + value for value in queryset.order_by().values_list(field, flat=True).distinct() if value + } context_pubkeys = valid_events.annotate( context_account_pubkey_hex=KeyTextTransform("account_pubkey_hex", "context_source")