161 lines
5.3 KiB
Python
161 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from backup_tool.cli import build_alembic_config, build_parser
|
|
from backup_tool.cli import main as cli_main
|
|
from backup_tool.db.engine import create_engine
|
|
from backup_tool.db.models import Repository, RepositoryDataKeyEpoch
|
|
from backup_tool.repository import initialize
|
|
from backup_tool.security.recovery_bundle import (
|
|
RecoveryBundleError,
|
|
RecoveryBundlePathError,
|
|
decrypt_bundle,
|
|
encrypt_bundle,
|
|
write_bundle_exclusive,
|
|
)
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
from tests.conftest import make_settings
|
|
|
|
PASSPHRASE = b"correct horse battery staple"
|
|
|
|
|
|
def _passphrase_fd(value: bytes) -> int:
|
|
read_fd, write_fd = os.pipe()
|
|
os.write(write_fd, value + b"\n")
|
|
os.close(write_fd)
|
|
return read_fd
|
|
|
|
|
|
def test_recovery_bundle_rejects_wrong_passphrase_tampering_and_invalid_kdf() -> None:
|
|
bundle = encrypt_bundle({"catalog": {"version": 1}, "keys": []}, PASSPHRASE)
|
|
tampered = bundle[:-1] + bytes([bundle[-1] ^ 1])
|
|
unsupported_kdf = bundle[:6] + b"\x02" + bundle[7:]
|
|
|
|
for encoded, passphrase in (
|
|
(bundle, b"wrong passphrase"),
|
|
(tampered, PASSPHRASE),
|
|
(unsupported_kdf, PASSPHRASE),
|
|
):
|
|
with pytest.raises(RecoveryBundleError, match="^recovery bundle is invalid$"):
|
|
decrypt_bundle(encoded, passphrase)
|
|
|
|
|
|
def test_recovery_bundle_output_is_exclusive_and_does_not_follow_symlinks(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
bundle = encrypt_bundle({"catalog": {"version": 1}, "keys": []}, PASSPHRASE)
|
|
existing = tmp_path / "existing.btrec"
|
|
existing.write_bytes(b"keep")
|
|
|
|
with pytest.raises(RecoveryBundlePathError, match="^recovery bundle output is unsafe$"):
|
|
write_bundle_exclusive(existing, bundle)
|
|
assert existing.read_bytes() == b"keep"
|
|
|
|
target = tmp_path / "target.btrec"
|
|
target.write_bytes(b"keep")
|
|
symlink = tmp_path / "link.btrec"
|
|
symlink.symlink_to(target)
|
|
with pytest.raises(RecoveryBundlePathError, match="^recovery bundle output is unsafe$"):
|
|
write_bundle_exclusive(symlink, bundle)
|
|
assert target.read_bytes() == b"keep"
|
|
|
|
|
|
def test_cli_recovery_export_validate_uses_fd_and_hides_plaintext_keys(
|
|
tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
settings = make_settings(tmp_path)
|
|
command.upgrade(build_alembic_config(settings), "head")
|
|
initialized = initialize(settings, "encrypted", "none", "aes-256-gcm")
|
|
assert initialized.data_key_id is not None
|
|
assert initialized.data_key_path is not None
|
|
|
|
async def create_repository() -> None:
|
|
engine = create_engine(settings)
|
|
sessions = async_sessionmaker(engine, expire_on_commit=False)
|
|
try:
|
|
async with sessions() as db:
|
|
repository = Repository(
|
|
name="encrypted",
|
|
root=str(initialized.root),
|
|
format_version=initialized.format_version,
|
|
compression=initialized.compression,
|
|
encryption=initialized.encryption,
|
|
signing_key_id=initialized.signing_key_id,
|
|
signing_public_key=initialized.signing_public_key,
|
|
active_data_key_id=initialized.data_key_id,
|
|
)
|
|
db.add(repository)
|
|
await db.flush()
|
|
db.add(
|
|
RepositoryDataKeyEpoch(
|
|
repository_id=repository.id,
|
|
key_id=initialized.data_key_id,
|
|
state="active",
|
|
)
|
|
)
|
|
await db.commit()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
asyncio.run(create_repository())
|
|
output = tmp_path / "recovery.btrec"
|
|
export_fd = _passphrase_fd(PASSPHRASE)
|
|
try:
|
|
assert (
|
|
cli_main(
|
|
[
|
|
"admin",
|
|
"recovery",
|
|
"export",
|
|
"--output",
|
|
str(output),
|
|
"--passphrase-fd",
|
|
str(export_fd),
|
|
],
|
|
settings=settings,
|
|
)
|
|
== 0
|
|
)
|
|
finally:
|
|
os.close(export_fd)
|
|
export_output = capsys.readouterr().out
|
|
bundle = output.read_bytes()
|
|
assert bundle.startswith(b"BTREC\x01")
|
|
assert initialized.data_key_path.read_bytes() not in bundle
|
|
signing_key_path = (
|
|
settings.data_dir / "repository-keys" / f"{initialized.repository_id}.ed25519"
|
|
)
|
|
assert signing_key_path.read_bytes() not in bundle
|
|
assert "key" not in export_output.lower()
|
|
|
|
validate_fd = _passphrase_fd(PASSPHRASE)
|
|
try:
|
|
assert (
|
|
cli_main(
|
|
[
|
|
"admin",
|
|
"recovery",
|
|
"validate",
|
|
"--input",
|
|
str(output),
|
|
"--passphrase-fd",
|
|
str(validate_fd),
|
|
],
|
|
settings=settings,
|
|
)
|
|
== 0
|
|
)
|
|
finally:
|
|
os.close(validate_fd)
|
|
assert capsys.readouterr().out == '{"repositories": 1, "status": "valid"}\n'
|
|
|
|
parser = build_parser()
|
|
with pytest.raises(SystemExit):
|
|
parser.parse_args(["admin", "recovery", "export", "--passphrase", "not-allowed"])
|