Files
archy/tests/regression/os-ota-installed-d09-admission.py

109 lines
6.1 KiB
Python

#!/usr/bin/env python3
"""Actual native release verification and production D09 admission on a VM.
An old signed reference is only a positive verifier control, never authority for
the future OS plan. Missing production signature must remain a real blocker.
"""
import json
from pathlib import Path
import shutil
import subprocess
import sys
import time
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'
PACKAGE=Path('/opt/ota-source')
sys.path[:0]=[str(PACKAGE/'scripts'),str(PACKAGE/'scripts/tests')]
import os_update_executor as executor
from os_update_plan import canonical,digest
import os_ota_d09_guest as d09
EVIDENCE=Path('/var/log/os-ota-stage')
native=Path('/opt/os-ota-native')
proof=json.loads(Path('/opt/os-ota-native-proof.json').read_text())
assert executor.file_hash(native)==proof['packaged_sha256']
assert native.read_bytes()[:4]==b'\x7fELF'
native.chmod(0o755)
positive=subprocess.run([str(native),'ceremony','verify',str(PACKAGE/'release-manifest.json')],capture_output=True,text=True)
(EVIDENCE/'native-signed-reference.log').write_text(positive.stdout+positive.stderr)
print('NATIVE_SIGNED_REFERENCE',positive.returncode,positive.stdout,positive.stderr,flush=True)
positive.check_returncode()
before=executor.inventory()
canaries=Path('/var/lib/archipelago/ota-synthetic-canaries')
preserved={p.name:executor.file_hash(p) for p in canaries.iterdir()}
context=d09.setup()
provision,binding=context
owner=provision.owner
# This admission attempt binds the actual compiled candidate, not the inert
# requester ELF used by the positive systemd component fixture.
native_sha=executor.file_hash(native)
for path in (owner.DESTINATIONS['backend'],owner.ROOT/'incoming/new-backend',owner.ROOT/'incoming/restore-backend'):
shutil.copyfile(native,path)
path.chmod(0o755)
binding['objects']['backend']={k:owner.object_digest(owner.DESTINATIONS['backend']) for k in ('old','new','restore')}
binding['artifacts']['backend']=binding['restore_artifacts']['backend']=native_sha
binding['prerequisite']['backend_sha256']=native_sha
unit=Path('/etc/systemd/system/archipelago.service')
unit.write_text(unit.read_text().replace('/usr/local/bin/archipelago /opt/ota-source/d09-os-requester.py','/usr/local/bin/archipelago'))
d09.run('systemctl','daemon-reload')
old_effective=digest(owner.effective_configuration().encode())
config=(PACKAGE/'image-recipe/configs/60-terminal-workspace.conf').read_bytes()
owner.DESTINATIONS['config'].write_bytes(config)
d09.run('systemctl','daemon-reload')
new_effective=digest(owner.effective_configuration().encode())
owner.DESTINATIONS['config'].unlink()
d09.run('systemctl','daemon-reload')
binding['effective_config']=dict(APPLY=new_effective,RESTORE=old_effective)
binding['prerequisite']['effective_config_digest']=old_effective
old=json.loads((EVIDENCE/'binding.json').read_text())
old_source=executor.ROOT/'plans'/old['os_update']['plan_sha256']
plan=json.loads((old_source/'plan.json').read_text())
preservation=json.loads((old_source/'preservation.json').read_text())
authority=digest(('native-D09-admission-'+str(time.time_ns())).encode())
preservation['authority_digest']=authority
plan['preservation_sha256']=digest(canonical(preservation))
sha=digest(canonical(plan))
version='os-ota-disposable-20261103'
components=[dict(name='archipelago',current_version=version,new_version=version,
sha256=binding['artifacts']['backend'],size_bytes=(owner.ROOT/'incoming/new-backend').stat().st_size,
download_url='https://example.invalid/private-fixture/archipelago'),
dict(name='os-update-plan.json',current_version=version,new_version=version,sha256=sha,
size_bytes=len(canonical(plan)),download_url='https://example.invalid/private-fixture/os-update-plan.json')]
manifest=dict(version=version,release_date='2026-11-03',changelog=['Private future OS ability test; NOT production approval'],
components=components,os_update=dict(version=1,plan_sha256=sha))
incoming=owner.ROOT/'incoming/os-update'
incoming.mkdir(mode=0o700)
for name,value in (('plan.json',plan),('manifest.json',manifest),('preservation.json',preservation)):
(incoming/name).write_bytes(canonical(value))
shutil.copyfile(native,incoming/'verifier')
binding['os_update']=dict(version=1,plan_sha256=sha,manifest_sha256=digest(canonical(manifest)),
authority_digest=authority,verifier_sha256=executor.file_hash(native))
binding['release_identity']=[version,sorted([[c[k] for k in ('name','new_version','sha256','size_bytes')] for c in components])]
admission=owner.ROOT/'incoming/admission.json'
admission.write_bytes(canonical(binding))
# Replace only the explicitly labelled private fixture verifier on this new
# independent disk, retaining it; never replace an installed native verifier.
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-fixture'))
result=dict(actual_native_verifier=True,native_artifact=proof,signed_reference_verified=True,
actual_native_candidate_bound=True,real_D09_provisioner_invoked=True,
production_approved=False,ota_apply_passed=False)
try:
provision.provision(admission)
except subprocess.CalledProcessError as error:
text=(error.stdout or '')+(error.stderr or '')
print('ACTUAL_D09_ADMISSION_REFUSAL',error.returncode,text,flush=True)
assert 'signature' in text.lower(),text
result.update(expected_refusal_passed=True,blocker='future OS candidate has no pinned-release-root signature',
verifier_exit=error.returncode,verifier_output=text)
else:
raise AssertionError('unsigned future OS candidate was admitted')
assert not owner.BINDING.exists() and not (owner.ROOT/'ledger.json').exists()
assert executor.inventory()==before
assert {p.name:executor.file_hash(p) for p in canaries.iterdir()}==preserved
result.update(no_D09_authority_published=True,dpkg_unchanged=True,canaries_preserved=True)
(EVIDENCE/'d09-native-admission-result.json').write_text(json.dumps(result,indent=2)+'\n')
print(json.dumps(result,indent=2),flush=True)