rfc-app/backend/app/otc.py

"""§6.2 / v0.7.0 / v0.8.0: email + one-time-code sign-in.

Replaces the Gitea OAuth gesture as the primary human-auth path. The
Gitea bot user + token are still needed for server-side git
operations (repo reads, PR creation); only the operator-facing
sign-in surface moves through this module.

The shape:

  * `request_code(email)` generates a 6-digit decimal code,
    hashes it (bcrypt), stores the hash + expiry in `otc_codes`,
    and dispatches a plain-text email via `email_otc.send`. It
    invalidates any prior unused codes for the same email so a
    re-request keeps the surface to one outstanding code per
    address. The TTL comes from `OTC_TTL_MINUTES` (default 10).
    A per-email cooldown (`OTC_REQUEST_COOLDOWN_SECONDS`, default
    60) refuses back-to-back requests inside the window.

  * `verify_code(email, code)` walks the most recent unconsumed
    non-expired row for the email, checks the bcrypt hash, marks
    the row consumed, and returns the linked or freshly-provisioned
    user row.

  * `provision_or_link_user(email)` is the migration path: if a
    `users` row already carries `email` (case-insensitive), it is
    reused — `gitea_id` is left alone so a grandfathered OAuth-era
    user keeps the linker intact. Otherwise a fresh contributor
    row is provisioned with `gitea_id = NULL`, `gitea_login = NULL`,
    and `permission_state = 'pending'` (v0.8.0 — see below).

The endpoints in `main.py` thin-wrap this module.

v0.8.0 (roadmap item #6) replaces the v0.3.0 `allowed_emails` gate at
the request surface. The request handler used to silently drop OTC
requests for emails not on the allowlist; now any valid email
receives a code. The admission gate moves to `permission_state` on
the freshly-provisioned `users` row: a fresh user lands in 'pending'
and waits for an admin grant before write endpoints accept them.
Read surfaces stay open (the same blast radius v0.6.0 / item #4
already audited for anonymous viewers).

The `allowed_emails` table itself stays in the schema as a
fast-path bypass — the admin UI from v0.3.0 continues to manage it,
and a future release (v0.9.0's admin user-management page) collapses
the two admission surfaces into one. The OTC request path no
longer consults the table.
"""
from __future__ import annotations

import logging
import os
import secrets
from dataclasses import dataclass

import bcrypt

from . import db
from .auth import SessionUser

log = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Tunables — env-driven with defaults so v0.7.0 needs no new secrets.
# ---------------------------------------------------------------------------


def _ttl_minutes() -> int:
    raw = os.environ.get("OTC_TTL_MINUTES", "").strip()
    if not raw:
        return 10
    try:
        return max(1, int(raw))
    except ValueError:
        return 10


def _cooldown_seconds() -> int:
    raw = os.environ.get("OTC_REQUEST_COOLDOWN_SECONDS", "").strip()
    if not raw:
        return 60
    try:
        return max(0, int(raw))
    except ValueError:
        return 60


# v0.25.0 / security audit 0026 (H1): per-email OTC verify lockout,
# mirroring the passcode path (passcode.py). Five consecutive wrong codes
# for an email lock its OTC verify for 15 minutes. The per-IP limiter in
# ratelimit.py is the primary brute-force brake; this is the durable,
# passcode-parity layer.
LOCKOUT_AFTER_FAILED_ATTEMPTS = 5
LOCKOUT_DURATION_MINUTES = 15


# ---------------------------------------------------------------------------
# Code generation + hashing
# ---------------------------------------------------------------------------


def _new_code() -> str:
    """Six decimal digits. `secrets.randbelow` is CSPRNG-backed so the
    code resists guessing even at the small (10^6) keyspace. The TTL
    + rate-limit are what carry the security weight — the entropy of a
    six-digit code by itself is intentionally human-readable."""
    return f"{secrets.randbelow(1_000_000):06d}"


def _hash_code(code: str) -> str:
    """bcrypt over the code bytes. The hash is stored at rest; the code
    itself only travels in the outbound email and the inbound verify
    body."""
    return bcrypt.hashpw(code.encode("utf-8"), bcrypt.gensalt()).decode("ascii")


def _check_code(code: str, code_hash: str) -> bool:
    try:
        return bcrypt.checkpw(code.encode("utf-8"), code_hash.encode("ascii"))
    except (ValueError, TypeError):
        return False


# ---------------------------------------------------------------------------
# Request path
#
# v0.8.0: the allowlist gate from v0.7.0 / v0.3.0 is removed here. Any
# valid email receives a code; the admission gate moved to
# `permission_state` on the freshly-provisioned `users` row (see
# `provision_or_link_user`). The `allowed_emails` table stays in the
# schema (admin UI from v0.3.0 still manages it); v0.9.0's admin
# user-management page will collapse the two surfaces.
# ---------------------------------------------------------------------------


