import hashlib,json,os,sqlite3,subprocess,sys,tempfile,unittest,types,stat,socket from pathlib import Path from unittest.mock import patch,Mock,MagicMock sys.path.insert(0,str(Path(__file__).resolve().parents[1])) import c01_retained_producer as retained class RetainedTests(unittest.TestCase): def selector(self): return dict(kind='retained-container-v1',unit='archipelago-retained-fixture-'+('a'*64)+'.service', container_id='a'*64,image_id='b'*64,config_sha256='c'*64, scope_kind='bound-libpod-scope',payload_principal=[1000,1000]) def test_exact_id_scope_and_keeper_unit_cannot_be_selected_by_name_only(self): value=self.selector();group=retained.PREFIX+'/app.slice/'+value['unit'] retained.validate(value,group) self.assertEqual(retained.payload_group(value,group),retained.PREFIX+'/user.slice/libpod-'+value['container_id']+'.scope/container') for change in ({'container_id':'d'*64},{'unit':'other.service'},{'scope_kind':'historical-pid-scope'}, {'payload_principal':[0,0]},{'command':['podman','run']}): with self.subTest(change=change),self.assertRaises(ValueError):retained.validate({**value,**change},group) def test_complete_config_and_running_state_are_required(self): value=self.selector() config=dict(rootfsImageID=value['image_id'],cgroupsMode='enabled',cgroupParent='user.slice',environment=['PRIVATE_FIXTURE=original']) value['config_sha256']=hashlib.sha256(json.dumps(config,sort_keys=True,separators=(',',':'),ensure_ascii=False).encode()).hexdigest() with tempfile.TemporaryDirectory() as directory: path=Path(directory)/'db.sql';db=sqlite3.connect(path) db.execute('CREATE TABLE ContainerConfig (ID TEXT, JSON TEXT)');db.execute('CREATE TABLE ContainerState (ID TEXT, JSON TEXT)') db.execute('INSERT INTO ContainerConfig VALUES (?,?)',(value['container_id'],json.dumps(config))) db.execute('INSERT INTO ContainerState VALUES (?,?)',(value['container_id'],json.dumps({'state':3,'pid':os.getpid()})));db.commit() with patch.object(retained,'DATABASE','file:'+str(path)+'?mode=ro'): self.assertEqual(retained.stored(value),os.getpid()) config['environment']=['PRIVATE_FIXTURE=changed'] db.execute('UPDATE ContainerConfig SET JSON=?',(json.dumps(config),));db.commit() with self.assertRaisesRegex(ValueError,'config/image'):retained.stored(value) db.close() def test_node_labels_are_data_but_unit_and_container_remain_exact(self): value=self.selector();value['unit']='archipelago-retained-node-a-'+value['container_id']+'.service' group=retained.PREFIX+'/app.slice/'+value['unit'];retained.validate(value,group) for label in ('../node','-node','node-','a'*65): invalid={**value,'unit':'archipelago-retained-'+label+'-'+value['container_id']+'.service'} with self.subTest(label=label),self.assertRaises(ValueError): retained.validate(invalid,retained.PREFIX+'/app.slice/'+invalid['unit']) def test_keeper_death_revokes_even_with_a_live_socket_owner(self): keeper=subprocess.Popen(['sleep','30']) guard=retained.Guard.__new__(retained.Guard);guard.fds=[] try: first=guard.hold(os.pidfd_open(keeper.pid));second=guard.hold(os.pidfd_open(os.getpid())) guard.processes=[{'pid':keeper.pid},{'pid':os.getpid()}] guard.namespace_bindings=[{'pidfd':first,'values':{}},{'pidfd':second,'values':{}}] guard.process=lambda pid:{'pid':pid};guard.next_check=float('inf') guard.check();keeper.terminate();keeper.wait() with self.assertRaisesRegex(ValueError,'lifetime lost'):guard.check() finally: if keeper.poll() is None:keeper.kill();keeper.wait() guard.close() def owner_fixture(self): guard=retained.Guard.__new__(retained.Guard) guard.value={**self.selector(),'payload_principal':[100099,100100]} guard.group=retained.PREFIX+'/app.slice/'+guard.value['unit'] guard.payload_group=retained.payload_group(guard.value,guard.group) guard.payload_net=(4,2) guard.policy=types.SimpleNamespace(value={'namespaces':{'runtime':[4,1]}}) actor={'pid':42,'group':guard.group,'uids':[1000]*4,'gids':[1000]*4,'net':(4,2)} return guard,actor def test_cross_namespace_helper_requires_protected_forwarder(self): guard,actor=self.owner_fixture() with self.assertRaisesRegex(ValueError,'unverified retained forwarder'):guard.assert_owner(actor) guard.forwarder=(actor['pid'],1,2) guard.assert_owner(actor) guard.forwarder=(actor['pid']+1,1,2) with self.assertRaisesRegex(ValueError,'unverified retained forwarder'):guard.assert_owner(actor) def test_cross_namespace_exception_cannot_admit_other_scope_principal_or_realm(self): guard,actor=self.owner_fixture() guard.forwarder=(actor['pid'],1,2) changes=({'uids':[100099]*4},{'gids':[100099]*4},{'group':retained.PREFIX+'/other.scope'}, {'group':guard.payload_group,'uids':[100099]*4,'gids':[100100]*4}) for change in changes: with self.subTest(change=change),patch.object(guard,'verify_forwarder') as verify: with self.assertRaises(ValueError):guard.assert_owner({**actor,**change}) verify.assert_not_called() def test_runtime_namespace_owner_keeps_existing_rules(self): guard,actor=self.owner_fixture() guard.forwarder=(actor['pid'],1,2) with patch.object(guard,'verify_forwarder') as verify: guard.assert_owner({**actor,'net':(4,1)});verify.assert_not_called() def test_forwarder_requires_same_root_owned_executable_inode(self): directory=types.SimpleNamespace(st_mode=stat.S_IFDIR|0o755,st_uid=0) expected=types.SimpleNamespace(st_mode=stat.S_IFREG|0o755,st_uid=0,st_dev=1,st_ino=2) for actual_inode in (2,3): guard,actor=self.owner_fixture();guard.fds=[] actual=types.SimpleNamespace(st_dev=1,st_ino=actual_inode) with patch.object(Path,'lstat',return_value=directory),patch.object(retained.os,'open',side_effect=[100001,100002]),patch.object(retained.os,'fstat',side_effect=[actual,expected]): if actual_inode==2: guard.verify_forwarder(actor);self.assertEqual(guard.forwarder,(42,1,2)) else: with self.assertRaisesRegex(ValueError,'protected rootless forwarder'):guard.verify_forwarder(actor) def test_replacing_forwarder_executable_revokes_existing_guard(self): guard,_=self.owner_fixture();guard.processes=[];guard.namespace_bindings=[];guard.next_check=float('inf') guard.forwarder=(42,1,2) actual=types.SimpleNamespace(st_dev=1,st_ino=2) replacement=types.SimpleNamespace(st_dev=1,st_ino=3,st_uid=0,st_mode=stat.S_IFREG|0o755) with patch.object(retained.os,'stat',return_value=actual),patch.object(Path,'lstat',return_value=replacement): with self.assertRaisesRegex(ValueError,'executable changed'):guard.check() def test_only_verified_namespace_switching_roles_may_change_thread_net(self): before={'pid':42,'start':1,'uids':[1000]*4,'group':'appointed','net':(4,1)} after={**before,'net':(4,2)} self.assertFalse(retained.same_process(before,after)) self.assertTrue(retained.same_process(before,after,thread_net=True)) for change in ({'pid':43},{'start':2},{'uids':[0]*4},{'group':'foreign'}): with self.subTest(change=change):self.assertFalse(retained.same_process(before,{**after,**change},thread_net=True)) def test_retained_forwarder_still_rejects_socket_from_wrong_realm(self): import c01_network_owner as owner with tempfile.TemporaryDirectory() as directory,socket.socket() as listener: listener.bind(('127.0.0.1',0));listener.listen(1) port=listener.getsockname()[1] group=retained.PREFIX+'/app.slice/fixture.service' route={'target':'runtime','target_uid':1000,'target_cgroup':group,'projection_sha256':'a'*64, 'protocol':'tcp','name':f'app/fixture/tcp/{port}/host','port':port,'target_port':port,'target_address':'127.0.0.1'} cookie=listener.getsockopt(socket.SOL_SOCKET,71,8) probe=MagicMock();probe.__enter__.return_value.getsockopt.return_value=bytes([cookie[0]^1])+cookie[1:] policy=types.SimpleNamespace(value={'namespaces':{'runtime':[4,1]}},runtime={'runtime_gid':1000}, manifests={'a'*64:{'producer':{'kind':'retained-container-v1'}}},socket=Mock(return_value=probe)) actor={'pid':os.getpid(),'uids':[1000]*4,'gids':[1000]*4,'group':group,'net':(4,2)} guard=Mock(owner_thread_net=True) original_open=os.open def open_group(path,flags): self.assertEqual(path,'/sys/fs/cgroup'+group) return original_open(directory,flags) with patch.object(owner,'process',return_value=actor),patch.object(retained,'Guard',return_value=guard),patch.object(owner.os,'open',side_effect=open_group),patch.object(owner.LIBC,'syscall',side_effect=lambda *args:os.dup(listener.fileno())): with self.assertRaisesRegex(ValueError,'foreign socket network namespace'): owner.Target(policy,route,os.getpid(),listener.fileno()) guard.close.assert_called_once() if __name__=='__main__':unittest.main()