256 lines
15 KiB
Python
256 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Real guest package effects through os_update_executor.stage, unsigned fixture.
|
|
|
|
The root-appointed verifier accepts ONLY this exact disposable fixture manifest.
|
|
This does not exercise production signature approval or full native D09 transport.
|
|
"""
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import lzma
|
|
import os
|
|
from pathlib import Path
|
|
import platform
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
|
|
assert 'QEMU' in Path('/sys/class/dmi/id/sys_vendor').read_text()
|
|
assert subprocess.check_output(['findmnt', '-n', '-o', 'SOURCE', '/'], text=True).strip() == '/dev/vda'
|
|
assert platform.release() == '6.12.107+deb13-amd64'
|
|
PACKAGE = Path('/opt/ota-source')
|
|
sys.path.insert(0, str(PACKAGE/'scripts'))
|
|
import os_update_executor as executor
|
|
import os_update_repository as repository
|
|
from os_update_plan import canonical, digest, protected, simulation, validate
|
|
|
|
os.environ['DEBIAN_FRONTEND'] = 'noninteractive'
|
|
EVIDENCE = Path('/var/log/os-ota-stage')
|
|
(executor.ROOT/'transactions').mkdir(parents=True, mode=0o700, exist_ok=True)
|
|
if EVIDENCE.exists():
|
|
# Retain failed planning attempts; never recycle a prepared authority.
|
|
if (EVIDENCE/'binding.json').exists():
|
|
previous = executor.Executor(json.loads((EVIDENCE/'binding.json').read_text())['os_update'])
|
|
assert previous.status['phase'] in (None, 'PREPARE', 'DOWNLOAD'), 'existing prepared/applied authority must not be reset'
|
|
for log in sorted(previous.root.glob('apt-*.log')):
|
|
print('RETAINED_FAILED_APT_LOG',str(log),log.read_text(),flush=True)
|
|
EVIDENCE.rename(EVIDENCE.with_name('os-ota-stage-failed-'+str(time.time_ns())))
|
|
EVIDENCE.mkdir(exist_ok=False)
|
|
RESULT = {'scope': 'actual privileged D09 OS executor stage in installed Debian VM',
|
|
'production_signature_approval': False, 'native_D09_transport': False,
|
|
'checks': [], 'passed': False}
|
|
|
|
|
|
def run(*argv, env=None, check=True):
|
|
print('COMMAND', argv, flush=True)
|
|
result = subprocess.run(argv, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
|
|
print(result.stdout, flush=True)
|
|
if check:
|
|
result.check_returncode()
|
|
return result
|
|
|
|
|
|
def hashes():
|
|
root = Path('/var/lib/archipelago/ota-synthetic-canaries')
|
|
return {p.name: executor.file_hash(p) for p in root.iterdir()}
|
|
|
|
|
|
def record(name, **details):
|
|
RESULT['checks'].append(dict(name=name, **details))
|
|
(EVIDENCE/'result.json').write_text(json.dumps(RESULT, indent=2)+'\n')
|
|
print('CHECK', name, json.dumps(details), flush=True)
|
|
|
|
|
|
def refused(name, function):
|
|
inventory = executor.inventory()
|
|
before = hashes()
|
|
try:
|
|
function()
|
|
except Exception as error:
|
|
assert inventory == executor.inventory(), 'refusal changed dpkg state'
|
|
assert before == hashes(), 'refusal changed protected canaries'
|
|
record(name, refused=True, error=str(error), exception=type(error).__name__)
|
|
return
|
|
raise AssertionError(name+' unexpectedly admitted')
|
|
|
|
|
|
try:
|
|
# Establish the maintenance writer exclusion in this single-purpose VM,
|
|
# rather than merely asserting it in a JSON receipt.
|
|
run('systemctl','mask','--now','apt-daily.service','apt-daily-upgrade.service',
|
|
'apt-daily.timer','apt-daily-upgrade.timer','unattended-upgrades.service')
|
|
policy = Path('/usr/sbin/policy-rc.d')
|
|
if policy.exists():
|
|
policy.rename(EVIDENCE/'inherited-container-policy-rc.d')
|
|
record('guest-maintenance-exclusion', apt_timers_masked=True,
|
|
inherited_container_service_suppression_removed=True)
|
|
# Establish a real old metapackage so dist-upgrade, not a manual kernel fix,
|
|
# selects its new exact kernel dependency. These are baseline setup effects.
|
|
run('apt-get', '-y', '--no-install-recommends', 'install',
|
|
'linux-image-amd64=6.12.107-1', 'gpgv', 'fdisk', 'e2fsprogs', 'python3-apt')
|
|
# Separate tiny disposable BIOS disk; installed root remains /dev/vda.
|
|
# Never accepts a caller-supplied device, and requires the harness's 64 MiB.
|
|
assert int(Path('/sys/class/block/sda/size').read_text()) == 64*1024**2//512
|
|
partition = subprocess.run(['sfdisk','/dev/sda'],input='label: dos\nstart=2048,type=83\n',
|
|
text=True,capture_output=True)
|
|
print(partition.stdout,partition.stderr,flush=True)
|
|
partition.check_returncode()
|
|
Path('/etc/default/grub').write_text('GRUB_TIMEOUT=1\nGRUB_TERMINAL=serial\nGRUB_SERIAL_COMMAND="serial --speed=115200"\nGRUB_CMDLINE_LINUX="console=ttyS0 net.ifnames=0 noresume"\n')
|
|
run('grub-install','--target=i386-pc','--boot-directory=/boot','/dev/sda')
|
|
run('update-grub')
|
|
assert not run('dpkg', '--audit').stdout.strip()
|
|
before_canaries = hashes()
|
|
(EVIDENCE/'canaries-before.json').write_bytes(canonical(before_canaries))
|
|
planning = Path('/var/lib/os-ota-fixture-planning-'+str(time.time_ns()))
|
|
planning.mkdir(mode=0o700, exist_ok=False)
|
|
repositories = []
|
|
metadata = {}
|
|
pockets = ('trixie','trixie-updates','trixie-security')
|
|
components = ('main','contrib','non-free','non-free-firmware')
|
|
for pocket in pockets:
|
|
base = 'https://deb.debian.org/' + ('debian-security' if pocket.endswith('-security') else 'debian')
|
|
with urllib.request.urlopen(base+'/dists/'+pocket+'/InRelease', timeout=120) as response:
|
|
raw = response.read(8*1024*1024)
|
|
signed = planning/(pocket+'.InRelease')
|
|
signed.write_bytes(raw)
|
|
verified = planning/(pocket+'.Release')
|
|
run('gpgv', '--keyring', repository.KEYRING, '--output', str(verified), str(signed))
|
|
fields = repository.paragraphs(verified.read_bytes())[0]
|
|
assert fields['Origin'] == 'Debian' and fields['Codename'] == pocket
|
|
entries = [line.split() for line in fields['SHA256'].splitlines() if line.strip()]
|
|
indexes = []
|
|
directory = planning/'mirror'/pocket/'dists'/pocket
|
|
directory.mkdir(parents=True)
|
|
(directory/'InRelease').write_bytes(raw)
|
|
for component in components:
|
|
sha, size, path = next(row for row in entries if row[2] == component+'/binary-amd64/Packages.xz')
|
|
indexes.append(dict(path=path, sha256=sha, size=int(size)))
|
|
index = repository.fetch(base+'/dists/'+pocket+'/'+path,sha,int(size))
|
|
(directory/path).parent.mkdir(parents=True,exist_ok=True)
|
|
(directory/path).write_bytes(index)
|
|
for row in repository.paragraphs(lzma.decompress(index,memlimit=256*1024**2)):
|
|
key = (pocket,row.get('Package'),row.get('Version'),row.get('Architecture'))
|
|
assert key not in metadata or metadata[key] == row
|
|
metadata[key] = row
|
|
repositories.append(dict(pocket=pocket,inrelease_sha256=digest(raw),indexes=indexes))
|
|
for name in ('empty', 'lists/partial', 'archives/partial'):
|
|
(planning/name).mkdir(parents=True)
|
|
sources = '\n'.join('deb [signed-by='+repository.KEYRING+'] file:'+str(planning/'mirror'/pocket)+' '+pocket+' '+' '.join(components)
|
|
for pocket in pockets)+'\n'
|
|
(planning/'sources.list').write_text(sources)
|
|
config = {'Dir::Etc::main': '-', 'Dir::Etc::parts': str(planning/'empty'),
|
|
'Dir::Etc::sourcelist': str(planning/'sources.list'), 'Dir::Etc::sourceparts': str(planning/'empty'),
|
|
'Dir::Etc::preferences': '-', 'Dir::Etc::preferencesparts': str(planning/'empty'),
|
|
'Dir::State::lists': str(planning/'lists'), 'Dir::Cache::archives': str(planning/'archives'),
|
|
'APT::Sandbox::User': 'root', 'APT::Get::AllowUnauthenticated': 'false',
|
|
'Acquire::Languages': 'none'}
|
|
(planning/'apt.conf').write_text('\n'.join(k+' '+json.dumps(v)+';' for k,v in config.items()))
|
|
env = dict(os.environ, APT_CONFIG=str(planning/'apt.conf'), LC_ALL='C')
|
|
run('apt-get', 'update', env=env)
|
|
solver_text = run('apt-get', '--simulate', 'dist-upgrade', env=env).stdout
|
|
(EVIDENCE/'planned-solver.log').write_text(solver_text)
|
|
transaction = simulation(solver_text)
|
|
packages = []
|
|
for item in transaction:
|
|
if item['operation'] != 'Inst':
|
|
continue
|
|
name = item['name'].split(':')[0]
|
|
matches = [(key, row) for key, row in metadata.items()
|
|
if key[1] == name and key[2] == item['version'] and key[3] in ('all', 'amd64')]
|
|
assert matches, item
|
|
key, row = matches[-1]
|
|
packages.append(dict(name=name, version=key[2], architecture=key[3],
|
|
pocket=key[0], sha256=row['SHA256'], size=int(row['Size']), filename=row['Filename']))
|
|
targets = [p['name'].removeprefix('linux-image-') for p in packages
|
|
if p['name'].startswith('linux-image-6.')]
|
|
assert len(targets) == 1, targets
|
|
changed = {x['name'].split(':')[0] for x in transaction if x['operation'] in ('Inst', 'Remv')}
|
|
installed = {line.split('\t')[0].split(':')[0] for line in executor.inventory().decode().splitlines()}
|
|
authority = digest(('os-ota-private-installed-fixture-20261103-'+str(time.time_ns())).encode())
|
|
preservation = dict(version=1, authority_digest=authority,
|
|
service_policy='reviewed-maintenance-window', exclusive_package_writer=True,
|
|
preserve_wallets_data_identity_stop_uninstall=True,
|
|
synthetic_canaries=before_canaries)
|
|
plan = dict(version=1, suite='trixie', architecture='amd64', before_sha256=digest(executor.inventory()),
|
|
repositories=repositories, packages=packages, transaction=transaction,
|
|
removals=sorted(x['name'] for x in transaction if x['operation']=='Remv'),
|
|
custody=[dict(name=n, disposition='upgrade-reviewed' if n in changed else 'retain')
|
|
for n in sorted(installed|changed) if protected(n)],
|
|
service_policy='reviewed-maintenance-window', preservation_sha256=digest(canonical(preservation)),
|
|
boot=dict(running_release=platform.release(), expected_release=targets[0],
|
|
package=next(p for p in packages if p['name']=='linux-image-'+targets[0]),
|
|
retained_files={p: executor.file_hash(p) for p in (
|
|
'/boot/vmlinuz-'+platform.release(), '/boot/initrd.img-'+platform.release(), '/boot/grub/grub.cfg')}))
|
|
# Explicit private-VM permission, measured before any plan hash/signature.
|
|
# This fixture covers the unchanged systemd package's libc-upgrade trigger.
|
|
assert not any(p['name']=='systemd' for p in packages), 'changed systemd requires reviewed target ELF provenance'
|
|
systemd_version,systemd_arch=run('dpkg-query','-W','-f=${Version}\t${Architecture}','systemd').stdout.strip().split('\t')
|
|
systemd_hash=executor.file_hash('/usr/lib/systemd/systemd')
|
|
assert executor.file_hash('/proc/1/exe')==systemd_hash
|
|
plan['manager_reexec']=dict(version=1,operation='debian-systemd-daemon-reexec',
|
|
package='systemd',package_version=systemd_version,architecture=systemd_arch,
|
|
expected_executable_sha256=systemd_hash,max_unavailable_seconds=120)
|
|
validate(plan)
|
|
del metadata
|
|
plan_sha = digest(canonical(plan))
|
|
manifest = dict(os_update=dict(version=1, plan_sha256=plan_sha),
|
|
qualification='ROOT-APPOINTED-DISPOSABLE-FIXTURE-NOT-PRODUCTION')
|
|
manifest_sha = digest(canonical(manifest))
|
|
source = executor.ROOT/'plans'/plan_sha
|
|
source.mkdir(parents=True, mode=0o700)
|
|
(executor.ROOT/'transactions').mkdir(mode=0o700, exist_ok=True)
|
|
for name, value in (('plan.json',plan), ('manifest.json',manifest), ('preservation.json',preservation)):
|
|
(source/name).write_bytes(canonical(value))
|
|
# Exact hash appointment in a private guest only, never a replacement for a
|
|
# production verifier. Wrong manifests and non-VM invocation fail closed.
|
|
if executor.VERIFIER.exists():
|
|
assert 'ROOT_APPOINTED_EXACT_FIXTURE_ONLY_NOT_PRODUCTION_SIGNATURE' in executor.VERIFIER.read_text()
|
|
executor.VERIFIER.rename(executor.VERIFIER.with_name(executor.VERIFIER.name+'.retained-'+str(time.time_ns())))
|
|
verifier = '''#!/usr/bin/python3
|
|
import hashlib,pathlib,sys
|
|
assert 'QEMU' in pathlib.Path('/sys/class/dmi/id/sys_vendor').read_text()
|
|
assert sys.argv[1:3] == ['ceremony', 'verify'] and len(sys.argv) == 4
|
|
assert hashlib.sha256(pathlib.Path(sys.argv[3]).read_bytes()).hexdigest() == %r
|
|
print('ROOT_APPOINTED_EXACT_FIXTURE_ONLY_NOT_PRODUCTION_SIGNATURE')
|
|
''' % manifest_sha
|
|
executor.VERIFIER.parent.mkdir(parents=True, exist_ok=True)
|
|
executor.VERIFIER.write_text(verifier)
|
|
executor.VERIFIER.chmod(0o755)
|
|
binding = dict(os_update=dict(version=1, plan_sha256=plan_sha, manifest_sha256=manifest_sha,
|
|
authority_digest=authority, verifier_sha256=executor.file_hash(executor.VERIFIER)))
|
|
(EVIDENCE/'binding.json').write_bytes(canonical(binding))
|
|
(EVIDENCE/'plan.json').write_bytes(canonical(plan))
|
|
record('authenticated-full-plan', packages=packages, transaction=transaction, target_kernel=targets[0])
|
|
status = executor.stage(binding, 'APPLY', 'prepare')
|
|
assert status['phase'] == 'VERIFIED'
|
|
record('actual-stage-prepare', phase=status['phase'])
|
|
# Repeated prepare must produce no new intent or package effects.
|
|
assert executor.stage(binding, 'APPLY', 'prepare') == status
|
|
refused('restore-refused-no-financial-rollback', lambda: executor.stage(binding, 'RESTORE', 'apply'))
|
|
refusal_binding = json.loads(json.dumps(binding))
|
|
refusal_binding['os_update']['plan_sha256'] = '0'*64
|
|
refused('wrong-plan-binding-refused', lambda: executor.stage(refusal_binding, 'APPLY', 'apply'))
|
|
instance = executor.Executor(binding['os_update'])
|
|
archive = next((instance.root/'archives').glob('*.deb'))
|
|
retained = archive.with_suffix('.retained')
|
|
archive.rename(retained)
|
|
try:
|
|
refused('missing-archive-refused', lambda: executor.stage(binding, 'APPLY', 'apply'))
|
|
archive.write_bytes(b'bad hash disposable fixture\n')
|
|
refused('bad-hash-refused', lambda: executor.stage(binding, 'APPLY', 'apply'))
|
|
archive.unlink()
|
|
finally:
|
|
retained.rename(archive)
|
|
# Hold the actual dpkg frontend lock, not a mocked lock check. This consumes
|
|
# APPLY_INTENT on current executor implementations; preserve its evidence.
|
|
# Do this only in a separate fault run; the successful transaction is never
|
|
# reset/replayed by rewriting its protected status.
|
|
record('faults-deferred-to-separate-authority', cases=['locked-apt','offline-download','space','missing-dependency','partial-dpkg','interrupted-before-stage'])
|
|
record('prepared-for-independent-fault-disks', phase='VERIFIED')
|
|
RESULT['passed'] = True
|
|
finally:
|
|
(EVIDENCE/'result.json').write_text(json.dumps(RESULT, indent=2)+'\n')
|
|
print(json.dumps(RESULT, indent=2), flush=True)
|