rfc-app/backend/app/invites.py

"""§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 "<first> <last>" (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
    ]