636 lines
40 KiB
Python
636 lines
40 KiB
Python
#!/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<p[1]<=65535 for p in doc['loopback_ports']):raise ValueError('manifest ports')
|
|
hashed=digest(canonical(doc))
|
|
if hashed in manifests:raise ValueError('duplicate projection')
|
|
manifests[hashed]=doc
|
|
self.routes={};ports=set()
|
|
self.manifests=manifests
|
|
for r in v['routes']:
|
|
validate_route(r,manifests,runtime,address_version=v['version'] if v['version'] in (3,4) else 1)
|
|
keys={(r['source'],r['protocol'],r['port'],address) for address in source_addresses(r)}
|
|
if r['name'] in self.routes or keys & ports:raise ValueError('duplicate/ambiguous route')
|
|
if any(source_binding_conflict(a,b) for a in keys for b in ports):raise ValueError('overlapping wildcard publication')
|
|
self.routes[r['name']]=r;ports.update(keys)
|
|
if len(ports)>(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'])<BUFFER else 0) | (selectors.EVENT_WRITE if r['buffer'] or r['connecting'] else 0)
|
|
if interest:
|
|
try:self.selector.modify(a,interest,('stream',name))
|
|
except KeyError:self.selector.register(a,interest,('stream',name))
|
|
else:
|
|
try:self.selector.unregister(a)
|
|
except KeyError:pass
|
|
if record['eof'] and self.streams[peer]['eof'] and not record['buffer'] and not self.streams[peer]['buffer']:self.drop(s)
|
|
except (OSError,ValueError,KeyError):self.drop(s)
|
|
for s,r in list(self.streams.items()):
|
|
if time.monotonic()>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<len(raw)<=MAX:raise ValueError('network control framing')
|
|
r=json.loads(raw,object_pairs_hook=exact)
|
|
if type(r) is not dict or r.get('policy_sha256')!=policy.sha:raise ValueError('network policy binding')
|
|
if host_reader and (set(r)!={'method','policy_sha256'} or r['method']!='gate-status'):
|
|
raise PermissionError('finite host gate observation only')
|
|
if actor['group']=='/system.slice/archipelago-c01-kiosk-endpoints.service' and (r.get('method') not in ('inspect','register') or r.get('name') not in (
|
|
'app/archipelago-kiosk/tcp/9222/managed','app/nostr-relay/tcp/7777/host','app/nostr-relay/tcp/7777/managed')):
|
|
raise PermissionError('finite kiosk registration scope')
|
|
if host_reader:
|
|
c.send(canonical(bridge.public_gate_status()));c.close()
|
|
elif set(r)=={'method','policy_sha256','name','pid','fd','previous'} and r['method']=='register':
|
|
if actor['group']=='/system.slice/archipelago.service/supervisor' and policy.routes.get(r['name'],{}).get('target')!='managed':raise ValueError('network registration scope')
|
|
if actor['group']=='/system.slice/archipelago-runtime-broker.service' and policy.routes.get(r['name'],{}).get('target')!='runtime':raise ValueError('runtime registration scope')
|
|
if r['previous'] is not None and (type(r['previous']) is not str or not re.fullmatch('[0-9a-f]{64}',r['previous'])):raise ValueError('endpoint generation token')
|
|
generation=bridge.register(r['name'],r['pid'],r['fd'],r['previous'])
|
|
c.send(canonical({'registered':r['name'],'policy_sha256':policy.sha,'generation':generation}));c.close()
|
|
elif set(r)=={'method','policy_sha256','name'} and r['method']=='inspect':
|
|
c.send(canonical({'policy_sha256':policy.sha,**bridge.registration_status(r['name'])}));c.close()
|
|
elif set(r)=={'method','policy_sha256','nonce'} and r['method']=='lease':
|
|
if actor['group']!='/system.slice/archipelago.service/supervisor':raise PermissionError('network lease role')
|
|
if len(bridge.leases)>=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()
|