bootstrap_retention_test.py (8920B)
1 #!/usr/bin/env python3 2 """Filesystem safety controls for the opt-in checkpoint retention command.""" 3 import copy 4 import fcntl 5 import json 6 import os 7 from pathlib import Path 8 import subprocess 9 import sys 10 import tempfile 11 import unittest 12 from unittest.mock import patch 13 14 import bootstrap_retention as r 15 from bootstrap_retention_lock import acquire, protect_generations 16 17 18 class RetentionTests(unittest.TestCase): 19 def setUp(self): 20 self.temp = tempfile.TemporaryDirectory(); self.addCleanup(self.temp.cleanup) 21 self.root = Path(self.temp.name) / '.bootstrap-run'; self.root.mkdir() 22 self.keep = self.root / 'current'; self.keep.mkdir() 23 (self.keep / 'compiler').write_bytes(b'protected compiler') 24 self.old = self.root / 'old/cas/objects/abc'; self.old.mkdir(parents=True) 25 (self.old / 'tree').write_bytes(b'old tree') 26 (self.old / 'receipt').write_bytes(b'arboricx.build-receipt.v1\nsource\t123\n') 27 self.out = self.root / 'checkpoint'; self.out.mkdir() 28 self.config = dict(protected=['current']) 29 30 def plan(self): 31 plan = r.make_plan(self.root, self.config, self.out) 32 r.save(self.out / 'plan.json', plan) 33 return plan 34 35 def test_only_objects_and_preserved_evidence(self): 36 (self.root / 'old/status.json').write_text('{"complete": true}') 37 (self.root / 'old/source.tri').write_text('t') 38 plan = self.plan() 39 self.assertEqual([x['objects'] for x in plan['stores']], ['old/cas/objects']) 40 r.prune(self.root, plan, self.out) 41 self.assertFalse(self.old.exists()) 42 self.assertTrue((self.keep / 'compiler').exists()) 43 self.assertTrue((self.root / 'old/status.json').exists()) 44 self.assertTrue((self.root / 'old/source.tri').exists()) 45 self.assertFalse(plan['complete']) # postchecks are mandatory 46 marker = json.loads((self.root / 'old/cas/PRUNED-CAS.json').read_text()) 47 self.assertFalse(marker['resumable']) 48 self.assertEqual(len(list((self.out / 'historical-metadata').iterdir())), 1) 49 50 def test_symlink_target_rejected(self): 51 (self.root / 'bad').mkdir() 52 (self.root / 'bad/objects').symlink_to(self.old.parent, target_is_directory=True) 53 with self.assertRaises(ValueError): self.plan() 54 55 def test_symlink_object_rejected(self): 56 (self.old / 'link').symlink_to(self.keep / 'compiler') 57 with self.assertRaises(ValueError): self.plan() 58 59 def test_path_escape_and_normalization(self): 60 for relative in ['../outside', '/tmp/objects', 'old/../current', 'old//cas', '']: 61 with self.assertRaises(ValueError): r.within(self.root, relative) 62 63 def test_mutated_candidate_blocks_all_deletion(self): 64 plan = self.plan(); (self.old / 'tree').write_bytes(b'changed tree') 65 with self.assertRaises(ValueError): r.prune(self.root, plan, self.out) 66 self.assertTrue((self.old / 'receipt').exists()) 67 68 def test_mutated_protected_input_blocks_deletion(self): 69 plan = self.plan(); (self.keep / 'compiler').write_bytes(b'changed') 70 with self.assertRaises(ValueError): r.prune(self.root, plan, self.out) 71 self.assertTrue(self.old.exists()) 72 73 def test_corrupt_metadata_blocks_deletion(self): 74 plan = self.plan() 75 next((self.out / 'historical-metadata').iterdir()).write_bytes(b'corrupt') 76 with self.assertRaises(ValueError): r.prune(self.root, plan, self.out) 77 self.assertTrue(self.old.exists()) 78 79 def test_protected_target_injected_in_plan_rejected(self): 80 (self.keep / 'objects').mkdir() 81 plan = self.plan() 82 plan['stores'].append(dict(objects='current/objects', files={}, metadata=[], state='pending')) 83 with self.assertRaises(ValueError): r.prune(self.root, plan, self.out) 84 self.assertTrue(self.old.exists()) 85 86 def test_hardlink_reclaim_and_retained_identity(self): 87 os.link(self.keep / 'compiler', self.old / 'shared') 88 external = Path(self.temp.name) / 'external'; os.link(self.old / 'tree', external) 89 plan = self.plan() 90 expected = (self.old / 'receipt').stat().st_blocks * 512 91 self.assertEqual(plan['reclaim_bytes'], expected) 92 r.prune(self.root, plan, self.out) 93 self.assertEqual((self.keep / 'compiler').read_bytes(), b'protected compiler') 94 self.assertEqual(external.read_bytes(), b'old tree') 95 96 def test_interrupted_directory_resumption(self): 97 plan = self.plan() 98 plan['stores'][0]['state'] = 'deleting' 99 (self.old / 'tree').unlink() # interrupted shutil.rmtree 100 r.prune(self.root, plan, self.out) 101 self.assertEqual(plan['stores'][0]['state'], 'removed') 102 r.prune(self.root, plan, self.out) # idempotent before postchecks 103 104 def test_missing_pending_directory_not_resumption(self): 105 plan = self.plan() 106 import shutil 107 shutil.rmtree(self.old.parent) 108 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out) 109 110 def test_removed_directory_reappearance_rejected(self): 111 plan = self.plan(); r.prune(self.root, plan, self.out) 112 self.old.mkdir(parents=True) 113 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out) 114 115 def test_tool_changed_plan_rejected(self): 116 plan = self.plan(); plan['tool_sha256'] = '0' * 64 117 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out) 118 119 def test_generation_cleanup_lock_exclusion(self): 120 handles = protect_generations([self.keep, self.old]) 121 self.addCleanup(lambda: [h.close() for h in handles]) 122 with self.assertRaises(ValueError): acquire(self.root, exclusive=True) 123 for h in handles: h.close() 124 lock = acquire(self.root, exclusive=True); self.addCleanup(lock.close) 125 with self.assertRaises(ValueError): protect_generations([self.keep]) 126 127 def test_generation_runner_rejects_pruned_input_and_output(self): 128 plan = self.plan(); r.prune(self.root, plan, self.out) 129 script = r.REPO / 'test/bootstrap_generations.py' 130 for args in [[str(self.root / 'old'), '--bootstrap-kit', str(self.root / 'kit')], 131 [str(self.root / 'old'), str(self.root / 'fresh')]]: 132 reply = subprocess.run([sys.executable, str(script), *args], 133 capture_output=True, text=True, timeout=30) 134 self.assertNotEqual(reply.returncode, 0) 135 self.assertIn('pruned generations are historical evidence', reply.stderr) 136 self.assertFalse((self.root / 'old/status.json').exists()) 137 self.assertFalse((self.root / 'fresh').exists()) 138 139 def test_pending_checkpoint_cannot_pass_completion_gate(self): 140 plan = self.plan() 141 with self.assertRaises(ValueError): r.validate_completion(self.root, plan, self.out) 142 r.prune(self.root, plan, self.out) 143 with self.assertRaises(ValueError): r.validate_completion(self.root, plan, self.out) 144 145 def test_completed_checkpoint_requires_both_check_reports(self): 146 plan = self.plan(); r.prune(self.root, plan, self.out) 147 plan.update(complete=True, phase='complete', prechecks='missing-before', postchecks='missing-after') 148 with self.assertRaises(ValueError): r.validate_completion(self.root, plan, self.out) 149 150 def test_plan_cannot_create_evidence_inside_protected_inputs(self): 151 evidence = self.keep / 'must-not-create' 152 args = ['bootstrap_retention.py', 'plan', '--root', str(self.root), 153 '--directory', str(evidence), '--config', str(self.root / 'config.json')] 154 with patch('sys.argv', args), patch.object(r, 'load_config', return_value=self.config): 155 with self.assertRaises(ValueError): r.main() 156 self.assertFalse(evidence.exists()) 157 158 def test_archived_harness_allows_non_destructive_new_version_check(self): 159 plan = self.plan() 160 original_sha = self.original_sha 161 with patch.object(r, 'sha', side_effect=lambda p: 'changed' if Path(p) == Path(r.__file__) else original_sha(p)): 162 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out) 163 r.validate_plan(self.root, plan, self.out, same_tools=False) 164 (self.out / 'harness/bootstrap_retention.py').write_bytes(b'corrupt') 165 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out, same_tools=False) 166 167 original_sha = staticmethod(r.sha) 168 169 def test_symlink_retained_directory_is_fingerprinted(self): 170 (self.keep / 'runtime').symlink_to('/nix/store/one', target_is_directory=True) 171 plan = self.plan() 172 (self.keep / 'runtime').unlink() 173 (self.keep / 'runtime').symlink_to('/nix/store/two', target_is_directory=True) 174 with self.assertRaises(ValueError): r.validate_plan(self.root, plan, self.out) 175 176 177 if __name__ == '__main__': 178 unittest.main()