jonas/kirche-einfach
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: authorize flow management, gate runs, token-gate callbacks, strict tenants

Browse source
This commit is contained in:
Jonas Miederer 2026-09-14 20:12:17 +02:00
parent cfcf0e8bd6
commit 94e5e53092
21 changed files with 662 additions and 160 deletions
Download patch file Download diff file Expand all files Collapse all files

125
backend/app/flows/api.py
View file

@ -1,19 +1,27 @@
"""Flows + approvals HTTP API."""
"""Flows + approvals HTTP API.
Auth model: flow *management* (create/version/deploy/sync/import) is
admin-only. Browsing (list/get/input-schema) is public so the website
renders. Runs, run reads and approval reads require login and are scoped
to the session tenant. Flow *callbacks* (LFX has no session) carry a
shared X-Callback-Token instead.
"""
from __future__ import annotations
import json
import secrets
from collections.abc import AsyncIterator, Iterator
from pathlib import Path
from typing import Annotated
from uuid import UUID
import httpx
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi import APIRouter, Depends, Header, HTTPException, Query
from fastapi.responses import StreamingResponse
from sqlmodel import Session
from app.auth.deps import get_optional_user, resolve_tenant
from app.auth.deps import get_current_user, require_admin
from app.db import get_engine, get_session
from app.models import User
from app.settings import settings
@ -127,7 +135,11 @@ def _read_task(t) -> ApprovalRead:
@flows_router.post("", response_model=FlowDefinitionRead, status_code=201)
def create_flow(data: FlowDefinitionCreate, session: Annotated[Session, Depends(get_db)]):
def create_flow(
data: FlowDefinitionCreate,
session: Annotated[Session, Depends(get_db)],
_admin: Annotated[User, Depends(require_admin)],
):
try:
return _read_definition(store.create_definition(session, data))
except store.ConflictError as exc:
@ -145,6 +157,7 @@ def list_flows(session: Annotated[Session, Depends(get_db)]):
async def sync_flows(
session: Annotated[Session, Depends(get_db)],
lfx: Annotated[LfxClient, Depends(get_lfx)],
_admin: Annotated[User, Depends(require_admin)],
):
"""Reconcile backend/flows/*.json into the DB + LFX (add/update/delete)."""
try:
@ -163,7 +176,10 @@ def get_flow(flow_id: UUID, session: Annotated[Session, Depends(get_db)]):
@flows_router.post("/{flow_id}/versions", response_model=FlowVersionRead, status_code=201)
def create_version(
flow_id: UUID, data: FlowVersionCreate, session: Annotated[Session, Depends(get_db)]
flow_id: UUID,
data: FlowVersionCreate,
session: Annotated[Session, Depends(get_db)],
_admin: Annotated[User, Depends(require_admin)],
):
try:
return _read_version(store.create_version(session, flow_id, data))
@ -174,7 +190,11 @@ def create_version(
@flows_router.get("/{flow_id}/versions", response_model=list[FlowVersionRead])
def list_versions(flow_id: UUID, session: Annotated[Session, Depends(get_db)]):
def list_versions(
flow_id: UUID,
session: Annotated[Session, Depends(get_db)],
_admin: Annotated[User, Depends(require_admin)],
):
try:
return [_read_version(v) for v in store.list_versions(session, flow_id)]
except store.NotFoundError as exc:
@ -187,6 +207,7 @@ async def deploy_version(
version_no: int,
session: Annotated[Session, Depends(get_db)],
lfx: Annotated[LfxClient, Depends(get_lfx)],
_admin: Annotated[User, Depends(require_admin)],
):
try:
version = store.get_version_by_no(session, flow_id, version_no)
@ -223,30 +244,32 @@ async def run_flow(
data: RunRequest,
session: Annotated[Session, Depends(get_db)],
lfx: Annotated[LfxClient, Depends(get_lfx)],
user: Annotated[User | None, Depends(get_optional_user)],
user: Annotated[User, Depends(get_current_user)],
stream: bool = Query(default=False),
):
# Tenant comes from the session; the body param stays as a deprecated
# fallback for unauthenticated callers.
data = data.model_copy(update={"tenant_id": resolve_tenant(user, data.tenant_id)})
# Tenant, callback URL and continuation always come from the server:
# the session, the settings, and the admin-authored definition.
tenant_id = user.tenant_id
try:
definition = store.get_definition(session, flow_id)
version = _resolve_version(session, flow_id, data.version_no)
except store.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except store.InvalidError as exc:
raise HTTPException(status_code=422, detail=str(exc)) from exc
if stream:
return await _run_stream(session, lfx, version, data)
return await _run_stream(session, lfx, version, data, user, definition)
try:
run = await runner.execute_run(
session,
lfx,
version,
payload=data.input,
tenant_id=data.tenant_id,
tenant_id=tenant_id,
session_id=data.session_id,
callback_base_url=data.callback_base_url or settings.callback_base_url,
continuation_flow_id=data.continuation_flow_id,
callback_base_url=settings.callback_base_url,
continuation_flow_id=definition.continuation_flow_id,
callback_token=settings.callback_token,
)
except LfxNotFound as exc:
raise HTTPException(status_code=502, detail=f"LFX drift: {exc}") from exc
@ -257,21 +280,24 @@ async def run_flow(
return _read_run(run)
async def _run_stream(session: Session, lfx: LfxClient, version, data: RunRequest):
async def _run_stream(
session: Session, lfx: LfxClient, version, data: RunRequest, user: User, definition
):
if not version.lfx_flow_id:
raise HTTPException(status_code=422, detail=f"version {version.id} is not deployed to LFX")
run = store.create_run(
session,
version,
input=data.input,
tenant_id=data.tenant_id,
tenant_id=user.tenant_id,
session_id=data.session_id,
)
envelope = runner.build_envelope(
payload=data.input,
run_id=run.id,
callback_base_url=data.callback_base_url or settings.callback_base_url,
continuation_flow_id=data.continuation_flow_id,
callback_base_url=settings.callback_base_url,
continuation_flow_id=definition.continuation_flow_id,
callback_token=settings.callback_token,
)
run_id = run.id
lfx_flow_id = version.lfx_flow_id
@ -314,21 +340,20 @@ async def _run_stream(session: Session, lfx: LfxClient, version, data: RunReques
def list_runs(
flow_id: UUID,
session: Annotated[Session, Depends(get_db)],
user: Annotated[User | None, Depends(get_optional_user)],
tenant_id: str | None = Query(default=None),
user: Annotated[User, Depends(get_current_user)],
):
# Authenticated callers are scoped to their session tenant unless they
# pass an explicit tenant (deprecated fallback for unauthenticated use).
if user is not None and tenant_id is None:
tenant_id = user.tenant_id
# Strict tenant scoping: the session decides, query params cannot widen.
try:
return [_read_run(r) for r in store.list_runs(session, flow_id, tenant_id)]
return [_read_run(r) for r in store.list_runs(session, flow_id, user.tenant_id)]
except store.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@flows_router.get("/langflow/available")
async def list_langflow_flows(langflow: Annotated[LangflowClient, Depends(get_langflow)]):
async def list_langflow_flows(
langflow: Annotated[LangflowClient, Depends(get_langflow)],
_admin: Annotated[User, Depends(require_admin)],
):
"""Browse flows on the Langflow authoring server (admin import source)."""
try:
return await langflow.list_flows()
@ -344,6 +369,7 @@ async def import_flow(
session: Annotated[Session, Depends(get_db)],
langflow: Annotated[LangflowClient, Depends(get_langflow)],
lfx: Annotated[LfxClient, Depends(get_lfx)],
_admin: Annotated[User, Depends(require_admin)],
):
"""Import a flow from Langflow: export -> version -> (deploy to LFX)."""
try:
@ -401,24 +427,38 @@ def get_input_schema(
def get_run(
run_id: UUID,
session: Annotated[Session, Depends(get_db)],
user: Annotated[User | None, Depends(get_optional_user)],
tenant_id: str | None = Query(default=None),
user: Annotated[User, Depends(get_current_user)],
):
"""Read a run. With tenant_id, cross-tenant reads 404 (v1 scoping)."""
if user is not None and tenant_id is None:
tenant_id = user.tenant_id
"""Read a run. Other tenants' runs 404 (strict session scoping)."""
try:
run = store.get_run(session, run_id)
except store.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
if tenant_id is not None and run.tenant_id != tenant_id:
if run.tenant_id != user.tenant_id:
raise HTTPException(status_code=404, detail=f"run {run_id} not found")
return _read_run(run)
CALLBACK_TOKEN_HEADER = "X-Callback-Token"
def _check_callback_token(token: str | None) -> None:
"""LFX-served flows prove they carry our run envelope (no session there)."""
expected = settings.callback_token
if not expected:
raise HTTPException(status_code=503, detail="flow callbacks are disabled")
if not token or not secrets.compare_digest(token, expected):
raise HTTPException(status_code=401, detail="invalid callback token")
@approvals_router.post("", response_model=ApprovalRead, status_code=201)
def create_approval(data: ApprovalCreate, session: Annotated[Session, Depends(get_db)]):
def create_approval(
data: ApprovalCreate,
session: Annotated[Session, Depends(get_db)],
callback_token: Annotated[str | None, Header(alias=CALLBACK_TOKEN_HEADER)] = None,
):
"""Flow callback contract: report a terminal approval state for a run."""
_check_callback_token(callback_token)
try:
return _read_task(store.create_approval_task(session, data))
except store.NotFoundError as exc:
@ -432,12 +472,9 @@ def create_approval(data: ApprovalCreate, session: Annotated[Session, Depends(ge
@approvals_router.get("", response_model=list[ApprovalRead])
def list_approvals(
session: Annotated[Session, Depends(get_db)],
user: Annotated[User | None, Depends(get_optional_user)],
tenant_id: str | None = Query(default=None),
user: Annotated[User, Depends(get_current_user)],
):
if user is not None and tenant_id is None:
tenant_id = user.tenant_id
return [_read_task(t) for t in store.list_approval_tasks(session, tenant_id)]
return [_read_task(t) for t in store.list_approval_tasks(session, user.tenant_id)]
@approvals_router.post("/{task_id}/decision", response_model=DecisionRead)
@ -446,8 +483,13 @@ async def decide_approval(
data: DecisionRequest,
session: Annotated[Session, Depends(get_db)],
lfx: Annotated[LfxClient, Depends(get_lfx)],
user: Annotated[User, Depends(get_current_user)],
):
"""Decide an open task. Approve starts the continuation flow (if any)."""
"""Decide an open task. Approve starts the continuation flow (if any).
Identity comes from the session: the approver is the logged-in user,
and the continuation is the task's own (never client-chosen).
"""
try:
task = store.get_task(session, task_id)
except store.NotFoundError as exc:
@ -467,7 +509,7 @@ async def decide_approval(
session,
task,
decision="approved" if data.decision == "approve" else "rejected",
approver=data.approver,
approver=user.email,
note=data.note,
)
except store.ConflictError as exc:
@ -489,7 +531,8 @@ async def decide_approval(
payload=payload,
tenant_id=task.tenant_id,
session_id=f"continuation-of-{task.run_id}",
callback_base_url=data.callback_base_url or settings.callback_base_url,
callback_base_url=settings.callback_base_url,
callback_token=settings.callback_token,
)
except LfxNotFound as exc:
raise HTTPException(status_code=502, detail=f"LFX drift: {exc}") from exc

4
backend/app/flows/runner.py
View file

@ -20,6 +20,7 @@ def build_envelope(
run_id: UUID,
callback_base_url: str,
continuation_flow_id: UUID | None,
callback_token: str,
) -> str:
return json.dumps(
{
@ -27,6 +28,7 @@ def build_envelope(
"run_id": str(run_id),
"callback_url": callback_base_url.rstrip("/"),
"continuation_flow_id": (str(continuation_flow_id) if continuation_flow_id else None),
"callback_token": callback_token,
}
)
@ -50,6 +52,7 @@ async def execute_run(
session_id: str | None = None,
callback_base_url: str,
continuation_flow_id: UUID | None = None,
callback_token: str = "",
) -> FlowRun:
"""Run a deployed version to a terminal state (no interrupts).
@ -68,6 +71,7 @@ async def execute_run(
run_id=run.id,
callback_base_url=callback_base_url,
continuation_flow_id=continuation_flow_id,
callback_token=callback_token,
)
try:
result = await lfx.run_flow(

16
backend/app/flows/schemas.py
View file

@ -52,16 +52,16 @@ class InputSchemaRead(BaseModel):
class RunRequest(BaseModel):
"""Start a flow run. `input` is the domain payload handed to the flow."""
"""Start a flow run. `input` is the domain payload handed to the flow.
Tenant, callback URL and continuation come from the server (session /
settings / task), never from the caller — they must not be
client-controllable (SSRF / cross-flow chaining).
"""
input: dict = Field(default_factory=dict)
version_no: int | None = None
tenant_id: str = Field(default="default", max_length=200)
session_id: str | None = None
# Base URL the flow uses for callbacks (e.g. approval tasks).
# Defaults to settings.callback_base_url; tests point it at a live server.
callback_base_url: str | None = None
continuation_flow_id: UUID | None = None
class RunRead(BaseModel):
@ -115,10 +115,10 @@ class FlowImportRead(BaseModel):
class DecisionRequest(BaseModel):
"""Decide a task. Identity comes from the session, not the body."""
decision: str = Field(pattern="^(approve|reject)$")
approver: str = Field(default="", max_length=200)
note: str = Field(default="", max_length=2000)
callback_base_url: str | None = None
class DecisionRead(BaseModel):