#!/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}))