"""§6.1 / v0.17.0: admin-create user with role + invite email (roadmap item #16). Distinguishes from the v0.8.0 self-serve beta-access flow: * **Self-serve (v0.8.0)** — anyone with an email can request OTC sign-in; a fresh `users` row lands in `permission_state='pending'`; an admin grants or revokes via the v0.9.0 user-management page. * **Admin-create (v0.17.0)** — an admin types first/last/email/role *before* the invitee has signed in. The framework provisions the `users` row with the chosen role and `permission_state='granted'` (the admin's hand is the grant) and `last_seen_at IS NULL` as the "invited but not yet arrived" discriminator. An invite-token row lands in `user_invite_tokens`; the admin's chosen `custom_message` (if any) rides in the email body alongside the claim link. * **Claim flow** — the invitee clicks the link, which lands them at `/invites/claim?token=…`. The page POSTs `/api/invites/claim` with the token. The framework verifies the token (not expired, not claimed, hash matches), marks the row claimed, signs the user in, and returns a payload telling the frontend whether to route to passcode-set (if v0.10.0 passcode flow is in play and the user has no passcode yet) or to `/`. **No OTC roundtrip** — clicking the unique token in the email is itself proof of email control, per the roadmap. This is the intentional UX shortcut for first sign-in; subsequent sign-ins use the standard OTC / passcode paths. The shape: * `create_invite(...)` — provision the invitee `users` row + the `user_invite_tokens` row, return the raw token for the admin endpoint to put in the outbound email link. * `claim(raw_token)` — validate the token, mark it claimed, return the `SessionUser` the endpoint signs in. Distinguishes the failure modes (`expired`, `claimed`, `unknown`, `invalid`) so the endpoint can map them to HTTP 410 vs HTTP 404 cleanly. * `list_pending_invites()` — return active invites for the admin listing surface. Filters out claimed + expired rows so the surface only shows live invites. Token shape: opaque DB token (256 bits of CSPRNG entropy via `secrets.token_urlsafe(32)`), bcrypt-hashed at rest. Opaque chosen over JWT because revocation is then a single SQL UPDATE — a JWT would be stateless but harder to invalidate, and admin-issued invites are exactly the kind of thing an admin should be able to yank back. The raw token only ever lives in the outbound email link and the inbound claim body; server-side storage is the hash. TTL: hard-coded to 7 days via `INVITE_TOKEN_TTL_DAYS`. Env-var configurability is a §19.2 candidate — the constant is exposed here as a single point of edit if a deployment wants to override. The 500-char ceiling on `custom_message` is enforced at the Pydantic body level in `api_admin.py`; this module trusts what the endpoint hands it. """ from __future__ import annotations import logging import secrets from dataclasses import dataclass import bcrypt from . import db from .auth import SessionUser log = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Tunables — intentionally hard-coded in v0.17.0 (§19.2 candidate to env-ify). # --------------------------------------------------------------------------- INVITE_TOKEN_TTL_DAYS = 7 # 256 bits of CSPRNG entropy. `secrets.token_urlsafe(32)` yields ~43 # URL-safe characters; the bcrypt hash is what's stored, so the raw # token only ever lives in the outbound email link. TOKEN_BYTES = 32 # Free-text ceiling for the admin's optional custom message. Matched # at the Pydantic body bound in `api_admin.py`; mentioned here so the # bound is documented in one place. CUSTOM_MESSAGE_MAX_LENGTH = 500 # --------------------------------------------------------------------------- # Token + hash helpers (mirror device_trust.py shape) # --------------------------------------------------------------------------- def _new_token() -> str: return secrets.token_urlsafe(TOKEN_BYTES) def _hash(token: str) -> str: return bcrypt.hashpw(token.encode("utf-8"), bcrypt.gensalt()).decode("ascii") def _check(token: str, token_hash: str) -> bool: try: return bcrypt.checkpw(token.encode("utf-8"), token_hash.encode("ascii")) except (ValueError, TypeError): return False # --------------------------------------------------------------------------- # Create # --------------------------------------------------------------------------- @dataclass class CreateOutcome: """The shape returned from `create_invite`. `raw_token` is what the admin endpoint puts in the outbound email link; it never appears in storage. `invite_id` is the surrogate key for the admin's "invites I've sent" listing. `invited_user_id` is the freshly-provisioned `users` row id so the admin surface can join through to the user-management page. """ raw_token: str invite_id: int invited_user_id: int def create_invite( *, email: str, first_name: str, last_name: str, role: str, custom_message: str, created_by_admin_id: int, ) -> CreateOutcome: """Provision the invitee `users` row + the `user_invite_tokens` row. Caller (`api_admin.py`) is responsible for the admin-only auth check, the self-email refusal (422), and the duplicate-email refusal (409). This function trusts what it's handed and writes both rows transactionally — the v0.10.0 `passcode.py` / v0.11.0 `device_trust.py` helpers follow the same separation-of-concerns pattern. The invitee `users` row is provisioned with: * `permission_state='granted'` — the admin's hand is the grant; the v0.8.0 self-serve `pending` queue is for the other path. * `created_at` / `last_seen_at` — NOT set here, so both fall through to the column default `datetime('now')` (the column is `NOT NULL`; see `migrations/001_users_and_audit.sql` and the longer note below). The "invited but not yet arrived" state is therefore NOT carried on the user row — it is the existence of an unclaimed `user_invite_tokens` row, surfaced as the listing's `pending_invite` field. Consumers that want a truthful last-seen MUST treat a pending-invite row as never-seen rather than trusting `last_seen_at` (every real sign-in path stamps it to now, but an unclaimed invite has never hit one). * `gitea_id = NULL`, `gitea_login = NULL` — same as a v0.7.0 OTC-provisioned user; the OAuth identity is grandfathered if the user ever lands through that path. * `display_name` defaults to " " (or local-part of email if both are empty) so the user-management page reads a sensible label before the user has signed in. * `first_name` / `last_name` / `beta_request_reason` — the first two from the admin's typed values; reason stays blank (this user did not self-request access). """ email_clean = email.strip() first_clean = (first_name or "").strip() last_clean = (last_name or "").strip() display = " ".join(p for p in (first_clean, last_clean) if p).strip() if not display: display = email_clean.split("@", 1)[0] or email_clean # 1. Provision the invitee users row. The grant is the admin's # hand; no permission_events row is necessary for the grant itself # (we are not transitioning from pending → granted, we are landing # a fresh row directly into granted). # # Note on the "pending invite" discriminator: the brief floated # `first_sign_in_at NULL` / `last_seen_at NULL` as the marker the # admin user-management page reads off the row to render the # "(pending invite)" badge. The schema didn't cooperate — the # existing `users.last_seen_at` column is NOT NULL with a # `datetime('now')` default (see `migrations/001_users_and_audit.sql`), # and there is no `first_sign_in_at` column. Rather than introduce # a schema migration to add one (the brief explicitly said "likely # no `users` table changes"), the discriminator is the existence of # an active row in `user_invite_tokens` joined on `invited_user_id`. # The admin listing's `pending_invite` field joins through that # table; the claim flow stamps `claimed_at` on the invite row, # which clears the badge naturally. This shape keeps the # discriminator scoped to the v0.17.0 surface and avoids # double-tracking against an existing column. cur = db.conn().execute( """ INSERT INTO users ( gitea_id, gitea_login, email, display_name, avatar_url, role, permission_state, first_name, last_name ) VALUES (NULL, NULL, ?, ?, '', ?, 'granted', ?, ?) """, (email_clean, display, role, first_clean, last_clean), ) invited_user_id = cur.lastrowid # 2. Mint the token, hash it, write the invite row. raw = _new_token() h = _hash(raw) cur = db.conn().execute( f""" INSERT INTO user_invite_tokens ( email, role, first_name, last_name, custom_message, token_hash, expires_at, created_by_admin_id, invited_user_id ) VALUES (?, ?, ?, ?, ?, ?, datetime('now', '+{INVITE_TOKEN_TTL_DAYS} days'), ?, ?) """, ( email_clean, role, first_clean, last_clean, (custom_message or "").strip(), h, created_by_admin_id, invited_user_id, ), ) invite_id = cur.lastrowid return CreateOutcome( raw_token=raw, invite_id=invite_id, invited_user_id=invited_user_id, ) # --------------------------------------------------------------------------- # Claim # --------------------------------------------------------------------------- @dataclass class ClaimOutcome: """The result of `claim`. `user` is populated only on success. `reason` distinguishes the failure modes so the endpoint can return distinct HTTP statuses (HTTP 410 for expired/claimed — the token is dead; HTTP 400 for unknown/invalid — the request shape is wrong). """ ok: bool user: SessionUser | None reason: str # 'ok' | 'invalid' | 'unknown' | 'expired' | 'claimed' invite_id: int | None = None def claim(raw_token: str) -> ClaimOutcome: """Validate the presented token and consume it. Walks the active invite rows looking for a bcrypt hash match. Mirrors `device_trust.lookup`: bcrypt's per-row salt means we cannot SELECT by hash, but the set is small (a deployment's outstanding invites at any moment) and bcrypt is cheap on the order of milliseconds. On a hit: * Mark the row claimed (stamp `claimed_at = now`, `claimed_by_user_id = invited_user_id` — the admin's pre-provisioned row is the claimant). * Stamp `last_seen_at = now` on the user row so the v0.9.0 admin user-management page no longer shows "(pending invite)". * Return a populated `SessionUser` for the endpoint to sign in. On a miss: * `unknown` — no row matched. The token may have been forged or the invite was admin-revoked. * `expired` — row matched but `expires_at` is in the past. * `claimed` — row matched but `claimed_at` is non-NULL. The token was already consumed; the user must contact the admin for a fresh invite. * `invalid` — the token string itself was empty or unparseable. """ raw = (raw_token or "").strip() if not raw: return ClaimOutcome(ok=False, user=None, reason="invalid") rows = db.conn().execute( """ SELECT id, token_hash, expires_at, claimed_at, invited_user_id, role FROM user_invite_tokens ORDER BY id DESC """ ).fetchall() matched = None for row in rows: if _check(raw, row["token_hash"]): matched = row break if matched is None: return ClaimOutcome(ok=False, user=None, reason="unknown") if matched["claimed_at"] is not None: return ClaimOutcome( ok=False, user=None, reason="claimed", invite_id=matched["id"], ) expired = db.conn().execute( "SELECT datetime(?) < datetime('now') AS expired", (matched["expires_at"],), ).fetchone()["expired"] if expired: return ClaimOutcome( ok=False, user=None, reason="expired", invite_id=matched["id"], ) # Consume the row before signing in so a parallel claim of the same # token cannot double-sign-in. (Mirrors `otc.verify_code`'s consume- # before-provision shape.) db.conn().execute( """ UPDATE user_invite_tokens SET claimed_at = datetime('now'), claimed_by_user_id = invited_user_id WHERE id = ? """, (matched["id"],), ) # Stamp last_seen_at on the user row so the user's activity stamp # is current after the claim (mirroring otc.verify_code's # last-seen update on the provision path). The "(pending invite)" # badge's clear is driven by the invite row's `claimed_at` # transition above; this update is for the general user-listing's # recency ordering. db.conn().execute( "UPDATE users SET last_seen_at = datetime('now') WHERE id = ?", (matched["invited_user_id"],), ) user_row = db.conn().execute( """ SELECT id, gitea_id, gitea_login, email, display_name, avatar_url, role, permission_state FROM users WHERE id = ? """, (matched["invited_user_id"],), ).fetchone() if user_row is None: # The invitee user row was deleted between create_invite and # claim (shouldn't happen under the FK ON DELETE CASCADE — the # cascade would drop the invite row too — be defensive anyway). return ClaimOutcome( ok=False, user=None, reason="unknown", invite_id=matched["id"], ) return ClaimOutcome( ok=True, user=SessionUser( user_id=user_row["id"], gitea_id=user_row["gitea_id"] or 0, gitea_login=user_row["gitea_login"] or "", display_name=user_row["display_name"], email=user_row["email"] or "", avatar_url=user_row["avatar_url"] or "", role=user_row["role"], permission_state=user_row["permission_state"] or "granted", ), reason="ok", invite_id=matched["id"], ) # --------------------------------------------------------------------------- # List pending invites — for the admin's review surface # --------------------------------------------------------------------------- @dataclass class PendingInviteRow: """The shape the `GET /api/admin/users/invites` endpoint returns. Note the absence of `token_hash` — the hash is structurally private, and the surface has no use for it. The raw token is also not on the listing; it lives only in the email link. """ id: int email: str role: str first_name: str last_name: str custom_message: str created_at: str expires_at: str created_by_admin_id: int invited_user_id: int def list_pending_invites() -> list[PendingInviteRow]: """Active invites (not claimed, not expired), freshest first. The admin's "I sent these but they haven't been claimed yet" view. Filters mirror the `device_trust.list_for_user` shape: the surface only shows live records the framework would actually accept on a presented token. """ rows = db.conn().execute( """ SELECT id, email, role, first_name, last_name, custom_message, created_at, expires_at, created_by_admin_id, invited_user_id FROM user_invite_tokens WHERE claimed_at IS NULL AND datetime(expires_at) > datetime('now') ORDER BY created_at DESC, id DESC """ ).fetchall() return [ PendingInviteRow( id=row["id"], email=row["email"], role=row["role"], first_name=row["first_name"] or "", last_name=row["last_name"] or "", custom_message=row["custom_message"] or "", created_at=row["created_at"], expires_at=row["expires_at"], created_by_admin_id=row["created_by_admin_id"], invited_user_id=row["invited_user_id"], ) for row in rows ]