tegwick
5fcd5952ad
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa
292 lines
20 KiB
Python
292 lines
20 KiB
Python
import base64
|
|
import importlib.util
|
|
import json
|
|
import io
|
|
import os
|
|
import tempfile
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
SOURCE=Path(__file__).parents[1]/'tools/bootstrap-recovery/isolated-keycape-session.py'
|
|
spec=importlib.util.spec_from_file_location('session',SOURCE)
|
|
s=importlib.util.module_from_spec(spec);spec.loader.exec_module(s)
|
|
|
|
class SessionTests(unittest.TestCase):
|
|
def test_tmpfs_plan_has_four_exact_original_mounts(self):
|
|
session=s.plan_session()
|
|
self.assertEqual({v['target'] for v in session['volumes'].values()},set(s.RUNROOTS))
|
|
self.assertTrue(all(v['options']==s.VOLUME_OPTIONS for v in session['volumes'].values()))
|
|
self.assertEqual(session['images'][session['issuer']],s.adapter.IMAGE)
|
|
def test_non_tmpfs_driver_options_refused(self):
|
|
session=s.plan_session();name,spec=next(iter(session['volumes'].items()))
|
|
wrong=[{'Name':name,'Driver':'local','Scope':'local','Options':{'type':'none','device':'/host/data'},'Labels':{s.adapter.LABEL:session['run_id']}}]
|
|
with patch.object(s,'run',return_value=json.dumps(wrong).encode()):
|
|
with self.assertRaises(s.adapter.Refused):s.verify_volume(name,spec,session['run_id'])
|
|
def test_owner_secret_is_not_docker_argv_and_import_is_after_inspection(self):
|
|
session=s.plan_session();events=[]
|
|
payload={'files':{'/etc/keycape/config.yaml':base64.b64encode(b'original').decode()},'env':{'KEYCAPE_TEST_SECRET':'fixture-sensitive'}}
|
|
def run(argv,**kwargs):
|
|
events.append(('run',argv,kwargs))
|
|
if argv[-1]=='{{.Id}}':return b'anchor'
|
|
return b''
|
|
with patch.object(s,'run',side_effect=run),patch.object(s,'verify_volume'),patch.object(s.adapter,'inspect_session',side_effect=lambda *args,**kwargs:events.append(('inspect',args,kwargs))):
|
|
s.provision(session,payload,{'/recovery-tls/trust.pem':b'original-trust'},'https://issuer.example')
|
|
joined=' '.join(' '.join(e[1]) for e in events if e[0]=='run')
|
|
self.assertNotIn('fixture-sensitive',joined)
|
|
private_index=next(i for i,e in enumerate(events) if e[0]=='run' and s.IMPORT in e[1])
|
|
inspect_index=next(i for i,e in enumerate(events) if e[0]=='inspect')
|
|
self.assertLess(inspect_index,private_index)
|
|
issuer_call=next(e for e in events if e[0]=='run' and e[1][-1]==s.adapter.IMAGE)
|
|
self.assertEqual(issuer_call[2]['env']['KEYCAPE_TEST_SECRET'],'fixture-sensitive')
|
|
self.assertIn('KEYCAPE_TEST_SECRET',issuer_call[1])
|
|
self.assertNotIn('--entrypoint',issuer_call[1])
|
|
def test_pinned_helper_image_volume_is_explicitly_masked_with_tmpfs(self):
|
|
calls=[]
|
|
def run(argv,**kwargs):calls.append(argv);return b'anchor' if argv[-1]=='{{.Id}}' else b''
|
|
with patch.object(s,'run',side_effect=run),patch.object(s,'verify_volume'),patch.object(s.adapter,'inspect_session'):
|
|
s.provision(s.plan_session(),{'files':{},'env':{}},{},'https://issuer.example')
|
|
worker=next(argv for argv in calls if s.WORKER in argv)
|
|
mask='/privacyidea/etc/persistent:rw,nosuid,nodev,noexec,size=16777216,uid=1000,gid=1000,mode=0700'
|
|
self.assertIn(mask,worker)
|
|
self.assertEqual(worker[worker.index(mask)-1],'--tmpfs')
|
|
issuer=next(argv for argv in calls if argv[-1]==s.adapter.IMAGE)
|
|
self.assertNotIn(mask,issuer)
|
|
def test_public_phase_sequence_reports_containment_before_private_import(self):
|
|
session=s.plan_session();phases=[]
|
|
def run(argv,**kwargs):
|
|
return b'anchor' if argv[-1]=='{{.Id}}' else b''
|
|
with patch.object(s,'run',side_effect=run),patch.object(s,'verify_volume'),patch.object(s.adapter,'inspect_session'):
|
|
s.provision(session,{'files':{},'env':{}},{},'https://issuer.example',phases.append)
|
|
self.assertEqual(phases,list(s.PROVISION_PHASES))
|
|
def test_worker_containment_refusal_keeps_exact_phase_and_skips_import(self):
|
|
phases=[];calls=[]
|
|
def run(argv,**kwargs):calls.append(argv);return b''
|
|
with patch.object(s,'run',side_effect=run),patch.object(s,'verify_volume'),patch.object(s.adapter,'inspect_session',side_effect=s.adapter.Refused):
|
|
with self.assertRaises(s.adapter.Refused):
|
|
s.provision(s.plan_session(),{'files':{},'env':{}},{},'https://issuer.example',phases.append)
|
|
self.assertEqual(phases[-1],'worker_containment')
|
|
self.assertFalse(any(s.IMPORT in argv for argv in calls))
|
|
def test_issuer_containment_refusal_does_not_claim_provision_complete(self):
|
|
phases=[]
|
|
def run(argv,**kwargs):return b'anchor' if argv[-1]=='{{.Id}}' else b''
|
|
with patch.object(s,'run',side_effect=run),patch.object(s,'verify_volume'),patch.object(s.adapter,'inspect_session',side_effect=[None,s.adapter.Refused()]):
|
|
with self.assertRaises(s.adapter.Refused):
|
|
s.provision(s.plan_session(),{'files':{},'env':{}},{},'https://issuer.example',phases.append)
|
|
self.assertEqual(phases[-1],'issuer_containment')
|
|
self.assertNotIn('complete',phases)
|
|
def test_non_original_https_port_refused_before_any_mutation(self):
|
|
with patch.object(s,'run') as run:
|
|
with self.assertRaises(s.adapter.Refused):s.provision(s.plan_session(),{'files':{},'env':{}},{},'http://issuer.example')
|
|
run.assert_not_called()
|
|
def test_container_failure_still_attempts_each_volume_cleanup(self):
|
|
session=s.plan_session();receipt={};removed=[]
|
|
def run(argv,**kwargs):
|
|
if argv[1:3]==['volume','rm']:removed.append(argv[-1])
|
|
if argv[1:3]==['volume','ls']:
|
|
name=next(x for x in argv if x.startswith('name=^')).removeprefix('name=^').removesuffix('$')
|
|
return (name+'\n').encode() if name not in removed else b''
|
|
return b''
|
|
with patch.object(s.adapter,'cleanup',side_effect=s.adapter.Refused),patch.object(s,'verify_volume'),patch.object(s,'run',side_effect=run):
|
|
with self.assertRaises(s.adapter.Refused):s.cleanup_all(session,receipt)
|
|
self.assertEqual(set(removed),set(session['volumes']))
|
|
self.assertFalse(receipt['container_cleanup_verified'])
|
|
self.assertEqual(len(receipt['tmpfs_volume_cleanup_obligations']),4)
|
|
def test_reaper_inherits_no_custody_or_service_environment(self):
|
|
session=s.plan_session();staging={'path':'/dev/shm/registered','run_id':session['run_id']}
|
|
with patch.object(s.subprocess,'Popen') as spawn,patch.object(s,'verify_execution'),patch.object(s.os,'pidfd_open',return_value=99),patch.object(s.os,'close'),patch.object(s,'register_workdir',return_value=staging),patch.object(s.select,'select',side_effect=lambda *args:([spawn.return_value.stdout],[],[])):
|
|
spawn.return_value.stdout.readline.return_value=b'registered\n'
|
|
s.start_reaper(session,staging,{'pid':123})
|
|
kwargs=spawn.call_args.kwargs
|
|
self.assertEqual(set(kwargs['env']),{'PATH','DOCKER_HOST'})
|
|
self.assertTrue(kwargs['close_fds']);self.assertTrue(kwargs['start_new_session']);self.assertEqual(kwargs['pass_fds'],(99,))
|
|
|
|
class TargetBindingTests(unittest.TestCase):
|
|
def test_every_direct_docker_command_is_explicitly_bound(self):
|
|
with patch.dict('os.environ',{},clear=True),patch.object(s.adapter.subprocess,'run',return_value=__import__('subprocess').CompletedProcess([],0,b'ok')) as run:
|
|
self.assertEqual(s.run(['docker','info']),b'ok')
|
|
argv=run.call_args.args[0]
|
|
self.assertEqual(argv[:3],['docker','--host',s.adapter.ENDPOINT])
|
|
self.assertEqual(run.call_args.kwargs['env']['DOCKER_HOST'],s.adapter.ENDPOINT)
|
|
def test_nondefault_unix_socket_is_refused_before_any_docker_call(self):
|
|
with patch.dict('os.environ',{'DOCKER_HOST':'unix:///tmp/other-daemon.sock'},clear=True),patch.object(s.adapter.subprocess,'run') as run:
|
|
with self.assertRaises(s.adapter.Refused):s.run(['docker','info'])
|
|
run.assert_not_called()
|
|
def test_conflicting_context_is_refused(self):
|
|
with patch.dict('os.environ',{'DOCKER_CONTEXT':'production'},clear=True),patch.object(s.adapter.subprocess,'run') as run:
|
|
with self.assertRaises(s.adapter.Refused):s.run(['docker','info'])
|
|
run.assert_not_called()
|
|
def test_conflicting_per_call_host_is_refused(self):
|
|
with patch.dict('os.environ',{},clear=True),patch.object(s.adapter.subprocess,'run') as run:
|
|
with self.assertRaises(s.adapter.Refused):s.run(['docker','info'],env={'DOCKER_HOST':'unix:///tmp/other.sock'})
|
|
run.assert_not_called()
|
|
def test_reused_preflight_uses_same_bound_command(self):
|
|
self.assertIs(s.local.command,s.adapter.command)
|
|
def test_probe_is_registered_before_session_creation(self):
|
|
session=s.plan_session()
|
|
self.assertEqual(session['public_image_probe'],'nk49-kc-'+session['run_id']+'-image-probe')
|
|
self.assertEqual(session['docker_endpoint'],s.adapter.ENDPOINT)
|
|
|
|
class PlaintextLifecycleTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
|
|
self.root=Path(self.tmp.name)
|
|
self.override=patch.object(s,'STAGING_ROOT',self.root);self.override.start();self.addCleanup(self.override.stop)
|
|
self.run_id='a'*32
|
|
self.work=self.root/('nk49-keycape-'+self.run_id+'-abcdefgh');self.work.mkdir(mode=0o700)
|
|
self.staging=s.register_workdir(self.work,self.run_id)
|
|
def test_exact_plaintext_directory_removed_and_recorded(self):
|
|
(self.work/'archive.tar').write_bytes(b'synthetic-private-archive')
|
|
restored=self.work/'restored';restored.mkdir(mode=0o700);(restored/'fixture').write_bytes(b'synthetic')
|
|
receipt={};s.cleanup_workdir(self.staging,receipt)
|
|
self.assertFalse(self.work.exists());self.assertTrue(receipt['temporary_plaintext_cleanup_verified'])
|
|
self.assertEqual(receipt['plaintext_cleanup_obligation']['status'],'absence_verified')
|
|
def test_docker_failure_does_not_skip_plaintext_cleanup(self):
|
|
(self.work/'archive.tar').write_bytes(b'synthetic')
|
|
receipt={}
|
|
with patch.object(s,'cleanup_all',side_effect=s.adapter.Refused):
|
|
with self.assertRaises(s.adapter.Refused):s.recover_obligations({},self.staging,receipt)
|
|
self.assertFalse(self.work.exists());self.assertTrue(receipt['temporary_plaintext_cleanup_verified'])
|
|
def test_replacement_inode_is_not_removed(self):
|
|
# Preserve original inode while replacing its registered pathname.
|
|
old=self.root/'retained-original';self.work.rename(old);self.work.mkdir(mode=0o700)
|
|
(self.work/'replacement').write_bytes(b'not-owned-by-this-obligation');receipt={}
|
|
with self.assertRaises(s.adapter.Refused):s.cleanup_workdir(self.staging,receipt)
|
|
self.assertTrue((self.work/'replacement').exists());self.assertTrue(old.exists())
|
|
self.assertEqual(receipt['plaintext_cleanup_obligation']['status'],'identity_mismatch_not_removed')
|
|
def test_symlink_root_is_not_followed(self):
|
|
self.work.rmdir();outside=self.root/'outside';outside.mkdir();(outside/'kept').write_bytes(b'keep')
|
|
self.work.symlink_to(outside,target_is_directory=True)
|
|
with self.assertRaises(s.adapter.Refused):s.cleanup_workdir(self.staging,{})
|
|
self.assertTrue((outside/'kept').exists())
|
|
def test_descendant_symlink_is_not_followed(self):
|
|
outside=self.root/'kept';outside.write_bytes(b'keep');(self.work/'link').symlink_to(outside)
|
|
with self.assertRaises(s.adapter.Refused):s.cleanup_workdir(self.staging,{})
|
|
self.assertEqual(outside.read_bytes(),b'keep')
|
|
def test_wrong_owner_identity_is_not_removed(self):
|
|
changed={**self.staging,'owner_uid':os.getuid()+1}
|
|
with self.assertRaises(s.adapter.Refused):s.cleanup_workdir(changed,{})
|
|
self.assertTrue(self.work.exists())
|
|
def test_unregistered_path_refused(self):
|
|
with self.assertRaises(s.adapter.Refused):s.cleanup_workdir({**self.staging,'path':str(self.root)}, {})
|
|
|
|
@unittest.skipUnless(hasattr(os,'pidfd_open') and hasattr(signal,'pidfd_send_signal'),'Linux pidfd required')
|
|
class ExecutionDeadlineTests(unittest.TestCase):
|
|
def start_synthetic_owner(self,root):
|
|
late=str(root/'late-creation')
|
|
code="""import importlib.util,json,subprocess,sys,time
|
|
spec=importlib.util.spec_from_file_location('session',sys.argv[1]);s=importlib.util.module_from_spec(spec);spec.loader.exec_module(s)
|
|
execution=s.isolate_execution()
|
|
child=subprocess.Popen([sys.executable,'-c',"import pathlib,sys,time;time.sleep(2);pathlib.Path(sys.argv[1]).write_text('late')",sys.argv[2]])
|
|
print(json.dumps({'execution':execution,'child_pid':child.pid}),flush=True)
|
|
time.sleep(20)
|
|
"""
|
|
parent=subprocess.Popen([sys.executable,'-c',code,str(SOURCE),late],stdout=subprocess.PIPE,stderr=subprocess.DEVNULL)
|
|
message=json.loads(parent.stdout.readline());parent.stdout.close()
|
|
def finish():
|
|
if parent.poll() is None:parent.kill()
|
|
parent.wait(timeout=5)
|
|
try:
|
|
before=s.process_facts(message['child_pid'])
|
|
if before is None or before['sid']!=message['execution']['sid'] or before['pgid']!=message['execution']['pgid']:return
|
|
fd=os.pidfd_open(message['child_pid'])
|
|
try:
|
|
after=s.process_facts(message['child_pid'])
|
|
if after is not None and all(after[k]==before[k] for k in ('pid','sid','pgid','start_ticks','uid')):
|
|
signal.pidfd_send_signal(fd,signal.SIGKILL)
|
|
except ProcessLookupError:pass
|
|
finally:os.close(fd)
|
|
except ProcessLookupError:pass
|
|
self.addCleanup(finish)
|
|
return parent,message['execution'],Path(late)
|
|
def test_deadline_quiesces_only_fresh_owner_group_before_cleanup(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent,execution,late=self.start_synthetic_owner(Path(directory));fd=os.pidfd_open(parent.pid)
|
|
try:
|
|
receipt={};s.quiesce_execution(execution,fd,receipt)
|
|
self.assertTrue(receipt['execution_quiescence_verified'])
|
|
self.assertFalse(s.group_members(execution));self.assertFalse(late.exists())
|
|
finally:os.close(fd)
|
|
def test_hard_parent_loss_stops_surviving_marker_creator(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
parent,execution,late=self.start_synthetic_owner(Path(directory));fd=os.pidfd_open(parent.pid)
|
|
try:
|
|
signal.pidfd_send_signal(fd,signal.SIGKILL)
|
|
poll=__import__('select').poll();poll.register(fd,__import__('select').POLLIN);self.assertTrue(poll.poll(2000))
|
|
receipt={};s.quiesce_execution(execution,fd,receipt)
|
|
self.assertTrue(receipt['execution_quiescence_verified']);self.assertFalse(late.exists())
|
|
self.assertFalse(s.group_members(execution))
|
|
finally:os.close(fd)
|
|
def test_expired_execution_cannot_start_any_docker_command(self):
|
|
binding={'pid':os.getpid(),'sid':os.getsid(0),'pgid':os.getpgrp(),'deadline':time.monotonic()-1,'staging':'/tmp/not-created'}
|
|
with patch.object(s.adapter,'EXECUTION',binding),patch.object(s.adapter.subprocess,'run') as run:
|
|
with self.assertRaises(s.adapter.Refused):s.run(['docker','create','fixed-image'])
|
|
run.assert_not_called()
|
|
def test_reused_or_foreign_session_refuses_signals(self):
|
|
binding={'pid':123,'sid':123,'pgid':123,'start_ticks':10,'uid':os.getuid()}
|
|
wrong={**binding,'sid':456,'state':'S'}
|
|
with patch.object(s,'process_facts',return_value=wrong),patch.object(s.signal,'pidfd_send_signal') as send:
|
|
with self.assertRaises(s.adapter.Refused):s.quiesce_execution(binding,99,{})
|
|
send.assert_not_called()
|
|
def test_inflight_uncertainty_survives_plaintext_directory_removal(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path=Path(directory)/'owned-stage';path.mkdir();marker=Path(str(path)+'-docker-inflight');marker.touch()
|
|
staging={'path':str(path)};self.assertTrue(s.daemon_uncertain(staging));path.rmdir()
|
|
self.assertTrue(s.daemon_uncertain(staging))
|
|
|
|
class ReaperOrderingTests(unittest.TestCase):
|
|
def exercise(self,quiescence_fails=False):
|
|
events=[];records=[]
|
|
def quiet(*args):
|
|
events.append('quiesce')
|
|
if quiescence_fails:raise s.adapter.Refused()
|
|
args[-1]['execution_quiescence_verified']=True
|
|
def recover(*args):events.append('cleanup')
|
|
with patch.object(s,'verify_execution'),patch.object(s,'pidfd_dead',return_value=False),patch.object(s.select,'poll') as poll,patch.object(s,'quiesce_execution',side_effect=quiet),patch.object(s,'daemon_uncertain',return_value=True),patch.object(s,'recover_obligations',side_effect=recover),patch.object(s.os,'close'),patch.object(s.os,'open',return_value=99),patch.object(s.os,'fdopen'),patch.object(s.json,'dump',side_effect=lambda r,*args:records.append(dict(r))),patch('sys.stdout',io.StringIO()):
|
|
poll.return_value.poll.return_value=[]
|
|
s.reaper_main({}, {'path':'/dev/shm/exact-staging'}, {'deadline':time.monotonic()-1},77)
|
|
return events,records[0]
|
|
def test_daemon_uncertainty_survives_verified_resource_readback(self):
|
|
events,receipt=self.exercise();self.assertEqual(events,['quiesce','cleanup'])
|
|
self.assertTrue(receipt['execution_quiescence_verified']);self.assertTrue(receipt['in_flight_daemon_request_uncertain'])
|
|
self.assertEqual(receipt['status'],'bounded_cleanup_uncertain')
|
|
def test_unverified_quiescence_prevents_any_cleanup(self):
|
|
events,receipt=self.exercise(True);self.assertEqual(events,['quiesce'])
|
|
self.assertFalse(receipt['execution_quiescence_verified']);self.assertEqual(receipt['status'],'bounded_cleanup_uncertain')
|
|
def test_checkpoint_rewrites_one_valid_json_document(self):
|
|
with tempfile.TemporaryFile() as stream:
|
|
s.persist_receipt(stream.fileno(),{'status':'registered','long':'x'*100})
|
|
s.persist_receipt(stream.fileno(),{'status':'failed'})
|
|
stream.seek(0);self.assertEqual(json.load(stream),{'status':'failed'})
|
|
def test_nonzero_docker_result_retains_uncertainty_before_inflight_clear(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
staging=str(Path(directory)/'stage');Path(staging).mkdir()
|
|
binding={'pid':os.getpid(),'sid':os.getsid(0),'pgid':os.getpgrp(),'deadline':time.monotonic()+10,'staging':staging}
|
|
def returned(*args,**kwargs):
|
|
self.assertTrue(Path(staging+'-docker-inflight').exists())
|
|
return subprocess.CompletedProcess(args[0],1,b'')
|
|
with patch.object(s.adapter,'EXECUTION',binding),patch.object(s.adapter.subprocess,'run',side_effect=returned):
|
|
with self.assertRaises(s.adapter.Refused):s.run(['docker','create','synthetic-already-accepted'])
|
|
self.assertFalse(Path(staging+'-docker-inflight').exists());self.assertTrue(s.daemon_uncertain({'path':staging}))
|
|
def test_only_policy_mount_is_writable_in_issuer(self):
|
|
plan=s.plan_session();args=s.mounts(plan,readonly=True)
|
|
mounts=args[1::2]
|
|
for value in mounts:
|
|
self.assertEqual(not value.endswith(',readonly'),'target=/var/lib/keycape-policy' in value)
|
|
def test_docker_timeout_retains_independent_uncertainty_marker(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
staging=str(Path(directory)/'stage');Path(staging).mkdir()
|
|
binding={'pid':os.getpid(),'sid':os.getsid(0),'pgid':os.getpgrp(),'deadline':time.monotonic()+10,'staging':staging}
|
|
def timeout(*args,**kwargs):
|
|
self.assertTrue(Path(staging+'-docker-inflight').exists())
|
|
raise subprocess.TimeoutExpired('synthetic Docker',1)
|
|
with patch.object(s.adapter,'EXECUTION',binding),patch.object(s.adapter.subprocess,'run',side_effect=timeout):
|
|
with self.assertRaises(subprocess.TimeoutExpired):s.run(['docker','create','fixed-image'])
|
|
self.assertFalse(Path(staging+'-docker-inflight').exists())
|
|
self.assertTrue(Path(staging+'-docker-uncertain').exists());self.assertTrue(s.daemon_uncertain({'path':staging}))
|
|
|
|
if __name__=='__main__':unittest.main()
|