172 lines
8.6 KiB
Python
172 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Real APT/dpkg transaction and failing postinst, ONLY in a fresh bwrap namespace.
|
|
|
|
Fixture repository is deliberately unsigned and trusted only in fixture config.
|
|
Production authentication is not bypassed: Executor.__init__/prepare are not used.
|
|
This qualifies apply/journal/reconciliation, not release admission or boot.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
def run(*args):
|
|
result = subprocess.run(args, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
if result.returncode:
|
|
raise RuntimeError(str(args) + '\n' + result.stdout + '\n' + result.stderr)
|
|
return result.stdout
|
|
|
|
|
|
def durable(path, raw, mode=0o600):
|
|
temporary = path.with_suffix('.new')
|
|
with open(temporary, 'wb') as out:
|
|
out.write(raw)
|
|
out.flush()
|
|
os.fsync(out.fileno())
|
|
os.replace(temporary, path)
|
|
fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
|
|
os.fsync(fd)
|
|
os.close(fd)
|
|
|
|
|
|
def inside():
|
|
assert os.getuid() == 0
|
|
os.environ['PATH'] = '/usr/sbin:/usr/bin:/sbin:/bin'
|
|
assert os.environ['OS_OTA_DISPOSABLE_PARENT_USERNS'] != str(os.stat('/proc/self/ns/user').st_ino)
|
|
assert any(line.split()[4] == '/' and 'ro' in line.split()[5].split(',') for line in pathlib.Path('/proc/self/mountinfo').read_text().splitlines())
|
|
sys.path.insert(0, '/mnt/src')
|
|
import os_update_executor as engine
|
|
from os_update_plan import simulation, digest
|
|
engine.atomic = durable # Fixture filesystem has mapped root, not host UID0 ancestry.
|
|
engine.read = lambda path: pathlib.Path(path).read_bytes()
|
|
root = pathlib.Path('/mnt/work')
|
|
for name in ('/var/lib/dpkg/updates', '/var/log', '/opt/preserved', '/mnt/work/repo'):
|
|
pathlib.Path(name).mkdir(parents=True, exist_ok=True)
|
|
pathlib.Path('/var/lib/dpkg/status').write_text('')
|
|
wallet = pathlib.Path('/opt/preserved/wallet')
|
|
wallet.write_text('wallet sentinel; never snapshot/restore')
|
|
decisions = pathlib.Path('/opt/preserved/decisions.json')
|
|
decisions.write_text('{"stopped":["bitcoin"],"uninstalled":["lnd"]}')
|
|
preservation = {str(p): digest(p.read_bytes()) for p in (wallet, decisions)}
|
|
|
|
def package(name, version, extra='', fail=False):
|
|
directory = root / ('build-' + name + '-' + version)
|
|
(directory / 'DEBIAN').mkdir(parents=True)
|
|
(directory / 'DEBIAN/control').write_text(
|
|
f'Package: {name}\nVersion: {version}\nArchitecture: all\nMaintainer: Disposable <fixture@example.invalid>\nDescription: OS updater disposable fixture\n' + extra)
|
|
destination = directory / 'opt/os-ota-fixture'
|
|
destination.mkdir(parents=True)
|
|
(destination / name).write_text(version)
|
|
if fail:
|
|
script = directory / 'DEBIAN/postinst'
|
|
script.write_text('#!/bin/sh\necho invoked >> /opt/os-ota-fixture/postinst-count\nexit 42\n')
|
|
script.chmod(0o755)
|
|
target = root / 'repo' / f'{name}_{version}_all.deb'
|
|
run('/usr/bin/dpkg-deb', '--build', '--root-owner-group', str(directory), str(target))
|
|
return target
|
|
|
|
old = package('os-ota-base', '1')
|
|
obsolete = package('os-ota-obsolete', '1')
|
|
run('/usr/bin/dpkg', '-i', str(old), str(obsolete))
|
|
base = package('os-ota-base', '2', 'Depends: os-ota-helper (= 1)\nConflicts: os-ota-obsolete\n')
|
|
helper = package('os-ota-helper', '1')
|
|
|
|
def repository(archives):
|
|
paragraphs = []
|
|
for archive in archives:
|
|
control = run('/usr/bin/dpkg-deb', '-f', str(archive)).strip()
|
|
paragraphs.append(control + f'\nFilename: {archive.name}\nSize: {archive.stat().st_size}\nSHA256: {digest(archive.read_bytes())}\n')
|
|
(root / 'repo/Packages').write_text('\n'.join(paragraphs))
|
|
|
|
def executor(name, archives):
|
|
obj = engine.Executor.__new__(engine.Executor)
|
|
obj.root = root / name
|
|
obj.root.mkdir()
|
|
for directory in ('lists/partial', 'archives/partial', 'empty'):
|
|
(obj.root / directory).mkdir(parents=True)
|
|
obj.status_path = obj.root / 'status.json'
|
|
(obj.root / 'before-inventory').write_bytes(engine.inventory())
|
|
obj.status = dict(version=1, plan_sha256='a'*64, authority_digest=digest(name.encode()), phase='VERIFIED', sequence=0, events=[])
|
|
sources = obj.root / 'sources.list'
|
|
sources.write_text('deb [trusted=yes] file:/mnt/work/repo ./\n')
|
|
config = dict(**{'Dir::Etc::main': '-', 'Dir::Etc::parts': str(obj.root / 'empty'),
|
|
'Dir::Etc::sourcelist': str(sources), 'Dir::Etc::sourceparts': str(obj.root / 'empty'),
|
|
'Dir::Etc::preferences': '-', 'Dir::Etc::preferencesparts': str(obj.root / 'empty'),
|
|
'Dir::State::lists': str(obj.root / 'lists'), 'Dir::Cache::archives': str(obj.root / 'archives'),
|
|
'Dir::Cache::pkgcache': '', 'Dir::Cache::srcpkgcache': '', 'APT::Sandbox::User': 'root'})
|
|
(obj.root / 'apt.conf').write_text('\n'.join(k + ' ' + json.dumps(v) + ';' for k, v in config.items()))
|
|
obj.apt('update')
|
|
transaction = simulation(obj.apt('--simulate', 'dist-upgrade'))
|
|
obj.plan = dict(transaction=transaction, packages=[dict(name=run('/usr/bin/dpkg-deb', '-f', str(a), 'Package').strip(),
|
|
version=run('/usr/bin/dpkg-deb', '-f', str(a), 'Version').strip(), architecture='all', sha256=digest(a.read_bytes())) for a in archives],
|
|
removals=[x['name'] for x in transaction if x['operation'] == 'Remv'], boot={'retained_files': {}})
|
|
obj.apt('--download-only', '--assume-yes', 'dist-upgrade')
|
|
# file repositories may use the original file instead of copying to cache.
|
|
for a in archives:
|
|
(obj.root / 'archives' / a.name).write_bytes(a.read_bytes())
|
|
obj.preflight = lambda: None # Host suite/kernel admission is out of this fixture's scope.
|
|
obj.boot_provenance = lambda: {'fixture_only': True}
|
|
return obj
|
|
|
|
repository([base, helper])
|
|
success = executor('success', [base, helper])
|
|
assert success.plan['removals'] == ['os-ota-obsolete']
|
|
success.apply()
|
|
assert success.status['phase'] == 'REBOOT_REQUIRED'
|
|
assert pathlib.Path('/opt/os-ota-fixture/os-ota-base').read_text() == '2'
|
|
assert not pathlib.Path('/opt/os-ota-fixture/os-ota-obsolete').exists()
|
|
before_repeat = engine.inventory()
|
|
success.apply()
|
|
assert engine.inventory() == before_repeat
|
|
broken = package('os-ota-base', '3', 'Depends: os-ota-helper (= 1)\n', fail=True)
|
|
repository([broken, helper])
|
|
failure = executor('failure', [broken])
|
|
try:
|
|
failure.apply()
|
|
raise AssertionError('failing maintainer script accepted')
|
|
except ValueError:
|
|
pass
|
|
assert failure.status['phase'] == 'RECONCILIATION_REQUIRED'
|
|
assert run('/usr/bin/dpkg', '--audit').strip()
|
|
failure.status = json.loads(failure.status_path.read_bytes())
|
|
try:
|
|
failure.apply()
|
|
raise AssertionError('replayed partial dpkg transaction')
|
|
except engine.ReconciliationRequired:
|
|
pass
|
|
assert pathlib.Path('/opt/os-ota-fixture/postinst-count').read_text() == 'invoked\n'
|
|
assert preservation == {p: digest(pathlib.Path(p).read_bytes()) for p in preservation}
|
|
result = dict(real_apt_dpkg=True, dependency_upgrade=True, explicit_conflict_removal=True,
|
|
partial_dpkg_reconciliation=True, maintainer_script_invocations=1, preserved_fixture_wallet_and_decisions=True,
|
|
production_authentication_tested=False, boot_tested=False)
|
|
(root / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
|
|
print(json.dumps(result, sort_keys=True))
|
|
|
|
|
|
def main():
|
|
if sys.argv[1:] == ['--inside']:
|
|
inside()
|
|
return
|
|
target = pathlib.Path(sys.argv[1]).resolve()
|
|
assert not target.exists() and target.parent.is_dir()
|
|
target.mkdir(mode=0o700)
|
|
source = pathlib.Path(__file__).resolve().parents[1]
|
|
env = dict(os.environ, OS_OTA_DISPOSABLE_PARENT_USERNS=str(os.stat('/proc/self/ns/user').st_ino),
|
|
CARGO_BUILD_JOBS='1', CARGO_INCREMENTAL='0')
|
|
args = ['bwrap', '--unshare-all', '--die-with-parent', '--ro-bind', '/', '/',
|
|
'--tmpfs', '/tmp', '--tmpfs', '/run', '--tmpfs', '/home', '--tmpfs', '/var',
|
|
'--tmpfs', '/opt', '--tmpfs', '/mnt', '--proc', '/proc', '--dev', '/dev', '--uid', '0', '--gid', '0',
|
|
'--ro-bind', str(source), '/mnt/src', '--bind', str(target), '/mnt/work',
|
|
'/usr/bin/python3', '/mnt/src/tests/os_update_disposable.py', '--inside']
|
|
result = subprocess.run(args, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
|
|
(target / 'integration.log').write_text(result.stdout)
|
|
print(result.stdout)
|
|
raise SystemExit(result.returncode)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|