@dataclass
class RequestOutcome:
    """The outcome of a `request_code` call.

    `code` is None whenever no code was generated — the cooldown
    window blocked the request or the email was syntactically
    invalid. The caller (the API endpoint) does not surface the
    invalid-email shape to the user; it returns 202 either way.
    The cooldown shape surfaces as a loud 429 per the v0.7.0
    contract.
    """
    sent: bool
    code: str | None
    reason: str  # 'sent' | 'cooldown' | 'invalid'


def request_code(email: str) -> RequestOutcome:
    email = (email or "").strip()
    if not email or "@" not in email:
        return RequestOutcome(sent=False, code=None, reason="invalid")

    # Cooldown: refuse if a code was issued for this email in the last
    # COOLDOWN_SECONDS. We surface it as a distinct outcome so the
    # endpoint can return 429 — the spec calls this out as a "loud
    # failure" so the abuse path is visible rather than swallowed.
    cooldown = _cooldown_seconds()
    if cooldown > 0:
        row = db.conn().execute(
            f"""
            SELECT 1 FROM otc_codes
             WHERE email = ?
               AND datetime(created_at, '+{cooldown} seconds') > datetime('now')
             LIMIT 1
            """,
            (email,),
        ).fetchone()
        if row is not None:
            return RequestOutcome(sent=False, code=None, reason="cooldown")

    # Invalidate prior unused codes for this email. A re-request is
    # always for the most recent code; older codes are dead.
    db.conn().execute(
        """
        UPDATE otc_codes
           SET consumed_at = datetime('now')
         WHERE email = ?
           AND consumed_at IS NULL
        """,
        (email,),
    )

    code = _new_code()
    code_hash = _hash_code(code)
    ttl = _ttl_minutes()
    db.conn().execute(
        f"""
        INSERT INTO otc_codes (email, code_hash, expires_at)
        VALUES (?, ?, datetime('now', '+{ttl} minutes'))
        """,
        (email, code_hash),
    )
    return RequestOutcome(sent=True, code=code, reason="sent")


# ---------------------------------------------------------------------------
# Verify path
# ---------------------------------------------------------------------------


@dataclass
class VerifyOutcome:
    """Result of a `verify_code` call.

    `user` is populated only on success. `reason` distinguishes the
    failure modes the UI can render — 'expired', 'consumed', 'wrong',
    'unknown' (no outstanding code at all). The endpoint maps the
    failure modes to a single 400 with a generic message; the reason
    is logged for the operator.
    """
    ok: bool
    user: SessionUser | None
    reason: str
    # v0.25.0 (H1): ISO-8601 stamp when reason == 'locked'.
    locked_until: str | None = None


def _verify_lockout_until(email: str) -> str | None:
    """Return the active lockout stamp for `email`, or None if not locked.

    Clears an elapsed lockout (and resets the counter) as a side effect so
    the next failure starts a fresh budget — mirrors passcode.verify_passcode.
    """
    row = db.conn().execute(
        "SELECT failed_attempts, locked_until FROM otc_verify_state WHERE email = ?",
        (email,),
    ).fetchone()
    if row is None or not row["locked_until"]:
        return None
    still_locked = db.conn().execute(
        "SELECT datetime(?) > datetime('now') AS locked", (row["locked_until"],),
    ).fetchone()["locked"]
    if still_locked:
        return row["locked_until"]
    db.conn().execute(
        "UPDATE otc_verify_state SET failed_attempts = 0, locked_until = NULL WHERE email = ?",
        (email,),
    )
    return None


def _record_verify_failure(email: str) -> None:
    """Increment the per-email failure counter; stamp a lockout once it
    crosses the threshold. Mirrors the passcode lockout shape."""
    db.conn().execute(
        """
        INSERT INTO otc_verify_state (email, failed_attempts)
        VALUES (?, 1)
        ON CONFLICT(email) DO UPDATE SET failed_attempts = failed_attempts + 1
        """,
        (email,),
    )
    count = db.conn().execute(
        "SELECT failed_attempts FROM otc_verify_state WHERE email = ?", (email,),
    ).fetchone()["failed_attempts"]
    if count >= LOCKOUT_AFTER_FAILED_ATTEMPTS:
        db.conn().execute(
            f"""
            UPDATE otc_verify_state
               SET locked_until = datetime('now', '+{LOCKOUT_DURATION_MINUTES} minutes')
             WHERE email = ?
            """,
            (email,),
        )


def _clear_verify_state(email: str) -> None:
    db.conn().execute("DELETE FROM otc_verify_state WHERE email = ?", (email,))


