archy/scripts/os_update_manager.py

"""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