scout-website/app/identity.py

"""
identity.py - units, people, roles, invites and sessions for greenlanescouts73.org

This is the identity layer the leader console (scout-control) and the member
document gate both sit on. It owns tables in scout73.db and, like store.py,
this app remains the only writer to them.

Three decisions are load-bearing and should not be quietly undone.

UNITS ARE A TABLE, NOT A STRING. A slug typed into six places is a slug that
will be typed wrong in one of them. Meeting nights live on the unit row rather
than in settings, because they are facts about a unit, and they are stored
structured (weekday + 24h time) rather than as display sentences, because the
site composes them five different ways.

ROLES SPLIT TWO WAYS. `leader` and `member` are per unit, in memberships.
`owner` and `admin` are site-wide, in people.global_role. If admin were a
membership row, granting it would mean one insert per unit, and the day a third
unit is added every existing admin would silently lose sight of it. Effective
capability is the union of the global set and the per-unit sets.

NOTHING IS HARD DELETED. People are disabled, invites are revoked or consumed,
sessions are revoked. Who had access, and when, has to stay answerable.

Stdlib only, as with store.py - scrypt ships with Python, so this adds no
image dependencies.
"""

import base64
import datetime
import hashlib
import hmac
import json
import os
import secrets
import sqlite3
import uuid
from pathlib import Path

DB_PATH = Path(os.environ.get("STORE_DB", "/data/scout73.db"))

SITE_BASE_URL = os.environ.get("SITE_BASE_URL", "https://greenlanescouts73.org").rstrip("/")
ADMIN_BOOTSTRAP_EMAIL = os.environ.get("ADMIN_BOOTSTRAP_EMAIL", "").strip().lower()

SESSION_ABSOLUTE_DAYS = 30
SESSION_IDLE_HOURS = 12
INVITE_TTL_DAYS = 14

# Login throttle. Counted from auth_events, so it survives a restart - an
# in-memory counter resets to zero on every deploy, which is not a throttle.
LOGIN_WINDOW_MINUTES = 15
LOGIN_MAX_FAILURES = 8

GLOBAL_ROLES = ("owner", "admin")
UNIT_ROLES = ("leader", "member")

# ---------------------------------------------------------------------------
# Capabilities
# ---------------------------------------------------------------------------
# One dictionary, consulted by one function. Scattered `if role == "admin"`
# checks are how a permission model rots: there has to be a single place to
# read to know who can do what.
#
# email:* are reserved and unused. The mail server is not connected yet, and
# keys minted before it lands should not need re-scoping afterwards.

CAPS = {
    "member": {
        "account:self",
        "documents:read_members",
    },
    "leader": {
        "account:self",
        "documents:read_members",
        "announcements:write",
        "leads:read",
        "nearby:write",
        "calendar:write",
        "unit:write_own",
        "roster:write",
        "fbposts:read",
        "apikeys:own",
        "api:docs",
        "email:draft",
    },
    "admin": {
        "account:self",
        "documents:read_members",
        "announcements:write",
        "leads:read",
        "nearby:write",
        "calendar:write",
        "unit:write_own",
        "roster:write",
        "fbposts:read",
        "fbposts:ingest",
        "units:write",
        "settings:write",
        "apikeys:own",
        "api:docs",
        "history:read",
        "people:invite_leader",
        "people:invite_admin",
        "email:draft",
        "email:send",
    },
}
CAPS["owner"] = CAPS["admin"] | {"people:manage", "secrets:rotate"}

SCHEMA = """
CREATE TABLE IF NOT EXISTS units (
    id            TEXT PRIMARY KEY,
    slug          TEXT NOT NULL UNIQUE,
    display_name  TEXT NOT NULL,
    short_name    TEXT NOT NULL,
    unit_type     TEXT NOT NULL,
    unit_number   TEXT NOT NULL,
    meets_weekday INTEGER,
    meets_time    TEXT,
    meets_at      TEXT,
    active        INTEGER NOT NULL DEFAULT 1,
    sort_order    INTEGER NOT NULL DEFAULT 100,
    updated_at    TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS people (
    id              TEXT PRIMARY KEY,
    email           TEXT NOT NULL UNIQUE COLLATE NOCASE,
    full_name       TEXT,
    preferred_name  TEXT,
    phone           TEXT,
    password_hash   TEXT,
    global_role     TEXT,
    bsa_member_id   TEXT,
    ypt_completed_on TEXT,
    registered_adult INTEGER NOT NULL DEFAULT 0,
    contact_pref    TEXT,
    created_at      TEXT NOT NULL,
    created_by      TEXT,
    last_login_at   TEXT,
    disabled_at     TEXT,
    disabled_reason TEXT
);

CREATE TABLE IF NOT EXISTS memberships (
    person_id  TEXT NOT NULL REFERENCES people(id),
    unit_id    TEXT NOT NULL REFERENCES units(id),
    role       TEXT NOT NULL,
    title      TEXT,
    created_at TEXT NOT NULL,
    created_by TEXT,
    PRIMARY KEY (person_id, unit_id)
);

CREATE TABLE IF NOT EXISTS invites (
    id          TEXT PRIMARY KEY,
    token_hash  TEXT NOT NULL UNIQUE,
    email       TEXT NOT NULL COLLATE NOCASE,
    global_role TEXT,
    units       TEXT NOT NULL DEFAULT '[]',
    created_at  TEXT NOT NULL,
    created_by  TEXT,
    expires_at  TEXT NOT NULL,
    consumed_at TEXT,
    person_id   TEXT,
    revoked_at  TEXT
);
CREATE INDEX IF NOT EXISTS idx_invites_email ON invites(email, consumed_at, revoked_at);

CREATE TABLE IF NOT EXISTS sessions (
    id           TEXT PRIMARY KEY,
    token_hash   TEXT NOT NULL UNIQUE,
    person_id    TEXT NOT NULL REFERENCES people(id),
    created_at   TEXT NOT NULL,
    expires_at   TEXT NOT NULL,
    last_seen_at TEXT NOT NULL,
    ip           TEXT,
    user_agent   TEXT,
    revoked_at   TEXT
);
CREATE INDEX IF NOT EXISTS idx_sessions_person ON sessions(person_id, revoked_at);

CREATE TABLE IF NOT EXISTS auth_events (
    id        TEXT PRIMARY KEY,
    at        TEXT NOT NULL,
    kind      TEXT NOT NULL,
    person_id TEXT,
    actor_id  TEXT,
    email     TEXT,
    detail    TEXT,
    ip        TEXT
);
CREATE INDEX IF NOT EXISTS idx_auth_events_at ON auth_events(at DESC);
CREATE INDEX IF NOT EXISTS idx_auth_events_kind ON auth_events(kind, email, at DESC);

CREATE TABLE IF NOT EXISTS settings (
    key        TEXT PRIMARY KEY,
    value      TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    updated_by TEXT
);

-- Password resets. Admin-issued one-time links (there is no outbound mail
-- yet, so the admin hands the link over however they talk to the person).
-- Same shape as invites: only the sha256 of the token is stored, single
-- use, short expiry, reissue revokes the prior link.
CREATE TABLE IF NOT EXISTS password_resets (
    id          TEXT PRIMARY KEY,
    token_hash  TEXT NOT NULL UNIQUE,
    person_id   TEXT NOT NULL REFERENCES people(id),
    created_at  TEXT NOT NULL,
    created_by  TEXT,
    expires_at  TEXT NOT NULL,
    consumed_at TEXT,
    revoked_at  TEXT
);
CREATE INDEX IF NOT EXISTS idx_password_resets_person ON password_resets(person_id, consumed_at, revoked_at);

-- API keys (P3). One row per key; only the sha256 of the key is stored, and
-- the visible prefix exists so a person can tell their keys apart. Scopes are
-- a JSON list and are a SUBSET of the owner's capabilities, checked at mint
-- time here and again at use time in can(), so a key never outlives its
-- owner's demotion. Revoked rows stay.
CREATE TABLE IF NOT EXISTS api_keys (
    id           TEXT PRIMARY KEY,
    person_id    TEXT NOT NULL REFERENCES people(id),
    label        TEXT NOT NULL,
    prefix       TEXT NOT NULL,
    key_hash     TEXT NOT NULL UNIQUE,
    scopes       TEXT NOT NULL,
    created_at   TEXT NOT NULL,
    last_used_at TEXT,
    expires_at   TEXT,
    revoked_at   TEXT
);
CREATE INDEX IF NOT EXISTS idx_api_keys_person ON api_keys(person_id, revoked_at);
"""

