212 lines
7.7 KiB
Python
212 lines
7.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
import pytest
|
|
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, **overrides: object) -> Settings:
|
|
key = tmp_path / "key"
|
|
key.write_bytes(b"x" * 32)
|
|
key.chmod(0o600)
|
|
root = tmp_path / "repos"
|
|
root.mkdir()
|
|
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.model_validate(values)
|
|
|
|
|
|
def test_partial_initialization_is_removed_on_publish_failure(tmp_path: Path) -> None:
|
|
settings = settings_for(tmp_path)
|
|
with (
|
|
patch("backup_tool.repository.os.replace", side_effect=OSError("disk failure")),
|
|
pytest.raises(OSError),
|
|
):
|
|
initialize(settings, "main", "none", "none")
|
|
assert not (settings.repository_roots[0] / "main").exists()
|
|
assert not list(settings.repository_roots[0].glob(".main.staging-*"))
|
|
assert not list((settings.data_dir / "repository-keys").glob("*"))
|
|
|
|
|
|
@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")
|
|
|
|
|
|
def test_initialization_supports_not_yet_created_nested_path(tmp_path: Path) -> None:
|
|
settings = settings_for(tmp_path)
|
|
result = initialize(settings, "nested/main", "none", "none")
|
|
assert result.root == settings.repository_roots[0] / "nested" / "main"
|
|
assert (result.root / "repository.json").is_file()
|
|
|
|
|
|
@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_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:
|
|
settings = settings_for(tmp_path)
|
|
result = initialize(settings, "main", "none", "none")
|
|
(result.root / "repository.json").write_text(
|
|
'{"encryption":"none","compression":"none","format_version":1}\n'
|
|
)
|
|
with pytest.raises(RepositoryError, match="canonical"):
|
|
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"]
|
|
first = await client.post(
|
|
"/api/v2/repositories",
|
|
json={
|
|
"name": "main",
|
|
"relative_path": "first",
|
|
"compression": "none",
|
|
"encryption": "none",
|
|
},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
assert first.status_code == 201
|
|
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 not (settings.repository_roots[0] / "second").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)
|
|
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"]
|
|
created = await client.post(
|
|
"/api/v2/repositories",
|
|
json={
|
|
"name": "main",
|
|
"relative_path": "main",
|
|
"compression": "none",
|
|
"encryption": "none",
|
|
},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
assert created.status_code == 201
|
|
listed = await client.get("/api/v2/repositories")
|
|
assert listed.status_code == 200
|
|
assert listed.json()["items"][0]["id"] == created.json()["id"]
|
|
inspection = await client.get(f"/api/v2/repositories/{created.json()['id']}/inspection")
|
|
assert inspection.status_code == 200
|
|
assert inspection.json()["format_version"] == 1
|
|
await app.state.engine.dispose()
|