"""The HTTP façade's only job: authenticate, deserialise, translate errors.
Everything in this module is machinery rather than feature. Three pieces matter,
and each one is a default that holds for routes nobody has written yet:
require_session a before_request guard built as an ALLOWLIST, so a route
added tomorrow is protected by default rather than by
someone remembering to protect it
check_csrf a token on every state-changing request from our own pages
endpoint turns a core error into the right status code, and rolls the
session back on every failure path
"""
import functools
import logging
import secrets
from flask import jsonify, redirect, request
from flask import session as flask_session
from sqlalchemy.exc import IntegrityError
import settings
from app.core import (
Conflict,
CoreError,
Forbidden,
Invalid,
NotFound,
Principal,
)
from app.core import auth as auth_core
from db.db_declarations import User
from utils.db_wrapper import db_session
logger = logging.getLogger(__name__)
USER_KEY = "user_id"
ORG_KEY = "org_id"
class Unauthorized(Forbidden):
"""Authentication failed (no/invalid credentials) - a 401, distinct from a
permission denial (403). Both are Forbidden to the core, which does not know
HTTP; the façade tells them apart here."""
# Reachable without a session - an allowlist, so a route added tomorrow is
# protected by default instead of by remembering to protect it. Every entry here
# is a decision: say in a comment why the path cannot demand a session, and what
# defends it instead.
PUBLIC_PATHS = frozenset({
# The sign-in page and its endpoint cannot require the session they exist to
# create. /api/login is CSRF-exempt for the same reason: there is no session
# cookie yet for a forged request to ride on.
"/login", "/api/login",
# Liveness. App Engine's frontend swallows /healthz before it reaches the
# app, so /health is the alias that actually works there.
"/health", "/healthz",
# Google sign-in: the OAuth redirect dance runs before any session exists.
# The callback's forgery defence is the OAuth `state` parameter, not our CSRF
# token - the standard mechanism for this leg. Both answer with a redirect to
# /login when GOOGLE_CLIENT_* is unset.
"/auth/google/start", "/auth/google/callback",
})
# How a core failure becomes a status code. The core does not know these numbers;
# this table is the whole of that translation. Unauthorized is listed ahead of
# its Forbidden base because the lookup is by exact type: a missing or bad
# credential is a 401, distinct from a 403 permission denial.
STATUS = {Unauthorized: 401, Invalid: 400, Forbidden: 403, NotFound: 404,
Conflict: 409}
# --- who is calling ---------------------------------------------------------
def principal() -> Principal:
"""The dashboard caller, from the signed session cookie.
Scopes are re-derived from the membership on every request rather than stored
in the session, so changing someone's role takes effect on their next click.
"""
user_id = flask_session.get(USER_KEY)
if not user_id:
raise Unauthorized("Sign in to continue.")
user = db_session.get(User, user_id)
if user is None or not user.is_active:
flask_session.clear()
raise Unauthorized("That account is no longer active.")
return auth_core.principal_for_user(
db_session, user, org_id=flask_session.get(ORG_KEY))
def is_public(path: str) -> bool:
return path in PUBLIC_PATHS or path.startswith("/static/")
# Hosts we will build a security-sensitive external link (password reset, email
# verification, a payment provider's return URL) from when no canonical
# PUBLIC_BASE_URL is configured. Only loopback names qualify: a developer running
# locally has no proxy to set PUBLIC_BASE_URL, and no attacker can steer a
# victim's browser at someone else's localhost. Every other host must be named
# explicitly by PUBLIC_BASE_URL, so a deployment cannot have its links poisoned
# through the attacker-controllable Host header.
_TRUSTED_LOCAL_HOSTS = frozenset({"localhost", "127.0.0.1"})
def public_base_url() -> str:
"""The canonical scheme+host that security-sensitive external links must be
built from, with no trailing slash.
Prefers PUBLIC_BASE_URL - a canonical host an operator set, which the request
cannot influence. Only when it is unset is the origin derived from the
request, and then only for a trusted loopback host. Any other host without
PUBLIC_BASE_URL fails closed rather than trusting the Host header: an emailed
reset link built from a spoofed Host is classic password-reset poisoning.
"""
base = settings.Config.PUBLIC_BASE_URL
if base:
return base.rstrip("/")
host = request.host
if host.split(":")[0] not in _TRUSTED_LOCAL_HOSTS:
raise Invalid(
"This deployment cannot build a secure external link: set "
"PUBLIC_BASE_URL to the canonical public URL."
)
# request.scheme only - never X-Forwarded-Proto. This is a security-sensitive
# link builder, and the forwarding header is attacker-controllable unless a
# trusted proxy has already sanitised it.
return f"{request.scheme}://{host}"
# --- CSRF -------------------------------------------------------------------
CSRF_KEY = "csrf_token"
CSRF_HEADER = "X-CSRF-Token"
SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
def csrf_token() -> str:
if CSRF_KEY not in flask_session:
flask_session[CSRF_KEY] = secrets.token_urlsafe(32)
return flask_session[CSRF_KEY]
def check_csrf():
"""Reject a state-changing request that did not come from our own page.
SameSite=Lax and a JSON content type are the first two layers; this is the
third, because the first two are properties of the client."""
if request.method in SAFE_METHODS or is_public(request.path):
return None
expected = flask_session.get(CSRF_KEY)
presented = request.headers.get(CSRF_HEADER, "")
if not expected or not secrets.compare_digest(expected, presented):
logger.warning("CSRF check failed for %s %s", request.method, request.path)
return jsonify({"error": "Stale page. Reload and try again."}), 403
return None
def require_session():
"""A before_request guard: authentication, then CSRF.
public - no proof needed (sign-in page, health)
dashboard - a signed session cookie, plus CSRF on anything that writes
When you add a machine-to-machine surface (an API token, a provider's
webhook), give it a predicate like `is_api_basic(path)` and return early
here so it can authenticate itself: it holds no cookie, and there is no
ambient credential for a forged request to ride, so CSRF does not apply to it.
"""
if not is_public(request.path) and not flask_session.get(USER_KEY):
# An unauthenticated API call is a JSON 401, not an HTML redirect - a
# fetch() that receives a login page reports a JSON parse error instead
# of the thing that actually happened.
if request.path.startswith("/api/"):
return jsonify({"error": "Sign in to continue."}), 401
return redirect("/login")
return check_csrf()
# --- error translation ------------------------------------------------------
def _humanise(e: IntegrityError) -> str:
"""A constraint violation in words a user can act on. The driver's message
names the constraint, which is useful in a log and meaningless in a form."""
text = str(getattr(e, "orig", e)).lower()
if "unique" in text or "duplicate" in text:
return "That already exists."
if "foreign key" in text:
return "That refers to something which no longer exists."
if "not null" in text:
return "A required value is missing."
return "That change was rejected by a database constraint."
def endpoint(fn):
"""Dashboard endpoints: turn exceptions into `{error}` JSON.
The session is rolled back on every failure path, or a failed write poisons
every later query in the same request. The bare `except` is deliberate: an
unhandled error must not leak an internal message to a browser, so it logs
the traceback and answers with a fixed string.
"""
@functools.wraps(fn)
def wrapper(*args, **kwargs):
try:
return fn(*args, **kwargs)
except CoreError as e:
db_session.rollback()
return jsonify({"error": str(e)}), STATUS.get(type(e), 400)
except IntegrityError as e:
db_session.rollback()
logger.info("constraint rejected a write: %s", e.orig)
return jsonify({"error": _humanise(e)}), 409
except Exception:
db_session.rollback()
logger.exception("unhandled error in %s", fn.__name__)
return jsonify({"error": "Something went wrong. The details are in the log."}), 500
return wrapper