385 lines
21 KiB
Python
385 lines
21 KiB
Python
"""Fixed manager-owned executor. All authoritative transitions go to live owner."""
|
|
import fcntl
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import socket
|
|
import struct
|
|
import time
|
|
|
|
from d09_store import atomic, read, secure, sync_dir
|
|
from d09_wire import MAX, canonical, parse
|
|
|
|
# Same implementation as the authority, loaded only from the protected package.
|
|
spec = importlib.util.spec_from_file_location('d09_owner', pathlib.Path(__file__).with_name('d09-lifecycle-owner.py'))
|
|
owner = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(owner)
|
|
|
|
|
|
def call(incarnation, method, **fields):
|
|
req = dict(incarnation_id=incarnation, method=method, **fields)
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as conn:
|
|
# A package checkpoint can wait for the explicitly bounded manager-bus
|
|
# proof window (max120s). This changes reply waiting, not authority life.
|
|
conn.settimeout(150 if method == 'os_checkpoint' else 10)
|
|
conn.connect(str(owner.PRIVATE))
|
|
pid, uid, gid = struct.unpack('3i', conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
|
|
if uid != 0:
|
|
raise PermissionError('owner peer')
|
|
conn.sendall(canonical(req))
|
|
conn.shutdown(socket.SHUT_WR)
|
|
raw = b''
|
|
while True:
|
|
block = conn.recv(8192)
|
|
if not block:
|
|
break
|
|
raw += block
|
|
if len(raw) > MAX:
|
|
raise ValueError('owner response size')
|
|
result = parse(raw)
|
|
if 'value' not in result:
|
|
raise PermissionError('owner refused continuity')
|
|
return result
|
|
|
|
|
|
class Worker:
|
|
def __init__(self, transaction, incarnation):
|
|
self.transaction, self.incarnation = transaction, incarnation
|
|
self.refresh()
|
|
if transaction != self.value['transaction_id']:
|
|
raise PermissionError('transaction mismatch')
|
|
|
|
def refresh(self):
|
|
result = call(self.incarnation, 'get')
|
|
self.binding, self.value = result['binding'], result['value']
|
|
|
|
def transition(self, phase, intent=None, consume=None, **fields):
|
|
result = call(self.incarnation, 'transition', updates=dict(phase=phase, **fields),
|
|
intent=intent, consume=consume)
|
|
self.binding, self.value = result['binding'], result['value']
|
|
|
|
def observed_start(self, action):
|
|
current = owner.service()
|
|
if current['ActiveState'] != 'active' or not current['InvocationID'] or current['MainPID'] == '0':
|
|
raise ValueError('activation not observed')
|
|
# Read the kernel's live executable reference, not the mutable pathname.
|
|
import hashlib
|
|
with open('/proc/' + str(int(current['MainPID'])) + '/exe', 'rb') as executable:
|
|
hashed = hashlib.file_digest(executable, 'sha256').hexdigest()
|
|
artifacts = self.binding['artifacts' if action == 'APPLY' else 'restore_artifacts']
|
|
if hashed != artifacts['backend']:
|
|
raise ValueError('unauthorized running artifact')
|
|
if current['InvocationID'] == self.value['receipt']['prior_invocation_id']:
|
|
raise ValueError('prior invocation still running')
|
|
try:
|
|
owner.PERMIT.unlink()
|
|
sync_dir(owner.RUN)
|
|
except FileNotFoundError:
|
|
pass
|
|
return current['InvocationID']
|
|
|
|
def disposition(self, name, action):
|
|
dest = owner.destinations(self.binding)[name]
|
|
spec = self.binding['objects'][name]
|
|
suffix = '.d09-' + self.transaction
|
|
staged = owner.snapshot(self.binding,name,'.new' if action=='APPLY' else '.restore')
|
|
retained = owner.snapshot(self.binding,name,'.old' if action=='APPLY' else '.failed')
|
|
wanted = spec['new' if action == 'APPLY' else 'restore']
|
|
before = spec['old' if action == 'APPLY' else 'new']
|
|
absent = owner.digest(b'ABSENT')
|
|
phase = 'CONFIG_INTENT' if name == 'config' else 'PAYLOAD_INTENT'
|
|
intent = dict(object=name, action=action, before=before, after=wanted,
|
|
stage=owner.object_digest(staged), retained=owner.object_digest(retained))
|
|
current = owner.object_digest(dest)
|
|
if action == 'RESTORE' and current == spec['old']:
|
|
before = current
|
|
old_intent = self.value['intent']
|
|
# Already-desired bytes are completion only after an attributable intent.
|
|
# Recovery may legitimately find the exact original preimage (including
|
|
# absent config), but still journals/restores/reloads before activation.
|
|
if current == wanted:
|
|
if action == 'APPLY' and wanted != spec['old'] and not (old_intent and old_intent.get('object') == name):
|
|
if self.value['receipt']['phase'] not in ('CONFIG_COMMITTED', 'PAYLOAD_COMMITTED', 'RELOAD_INTENT', 'RELOADED', 'START_INTENT', 'AWAITING_OBSERVATION'):
|
|
raise ValueError('unattributed postimage')
|
|
self.transition(phase, intent=intent)
|
|
return
|
|
if current != before and not (current == absent and owner.object_digest(retained) == before and old_intent and old_intent.get('object') == name):
|
|
raise ValueError('foreign payload/config')
|
|
if wanted != absent and owner.object_digest(staged) != wanted:
|
|
raise ValueError('missing/foreign staged payload')
|
|
self.transition(phase, intent=intent)
|
|
fd = secure(dest.parent, True)
|
|
stage_fd = None
|
|
try:
|
|
stage_fd = secure(staged.parent, True)
|
|
# Verify again under the externally provisioned writer exclusion.
|
|
if owner.object_digest(dest) == before and before != absent:
|
|
if retained.exists() or retained.is_symlink():
|
|
raise ValueError('retained evidence collision')
|
|
os.rename(dest.name, retained.name, src_dir_fd=fd, dst_dir_fd=stage_fd)
|
|
os.fsync(fd)
|
|
os.fsync(stage_fd)
|
|
if wanted != absent:
|
|
os.rename(staged.name, dest.name, src_dir_fd=stage_fd, dst_dir_fd=fd)
|
|
os.fsync(fd)
|
|
os.fsync(stage_fd)
|
|
if owner.object_digest(dest) != wanted:
|
|
raise ValueError('postimage mismatch')
|
|
finally:
|
|
if stage_fd is not None:
|
|
os.close(stage_fd)
|
|
os.close(fd)
|
|
|
|
def await_start(self, action):
|
|
end = time.monotonic() + 130
|
|
while time.monotonic() < end:
|
|
self.refresh()
|
|
current = owner.service()
|
|
if current['ActiveState'] == 'active':
|
|
return self.observed_start(action)
|
|
if (self.value.get('intent') or {}).get('start_submission')=='ACKNOWLEDGED' and current['ActiveState'] == 'failed' and current['MainPID'] == '0' and current['Job'] in ('', '0', '0 /'):
|
|
raise ValueError('manager observed failed start')
|
|
time.sleep(.1)
|
|
raise StartPending('one submitted start remains unresolved')
|
|
|
|
def generation(self, method):
|
|
result=call(self.incarnation,'access_generation',action=method)
|
|
self.binding,self.value=result['binding'],result['value']
|
|
return result['access_generation']
|
|
|
|
def prepare_generation(self):
|
|
fresh=self.value['receipt']['phase']=='STOPPED'
|
|
if fresh:
|
|
# Inspect until the actual old supervisor/role lifetime is retired.
|
|
# No deadline can turn an unreaped role into transition authority.
|
|
while not self.generation('inspect')['quiescent']:
|
|
time.sleep(.1)
|
|
self.transition('ACCESS_INTENT')
|
|
state=self.generation('begin' if fresh else 'inspect')
|
|
if state['state']!='PENDING':
|
|
raise ValueError('C01 generation begin outcome requires external reconciliation')
|
|
self.transition('ACCESS_PREPARED')
|
|
|
|
def commit_generation(self):
|
|
fresh=self.value['receipt']['phase']=='PAYLOAD_COMMITTED'
|
|
if fresh:
|
|
self.transition('ACCESS_COMMIT_INTENT')
|
|
state=self.generation('commit' if fresh else 'inspect')
|
|
if state['state']!='COMMITTED':
|
|
raise ValueError('C01 generation commit outcome requires external reconciliation')
|
|
self.transition('ACCESS_COMMITTED')
|
|
|
|
def execute(self, key, request):
|
|
if request['operation'] == 'ACK_OBSERVATION':
|
|
result = call(self.incarnation, 'observation', digest=request['observer_receipt_digest'])
|
|
obs = result['observation']
|
|
self.transition('COMPLETE', consume=key, outcome='COMPLETE', reason='NONE',
|
|
namespace_observation_digest=obs['namespace_observation_digest'],
|
|
preservation_observation_digest=obs['preservation_observation_digest'])
|
|
return
|
|
action = 'APPLY' if request['operation'] == 'APPLY' else 'RESTORE'
|
|
if 'os_update' in self.binding:
|
|
import os_update_executor
|
|
try:
|
|
os_update_executor.stage(self.binding, action, 'prepare',
|
|
checkpoint=lambda: call(self.incarnation, 'os_checkpoint'))
|
|
except Exception as exc:
|
|
if self.value['receipt']['phase'] in ('DISPATCH_INTENT', 'HANDOFF_READY', 'EXECUTING'):
|
|
raise OSPreparationRefused('OS preparation failed before service effects') from exc
|
|
raise
|
|
r = self.value['receipt']
|
|
if r['phase'] in ('AWAITING_OBSERVATION', 'COMPLETE', 'RETAINED') and r['config_action'] == action:
|
|
if action == 'RESTORE' and request['reason'] == 'AUTOMATIC_FAILURE':
|
|
self.transition('CONFLICT', consume=key, outcome='RECOVERY_REQUIRED', reason='PARTIAL_RECOVERY')
|
|
return
|
|
if owner.service()['InvocationID'] != r['observed_invocation_id']:
|
|
raise ValueError('no repeat activation under stale invocation')
|
|
self.transition(r['phase'], consume=key)
|
|
return
|
|
# A worker lost around start must observe the existing exact successor;
|
|
# it may never stop it and blindly repeat the activation.
|
|
if r['phase'] == 'START_INTENT':
|
|
current = owner.service()
|
|
if current['ActiveState'] == 'active' and current['InvocationID'] != r['prior_invocation_id']:
|
|
self.transition('AWAITING_OBSERVATION', consume=key,
|
|
observed_invocation_id=self.observed_start(action))
|
|
return
|
|
invocation = self.await_start(action)
|
|
self.transition('AWAITING_OBSERVATION', consume=key, observed_invocation_id=invocation)
|
|
return
|
|
if r['phase'] in ('DISPATCH_INTENT', 'HANDOFF_READY', 'COMPLETE', 'AWAITING_OBSERVATION'):
|
|
self.transition('EXECUTING', config_action=action, outcome='PENDING', reason='NONE',
|
|
namespace_observation_digest=None, preservation_observation_digest=None,
|
|
observed_invocation_id=None)
|
|
if self.value['receipt']['phase'] in ('EXECUTING', 'STOP_INTENT'):
|
|
current = owner.service()
|
|
stopped = current['MainPID'] == '0' and current['ActiveState'] in ('inactive', 'failed') and current['Job'] in ('', '0', '0 /')
|
|
if self.value['receipt']['phase'] == 'STOP_INTENT' and not stopped and current['Job'] not in ('', '0', '0 /'):
|
|
raise ValueError('outstanding stop job must resolve before retry')
|
|
self.transition('STOP_INTENT', prior_invocation_id=current['InvocationID'] or None,
|
|
config_action=action)
|
|
try:
|
|
owner.PERMIT.unlink()
|
|
sync_dir(owner.RUN)
|
|
except FileNotFoundError:
|
|
pass
|
|
if not stopped:
|
|
owner.command('/usr/bin/systemctl', 'stop', '--no-block', owner.SERVICE)
|
|
while True:
|
|
self.refresh()
|
|
current = owner.service()
|
|
if current['MainPID'] == '0' and current['ActiveState'] in ('inactive', 'failed') and current['Job'] in ('', '0', '0 /'):
|
|
break
|
|
time.sleep(.1)
|
|
self.transition('STOPPED')
|
|
if 'c01' in self.binding and self.value['receipt']['phase'] in ('STOPPED','ACCESS_INTENT'):
|
|
self.prepare_generation()
|
|
if 'os_update' in self.binding and self.value['receipt']['phase'] in ('STOPPED', 'ACCESS_PREPARED'):
|
|
import os_update_executor
|
|
os_update_executor.stage(self.binding, action, 'apply',
|
|
checkpoint=lambda: call(self.incarnation, 'os_checkpoint'))
|
|
phase = self.value['receipt']['phase']
|
|
if phase in ('STOPPED', 'ACCESS_PREPARED', 'CONFIG_INTENT'):
|
|
if action == 'APPLY':
|
|
helper = pathlib.Path('/usr/libexec/archipelago-d09-prepare-workspace')
|
|
if owner.digest(read(helper)) != self.binding['helper_digest']:
|
|
raise ValueError('helper changed after admission')
|
|
# This helper deliberately rejects root and preserves existing
|
|
# Work/history. Its executable and HOME are fixed, not wire data.
|
|
if 'c01' in self.binding:
|
|
import d09_c01
|
|
d09_c01.workspace(self.binding)
|
|
else:
|
|
owner.command('/usr/bin/setpriv', '--reuid=' + str(self.binding['requester_uid']),
|
|
'--regid=' + str(self.binding['requester_gid']), '--init-groups',
|
|
'/usr/bin/env', '-i', 'HOME=/home/archipelago', 'PATH=/usr/bin:/bin',
|
|
'/bin/bash', str(helper), '/home/archipelago')
|
|
spec = self.binding['objects']['config']
|
|
self.transition('CONFIG_INTENT', intent=self.value['intent'], config_action=action,
|
|
config_pre_digest=spec['old'], config_post_digest=spec['new'],
|
|
config_current_digest=owner.object_digest(owner.DESTINATIONS['config']))
|
|
self.disposition('config', action)
|
|
self.transition('CONFIG_COMMITTED', config_current_digest=owner.object_digest(owner.DESTINATIONS['config']))
|
|
if self.value['receipt']['phase'] in ('CONFIG_COMMITTED', 'PAYLOAD_INTENT'):
|
|
start = (self.value['intent'] or {}).get('object')
|
|
names = ['frontend', 'backend'] + (['launch'] if 'c01' in self.binding else [])
|
|
if start in names:
|
|
names = names[names.index(start):]
|
|
for name in names:
|
|
self.disposition(name, action)
|
|
# A committed per-object intent is retained until the next one.
|
|
self.transition('PAYLOAD_INTENT', intent=dict(object=name, action=action, committed=True))
|
|
self.transition('PAYLOAD_COMMITTED', frontend_restoration='SUCCEEDED' if action == 'RESTORE' else 'NOT_REQUIRED',
|
|
backend_restoration='SUCCEEDED' if action == 'RESTORE' else 'NOT_REQUIRED')
|
|
if 'c01' in self.binding and self.value['receipt']['phase'] in ('PAYLOAD_COMMITTED','ACCESS_COMMIT_INTENT'):
|
|
self.commit_generation()
|
|
if self.value['receipt']['phase'] in ('PAYLOAD_COMMITTED', 'ACCESS_COMMITTED', 'RELOAD_INTENT'):
|
|
reopening = self.value['receipt']['phase'] == 'RELOAD_INTENT'
|
|
self.transition('RELOAD_INTENT')
|
|
if reopening:
|
|
if owner.command('/usr/bin/systemctl', 'show', owner.SERVICE, '-p', 'NeedDaemonReload', '--value') != 'no':
|
|
raise ValueError('reload result unresolved; no blind manager retry')
|
|
else:
|
|
owner.command('/usr/bin/systemctl', 'daemon-reload')
|
|
effective = owner.effective_configuration()
|
|
expected = self.binding['effective_config'][action]
|
|
if owner.digest(effective.encode()) != expected:
|
|
raise ValueError('effective config differs from admission')
|
|
self.transition('RELOADED', effective_config_observation_digest=expected)
|
|
if self.value['receipt']['phase'] == 'RELOADED':
|
|
self.transition('START_INTENT',intent={'start_submission':'INTENT','action':action})
|
|
atomic(owner.PERMIT, canonical({'transaction_id': self.transaction}), 0o600)
|
|
# Exactly one ordinary start. Lost replies leave START_INTENT.
|
|
owner.command('/usr/bin/systemctl', 'start', '--no-block', owner.SERVICE)
|
|
self.transition('START_INTENT',intent={'start_submission':'ACKNOWLEDGED','action':action,'manager_job':owner.service()['Job']})
|
|
invocation = self.await_start(action)
|
|
self.transition('AWAITING_OBSERVATION', consume=key, observed_invocation_id=invocation)
|
|
|
|
def reactivate(self):
|
|
"""Externally consumed one-use grant for the retained exact generation."""
|
|
r = self.value['receipt']
|
|
action = r['config_action']
|
|
if r['phase'] == 'START_INTENT':
|
|
# Lost manager/owner acknowledgement: observation only, never start.
|
|
invocation = self.observed_start(action)
|
|
self.transition('AWAITING_OBSERVATION', observed_invocation_id=invocation)
|
|
return
|
|
if r['phase'] != 'REACTIVATION_READY' or action not in ('APPLY', 'RESTORE'):
|
|
raise ValueError('invalid external activation phase')
|
|
current = owner.service()
|
|
if current['MainPID'] != '0' or current['ActiveState'] not in ('inactive', 'failed') or current['Job'] not in ('', '0', '0 /'):
|
|
raise ValueError('activation process/job not drained')
|
|
generation = 'new' if action == 'APPLY' else 'restore'
|
|
for name, dest in owner.destinations(self.binding).items():
|
|
if owner.object_digest(dest) != self.binding['objects'][name][generation]:
|
|
raise ValueError('retained generation changed')
|
|
effective = owner.effective_configuration()
|
|
if owner.digest(effective.encode()) != self.binding['effective_config'][action]:
|
|
raise ValueError('retained effective configuration changed')
|
|
self.transition('START_INTENT', intent={'start_submission':'INTENT','action':action}, effective_config_observation_digest=owner.digest(effective.encode()))
|
|
atomic(owner.PERMIT, canonical({'transaction_id': self.transaction,
|
|
'execution_grant_digest': self.value['activation']['execution_grant_digest']}))
|
|
owner.command('/usr/bin/systemctl', 'start', '--no-block', owner.SERVICE)
|
|
self.transition('START_INTENT',intent={'start_submission':'ACKNOWLEDGED','action':action,'manager_job':owner.service()['Job']})
|
|
self.transition('AWAITING_OBSERVATION', observed_invocation_id=self.await_start(action))
|
|
|
|
|
|
class StartPending(Exception):
|
|
pass
|
|
|
|
|
|
class OSPreparationRefused(Exception):
|
|
pass
|
|
|
|
|
|
def run(transaction, incarnation):
|
|
lock = os.open(owner.ROOT / 'lifecycle.lock', os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
|
|
fcntl.flock(lock, fcntl.LOCK_EX)
|
|
worker = Worker(transaction, incarnation)
|
|
try:
|
|
if worker.value.get('activation'):
|
|
worker.reactivate()
|
|
worker.refresh()
|
|
while worker.value['inbox']:
|
|
key = worker.value['inbox'][0]
|
|
worker.execute(key, worker.value['requests'][key]['request'])
|
|
worker.refresh()
|
|
except StartPending:
|
|
# Preserve the exact intent/permit/notice. No timeout authorizes a start.
|
|
return
|
|
except OSPreparationRefused:
|
|
# A rejected dependency/repository plan must not enter the generic
|
|
# post-effects recovery path, which deliberately stops the manager.
|
|
worker.transition('CONFLICT', intent=worker.value['intent'], outcome='RECOVERY_REQUIRED', reason='PARTIAL_RECOVERY')
|
|
raise
|
|
except Exception:
|
|
# A definitively failed ordinary activation may use the admitted failure
|
|
# scope once. Unknown manager jobs/processes must remain held instead.
|
|
r = worker.value['receipt']
|
|
if worker.value.get('activation') and r['phase'] == 'START_INTENT':
|
|
try:
|
|
invocation = worker.observed_start(r['config_action'])
|
|
worker.transition('AWAITING_OBSERVATION', observed_invocation_id=invocation)
|
|
return
|
|
except Exception:
|
|
pass
|
|
current = owner.service()
|
|
if not worker.value.get('activation') and (worker.value.get('intent') or {}).get('start_submission')=='ACKNOWLEDGED' and r['phase'] == 'START_INTENT' and r['config_action'] == 'APPLY' and current['MainPID'] == '0' and current['ActiveState'] in ('inactive', 'failed') and current['Job'] in ('', '0', '0 /'):
|
|
try:
|
|
worker.transition('EXECUTING', config_action='RESTORE', outcome='PENDING', reason='AUTOMATIC_FAILURE')
|
|
worker.execute(key, dict(worker.value['requests'][key]['request'], operation='RECOVER'))
|
|
return
|
|
except Exception:
|
|
pass
|
|
# Preserve all bytes and the outstanding notice. No local downgrade or
|
|
# cleanup, even if the authority was lost while the worker was alive.
|
|
try:
|
|
owner.PERMIT.unlink()
|
|
sync_dir(owner.RUN)
|
|
except FileNotFoundError:
|
|
pass
|
|
owner.command('/usr/bin/systemctl', '--no-block', 'stop', owner.SERVICE)
|
|
worker.transition('CONFLICT', intent=worker.value['intent'], outcome='RECOVERY_REQUIRED', reason='PARTIAL_RECOVERY')
|
|
raise
|