# Seeded at boot, idempotent by slug. Values lifted from the site constants in
# app.py so the two cannot disagree on day one. app.py keeps its constants
# until P2 moves the rendering over.
SEED_UNITS = [
    dict(slug="pack73", display_name="Cub Scout Pack 73", short_name="Pack 73",
         unit_type="pack", unit_number="73", meets_weekday=2, meets_time="18:00",
         meets_at="St. Luke's Lutheran Church, Zieglerville, PA", sort_order=10),
    dict(slug="troop73", display_name="Scouts BSA Troop 73", short_name="Troop 73",
         unit_type="troop", unit_number="73", meets_weekday=2, meets_time="19:30",
         meets_at="St. Luke's Lutheran Church, Zieglerville, PA", sort_order=20),
]


class IdentityError(Exception):
    """Carries the HTTP status the route should return, so rules live here."""

    def __init__(self, status, detail):
        super().__init__(detail)
        self.status = status
        self.detail = detail


def _now():
    return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")


def _plus(**kw):
    return (datetime.datetime.now(datetime.timezone.utc)
            + datetime.timedelta(**kw)).isoformat(timespec="seconds")


def connect():
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    con = sqlite3.connect(DB_PATH, timeout=10)
    con.row_factory = sqlite3.Row
    con.execute("PRAGMA foreign_keys=ON")
    return con


def init():
    """Create the schema and seed units. Safe on every boot."""
    con = connect()
    try:
        con.executescript(SCHEMA)
        for u in SEED_UNITS:
            con.execute(
                "INSERT INTO units (id, slug, display_name, short_name, unit_type,"
                " unit_number, meets_weekday, meets_time, meets_at, active, sort_order, updated_at)"
                " VALUES (?,?,?,?,?,?,?,?,?,1,?,?)"
                " ON CONFLICT(slug) DO NOTHING",
                (str(uuid.uuid4()), u["slug"], u["display_name"], u["short_name"],
                 u["unit_type"], u["unit_number"], u["meets_weekday"], u["meets_time"],
                 u["meets_at"], u["sort_order"], _now()))
        con.commit()
    finally:
        con.close()


# ---------------------------------------------------------------------------
# Passwords
# ---------------------------------------------------------------------------

SCRYPT_N, SCRYPT_R, SCRYPT_P = 2 ** 14, 8, 1


def hash_password(password):
    if not password or len(password) < 12:
        raise IdentityError(422, "password must be at least 12 characters")
    salt = secrets.token_bytes(16)
    dk = hashlib.scrypt(password.encode(), salt=salt, n=SCRYPT_N, r=SCRYPT_R,
                        p=SCRYPT_P, dklen=32)
    return "scrypt$%d$%d$%d$%s$%s" % (
        SCRYPT_N, SCRYPT_R, SCRYPT_P,
        base64.b64encode(salt).decode(), base64.b64encode(dk).decode())


def verify_password(password, stored):
    """Constant-time check. False on anything malformed rather than raising -
    a corrupt hash must read as a failed login, never as a pass."""
    try:
        scheme, n, r, p, salt_b64, dk_b64 = stored.split("$")
        if scheme != "scrypt":
            return False
        dk = hashlib.scrypt(password.encode(), salt=base64.b64decode(salt_b64),
                            n=int(n), r=int(r), p=int(p), dklen=32)
        return hmac.compare_digest(dk, base64.b64decode(dk_b64))
    except Exception:
        return False


def _hash_token(tok):
    return hashlib.sha256(tok.encode()).hexdigest()


# ---------------------------------------------------------------------------
# Audit
# ---------------------------------------------------------------------------

def log_event(kind, person_id=None, actor_id=None, email=None, detail=None, ip=None, con=None):
    own = con is None
    con = con or connect()
    try:
        con.execute(
            "INSERT INTO auth_events (id, at, kind, person_id, actor_id, email, detail, ip)"
            " VALUES (?,?,?,?,?,?,?,?)",
            (str(uuid.uuid4()), _now(), kind, person_id, actor_id,
             (email or "").lower() or None, detail, ip))
        if own:
            con.commit()
    finally:
        if own:
            con.close()


# ---------------------------------------------------------------------------
# Units and people
# ---------------------------------------------------------------------------

def list_units(include_inactive=False):
    con = connect()
    try:
        sql = "SELECT * FROM units"
        if not include_inactive:
            sql += " WHERE active = 1"
        sql += " ORDER BY sort_order, slug"
        return [dict(r) for r in con.execute(sql)]
    finally:
        con.close()


def get_unit(slug_or_id):
    con = connect()
    try:
        r = con.execute("SELECT * FROM units WHERE slug=? OR id=?",
                        (slug_or_id, slug_or_id)).fetchone()
        return dict(r) if r else None
    finally:
        con.close()


# The only unit fields the admin API lets a leader change. Renaming a unit or
# changing its type is a rechartering event, not a Tuesday edit, and stays a
# code change.
UNIT_MEETS_FIELDS = ("meets_weekday", "meets_time", "meets_at")

WEEKDAY_NAMES = ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday")


def clock_words(hhmm):
    """"18:00" -> ("6:00 PM", "6:00"). Returns None on anything unparseable so
    the caller keeps its fallback rather than rendering a blank."""
    try:
        t = datetime.datetime.strptime(str(hhmm), "%H:%M")
    except (TypeError, ValueError):
        return None
    h12 = t.hour % 12 or 12
    clock = "%d:%02d" % (h12, t.minute)
    return ("%s %s" % (clock, "AM" if t.hour < 12 else "PM"), clock)


