import copy import hashlib import importlib.util import json import os from pathlib import Path import tempfile import pty import subprocess import unittest from unittest.mock import patch ROOT=Path(__file__).parents[1] spec=importlib.util.spec_from_file_location('prepare',ROOT/'tools/bootstrap-recovery/prepare-isolated-keycape.py') p=importlib.util.module_from_spec(spec);spec.loader.exec_module(p) class PreparationTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup) self.path=Path(self.temp.name)/'ciphertext';self.path.write_bytes(b'synthetic ciphertext');self.path.chmod(0o600) self.transfer={'schema':'netkingdom.identity-offsite.v1','status':'readback_verified','provider':'Storage Share', 'folder':'Backup/railiance-backups/net-kingdom-identity','artifact':'net-kingdom-identity-20261016T063159Z-5882771aa2ce.tar.age', 'ciphertext_sha256':'a'*64,'ciphertext_bytes':len(b'synthetic ciphertext'),'etag':'"synthetic"','fetched_path':str(self.path), 'independent_owner_retrieval_verified':True} self.admission={name:'a'*64 for name in ('ciphertext_sha256','manifest_sha256','verifier_sha256','trust_sha256','tls_cert_sha256')} self.admission.update(issuer='https://issuer.example',consumer=dict(client_id='existing',secret_env='KEYCAPE_EXISTING_SECRET',audience='existing-api',scope='read',subject='svc:existing',tenant='existing',roles='reader',deny_scope='write'), source_sha256={name:'sha256:'+hashlib.sha256((ROOT/name).read_bytes()).hexdigest() for name in p.SOURCE_PATHS}) def test_complete_public_bindings_admitted_without_reading_ciphertext(self): original=Path.read_bytes def read(path): self.assertNotEqual(path,self.path) return original(path) with patch.object(Path,'read_bytes',read):self.assertTrue(p.validate(self.transfer,self.admission,ROOT)) def test_oversize_metadata_refused_before_any_content_read(self): self.path.write_bytes(b'x'*(128*1024+1)) with patch.object(Path,'read_bytes',side_effect=AssertionError('unbounded read')): with self.assertRaises(p.Refused):p.read_metadata(self.path) def test_duplicate_json_keys_refused_at_every_level(self): for raw in (b'{"issuer":"one","issuer":"two"}',b'{"consumer":{"client_id":"one","client_id":"two"}}'): with self.subTest(raw=raw),self.assertRaises(p.Refused):p.strict_json(raw) def test_bounded_metadata_reader_accepts_exact_limit(self): self.path.write_bytes(b'x'*128) self.assertEqual(p.read_metadata(self.path,128),b'x'*128) with self.assertRaises(p.Refused):p.read_metadata(self.path,127) def test_wrong_archive_binding_refused(self): self.admission['ciphertext_sha256']='b'*64 with self.assertRaises(p.Refused):p.validate(self.transfer,self.admission,ROOT) def test_ciphertext_size_mismatch_refused_without_decryption(self): self.transfer['ciphertext_bytes']+=1 with self.assertRaises(p.Refused):p.validate(self.transfer,self.admission,ROOT) def test_missing_independent_transfer_refused(self): self.transfer['independent_owner_retrieval_verified']=False with self.assertRaises(p.Refused):p.validate(self.transfer,self.admission,ROOT) def test_missing_or_drifted_source_refused(self): for source in p.SOURCE_PATHS: bad=copy.deepcopy(self.admission);bad['source_sha256'][source]='0'*64 with self.subTest(source=source),self.assertRaises(p.Refused):p.validate(self.transfer,bad,ROOT) bad=copy.deepcopy(self.admission);bad['source_sha256'].pop(next(iter(p.SOURCE_PATHS))) with self.assertRaises(p.Refused):p.validate(self.transfer,bad,ROOT) def test_unreviewed_extra_dependency_refused(self): self.admission['source_sha256']['unreviewed.py']='a'*64 with self.assertRaises(p.Refused):p.validate(self.transfer,self.admission,ROOT) def test_ambiguous_or_unsupported_issuer_refused(self): for issuer in ('http://issuer.example','https://u:p@issuer.example','https://issuer.example:8443','https://issuer.example/path','https://issuer.example?x=1','https://issuer.example#fragment'): self.admission['issuer']=issuer with self.subTest(issuer=issuer),self.assertRaises(p.Refused):p.validate(self.transfer,self.admission,ROOT) def test_missing_consumer_or_overlapping_denial_refused(self): self.admission['consumer']['deny_scope']='read' with self.assertRaises(p.Refused):p.validate(self.transfer,self.admission,ROOT) self.admission['consumer'].pop('roles') with self.assertRaises(p.Refused):p.validate(self.transfer,self.admission,ROOT) def test_symlink_private_input_refused_without_read(self): link=self.path.parent/'link';link.symlink_to(self.path) with self.assertRaises(p.Refused):p.metadata_path(link) def test_publicly_readable_private_input_refused(self): self.path.chmod(0o644) with self.assertRaises(p.Refused):p.metadata_path(self.path) def test_bad_admission_digest_never_reads_tls_or_starts_custody(self): admission=self.path.parent/'admission';admission.write_text(json.dumps(self.admission));admission.chmod(0o600) argv=['prepare','--transfer-receipt',str(self.path),'--admission',str(admission),'--admission-sha256','b'*64, '--tls-cert','/never/read-cert','--tls-key','/never/read-key','--trust','/never/read-trust'] with patch('sys.argv',argv),patch('builtins.print') as output: self.assertEqual(p.main(),1) self.assertEqual(output.call_args.args,('ERROR: reviewed metadata/source/dependency preflight refused; no key requested.',)) def test_actual_launcher_bad_digest_stops_before_hidden_prompt(self): master,slave=pty.openpty() try: args=['bash',str(ROOT/'tools/bootstrap-recovery/run-isolated-keycape.sh'),str(self.path),str(self.path),'b'*64,str(self.path),str(self.path),str(self.path)] process=subprocess.Popen(args,stdin=slave,stdout=slave,stderr=slave) process.wait(timeout=5);os.close(slave);slave=None output=b'' while True: try: block=os.read(master,4096) if not block:break output+=block except OSError:break self.assertEqual(process.returncode,1) self.assertIn(b'preflight refused; no key requested.',output) self.assertNotIn(b'Paste custody',output) self.assertNotIn(b'Enter only',output) finally: if slave is not None:os.close(slave) os.close(master) def test_launcher_prepares_before_sealed_descriptor_custody(self): script=(ROOT/'tools/bootstrap-recovery/run-isolated-keycape.sh').read_text() self.assertLess(script.index('python3 tools/bootstrap-recovery/prepare-isolated-keycape.py'),script.index('bash sso-mfa/bootstrap/sops-custody-unlock.sh')) self.assertIn('--sealed-fd --prompt -- python3',script)