"""Multi-mode authentication: register, login, logout, and session setup. Single mode never registers these routes' behavior — every handler short-circuits with 404 unless FEEDECHO_MODE=multi. Sessions are stateless HMAC tokens from security.py; the shared-secret AuthMiddleware in app.py routes to this flow only when settings.MULTI is set. """ from __future__ import annotations import ipaddress import logging import threading import time from datetime import datetime, timedelta, timezone from fastapi import Form, HTTPException, Request from fastapi.responses import HTMLResponse, RedirectResponse import disposable_emails import settings import invites import plans from database import get_db from security import SESSION_TTL_SECONDS, hash_password, read_session, sign_session, verify_password from utils import EMAIL_RE as _EMAIL_RE # Minimal in-memory login throttle: 5 failed attempts per IP per 5 minutes. # Deliberately cheap and boring; real abuse controls (per-user caps, IP # reputation) land with the billing/abuse phase. _MAX_LOGIN_ATTEMPTS = 5 _LOGIN_WINDOW_SECONDS = 5 * 60 _login_attempts: dict[str, list[float]] = {} _login_lock = threading.Lock() # Registration gets its own, more generous bucket: scrypt is expensive and # the route is public, so cap valid-form submissions per IP. _MAX_REGISTER_ATTEMPTS = 10 _REGISTER_WINDOW_SECONDS = 10 * 60 _register_attempts: dict[str, list[float]] = {} # Deployment-wide registration brake (settings.REGISTER_HOURLY_CAP / # REGISTER_DAILY_CAP): the per-IP bucket bounds one source, but a # rotating-IP spray has no ceiling, and every valid submission costs a # scrypt hash plus a verification email to an address the registrant chooses. # Monotonic timestamps, so the bucket resets on restart — acceptable: a # restart is not the abuse case, and the DB row count is the backstop. _REGISTER_GLOBAL_HOUR = 3600.0 _REGISTER_GLOBAL_DAY = 86400.0 _register_global: list[float] = [] def _register_global_capped() -> bool: """Whether the deployment-wide signup brake has tripped. Two ceilings bound both a short burst (hourly) and a slow drip (daily); entries older than a day are pruned so the list cannot grow without bound. Recording happens via _record_register_global on submission. """ now = time.monotonic() with _login_lock: day_cutoff = now - _REGISTER_GLOBAL_DAY while _register_global and _register_global[0] < day_cutoff: _register_global.pop(0) if len(_register_global) >= settings.REGISTER_DAILY_CAP: return True hour_cutoff = now - _REGISTER_GLOBAL_HOUR recent = sum(1 for t in _register_global if t >= hour_cutoff) return recent >= settings.REGISTER_HOURLY_CAP def _record_register_global() -> None: with _login_lock: _register_global.append(time.monotonic()) # Forgot-password IP throttle: 5 requests per IP per 10 minutes. Keeps an # unauthenticated email-sending endpoint from rotating across the whole # address space (per-user token throttle alone doesn't bound total sends). _forgot_attempts: dict[str, list[float]] = {} _FORGOT_LIMIT = 5 _FORGOT_WINDOW = 600 _MIN_PASSWORD_LENGTH = 8 _MAX_PASSWORD_LENGTH = 1024 _TRIAL_DAYS = 14 COOKIE_NAME = "feedecho_session" # Single-mode shared-secret cookie. Named here so the middleware, the login # handler and logout cannot drift apart — a rename that missed logout is what # made single-mode logout a silent no-op. AUTH_COOKIE_NAME = "feedecho_auth" # Precomputed once so unknown-email login attempts pay the same scrypt # cost as known-email attempts (no timing-based user enumeration). _DUMMY_HASH = hash_password("feedecho-timing-equalizer") # Holds strong references to fire-and-forget asyncio Tasks (currently just # the password-reset email dispatch below) so the event loop can't garbage # -collect one mid-execution, per the asyncio docs' own warning about # unreferenced create_task() results. Exception-safety doesn't depend on # this (_send_reset_email already guards its entire body), so this is # hygiene, not a live-bug fix. Error-handling audit finding 3.1 # (docs/reviews/2026-09-08-error-handling-audit.md). _background_tasks: set = set() def current_user_id(request: Request) -> int: """The authenticated user for this request (multi mode), or the singleton user 1 in single mode.""" if settings.MULTI: uid = getattr(request.state, "user_id", None) if uid is None: raise HTTPException(status_code=401, detail="Authentication required") return uid return 1 def _require_multi() -> None: if not settings.MULTI: raise HTTPException(status_code=404, detail="Not found") class _InviteRejected(Exception): """Internal: invite consume failed mid-transaction. Raised so the get_db context manager rolls back the whole signup (user row + code consume together); caught just outside the block, where it is safe to render the HTML error banner. """ def is_admin(uid: int) -> bool: """Whether the given user is an admin. Read fresh from the database on every call — never from session claims — so demotions take effect immediately even with a valid session token. Non-existent users are not admins. """ with get_db() as db: row = db.execute( "SELECT is_admin FROM users WHERE id = ?", (uid,) ).fetchone() return bool(row and row["is_admin"]) def _trial_end() -> str: return (datetime.now(timezone.utc) + timedelta(days=_TRIAL_DAYS)).strftime( "%Y-%m-%d %H:%M:%S" ) _trusted_networks_cache: tuple[tuple[str, ...], tuple] | None = None def _trusted_networks() -> tuple: """Parsed settings.TRUSTED_PROXIES networks, cached, bad entries skipped. The access logger now calls _client_ip on every request, so parsing per call is wasted work; and an unparseable entry used to raise, which would turn one typo'd CIDR into a 500 on every request. Bad entries are ignored with a warning, failing to the SAFE side: X-Forwarded-For is simply not believed, exactly as if no proxy were configured. """ global _trusted_networks_cache raw = tuple(settings.TRUSTED_PROXIES) if _trusted_networks_cache and _trusted_networks_cache[0] == raw: return _trusted_networks_cache[1] nets = [] for cidr in raw: try: nets.append(ipaddress.ip_network(cidr, strict=False)) except ValueError: logging.getLogger("feedecho").warning( "Ignoring unparseable FEEDECHO_TRUSTED_PROXIES entry: %r", cidr ) _trusted_networks_cache = (raw, tuple(nets)) return _trusted_networks_cache[1] def _client_ip(request: Request) -> str: """The client IP for rate limiting and access logs. Behind a trusted reverse proxy (settings.TRUSTED_PROXIES), the TCP peer is the proxy, so the rightmost X-Forwarded-For entry is used. Without trusted proxies configured (or with only unparseable entries), X-Forwarded-For is ignored entirely (it is trivially spoofable). """ peer = request.client.host if request.client else "unknown" nets = _trusted_networks() if not nets: return peer try: peer_ip = ipaddress.ip_address(peer) except ValueError: return peer if not any(peer_ip in net for net in nets): return peer forwarded = request.headers.get("x-forwarded-for", "") entries = [e.strip() for e in forwarded.split(",") if e.strip()] if not entries: return peer candidate = entries[-1] # Only a parseable IP is ever used: a proxy that appends garbage (or a # header we do not actually trust) must not become a log field or a # throttle bucket key. try: ipaddress.ip_address(candidate) except ValueError: return peer return candidate def _prune(bucket: dict[str, list[float]], key: str, window: float) -> list[float]: now = time.monotonic() attempts = [t for t in bucket.get(key, []) if now - t < window] if attempts: bucket[key] = attempts else: bucket.pop(key, None) # evict empty keys so the dict can't grow return attempts def _throttled(ip: str) -> bool: with _login_lock: return len(_prune(_login_attempts, ip, _LOGIN_WINDOW_SECONDS)) >= _MAX_LOGIN_ATTEMPTS def _register_throttled(ip: str) -> bool: with _login_lock: return ( len(_prune(_register_attempts, ip, _REGISTER_WINDOW_SECONDS)) >= _MAX_REGISTER_ATTEMPTS ) def _record_failure(ip: str) -> None: with _login_lock: _login_attempts.setdefault(ip, []).append(time.monotonic()) def _record_register(ip: str) -> None: with _login_lock: _register_attempts.setdefault(ip, []).append(time.monotonic()) def _clear_failures(ip: str) -> None: """A successful login resets the failure bucket for that IP.""" with _login_lock: _login_attempts.pop(ip, None) def _set_session_cookie( response: RedirectResponse, user_id: int, email: str, request: Request, epoch: int = 0, ) -> None: response.set_cookie( key=COOKIE_NAME, value=sign_session(user_id, email, epoch), httponly=True, samesite="lax", secure=request.url.scheme == "https" or settings.FORCE_SECURE_COOKIE, max_age=SESSION_TTL_SECONDS, ) def _render_auth(request: Request, template: str, status_code: int = 200, **kwargs): from app import render return render(template, request, status_code=status_code, **kwargs) def _absolute_link(path: str) -> str: """An absolute URL for an emailed link, or "" when BASE_URL is unset. Better to skip the send and log it than to mail a relative link the recipient cannot open; validate_config warns about this at startup. """ base = settings.BASE_URL.rstrip("/") if not base: logging.getLogger("feedecho").error( "Cannot build %s link: FEEDECHO_BASE_URL is not set; email not sent", path ) return "" return f"{base}{path}" def register_page(request: Request): _require_multi() return _render_auth(request, "register.html", invites_required=invites.invites_required()) def register_submit( request: Request, email: str = Form(...), password: str = Form(...), confirm: str = Form(...), invite_code: str = Form(""), ): _require_multi() email = email.strip().lower() invite_code = (invite_code or "").strip() errors: list[str] = [] if not _EMAIL_RE.match(email): errors.append("Enter a valid email address.") if len(password) < _MIN_PASSWORD_LENGTH: errors.append(f"Password must be at least {_MIN_PASSWORD_LENGTH} characters.") if len(password) > _MAX_PASSWORD_LENGTH: errors.append(f"Password must be at most {_MAX_PASSWORD_LENGTH} characters.") if password != confirm: errors.append("Passwords do not match.") if settings.DISPOSABLE_EMAIL_BLOCK and disposable_emails.is_disposable_email(email): errors.append( "That email provider isn't supported. Please use a permanent email address." ) if errors: return _render_auth( request, "register.html", error=" ".join(errors), email=email, invite_code=invite_code, invites_required=invites.invites_required(), ) ip = _client_ip(request) if _register_throttled(ip): return _render_auth( request, "register.html", error="Too many signup attempts from this address. Try again later.", email=email, invite_code=invite_code, invites_required=invites.invites_required(), ) if _register_global_capped(): # Deployment-wide brake: the per-IP bucket above cannot see a # rotating-IP spray, and each valid submission costs a scrypt hash # plus a verification email to an attacker-chosen address. return _render_auth( request, "register.html", error="Signups are temporarily paused. Please try again later.", email=email, invite_code=invite_code, invites_required=invites.invites_required(), ) try: with get_db() as db: existing = db.execute( "SELECT id FROM users WHERE email = ?", (email,) ).fetchone() if existing: return _render_auth( request, "register.html", error="An account with that email already exists.", email=email, invite_code=invite_code, invites_required=invites.invites_required(), ) # Early UX check (non-consuming): renders the friendly banner for # the common failure before doing any work. The authoritative gate # is the atomic consume below — this pre-check creates no claim. if invites.invites_required() and not invites.looks_usable(db, invite_code): return _render_auth( request, "register.html", error="That invite code is not valid.", email=email, invite_code=invite_code, invites_required=True, ) _record_register(ip) _record_register_global() # Card-gated trial (hosted billing): the trial clock does NOT start # at registration — a past trial_ends_at pauses posting through the # existing expired-trial path until the card is collected, when the # billing webhook sets the real trial_end from Stripe. trial_end = plans.TRIAL_PENDING if settings.BILLING_ENABLED else _trial_end() try: db.execute( """ INSERT INTO users (email, password_hash, plan, trial_ends_at, email_verified) VALUES (?, ?, 'trial', ?, 0) """, (email, hash_password(password), trial_end), ) except Exception as e: # Duplicate-email race between SELECT and INSERT: the UNIQUE # constraint fires as IntegrityError (sqlite) / UniqueViolation # (psycopg). Render the same friendly message as the SELECT path. # No invite code has been consumed yet (consume happens below, # after the INSERT succeeded), so nothing to roll back. if e.__class__.__name__ in ("IntegrityError", "UniqueViolation"): return _render_auth( request, "register.html", error="An account with that email already exists.", email=email, invite_code=invite_code, invites_required=invites.invites_required(), ) raise user = db.execute( "SELECT id FROM users WHERE email = ?", (email,) ).fetchone() # Invite gate: consume AFTER the user row exists, SAME transaction. # The single conditional UPDATE stamps used_by with the new user's # id; rowcount 1 is the entire validation. If it fails, raise out of # the context manager: the whole transaction — user row included — # rolls back, so a rejected code leaves zero trace. Two simultaneous # signups on one code: the first UPDATE wins, the second matches # zero rows and rolls its user INSERT back. No hold phase, no window. if invites.invites_required(): try: invites.validate_and_consume(db, invite_code, user["id"]) except invites.InviteError: # Abort the whole transaction (user row included), then the # except below renders the banner once we're outside it. raise _InviteRejected() except _InviteRejected: # Outside the DB context: rollback already happened, safe to render. return _render_auth( request, "register.html", error="That invite code is not valid.", email=email, invite_code=invite_code, invites_required=True, ) # Verification email: best-effort. Signup succeeds even if system SMTP # is unconfigured or the send fails; the dashboard banner offers resend. try: import logging import verification from email_sender import send_system_email token = verification.issue_token(user["id"], "verify") link = _absolute_link(f"/verify-email?token={token}") if link: send_system_email( email, "Verify your FeedEcho account", f"Welcome to FeedEcho. Verify your email address by opening this link:\n\n{link}\n\n" "This link expires in 24 hours. If you did not create this account, you can ignore this email.", ) except Exception as exc: # noqa: BLE001 — verification must not block signup # ERROR + exc_info=True to match _send_reset_email's logging below — # previously WARNING with no traceback, so a genuine bug here (as # opposed to the expected "SMTP unconfigured" case) had no stack # trace anywhere to diagnose it. Error-handling audit finding 5.2 # (docs/reviews/2026-09-08-error-handling-audit.md). Still non-blocking: signup # succeeds either way, per the comment above. logging.getLogger("feedecho").error( "Verification email for %s failed: %s", email, exc, exc_info=True ) # Card-gated trial: send a freshly registered user straight into Stripe # Checkout to collect the card (the billing GET checkout route). Without # billing, the original landing redirect is unchanged. url = "/api/billing/checkout?interval=monthly" if settings.BILLING_ENABLED else "/" response = RedirectResponse(url=url, status_code=302) _set_session_cookie(response, user["id"], email, request) return response def login_page(request: Request): if not settings.MULTI: return _render_auth(request, "login.html") return _render_auth(request, "login.html", multi=True) def _login_single_mode(request: Request, token: str): """Shared-secret token login (original single-tenant behavior).""" import secrets as _secrets if not settings.AUTH_TOKEN: return RedirectResponse(url="/", status_code=302) ip = _client_ip(request) # Verify first, throttle second. Gating the comparison on the bucket # would let anyone who can reach /login keep the operator out of their # own instance indefinitely by POSTing wrong tokens: single mode has no # second credential and no account recovery. A correct token always # works and clears the bucket; wrong ones are what get rate limited. # # Compare bytes: compare_digest raises TypeError on non-ASCII str. if token and _secrets.compare_digest( token.encode("utf-8", "surrogatepass"), settings.AUTH_TOKEN.encode("utf-8", "surrogatepass"), ): _clear_failures(ip) response = RedirectResponse(url="/", status_code=302) response.set_cookie( key=AUTH_COOKIE_NAME, value=token, httponly=True, samesite="lax", secure=request.url.scheme == "https" or settings.FORCE_SECURE_COOKIE, ) return response _record_failure(ip) if _throttled(ip): return _render_auth( request, "login.html", error="Too many failed attempts. Try again in a few minutes.", ) return _render_auth(request, "login.html", error="Invalid token") def _login_multi_mode(request: Request, email: str, password: str): """Email + password + session login (multi-tenant behavior).""" ip = _client_ip(request) if _throttled(ip): return _render_auth( request, "login.html", multi=True, error="Too many failed attempts. Try again in a few minutes.", ) email = email.strip().lower() with get_db() as db: user = db.execute( "SELECT id, email, password_hash, suspended, session_epoch" " FROM users WHERE email = ?", (email,), ).fetchone() if user is None: # Pay the same scrypt cost as a known-email attempt so response # timing does not reveal whether an account exists. verify_password(password, _DUMMY_HASH) _record_failure(ip) return _render_auth( request, "login.html", multi=True, error="Invalid email or password" ) if not verify_password(password, user["password_hash"]): _record_failure(ip) return _render_auth( request, "login.html", multi=True, error="Invalid email or password" ) _clear_failures(ip) if user["suspended"]: return _render_auth( request, "login.html", multi=True, error="This account is suspended." ) response = RedirectResponse(url="/", status_code=302) _set_session_cookie(response, user["id"], user["email"], request, user["session_epoch"]) return response def login_submit( request: Request, email: str = Form(""), password: str = Form(""), token: str = Form(""), ): """Dispatch to the mode-appropriate login strategy. Single-tenant and multi-tenant login are different algorithms (a shared-secret token compare vs. an email/password/session lookup) that happened to share one function; split per the design-patterns audit's Strategy-pattern finding 3.2 so each mode reads as its own self-contained flow instead of one function branching for its entire body. """ if not settings.MULTI: return _login_single_mode(request, token) return _login_multi_mode(request, email, password) def logout(request: Request): response = RedirectResponse(url="/login", status_code=302) # Both cookies: multi mode authenticates on the signed session, single # mode on the shared-secret cookie. Clearing only one made logout a # no-op in the other mode. Both are set with Starlette's default path # ("/") and no domain, so these deletions match them. # # The deletion MUST carry the same Secure/HttpOnly flags the login flow # sets. A cookie set with Secure is not reliably evicted by a deletion # that omits Secure on WebKit/Safari — the browser keeps the stale # session cookie, and even though the epoch bump below makes the server # reject it, the browser re-sends it until the user clears it. Matching # the flags removes the cookie for real. _secure = request.url.scheme == "https" or settings.FORCE_SECURE_COOKIE for name in (COOKIE_NAME, AUTH_COOKIE_NAME): response.delete_cookie(name, secure=_secure, httponly=True, samesite="lax") # S10a: bump session_epoch so all existing sessions for this user are # invalidated immediately. The cookie is deleted either way, but this # also kills any session token an attacker may have stored before the # user logged out. /logout is auth-exempt, so request.state.user_id is # unset here — read the signed session cookie directly instead. if settings.MULTI: token = request.cookies.get(COOKIE_NAME) # ignore_expiry: logout must invalidate all of the user's sessions # even when this cookie has already expired — otherwise a fresher # session on another device (or a stored token) survives a logout the # user believes happened. The signature is still verified. claims = read_session(token, ignore_expiry=True) if token else None uid = claims["user_id"] if claims else None if uid: with get_db() as db: db.execute( "UPDATE users SET session_epoch = session_epoch + 1 WHERE id = ?", (uid,), ) return response # ── Password reset ──────────────────────────────────────────────────────────── def _forgot_throttled(ip: str) -> bool: # Shares _prune's eviction and the lock: reassigning an empty bucket left # a permanent entry per IP that ever hit the endpoint, and the dict was # mutated from request threads without synchronisation. with _login_lock: return len(_prune(_forgot_attempts, ip, _FORGOT_WINDOW)) >= _FORGOT_LIMIT def _record_forgot(ip: str) -> None: with _login_lock: _forgot_attempts.setdefault(ip, []).append(time.monotonic()) def forgot_page(request: Request): _require_multi() return _render_auth(request, "forgot_password.html") def _send_reset_email(user_id: int, email: str) -> None: """Issue a reset token and email the link (runs in a worker thread). Called fire-and-forget so the HTTP response never waits on SMTP — this also removes the timing oracle between known/unknown addresses. """ import verification from email_sender import send_system_email try: if not verification.resend_allowed(user_id, "reset"): return token = verification.issue_token(user_id, "reset") link = _absolute_link(f"/reset-password?token={token}") if not link: return send_system_email( email, "Reset your FeedEcho password", f"Reset your FeedEcho password by opening this link:\n\n{link}\n\n" "This link expires in 24 hours. If you did not request a reset, you can ignore this email.", ) except Exception: # noqa: BLE001 — enumeration-safe; ERROR level so a logging.getLogger("feedecho").error( # broken mailer is detectable "Password reset email failed for user %s", user_id, exc_info=True ) def forgot_submit(request: Request, email: str = Form("")): _require_multi() email = email.strip().lower() # Uniform response: never reveal whether an account exists. The email # dispatch happens in a background thread (see _send_reset_email), so # the response time is identical for known and unknown addresses. ip = _client_ip(request) if not _forgot_throttled(ip): with get_db() as db: user = db.execute( "SELECT id, email FROM users WHERE email = ?", (email,) ).fetchone() if user is not None: _record_forgot(ip) import asyncio task = asyncio.create_task( asyncio.to_thread(_send_reset_email, user["id"], user["email"]) ) _background_tasks.add(task) task.add_done_callback(_background_tasks.discard) return _render_auth(request, "forgot_password.html", sent=True) def reset_page(request: Request, token: str = ""): _require_multi() if not token: return _render_auth(request, "reset_password.html", error="Missing reset token.") import verification if verification.peek_token(token, "reset") is None: return _render_auth( request, "reset_password.html", error="This reset link is invalid or has expired.", ) return _render_auth(request, "reset_password.html", token=token) def reset_submit( request: Request, token: str = Form(""), password: str = Form(""), confirm: str = Form(""), ): _require_multi() import verification # Peek first: validation errors re-render with the SAME token (no # reissue, no throttle bypass, no reset-window extension). The token # is consumed only after the new password passes validation. uid = verification.peek_token(token, "reset") if token else None if uid is None: return _render_auth( request, "reset_password.html", error="This reset link is invalid or has expired.", ) errors: list[str] = [] if len(password) < _MIN_PASSWORD_LENGTH: errors.append(f"Password must be at least {_MIN_PASSWORD_LENGTH} characters.") if len(password) > _MAX_PASSWORD_LENGTH: errors.append(f"Password must be at most {_MAX_PASSWORD_LENGTH} characters.") if password != confirm: errors.append("Passwords do not match.") if errors: return _render_auth( request, "reset_password.html", error=" ".join(errors), token=token, ) if verification.consume_token(token, "reset") is None: return _render_auth( request, "reset_password.html", error="This reset link is invalid or has expired.", ) with get_db() as db: # Bump the session epoch: every previously issued session cookie # for this user dies with the password change. db.execute( "UPDATE users SET password_hash = ?," " session_epoch = session_epoch + 1 WHERE id = ?", (hash_password(password), uid), ) logging.getLogger("feedecho").info("User %s reset their password", uid) return RedirectResponse(url="/login?reset=1", status_code=302)