rfc-app/backend/app/passcode.py

"""§6.2 / v0.10.0: user-set passcodes after OTC (roadmap item #8).

After a successful OTC sign-in, a contributor may set a passcode and
use email + passcode for subsequent sign-ins. OTC remains the fallback
— a forgotten passcode is recovered by requesting a fresh OTC.

This module is the state machine behind the four `/auth/passcode/*`
endpoints (`set`, `clear`, `verify`, `check`). The endpoints in
`main.py` thin-wrap these helpers in the same shape the OTC module
uses (see `otc.py`).

Shape:

  * `set_passcode(user_id, passcode)` — bcrypt-hash the passcode and
    write it to `users.passcode_hash` + `users.passcode_set_at`.
    Validation (length, denylist) happens here, not at the endpoint,
    so the rule lives in one place. Replaces any prior passcode.
  * `clear_passcode(user_id)` — null out `passcode_hash` and
    `passcode_set_at`. The user is back to OTC-only.
  * `verify_passcode(email, passcode)` — locate the user by email,
    check lockout, compare via bcrypt, manage the failure counter,
    and return a populated `SessionUser` on success.
  * `passcode_status(email)` — does this email have a passcode set?
    Used by the `/auth/passcode/check` endpoint that the Login.jsx
    flow consults after the user types their email.

Lockout is a v1 shape: 5 consecutive failures sets
`passcode_locked_until` to `now + 15 minutes`, after which a verify
attempt that lands inside the window returns HTTP 423. The OTC path
is unaffected by the lockout — a user can request and verify a fresh
OTC to sign in while their passcode is locked out, and `verify_code`
in `otc.py` does not consult these columns.

The lockout window and the failure threshold are hard-coded here.
Tuning them via env vars (or moving to per-IP rate-limiting) is a
§19.2 candidate; see SPEC §19.2.
"""
from __future__ import annotations

import logging
from dataclasses import dataclass

import bcrypt

from . import db
from .auth import SessionUser

log = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Tunables — intentionally hard-coded in v0.10.0 (see module docstring).
# ---------------------------------------------------------------------------

LOCKOUT_AFTER_FAILED_ATTEMPTS = 5
LOCKOUT_DURATION_MINUTES = 15

PASSCODE_MIN_LENGTH = 4
PASSCODE_MAX_LENGTH = 20

# A small denylist of patterns we never want a passcode to be. The
# rule is "no obvious patterns"; the list is deliberately small —
# every entry here is a verbatim string match. A heavier check
# (sequential digits, single-character runs of length >= N, etc.)
# is a §19.2 candidate.
PASSCODE_DENYLIST: frozenset[str] = frozenset(
    {
        "0000",
        "1111",
        "2222",
        "3333",
        "4444",
        "5555",
        "6666",
        "7777",
        "8888",
        "9999",
        "1234",
        "12345",
        "123456",
        "1234567",
        "12345678",
        "123456789",
        "1234567890",
        "0123",
        "01234",
        "012345",
        "0123456",
        "01234567",
        "012345678",
        "0123456789",
        "abcd",
        "abcde",
        "abcdef",
        "qwer",
        "qwerty",
        "asdf",
        "asdfg",
        "asdfgh",
        "aaaa",
        "bbbb",
        "cccc",
        "password",
        "letmein",
    }
)


# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------


class PasscodeValidationError(Exception):
    """The proposed passcode failed validation. The endpoint surface
    maps this to HTTP 422 with the message intact."""


