invites.py

Rama master · réplica de lectura del repositorio original.

"""Invite codes: the hosted beta's registration gate.

An invite code is a random opaque token the operator generates in /admin and
hands to a specific person. Registration (multi mode, when
settings.INVITES_REQUIRED) consumes one with a SINGLE conditional UPDATE
stamped with the new user's id, inside the same transaction as the user
INSERT. There is no hold phase: either the user row and the consumed code
commit together, or (duplicate-email race, code failure) neither does.
Two simultaneous signups with the same code: the first UPDATE wins, the
second matches zero rows and aborts its signup. Race structurally closed.

Codes are not hashed: they are operator-minted, single-use, revocable, and
short-lived in practice; a DB leak is already game over for session secrets.
"""

from __future__ import annotations

import secrets
import settings
from utils import utc_now_str as _now_str


class InviteError(ValueError):
    """Raised when a supplied invite code cannot be used."""


def generate_code() -> str:
    """A new invite code: 12 URL-safe characters, no padding."""
    return secrets.token_urlsafe(9)[:12]


def _normalize(code: str) -> str:
    """Codes are compared case-insensitively and ignore surrounding space."""
    return (code or "").strip().upper()


def create_codes(db, count: int, created_by: int) -> list[str]:
    """Insert ``count`` fresh codes. Returns the plaintext codes."""
    codes: list[str] = []
    for _ in range(max(1, min(count, 50))):
        code = generate_code()
        db.execute(
            "INSERT INTO invite_codes (code, created_by) VALUES (?, ?)",
            (_normalize(code), created_by),
        )
        codes.append(_normalize(code))
    return codes


def validate_and_consume(db, code: str, user_id: int) -> str:
    """Atomically consume an invite code, stamping it with its consumer.

    Called AFTER the user row exists, inside the same transaction. The
    single conditional UPDATE (unused + unrevoked + matching code, rowcount
    1) is the entire validation and the entire claim: there is no hold
    phase, no SELECT-then-UPDATE window, and a rolled-back user INSERT
    rolls the consume back with it. Raises InviteError when the code is
    unknown/used/revoked — the caller must abort the signup.
    """
    normalized = _normalize(code)
    if not normalized:
        raise InviteError("An invite code is required to sign up.")
    result = db.execute(
        """
        UPDATE invite_codes
           SET used_by = ?, used_at = ?
         WHERE code = ? AND used_by IS NULL AND revoked = 0
        """,
        (user_id, _now_str(), normalized),
    )
    if result.rowcount != 1:
        # One message for every failure mode: no code-state enumeration for
        # visitors probing the endpoint.
        raise InviteError("That invite code is not valid.")
    return normalized


def looks_usable(db, code: str) -> bool:
    """Non-consuming pre-check for early form validation UX only.

    NEVER authoritative — the atomic consume in validate_and_consume decides.
    Exists so the register form can render its error banner (HTML) instead of
    a JSON 400 from mid-transaction, without creating any claim on the code.
    """
    normalized = _normalize(code)
    if not normalized:
        return False
    row = db.execute(
        "SELECT 1 FROM invite_codes WHERE code = ? AND used_by IS NULL AND revoked = 0",
        (normalized,),
    ).fetchone()
    return row is not None


def list_codes(db) -> list:
    """All codes, newest first, for the admin table."""
    return db.execute(
        """
        SELECT code, created_by, used_by, used_at, revoked, created_at
          FROM invite_codes
         ORDER BY created_at DESC
        """
    ).fetchall()


def revoke(db, code: str, admin_uid: int) -> bool:
    """Revoke an unused code. Returns False for unknown/already-used codes."""
    result = db.execute(
        """
        UPDATE invite_codes SET revoked = 1
         WHERE code = ? AND used_by IS NULL AND revoked = 0
        """,
        (_normalize(code),),
    )
    if result.rowcount == 1:
        return True
    return False


def invites_required() -> bool:
    """Whether /register demands an invite code (multi mode only)."""
    return settings.MULTI and settings.INVITES_REQUIRED