1164 lines
64 KiB
Python
Executable File
1164 lines
64 KiB
Python
Executable File
#!/usr/bin/python3 -I
|
|
"""Fixed, outside-updater D09 authority. Public transport: submit and inspect.
|
|
|
|
The manager-owned worker holds the lifecycle lock across stop/config/payload/
|
|
reload/start. Only this still-live authority may advance its protected journal.
|
|
Neither worker nor managed updater can seed continuity from a restored ledger.
|
|
"""
|
|
import argparse
|
|
import copy
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import secrets
|
|
import socket
|
|
import stat
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import traceback
|
|
|
|
# -I excludes caller PYTHONPATH and cwd. This directory is installed root-owned.
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
|
|
from d09_wire import MAX, canonical, decode, digest, parse, validate, release_digest
|
|
from d09_store import Store, ContinuityError, atomic, read, secure, sync_dir
|
|
|
|
ROOT = pathlib.Path('/var/lib/archipelago-d09')
|
|
RUN = pathlib.Path('/run/archipelago-d09')
|
|
BINDING = pathlib.Path('/etc/archipelago-d09/binding.json')
|
|
SOCKET = RUN / 'lifecycle.sock'
|
|
PRIVATE = RUN / 'owner.sock'
|
|
SERVICE = 'archipelago.service'
|
|
PROGRAM = '/usr/libexec/archipelago-d09-owner'
|
|
PERMIT = RUN / 'start-permit'
|
|
DESTINATIONS = {'frontend': pathlib.Path('/opt/archipelago/web-ui'),
|
|
'backend': pathlib.Path('/usr/local/bin/archipelago'),
|
|
'config': pathlib.Path('/etc/systemd/system/archipelago.service.d/60-terminal-workspace.conf')}
|
|
|
|
|
|
def destinations(binding):
|
|
return dict(DESTINATIONS, launch=pathlib.Path('/etc/archipelago-c01/launch.json')) if 'c01' in binding else DESTINATIONS
|
|
|
|
|
|
def snapshot(binding, name, suffix):
|
|
dest=destinations(binding)[name]
|
|
if 'c01' in binding and name=='config':
|
|
return dest.parent/'.d09-staging'/(binding['transaction_id']+suffix)
|
|
return dest.with_name(dest.name+'.d09-'+binding['transaction_id']+suffix)
|
|
|
|
|
|
def boot_id():
|
|
return pathlib.Path('/proc/sys/kernel/random/boot_id').read_text().strip().replace('-', '')
|
|
|
|
|
|
def process(pid):
|
|
raw = pathlib.Path('/proc/%d/stat' % pid).read_text()
|
|
return {'pid': pid, 'start_ticks': int(raw[raw.rindex(')') + 2:].split()[19]),
|
|
'boot_id': boot_id()}
|
|
|
|
|
|
def command(*args):
|
|
return subprocess.check_output(args, timeout=30, env={'PATH': '/usr/sbin:/usr/bin:/sbin:/bin', 'LANG': 'C'}, text=True).strip()
|
|
|
|
|
|
def service():
|
|
return dict(line.split('=', 1) for line in command('/usr/bin/systemctl', 'show', SERVICE,
|
|
'-p', 'InvocationID', '-p', 'MainPID', '-p', 'ActiveState', '-p', 'Job').splitlines())
|
|
|
|
|
|
def worker_disposition(execution):
|
|
unit=execution['unit']
|
|
if not re.fullmatch(r'archipelago-d09-[0-9a-f]{32}-[0-9a-f]{32}\.service',unit):
|
|
raise ValueError('invalid retained worker unit')
|
|
group=pathlib.Path('/sys/fs/cgroup/system.slice')/unit
|
|
try:
|
|
meta=group.lstat()
|
|
except FileNotFoundError:
|
|
return None
|
|
if not stat.S_ISDIR(meta.st_mode) or meta.st_uid!=0 or meta.st_mode & 0o022:
|
|
raise ValueError('worker cgroup ownership changed')
|
|
events=dict(line.split() for line in (group/'cgroup.events').read_text().splitlines())
|
|
if events.get('populated') not in ('0','1'):
|
|
raise ValueError('worker subtree disposition unknown')
|
|
return {'device':meta.st_dev,'inode':meta.st_ino,'populated':events['populated']}
|
|
|
|
|
|
def effective_configuration():
|
|
raw = command('/usr/bin/systemctl', 'show', SERVICE,
|
|
'-p', 'ReadWritePaths', '-p', 'ProtectHome', '-p', 'ExecStart', '-p', 'DropInPaths')
|
|
lines = []
|
|
for line in raw.splitlines():
|
|
if line.startswith('ExecStart='):
|
|
# systemd mixes configuration with last-run timestamps/PID/status in
|
|
# this property. Hash only the exact configured command prefix;
|
|
# process/job identity is authenticated separately, never ignored.
|
|
match = re.fullmatch(r'(ExecStart=\{ path=.*? ; argv\[\]=.*? ; ignore_errors=(?:yes|no)) ; start_time=.* \}', line)
|
|
if not match:
|
|
raise ValueError('unsupported manager ExecStart representation')
|
|
line = match[1] + ' }'
|
|
lines.append(line)
|
|
return '\n'.join(lines)
|
|
|
|
|
|
def receive(conn):
|
|
conn.settimeout(5)
|
|
raw = b''
|
|
while True:
|
|
chunk = conn.recv(min(8192, MAX + 1 - len(raw)))
|
|
if not chunk:
|
|
break
|
|
raw += chunk
|
|
if len(raw) > MAX:
|
|
raise ValueError('message size')
|
|
return raw
|
|
|
|
|
|
def refusal(reason, hashed=None, reservation='NONE'):
|
|
return {'schema_version': 1, 'status': 'REFUSED', 'outcome': 'REFUSED',
|
|
'reason': reason, 'reservation': reservation, 'request_digest': hashed,
|
|
'acceptance': None, 'receipt': None, 'receipt_digest': None}
|
|
|
|
|
|
def consumption_history(value):
|
|
# Includes prerequisite attempts, predecessor bindings and pending execution
|
|
# metadata. Matching a receipt alone cannot authenticate these at readmission.
|
|
return {k: v for k, v in value.items() if k not in ('receipt', 'journal', 'intents')}
|
|
|
|
|
|
def object_digest(path):
|
|
"""Exact bytes/type/mode/ownership; symlinks and hardlinks are not admitted."""
|
|
path = pathlib.Path(path)
|
|
if not path.exists():
|
|
if path.is_symlink():
|
|
raise PermissionError('foreign symlink')
|
|
return digest(b'ABSENT')
|
|
entries = []
|
|
def visit(p):
|
|
st = p.lstat()
|
|
if not (stat.S_ISDIR(st.st_mode) or stat.S_ISREG(st.st_mode)):
|
|
raise PermissionError('foreign object type')
|
|
fd = secure(p, stat.S_ISDIR(st.st_mode))
|
|
try:
|
|
item = [str(p.relative_to(path)), stat.S_IMODE(st.st_mode), st.st_uid, st.st_gid]
|
|
if stat.S_ISDIR(st.st_mode):
|
|
item.append('directory')
|
|
entries.append(item)
|
|
for child in sorted(p.iterdir()):
|
|
visit(child)
|
|
else:
|
|
h = __import__('hashlib').sha256()
|
|
with os.fdopen(os.dup(fd), 'rb') as f:
|
|
for block in iter(lambda: f.read(1024 * 1024), b''):
|
|
h.update(block)
|
|
item.append(h.hexdigest())
|
|
entries.append(item)
|
|
finally:
|
|
os.close(fd)
|
|
visit(path)
|
|
return digest(canonical(entries))
|
|
|
|
|
|
class Owner:
|
|
def __init__(self):
|
|
binding_raw = read(BINDING)
|
|
self.binding = json.loads(binding_raw)
|
|
self.binding_digest = digest(binding_raw)
|
|
self.mutex = threading.RLock()
|
|
self.dispatch_mutex = threading.Lock()
|
|
self.incarnation = secrets.token_hex(16)
|
|
self.boot = boot_id()
|
|
# Only the appointed provisioner's one-use volatile bootstrap can admit
|
|
# sequence zero. A missing ledger is never provisioned by this process.
|
|
bootstrap = RUN / 'bootstrap.json'
|
|
admitted = False
|
|
if bootstrap.exists():
|
|
token = json.loads(read(bootstrap))
|
|
admitted = token == {'boot_id': self.boot, 'binding_digest': digest(read(BINDING)),
|
|
'ledger_digest': digest(read(ROOT / 'ledger.json'))}
|
|
bootstrap.unlink()
|
|
sync_dir(RUN)
|
|
self.store = Store(ROOT, self.binding, admitted=admitted)
|
|
self.binding = self.store.binding
|
|
self.handoff = None
|
|
self.manager_window = None # Never reconstruct live references from disk.
|
|
self.manager_blocked = bool(self.store.value.get('manager_fault'))
|
|
self.manager_condition = threading.Condition(self.mutex)
|
|
self.hold()
|
|
|
|
def check(self):
|
|
self.store.check()
|
|
if hasattr(self, 'binding_digest') and digest(read(BINDING)) != self.binding_digest:
|
|
self.store.anchor = None
|
|
raise ContinuityError('provisioned authority binding changed')
|
|
|
|
def handoff_service(self):
|
|
# No role registration/continued invocation proof is available during an
|
|
# OS manager window. Convert failed observation to denial, not a dead
|
|
# descriptor-listener thread or a cached successful service observation.
|
|
if getattr(self, 'manager_blocked', False) or getattr(self, 'manager_window', None) is not None:
|
|
raise PermissionError('manager maintenance excludes role handoff')
|
|
try:
|
|
return service()
|
|
except (OSError, subprocess.SubprocessError) as exc:
|
|
raise PermissionError('manager invocation observation unavailable') from exc
|
|
|
|
def hold(self, request_stop=True):
|
|
if getattr(self, 'handoff', None) is not None:
|
|
self.handoff.retire()
|
|
try:
|
|
PERMIT.unlink()
|
|
sync_dir(RUN)
|
|
except FileNotFoundError:
|
|
pass
|
|
# BindsTo + the prerequisite ConditionPathExists exclude activation.
|
|
# Only an already admitted transaction carries stop/hold authority.
|
|
if request_stop and (self.store.value['receipt'] is not None or self.store.value.get('prerequisite') is not None):
|
|
try:
|
|
command('/usr/bin/systemctl', '--no-block', 'stop', SERVICE)
|
|
except (OSError, subprocess.SubprocessError):
|
|
# Unavailable control transport is NOT proof of a stopped unit.
|
|
# In particular this error path must not kill the live owner.
|
|
print('D09 hold: manager stop unobserved', file=sys.stderr, flush=True)
|
|
return False
|
|
return True
|
|
|
|
def fail_manager(self, error):
|
|
"""Non-admitting failure path, itself safe when the manager is absent."""
|
|
with self.mutex:
|
|
self.manager_blocked = True
|
|
try:
|
|
window = getattr(self, 'manager_window', None)
|
|
if window is not None:
|
|
window.fail(type(error).__name__)
|
|
else:
|
|
from os_update_manager import record
|
|
if not self.store.value.get('manager_fault'):
|
|
record(self, 'HELD', reason=type(error).__name__)
|
|
except Exception:
|
|
# A lost/damaged anchor is never repaired using local bytes.
|
|
pass
|
|
try:
|
|
self.hold(request_stop=getattr(self, 'manager_window', None) is None)
|
|
except Exception:
|
|
print('D09 hold remains unobserved', file=sys.stderr, flush=True)
|
|
|
|
def run_iteration(self):
|
|
try:
|
|
with self.mutex:
|
|
self.check()
|
|
if getattr(self, 'manager_blocked', False):
|
|
return
|
|
window = getattr(self, 'manager_window', None)
|
|
if window is not None:
|
|
if self.revoked():
|
|
window.fail('AUTHORITY_REVOKED')
|
|
else:
|
|
window.poll()
|
|
if window.state == 'HELD':
|
|
# The app was already drained before arming. Do not send
|
|
# commands through an unavailable/replaced manager peer.
|
|
self.hold(request_stop=False)
|
|
return # Existing worker owns this stage; never redispatch.
|
|
self.dispatch()
|
|
except Exception as error:
|
|
self.fail_manager(error)
|
|
|
|
def wait_manager(self):
|
|
"""Called under mutex by the exact authenticated existing worker."""
|
|
window = getattr(self, 'manager_window', None)
|
|
while window is not None:
|
|
self.check()
|
|
if self.revoked() or getattr(self, 'manager_blocked', False) or window.state == 'HELD':
|
|
raise PermissionError('manager continuity requires external reconciliation')
|
|
if window.poll():
|
|
return
|
|
if window.state == 'HELD':
|
|
raise PermissionError('manager lifetime/deadline unresolved')
|
|
self.manager_condition.wait(.1) # Releases mutex; no lifetime extension.
|
|
|
|
def manager_readmitted(self, previous_epoch):
|
|
"""Only after existing external readmission verification succeeds."""
|
|
if self.store.value['epoch'] == previous_epoch:
|
|
return # An idempotent OLD record cannot clear a later interruption.
|
|
if getattr(self, 'manager_blocked', False) or getattr(self, 'manager_window', None):
|
|
from os_update_manager import record
|
|
record(self, 'EXTERNALLY_RECONCILED')
|
|
window = getattr(self, 'manager_window', None)
|
|
if window is not None:
|
|
window.close()
|
|
self.manager_window = None
|
|
self.manager_blocked = False
|
|
|
|
def prerequisite_artifact(self, current):
|
|
if current['ActiveState'] != 'active' or current['MainPID'] == '0' or not current['InvocationID']:
|
|
raise ValueError('prerequisite invocation not observed')
|
|
with open('/proc/%d/exe' % int(current['MainPID']), 'rb') as executable:
|
|
hashed = hashlib.file_digest(executable, 'sha256').hexdigest()
|
|
if hashed != self.binding['prerequisite']['backend_sha256']:
|
|
raise ValueError('wrong running prerequisite artifact')
|
|
return dict(invocation_id=current['InvocationID'], process=process(int(current['MainPID'])))
|
|
|
|
def activate_prerequisite(self):
|
|
# Called only by the appointed root provisioner on the private socket.
|
|
# Keep the ordinary public submit schema and grant-B exclusion intact.
|
|
with self.mutex:
|
|
self.check()
|
|
v = self.store.value
|
|
if v['sequence'] != 0 or v['receipt'] is not None or any(
|
|
v[k] for k in ('requests', 'inbox', 'journal', 'intents', 'readmissions')):
|
|
raise PermissionError('grant A cannot activate an admitted transaction')
|
|
grant = self.binding['prerequisite']
|
|
if self.revoked():
|
|
raise PermissionError('authority revoked')
|
|
prior = v.get('prerequisite')
|
|
if prior is not None:
|
|
if prior['state'] == 'START_INTENT' and prior['owner_incarnation'] == self.incarnation and prior['boot_id'] == self.boot:
|
|
return self.observe_prerequisite()
|
|
if prior['state'] != 'STARTED' or prior['owner_incarnation'] != self.incarnation or prior['boot_id'] != self.boot:
|
|
raise PermissionError('prerequisite start outcome requires external resolution')
|
|
observed = self.prerequisite_artifact(service())
|
|
if observed != prior['observation']:
|
|
raise PermissionError('prerequisite invocation changed')
|
|
return {'prerequisite_started': True, 'invocation_id': observed['invocation_id']}
|
|
current = service()
|
|
if current['MainPID'] != '0' or current['ActiveState'] not in ('inactive', 'failed') or current['Job'] not in ('', '0', '0 /'):
|
|
raise PermissionError('prerequisite service/job not drained')
|
|
self.preflight()
|
|
if digest(read(DESTINATIONS['backend'])) != grant['backend_sha256']:
|
|
raise ValueError('prerequisite backend differs from grant A')
|
|
effective = effective_configuration()
|
|
if digest(effective.encode()) != grant['effective_config_digest']:
|
|
raise ValueError('prerequisite effective configuration differs from grant A')
|
|
record = dict(state='START_INTENT', grant_digest=grant['grant_digest'],
|
|
owner_incarnation=self.incarnation, boot_id=self.boot,
|
|
backend_sha256=grant['backend_sha256'], observation=None, submission='INTENT')
|
|
# Consumption precedes the permit and manager call. Any lost outcome
|
|
# is retained and cannot issue another start, including after reboot.
|
|
self.store.commit_prerequisite(record)
|
|
if getattr(self, 'handoff', None):
|
|
self.handoff.expect('old')
|
|
atomic(PERMIT, canonical({'prerequisite_grant_digest': grant['grant_digest']}))
|
|
# This acknowledges submission, NOT READY. A lost submission reply is
|
|
# still START_INTENT: neither retry nor the poller submits another start.
|
|
command('/usr/bin/systemctl', 'start', '--no-block', SERVICE)
|
|
self.acknowledge_prerequisite_submission()
|
|
return self.observe_prerequisite()
|
|
|
|
def acknowledge_prerequisite_submission(self):
|
|
with self.mutex:
|
|
self.check()
|
|
v=copy.deepcopy(self.store.value)
|
|
prior=v.get('prerequisite')
|
|
if prior and prior['state']=='START_INTENT' and prior['owner_incarnation']==self.incarnation and prior['boot_id']==self.boot:
|
|
prior.update(submission='ACKNOWLEDGED',manager_job=service()['Job'])
|
|
self.store.commit_head(v)
|
|
|
|
def observe_prerequisite(self):
|
|
with self.mutex:
|
|
self.check()
|
|
v = copy.deepcopy(self.store.value)
|
|
prior = v.get('prerequisite')
|
|
if v['receipt'] is not None or not prior or prior['state'] not in ('START_INTENT', 'STARTED'):
|
|
return {'prerequisite_started': False, 'invocation_id': None}
|
|
if prior['owner_incarnation'] != self.incarnation or prior['boot_id'] != self.boot:
|
|
return {'prerequisite_started': False, 'invocation_id': None}
|
|
current = service()
|
|
# A failed dependency can consume the manager job without ever
|
|
# activating this unit: systemd leaves it inactive, not failed.
|
|
# Only the acknowledged, same-owner/same-boot submission above may
|
|
# be fenced. Never infer permission to dispatch again from no PID.
|
|
if prior['state']=='START_INTENT' and prior.get('submission')=='ACKNOWLEDGED' and current['ActiveState'] in ('inactive','failed') and current['MainPID']=='0' and current['Job'] in ('','0','0 /'):
|
|
v['prerequisite']['state']='START_UNCERTAIN'
|
|
self.store.commit_head(v)
|
|
self.hold()
|
|
return {'prerequisite_started':False,'invocation_id':None}
|
|
if current['ActiveState'] != 'active':
|
|
return {'prerequisite_started': False, 'invocation_id': None}
|
|
observed = self.prerequisite_artifact(current)
|
|
if prior['state'] == 'START_INTENT':
|
|
v['prerequisite'].update(state='STARTED', observation=observed)
|
|
self.store.commit_head(v)
|
|
try:
|
|
PERMIT.unlink()
|
|
sync_dir(RUN)
|
|
except FileNotFoundError:
|
|
pass
|
|
elif observed != prior['observation']:
|
|
raise PermissionError('prerequisite invocation changed')
|
|
return {'prerequisite_started': True, 'invocation_id': observed['invocation_id']}
|
|
|
|
def authorize(self, req, peer):
|
|
pid, uid, gid = peer
|
|
if uid not in (0, self.binding['requester_uid']):
|
|
raise PermissionError('caller')
|
|
if 'operation' not in req:
|
|
return
|
|
route = req['route']
|
|
if pid == os.getpid() and uid == 0 and route == 'AUTOMATIC':
|
|
# Sole reconciler's failure notice, never a JSON caller identity.
|
|
return
|
|
if route in ('OBSERVER', 'BOOT'):
|
|
if uid != 0 or route == 'BOOT':
|
|
raise PermissionError('external owner/observer')
|
|
else:
|
|
cgroup = pathlib.Path('/proc/%d/cgroup' % pid).read_text()
|
|
if not any(line.split(':', 2)[-1] == '/system.slice/' + SERVICE for line in cgroup.splitlines()):
|
|
raise PermissionError('caller outside bound invocation')
|
|
if req['requester_invocation_id'] != service()['InvocationID']:
|
|
raise PermissionError('caller invocation')
|
|
if route == 'GUARD' and uid != 0:
|
|
raise PermissionError('guard principal')
|
|
|
|
def receipt(self, req):
|
|
r = self.store.value['receipt']
|
|
if r is None:
|
|
r = dict(schema_version=1, transaction_id=req['transaction_id'],
|
|
target_binding_digest=req['target_binding_digest'], composition_digest=req['composition_digest'],
|
|
boot_id=self.boot, executor=None, prior_invocation_id=req['requester_invocation_id'],
|
|
observed_invocation_id=None, phase='HANDOFF_READY', outcome='PENDING', reason='NONE',
|
|
config_action='NONE', config_pre_digest=None, config_post_digest=None,
|
|
config_current_digest=None, frontend_restoration='NOT_ATTEMPTED',
|
|
backend_restoration='NOT_ATTEMPTED', effective_config_observation_digest=None,
|
|
namespace_observation_digest=None, preservation_observation_digest=None,
|
|
staging_generation=req['staging_generation'],
|
|
sequence=1, previous_receipt_digest=None)
|
|
r = copy.deepcopy(r)
|
|
r.update(operation=req['operation'], request_digest=digest(canonical(req)),
|
|
grant_digest=req['grant_digest'], admission_epoch_id=self.store.value['epoch'])
|
|
return r
|
|
|
|
def accepted(self, entry):
|
|
self.check()
|
|
r = self.store.value['receipt']
|
|
return dict(schema_version=1, status='ACCEPTED', outcome=r['outcome'], reason=r['reason'],
|
|
reservation='RETAINED', request_digest=entry['acceptance']['request_digest'],
|
|
acceptance=entry['acceptance'], receipt=r, receipt_digest=digest(canonical(r)))
|
|
|
|
def revoked(self):
|
|
path = ROOT / 'revocation.json'
|
|
if not path.exists():
|
|
return False
|
|
value = json.loads(read(path))
|
|
if set(value) != {'target_binding_digest', 'transaction_id', 'grant_digest'} or any(
|
|
value[k] != self.binding[k] for k in ('target_binding_digest', 'transaction_id')):
|
|
raise ContinuityError('uncertain revocation authority')
|
|
if value['grant_digest'] not in (*self.binding['grants'].values(), self.binding.get('prerequisite', {}).get('grant_digest')):
|
|
raise ContinuityError('unknown revocation grant')
|
|
r = self.store.value['receipt']
|
|
if r and r['reason'] != 'AUTHORITY_REVOKED':
|
|
r = dict(r, outcome='RECOVERY_REQUIRED', phase='READMISSION_REQUIRED', reason='AUTHORITY_REVOKED')
|
|
self.store.commit(self.store.value, r)
|
|
return True
|
|
|
|
def preflight(self):
|
|
# Initial admission is non-consuming and must leave the running
|
|
# prerequisite invocation untouched if any admitted bytes changed.
|
|
for name, dest in destinations(self.binding).items():
|
|
obj = self.binding['objects'][name]
|
|
if object_digest(dest) != obj['old']:
|
|
raise ValueError('foreign preimage')
|
|
for generation, suffix in [('new', '.new'), ('restore', '.restore')]:
|
|
staged = snapshot(self.binding,name,suffix)
|
|
if object_digest(staged) != obj[generation]:
|
|
raise ValueError('admitted snapshot changed')
|
|
if digest(read(pathlib.Path('/usr/libexec/archipelago-d09-prepare-workspace'))) != self.binding['helper_digest']:
|
|
raise ValueError('helper changed')
|
|
if 'os_update' in self.binding:
|
|
import os_update_owner
|
|
os_update_owner.qualify(self.binding, lock_probe=True)
|
|
|
|
def submit(self, req, peer, session=None):
|
|
validate('submit_request', req)
|
|
hashed = digest(canonical(req))
|
|
try:
|
|
if session is None:
|
|
self.authorize(req, peer)
|
|
else:
|
|
if not getattr(self, 'handoff', None) or self.handoff.session is not session:
|
|
raise PermissionError('foreign descriptor session')
|
|
session.check(peer)
|
|
if req['route'] not in ('UPDATER', 'MANUAL', 'AUTOMATIC') or req['requester_invocation_id'] != session.invocation:
|
|
raise PermissionError('protected mutation invocation/scope')
|
|
except PermissionError:
|
|
return refusal('AUTHORIZATION', hashed)
|
|
for key, reason in [('target_binding_digest', 'TARGET_MISMATCH'), ('composition_digest', 'COMPOSITION_MISMATCH')]:
|
|
if req[key] != self.binding[key]:
|
|
return refusal(reason, hashed)
|
|
if req['operation'] == 'APPLY' and req['selected_release_digest'] != release_digest(self.binding['release_identity']):
|
|
return refusal('COMPOSITION_MISMATCH', hashed)
|
|
with self.mutex:
|
|
try:
|
|
self.check()
|
|
revoked = self.revoked()
|
|
except ContinuityError:
|
|
return refusal('CONTINUITY_REQUIRED', hashed, 'UNKNOWN')
|
|
v = copy.deepcopy(self.store.value)
|
|
key = ':'.join(req[k] for k in ('transaction_id', 'operation', 'request_id'))
|
|
if key in v['requests']:
|
|
entry = v['requests'][key]
|
|
if canonical(entry['request']) != canonical(req):
|
|
return refusal('IDEMPOTENCY_MISMATCH', hashed, 'RETAINED')
|
|
# Lookup precedes sequence/grant/inbox admission; no mutation.
|
|
return self.accepted(entry)
|
|
if getattr(self, 'manager_blocked', False) or getattr(self, 'manager_window', None) is not None:
|
|
return refusal('DISPATCH_PENDING', hashed, 'RETAINED' if v['receipt'] else 'NONE')
|
|
if revoked:
|
|
return refusal('AUTHORITY_REVOKED', hashed, 'RETAINED' if v['receipt'] else 'NONE')
|
|
if req['expected_sequence'] != v['sequence']:
|
|
return refusal('STALE_SEQUENCE', hashed, 'RETAINED' if v['receipt'] else 'NONE')
|
|
if req['transaction_id'] != v['transaction_id']:
|
|
return refusal('ACTIVE_TRANSACTION', hashed)
|
|
if v['sequence'] >= 9007199254740991:
|
|
return refusal('SEQUENCE_EXHAUSTED', hashed)
|
|
current_grant = v['notice_grant'] or self.binding['grants'].get(req['route'])
|
|
if req['grant_digest'] != current_grant:
|
|
return refusal('GRANT_INVALID', hashed)
|
|
if req['operation'] == 'APPLY' and (v['receipt'] or (req['grant_digest'] in v['grants'] and req['grant_digest'] != v['notice_grant'])):
|
|
return refusal('GRANT_REPLAY', hashed, 'RETAINED')
|
|
if req['operation'] == 'APPLY':
|
|
prerequisite = v.get('prerequisite')
|
|
if 'prerequisite' not in self.binding or not prerequisite or prerequisite['state'] != 'STARTED':
|
|
return refusal('DISPATCH_PENDING', hashed)
|
|
if req['requester_invocation_id'] != prerequisite['observation']['invocation_id']:
|
|
return refusal('IDENTITY_MISMATCH', hashed)
|
|
try:
|
|
self.preflight()
|
|
except (OSError, ValueError, KeyError, subprocess.SubprocessError, ImportError):
|
|
return refusal('FOREIGN_STATE', hashed)
|
|
if req['operation'] != 'APPLY' and v['receipt'] is None:
|
|
return refusal('UNKNOWN_TRANSACTION', hashed)
|
|
if len(v['inbox']) >= 64:
|
|
return refusal('INBOX_FULL', hashed, 'RETAINED')
|
|
if req['operation'] == 'ACK_OBSERVATION':
|
|
try:
|
|
self.observation(req['observer_receipt_digest'])
|
|
except (ValueError, OSError, KeyError):
|
|
return refusal('OBSERVATION_MISSING', hashed, 'RETAINED')
|
|
a = {k: req[k] for k in ('schema_version', 'transaction_id', 'request_id', 'operation',
|
|
'target_binding_digest', 'composition_digest', 'grant_digest')}
|
|
a.update(request_digest=hashed, admission_epoch_id=v['epoch'], accepted_sequence=v['sequence'] + 1)
|
|
v['requests'][key] = {'request': req, 'acceptance': a}
|
|
v['inbox'].append(key)
|
|
if req['operation'] == 'APPLY':
|
|
v['grants'].append(req['grant_digest'])
|
|
r = self.receipt(req)
|
|
self.store.commit(v, r)
|
|
return self.accepted(v['requests'][key])
|
|
|
|
def inspect(self, req, peer, read_only=False):
|
|
validate('inspect_request', req)
|
|
self.authorize(req, peer)
|
|
result = dict(schema_version=1, challenge=req['challenge'], target_binding_digest=req['target_binding_digest'],
|
|
transaction_id=req['transaction_id'], status='UNKNOWN', reason='CONTINUITY_REQUIRED',
|
|
admission_epoch_id=None, head_sequence=None, head_digest=None, receipt=None, receipt_digest=None)
|
|
with self.mutex:
|
|
if req['target_binding_digest'] != self.binding['target_binding_digest']:
|
|
result['reason'] = 'TARGET_MISMATCH'
|
|
return result
|
|
try:
|
|
self.check()
|
|
if read_only:
|
|
if (ROOT / 'revocation.json').exists():
|
|
return result
|
|
else:
|
|
self.revoked()
|
|
except ContinuityError:
|
|
return result
|
|
v = self.store.value
|
|
if getattr(self, 'manager_blocked', False) or getattr(getattr(self, 'manager_window', None), 'state', None) in ('UNCERTAIN', 'HELD'):
|
|
result['reason'] = 'DISPATCH_PENDING'
|
|
return result
|
|
if req['transaction_id'] not in (None, v['transaction_id']):
|
|
result['reason'] = 'UNKNOWN_TRANSACTION'
|
|
return result
|
|
r = v['receipt']
|
|
if r and r['outcome'] == 'COMPLETE' and service()['InvocationID'] != r['observed_invocation_id']:
|
|
return result
|
|
result.update(status='TRANSACTION' if r else 'NO_D09_TRANSACTION', reason=r['reason'] if r else 'NONE',
|
|
admission_epoch_id=v['epoch'], head_sequence=v['sequence'],
|
|
transaction_id=v['transaction_id'] if r else None, receipt=r,
|
|
head_digest=digest(canonical(r)) if r else digest(self.store.raw),
|
|
receipt_digest=digest(canonical(r)) if r else None)
|
|
if 'os_update' in self.binding:
|
|
import os_update_owner
|
|
result['os_update'] = os_update_owner.projection(self.binding, v.get('os_checkpoints', []))
|
|
return result
|
|
|
|
def lookup(self, request):
|
|
# Only Handoff calls this after authenticating the CURRENT application.
|
|
# The original request is evidence, never submitted or rewritten here.
|
|
validate('lookup_request', request)
|
|
original = request['request']
|
|
if original['route'] not in ('UPDATER','MANUAL','AUTOMATIC'):
|
|
raise PermissionError('historical operation outside application scope')
|
|
hashed = digest(canonical(original))
|
|
if request['request_digest'] != hashed:
|
|
raise ValueError('historical request digest mismatch')
|
|
with self.mutex:
|
|
self.check()
|
|
if original['target_binding_digest'] != self.binding['target_binding_digest']:
|
|
raise PermissionError('historical target mismatch')
|
|
key = ':'.join(original[k] for k in ('transaction_id', 'operation', 'request_id'))
|
|
entry = self.store.value['requests'].get(key)
|
|
if entry is not None and canonical(entry['request']) != canonical(original):
|
|
raise PermissionError('historical body mismatch')
|
|
current = self.inspect(dict(schema_version=1, target_binding_digest=self.binding['target_binding_digest'],
|
|
transaction_id=None, challenge=request['challenge']), (os.getpid(), 0, 0), read_only=True)
|
|
return dict(schema_version=1, challenge=request['challenge'], request_digest=hashed,
|
|
status='FOUND' if entry else 'UNKNOWN', acceptance=entry['acceptance'] if entry else None,
|
|
current=current)
|
|
|
|
def observation(self, hashed):
|
|
raw = read(ROOT / 'observations' / (hashed + '.json'))
|
|
if digest(raw) != hashed:
|
|
raise ValueError('observer receipt digest')
|
|
obs = json.loads(raw)
|
|
fields = {'transaction_id', 'target_binding_digest', 'composition_digest', 'observed_invocation_id',
|
|
'effective_config_observation_digest', 'config_current_digest',
|
|
'namespace_observation_digest', 'preservation_observation_digest'}
|
|
if set(obs) != fields:
|
|
raise ValueError('observer receipt fields')
|
|
r = self.store.value['receipt']
|
|
if not r or r['phase'] != 'AWAITING_OBSERVATION':
|
|
raise ValueError('observation phase')
|
|
for key in ('transaction_id', 'target_binding_digest', 'composition_digest', 'observed_invocation_id',
|
|
'effective_config_observation_digest', 'config_current_digest'):
|
|
if obs[key] != r[key]:
|
|
raise ValueError('observation binding')
|
|
if service()['InvocationID'] != obs['observed_invocation_id']:
|
|
raise ValueError('stale invocation')
|
|
for key in ('namespace_observation_digest', 'preservation_observation_digest'):
|
|
if len(obs[key]) != 64 or any(c not in '0123456789abcdef' for c in obs[key]):
|
|
raise ValueError('observation digest')
|
|
evidence = read(ROOT / 'observations' / (obs[key] + '.json'))
|
|
if digest(evidence) != obs[key]:
|
|
raise ValueError('independent evidence missing or altered')
|
|
detail = json.loads(evidence)
|
|
if detail['observed_invocation_id'] != obs['observed_invocation_id'] or detail['transaction_id'] != obs['transaction_id']:
|
|
raise ValueError('independent evidence invocation')
|
|
if key == 'namespace_observation_digest':
|
|
if detail['default_terminal_created'] is not True or detail['default_terminal_workspace'] != '/home/archipelago/Work' or detail['namespace_writable'] != ['/home/archipelago/Work', '/home/archipelago/.local/state/archipelago/sessions']:
|
|
raise ValueError('namespace/default-terminal not observed')
|
|
else:
|
|
if set(detail['preserved']) != {'work', 'history', 'custom_fragments'} or any(
|
|
item['before'] != item['after'] or len(item['before']) != 64 for item in detail['preserved'].values()):
|
|
raise ValueError('preservation not observed')
|
|
return obs
|
|
|
|
def worker_call(self, request, peer):
|
|
# Private socket is root-only; additionally pin exact PID/start/cgroup
|
|
# to the manager-owned, durably named worker before granting any step.
|
|
if peer[1] != 0:
|
|
raise PermissionError('worker uid')
|
|
if request == {'method': 'activate_prerequisite'}:
|
|
if getattr(self, 'manager_blocked', False) or getattr(self, 'manager_window', None) is not None:
|
|
raise PermissionError('manager continuity held')
|
|
return self.activate_prerequisite()
|
|
with self.mutex:
|
|
if request == {'method': 'os_postboot'}:
|
|
self.check()
|
|
if getattr(self, 'manager_blocked', False) or getattr(self, 'manager_window', None) is not None:
|
|
raise PermissionError('manager continuity not released')
|
|
if self.revoked():
|
|
raise PermissionError('OS observer authority revoked')
|
|
import os_update_owner
|
|
observation = os_update_owner.postboot(self.binding)
|
|
import os_update_executor
|
|
spec = self.binding['os_update']
|
|
status = json.loads(read(os_update_executor.ROOT / 'transactions' / spec['authority_digest'] / 'status.json'))
|
|
history = [c for c in self.store.value.get('os_checkpoints', []) if c['authority_digest'] == spec['authority_digest']]
|
|
if not history or history[-1]['status_sha256'] != digest(canonical(status)):
|
|
self.store.checkpoint_os(status)
|
|
value = copy.deepcopy(self.store.value)
|
|
value.setdefault('os_observations', []).append(observation)
|
|
self.store.commit_head(value)
|
|
return dict(os_postboot_verified=True, observation=observation)
|
|
if request == {'method': 'readmit'}:
|
|
previous_epoch = self.store.value['epoch']
|
|
result = self.readmit()
|
|
self.manager_readmitted(previous_epoch)
|
|
return result
|
|
if request == {'method': 'reactivate'}:
|
|
previous_epoch = self.store.value['epoch']
|
|
result = self.readmit(activation=True)
|
|
self.manager_readmitted(previous_epoch)
|
|
return result
|
|
if request == {'method': 'successor'}:
|
|
if getattr(self, 'manager_blocked', False) or getattr(self, 'manager_window', None) is not None:
|
|
raise PermissionError('manager reconciliation precedes successor authority')
|
|
return self.admit_successor()
|
|
self.check()
|
|
if self.revoked():
|
|
raise PermissionError('execution authority revoked')
|
|
v = copy.deepcopy(self.store.value)
|
|
execution = v['execution']
|
|
if not execution or request['incarnation_id'] != execution['incarnation_id']:
|
|
raise PermissionError('superseded worker')
|
|
cgroup = pathlib.Path('/proc/%d/cgroup' % peer[0]).read_text()
|
|
if not any(line.split(':', 2)[-1] == '/system.slice/' + execution['unit'] for line in cgroup.splitlines()):
|
|
raise PermissionError('worker manager identity')
|
|
actual = process(peer[0])
|
|
actual['incarnation_id'] = request['incarnation_id']
|
|
if execution.get('executor') not in (None, actual):
|
|
raise PermissionError('worker incarnation')
|
|
execution['executor'] = actual
|
|
if request['method'] == 'os_checkpoint':
|
|
import os_update_executor
|
|
spec = self.binding['os_update']
|
|
status = json.loads(read(os_update_executor.ROOT / 'transactions' / spec['authority_digest'] / 'status.json'))
|
|
if getattr(self, 'manager_blocked', False):
|
|
raise PermissionError('manager held for reconciliation')
|
|
if status['phase'] == 'APPLY_INTENT':
|
|
import os_update_owner
|
|
executor, _ = os_update_owner.qualify(self.binding)
|
|
policy = executor.plan.get('manager_reexec')
|
|
if policy is not None:
|
|
if getattr(self, 'manager_window', None) is not None:
|
|
raise PermissionError('manager operation already armed')
|
|
from os_update_manager import Window
|
|
self.manager_window = Window(self, peer[0], policy)
|
|
self.manager_condition = threading.Condition(self.mutex)
|
|
self.wait_manager()
|
|
self.store.checkpoint_os(status)
|
|
if status['phase'] == 'REBOOT_REQUIRED' and getattr(self, 'manager_window', None) is not None:
|
|
from os_update_manager import record
|
|
record(self, 'COMPLETED', deadline_ms=self.manager_window.deadline)
|
|
self.manager_window.close()
|
|
self.manager_window = None
|
|
return self.worker_snapshot(self.store.value)
|
|
if request['method'] == 'observation':
|
|
obs = self.observation(request['digest'])
|
|
return dict(self.worker_snapshot(v), observation=obs)
|
|
if request['method'] == 'get':
|
|
return self.worker_snapshot(v)
|
|
holding = (request['method'] == 'transition' and request.get('updates', {}).get('phase') in ('CONFLICT', 'READMISSION_REQUIRED')
|
|
and request.get('updates', {}).get('outcome') == 'RECOVERY_REQUIRED')
|
|
if (getattr(self, 'manager_blocked', False) or getattr(self, 'manager_window', None) is not None) and not holding:
|
|
raise PermissionError('manager stage not released for lifecycle continuation')
|
|
if request['method'] == 'access_generation':
|
|
import c01_access_generation
|
|
action=v['receipt']['config_action']
|
|
method=request.get('action')
|
|
if method not in ('inspect','begin','commit') or 'c01' not in self.binding:
|
|
raise ValueError('access generation operation')
|
|
hashed=self.binding['c01']['transitions'][action]
|
|
return dict(self.worker_snapshot(v), access_generation=c01_access_generation.request(method,hashed))
|
|
if request['method'] != 'transition':
|
|
raise ValueError('worker method')
|
|
r = copy.deepcopy(v['receipt'])
|
|
prior_phase=r['phase']
|
|
updates = request['updates']
|
|
if set(updates) - set(r):
|
|
raise ValueError('receipt fields')
|
|
r.update(updates)
|
|
if v.get('activation') and r['phase'] == 'AWAITING_OBSERVATION':
|
|
v['activation'] = None
|
|
r['executor'] = actual
|
|
r['boot_id'] = self.boot
|
|
v['intent'] = request['intent']
|
|
if request.get('consume') is not None:
|
|
key = request['consume']
|
|
if key not in v['inbox']:
|
|
raise ValueError('notice already consumed')
|
|
v['inbox'].remove(key)
|
|
v['consumed'].append(key)
|
|
self.store.commit(v, r)
|
|
if r['phase'] == 'START_INTENT' and prior_phase != 'START_INTENT' and getattr(self, 'handoff', None):
|
|
self.handoff.expect('new' if r['config_action'] == 'APPLY' else 'restore')
|
|
return self.worker_snapshot(self.store.value)
|
|
|
|
def worker_snapshot(self, value):
|
|
# The private pipe has the same bound as public transport. Historical
|
|
# acceptances/64-entry inbox stay in the owner; expose only the next
|
|
# notice and current execution state, never an unbounded full ledger.
|
|
keys = value['inbox'][:1]
|
|
v = {k: value[k] for k in ('transaction_id', 'receipt', 'intent', 'execution')}
|
|
v.update(inbox=keys, requests={k: value['requests'][k] for k in keys}, activation=value.get('activation'))
|
|
return {'binding': self.binding, 'value': v}
|
|
|
|
def drained_disposition(self, value):
|
|
current = service()
|
|
if current['MainPID'] != '0' or current['ActiveState'] not in ('inactive', 'failed') or current['Job'] not in ('', '0', '0 /'):
|
|
raise ValueError('management process/job not drained')
|
|
prerequisite_processes = []
|
|
for attempt in [value.get('prerequisite'), *value.get('prerequisite_history', [])]:
|
|
previous = ((attempt or {}).get('observation') or {}).get('process')
|
|
if previous and previous['boot_id'] == boot_id():
|
|
try:
|
|
actual = process(previous['pid'])
|
|
if actual['start_ticks'] == previous['start_ticks']:
|
|
raise ValueError('prerequisite process still exists')
|
|
except FileNotFoundError:
|
|
actual = None
|
|
prerequisite_processes.append(dict(previous=previous, actual=actual))
|
|
execution = value['execution']
|
|
worker = None
|
|
subtree = None
|
|
if execution:
|
|
subtree=worker_disposition(execution)
|
|
if subtree and subtree['populated']!='0':
|
|
raise ValueError('worker descendants not retired')
|
|
worker = command('/usr/bin/systemctl', 'show', execution['unit'], '-p', 'ActiveState', '-p', 'MainPID', '-p', 'Job')
|
|
fields = dict(line.split('=', 1) for line in worker.splitlines())
|
|
if fields.get('ActiveState') not in ('inactive', 'failed') or fields.get('MainPID') != '0' or fields.get('Job') not in ('', '0', '0 /'):
|
|
raise ValueError('worker process/job not drained')
|
|
executor = execution.get('executor')
|
|
if executor and executor['boot_id'] == boot_id():
|
|
try:
|
|
if process(executor['pid'])['start_ticks'] == executor['start_ticks']:
|
|
raise ValueError('worker process still exists')
|
|
except FileNotFoundError:
|
|
pass
|
|
return digest(canonical(dict(boot_id=boot_id(), service=current, execution=execution, worker=worker,
|
|
worker_subtree=subtree, prerequisite_processes=prerequisite_processes)))
|
|
|
|
def admit_successor(self):
|
|
self.check()
|
|
record = parse(read(ROOT / 'successor.json'))
|
|
validate('external_successor_record', record)
|
|
binding_raw = read(ROOT / 'successor-binding.json')
|
|
binding = json.loads(binding_raw)
|
|
if digest(binding_raw) != record['successor_binding_digest']:
|
|
raise ValueError('successor admission binding')
|
|
hashed = digest(canonical(record))
|
|
v = self.store.value
|
|
if hashed in v.get('successions', []):
|
|
# Lost acknowledgement is a read, including after another successor.
|
|
return {'admitted': True}
|
|
if self.revoked():
|
|
raise ValueError('predecessor authority revoked')
|
|
if 'os_update' in self.binding:
|
|
import os_update_owner
|
|
if os_update_owner.projection(self.binding, self.store.value.get('os_checkpoints', []))['phase'] != 'POSTBOOT_CONFIRMED':
|
|
raise ValueError('predecessor OS boot/preservation still unresolved')
|
|
r = v['receipt']
|
|
if not r or r['phase'] != 'COMPLETE' or r['outcome'] != 'COMPLETE' or v['inbox'] or v.get('activation'):
|
|
raise ValueError('terminal independently observed predecessor required')
|
|
index = {k: self.binding[k] for k in ('target_binding_digest', 'composition_digest', 'transaction_id')}
|
|
history = consumption_history(v)
|
|
expected = dict(index, predecessor_epoch_id=v['epoch'], head_sequence=v['sequence'],
|
|
observed_boot_id=boot_id(), preserved_head_digest=digest(canonical(r)),
|
|
preserved_journal_digest=digest(canonical({'receipts': v['journal'], 'intents': v['intents']})),
|
|
preserved_consumed_history_digest=digest(canonical(history)),
|
|
preserved_target_index_digest=digest(canonical(index)), terminal_disposition=r['config_action'])
|
|
if any(record[k] != x for k, x in expected.items()):
|
|
raise ValueError('successor predecessor/history binding')
|
|
if binding['target_binding_digest'] != self.binding['target_binding_digest'] or binding['transaction_id'] in (v['transaction_id'], *v.get('predecessors', {})) or binding['admission_epoch_id'] in v['epochs']:
|
|
raise ValueError('successor target/identity replay')
|
|
used = set(v['grants']) | set(self.binding['grants'].values())
|
|
for predecessor in v.get('predecessors', {}).values():
|
|
used.update(predecessor['binding']['grants'].values())
|
|
fresh = list(binding['grants'].values()) + [binding['prerequisite']['grant_digest']]
|
|
if len(set(fresh)) != len(fresh) or used.intersection(fresh):
|
|
raise ValueError('fresh scoped successor grants required')
|
|
if 'os_update' in binding:
|
|
previous_bindings = [self.binding] + [p['binding'] for p in v.get('predecessors', {}).values()]
|
|
if any(p.get('os_update', {}).get('authority_digest') == binding['os_update']['authority_digest'] for p in previous_bindings):
|
|
raise ValueError('fresh OS maintenance authority required')
|
|
generation = 'new' if r['config_action'] == 'APPLY' else 'restore'
|
|
artifacts = self.binding['artifacts' if generation == 'new' else 'restore_artifacts']
|
|
if binding['restore_artifacts'] != artifacts or binding['prerequisite']['backend_sha256'] != artifacts['backend']:
|
|
raise ValueError('successor rollback provenance')
|
|
for name in destinations(binding):
|
|
if binding['objects'][name]['old'] != self.binding['objects'][name][generation] or binding['objects'][name]['restore'] != binding['objects'][name]['old']:
|
|
raise ValueError('successor must retain exact predecessor generation')
|
|
current = service()
|
|
if current['InvocationID'] != r['observed_invocation_id']:
|
|
raise ValueError('predecessor invocation changed')
|
|
if v['execution']:
|
|
# Reuse the drain proof for the old worker without stopping the
|
|
# independently observed management invocation.
|
|
execution = v['execution']
|
|
subtree=worker_disposition(execution)
|
|
if subtree and subtree['populated']!='0':
|
|
raise ValueError('predecessor worker descendants remain')
|
|
state = command('/usr/bin/systemctl', 'show', execution['unit'], '-p', 'ActiveState', '-p', 'MainPID', '-p', 'Job')
|
|
fields = dict(line.split('=', 1) for line in state.splitlines())
|
|
if fields.get('ActiveState') not in ('inactive', 'failed') or fields.get('MainPID') != '0' or fields.get('Job') not in ('', '0', '0 /'):
|
|
raise ValueError('predecessor worker not drained')
|
|
executor = execution.get('executor')
|
|
if executor and executor['boot_id'] == boot_id():
|
|
try:
|
|
if process(executor['pid'])['start_ticks'] == executor['start_ticks']:
|
|
raise ValueError('predecessor executor still alive')
|
|
except FileNotFoundError:
|
|
pass
|
|
old_binding = self.binding
|
|
try:
|
|
self.binding = binding
|
|
self.preflight()
|
|
observed = self.prerequisite_artifact(current)
|
|
effective = effective_configuration()
|
|
if digest(effective.encode()) != binding['prerequisite']['effective_config_digest']:
|
|
raise ValueError('successor prerequisite configuration')
|
|
prerequisite = dict(state='STARTED', grant_digest=binding['prerequisite']['grant_digest'],
|
|
owner_incarnation=self.incarnation, boot_id=self.boot,
|
|
backend_sha256=artifacts['backend'], observation=observed)
|
|
os_admission = None
|
|
if 'os_update' in binding:
|
|
import os_update_owner
|
|
_, os_admission = os_update_owner.qualify(binding, lock_probe=True)
|
|
self.store.commit_successor(binding, hashed, prerequisite, os_admission=os_admission)
|
|
except BaseException:
|
|
self.binding = old_binding
|
|
raise
|
|
return {'admitted': True}
|
|
|
|
def readmit(self, activation=False):
|
|
record = parse(read(ROOT / ('activation.json' if activation else 'readmission.json')))
|
|
validate('external_activation_record' if activation else 'external_readmission_record', record)
|
|
# Root-only appointed-provisioner channel authenticated this exact record
|
|
# against external latest-consumption history before installation.
|
|
preserved = Store(ROOT, self.binding, admitted=False)
|
|
previous = self.store
|
|
try:
|
|
return self._readmit(record, preserved, activation)
|
|
finally:
|
|
obsolete = previous if self.store is preserved else preserved
|
|
os.close(obsolete.fd)
|
|
os.close(obsolete.head_fd)
|
|
|
|
def _readmit(self, record, preserved, activation):
|
|
v = copy.deepcopy(preserved.value)
|
|
r = v['receipt']
|
|
hashed = digest(canonical(record))
|
|
if hashed in v['readmissions']:
|
|
# A lost acknowledgement may be reread only through this still-live
|
|
# anchor. It never consumes the record again or admits a new owner.
|
|
self.check()
|
|
if record['new_epoch_id'] != v['epoch']:
|
|
raise ValueError('stale readmission record')
|
|
return {'readmitted': True}
|
|
if r is None and not activation:
|
|
raise ValueError('empty/missing historical transaction cannot be readmitted')
|
|
index = {k: self.binding[k] for k in ('target_binding_digest', 'composition_digest', 'transaction_id')}
|
|
history = consumption_history(v)
|
|
expected = dict(index, predecessor_epoch_id=v['epoch'], head_sequence=v['sequence'],
|
|
observed_boot_id=boot_id(), preserved_head_digest=digest(canonical(r)) if r else digest(preserved.raw),
|
|
preserved_journal_digest=digest(canonical({'receipts': v['journal'], 'intents': v['intents']})),
|
|
preserved_consumed_history_digest=digest(canonical(history)),
|
|
preserved_target_index_digest=digest(canonical(index)))
|
|
if any(record[k] != value for k, value in expected.items()):
|
|
raise ValueError('readmission preserved-state binding')
|
|
if record['new_epoch_id'] in v['epochs'] or record['new_grant_digest'] in v['grants'] or hashed in v['readmissions']:
|
|
raise ValueError('readmission replay')
|
|
if activation:
|
|
if (ROOT / 'revocation.json').exists():
|
|
raise ValueError('revocation requires separate external disposition')
|
|
if record['disposition_digest'] != self.drained_disposition(v):
|
|
raise ValueError('external process/boot/job disposition mismatch')
|
|
used = set(v['grants']) | set(self.binding['grants'].values()) | {self.binding['prerequisite']['grant_digest']}
|
|
for predecessor in v.get('predecessors', {}).values():
|
|
used.update(predecessor['binding']['grants'].values())
|
|
if record['execution_grant_digest'] in used or record['new_grant_digest'] in used or record['execution_grant_digest'] == record['new_grant_digest']:
|
|
raise ValueError('fresh distinct external authority required')
|
|
if v['inbox']:
|
|
raise ValueError('activation cannot supersede outstanding notices')
|
|
if record['allowed_action'] == 'REACTIVATE_PREREQUISITE':
|
|
if r is not None or (v['sequence'] != 0 and not any(
|
|
p['receipt']['outcome'] == 'COMPLETE' and digest(canonical(p['receipt'])) == v['journal'][-1]
|
|
for p in v.get('predecessors', {}).values())):
|
|
raise ValueError('prerequisite reconciliation requires an unreserved admitted target')
|
|
self.preflight()
|
|
grant = self.binding['prerequisite']
|
|
if digest(read(DESTINATIONS['backend'])) != grant['backend_sha256']:
|
|
raise ValueError('prerequisite executable changed')
|
|
effective = effective_configuration()
|
|
if digest(effective.encode()) != grant['effective_config_digest']:
|
|
raise ValueError('prerequisite effective configuration changed')
|
|
elif not r or (r['phase'] not in ('COMPLETE', 'AWAITING_OBSERVATION') and not (v.get('activation') and r['phase'] in ('REACTIVATION_READY', 'START_INTENT', 'CONFLICT'))) or r['config_action'] not in ('APPLY', 'RESTORE'):
|
|
raise ValueError('retained activation requires exact completed/applied generation')
|
|
if record['allowed_action'] == 'RESUME_SAME_TRANSACTION':
|
|
if not record['execution_grant_digest'] or record['execution_grant_digest'] in v['grants']:
|
|
raise ValueError('new execution authority required')
|
|
if r['phase'] in ('CONFLICT', 'READMISSION_REQUIRED'):
|
|
raise ValueError('readmission does not reconcile a conflict')
|
|
v['epoch'] = record['new_epoch_id']
|
|
v['notice_grant'] = record['new_grant_digest']
|
|
v['execution_grant'] = record['execution_grant_digest']
|
|
v['epochs'].append(v['epoch'])
|
|
v['grants'].append(record['new_grant_digest'])
|
|
if record['execution_grant_digest']:
|
|
v['grants'].append(record['execution_grant_digest'])
|
|
v['readmissions'].append(hashed)
|
|
if activation and r is None:
|
|
v.setdefault('prerequisite_history', []).append(copy.deepcopy(v['prerequisite']))
|
|
v['prerequisite'] = dict(state='START_INTENT', grant_digest=record['execution_grant_digest'],
|
|
owner_incarnation=self.incarnation, boot_id=self.boot,
|
|
backend_sha256=self.binding['prerequisite']['backend_sha256'], observation=None,submission='INTENT')
|
|
preserved.anchor = digest(read(ROOT / 'ledger.json'))
|
|
preserved.commit_head(v)
|
|
self.store = preserved
|
|
# Consumption and intent are durable before permit publication. An
|
|
# uncertain outcome needs another external reconciliation, not retry.
|
|
if getattr(self, 'handoff', None):
|
|
self.handoff.expect('old')
|
|
atomic(PERMIT, canonical({'execution_grant_digest': record['execution_grant_digest']}))
|
|
command('/usr/bin/systemctl', 'start', '--no-block', SERVICE)
|
|
self.acknowledge_prerequisite_submission()
|
|
# The main progress loop observes READY outside the private handler.
|
|
# Keep the permit until that observation or an explicit hold.
|
|
return {'readmitted': True}
|
|
# Existing worker identity is retained until actual death is proven by
|
|
# dispatch. A readmission is not a worker supersession or lifecycle reset.
|
|
r = copy.deepcopy(r)
|
|
r['admission_epoch_id'] = v['epoch']
|
|
r['boot_id'] = boot_id()
|
|
r['outcome'] = 'RECOVERY_REQUIRED'
|
|
r['reason'] = 'CONTINUITY_REQUIRED'
|
|
if activation:
|
|
v['activation'] = dict(record_digest=hashed, execution_grant_digest=record['execution_grant_digest'])
|
|
r.update(phase='REACTIVATION_READY', prior_invocation_id=r['observed_invocation_id'],
|
|
observed_invocation_id=None, namespace_observation_digest=None,
|
|
preservation_observation_digest=None, outcome='PENDING')
|
|
if record['allowed_action'] == 'PRESERVE_ONLY':
|
|
r.update(phase='READMISSION_REQUIRED', outcome='RECOVERY_REQUIRED', reason='CONTINUITY_REQUIRED')
|
|
preserved.anchor = digest(read(ROOT / 'ledger.json'))
|
|
preserved.commit(v, r)
|
|
self.store = preserved
|
|
return {'readmitted': True}
|
|
|
|
def dispatch(self):
|
|
with self.dispatch_mutex, self.mutex:
|
|
if getattr(self, 'manager_blocked', False) or getattr(self, 'manager_window', None) is not None:
|
|
return
|
|
self.check()
|
|
if self.store.value['receipt'] is None:
|
|
self.observe_prerequisite()
|
|
if self.revoked():
|
|
self.hold()
|
|
return
|
|
v = copy.deepcopy(self.store.value)
|
|
if v['receipt'] and v['receipt']['phase'] in ('CONFLICT', 'READMISSION_REQUIRED'):
|
|
self.hold()
|
|
return
|
|
if not v['inbox'] and not v.get('activation'):
|
|
r = v['receipt']
|
|
if r and r['phase'] == 'AWAITING_OBSERVATION':
|
|
current = service()
|
|
if current['ActiveState'] in ('inactive', 'failed') and current['MainPID'] == '0' and current['Job'] in ('', '0', '0 /'):
|
|
if r['config_action'] == 'RESTORE':
|
|
self.store.commit(v, dict(r, phase='CONFLICT', outcome='RECOVERY_REQUIRED', reason='PARTIAL_RECOVERY'))
|
|
self.hold()
|
|
return
|
|
request = dict(schema_version=1, operation='RECOVER', route='AUTOMATIC', reason='AUTOMATIC_FAILURE',
|
|
transaction_id=v['transaction_id'], request_id=secrets.token_hex(16),
|
|
target_binding_digest=self.binding['target_binding_digest'],
|
|
composition_digest=self.binding['composition_digest'],
|
|
grant_digest=v['notice_grant'] or self.binding['grants']['AUTOMATIC'], expected_sequence=v['sequence'],
|
|
requester_invocation_id=r['observed_invocation_id'], observer_receipt_digest=None,
|
|
selected_release_digest=None, staging_generation=None)
|
|
self.submit(request, (os.getpid(), 0, 0))
|
|
return
|
|
if v['receipt'] and v['receipt']['phase'] == 'START_INTENT' and service()['ActiveState'] == 'activating':
|
|
return # one owned manager job; no worker retries while READY is pending
|
|
if v['execution']:
|
|
unit = v['execution']['unit']
|
|
state = command('/usr/bin/systemctl', 'show', unit, '-p', 'ActiveState', '--value')
|
|
if state not in ('inactive', 'failed'):
|
|
return # queued/alive/unknown never duplicates a worker
|
|
subtree=worker_disposition(v['execution'])
|
|
if subtree and subtree['populated']!='0':
|
|
return
|
|
executor = v['execution'].get('executor')
|
|
if executor and executor['boot_id']==self.boot:
|
|
try:
|
|
if process(executor['pid'])['start_ticks'] == executor['start_ticks']:
|
|
return
|
|
except FileNotFoundError:
|
|
pass
|
|
incarnation = secrets.token_hex(16)
|
|
unit = 'archipelago-d09-' + v['transaction_id'] + '-' + incarnation + '.service'
|
|
v['execution'] = {'unit': unit, 'incarnation_id': incarnation, 'executor': None}
|
|
r = dict(v['receipt'])
|
|
# Reopen retains the side-effect phase and intent; never rewind it.
|
|
if r['phase'] == 'HANDOFF_READY':
|
|
r['phase'] = 'DISPATCH_INTENT'
|
|
lockfd = os.open(ROOT / 'lifecycle.lock', os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
|
|
try:
|
|
try:
|
|
fcntl.flock(lockfd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
return
|
|
self.store.commit(v, r)
|
|
finally:
|
|
os.close(lockfd)
|
|
# Fixed system-manager service; never a scope or updater child.
|
|
command('/usr/bin/systemd-run', '--unit=' + unit, '--no-block',
|
|
'--property=Type=exec', '--property=UMask=0077',
|
|
'--property=StandardInput=null', '--property=StandardOutput=journal',
|
|
'--property=RuntimeMaxSec=' + ('21600' if 'os_update' in self.binding else 'infinity'),
|
|
'--property=TimeoutStartSec=infinity', '--property=KillMode=control-group',
|
|
'--property=CPUQuota=200%', '--property=MemoryMax=8G', '--property=MemorySwapMax=0',
|
|
PROGRAM, '--worker', v['transaction_id'], '--incarnation', incarnation)
|
|
|
|
|
|
def serve_socket(path, handler, mode, gid):
|
|
try:
|
|
path.unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
listener.bind(str(path))
|
|
os.chown(path, 0, gid)
|
|
os.chmod(path, mode)
|
|
listener.listen(16)
|
|
while True:
|
|
conn, _ = listener.accept()
|
|
try:
|
|
peer = struct.unpack('3i', conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
|
|
response = handler(parse(receive(conn)), peer)
|
|
conn.sendall(canonical(response))
|
|
except Exception as error:
|
|
# No malformed request can turn into a success or an absent ledger.
|
|
# Log only the exception type and source location: no request,
|
|
# grant, response body or exception message enters the journal.
|
|
frames = traceback.extract_tb(error.__traceback__)
|
|
location = frames[-1] if frames else None
|
|
print('D09 request refused: ' + type(error).__name__ +
|
|
((' at ' + pathlib.Path(location.filename).name + ':' + str(location.lineno))
|
|
if location else ''), file=sys.stderr, flush=True)
|
|
try:
|
|
conn.sendall(canonical(refusal('INTERNAL_UNCERTAIN', reservation='UNKNOWN')))
|
|
except OSError:
|
|
pass
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def main():
|
|
if os.geteuid() != 0:
|
|
raise SystemExit('root required')
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--worker')
|
|
parser.add_argument('--incarnation')
|
|
args = parser.parse_args()
|
|
if args.worker:
|
|
from d09_worker import run
|
|
run(args.worker, args.incarnation)
|
|
return
|
|
lock = os.open(ROOT / 'owner.lock', os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
owner = Owner()
|
|
from d09_handoff import Handoff
|
|
owner.handoff = Handoff(owner, owner.handoff_service, ROOT)
|
|
threading.Thread(target=owner.handoff.serve, daemon=True).start()
|
|
threading.Thread(target=serve_socket, args=(PRIVATE, owner.worker_call, 0o600, 0), daemon=True).start()
|
|
def public(req, peer):
|
|
if type(req) is not dict:
|
|
return refusal('MALFORMED')
|
|
return owner.submit(req, peer) if 'operation' in req else owner.inspect(req, peer)
|
|
threading.Thread(target=serve_socket, args=(SOCKET, public, 0o660, owner.binding['requester_gid']), daemon=True).start()
|
|
while True:
|
|
owner.run_iteration()
|
|
time.sleep(0.5)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|