137 lines
4.2 KiB
Python
137 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from backup_tool.gc import purge_repository
|
|
from backup_tool.security.repository_crypto import encrypt_object, object_aad
|
|
|
|
|
|
def write_manifest(path: Path, digests: list[str]) -> None:
|
|
path.write_text(
|
|
json.dumps({"entries": [{"blob_digest": digest} for digest in digests]}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def age(path: Path, days: int = 8) -> None:
|
|
stamp = (datetime.now(UTC) - timedelta(days=days)).timestamp()
|
|
os.utime(path, (stamp, stamp))
|
|
|
|
|
|
def test_gc_purges_tombstoned_manifest_and_only_unreferenced_old_blob(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
root = tmp_path / "repository"
|
|
manifests = root / "manifests"
|
|
blobs = root / "blobs" / "sha256"
|
|
manifests.mkdir(parents=True)
|
|
blobs.mkdir(parents=True)
|
|
kept = "a" * 64
|
|
removed = "b" * 64
|
|
write_manifest(manifests / "kept.json", [kept])
|
|
write_manifest(manifests / "deleted.json", [removed])
|
|
(blobs / kept).write_bytes(b"kept")
|
|
(blobs / removed).write_bytes(b"removed")
|
|
age(manifests / "deleted.json")
|
|
age(blobs / removed)
|
|
|
|
report = purge_repository(root, {"deleted"})
|
|
|
|
assert report.purged_manifests == 1
|
|
assert report.purged_blobs == 1
|
|
assert (manifests / "kept.json").exists()
|
|
assert (blobs / kept).exists()
|
|
assert not (blobs / removed).exists()
|
|
|
|
|
|
@pytest.mark.parametrize("encrypted", [False, True], ids=["corrupt", "encrypted"])
|
|
def test_gc_fails_closed_for_unreadable_manifest(tmp_path: Path, encrypted: bool) -> None:
|
|
root = tmp_path / "repository"
|
|
manifests = root / "manifests"
|
|
blobs = root / "blobs" / "sha256"
|
|
manifests.mkdir(parents=True)
|
|
blobs.mkdir(parents=True)
|
|
kept = "a" * 64
|
|
deleted = "b" * 64
|
|
write_manifest(manifests / "kept.json", [kept])
|
|
write_manifest(manifests / "deleted.json", [deleted])
|
|
unreadable = manifests / "unreadable.json"
|
|
unreadable.write_bytes(
|
|
encrypt_object(
|
|
b"k" * 32,
|
|
object_aad("repository", "key", "manifest", "unreadable"),
|
|
b'{"entries": []}',
|
|
)
|
|
if encrypted
|
|
else b"not json"
|
|
)
|
|
(blobs / kept).write_bytes(b"kept")
|
|
(blobs / deleted).write_bytes(b"deleted")
|
|
age(manifests / "deleted.json")
|
|
age(blobs / deleted)
|
|
|
|
report = purge_repository(root, {"deleted"})
|
|
|
|
assert report.purged_manifests == 0
|
|
assert report.purged_blobs == 0
|
|
assert (manifests / "deleted.json").exists()
|
|
assert (blobs / deleted).exists()
|
|
|
|
|
|
def test_gc_purges_encrypted_manifests_with_known_epoch_keys(tmp_path: Path) -> None:
|
|
root = tmp_path / "repository"
|
|
manifests = root / "manifests"
|
|
blobs = root / "blobs" / "sha256"
|
|
manifests.mkdir(parents=True)
|
|
blobs.mkdir(parents=True)
|
|
key = b"k" * 32
|
|
repository_id = "repository"
|
|
key_id = "epoch"
|
|
kept = "a" * 64
|
|
deleted = "b" * 64
|
|
for manifest_id, digests in (("kept", [kept]), ("deleted", [deleted])):
|
|
plaintext = json.dumps(
|
|
{"entries": [{"blob_digest": digest} for digest in digests]}
|
|
).encode()
|
|
(manifests / f"{manifest_id}.json").write_bytes(
|
|
encrypt_object(
|
|
key,
|
|
object_aad(repository_id, key_id, "manifest", manifest_id),
|
|
plaintext,
|
|
)
|
|
)
|
|
(blobs / kept).write_bytes(b"kept")
|
|
(blobs / deleted).write_bytes(b"deleted")
|
|
age(manifests / "deleted.json")
|
|
age(blobs / deleted)
|
|
|
|
report = purge_repository(
|
|
root,
|
|
{"deleted"},
|
|
repository_id=repository_id,
|
|
manifest_keys={"kept": (key_id, key), "deleted": (key_id, key)},
|
|
)
|
|
|
|
assert report.purged_manifests == 1
|
|
assert report.purged_blobs == 1
|
|
assert (blobs / kept).exists()
|
|
assert not (blobs / deleted).exists()
|
|
|
|
|
|
def test_gc_quarantines_unknown_blob_name(tmp_path: Path) -> None:
|
|
root = tmp_path / "repository"
|
|
blobs = root / "blobs" / "sha256"
|
|
blobs.mkdir(parents=True)
|
|
unknown = blobs / "not-a-digest"
|
|
unknown.write_bytes(b"unknown")
|
|
|
|
report = purge_repository(root, set())
|
|
|
|
assert report.quarantined == 1
|
|
assert not unknown.exists()
|
|
assert (root / "quarantine" / "not-a-digest").exists()
|