def meeting_words(units, defaults):
    """The seven seasonal strings app.py renders, derived from the units rows.

    `defaults` is the dict of the code constants; any word a row cannot supply
    keeps its default, so a half-filled unit row degrades to today's copy and
    never to a blank page. The day words come from the PACK row: the site's
    copy assumes both units meet the same night ("right after the pack"), and
    if that ever changes the copy needs rewriting, not a bigger constant.
    """
    out = dict(defaults)
    by = {u.get("slug"): u for u in units}
    pack, troop = by.get("pack73"), by.get("troop73")
    day = (pack or troop or {}).get("meets_weekday")
    if isinstance(day, int) and 1 <= day <= 7:
        name = WEEKDAY_NAMES[day - 1]
        out["MEETING_DAY"], out["MEETING_DAYS"], out["MEETING_DAY_ABBR"] = name, name + "s", name[:3]
    for row, tkey, ckey in ((pack, "PACK_TIME", "PACK_CLOCK"), (troop, "TROOP_TIME", "TROOP_CLOCK")):
        w = clock_words((row or {}).get("meets_time"))
        if w:
            out[tkey], out[ckey] = w
    return out


def update_unit_meets(slug_or_id, fields):
    """Change when and where a unit meets.

    meets_weekday is ISO (Monday=1 .. Sunday=7) and meets_time is 24h "HH:MM",
    matching how the seed rows store them; the site composes the display
    sentence, so a display string is never accepted here. Any field may be set
    to null - a unit between meeting places is a real state.
    """
    unit = get_unit(slug_or_id)
    if not unit:
        raise IdentityError(404, "no such unit")
    if not fields:
        raise IdentityError(422, "nothing to update")
    unknown = sorted(set(fields) - set(UNIT_MEETS_FIELDS))
    if unknown:
        raise IdentityError(422, "only meeting fields are editable here: %s"
                            % (", ".join(UNIT_MEETS_FIELDS)))

    out = dict(fields)
    if "meets_weekday" in out and out["meets_weekday"] is not None:
        v = out["meets_weekday"]
        if isinstance(v, bool) or not isinstance(v, int) or not 1 <= v <= 7:
            raise IdentityError(422, "meets_weekday is ISO: 1 (Monday) to 7 (Sunday), or null")
    if "meets_time" in out and out["meets_time"] is not None:
        try:
            datetime.datetime.strptime(str(out["meets_time"]), "%H:%M")
        except ValueError:
            raise IdentityError(422, 'meets_time must be 24h "HH:MM", or null')
        out["meets_time"] = str(out["meets_time"])
    if "meets_at" in out and out["meets_at"] is not None:
        out["meets_at"] = str(out["meets_at"]).strip() or None

    out["updated_at"] = _now()
    con = connect()
    try:
        con.execute("UPDATE units SET %s WHERE id=?"
                    % ", ".join("%s=?" % k for k in out),
                    list(out.values()) + [unit["id"]])
        con.commit()
    finally:
        con.close()
    return get_unit(unit["id"])


def _person_row(con, r):
    if not r:
        return None
    p = dict(r)
    p.pop("password_hash", None)
    p["memberships"] = [dict(m) for m in con.execute(
        "SELECT m.unit_id, m.role, m.title, u.slug, u.unit_type, u.short_name, u.display_name"
        " FROM memberships m JOIN units u ON u.id = m.unit_id"
        " WHERE m.person_id = ? ORDER BY u.sort_order", (p["id"],))]
    p["capabilities"] = sorted(effective_caps(p))
    return p


def get_person(person_id):
    con = connect()
    try:
        return _person_row(con, con.execute(
            "SELECT * FROM people WHERE id=?", (person_id,)).fetchone())
    finally:
        con.close()


def get_person_by_email(email):
    con = connect()
    try:
        return _person_row(con, con.execute(
            "SELECT * FROM people WHERE email=?", ((email or "").lower(),)).fetchone())
    finally:
        con.close()


def people_count():
    con = connect()
    try:
        return con.execute("SELECT COUNT(*) c FROM people").fetchone()["c"]
    finally:
        con.close()


def effective_caps(person):
    """Union of the global role's capabilities and every membership's.

    Union, not precedence: an admin who is also a den leader should not lose
    anything by holding both, and a leader in one unit is not thereby a leader
    in another - that part is answered by can(), which takes a unit.
    """
    caps = set()
    if person.get("disabled_at"):
        return caps
    if person.get("global_role"):
        caps |= CAPS.get(person["global_role"], set())
    for m in person.get("memberships", []):
        caps |= CAPS.get(m["role"], set())
    return caps


def can(person, capability, unit_id=None):
    """Does this person hold `capability`, optionally within a specific unit?

    A global role satisfies a unit-scoped check for EVERY unit, including units
    created after the role was granted. That is the entire reason global_role
    is a column rather than a membership row.
    """
    if not person or person.get("disabled_at"):
        return False
    # A person reached through an API key is that person narrowed to the
    # key's scopes. Checked here, in the one place that decides, so a route
    # cannot forget it and a demoted owner's key loses what the owner lost.
    scopes = person.get("key_scopes")
    if scopes is not None and capability not in scopes:
        return False
    if person.get("global_role") and capability in CAPS.get(person["global_role"], set()):
        return True
    for m in person.get("memberships", []):
        if unit_id and m["unit_id"] != unit_id:
            continue
        if capability in CAPS.get(m["role"], set()):
            return True
    return False


# ---------------------------------------------------------------------------
# Invites
# ---------------------------------------------------------------------------
#
# Only the sha256 of the token is stored. A database read, a backup on the NAS
# or a stray SELECT must not hand over live invitations.
#
# Issuing a new invite for an address revokes the prior unconsumed one, which
# is what "the URL can be recreated until it is used" means in practice. Single
# use is enforced by setting consumed_at in the SAME transaction that creates
# the person, so a double submit cannot mint two accounts.

