Make changing engine-run code superuser-only

Writing node source, installing modules and any flow save that adds,
retypes or repoints a code node (Function, Command), edits a Command
node's settings or changes a node's resources.env now answer 403 for
everyone but a superuser. MCP tools and the assistant reach the same
routes, so they inherit the gate. Node types report runs_code.

The editor shows code read-only with a note, disables a Command node's
settings, leaves code nodes and reusable nodes out of the palette and
paste, and the Modules page offers no install.

backend/fluksio/api/deps.py

@@ -444,3 +444,19 @@ def get_current_active_superuser(current_user: CurrentUser) -> User:
             status_code=403, detail="The user doesn't have enough privileges"
         )
     return current_user
+
+
+#: Why a non-superuser is refused a change to what the engine executes.
+CODE_IS_SUPERUSERS = "Only a superuser can change code the engine runs"
+
+
+def get_code_author(current_user: CurrentUser) -> User:
+    """A superuser, the only account that may change code the engine runs.
+
+    Node code, a command, an installed package: each runs as the engine, with
+    its signing key and every secret in reach, so whoever writes one holds
+    the instance whatever their account says.
+    """
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail=CODE_IS_SUPERUSERS)
+    return current_user

backend/fluksio/api/routes/flows.py

@@ -19,16 +19,18 @@ from pydantic import BaseModel
 from sqlmodel import Session, col, select
 
 from fluksio.api.deps import (
+    CODE_IS_SUPERUSERS,
     CurrentUser,
     FlowControllerDep,
     SessionDep,
     decode_token,
+    get_code_author,
     get_current_user,
     user_from_token,
 )
 from fluksio.core.db import engine
 from fluksio.flow.artifacts import is_reference
-from fluksio.flow.controller import FlowController
+from fluksio.flow.controller import NODE_TYPES, FlowController
 from fluksio.flow.dashboards import DashboardStore
 from fluksio.flow.events import event_bus
 from fluksio.flow.messages import flow_of, qualify
@@ -46,6 +48,7 @@ from fluksio.flow.schemas import (
     LibraryNode,
     MessageHistory,
     MessageValue,
+    NodeDef,
     NodeSource,
     NodeStatusPublic,
     NodeTypeInfo,
@@ -377,11 +380,56 @@ def _check_flavors(definition: FlowDef) -> None:
         )
 
 
+def _worker_env(node: NodeDef | None) -> dict[str, str]:
+    return node.resources.env if node is not None and node.resources else {}
+
+
+def _changes_code(node: NodeDef, old: NodeDef | None) -> bool:
+    """Whether saving *node* over *old* changes code the engine would run.
+
+    A function node's body has its own route, gated there. What the document
+    carries is which nodes run code at all, which shared source each runs, the
+    command in an exec node's settings, and the environment a node's worker
+    starts with — where `PYTHONPATH` alone is enough to import a file of
+    somebody's choosing.
+    """
+    if _worker_env(old) != _worker_env(node):
+        return True
+    spec = NODE_TYPES.get(node.type)
+    if spec is None or not spec.runs_code:
+        return False
+    return (
+        old is None
+        or old.type != node.type
+        or old.source_ref != node.source_ref
+        or (not spec.has_source and old.params != node.params)
+    )
+
+
+def _refuse_code_changes(
+    controller: FlowController, name: str, definition: FlowDef
+) -> None:
+    """Refuse a non-superuser a change to code this flow would run.
+
+    Wiring, titles, a function node's settings and every other node stay
+    anyone's.
+    """
+    before = _read_flow(controller, name) if controller.store.exists(name) else None
+    previous = {node.id: node for node in before.nodes} if before else {}
+    for node in definition.nodes:
+        if _changes_code(node, previous.get(node.id)):
+            raise HTTPException(
+                status_code=403,
+                detail=f"{CODE_IS_SUPERUSERS} (node '{node.id}')",
+            )
+
+
 @router.put("/{name}", response_model=FlowDetail)
 async def save_flow(
     name: str,
     definition: FlowDef,
     controller: FlowControllerDep,
+    user: CurrentUser,
 ) -> Any:
     """Save unpublished changes to a flow.
 
@@ -400,6 +448,8 @@ async def save_flow(
         raise HTTPException(status_code=400, detail="Node names must be unique")
 
     _check_flavors(definition)
+    if not user.is_superuser:
+        _refuse_code_changes(controller, name, definition)
 
     try:
         stored = await run_in_threadpool(
@@ -548,7 +598,11 @@ def read_node_source(
     )
 
 
-@router.put("/{name}/nodes/{node_id}/source", response_model=NodeStatusPublic)
+@router.put(
+    "/{name}/nodes/{node_id}/source",
+    response_model=NodeStatusPublic,
+    dependencies=[Depends(get_code_author)],
+)
 async def save_node_source(
     name: str,
     node_id: str,

backend/fluksio/api/routes/modules.py

@@ -16,6 +16,7 @@ from fluksio.api.deps import (
     CurrentUser,
     FlowControllerDep,
     WorkerPoolDep,
+    get_code_author,
     get_current_user,
 )
 from fluksio.flow import modules
@@ -34,7 +35,9 @@ async def read_modules(controller: FlowControllerDep) -> Any:
     return await run_in_threadpool(modules.info, controller.store)
 
 
-@router.post("/apply", response_model=ApplyResult)
+@router.post(
+    "/apply", response_model=ApplyResult, dependencies=[Depends(get_code_author)]
+)
 async def apply_modules(
     body: ApplyRequest,
     controller: FlowControllerDep,
@@ -43,6 +46,8 @@ async def apply_modules(
 ) -> Any:
     """Install exactly these requirements, then hand them to the workers.
 
