#!/usr/bin/python3 """Independent named TCP route owner. Never accepts an address/path/argv RPC. Targets are explicitly registered kernel socket capabilities, not discovered by blindly reconnecting to a port. The duplicated listener reserves its port across producer death; pidfd/cgroup/namespace/policy drift closes affected streams. Both TCP and bounded UDP associations retain exact producer/socket identity. """ import array import ctypes import errno import hashlib import json import re import os import pwd from pathlib import Path import selectors import select import socket import stat import struct import subprocess import time import traceback import types import sys if __name__=='__main__': for name in ('c01_network','c01_endpoints','runtime_uplink','c01_uplink','c01_network_mdns','c01_retained_producer','c01_network_plan'): source=Path(__file__).resolve().with_name(name+'.py') for part in [source,*source.parents]: meta=part.lstat() if meta.st_uid!=0 or meta.st_mode & 0o022 or stat.S_ISLNK(meta.st_mode):raise ValueError('unprotected network module') module=types.ModuleType(name);module.__file__=str(source);sys.modules[name]=module exec(compile(source.read_bytes(),str(source),'exec'),module.__dict__) import c01_network as namespaces from c01_endpoints import validate_route, source_addresses, target_address, target_port, source_binding_conflict import c01_uplink import c01_network_plan import c01_network_mdns import c01_retained_producer def gate_peer_header(client): """Kernel-observed peer, not an HTTP header supplied by the caller.""" family=client.family source=client.getpeername();destination=client.getsockname() if family not in (socket.AF_INET,socket.AF_INET6):raise ValueError('gate peer family') body=socket.inet_pton(family,source[0])+socket.inet_pton(family,destination[0])+struct.pack('!HH',source[1],destination[1]) return b'\r\n\r\n\0\r\nQUIT\n'+bytes((0x21,0x11 if family==socket.AF_INET else 0x21))+struct.pack('!H',len(body))+body def producer_group(policy,route,actor): expected=route['target_cgroup'];actual=actor['group'] doc=getattr(policy,'manifests',{}).get(route.get('projection_sha256'),{}) selector=doc.get('producer',{'kind':'exact'}) if selector['kind']=='exact': if actual!=expected:raise ValueError('endpoint producer cgroup') return actual if route['target']!='runtime' or selector['kind']!='container-payload':raise ValueError('producer selector realm') match=re.fullmatch(re.escape(expected)+r'/libpod-payload-([0-9a-f]{64})',actual) if not match:raise ValueError('payload outside appointed service') import sqlite3 try:connection=sqlite3.connect('file:/var/lib/archipelago/containers/storage/db.sql?mode=ro',uri=True,timeout=.25) except sqlite3.Error:raise ValueError('runtime producer metadata unavailable') from None try: connection.execute('PRAGMA query_only=ON') config=connection.execute('SELECT JSON FROM ContainerConfig WHERE ID=?',(match[1],)).fetchone() state=connection.execute('SELECT JSON FROM ContainerState WHERE ID=?',(match[1],)).fetchone() if config is None or state is None:raise ValueError('payload container absent from retained runtime') config=json.loads(config[0]);state=json.loads(state[0]) if config.get('name')!=selector['container_name'] or config.get('rootfsImageID')!=selector['image_id'] or config.get('id')!=match[1]:raise ValueError('payload configuration identity differs') init=state.get('pid') if type(init) is not int or init<=1 or state.get('state')!=3:raise ValueError('payload container is not running') if os.readlink(f'/proc/{init}/ns/pid')!=os.readlink(f'/proc/{actor["pid"]}/ns/pid'):raise ValueError('payload PID namespace differs') initial=process(init) if initial['group']!=actual:raise ValueError('payload init cgroup differs') if 'payload_principal' in selector: uid,gid=selector['payload_principal'] if actor['uids']!=[uid]*4 or actor['gids']!=[gid]*4 or initial['uids']!=[uid]*4 or initial['gids']!=[gid]*4: raise ValueError('mapped payload principal differs') if os.readlink(f'/proc/{init}/ns/user')!=os.readlink(f'/proc/{actor["pid"]}/ns/user'):raise ValueError('payload user namespace differs') except sqlite3.Error:raise ValueError('runtime producer metadata unavailable') from None finally:connection.close() return actual CONFIG=Path('/etc/archipelago-c01/endpoints.json') CONTROL=Path('/run/archipelago-c01-network/control.sock') RUNTIME_CONFIG=Path('/etc/archipelago-c01/runtime.json') LIBC=ctypes.CDLL(None,use_errno=True) MAX=65536 MAX_STREAMS=128 BUFFER=65536 def canonical(value):return json.dumps(value,sort_keys=True,separators=(',',':'),ensure_ascii=True,allow_nan=False).encode('ascii') def digest(raw):return hashlib.sha256(raw).hexdigest() def exact(pairs): result={} for k,v in pairs: if k in result:raise ValueError('duplicate endpoint field') result[k]=v return result def protected_read(path): for parent in reversed(path.parents): m=parent.lstat() if not stat.S_ISDIR(m.st_mode) or m.st_uid!=0 or m.st_mode&0o022:raise ValueError('endpoint policy ancestry') fd=os.open(path,os.O_RDONLY|os.O_NOFOLLOW|os.O_CLOEXEC) with os.fdopen(fd,'rb') as stream: m=os.fstat(stream.fileno()) if not stat.S_ISREG(m.st_mode) or m.st_uid!=0 or m.st_mode&0o077 or m.st_nlink!=1 or m.st_size>512*1024:raise ValueError('endpoint policy object') raw=stream.read(512*1024+1) n=os.fstat(stream.fileno()) if (m.st_dev,m.st_ino,m.st_size,m.st_mtime_ns,m.st_ctime_ns)!=(n.st_dev,n.st_ino,n.st_size,n.st_mtime_ns,n.st_ctime_ns):raise ValueError('endpoint policy changed') return raw def process(pid): raw=Path(f'/proc/{pid}/stat').read_text() status=dict(line.split(':',1) for line in Path(f'/proc/{pid}/status').read_text().splitlines() if ':' in line) group=Path(f'/proc/{pid}/cgroup').read_text().strip() if not group.startswith('0::/'):raise ValueError('endpoint unified cgroup') return {'pid':pid,'start':int(raw[raw.rindex(')')+2:].split()[19]),'parent':int(status['PPid']), 'uids':list(map(int,status['Uid'].split())),'gids':list(map(int,status['Gid'].split())), 'group':group[3:], 'net':namespaces.process_namespace(pid)} def require_main(pid, unit, group): """Authenticate the service leader, never just a descendant in its cgroup.""" result=subprocess.run(['/usr/bin/systemctl','show',unit,'-p','MainPID','--value'], capture_output=True,text=True,timeout=2,env={'PATH':'/usr/bin:/bin','LANG':'C'}) actor=process(pid) if result.returncode or result.stdout.strip()!=str(pid) or actor['group']!=group or actor['uids']!=[0]*4 or actor['gids']!=[0]*4: raise PermissionError('network root MainPID appointment') return actor class Policy: def __init__(self,path=CONFIG): self.path=path;raw=protected_read(path);self.sha=digest(raw) self.value=json.loads(raw,object_pairs_hook=exact) v=self.value if type(v) is not dict or type(v.get('version')) is not int or v['version'] not in (1,2,3,4) or set(v)!=({'version','namespaces','manifests','routes'} | ({'uplink'} if v['version'] in (2,3,4) else set()) | ({'mdns'} if v['version']==4 and 'mdns' in v else set())): raise ValueError('endpoint policy schema') self.uplink=None if v['version'] in (2,3,4):c01_uplink.validate(v['uplink']) bound=128 if v['version']==4 else 64 if type(v['namespaces']) is not dict or set(v['namespaces'])!={'host','managed','runtime'} or type(v['manifests']) is not list or len(v['manifests'])>128 or type(v['routes']) is not list or not 1<=len(v['routes'])<=bound: raise ValueError('endpoint policy bounds') self.runtime_raw=None runtime=None if any('runtime' in (r.get('source'),r.get('target')) or r.get('name','').startswith(('gate/','native/')) for r in v['routes'] if type(r) is dict): self.runtime_raw=protected_read(RUNTIME_CONFIG) runtime=json.loads(self.runtime_raw,object_pairs_hook=exact) uid=runtime['runtime_uid'] if type(uid) is not int or uid<=0 or runtime['user_cgroup']!=f'/user.slice/user-{uid}.slice/user@{uid}.service': raise ValueError('runtime policy appointment') account=pwd.getpwnam(runtime['runtime_user']) if (account.pw_uid,account.pw_gid)!=(uid,runtime['runtime_gid']):raise ValueError('runtime account differs from appointment') self.runtime=runtime self.retained_manager=None if v['version']==4:c01_network_plan.validate(v,runtime) manifests={} for doc in v['manifests']: fields={'app_id','manifest_sha256','loopback_ports','target_cgroup'} if v['version']==4 and 'producer' in doc:fields.update(('producer','source_bindings','target_addresses')) if type(doc) is not dict or set(doc)!=fields or not isinstance(doc['loopback_ports'],list):raise ValueError('manifest projection schema') if type(doc['manifest_sha256']) is not str or len(doc['manifest_sha256'])!=64 or any(c not in '0123456789abcdef' for c in doc['manifest_sha256']):raise ValueError('manifest byte digest') if type(doc['app_id']) is not str or not re.fullmatch('[a-z0-9][a-z0-9-]{0,63}',doc['app_id']):raise ValueError('manifest application identity') if not all(type(p) is list and len(p)==2 and p[0] in ('tcp','udp') and type(p[1]) is int and 0(512 if v['version']==4 else bound):raise ValueError('endpoint listener capability bound') self.fds={} try: for role in ('host','managed','runtime'): if type(v['namespaces'][role]) is not list or len(v['namespaces'][role])!=2 or any(type(x) is not int or x<=0 for x in v['namespaces'][role]):raise ValueError('namespace identity type') path=Path('/proc/1/ns/net') if role=='host' else namespaces.NAMES[role] if role!='host':namespaces.appointed(role) fd=os.open(path,os.O_RDONLY|os.O_CLOEXEC| (0 if role=='host' else os.O_NOFOLLOW)) self.fds[role]=fd if list(namespaces.identity(fd))!=v['namespaces'][role]:raise ValueError('endpoint namespace appointment mismatch') if len({tuple(x) for x in v['namespaces'].values()})!=3:raise ValueError('endpoint namespaces must be distinct') self.home=os.open('/proc/self/ns/net',os.O_RDONLY|os.O_CLOEXEC) self.next_address_check=0 except BaseException: for fd in self.fds.values():os.close(fd) raise def check(self): if digest(protected_read(self.path))!=self.sha:raise ValueError('endpoint policy lifetime replaced') if self.runtime_raw is not None and protected_read(RUNTIME_CONFIG)!=self.runtime_raw:raise ValueError('runtime appointment replaced') for role,expected in self.value['namespaces'].items(): actual=namespaces.process_namespace(1) if role=='host' else namespaces.appointed(role) if list(actual)!=expected:raise ValueError('endpoint namespace lifetime replaced') if self.uplink is not None:self.uplink.check() if self.value['version']==4 and time.monotonic()>=self.next_address_check: import ipaddress observed=namespaces.namespace_operation(self.fds['host'],'host-addresses') actual=set(observed['addresses']) if 'mdns' in self.value and observed['interfaces'].get(self.value['mdns']['interface'])!=self.value['mdns']['ifindex']:raise ValueError('mDNS host interface replaced') required={a for route in self.routes.values() if route['source']=='host' for a in source_addresses(route) if not ipaddress.ip_address(a).is_loopback and not ipaddress.ip_address(a).is_unspecified} if not required<=actual:raise ValueError('appointed host ingress address disappeared') fips={a for route in self.routes.values() if route['name'].startswith(('native/peer/','native/website-fips/')) for a in source_addresses(route)} if not fips<=set(observed['fips_addresses']):raise ValueError('FIPS ingress address is not on the actual host FIPS interface') self.next_address_check=time.monotonic()+1 def socket(self,role,kind=socket.SOCK_STREAM,family=socket.AF_INET): if kind not in (socket.SOCK_STREAM,socket.SOCK_DGRAM):raise ValueError('finite endpoint socket kind') if family not in (socket.AF_INET,socket.AF_INET6):raise ValueError('finite endpoint socket family') return namespaces.namespace_operation(self.fds[role],('tcp' if kind==socket.SOCK_STREAM else 'udp')+('6' if family==socket.AF_INET6 else '')) def close(self): if self.retained_manager is not None:self.retained_manager.close();self.retained_manager=None if self.uplink is not None:self.uplink.close() for fd in [*self.fds.values(),self.home]:os.close(fd) def listener_address_matches(listener,route): if route.get('name')=='native/mdns4/udp/5353/host' and listener.family!=socket.AF_INET:return False address=listener.getsockname()[:2];expected=target_address(route);port=target_port(route) if ':' in expected: return listener.family==socket.AF_INET6 and address in ((expected,port),('::',port)) if listener.family==socket.AF_INET: return address in ((expected,port),('0.0.0.0',port)) return (listener.family==socket.AF_INET6 and address==('::',port) and listener.getsockopt(socket.IPPROTO_IPV6,socket.IPV6_V6ONLY)==0) class Target: def __init__(self,policy,route,pid,number): self.policy,self.route=policy,route self.number=number self.guard=None self.pidfd=os.pidfd_open(pid);self.listener=None;self.group=None try: self.identity=process(pid) selector=getattr(policy,'manifests',{}).get(route.get('projection_sha256'),{}).get('producer',{}) if self.identity['uids']!=[route['target_uid']]*4:raise ValueError('endpoint producer identity') # A retained rootless port-forwarder may enter its payload namespace # after creating the listener. Its guard pins the user-manager and # payload realms plus the trusted executable; the socket's own # namespace cookie is still checked below. if selector.get('kind')!='retained-container-v1' and list(self.identity['net'])!=policy.value['namespaces'][route['target']]:raise ValueError('endpoint producer identity') if selector.get('kind')=='retained-container-v1': if route['target']!='runtime':raise ValueError('retained producer realm') self.guard=c01_retained_producer.Guard(policy,route,self.identity,process) self.actual_group=self.identity['group'] else:self.actual_group=producer_group(policy,route,self.identity) if route['target']=='runtime' and self.guard is None: expected_gid=(selector['payload_principal'][1] if selector.get('kind')=='container-payload' and 'payload_principal' in selector else policy.runtime['runtime_gid']) if self.identity['gids']!=[expected_gid]*4:raise ValueError('runtime endpoint group identity') if route['target']=='managed': expected_gid=(policy.runtime or {}).get('manager_gid',route['target_uid']) if self.identity['gids']!=[expected_gid]*4:raise ValueError('managed endpoint group identity') require_main(self.identity['parent'],'archipelago.service','/system.slice/archipelago.service/supervisor') elif route['target']=='host': result=subprocess.run(['/usr/bin/systemctl','show',route['target_cgroup'].split('/')[-1],'-p','MainPID','--value'],capture_output=True,text=True,timeout=2) if result.returncode or result.stdout.strip()!=str(pid):raise ValueError('host endpoint must be service MainPID') self.group=os.open('/sys/fs/cgroup'+self.actual_group,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW|os.O_CLOEXEC) self.group_id=(os.fstat(self.group).st_dev,os.fstat(self.group).st_ino) raw=LIBC.syscall(438,self.pidfd,number,0) # pidfd_getfd: pins the exact listening socket if raw<0:raise OSError(ctypes.get_errno(),'endpoint listener capability unavailable') self.listener=socket.socket(fileno=raw);self.listener.set_inheritable(False) kind=socket.SOCK_STREAM if route['protocol']=='tcp' else socket.SOCK_DGRAM multicast=route['name'] in ('native/mdns4/udp/5353/host','native/mdns6/udp/5353/host') if self.listener.getsockopt(socket.SOL_SOCKET,socket.SO_TYPE)!=kind or (kind==socket.SOCK_STREAM and self.listener.getsockopt(socket.SOL_SOCKET,socket.SO_ACCEPTCONN)!=1) or (not multicast and (self.listener.getsockopt(socket.SOL_SOCKET,socket.SO_REUSEPORT)!=0 or (kind==socket.SOCK_DGRAM and self.listener.getsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR)!=0))): raise ValueError('endpoint must have exclusive bind ownership') if not listener_address_matches(self.listener,route):raise ValueError('endpoint listener address differs') # Socket namespace cookie is compared with a root-created socket in # the appointed target namespace, not inferred only from process netns. with policy.socket(route['target'],family=self.listener.family) as probe: if self.listener.getsockopt(socket.SOL_SOCKET,71,8)!=probe.getsockopt(socket.SOL_SOCKET,71,8):raise ValueError('foreign socket network namespace') if self.guard is not None:self.guard.check(force=True) self.check() self.token=digest(canonical({'process':self.identity,'group':self.group_id, 'socket_inode':os.fstat(self.listener.fileno()).st_ino,'policy':policy.sha,'route':route['name'], **({'retained':self.guard.binding} if self.guard is not None else {})})) except BaseException:self.close();raise def check(self): if self.guard is not None:self.guard.check() thread_net=self.guard is not None and self.guard.owner_thread_net if select.select([self.pidfd],[],[],0)[0] or not c01_retained_producer.same_process(self.identity,process(self.identity['pid']),thread_net):raise ValueError('endpoint producer lifetime lost') if os.readlink(f'/proc/{self.identity["pid"]}/fd/{self.number}')!=f'socket:[{os.fstat(self.listener.fileno()).st_ino}]': raise ValueError('endpoint producer closed or replaced its appointed descriptor') m=os.stat('/sys/fs/cgroup'+self.actual_group) if (m.st_dev,m.st_ino)!=self.group_id:raise ValueError('endpoint producer cgroup replaced') def close(self): if self.guard is not None:self.guard.close();self.guard=None if self.listener is not None:self.listener.close();self.listener=None if self.pidfd is not None:os.close(self.pidfd);self.pidfd=None if self.group is not None:os.close(self.group);self.group=None class Bridge: def __init__(self,policy): self.policy=policy;self.listeners={};self.targets={};self.epochs={};self.streams={};self.datagrams={};self.leases={};self.selector=selectors.DefaultSelector();self.mdns={} try: for name,r in policy.routes.items(): if name.startswith('native/mdns'): family=socket.AF_INET6 if name.startswith('native/mdns6/') else socket.AF_INET scope=policy.value['mdns'] source=c01_network_mdns.make_socket(family,scope['interface'],scope['ipv4']) self.listeners[source]=name private=namespaces.namespace_operation(policy.fds['managed'],'mdns6' if family==socket.AF_INET6 else 'mdns4') self.mdns[name]=c01_network_mdns.Link(source,private,scope['ifindex']) self.selector.register(source,selectors.EVENT_READ,('mdns-host',name)) self.selector.register(private,selectors.EVENT_READ,('mdns-managed',name)) continue for address in source_addresses(r): s=policy.socket(r['source'],socket.SOCK_STREAM if r['protocol']=='tcp' else socket.SOCK_DGRAM, family=socket.AF_INET6 if ':' in address else socket.AF_INET) try: if s.family==socket.AF_INET6:s.setsockopt(socket.IPPROTO_IPV6,socket.IPV6_V6ONLY,1) s.bind((address,r['port'])) if r['protocol']=='tcp':s.listen(32) s.setblocking(False) self.listeners[s]=name;self.selector.register(s,selectors.EVENT_READ,('listen' if r['protocol']=='tcp' else 'udp-listen',name)) except BaseException:s.close();raise except BaseException:self.close();raise def register(self,name,pid,number,previous=None): if name not in self.policy.routes or type(pid) is not int or pid<=1 or type(number) is not int or not 0<=number<=65535:raise ValueError('named endpoint registration') self.policy.check() if previous!=self.epochs.get(name):raise ValueError('endpoint generation changed; no implicit retarget') candidate=Target(self.policy,self.policy.routes[name],pid,number) if name in self.targets: candidate.close() raise ValueError('live endpoint cannot be replaced') self.retire(name) self.targets[name]=candidate self.epochs[name]=candidate.token return candidate.token def register_host_services(self): """One initial appointment to actual service MainPID/listener objects. No retarget after loss: service restart requires new owner admission. Existing Target checks still prove UID, cgroup, executable-independent service MainPID, exclusive socket bind and target namespace. """ for name,route in self.policy.routes.items(): if route['target']!='host':continue unit=route['target_cgroup'].split('/')[-1] result=subprocess.run(['/usr/bin/systemctl','show',unit,'-p','MainPID','--value'], capture_output=True,text=True,timeout=2,check=True) pid=int(result.stdout.strip()) if pid<=1:raise ValueError('appointed host service absent: '+name) entries=list(Path(f'/proc/{pid}/fd').iterdir()) if len(entries)>1024:raise ValueError('host descriptor inventory exceeds bound') candidates=[] try: for entry in entries: try: if not os.readlink(entry).startswith('socket:['):continue candidate=Target(self.policy,route,pid,int(entry.name)) except (OSError,ValueError):continue candidates.append(candidate) if len(candidates)!=1:raise ValueError('unique appointed host listener unavailable: '+name) candidate=candidates.pop() self.targets[name]=candidate;self.epochs[name]=candidate.token finally: for candidate in candidates:candidate.close() def retire_management_lease(self): for name,target in list(self.targets.items()): if target.route['target']=='managed':self.retire(name) def public_gate_status(self): self.policy.check() gates=[] for name,route in self.policy.routes.items(): if not name.startswith('gate/'):continue status=self.registration_status(name) producer=status['producer'] if producer is None:continue gates.append({'app_id':route['app_id'],'port':route['port'],'target_port':target_port(route), 'addresses':source_addresses(route),'pid':producer['pid']}) value={'policy_sha256':self.policy.sha,'gates':gates} if len(canonical(value))>8192:raise ValueError('public gate observation exceeds bound') return value def registration_status(self,name): self.policy.check() if name not in self.policy.routes:raise ValueError('unknown endpoint route') target=self.targets.get(name) if target: try:target.check() except (OSError,ValueError):self.retire(name);target=None value={'name':name,'generation':self.epochs.get(name), 'producer':target.identity if target else None} if self.policy.value['version']==4: value['listener']={'number':target.number,'inode':os.fstat(target.listener.fileno()).st_ino} if target else None return value def retire(self,name): for s,r in list(self.datagrams.items()): if r['name']==name:self.selector.unregister(s);s.close();del self.datagrams[s] for stream in list(self.streams): if stream in self.streams and self.streams[stream]['name']==name:self.drop(stream) old=self.targets.pop(name,None) if old:old.close() def drop(self,s): record=self.streams.pop(s,None) if record: peer=record['peer'];self.streams.pop(peer,None) for channel in (s,peer): try:self.selector.unregister(channel) except (KeyError,ValueError):pass channel.close() def poll(self,timeout=.05): self.policy.check() for name,target in list(self.targets.items()): try:target.check() except (OSError,ValueError):self.retire(name) for key,events in self.selector.select(timeout): s=key.fileobj;kind,name=key.data if kind.startswith('mdns-'): try:self.mdns[name].receive(kind[5:],name in self.targets) except BlockingIOError:pass continue if kind=='udp-listen': data,address=s.recvfrom(4097) if len(data)>4096 or name not in self.targets or len(self.datagrams)>=64:continue upstream=None try: target=self.targets[name];target.check() upstream=self.policy.socket(target.route['target'],socket.SOCK_DGRAM,socket.AF_INET6 if ':' in target_address(target.route) else socket.AF_INET);upstream.setblocking(False) upstream.connect((target_address(target.route),target_port(target.route)));target.check() upstream.send(data) self.datagrams[upstream]={'name':name,'source':s,'address':address,'deadline':time.monotonic()+5} self.selector.register(upstream,selectors.EVENT_READ,('udp-reply',name)) except (OSError,ValueError): if upstream:upstream.close() continue if kind=='udp-reply': r=self.datagrams.pop(s,None) if r: try: self.targets[name].check();data=s.recv(4097) if len(data)<=4096:r['source'].sendto(data,r['address']) except (OSError,ValueError,KeyError):pass self.selector.unregister(s);s.close() continue if kind=='listen': client,_=s.accept();client.setblocking(False) if name not in self.targets or len(self.streams)>=2*MAX_STREAMS:client.close();continue upstream=None try: target=self.targets[name];target.check() upstream=self.policy.socket(target.route['target'],family=socket.AF_INET6 if ':' in target_address(target.route) else socket.AF_INET);upstream.setblocking(False) error=upstream.connect_ex((target_address(target.route),target_port(target.route))) if error not in (0,errno.EINPROGRESS):raise OSError(error,'endpoint connect') target.check() for a,b,connecting in ((client,upstream,False),(upstream,client,error!=0)): prefix=gate_peer_header(client) if a is upstream and name.startswith('gate/') else b'' self.streams[a]={'peer':b,'name':name,'buffer':bytearray(prefix),'connecting':connecting,'eof':False,'deadline':time.monotonic()+120} self.selector.register(a,selectors.EVENT_READ| (selectors.EVENT_WRITE if connecting or prefix else 0),('stream',name)) except (OSError,ValueError): self.drop(client);client.close() if upstream is not None:upstream.close() continue if s not in self.streams:continue record=self.streams[s];peer=record['peer'] try: self.targets[name].check() if events & selectors.EVENT_WRITE: if record['connecting']: error=s.getsockopt(socket.SOL_SOCKET,socket.SO_ERROR) if error:raise OSError(error,'endpoint connect failed') record['connecting']=False if record['buffer']: sent=s.send(record['buffer']);del record['buffer'][:sent] if sent: record['deadline']=self.streams[peer]['deadline']=time.monotonic()+120 if events & selectors.EVENT_READ: other=self.streams[peer] capacity=BUFFER-len(other['buffer']) if capacity: data=s.recv(min(16384,capacity)) if not data:record['eof']=True else: other['buffer'].extend(data) record['deadline']=other['deadline']=time.monotonic()+120 for a in (s,peer): r=self.streams[a];other=self.streams[r['peer']] if r['eof'] and not other['buffer']: try:r['peer'].shutdown(socket.SHUT_WR) except OSError:pass interest=(selectors.EVENT_READ if not r['eof'] and len(other['buffer'])r['deadline']:self.drop(s) for s,r in list(self.datagrams.items()): if time.monotonic()>r['deadline']:self.selector.unregister(s);s.close();del self.datagrams[s] def close(self): for link in self.mdns.values():link.close() self.mdns.clear() for s in self.datagrams:s.close() self.datagrams.clear() for s in list(self.streams):self.drop(s) for target in self.targets.values():target.close() self.targets.clear() for s in self.listeners:s.close() self.listeners.clear();self.selector.close() def main(): if os.geteuid()!=0 or len(sys.argv)!=1:raise ValueError('fixed root network owner only') if namespaces.process_namespace(os.getpid())!=namespaces.process_namespace(1):raise ValueError('network owner must use the actual host namespace') policy=Policy() try: if policy.value['version'] in (2,3,4): factory=c01_uplink.Uplinks if policy.value['uplink'].get('routing')=='host-dual-stack' else c01_uplink.Uplink policy.uplink=factory(policy,policy.value['uplink']) bridge=Bridge(policy) bridge.register_host_services() except BaseException:policy.close();raise CONTROL.parent.mkdir(mode=0o700,exist_ok=True) control=socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET|socket.SOCK_CLOEXEC) control.bind(str(CONTROL));os.chmod(CONTROL,0o600);control.listen(8);control.setblocking(False) pending={};renew_at=0 try: # After= must order consumers after actual route/owner readiness, not # merely after Python exec. Only this service MainPID may notify PID1. policy.check() address=os.environ.get('NOTIFY_SOCKET') if address: if not address.startswith(('/', '@')):raise ValueError('network notification address') with socket.socket(socket.AF_UNIX,socket.SOCK_DGRAM) as notify: notify.connect('\0'+address[1:] if address.startswith('@') else address) notify.send(b'READY=1') while True: bridge.poll() if time.monotonic()>=renew_at: renewal_value={'alive_until_boottime_ns':time.clock_gettime_ns(time.CLOCK_BOOTTIME)+3_000_000_000} if policy.value['version']==4:renewal_value['generations']={name:bridge.epochs.get(name) if name in bridge.targets else None for name in policy.routes} renewal=canonical(renewal_value) if len(renewal)>65536:raise ValueError('network progress bound') for channel,pidfd in list(bridge.leases.items()): try: if channel.send(renewal,socket.MSG_DONTWAIT|socket.MSG_NOSIGNAL)!=len(renewal):raise OSError('network renewal incomplete') except OSError: channel.close();os.close(pidfd);del bridge.leases[channel] bridge.retire_management_lease() renew_at=time.monotonic()+.5 ready,_,_=select.select([control,*pending,*bridge.leases,*bridge.leases.values()],[],[],0) if control in ready: c,_=control.accept();c.setblocking(False) if len(pending)>=8:c.close() else:pending[c]=time.monotonic()+2 for c,pidfd in list(bridge.leases.items()): if c in ready or pidfd in ready: c.close();os.close(pidfd);del bridge.leases[c] bridge.retire_management_lease() for c,end in list(pending.items()): if time.monotonic()>end:c.close();del pending[c];continue if c not in ready:continue try: pid,uid,gid=struct.unpack('3i',c.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12)) actor=process(pid) if (uid,gid)!=(0,0) or actor['uids']!=[0]*4 or actor['gids']!=[0]*4 or actor['group'] not in ( '/system.slice/archipelago.service/supervisor', '/system.slice/archipelago.service/c01-host-io', '/system.slice/archipelago-c01-network-provision.service', '/system.slice/archipelago-c01-kiosk-endpoints.service', '/system.slice/archipelago-runtime-broker.service'):raise PermissionError('network actor') host_reader=actor['group']=='/system.slice/archipelago.service/c01-host-io' if host_reader:require_main(actor['parent'],'archipelago.service','/system.slice/archipelago.service/supervisor') else: unit='archipelago.service' if actor['group'].endswith('/supervisor') else actor['group'].split('/')[-1] require_main(pid,unit,actor['group']) raw,anc,flags,_=c.recvmsg(MAX+1,0) if anc or flags or not 0=1 or type(r['nonce']) is not str or not re.fullmatch('[0-9a-f]{64}',r['nonce']):raise ValueError('network lease bound') descriptors=[s.fileno() for s in bridge.listeners] if policy.value['version']==4: # Linux SCM_MAX_FD is 253; preserve larger exact # dual-stack inventories with ordered 64-FD chunks. for offset in range(0,len(descriptors),64): header=canonical({**r,'offset':offset,'total':len(descriptors)}) rights=array.array('i',descriptors[offset:offset+64]) if c.sendmsg([header],[(socket.SOL_SOCKET,socket.SCM_RIGHTS,rights)])!=len(header):raise OSError('partial listener lease') else: rights=array.array('i',descriptors) c.sendmsg([canonical(r)],[(socket.SOL_SOCKET,socket.SCM_RIGHTS,rights)]) bridge.leases[c]=os.pidfd_open(pid) # not kept alive by a forked socket copy else:raise ValueError('network operation not supported') except (OSError,ValueError,KeyError,TypeError) as error: frames=traceback.extract_tb(error.__traceback__) location=frames[-1] if frames else None print('C01 network request refused: '+type(error).__name__+ (' errno='+str(error.errno) if isinstance(error,OSError) else '')+ (' at '+Path(location.filename).name+':'+str(location.lineno) if location else ''), file=sys.stderr,flush=True) c.close() del pending[c] finally: for c in [control,*pending,*bridge.leases]:c.close() for pidfd in bridge.leases.values():os.close(pidfd) bridge.close();policy.close() if __name__=='__main__':main()