def create_invite(email, global_role=None, units=None, created_by=None, ttl_days=INVITE_TTL_DAYS):
    """Revoke any live invite for this address, mint a new one, return (row, token).

    The raw token is returned exactly once and never stored.
    """
    email = (email or "").strip().lower()
    if "@" not in email:
        raise IdentityError(422, "a valid email address is required")
    if global_role and global_role not in GLOBAL_ROLES:
        raise IdentityError(422, "global_role must be one of %s" % (GLOBAL_ROLES,))

    units = units or []
    for u in units:
        if u.get("role") not in UNIT_ROLES:
            raise IdentityError(422, "unit role must be one of %s" % (UNIT_ROLES,))
        if not get_unit(u.get("unit_id") or ""):
            raise IdentityError(422, "unknown unit %r" % (u.get("unit_id"),))
    if not global_role and not units:
        raise IdentityError(422, "an invite needs a global role, a unit membership, or both")

    tok = secrets.token_urlsafe(32)
    iid = str(uuid.uuid4())
    con = connect()
    try:
        con.execute("UPDATE invites SET revoked_at=? WHERE email=? AND consumed_at IS NULL"
                    " AND revoked_at IS NULL", (_now(), email))
        con.execute(
            "INSERT INTO invites (id, token_hash, email, global_role, units, created_at,"
            " created_by, expires_at, consumed_at, person_id, revoked_at)"
            " VALUES (?,?,?,?,?,?,?,?,NULL,NULL,NULL)",
            (iid, _hash_token(tok), email, global_role,
             json.dumps(units, ensure_ascii=False), _now(), created_by,
             _plus(days=ttl_days)))
        log_event("invite.created", actor_id=created_by, email=email,
                  detail="role=%s units=%d" % (global_role or "-", len(units)), con=con)
        con.commit()
        row = dict(con.execute("SELECT * FROM invites WHERE id=?", (iid,)).fetchone())
    finally:
        con.close()
    return row, tok


def invite_url(token):
    return "%s/invite/%s" % (SITE_BASE_URL, token)


def live_invite_for(email):
    """The current unconsumed, unrevoked, unexpired invite for an address, if any.

    Used by bootstrap so a container restart does not invalidate a link somebody
    is already holding.
    """
    con = connect()
    try:
        r = con.execute(
            "SELECT * FROM invites WHERE email=? AND consumed_at IS NULL"
            " AND revoked_at IS NULL AND expires_at > ?"
            " ORDER BY created_at DESC LIMIT 1", ((email or "").lower(), _now())).fetchone()
        return dict(r) if r else None
    finally:
        con.close()


def peek_invite(token):
    """Read an invite by raw token without consuming it. None if unusable.

    Expired, revoked and consumed all return None on purpose. Telling the
    difference tells a stranger which addresses are real.
    """
    con = connect()
    try:
        r = con.execute(
            "SELECT * FROM invites WHERE token_hash=? AND consumed_at IS NULL"
            " AND revoked_at IS NULL AND expires_at > ?",
            (_hash_token(token), _now())).fetchone()
        return dict(r) if r else None
    finally:
        con.close()


def consume_invite(token, full_name, password, preferred_name=None, phone=None, ip=None):
    """Create the person and burn the invite in one transaction.

    Everything below happens or nothing does. A half-applied invite would leave
    an account with no memberships and a token that still looks live.
    """
    full_name = (full_name or "").strip()
    if not full_name:
        raise IdentityError(422, "name is required")
    pw_hash = hash_password(password)

    con = connect()
    try:
        con.execute("BEGIN IMMEDIATE")
        r = con.execute(
            "SELECT * FROM invites WHERE token_hash=? AND consumed_at IS NULL"
            " AND revoked_at IS NULL AND expires_at > ?",
            (_hash_token(token), _now())).fetchone()
        if not r:
            con.rollback()
            raise IdentityError(410, "this invitation is no longer valid")
        inv = dict(r)

        if con.execute("SELECT 1 FROM people WHERE email=?", (inv["email"],)).fetchone():
            con.rollback()
            raise IdentityError(409, "an account already exists for this address")

        pid = str(uuid.uuid4())
        con.execute(
            "INSERT INTO people (id, email, full_name, preferred_name, phone, password_hash,"
            " global_role, registered_adult, created_at, created_by)"
            " VALUES (?,?,?,?,?,?,?,0,?,?)",
            (pid, inv["email"], full_name, (preferred_name or "").strip() or None,
             (phone or "").strip() or None, pw_hash, inv["global_role"], _now(),
             inv["created_by"]))
        for u in json.loads(inv["units"] or "[]"):
            con.execute(
                "INSERT INTO memberships (person_id, unit_id, role, title, created_at, created_by)"
                " VALUES (?,?,?,?,?,?)",
                (pid, u["unit_id"], u["role"], u.get("title"), _now(), inv["created_by"]))
        con.execute("UPDATE invites SET consumed_at=?, person_id=? WHERE id=?",
                    (_now(), pid, inv["id"]))
        log_event("invite.consumed", person_id=pid, email=inv["email"], ip=ip, con=con)
        con.commit()
    finally:
        con.close()
    return get_person(pid)


# ---------------------------------------------------------------------------
# Sessions
# ---------------------------------------------------------------------------
#
# Rows, not signed cookies. A leader stepping down has to be revocable now
# rather than at token expiry, and "sign out everywhere" has to be possible.

def safe_next(value, default="/account"):
    """Where to send someone after login. Only a same-origin relative path
    survives: a single leading slash, no scheme, no protocol-relative `//`,
    no backslash or control character that a browser might normalise into
    one. Anything else falls back to default. This is what keeps /login from
    being an open redirect the moment it starts honouring `next`."""
    v = (value or "").strip()
    if not v.startswith("/") or v.startswith("//") or v.startswith("/\\"):
        return default
    if any(ord(c) < 32 or c in "\\" for c in v):
        return default
    return v


def start_session(person_id, ip=None, user_agent=None):
    tok = secrets.token_urlsafe(32)
    con = connect()
    try:
        con.execute(
            "INSERT INTO sessions (id, token_hash, person_id, created_at, expires_at,"
            " last_seen_at, ip, user_agent, revoked_at) VALUES (?,?,?,?,?,?,?,?,NULL)",
            (str(uuid.uuid4()), _hash_token(tok), person_id, _now(),
             _plus(days=SESSION_ABSOLUTE_DAYS), _now(), ip, (user_agent or "")[:200]))
        con.execute("UPDATE people SET last_login_at=? WHERE id=?", (_now(), person_id))
        con.commit()
    finally:
        con.close()
    return tok


def session_person(token):
    """The person behind a session cookie, or None.

    Enforces both bounds: an absolute expiry and an idle timeout. Touches
    last_seen_at on success, so the idle clock tracks use rather than login.
    """
    if not token:
        return None
    con = connect()
    try:
        r = con.execute(
            "SELECT * FROM sessions WHERE token_hash=? AND revoked_at IS NULL",
            (_hash_token(token),)).fetchone()
        if not r:
            return None
        now = _now()
        if r["expires_at"] <= now:
            return None
        idle_cutoff = (datetime.datetime.now(datetime.timezone.utc)
                       - datetime.timedelta(hours=SESSION_IDLE_HOURS)).isoformat(timespec="seconds")
        if r["last_seen_at"] <= idle_cutoff:
            return None
        p = _person_row(con, con.execute(
            "SELECT * FROM people WHERE id=?", (r["person_id"],)).fetchone())
        if not p or p.get("disabled_at"):
            return None
        con.execute("UPDATE sessions SET last_seen_at=? WHERE id=?", (now, r["id"]))
        con.commit()
        return p
    finally:
        con.close()


