"""Finite, externally appointed D09 launch-generation transitions. The access policy/runtime appointment never changes. Only native_sha256 may change in the launch JSON. An owned pending transition permits those exact two launch objects while refusing a new supervisor lease; runtime leases continue. This head is a transaction locator, not restart authority: Journal's existing offline requalification fence still governs access-owner restart. """ import copy import json import os from pathlib import Path import select import socket import struct import stat import ipaddress import c01_access_policy as policy DIRECTORY = Path('/var/lib/archipelago-c01/access/generations') HEAD = Path('/var/lib/archipelago-c01/access/generation-head.json') WORKSPACE = Path('/etc/systemd/system/archipelago.service.d/60-terminal-workspace.conf') STAGING = WORKSPACE.parent / '.d09-staging' def object_digest(path): if not os.path.lexists(path): return policy.sha(b'ABSENT') meta=policy.protected(path) return policy.sha(canonical([['.',stat.S_IMODE(meta.st_mode),meta.st_uid,meta.st_gid,policy.sha(policy.read(path))]])) def validate_launch(value): paths={'manager_backing','snapshot_backing','authority_backing','runtime_data_dir','snapshot_key','authority_key','platform_profile','proxy_path'} fields=paths|{'version','application_user','authority_user','runtime_uid','runtime_gid','binding','platform_profile_sha256','clock_domain','host_ip','native_sha256'} if value.get('version') in (4,5): fields.update(('epoch_sha256','power_sha256')) if value.get('version')==5: fields.add('admission_sha256') policy.require(type(value) is dict and set(value)==fields and type(value['version']) is int and value['version'] in (3,4,5), 'exact C01 launch schema') for key in paths: policy.require(type(value[key]) is str and len(value[key])<=4096 and Path(value[key]).is_absolute() and '..' not in Path(value[key]).parts,'C01 launch path') for key in ('application_user','authority_user'): policy.require(type(value[key]) is str and 00,'C01 clock domain') ipaddress.ip_address(value['host_ip']) b=value['binding'] policy.require(type(b) is dict and set(b)=={'local_identity','incarnation','reader_schema','writer_schema'},'C01 clock binding') for key in ('local_identity','incarnation'): policy.require(type(b[key]) is list and len(b[key])==32 and all(type(n) is int and 0<=n<=255 for n in b[key]) and any(b[key]),'C01 identity bytes') for key in ('reader_schema','writer_schema'): policy.require(type(b[key]) is int and b[key]==2,'C01 storage/clock ABI unchanged') def canonical(value): return json.dumps(value,sort_keys=True,separators=(',',':'),ensure_ascii=True,allow_nan=False).encode('ascii') class Generations: def __init__(self, value, digest, launch, launch_digest): self.policy, self.policy_digest = value, digest self.value = {'version':1,'policy_sha256':digest,'active_launch_sha256':value['launch_sha256'], 'active_workspace_digest':value.get('d09_workspace',{}).get('initial_digest'), 'pending':None,'consumed':[], 'journal':[]} self.anchor = None if HEAD.exists(): raw=policy.read(HEAD,True) self.value=json.loads(raw,object_pairs_hook=policy.exact) self.validate_head(self.value) self.anchor=policy.sha(raw) policy.require(launch_digest in self.allowed(),'unappointed launch generation') def validate_head(self,value): policy.require(type(value) is dict and set(value)=={'version','policy_sha256','active_launch_sha256','active_workspace_digest','pending','consumed','journal'} and type(value['version']) is int and value['version']==1 and value['policy_sha256']==self.policy_digest,'generation head binding') policy.require(type(value['consumed']) is list and type(value['journal']) is list,'generation history types') active=self.policy['launch_sha256'];workspace=self.policy.get('d09_workspace',{}).get('initial_digest') pending=None;consumed=[];grants=set() for event in value['journal']: policy.require(type(event) is dict and set(event)=={'operation','record_digest','grant_digest'} and event['operation'] in ('BEGIN','COMMIT') and policy.hex_digest(event['record_digest']) and policy.hex_digest(event['grant_digest']), 'generation event schema') record=self.record(event['record_digest']) policy.require(record['grant_digest']==event['grant_digest'],'generation event authority') if event['operation']=='BEGIN': policy.require(pending is None and event['grant_digest'] not in grants and record['from_launch_sha256']==active and record['from_workspace_digest']==workspace,'generation history replay/preimage') grants.add(event['grant_digest']);pending=event['record_digest'] else: policy.require(pending==event['record_digest'],'generation commit without owned intent') consumed.append(pending);pending=None active,workspace=record['to_launch_sha256'],record['to_workspace_digest'] expected=None if pending is None else {'record_digest':pending,'to_launch_sha256':self.record(pending)['to_launch_sha256']} policy.require(value['consumed']==consumed and value['pending']==expected and value['active_launch_sha256']==active and value['active_workspace_digest']==workspace,'generation head/history disagreement') def allowed(self): pending=self.value['pending'] return {self.value['active_launch_sha256']} | ({pending['to_launch_sha256']} if pending else set()) def check(self): if self.anchor is None: policy.require(not HEAD.exists(),'generation head appeared') else: policy.require(policy.sha(policy.read(HEAD,True))==self.anchor,'generation head continuity lost') def refresh(self, expected): if expected is None: policy.require(not HEAD.exists(),'unexpected generation head') return raw=policy.read(HEAD,True) policy.require(policy.sha(raw)==expected,'generation checker head binding') value=json.loads(raw,object_pairs_hook=policy.exact) self.validate_head(value) self.value,self.anchor=value,expected def save(self, value, event): self.check() value['journal'].append(event) self.validate_head(value) raw=canonical(value) policy.require(len(raw)<=policy.LIMIT,'generation history capacity; preserve existing authority') policy.atomic(HEAD,raw) self.value,self.anchor=value,policy.sha(raw) def record(self, hashed): policy.require(policy.hex_digest(hashed),'generation record digest') raw=policy.read(DIRECTORY/(hashed+'.json'),True) policy.require(policy.sha(raw)==hashed,'generation record changed') r=json.loads(raw,object_pairs_hook=policy.exact) fields={'version','policy_sha256','runtime_sha256','transaction_id','action','grant_digest', 'from_launch','to_launch','from_launch_sha256','to_launch_sha256','external_history_digest', 'from_workspace_digest','to_workspace_digest'} policy.require(set(r)==fields and type(r['version']) is int and r['version']==1 and r['action'] in ('APPLY','RESTORE') and r['policy_sha256']==self.policy_digest and r['runtime_sha256']==self.policy['runtime_sha256'], 'external generation authority binding') for key in ('grant_digest','from_launch_sha256','to_launch_sha256','external_history_digest','from_workspace_digest','to_workspace_digest'): policy.require(policy.hex_digest(r[key]),'generation digest domain') policy.require(type(r['transaction_id']) is str and len(r['transaction_id'])==32 and all(c in '0123456789abcdef' for c in r['transaction_id']) and int(r['transaction_id'],16)>0,'generation transaction') for key in ('from_launch','to_launch'): validate_launch(r[key]) policy.require(policy.sha(canonical(r[key]))==r[key+'_sha256'] and policy.hex_digest(r[key]['native_sha256']),'exact launch/native generation') policy.require({k:v for k,v in r['from_launch'].items() if k!='native_sha256'} == {k:v for k,v in r['to_launch'].items() if k!='native_sha256'}, 'generation cannot mutate principals, keys, proxy, data or runtime policy') return r def workspace_exclusions(self): if 'd09_workspace' not in self.policy: return set() delegation=self.policy['d09_workspace'] policy.require(set(delegation)=={'path','initial_digest'} and delegation['path']==str(WORKSPACE) and policy.hex_digest(delegation['initial_digest']),'finite workspace delegation') allowed={self.value['active_workspace_digest']} if self.value['pending']: r=self.record(self.value['pending']['record_digest']) allowed.update((r['to_workspace_digest'],policy.sha(b'ABSENT'))) policy.require(object_digest(WORKSPACE) in allowed,'unappointed workspace object') policy.protected(STAGING,private=True,directory=True) # This private non-unit directory is the sole staging namespace, not a # wildcard exception for other systemd units/drop-ins. items=list(STAGING.iterdir()) policy.require(len(items)<=4096,'workspace retained stage bound') for item in items: policy.protected(item) return {WORKSPACE,STAGING} def handle(self, request, quiescent): policy.require(set(request)=={'protocol','method','record_digest'} and request['protocol']=='d09-access-generation-v1' and request['method'] in ('inspect','begin','commit'),'generation operation schema') self.check() r=self.record(request['record_digest']) token=request['record_digest'] pending=self.value['pending'] state='PENDING' if pending and pending['record_digest']==token else 'COMMITTED' if token in self.value['consumed'] else 'UNKNOWN' method=request['method'] if method=='inspect': return {'state':state,'record_digest':token,'active_launch_sha256':self.value['active_launch_sha256'],'quiescent':quiescent} policy.require(quiescent,'old management retirement not observed') if method=='begin': policy.require(state=='UNKNOWN' and pending is None and r['from_launch_sha256']==self.value['active_launch_sha256'] and policy.json_read(policy.LAUNCH)[1]==r['from_launch_sha256'] and 'd09_workspace' in self.policy and r['from_workspace_digest']==self.value['active_workspace_digest'] and object_digest(WORKSPACE)==r['from_workspace_digest'], 'generation begin/replay/preimage') policy.require(not any(entry.get('grant_digest')==r['grant_digest'] for entry in self.value['journal']), 'generation grant consumed') value=copy.deepcopy(self.value) value['pending']={'record_digest':token,'to_launch_sha256':r['to_launch_sha256']} self.save(value,dict(operation='BEGIN',record_digest=token,grant_digest=r['grant_digest'])) state='PENDING' else: policy.require(state=='PENDING' and policy.json_read(policy.LAUNCH)[1]==r['to_launch_sha256'] and object_digest(WORKSPACE)==r['to_workspace_digest'], 'generation commit requires exact published launch') value=copy.deepcopy(self.value) value['active_launch_sha256']=r['to_launch_sha256'] value['active_workspace_digest']=r['to_workspace_digest'] value['consumed'].append(token) value['pending']=None self.save(value,dict(operation='COMMIT',record_digest=token,grant_digest=r['grant_digest'])) state='COMMITTED' return {'state':state,'record_digest':token,'active_launch_sha256':self.value['active_launch_sha256'],'quiescent':True} def request(method, record_digest): # Called only by D09's authenticated root worker handler; no app path input. from c01_access_client import SOCKET policy.protected(SOCKET.parent,directory=True) with socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET|socket.SOCK_CLOEXEC) as channel: channel.settimeout(3) channel.connect(str(SOCKET)) pid,uid,gid=struct.unpack('3i',channel.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12)) policy.require(pid>1 and (uid,gid)==(0,0),'access generation owner peer') raw=canonical({'protocol':'d09-access-generation-v1','method':method,'record_digest':record_digest}) policy.require(channel.send(raw)==len(raw),'generation request uncertain') raw,ancillary,flags,_=channel.recvmsg(8193,0) policy.require(not ancillary and not flags and 0