app/backend/fluksio/api/deps.py

import uuid
from collections.abc import Generator
from typing import Annotated, Any
from urllib.parse import unquote

import jwt
from fastapi import Depends, HTTPException, Request, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pydantic import ValidationError
from sqlmodel import Session, select

from fluksio.cloud import config as cloud_config
from fluksio.cloud import enroll
from fluksio.core import security
from fluksio.core.config import settings
from fluksio.core.db import engine
from fluksio.flow import devices
from fluksio.flow.artifacts import is_reference
from fluksio.flow.controller import FlowController
from fluksio.flow.dashboards import DashboardDef, DashboardStore
from fluksio.flow.devices import DeviceDef, DeviceStore, DeviceUnavailable, presence
from fluksio.flow.events import event_bus
from fluksio.flow.workers import PythonWorkerPool
from fluksio.models import OAuthClient, TokenPayload, User, WorkerToken

reusable_oauth2 = OAuth2PasswordBearer(
    tokenUrl=f"{settings.API_V1_STR}/login/access-token"
)


def get_db() -> Generator[Session, None, None]:
    with Session(engine) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_db)]
TokenDep = Annotated[str, Depends(reusable_oauth2)]


#: What a `draft` query parameter has to say to mean "no". Pydantic's own false
#: literals; anything else is refused rather than guessed at.
_NOT_DRAFT = {"false", "0", "off", "f", "n", "no", ""}

#: The one route under a message name a device reads.
_HISTORY = "/history"


def _device_for(payload: dict[str, Any], app: Any) -> DeviceDef:
    """The device a credential names, if that credential still stands.

    Two ways it stops standing. The device was deleted, which revokes every
    credential ever minted for it, read from disk on each call so it takes
    effect at once. Or the device's nonce moved on, which revokes exactly one
    screen's and leaves the device and its dashboards alone.

    The nonce is held against a credential the portal minted for a remote
    screen as well, which carries the nonce this instance named when it asked
    for it. One from an older portal carries none and reads as 0, so it
    stands only until the device is first unpaired or deleted.

    The claim is still called ``panel``: tokens in circulation say it, and
    nothing pairs again for a rename.

    What cannot be checked right now raises ``DeviceUnavailable`` rather than
    refusing: a device told its credential is refused forgets it.
    """
    if app is None:
        raise InvalidTokenError("Nothing to check a device credential against")
    store: DeviceStore | None = getattr(app.state, "device_store", None)
    if store is None:
        raise DeviceUnavailable("The device store is not running")
    device = store.find(str(payload.get("panel", "")))
    if device is None:
        raise InvalidTokenError("This device no longer exists")
    try:
        nonce = store.nonce(device.id)
    except (OSError, ValueError) as exc:
        raise DeviceUnavailable(f"The device nonces are unreadable: {exc}")
    if int(payload.get("pnc") or 0) != nonce:
        raise InvalidTokenError("This device was paired again")
    presence.seen(device.id)
    return device


def _device_messages(device: DeviceDef, request: Request) -> set[str]:
    """Every message this device's widgets and ports read or write.

    The same walk that bounds its socket, so the two surfaces a screen has
    agree on what it is entitled to. No store means nothing resolves, which is
    the answer to give when the answer cannot be worked out.
    """
    store: DashboardStore | None = getattr(request.app.state, "dashboard_store", None)
    owned: DeviceStore | None = getattr(request.app.state, "device_store", None)
    if store is None or owned is None:
        return set()
    return devices.messages_for(device, store, owned)


def _device_in(device: DeviceDef, request: Request) -> set[str]:
    """The messages this device's own ``in`` ports are wired to publish."""
    owned: DeviceStore | None = getattr(request.app.state, "device_store", None)
    return devices.io_in(device, owned) if owned is not None else set()


def _locked(defn: DashboardDef, live: dict[str, Any]) -> bool:
    """Whether this dashboard is read-only right now.

    Read the way the client reads it (``Dashboard/settings.tsx``): the bound
    message if it is carrying something, and the stored value otherwise. Live,
    because a lock a flow drives is the case the binding exists for — judging
    it from the stored fallback alone would refuse every control on a
    dashboard its flow has unlocked.
    """
    setting = defn.settings.get("locked")
    if setting is None:
        return False
    value = live.get(setting.message) if setting.message else None
    return (setting.value if value is None else value) is True


