thethreemagi
5c7a8dd785
The console's 401 redirect carries next=/leaders/ and /login dropped it, so a leader arriving cold landed on /account. next now rides the form as a hidden field and is applied after a successful POST and when an already-signed-in person hits /login. identity.safe_next() admits only a relative path with a single leading slash - no scheme, no //, no backslash, no control characters - so the login page cannot become an open redirect. Ten checks added to tests/smoke_identity.py; the suite is 59 + 24 + 53, all green.
185 lines
8.2 KiB
Python
185 lines
8.2 KiB
Python
"""
|
|
smoke_identity.py - end-to-end check of the identity layer 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_identity.py
|
|
|
|
It covers the rules that are expensive to get wrong and invisible when they
|
|
are: single-use invites, reissue revoking the previous link, the idle and
|
|
absolute session bounds, login throttling, and the global-versus-unit
|
|
capability split.
|
|
"""
|
|
|
|
import datetime, os, sys, tempfile, uuid
|
|
|
|
DB = os.environ.get("STORE_DB") or os.path.join(tempfile.mkdtemp(), "smoke.db")
|
|
os.environ["STORE_DB"] = DB
|
|
os.environ["ADMIN_BOOTSTRAP_EMAIL"] = "owner@example.test"
|
|
os.environ["SITE_BASE_URL"] = "https://greenlanescouts73.org"
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "app"))
|
|
|
|
import identity as I
|
|
|
|
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, fn, *a, **kw):
|
|
try:
|
|
fn(*a, **kw)
|
|
except I.IdentityError 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)
|
|
|
|
print("schema and unit seed")
|
|
I.init()
|
|
I.init()
|
|
units = I.list_units()
|
|
check("two units seeded", len(units) == 2)
|
|
check("init is idempotent", len(I.list_units()) == 2)
|
|
pack = I.get_unit("pack73"); troop = I.get_unit("troop73")
|
|
check("pack73 by slug", pack and pack["short_name"] == "Pack 73")
|
|
check("pack meets Tuesday 18:00", pack["meets_weekday"] == 2 and pack["meets_time"] == "18:00")
|
|
check("troop meets 19:30", troop["meets_time"] == "19:30")
|
|
|
|
print("\npasswords")
|
|
h = I.hash_password("correct horse battery staple")
|
|
check("verify accepts", I.verify_password("correct horse battery staple", h))
|
|
check("verify rejects", not I.verify_password("wrong", h))
|
|
check("verify rejects corrupt hash", not I.verify_password("x", "garbage"))
|
|
raises("short password", 422, I.hash_password, "short")
|
|
|
|
print("\nbootstrap")
|
|
url, minted = I.bootstrap()
|
|
check("mints on empty db", minted and url)
|
|
tok = url.rsplit("/", 1)[-1]
|
|
url2, minted2 = I.bootstrap()
|
|
check("reuses live invite on restart", not minted2 and url2 is None)
|
|
check("original token still live", I.peek_invite(tok) is not None)
|
|
|
|
print("\ninvite consumption")
|
|
check("peek does not consume", I.peek_invite(tok)["email"] == "owner@example.test")
|
|
owner = I.consume_invite(tok, "Test Owner", "a-long-enough-password", phone="555")
|
|
check("person created", owner["email"] == "owner@example.test")
|
|
check("global_role owner", owner["global_role"] == "owner")
|
|
check("password not returned", "password_hash" not in owner)
|
|
raises("second use of same token", 410, I.consume_invite, tok, "Impostor", "a-long-enough-password")
|
|
check("bootstrap now inert", I.bootstrap() == (None, False))
|
|
|
|
print("\nreissue revokes the previous link")
|
|
_, t1 = I.create_invite("leader@example.test", units=[{"unit_id": pack["id"], "role": "leader"}])
|
|
_, t2 = I.create_invite("leader@example.test", units=[{"unit_id": pack["id"], "role": "leader"}])
|
|
check("old token dead", I.peek_invite(t1) is None)
|
|
check("new token live", I.peek_invite(t2) is not None)
|
|
raises("invite with no role at all", 422, I.create_invite, "x@example.test")
|
|
raises("invite to unknown unit", 422, I.create_invite, "x@example.test",
|
|
None, [{"unit_id": "nope", "role": "leader"}])
|
|
raises("invite with bad unit role", 422, I.create_invite, "x@example.test",
|
|
None, [{"unit_id": pack["id"], "role": "wizard"}])
|
|
|
|
print("\nexpiry")
|
|
_, t3 = I.create_invite("expired@example.test", global_role="admin", ttl_days=-1)
|
|
check("expired invite unusable", I.peek_invite(t3) is None)
|
|
raises("expired invite cannot be consumed", 410, I.consume_invite, t3, "N", "a-long-enough-password")
|
|
|
|
leader = I.consume_invite(t2, "Den Leader", "another-long-password")
|
|
|
|
print("\ncapabilities")
|
|
check("leader can write calendar in own unit", I.can(leader, "calendar:write", pack["id"]))
|
|
check("leader cannot in the other unit", not I.can(leader, "calendar:write", troop["id"]))
|
|
check("leader cannot invite", not I.can(leader, "people:invite_leader"))
|
|
check("owner can manage people", I.can(owner, "people:manage"))
|
|
check("owner spans both units", I.can(owner, "calendar:write", pack["id"])
|
|
and I.can(owner, "calendar:write", troop["id"]))
|
|
|
|
con = I.connect()
|
|
con.execute("INSERT INTO units (id, slug, display_name, short_name, unit_type, unit_number,"
|
|
" active, sort_order, updated_at) VALUES (?,?,?,?,?,?,1,30,?)",
|
|
(str(uuid.uuid4()), "crew73", "Venturing Crew 73", "Crew 73", "crew", "73", I._now()))
|
|
con.commit(); con.close()
|
|
crew = I.get_unit("crew73")
|
|
check("owner reaches a unit created after the grant", I.can(I.get_person(owner["id"]),
|
|
"calendar:write", crew["id"]))
|
|
check("leader does not", not I.can(I.get_person(leader["id"]), "calendar:write", crew["id"]))
|
|
|
|
print("\nlogin and sessions")
|
|
raises("wrong password", 401, I.authenticate, "owner@example.test", "nope")
|
|
raises("unknown account", 401, I.authenticate, "ghost@example.test", "whatever")
|
|
p, stok = I.authenticate("owner@example.test", "a-long-enough-password")
|
|
check("authenticates", p["id"] == owner["id"])
|
|
check("session resolves", I.session_person(stok)["id"] == owner["id"])
|
|
check("junk cookie resolves to nobody", I.session_person("junk") is None)
|
|
I.end_session(stok)
|
|
check("revoked session is dead", I.session_person(stok) is None)
|
|
|
|
_, stok2 = I.authenticate("owner@example.test", "a-long-enough-password")
|
|
con = I.connect()
|
|
stale = (datetime.datetime.now(datetime.timezone.utc)
|
|
- datetime.timedelta(hours=I.SESSION_IDLE_HOURS + 1)).isoformat(timespec="seconds")
|
|
con.execute("UPDATE sessions SET last_seen_at=? WHERE token_hash=?",
|
|
(stale, I._hash_token(stok2)))
|
|
con.commit(); con.close()
|
|
check("idle timeout enforced", I.session_person(stok2) is None)
|
|
|
|
_, stok3 = I.authenticate("owner@example.test", "a-long-enough-password")
|
|
con = I.connect()
|
|
con.execute("UPDATE people SET disabled_at=? WHERE id=?", (I._now(), owner["id"]))
|
|
con.commit(); con.close()
|
|
check("disabled person has no session", I.session_person(stok3) is None)
|
|
check("disabled person holds no capabilities", I.effective_caps(I.get_person(owner["id"])) == set())
|
|
raises("disabled person cannot log in", 401, I.authenticate,
|
|
"owner@example.test", "a-long-enough-password")
|
|
con = I.connect(); con.execute("UPDATE people SET disabled_at=NULL WHERE id=?", (owner["id"],))
|
|
con.commit(); con.close()
|
|
|
|
print("\nthrottle")
|
|
for _ in range(I.LOGIN_MAX_FAILURES):
|
|
try:
|
|
I.authenticate("throttle@example.test", "bad")
|
|
except I.IdentityError:
|
|
pass
|
|
raises("locks out after %d failures" % I.LOGIN_MAX_FAILURES, 429,
|
|
I.authenticate, "throttle@example.test", "bad")
|
|
check("other accounts unaffected", I.authenticate("leader@example.test",
|
|
"another-long-password")[0] is not None)
|
|
|
|
print("\naudit")
|
|
con = I.connect()
|
|
kinds = {r["kind"] for r in con.execute("SELECT DISTINCT kind FROM auth_events")}
|
|
con.close()
|
|
for k in ("invite.created", "invite.consumed", "login.ok", "login.failed", "login.throttled"):
|
|
check("auth_events records %s" % k, k in kinds)
|
|
|
|
print("\nsafe_next")
|
|
check("relative path passes", I.safe_next("/leaders/") == "/leaders/")
|
|
check("query string kept", I.safe_next("/leaders/nearby?x=1") == "/leaders/nearby?x=1")
|
|
check("empty falls back", I.safe_next("") == "/account")
|
|
check("None falls back", I.safe_next(None) == "/account")
|
|
check("absolute URL rejected", I.safe_next("https://evil.example/") == "/account")
|
|
check("protocol-relative rejected", I.safe_next("//evil.example/") == "/account")
|
|
check("backslash form rejected", I.safe_next("/\\evil.example") == "/account")
|
|
check("control char rejected", I.safe_next("/leaders\r\nX: y") == "/account")
|
|
check("no leading slash rejected", I.safe_next("leaders/") == "/account")
|
|
check("custom default honoured", I.safe_next("nope", default="/") == "/")
|
|
|
|
print("\n%d passed, %d failed" % (PASS, FAIL))
|
|
sys.exit(1 if FAIL else 0)
|