Files
app/backend/fluksio/api/deps.py
T
stroblmeandClaude Opus 5 73eeec29b1 Keep the engine's state in SQLite, not Postgres
One process owns this database — the image has run a single uvicorn
worker for that reason since the four-engines bug — so a file beside the
flows is the honest shape for it, and it is what lets `fluksio serve`
need no infrastructure at all. Live values, node execution and the work
queue never came here anyway; what does is a rollup a minute at a time,
a row per cascade and the run history, and WAL keeps the readers going
while that one writer works.

DATA_DIR is now the one setting that moves everything an installation
keeps; the rest derive from it and the images still spell theirs out.
The schema is prepared in-process at startup, so the prestart service is
gone, and the ten Postgres-only revisions collapse into one portable
baseline.

Three things only worked because psycopg was casting for us: a token's
subject arriving as a string where the column is a UUID, `greatest`, and
`date_bin`. The timestamps needed a column type of their own — SQLite
stores no offset, and a naive datetime read back either raises against an
aware `now` or serialises as local time.

Postgres stays in the stack only for Umami, behind the analytics profile.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-21 22:19:45 +02:00

259 lines
10 KiB
Python

import uuid
from collections.abc import Generator
from typing import Annotated, Any
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.core import security
from fluksio.core.config import settings
from fluksio.core.db import engine
from fluksio.flow import panels
from fluksio.flow.controller import FlowController
from fluksio.flow.dashboards import DashboardStore
from fluksio.flow.workers import PythonWorkerPool
from fluksio.models import TokenPayload, User
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", ""}
def _panel_may(payload: dict[str, Any], request: Request) -> None:
"""Refuse anything a wall panel has no business asking for.
A panel credential names the account that approved the pairing, so without
this it would be that person's session hanging on a wall. What a panel
genuinely needs is small and worth writing out: the dashboards it was
assigned, its own definition, and the message endpoints its widgets speak.
Publishing a message is in the list because a panel cannot be strictly
read-only — a querying chart asks for its window by publishing the request
— and a control on a panel is the point of putting one there.
"""
panel = panels.find(str(payload.get("panel", "")))
if panel is None:
# Deleting a panel is how its credential is revoked, so a token naming
# one that is gone is a token to stop trusting.
raise InvalidTokenError("This panel no longer exists")
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 panel.dashboards and published
elif method == "GET" and path == f"{api}/panels/{panel.id}":
allowed = True
elif method in ("GET", "POST") and path.startswith(f"{api}/messages/"):
allowed = True
if not allowed:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="A panel credential cannot do that",
)
def decode_token(token: str, request: Request | None = 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 ``_panel_may``
guards.
The last branch is the seam a hosted deployment widens: a portal this
installation 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 installation 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)
try:
return security.decode_oauth_token(token)
except InvalidTokenError:
return _gate_panel(cloud_config.decode_portal_token(token), request)
def _gate_panel(payload: dict[str, Any], request: Request | None) -> dict[str, Any]:
"""Scope a payload that names a panel; pass anything else through.
Without a request — from the websocket, which has no route to scope — the
check is only that the panel still exists, which is what makes deleting one
revoke its credential.
"""
if not payload.get("panel"):
return payload
if request is not None:
_panel_may(payload, request)
elif panels.find(str(payload["panel"])) is None:
raise InvalidTokenError("This panel no longer exists")
return payload
def _user_for(session: Session, token_data: TokenPayload) -> User | None:
"""The local account a payload names, by id or by portal identity.
A token the portal minted names a person on the portal, not a user here, so
the mapping a superuser made when they admitted them is what turns one into
the other. No mapping, no user — the caller answers that the same way it
answers a token naming a deleted account, which is what makes deleting the
local user the whole of the revocation.
"""
if token_data.portal_sub:
return session.exec(
select(User).where(User.portal_sub == token_data.portal_sub)
).first()
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
) -> 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 panel's credential is
scoped by route, and without it the scope check cannot run.
"""
try:
token_data = TokenPayload(**decode_token(token, request))
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_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_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