def end_session(token):
    con = connect()
    try:
        con.execute("UPDATE sessions SET revoked_at=? WHERE token_hash=? AND revoked_at IS NULL",
                    (_now(), _hash_token(token)))
        con.commit()
    finally:
        con.close()


def end_all_sessions(person_id):
    con = connect()
    try:
        cur = con.execute("UPDATE sessions SET revoked_at=? WHERE person_id=? AND revoked_at IS NULL",
                          (_now(), person_id))
        con.commit()
        return cur.rowcount
    finally:
        con.close()


# ---------------------------------------------------------------------------
# Login
# ---------------------------------------------------------------------------

def recent_failures(email):
    cutoff = (datetime.datetime.now(datetime.timezone.utc)
              - datetime.timedelta(minutes=LOGIN_WINDOW_MINUTES)).isoformat(timespec="seconds")
    con = connect()
    try:
        return con.execute(
            "SELECT COUNT(*) c FROM auth_events WHERE kind='login.failed' AND email=? AND at > ?",
            ((email or "").lower(), cutoff)).fetchone()["c"]
    finally:
        con.close()


def authenticate(email, password, ip=None, user_agent=None):
    """Returns (person, session_token). Raises IdentityError on any failure.

    One message for every failure mode. Distinguishing "no such account" from
    "wrong password" enumerates the address list of a volunteer organisation.
    """
    email = (email or "").strip().lower()
    if recent_failures(email) >= LOGIN_MAX_FAILURES:
        log_event("login.throttled", email=email, ip=ip)
        raise IdentityError(429, "too many attempts. Wait %d minutes and try again."
                            % LOGIN_WINDOW_MINUTES)

    con = connect()
    try:
        r = con.execute("SELECT * FROM people WHERE email=?", (email,)).fetchone()
        stored = r["password_hash"] if r else None
    finally:
        con.close()

    ok = bool(stored) and verify_password(password or "", stored)
    if not ok or (r and r["disabled_at"]):
        log_event("login.failed", person_id=(r["id"] if r else None), email=email, ip=ip,
                  detail="disabled" if (r and r["disabled_at"]) else "bad credentials")
        raise IdentityError(401, "that email and password do not match an account")

    tok = start_session(r["id"], ip=ip, user_agent=user_agent)
    log_event("login.ok", person_id=r["id"], email=email, ip=ip)
    return get_person(r["id"]), tok


# ---------------------------------------------------------------------------
# Bootstrap
# ---------------------------------------------------------------------------
#
# The first owner has nobody to invite them, so the seed is a boot action and
# never a UI one. It goes inert the moment any person exists, and stays in the
# code as a disaster path rather than being deleted.
#
# A live invite is REUSED across restarts. Minting a fresh token on every boot
# would invalidate the link the recipient is already holding, every deploy.

def bootstrap():
    """Returns (url, minted) or (None, False). Never raises: a boot path that
    can take the site down over a misconfigured email address is worse than
    one that logs and carries on."""
    try:
        if people_count() > 0:
            return None, False
        if not ADMIN_BOOTSTRAP_EMAIL:
            print("identity: no people and ADMIN_BOOTSTRAP_EMAIL is unset, "
                  "nobody can sign in", flush=True)
            return None, False

        live = live_invite_for(ADMIN_BOOTSTRAP_EMAIL)
        if live:
            print("identity: owner invite already outstanding for %s, expires %s"
                  % (ADMIN_BOOTSTRAP_EMAIL, live["expires_at"]), flush=True)
            return None, False

        _, tok = create_invite(ADMIN_BOOTSTRAP_EMAIL, global_role="owner",
                              created_by="bootstrap")
        url = invite_url(tok)
        print("identity: BOOTSTRAP OWNER INVITE for %s -> %s"
              % (ADMIN_BOOTSTRAP_EMAIL, url), flush=True)
        return url, True
    except Exception as e:
        print("identity: bootstrap failed: %s" % e, flush=True)
        return None, False


# ---------------------------------------------------------------------------
# Site settings
# ---------------------------------------------------------------------------
#
# Typed key-value pairs for the handful of site-wide values that change more
# often than the code does. SETTINGS_KEYS is the whole contract: a key not in
# it cannot be written, and a key absent from the table - or holding a value
# its checker no longer accepts - falls back to the code default. A mangled
# row can make the site stale, never make it crash.
#
# Setting a value to null clears the row, which IS the fallback: there is no
# stored-but-empty state to reason about.

def _setting_text(v):
    v = str(v).strip()
    if not v:
        raise IdentityError(422, "value must not be blank; send null to clear to the default")
    return v


def _setting_https_url(v):
    v = _setting_text(v)
    if not v.startswith("https://") or any(c in v for c in " \"'<>"):
        raise IdentityError(422, "value must be a plain https:// URL")
    return v


def _setting_choice(*allowed):
    def check(v):
        v = _setting_text(v).lower()
        if v not in allowed:
            raise IdentityError(422, "value must be one of %s" % ", ".join(allowed))
        return v
    return check


# key -> (code default, checker)
SETTINGS_KEYS = {
    "nearby_source_name": ("Continental District unit list", _setting_text),
    "nearby_source_url": ("https://tinyurl.com/ContinentalScouts", _setting_https_url),
    # Where API keys may be used from. "lan" is the historical rule, kept as
    # the default. Sessions are never restricted (the console IS a session
    # from anywhere) and the break-glass token is ALWAYS LAN-only - that one
    # is not a choice, so it is not a setting. See admin_api.client_is_lan.
    "api_keys_from": ("lan", _setting_choice("lan", "anywhere")),
}


def get_setting(key):
    """The effective value: the stored one if present and still valid, else
    the code default. Unknown keys are a programming error and raise."""
    default, check = SETTINGS_KEYS[key]
    con = connect()
    try:
        r = con.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone()
    finally:
        con.close()
    if not r:
        return default
    try:
        return check(r["value"])
    except IdentityError:
        return default


def all_settings():
    """Every known key with its effective value, for the settings screen.
    Rows for keys the registry no longer knows are not shown - they are dead
    weight, not settings."""
    con = connect()
    try:
        stored = {r["key"]: dict(r) for r in con.execute("SELECT * FROM settings")}
    finally:
        con.close()
    out = []
    for key, (default, _check) in SETTINGS_KEYS.items():
        row = stored.get(key)
        out.append({
            "key": key,
            "value": get_setting(key),
            "default": default,
            "is_set": row is not None,
            "updated_at": row["updated_at"] if row else None,
            "updated_by": row["updated_by"] if row else None,
        })
    return out


def set_setting(key, value, actor=None):
    """Write one setting, or clear it back to the default with value=null."""
    if key not in SETTINGS_KEYS:
        raise IdentityError(422, "unknown setting %r. Known keys: %s"
                            % (key, ", ".join(sorted(SETTINGS_KEYS))))
    default, check = SETTINGS_KEYS[key]
    con = connect()
    try:
        if value is None:
            con.execute("DELETE FROM settings WHERE key=?", (key,))
        else:
            value = check(value)
            con.execute(
                "INSERT INTO settings (key, value, updated_at, updated_by)"
                " VALUES (?,?,?,?) ON CONFLICT(key) DO UPDATE SET"
                " value=excluded.value, updated_at=excluded.updated_at,"
                " updated_by=excluded.updated_by",
                (key, value, _now(), actor))
        con.commit()
    finally:
        con.close()
    return [s for s in all_settings() if s["key"] == key][0]


