262 lines
9.3 KiB
Python
262 lines
9.3 KiB
Python
"""M15 recovery path, archive and secret-boundary security tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import tarfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from modelforge_api.domain.recovery import redact_database_url
|
|
from modelforge_api.services.recovery_paths import (
|
|
MAX_ARCHIVE_MEMBERS,
|
|
RecoveryPathError,
|
|
extract_archive,
|
|
normalise_root,
|
|
resolve_within,
|
|
sha256_bytes,
|
|
sha256_file,
|
|
validate_archive,
|
|
)
|
|
from modelforge_api.services.recovery_postgres import PostgresToolError, parse_target
|
|
|
|
|
|
@pytest.fixture
|
|
def root(tmp_path: Path) -> Path:
|
|
target = tmp_path / "backups"
|
|
target.mkdir()
|
|
return target
|
|
|
|
|
|
# --------------------------------------------------------------------- path safety
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"candidate",
|
|
[
|
|
"../escape.dump",
|
|
"nested/../../escape.dump",
|
|
"..",
|
|
"a/../../b",
|
|
"~/escape.dump",
|
|
"/etc/passwd",
|
|
"C:/Windows/System32/config/SAM",
|
|
"\\\\server\\share\\payload.dump",
|
|
"",
|
|
" ",
|
|
],
|
|
)
|
|
def test_traversal_and_absolute_paths_are_rejected(root: Path, candidate: str) -> None:
|
|
with pytest.raises(RecoveryPathError) as error:
|
|
resolve_within(root, candidate)
|
|
assert error.value.code == "PATH_NOT_ALLOWED"
|
|
|
|
|
|
def test_a_legitimate_relative_path_resolves_inside_the_root(root: Path) -> None:
|
|
resolved = resolve_within(root, "m15-backup/database.dump")
|
|
assert resolved.parent.parent == root.resolve()
|
|
assert resolved.name == "database.dump"
|
|
|
|
|
|
def test_a_missing_payload_reports_payload_missing_rather_than_path_denied(root: Path) -> None:
|
|
with pytest.raises(RecoveryPathError) as error:
|
|
resolve_within(root, "m15-backup/absent.dump", must_exist=True)
|
|
assert error.value.code == "PAYLOAD_MISSING"
|
|
|
|
|
|
def test_a_symlink_out_of_the_root_is_refused(root: Path, tmp_path: Path) -> None:
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
(outside / "secret.dump").write_bytes(b"not-ours")
|
|
link = root / "escape"
|
|
try:
|
|
link.symlink_to(outside, target_is_directory=True)
|
|
except (OSError, NotImplementedError):
|
|
pytest.skip("symlink creation is not permitted in this environment")
|
|
|
|
with pytest.raises(RecoveryPathError) as error:
|
|
resolve_within(root, "escape/secret.dump", must_exist=True)
|
|
assert error.value.code == "PATH_NOT_ALLOWED"
|
|
|
|
|
|
def test_a_relative_recovery_root_is_refused() -> None:
|
|
with pytest.raises(RecoveryPathError) as error:
|
|
normalise_root(Path("relative/backups"))
|
|
assert error.value.code == "PATH_NOT_ALLOWED"
|
|
|
|
|
|
def test_hashing_helpers_agree_on_the_same_content(root: Path) -> None:
|
|
payload = b"modelforge-recovery-payload"
|
|
path = root / "payload.bin"
|
|
path.write_bytes(payload)
|
|
assert sha256_file(path) == sha256_bytes(payload)
|
|
|
|
|
|
# --------------------------------------------------------------------- archive safety
|
|
|
|
|
|
def build_archive(path: Path, members: list[tuple[str, bytes]]) -> Path:
|
|
with tarfile.open(path, "w") as handle:
|
|
for name, payload in members:
|
|
info = tarfile.TarInfo(name)
|
|
info.size = len(payload)
|
|
handle.addfile(info, io.BytesIO(payload))
|
|
return path
|
|
|
|
|
|
def test_a_benign_archive_validates_and_extracts_inside_the_destination(
|
|
root: Path, tmp_path: Path
|
|
) -> None:
|
|
archive = build_archive(
|
|
tmp_path / "good.tar", [("config/policies.yaml", b"policy: value"), ("manifest.json", b"{}")]
|
|
)
|
|
destination = root / "extracted"
|
|
names = validate_archive(archive, destination)
|
|
assert sorted(names) == ["config/policies.yaml", "manifest.json"]
|
|
extract_archive(archive, destination)
|
|
assert (destination / "config" / "policies.yaml").read_bytes() == b"policy: value"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"member",
|
|
["../escape.txt", "nested/../../escape.txt", "/absolute/escape.txt"],
|
|
)
|
|
def test_a_traversing_archive_member_is_refused_before_extraction(
|
|
root: Path, tmp_path: Path, member: str
|
|
) -> None:
|
|
archive = build_archive(tmp_path / "evil.tar", [(member, b"payload")])
|
|
destination = root / "extracted"
|
|
with pytest.raises(RecoveryPathError) as error:
|
|
validate_archive(archive, destination)
|
|
assert error.value.code == "ARCHIVE_UNSAFE"
|
|
assert not any(destination.iterdir())
|
|
|
|
|
|
def test_a_symlinked_archive_member_is_refused(root: Path, tmp_path: Path) -> None:
|
|
archive = tmp_path / "link.tar"
|
|
with tarfile.open(archive, "w") as handle:
|
|
info = tarfile.TarInfo("evil-link")
|
|
info.type = tarfile.SYMTYPE
|
|
info.linkname = "/etc/passwd"
|
|
handle.addfile(info)
|
|
with pytest.raises(RecoveryPathError) as error:
|
|
validate_archive(archive, root / "extracted")
|
|
assert error.value.code == "ARCHIVE_UNSAFE"
|
|
|
|
|
|
def test_a_device_archive_member_is_refused(root: Path, tmp_path: Path) -> None:
|
|
archive = tmp_path / "device.tar"
|
|
with tarfile.open(archive, "w") as handle:
|
|
info = tarfile.TarInfo("evil-device")
|
|
info.type = tarfile.CHRTYPE
|
|
handle.addfile(info)
|
|
with pytest.raises(RecoveryPathError) as error:
|
|
validate_archive(archive, root / "extracted")
|
|
assert error.value.code == "ARCHIVE_UNSAFE"
|
|
|
|
|
|
def test_an_unbounded_archive_member_count_is_refused(root: Path, tmp_path: Path) -> None:
|
|
archive = tmp_path / "many.tar"
|
|
with tarfile.open(archive, "w") as handle:
|
|
for index in range(MAX_ARCHIVE_MEMBERS + 5):
|
|
info = tarfile.TarInfo(f"file-{index}.txt")
|
|
info.size = 0
|
|
handle.addfile(info, io.BytesIO(b""))
|
|
with pytest.raises(RecoveryPathError) as error:
|
|
validate_archive(archive, root / "extracted")
|
|
assert error.value.code == "ARCHIVE_UNSAFE"
|
|
|
|
|
|
def test_a_declared_extraction_bomb_is_refused(root: Path, tmp_path: Path) -> None:
|
|
archive = tmp_path / "bomb.tar"
|
|
# A header that claims 65 GiB; tarfile itself refuses to write one, so it is crafted here.
|
|
info = tarfile.TarInfo("huge.bin")
|
|
info.size = 65 * 1024**3
|
|
archive.write_bytes(info.tobuf() + bytes(1024))
|
|
with pytest.raises(RecoveryPathError) as error:
|
|
validate_archive(archive, root / "extracted")
|
|
assert error.value.code == "ARCHIVE_UNSAFE"
|
|
|
|
|
|
# --------------------------------------------------------------------- destination safety
|
|
|
|
|
|
def test_only_postgresql_destinations_with_safe_identifiers_are_accepted() -> None:
|
|
target = parse_target("postgresql+psycopg://modelforge:pw@postgres:5544/mf_restore")
|
|
assert (target.host, target.port, target.database) == ("postgres", 5544, "mf_restore")
|
|
assert target.redacted == "postgresql://modelforge@postgres:5544/mf_restore"
|
|
assert "pw" not in target.redacted
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url",
|
|
[
|
|
"sqlite:///local.db",
|
|
"mysql://user:pw@host/db",
|
|
'postgresql://user:pw@host:5432/db";drop table models;--',
|
|
"postgresql://user:pw@host:5432/",
|
|
],
|
|
)
|
|
def test_unsupported_or_unsafe_destinations_are_refused(url: str) -> None:
|
|
with pytest.raises(PostgresToolError):
|
|
parse_target(url)
|
|
|
|
|
|
def test_a_redacted_url_never_carries_the_password() -> None:
|
|
redacted = redact_database_url("postgresql+psycopg://modelforge:hunter2@postgres:5432/mf")
|
|
assert "hunter2" not in redacted
|
|
assert redacted == "postgresql+psycopg://modelforge:***@postgres:5432/mf"
|
|
|
|
|
|
# --------------------------------------------------------------------- secret boundary
|
|
|
|
|
|
def test_a_configuration_manifest_names_secrets_without_carrying_their_values(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from modelforge_api.persistence.models import Base
|
|
from modelforge_api.services.recovery import RecoveryService
|
|
from modelforge_api.settings import Settings
|
|
|
|
config_root = tmp_path / "config"
|
|
(config_root / "policies").mkdir(parents=True)
|
|
(config_root / "policies" / "defaults.yaml").write_bytes(b"retention: 30\n")
|
|
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
service = RecoveryService(
|
|
session,
|
|
Settings(
|
|
backup_root=tmp_path / "backups",
|
|
config_root=config_root,
|
|
backup_encryption_key="a-local-rehearsal-key",
|
|
operator_api_key="super-secret-operator-key",
|
|
hf_token="hf_super_secret_token", # noqa: S106 - deliberate leak probe
|
|
database_url="postgresql+psycopg://modelforge:dbpassword@postgres:5432/modelforge",
|
|
),
|
|
)
|
|
manifest = service._configuration_manifest()
|
|
|
|
body = json.dumps(manifest)
|
|
assert "super-secret-operator-key" not in body
|
|
assert "hf_super_secret_token" not in body
|
|
assert "a-local-rehearsal-key" not in body
|
|
assert "dbpassword" not in body
|
|
keys = {item["key"] for item in manifest["secrets"]}
|
|
assert keys == {
|
|
"MODELFORGE_OPERATOR_API_KEY",
|
|
"MODELFORGE_BACKUP_ENCRYPTION_KEY",
|
|
"MODELFORGE_HF_TOKEN",
|
|
}
|
|
recoveries = {item["recovery"] for item in manifest["secrets"]}
|
|
assert recoveries == {"ROTATABLE_SECRET", "NON_EXPORTABLE_SECRET"}
|
|
files = {item["relative_path"]: item for item in manifest["files"]}
|
|
assert files["policies/defaults.yaml"]["classification"] == "SOURCE_CONTROLLED"
|
|
assert files["policies/defaults.yaml"]["sha256"] == sha256_bytes(b"retention: 30\n")
|