PacketStation/server/packetstation/api/deps.py

"""Request dependencies: the database connection, the directory, the session key and the signed-in Principal."""

from __future__ import annotations

import ipaddress
import logging
import uuid
from collections.abc import Iterator
from contextlib import AbstractContextManager
from dataclasses import dataclass
from datetime import timedelta
from typing import Any

from fastapi import Depends, Request, Response
from psycopg.types.json import Jsonb

from .. import audit, db, events
from ..auth import devices, sessions
from ..auth import directory as userdir
from ..auth import groups as groupmap
from ..auth.ldap import Directory, DirectoryProtocol, LdapUnavailable
from ..proxy import client_ip
from ..settings import SecretFileError, Settings
from .errors import ApiError

log = logging.getLogger(__name__)

LAST_SEEN_SLACK = timedelta(seconds=30)


def desk_roles(roles: list[str]) -> bool:
    """Admin and Clerical get the longer desk idle limits."""
    return any(r == "admin" or r.startswith("clerical") for r in roles)


def idle_minutes(s: Settings, roles: list[str]) -> tuple[int, int]:
    """(sign-out, screen lock) idle minutes for a session with these roles."""
    if desk_roles(roles):
        return max(s.session_idle_minutes, s.session_desk_minutes), max(s.session_lock_minutes, s.session_desk_minutes)
    return s.session_idle_minutes, s.session_lock_minutes


# ---------------------------------------------------------------- plumbing


def settings_of(request: Request) -> Settings:
    s: Settings = request.app.state.settings
    return s


def directory_of(request: Request) -> DirectoryProtocol:
    d: DirectoryProtocol | None = request.app.state.directory
    if d is None:
        d = Directory(settings_of(request))
        request.app.state.directory = d
    return d


def session_key(request: Request) -> bytes:
    key: bytes | None = getattr(request.app.state, "session_key", None)
    if key is None:
        try:
            key = settings_of(request).session_key()
        except SecretFileError as e:
            log.error("session key unavailable: %s", e)
            raise ApiError(503, "server_misconfigured", "The server is missing its session key.") from None
        request.app.state.session_key = key
    return key


class Db:
    """One pooled app-role connection per request, opened on first use, in autocommit mode."""

    def __init__(self, settings: Settings):
        self.settings = settings
        self._cm: AbstractContextManager[Any] | None = None
        self._conn: Any = None

    @property
    def conn(self) -> Any:
        if self._conn is None:
            try:
                cm = db.get_pool(self.settings).connection(timeout=5.0)
                conn = cm.__enter__()
            except Exception as e:
                log.warning("database unavailable: %s", type(e).__name__)
                raise ApiError(503, "database_unavailable", "The database is not available.") from None
            conn.autocommit = True
            self._cm, self._conn = cm, conn
        return self._conn

    def close(self) -> None:
        if self._cm is not None:
            cm, self._cm, self._conn = self._cm, None, None
            cm.__exit__(None, None, None)


def get_db(request: Request) -> Iterator[Db]:
    d = Db(settings_of(request))
    try:
        yield d
    finally:
        d.close()


def ip_of(request: Request) -> str | None:
    """The client address as text, or None when it is not an IP address (test clients)."""
    host = client_ip(request)
    if not host:
        return None
    try:
        return str(ipaddress.ip_address(host))
    except ValueError:
        return None


def user_agent_of(request: Request) -> str | None:
    ua = request.headers.get("user-agent")
    return ua[:400] if ua else None


# ---------------------------------------------------------------- cookies


def set_session_cookie(request: Request, response: Response, token: str) -> None:
    response.set_cookie(
        sessions.SESSION_COOKIE,
        token,
        httponly=True,
        samesite="strict",
        secure=request.url.scheme == "https",
        path="/",
    )


def clear_session_cookie(request: Request, response: Response) -> None:
    response.delete_cookie(
        sessions.SESSION_COOKIE, path="/", httponly=True, samesite="strict", secure=request.url.scheme == "https"
    )


def set_device_cookie(request: Request, response: Response, token: str, days: int) -> None:
    response.set_cookie(
        sessions.DEVICE_COOKIE,
        token,
        max_age=days * 86400,
        httponly=True,
        samesite="strict",
        secure=request.url.scheme == "https",
        path="/",
    )


