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