thethreemagi/scout-website
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

P1: gate members documents per unit, session auth on the admin API

Browse Source 
documents.visible() now takes the viewer's unit set. A members document is
served and listed only to a signed-in member of the matching unit; 'both'
reaches any member; owner and admin reach everything including unit types
added later. Everyone else gets 404, never 403.

The gate moved INSIDE find(), so there is one path from a slug to a file and
no route can forget to check. listed() and visible() stay separate functions.

admin_api takes a session first and falls back to X-Admin-Token as break
glass. Still fails closed: no session and no ADMIN_TOKEN is 503. Capability,
not role, decides per route. announcements.created_by now comes from the
session and ignores any value in the request body.

tests/smoke_documents.py, 24 checks, including the invariant that the index
can never list something serving would refuse.
This commit is contained in:
thethreemagi
2026-09-04 12:09:19 -04:00
parent 8ddcccf293
commit 8c8dab12e3
5 changed files with 262 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/admin_api.py
+56 -25
View File
@@ -19,17 +19,31 @@ shows, it comes down. Writing it is the entire feature. The caps that keep
the banner from becoming a mess live in store.py, not here, so the panel and
any future client inherit them rather than reimplementing them.
Auth: every route requires the X-Admin-Token header to match ADMIN_TOKEN.
If ADMIN_TOKEN is unset the whole router returns 503 - it FAILS CLOSED. These
endpoints expose parent names, emails and phone numbers for minors' families,
so an unconfigured deployment must not serve them.
Auth, two ways, session first.
A signed-in person holding the required capability is allowed. Otherwise the
X-Admin-Token header must match ADMIN_TOKEN, which is retained as BREAK GLASS:
if the identity layer is broken there has to be a way in that does not depend
on the identity layer.
Still FAILS CLOSED. With no session and no ADMIN_TOKEN set, every route returns
503. These endpoints expose parent names, emails and phone numbers for minors'
families, so an unconfigured deployment must not serve them. Adding sessions
widened who may read; it did not soften what happens when nothing is
configured.
Capability, not role, decides. Lead routes need leads:read, announcement routes
need announcements:write, and both are answered by identity.can() against the
one CAPS dictionary - never by a role comparison here.
"""
import hmac
import os
from fastapi import APIRouter, Body, Header, HTTPException, Query
from fastapi import APIRouter, Body, Header, HTTPException, Query, Request
import auth
import identity
import store
ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "").strip()
@@ -37,30 +51,42 @@ ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "").strip()
router = APIRouter(prefix="/api/admin", tags=["admin"])
def _auth(token):
def _auth(request, token, capability):
"""Authorise, and return a label naming who acted, for created_by.
Order matters: session first, so a normal signed-in leader never depends on
the shared token, and the token stays a fallback rather than the everyday
path.
"""
person = auth.current_person(request)
if person:
if not identity.can(person, capability):
raise HTTPException(403, "your account does not have %s" % capability)
return person["email"]
if not ADMIN_TOKEN:
raise HTTPException(503, "admin API disabled: ADMIN_TOKEN is not set")
raise HTTPException(503, "admin API disabled: sign in, or set ADMIN_TOKEN")
if not token or not hmac.compare_digest(token, ADMIN_TOKEN):
raise HTTPException(401, "bad or missing X-Admin-Token")
raise HTTPException(401, "sign in, or send a valid X-Admin-Token")
return "admin-token"
@router.get("/summary")
def get_summary(x_admin_token: str = Header(None)):
_auth(x_admin_token)
def get_summary(request: Request, x_admin_token: str = Header(None)):
_auth(request, x_admin_token, "leads:read")
return store.summary()
@router.get("/leads")
def get_leads(since: str = None, q: str = None,
def get_leads(request: Request, since: str = None, q: str = None,
limit: int = Query(100, ge=1, le=500), offset: int = 0,
x_admin_token: str = Header(None)):
_auth(x_admin_token)
_auth(request, x_admin_token, "leads:read")
return {"leads": store.list_leads(since=since, q=q, limit=limit, offset=offset)}
@router.get("/leads/{record_id}")
def get_one(record_id: str, x_admin_token: str = Header(None)):
_auth(x_admin_token)
def get_one(request: Request, record_id: str, x_admin_token: str = Header(None)):
_auth(request, x_admin_token, "leads:read")
rec = store.get_lead(record_id)
if not rec:
raise HTTPException(404, "no such lead")
@@ -68,16 +94,16 @@ def get_one(record_id: str, x_admin_token: str = Header(None)):
@router.get("/mirrors/failed")
def failed(target: str = "google_sheet", x_admin_token: str = Header(None)):
_auth(x_admin_token)
def failed(request: Request, target: str = "google_sheet", x_admin_token: str = Header(None)):
_auth(request, x_admin_token, "leads:read")
return {"target": target, "leads": store.failed_mirror_records(target)}
@router.post("/mirrors/retry")
def retry(target: str = "google_sheet", x_admin_token: str = Header(None)):
def retry(request: Request, target: str = "google_sheet", x_admin_token: str = Header(None)):
"""Replay leads whose copy to an external target failed. Idempotent-ish:
a lead already marked ok is never retried."""
_auth(x_admin_token)
_auth(request, x_admin_token, "leads:read")
if target != "google_sheet":
raise HTTPException(422, "only google_sheet retry is implemented")
import app as main_app
@@ -106,25 +132,25 @@ def _reject(e):
@router.get("/announcements")
def list_announcements(include_expired: bool = False,
def list_announcements(request: Request, include_expired: bool = False,
limit: int = Query(100, ge=1, le=500),
x_admin_token: str = Header(None)):
"""Every announcement with its computed state: live, scheduled, expired,
revoked, or over_cap. State is returned rather than left to be inferred
from what the homepage happens to render."""
_auth(x_admin_token)
_auth(request, x_admin_token, "announcements:write")
return {"announcements": store.list_announcements(
include_expired=include_expired, limit=limit)}
@router.post("/announcements", status_code=201)
def create_announcement(payload: dict = Body(...), x_admin_token: str = Header(None)):
def create_announcement(request: Request, payload: dict = Body(...), x_admin_token: str = Header(None)):
"""Post a notice. ends_at is required.
422 if the message is over the character cap or the window is invalid.
409 if the live cap is already reached, listing what is up so you can
decide what to revoke."""
_auth(x_admin_token)
_actor = _auth(request, x_admin_token, "announcements:write")
try:
return store.create_announcement(
message=payload.get("message"),
@@ -133,16 +159,21 @@ def create_announcement(payload: dict = Body(...), x_admin_token: str = Header(N
level=payload.get("level", "info"),
link_url=payload.get("link_url"),
link_text=payload.get("link_text"),
created_by=payload.get("created_by"),
# From the session, never from the body. A caller must not be able
# to attribute a public notice to somebody else. Stored as the
# email rather than the uuid: the column is display-facing, people
# are never hard deleted, and a uuid in a banner audit trail helps
# nobody read it.
created_by=_actor,
)
except store.AnnouncementRejected as e:
raise _reject(e)
@router.delete("/announcements/{announcement_id}")
def revoke_announcement(announcement_id: str, x_admin_token: str = Header(None)):
def revoke_announcement(request: Request, announcement_id: str, x_admin_token: str = Header(None)):
"""Take one down early. Sets revoked_at; never deletes the row."""
_auth(x_admin_token)
_auth(request, x_admin_token, "announcements:write")
if not store.get_announcement(announcement_id):
raise HTTPException(404, "no such announcement")
if not store.revoke_announcement(announcement_id):