# ---------------------------------------------------------------------------
# API keys (P3)
# ---------------------------------------------------------------------------
#
# For scripts, not for people signing in every Tuesday. A key is the person
# who minted it, narrowed to the scopes they chose. It cannot hold more than
# they hold, it cannot mint further keys, and it dies with them: disabling a
# person disables their keys through can(), with no separate flag to forget.

KEY_PREFIX = "gls73_"
KEY_PREFIX_SHOWN = len(KEY_PREFIX) + 6          # "gls73_ab12cd" - enough to tell keys apart
KEY_MAX_DAYS = 365

# Scopes a key may never carry, whatever the owner holds. Minting keys from a
# key is a loop; the rest are owner powers that belong to a person at a screen.
KEY_UNSCOPABLE = {"account:self", "apikeys:own", "api:docs", "history:read", "people:manage",
                  "secrets:rotate", "people:invite_leader", "people:invite_admin",
                  "roster:write"}   # minors' names never ride a script key


def scopable_caps(person):
    """The scopes this person may put on a key: what they hold, minus the
    ones a key may never carry."""
    return sorted(effective_caps(person) - KEY_UNSCOPABLE)


def _key_row(r):
    d = dict(r)
    d.pop("key_hash", None)
    d["scopes"] = json.loads(d["scopes"]) if isinstance(d.get("scopes"), str) else (d.get("scopes") or [])
    now = _now()
    if d.get("revoked_at"):
        d["state"] = "revoked"
    elif d.get("expires_at") and d["expires_at"] <= now:
        d["state"] = "expired"
    else:
        d["state"] = "active"
    return d


def mint_api_key(person, label, scopes, expires_days=None):
    """Create a key for `person`. Returns (full_key, row). The full key is
    returned exactly once and never stored."""
    if not person or person.get("disabled_at"):
        raise IdentityError(403, "keys belong to a signed-in, enabled person")
    if person.get("key_scopes") is not None:
        raise IdentityError(403, "a key cannot mint keys")
    label = (label or "").strip()
    if not label or len(label) > 60:
        raise IdentityError(422, "label is required, up to 60 characters")
    if not isinstance(scopes, (list, tuple, set)) or not scopes:
        raise IdentityError(422, "scopes is required: a non-empty list")
    scopes = sorted({str(s).strip() for s in scopes if str(s).strip()})
    allowed = set(scopable_caps(person))
    bad = [s for s in scopes if s not in allowed]
    if bad:
        raise IdentityError(422, "scopes not available to you or not allowed on a key: %s. "
                                 "Choose from %s" % (", ".join(bad), ", ".join(sorted(allowed))))
    expires_at = None
    if expires_days not in (None, ""):
        try:
            days = int(expires_days)
        except (TypeError, ValueError):
            raise IdentityError(422, "expires_days must be a whole number of days")
        if not 1 <= days <= KEY_MAX_DAYS:
            raise IdentityError(422, "expires_days must be 1 to %d" % KEY_MAX_DAYS)
        expires_at = (datetime.datetime.now(datetime.timezone.utc)
                      + datetime.timedelta(days=days)).isoformat(timespec="seconds")
    full = KEY_PREFIX + secrets.token_urlsafe(32)
    kid = str(uuid.uuid4())
    con = connect()
    try:
        con.execute(
            "INSERT INTO api_keys (id, person_id, label, prefix, key_hash, scopes, created_at,"
            " last_used_at, expires_at, revoked_at) VALUES (?,?,?,?,?,?,?,NULL,?,NULL)",
            (kid, person["id"], label, full[:KEY_PREFIX_SHOWN], _hash_token(full),
             json.dumps(scopes), _now(), expires_at))
        log_event("apikey.minted", person_id=person["id"], actor_id=person["id"],
                  email=person.get("email"), detail="%s [%s] %s" % (label, ", ".join(scopes), kid), con=con)
        con.commit()
        r = con.execute("SELECT * FROM api_keys WHERE id=?", (kid,)).fetchone()
    finally:
        con.close()
    return full, _key_row(r)


def list_api_keys(person_id):
    con = connect()
    try:
        return [_key_row(r) for r in con.execute(
            "SELECT * FROM api_keys WHERE person_id=? ORDER BY created_at DESC", (person_id,))]
    finally:
        con.close()


def revoke_api_key(person, key_id):
    """Revoke one of the person's OWN keys. None if it is not theirs (the
    caller answers 404, never 403 - a foreign key id is not confirmed)."""
    con = connect()
    try:
        r = con.execute("SELECT * FROM api_keys WHERE id=? AND person_id=?",
                        (key_id, person["id"])).fetchone()
        if not r:
            return None
        if r["revoked_at"]:
            raise IdentityError(409, "already revoked")
        con.execute("UPDATE api_keys SET revoked_at=? WHERE id=?", (_now(), key_id))
        log_event("apikey.revoked", person_id=person["id"], actor_id=person["id"],
                  email=person.get("email"), detail="%s %s" % (r["label"], key_id), con=con)
        con.commit()
        return _key_row(con.execute("SELECT * FROM api_keys WHERE id=?", (key_id,)).fetchone())
    finally:
        con.close()


def api_key_person(bearer):
    """The person behind an Authorization: Bearer value, narrowed to the key's
    scopes, or None. Touches last_used_at. Never raises: an unknown, revoked,
    expired or foreign-prefixed value is simply not a person."""
    if not bearer or not str(bearer).startswith(KEY_PREFIX):
        return None
    con = connect()
    try:
        r = con.execute("SELECT * FROM api_keys WHERE key_hash=? AND revoked_at IS NULL",
                        (_hash_token(str(bearer)),)).fetchone()
        if not r:
            return None
        now = _now()
        if r["expires_at"] and r["expires_at"] <= now:
            return None
        p = _person_row(con, con.execute("SELECT * FROM people WHERE id=?", (r["person_id"],)).fetchone())
        if not p or p.get("disabled_at"):
            return None
        con.execute("UPDATE api_keys SET last_used_at=? WHERE id=?", (now, r["id"]))
        con.commit()
        p["key_scopes"] = set(json.loads(r["scopes"]))
        p["key_id"] = r["id"]
        p["key_prefix"] = r["prefix"]
        p["capabilities"] = sorted(effective_caps(p) & p["key_scopes"])
        return p
    finally:
        con.close()


# ---------------------------------------------------------------------------
# History. auth_events is the one action log: logins and invites since P0,
# keys since P3, calendar since P4, and every P2 write since 2026-09-04.
# One table, one screen. Not per-row history tables, and not the dropped
# generic audit table with workflow columns.
# ---------------------------------------------------------------------------