+    A superuser's: a package runs its own code on install and on import.
+
     A manifest that does not resolve changes nothing: the venv is left as it
     was and the stored manifest is only written once the install succeeded.
 

backend/fluksio/flow/controller.py

@@ -303,6 +303,9 @@ class NodeType:
     # Constructor signatures differ per node type.
     cls: Any
     has_source: bool = False
+    #: Runs code its definition supplies — a source file, or a command among
+    #: its settings — with everything the engine can reach.
+    runs_code: bool = False
     free_params: bool = False
     params_schema: dict[str, Any] = field(default_factory=dict)
     #: Which installed package supplied this type, for the ones that are not
@@ -321,6 +324,7 @@ NODE_TYPES: dict[str, NodeType] = {
         description="Your own Python code, run on every incoming message.",
         cls=Node,
         has_source=True,
+        runs_code=True,
         free_params=True,
     ),
     "mqtt": NodeType(
@@ -399,6 +403,7 @@ NODE_TYPES: dict[str, NodeType] = {
         title="Command",
         description="Run a command in the engine's container and read its output.",
         cls=ExecNode,
+        runs_code=True,
         params_schema=_schema_of(ExecNode),
     ),
     "file": NodeType(
@@ -424,6 +429,7 @@ def node_type_info() -> list[NodeTypeInfo]:
             description=spec.description,
             params_schema=spec.params_schema,
             has_source=spec.has_source,
+            runs_code=spec.runs_code,
             free_params=spec.free_params,
             plugin=spec.plugin,
         )

backend/fluksio/flow/schemas.py

@@ -476,6 +476,10 @@ class NodeTypeInfo(BaseModel):
     description: str
     params_schema: dict[str, Any] = Field(default_factory=dict)
     has_source: bool = False
+    #: Whether a node of this type runs code its definition supplies: its
+    #: source, or a command among its settings. Only a superuser may add one
+    #: or change that code.
+    runs_code: bool = False
     #: Whether this type takes settings beyond the ones its schema declares.
     #: A function node's settings are its author's to name, and reach `process`
     #: as keyword arguments beside its ports.

backend/tests/api/routes/test_code_rights.py

@@ -0,0 +1,138 @@
+"""Code the engine runs is a superuser's to change; everything else is not."""
+
+from collections.abc import Callable
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+from fluksio.api.deps import CODE_IS_SUPERUSERS
+from fluksio.api.routes import modules as modules_route
+from fluksio.core.config import settings
+
+PREFIX = f"{settings.API_V1_STR}/flows"
+FLOW = "code_rights"
+
+
+def a_flow(version: int, **changes: Any) -> dict[str, Any]:
+    nodes: dict[str, dict[str, Any]] = {
+        "fn": {
+            "id": "fn",
+            "type": "python",
+            "params": {"gain": 1},
+            "provides": [{"name": "reading", "dtype": "float"}],
+        },
+        "cmd": {"id": "cmd", "type": "exec", "params": {"command": "echo hi"}},
+        "hold": {"id": "hold", "type": "delay", "params": {"delay": 1}},
+    }
+    for node_id, patch in changes.items():
+        nodes[node_id] = {**nodes.get(node_id, {"id": node_id}), **patch}
+    return {"name": FLOW, "version": version, "nodes": list(nodes.values())}
+
+
+@pytest.fixture
+def save(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> Callable[..., Any]:
+    """Seed the flow as a superuser, then save a change as someone else."""
+    client.delete(f"{PREFIX}/{FLOW}", headers=superuser_token_headers)
+    seeded = client.put(
+        f"{PREFIX}/{FLOW}", headers=superuser_token_headers, json=a_flow(0)
+    )
+    assert seeded.status_code == 200, seeded.text
+    version = seeded.json()["definition"]["version"]
+
+    def put(headers: dict[str, str], **changes: Any) -> Any:
+        return client.put(
+            f"{PREFIX}/{FLOW}", headers=headers, json=a_flow(version, **changes)
+        )
+
+    return put
+
+
+@pytest.mark.parametrize(
+    "changes",
+    [
+        {"new": {"type": "python"}},
+        {"new": {"type": "exec", "params": {"command": "id"}}},
+        {"cmd": {"params": {"command": "cat /proc/self/environ"}}},
+        {"hold": {"type": "exec", "params": {"command": "id"}}},
+        {"fn": {"source_ref": "someone_elses"}},
+        {"fn": {"resources": {"env": {"PYTHONPATH": "/data/files"}}}},
+    ],
+    ids=["add-function", "add-exec", "exec-command", "retype", "repoint", "env"],
+)
+def test_a_user_cannot_change_code_in_a_flow(
+    save: Callable[..., Any],
+    normal_user_token_headers: dict[str, str],
+    superuser_token_headers: dict[str, str],
+    changes: dict[str, Any],
+) -> None:
+    refused = save(normal_user_token_headers, **changes)
+    assert refused.status_code == 403
+    assert refused.json()["detail"].startswith(CODE_IS_SUPERUSERS)
+    assert save(superuser_token_headers, **changes).status_code == 200
+
+
+def test_a_user_keeps_everything_but_the_code(
+    save: Callable[..., Any], normal_user_token_headers: dict[str, str]
+) -> None:
+    saved = save(
+        normal_user_token_headers,
+        fn={
+            "title": "Renamed",
+            "params": {"gain": 2},
+            "requires": [{"name": "trigger", "dtype": "float"}],
+        },
+        hold={"params": {"delay": 5}},
+        late={"type": "delay", "params": {"delay": 2}},
+    )
+    assert saved.status_code == 200, saved.text
+
+
+@pytest.mark.usefixtures("save")
+def test_node_source_is_a_superusers_to_write(
+    client: TestClient,
+    normal_user_token_headers: dict[str, str],
+    superuser_token_headers: dict[str, str],
+) -> None:
+    url = f"{PREFIX}/{FLOW}/nodes/fn/source"
+    body = {"code": "def process():\n    return {'reading': 1.0}\n"}
+
+    refused = client.put(url, headers=normal_user_token_headers, json=body)
+    assert refused.status_code == 403
+    assert refused.json()["detail"] == CODE_IS_SUPERUSERS
+    # Reading it is still anyone's.
+    assert client.get(url, headers=normal_user_token_headers).status_code == 200
+
+    saved = client.put(url, headers=superuser_token_headers, json=body)
+    assert saved.status_code == 200, saved.text
+
+
+def test_installing_modules_is_a_superusers(
+    client: TestClient,
+    monkeypatch: pytest.MonkeyPatch,
+    normal_user_token_headers: dict[str, str],
+    superuser_token_headers: dict[str, str],
+) -> None:
+    installs: list[str] = []
+
+    def sync(requirements: str) -> tuple[bool, str]:
+        installs.append(requirements)
+        return True, ""
+
+    async def retire(*_: Any) -> None:
+        return None
+
+    monkeypatch.setattr(modules_route.modules, "sync", sync)
+    monkeypatch.setattr(modules_route, "_retire", retire)
+    url = f"{settings.API_V1_STR}/modules/apply"
+    body = {"requirements": ""}
+
+    refused = client.post(url, headers=normal_user_token_headers, json=body)
+    assert refused.status_code == 403
+    assert refused.json()["detail"] == CODE_IS_SUPERUSERS
+    assert installs == []
+
+    assert client.post(url, headers=superuser_token_headers, json=body).is_success
+    assert installs == [""]

backend/tests/mcp/test_mcp_http.py

@@ -1,6 +1,7 @@
 """The MCP endpoint: who it lets in, and what happens once they are in."""
 
 import asyncio
+import json
 import uuid
 from collections.abc import Awaitable, Callable, Generator
 from datetime import timedelta
@@ -11,6 +12,7 @@ import pytest
 from fastapi.testclient import TestClient
 from sqlmodel import Session
 
+from fluksio.api.deps import CODE_IS_SUPERUSERS
 from fluksio.core import security
 from fluksio.core.config import settings
 from fluksio.main import app
@@ -170,3 +172,31 @@ def test_revoking_an_agent_stops_the_token_it_already_holds(
     before, after = over_mcp(block)
     assert before.status_code == 200
     assert after.status_code == 401
+
+
+def test_an_agent_cannot_write_code_its_person_may_not(
+    over_mcp,
+    client: TestClient,
+    normal_user_token_headers: dict[str, str],
+    agent: uuid.UUID,
+) -> None:
+    """The tools call the same routes, so an agent inherits the code gate."""
+    me = client.get(
+        f"{settings.API_V1_STR}/users/me", headers=normal_user_token_headers
+    ).json()
+    assert me["is_superuser"] is False
+    token = mcp_token(uuid.UUID(me["id"]), agent)
+
+    async def block(http: httpx.AsyncClient) -> httpx.Response:
+        return await http.post(
+            "/mcp",
+            **rpc(
+                token,
+                "tools/call",
+                name="save_node_source",
+                arguments={"name": "any", "node_id": "fn", "code": "import os"},
+            ),
+        )
+
+    answer = json.loads(over_mcp(block).json()["result"]["content"][0]["text"])
+    assert answer == {"error": CODE_IS_SUPERUSERS, "status": 403}