"""Generic archive -> immutable installation -> real stdio SDK, inside Computer.""" import hashlib import io import json import os from pathlib import Path import subprocess import time import tarfile import unittest import uuid SERVER = r''' import json,os,sys from pathlib import Path for line in sys.stdin: request=json.loads(line) if 'id' not in request: continue method=request['method'] if method=='initialize': result={'protocolVersion':request['params']['protocolVersion'],'capabilities':{'tools':{}},'serverInfo':{'name':'generic-fixture','version':'1'}} elif method=='tools/list': result={'tools':[{'name':'diagnose','description':'Check binding isolation','inputSchema':{'type':'object'}}]} elif method=='tools/call': args=request['params'].get('arguments',{}) if args.get('delay'): import time Path('inflight').touch() time.sleep(args['delay']) count=Path('count') value=int(count.read_text())+1 if count.exists() else 1 count.write_text(str(value)) def readable(path): try: Path(path).read_bytes(); return True except PermissionError: return False if args.get('fork'): import subprocess,time ready=Path('child-ready') ready.unlink(missing_ok=True) child_code="import ctypes,time; from pathlib import Path; ctypes.CDLL(None).prctl(4,0); Path('child-ready').touch(); time.sleep(30)" subprocess.Popen([sys.executable,'-c',child_code],start_new_session=True,stdin=subprocess.DEVNULL,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL) while not ready.exists(): time.sleep(.005) data={'uid':os.getuid(),'groups':os.getgroups(),'home':os.environ['HOME'],'count':value, 'rootReadable':readable('/var/lib/lazyboy-runner/jobs.sqlite'), 'peerReadable':readable(args['peer']) if args.get('peer') else False, 'token':os.environ.get('LAZYBOY_CONTROL_TOKEN')} result={'content':[{'type':'text','text':json.dumps(data)}]} else: result={} print(json.dumps({'jsonrpc':'2.0','id':request['id'],'result':result}),flush=True) ''' class Packages(unittest.TestCase): def setUp(self): if os.environ.get('LAZYBOY_RUNNER_TEST_CONTAINER') != '1' or not Path('/.dockerenv').exists(): self.fail('requires disposable Computer') self.identity = {'computer_id':'fixture','bot_id':'package-agent','generation':1} self.artifacts = [] def tearDown(self): for path in self.artifacts: path.unlink(missing_ok=True) def call(self, action, body, identity=None, check=True): result = subprocess.run(['python3','-I','/var/lib/lazyboy-runner/runner_jobs.py'], input=json.dumps({'action':action,'identity':identity or self.identity,'input':json.dumps(body)}), text=True,capture_output=True,timeout=55) if check: self.assertEqual(result.returncode,0,result.stdout+result.stderr) return json.loads(result.stdout) def archive(self, unsafe=None, server=SERVER): data=io.BytesIO() with tarfile.open(fileobj=data,mode='w:gz') as archive: member=tarfile.TarInfo('server.py') content=server.encode() member.size=len(content) archive.addfile(member,io.BytesIO(content)) if unsafe: member=tarfile.TarInfo(unsafe) member.type=tarfile.SYMTYPE if unsafe=='link' else tarfile.LNKTYPE if unsafe=='hardlink' else tarfile.REGTYPE member.linkname='/etc/passwd' payload=b'x'*(33*1024*1024) if unsafe=='oversize' else b'' member.size=len(payload) archive.addfile(member,io.BytesIO(payload)) blob=data.getvalue() path=Path('/home/lazyboy')/('package-'+uuid.uuid4().hex+'.tar.gz') path.write_bytes(blob) self.artifacts.append(path) return {'artifact_relative':path.name,'binding_id':uuid.uuid4().hex, 'manifest':{'id':'fixture.'+uuid.uuid4().hex,'version':'1.0.0','sha256':hashlib.sha256(blob).hexdigest(), 'entrypoint':['python3','./server.py'],'share_immutable_package':True}} def invoke(self, binding, args=None, identity=None): reply=self.call('package_call',{'binding_id':binding,'method':'tools/call','name':'diagnose','arguments':args or {}},identity) return json.loads(json.loads(reply['stdout'])['content'][0]['text']) def test_generic_install_private_uid_state_and_exact_replay(self): package=self.archive() self.call('package_install',package) self.call('package_install',package) listed=self.call('package_call',{'binding_id':package['binding_id'],'method':'tools/list'}) self.assertEqual(json.loads(listed['stdout'])[0]['name'],'diagnose') first=self.invoke(package['binding_id'],{'fork':True}) self.assertGreaterEqual(first['uid'],20000) self.assertEqual(first['groups'],[]) self.assertFalse(first['rootReadable']) self.assertIsNone(first['token']) second=self.invoke(package['binding_id']) self.assertEqual(second['count'],2) peer=dict(self.identity,bot_id='other-agent') denied=self.call('package_call',{'binding_id':package['binding_id'],'method':'tools/list'},peer,False) self.assertEqual(denied['error'],'BINDING_UNAVAILABLE') another=dict(package,binding_id=uuid.uuid4().hex) self.call('package_install',another,peer) other=self.invoke(another['binding_id'],{'peer':first['home']+'/count'},peer) self.assertNotEqual(first['uid'],other['uid']) self.assertFalse(other['peerReadable']) self.assertEqual(other['count'],1) for process in Path('/proc').iterdir(): if process.name.isdigit(): try: uid_line=next(line for line in process.joinpath('status').read_text().splitlines() if line.startswith('Uid:')) if int(uid_line.split()[2])==first['uid']: self.assertIn(process.joinpath('stat').read_text().rsplit(')',1)[1].split()[0],('Z','X')) except FileNotFoundError: pass def test_digest_paths_links_and_immutable_version_are_enforced(self): package=self.archive() unknown=dict(package,manifest=dict(package['manifest'],needs_root=True)) self.assertEqual(self.call('package_install',unknown,check=False)['error'],'UNSUPPORTED_MANIFEST_PROFILE') wrong=dict(package,manifest=dict(package['manifest'],sha256='0'*64)) self.assertEqual(self.call('package_install',wrong,check=False)['error'],'PACKAGE_DIGEST_MISMATCH') for path in ('../escape','/tmp/escape','link','hardlink','oversize'): rejected=self.call('package_install',self.archive(path),check=False) self.assertTrue(rejected.get('error')) self.call('package_install',package) changed=dict(package,manifest=dict(package['manifest'],entrypoint=['python3','./server.py','extra'])) self.assertEqual(self.call('package_install',changed,check=False)['error'],'IMMUTABLE_VERSION_CONFLICT') source=Path('/opt/lazyboy-tools')/hashlib.sha256(json.dumps(package['manifest'],sort_keys=True).encode()).hexdigest()/'server.py' source.write_text('changed by root fault injection') rejected=self.call('package_call',{'binding_id':package['binding_id'],'method':'tools/list'},check=False) self.assertEqual(rejected['error'],'PACKAGE_INTEGRITY_FAILURE') def test_private_package_source_is_not_readable_by_peer_binding(self): package=self.archive() package['manifest']['share_immutable_package']=False self.call('package_install',package) self.assertEqual(self.invoke(package['binding_id'])['count'],1) other=dict(package,binding_id=uuid.uuid4().hex) peer=dict(self.identity,bot_id='private-peer') self.call('package_install',other,peer) result=self.invoke(other['binding_id'],{'peer':'/opt/lazyboy-private-tools/'+package['binding_id']+'/server.py'},peer) self.assertFalse(result['peerReadable']) def test_versions_pin_manifest_and_revocation_is_binding_scoped(self): package=self.archive() self.call('package_install',package) another=dict(package,binding_id=uuid.uuid4().hex,manifest=dict(package['manifest'],version='2.0.0')) self.call('package_install',another) wrong=self.call('package_call',{'binding_id':package['binding_id'],'method':'tools/list','expected_version':'2.0.0'},check=False) self.assertEqual(wrong['error'],'STALE_PACKAGE_BINDING') original=self.invoke(package['binding_id']) self.assertEqual(original['count'],1) denied=self.call('package_revoke',{'binding_id':package['binding_id']},dict(self.identity,bot_id='peer'),check=False) self.assertEqual(denied['error'],'BINDING_UNAVAILABLE') self.call('package_revoke',{'binding_id':package['binding_id']}) denied=self.call('package_call',{'binding_id':package['binding_id'],'method':'tools/list'},check=False) self.assertEqual(denied['error'],'BINDING_UNAVAILABLE') self.assertEqual(self.invoke(another['binding_id'])['count'],1) denied=self.call('package_install',package,check=False) self.assertEqual(denied['error'],'BINDING_REVOKED') self.call('package_revoke',{'binding_id':uuid.uuid4().hex}) fresh=dict(package,binding_id=uuid.uuid4().hex) self.call('package_install',fresh) renewed=self.invoke(fresh['binding_id']) self.assertEqual(renewed['count'],1) self.assertNotEqual(original['uid'],renewed['uid']) denied=self.call('package_call',{'binding_id':package['binding_id'],'method':'tools/list'},check=False) self.assertEqual(denied['error'],'BINDING_UNAVAILABLE') def test_atomic_version_switch_pins_calls_and_replays_without_switching_again(self): source=self.archive() source['manifest']['share_immutable_package']=False self.call('package_install',source) original=self.invoke(source['binding_id']) target=self.archive(server=SERVER.replace("'count':value,", "'count':value,'version':'v2',")) target['manifest'].update(id=source['manifest']['id'],version='2.0.0',share_immutable_package=False) self.call('package_install',target) switch={'binding_id':source['binding_id'],'transition_id':uuid.uuid4().hex, 'expected_version':'1.0.0','expected_sha256':source['manifest']['sha256'], 'target_manifest':target['manifest']} before=json.loads(self.call('package_transition_status',switch,dict(self.identity,generation=2))['stdout']) self.assertFalse(before['recorded']) self.assertEqual(before['version'],'1.0.0') payload={'action':'package_call','identity':self.identity,'input':json.dumps({'binding_id':source['binding_id'],'method':'tools/call','name':'diagnose','arguments':{'delay':2}})} worker=subprocess.Popen(['python3','-I','/var/lib/lazyboy-runner/runner_jobs.py'],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True) try: worker.stdin.write(json.dumps(payload));worker.stdin.close();worker.stdin=None deadline=time.monotonic()+5 while not (Path(original['home'])/'inflight').exists() and time.monotonic()