def _device_writable(device: DeviceDef, request: Request) -> set[str]:
    """The messages this device may publish to.

    What its dashboards' inputs target, which is narrower than what it may
    read: a chart, a reading or a bound setting is something a screen shows,
    and showing a message does not entitle it to set one.

    A locked dashboard contributes less still: ``locked`` is a dashboard
    saying it is there to be looked at, and until this it stopped only the
    client drawing the control — the server took the publish from a screen
    that asked anyway. Not nothing, though: a querying chart publishes the
    request it reads by, and a locked dashboard whose charts cannot ask goes
    blank rather than read-only. That request is the exception
    ``devices.requests_of`` names.

    Union, exactly as the allowlist itself is: a message one dashboard on this
    panel displays and another controls stays writable, because the unlocked
    one is what entitles the screen to it. Reads are untouched — read-only is
    not blind, and a locked dashboard has to keep drawing live data.

    The device's own ports, as flows wire them, are added outside the lock:
    it covers the controls on a dashboard, not the device itself. Its ``in``
    ports are its sensors; its ``out`` ports it publishes when set on the
    device (a panel's brightness on its config page), so what follows the
    message hears of it.
    """
    store: DashboardStore | None = getattr(request.app.state, "dashboard_store", None)
    if store is None:
        return set()
    owned: DeviceStore | None = getattr(request.app.state, "device_store", None)
    writable = (
        {b.message for b in owned.bound(device.id)} if owned is not None else set()
    )
    defns = devices.dashboards_for(device, store)
    bound = sorted(
        {s.message for d in defns if (s := d.settings.get("locked")) and s.message}
    )
    controller: FlowController | None = getattr(
        request.app.state, "flow_controller", None
    )
    live = controller.state.get_present(bound) if bound and controller else {}
    for defn in defns:
        writable |= (
            devices.requests_of(defn)
            if _locked(defn, live)
            else devices.targets_of(defn)
        )
    return writable


def _device_digests(device: DeviceDef, request: Request) -> set[str]:
    """The artifacts this panel's messages are pointing at, now or just before.

    A media tile fetches the bytes its message names, so the messages already
    bounding the panel bound this too — one step further along, through
    whatever those messages currently hold. And through the last few they
    held: a screen through the portal fetches a camera frame by frame, and by
    the time its fetch arrives the camera may have published the next one.
    """
    controller: FlowController | None = getattr(
        request.app.state, "flow_controller", None
    )
    return digests_of(controller, _device_messages(device, request))


def digests_of(controller: FlowController | None, names: set[str]) -> set[str]:
    """The artifacts these messages point at now, and the last few before."""
    if controller is None or not names:
        return set()
    found: set[str] = set()
    for value in controller.state.get_present(sorted(names)).values():
        if is_reference(value):
            found.add(str(value["digest"]))
    for name in names:
        found.update(event_bus.digests.get(name, ()))
    return found


def hidden_from(user: User, app: Any, *, device: bool = False) -> frozenset[str]:
    """The camera messages ``user`` may not see (``DeviceDef.viewers``).

    None for a superuser, and none for a device credential, which its
    dashboards bound instead: a superuser assigned them, camera tiles
    included. ``device`` says the caller is one.
    """
    if device or user.is_superuser:
        return frozenset()
    store: DeviceStore | None = getattr(app.state, "device_store", None)
    return store.restricted() if store is not None else frozenset()


