249 lines
15 KiB
Python
249 lines
15 KiB
Python
"""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 0<len(value[key])<=32 and value[key].replace('-','').replace('_','').isalnum(),'C01 launch account')
|
|
for key in ('runtime_uid','runtime_gid'):
|
|
policy.require(type(value[key]) is int and 0<value[key]<2**32-1,'C01 runtime identity')
|
|
for key in ('platform_profile_sha256','native_sha256'):
|
|
policy.require(policy.hex_digest(value[key]),'C01 launch digest')
|
|
if value['version'] in (4,5): policy.require(policy.hex_digest(value['epoch_sha256']),'C01 epoch configuration digest')
|
|
if value['version'] in (4,5): policy.require(policy.hex_digest(value['power_sha256']),'C01 power configuration digest')
|
|
if value['version']==5: policy.require(policy.hex_digest(value['admission_sha256']),'C01 admission digest')
|
|
domain=value['clock_domain']
|
|
policy.require(type(domain) is str and len(domain)==32 and all(c in '0123456789abcdef' for c in domain) and int(domain,16)>0,'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<len(raw)<=8192,'generation reply framing')
|
|
response=json.loads(raw,object_pairs_hook=policy.exact)
|
|
policy.require(set(response)=={'state','record_digest','active_launch_sha256','quiescent'} and
|
|
response['record_digest']==record_digest and response['state'] in ('UNKNOWN','PENDING','COMMITTED') and
|
|
type(response['quiescent']) is bool and policy.hex_digest(response['active_launch_sha256']), 'generation reply binding')
|
|
return response
|
|
|
|
|
|
def worker(control, generations, supervisor_group, send, receive):
|
|
import c01_access_kernel as kernel
|
|
import c01_children
|
|
group=Path(supervisor_group)
|
|
while True:
|
|
if not select.select([control],[],[],1)[0]:
|
|
continue
|
|
message=receive(control)
|
|
try:
|
|
quiescent=not group.exists() or (not kernel.members(group) and not any(
|
|
(group/('c01-'+role)).exists() for role in c01_children.ROLES))
|
|
result=generations.handle(message,quiescent)
|
|
state={k:generations.value[k] for k in ('active_launch_sha256','active_workspace_digest','pending')}
|
|
send(control,{'result':result,'head':generations.anchor,'state':state})
|
|
except (OSError,ValueError,KeyError,TypeError):
|
|
# An uncertain write cannot be adopted by this helper on retry.
|
|
send(control,{'failed':True})
|