def current_device(request: Request, conn: Any) -> dict[str, Any] | None:
    token = request.cookies.get(sessions.DEVICE_COOKIE)
    if not token:
        return None
    return devices.find(conn, session_key(request), token)


# ---------------------------------------------------------------- principal


@dataclass
class Principal:
    session: dict[str, Any]
    user: dict[str, Any]
    roles: list[str]
    groups: list[str]
    screens: list[str]
    is_admin: bool
    kind: str
    device: dict[str, Any] | None
    ip: str | None
    lock_minutes: int = 3

    @property
    def session_id(self) -> uuid.UUID:
        sid: uuid.UUID = self.session["id"]
        return sid

    @property
    def user_id(self) -> uuid.UUID:
        uid: uuid.UUID = self.user["id"]
        return uid

    @property
    def locked(self) -> bool:
        return self.session.get("locked_since") is not None

    @property
    def role_snapshot(self) -> str:
        return ",".join(self.roles)

    @property
    def actor(self) -> sessions.Actor:
        return sessions.Actor(
            session_id=self.session_id,
            user_id=self.user_id,
            sam_account=self.user["sam_account"],
            display_name=self.user["display_name"],
            kind=self.kind,
            roles=tuple(self.roles),
        )


def _expire(conn: Any, row: dict[str, Any], reason: str, ip: str | None) -> None:
    with conn.transaction():
        if sessions.revoke(conn, row["id"], reason):
            snap = row.get("roles_snapshot") or {}
            audit.append(
                conn,
                action="auth.session_ended",
                user_id=row["user_id"],
                session_id=row["id"],
                role_snapshot=",".join(snap.get("roles", [])),
                detail={"reason": reason},
                ip=ip,
            )


def _refresh_roles(
    conn: Any, directory: DirectoryProtocol, row: dict[str, Any], ip: str | None
) -> tuple[list[str], list[str]]:
    """Re-resolve the session's groups from AD. Raises 401 session_expired when the roles are gone."""
    uid = row["user_id"]
    try:
        u = directory.get_user_by_guid(uid)
        found = sorted(directory.groups_of(u.dn)) if u is not None and not u.disabled else []
    except LdapUnavailable as e:
        # Keep the cached roles; try again after the next interval rather than on every request
        log.warning("roles refresh skipped, AD unavailable: %s", e)
        conn.execute("UPDATE sessions SET roles_checked_at = now() WHERE id = %s", (row["id"],))
        snap = row.get("roles_snapshot") or {}
        return list(snap.get("roles", [])), list(snap.get("groups", []))
    roles = groupmap.roles_from_groups(found)
    if u is None or u.disabled or not roles:
        with conn.transaction():
            if u is None or u.disabled:
                userdir.mark_inactive(conn, uid, reason="roles_lost", gone=u is None)
            else:
                userdir.set_roles(conn, uid, [])
        _expire(conn, row, "roles_lost", ip)
        raise ApiError(401, "session_expired", "Your Packet Station roles have changed. Sign in again.")
    with conn.transaction():
        userdir.upsert_user(conn, u, groups=found)
        userdir.set_roles(conn, uid, roles)
        conn.execute(
            "UPDATE sessions SET roles_snapshot = %s, roles_checked_at = now() WHERE id = %s",
            (Jsonb({"roles": roles, "groups": found}), row["id"]),
        )
    return roles, found