def _device_may(payload: dict[str, Any], request: Request) -> None:
    """Refuse anything a wall panel has no business asking for.

    A panel credential acts as the device itself, an account of no rights
    (``device_principal``), and what that account may reach is decided here
    and nowhere else. What a panel
    genuinely needs is small and worth writing out: the dashboards it was
    assigned, its own definition and report, the messages its own widgets and
    ports bind to, and the bytes those messages currently point at, for a tile
    drawing a camera frame — plus an upload named after one of its ``in``
    ports, for a microphone whose recording is the value it publishes.

    Publishing is in the list because a panel cannot be strictly read-only — a
    control on a panel is the point of putting one there, and a querying chart
    asks for its window by publishing a request, which is why that request
    counts as one of its widget's targets. Bounded to those, though: a screen
    on a wall has no business reaching a message no tile on it draws, and the
    catalogue behind ``GET /messages/`` is the whole namespace at once. It
    reads every message its tiles draw and writes only what its inputs target.
    A dashboard that says it is locked is bounded further still — it entitles
    a panel to read every message it names and to publish to none of them but
    the requests its own charts ask by.

    Its own ports are the device nodes the published flows hold for it, so
    wiring a port in a flow is what lets the device publish on it.

    The device is left on ``request.state`` for the route, which credits a
    value published on one of its ports to the node holding that port.

    The device file is read per request, which is what makes deleting a
    device revoke it at once. The nonces are cached by the file's stat and
    the published dashboards until the flow store's next commit.
    """
    device = _device_for(payload, request.app)
    request.state.device = device

    api = settings.API_V1_STR
    path = request.url.path
    method = request.method

    allowed = False
    if method == "GET" and path.startswith(f"{api}/dashboards/"):
        name = path[len(f"{api}/dashboards/") :]
        # One segment only: a dashboard it shows, published — never a draft,
        # and never a sibling route like `/publish`. The flag is read the way
        # the endpoint reads it rather than by presence, because the generated
        # client spells the default out as `?draft=false` — and it is read
        # fail-closed, so a spelling neither side recognises is a draft.
        draft = request.query_params.get("draft")
        published = draft is None or draft.lower() in _NOT_DRAFT
        allowed = "/" not in name and name in device.dashboards and published
    elif method == "GET" and path in (
        f"{api}/devices/{device.id}",
        # What firmware from before devices reads.
        f"{api}/panels/{device.id}",
    ):
        allowed = True
    elif method == "PUT" and path == f"{api}/devices/{device.id}/report":
        # What it has, said about itself and nobody else.
        allowed = True
    elif method == "POST" and path == f"{api}/flows/ws-ticket":
        # How its live socket opens; the socket applies the panel's own scope.
        allowed = True
    elif path.startswith(f"{api}/messages/"):
        # The two a widget speaks: a chart reads a series, a control puts a
        # value in. Percent-decoded, because a message name is a path segment
        # here and the client encodes it as one.
        rest = unquote(path[len(f"{api}/messages/") :])
        scope = _device_messages
        if method == "GET" and rest.endswith(_HISTORY):
            name = rest[: -len(_HISTORY)]
        elif method == "POST":
            name = rest
            # A locked dashboard entitles the panel to nothing it can publish.
            scope = _device_writable
        else:
            name = ""
        allowed = bool(name) and name in scope(device, request)
    elif method == "GET" and path.startswith(f"{api}/artifacts/"):
        # The bytes behind a media message a tile on this panel is drawing.
        # Scoped to what those messages hold *now*, which is exactly what a
        # live widget asks for — a screen has no business reading an artifact
        # off an old run because it happens to know the digest.
        digest = unquote(path[len(f"{api}/artifacts/") :])
        allowed = "/" not in digest and digest in _device_digests(device, request)
    elif method == "PUT" and path == f"{api}/artifacts":
        # What the device's own sensor recorded — a microphone's utterance —
        # uploaded under the ``in`` message it is about to publish it to.
        allowed = request.query_params.get("name", "") in _device_in(device, request)

    if not allowed:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="A device credential cannot do that",
        )


def decode_token(
    token: str, request: Request | None = None, app: Any = None
) -> dict[str, Any]:
    """Read a bearer token, whichever channel issued it.

    A browser session is signed with the app's own secret; an agent's token is
    signed with the OAuth keypair, so that set can be revoked on its own and
    the public half published. Both name a user, and both grant that user's
    rights — the difference is only in who is holding it, which the MCP
    endpoint checks separately.

    A paired wall panel's token is signed with the app's secret too, and is
    told apart from a session by its audience: the session decode above
    refuses it outright, so the only door it fits is the one ``_device_may``
    guards. Where there is no request to scope (a socket handshake, the
    tunnel), ``app`` is what a device credential is checked against.

    An agent's token is held to one thing more: the client it names has to
    still be registered, which is what makes revoking an agent take effect now
    rather than whenever its stateless token happens to expire.

    The last branch is the seam a hosted deployment widens: a portal this
    instance was enrolled with signs tokens with a key pinned at
    enrolment, and they name the portal account holding them, which resolves
    to whichever local account was mapped to it — the superuser who enrolled,
    or a remote user one of them admitted since. With no enrolment the branch
    raises immediately, so an offline instance pays nothing for the
    possibility. A screen that paired through that portal
    arrives there too, naming a panel — which is why the gate below is applied
    to whichever branch produced the claims rather than to one of them: where a
    panel credential was minted is not what decides what it may read.
    """
    try:
        session: dict[str, Any] = jwt.decode(
            token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
        )
        return session
    except InvalidTokenError:
        pass
    try:
        panel_token = security.decode_panel_token(token)
    except InvalidTokenError:
        pass
    else:
        if not panel_token.get("panel"):
            raise InvalidTokenError("a panel token must name its panel")
        return _gate_panel(panel_token, request, app)
    try:
        agent = security.decode_oauth_token(token)
    except InvalidTokenError:
        return _gate_panel(cloud_config.decode_portal_token(token), request, app)
    return _gate_agent(agent)