def _validate(passcode: str) -> str:
    """Return the normalized passcode (stripped) or raise.

    Rules:
      * 4-20 characters after stripping leading/trailing whitespace.
      * Not on the small denylist of obvious patterns.

    No character-class restriction beyond that — the spec says
    "numeric PIN or short alphanumeric"; we don't refuse other
    characters because the entropy isn't load-bearing (the per-account
    lockout is what carries the security weight, mirroring the OTC
    shape from v0.7.0).
    """
    pc = (passcode or "").strip()
    if not pc:
        raise PasscodeValidationError("Passcode is required")
    if len(pc) < PASSCODE_MIN_LENGTH:
        raise PasscodeValidationError(
            f"Passcode must be at least {PASSCODE_MIN_LENGTH} characters"
        )
    if len(pc) > PASSCODE_MAX_LENGTH:
        raise PasscodeValidationError(
            f"Passcode must be at most {PASSCODE_MAX_LENGTH} characters"
        )
    if pc.lower() in PASSCODE_DENYLIST:
        raise PasscodeValidationError("Passcode is too common; pick something less obvious")
    return pc


# ---------------------------------------------------------------------------
# Hashing
# ---------------------------------------------------------------------------


def _hash(passcode: str) -> str:
    return bcrypt.hashpw(passcode.encode("utf-8"), bcrypt.gensalt()).decode("ascii")


def _check(passcode: str, passcode_hash: str) -> bool:
    try:
        return bcrypt.checkpw(passcode.encode("utf-8"), passcode_hash.encode("ascii"))
    except (ValueError, TypeError):
        return False


# ---------------------------------------------------------------------------
# Set / clear
# ---------------------------------------------------------------------------


def set_passcode(user_id: int, passcode: str) -> None:
    """Hash and store the passcode. Replaces any prior passcode on the
    same row; clears the failure counter and lockout (a user setting a
    fresh passcode is implicitly re-authenticating their account)."""
    pc = _validate(passcode)
    h = _hash(pc)
    db.conn().execute(
        """
        UPDATE users
           SET passcode_hash = ?,
               passcode_set_at = datetime('now'),
               passcode_failed_attempts = 0,
               passcode_locked_until = NULL
         WHERE id = ?
        """,
        (h, user_id),
    )


def clear_passcode(user_id: int) -> None:
    """Remove the passcode. The user is back to OTC-only on next sign-in."""
    db.conn().execute(
        """
        UPDATE users
           SET passcode_hash = NULL,
               passcode_set_at = NULL,
               passcode_failed_attempts = 0,
               passcode_locked_until = NULL
         WHERE id = ?
        """,
        (user_id,),
    )


# ---------------------------------------------------------------------------
# Check (status surface for the Login.jsx flow)
# ---------------------------------------------------------------------------


@dataclass
class PasscodeStatus:
    """The shape `/auth/passcode/check` returns.

    `has_passcode` is the only signal the frontend needs to decide
    whether to show a passcode input or an OTC request step. We do
    not leak the hash, the set-at timestamp, or the lockout state —
    a probing client that wants to know "is this account locked
    out" can attempt a verify and read the 423.
    """
    has_passcode: bool


def passcode_status(email: str) -> PasscodeStatus:
    email = (email or "").strip()
    if not email or "@" not in email:
        return PasscodeStatus(has_passcode=False)
    row = db.conn().execute(
        "SELECT passcode_hash FROM users WHERE email = ? COLLATE NOCASE",
        (email,),
    ).fetchone()
    if row is None:
        return PasscodeStatus(has_passcode=False)
    return PasscodeStatus(has_passcode=bool(row["passcode_hash"]))


