rfc-app/backend/app/otc.py

"""§6.2 / v0.7.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`.

The endpoints in `main.py` thin-wrap this module. The allowlist gate
from v0.3.0 is consulted at request time — if `allowed_emails` is
populated and the requested address isn't on it, the request returns
202 as usual but no email is sent. This intentionally does not leak
allowlist state to the caller; the §19.2 candidate for v0.8.0
replaces this gate with an admin-grant flow.
"""
from __future__ import annotations

import logging
import os
import secrets
from dataclasses import dataclass

import bcrypt

from . import db
from .auth import SessionUser, allowlist_is_active

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


# ---------------------------------------------------------------------------
# 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


# ---------------------------------------------------------------------------
# Allowlist gate — shared with the OAuth flow.
# ---------------------------------------------------------------------------


def _allowlist_admits(email: str) -> bool:
    """The same allowlist v0.3.0 introduced for OAuth, applied to OTC
    requests. If the allowlist is populated and the email is not on it,
    we still respond 202 to the caller, but no code is sent."""
    if not allowlist_is_active():
        return True
    row = db.conn().execute(
        "SELECT 1 FROM allowed_emails WHERE email = ? LIMIT 1", (email,)
    ).fetchone()
    return row is not None


# ---------------------------------------------------------------------------
# Request path
# ---------------------------------------------------------------------------


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

    `code` is None whenever no code was generated — either because the
    allowlist denied the email or because the cooldown window blocked
    the request. The caller (the API endpoint) does not surface this
    distinction to the user; it returns 202 either way.
    """
    sent: bool
    code: str | None
    reason: str  # 'sent' | 'allowlist' | '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")

    # Allowlist: silently drop the send if the email isn't on the list.
    # The row is not written either — there's nothing for verify to
    # match against, so the user-facing experience is "I never got an
    # email", which is the intended shape for the private-beta gate.
    if not _allowlist_admits(email):
        return RequestOutcome(sent=False, code=None, reason="allowlist")

    # 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


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")

    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:
        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"],),
    )
    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.
      2. Otherwise: a fresh contributor row with `gitea_id = NULL`,
         `gitea_login = NULL`. The display name defaults to the local
         part of the email (everything before the `@`) — users can
         rename later via the §19.2 first-OTC profile-capture flow
         that v0.8.0 introduces.

    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"],
        )

    display = email.split("@", 1)[0] or email
    cur = db.conn().execute(
        """
        INSERT INTO users (gitea_id, gitea_login, email, display_name, avatar_url, role)
        VALUES (NULL, NULL, ?, ?, '', 'contributor')
        """,
        (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",
    )