e1918b3f88
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
"""Kit integrity failures must refuse before extracting executable material."""
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import sys
|
|
import tarfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
|
import application_recovery_drill as drill
|
|
|
|
|
|
def kit(path, mutation=None):
|
|
payload = b"synthetic recovery code"
|
|
manifest = {"schema": "platform.coulomb-recovery-kit-manifest.v1",
|
|
"private_identity_included": False, "api_pairs_included": False,
|
|
"files": {"tools/recover.py": {"bytes": len(payload),
|
|
"sha256": hashlib.sha256(payload).hexdigest()}}}
|
|
name = "../escape" if mutation == "traversal" else "tools/recover.py"
|
|
if mutation == "credential":
|
|
manifest["private_identity_included"] = True
|
|
with tarfile.open(path, "w") as archive:
|
|
data = json.dumps(manifest).encode()
|
|
meta = tarfile.TarInfo("manifest.json")
|
|
meta.size = len(data)
|
|
archive.addfile(meta, io.BytesIO(data))
|
|
item = tarfile.TarInfo(name)
|
|
item.size = len(payload)
|
|
if mutation == "link":
|
|
item.type, item.linkname, item.size = tarfile.SYMTYPE, "/etc/passwd", 0
|
|
actual = b"X" * len(payload) if mutation == "hash" else payload
|
|
archive.addfile(item, io.BytesIO(actual))
|
|
if mutation == "duplicate":
|
|
archive.addfile(item, io.BytesIO(actual))
|
|
|
|
|
|
def test_every_member_is_verified_before_extraction(tmp_path):
|
|
source = tmp_path / "kit.tar"
|
|
kit(source)
|
|
result = drill.extract_verified(source, tmp_path / "extracted")
|
|
assert (tmp_path / "extracted/tools/recover.py").read_bytes() == b"synthetic recovery code"
|
|
assert result["api_pairs_included"] is False
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["traversal", "link", "duplicate", "hash", "credential"])
|
|
def test_unsafe_or_corrupt_kit_never_extracts_code(tmp_path, mutation):
|
|
source = tmp_path / "kit.tar"
|
|
kit(source, mutation)
|
|
output = tmp_path / "extracted"
|
|
with pytest.raises(drill.Refused):
|
|
drill.extract_verified(source, output)
|
|
assert not output.exists()
|
|
|
|
|
|
def test_existing_extraction_is_preserved(tmp_path):
|
|
source = tmp_path / "kit.tar"
|
|
kit(source)
|
|
output = tmp_path / "existing"
|
|
output.mkdir()
|
|
(output / "keep").write_text("keep")
|
|
with pytest.raises(drill.Refused):
|
|
drill.extract_verified(source, output)
|
|
assert (output / "keep").read_text() == "keep"
|