Refuse names that are not names at the store boundary

DELETE /flows/%2e%2e reached FlowStore.delete_flow(".."), which removed
the data directory, and /flows/_lib removed every shared node. Every
path the flow and dashboard stores build from a flow, node, library or
dashboard name now goes through checked(), which raises InvalidName
(400) for anything _validate_name refuses. _validate_name uses
fullmatch, so a trailing newline is refused too.

backend/fluksio/flow/dashboards.py

@@ -26,7 +26,7 @@ from pydantic import BaseModel, Field, field_validator, model_validator
 
 from fluksio.flow.messages import DType, qualify
 from fluksio.flow.schemas import FlowDef, _validate_name
-from fluksio.flow.store import FlowStore, StaleVersion, _updated_at
+from fluksio.flow.store import FlowStore, StaleVersion, _updated_at, checked
 
 #: Sibling of the shared-node library, and likewise not a flow.
 DASHBOARD_DIR = "_dashboards"
@@ -612,10 +612,10 @@ class DashboardStore:
         self._lock = threading.Lock()
 
     def _file(self, name: str) -> Path:
-        return self.root / name / "dashboard.json"
+        return self.root / checked(name) / "dashboard.json"
 
     def _draft_file(self, name: str) -> Path:
-        return self.root / name / "dashboard.draft.json"
+        return self.root / checked(name) / "dashboard.draft.json"
 
     @staticmethod
     def _dump(defn: DashboardDef) -> str:

backend/fluksio/flow/schemas.py

@@ -24,7 +24,8 @@ _SPAN_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400}
 
 
 def _validate_name(value: str) -> str:
-    if not NAME_PATTERN.match(value):
+    # fullmatch: `$` alone lets a trailing newline through.
+    if not NAME_PATTERN.fullmatch(value):
         raise ValueError(
             "Use lowercase letters, digits and underscores, starting with a letter"
         )

backend/fluksio/flow/store.py

@@ -23,7 +23,7 @@ from collections.abc import Iterable
 from datetime import UTC, datetime
 from pathlib import Path
 
-from fluksio.flow.schemas import FlowDef
+from fluksio.flow.schemas import FlowDef, _validate_name
 
 logger = logging.getLogger(__name__)
 
