292 lines
17 KiB
Python
292 lines
17 KiB
Python
#!/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)
|