diff --git a/tests/integration/test_repository_safety.py b/tests/integration/test_repository_safety.py index 21f67c9..b141e18 100644 --- a/tests/integration/test_repository_safety.py +++ b/tests/integration/test_repository_safety.py @@ -1,7 +1,9 @@ from __future__ import annotations +import json +import shutil from pathlib import Path -from unittest.mock import patch +from unittest.mock import AsyncMock, patch import httpx import pytest @@ -9,22 +11,26 @@ from backup_tool.api.app import create_app from backup_tool.config import Settings from backup_tool.repository import RepositoryError, initialize, inspect_repository +PASSWORD = "a secure password" -def settings_for(tmp_path: Path) -> Settings: + +def settings_for(tmp_path: Path, **overrides: object) -> Settings: key = tmp_path / "key" key.write_bytes(b"x" * 32) key.chmod(0o600) root = tmp_path / "repos" root.mkdir() - return Settings( - repository_roots=(root,), - local_source_roots=(tmp_path,), - restore_roots=(tmp_path,), - master_key_file=key, - data_dir=tmp_path, - database_url=f"sqlite+aiosqlite:///{tmp_path / 'db.sqlite'}", - min_free_bytes=1, - ) + values: dict[str, object] = { + "repository_roots": (root,), + "local_source_roots": (tmp_path,), + "restore_roots": (tmp_path,), + "master_key_file": key, + "data_dir": tmp_path, + "database_url": f"sqlite+aiosqlite:///{tmp_path / 'db.sqlite'}", + "min_free_bytes": 1, + } + values.update(overrides) + return Settings(**values) def test_partial_initialization_is_removed_on_publish_failure(tmp_path: Path) -> None: @@ -38,6 +44,41 @@ def test_partial_initialization_is_removed_on_publish_failure(tmp_path: Path) -> assert not list(settings.repository_roots[0].glob(".main.staging-*")) +@pytest.mark.parametrize("relative_path", ["/absolute", "../escape"]) +def test_initialization_rejects_absolute_and_traversal_paths( + tmp_path: Path, relative_path: str +) -> None: + with pytest.raises(RepositoryError, match="path"): + initialize(settings_for(tmp_path), relative_path, "none", "none") + + +def test_initialization_rejects_symlinked_destination_component(tmp_path: Path) -> None: + settings = settings_for(tmp_path) + outside = tmp_path / "outside" + outside.mkdir() + (settings.repository_roots[0] / "linked").symlink_to(outside, target_is_directory=True) + with pytest.raises(RepositoryError, match="symlink"): + initialize(settings, "linked/child", "none", "none") + + +@pytest.mark.parametrize( + ("minimum", "usage"), + [ + ("min_free_bytes", shutil._ntuple_diskusage(total=100, used=99, free=1)), + ("min_free_percent", shutil._ntuple_diskusage(total=100, used=96, free=4)), + ], +) +def test_initialization_rejects_insufficient_capacity( + tmp_path: Path, minimum: str, usage: shutil._ntuple_diskusage +) -> None: + settings = settings_for(tmp_path, **{minimum: 5}) + with ( + patch("backup_tool.repository.shutil.disk_usage", return_value=usage), + pytest.raises(RepositoryError, match="capacity"), + ): + initialize(settings, "main", "none", "none") + + def test_repository_inspection_rejects_noncanonical_metadata(tmp_path: Path) -> None: settings = settings_for(tmp_path) result = initialize(settings, "main", "none", "none") @@ -48,6 +89,55 @@ def test_repository_inspection_rejects_noncanonical_metadata(tmp_path: Path) -> inspect_repository(settings, result.root) +def test_repository_inspection_rejects_swapped_symlink_and_out_of_allowlist( + tmp_path: Path, +) -> None: + settings = settings_for(tmp_path) + result = initialize(settings, "main", "none", "none") + outside = tmp_path / "outside" + outside.mkdir() + (outside / "repository.json").write_text(json.dumps({"not": "a repository"}), encoding="utf-8") + result.root.rename(settings.repository_roots[0] / "real-main") + result.root.symlink_to(outside, target_is_directory=True) + with pytest.raises(RepositoryError, match="escapes"): + inspect_repository(settings, result.root) + with pytest.raises(RepositoryError, match="escapes"): + inspect_repository(settings, outside) + + +@pytest.mark.asyncio +async def test_database_failure_removes_published_repository(tmp_path: Path) -> None: + settings = settings_for(tmp_path) + app = create_app(settings) + from backup_tool.db.models import Base + + async with app.state.engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="https://test" + ) as client: + assert ( + await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD}) + ).status_code == 201 + csrf = client.cookies["backup_tool_csrf"] + with patch( + "backup_tool.api.app.audit", new=AsyncMock(side_effect=RuntimeError("db failure")) + ): + response = await client.post( + "/api/v2/repositories", + json={ + "name": "main", + "relative_path": "main", + "compression": "none", + "encryption": "none", + }, + headers={"X-CSRF-Token": csrf}, + ) + assert response.status_code == 409 + assert not (settings.repository_roots[0] / "main").exists() + await app.state.engine.dispose() + + @pytest.mark.asyncio async def test_admin_can_list_and_inspect_repositories(tmp_path: Path) -> None: settings = settings_for(tmp_path) @@ -60,10 +150,7 @@ async def test_admin_can_list_and_inspect_repositories(tmp_path: Path) -> None: transport=httpx.ASGITransport(app=app), base_url="https://test" ) as client: assert ( - await client.post( - "/api/v2/setup", - json={"username": "admin", "password": "a secure password"}, - ) + await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD}) ).status_code == 201 csrf = client.cookies["backup_tool_csrf"] created = await client.post(