@@ -38,6 +38,33 @@ def process():
 '''
 
 
+class InvalidName(ValueError):
+    """A name that cannot be a file or directory of the store."""
+
+    def __init__(self, name: str) -> None:
+        super().__init__(name)
+        self.name = name
+
+    def __str__(self) -> str:
+        return (
+            f"'{self.name}' is not a valid name: use lowercase letters, digits "
+            "and underscores, starting with a letter"
+        )
+
+
+def checked(name: str) -> str:
+    """*name*, once it is certain to be one path segment inside the store.
+
+    Every path the stores build starts from a name the API was handed,
+    percent-decoded on the way in, so without this `..` or `_lib` reached the
+    filesystem as written: `delete_flow("..")` removed the data directory.
+    """
+    try:
+        return _validate_name(name)
+    except ValueError:
+        raise InvalidName(name) from None
+
+
 class FlowNotFound(KeyError):
     def __init__(self, name: str) -> None:
         super().__init__(name)
@@ -206,8 +233,11 @@ class FlowStore:
     # Paths
     # -------------------------------------------------------------------------
 
+    # Every path below starts in one of these three, and each name in it has
+    # been through `checked`.
+
     def _flow_dir(self, name: str) -> Path:
-        return self.root / name
+        return self.root / checked(name)
 
     def _flow_file(self, name: str) -> Path:
         return self._flow_dir(name) / "flow.json"
@@ -216,19 +246,19 @@ class FlowStore:
         return self._flow_dir(name) / "flow.draft.json"
 
     def _node_file(self, flow: str, node_id: str) -> Path:
-        return self._flow_dir(flow) / "nodes" / f"{node_id}.py"
+        return self._flow_dir(flow) / "nodes" / f"{checked(node_id)}.py"
 
     def _draft_nodes_dir(self, flow: str) -> Path:
         return self._flow_dir(flow) / "nodes.draft"
 
     def _draft_node_file(self, flow: str, node_id: str) -> Path:
-        return self._draft_nodes_dir(flow) / f"{node_id}.py"
+        return self._draft_nodes_dir(flow) / f"{checked(node_id)}.py"
 
     def _runtime_file(self, name: str) -> Path:
         return self._flow_dir(name) / "runtime.json"
 
     def _lib_file(self, name: str) -> Path:
-        return self.root / LIB_DIR / f"{name}.py"
+        return self.root / LIB_DIR / f"{checked(name)}.py"
 
     def head(self) -> str:
         """The commit the store is at, so a result can name the code it ran."""

backend/fluksio/main.py

@@ -49,7 +49,7 @@ from fluksio.flow.resources import (
 from fluksio.flow.runs import RUN_STATE_TTL, RunService, sweep_artifacts
 from fluksio.flow.secrets import init_secrets
 from fluksio.flow.state import MemoryState, RedisState, StateBackend
-from fluksio.flow.store import FlowStore
+from fluksio.flow.store import FlowStore, InvalidName
 from fluksio.flow.watchdog import LoopWatchdog
 from fluksio.flow.workers import PythonWorkerPool
 
@@ -436,6 +436,12 @@ if settings.all_cors_origins:
 app.include_router(api_router, prefix=settings.API_V1_STR)
 
 
+@app.exception_handler(InvalidName)
+async def invalid_name(request: Request, exc: InvalidName) -> JSONResponse:  # noqa: ARG001
+    """A flow, node, library or dashboard name no store will turn into a path."""
+    return JSONResponse(status_code=400, content={"detail": str(exc)})
+
+
 @app.exception_handler(RebuildBusy)
 async def rebuild_busy(request: Request, exc: RebuildBusy) -> JSONResponse:  # noqa: ARG001
     """Every route that deploys something answers a wedged rebuild the same way.

backend/tests/api/routes/test_path_names.py

@@ -0,0 +1,68 @@
+"""A name from a URL never becomes a path outside the one it names."""
+
+from pathlib import Path
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+from fluksio.core.config import settings
+from fluksio.flow import store as store_module
+from fluksio.flow.dashboards import DashboardStore
+from fluksio.flow.store import FlowStore, InvalidName
+
+API = settings.API_V1_STR
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/flows/%2e%2e",
+        "/flows/.",
+        "/flows/_lib",
+        "/flows/_dashboards",
+        "/dashboards/%2e%2e",
+        "/dashboards/_x",
+    ],
+)
+def test_a_delete_by_a_name_that_is_not_one_touches_nothing(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    monkeypatch: pytest.MonkeyPatch,
+    path: str,
+) -> None:
+    removed: list[Any] = []
+    monkeypatch.setattr(store_module.shutil, "rmtree", removed.append)
+    flows = settings.FLOWS_DIR
+    canaries = [flows.parent / "canary", flows / "_lib" / "canary.py"]
+    for canary in canaries:
+        canary.parent.mkdir(parents=True, exist_ok=True)
+        canary.write_text("still here")
+
+    try:
+        response = client.delete(f"{API}{path}", headers=superuser_token_headers)
+
+        assert 400 <= response.status_code < 500, response.text
+        assert removed == []
+        assert all(canary.read_text() == "still here" for canary in canaries)
+        assert flows.is_dir()
+    finally:
+        for canary in canaries:
+            canary.unlink(missing_ok=True)
+
+
+def test_the_stores_refuse_them_themselves(tmp_path: Path) -> None:
+    """The routes need not check: no store method builds a path from one."""
+    flows = FlowStore(tmp_path / "flows")
+    dashboards = DashboardStore(flows)
+    for call in (
+        lambda: flows.delete_flow(".."),
+        lambda: flows.read_flow("_lib"),
+        lambda: flows.rename_flow("..", "elsewhere"),
+        lambda: flows.write_node_source("demo", "../x", "code"),
+        lambda: flows.write_lib_source("..", "code"),
+        lambda: flows.exists("demo\n"),
+        lambda: dashboards.delete(".."),
+    ):
+        with pytest.raises(InvalidName):
+            call()