"""Real socket transport regressions; policy/namespace admission is separate.""" import selectors import socket import sys import time import unittest from unittest.mock import patch from pathlib import Path from types import SimpleNamespace sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from c01_network_owner import Bridge import c01_network_owner as network from c01_endpoints import validate_route import c01_supervisor as supervisor import c01_network_client as client import os import json import array class AppointmentTests(unittest.TestCase): def test_required_generation_loss_refuses_without_waiting_for_socket_pid_exit(self): left,right=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET) lease=client.NetworkLease.__new__(client.NetworkLease) lease.channel=left;lease.pidfd=os.pidfd_open(os.getpid());lease.owner=network.process(os.getpid()) lease.policy=SimpleNamespace(routes={'required':{}},check=lambda:None,close=lambda:None) lease.protocol=4;lease.required_targets={};lease.required_sockets={};lease.listeners=[] lease.required_generations={'required':'a'*64};lease.required_since={'required':0} lease.expiry=0;lease.start_deadline=time.clock_gettime_ns(time.CLOCK_BOOTTIME)+3_000_000_000 try: right.send(network.canonical({'alive_until_boottime_ns':time.clock_gettime_ns(time.CLOCK_BOOTTIME)+3_000_000_000, 'generations':{'required':None}})) with self.assertRaisesRegex(ValueError,'generation lost'):lease.check() finally:lease.close();right.close() def test_dual_stack_inventory_larger_than_scm_limit_is_complete_and_ordered(self): left,right=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET) listeners=[];sources=[] request={'method':'lease','policy_sha256':'a'*64,'nonce':'b'*64} try: for _ in range(270):sources.append(socket.socket()) for offset in range(0,len(sources),64): rights=array.array('i',[s.fileno() for s in sources[offset:offset+64]]) right.sendmsg([network.canonical({**request,'offset':offset,'total':len(sources)})], [(socket.SOL_SOCKET,socket.SCM_RIGHTS,rights)]) client.receive_listeners(left,request,len(sources),4,listeners) self.assertEqual([os.fstat(fd).st_ino for fd in listeners], [os.fstat(s.fileno()).st_ino for s in sources]) self.assertTrue(all(not os.get_inheritable(fd) for fd in listeners)) finally: for fd in listeners:os.close(fd) for s in sources:s.close() left.close();right.close() def test_listener_chunk_cannot_reorder_or_change_total(self): for change in ({'offset':64},{'total':2}): left,right=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET) received=[];request={'method':'lease','policy_sha256':'a'*64,'nonce':'b'*64} try: right.sendmsg([network.canonical({**request,'offset':0,'total':1,**change})], [(socket.SOL_SOCKET,socket.SCM_RIGHTS,array.array('i',[right.fileno()]))]) with self.assertRaisesRegex(ValueError,'binding/rights'): client.receive_listeners(left,request,1,4,received) finally: for fd in received:os.close(fd) left.close();right.close() def test_managed_registration_discovers_exact_listener_and_preserves_generation(self): with socket.socket() as listener: listener.bind(('127.0.0.2',0));listener.listen(1) route=dict(target='managed',protocol='tcp',target_address='127.0.0.2',target_port=listener.getsockname()[1],port=61001) lease=client.NetworkLease.__new__(client.NetworkLease) lease.policy=SimpleNamespace(routes={'gate/fixture':route}) registered=[] def request(body): if body['method']=='inspect':return {'producer':None,'generation':'a'*64} registered.append(body) return {'registered':body['name'],'policy_sha256':'b'*64,'generation':'c'*64} lease.request=request with patch.object(client.LIBC,'syscall',side_effect=lambda syscall,pidfd,fd,flags:os.dup(fd)): lease.register_managed_listeners(os.getpid(),['gate/fixture']) self.assertEqual(registered,[{'method':'register','name':'gate/fixture','pid':os.getpid(), 'fd':listener.fileno(),'previous':'a'*64}]) def test_absent_gate_is_not_registered_to_an_unrelated_port(self): lease=client.NetworkLease.__new__(client.NetworkLease) lease.policy=SimpleNamespace(routes={'gate/fixture':dict(target='managed',protocol='tcp',target_address='127.0.0.2',target_port=1,port=61001)}) calls=[] def request(body): calls.append(body) return {'producer':None,'generation':None} lease.request=request with patch.object(client.LIBC,'syscall',side_effect=lambda syscall,pidfd,fd,flags:os.dup(fd)): lease.register_managed_listeners(os.getpid(),['gate/fixture']) self.assertEqual(calls,[{'method':'inspect','name':'gate/fixture'}]) def test_gate_relay_cannot_bypass_authentication_or_change_producer(self): group='/system.slice/archipelago.service/c01-application' runtime={'manager_uid':61001} manifests={'a'*64:dict(app_id='fixture',loopback_ports=[['tcp',7778]],target_cgroup=group)} route=dict(name='gate/fixture/tcp/61001/host',source='host',target='managed',protocol='tcp', port=61001,target_port=7778,app_id='fixture',projection_sha256='a'*64, target_uid=61001,target_cgroup=group,source_addresses=['127.0.0.1'],target_address='127.0.0.2') self.assertEqual(validate_route(route,manifests,runtime,address_version=4),route) ingress={**route,'name':'gate/fixture/tcp/7778/host','port':7778, 'source_addresses':['192.168.63.100','fd79:1aa:b9e9:4c9f:1f80:5376:9385:1824']} self.assertEqual(validate_route(ingress,manifests,runtime,address_version=4),ingress) for change in ({'target':'runtime'},{'target_address':'127.0.0.1'}, {'target_port':7779},{'target_uid':1000},{'projection_sha256':'b'*64}, {'source_addresses':['0.0.0.0']},{'target_cgroup':'/system.slice/other.service'}): with self.subTest(change=change),self.assertRaises(ValueError): validate_route({**route,**change},manifests,runtime,address_version=4) def test_native_peer_interface_is_fixed(self): for name,address,port in (('peer','127.0.0.1',5679),): route=dict(name=f'native/{name}/tcp/5679/host',source='host',target='managed',protocol='tcp', port=5679,target_port=port,app_id=None,projection_sha256=None,target_uid=61001, target_cgroup='/system.slice/archipelago.service/c01-application', source_addresses=['fd00::1'],target_address=address) self.assertEqual(validate_route(route,{}, {'manager_uid':61001},address_version=4),route) for change in ({'target_port':5678},{'target_uid':1000},{'target':'runtime'}, {'target_address':'192.168.1.1'},{'protocol':'udp'}): with self.subTest(name=name,change=change),self.assertRaises(ValueError): validate_route({**route,**change},{},{'manager_uid':61001},address_version=4) def test_8443_preserves_fips_transport_not_fabricated_https(self): route=dict(name='fips-tcp6',source='host',target='host',protocol='tcp',port=8443,target_port=8443, app_id=None,projection_sha256=None,target_uid=0,target_cgroup='/system.slice/archipelago-fips.service', source_addresses=['::'],target_address='127.0.0.1') self.assertEqual(validate_route(route,{},address_version=4),route) for change in ({'target':'managed'},{'target_uid':61001},{'target_port':5678}, {'target_cgroup':'/system.slice/nginx.service'},{'source_addresses':['0.0.0.0']}, {'name':'native/https/tcp/8443/host'}): with self.subTest(change=change),self.assertRaises(ValueError):validate_route({**route,**change},{},address_version=4) def test_website_publication_keeps_fips_and_tor_exposures_separate(self): for kind,port,addresses in (('website-fips',32000,['fd00::1']),('website-tor',32100,['127.0.0.1'])): route=dict(name=f'native/{kind}/tcp/{port}/host',source='host',target='managed',protocol='tcp', port=port,target_port=port,app_id=None,projection_sha256=None,target_uid=61001, target_cgroup='/system.slice/archipelago.service/c01-application',source_addresses=addresses,target_address='127.0.0.1') self.assertEqual(validate_route(route,{}, {'manager_uid':61001},address_version=4),route) with self.assertRaises(ValueError):validate_route({**route,'source_addresses':['192.168.1.1']},{},{'manager_uid':61001},address_version=4) def test_kiosk_web_routes_only_reach_fixed_host_nginx(self): runtime = {'runtime_uid': 1000, 'user_cgroup': '/user.slice/user-1000.slice/user@1000.service'} for name, port in (('kiosk-http', 80), ('kiosk-https', 443)): route = dict(name=name, source='runtime', target='host', protocol='tcp', port=port, app_id=None, projection_sha256=None, target_uid=0, target_cgroup='/system.slice/nginx.service', source_addresses=['127.0.0.1'], target_address='127.0.0.1') self.assertEqual(validate_route(route, {}, runtime, address_version=3), route) with self.assertRaises(ValueError):validate_route(route, {}, address_version=3) for change in ({'port':22}, {'target_uid':1000}, {'protocol':'udp'}, {'target_cgroup':'/system.slice/ssh.service'}, {'source':'managed'}, {'source_addresses':['0.0.0.0']}, {'target_address':'127.0.0.2'}, {'app_id':'nginx'}, {'projection_sha256':'a'*64}): with self.subTest(name=name, change=change), self.assertRaises(ValueError): validate_route({**route, **change}, {}, runtime, address_version=3) def test_kiosk_cdp_uses_existing_exact_runtime_projection(self): runtime = {'runtime_uid':1000,'user_cgroup':'/user.slice/user-1000.slice/user@1000.service'} group = runtime['user_cgroup'] + '/app.slice/archipelago-kiosk-browser.service' route = dict(name='app/archipelago-kiosk/tcp/9222/managed', source='managed', target='runtime', protocol='tcp', port=9222, app_id='archipelago-kiosk', projection_sha256='a'*64, target_uid=1000, target_cgroup=group) manifests = {'a'*64:dict(app_id='archipelago-kiosk',loopback_ports=[['tcp',9222]],target_cgroup=group)} self.assertEqual(validate_route(route, manifests, runtime), route) with self.assertRaises(ValueError):validate_route(route, {}, runtime) with self.assertRaises(ValueError):validate_route({**route,'target_cgroup':'/system.slice/archipelago-kiosk.service'}, manifests, runtime) def test_network_lease_rejects_buffered_expired_progress(self): left,right=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET) lease=client.NetworkLease.__new__(client.NetworkLease) lease.channel=left;lease.pidfd=os.pidfd_open(os.getpid());lease.owner=network.process(os.getpid()) lease.policy=SimpleNamespace(check=lambda:None,close=lambda:None) lease.required_targets={};lease.listeners=[];lease.expiry=0 lease.start_deadline=time.clock_gettime_ns(time.CLOCK_BOOTTIME)+3_000_000_000 try: right.send(json.dumps({'alive_until_boottime_ns':time.clock_gettime_ns(time.CLOCK_BOOTTIME)+2_000_000_000}).encode()) lease.check() lease.expiry=0;lease.start_deadline=0 right.send(json.dumps({'alive_until_boottime_ns':time.clock_gettime_ns(time.CLOCK_BOOTTIME)-1}).encode()) with self.assertRaisesRegex(ValueError,'progress expired'):lease.check() finally:lease.close();right.close() def test_runtime_route_uses_appointed_uid_and_exact_manifest_producer(self): runtime = {'runtime_uid': 21002, 'user_cgroup': '/user.slice/user-21002.slice/user@21002.service'} group = runtime['user_cgroup'] + '/app.slice/ollama.service' route = dict(name='ollama', source='managed', target='runtime', protocol='tcp', port=11434, app_id='ollama', projection_sha256='a'*64, target_uid=21002, target_cgroup=group) manifests = {'a'*64: dict(app_id='ollama', loopback_ports=[['tcp',11434]], target_cgroup=group)} self.assertEqual(validate_route(route, manifests, runtime), route) for changes in ({'target_uid':1000}, {'port':11435}, {'app_id':'other'}, {'target_cgroup':runtime['user_cgroup']+'/app.slice/other.service'}): with self.assertRaises(ValueError):validate_route({**route, **changes}, manifests, runtime) with self.assertRaises(ValueError):validate_route(route, manifests) def test_same_cgroup_descendant_is_not_root_mainpid(self): actor = dict(group='/system.slice/archipelago.service/supervisor', uids=[0]*4, gids=[0]*4) with patch.object(network, 'process', return_value=actor), patch.object(network.subprocess, 'run', return_value=SimpleNamespace(returncode=0, stdout='123\n')): network.require_main(123, 'archipelago.service', actor['group']) with self.assertRaises(PermissionError):network.require_main(124, 'archipelago.service', actor['group']) actor['gids']=[1000]*4 with self.assertRaises(PermissionError):network.require_main(123, 'archipelago.service', actor['group']) def test_reviewed_legacy_loopback_addresses_do_not_create_arbitrary_host_routes(self): runtime={'runtime_uid':1000,'user_cgroup':'/user.slice/user-1000.slice/user@1000.service'} group=runtime['user_cgroup']+'/app.slice/fixture.service' route=dict(name='app/fixture/tcp/7778/host',source='host',target='runtime',protocol='tcp',port=7778, app_id='fixture',projection_sha256='a'*64,target_uid=1000,target_cgroup=group, source_addresses=['127.0.0.1','127.0.0.2'],target_address='127.0.0.2') manifests={'a'*64:dict(app_id='fixture',loopback_ports=[['tcp',7778]],target_cgroup=group)} self.assertEqual(validate_route(route,manifests,runtime,address_version=3),route) with self.assertRaises(ValueError):validate_route(route,manifests,runtime) for change in ({'source_addresses':['0.0.0.0']},{'source_addresses':['127.0.0.2']*2}, {'target_address':'192.168.63.1'},{'target_address':'127.0.0.3'}, {'target_uid':0},{'target_cgroup':runtime['user_cgroup']+'/app.slice/other.service'}): with self.subTest(change=change),self.assertRaises(ValueError): validate_route({**route,**change},manifests,runtime,address_version=3) def test_network_loss_stops_roles_but_keeps_lease_until_reaped(self): events=[] owner=supervisor.Supervisor.__new__(supervisor.Supervisor) children=SimpleNamespace(stopping=False, results={}, children={'application':object()}, reserved={}, poll=lambda: [], close=lambda:events.append('reaped')) def stop(): children.stopping=True events.append('stop') children.request_stop=stop owner.children=children;owner.watch=None;owner.failure=None;owner.completed_one_shots=set() owner.power_lease=None;owner.power_channel=None;owner.prior_power_signal=supervisor.signal.SIG_DFL owner.network_settings_watch=None;owner.tor_watch=None owner.runtime_lease=None;owner.access_lease=None def lost():raise ValueError('owner died') owner.network_lease=SimpleNamespace(check=lost,close=lambda:events.append('lease closed')) self.assertFalse(owner.tick()) self.assertEqual(owner.failure, 'network lease lost') self.assertNotIn('lease closed',events) children.children={} owner.prior_wakeup=-1;owner.wake_read=10;owner.wake_write=11;owner.notifier=None owner.lock=SimpleNamespace(close=lambda:events.append('lock closed')) with patch.object(supervisor.signal,'set_wakeup_fd'), patch.object(supervisor.signal,'signal'), patch.object(supervisor.os,'close'): owner.retire() self.assertLess(events.index('reaped'),events.index('lease closed')) self.assertLess(events.index('lease closed'),events.index('lock closed')) class RegistrationTests(unittest.TestCase): def test_live_process_cannot_keep_closed_producer_socket_admitted(self): original=socket.socket();original.bind(('127.0.0.1',0));original.listen(1) target=network.Target.__new__(network.Target) target.guard=None target.number=original.fileno();target.listener=socket.socket(fileno=os.dup(original.fileno())) target.pidfd=os.pidfd_open(os.getpid());target.group=None target.identity={'pid':os.getpid()};target.route={'target_cgroup':'/fixture'};target.actual_group='/fixture';target.group_id=(1,2) try: with patch.object(network,'process',return_value=target.identity),patch.object(network.os,'stat',return_value=SimpleNamespace(st_dev=1,st_ino=2)): target.check() original.close() with self.assertRaises((FileNotFoundError,ValueError)):target.check() finally:original.close();target.close() def test_manager_retirement_does_not_replace_external_service_producers(self): bridge=Bridge(SimpleNamespace(routes={},check=lambda:None)) retired=[] try: for name,realm in (('management','managed'),('tor-socks','host'),('lnd-rest','runtime')): bridge.targets[name]=SimpleNamespace(route={'target':realm},close=lambda name=name:retired.append(name)) before=bridge.targets['tor-socks'] bridge.retire_management_lease() self.assertEqual(retired,['management']) self.assertIs(bridge.targets['tor-socks'],before) self.assertIn('lnd-rest',bridge.targets) finally:bridge.close() def test_dead_producer_requires_explicit_generation_for_replacement(self): policy = SimpleNamespace(routes={}, check=lambda: None, value={'version':1}) bridge = Bridge(policy) policy.routes['fixture'] = {'target': 'runtime'} def target(token, pid): return SimpleNamespace(token=token, identity={'pid': pid}, check=lambda: None, close=lambda: None) first = target('a' * 64, 100) second = target('b' * 64, 101) try: with patch.object(network, 'Target', return_value=first): self.assertEqual(bridge.register('fixture', 100, 3), 'a' * 64) self.assertEqual(bridge.registration_status('fixture')['producer'], {'pid': 100}) with patch.object(network, 'Target', return_value=second): with self.assertRaisesRegex(ValueError, 'live endpoint'): bridge.register('fixture', 101, 3, 'a' * 64) bridge.retire('fixture') self.assertEqual(bridge.registration_status('fixture'), {'name': 'fixture', 'generation': 'a' * 64, 'producer': None}) with self.assertRaisesRegex(ValueError, 'generation changed'): bridge.register('fixture', 101, 3) self.assertEqual(bridge.register('fixture', 101, 3, 'a' * 64), 'b' * 64) with self.assertRaisesRegex(ValueError, 'generation changed'): bridge.register('fixture', 101, 3, 'a' * 64) finally: bridge.close() class StreamTests(unittest.TestCase): def setUp(self): self.bridge = Bridge(SimpleNamespace(routes={}, check=lambda: None)) self.a, self.client = socket.socketpair() self.b, self.server = socket.socketpair() for channel in (self.a, self.b, self.client, self.server): channel.setblocking(False) self.bridge.targets['fixture'] = SimpleNamespace(check=lambda: None, close=lambda: None) for channel, peer in ((self.a, self.b), (self.b, self.a)): self.bridge.streams[channel] = dict(peer=peer, name='fixture', buffer=bytearray(), connecting=False, eof=False, deadline=time.monotonic()+120) self.bridge.selector.register(channel, selectors.EVENT_READ, ('stream', 'fixture')) def tearDown(self): self.bridge.close() self.client.close() self.server.close() def test_one_way_progress_renews_both_stream_lifetimes(self): self.bridge.streams[self.a]['deadline'] = time.monotonic()-1 self.server.send(b'active media') self.bridge.poll(0) self.assertEqual(len(self.bridge.streams), 2) for record in self.bridge.streams.values(): self.assertGreater(record['deadline'], time.monotonic()) self.bridge.poll(0) self.assertEqual(self.client.recv(100), b'active media') def test_dual_half_close_drains_buffered_bytes_before_retirement(self): payload = b'retained final bytes' * 2048 self.bridge.streams[self.a]['buffer'].extend(payload) self.bridge.streams[self.a]['eof'] = True self.bridge.streams[self.b]['eof'] = True self.bridge.selector.modify(self.a, selectors.EVENT_WRITE, ('stream', 'fixture')) received = bytearray() deadline = time.monotonic()+3 while time.monotonic() < deadline: self.bridge.poll(0) try: block = self.client.recv(65536) if not block: break received.extend(block) except BlockingIOError: pass self.assertEqual(bytes(received), payload) self.assertEqual(self.bridge.streams, {}) def test_idle_streams_are_retired(self): self.bridge.streams[self.a]['deadline'] = time.monotonic()-1 self.bridge.poll(0) self.assertEqual(self.bridge.streams, {}) class AddressTransportTests(unittest.TestCase): def test_tcp_forwarding_reaches_the_preserved_second_loopback_address(self): self.forwarding('127.0.0.2',socket.AF_INET) def test_tcp_forwarding_reaches_ipv6_on_distinct_target_port(self): self.forwarding('::1',socket.AF_INET6,translated=True) def test_ipv6_ingress_reaches_the_authenticated_second_loopback_target(self): self.forwarding('127.0.0.2',socket.AF_INET,translated=True,source_address='::1') def forwarding(self,address,family,translated=False,source_address='127.0.0.1'): with socket.socket(family,socket.SOCK_STREAM) as server: server.bind((address,0));server.listen(1);server.setblocking(False) port=server.getsockname()[1] route=dict(source='managed',target='runtime',protocol='tcp',port=0 if translated else port, target_port=port,source_addresses=[source_address],target_address=address) policy=SimpleNamespace(routes={'fixture':route},check=lambda:None, socket=lambda role,kind=socket.SOCK_STREAM,family=socket.AF_INET:socket.socket(family,kind)) bridge=Bridge(policy);remote=None source_port=next(iter(bridge.listeners)).getsockname()[1] bridge.targets['fixture']=SimpleNamespace(route=route,check=lambda:None,close=lambda:None) try: with socket.create_connection((source_address,source_port)) as caller: caller.setblocking(False);caller.sendall(b'preserved-loopback') result=b'';end=time.monotonic()+3 while time.monotonic()