def verify_code(email: str, code: str) -> VerifyOutcome:
    email = (email or "").strip()
    code = (code or "").strip()
    if not email or not code:
        return VerifyOutcome(ok=False, user=None, reason="invalid")

    # v0.25.0 (H1): refuse before spending any bcrypt if this email is in
    # its OTC-verify lockout window. The passcode path is unaffected — a
    # locked-out OTC user can still set/use a passcode, and vice versa.
    locked_until = _verify_lockout_until(email)
    if locked_until:
        return VerifyOutcome(ok=False, user=None, reason="locked", locked_until=locked_until)

    rows = db.conn().execute(
        """
        SELECT id, code_hash, expires_at, consumed_at
          FROM otc_codes
         WHERE email = ?
         ORDER BY id DESC
         LIMIT 5
        """,
        (email,),
    ).fetchall()
    if not rows:
        return VerifyOutcome(ok=False, user=None, reason="unknown")

    # Walk the recent rows so a user who pasted an older code still
    # gets a sensible error — without this, the most-recent-row check
    # would mask "you entered yesterday's code" as "wrong code".
    matched = None
    for row in rows:
        if _check_code(code, row["code_hash"]):
            matched = row
            break

    if matched is None:
        # A genuine wrong guess against this email — the brute-force
        # signal. Count it toward the lockout threshold (H1).
        _record_verify_failure(email)
        return VerifyOutcome(ok=False, user=None, reason="wrong")

    if matched["consumed_at"] is not None:
        return VerifyOutcome(ok=False, user=None, reason="consumed")

    expired = db.conn().execute(
        "SELECT datetime(?) < datetime('now') AS expired",
        (matched["expires_at"],),
    ).fetchone()["expired"]
    if expired:
        return VerifyOutcome(ok=False, user=None, reason="expired")

    # Stamp consumed before provisioning so a parallel verify of the
    # same row can't double-sign-in.
    db.conn().execute(
        "UPDATE otc_codes SET consumed_at = datetime('now') WHERE id = ?",
        (matched["id"],),
    )
    # Success wipes the per-email failure counter (H1).
    _clear_verify_state(email)
    user = provision_or_link_user(email)
    return VerifyOutcome(ok=True, user=user, reason="ok")


# ---------------------------------------------------------------------------
# Provisioning — the migration path from OAuth identity to email identity.
# ---------------------------------------------------------------------------


def provision_or_link_user(email: str) -> SessionUser:
    """Link the OTC sign-in to a `users` row.

    Match order:
      1. An existing row whose email equals (case-insensitive) the
         requested email — the OAuth-era user is grandfathered in via
         this path. `gitea_id` is preserved so a future OAuth round
         trip still resolves the same row. `permission_state` is
         read off the row as-is — grandfathered users come through
         migration with 'granted' (the column default), so their
         contributor capabilities are unaffected.
      2. Otherwise: a fresh contributor row with `gitea_id = NULL`,
         `gitea_login = NULL`, and `permission_state = 'pending'`
         (v0.8.0). The display name defaults to the local part of
         the email (everything before the `@`); a separate
         `POST /auth/me/beta-request` call lands first name / last
         name / "why I want access" on the same row.

    The §6.1 owner-zero bootstrap still applies: if the email matches
    the configured `OWNER_GITEA_LOGIN`-derived owner identity, the row
    is provisioned with role='owner'. v0.7.0 keeps that field as the
    Gitea login (so existing deployments don't break); a future
    release may add a parallel `OWNER_EMAIL` env if the OAuth route is
    dropped entirely.
    """
    email = email.strip()
    existing = db.conn().execute(
        "SELECT * FROM users WHERE email = ? COLLATE NOCASE",
        (email,),
    ).fetchone()
    if existing is not None:
        db.conn().execute(
            "UPDATE users SET last_seen_at = datetime('now') WHERE id = ?",
            (existing["id"],),
        )
        return SessionUser(
            user_id=existing["id"],
            gitea_id=existing["gitea_id"] or 0,
            gitea_login=existing["gitea_login"] or "",
            display_name=existing["display_name"],
            email=existing["email"] or email,
            avatar_url=existing["avatar_url"] or "",
            role=existing["role"],
            permission_state=existing["permission_state"] or "granted",
        )

    display = email.split("@", 1)[0] or email
    # v0.8.0: 'pending' is the explicit insert value; the migration
    # default of 'granted' is what passes grandfathered users
    # through. A fresh OTC user lands in 'pending' regardless of
    # what the migration default says, so the gate engages reliably
    # even if a future migration changes the default.
    cur = db.conn().execute(
        """
        INSERT INTO users (gitea_id, gitea_login, email, display_name, avatar_url, role, permission_state)
        VALUES (NULL, NULL, ?, ?, '', 'contributor', 'pending')
        """,
        (email, display),
    )
    user_id = cur.lastrowid
    return SessionUser(
        user_id=user_id,
        gitea_id=0,
        gitea_login="",
        display_name=display,
        email=email,
        avatar_url="",
        role="contributor",
        permission_state="pending",
    )