# ---------------------------------------------------------------------------
# Verify
# ---------------------------------------------------------------------------


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

    `reason` distinguishes the failure modes the endpoint surfaces as
    distinct HTTP shapes:
      * 'ok'        — populated `user`, HTTP 200.
      * 'unknown'   — no user with this email, HTTP 400 (generic).
      * 'no_passcode' — user exists but never set a passcode, HTTP 400
        (the frontend should fall back to OTC).
      * 'locked'    — user is currently in the lockout window, HTTP
        423. `locked_until` carries the ISO-8601 stamp for the client.
      * 'wrong'     — passcode didn't match. HTTP 400. If the failure
        crossed the lockout threshold the row is now locked; the
        endpoint surfaces this as a fresh `locked` response on the
        next attempt rather than collapsing the two states here.
    """
    ok: bool
    user: SessionUser | None
    reason: str
    locked_until: str | None = None


def verify_passcode(email: str, passcode: str) -> VerifyOutcome:
    email = (email or "").strip()
    passcode = (passcode or "").strip()
    if not email or not passcode:
        return VerifyOutcome(ok=False, user=None, reason="unknown")

    row = db.conn().execute(
        """
        SELECT id, gitea_id, gitea_login, email, display_name, avatar_url, role,
               passcode_hash, passcode_failed_attempts, passcode_locked_until
          FROM users
         WHERE email = ? COLLATE NOCASE
        """,
        (email,),
    ).fetchone()
    if row is None:
        return VerifyOutcome(ok=False, user=None, reason="unknown")
    if not row["passcode_hash"]:
        return VerifyOutcome(ok=False, user=None, reason="no_passcode")

    # Lockout check: if `passcode_locked_until` is populated and in the
    # future, the verify is refused without touching the hash. Once the
    # window has elapsed we let the verify proceed; the failed-attempts
    # counter is also reset so the user gets a fresh 5-attempt budget.
    locked_until = row["passcode_locked_until"]
    if locked_until:
        still_locked = db.conn().execute(
            "SELECT datetime(?) > datetime('now') AS still_locked",
            (locked_until,),
        ).fetchone()["still_locked"]
        if still_locked:
            return VerifyOutcome(
                ok=False,
                user=None,
                reason="locked",
                locked_until=locked_until,
            )
        # Lockout expired — clear the counter so the next failure starts
        # from zero, and continue with the verify.
        db.conn().execute(
            """
            UPDATE users
               SET passcode_failed_attempts = 0,
                   passcode_locked_until = NULL
             WHERE id = ?
            """,
            (row["id"],),
        )

    if _check(passcode, row["passcode_hash"]):
        # Success: clear the counter (a single success wipes the
        # accumulated failures — the threshold tracks *consecutive*
        # failures).
        db.conn().execute(
            """
            UPDATE users
               SET passcode_failed_attempts = 0,
                   passcode_locked_until = NULL,
                   last_seen_at = datetime('now')
             WHERE id = ?
            """,
            (row["id"],),
        )
        return VerifyOutcome(
            ok=True,
            user=SessionUser(
                user_id=row["id"],
                gitea_id=row["gitea_id"] or 0,
                gitea_login=row["gitea_login"] or "",
                display_name=row["display_name"],
                email=row["email"] or email,
                avatar_url=row["avatar_url"] or "",
                role=row["role"],
            ),
            reason="ok",
        )

    # Failure: increment the counter. If this push crosses the
    # threshold, stamp the lockout. The next verify attempt against
    # the same row returns 423 with the `locked_until` stamp.
    next_count = (row["passcode_failed_attempts"] or 0) + 1
    if next_count >= LOCKOUT_AFTER_FAILED_ATTEMPTS:
        db.conn().execute(
            f"""
            UPDATE users
               SET passcode_failed_attempts = ?,
                   passcode_locked_until = datetime('now', '+{LOCKOUT_DURATION_MINUTES} minutes')
             WHERE id = ?
            """,
            (next_count, row["id"]),
        )
        new_locked_until = db.conn().execute(
            "SELECT passcode_locked_until FROM users WHERE id = ?",
            (row["id"],),
        ).fetchone()["passcode_locked_until"]
        return VerifyOutcome(
            ok=False,
            user=None,
            reason="locked",
            locked_until=new_locked_until,
        )
    db.conn().execute(
        "UPDATE users SET passcode_failed_attempts = ? WHERE id = ?",
        (next_count, row["id"]),
    )
    return VerifyOutcome(ok=False, user=None, reason="wrong")