diff --git a/.env.example b/.env.example
index 58f66f3..a24a541 100644
--- a/.env.example
+++ b/.env.example
@@ -24,6 +24,9 @@ LFX_API_KEY=lfx-
# --- API self-reference for flow callbacks (must be reachable FROM lfx) ---
CALLBACK_BASE_URL=http://api:8000
+# Shared secret for flow callbacks (POST /api/approvals carries it as
+# X-Callback-Token). Long random value; empty disables callbacks.
+CALLBACK_TOKEN=
# --- frontend mode: dev server with HMR (prod .env must set `preview`) ---
FRONTEND_MODE=dev
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b5e6d5f..687b662 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,15 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [Unreleased]
+### Security
+- API authorization: flow management is admin-only, runs and run/approval
+ reads require login with strict session-tenant scoping, approval
+ decisions record the session user, and flow callbacks authenticate with
+ a shared `X-Callback-Token` (`CALLBACK_TOKEN`). Request-controlled
+ `tenant_id`, `callback_base_url` and `continuation_flow_id` were
+ removed (SSRF / cross-tenant / flow-chaining hardening);
+ continuations now resolve from the admin-authored definition.
+- Frontend: `/app/*` routes require login; approver is the session user.
### Added
- Landing page: status-quo problem block, 5-step process with quote,
portal overview with admin areas, AI platform parts, exception tiers,
diff --git a/backend/app/flows/api.py b/backend/app/flows/api.py
index 135c461..9f74ace 100644
--- a/backend/app/flows/api.py
+++ b/backend/app/flows/api.py
@@ -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
diff --git a/backend/app/flows/runner.py b/backend/app/flows/runner.py
index 1859b70..4c894b9 100644
--- a/backend/app/flows/runner.py
+++ b/backend/app/flows/runner.py
@@ -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(
diff --git a/backend/app/flows/schemas.py b/backend/app/flows/schemas.py
index c3a4146..d32dfcb 100644
--- a/backend/app/flows/schemas.py
+++ b/backend/app/flows/schemas.py
@@ -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):
diff --git a/backend/app/settings.py b/backend/app/settings.py
index 9ccb6e1..fed2bc9 100644
--- a/backend/app/settings.py
+++ b/backend/app/settings.py
@@ -31,6 +31,11 @@ class Settings(BaseSettings):
# Base URL flows use for callbacks into this API. Must be reachable
# FROM the LFX process (compose: http://api:8000).
callback_base_url: str = "http://localhost:8000"
+ # Shared secret that authenticates flow callbacks (POST /api/approvals).
+ # The runner injects it into every run envelope; LFX-served flows send
+ # it back as X-Callback-Token. Empty disables callbacks (503) — set a
+ # long random value in prod (.env CALLBACK_TOKEN).
+ callback_token: str = ""
# --- access management (magic-link auth, see app/auth) ---
# Public frontend URL (magic links point at /login?token= there).
diff --git a/backend/flows/approval_handoff.json b/backend/flows/approval_handoff.json
index cb70cf3..fbc7091 100644
--- a/backend/flows/approval_handoff.json
+++ b/backend/flows/approval_handoff.json
@@ -292,7 +292,7 @@
"list": false,
"show": true,
"multiline": true,
- "value": "from lfx.custom import Component\nfrom lfx.io import MessageTextInput, Output\nfrom lfx.schema.message import Message\nimport json\n\n\nclass EnvelopeComponent(Component):\n display_name = \"Envelope\"\n description = \"Zerlegt die Eingabe-Huelle in Payload und Laufzeit-Kontext.\"\n\n inputs = [\n MessageTextInput(name=\"raw\", display_name=\"Raw Input\"),\n ]\n outputs = [\n Output(display_name=\"Payload\", name=\"payload\", method=\"get_payload\"),\n Output(display_name=\"Run ID\", name=\"run_id\", method=\"get_run_id\"),\n Output(\n display_name=\"Callback URL\",\n name=\"callback_url\",\n method=\"get_callback_url\",\n ),\n Output(\n display_name=\"Continuation\",\n name=\"continuation\",\n method=\"get_continuation\",\n ),\n ]\n\n def _text(self, value) -> str:\n if value is None:\n return \"\"\n if hasattr(value, \"text\"):\n return value.text or \"\"\n return str(value)\n\n def _env(self) -> dict:\n if not hasattr(self, \"_env_cache\"):\n try:\n self._env_cache = json.loads(self._text(self.raw) or \"{}\")\n except Exception:\n self._env_cache = {}\n if not isinstance(self._env_cache, dict):\n self._env_cache = {}\n return self._env_cache\n\n def get_payload(self) -> Message:\n return Message(text=json.dumps(self._env().get(\"payload\", {})))\n\n def get_run_id(self) -> Message:\n return Message(text=str(self._env().get(\"run_id\", \"\")))\n\n def get_callback_url(self) -> Message:\n return Message(text=str(self._env().get(\"callback_url\", \"\")))\n\n def get_continuation(self) -> Message:\n value = self._env().get(\"continuation_flow_id\")\n return Message(text=str(value) if value else \"\")\n",
+ "value": "from lfx.custom import Component\nfrom lfx.io import MessageTextInput, Output\nfrom lfx.schema.message import Message\nimport json\n\n\nclass EnvelopeComponent(Component):\n display_name = \"Envelope\"\n description = \"Zerlegt die Eingabe-Huelle in Payload und Laufzeit-Kontext.\"\n\n inputs = [\n MessageTextInput(name=\"raw\", display_name=\"Raw Input\"),\n ]\n outputs = [\n Output(display_name=\"Payload\", name=\"payload\", method=\"get_payload\"),\n Output(display_name=\"Run ID\", name=\"run_id\", method=\"get_run_id\"),\n Output(\n display_name=\"Callback URL\",\n name=\"callback_url\",\n method=\"get_callback_url\",\n ),\n Output(\n display_name=\"Continuation\",\n name=\"continuation\",\n method=\"get_continuation\",\n ),\n Output(\n display_name=\"Callback Token\",\n name=\"callback_token\",\n method=\"get_callback_token\",\n ),\n ]\n\n def _text(self, value) -> str:\n if value is None:\n return \"\"\n if hasattr(value, \"text\"):\n return value.text or \"\"\n return str(value)\n\n def _env(self) -> dict:\n if not hasattr(self, \"_env_cache\"):\n try:\n self._env_cache = json.loads(self._text(self.raw) or \"{}\")\n except Exception:\n self._env_cache = {}\n if not isinstance(self._env_cache, dict):\n self._env_cache = {}\n return self._env_cache\n\n def get_payload(self) -> Message:\n return Message(text=json.dumps(self._env().get(\"payload\", {})))\n\n def get_run_id(self) -> Message:\n return Message(text=str(self._env().get(\"run_id\", \"\")))\n\n def get_callback_url(self) -> Message:\n return Message(text=str(self._env().get(\"callback_url\", \"\")))\n\n def get_continuation(self) -> Message:\n value = self._env().get(\"continuation_flow_id\")\n return Message(text=str(value) if value else \"\")\n\n def get_callback_token(self) -> Message:\n return Message(text=str(self._env().get(\"callback_token\", \"\")))\n",
"fileTypes": [],
"file_path": "",
"password": false,
@@ -399,6 +399,20 @@
"allows_loop": false,
"group_outputs": false,
"tool_mode": true
+ },
+ {
+ "types": [
+ "Message"
+ ],
+ "selected": "Message",
+ "name": "callback_token",
+ "display_name": "Callback Token",
+ "method": "get_callback_token",
+ "value": "__UNDEFINED__",
+ "cache": true,
+ "allows_loop": false,
+ "group_outputs": false,
+ "tool_mode": true
}
],
"field_order": [
@@ -409,7 +423,7 @@
"edited": false,
"metadata": {
"module": "custom_components.envelope",
- "code_hash": "2b0bd3c3d529",
+ "code_hash": "02e69613e210",
"dependencies": {
"total_dependencies": 1,
"dependencies": [
@@ -858,6 +872,32 @@
"node": {
"template": {
"_type": "Component",
+ "callback_token": {
+ "tool_mode": false,
+ "trace_as_input": true,
+ "trace_as_metadata": true,
+ "load_from_db": false,
+ "list": false,
+ "list_add_label": "Add More",
+ "override_skip": false,
+ "required": false,
+ "placeholder": "",
+ "show": true,
+ "name": "callback_token",
+ "value": "",
+ "display_name": "Callback Token",
+ "advanced": false,
+ "api_editable": false,
+ "input_types": [
+ "Message"
+ ],
+ "dynamic": false,
+ "info": "",
+ "title_case": false,
+ "track_in_telemetry": false,
+ "type": "str",
+ "_input_type": "MessageTextInput"
+ },
"callback_url": {
"tool_mode": false,
"trace_as_input": true,
@@ -891,7 +931,7 @@
"list": false,
"show": true,
"multiline": true,
- "value": "from lfx.custom import Component\nfrom lfx.io import MessageTextInput, Output\nfrom lfx.schema.message import Message\n\n\nclass ApprovalCallbackComponent(Component):\n display_name = \"Approval Callback\"\n description = \"Meldet einen Sonderfall per HTTP an unsere API und terminiert.\"\n\n inputs = [\n MessageTextInput(name=\"trigger\", display_name=\"Trigger\"),\n MessageTextInput(name=\"run_id\", display_name=\"Run ID\"),\n MessageTextInput(name=\"callback_url\", display_name=\"Callback URL\"),\n MessageTextInput(\n name=\"continuation_flow_id\",\n display_name=\"Continuation Flow ID\",\n ),\n MessageTextInput(name=\"facts\", display_name=\"Facts\"),\n ]\n outputs = [\n Output(display_name=\"Result\", name=\"result\", method=\"notify\"),\n ]\n\n def _text(self, value) -> str:\n if value is None:\n return \"\"\n if hasattr(value, \"text\"):\n return value.text or \"\"\n return str(value)\n\n def notify(self) -> Message:\n import httpx\n import json\n\n trigger = self._text(self.trigger).strip()\n if not trigger:\n # Zweig wurde nicht genommen: kein Seiteneffekt.\n return Message(text=\"\")\n try:\n facts = json.loads(self._text(self.facts) or \"{}\")\n except Exception:\n facts = {}\n url = self._text(self.callback_url).rstrip(\"/\") + \"/api/approvals\"\n body = {\n \"run_id\": self._text(self.run_id),\n \"title\": f\"Freigabe erforderlich ({trigger})\",\n \"payload\": {\"verdict\": trigger, \"facts\": facts},\n \"continuation_flow_id\": self._text(self.continuation_flow_id) or None,\n }\n try:\n response = httpx.post(url, json=body, timeout=10.0)\n response.raise_for_status()\n except Exception as exc:\n return Message(text=f\"CALLBACK_FAILED: {exc}\")\n return Message(text=\"PENDING_HUMAN_APPROVAL\")\n",
+ "value": "from lfx.custom import Component\nfrom lfx.io import MessageTextInput, Output\nfrom lfx.schema.message import Message\n\n\nclass ApprovalCallbackComponent(Component):\n display_name = \"Approval Callback\"\n description = \"Meldet einen Sonderfall per HTTP an unsere API und terminiert.\"\n\n inputs = [\n MessageTextInput(name=\"trigger\", display_name=\"Trigger\"),\n MessageTextInput(name=\"run_id\", display_name=\"Run ID\"),\n MessageTextInput(name=\"callback_url\", display_name=\"Callback URL\"),\n MessageTextInput(\n name=\"callback_token\",\n display_name=\"Callback Token\",\n ),\n MessageTextInput(\n name=\"continuation_flow_id\",\n display_name=\"Continuation Flow ID\",\n ),\n MessageTextInput(name=\"facts\", display_name=\"Facts\"),\n ]\n outputs = [\n Output(display_name=\"Result\", name=\"result\", method=\"notify\"),\n ]\n\n def _text(self, value) -> str:\n if value is None:\n return \"\"\n if hasattr(value, \"text\"):\n return value.text or \"\"\n return str(value)\n\n def notify(self) -> Message:\n import httpx\n import json\n\n trigger = self._text(self.trigger).strip()\n if not trigger:\n # Zweig wurde nicht genommen: kein Seiteneffekt.\n return Message(text=\"\")\n try:\n facts = json.loads(self._text(self.facts) or \"{}\")\n except Exception:\n facts = {}\n url = self._text(self.callback_url).rstrip(\"/\") + \"/api/approvals\"\n body = {\n \"run_id\": self._text(self.run_id),\n \"title\": f\"Freigabe erforderlich ({trigger})\",\n \"payload\": {\"verdict\": trigger, \"facts\": facts},\n \"continuation_flow_id\": self._text(self.continuation_flow_id) or None,\n }\n headers = {\"X-Callback-Token\": self._text(self.callback_token)}\n try:\n response = httpx.post(url, json=body, headers=headers, timeout=10.0)\n response.raise_for_status()\n except Exception as exc:\n return Message(text=f\"CALLBACK_FAILED: {exc}\")\n return Message(text=\"PENDING_HUMAN_APPROVAL\")\n",
"fileTypes": [],
"file_path": "",
"password": false,
@@ -1040,6 +1080,7 @@
"trigger",
"run_id",
"callback_url",
+ "callback_token",
"continuation_flow_id",
"facts"
],
@@ -1048,7 +1089,7 @@
"edited": false,
"metadata": {
"module": "custom_components.approval_callback",
- "code_hash": "a334376410f7",
+ "code_hash": "28b2f5894566",
"dependencies": {
"total_dependencies": 2,
"dependencies": [
@@ -1804,6 +1845,34 @@
"target": "CustomComponent-callback",
"targetHandle": "{œfieldNameœ: œcallback_urlœ, œidœ: œCustomComponent-callbackœ, œinputTypesœ: [œMessageœ], œtypeœ: œstrœ}"
},
+ {
+ "animated": false,
+ "className": "",
+ "data": {
+ "sourceHandle": {
+ "dataType": "CustomComponent",
+ "id": "CustomComponent-envelope",
+ "name": "callback_token",
+ "output_types": [
+ "Message"
+ ]
+ },
+ "targetHandle": {
+ "fieldName": "callback_token",
+ "id": "CustomComponent-callback",
+ "inputTypes": [
+ "Message"
+ ],
+ "type": "str"
+ }
+ },
+ "id": "xy-edge__{œdataTypeœ: œCustomComponentœ, œidœ: œCustomComponent-envelopeœ, œnameœ: œcallback_tokenœ, œoutput_typesœ: [œMessageœ]}-{œfieldNameœ: œcallback_tokenœ, œidœ: œCustomComponent-callbackœ, œinputTypesœ: [œMessageœ], œtypeœ: œstrœ}",
+ "selected": false,
+ "source": "CustomComponent-envelope",
+ "sourceHandle": "{œdataTypeœ: œCustomComponentœ, œidœ: œCustomComponent-envelopeœ, œnameœ: œcallback_tokenœ, œoutput_typesœ: [œMessageœ]}",
+ "target": "CustomComponent-callback",
+ "targetHandle": "{œfieldNameœ: œcallback_tokenœ, œidœ: œCustomComponent-callbackœ, œinputTypesœ: [œMessageœ], œtypeœ: œstrœ}"
+ },
{
"animated": false,
"className": "",
diff --git a/backend/scripts/make_flow_fixtures.py b/backend/scripts/make_flow_fixtures.py
index 3237c82..49e309d 100644
--- a/backend/scripts/make_flow_fixtures.py
+++ b/backend/scripts/make_flow_fixtures.py
@@ -302,6 +302,11 @@ class EnvelopeComponent(Component):
name="continuation",
method="get_continuation",
),
+ Output(
+ display_name="Callback Token",
+ name="callback_token",
+ method="get_callback_token",
+ ),
]
def _text(self, value) -> str:
@@ -333,6 +338,9 @@ class EnvelopeComponent(Component):
def get_continuation(self) -> Message:
value = self._env().get("continuation_flow_id")
return Message(text=str(value) if value else "")
+
+ def get_callback_token(self) -> Message:
+ return Message(text=str(self._env().get("callback_token", "")))
"""
APPROVAL_CALLBACK_CODE = """from lfx.custom import Component
@@ -348,6 +356,10 @@ class ApprovalCallbackComponent(Component):
MessageTextInput(name="trigger", display_name="Trigger"),
MessageTextInput(name="run_id", display_name="Run ID"),
MessageTextInput(name="callback_url", display_name="Callback URL"),
+ MessageTextInput(
+ name="callback_token",
+ display_name="Callback Token",
+ ),
MessageTextInput(
name="continuation_flow_id",
display_name="Continuation Flow ID",
@@ -384,8 +396,9 @@ class ApprovalCallbackComponent(Component):
"payload": {"verdict": trigger, "facts": facts},
"continuation_flow_id": self._text(self.continuation_flow_id) or None,
}
+ headers = {"X-Callback-Token": self._text(self.callback_token)}
try:
- response = httpx.post(url, json=body, timeout=10.0)
+ response = httpx.post(url, json=body, headers=headers, timeout=10.0)
response.raise_for_status()
except Exception as exc:
return Message(text=f"CALLBACK_FAILED: {exc}")
@@ -487,6 +500,7 @@ def build_approval_handoff() -> dict:
edge(router, "true_result", callback, "trigger"),
edge(envelope, "run_id", callback, "run_id"),
edge(envelope, "callback_url", callback, "callback_url"),
+ edge(envelope, "callback_token", callback, "callback_token"),
edge(envelope, "continuation", callback, "continuation_flow_id"),
edge(envelope, "payload", callback, "facts"),
edge(callback, "result", out_pending, "input_value"),
diff --git a/backend/scripts/smoke_e2e.py b/backend/scripts/smoke_e2e.py
index 358e58d..f90b81e 100644
--- a/backend/scripts/smoke_e2e.py
+++ b/backend/scripts/smoke_e2e.py
@@ -10,6 +10,8 @@ from __future__ import annotations
import json
import os
+import shutil
+import socket
import subprocess
import sys
import tempfile
@@ -18,11 +20,21 @@ import urllib.request
from pathlib import Path
BACKEND = Path(__file__).resolve().parent.parent
-VENV_BIN = BACKEND / ".venv" / "bin"
+# Local venv (uv run) or image venv (/opt/venv in containers): prefer PATH.
+VENV_BIN = Path(shutil.which("lfx") or "").parent or (BACKEND / ".venv" / "bin")
FIXTURES = BACKEND / "tests" / "fixtures" / "flows"
-LFX_PORT = 6004
-API_PORT = 8000
+
+def _free_port() -> int:
+ # Ephemeral ports: the dev API/LFX may already occupy the defaults
+ # (e.g. when smoke runs inside the api container itself).
+ with socket.socket() as s:
+ s.bind(("127.0.0.1", 0))
+ return s.getsockname()[1]
+
+
+LFX_PORT = _free_port()
+API_PORT = _free_port()
LFX_KEY = "lfx-759GsfJKjAeOBZ_44-cll9WdHp_yfYn_tIg7eHG38Rs"
@@ -42,6 +54,9 @@ def api(method: str, path: str, body: dict | None = None) -> tuple[int, dict | s
import httpx
with httpx.Client(base_url=f"http://127.0.0.1:{API_PORT}", timeout=120) as c:
+ # Management + runs are login-gated; the smoke admin session is
+ # planted directly in the throwaway DB (see main()).
+ c.cookies.set("kirche_session", "smoke-session", domain="127.0.0.1")
r = c.request(method, path, json=body)
try:
return r.status_code, r.json()
@@ -68,8 +83,36 @@ def main() -> None:
cfg = Config()
cfg.set_main_option("script_location", str(BACKEND / "migrations"))
cfg.set_main_option("sqlalchemy.url", db_url)
+ # migrations/env.py takes the URL from settings (not the ini value):
+ # point this process at the throwaway DB before migrating.
+ from app.settings import settings as app_settings
+
+ app_settings.database_url = db_url
alembic_command.upgrade(cfg, "head")
+ # Plant an admin session: management + runs require login, and there is
+ # no SMTP in smoke to deliver a magic link.
+ from datetime import UTC, datetime, timedelta
+
+ from sqlmodel import Session, create_engine
+
+ from app.models import User, UserSession
+
+ engine = create_engine(db_url)
+ with Session(engine) as s:
+ user = User(name="smoke", email="smoke@example.com", role="admin")
+ s.add(user)
+ s.flush()
+ s.add(
+ UserSession(
+ id="smoke-session",
+ user_id=user.id,
+ expires_at=datetime.now(UTC) + timedelta(hours=1),
+ )
+ )
+ s.commit()
+ engine.dispose()
+
env = {
**os.environ,
# lfx prefers LANGFLOW_API_KEY over LFX_API_KEY; the repo .env may set
@@ -80,6 +123,7 @@ def main() -> None:
"FLOWS_WATCH_ENABLED": "false",
"LFX_URL": f"http://127.0.0.1:{LFX_PORT}",
"CALLBACK_BASE_URL": f"http://127.0.0.1:{API_PORT}",
+ "CALLBACK_TOKEN": "smoke-callback-token",
}
lfx = subprocess.Popen(
[
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index 7f2ec91..6af8fd5 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -17,6 +17,8 @@ TEST_AREA = Path(tempfile.mkdtemp(prefix="kirche-tests-"))
# Must precede all `app.*` imports: Settings is instantiated at import time.
os.environ["DATABASE_URL"] = f"sqlite:///{TEST_AREA / 'test.db'}"
os.environ["FLOWS_DIR"] = str(TEST_AREA / "flows")
+# Flow callbacks (POST /api/approvals) authenticate with this token.
+os.environ["CALLBACK_TOKEN"] = "test-callback-token"
# The embedded watcher + startup sync stay off in tests; sync is exercised
# directly (sync_from_directory) with per-test tmp dirs instead.
os.environ["FLOWS_WATCH_ENABLED"] = "false"
@@ -81,6 +83,48 @@ async def client():
yield c
+@pytest.fixture()
+def login(client, session):
+ """Log the test client in as a fresh user: ``await login()`` (admin)
+ or ``await login(role="member", tenant_id="pfarramt-a")``."""
+
+ async def _login(
+ email: str = "admin@example.com",
+ role: str = "admin",
+ tenant_id: str = "default",
+ ):
+ from datetime import UTC, datetime, timedelta
+ from uuid import uuid4
+
+ from sqlmodel import select
+
+ from app.auth.deps import SESSION_COOKIE
+ from app.models import User, UserSession
+
+ # Idempotent: re-login updates the existing row (tests switch
+ # identity mid-test by calling login() again).
+ user = session.exec(select(User).where(User.email == email)).first()
+ if user is None:
+ user = User(name="Test", email=email, role=role, tenant_id=tenant_id)
+ session.add(user)
+ else:
+ user.role = role
+ user.tenant_id = tenant_id
+ session.add(user)
+ session.flush()
+ user_session = UserSession(
+ id=f"test-session-{uuid4().hex}",
+ user_id=user.id,
+ expires_at=datetime.now(UTC) + timedelta(days=1),
+ )
+ session.add(user_session)
+ session.commit()
+ client.cookies.set(SESSION_COOKIE, user_session.id)
+ return user
+
+ return _login
+
+
@pytest.fixture(scope="session")
def lfx_server(tmp_path_factory):
"""Real `lfx serve` subprocess on a free port; points settings at it.
diff --git a/backend/tests/fixtures/flows/approval_handoff.json b/backend/tests/fixtures/flows/approval_handoff.json
index cb70cf3..fbc7091 100644
--- a/backend/tests/fixtures/flows/approval_handoff.json
+++ b/backend/tests/fixtures/flows/approval_handoff.json
@@ -292,7 +292,7 @@
"list": false,
"show": true,
"multiline": true,
- "value": "from lfx.custom import Component\nfrom lfx.io import MessageTextInput, Output\nfrom lfx.schema.message import Message\nimport json\n\n\nclass EnvelopeComponent(Component):\n display_name = \"Envelope\"\n description = \"Zerlegt die Eingabe-Huelle in Payload und Laufzeit-Kontext.\"\n\n inputs = [\n MessageTextInput(name=\"raw\", display_name=\"Raw Input\"),\n ]\n outputs = [\n Output(display_name=\"Payload\", name=\"payload\", method=\"get_payload\"),\n Output(display_name=\"Run ID\", name=\"run_id\", method=\"get_run_id\"),\n Output(\n display_name=\"Callback URL\",\n name=\"callback_url\",\n method=\"get_callback_url\",\n ),\n Output(\n display_name=\"Continuation\",\n name=\"continuation\",\n method=\"get_continuation\",\n ),\n ]\n\n def _text(self, value) -> str:\n if value is None:\n return \"\"\n if hasattr(value, \"text\"):\n return value.text or \"\"\n return str(value)\n\n def _env(self) -> dict:\n if not hasattr(self, \"_env_cache\"):\n try:\n self._env_cache = json.loads(self._text(self.raw) or \"{}\")\n except Exception:\n self._env_cache = {}\n if not isinstance(self._env_cache, dict):\n self._env_cache = {}\n return self._env_cache\n\n def get_payload(self) -> Message:\n return Message(text=json.dumps(self._env().get(\"payload\", {})))\n\n def get_run_id(self) -> Message:\n return Message(text=str(self._env().get(\"run_id\", \"\")))\n\n def get_callback_url(self) -> Message:\n return Message(text=str(self._env().get(\"callback_url\", \"\")))\n\n def get_continuation(self) -> Message:\n value = self._env().get(\"continuation_flow_id\")\n return Message(text=str(value) if value else \"\")\n",
+ "value": "from lfx.custom import Component\nfrom lfx.io import MessageTextInput, Output\nfrom lfx.schema.message import Message\nimport json\n\n\nclass EnvelopeComponent(Component):\n display_name = \"Envelope\"\n description = \"Zerlegt die Eingabe-Huelle in Payload und Laufzeit-Kontext.\"\n\n inputs = [\n MessageTextInput(name=\"raw\", display_name=\"Raw Input\"),\n ]\n outputs = [\n Output(display_name=\"Payload\", name=\"payload\", method=\"get_payload\"),\n Output(display_name=\"Run ID\", name=\"run_id\", method=\"get_run_id\"),\n Output(\n display_name=\"Callback URL\",\n name=\"callback_url\",\n method=\"get_callback_url\",\n ),\n Output(\n display_name=\"Continuation\",\n name=\"continuation\",\n method=\"get_continuation\",\n ),\n Output(\n display_name=\"Callback Token\",\n name=\"callback_token\",\n method=\"get_callback_token\",\n ),\n ]\n\n def _text(self, value) -> str:\n if value is None:\n return \"\"\n if hasattr(value, \"text\"):\n return value.text or \"\"\n return str(value)\n\n def _env(self) -> dict:\n if not hasattr(self, \"_env_cache\"):\n try:\n self._env_cache = json.loads(self._text(self.raw) or \"{}\")\n except Exception:\n self._env_cache = {}\n if not isinstance(self._env_cache, dict):\n self._env_cache = {}\n return self._env_cache\n\n def get_payload(self) -> Message:\n return Message(text=json.dumps(self._env().get(\"payload\", {})))\n\n def get_run_id(self) -> Message:\n return Message(text=str(self._env().get(\"run_id\", \"\")))\n\n def get_callback_url(self) -> Message:\n return Message(text=str(self._env().get(\"callback_url\", \"\")))\n\n def get_continuation(self) -> Message:\n value = self._env().get(\"continuation_flow_id\")\n return Message(text=str(value) if value else \"\")\n\n def get_callback_token(self) -> Message:\n return Message(text=str(self._env().get(\"callback_token\", \"\")))\n",
"fileTypes": [],
"file_path": "",
"password": false,
@@ -399,6 +399,20 @@
"allows_loop": false,
"group_outputs": false,
"tool_mode": true
+ },
+ {
+ "types": [
+ "Message"
+ ],
+ "selected": "Message",
+ "name": "callback_token",
+ "display_name": "Callback Token",
+ "method": "get_callback_token",
+ "value": "__UNDEFINED__",
+ "cache": true,
+ "allows_loop": false,
+ "group_outputs": false,
+ "tool_mode": true
}
],
"field_order": [
@@ -409,7 +423,7 @@
"edited": false,
"metadata": {
"module": "custom_components.envelope",
- "code_hash": "2b0bd3c3d529",
+ "code_hash": "02e69613e210",
"dependencies": {
"total_dependencies": 1,
"dependencies": [
@@ -858,6 +872,32 @@
"node": {
"template": {
"_type": "Component",
+ "callback_token": {
+ "tool_mode": false,
+ "trace_as_input": true,
+ "trace_as_metadata": true,
+ "load_from_db": false,
+ "list": false,
+ "list_add_label": "Add More",
+ "override_skip": false,
+ "required": false,
+ "placeholder": "",
+ "show": true,
+ "name": "callback_token",
+ "value": "",
+ "display_name": "Callback Token",
+ "advanced": false,
+ "api_editable": false,
+ "input_types": [
+ "Message"
+ ],
+ "dynamic": false,
+ "info": "",
+ "title_case": false,
+ "track_in_telemetry": false,
+ "type": "str",
+ "_input_type": "MessageTextInput"
+ },
"callback_url": {
"tool_mode": false,
"trace_as_input": true,
@@ -891,7 +931,7 @@
"list": false,
"show": true,
"multiline": true,
- "value": "from lfx.custom import Component\nfrom lfx.io import MessageTextInput, Output\nfrom lfx.schema.message import Message\n\n\nclass ApprovalCallbackComponent(Component):\n display_name = \"Approval Callback\"\n description = \"Meldet einen Sonderfall per HTTP an unsere API und terminiert.\"\n\n inputs = [\n MessageTextInput(name=\"trigger\", display_name=\"Trigger\"),\n MessageTextInput(name=\"run_id\", display_name=\"Run ID\"),\n MessageTextInput(name=\"callback_url\", display_name=\"Callback URL\"),\n MessageTextInput(\n name=\"continuation_flow_id\",\n display_name=\"Continuation Flow ID\",\n ),\n MessageTextInput(name=\"facts\", display_name=\"Facts\"),\n ]\n outputs = [\n Output(display_name=\"Result\", name=\"result\", method=\"notify\"),\n ]\n\n def _text(self, value) -> str:\n if value is None:\n return \"\"\n if hasattr(value, \"text\"):\n return value.text or \"\"\n return str(value)\n\n def notify(self) -> Message:\n import httpx\n import json\n\n trigger = self._text(self.trigger).strip()\n if not trigger:\n # Zweig wurde nicht genommen: kein Seiteneffekt.\n return Message(text=\"\")\n try:\n facts = json.loads(self._text(self.facts) or \"{}\")\n except Exception:\n facts = {}\n url = self._text(self.callback_url).rstrip(\"/\") + \"/api/approvals\"\n body = {\n \"run_id\": self._text(self.run_id),\n \"title\": f\"Freigabe erforderlich ({trigger})\",\n \"payload\": {\"verdict\": trigger, \"facts\": facts},\n \"continuation_flow_id\": self._text(self.continuation_flow_id) or None,\n }\n try:\n response = httpx.post(url, json=body, timeout=10.0)\n response.raise_for_status()\n except Exception as exc:\n return Message(text=f\"CALLBACK_FAILED: {exc}\")\n return Message(text=\"PENDING_HUMAN_APPROVAL\")\n",
+ "value": "from lfx.custom import Component\nfrom lfx.io import MessageTextInput, Output\nfrom lfx.schema.message import Message\n\n\nclass ApprovalCallbackComponent(Component):\n display_name = \"Approval Callback\"\n description = \"Meldet einen Sonderfall per HTTP an unsere API und terminiert.\"\n\n inputs = [\n MessageTextInput(name=\"trigger\", display_name=\"Trigger\"),\n MessageTextInput(name=\"run_id\", display_name=\"Run ID\"),\n MessageTextInput(name=\"callback_url\", display_name=\"Callback URL\"),\n MessageTextInput(\n name=\"callback_token\",\n display_name=\"Callback Token\",\n ),\n MessageTextInput(\n name=\"continuation_flow_id\",\n display_name=\"Continuation Flow ID\",\n ),\n MessageTextInput(name=\"facts\", display_name=\"Facts\"),\n ]\n outputs = [\n Output(display_name=\"Result\", name=\"result\", method=\"notify\"),\n ]\n\n def _text(self, value) -> str:\n if value is None:\n return \"\"\n if hasattr(value, \"text\"):\n return value.text or \"\"\n return str(value)\n\n def notify(self) -> Message:\n import httpx\n import json\n\n trigger = self._text(self.trigger).strip()\n if not trigger:\n # Zweig wurde nicht genommen: kein Seiteneffekt.\n return Message(text=\"\")\n try:\n facts = json.loads(self._text(self.facts) or \"{}\")\n except Exception:\n facts = {}\n url = self._text(self.callback_url).rstrip(\"/\") + \"/api/approvals\"\n body = {\n \"run_id\": self._text(self.run_id),\n \"title\": f\"Freigabe erforderlich ({trigger})\",\n \"payload\": {\"verdict\": trigger, \"facts\": facts},\n \"continuation_flow_id\": self._text(self.continuation_flow_id) or None,\n }\n headers = {\"X-Callback-Token\": self._text(self.callback_token)}\n try:\n response = httpx.post(url, json=body, headers=headers, timeout=10.0)\n response.raise_for_status()\n except Exception as exc:\n return Message(text=f\"CALLBACK_FAILED: {exc}\")\n return Message(text=\"PENDING_HUMAN_APPROVAL\")\n",
"fileTypes": [],
"file_path": "",
"password": false,
@@ -1040,6 +1080,7 @@
"trigger",
"run_id",
"callback_url",
+ "callback_token",
"continuation_flow_id",
"facts"
],
@@ -1048,7 +1089,7 @@
"edited": false,
"metadata": {
"module": "custom_components.approval_callback",
- "code_hash": "a334376410f7",
+ "code_hash": "28b2f5894566",
"dependencies": {
"total_dependencies": 2,
"dependencies": [
@@ -1804,6 +1845,34 @@
"target": "CustomComponent-callback",
"targetHandle": "{œfieldNameœ: œcallback_urlœ, œidœ: œCustomComponent-callbackœ, œinputTypesœ: [œMessageœ], œtypeœ: œstrœ}"
},
+ {
+ "animated": false,
+ "className": "",
+ "data": {
+ "sourceHandle": {
+ "dataType": "CustomComponent",
+ "id": "CustomComponent-envelope",
+ "name": "callback_token",
+ "output_types": [
+ "Message"
+ ]
+ },
+ "targetHandle": {
+ "fieldName": "callback_token",
+ "id": "CustomComponent-callback",
+ "inputTypes": [
+ "Message"
+ ],
+ "type": "str"
+ }
+ },
+ "id": "xy-edge__{œdataTypeœ: œCustomComponentœ, œidœ: œCustomComponent-envelopeœ, œnameœ: œcallback_tokenœ, œoutput_typesœ: [œMessageœ]}-{œfieldNameœ: œcallback_tokenœ, œidœ: œCustomComponent-callbackœ, œinputTypesœ: [œMessageœ], œtypeœ: œstrœ}",
+ "selected": false,
+ "source": "CustomComponent-envelope",
+ "sourceHandle": "{œdataTypeœ: œCustomComponentœ, œidœ: œCustomComponent-envelopeœ, œnameœ: œcallback_tokenœ, œoutput_typesœ: [œMessageœ]}",
+ "target": "CustomComponent-callback",
+ "targetHandle": "{œfieldNameœ: œcallback_tokenœ, œidœ: œCustomComponent-callbackœ, œinputTypesœ: [œMessageœ], œtypeœ: œstrœ}"
+ },
{
"animated": false,
"className": "",
diff --git a/backend/tests/test_approvals.py b/backend/tests/test_approvals.py
index 0f80b6b..2ee5661 100644
--- a/backend/tests/test_approvals.py
+++ b/backend/tests/test_approvals.py
@@ -30,16 +30,14 @@ async def _version_and_deploy(client, flow_id, fixture="simple_rule.json"):
return response.json()
-async def _handoff_task(client, live_server, continuation_flow_id=None):
+async def _handoff_task(client, live_server, monkeypatch):
"""Run approval_handoff over threshold; returns (run, task)."""
+ from app.settings import settings
+
+ monkeypatch.setattr(settings, "callback_base_url", live_server)
flow = await _create_flow(client, f"handoff-{uuid4().hex[:8]}")
await _version_and_deploy(client, flow["id"], fixture="approval_handoff.json")
- body = {
- "input": {"betrag": 250000, "schwelle": 200000},
- "callback_base_url": live_server,
- }
- if continuation_flow_id is not None:
- body["continuation_flow_id"] = continuation_flow_id
+ body = {"input": {"betrag": 250000, "schwelle": 200000}}
run = (await client.post(f"/api/flows/{flow['id']}/run", json=body)).json()
assert run["status"] == "pending_approval"
tasks = (await client.get("/api/approvals")).json()
@@ -47,30 +45,55 @@ async def _handoff_task(client, live_server, continuation_flow_id=None):
return run, task
-async def test_approve_without_continuation(client, lfx_server, live_server):
- _, task = await _handoff_task(client, live_server)
+async def test_approve_without_continuation(client, lfx_server, live_server, login, monkeypatch):
+ await login()
+ _, task = await _handoff_task(client, live_server, monkeypatch)
response = await client.post(
f"/api/approvals/{task['id']}/decision",
- json={"decision": "approve", "approver": "sachbearbeiter", "note": "ok"},
+ json={"decision": "approve", "note": "ok"},
)
assert response.status_code == 200, response.text
body = response.json()
assert body["task"]["status"] == "approved"
- assert body["task"]["approver"] == "sachbearbeiter"
+ assert body["task"]["approver"] == "admin@example.com"
assert body["task"]["decided_at"] is not None
assert body["continuation_run"] is None
-async def test_approve_with_continuation_runs_facts(client, lfx_server, live_server):
+async def test_approve_with_continuation_runs_facts(
+ client, lfx_server, live_server, login, monkeypatch
+):
+ await login()
cont = await _create_flow(client, f"cont-{uuid4().hex[:8]}")
await _version_and_deploy(client, cont["id"], fixture="simple_rule.json")
- _, task = await _handoff_task(client, live_server, continuation_flow_id=cont["id"])
+ # Continuation is authoring-time wiring on the definition (admin-only);
+ # run bodies cannot set it.
+ response = await client.post(
+ "/api/flows",
+ json={"name": f"handoff-{uuid4().hex[:8]}", "continuation_flow_id": cont["id"]},
+ )
+ assert response.status_code == 201, response.text
+ flow = response.json()
+ await _version_and_deploy(client, flow["id"], fixture="approval_handoff.json")
+
+ from app.settings import settings
+
+ monkeypatch.setattr(settings, "callback_base_url", live_server)
+ run = (
+ await client.post(
+ f"/api/flows/{flow['id']}/run",
+ json={"input": {"betrag": 250000, "schwelle": 200000}},
+ )
+ ).json()
+ assert run["status"] == "pending_approval"
+ tasks = (await client.get("/api/approvals")).json()
+ task = next(t for t in tasks if t["run_id"] == run["id"])
assert task["continuation_flow_id"] == cont["id"]
response = await client.post(
f"/api/approvals/{task['id']}/decision",
- json={"decision": "approve", "approver": "chef"},
+ json={"decision": "approve"},
)
assert response.status_code == 200, response.text
body = response.json()
@@ -83,21 +106,51 @@ async def test_approve_with_continuation_runs_facts(client, lfx_server, live_ser
assert "Sonderfall" in cont_run["output"]["result"]
-async def test_reject_task(client, lfx_server, live_server):
- _, task = await _handoff_task(client, live_server)
+async def test_client_continuation_is_ignored(client, lfx_server, live_server, login, monkeypatch):
+ """A run body cannot chain arbitrary flows (G)."""
+ from app.settings import settings
+
+ await login()
+ monkeypatch.setattr(settings, "callback_base_url", live_server)
+ cont = await _create_flow(client, f"cont-{uuid4().hex[:8]}")
+ await _version_and_deploy(client, cont["id"], fixture="simple_rule.json")
+ flow = await _create_flow(client, f"handoff-{uuid4().hex[:8]}")
+ await _version_and_deploy(client, flow["id"], fixture="approval_handoff.json")
+ run = (
+ await client.post(
+ f"/api/flows/{flow['id']}/run",
+ json={
+ "input": {"betrag": 250000, "schwelle": 200000},
+ "continuation_flow_id": cont["id"],
+ },
+ )
+ ).json()
+ assert run["status"] == "pending_approval"
+ tasks = (await client.get("/api/approvals")).json()
+ task = next(t for t in tasks if t["run_id"] == run["id"])
+ assert task["continuation_flow_id"] is None
+
+
+async def test_reject_task(client, lfx_server, live_server, login, monkeypatch):
+ await login()
+ _, task = await _handoff_task(client, live_server, monkeypatch)
+ # Any logged-in user (not just admins) can decide; identity is the session.
+ await login(email="chef@example.com", role="member")
response = await client.post(
f"/api/approvals/{task['id']}/decision",
- json={"decision": "reject", "approver": "chef", "note": "zu teuer"},
+ json={"decision": "reject", "note": "zu teuer"},
)
assert response.status_code == 200, response.text
body = response.json()
assert body["task"]["status"] == "rejected"
+ assert body["task"]["approver"] == "chef@example.com"
assert body["task"]["decision_note"] == "zu teuer"
assert body["continuation_run"] is None
-async def test_decision_guards(client, lfx_server, live_server):
- _, task = await _handoff_task(client, live_server)
+async def test_decision_guards(client, lfx_server, live_server, login, monkeypatch):
+ await login()
+ _, task = await _handoff_task(client, live_server, monkeypatch)
url = f"/api/approvals/{task['id']}/decision"
response = await client.post(url, json={"decision": "approve"})
@@ -108,7 +161,7 @@ async def test_decision_guards(client, lfx_server, live_server):
response = await client.post(f"/api/approvals/{uuid4()}/decision", json={"decision": "approve"})
assert response.status_code == 404
- _, task2 = await _handoff_task(client, live_server)
+ _, task2 = await _handoff_task(client, live_server, monkeypatch)
response = await client.post(
f"/api/approvals/{task2['id']}/decision", json={"decision": "maybe"}
)
@@ -122,9 +175,10 @@ async def test_decision_guards(client, lfx_server, live_server):
assert response.status_code == 200
-async def test_decision_bad_continuation_reference(client, session):
+async def test_decision_bad_continuation_reference(client, session, login):
from app.flows import store as flow_store
+ await login()
flow = await _create_flow(client, f"badcont-{uuid4().hex[:8]}")
version = (
await client.post(
@@ -147,39 +201,45 @@ async def test_decision_bad_continuation_reference(client, session):
assert response.status_code == 422
-async def test_tenant_isolation(client, lfx_server):
+async def test_tenant_isolation(client, lfx_server, login):
+ """Strict session scoping: explicit ?tenant_id= cannot widen reads (E)."""
+ await login()
flow = await _create_flow(client, f"tenant-{uuid4().hex[:8]}")
await _version_and_deploy(client, flow["id"])
- for tenant in ("pfarramt-a", "pfarramt-b"):
- response = await client.post(
- f"/api/flows/{flow['id']}/run",
- json={"input": {"betrag": 1}, "tenant_id": tenant},
- )
- assert response.status_code == 200
+ await login(email="a@example.com", role="member", tenant_id="pfarramt-a")
+ response = await client.post(f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 1}})
+ assert response.status_code == 200
+ run_a = response.json()
+ assert run_a["tenant_id"] == "pfarramt-a"
+ await login(email="b@example.com", role="member", tenant_id="pfarramt-b")
+ response = await client.post(f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 1}})
+ assert response.status_code == 200
+
+ # A forged tenant_id query param is ignored: b sees only b's runs.
response = await client.get(f"/api/flows/{flow['id']}/runs", params={"tenant_id": "pfarramt-a"})
- runs_a = response.json()
- assert len(runs_a) == 1
- assert runs_a[0]["tenant_id"] == "pfarramt-a"
+ runs_b = response.json()
+ assert len(runs_b) == 1
+ assert runs_b[0]["tenant_id"] == "pfarramt-b"
- response = await client.get(f"/api/flows/{flow['id']}/runs")
- assert len(response.json()) == 2
-
- # Cross-tenant read with scoping 404s.
- response = await client.get(
- f"/api/flows/runs/{runs_a[0]['id']}", params={"tenant_id": "pfarramt-b"}
- )
+ # Cross-tenant run read 404s with or without the forged param.
+ response = await client.get(f"/api/flows/runs/{run_a['id']}")
assert response.status_code == 404
response = await client.get(
- f"/api/flows/runs/{runs_a[0]['id']}", params={"tenant_id": "pfarramt-a"}
+ f"/api/flows/runs/{run_a['id']}", params={"tenant_id": "pfarramt-a"}
)
+ assert response.status_code == 404
+
+ await login(email="a@example.com", role="member", tenant_id="pfarramt-a")
+ response = await client.get(f"/api/flows/runs/{run_a['id']}")
assert response.status_code == 200
-async def test_lfx_down_maps_to_503(client, lfx_server, monkeypatch):
+async def test_lfx_down_maps_to_503(client, lfx_server, monkeypatch, login):
from app.settings import settings
+ await login()
flow = await _create_flow(client, f"down-{uuid4().hex[:8]}")
await client.post(
f"/api/flows/{flow['id']}/versions",
@@ -198,10 +258,11 @@ async def test_lfx_down_maps_to_503(client, lfx_server, monkeypatch):
assert response.status_code == 503
-async def test_lfx_drift_maps_to_502(client, lfx_server):
+async def test_lfx_drift_maps_to_502(client, lfx_server, login):
from app.flows.lfx_client import LfxClient
from app.settings import settings
+ await login()
flow = await _create_flow(client, f"drift-{uuid4().hex[:8]}")
deployed = await _version_and_deploy(client, flow["id"])
await LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None).delete_flow(
@@ -211,7 +272,8 @@ async def test_lfx_drift_maps_to_502(client, lfx_server):
assert response.status_code == 502
-async def test_corrupt_flow_json_deploy_maps_to_422(client, lfx_server):
+async def test_corrupt_flow_json_deploy_maps_to_422(client, lfx_server, login):
+ await login()
flow = await _create_flow(client, f"corrupt-{uuid4().hex[:8]}")
bogus = {
"name": "bogus",
diff --git a/backend/tests/test_deploy.py b/backend/tests/test_deploy.py
index 6315d1a..a8430a4 100644
--- a/backend/tests/test_deploy.py
+++ b/backend/tests/test_deploy.py
@@ -24,8 +24,11 @@ def _export_doc(name="importiert"):
}
-async def test_import_creates_definition_version_and_deploys(client, lfx_server, override_langflow):
+async def test_import_creates_definition_version_and_deploys(
+ client, login, lfx_server, override_langflow
+):
override_langflow(_export_doc())
+ await login()
response = await client.post(
"/api/flows/import",
json={"langflow_flow_id": LANGFLOW_ID, "author": "admin"},
@@ -46,8 +49,9 @@ async def test_import_creates_definition_version_and_deploys(client, lfx_server,
assert response.json()["status"] == "completed"
-async def test_import_same_content_is_idempotent(client, lfx_server, override_langflow):
+async def test_import_same_content_is_idempotent(client, login, lfx_server, override_langflow):
override_langflow(_export_doc())
+ await login()
first = await client.post("/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID})
second = await client.post("/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID})
assert first.status_code == 201
@@ -57,8 +61,9 @@ async def test_import_same_content_is_idempotent(client, lfx_server, override_la
assert second.json()["version"]["version_no"] == 1
-async def test_import_changed_content_bumps_version(client, lfx_server, override_langflow):
+async def test_import_changed_content_bumps_version(client, login, lfx_server, override_langflow):
override_langflow(_export_doc())
+ await login()
first = await client.post("/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID})
assert first.status_code == 201
changed = _export_doc()
@@ -70,12 +75,13 @@ async def test_import_changed_content_bumps_version(client, lfx_server, override
async def test_import_writes_canonical_file(
- client, lfx_server, override_langflow, tmp_path, monkeypatch
+ client, login, lfx_server, override_langflow, tmp_path, monkeypatch
):
from app.settings import settings
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
override_langflow(_export_doc())
+ await login()
response = await client.post("/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID})
assert response.status_code == 201, response.text
written = tmp_path / "importiert.json"
@@ -83,8 +89,9 @@ async def test_import_writes_canonical_file(
assert json.loads(written.read_text())["name"] == "importiert"
-async def test_import_with_explicit_name(client, lfx_server, override_langflow):
+async def test_import_with_explicit_name(client, login, lfx_server, override_langflow):
override_langflow(_export_doc(name="langflow-name"))
+ await login()
response = await client.post(
"/api/flows/import",
json={"langflow_flow_id": LANGFLOW_ID, "name": "wunschname"},
@@ -93,8 +100,9 @@ async def test_import_with_explicit_name(client, lfx_server, override_langflow):
assert response.json()["definition"]["name"] == "wunschname"
-async def test_import_without_deploy(client, override_langflow):
+async def test_import_without_deploy(client, login, override_langflow):
override_langflow(_export_doc())
+ await login()
response = await client.post(
"/api/flows/import",
json={"langflow_flow_id": LANGFLOW_ID, "deploy": False},
@@ -105,20 +113,23 @@ async def test_import_without_deploy(client, override_langflow):
assert body["version"]["deployed_at"] is None
-async def test_import_langflow_not_found(client, override_langflow):
+async def test_import_langflow_not_found(client, login, override_langflow):
override_langflow(status=404)
+ await login()
response = await client.post("/api/flows/import", json={"langflow_flow_id": LANGFLOW_ID})
assert response.status_code == 404
-async def test_import_langflow_unreachable(client, override_langflow):
+async def test_import_langflow_unreachable(client, login, override_langflow):
override_langflow(status=0)
+ await login()
response = await client.post("/api/flows/import", json={"langflow_flow_id": str(uuid4())})
assert response.status_code == 503
-async def test_list_langflow_flows(client, override_langflow):
+async def test_list_langflow_flows(client, login, override_langflow):
override_langflow(_export_doc())
+ await login()
response = await client.get("/api/flows/langflow/available")
assert response.status_code == 200
assert response.json() == [{"id": "fake-1", "name": "fake"}]
diff --git a/backend/tests/test_flows_api.py b/backend/tests/test_flows_api.py
index 3ca947e..cc29aff 100644
--- a/backend/tests/test_flows_api.py
+++ b/backend/tests/test_flows_api.py
@@ -7,6 +7,7 @@ from uuid import UUID, uuid4
from app.flows.sync import lfx_id_for
FIXTURES = Path(__file__).parent / "fixtures" / "flows"
+CALLBACK_HEADERS = {"X-Callback-Token": "test-callback-token"}
def load_fixture(name: str) -> dict:
@@ -34,7 +35,8 @@ async def _deploy(client, flow_id, version_no=1):
return response.json()
-async def test_flow_crud(client):
+async def test_flow_crud(client, login):
+ await login()
created = await _create_flow(client, "crud-flow")
assert created["status"] == "active"
@@ -53,7 +55,63 @@ async def test_flow_crud(client):
assert response.status_code == 404
-async def test_version_validation_and_numbering(client):
+async def test_management_endpoints_require_admin(client, login):
+ """Anonymous -> 401, member -> 403, admin -> through (A)."""
+ flow_id = uuid4()
+ probes = [
+ ("POST", "/api/flows", {"name": "x"}),
+ ("POST", "/api/flows/sync", None),
+ ("POST", f"/api/flows/{flow_id}/versions", {"langflow_json": {}}),
+ ("GET", f"/api/flows/{flow_id}/versions", None),
+ ("POST", f"/api/flows/{flow_id}/versions/1/deploy", None),
+ ("POST", "/api/flows/import", {"langflow_flow_id": "x"}),
+ ("GET", "/api/flows/langflow/available", None),
+ ]
+ for method, path, body in probes:
+ if method == "POST":
+ response = await client.post(path, json=body or {})
+ else:
+ response = await client.get(path)
+ assert response.status_code == 401, (method, path, response.text)
+
+ await login(email="member@example.com", role="member")
+ for method, path, body in probes:
+ if method == "POST":
+ response = await client.post(path, json=body or {})
+ else:
+ response = await client.get(path)
+ assert response.status_code == 403, (method, path, response.text)
+
+
+async def test_runs_and_reads_require_login(client, login):
+ """Run, run reads and approval reads are login-gated (A)."""
+ await login()
+ flow = await _create_flow(client, "gated")
+ run_id = uuid4()
+ authed_paths = [
+ ("POST", f"/api/flows/{flow['id']}/run", {"input": {}}),
+ ("GET", f"/api/flows/{flow['id']}/runs", None),
+ ("GET", f"/api/flows/runs/{run_id}", None),
+ ("GET", "/api/approvals", None),
+ ("POST", f"/api/approvals/{uuid4()}/decision", {"decision": "approve"}),
+ ]
+ # Fresh client without a session cookie.
+ from httpx import ASGITransport, AsyncClient
+
+ from app.main import app
+
+ transport = ASGITransport(app=app)
+ async with AsyncClient(transport=transport, base_url="http://test") as anon:
+ for method, path, body in authed_paths:
+ if method == "POST":
+ response = await anon.post(path, json=body or {})
+ else:
+ response = await anon.get(path)
+ assert response.status_code == 401, (method, path, response.text)
+
+
+async def test_version_validation_and_numbering(client, login):
+ await login()
flow = await _create_flow(client, "versioned")
v1 = await _create_version(client, flow["id"])
@@ -77,7 +135,8 @@ async def test_version_validation_and_numbering(client):
assert [v["version_no"] for v in response.json()] == [1, 2]
-async def test_deploy_and_run_standard_and_sonderfall(client, lfx_server):
+async def test_deploy_and_run_standard_and_sonderfall(client, lfx_server, login):
+ await login()
flow = await _create_flow(client, "dachsanierung")
await _create_version(client, flow["id"])
deployed = await _deploy(client, flow["id"])
@@ -108,14 +167,16 @@ async def test_deploy_and_run_standard_and_sonderfall(client, lfx_server):
assert response.json()["id"] == run["id"]
-async def test_run_undeployed_version_fails(client):
+async def test_run_undeployed_version_fails(client, login):
+ await login()
flow = await _create_flow(client, "undeployed")
await _create_version(client, flow["id"])
response = await client.post(f"/api/flows/{flow['id']}/run", json={"input": {"betrag": 1}})
assert response.status_code == 422
-async def test_run_explicit_version_no(client, lfx_server):
+async def test_run_explicit_version_no(client, lfx_server, login):
+ await login()
flow = await _create_flow(client, "explicit-version")
await _create_version(client, flow["id"])
await _create_version(client, flow["id"])
@@ -128,7 +189,8 @@ async def test_run_explicit_version_no(client, lfx_server):
assert response.json()["status"] == "completed"
-async def test_run_stream_events(client, lfx_server):
+async def test_run_stream_events(client, lfx_server, login):
+ await login()
flow = await _create_flow(client, "streamed")
await _create_version(client, flow["id"])
await _deploy(client, flow["id"])
@@ -161,17 +223,18 @@ async def test_run_stream_events(client, lfx_server):
assert response.json()["status"] == "completed"
-async def test_approval_handoff_end_to_end(client, lfx_server, live_server):
+async def test_approval_handoff_end_to_end(client, lfx_server, live_server, login, monkeypatch):
+ from app.settings import settings
+
+ await login()
+ monkeypatch.setattr(settings, "callback_base_url", live_server)
flow = await _create_flow(client, "freigabe")
await _create_version(client, flow["id"], fixture="approval_handoff.json")
await _deploy(client, flow["id"])
response = await client.post(
f"/api/flows/{flow['id']}/run",
- json={
- "input": {"betrag": 250000, "schwelle": 200000},
- "callback_base_url": live_server,
- },
+ json={"input": {"betrag": 250000, "schwelle": 200000}},
)
assert response.status_code == 200, response.text
run = response.json()
@@ -188,17 +251,20 @@ async def test_approval_handoff_end_to_end(client, lfx_server, live_server):
assert task["payload"]["facts"]["betrag"] == 250000
-async def test_approval_handoff_auto_creates_no_task(client, lfx_server, live_server):
+async def test_approval_handoff_auto_creates_no_task(
+ client, lfx_server, live_server, login, monkeypatch
+):
+ from app.settings import settings
+
+ await login()
+ monkeypatch.setattr(settings, "callback_base_url", live_server)
flow = await _create_flow(client, "freigabe-auto")
await _create_version(client, flow["id"], fixture="approval_handoff.json")
await _deploy(client, flow["id"])
response = await client.post(
f"/api/flows/{flow['id']}/run",
- json={
- "input": {"betrag": 50000, "schwelle": 200000},
- "callback_base_url": live_server,
- },
+ json={"input": {"betrag": 50000, "schwelle": 200000}},
)
assert response.status_code == 200, response.text
assert response.json()["status"] == "completed"
@@ -208,11 +274,12 @@ async def test_approval_handoff_auto_creates_no_task(client, lfx_server, live_se
assert response.json() == []
-async def test_approval_callback_contract(client, session):
+async def test_approval_callback_contract(client, session, login):
from uuid import UUID
from app.flows import store as flow_store
+ await login()
flow = await _create_flow(client, "callback-target")
version = await _create_version(client, flow["id"])
version_obj = flow_store.get_version(session, UUID(version["id"]))
@@ -222,6 +289,7 @@ async def test_approval_callback_contract(client, session):
response = await client.post(
"/api/approvals",
json={"run_id": str(run.id), "title": "bitte pruefen", "payload": {"x": 1}},
+ headers=CALLBACK_HEADERS,
)
assert response.status_code == 201, response.text
task = response.json()
@@ -229,18 +297,46 @@ async def test_approval_callback_contract(client, session):
assert task["tenant_id"] == "default"
assert task["version_id"] == version["id"]
- response = await client.get(f"/api/flows/runs/{run.id}")
+ response = await client.get(f"/api/flows/runs/{run.id}", headers=CALLBACK_HEADERS)
assert response.json()["status"] == "pending_approval"
- response = await client.post("/api/approvals", json={"run_id": str(run.id)})
+ response = await client.post(
+ "/api/approvals", json={"run_id": str(run.id)}, headers=CALLBACK_HEADERS
+ )
assert response.status_code == 409
run2 = flow_store.create_run(session, version_obj, input={}, tenant_id="default")
response = await client.post(
"/api/approvals",
json={"run_id": str(run2.id), "continuation_flow_id": str(uuid4())},
+ headers=CALLBACK_HEADERS,
)
assert response.status_code == 422
- response = await client.post("/api/approvals", json={"run_id": str(uuid4())})
+ response = await client.post(
+ "/api/approvals", json={"run_id": str(uuid4())}, headers=CALLBACK_HEADERS
+ )
assert response.status_code == 404
+
+
+async def test_approval_callback_token_required(client, session, login):
+ """Forged callbacks without the shared token are rejected (C)."""
+ from uuid import UUID
+
+ from app.flows import store as flow_store
+
+ await login()
+ flow = await _create_flow(client, "callback-gated")
+ version = await _create_version(client, flow["id"])
+ version_obj = flow_store.get_version(session, UUID(version["id"]))
+ run = flow_store.create_run(session, version_obj, input={}, tenant_id="default")
+
+ response = await client.post("/api/approvals", json={"run_id": str(run.id)})
+ assert response.status_code == 401
+
+ response = await client.post(
+ "/api/approvals",
+ json={"run_id": str(run.id)},
+ headers={"X-Callback-Token": "wrong"},
+ )
+ assert response.status_code == 401
diff --git a/backend/tests/test_form_components.py b/backend/tests/test_form_components.py
index 0977747..3d047b5 100644
--- a/backend/tests/test_form_components.py
+++ b/backend/tests/test_form_components.py
@@ -77,7 +77,8 @@ def test_intake_fixture_shape():
assert ("CustomInputForm-intake", "ChatOutput-collected") in edges
-async def test_intake_deploy_and_run_with_values(client, lfx_server):
+async def test_intake_deploy_and_run_with_values(client, lfx_server, login):
+ await login()
doc = json.loads((FIXTURES / "intake_form.json").read_text())
created = await client.post("/api/flows", json={"name": "intake-e2e"})
assert created.status_code == 201, created.text
diff --git a/backend/tests/test_input_schema.py b/backend/tests/test_input_schema.py
index 81c974d..b81e3fd 100644
--- a/backend/tests/test_input_schema.py
+++ b/backend/tests/test_input_schema.py
@@ -170,9 +170,10 @@ async def test_sync_caches_schema_per_version(session, lfx_server, tmp_path):
assert "form_values" not in cached["json_schema"]["properties"]
-async def test_input_schema_endpoint(client, lfx_server):
+async def test_input_schema_endpoint(client, lfx_server, login):
assert (await client.get(f"/api/flows/{uuid4()}/input-schema")).status_code == 404
+ await login()
doc = json.loads((FIXTURES / "intake_form.json").read_text())
created = await client.post("/api/flows", json={"name": "schema-e2e"})
flow_id = created.json()["id"]
diff --git a/backend/tests/test_sync.py b/backend/tests/test_sync.py
index 1cb0796..bbff7ca 100644
--- a/backend/tests/test_sync.py
+++ b/backend/tests/test_sync.py
@@ -118,7 +118,8 @@ async def test_sync_rename_updates_definition_name(session, lfx_server, tmp_path
assert definition.name == "umbenannt"
-async def test_sync_endpoint_returns_report(client, lfx_server, tmp_path, monkeypatch):
+async def test_sync_endpoint_returns_report(client, lfx_server, tmp_path, monkeypatch, login):
+ await login()
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
_seed_dir(tmp_path)
response = await client.post("/api/flows/sync")
diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx
index fbbb7bb..b32fcf4 100644
--- a/frontend/src/App.tsx
+++ b/frontend/src/App.tsx
@@ -1,4 +1,4 @@
-import { Link, Route, Routes, useNavigate } from "react-router-dom";
+import { Link, Navigate, Route, Routes, useNavigate } from "react-router-dom";
import {
Anchor,
AppShell,
@@ -10,6 +10,7 @@ import {
Text,
Title,
} from "@mantine/core";
+import type { ReactNode } from "react";
import { AuthProvider, useAuth } from "./auth";
import SchemeToggle from "./components/SchemeToggle";
import StartPage from "./pages/StartPage";
@@ -66,6 +67,21 @@ function Nav() {
);
}
+function RequireAuth({ children }: { children: ReactNode }) {
+ const { me, loading } = useAuth();
+ if (loading) {
+ return (
+
+
+
+ );
+ }
+ // Runs, run history and approvals are login-gated server-side;
+ // anonymous visitors go to the magic-link login.
+ if (!me) return ;
+ return <>{children};
+}
+
function Shell() {
const { loading } = useAuth();
@@ -103,9 +119,30 @@ function Shell() {
} />
} />
} />
- } />
- } />
- } />
+
+
+
+ }
+ />
+
+
+
+ }
+ />
+
+
+
+ }
+ />
} />
)}
diff --git a/frontend/src/api.ts b/frontend/src/api.ts
index 3746cd6..f0af0f7 100644
--- a/frontend/src/api.ts
+++ b/frontend/src/api.ts
@@ -153,15 +153,11 @@ export const api = {
req(`/api/flows/${flowId}/input-schema`),
listRuns: (flowId: string) => req(`/api/flows/${flowId}/runs`),
listApprovals: () => req("/api/approvals"),
- decide: (
- taskId: string,
- decision: "approve" | "reject",
- approver: string,
- note: string,
- ) =>
+ decide: (taskId: string, decision: "approve" | "reject", note: string) =>
req(`/api/approvals/${taskId}/decision`, {
method: "POST",
- body: JSON.stringify({ decision, approver, note }),
+ // Identity comes from the session; the server records the user.
+ body: JSON.stringify({ decision, note }),
}),
importFlow: (langflowFlowId: string, name?: string) =>
req<{ definition: FlowDefinition; version: FlowVersion }>(
diff --git a/frontend/src/components/ApprovalsTab.tsx b/frontend/src/components/ApprovalsTab.tsx
index ac525d0..0412f32 100644
--- a/frontend/src/components/ApprovalsTab.tsx
+++ b/frontend/src/components/ApprovalsTab.tsx
@@ -12,11 +12,12 @@ import {
} from "@mantine/core";
import { IconAlertCircle, IconCheck, IconRefresh, IconX } from "@tabler/icons-react";
import { ApiError, api, type ApprovalTask } from "../api";
+import { useAuth } from "../auth";
import StatusBadge from "./StatusBadge";
export default function ApprovalsTab() {
+ const { me } = useAuth();
const [tasks, setTasks] = useState([]);
- const [approver, setApprover] = useState("");
const [note, setNote] = useState("");
const [error, setError] = useState("");
const [busyId, setBusyId] = useState("");
@@ -37,7 +38,7 @@ export default function ApprovalsTab() {
setBusyId(taskId);
setError("");
try {
- const result = await api.decide(taskId, decision, approver, note);
+ const result = await api.decide(taskId, decision, note);
setTasks((prev) =>
prev.map((t) => (t.id === taskId ? result.task : t)),
);
@@ -61,12 +62,11 @@ export default function ApprovalsTab() {
{error}
)}
+
+ Angemeldet als {me?.email ?? "–"} — Entscheidungen werden diesem
+ Konto zugeordnet.
+
- setApprover(e.target.value)}
- />
([]);
const [flowId, setFlowId] = useState("");
const [schema, setSchema] = useState(null);
@@ -86,11 +84,6 @@ export default function WorkflowsPage() {
Workflows
- {!me && (
-
- Sie sind nicht angemeldet — Läufe nutzen den Standard-Mandanten.
-
- )}
{error && (
} mb="md">
{error}