ai-agent/backend/users/caller.py

"""
Who is calling: the :class:`Caller` every ``/api`` request carries in
``request.state.caller`` (``core/auth.py`` resolves it), and what a guest may
reach.

Two kinds of people:

* the **owner** — whoever reaches the admin host (Authelia at Traefik), the
  trusted loopback / docker gateway, or presents ``INTERNAL_API_TOKEN``. This is
  every caller the app had before guests existed, and it keeps full access.
* a **guest** — a user from ``users.json`` holding a session cookie (the
  built-in login). Guests reach only the routes in :data:`GUEST_ROUTES`
  (default deny), and within them only the conversations they own or were
  shared and the projects they were granted.

The **guest surface** (``GUEST_HOSTS``, e.g. ``claude.lab.gabvdl.xyz``) is a
host on which nothing is ever owner: a request without a session there is
anonymous and may only log in.
"""

from __future__ import annotations

import dataclasses
import os
import re
import typing

from fastapi import Depends, HTTPException, Request

from users.store import LEVELS

OWNER_ID = "owner"
OWNER_NAME = os.environ.get("OWNER_NAME", "Gabriel")
SESSION_COOKIE = "aia_session"
GUEST_HOSTS = frozenset(h.strip().lower() for h in
                        os.environ.get("GUEST_HOSTS", "").split(",") if h.strip())


@dataclasses.dataclass(frozen=True)
class Caller:
    id: str
    role: str                      # "owner" | "guest"
    name: str
    email: str | None = None
    readonly: bool = False
    projects: dict = dataclasses.field(default_factory=dict)   # slug → level
    models: tuple = ()
    budget_usd: float | None = None

    @property
    def is_owner(self) -> bool:
        return self.role == "owner"

    def level(self, slug: str | None) -> str | None:
        if self.is_owner:
            return "ship"
        return self.projects.get(slug or "")

    def can(self, slug: str | None, need: str = "view") -> bool:
        lvl = self.level(slug)
        return lvl is not None and LEVELS.index(lvl) >= LEVELS.index(need)


OWNER = Caller(id=OWNER_ID, role="owner", name=OWNER_NAME)
OWNER_READONLY = dataclasses.replace(OWNER, readonly=True)


def guest_from_record(u: dict) -> Caller:
    return Caller(id=u["id"], role="guest", name=u.get("name") or u["id"],
                  email=u.get("email"), projects=dict(u.get("projects") or {}),
                  models=tuple(u.get("models") or ()),
                  budget_usd=u.get("budgetUsd"))


def request_host(request: Request) -> str:
    return (request.headers.get("host") or "").split(":")[0].strip().lower()


def on_guest_surface(request: Request) -> bool:
    return bool(GUEST_HOSTS) and request_host(request) in GUEST_HOSTS


# ── what a guest may reach (default deny) ────────────────────────────────────
# (methods, path regex). Everything else answers 403 to a guest before any
# handler runs; the handlers listed here filter by conversation / project.
_R = typing.Tuple[frozenset, re.Pattern]


def _r(methods: str, pattern: str) -> _R:
    return frozenset(methods.split()), re.compile(pattern)


GUEST_ROUTES: tuple[_R, ...] = (
    _r("GET PUT", r"^/api/me$"),
    # A signed-in browser may still open a sign-in link or ask for one (a
    # second device's link, a reopened mail): refusing them stranded a guest
    # whose first sign-in had worked but whose page never left /login.
    _r("POST", r"^/api/auth/(request-link|login|logout)$"),
    _r("GET", r"^/api/events$"),
    _r("GET", r"^/api/conversations$"),
    _r("GET", r"^/api/conversation$"),
    _r("GET", r"^/api/conversation-summary$"),
    _r("GET", r"^/api/search-index$"),
    _r("GET", r"^/api/subagents$"),
    _r("GET", r"^/api/conversation-meta$"),
    _r("GET PUT", r"^/api/conversation-share$"),
    _r("POST", r"^/api/(spawn|resume|interrupt|fork)$"),
    _r("GET", r"^/api/forms$"),
    _r("GET", r"^/api/forms/[^/]+$"),
    _r("POST", r"^/api/forms/[^/]+/(submit|cancel)$"),
    _r("GET", r"^/api/notifications$"),
    _r("POST", r"^/api/notifications/(seen|seed)$"),
    _r("GET", r"^/api/notify/[^/]+$"),
    _r("GET", r"^/api/notify-log$"),
    _r("GET", r"^/api/ask/[^/]+$"),
    _r("POST", r"^/api/ask/[^/]+/answer$"),
    _r("GET", r"^/api/projects$"),
    _r("GET", r"^/api/project$"),
    _r("GET", r"^/api/(conversation-commits|conversation-diff|"
              r"conversation-uncommitted-diff|commit-diff|diff-blob)$"),
    _r("GET", r"^/api/models$"),
    _r("GET PUT DELETE", r"^/api/ui-state/[^/]+$"),
    _r("POST", r"^/api/upload$"),
    _r("GET", r"^/api/upload-file$"),
    _r("GET", r"^/api/(project-asset|artefact)$"),
    _r("GET", r"^/api/(avatar/manifest|avatar-asset)$"),
    _r("GET", r"^/api/guest-skills$"),
    _r("GET", r"^/api/deploy-status$"),
)

# What an anonymous caller on the guest surface may do: log in.
ANON_ROUTES: tuple[_R, ...] = (
    _r("POST", r"^/api/auth/(request-link|login|logout)$"),
)


def allowed(routes: tuple[_R, ...], method: str, path: str) -> bool:
    if method in ("HEAD", "OPTIONS"):
        method = "GET"
    return any(method in ms and rx.match(path) for ms, rx in routes)


# ── route dependencies ───────────────────────────────────────────────────────
def get_caller(request: Request) -> Caller:
    c = getattr(request.state, "caller", None)
    if c is None:
        raise HTTPException(401, {"message": "login required", "kind": "login_required"})
    return c


CallerDep = typing.Annotated[Caller, Depends(get_caller)]


def require_owner(request: Request) -> Caller:
    c = get_caller(request)
    if not c.is_owner:
        raise HTTPException(403, "owner only")
    return c


OwnerDep = typing.Annotated[Caller, Depends(require_owner)]