151 lines
12 KiB
Python
151 lines
12 KiB
Python
#!/usr/bin/python3
|
|
"""Emit network-only appointments from measured sockets and reviewed producers.
|
|
|
|
Producer bindings are supplied by the node/manual-unit planner. This emitter
|
|
never creates units, guesses future container IDs, or changes app decisions.
|
|
"""
|
|
import argparse,hashlib,ipaddress,json,re
|
|
from pathlib import Path
|
|
from c01_network_plan import exact,validate
|
|
from c01_retained_producer import NODE_LABEL
|
|
|
|
def canonical(value):return json.dumps(value,sort_keys=True,separators=(',',':'),ensure_ascii=True).encode()
|
|
def digest(value):return hashlib.sha256(canonical(value)).hexdigest()
|
|
|
|
def emit(inventory,settings):
|
|
fields={'version','node','hostname','boot_id','manager_pid','runtime','namespaces','uplink','mdns','producers','additional_fragments'}
|
|
if set(settings)!=fields or settings['version']!=1 or inventory['node']!=settings['hostname']:raise ValueError('network emitter input binding')
|
|
if type(settings['node']) is not str or not re.fullmatch(NODE_LABEL,settings['node']):raise ValueError('node binding')
|
|
import uuid
|
|
if str(uuid.UUID(settings['boot_id']))!=settings['boot_id']:raise ValueError('boot identity')
|
|
runtime=settings['runtime'];manager=settings['manager_pid'];appgroup='/system.slice/archipelago.service/c01-application'
|
|
by_group={}
|
|
for item in settings['producers']:
|
|
if set(item)!={'source_cgroup','source_pid','app_id','target_cgroup','manifest_sha256','producer','required'} or type(item['required']) is not bool:raise ValueError('producer binding fields')
|
|
by_group.setdefault(item['source_cgroup'],[]).append(item)
|
|
rows=[];seen=set();unattributed=[]
|
|
for row in inventory['listeners']:
|
|
if not row['processes']:unattributed.append(row);continue
|
|
selected=[p for p in row['processes'] if any(b['source_pid'] in (None,p['pid']) for b in by_group.get(p['cgroup'],[]))]
|
|
for actor in (selected or row['processes']):
|
|
s=actor.get('socket')
|
|
if s is None:raise ValueError('measured socket family/identity required')
|
|
identity=(s['net_cookie'],s['inode'])
|
|
if identity in seen:continue
|
|
seen.add(identity);rows.append((row['protocol'],actor,s))
|
|
gates={s['port'] for proto,p,s in rows if p['pid']==manager and proto=='tcp' and s['address']=='127.0.0.2'}
|
|
routes={};projections={};required=set();retained=[];used=set();runtime_ports={}
|
|
def append(name,source,target,protocol,port,addresses,address,target_port,uid,group,doc=None,mandatory=False):
|
|
projected=None
|
|
if doc is not None:
|
|
projected=digest(doc);projections[projected]=doc
|
|
item=dict(name=name,source=source,target=target,protocol=protocol,port=port,target_port=target_port,
|
|
app_id=doc['app_id'] if doc else None,projection_sha256=projected,target_uid=uid,target_cgroup=group,
|
|
source_addresses=sorted(set(addresses)),target_address=address)
|
|
if name in routes:
|
|
previous=routes[name]
|
|
if {k:v for k,v in previous.items() if k!='source_addresses'}!={k:v for k,v in item.items() if k!='source_addresses'}:raise ValueError('ambiguous route identity: '+name)
|
|
previous['source_addresses']=sorted(set(previous['source_addresses'])|set(addresses))
|
|
else:routes[name]=item
|
|
if mandatory:required.add(name)
|
|
for protocol,actor,sock in rows:
|
|
group=actor['cgroup']
|
|
migrated=group.startswith(runtime['user_cgroup']+'/') or group=='/system.slice/nostr-relay.service'
|
|
if not migrated:continue
|
|
matches=[b for b in by_group.get(group,[]) if b['source_pid'] in (None,actor['pid'])]
|
|
if len(matches)!=1:raise ValueError('exact reviewed producer binding required: '+group)
|
|
binding=matches[0];used.add((group,binding['source_pid']))
|
|
port=sock['port'];address=sock['address'];ip=ipaddress.ip_address(address)
|
|
public=[address]
|
|
if sock['family']==10 and address=='::' and sock['v6only']==0:public=['0.0.0.0','::']
|
|
target=('127.0.0.1' if sock['family']==2 or sock['v6only']==0 else '::1') if ip.is_unspecified else address
|
|
# The current native client interface remains IPv4 loopback. Exact
|
|
# private destinations retain explicit runtime binds where necessary.
|
|
for realm,addresses in (('host',public),('managed',['127.0.0.1'])):
|
|
doc=dict(app_id=binding['app_id'],manifest_sha256=binding['manifest_sha256'],loopback_ports=[[protocol,port]],
|
|
target_cgroup=binding['target_cgroup'],source_bindings=[[realm,protocol,port,a] for a in sorted(addresses)],
|
|
target_addresses=[target],producer=binding['producer'])
|
|
append(f'app/{binding["app_id"]}/{protocol}/{port}/{realm}',realm,'runtime',protocol,port,addresses,target,port,
|
|
actor['uid'][0],binding['target_cgroup'],doc,binding['required'])
|
|
runtime_ports.setdefault((protocol,port),[]).append(binding)
|
|
for protocol,actor,sock in rows:
|
|
port=sock['port'];address=sock['address'];group=actor['cgroup']
|
|
if actor['pid']==manager:
|
|
if protocol=='tcp' and port in gates:
|
|
candidates=runtime_ports.get(('tcp',port),[])
|
|
identities={(item['app_id'],item['manifest_sha256']) for item in candidates}
|
|
if len(identities)!=1:raise ValueError('gate backend identity is ambiguous: '+str(port))
|
|
app,manifest=next(iter(identities))
|
|
doc=dict(app_id=app,manifest_sha256=manifest,loopback_ports=[['tcp',port]],target_cgroup=appgroup)
|
|
append(f'gate/{app}/tcp/{port}/host','host','managed','tcp',port,[address],'127.0.0.2',port,
|
|
runtime['manager_uid'],appgroup,doc,True)
|
|
elif protocol=='tcp' and port==5678:
|
|
append('management','host','managed','tcp',port,['127.0.0.1'],'127.0.0.1',port,runtime['manager_uid'],appgroup,mandatory=True)
|
|
elif protocol=='tcp' and port==5679:
|
|
append('native/peer/tcp/5679/host','host','managed','tcp',port,[address],'127.0.0.1',port,runtime['manager_uid'],appgroup,mandatory=True)
|
|
elif protocol=='tcp' and 32000<=port<32132 and (port<32032 or port>=32100):
|
|
kind='website-fips' if port<32032 else 'website-tor'
|
|
append(f'native/{kind}/tcp/{port}/host','host','managed','tcp',port,[address],'127.0.0.1',port,runtime['manager_uid'],appgroup,mandatory=True)
|
|
elif protocol=='tcp' and port==7777:
|
|
# Preserve the old native IPv6 relay to the retained real relay.
|
|
candidates=runtime_ports.get(('tcp',7777),[])
|
|
if len(candidates)!=1 or candidates[0]['app_id']!='nostr-relay':raise ValueError('retained relay producer required')
|
|
binding=candidates[0];name='app/nostr-relay/tcp/7777/host';old=routes.pop(name)
|
|
addresses=sorted(set(old['source_addresses'])|{address})
|
|
doc=dict(app_id='nostr-relay',manifest_sha256=binding['manifest_sha256'],loopback_ports=[['tcp',7777]],
|
|
target_cgroup=binding['target_cgroup'],source_bindings=[['host','tcp',7777,a] for a in addresses],
|
|
target_addresses=[old['target_address']],producer=binding['producer'])
|
|
append(name,'host','runtime','tcp',7777,addresses,old['target_address'],7777,runtime['runtime_uid'],binding['target_cgroup'],doc,True)
|
|
elif protocol=='tcp' and port==8443:
|
|
append('fips-tcp6','host','host','tcp',8443,['::'],'127.0.0.1',8443,0,'/system.slice/archipelago-fips.service',mandatory=True)
|
|
elif protocol=='udp' and port==5353:
|
|
family=4 if sock['family']==2 else 6
|
|
append(f'native/mdns{family}/udp/5353/host','host','managed','udp',5353,['0.0.0.0' if family==4 else '::'],
|
|
'127.0.0.1' if family==4 else '::1',5353,runtime['manager_uid'],appgroup,mandatory=True)
|
|
elif protocol=='udp':retained.append({'socket':sock,'disposition':'native ephemeral outbound socket; regenerated by native transport'})
|
|
else:raise ValueError('unclassified native listener: '+str(port))
|
|
elif group.startswith(runtime['user_cgroup']+'/') or group=='/system.slice/nostr-relay.service':continue
|
|
elif group=='/system.slice/archipelago.service':raise ValueError('unclassified legacy native child producer')
|
|
elif group=='/system.slice/archipelago-kiosk.service' and protocol=='tcp' and port==9222:
|
|
retained.append({'socket':sock,'producer':actor,'disposition':'replaced by approved private runtime kiosk route'})
|
|
else:
|
|
retained.append({'socket':sock,'producer':actor,'disposition':'retain independent host producer'})
|
|
if group.endswith('/tor@default.service') and port==9050 and protocol=='tcp':
|
|
append('tor-socks','managed','host','tcp',port,['127.0.0.1'],'127.0.0.1',port,actor['uid'][0],group,mandatory=True)
|
|
if group=='/system.slice/archipelago-fips.service' and port==5354 and protocol=='udp':
|
|
append('fips-dns','managed','host','udp',port,['127.0.0.1'],'127.0.0.1',port,actor['uid'][0],group,mandatory=True)
|
|
for fragment in settings['additional_fragments']:
|
|
if set(fragment)!={'route','projection','required','provenance_sha256'} or len(fragment['provenance_sha256'])!=64:raise ValueError('qualified additional fragment')
|
|
r=fragment['route'];doc=fragment['projection']
|
|
if r['name'] in routes:raise ValueError('additional fragment collides with measured route')
|
|
routes[r['name']]=r
|
|
if doc is not None:projections[digest(doc)]=doc
|
|
if fragment['required']:required.add(r['name'])
|
|
policy=dict(version=4,namespaces=settings['namespaces'],uplink=settings['uplink'],
|
|
routes=sorted(routes.values(),key=lambda r:r['name']))
|
|
used_projections={r['projection_sha256'] for r in policy['routes'] if r['projection_sha256']}
|
|
policy['manifests']=[projections[key] for key in sorted(used_projections)]
|
|
if settings['mdns'] is not None:policy['mdns']=settings['mdns']
|
|
validate(policy,runtime)
|
|
if any(row['disposition']=='replaced by approved private runtime kiosk route' for row in retained) and 'app/archipelago-kiosk/tcp/9222/managed' not in routes:raise ValueError('approved kiosk replacement route missing')
|
|
unused=[item for item in settings['producers'] if (item['source_cgroup'],item['source_pid']) not in used]
|
|
if any(item['required'] for item in unused):raise ValueError('required producer lacks an observed publication')
|
|
return ({'node':settings['node'],'boot_id':settings['boot_id'],'value':{'policy':policy,'required_routes':sorted(required)}},
|
|
{'source_inventory_sha256':digest(inventory),'producer_bindings_sha256':digest(settings),
|
|
'retained_host_listeners':retained,'unattributed_host_sockets':unattributed,
|
|
'unused_producer_bindings':unused,'live_admission':False})
|
|
|
|
if __name__=='__main__':
|
|
parser=argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--inventory',required=True);parser.add_argument('--settings',required=True);parser.add_argument('--output',required=True)
|
|
args=parser.parse_args()
|
|
def load(path):
|
|
with open(path,'rb') as stream:raw=stream.read(4*1024*1024+1)
|
|
if len(raw)>4*1024*1024:raise ValueError('emitter input bound')
|
|
return json.loads(raw,object_pairs_hook=exact)
|
|
result,audit=emit(load(args.inventory),load(args.settings))
|
|
with open(args.output,'xb') as stream:stream.write(canonical(result)+b'\n')
|
|
with open(args.output+'.audit.json','xb') as stream:stream.write(canonical(audit)+b'\n')
|
|
print(json.dumps({'output':args.output,'routes':len(result['value']['policy']['routes']),'required':len(result['value']['required_routes']),
|
|
'unattributed':len(audit['unattributed_host_sockets']),'live_admission':False}))
|