def oauth_client_lives(client_id: str) -> bool:
    """Is the agent a token names still a registered client? Blocking.

    An MCP access token is a stateless JWT good until it expires, so deleting
    the client row revoked nothing that had already been handed out. The
    client id is a claim the token has always carried, so this holds the ones
    already in circulation just as well as the next one minted.

    One primary-key read per agent request, against a table with a row per
    registered agent. The request it gates goes on to look its user up the
    same way, and the device gate next door re-reads the device's file and every
    published dashboard from disk, so this is the cheapest check in here.
    """
    try:
        key = uuid.UUID(client_id)
    except ValueError:
        return False
    with Session(engine) as session:
        return session.get(OAuthClient, key) is not None


def worker_claims(token: str) -> dict[str, Any]:
    """A worker credential's claims, if it is still good. Blocking.

    Raises ``InvalidTokenError`` for one that does not verify, one minted
    before tokens carried a `jti` and their labels, and one a superuser has
    revoked. A token whose `jti` has no row is one the engine minted for
    itself, a provisioned machine's, and is good until it expires.
    """
    claims = security.decode_worker_token(token)
    jti = claims.get("jti")
    if not jti:
        raise InvalidTokenError("this worker token predates revocation; mint a new one")
    with Session(engine) as session:
        row = session.get(WorkerToken, str(jti))
    if row is not None and row.revoked_at is not None:
        raise InvalidTokenError("this worker token has been revoked")
    return claims


def _gate_agent(payload: dict[str, Any]) -> dict[str, Any]:
    """Refuse a token whose agent has been revoked."""
    if not oauth_client_lives(str(payload.get("client_id") or "")):
        raise InvalidTokenError("this agent's registration was withdrawn")
    return payload


def _gate_panel(
    payload: dict[str, Any], request: Request | None, app: Any = None
) -> dict[str, Any]:
    """Scope a payload that names a device; pass anything else through.

    Without a request (the websocket, which has no route to scope) the check
    is only that the credential still stands, which is what makes deleting a
    device, or bumping its nonce, revoke one. Without an app either there is
    nothing to check it against, and it is refused.
    """
    if not payload.get("panel"):
        return payload
    if request is not None:
        _device_may(payload, request)
    else:
        _device_for(payload, app)
    return payload


#: Where the ids of device principals come from, so each device has its own.
_DEVICE_IDS = uuid.UUID("5e1f2a0c-6d3b-4c1e-9f0a-0d3c1e5b7a20")


def device_principal(device_id: str) -> User:
    """The account a device credential acts as: the device, and no person.

    Not stored. A device's credential is checked against the device itself
    (``_device_for``) and bounded by route (``_device_may``), so no account's
    rights are in play, and deleting or deactivating the superuser who
    approved the pairing leaves the device paired. Never a superuser, so a
    route gated on one refuses it even past the scope check.
    """
    return User(
        id=uuid.uuid5(_DEVICE_IDS, device_id),
        email=f"{device_id}@device.invalid",
        hashed_password="",
        full_name=device_id,
        is_active=True,
        is_superuser=False,
    )


def _user_for(session: Session, token_data: TokenPayload) -> User | None:
    """The local account a payload names, by id or by portal identity.

    A device credential, already checked by ``decode_token``, names the
    device instead, which acts as itself.

    A token the portal minted names a person on the portal, not a user here, so
    a mapping is what turns one into the other. There are two ways one comes to
    exist: a superuser typed their code in, which makes the account up front,
    or the owner shared a link from the portal, which admits them there and
    leaves this instance to find out when they first arrive. So an unmapped
    identity is checked once against the portal's own list of who may reach
    this instance, and adopted only if the portal vouches for it.

    Deleting the local user stays the whole of the revocation: the portal is
    asked rather than the token believed, so a still-valid token cannot rebuild
    the account it named, and a person dropped at the portal is not adopted
    again.
    """
    if token_data.panel:
        return device_principal(token_data.panel)
    if token_data.portal_sub:
        user = session.exec(
            select(User).where(User.portal_sub == token_data.portal_sub)
        ).first()
        if user is None:
            return enroll.adopt_member(session, token_data.portal_sub)
        return user
    if not token_data.sub:
        return None
    try:
        # The subject is a string on the wire. Postgres cast it on the way in;
        # nothing else does, so parse it here rather than hand a driver a
        # string where it wants a UUID.
        user_id = uuid.UUID(token_data.sub)
    except ValueError:
        return None
    return session.get(User, user_id)


