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.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013GAeg4Jo6xjZzjwy9QDWMQ
This commit is contained in:
@@ -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()
|
||||
Reference in New Issue
Block a user