scout-website/tests/smoke_identity.py

"""
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("\napi keys")
raises("label required", 422, I.mint_api_key, leader, "", ["leads:read"])
raises("scopes required", 422, I.mint_api_key, leader, "script", [])
raises("scope the person does not hold", 422, I.mint_api_key, leader, "script", ["settings:write"])
raises("unscopable scope refused even for an owner", 422, I.mint_api_key, owner, "script", ["apikeys:own"])
raises("bad expiry", 422, I.mint_api_key, leader, "script", ["leads:read"], "soon")
raises("expiry over the cap", 422, I.mint_api_key, leader, "script", ["leads:read"], 9999)
full, row = I.mint_api_key(leader, "roundup script", ["leads:read", "nearby:write"], 30)
check("key has the prefix and is not stored", full.startswith("gls73_") and "key_hash" not in row
      and row["prefix"] == full[:12] and row["state"] == "active" and row["expires_at"])
check("scopes stored sorted", row["scopes"] == ["leads:read", "nearby:write"])
kp = I.api_key_person(full)
check("key resolves to its owner, narrowed", kp and kp["email"] == "leader@example.test"
      and kp["key_scopes"] == {"leads:read", "nearby:write"} and kp["key_prefix"] == row["prefix"])
check("can() honours the narrowing", I.can(kp, "leads:read") and I.can(kp, "nearby:write")
      and not I.can(kp, "announcements:write") and not I.can(kp, "apikeys:own"))
check("unit scope still applies through a key", I.can(kp, "nearby:write", pack["id"]))
check("capabilities list reflects the key", kp["capabilities"] == ["leads:read", "nearby:write"])
check("last_used_at touched", I.list_api_keys(leader["id"])[0]["last_used_at"])
raises("a key cannot mint keys", 403, I.mint_api_key, kp, "nested", ["leads:read"])
check("unknown key is nobody", I.api_key_person("gls73_nope") is None)
check("foreign-prefixed value is nobody", I.api_key_person("tk_" + full[6:]) is None)
check("empty is nobody", I.api_key_person("") is None and I.api_key_person(None) is None)
check("not another person's to revoke", I.revoke_api_key(owner, row["id"]) is None)
rev = I.revoke_api_key(leader, row["id"])
check("revoked by its owner", rev["state"] == "revoked" and rev["revoked_at"])
check("revoked key is nobody", I.api_key_person(full) is None)
raises("second revoke", 409, I.revoke_api_key, leader, row["id"])
full2, row2 = I.mint_api_key(leader, "no expiry", ["leads:read"])
check("no expiry allowed", row2["expires_at"] is None and I.api_key_person(full2) is not None)
con = I.connect()
con.execute("UPDATE api_keys SET expires_at='2000-01-01T00:00:00+00:00' WHERE id=?", (row2["id"],)); con.commit(); con.close()
check("expired key is nobody, and lists as expired", I.api_key_person(full2) is None
      and I.list_api_keys(leader["id"])[0]["state"] == "expired")
full3, row3 = I.mint_api_key(leader, "survives?", ["leads:read"])
con = I.connect()
con.execute("UPDATE people SET disabled_at=? WHERE id=?", (I._now(), leader["id"])); con.commit(); con.close()
check("disabling the person kills the key", I.api_key_person(full3) is None)
con = I.connect()
con.execute("UPDATE people SET disabled_at=NULL WHERE id=?", (leader["id"],)); con.commit(); con.close()
check("scopable set for a leader excludes the unscopable", "apikeys:own" not in I.scopable_caps(leader)
      and "leads:read" in I.scopable_caps(leader) and "settings:write" not in I.scopable_caps(leader))
con = I.connect()
kinds = {r["kind"] for r in con.execute("SELECT DISTINCT kind FROM auth_events")}
con.close()
check("auth_events records mint and revoke", "apikey.minted" in kinds and "apikey.revoked" in kinds)

print("\npeople management")
owner_p = I.get_person(owner["id"]); leader_p = I.get_person(leader["id"])
g = I.grantable_roles(owner_p); check("owner grants anything", g["global"] == ["owner", "admin"] and g["unit"] == ["leader", "member"])
check("leader grants nothing (invites are admin and above)", I.grantable_roles(leader_p) == {"global": [], "unit": []})
raises("leader cannot invite at all", 403, I.invite_person, leader_p, "new@example.test", None,
       [{"unit_id": pack["id"], "role": "member"}])
raises("admin cannot grant owner", 403, I.invite_person, dict(owner_p, global_role="admin"), "new@example.test", "owner", [])
raises("existing address cannot be re-invited", 409, I.invite_person, owner_p, "leader@example.test", None,
       [{"unit_id": pack["id"], "role": "member"}])
row, url = I.invite_person(dict(owner_p, global_role="admin"), "den2@example.test", None, [{"unit_id": pack["id"], "role": "leader", "title": "Bear Den"}])
check("invite url minted once", url.startswith(I.SITE_BASE_URL + "/invite/") and "token_hash" in row)
inv = [i for i in I.list_open_invites() if i["email"] == "den2@example.test"]
check("open invite listed, units decoded, not expired", len(inv) == 1 and inv[0]["units"][0]["title"] == "Bear Den" and not inv[0]["expired"])
check("revoke invite", I.revoke_invite(owner_p, inv[0]["id"])["revoked_at"] and not [i for i in I.list_open_invites() if i["email"] == "den2@example.test"])
raises("roles need people:manage", 403, I.set_roles, leader_p, leader["id"], "admin", [])
raises("last owner cannot be demoted", 409, I.set_roles, owner_p, owner["id"], "admin", [])
raises("cannot disable yourself", 409, I.disable_person, owner_p, owner["id"])
raises("last owner cannot be disabled", 409, I.disable_person, owner_p, owner["id"])
after = I.set_roles(owner_p, leader["id"], "admin", [{"unit_id": troop["id"], "role": "leader", "title": "SM"}])
check("roles replaced in full", after["global_role"] == "admin" and [m["slug"] for m in after["memberships"]] == ["troop73"])
after = I.set_roles(owner_p, leader["id"], None, [{"unit_id": pack["id"], "role": "leader"}])
check("and back", after["global_role"] is None and [m["slug"] for m in after["memberships"]] == ["pack73"])
ppl = I.list_people(); me = [p for p in ppl if p["id"] == leader["id"]][0]
check("list carries counts", "active_sessions" in me and "active_keys" in me and "password_hash" not in me)
raises("leader cannot issue a reset", 403, I.create_reset, leader_p, owner["id"])
raises("admin cannot reset an owner", 403, I.create_reset, dict(owner_p, global_role="admin", capabilities=[]), owner["id"])
url = I.create_reset(owner_p, leader["id"]); rtok = url.rsplit("/", 1)[-1]
check("reset link minted", "/reset/" in url and I.peek_reset(rtok)["id"] == leader["id"])
stale = I.create_reset(owner_p, leader["id"]).rsplit("/", 1)[-1]
check("reissue revokes the prior link", I.peek_reset(rtok) is None and I.peek_reset(stale) is not None)
s_before = I.start_session(leader["id"])
raises("short password on reset", 422, I.consume_reset, stale, "short")
I.consume_reset(stale, "a-brand-new-long-password")
check("reset sets the password and ends sessions", I.authenticate("leader@example.test", "a-brand-new-long-password")[0] is not None
      and I.session_person(s_before) is None and I.peek_reset(stale) is None)
raises("used link is dead", 410, I.consume_reset, stale, "another-long-password-here")
raises("change password needs the current one", 403, I.change_password, leader["id"], "wrong-password-here", "yet-another-long-pw")
I.change_password(leader["id"], "a-brand-new-long-password", "yet-another-long-password")
check("self change works", I.authenticate("leader@example.test", "yet-another-long-password")[0] is not None)
raises("details need a name", 422, I.update_own_details, leader["id"], "")
check("details update", I.update_own_details(leader["id"], "Den Leader Two", "Deb", "555-0100")["preferred_name"] == "Deb")
d = I.disable_person(owner_p, leader["id"], "moved away")
check("disabled: row kept, reason, sessions gone, auth refused", d["disabled_at"] and d["disabled_reason"] == "moved away"
      and I.session_person(I.start_session(leader["id"])) is None)
raises("reset for a disabled person", 409, I.create_reset, owner_p, leader["id"])
raises("disable twice", 409, I.disable_person, owner_p, leader["id"])
check("enabled again", I.enable_person(owner_p, leader["id"])["disabled_at"] is None)
kinds = {r["kind"] for r in I.list_events(limit=500)}
check("people actions in the log", {"invite.revoked", "person.roles", "person.disabled", "person.enabled",
      "password.reset_issued", "password.reset", "password.changed"} <= kinds)

print("\nfooter sign-in")
import os as _os
_src = open(_os.path.join(_os.path.dirname(__file__), "..", "app", "app.py")).read()
_foot = _src[_src.find("<footer"):_src.find("</footer>")]
check("public footer carries one quiet sign-in link before Join", _foot.count('href="/login"') == 1 and _foot.index('href="/login"') < _foot.index('href="/join"'))
check("sign-in lands on the console by default; unsafe next still refused",
      I.safe_next("") == "/leaders/" and I.safe_next("//evil") == "/leaders/" and I.safe_next("/leaders/roster") == "/leaders/roster")

print("\nmeeting words")
D = dict(MEETING_DAY="Tuesday", MEETING_DAYS="Tuesdays", MEETING_DAY_ABBR="Tue", PACK_TIME="6:00 PM",
         TROOP_TIME="7:30 PM", PACK_CLOCK="6:00", TROOP_CLOCK="7:30")
check("clock words", I.clock_words("18:00") == ("6:00 PM", "6:00") and I.clock_words("07:05") == ("7:05 AM", "7:05")
      and I.clock_words("00:30") == ("12:30 AM", "12:30") and I.clock_words("12:00") == ("12:00 PM", "12:00"))
check("clock words refuse junk", I.clock_words(None) is None and I.clock_words("6pm") is None)
w = I.meeting_words([{"slug": "pack73", "meets_weekday": 4, "meets_time": "18:15"},
                     {"slug": "troop73", "meets_weekday": 4, "meets_time": "19:45"}], D)
check("day and both times derived", w["MEETING_DAY"] == "Thursday" and w["MEETING_DAYS"] == "Thursdays"
      and w["MEETING_DAY_ABBR"] == "Thu" and w["PACK_TIME"] == "6:15 PM" and w["PACK_CLOCK"] == "6:15"
      and w["TROOP_TIME"] == "7:45 PM" and w["TROOP_CLOCK"] == "7:45")
w2 = I.meeting_words([{"slug": "pack73", "meets_weekday": None, "meets_time": None},
                      {"slug": "troop73", "meets_weekday": 2, "meets_time": "19:30"}], D)
check("missing pack values keep the defaults, day falls to the troop row",
      w2["MEETING_DAY"] == "Tuesday" and w2["PACK_TIME"] == "6:00 PM" and w2["TROOP_TIME"] == "7:30 PM")
check("no rows at all is the defaults", I.meeting_words([], D) == D)
check("defaults dict is not mutated", D["PACK_TIME"] == "6:00 PM")
check("live seed rows reproduce the constants", I.meeting_words(I.list_units(), D) == D)

print("\nwhoami payload (what scout-finance reads)")
o = I.get_person(owner["id"])
l = I.get_person(leader["id"])
wo = I.whoami_payload(o)
wl = I.whoami_payload(l)
check("carries the person id, for entered_by", wo["id"] == owner["id"])
check("carries a display name, for entered_by_name",
      wl["preferred_name"] or wl["full_name"])
check("finance:read and finance:write are a leader capability",
      "finance:read" in wl["capabilities"] and "finance:write" in wl["capabilities"])
check("finance:admin is not", "finance:admin" not in wl["capabilities"])
check("finance:admin is an owner capability", I.can(o, "finance:admin"))
check("a leader's membership carries its own capability set",
      "finance:write" in wl["memberships"][0]["capabilities"])
check("a leader has no site-wide grant", wl["global_capabilities"] == [])
check("an owner's site-wide grant is not empty",
      "finance:write" in wo["global_capabilities"])
check("membership rows name the unit for a screen",
      wl["memberships"][0]["unit_id"] and wl["memberships"][0]["role"] == "leader")
check("payload capabilities agree with can()",
      all(I.can(l, c) for c in wl["capabilities"]))
check("a capability held only per unit does not appear site-wide for others",
      not I.can(l, "finance:write", troop["id"]))
rawkey, _krow = I.mint_api_key(l, "finance reader", ["finance:read"])
kp = I.api_key_person(rawkey)
wk = I.whoami_payload(kp, via="key")
check("a key's membership capabilities are narrowed to its scopes too",
      wk["memberships"][0]["capabilities"] == ["finance:read"])
check("a key cannot appear to hold what can() would refuse",
      not I.can(kp, "finance:write") and "finance:write" not in wk["capabilities"])

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("") == "/leaders/")
check("None falls back", I.safe_next(None) == "/leaders/")
check("absolute URL rejected", I.safe_next("https://evil.example/") == "/leaders/")
check("protocol-relative rejected", I.safe_next("//evil.example/") == "/leaders/")
check("backslash form rejected", I.safe_next("/\\evil.example") == "/leaders/")
check("control char rejected", I.safe_next("/leaders\r\nX: y") == "/leaders/")
check("no leading slash rejected", I.safe_next("leaders/") == "/leaders/")
check("custom default honoured", I.safe_next("nope", default="/") == "/")

print("\n%d passed, %d failed" % (PASS, FAIL))
sys.exit(1 if FAIL else 0)