#!/usr/bin/env python3 """Production state/protocol regressions. OS adapters never call host services.""" import array import copy import importlib.util import json import os from pathlib import Path import socket import struct import sys import tempfile import unittest from unittest.mock import patch ROOT=Path(__file__).resolve().parents[2] sys.path.insert(0,str(ROOT/'scripts')) import d09_handoff as handoff import c01_d09 as transfer import c01_access_generation as generation import c01_access_policy as policy spec=importlib.util.spec_from_file_location('corrections',Path(__file__).with_name('d09-review-corrections.py')) corrections=importlib.util.module_from_spec(spec);spec.loader.exec_module(corrections) owner,store,wire=corrections.owner,corrections.store,corrections.wire class ProtocolTests(unittest.TestCase): def test_retiring_old_session_cannot_revoke_new_start_scope(self): import threading from types import SimpleNamespace from unittest.mock import Mock h=handoff.Handoff.__new__(handoff.Handoff) h.lock=threading.RLock();h.valid=True;h.expected={} old,new=Mock(),Mock();h.session=new h.retire(old) self.assertIs(h.session,new);self.assertTrue(h.valid);new.close.assert_not_called() h.owner=SimpleNamespace(binding={'c01':{'generations':{'old':{}},'application':{},'runtime':{}}}) h.expect('old') new.close.assert_called_once();self.assertIsNone(h.session);self.assertTrue(h.valid) h.retire(new) self.assertTrue(h.valid) def test_socketpair_creator_is_not_packet_sender(self): left,right=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET) for channel in (left,right):channel.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1) creator=struct.unpack('3i',left.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12)) pid=os.fork() if pid==0: try: left.close();right.send(wire.canonical({'native':True}));os._exit(0) except BaseException:os._exit(1) try: value,rights=handoff.receive(left,(pid,os.getuid(),os.getgid())) self.assertEqual(value,{'native':True});self.assertEqual(rights,[]) self.assertEqual(creator[0],os.getpid());self.assertNotEqual(creator[0],pid) self.assertEqual(os.waitpid(pid,0)[1],0) finally:left.close();right.close() def test_surplus_rights_and_wrong_credentials_close_received_fds(self): for receive in (handoff.receive,transfer.receive): a,b=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET) a.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1) baseline=len(os.listdir('/proc/self/fd')) for expected,nfds in [((os.getpid(),os.getuid(),os.getgid()),2),((os.getpid()+1,os.getuid(),os.getgid()),1)]: b.sendmsg([wire.canonical({'method':'D09_FD'})],[(socket.SOL_SOCKET,socket.SCM_RIGHTS,array.array('i',[b.fileno()]*nfds))]) with self.assertRaises((ValueError,PermissionError)): receive(a,expected,1) self.assertEqual(len(os.listdir('/proc/self/fd')),baseline) a.close();b.close() class LifecycleTests(unittest.TestCase): def setUp(self): self.case=corrections.Corrections();self.case.setUp() self.o,self.root=self.case.o,self.case.root def tearDown(self):self.case.tearDown() def test_pending_start_is_nonblocking_permit_retained_and_never_resubmitted(self): self.case.prerequisite_fixture() original=owner.command.side_effect def delayed(*args): result=original(*args) if args[1]=='start':self.case.state['ActiveState']='activating' return result with patch.object(owner,'command',side_effect=delayed): self.assertFalse(self.o.activate_prerequisite()['prerequisite_started']) self.assertTrue(owner.PERMIT.exists()) before=self.o.store.raw for _ in range(3): self.assertFalse(self.o.activate_prerequisite()['prerequisite_started']) inspection=self.o.inspect(dict(schema_version=1,target_binding_digest=self.case.binding['target_binding_digest'],transaction_id=None,challenge='d'*64),self.case.case.peer) self.assertEqual(inspection['status'],'NO_D09_TRANSACTION') self.assertEqual(before,self.o.store.raw) self.assertEqual(sum(c[1]=='start' for c in self.case.commands),1) self.assertEqual(self.o.submit(self.case.case.request(),self.case.case.peer)['reason'],'DISPATCH_PENDING') self.case.state['ActiveState']='active' self.assertTrue(self.o.observe_prerequisite()['prerequisite_started']) self.assertFalse(owner.PERMIT.exists()) def test_old_failed_state_does_not_remove_permit_before_submission_ack(self): self.case.prerequisite_fixture() self.case.state['ActiveState']='failed' original=owner.command.side_effect def evaluated_failure(*args): if args[1]=='start': self.case.commands.append(args) self.assertFalse(self.o.observe_prerequisite()['prerequisite_started']) self.assertTrue(owner.PERMIT.exists()) return '' # accepted job has now been evaluated and failed return original(*args) with patch.object(owner,'command',side_effect=evaluated_failure): self.assertFalse(self.o.activate_prerequisite()['prerequisite_started']) self.assertFalse(owner.PERMIT.exists()) self.assertEqual(self.o.store.value['prerequisite']['state'],'START_UNCERTAIN') self.assertEqual(self.o.store.value['prerequisite']['submission'],'ACKNOWLEDGED') with self.assertRaises(PermissionError):self.o.activate_prerequisite() self.assertEqual(sum(c[1]=='start' for c in self.case.commands),1) def test_lookup_cross_invocation_is_nonmutating_and_does_not_relax_submit(self): req=self.case.case.request() acceptance=self.case.case.submit(req)['acceptance'] import types self.o.authorize=types.MethodType(owner.Owner.authorize,self.o) before=self.o.store.raw lookup=dict(schema_version=1,challenge='d'*64,request=req,request_digest=wire.digest(wire.canonical(req))) response=self.o.lookup(lookup) wire.validate('lookup_response',response) self.assertEqual(response['acceptance'],acceptance) self.assertEqual(response['status'],'FOUND') self.assertEqual(self.o.submit(req,self.case.case.peer)['reason'],'AUTHORIZATION') self.assertEqual(before,self.o.store.raw) changed=dict(req,expected_sequence=123) with self.assertRaises(PermissionError): self.o.lookup(dict(lookup,request=changed,request_digest=wire.digest(wire.canonical(changed)))) unknown=dict(req,request_id='e'*32) self.assertEqual(self.o.lookup(dict(lookup,request=unknown,request_digest=wire.digest(wire.canonical(unknown))))['status'],'UNKNOWN') self.assertEqual(before,self.o.store.raw) def test_acknowledged_dependency_failure_fences_without_redispatch(self): self.case.prerequisite_fixture() original=owner.command.side_effect def blocked(*args): if args[1]=='start': self.case.commands.append(args) self.case.state.update(ActiveState='inactive',MainPID='0',Job='1439') return '' return original(*args) with patch.object(owner,'command',side_effect=blocked): self.assertFalse(self.o.activate_prerequisite()['prerequisite_started']) self.assertEqual(self.o.store.value['prerequisite']['state'],'START_INTENT') self.assertTrue(owner.PERMIT.exists()) self.case.state['Job']='' self.assertFalse(self.o.observe_prerequisite()['prerequisite_started']) value=copy.deepcopy(self.o.store.value) self.assertEqual(value['prerequisite']['state'],'START_UNCERTAIN') self.assertIsNone(value['receipt']) self.assertFalse(owner.PERMIT.exists()) with self.assertRaises(PermissionError):self.o.activate_prerequisite() self.o.observe_prerequisite() self.assertEqual(self.o.store.value,value) self.assertEqual(sum(c[1]=='start' for c in self.case.commands),1) def test_lookup_after_successor_returns_original_acceptance_and_separate_current_head(self): req=self.case.case.request();a=self.case.case.submit(req)['acceptance'] v=copy.deepcopy(self.o.store.value) v['consumed']+=v['inbox'];v['inbox']=[] r=dict(v['receipt'],phase='COMPLETE',outcome='COMPLETE',config_action='APPLY', executor=dict(pid=123,start_ticks=1,boot_id='3'*32,incarnation_id='4'*32),observed_invocation_id='7'*32, namespace_observation_digest='a'*64,preservation_observation_digest='b'*64, effective_config_observation_digest='c'*64,config_pre_digest='d'*64,config_post_digest='e'*64,config_current_digest='e'*64) self.o.store.commit(v,r) b=dict(self.case.binding,transaction_id='8'*32,admission_epoch_id='9'*32,composition_digest='f'*64) self.o.store.commit_successor(b,'e'*64,v['prerequisite']);self.o.binding=b before=self.o.store.raw result=self.o.lookup(dict(schema_version=1,challenge='d'*64,request=req,request_digest=wire.digest(wire.canonical(req)))) self.assertEqual(result['acceptance'],a) self.assertEqual(result['current']['status'],'NO_D09_TRANSACTION') self.assertIsNone(result['current']['receipt']) self.assertEqual(before,self.o.store.raw) class GenerationTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory(prefix='d09-access-generation-') self.root=Path(self.temp.name) self.launch=self.root/'launch.json';self.head=self.root/'generation-head.json';self.records=self.root/'records' self.records.mkdir(mode=0o700) self.workspace=self.root/'60-terminal-workspace.conf';self.stage=self.root/'.d09-staging';self.stage.mkdir(mode=0o700) def protected(path,*args,**kwargs): fd=corrections.fixture.fixture_secure(path,kwargs.get('directory',False));os.close(fd);return Path(path).lstat() def atomic(path,raw,*args): Path(path).write_bytes(raw);Path(path).chmod(0o600) self.patches=[patch.object(generation,'HEAD',self.head),patch.object(generation,'DIRECTORY',self.records), patch.object(generation,'WORKSPACE',self.workspace),patch.object(generation,'STAGING',self.stage), patch.object(policy,'LAUNCH',self.launch),patch.object(policy,'read',side_effect=lambda p,*a:Path(p).read_bytes()), patch.object(policy,'protected',side_effect=protected),patch.object(policy,'atomic',side_effect=atomic)] for p in self.patches:p.start() self.a={'version':3,'native_sha256':'a'*64,'application_user':'protected','authority_user':'authority','runtime_uid':1000,'runtime_gid':1000, **{k:'/fixture/'+k for k in ('manager_backing','snapshot_backing','authority_backing','runtime_data_dir','snapshot_key','authority_key','platform_profile','proxy_path')}, 'binding':{'local_identity':[1]*32,'incarnation':[2]*32,'reader_schema':2,'writer_schema':2}, 'platform_profile_sha256':'f'*64,'clock_domain':'f'*32,'host_ip':'192.0.2.1'} self.launch.write_bytes(generation.canonical(self.a));self.launch.chmod(0o600) self.policy={'launch_sha256':policy.sha(self.launch.read_bytes()),'runtime_sha256':'d'*64, 'd09_workspace':{'path':str(self.workspace),'initial_digest':policy.sha(b'ABSENT')}} self.controller=generation.Generations(self.policy,'e'*64,self.a,self.policy['launch_sha256']) def tearDown(self): for p in reversed(self.patches):p.stop() self.temp.cleanup() def record(self,before,after,token): r=dict(version=1,policy_sha256='e'*64,runtime_sha256='d'*64,transaction_id=token*32,action='APPLY',grant_digest=token*64, from_launch=before,to_launch=after,from_launch_sha256=policy.sha(generation.canonical(before)), to_launch_sha256=policy.sha(generation.canonical(after)),external_history_digest='f'*64, from_workspace_digest=policy.sha(b'ABSENT'),to_workspace_digest=policy.sha(b'ABSENT')) raw=generation.canonical(r);hashed=policy.sha(raw);(self.records/(hashed+'.json')).write_bytes(raw) return {'protocol':'d09-access-generation-v1','method':'begin','record_digest':hashed} def test_a_b_c_b_changes_only_appointed_launch_preserves_runtime_and_replay_history(self): b=dict(self.a,native_sha256='b'*64);c=dict(self.a,native_sha256='c'*64) original=copy.deepcopy(self.policy) for before,after,token in [(self.a,b,'1'),(b,c,'2'),(c,b,'3')]: req=self.record(before,after,token) with self.assertRaises(ValueError):self.controller.handle(req,False) self.assertEqual(self.controller.handle(req,True)['state'],'PENDING') # Lost begin acknowledgement is read, not repeated BEGIN. self.assertEqual(self.controller.handle(dict(req,method='inspect'),False)['state'],'PENDING') with self.assertRaises(ValueError):self.controller.handle(req,True) self.launch.write_bytes(generation.canonical(after)) self.controller.workspace_exclusions() self.assertEqual(self.controller.handle(dict(req,method='commit'),True)['state'],'COMMITTED') head=self.head.read_bytes() self.assertEqual(self.controller.handle(dict(req,method='inspect'),False)['state'],'COMMITTED') self.assertEqual(self.head.read_bytes(),head) self.assertEqual(self.policy,original) self.assertEqual(len(self.controller.value['consumed']),3) self.assertEqual(len(self.controller.value['journal']),6) def test_changed_roles_unknown_workspace_and_unpublished_launch_refuse(self): bad=dict(self.a,native_sha256='b'*64,runtime_uid=999) with self.assertRaises(ValueError):self.controller.handle(self.record(self.a,bad,'1'),True) b=dict(self.a,native_sha256='b'*64);req=self.record(self.a,b,'2') self.controller.handle(req,True) with self.assertRaises(ValueError):self.controller.handle(dict(req,method='commit'),True) self.workspace.write_bytes(b'foreign unit');self.workspace.chmod(0o600) with self.assertRaises(ValueError):self.controller.workspace_exclusions() def test_checker_refresh_requires_parent_owned_exact_head(self): req=self.record(self.a,dict(self.a,native_sha256='b'*64),'1');self.controller.handle(req,True) with self.assertRaises(ValueError):self.controller.refresh('0'*64) self.controller.refresh(policy.sha(self.head.read_bytes())) corrupted=copy.deepcopy(self.controller.value);corrupted['consumed']=['f'*64] with self.assertRaises(ValueError):self.controller.validate_head(corrupted) corrupted=copy.deepcopy(self.controller.value);corrupted['version']=True with self.assertRaises(ValueError):self.controller.validate_head(corrupted) def test_admitted_rename_gap_does_not_revoke_unchanged_runtime_policy(self): p=policy.Policy.__new__(policy.Policy) p.value=dict(self.policy,files={});p.digest='e'*64;p.generations=self.controller original=policy.json_read def read_json(path): if path==policy.RUNTIME:return {},'d'*64 if path==policy.CONFIG:return {},'e'*64 return original(path) with patch.object(policy,'json_read',side_effect=read_json), patch.object(policy,'TARGETS',{}): p.verify_files() req=self.record(self.a,dict(self.a,native_sha256='b'*64),'1') self.controller.handle(req,True) self.launch.unlink() p.verify_files() # exact absent interval is authorized only by pending self.launch.write_bytes(generation.canonical(dict(self.a,native_sha256='f'*64))) with self.assertRaises(ValueError):p.verify_files() self.launch.write_bytes(generation.canonical(dict(self.a,native_sha256='b'*64))) self.controller.handle(dict(req,method='commit'),True) p.verify_files() self.launch.unlink() with self.assertRaises(ValueError):p.verify_files() class PackageTests(unittest.TestCase): def test_all_runtime_helpers_are_in_reproducible_immutable_package(self): spec=importlib.util.spec_from_file_location('packager',ROOT/'scripts/package-d09-c01.py') packager=importlib.util.module_from_spec(spec);spec.loader.exec_module(packager) import tarfile with tempfile.TemporaryDirectory() as tmp: a,b=Path(tmp)/'a.tar',Path(tmp)/'b.tar' manifest=packager.package(a);packager.package(b) self.assertEqual(a.read_bytes(),b.read_bytes()) for name in ('d09_handoff.py','d09_c01.py','c01_d09.py','c01_access_generation.py','c01_runtime_view.py'): self.assertIn('usr/libexec/'+name,manifest) with tarfile.open(a) as archive: for path,hashed in manifest.items(): self.assertEqual(wire.digest(archive.extractfile(path).read()),hashed) self.assertEqual(archive.getmember(path).uid,0) if __name__=='__main__':unittest.main(verbosity=2)