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