"""Live-only, bounded manager reexec window. No disk reconstruction or dispatch.""" import copy import hashlib import os import pathlib import select import time from d09_store import secure from os_update_custody import Pin from os_update_plan import require from os_update_bus import Bus, Unavailable OWNER_UNIT = 'archipelago-d09-lifecycle.service' APP_UNIT = 'archipelago.service' FRONTEND = pathlib.Path('/var/lib/dpkg/lock-frontend') def now(): return time.clock_gettime_ns(time.CLOCK_BOOTTIME) // 1_000_000 def executable(pid): with open('/proc/%d/exe' % pid, 'rb') as source: return hashlib.file_digest(source, 'sha256').hexdigest() def process(pid): raw = pathlib.Path('/proc/%d/stat' % pid).read_text() group = pathlib.Path('/proc/%d/cgroup' % pid).read_text().splitlines() unified = [line[3:] for line in group if line.startswith('0::')] require(len(unified) == 1, 'unified process cgroup required') return dict(pid=pid, start_ticks=int(raw[raw.rindex(')') + 2:].split()[19]), cgroup=unified[0]) class Lifetime: def __init__(self, pid): self.pidfd = os.pidfd_open(pid) self.group = None try: self.process = process(pid) self.group = Pin(pathlib.Path('/sys/fs/cgroup') / self.process['cgroup'].lstrip('/')) self.check() except BaseException: self.close() raise def check(self): poll = select.poll() poll.register(self.pidfd, select.POLLIN | select.POLLHUP | select.POLLERR) require(not poll.poll(0), 'retained process lifetime ended') require(process(self.process['pid']) == self.process, 'process/cgroup changed') self.group.check() def close(self): if self.group is not None: self.group.close() if self.pidfd is not None: os.close(self.pidfd) self.pidfd = None def record(owner, state, **details): value = copy.deepcopy(owner.store.value) entry = dict(state=state, owner_incarnation=owner.incarnation, boot_id=owner.boot, boottime_ms=now(), **details) value.setdefault('manager_events', []).append(entry) value['manager_fault'] = entry if state in ('UNCERTAIN', 'HELD') else None owner.store.commit_head(value) class Window: def __init__(self, owner, worker_pid, policy): self.owner, self.policy = owner, copy.deepcopy(policy) self.boot = owner.boot self.epoch = owner.store.value['epoch'] self.execution = copy.deepcopy(owner.store.value['execution']) self.deadline = None self.state = 'ARMED' self.refs = [] self.lock_fd = None try: for pid in (1, os.getpid(), worker_pid): self.refs.append(Lifetime(pid)) self.initial_executable = executable(1) self.lock_fd = secure(FRONTEND) self.lock_identity = (os.fstat(self.lock_fd).st_dev, os.fstat(self.lock_fd).st_ino) self.kernel_check() with Bus() as bus: self.units = self.snapshot(bus) self.validate_units(self.units) record(owner, 'ARMED', policy=self.policy, execution=self.execution, processes=[r.process for r in self.refs], units=self.units, initial_executable_sha256=self.initial_executable) except BaseException: self.close() raise def snapshot(self, bus): return {name: bus.unit(name) for name in (OWNER_UNIT, self.execution['unit'], APP_UNIT)} def validate_units(self, units): for name, pid in ((OWNER_UNIT, os.getpid()), (self.execution['unit'], self.refs[2].process['pid'])): unit = units[name] require(unit['Id'] == name and unit['LoadState'] == 'loaded' and unit['ActiveState'] == 'active' and unit['SubState'] == 'running' and unit['MainPID'] == pid and unit['Job'] == [0, '/'] and len(unit['InvocationID']) == 32, 'manager unit/process/job not exact') require(unit['ControlGroup'] == process(pid)['cgroup'], 'manager cgroup differs from kernel') app = units[APP_UNIT] require(app['Id'] == APP_UNIT and app['MainPID'] == 0 and app['ActiveState'] in ('inactive', 'failed') and app['Job'] == [0, '/'], 'application was not stopped before OS effects') def kernel_check(self): self.owner.check() require(self.owner.boot == self.boot and pathlib.Path('/proc/sys/kernel/random/boot_id').read_text().strip().replace('-', '') == self.boot, 'boot lifetime changed') require(self.owner.store.value['epoch'] == self.epoch, 'admission epoch changed') execution = self.owner.store.value['execution'] require(execution and all(execution[k] == self.execution[k] for k in ('unit', 'incarnation_id')), 'execution grant changed') for ref in self.refs: ref.check() st = FRONTEND.stat() require((st.st_dev, st.st_ino) == self.lock_identity, 'frontend lock inode changed') key = '%02x:%02x:%d' % (os.major(st.st_dev), os.minor(st.st_dev), st.st_ino) pid = self.refs[2].process['pid'] rows = [line.split() for line in pathlib.Path('/proc/locks').read_text().splitlines()] require(any(key in row and row[1:5] == ['POSIX', 'ADVISORY', 'WRITE', str(pid)] for row in rows), 'original worker no longer owns package frontend lock') def fail(self, reason): if self.state != 'HELD': self.state = 'HELD' self.owner.manager_blocked = True record(self.owner, 'HELD', reason=reason, deadline_ms=self.deadline) return False def poll(self): if self.state == 'HELD': return False try: self.kernel_check() if self.state == 'UNCERTAIN' and now() >= self.deadline: return self.fail('RECONNECT_DEADLINE') with Bus() as bus: units = self.snapshot(bus) self.validate_units(units) require(units == self.units, 'unit invocation/job/configuration changed across manager interruption') require(executable(1) in (self.initial_executable, self.policy['expected_executable_sha256']), 'unappointed manager executable') if self.state == 'UNCERTAIN': record(self.owner, 'RECOVERED', deadline_ms=self.deadline, units=units, manager_executable_sha256=executable(1)) self.state = 'ARMED' return True except Unavailable: if self.deadline is None: self.deadline = now() + self.policy['max_unavailable_seconds'] * 1000 if now() >= self.deadline: return self.fail('RECONNECT_DEADLINE') if self.state != 'UNCERTAIN': self.state = 'UNCERTAIN' record(self.owner, 'UNCERTAIN', deadline_ms=self.deadline) return False except Exception as exc: return self.fail(type(exc).__name__) def complete(self): require(self.poll(), 'manager continuity unresolved at OS completion') record(self.owner, 'COMPLETED', deadline_ms=self.deadline) self.close() def close(self): for ref in self.refs: ref.close() self.refs = [] if self.lock_fd is not None: os.close(self.lock_fd) self.lock_fd = None