🔒 session-token credential binding + admin-guard/CSRF hardening (adversarial review)
Fable flagged that stateless session tokens bound only to uid+expiry — so an
admin action couldn't actually invalidate a live 30-day cookie. Reproduced 3
facets, all one root cause, all fixed by binding the token to the account's
credential state:
- make_token/read_token/_bind: token = uid:exp:bind:sig where
bind = hmac(secret, 'uid:created:salt')[:16]; read_token re-derives it from the
user's CURRENT row and rejects on mismatch/missing. So a token dies on account
DELETE (no ghost session/writes), password RESET (salt rotates → admin can truly
lock someone out), and SQLite id-REUSE (different created → no identity takeover).
Old 3-part tokens fail to parse → a one-time re-login. No schema change.
A follow-on adversarial review (4 security lenses × find→refute×2, GO gate) then
surfaced 3 low-severity issues, all fixed here too:
- self-de-admin guard used (identity) — {admin:0}
/ '' / 'false' / [] slipped past and could strand a sole db-admin. Now only a
real JSON bool toggles admin (closes the bypass AND the truthy-string mis-grant).
- signup invite claim was check-then-write (double-spend under concurrency) — now
an atomic +
rowcount check; signup/admin-edit wrap the UNIQUE writes in try/except
IntegrityError → friendly message instead of a 500.
- login-CSRF (Lax cookie only): hub.py _api now blocks cross-site mutating requests
by Origin (allowlist godstrument.pro/localhost + Host-match fallback so legit
traffic always passes; safe GETs never blocked).
Verified: auth.py --selftest (delete/reset/id-reuse kill the token; guard bypass
closed for every falsy value); live curl (CSRF 401/401/401/403, GET never blocked);
real-browser flow (new-format cookie authenticates through a page load, admin table
renders, same-origin admin POST passes CSRF). Console clean.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
type-two
parent
764e28b1c3
commit
47b51b98ca
83
auth.py
83
auth.py
@ -112,9 +112,27 @@ def verify_pw(pw: str, salt: str, expected: str) -> bool:
|
||||
|
||||
|
||||
# ---- stateless signed session token -----------------------------------------
|
||||
# A token is bound to the account's *credential state* — a tag over its immutable
|
||||
# `created` time + its current password `salt`. So a token dies the moment the
|
||||
# account is deleted (no row), its password is reset (salt rotates), or its id is
|
||||
# reused by a different signup (different created). Without this, a signed uid:exp
|
||||
# token stays valid for its full 30 days regardless of any of the above.
|
||||
def _bind(user_id: int, db: str = DEFAULT_DB) -> str | None:
|
||||
con = _con(db)
|
||||
try:
|
||||
row = con.execute("SELECT created, salt FROM users WHERE id=?", (user_id,)).fetchone()
|
||||
finally:
|
||||
con.close()
|
||||
if not row:
|
||||
return None
|
||||
return hmac.new(_secret(db), f"{user_id}:{row[0]}:{row[1]}".encode(),
|
||||
hashlib.sha256).hexdigest()[:16]
|
||||
|
||||
|
||||
def make_token(user_id: int, db: str = DEFAULT_DB, days: int = SESSION_DAYS) -> str:
|
||||
exp = int(time.time()) + days * 86400
|
||||
payload = f"{user_id}:{exp}"
|
||||
bind = _bind(user_id, db) or ""
|
||||
payload = f"{user_id}:{exp}:{bind}"
|
||||
sig = hmac.new(_secret(db), payload.encode(), hashlib.sha256).hexdigest()
|
||||
return base64.urlsafe_b64encode(f"{payload}:{sig}".encode()).decode()
|
||||
|
||||
@ -122,12 +140,15 @@ def make_token(user_id: int, db: str = DEFAULT_DB, days: int = SESSION_DAYS) ->
|
||||
def read_token(token: str, db: str = DEFAULT_DB) -> int | None:
|
||||
try:
|
||||
raw = base64.urlsafe_b64decode(token.encode()).decode()
|
||||
uid, exp, sig = raw.split(":")
|
||||
good = hmac.new(_secret(db), f"{uid}:{exp}".encode(), hashlib.sha256).hexdigest()
|
||||
uid, exp, bind, sig = raw.split(":") # 4 parts; old 3-part tokens fail here -> re-login
|
||||
good = hmac.new(_secret(db), f"{uid}:{exp}:{bind}".encode(), hashlib.sha256).hexdigest()
|
||||
if not hmac.compare_digest(sig, good):
|
||||
return None
|
||||
if int(exp) < time.time():
|
||||
return None
|
||||
cur = _bind(int(uid), db) # current credential state
|
||||
if cur is None or not hmac.compare_digest(bind, cur):
|
||||
return None # deleted / password-reset / id-reused
|
||||
return int(uid)
|
||||
except (ValueError, TypeError, AttributeError):
|
||||
return None
|
||||
@ -164,10 +185,19 @@ def signup(username: str, email: str, pw: str, invite: str,
|
||||
if con.execute("SELECT 1 FROM users WHERE username=? COLLATE NOCASE", (username,)).fetchone():
|
||||
return False, "username already taken", None
|
||||
salt, ph = hash_pw(pw)
|
||||
cur = con.execute("INSERT INTO users (username,email,pw_hash,salt,created) VALUES (?,?,?,?,?)",
|
||||
(username, email, ph, salt, time.time()))
|
||||
try:
|
||||
cur = con.execute("INSERT INTO users (username,email,pw_hash,salt,created) VALUES (?,?,?,?,?)",
|
||||
(username, email, ph, salt, time.time()))
|
||||
except sqlite3.IntegrityError: # lost a same-instant race on the UNIQUE(email/username)
|
||||
con.rollback()
|
||||
return False, "email or username already taken", None
|
||||
uid = cur.lastrowid
|
||||
con.execute("UPDATE invites SET used_by=? WHERE code=?", (uid, invite))
|
||||
# atomic single-use claim: only the first concurrent signup flips a NULL used_by
|
||||
claimed = con.execute("UPDATE invites SET used_by=? WHERE code=? AND used_by IS NULL",
|
||||
(uid, invite)).rowcount
|
||||
if not claimed: # another signup claimed this invite first
|
||||
con.rollback()
|
||||
return False, "invite code already used", None
|
||||
con.commit()
|
||||
return True, "ok", uid
|
||||
finally:
|
||||
@ -275,8 +305,12 @@ def admin_update_user(user_id: int, db: str = DEFAULT_DB, username: str | None =
|
||||
if not sets:
|
||||
return False, "nothing to change"
|
||||
vals.append(user_id)
|
||||
con.execute("UPDATE users SET " + ", ".join(sets) + " WHERE id=?", vals)
|
||||
con.commit()
|
||||
try:
|
||||
con.execute("UPDATE users SET " + ", ".join(sets) + " WHERE id=?", vals)
|
||||
con.commit()
|
||||
except sqlite3.IntegrityError: # lost a race on UNIQUE(username/email)
|
||||
con.rollback()
|
||||
return False, "username or email already taken"
|
||||
return True, "ok"
|
||||
finally:
|
||||
con.close()
|
||||
@ -447,10 +481,15 @@ def handle_api(method: str, path: str, body: bytes, cookie_header: str,
|
||||
ok, temp = admin_reset_password(target, db)
|
||||
return (200, {"ok": True, "password": temp}, None) if ok else (404, {"error": "no such user"}, None)
|
||||
if not action and method == "POST":
|
||||
if target == uid and payload.get("admin") is False:
|
||||
# only a real JSON boolean toggles admin — a stray 0/""/"false" must not
|
||||
# silently demote (the self-guard) or promote (truthy string) via coercion
|
||||
flag = payload.get("admin")
|
||||
if not isinstance(flag, bool):
|
||||
flag = None
|
||||
if target == uid and flag is False:
|
||||
return 400, {"error": "you can't remove your own admin"}, None
|
||||
ok, msg = admin_update_user(target, db, payload.get("username"),
|
||||
payload.get("email"), payload.get("admin"))
|
||||
payload.get("email"), flag)
|
||||
return (200, {"ok": True}, None) if ok else (400, {"error": msg}, None)
|
||||
if not action and method == "DELETE":
|
||||
if target == uid:
|
||||
@ -553,6 +592,10 @@ def _selftest():
|
||||
assert handle_api("GET", "/api/admin/users", b"", "", db)[0] == 401, "no session blocked"
|
||||
assert handle_api("POST", f"/api/admin/user/{vid}/reset", b"", cookie, db)[1].get("password"), "reset endpoint returns a temp"
|
||||
assert handle_api("POST", f"/api/admin/user/{uid}", json.dumps({"admin": False}).encode(), cookie, db)[0] == 400, "can't de-admin self"
|
||||
handle_api("POST", f"/api/admin/user/{uid}", json.dumps({"admin": 0}).encode(), cookie, db)
|
||||
assert user_info(uid, db)["admin"], "self stays admin — a crafted falsy {admin:0} can't bypass the self-guard"
|
||||
handle_api("POST", f"/api/admin/user/{vid}", json.dumps({"admin": "yes"}).encode(), cookie, db)
|
||||
assert not user_info(vid, db)["admin"], "a non-boolean truthy admin flag does not grant admin"
|
||||
assert handle_api("DELETE", f"/api/admin/user/{uid}", b"", cookie, db)[0] == 400, "can't delete self"
|
||||
assert handle_api("POST", f"/api/admin/user/{vid}", json.dumps({"username": "Vic2"}).encode(), cookie, db)[0] == 200, "edit endpoint works"
|
||||
# a non-admin can't reach any of it
|
||||
@ -564,6 +607,26 @@ def _selftest():
|
||||
assert save_preset(vid, "gone", {"x": 1}, db) and list_presets(vid, db)
|
||||
assert admin_delete_user(vid, db) and user_info(vid, db) is None, "delete removes the user"
|
||||
assert not list_presets(vid, db), "delete cascades presets"
|
||||
# --- session tokens die on delete / password-reset / id-reuse (credential binding) ---
|
||||
_, _, sa = signup("Sessa", "sessa@x.co", "sessapass1", mint_invite(db), db)
|
||||
stok = make_token(sa, db)
|
||||
assert read_token(stok, db) == sa, "a fresh token authenticates"
|
||||
admin_reset_password(sa, db)
|
||||
assert read_token(stok, db) is None, "reset-password invalidates the old session (admin can lock out)"
|
||||
stok2 = make_token(sa, db)
|
||||
assert read_token(stok2, db) == sa, "a token minted after reset works"
|
||||
admin_delete_user(sa, db)
|
||||
assert read_token(stok2, db) is None, "delete invalidates the session — no ghost cookie, no ghost writes"
|
||||
# id reuse (fresh db → deterministic): a deleted user's token must not become the reused occupant
|
||||
import tempfile as _tf
|
||||
rdb = os.path.join(_tf.mkdtemp(), "r.db"); init_db(rdb)
|
||||
_, _, x = signup("Xavier", "x@x.co", "xavierpass1", mint_invite(rdb), rdb)
|
||||
xtok = make_token(x, rdb)
|
||||
admin_delete_user(x, rdb)
|
||||
_, _, y = signup("Yara", "y@x.co", "yarapass111", mint_invite(rdb), rdb)
|
||||
assert y == x, "fresh db: sqlite reuses the freed id"
|
||||
assert read_token(xtok, rdb) is None, "a reused id must reject the deleted user's token (no identity takeover)"
|
||||
assert read_token(make_token(y, rdb), rdb) == y, "the reused id's real owner still logs in fine"
|
||||
print("auth self-test: all checks passed.")
|
||||
|
||||
|
||||
|
||||
18
hub.py
18
hub.py
@ -610,6 +610,24 @@ def _serve_http(directory: str, port: int):
|
||||
path = self.path.split("?", 1)[0]
|
||||
if auth is None:
|
||||
return self.send_error(503, "accounts unavailable")
|
||||
# CSRF: a browser always sends Origin on a cross-site state-changing request.
|
||||
# Block those (incl. login-CSRF); a same-origin request or a non-browser
|
||||
# client (no Origin) passes. Allowlist the canonical host + localhost so a
|
||||
# legitimate request always passes even if the tunnel rewrites Host.
|
||||
if method in ("POST", "PUT", "DELETE"):
|
||||
origin = self.headers.get("Origin", "")
|
||||
if origin:
|
||||
from urllib.parse import urlparse
|
||||
o = urlparse(origin)
|
||||
ok = (o.hostname in ("godstrument.pro", "localhost", "127.0.0.1")
|
||||
or o.netloc == self.headers.get("Host", ""))
|
||||
if not ok:
|
||||
blob = json.dumps({"error": "cross-site request blocked"}).encode()
|
||||
self.send_response(403)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(blob)))
|
||||
self.end_headers(); self.wfile.write(blob)
|
||||
return
|
||||
length = int(self.headers.get("Content-Length", 0) or 0)
|
||||
if length > 512 * 1024:
|
||||
return self.send_error(413, "too large")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user