def resolve(request: Request, d: Db, *, allow_locked: bool, token: str | None = None) -> Principal:
    """Load and check the session for this request (or for `token`). See the contract's "Sessions and cookies"."""
    token = token or request.cookies.get(sessions.SESSION_COOKIE)
    if not token:
        raise ApiError(401, "no_session", "Not signed in.")
    s = settings_of(request)
    key = session_key(request)
    conn = d.conn
    ip = ip_of(request)
    row = sessions.load(conn, key, token)
    if row is None or row["revoked_at"] is not None or row["expires_at"] <= row["db_now"]:
        raise ApiError(401, "no_session", "Not signed in.")
    now = row["db_now"]
    idle = now - row["last_seen_at"]
    idle_min, lock_min = idle_minutes(s, list((row.get("roles_snapshot") or {}).get("roles", [])))
    if idle > timedelta(minutes=idle_min):
        _expire(conn, row, "idle", ip)
        raise ApiError(401, "session_expired", "Signed out after a period of no activity.")
    lock_source: str | None = None
    if row["locked_since"] is None and idle > timedelta(minutes=lock_min):
        lock_source = "idle"
    hidden_since = row.get("hidden_since")
    if hidden_since is not None:
        # The browser reported a hidden tab. Any request means it is back (or was never really
        # away, as on a reload): lock only when it stayed hidden for the old build's minute.
        if row["locked_since"] is None and now - hidden_since > timedelta(seconds=s.session_hidden_lock_seconds):
            lock_source = lock_source or "hidden"
        conn.execute("UPDATE sessions SET hidden_since = NULL WHERE id = %s", (row["id"],))
    if lock_source is not None:
        with conn.transaction():
            locked = conn.execute(
                "UPDATE sessions SET locked_since = %s WHERE id = %s AND locked_since IS NULL RETURNING id",
                (now, row["id"]),
            ).fetchone()
            if locked is not None:
                # Presence drops a locked session at once (events.EventHub handles the event).
                events.publish(conn, {"event": "session", "session_id": str(row["id"]), "action": "locked"})
            snap = row.get("roles_snapshot") or {}
            audit.append(
                conn,
                action="auth.lock",
                user_id=row["user_id"],
                session_id=row["id"],
                role_snapshot=",".join(snap.get("roles", [])),
                detail={"source": lock_source},
                ip=ip,
            )
        row["locked_since"] = now
    if row["locked_since"] is not None and not allow_locked:
        raise ApiError(423, "locked", "The session is locked.")
    if row["locked_since"] is None and idle > LAST_SEEN_SLACK:
        conn.execute("UPDATE sessions SET last_seen_at = now() WHERE id = %s", (row["id"],))
    user = userdir.get_user(conn, row["user_id"])
    if user is None:
        raise ApiError(401, "no_session", "Not signed in.")
    snap = row.get("roles_snapshot") or {}
    roles, groups = list(snap.get("roles", [])), list(snap.get("groups", []))
    if now - row["roles_checked_at"] > timedelta(minutes=s.roles_refresh_minutes):
        roles, groups = _refresh_roles(conn, directory_of(request), row, ip)
    device = devices.get(conn, row["device_id"]) if row.get("device_id") else None
    return Principal(
        session=row,
        user=user,
        roles=roles,
        groups=groups,
        screens=groupmap.screens_for(roles),
        is_admin=groupmap.is_admin(roles),
        kind=str(row["kind"]),
        device=device,
        ip=ip,
        lock_minutes=idle_minutes(s, roles)[1],
    )


def current_principal(request: Request, d: Db = Depends(get_db)) -> Principal:  # noqa: B008
    """A signed-in, unlocked session."""
    return resolve(request, d, allow_locked=False)


def current_principal_any(request: Request, d: Db = Depends(get_db)) -> Principal:  # noqa: B008
    """A signed-in session, locked or not (me, unlock, signout, visibility, events)."""
    return resolve(request, d, allow_locked=True)


def resolve_standalone(request: Request) -> Principal:
    """For long-lived responses (SSE): check the session on its own connection and release it at once."""
    d = Db(settings_of(request))
    try:
        return resolve(request, d, allow_locked=True)
    finally:
        d.close()


# ---------------------------------------------------------------- policy helpers


def require_password_session(p: Principal) -> Principal:
    if p.kind != "password":
        raise ApiError(403, "password_session_required", "Sign in with your password to do this. A PIN session cannot.")
    return p


def require_admin(p: Principal) -> Principal:
    if not p.is_admin:
        raise ApiError(403, "forbidden", "Admin only.")
    return p


def can_register_device(p: Principal) -> bool:
    return p.kind == "password" and (p.is_admin or groupmap.is_shop_supervisor(p.roles))


def require_device_registrar(p: Principal) -> Principal:
    require_password_session(p)
    if not can_register_device(p):
        raise ApiError(403, "forbidden", "Only an Admin or a shop Supervisor can register a device.")
    return p


def admin_principal(p: Principal = Depends(current_principal)) -> Principal:  # noqa: B008
    require_admin(p)
    return require_password_session(p)


def password_principal(p: Principal = Depends(current_principal)) -> Principal:  # noqa: B008
    return require_password_session(p)