300 lines
15 KiB
Python
300 lines
15 KiB
Python
#!/usr/bin/python3
|
|
"""Real installed-PID1 manager continuity component qualification, private VM only.
|
|
|
|
Uses actual Owner/Store/Window and Debian libc-upgrade systemd trigger. Synthetic
|
|
root-appointed control tuples are NOT release crypto, native C01 or full OTA UAT.
|
|
"""
|
|
import copy
|
|
import fcntl
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
assert os.geteuid() == 0 and 'QEMU' in Path('/sys/class/dmi/id/sys_vendor').read_text()
|
|
assert subprocess.check_output(['findmnt', '-n', '-o', 'SOURCE', '/'], text=True).strip() == '/dev/vda'
|
|
SCRIPTS = Path('/opt/ota-source/scripts')
|
|
SCRIPT = '/opt/ota-source/tests/regression/os-ota-manager-bus.py'
|
|
BASE = Path('/var/lib/os-manager-fixture')
|
|
EVIDENCE = Path('/var/log/os-ota-manager-bus')
|
|
CASES = ('good', 'timeout', 'bad-peer', 'owner-loss', 'worker-loss', 'unplanned')
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
import d09_store as store
|
|
from d09_wire import canonical, digest
|
|
import os_update_manager as manager
|
|
|
|
|
|
def command(*args, **kwargs):
|
|
return subprocess.run(args, check=True, text=True, stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT, env=dict(os.environ, PATH='/usr/sbin:/usr/bin:/sbin:/bin', LC_ALL='C'), **kwargs).stdout
|
|
|
|
|
|
def unit(case):
|
|
return 'archipelago-d09-' + digest(case.encode())[:32] + '-' + 'c'*32 + '.service'
|
|
|
|
|
|
def call(case, operation):
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as conn:
|
|
conn.settimeout(150)
|
|
conn.connect(str(BASE / case / 'control'))
|
|
conn.sendall(json.dumps({'operation':operation}).encode())
|
|
conn.shutdown(socket.SHUT_WR)
|
|
raw = b''
|
|
while True:
|
|
block = conn.recv(65536)
|
|
if not block:
|
|
break
|
|
raw += block
|
|
return json.loads(raw)
|
|
|
|
|
|
def wait(function, timeout=60):
|
|
end = time.monotonic() + timeout
|
|
while time.monotonic() < end:
|
|
try:
|
|
result = function()
|
|
if result:
|
|
return result
|
|
except (OSError, ValueError):
|
|
pass
|
|
time.sleep(.1)
|
|
raise TimeoutError('fixture observation deadline')
|
|
|
|
|
|
def owner_mode(case):
|
|
spec = importlib.util.spec_from_file_location('manager_fixture_owner', SCRIPTS / 'd09-lifecycle-owner.py')
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
root = BASE / case
|
|
module.ROOT = root / 'ledger'
|
|
module.RUN = root / 'run'
|
|
module.BINDING = root / 'binding.json'
|
|
module.PERMIT = root / 'run/permit'
|
|
owner = module.Owner() # Real one-use bootstrap; restart never reinitializes it.
|
|
listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
(root / 'control').unlink(missing_ok=True)
|
|
listener.bind(str(root / 'control'))
|
|
os.chmod(root / 'control', 0o600)
|
|
listener.listen(8)
|
|
|
|
def handle(conn):
|
|
try:
|
|
peer = struct.unpack('3i', conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
|
|
assert peer[1] == 0
|
|
request = json.loads(conn.recv(4096))
|
|
with owner.mutex:
|
|
operation = request['operation']
|
|
if operation == 'inspect':
|
|
result = dict(ok=True, pid=os.getpid(), incarnation=owner.incarnation,
|
|
anchor=owner.store.anchor is not None, epoch=owner.store.value['epoch'],
|
|
state=owner.manager_window.state if owner.manager_window else None,
|
|
blocked=owner.manager_blocked, refs=len(owner.manager_window.refs) if owner.manager_window else 0,
|
|
events=owner.store.value.get('manager_events', []))
|
|
else:
|
|
owner.check()
|
|
if operation == 'arm':
|
|
if case == 'unplanned':
|
|
value = copy.deepcopy(owner.store.value)
|
|
value['prerequisite'] = dict(state='STARTED', owner_incarnation=owner.incarnation,
|
|
boot_id=owner.boot)
|
|
owner.store.commit_head(value)
|
|
else:
|
|
assert owner.manager_window is None
|
|
policy = json.loads((root / 'policy.json').read_text())
|
|
owner.manager_window = manager.Window(owner, peer[0], policy)
|
|
result = dict(ok=True)
|
|
elif operation in ('checkpoint', 'complete'):
|
|
if owner.manager_blocked:
|
|
raise PermissionError('held')
|
|
owner.wait_manager()
|
|
if operation == 'complete':
|
|
manager.record(owner, 'COMPLETED', deadline_ms=owner.manager_window.deadline)
|
|
owner.manager_window.close()
|
|
owner.manager_window = None
|
|
result = dict(ok=True)
|
|
else:
|
|
raise ValueError('fixture operation')
|
|
conn.sendall(json.dumps(result).encode())
|
|
except Exception as exc:
|
|
import traceback
|
|
with (EVIDENCE / 'owner-errors.log').open('a') as log:
|
|
traceback.print_exc(file=log)
|
|
conn.sendall(json.dumps(dict(ok=False, error=type(exc).__name__, detail=str(exc))).encode())
|
|
finally:
|
|
conn.close()
|
|
|
|
def accept():
|
|
while True:
|
|
conn, _ = listener.accept()
|
|
threading.Thread(target=handle, args=(conn,), daemon=True).start()
|
|
threading.Thread(target=accept, daemon=True).start()
|
|
while True:
|
|
if owner.manager_window is not None or owner.manager_blocked or (root / 'run-unplanned').exists() or owner.store.anchor is None:
|
|
owner.run_iteration()
|
|
time.sleep(.05)
|
|
|
|
|
|
def worker_mode(case):
|
|
root = BASE / case
|
|
# A deliberate service replacement must not repeat the fixture package step.
|
|
attempt = root / 'worker-once'
|
|
try:
|
|
fd = os.open(attempt, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
|
|
os.close(fd)
|
|
except FileExistsError:
|
|
time.sleep(15)
|
|
return
|
|
fd = os.open('/var/lib/dpkg/lock-frontend', os.O_RDWR | os.O_CREAT, 0o640)
|
|
fcntl.lockf(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
result = dict(case=case, dpkg_invocations=0, completed=False)
|
|
try:
|
|
result['arm'] = call(case, 'arm')
|
|
assert result['arm']['ok'], result
|
|
(root / 'worker-ready').touch()
|
|
wait(lambda: (root / 'go').exists(), 120)
|
|
result['checkpoint'] = call(case, 'checkpoint')
|
|
if result['checkpoint']['ok'] and case == 'good':
|
|
result['dpkg_invocations'] = 1
|
|
env = dict(os.environ, PATH='/usr/sbin:/usr/bin:/sbin:/bin', LC_ALL='C', DPKG_FRONTEND_LOCKED='true')
|
|
applied = subprocess.run(['/usr/bin/dpkg', '-i', str(BASE / 'package-2.deb')],
|
|
env=env, capture_output=True, text=True)
|
|
(EVIDENCE / 'dpkg.log').write_text(applied.stdout + applied.stderr)
|
|
assert applied.returncode == 0, applied.stdout + applied.stderr
|
|
assert 'Processing triggers for systemd' in applied.stdout
|
|
result['complete'] = call(case, 'complete')
|
|
result['completed'] = result['complete']['ok']
|
|
except Exception as exc:
|
|
result['error'] = type(exc).__name__
|
|
finally:
|
|
(root / 'worker-result.json').write_text(json.dumps(result, indent=2))
|
|
# Controller inspects live retained references before ending this lifetime.
|
|
wait(lambda: (root / 'release').exists(), 120)
|
|
os.close(fd)
|
|
|
|
|
|
def driver():
|
|
assert not BASE.exists(), 'new independent disposable disk required'
|
|
BASE.mkdir(mode=0o700)
|
|
EVIDENCE.mkdir(mode=0o700)
|
|
result = dict(scope='real PID1 manager continuity component; NOT release crypto/native C01/full OTA', passed=False, cases={})
|
|
try:
|
|
# Real dpkg upgrade whose postinst activates Debian's libc-upgrade trigger.
|
|
for version in ('1', '2'):
|
|
package = BASE / ('build-' + version)
|
|
(package / 'DEBIAN').mkdir(parents=True)
|
|
(package / 'DEBIAN/control').write_text('Package: os-manager-fixture\nVersion: '+version+'\nArchitecture: all\nMaintainer: Fixture <fixture@example.invalid>\nDescription: private PID1 trigger fixture\n')
|
|
(package / 'opt').mkdir()
|
|
(package / 'opt/os-manager-canary').write_text(version)
|
|
if version == '2':
|
|
postinst = package / 'DEBIAN/postinst'
|
|
postinst.write_text('#!/bin/sh\nset -e\ndpkg-trigger --no-await libc-upgrade\n')
|
|
postinst.chmod(0o755)
|
|
command('dpkg-deb', '--build', str(package), str(BASE / ('package-' + version + '.deb')))
|
|
command('dpkg', '-i', str(BASE / 'package-1.deb'))
|
|
systemd_version, arch = command('dpkg-query', '-W', '-f=${Version}\t${Architecture}', 'systemd').split('\t')
|
|
Path('/etc/systemd/system/archipelago.service').write_text('[Unit]\nDescription=Stopped manager-proof fixture app\n[Service]\nExecStart=/bin/sleep infinity\n')
|
|
command('systemctl', 'daemon-reload')
|
|
command('systemctl', 'stop', 'archipelago.service')
|
|
for case in CASES:
|
|
root = BASE / case
|
|
root.mkdir(mode=0o700)
|
|
for name in ('ledger', 'run'):
|
|
(root / name).mkdir(mode=0o700)
|
|
binding = dict(transaction_id=digest(case.encode())[:32], admission_epoch_id='b'*32)
|
|
value = store.initial(binding)
|
|
value['execution'] = dict(unit=unit(case), incarnation_id='c'*32, executor=None)
|
|
store.atomic(root / 'binding.json', canonical(binding))
|
|
store.atomic(root / 'ledger/ledger.json', canonical(value))
|
|
boot = Path('/proc/sys/kernel/random/boot_id').read_text().strip().replace('-', '')
|
|
store.atomic(root / 'run/bootstrap.json', canonical(dict(boot_id=boot,
|
|
binding_digest=digest(canonical(binding)), ledger_digest=digest(canonical(value)))))
|
|
policy = dict(version=1, operation='debian-systemd-daemon-reexec', package='systemd',
|
|
package_version=systemd_version, architecture=arch,
|
|
expected_executable_sha256=manager.executable(1), max_unavailable_seconds=2 if case == 'timeout' else 120)
|
|
store.atomic(root / 'policy.json', canonical(policy))
|
|
Path('/etc/systemd/system/archipelago-d09-lifecycle.service').write_text(
|
|
'[Service]\nType=simple\nExecStart=/usr/bin/python3 -B '+SCRIPT+' --owner '+case+'\nRestart=on-failure\nRestartSec=0.2\n')
|
|
Path('/etc/systemd/system/' + unit(case)).write_text(
|
|
'[Service]\nType=exec\nExecStart=/usr/bin/python3 -B '+SCRIPT+' --worker '+case+'\n')
|
|
command('systemctl', 'daemon-reload')
|
|
command('systemctl', 'start', manager.OWNER_UNIT)
|
|
before = wait(lambda: call(case, 'inspect'))
|
|
assert before['anchor'] and not before['blocked']
|
|
command('systemctl', 'start', '--no-block', unit(case))
|
|
wait(lambda: (root / 'worker-ready').exists() or (root / 'worker-result.json').exists(), 30)
|
|
assert (root / 'worker-ready').exists(), (root / 'worker-result.json').read_text()
|
|
armed = call(case, 'inspect')
|
|
if case != 'unplanned':
|
|
assert armed['state'] == 'ARMED' and armed['refs'] == 3, armed
|
|
private = Path('/run/systemd/private')
|
|
retained = Path('/run/systemd/private.manager-fixture-retained')
|
|
fake = None
|
|
if case in ('good', 'timeout', 'bad-peer', 'unplanned'):
|
|
private.rename(retained)
|
|
try:
|
|
if case == 'bad-peer':
|
|
fake = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
fake.bind(str(private))
|
|
fake.listen(8)
|
|
if case == 'unplanned':
|
|
(root / 'run-unplanned').touch()
|
|
observed = wait(lambda: (x if (x := call(case, 'inspect'))['blocked'] or x['state'] == 'UNCERTAIN' else None))
|
|
if case == 'timeout':
|
|
observed = wait(lambda: (x if (x := call(case, 'inspect'))['blocked'] else None), 15)
|
|
assert observed['pid'] == before['pid'] and observed['anchor'], observed
|
|
if case != 'unplanned':
|
|
assert observed['refs'] == 3, observed
|
|
finally:
|
|
if fake is not None:
|
|
fake.close()
|
|
private.unlink()
|
|
retained.rename(private)
|
|
elif case == 'owner-loss':
|
|
os.kill(before['pid'], signal.SIGKILL)
|
|
observed = wait(lambda: (x if (x := call(case, 'inspect'))['pid'] != before['pid'] else None), 30)
|
|
assert not observed['anchor'] and observed['state'] is None, observed
|
|
else:
|
|
command('systemctl', 'restart', unit(case))
|
|
observed = wait(lambda: (x if (x := call(case, 'inspect'))['blocked'] else None), 15)
|
|
assert observed['anchor'] and observed['refs'] == 3
|
|
(root / 'go').touch()
|
|
if case != 'worker-loss':
|
|
worker = wait(lambda: json.loads((root / 'worker-result.json').read_text()), 90)
|
|
assert worker['completed'] == (case == 'good'), worker
|
|
assert worker['dpkg_invocations'] == (1 if case == 'good' else 0), worker
|
|
else:
|
|
worker = dict(replaced_lifetime_refused=True)
|
|
after = call(case, 'inspect')
|
|
if case == 'good':
|
|
states = [e['state'] for e in after['events']]
|
|
assert 'UNCERTAIN' in states and 'RECOVERED' in states and states[-1] == 'COMPLETED', states
|
|
assert after['pid'] == before['pid'] and after['incarnation'] == before['incarnation'] and after['epoch'] == before['epoch']
|
|
assert Path('/opt/os-manager-canary').read_text() == '2'
|
|
else:
|
|
assert after['blocked'] or not after['anchor'], after
|
|
result['cases'][case] = dict(before=before, after=after, worker=worker)
|
|
(root / 'release').touch()
|
|
command('systemctl', 'stop', unit(case), manager.OWNER_UNIT)
|
|
journal = command('journalctl', '_PID=1', '--no-pager')
|
|
(EVIDENCE / 'pid1-journal.log').write_text(journal)
|
|
assert 'Reexecuting' in journal, 'actual PID1 reexec observation missing'
|
|
result['passed'] = True
|
|
finally:
|
|
(EVIDENCE / 'result.json').write_text(json.dumps(result, indent=2))
|
|
print(json.dumps(dict(passed=result['passed'], cases=list(result['cases'])), sort_keys=True), flush=True)
|
|
|
|
|
|
if len(sys.argv) == 3 and sys.argv[2] in CASES and sys.argv[1] in ('--owner', '--worker'):
|
|
(owner_mode if sys.argv[1] == '--owner' else worker_mode)(sys.argv[2])
|
|
elif len(sys.argv) == 1:
|
|
driver()
|
|
else:
|
|
raise SystemExit('fixture invocation')
|