P2: nearby units CRUD, unit meeting times, site settings - the first admin writes

Nearby rows bump verified_at on every save and deactivate rather than delete.
Unit edits are meeting fields only, gated by unit:write_own scoped to the unit
in the URL. Settings are a typed key registry falling back to code defaults;
find-a-unit now reads its source name and URL from it. link_url and contact
reject attribute-breakout characters and the nearby renderer escapes quotes.
Covered by tests/smoke_admin.py, 53 checks in-process.
This commit is contained in:
2026-09-04 14:10:29 -04:00
parent 8c8dab12e3
commit 32e1c67a6f
5 changed files with 609 additions and 14 deletions
+117 -3
View File
@@ -51,17 +51,23 @@ ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "").strip()
router = APIRouter(prefix="/api/admin", tags=["admin"])
def _auth(request, token, capability):
def _auth(request, token, capability, unit_id=None):
"""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.
unit_id scopes a membership capability to one unit: a pack leader holds
unit:write_own, but only within the pack. Global roles pass a scoped check
for every unit, and the break-glass token reaches everything - it exists
for when identity is broken, so it cannot depend on identity's scoping.
"""
person = auth.current_person(request)
if person:
if not identity.can(person, capability):
raise HTTPException(403, "your account does not have %s" % capability)
if not identity.can(person, capability, unit_id):
raise HTTPException(403, "your account does not have %s" % capability
+ (" for this unit" if unit_id else ""))
return person["email"]
if not ADMIN_TOKEN:
raise HTTPException(503, "admin API disabled: sign in, or set ADMIN_TOKEN")
@@ -179,3 +185,111 @@ def revoke_announcement(request: Request, announcement_id: str, x_admin_token: s
if not store.revoke_announcement(announcement_id):
raise HTTPException(409, "already revoked")
return store.get_announcement(announcement_id)
# ----------------------------------------------------------------------------
# Nearby units - the /find-a-unit directory, and the first writable directory
# data. The verified_at and deactivate-not-delete rules live in store.py so
# any future client inherits them.
# ----------------------------------------------------------------------------
@router.get("/nearby")
def list_nearby(request: Request, include_inactive: bool = False,
x_admin_token: str = Header(None)):
"""The editing view, so deactivated rows are reachable. nearby:write
rather than a read capability: the public page IS the read surface, and
this list exists only to be edited."""
_auth(request, x_admin_token, "nearby:write")
return {"nearby_units": store.list_nearby(include_inactive=include_inactive)}
@router.post("/nearby", status_code=201)
def create_nearby(request: Request, payload: dict = Body(...),
x_admin_token: str = Header(None)):
_auth(request, x_admin_token, "nearby:write")
try:
return store.create_nearby(payload)
except store.NearbyRejected as e:
raise _reject(e)
@router.patch("/nearby/{nearby_id}")
def update_nearby(request: Request, nearby_id: str, payload: dict = Body(...),
x_admin_token: str = Header(None)):
"""Partial update. Saving bumps verified_at to today unless the payload
carries an explicit date - see store.py for why saving is verifying."""
_auth(request, x_admin_token, "nearby:write")
try:
rec = store.update_nearby(nearby_id, payload)
except store.NearbyRejected as e:
raise _reject(e)
if not rec:
raise HTTPException(404, "no such nearby unit")
return rec
@router.delete("/nearby/{nearby_id}")
def deactivate_nearby(request: Request, nearby_id: str, x_admin_token: str = Header(None)):
"""Take a unit off the page. Sets active=0; never deletes the row."""
_auth(request, x_admin_token, "nearby:write")
if not store.get_nearby(nearby_id):
raise HTTPException(404, "no such nearby unit")
if not store.deactivate_nearby(nearby_id):
raise HTTPException(409, "already inactive")
return store.get_nearby(nearby_id)
# ----------------------------------------------------------------------------
# Our own units - meeting time and place only. unit:write_own is checked
# against the unit in the URL, so a pack leader edits the pack and not the
# troop; admins hold it globally and reach both.
# ----------------------------------------------------------------------------
@router.get("/units")
def list_units(request: Request, x_admin_token: str = Header(None)):
"""The units the caller may edit, which is what the screen this feeds
shows. A pack leader gets the pack; a global role or the break-glass
token gets everything. The answer IS the scope - no second filter for
the console to get wrong."""
_auth(request, x_admin_token, "unit:write_own")
units = identity.list_units(include_inactive=True)
person = auth.current_person(request)
if person:
units = [u for u in units if identity.can(person, "unit:write_own", u["id"])]
return {"units": units}
@router.patch("/units/{slug_or_id}")
def update_unit(request: Request, slug_or_id: str, payload: dict = Body(...),
x_admin_token: str = Header(None)):
unit = identity.get_unit(slug_or_id)
# Scope to the unit when it exists; an unknown slug still goes through
# _auth first, so probing paths answers 401/403 before it answers 404.
_auth(request, x_admin_token, "unit:write_own",
unit_id=unit["id"] if unit else None)
try:
return identity.update_unit_meets(slug_or_id, payload)
except identity.IdentityError as e:
raise HTTPException(e.status, e.detail)
# ----------------------------------------------------------------------------
# Site settings - the typed key registry lives in identity.SETTINGS_KEYS;
# unknown keys are rejected there, not silently stored.
# ----------------------------------------------------------------------------
@router.get("/settings")
def list_settings(request: Request, x_admin_token: str = Header(None)):
_auth(request, x_admin_token, "settings:write")
return {"settings": identity.all_settings()}
@router.put("/settings/{key}")
def put_setting(request: Request, key: str, payload: dict = Body(...),
x_admin_token: str = Header(None)):
"""Set one value, or clear it back to the code default with value: null."""
actor = _auth(request, x_admin_token, "settings:write")
try:
return identity.set_setting(key, payload.get("value"), actor=actor)
except identity.IdentityError as e:
raise HTTPException(e.status, e.detail)