T11.4: usage export endpoint (raw + sanitized CSV)

CR-019 / wave 11. Adds GET /api/usage/export, reusing _usage_query()'s
scoped filters (from/to/project_id/username/tool) and the same
require_user_manager gate as /api/usage/summary. Two modes:

- raw (default): real usernames, for internal admin use.
- sanitize=true: usernames replaced with an HMAC-SHA256 pseudonym
  (keyed with auth.SECRET_KEY, 16 hex chars, 'u_' prefix) so the file
  can be fed into PowerBI or another external reporting tool without
  carrying real identities. HMAC chosen over a plain hash since the
  username space is small enough to brute-force a bare digest.

Both modes emit at, username, project_id, tool, event as columns and
deliberately omit the detail JSON column in both modes to avoid an
identity leak riding along inside free-form detail data. Response is
returned with a Content-Disposition: attachment header and a filename
that encodes mode + date.

Verified locally against a throwaway SQLite DB with two seeded users
and four seeded UsageEvent rows:
  - raw export contains the real usernames and matches the summary
    endpoint's event_count for the same session state
  - sanitized export contains no real username or email anywhere in
    the file body, across two independently-issued export calls
  - the same real user maps to the same pseudonym both within one
    export and across the two separate export calls
  - raw and sanitized rows line up 1:1 on at/tool/event for the same
    filter set
  - the tool= filter narrows the export the same way it narrows the
    summary
  - a plain project_user is refused with 403; an unauthenticated
    request is refused with 401
  - full smoke test (27/27) and seed_demo.py both still pass
This commit is contained in:
2026-09-23 12:25:05 -07:00
parent 8e863ae7d0
commit 2de76d52e6

View File

@@ -9,6 +9,10 @@ Run (prod): gunicorn -k uvicorn.workers.UvicornWorker -b 127.0.0.1:8000 serve
Interactive docs: http://<host>/api/docs
"""
import base64
import csv
import hashlib
import hmac
import io
import logging
import os
import re
@@ -3509,6 +3513,59 @@ def usage_summary(
}
def _pseudonym(username: str) -> str:
"""A stable per-user id for the sanitized export — the SAME input always
produces the SAME output, within one export and across separate export
runs, so an external system (Power BI or similar) can still group and
trend "by user" without ever receiving a real name. HMAC rather than a
plain hash: a plain sha256(username) is trivially reversed against a
wordlist of the handful of usernames this app actually has; keying it
with AUTH_SECRET_KEY (already a real secret, already required in
production — see auth.py) means recovering a username from its
pseudonym requires the signing key, not just guessing."""
digest = hmac.new(auth.SECRET_KEY.encode(), username.encode(), hashlib.sha256).hexdigest()
return "u_" + digest[:16]
@app.get("/api/usage/export")
def usage_export(
date_from: Optional[str] = Query(None, alias="from"),
date_to: Optional[str] = Query(None, alias="to"),
project_id: Optional[str] = Query(None),
username: Optional[str] = Query(None),
tool: Optional[str] = Query(None),
sanitize: bool = Query(False),
caller: models.User = Depends(require_user_manager),
db: Session = Depends(get_db),
):
"""CR-019. Same gate, same filters, same underlying row set as
usage_summary() (_usage_query) — the export can never show a different
slice of data than what the console counted for the same filters.
sanitize=true replaces `username` with a stable pseudonym (_pseudonym)
and — deliberately — the `detail` column is not exported in EITHER mode.
Every event this app writes today (login, page_open) leaves `detail`
empty, so this costs nothing now, but it also means a future event type
that DOES populate `detail` can't accidentally leak a real name into a
sanitized file through a column nobody thought to scrub. If `detail`
is ever needed in the export, it has to be sanitized explicitly, not
assumed safe because the rest of the row was."""
rows = db.scalars(_usage_query(db, caller, date_from, date_to, project_id, username, tool)).all()
buf = io.StringIO()
w = csv.writer(buf)
w.writerow(["at", "username", "project_id", "tool", "event"])
for e in rows:
who = _pseudonym(e.username) if sanitize else e.username
w.writerow([models._iso(e.at), who, e.project_id or "", e.tool, e.event])
filename = "usage_export_%s_%s.csv" % (
"sanitized" if sanitize else "raw", datetime.now(timezone.utc).strftime("%Y%m%d"),
)
return Response(
content=buf.getvalue(), media_type="text/csv",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
@app.get("/api/comments")
def list_comments(
source: Optional[str] = Query(None),