201 lines
13 KiB
Python
201 lines
13 KiB
Python
"""Live root supervisor lease; pin source listeners through complete retirement."""
|
|
import array
|
|
import ctypes
|
|
import errno
|
|
import json
|
|
import os
|
|
import select
|
|
import socket
|
|
import struct
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from c01_network_owner import CONFIG,CONTROL,Policy,canonical,exact,protected_read,digest,process,require_main,LIBC,listener_address_matches
|
|
from c01_endpoints import source_addresses
|
|
|
|
|
|
def receive_listeners(channel,request,total,version,listeners):
|
|
if not 1<=total<=(512 if version==4 else 64):raise ValueError('listener lease bound')
|
|
for offset in range(0,total,64):
|
|
raw,anc,flags,_=channel.recvmsg(8193,socket.CMSG_SPACE(64*4),socket.MSG_CMSG_CLOEXEC)
|
|
unexpected=False;before=len(listeners)
|
|
for level,kind,data in anc:
|
|
if (level,kind)==(socket.SOL_SOCKET,socket.SCM_RIGHTS):
|
|
fds=array.array('i');fds.frombytes(data[:len(data)-len(data)%4]);listeners.extend(fds)
|
|
unexpected |= len(data)%4!=0
|
|
else:unexpected=True
|
|
expected={**request,'offset':offset,'total':total} if version==4 else request
|
|
if unexpected or flags & ~socket.MSG_CMSG_CLOEXEC or len(raw)>8192 or json.loads(raw,object_pairs_hook=exact)!=expected or len(listeners)-before!=min(64,total-offset):
|
|
raise ValueError('network lease binding/rights')
|
|
|
|
|
|
class NetworkLease:
|
|
def __init__(self, manager_uid=None, runtime_uid=None):
|
|
self.channel=socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET|socket.SOCK_CLOEXEC)
|
|
self.pidfd=None;self.listeners=[];self.policy=None;self.management=None;self.required_targets={}
|
|
self.required_sockets={}
|
|
self.required_generations={};self.required_since={}
|
|
try:
|
|
self.policy=Policy();self.policy_digest=self.policy.sha
|
|
policy=self.policy.value
|
|
self.protocol=policy['version']
|
|
if manager_uid is not None:
|
|
route=self.policy.routes.get('management')
|
|
if route is None or route['target_uid']!=manager_uid:raise ValueError('network manager appointment')
|
|
# Policy has already validated each route's explicit producer
|
|
# principal. The runtime account need not equal a container's
|
|
# subordinate UID; bind the account appointment itself here.
|
|
# Management-only legacy policies have no runtime authority to bind.
|
|
if runtime_uid is not None:
|
|
if self.policy.runtime is None:
|
|
if any(r['target']=='runtime' for r in policy['routes']):raise ValueError('network runtime appointment')
|
|
elif self.policy.runtime['runtime_uid']!=runtime_uid:raise ValueError('network runtime appointment')
|
|
self.channel.settimeout(3);self.channel.connect(str(CONTROL))
|
|
pid,uid,gid=struct.unpack('3i',self.channel.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12))
|
|
if (uid,gid)!=(0,0) or process(pid)['group']!='/system.slice/archipelago-c01-network.service':raise PermissionError('network owner identity')
|
|
self.owner= require_main(pid,'archipelago-c01-network.service','/system.slice/archipelago-c01-network.service')
|
|
self.pidfd=self.channel.getsockopt(socket.SOL_SOCKET,77);os.set_inheritable(self.pidfd,False)
|
|
request={'method':'lease','policy_sha256':self.policy_digest,'nonce':os.urandom(32).hex()}
|
|
self.expiry=0;self.start_deadline=time.clock_gettime_ns(time.CLOCK_BOOTTIME)+3_000_000_000
|
|
self.channel.sendall(canonical(request))
|
|
routes=[(route,address) for route in policy['routes'] for address in source_addresses(route)]
|
|
receive_listeners(self.channel,request,len(routes),policy['version'],self.listeners)
|
|
for fd,(route,address) in zip(self.listeners,routes):
|
|
with socket.socket(fileno=os.dup(fd)) as listener:
|
|
kind=socket.SOCK_STREAM if route['protocol']=='tcp' else socket.SOCK_DGRAM
|
|
if listener.getsockopt(socket.SOL_SOCKET,socket.SO_TYPE)!=kind or listener.getsockname()[:2]!=(address,route['port']):raise ValueError('network listener capability')
|
|
with self.policy.socket(route['source'],kind,family=listener.family) as probe:
|
|
if listener.getsockopt(socket.SOL_SOCKET,71,8)!=probe.getsockopt(socket.SOL_SOCKET,71,8):raise ValueError('network source namespace')
|
|
if kind==socket.SOCK_STREAM and not listener.getsockopt(socket.SOL_SOCKET,socket.SO_ACCEPTCONN):raise ValueError('network ingress not listening')
|
|
self.channel.setblocking(False);self.check()
|
|
except BaseException:self.close();raise
|
|
def check(self):
|
|
if self.pidfd is None or select.select([self.pidfd],[],[],0)[0]:raise ValueError('network capability lifetime lost')
|
|
for _ in range(16):
|
|
if not select.select([self.channel],[],[],0)[0]:break
|
|
bound=65536 if getattr(self,'protocol',1)==4 else 256
|
|
raw,anc,flags,_=self.channel.recvmsg(bound+1,0)
|
|
if anc or flags or not 0<len(raw)<=bound:raise ValueError('network renewal framing/lifetime lost')
|
|
value=json.loads(raw,object_pairs_hook=exact)
|
|
fields={'alive_until_boottime_ns'}|({'generations'} if getattr(self,'protocol',1)==4 else set())
|
|
if set(value)!=fields:raise ValueError('network renewal schema')
|
|
expiry=value['alive_until_boottime_ns']
|
|
if type(expiry) is not int or expiry<self.expiry:raise ValueError('network renewal regressed')
|
|
self.expiry=expiry
|
|
if getattr(self,'protocol',1)==4:
|
|
generations=value['generations']
|
|
if type(generations) is not dict or set(generations)!=set(self.policy.routes):raise ValueError('network generation inventory')
|
|
for name,expected in self.required_generations.items():
|
|
if expiry-3_000_000_000>=self.required_since[name] and generations[name]!=expected:
|
|
raise ValueError('required producer generation lost: '+name)
|
|
stamp=time.clock_gettime_ns(time.CLOCK_BOOTTIME)
|
|
if self.expiry:
|
|
if not stamp<self.expiry<=stamp+3_000_000_000:raise ValueError('network owner progress expired')
|
|
elif stamp>=self.start_deadline:raise ValueError('network initial progress absent')
|
|
if process(self.owner['pid'])!=self.owner:raise ValueError('network owner identity changed')
|
|
self.policy.check()
|
|
for name,(pidfd,expected) in self.required_targets.items():
|
|
if select.select([pidfd],[],[],0)[0] or canonical(process(expected['pid']))!=canonical(expected):
|
|
raise ValueError('required network producer lifetime lost: '+name)
|
|
if name in self.required_sockets:
|
|
number,inode=self.required_sockets[name]
|
|
if os.readlink(f'/proc/{expected["pid"]}/fd/{number}')!=f'socket:[{inode}]':
|
|
raise ValueError('required network producer descriptor lost: '+name)
|
|
def require_targets(self,names):
|
|
for name in names:
|
|
if name not in self.policy.routes:raise ValueError('required named route absent')
|
|
if name in self.required_targets:continue
|
|
status=self.request({'method':'inspect','name':name})
|
|
expected_fields={'policy_sha256','name','generation','producer'}|({'listener'} if self.policy.value['version']==4 else set())
|
|
if set(status)!=expected_fields or status['name']!=name:
|
|
raise ValueError('required named producer reply binding')
|
|
producer=status.get('producer')
|
|
if type(producer) is not dict or type(producer.get('pid')) is not int:raise ValueError('required named producer absent')
|
|
fd=os.pidfd_open(producer['pid'])
|
|
try:
|
|
if select.select([fd],[],[],0)[0] or canonical(process(producer['pid']))!=canonical(producer):
|
|
raise ValueError('required named producer changed')
|
|
self.required_targets[name]=(fd,producer)
|
|
if self.policy.value['version']==4:
|
|
listener=status['listener']
|
|
if type(listener) is not dict or set(listener)!={'number','inode'} or type(listener['number']) is not int or not 0<=listener['number']<=65535 or type(listener['inode']) is not int or listener['inode']<=0:
|
|
del self.required_targets[name]
|
|
raise ValueError('required producer descriptor binding')
|
|
self.required_sockets[name]=(listener['number'],listener['inode'])
|
|
if type(status['generation']) is not str or len(status['generation'])!=64:raise ValueError('required producer generation')
|
|
self.required_generations[name]=status['generation'];self.required_since[name]=time.clock_gettime_ns(time.CLOCK_BOOTTIME)
|
|
except BaseException:os.close(fd);raise
|
|
self.check()
|
|
def request(self, body):
|
|
self.check()
|
|
with socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET|socket.SOCK_CLOEXEC) as channel:
|
|
channel.settimeout(3);channel.connect(str(CONTROL))
|
|
if struct.unpack('3i',channel.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12))!=(self.owner['pid'],0,0):raise ValueError('network owner replaced')
|
|
channel.sendall(canonical({'policy_sha256':self.policy_digest,**body}))
|
|
raw,anc,flags,_=channel.recvmsg(8193,0)
|
|
if anc or flags or not 0<len(raw)<=8192:raise ValueError('network reply framing')
|
|
reply=json.loads(raw,object_pairs_hook=exact)
|
|
if reply.get('policy_sha256')!=self.policy_digest:raise ValueError('network reply policy')
|
|
self.check()
|
|
return reply
|
|
def register_management(self, pid, fd):
|
|
if self.management is not None:raise ValueError('management already registered')
|
|
status=self.request({'method':'inspect','name':'management'})
|
|
if status.get('producer') is not None:raise ValueError('previous management producer active')
|
|
reply=self.request({'method':'register','name':'management','pid':pid,'fd':fd,'previous':status['generation']})
|
|
if set(reply)!={'registered','policy_sha256','generation'} or reply['registered']!='management':raise ValueError('management registration reply')
|
|
self.management=reply['generation']
|
|
|
|
def register_managed_listeners(self,pid,names):
|
|
"""Discover only this native child's finite appointed listening sockets.
|
|
|
|
The owner independently authenticates each producer and pins its socket.
|
|
No connection is redirected to a runtime port when a gate is absent.
|
|
"""
|
|
routes={name:self.policy.routes[name] for name in names
|
|
if name!='management' and self.policy.routes[name]['target']=='managed'}
|
|
pending={}
|
|
for name,route in routes.items():
|
|
status=self.request({'method':'inspect','name':name})
|
|
if status.get('producer') is None:pending[name]=(route,status['generation'])
|
|
elif status['producer']['pid']!=pid:raise ValueError('managed producer differs from native child')
|
|
if not pending:return
|
|
entries=list(Path(f'/proc/{pid}/fd').iterdir())
|
|
if len(entries)>4096:raise ValueError('native listener inventory exceeds bound')
|
|
pidfd=os.pidfd_open(pid);matches={name:{} for name in pending}
|
|
try:
|
|
for entry in entries:
|
|
try:
|
|
if not os.readlink(entry).startswith('socket:['):continue
|
|
except FileNotFoundError:continue
|
|
fd=LIBC.syscall(438,pidfd,int(entry.name),0)
|
|
if fd<0:
|
|
error=ctypes.get_errno()
|
|
if error==errno.EBADF:continue
|
|
raise OSError(error,'native socket inventory unavailable')
|
|
with socket.socket(fileno=fd) as listener:
|
|
if listener.family not in (socket.AF_INET,socket.AF_INET6):continue
|
|
kind=listener.getsockopt(socket.SOL_SOCKET,socket.SO_TYPE)
|
|
if kind==socket.SOCK_STREAM and not listener.getsockopt(socket.SOL_SOCKET,socket.SO_ACCEPTCONN):continue
|
|
if kind not in (socket.SOCK_STREAM,socket.SOCK_DGRAM):continue
|
|
for name,(route,_) in pending.items():
|
|
if kind==(socket.SOCK_STREAM if route['protocol']=='tcp' else socket.SOCK_DGRAM) and listener_address_matches(listener,route):matches[name].setdefault(os.fstat(listener.fileno()).st_ino,int(entry.name))
|
|
if select.select([pidfd],[],[],0)[0]:raise ValueError('native producer exited during inventory')
|
|
for name,numbers in matches.items():
|
|
if len(numbers)>1:raise ValueError('ambiguous native listener')
|
|
if not numbers:continue
|
|
reply=self.request({'method':'register','name':name,'pid':pid,'fd':next(iter(numbers.values())),'previous':pending[name][1]})
|
|
if set(reply)!={'registered','policy_sha256','generation'} or reply['registered']!=name:
|
|
raise ValueError('native listener registration reply')
|
|
finally:os.close(pidfd)
|
|
def close(self):
|
|
self.channel.close()
|
|
for fd,_ in self.required_targets.values():os.close(fd)
|
|
self.required_targets={}
|
|
self.required_sockets={}
|
|
self.required_generations={};self.required_since={}
|
|
for fd in self.listeners:os.close(fd)
|
|
self.listeners=[]
|
|
if self.pidfd is not None:os.close(self.pidfd);self.pidfd=None
|
|
if self.policy is not None:self.policy.close();self.policy=None
|