|
| 1 | +"""Synthetic release envelopes and mocked gh; no production requests or signing.""" |
| 2 | +from datetime import datetime, timedelta, timezone |
| 3 | +import copy |
| 4 | +import hashlib |
| 5 | +import importlib.util |
| 6 | +import json |
| 7 | +import os |
| 8 | +from pathlib import Path |
| 9 | +import subprocess |
| 10 | +import tempfile |
| 11 | +import unittest |
| 12 | +from unittest.mock import patch |
| 13 | + |
| 14 | +spec = importlib.util.spec_from_file_location("keyless_rules", Path(__file__).parent / "tools/signed_rules/keyless_rules.py") |
| 15 | +k = importlib.util.module_from_spec(spec) |
| 16 | +spec.loader.exec_module(k) |
| 17 | + |
| 18 | + |
| 19 | +class KeylessTests(unittest.TestCase): |
| 20 | + def setUp(self): |
| 21 | + self.temp = tempfile.TemporaryDirectory() |
| 22 | + self.addCleanup(self.temp.cleanup) |
| 23 | + self.root = Path(self.temp.name) |
| 24 | + self.artifact, self.bundle, self.trust, self.state, self.output = [self.root / name for name in ("rules.json", "bundle.json", "roots.jsonl", "state.json", "review.json")] |
| 25 | + self.bundle.write_text("synthetic-bundle", encoding="utf-8") |
| 26 | + self.trust.write_text("independent-synthetic-root", encoding="utf-8") |
| 27 | + self.now = datetime(2026, 10, 26, tzinfo=timezone.utc) |
| 28 | + self.value = k.build(1001, "a" * 40, self.now) |
| 29 | + self.artifact.write_bytes(k.encoded(self.value)) |
| 30 | + |
| 31 | + def verify(self, value=None, **kwargs): |
| 32 | + if value is not None: |
| 33 | + self.artifact.write_bytes(k.encoded(value)) |
| 34 | + with patch("socket.socket", side_effect=AssertionError("offline only")): |
| 35 | + with patch.object(k, "verify_provenance") as proof: |
| 36 | + result = k.verify(self.artifact, self.bundle, self.trust, self.state, self.output, now=self.now, **kwargs) |
| 37 | + proof.assert_called_once() |
| 38 | + return result |
| 39 | + |
| 40 | + def test_build_matches_embedded_snapshot_and_never_changes_checker(self): |
| 41 | + import node2shell_check as checker |
| 42 | + before = Path(checker.__file__).read_bytes() |
| 43 | + self.assertEqual({str(a): list(b) for a, b in checker.NODE2SHELL_FIXES.items()}, self.value["rules"]["fixes"]) |
| 44 | + self.assertEqual(hashlib.sha256(before).hexdigest(), self.value["rules"]["checker_sha256"]) |
| 45 | + self.verify() |
| 46 | + self.assertEqual(self.value["rules"], json.loads(self.output.read_bytes())) |
| 47 | + self.assertEqual(before, Path(checker.__file__).read_bytes()) |
| 48 | + self.assertEqual(1001, json.loads(self.state.read_bytes())["sequence"]) |
| 49 | + |
| 50 | + def test_unsigned_tampered_and_untrusted_rejections_preserve_previous_output(self): |
| 51 | + self.verify() |
| 52 | + before = self.output.read_bytes() |
| 53 | + for error in (ValueError("wrong publisher"), subprocess.CalledProcessError(1, ["gh"]), subprocess.TimeoutExpired(["gh"], 1)): |
| 54 | + with patch.object(k, "verify_provenance", side_effect=error): |
| 55 | + with self.assertRaises((ValueError, subprocess.SubprocessError)): |
| 56 | + k.verify(self.artifact, self.bundle, self.trust, self.state, self.output, True, self.now) |
| 57 | + self.assertEqual(before, self.output.read_bytes()) |
| 58 | + |
| 59 | + def test_expiry_future_timestamps_and_schema_uncertainty_fail_closed(self): |
| 60 | + self.verify() |
| 61 | + before = self.output.read_bytes() |
| 62 | + cases = [] |
| 63 | + for key, value in [("expires_at", k.timestamp(self.now)), ("issued_at", k.timestamp(self.now + timedelta(minutes=6))), ("sequence", True), ("source_commit", "../x")]: |
| 64 | + item = copy.deepcopy(self.value); item[key] = value; cases.append(item) |
| 65 | + for key, value in [("exploitability", "confirmed"), ("fixes", {"24": [22, 0, 1]}), ("as_of", "2099-01-01"), ("sources", ["https://private.invalid"] )]: |
| 66 | + item = copy.deepcopy(self.value); item["rules"][key] = value; cases.append(item) |
| 67 | + for value in cases: |
| 68 | + with self.assertRaises(ValueError): self.verify(value, force=True) |
| 69 | + self.assertEqual(before, self.output.read_bytes()) |
| 70 | + |
| 71 | + def test_rollback_sequence_reuse_and_same_artifact_retry(self): |
| 72 | + self.verify() |
| 73 | + self.verify(force=True) |
| 74 | + for sequence in (1000, 1001): |
| 75 | + value = copy.deepcopy(self.value); value["sequence"] = sequence |
| 76 | + value["rules"]["checker_sha256"] = "b" * 64 |
| 77 | + with self.assertRaisesRegex(ValueError, "rollback|reuse"): self.verify(value, force=True) |
| 78 | + newer = copy.deepcopy(self.value); newer["sequence"] = 2001 |
| 79 | + self.verify(newer, force=True) |
| 80 | + self.assertEqual(2001, json.loads(self.state.read_bytes())["sequence"]) |
| 81 | + |
| 82 | + def test_clock_rollback_invalid_state_and_existing_lock_are_not_bypassed(self): |
| 83 | + self.verify() |
| 84 | + prior = json.loads(self.state.read_bytes()) |
| 85 | + prior["accepted_at"] = k.timestamp(self.now + timedelta(minutes=10)) |
| 86 | + self.state.write_bytes(k.encoded(prior)) |
| 87 | + with patch.object(k, "verify_provenance"): |
| 88 | + with self.assertRaisesRegex(ValueError, "Clock rollback"): |
| 89 | + k.verify(self.artifact, self.bundle, self.trust, self.state, self.output, True, self.now + timedelta(minutes=4)) |
| 90 | + self.state.write_text('{"sequence":1}', encoding="utf-8") |
| 91 | + with self.assertRaisesRegex(ValueError, "durable state"): self.verify(force=True) |
| 92 | + lock = self.state.with_name(self.state.name + ".lock"); lock.write_text("other owner", encoding="utf-8") |
| 93 | + with self.assertRaises(FileExistsError): self.verify(force=True) |
| 94 | + self.assertEqual("other owner", lock.read_text()) |
| 95 | + |
| 96 | + def test_protected_paths_hardlinks_and_no_clobber(self): |
| 97 | + self.verify() |
| 98 | + with self.assertRaises(ValueError): self.verify() |
| 99 | + for path in (self.artifact, self.bundle, self.trust, self.state, self.state.with_name(self.state.name + ".lock"), Path(__file__).parent / "node2shell_check.py"): |
| 100 | + with self.assertRaises(ValueError): |
| 101 | + k.verify(self.artifact, self.bundle, self.trust, self.state, path, True, self.now) |
| 102 | + alias = self.root / "alias"; os.link(self.artifact, alias) |
| 103 | + with self.assertRaises(ValueError): |
| 104 | + k.verify(self.artifact, self.bundle, self.trust, self.state, alias, True, self.now) |
| 105 | + |
| 106 | + def test_failed_output_fences_rollback_and_allows_verified_retry(self): |
| 107 | + real_atomic = k.atomic |
| 108 | + def failure(path, data, replace=False): |
| 109 | + if path.resolve() == self.output.resolve(): raise OSError("output unavailable") |
| 110 | + return real_atomic(path, data, replace) |
| 111 | + with patch.object(k, "atomic", side_effect=failure): |
| 112 | + with self.assertRaises(OSError): self.verify() |
| 113 | + self.assertEqual(1001, json.loads(self.state.read_bytes())["sequence"]) |
| 114 | + self.assertFalse(self.output.exists()) |
| 115 | + self.verify() |
| 116 | + self.assertEqual([], list(self.root.glob(".rules-*"))) |
| 117 | + |
| 118 | + def test_gh_policy_checks_cert_numeric_ids_and_exact_source_identity(self): |
| 119 | + good = [{"verificationResult": {"signature": {"certificate": {"sourceRepositoryIdentifier": k.REPOSITORY_ID, "sourceRepositoryOwnerIdentifier": k.OWNER_ID}}}}] |
| 120 | + for wrong in (False, True): |
| 121 | + value = copy.deepcopy(good) |
| 122 | + if wrong: value[0]["verificationResult"]["signature"]["certificate"]["sourceRepositoryIdentifier"] = "123" |
| 123 | + def run(command, **kwargs): |
| 124 | + self.assertIn("--custom-trusted-root", command) |
| 125 | + self.assertIn("--bundle", command) |
| 126 | + for expected in (k.IDENTITY, k.ISSUER, "--deny-self-hosted-runners", "--source-digest", "--signer-digest", "refs/heads/main"): |
| 127 | + self.assertIn(expected, command) |
| 128 | + self.assertNotIn("shell", kwargs) |
| 129 | + kwargs["stdout"].write(k.encoded(value)) |
| 130 | + with patch.object(k.shutil, "which", return_value="trusted-gh"), patch.object(k.subprocess, "run", side_effect=run): |
| 131 | + if wrong: |
| 132 | + with self.assertRaisesRegex(ValueError, "numeric identity"): |
| 133 | + k.verify_provenance(self.artifact, self.bundle, self.trust, "a" * 40) |
| 134 | + else: k.verify_provenance(self.artifact, self.bundle, self.trust, "a" * 40) |
| 135 | + |
| 136 | + def test_oversize_duplicate_json_and_malformed_output_never_publish(self): |
| 137 | + self.artifact.write_bytes(b" " * (k.MAX_BYTES + 1)) |
| 138 | + with self.assertRaises(ValueError): self.verify() |
| 139 | + self.artifact.write_bytes(b'{"schema":1,"schema":2}') |
| 140 | + with self.assertRaises(ValueError): self.verify() |
| 141 | + self.assertFalse(self.state.exists()); self.assertFalse(self.output.exists()) |
| 142 | + |
| 143 | + |
| 144 | +if __name__ == "__main__": |
| 145 | + unittest.main() |
0 commit comments