"""Authentication: password hashing, sessions, CSRF, login lockout. argon2id hashing with constant-time verify. No account enumeration — an unknown email and a wrong password give the same error, and an unknown email still pays a dummy verify so the timing matches. Sessions are server-side: a 256-bit token in an HttpOnly SameSite=Lax cookie, of which only the SHA-256 is stored, with sliding expiry. CSRF is session-bound, except pre-auth forms which double-submit a cookie token. API keys hash the same way and scope to one portfolio. Every refusal lives in `load_session`, never in a route (decisions.md #3). """ from __future__ import annotations import datetime as dt import hashlib import hmac import ipaddress import logging import secrets from contextlib import contextmanager from dataclasses import dataclass from typing import Iterator from argon2 import PasswordHasher from argon2.exceptions import InvalidHashError, VerifyMismatchError from fastapi import HTTPException, Request from sqlalchemy import select from sqlalchemy.orm import Session as DbSession from appcore import ensure_utc from . import tenancy from .models import ( WRITE_ROLES, ApiKey, LoginAttempt, Portfolio, PortfolioMember, User, UserSession, row_id, ) from .settings import PortfolioSettings log = logging.getLogger(__name__) _ph = PasswordHasher() # Equalises timing when the email is unknown — verifying this costs the same as # verifying a real hash, so a failure reveals nothing about account existence. _DUMMY_HASH = _ph.hash("portfolio-timing-equaliser") PRE_AUTH_CSRF_COOKIE = "pf_csrf" def _utcnow() -> dt.datetime: return dt.datetime.now(dt.timezone.utc) # --------------------------------------------------------------------------- # # Passwords # --------------------------------------------------------------------------- # def hash_password(password: str) -> str: return _ph.hash(password) def verify_password(stored_hash: str | None, password: str) -> bool: """Check a password, in constant-ish time whether or not the account exists. The `is None` test is load-bearing and must not become a falsy test. An EMPTY stored hash is falsy but is still a stored hash: treating it as "absent" verifies the supplied password against `_DUMMY_HASH` and then reports success, so anyone who knows the equaliser passphrase — a literal in this file, and public the moment this repository is — could sign in to any account whose hash was blank. """ target = _DUMMY_HASH if stored_hash is None else stored_hash try: _ph.verify(target, password) return stored_hash is not None except (VerifyMismatchError, InvalidHashError): return False def needs_rehash(stored_hash: str) -> bool: try: return _ph.check_needs_rehash(stored_hash) except InvalidHashError: return True def password_problem(password: str) -> str | None: """Length only: at least 10 characters, at most 64. No mix of character types and no banned list — NIST 800-63B-4 and ASVS both rule out the first, and how strong a password is gets shown, not enforced. `len` counts code points, as NIST asks. Checked where a password is chosen, never at sign-in, so an account with a longer one from before still gets in.""" if len(password or "") < 10: return "Password must be at least 10 characters." if len(password) > 64: return "Password must be at most 64 characters." return None def email_problem(email: str) -> str | None: email = (email or "").strip() if "@" not in email or "." not in email.split("@")[-1] or len(email) < 6: return "Enter a valid email address." # Measured as it will be stored, lowercased: some characters get longer # ("İ".lower() is two), so 300 of them pass before and overflow after. if len(email.lower()) > User.email.type.length: return "That email address is too long." return None def email_fits(email: str) -> bool: """Whether an email can be recorded against a sign-in attempt at all. The sign-in and recovery forms take an email from anyone, and write it to `login_attempt` before they know whether the account exists. One longer than the column would be a 500 on Postgres, so those forms refuse it first, with the answer they give for any wrong email. """ return len(email) <= LoginAttempt.email.type.length # --------------------------------------------------------------------------- # # Tokens / sessions # --------------------------------------------------------------------------- # def new_token() -> tuple[str, str]: """(raw_token, sha256_hex) — store the hash, hand out the raw token.""" raw = secrets.token_urlsafe(32) return raw, hash_token(raw) def hash_token(raw: str) -> str: return hashlib.sha256(raw.encode()).hexdigest() def create_session( db: DbSession, user: User, settings: PortfolioSettings, active_portfolio_id: int | None = None, awaiting_totp: bool = False, ip: str | None = None, user_agent: str | None = None, recovery_spent: bool = False, ) -> str: raw, token_hash = new_token() now = _utcnow() db.add( UserSession( token_hash=token_hash, user_id=user.id, active_portfolio_id=active_portfolio_id, csrf_token=secrets.token_urlsafe(32), created_at=now, last_seen_at=now, expires_at=now + dt.timedelta(days=settings.auth.session_ttl_days), awaiting_totp=awaiting_totp, # A code already spent to get here cannot be followed by a second # one at the 2FA step — see UserSession.recovery_spent. recovery_spent=recovery_spent, ip=ip, user_agent=(user_agent or "")[:200] or None, ) ) db.flush() return raw def session_deadlines( row: UserSession, settings: PortfolioSettings ) -> tuple[dt.datetime, dt.datetime]: """When this session dies from inactivity, and when it dies regardless. Returned together because the effective expiry is whichever comes first, and every caller — the middleware, the keepalive endpoint, the countdown the browser renders — has to agree on that answer. """ idle_deadline = ensure_utc(row.last_seen_at) + dt.timedelta( minutes=settings.auth.session_idle_minutes ) hard_deadline = ensure_utc(row.created_at) + dt.timedelta( days=settings.auth.session_absolute_days ) return idle_deadline, hard_deadline def session_expired(row: UserSession, settings: PortfolioSettings) -> bool: """Whether this session is over, by any of the three clocks. Three, not one: the stored `expires_at`, the idle window, and the absolute cap. The sliding renewal can push the first two out indefinitely — a session used every day would otherwise live forever, which is precisely the stolen-cookie case the absolute cap exists for. """ now = _utcnow() if ensure_utc(row.expires_at) <= now: return True idle_deadline, hard_deadline = session_deadlines(row, settings) return idle_deadline <= now or hard_deadline <= now def _live_session( db: DbSession, raw_token: str | None, settings: PortfolioSettings ) -> UserSession | None: """The session row for this token if it exists and hasn't expired. Says nothing about whether it is *authenticated* — see `load_session`. Split out so the second-factor step has a way to reach its own half-authenticated session without weakening the check everything else goes through. """ if not raw_token: return None row = db.scalar( select(UserSession).where(UserSession.token_hash == hash_token(raw_token)) ) if row is None: return None # `settings` is REQUIRED rather than defaulted. Making it optional would # mean a caller that forgot it silently lost the idle and absolute # timeouts, and a security control that fails open when you forget an # argument is worse than no control — you would believe you had one. if session_expired(row, settings): db.delete(row) db.flush() return None return row def load_session( db: DbSession, raw_token: str | None, settings: PortfolioSettings ) -> UserSession | None: """The session behind this token, or None. A session still `awaiting_totp` is treated as no session at all. That refusal lives HERE, at the one place every route resolves a cookie, rather than in each route — the password step alone must never reach a page, and a check that has to be remembered is a check that will eventually be forgotten. The same argument applies to the timeouts. """ row = _live_session(db, raw_token, settings) if row is None or row.awaiting_totp: return None return row def load_pending_session( db: DbSession, raw_token: str | None, settings: PortfolioSettings ) -> UserSession | None: """The half-authenticated session, for the code-entry step ONLY. Deliberately narrow and deliberately greppable: this is the single function that can see a session that has not finished authenticating, and it refuses a fully-authenticated one so it can't be used to sidestep anything. """ row = _live_session(db, raw_token, settings) if row is None or not row.awaiting_totp: return None return row def destroy_session(db: DbSession, raw_token: str | None) -> None: if not raw_token: return row = db.scalar( select(UserSession).where(UserSession.token_hash == hash_token(raw_token)) ) if row is not None: db.delete(row) db.flush() def destroy_sessions_for(db: DbSession, user_id: int) -> None: """Log a user out everywhere — after a password change or deactivation.""" for row in db.scalars(select(UserSession).where(UserSession.user_id == user_id)): db.delete(row) db.flush() # --------------------------------------------------------------------------- # # Lockout # --------------------------------------------------------------------------- # def is_locked(db: DbSession, email: str, settings: PortfolioSettings) -> bool: """Whether this account has failed too often to try again yet. Counted per ACCOUNT, across every source address. It was per (email, address), so that nobody could lock somebody else out from outside — and that meant anyone who could vary the address got a fresh allowance from each one: a spoofed X-Forwarded-For, or simply many machines. At the six-digit code step that is a brute force by someone who already has the password. GHSA-gqj4-vm54-mp3j. The cost is accepted deliberately: anybody who knows an email address can keep that account's password and code steps locked by failing on purpose. Passkeys and single sign-on do not pass through here, so they still work while it lasts. The address is still recorded with every attempt; it just no longer decides anything. """ rl = settings.auth.rate_limit window_start = _utcnow() - dt.timedelta(minutes=rl.window_minutes) fails = db.scalars( select(LoginAttempt.created_at) .where( LoginAttempt.email == email.lower(), LoginAttempt.success.is_(False), LoginAttempt.created_at >= window_start, ) .order_by(LoginAttempt.created_at.desc()) ).all() if len(fails) < rl.max_attempts: return False return _utcnow() < ensure_utc(fails[0]) + dt.timedelta(minutes=rl.lockout_minutes) def record_attempt(db: DbSession, email: str, ip: str, success: bool) -> None: db.add(LoginAttempt(email=email.lower(), ip=ip, success=success, created_at=_utcnow())) db.flush() # --------------------------------------------------------------------------- # # Request-scoped context # --------------------------------------------------------------------------- # @dataclass class Membership: portfolio_id: int name: str role: str @dataclass class AuthContext: user: User session: UserSession memberships: list[Membership] active_portfolio_id: int | None role: str | None # role in the ACTIVE portfolio @property def is_admin(self) -> bool: """App-wide account administration (adding people), not portfolio rights.""" return bool(self.user.is_admin) @property def is_owner(self) -> bool: return self.role == "owner" @property def can_write(self) -> bool: """Viewers read everything and change nothing.""" return self.role in WRITE_ROLES @property def active(self) -> Membership | None: return next( (m for m in self.memberships if m.portfolio_id == self.active_portfolio_id), None, ) def _is_trusted(peer: str, trusted: list[str]) -> bool: """Whether a request arrived from a peer allowed to speak for someone else.""" if not trusted: return False try: address = ipaddress.ip_address(peer) except ValueError: return False for entry in trusted: try: if address in ipaddress.ip_network(entry, strict=False): return True except ValueError: log.warning("ignoring unparseable auth.trusted_proxies entry %r", entry) return False def is_secure_request(request: Request, settings: PortfolioSettings) -> bool: """Whether this request actually reached us over TLS. Behind a proxy the app always sees plain HTTP, so the answer comes from `X-Forwarded-Proto` — and that header is believed on exactly the same terms as `X-Forwarded-For`: only from a configured proxy. Anyone can set it, and believing it from an arbitrary peer would let a caller assert their connection is encrypted when it is not. """ if request.url.scheme == "https": return True peer = request.client.host if request.client else None trusted = list(settings.auth.trusted_proxies) if peer and _is_trusted(peer, trusted): return request.headers.get("x-forwarded-proto", "").lower() == "https" return False def insecure_login_problem(request: Request, settings: PortfolioSettings) -> str | None: """Why signing in here cannot work, or None. The failure this catches is silent and baffling. With `cookie_secure` on, a browser on plain HTTP **discards the session cookie**: the password is accepted, `Set-Cookie` is sent, the browser throws it away, and the person lands back on the login page with nothing to explain it. They will try their password again, and again. No second factor helps — the cookie cannot be stored however you authenticated — so the only useful response is to say so before they try. """ if not settings.auth.cookie_secure or is_secure_request(request, settings): return None return ( "This instance is configured for HTTPS, but you have reached it over " "plain HTTP. Your browser would throw the session cookie away, so " "signing in cannot work from this address — you would simply arrive " "back here. Use the https:// address instead. " "(If you are the administrator and there is no HTTPS yet, set " "auth.cookie_secure to false.)" ) def client_ip(request: Request) -> str: """The address this request came from, for lockout counting. X-Forwarded-For is set by whoever sends it, so it is believed ONLY when the immediate peer is a configured proxy. Otherwise anyone could pick their own address per request — which both defeats lockout entirely (a fresh identity each attempt) and turns it into a weapon (forge somebody else's address and lock them out). **Read from the right, not the left.** A proxy that appends (nginx's `$proxy_add_x_forwarded_for`, and so Nginx Proxy Manager) adds the peer it received from to whatever the client sent. The leftmost entry is then the client's own claim, and taking it hands back exactly the choice this function exists to remove. Only the entries our own proxies wrote can be believed, so the answer is the first one from the right that is not itself a trusted proxy: the address the outermost trusted proxy saw. That is right for proxies that overwrite the header too, where there is only one entry. Repeated header lines are joined first, in order. A client's line comes before the one the proxy adds, and reading only the first would be the leftmost-entry mistake again. """ settings = getattr(getattr(request.app, "state", None), "settings", None) trusted = list(settings.auth.trusted_proxies) if settings is not None else [] peer = request.client.host if request.client else None if peer and _is_trusted(peer, trusted): hops = [hop.strip() for line in request.headers.getlist("x-forwarded-for") for hop in line.split(",") if hop.strip()] for hop in reversed(hops): if not _is_trusted(hop, trusted): return hop # Every hop is one of our own proxies: a request that started inside # the trusted range, such as the proxy's own health check. if hops: return hops[0] return peer or "unknown" # Paths that must NOT count as activity — decisions.md #19. Anything the page # polls on a timer belongs here; `/session/keepalive` deliberately does not, # because a button press is a person. SLIDING_EXEMPT = frozenset( {"/session/status", "/healthz", "/readyz", "/feed/status", # The log console tails on a self-rescheduling timer, so leaving Admin → # Settings open renewed the session indefinitely. Found by deriving the # list from the scripts instead of maintaining it by hand. "/admin/logs.json"} ) def slides_window(path: str) -> bool: """Whether a request to this path counts as the user being present.""" return path not in SLIDING_EXEMPT and not path.startswith("/static/") def load_auth(request: Request, db: DbSession) -> AuthContext | None: """Resolve the logged-in user from the cookie, sliding the expiry. "Sliding" means real activity only — see `SLIDING_EXEMPT`. """ settings: PortfolioSettings = request.app.state.settings row = load_session(db, request.cookies.get(settings.auth.cookie_name), settings) if row is None: return None user = db.get(User, row.user_id) if user is None or not user.is_active: db.delete(row) db.flush() return None now = _utcnow() if slides_window(request.url.path): row.last_seen_at = now # Never past the absolute cap: the whole point of that cap is that # renewal cannot push it out. Without this clamp a daily user's session # would be immortal, which is the stolen-cookie case it exists for. _, hard_deadline = session_deadlines(row, settings) row.expires_at = min( now + dt.timedelta(days=settings.auth.session_ttl_days), hard_deadline ) rows = db.execute( select(PortfolioMember, Portfolio) .join(Portfolio, Portfolio.id == PortfolioMember.portfolio_id) .where(PortfolioMember.user_id == user.id) .order_by(Portfolio.name) ).all() memberships = [Membership(p.id, p.name, m.role) for (m, p) in rows] active_id = row.active_portfolio_id valid = {m.portfolio_id for m in memberships} if active_id not in valid: # Removed from that portfolio (or it was deleted) — fall back rather # than leaving the session pointed at something it can't read. active_id = next(iter(sorted(valid)), None) row.active_portfolio_id = active_id role = next((m.role for m in memberships if m.portfolio_id == active_id), None) return AuthContext( user=user, session=row, memberships=memberships, active_portfolio_id=active_id, role=role, ) # --------------------------------------------------------------------------- # # CSRF # --------------------------------------------------------------------------- # def expected_csrf(request: Request, db: DbSession) -> str | None: settings: PortfolioSettings = request.app.state.settings row = load_session(db, request.cookies.get(settings.auth.cookie_name), settings) if row is not None: return row.csrf_token return request.cookies.get(PRE_AUTH_CSRF_COOKIE) async def verify_csrf(request: Request, db: DbSession) -> None: """403 unless the request echoes the session's CSRF token (X-CSRF-Token header or a `_csrf` form field).""" provided = request.headers.get("x-csrf-token") if provided is None: form = await request.form() value = form.get("_csrf") provided = value if isinstance(value, str) else None expected = expected_csrf(request, db) if not expected or not provided or not hmac.compare_digest(provided, expected): raise HTTPException(status_code=403, detail="Invalid or missing CSRF token") def ensure_pre_auth_csrf(request: Request) -> str: return request.cookies.get(PRE_AUTH_CSRF_COOKIE) or secrets.token_urlsafe(32) async def verify_pre_auth_csrf(request: Request) -> None: """The double-submit check on its own, with no database in the picture. `verify_csrf` prefers a session token and needs a session to look one up. The setup wizard's first steps run before there is a database at all, let alone a session — so they need the half of the check that does not touch one. Same guarantee for those steps: a cross-site form post carries no cookie it can echo, so it cannot produce a match. """ form = await request.form() value = form.get("_csrf") provided = value if isinstance(value, str) else request.headers.get("x-csrf-token") expected = request.cookies.get(PRE_AUTH_CSRF_COOKIE) if not expected or not provided or not hmac.compare_digest(provided, expected): raise HTTPException(status_code=403, detail="Invalid or missing CSRF token") # --------------------------------------------------------------------------- # # Session helpers shared by every router # --------------------------------------------------------------------------- # @contextmanager def scoped_session(request: Request) -> Iterator[tuple[AuthContext, DbSession]]: """DB session confined to the logged-in user, plus their auth context. Commits on success. Routes reach this only behind the require_login middleware, so a missing session means it expired mid-request — 401 beats quietly running the query unscoped. """ db = request.app.state.session_factory() try: ctx = load_auth(request, db) if ctx is None: raise HTTPException(401, "session expired — reload and log in") if ctx.active_portfolio_id is None: raise HTTPException(403, "your account isn't a member of any portfolio") tenancy.bind(db, ctx.active_portfolio_id, ctx.user.id) yield ctx, db db.commit() except Exception: db.rollback() raise finally: db.close() @contextmanager def open_session(request: Request) -> Iterator[DbSession]: """Unbound session for the pre-login flows (users, sessions, attempts). Personal-data queries raise on it — see tenancy.py.""" db = request.app.state.session_factory() try: yield db db.commit() except Exception: db.rollback() raise finally: db.close() # --------------------------------------------------------------------------- # # API keys — machine access for the other apps in the family # --------------------------------------------------------------------------- # API_KEY_PREFIX = "pfk" def new_api_key() -> tuple[str, str, str]: """Return (raw_key, sha256_hex, public_prefix). Format `pfk_<8-char id>_`: the middle chunk is stored so a key can be named and revoked from the UI without keeping anything usable. """ ident = secrets.token_hex(4) secret = secrets.token_urlsafe(32) raw = f"{API_KEY_PREFIX}_{ident}_{secret}" return raw, hash_token(raw), f"{API_KEY_PREFIX}_{ident}" def key_from_request(request: Request) -> str | None: """`Authorization: Bearer ` or `X-API-Key: `.""" header = request.headers.get("authorization", "") if header.lower().startswith("bearer "): return header[7:].strip() or None return request.headers.get("x-api-key") or None def load_api_key(db: DbSession, raw: str | None) -> ApiKey | None: """Resolve a raw key to a live ApiKey row, touching last_used_at.""" if not raw: return None row = db.scalar(select(ApiKey).where(ApiKey.key_hash == hash_token(raw))) if row is None or row.revoked_at is not None: return None row.last_used_at = _utcnow() return row def key_reach(db: DbSession, key: ApiKey) -> list[tuple[Portfolio, str | None]]: """Every portfolio this key was given, each with the role its creator holds there NOW — None if they are no longer a member. A creator who is no longer active never gets this far: `api_session` refuses their key first, and the profile page lists only the signed-in person's own keys. A key belongs to the person who made it and never does more than they can do today — GHSA-cjcx-g69x-63jq, decisions.md #128. They saw the raw key, so a key that outlived their access would be their access, kept. Checked on every use rather than by revoking keys when someone leaves: that would need remembering at every way access ends (deactivation, removal, a demotion to viewer) and would still miss the next one added. """ roles: dict[int, str] = {} if key.created_by is not None: roles = dict(db.execute( select(PortfolioMember.portfolio_id, PortfolioMember.role) .where(PortfolioMember.user_id == key.created_by) ).all()) return [(portfolio, roles.get(portfolio.id)) for portfolio in key.portfolios] @dataclass class ApiAccess: """What one API request is acting on. A write it may not make never gets this far: `api_session` refuses it.""" key: ApiKey portfolio_id: int | None # None only where no portfolio is chosen role: str | None = None # its creator's role there, for the owner's writes def _chosen_portfolio(request: Request, reach: list[tuple[Portfolio, str | None]]) -> int: """Which of the key's portfolios this request means. `?portfolio=`, on every method. Optional while the key was GIVEN only one, whether or not its creator can still reach it — the contract a client is written against must not change shape because somebody's membership did. """ given = request.query_params.get("portfolio", "").strip() linked = [portfolio.id for portfolio, _role in reach] if not given: if len(linked) == 1: return linked[0] if not linked: raise HTTPException(403, "this key does not reach any portfolio") raise HTTPException( 400, "this key reaches more than one portfolio: say which with " "?portfolio= (GET /api/v1/portfolios lists them)") chosen = row_id(given) if chosen is None or chosen not in linked: # One answer for "not one of this key's" and "no such portfolio", so a # key cannot be used to learn which ids exist. raise HTTPException(404, "this key does not reach that portfolio") return chosen @contextmanager def api_session(request: Request, write: bool = False, choose: bool = True) -> Iterator[tuple[ApiAccess, DbSession]]: """A session bound to the portfolio this request means. 401 without a valid key, or when its creator's account is gone or deactivated — the key is dead everywhere. 403 when its creator is no longer a member of the chosen portfolio, or a write is beyond the key or beyond them. `choose=False` is for the one route that lists portfolios rather than acting in one; its session is bound to none. """ db = request.app.state.session_factory() try: key = load_api_key(db, key_from_request(request)) if key is None: raise HTTPException( 401, "missing or invalid API key (Authorization: Bearer pfk_...)" ) creator = db.get(User, key.created_by) if key.created_by is not None else None if creator is None or not creator.is_active: raise HTTPException( 401, "the person who created this key no longer has an active account, " "so the key has stopped working" ) access = ApiAccess(key=key, portfolio_id=None) if choose: reach = key_reach(db, key) portfolio_id = _chosen_portfolio(request, reach) role = dict((portfolio.id, r) for portfolio, r in reach)[portfolio_id] if role is None: raise HTTPException( 403, "the person who created this key no longer has access to " "this portfolio") if write and not key.can_write: raise HTTPException(403, "this key is read-only") if write and role not in WRITE_ROLES: raise HTTPException( 403, "the person who created this key can no longer write to this " "portfolio, so the key is read-only here") tenancy.bind(db, portfolio_id, key.created_by) access = ApiAccess(key=key, portfolio_id=portfolio_id, role=role) yield access, db db.commit() except Exception: db.rollback() raise finally: db.close()