def list_events(limit=200, kind_prefix=None, before=None):
    con = connect()
    try:
        sql = "SELECT e.*, p.email AS actor_email FROM auth_events e LEFT JOIN people p ON p.id = e.actor_id"
        where, vals = [], []
        if kind_prefix:
            where.append("e.kind LIKE ?"); vals.append(kind_prefix + "%")
        if before:
            where.append("e.at < ?"); vals.append(before)
        if where:
            sql += " WHERE " + " AND ".join(where)
        # rowid breaks ties: `at` is second-resolution and a screen can write
        # several rows in one second.
        sql += " ORDER BY e.at DESC, e.rowid DESC LIMIT ?"
        vals.append(max(1, min(int(limit), 500)))
        return [dict(r) for r in con.execute(sql, vals)]
    finally:
        con.close()


# ---------------------------------------------------------------------------
# People management (post-P4, 2026-09-04). Who exists, what they hold, and
# the levers: invite, change roles, disable, enable, reset a password.
# Rules that must hold whatever the caller does live here, not in routes:
# never zero owners, never disable yourself, an admin cannot grant what they
# do not hold.
# ---------------------------------------------------------------------------

RESET_TTL_HOURS = 24


def list_people(include_disabled=True):
    con = connect()
    try:
        sql = "SELECT * FROM people"
        if not include_disabled:
            sql += " WHERE disabled_at IS NULL"
        sql += " ORDER BY COALESCE(global_role, 'z'), email"
        rows = [_person_row(con, r) for r in con.execute(sql)]
        for p in rows:
            p["active_sessions"] = con.execute(
                "SELECT count(*) FROM sessions WHERE person_id=? AND revoked_at IS NULL AND expires_at > ?",
                (p["id"], _now())).fetchone()[0]
            p["active_keys"] = con.execute(
                "SELECT count(*) FROM api_keys WHERE person_id=? AND revoked_at IS NULL", (p["id"],)).fetchone()[0]
        return rows
    finally:
        con.close()


def list_open_invites():
    """Unconsumed, unrevoked invites, expired ones included and marked, so an
    admin can see who has not finished signing up."""
    con = connect()
    try:
        now = _now()
        out = []
        for r in con.execute("SELECT * FROM invites WHERE consumed_at IS NULL AND revoked_at IS NULL"
                             " ORDER BY created_at DESC"):
            d = dict(r); d.pop("token_hash", None)
            d["units"] = json.loads(d.get("units") or "[]")
            d["expired"] = d["expires_at"] <= now
            out.append(d)
        return out
    finally:
        con.close()


def _owner_count(con, excluding=None):
    sql = "SELECT count(*) FROM people WHERE global_role='owner' AND disabled_at IS NULL"
    vals = ()
    if excluding:
        sql += " AND id<>?"; vals = (excluding,)
    return con.execute(sql, vals).fetchone()[0]


def grantable_roles(actor):
    """What this actor may hand out. people:manage (owner) grants anything;
    people:invite_admin grants admin and below; people:invite_leader grants
    unit roles only."""
    if can(actor, "people:manage"):
        return {"global": ["owner", "admin"], "unit": list(UNIT_ROLES)}
    if can(actor, "people:invite_admin"):
        return {"global": ["admin"], "unit": list(UNIT_ROLES)}
    if can(actor, "people:invite_leader"):
        return {"global": [], "unit": list(UNIT_ROLES)}
    return {"global": [], "unit": []}


def _check_grant(actor, global_role, units):
    g = grantable_roles(actor)
    if global_role and global_role not in g["global"]:
        raise IdentityError(403, "you cannot grant the %s role" % global_role)
    for u in units or []:
        if u.get("role") not in g["unit"]:
            raise IdentityError(403, "you cannot grant the unit role %r" % u.get("role"))


def invite_person(actor, email, global_role=None, units=None):
    """create_invite, gated on what the actor may grant. Returns (row, url)."""
    _check_grant(actor, global_role, units)
    if get_person_by_email(email):
        raise IdentityError(409, "that address already has an account; reset its password or change its roles instead")
    row, tok = create_invite(email, global_role=global_role, units=units, created_by=actor["id"])
    return row, invite_url(tok)


def revoke_invite(actor, invite_id):
    con = connect()
    try:
        r = con.execute("SELECT * FROM invites WHERE id=? AND consumed_at IS NULL AND revoked_at IS NULL",
                        (invite_id,)).fetchone()
        if not r:
            return None
        con.execute("UPDATE invites SET revoked_at=? WHERE id=?", (_now(), invite_id))
        log_event("invite.revoked", actor_id=actor["id"], email=r["email"], con=con)
        con.commit()
        return dict(con.execute("SELECT * FROM invites WHERE id=?", (invite_id,)).fetchone())
    finally:
        con.close()


def set_roles(actor, person_id, global_role=None, units=None):
    """Replace a person's global role and unit memberships in one transaction.
    `units` is the full desired list [{unit_id, role, title}]; absent
    memberships are removed. Requires people:manage. Never leaves zero owners."""
    if not can(actor, "people:manage"):
        raise IdentityError(403, "changing roles needs people:manage")
    if global_role and global_role not in GLOBAL_ROLES:
        raise IdentityError(422, "global_role must be one of %s, or null" % (GLOBAL_ROLES,))
    units = units or []
    for u in units:
        if u.get("role") not in UNIT_ROLES:
            raise IdentityError(422, "unit role must be one of %s" % (UNIT_ROLES,))
        if not get_unit(u.get("unit_id") or ""):
            raise IdentityError(422, "unknown unit %r" % (u.get("unit_id"),))
    con = connect()
    try:
        before = _person_row(con, con.execute("SELECT * FROM people WHERE id=?", (person_id,)).fetchone())
        if not before:
            return None
        if before.get("global_role") == "owner" and global_role != "owner" and _owner_count(con, excluding=person_id) == 0:
            raise IdentityError(409, "that is the last owner; make someone else owner first")
        con.execute("UPDATE people SET global_role=? WHERE id=?", (global_role or None, person_id))
        con.execute("DELETE FROM memberships WHERE person_id=?", (person_id,))
        for u in units:
            con.execute("INSERT INTO memberships (person_id, unit_id, role, title, created_at, created_by)"
                        " VALUES (?,?,?,?,?,?)", (person_id, u["unit_id"], u["role"], (u.get("title") or "").strip() or None,
                                                  _now(), actor["id"]))
        after = _person_row(con, con.execute("SELECT * FROM people WHERE id=?", (person_id,)).fetchone())
        log_event("person.roles", person_id=person_id, actor_id=actor["id"], email=before["email"],
                  detail="global %s -> %s; units %s -> %s" % (
                      before.get("global_role") or "-", global_role or "-",
                      ", ".join("%s:%s" % (m["slug"], m["role"]) for m in before["memberships"]) or "-",
                      ", ".join("%s:%s" % (m["slug"], m["role"]) for m in after["memberships"]) or "-"), con=con)
        con.commit()
        return after
    finally:
        con.close()


