tegwick
4f65e41ac5
Some checks failed
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa
108 lines
7 KiB
Python
108 lines
7 KiB
Python
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)
|