ai-agent/backend/core/auth.py

"""The trusted-caller gate: every /api request must come from Traefik, the
docker gateway, localhost or a caller holding INTERNAL_API_TOKEN — or carry a
guest's session cookie (users/). It also decides *who* the caller is
(``request.state.caller``, a :class:`users.caller.Caller`)."""

import dataclasses
import os
import pathlib
import socket
import threading
import time

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from users.caller import (ANON_ROUTES, GUEST_ROUTES, OWNER, OWNER_READONLY,
                          SESSION_COOKIE, allowed, guest_from_record,
                          on_guest_surface, request_host)

# ── trusted-caller gate ──────────────────────────────────────────────────────
# The backend has no auth of its own: the browser reaches it through Traefik
# (behind Authelia) and host tooling through the 127.0.0.1-published port. But
# the container also sits on the shared `main` docker network, where any other
# container could hit it directly — and a PUT into `.claude/` (hooks) or a
# /api/spawn is host-level code execution. So every /api request must come from
# a trusted peer: Traefik (the Authelia-guarded edge), the docker gateway (how
# host-originated connections to the published port appear), localhost, or a
# caller presenting the shared INTERNAL_API_TOKEN. Only /api/health stays open
# (deploy probes). Static assets are public-shell only, so they stay open too.
INTERNAL_API_TOKEN = os.environ.get("INTERNAL_API_TOKEN", "")
# A read-ONLY API key: a caller presenting it (X-API-Key) may hit **safe**
# (GET/HEAD/OPTIONS) /api/* routes only — every mutating method is refused. This
# is what the desk phone holds to fetch unread notifications: it must never be
# able to spawn a session or PUT into .claude/ (that's host-level code exec).
# Comma-separated to allow more than one key. Empty ⇒ feature off.
READONLY_API_KEYS = frozenset(
    k.strip() for k in os.environ.get("READONLY_API_KEY", "").split(",")
    if k.strip())
# Methods a read-only key is allowed to use.
_SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
TRAEFIK_HOST = os.environ.get("TRAEFIK_HOST", "traefik")
_TRUSTED_TTL_S = 30.0
_trusted_lock = threading.Lock()
_trusted_cache: tuple[float, frozenset[str]] = (0.0, frozenset())


def _default_gateway_ips() -> set[str]:
    """The container's default-gateway IP(s) (/proc/net/route, IPv4)."""
    ips: set[str] = set()
    try:
        for line in pathlib.Path("/proc/net/route").read_text().splitlines()[1:]:
            f = line.split()
            if len(f) >= 3 and f[1] == "00000000":  # default route
                ips.add(socket.inet_ntoa(bytes.fromhex(f[2])[::-1]))
    except (OSError, ValueError):
        pass
    return ips


def _trusted_ips() -> frozenset[str]:
    """Gateway + Traefik IPs, re-resolved at most every _TRUSTED_TTL_S."""
    global _trusted_cache
    now = time.monotonic()
    with _trusted_lock:
        ts, ips = _trusted_cache
        if now - ts < _TRUSTED_TTL_S and ips:
            return ips
    fresh = {"127.0.0.1", "::1"} | _default_gateway_ips()
    try:
        for info in socket.getaddrinfo(TRAEFIK_HOST, None):
            fresh.add(info[4][0])
    except OSError:
        pass
    out = frozenset(fresh)
    with _trusted_lock:
        _trusted_cache = (now, out)
    return out


def _same_origin(request: Request) -> bool:
    """A browser's mutating request must come from this very host. The
    session cookie is SameSite=Lax already; this closes the rest (a sibling
    subdomain, an old browser). No Origin ⇒ not a browser ⇒ nothing to check."""
    origin = request.headers.get("origin")
    if not origin:
        return True
    host = origin.split("://", 1)[-1].split("/", 1)[0].split(":")[0].lower()
    return host == request_host(request)


def _guest_session(request: Request):
    """The guest a valid session cookie belongs to, else None."""
    token = request.cookies.get(SESSION_COOKIE)
    if not token:
        return None
    deps = request.app.state.ai
    rec = deps.auth.session(token)
    u = deps.users.active(rec["user"]) if rec and rec.get("user") else None
    if not u:
        return None
    guest = guest_from_record(u)
    # The owner impersonating this guest: the guest in every respect, but
    # /api/me says so (the PWA's "back to admin" bar).
    return dataclasses.replace(guest, impersonator=rec["imp"]) if rec.get("imp") else guest


async def _trusted_caller_gate(request: Request, call_next):
    path = request.url.path
    if not path.startswith("/api/") or path == "/api/health":
        return await call_next(request)

    # A guest's session (the built-in login, users/) decides the request
    # outright, wherever it comes from: only GUEST_ROUTES, and handlers then
    # filter by conversation and project.
    guest = _guest_session(request)
    if guest is not None:
        if not allowed(GUEST_ROUTES, request.method, path):
            return JSONResponse({"detail": "not available to guests"}, status_code=403)
        if request.method not in _SAFE_METHODS and not _same_origin(request):
            return JSONResponse({"detail": "cross-site request refused"},
                                status_code=403)
        request.state.caller = guest
        return await call_next(request)

    # The guest surface (GUEST_HOSTS): nothing reached through it is ever the
    # owner, even though it arrives from Traefik — without a session, the only
    # thing to do there is log in.
    if on_guest_surface(request):
        if allowed(ANON_ROUTES, request.method, path):
            request.state.caller = None
            return await call_next(request)
        return JSONResponse({"detail": {"message": "login required",
                                        "kind": "login_required"}},
                            status_code=401)

    # A read-only API key downgrades the caller to safe methods, whatever the
    # source IP — the phone reaches us over a "trusted" host loopback but is
    # only allowed to read (fetch notifications), never to spawn/mutate. So
    # this check comes before the IP one and, when the key matches, it decides
    # the request outright (a read-only key on a POST is refused, not silently
    # upgraded by a trusted IP).
    api_key = request.headers.get("x-api-key", "")
    if api_key and api_key in READONLY_API_KEYS:
        if request.method not in _SAFE_METHODS:
            return JSONResponse(
                {"detail": "read-only api key: method not allowed"},
                status_code=403)
        request.state.caller = OWNER_READONLY
        return await call_next(request)

    client = request.client.host if request.client else ""
    if client not in _trusted_ips():
        tok = request.headers.get("x-internal-auth", "")
        if not (INTERNAL_API_TOKEN and tok == INTERNAL_API_TOKEN):
            return JSONResponse({"detail": "forbidden"}, status_code=403)
    request.state.caller = OWNER
    return await call_next(request)


def install(app: FastAPI) -> None:
    """Register the gate as the app's HTTP middleware."""
    app.middleware("http")(_trusted_caller_gate)