def disable_person(actor, person_id, reason=None):
    """Disable: sessions end now, keys stop through can(), documents close.
    The row stays. Cannot disable yourself or the last owner."""
    if not can(actor, "people:manage"):
        raise IdentityError(403, "disabling a person needs people:manage")
    if person_id == actor["id"]:
        raise IdentityError(409, "you cannot disable yourself")
    con = connect()
    try:
        p = _person_row(con, con.execute("SELECT * FROM people WHERE id=?", (person_id,)).fetchone())
        if not p:
            return None
        if p.get("disabled_at"):
            raise IdentityError(409, "already disabled")
        if p.get("global_role") == "owner" and _owner_count(con, excluding=person_id) == 0:
            raise IdentityError(409, "that is the last owner")
        con.execute("UPDATE people SET disabled_at=?, disabled_reason=? WHERE id=?",
                    (_now(), (reason or "").strip() or None, person_id))
        con.execute("UPDATE sessions SET revoked_at=? WHERE person_id=? AND revoked_at IS NULL", (_now(), person_id))
        log_event("person.disabled", person_id=person_id, actor_id=actor["id"], email=p["email"],
                  detail=(reason or "").strip() or None, con=con)
        con.commit()
        return _person_row(con, con.execute("SELECT * FROM people WHERE id=?", (person_id,)).fetchone())
    finally:
        con.close()


def enable_person(actor, person_id):
    if not can(actor, "people:manage"):
        raise IdentityError(403, "enabling a person needs people:manage")
    con = connect()
    try:
        p = _person_row(con, con.execute("SELECT * FROM people WHERE id=?", (person_id,)).fetchone())
        if not p:
            return None
        if not p.get("disabled_at"):
            raise IdentityError(409, "not disabled")
        con.execute("UPDATE people SET disabled_at=NULL, disabled_reason=NULL WHERE id=?", (person_id,))
        log_event("person.enabled", person_id=person_id, actor_id=actor["id"], email=p["email"], con=con)
        con.commit()
        return _person_row(con, con.execute("SELECT * FROM people WHERE id=?", (person_id,)).fetchone())
    finally:
        con.close()


def create_reset(actor, person_id):
    """Mint a one-time password-reset link for a person. An owner may reset
    anyone; an admin may reset anyone who is not an owner. The token is
    returned once. Reissue revokes the prior link."""
    con = connect()
    try:
        p = _person_row(con, con.execute("SELECT * FROM people WHERE id=?", (person_id,)).fetchone())
        if not p:
            return None
        if p.get("disabled_at"):
            raise IdentityError(409, "person is disabled; enable them first")
        if p.get("global_role") == "owner" and not can(actor, "people:manage"):
            raise IdentityError(403, "only an owner can reset an owner's password")
        if not (can(actor, "people:manage") or can(actor, "people:invite_admin")):
            raise IdentityError(403, "resetting a password needs people:invite_admin")
        tok = secrets.token_urlsafe(32)
        con.execute("UPDATE password_resets SET revoked_at=? WHERE person_id=? AND consumed_at IS NULL AND revoked_at IS NULL",
                    (_now(), person_id))
        con.execute("INSERT INTO password_resets (id, token_hash, person_id, created_at, created_by, expires_at)"
                    " VALUES (?,?,?,?,?,?)", (str(uuid.uuid4()), _hash_token(tok), person_id, _now(), actor["id"],
                                             _plus(hours=RESET_TTL_HOURS)))
        log_event("password.reset_issued", person_id=person_id, actor_id=actor["id"], email=p["email"], con=con)
        con.commit()
    finally:
        con.close()
    return "%s/reset/%s" % (SITE_BASE_URL, tok)


def peek_reset(token):
    """The person behind a live reset token, or None. Expired, consumed,
    revoked and never-existed are all None, deliberately."""
    con = connect()
    try:
        r = con.execute("SELECT * FROM password_resets WHERE token_hash=? AND consumed_at IS NULL"
                        " AND revoked_at IS NULL AND expires_at > ?", (_hash_token(token or ""), _now())).fetchone()
        if not r:
            return None
        p = _person_row(con, con.execute("SELECT * FROM people WHERE id=?", (r["person_id"],)).fetchone())
        if not p or p.get("disabled_at"):
            return None
        return p
    finally:
        con.close()


def consume_reset(token, new_password, ip=None):
    """Set the password, burn the token, end every session, in one
    transaction. Returns the person."""
    pw_hash = hash_password(new_password)
    con = connect()
    try:
        r = con.execute("SELECT * FROM password_resets WHERE token_hash=? AND consumed_at IS NULL"
                        " AND revoked_at IS NULL AND expires_at > ?", (_hash_token(token or ""), _now())).fetchone()
        if not r:
            raise IdentityError(410, "this reset link is no longer valid")
        p = con.execute("SELECT * FROM people WHERE id=?", (r["person_id"],)).fetchone()
        if not p or p["disabled_at"]:
            raise IdentityError(410, "this reset link is no longer valid")
        con.execute("UPDATE people SET password_hash=? WHERE id=?", (pw_hash, p["id"]))
        con.execute("UPDATE password_resets SET consumed_at=? WHERE id=?", (_now(), r["id"]))
        con.execute("UPDATE sessions SET revoked_at=? WHERE person_id=? AND revoked_at IS NULL", (_now(), p["id"]))
        log_event("password.reset", person_id=p["id"], email=p["email"], ip=ip, con=con)
        con.commit()
        return _person_row(con, con.execute("SELECT * FROM people WHERE id=?", (p["id"],)).fetchone())
    finally:
        con.close()


def change_password(person_id, current, new, ip=None):
    """Self-service. Needs the current password; ends the OTHER sessions."""
    con = connect()
    try:
        r = con.execute("SELECT * FROM people WHERE id=?", (person_id,)).fetchone()
        if not r or not verify_password(current or "", r["password_hash"] or ""):
            raise IdentityError(403, "current password is wrong")
        pw_hash = hash_password(new)
        con.execute("UPDATE people SET password_hash=? WHERE id=?", (pw_hash, person_id))
        log_event("password.changed", person_id=person_id, actor_id=person_id, email=r["email"], ip=ip, con=con)
        con.commit()
    finally:
        con.close()


def update_own_details(person_id, full_name=None, preferred_name=None, phone=None):
    full_name = (full_name or "").strip()
    if not full_name:
        raise IdentityError(422, "full name is required")
    con = connect()
    try:
        con.execute("UPDATE people SET full_name=?, preferred_name=?, phone=? WHERE id=?",
                    (full_name[:80], (preferred_name or "").strip()[:40] or None, (phone or "").strip()[:30] or None, person_id))
        con.commit()
        return _person_row(con, con.execute("SELECT * FROM people WHERE id=?", (person_id,)).fetchone())
    finally:
        con.close()