154 lines
6.2 KiB
Python
154 lines
6.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
META_ROOT = Path(__file__).resolve().parents[1]
|
|
RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release"
|
|
if str(RELEASE_TOOLS_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(RELEASE_TOOLS_ROOT))
|
|
|
|
from govoplan_release.candidate_artifact import ( # noqa: E402
|
|
CandidateArtifactError,
|
|
candidate_output_path,
|
|
issue_candidate_id,
|
|
issue_candidate_receipt,
|
|
verify_candidate_receipt,
|
|
validate_release_channel,
|
|
)
|
|
|
|
|
|
class CandidateArtifactTests(unittest.TestCase):
|
|
def test_channel_is_one_canonical_basename(self) -> None:
|
|
self.assertEqual("stable_1", validate_release_channel("stable_1"))
|
|
for value in ("../stable", "/stable", ".", "..", "Stable", "stable/name"):
|
|
with self.subTest(value=value), self.assertRaises(CandidateArtifactError):
|
|
validate_release_channel(value)
|
|
|
|
def test_handle_is_deterministic_and_can_precede_output(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir) / "not-created" / "release-candidates"
|
|
candidate_id = issue_candidate_id("rr-123", "attempt-456")
|
|
|
|
first = candidate_output_path(root, candidate_id)
|
|
second = candidate_output_path(root, candidate_id)
|
|
|
|
self.assertEqual(first, second)
|
|
self.assertEqual(candidate_id, first.name)
|
|
self.assertRegex(candidate_id, r"^candidate-[0-9a-f]{32}$")
|
|
|
|
def test_receipt_rejects_catalog_drift(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir) / "release-candidates"
|
|
candidate_id = issue_candidate_id("run", "attempt")
|
|
catalog = self._write_candidate(root / candidate_id)
|
|
receipt = issue_candidate_receipt(
|
|
root=root, candidate_id=candidate_id, channel="stable"
|
|
)
|
|
|
|
resolved = verify_candidate_receipt(
|
|
root=root,
|
|
candidate_id=receipt.candidate_id,
|
|
catalog_sha256=receipt.catalog_sha256,
|
|
channel="stable",
|
|
)
|
|
catalog.write_text(
|
|
json.dumps({"channel": "stable", "sequence": 2, "signatures": [{}]}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
self.assertEqual(root / candidate_id, resolved)
|
|
with self.assertRaisesRegex(CandidateArtifactError, "no longer matches"):
|
|
verify_candidate_receipt(
|
|
root=root,
|
|
candidate_id=receipt.candidate_id,
|
|
catalog_sha256=receipt.catalog_sha256,
|
|
channel="stable",
|
|
)
|
|
|
|
def test_unissued_ids_traversal_and_symlinks_fail_closed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir) / "release-candidates"
|
|
root.mkdir(mode=0o700)
|
|
candidate_id = issue_candidate_id("run", "attempt")
|
|
outside = Path(temp_dir) / "outside"
|
|
outside.mkdir()
|
|
(root / candidate_id).symlink_to(outside, target_is_directory=True)
|
|
|
|
with self.assertRaises(CandidateArtifactError):
|
|
candidate_output_path(root, "../candidate-" + "0" * 32)
|
|
with self.assertRaisesRegex(CandidateArtifactError, "real directory"):
|
|
candidate_output_path(root, candidate_id)
|
|
|
|
def test_catalog_and_channel_components_must_not_be_symlinks(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir) / "release-candidates"
|
|
candidate_id = issue_candidate_id("run", "attempt")
|
|
candidate = root / candidate_id
|
|
target = Path(temp_dir) / "target"
|
|
target.mkdir()
|
|
(target / "stable.json").write_text(
|
|
json.dumps({"signatures": [{}]}), encoding="utf-8"
|
|
)
|
|
root.mkdir(mode=0o700)
|
|
candidate.mkdir(mode=0o700)
|
|
(candidate / "channels").symlink_to(target, target_is_directory=True)
|
|
|
|
with self.assertRaisesRegex(CandidateArtifactError, "real directory"):
|
|
issue_candidate_receipt(
|
|
root=root, candidate_id=candidate_id, channel="stable"
|
|
)
|
|
|
|
def test_shared_candidate_roots_and_catalog_files_fail_closed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir) / "release-candidates"
|
|
root.mkdir(mode=0o755)
|
|
candidate_id = issue_candidate_id("run", "attempt")
|
|
with self.assertRaisesRegex(CandidateArtifactError, "another user"):
|
|
candidate_output_path(root, candidate_id)
|
|
|
|
root.chmod(0o700)
|
|
catalog = self._write_candidate(root / candidate_id)
|
|
catalog.chmod(0o640)
|
|
with self.assertRaisesRegex(CandidateArtifactError, "another user"):
|
|
issue_candidate_receipt(
|
|
root=root, candidate_id=candidate_id, channel="stable"
|
|
)
|
|
|
|
def test_receipt_rejects_catalog_with_inconsistent_channel(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir) / "release-candidates"
|
|
candidate_id = issue_candidate_id("run", "attempt")
|
|
catalog = self._write_candidate(root / candidate_id)
|
|
catalog.write_text(
|
|
json.dumps({"channel": "next", "signatures": [{}]}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with self.assertRaisesRegex(CandidateArtifactError, "does not match"):
|
|
issue_candidate_receipt(
|
|
root=root, candidate_id=candidate_id, channel="stable"
|
|
)
|
|
|
|
@staticmethod
|
|
def _write_candidate(candidate: Path) -> Path:
|
|
candidate.parent.mkdir(mode=0o700, exist_ok=True)
|
|
candidate.mkdir(mode=0o700)
|
|
channels = candidate / "channels"
|
|
channels.mkdir(mode=0o700)
|
|
catalog = channels / "stable.json"
|
|
catalog.write_text(
|
|
json.dumps({"channel": "stable", "sequence": 1, "signatures": [{}]}),
|
|
encoding="utf-8",
|
|
)
|
|
catalog.chmod(0o600)
|
|
return catalog
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|