fix(v2): enforce repository safety failures

This commit is contained in:
2026-07-27 21:16:44 +02:00
parent cccd638be2
commit 14b4347806
2 changed files with 56 additions and 19 deletions
+12 -3
View File
@@ -29,10 +29,19 @@ def blob_digest(content: bytes) -> str:
def _contained(root: Path, relative_path: str) -> Path: def _contained(root: Path, relative_path: str) -> Path:
candidate = (root / relative_path).resolve() relative = Path(relative_path)
if not relative_path or Path(relative_path).is_absolute() or candidate == root: if not relative_path or relative.is_absolute():
raise RepositoryError("repository path must be a non-empty relative child") raise RepositoryError("repository path must be a non-empty relative child")
if root not in candidate.parents: lexical = root
for part in relative.parts:
lexical /= part
if lexical.is_symlink():
raise RepositoryError("repository path contains a symlink")
candidate = lexical.resolve()
resolved_root = root.resolve()
if candidate == resolved_root:
raise RepositoryError("repository path must be a non-empty relative child")
if resolved_root not in candidate.parents:
raise RepositoryError("repository path escapes configured roots") raise RepositoryError("repository path escapes configured roots")
return candidate return candidate
+44 -16
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
import json import json
import shutil import shutil
from pathlib import Path from pathlib import Path
from unittest.mock import AsyncMock, patch from unittest.mock import patch
import httpx import httpx
import pytest import pytest
@@ -30,7 +30,7 @@ def settings_for(tmp_path: Path, **overrides: object) -> Settings:
"min_free_bytes": 1, "min_free_bytes": 1,
} }
values.update(overrides) values.update(overrides)
return Settings(**values) return Settings.model_validate(values)
def test_partial_initialization_is_removed_on_publish_failure(tmp_path: Path) -> None: def test_partial_initialization_is_removed_on_publish_failure(tmp_path: Path) -> None:
@@ -79,6 +79,26 @@ def test_initialization_rejects_insufficient_capacity(
initialize(settings, "main", "none", "none") initialize(settings, "main", "none", "none")
def test_initialization_emits_complete_repository_protocol_metadata(tmp_path: Path) -> None:
result = initialize(settings_for(tmp_path), "main", "none", "none")
try:
payload = json.loads((result.root / "repository.json").read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
raise AssertionError("repository metadata was not readable JSON") from error
assert set(payload) == {
"repository_id",
"format_version",
"digest_algorithm",
"compression",
"encryption",
"created_at",
}
assert payload["repository_id"][14] == "7"
assert payload["digest_algorithm"] == "sha256"
assert payload["encryption"] == {"mode": "none", "key_id": None}
assert payload["created_at"].endswith("Z")
def test_repository_inspection_rejects_noncanonical_metadata(tmp_path: Path) -> None: def test_repository_inspection_rejects_noncanonical_metadata(tmp_path: Path) -> None:
settings = settings_for(tmp_path) settings = settings_for(tmp_path)
result = initialize(settings, "main", "none", "none") result = initialize(settings, "main", "none", "none")
@@ -120,21 +140,29 @@ async def test_database_failure_removes_published_repository(tmp_path: Path) ->
await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD}) await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD})
).status_code == 201 ).status_code == 201
csrf = client.cookies["backup_tool_csrf"] csrf = client.cookies["backup_tool_csrf"]
with patch( first = await client.post(
"backup_tool.api.app.audit", new=AsyncMock(side_effect=RuntimeError("db failure")) "/api/v2/repositories",
): json={
response = await client.post( "name": "main",
"/api/v2/repositories", "relative_path": "first",
json={ "compression": "none",
"name": "main", "encryption": "none",
"relative_path": "main", },
"compression": "none", headers={"X-CSRF-Token": csrf},
"encryption": "none", )
}, assert first.status_code == 201
headers={"X-CSRF-Token": csrf}, response = await client.post(
) "/api/v2/repositories",
json={
"name": "main",
"relative_path": "second",
"compression": "none",
"encryption": "none",
},
headers={"X-CSRF-Token": csrf},
)
assert response.status_code == 409 assert response.status_code == 409
assert not (settings.repository_roots[0] / "main").exists() assert not (settings.repository_roots[0] / "second").exists()
await app.state.engine.dispose() await app.state.engine.dispose()