A key is the person who minted it, narrowed to the scopes they chose. Only the sha256 is stored; the full key is returned once. Scopes must be a subset of the owner's capabilities at mint time and are enforced again at use time inside identity.can(), the one place that decides, so a key never outlives its owner's demotion and disabling a person disables their keys with no separate flag. A key cannot carry apikeys:own or the owner powers, so it cannot mint keys. Revoked rows stay; a foreign key id is 404, never 403. Bearer keys are honoured ONLY on /api/admin. The rest of the site reads sessions alone, so a scoped key never widens into a browser identity. X-Admin-Token remains break glass and, having no person, cannot own a key. /api/docs is generated from the router on every request: path, methods and docstring from the route objects, and the capability read out of each handler's own _auth() call so it cannot drift from the check. Gated on a new api:docs capability (leader and above). GET /api/admin/whoami answers who the API thinks you are and what you can do. Tests: smoke_identity 66 -> 92, smoke_admin 53 -> 58 (registry has a capability for every route, docs page renders every route). Driven end to end on a throwaway site with a DB copy: mint, whoami via key, scoped 200s and a 403 that names the narrowing, key-mints-key 403, garbage key 401, admin token on /keys 403, key on /account is not a session, revoke then 401, second revoke 409.
179 lines
8.7 KiB
Python
179 lines
8.7 KiB
Python
"""
|
|
smoke_admin.py - the P2 admin write paths against a throwaway DB.
|
|
|
|
Runs in-process with no container, no network and no dependencies beyond the
|
|
stdlib, so it can be run before anything is committed.
|
|
|
|
STORE_DB=/tmp/x.db python3 tests/smoke_admin.py
|
|
|
|
It covers the rules that are expensive to get wrong and invisible when they
|
|
are: nearby rows bump verified_at on every save and deactivate rather than
|
|
delete, unit edits are meeting fields only and unit-scoped, and settings
|
|
accept only registered keys and fall back to the code default on anything
|
|
absent or mangled.
|
|
"""
|
|
|
|
import os, sys, tempfile
|
|
|
|
DB = os.environ.get("STORE_DB") or os.path.join(tempfile.mkdtemp(), "smoke.db")
|
|
os.environ["STORE_DB"] = DB
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "app"))
|
|
|
|
import identity as I
|
|
import nearby as N
|
|
import store as S
|
|
|
|
PASS = FAIL = 0
|
|
|
|
|
|
def check(label, cond):
|
|
global PASS, FAIL
|
|
if cond:
|
|
PASS += 1
|
|
print(" ok %s" % label)
|
|
else:
|
|
FAIL += 1
|
|
print(" FAIL %s" % label)
|
|
|
|
|
|
def raises(label, status, exc, fn, *a, **kw):
|
|
try:
|
|
fn(*a, **kw)
|
|
except exc as e:
|
|
check("%s -> %d" % (label, status), e.status == status)
|
|
return
|
|
except Exception as e:
|
|
check("%s -> %d (got %r)" % (label, status, e), False)
|
|
return
|
|
check("%s -> %d (no error raised)" % (label, status), False)
|
|
|
|
|
|
print("db: %s\n" % DB)
|
|
S.init()
|
|
I.init()
|
|
|
|
print("nearby: writes are validated")
|
|
raises("bad unit_type", 422, S.NearbyRejected, S.create_nearby,
|
|
{"unit_type": "trop", "unit_number": "1"})
|
|
raises("missing unit_number", 422, S.NearbyRejected, S.create_nearby,
|
|
{"unit_type": "pack"})
|
|
raises("bad serves", 422, S.NearbyRejected, S.create_nearby,
|
|
{"unit_type": "pack", "unit_number": "1", "serves": "everyone"})
|
|
raises("http link", 422, S.NearbyRejected, S.create_nearby,
|
|
{"unit_type": "pack", "unit_number": "1", "link_url": "http://x.test"})
|
|
raises("link with an attribute breakout", 422, S.NearbyRejected, S.create_nearby,
|
|
{"unit_type": "pack", "unit_number": "1",
|
|
"link_url": 'https://x.test" onmouseover="alert(1)'})
|
|
raises("contact without @", 422, S.NearbyRejected, S.create_nearby,
|
|
{"unit_type": "pack", "unit_number": "1", "contact": "call Steve"})
|
|
raises("contact with an attribute breakout", 422, S.NearbyRejected, S.create_nearby,
|
|
{"unit_type": "pack", "unit_number": "1",
|
|
"contact": 'a@b.test" onfocus="alert(1)'})
|
|
check("renderer escapes quotes as a second line", N._esc('a"b') == "a"b")
|
|
raises("unknown field rejected, not dropped", 422, S.NearbyRejected, S.create_nearby,
|
|
{"unit_type": "pack", "unit_number": "1", "unit_typo": "pack"})
|
|
raises("verified_at must be a date", 422, S.NearbyRejected, S.create_nearby,
|
|
{"unit_type": "pack", "unit_number": "1", "verified_at": "yesterday"})
|
|
raises("sort_order must be an int", 422, S.NearbyRejected, S.create_nearby,
|
|
{"unit_type": "pack", "unit_number": "1", "sort_order": "soon"})
|
|
|
|
print("\nnearby: create, update, verified_at")
|
|
a = S.create_nearby({"unit_type": "pack", "unit_number": "244", "town": "Harleysville",
|
|
"area": "North Penn", "serves": "family"})
|
|
check("row created", a and a["unit_number"] == "244")
|
|
check("active and sort_order defaulted", a["active"] == 1 and a["sort_order"] == 100)
|
|
check("verified_at defaults to today", a["verified_at"] == S._today())
|
|
|
|
b = S.create_nearby({"unit_type": "troop", "unit_number": "27", "area": "North Penn",
|
|
"verified_at": "2026-08-01"})
|
|
check("explicit verified_at kept", b["verified_at"] == "2026-08-01")
|
|
|
|
b2 = S.update_nearby(b["id"], {"town": "Lansdale"})
|
|
check("partial update lands", b2["town"] == "Lansdale" and b2["unit_number"] == "27")
|
|
check("saving bumps verified_at", b2["verified_at"] == S._today())
|
|
b3 = S.update_nearby(b["id"], {"notes": "meets in the annexe", "verified_at": "2026-08-15"})
|
|
check("explicit verified_at wins on update", b3["verified_at"] == "2026-08-15")
|
|
check("update of missing row is None", S.update_nearby("nope", {"town": "x"}) is None)
|
|
raises("empty update", 422, S.NearbyRejected, S.update_nearby, b["id"], {})
|
|
|
|
print("\nnearby: deactivate, never delete")
|
|
check("deactivate", S.deactivate_nearby(a["id"]))
|
|
check("second deactivate is a no-op", not S.deactivate_nearby(a["id"]))
|
|
check("row survives", S.get_nearby(a["id"])["active"] == 0)
|
|
check("default list hides it", all(r["id"] != a["id"] for r in S.list_nearby()))
|
|
check("include_inactive shows it",
|
|
any(r["id"] == a["id"] for r in S.list_nearby(include_inactive=True)))
|
|
back = S.update_nearby(a["id"], {"active": 1})
|
|
check("reactivate via update", back["active"] == 1)
|
|
groups = N.listing()
|
|
check("public page groups the live rows",
|
|
len(groups) == 1 and groups[0][0] == "North Penn" and len(groups[0][1]) == 2)
|
|
|
|
print("\nunits: meeting fields only, structured")
|
|
pack = I.get_unit("pack73")
|
|
check("seed row present", pack and pack["meets_weekday"] == 2)
|
|
u = I.update_unit_meets("pack73", {"meets_time": "18:30"})
|
|
check("time updated", u["meets_time"] == "18:30")
|
|
u = I.update_unit_meets(pack["id"], {"meets_weekday": 4, "meets_at": None})
|
|
check("update by id, null allowed", u["meets_weekday"] == 4 and u["meets_at"] is None)
|
|
check("other fields untouched", u["display_name"] == pack["display_name"])
|
|
raises("weekday 8", 422, I.IdentityError, I.update_unit_meets, "pack73", {"meets_weekday": 8})
|
|
raises("weekday as bool", 422, I.IdentityError, I.update_unit_meets, "pack73", {"meets_weekday": True})
|
|
raises("display time string", 422, I.IdentityError, I.update_unit_meets, "pack73", {"meets_time": "7pm"})
|
|
raises("renaming is not a meeting edit", 422, I.IdentityError, I.update_unit_meets,
|
|
"pack73", {"display_name": "Pack 99"})
|
|
raises("unknown unit", 404, I.IdentityError, I.update_unit_meets, "pack99", {"meets_time": "18:00"})
|
|
raises("nothing to update", 422, I.IdentityError, I.update_unit_meets, "pack73", {})
|
|
|
|
print("\nsettings: typed keys, default fallback")
|
|
check("default when unset",
|
|
I.get_setting("nearby_source_name") == "Continental District unit list")
|
|
s = I.set_setting("nearby_source_url", "https://example.test/units", actor="a@example.test")
|
|
check("set and read back", I.get_setting("nearby_source_url") == "https://example.test/units")
|
|
check("set_setting reports itself", s["is_set"] and s["updated_by"] == "a@example.test")
|
|
raises("unknown key", 422, I.IdentityError, I.set_setting, "nearby_src_url", "https://x.test")
|
|
raises("blank value", 422, I.IdentityError, I.set_setting, "nearby_source_name", " ")
|
|
raises("http url", 422, I.IdentityError, I.set_setting, "nearby_source_url", "http://x.test")
|
|
cleared = I.set_setting("nearby_source_url", None)
|
|
check("null clears to default", not cleared["is_set"]
|
|
and I.get_setting("nearby_source_url") == "https://tinyurl.com/ContinentalScouts")
|
|
|
|
con = I.connect()
|
|
con.execute("INSERT INTO settings (key, value, updated_at) VALUES (?,?,?)",
|
|
("nearby_source_url", "ftp://mangled", I._now()))
|
|
con.commit(); con.close()
|
|
check("mangled row falls back to default",
|
|
I.get_setting("nearby_source_url") == "https://tinyurl.com/ContinentalScouts")
|
|
check("all_settings covers the registry",
|
|
{s["key"] for s in I.all_settings()} == set(I.SETTINGS_KEYS))
|
|
|
|
print("\ncapabilities behind the new routes")
|
|
leader = {"memberships": [{"unit_id": "u1", "role": "leader"}]}
|
|
member = {"memberships": [{"unit_id": "u1", "role": "member"}]}
|
|
admin = {"global_role": "admin", "memberships": []}
|
|
check("leader can edit nearby", I.can(leader, "nearby:write"))
|
|
check("member cannot", not I.can(member, "nearby:write"))
|
|
check("leader edits own unit", I.can(leader, "unit:write_own", "u1"))
|
|
check("but not the other one", not I.can(leader, "unit:write_own", "u2"))
|
|
check("leader cannot touch settings", not I.can(leader, "settings:write"))
|
|
check("admin spans units", I.can(admin, "unit:write_own", "u1") and I.can(admin, "unit:write_own", "u2"))
|
|
check("admin holds settings:write", I.can(admin, "settings:write"))
|
|
|
|
print("\napi docs registry")
|
|
import admin_api as A
|
|
reg = A.describe_routes()
|
|
check("registry is non-trivial", len(reg) >= 18)
|
|
missing = [d for d in reg if not d["capability"] and d["path"] != "/api/admin/whoami"]
|
|
check("every route's capability is read from its own _auth call (except whoami): %s"
|
|
% ", ".join(d["path"] for d in missing), not missing)
|
|
check("key routes gate on apikeys:own", all(d["capability"] == "apikeys:own" for d in reg if "/keys" in d["path"]))
|
|
check("docs carry the handler docstring", all(d["doc"] for d in reg if "/keys" in d["path"]))
|
|
class _R:
|
|
headers = {"authorization": ""}; cookies = {}
|
|
os.environ["ADMIN_TOKEN"] = "t"; A.ADMIN_TOKEN = "t"
|
|
page = A.api_docs(_R(), "t").body.decode()
|
|
check("docs page renders every route", page.count("<tr><td><code>") == len(reg))
|
|
|
|
print("\n%d passed, %d failed" % (PASS, FAIL))
|
|
sys.exit(1 if FAIL else 0)
|