auth.py
Rama master · réplica de lectura del repositorio original.
"""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)