menzeljandClaude Opus 5
60a7ccff93
The repo had no tests, no lint config, and a CI that went straight from push
to docker push. That is the reason F1 could ship: authorization lives in the
routers, each of 171 routes independently picks require_admin or
get_current_user, and nothing checked the choice was right.
670 tests, no Docker daemon needed. The app is driven through TestClient
without entering it as a context manager, which skips the lifespan — no
background loops, no socket — and conftest points DATA_DIR/STACKS_DIR at a
temp directory before anything is imported.
test_route_authorization.py is the load-bearing one. Rather than 171 implied
decisions it states the policy once — every route requires admin unless it is
listed in USER_READABLE or PUBLIC — and fails on any route that disagrees. A
new route defaults to admin, which is the safe direction; what it catches is a
route written with get_current_user that nobody weighed against "can this
return a credential". Writing the allowlist meant auditing all 53 user-readable
routes, which turned up one more leak: GET /api/templates/{id} returns a
template's env, and "save stack as template" snapshots the stack's real .env
into it. Now admin-only; the listing stays open.
test_agent_authorization.py pins the same invariant on the agent, where the
whole access model is one shared token declared per route and a single
forgotten Depends(verify_token) would hand over the host.
Both were checked by reintroducing the bug: re-opening /api/files/read fails
three tests with actionable messages, dropping a token guard fails two.
test_bundled_templates.py covers the 83 templates — parse, image per service,
.env.example in sync with what compose reads, every bind-mounted file actually
shipped, and no working default password. It found one on its first run:
authentik shipped PG_PASS=change-me and AUTHENTIK_SECRET_KEY=change-me against
a compose that marks both required, so the stack would have come up with a
known password instead of refusing to start. Fixed.
The rest ports the ad-hoc harnesses from 0.44.0 into permanent tests (crypto
round-trip incl. plaintext passthrough and key-loss handling, the browse
sandbox) and covers compose_service's slug/status/file handling and
secret_service's name validation.
ruff is configured as a floor, not a style bar: F, E9 and B only. Import
sorting is deliberately out — it is style, and enabling it would rewrite the
imports of nine files that have nothing else wrong. The 12 findings it did have
are fixed here (unused imports, an unused local, four raise-without-from that
were swallowing exception context).
CI now runs check (ruff, pytest, tsc) and only builds if it passes.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Dk43rmEeRfYi5wsLDbmfyG
162 lines
5.7 KiB
Python
162 lines
5.7 KiB
Python
"""Stack storage and status derivation.
|
|
|
|
``compose_service`` is where the file-is-the-truth model lives: slugs become
|
|
directory names, directory names become compose project names, and container
|
|
states become the one status the UI shows. All three are pure enough to test
|
|
without a Docker daemon.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def svc():
|
|
from services import compose_service
|
|
|
|
return compose_service
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Slugs — these become directory names and compose project names
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"name,expected",
|
|
[
|
|
("Jellyfin", "jellyfin"),
|
|
("My Media Server", "my-media-server"),
|
|
("Paperless-NGX", "paperless-ngx"),
|
|
(" spaces ", "spaces"),
|
|
("Wiki.js", "wiki-js"),
|
|
("a---b", "a-b"),
|
|
("--leading-and-trailing--", "leading-and-trailing"),
|
|
("Ümlaut Stack", "mlaut-stack"),
|
|
],
|
|
)
|
|
def test_slugify(svc, name, expected):
|
|
assert svc.slugify(name) == expected
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["", " ", "///", "..."])
|
|
def test_slugify_never_returns_an_empty_or_traversing_slug(svc, name):
|
|
"""The slug is joined onto STACKS_DIR, so an empty or dotted result would
|
|
point the stack directory at the root itself."""
|
|
slug = svc.slugify(name)
|
|
assert slug
|
|
assert slug not in (".", "..")
|
|
assert "/" not in slug
|
|
|
|
|
|
def test_stack_dir_stays_under_the_root(svc):
|
|
root = svc.stacks_root()
|
|
assert svc.stack_dir("jellyfin").startswith(root + os.sep)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Status derivation
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"states,expected",
|
|
[
|
|
([], "stopped"),
|
|
(["running"], "running"),
|
|
(["running", "running"], "running"),
|
|
(["running", "exited"], "partial"),
|
|
(["exited", "exited"], "stopped"),
|
|
(["created"], "stopped"),
|
|
(["running", "dead"], "error"),
|
|
(["dead"], "error"),
|
|
],
|
|
)
|
|
def test_status_from_states(svc, states, expected):
|
|
assert svc._status_from_states(states) == expected
|
|
|
|
|
|
def test_busy_stacks_report_as_updating(svc):
|
|
"""The busy flag has to win over the container states, or a stack shows
|
|
'stopped' for the moment between `down` and `up` during an update."""
|
|
svc.mark_busy("busy-stack")
|
|
try:
|
|
assert svc.compute_status("busy-stack", containers=[]) == "updating"
|
|
finally:
|
|
svc.clear_busy("busy-stack")
|
|
assert svc.compute_status("busy-stack", containers=[]) == "stopped"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Reading and writing stack files
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_write_compose_keeps_a_backup_of_the_previous_version(svc, tmp_path):
|
|
"""Every save writes a .bak — the raw material for a rollback feature."""
|
|
stack_id = "backup-check"
|
|
svc.write_compose(stack_id, "services:\n a:\n image: alpine\n", override=str(tmp_path))
|
|
svc.write_compose(stack_id, "services:\n b:\n image: nginx\n", override=str(tmp_path))
|
|
|
|
directory = tmp_path / stack_id
|
|
assert "image: nginx" in (directory / "compose.yaml").read_text()
|
|
assert "image: alpine" in (directory / "compose.yaml.bak").read_text()
|
|
|
|
|
|
def test_reading_a_missing_stack_returns_empty_not_an_error(svc, tmp_path):
|
|
assert svc.read_compose("does-not-exist", override=str(tmp_path)) == ""
|
|
assert svc.read_env("does-not-exist", override=str(tmp_path)) == ""
|
|
|
|
|
|
def test_discover_stacks_finds_only_directories_with_a_compose_file(svc, tmp_path):
|
|
(tmp_path / "real").mkdir()
|
|
(tmp_path / "real" / "compose.yaml").write_text("services: {}\n")
|
|
(tmp_path / "legacy").mkdir()
|
|
(tmp_path / "legacy" / "docker-compose.yml").write_text("services: {}\n")
|
|
(tmp_path / "not-a-stack").mkdir()
|
|
(tmp_path / "not-a-stack" / "readme.txt").write_text("hi\n")
|
|
|
|
assert svc.discover_stacks(override=str(tmp_path)) == ["legacy", "real"]
|
|
|
|
|
|
def test_clone_refuses_to_overwrite_an_existing_stack(svc, tmp_path):
|
|
svc.write_compose("source", "services: {}\n", override=str(tmp_path))
|
|
svc.write_compose("target", "services: {}\n", override=str(tmp_path))
|
|
with pytest.raises(svc.StackFileError):
|
|
svc.clone_stack_files("source", "target", override=str(tmp_path))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Per-stack secrets
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["../escape", "a/b", ".hidden", "", "na me"])
|
|
def test_secret_names_reject_traversal_and_hidden_files(name):
|
|
"""Secret names become filenames inside the stack's .secrets directory."""
|
|
from services import secret_service
|
|
|
|
with pytest.raises(secret_service.SecretError):
|
|
secret_service._check_name(name)
|
|
|
|
|
|
def test_secret_files_are_written_owner_only(tmp_path):
|
|
from services import secret_service
|
|
|
|
secret_service.write_secret("s", "secret", "db_password", "hunter2", override=str(tmp_path))
|
|
path = tmp_path / "s" / ".secrets" / "db_password"
|
|
assert path.read_text() == "hunter2"
|
|
assert oct(path.stat().st_mode)[-3:] == "600"
|
|
assert oct(path.parent.stat().st_mode)[-3:] == "700"
|
|
|
|
|
|
def test_listing_secrets_never_returns_their_content(tmp_path):
|
|
from services import secret_service
|
|
|
|
secret_service.write_secret("s2", "secret", "token", "top-secret", override=str(tmp_path))
|
|
listed = secret_service.list_secrets("s2", "secret", override=str(tmp_path))
|
|
assert [item["name"] for item in listed] == ["token"]
|
|
assert "top-secret" not in str(listed)
|