def user_from_token(
    session: Session, token: str, request: Request | None = None, app: Any = None
) -> User | None:
    """Resolve a bearer token to its user, or None if it does not hold up.

    Shared with the websocket, which cannot use the HTTP security scheme, and
    with the artifact endpoint, which accepts a worker's credential as well as
    a person's. Pass the request wherever there is one: a device's credential is
    scoped by route, and without it the scope check cannot run. Without one,
    pass the app, or a device credential is refused.
    """
    try:
        token_data = TokenPayload(**decode_token(token, request, app))
    except (InvalidTokenError, ValidationError):
        return None
    user = _user_for(session, token_data)
    if user is None or not user.is_active:
        return None
    return user


def get_current_user(request: Request, session: SessionDep, token: TokenDep) -> User:
    """Resolve the bearer token to its user.

    Every failure here is an authentication failure, so all of them answer 401:
    a token naming a user who no longer exists is a session to log in again,
    not a missing resource to report. A panel credential reaching past what it
    is allowed is the one exception — that is a 403 raised inside
    ``decode_token``, and it travels through here untouched.
    """
    try:
        token_data = TokenPayload(**decode_token(token, request))
    except (InvalidTokenError, ValidationError):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
        )
    user = _user_for(session, token_data)
    if user is None or not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="This session is no longer valid. Please log in again",
        )
    return user


CurrentUser = Annotated[User, Depends(get_current_user)]


def get_hidden(request: Request, user: CurrentUser) -> frozenset[str]:
    """The camera messages the caller of this request may not see."""
    device = getattr(request.state, "device", None) is not None
    return hidden_from(user, request.app, device=device)


HiddenDep = Annotated[frozenset[str], Depends(get_hidden)]

#: Why a camera's values, bytes or live view are refused.
CAMERA_IS_SUPERUSERS = "Only a superuser may see this camera"


def get_flow_controller(request: Request) -> FlowController:
    controller: FlowController | None = getattr(
        request.app.state, "flow_controller", None
    )
    if controller is None:
        raise HTTPException(status_code=503, detail="The flow engine is not running")
    return controller


FlowControllerDep = Annotated[FlowController, Depends(get_flow_controller)]


def get_dashboard_store(request: Request) -> DashboardStore:
    store: DashboardStore | None = getattr(request.app.state, "dashboard_store", None)
    if store is None:
        raise HTTPException(status_code=503, detail="The flow engine is not running")
    return store


DashboardStoreDep = Annotated[DashboardStore, Depends(get_dashboard_store)]


def get_device_store(request: Request) -> DeviceStore:
    store: DeviceStore | None = getattr(request.app.state, "device_store", None)
    if store is None:
        raise HTTPException(status_code=503, detail="The flow engine is not running")
    return store


DeviceStoreDep = Annotated[DeviceStore, Depends(get_device_store)]


def get_worker_pool(request: Request) -> PythonWorkerPool:
    pool: PythonWorkerPool | None = getattr(request.app.state, "worker_pool", None)
    if pool is None:
        raise HTTPException(status_code=503, detail="The flow engine is not running")
    return pool


WorkerPoolDep = Annotated[PythonWorkerPool, Depends(get_worker_pool)]


def get_current_active_superuser(current_user: CurrentUser) -> User:
    if not current_user.is_superuser:
        raise HTTPException(
            status_code=403, detail="The user doesn't have enough privileges"
        )
    return current_user


#: Why a non-superuser is refused a change to what the engine executes.
CODE_IS_SUPERUSERS = "Only a superuser can change code the engine runs"


def get_code_author(current_user: CurrentUser) -> User:
    """A superuser, the only account that may change code the engine runs.

    Node code, a command, an installed package: each runs as the engine, with
    its signing key and every secret in reach, so whoever writes one holds
    the instance whatever their account says.
    """
    if not current_user.is_superuser:
        raise HTTPException(status_code=403, detail=CODE_IS_SUPERUSERS)
    return current_user