244 lines
12 KiB
Python
244 lines
12 KiB
Python
"""Installed-VM D09 integration using the shipped client, owner and worker.
|
|
|
|
The requester daemon is an inert fixture. This proves the real privileged D09
|
|
route, not native application/C01 admission or production release signatures.
|
|
No production function, command, APT operation or owner response is replaced.
|
|
"""
|
|
import fcntl
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import pwd
|
|
import secrets
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
PACKAGE=Path('/opt/ota-source')
|
|
EVIDENCE=Path('/var/log/os-ota-stage')
|
|
CONTROL=Path('/run/os-ota-d09-client')
|
|
|
|
|
|
def run(*args,check=True):
|
|
result=subprocess.run(args,capture_output=True,text=True,timeout=90)
|
|
print('D09_COMMAND',args,result.returncode,result.stdout,result.stderr,flush=True)
|
|
if check:
|
|
result.check_returncode()
|
|
return result
|
|
|
|
|
|
def write(path,raw,mode=0o644):
|
|
path=Path(path)
|
|
path.parent.mkdir(parents=True,exist_ok=True)
|
|
path.write_bytes(raw.encode() if isinstance(raw,str) else raw)
|
|
path.chmod(mode)
|
|
|
|
|
|
def wait(predicate,label,seconds=1200):
|
|
end=time.monotonic()+seconds
|
|
while time.monotonic()<end:
|
|
value=predicate()
|
|
if value:
|
|
return value
|
|
time.sleep(.2)
|
|
raise TimeoutError(label)
|
|
|
|
|
|
def setup():
|
|
assert 'QEMU' in Path('/sys/class/dmi/id/sys_vendor').read_text()
|
|
assert run('findmnt','-n','-o','SOURCE','/').stdout.strip()=='/dev/vda'
|
|
sys.path.insert(0,str(PACKAGE/'scripts'))
|
|
import d09_wire as wire
|
|
import d09_store as store
|
|
spec=importlib.util.spec_from_file_location('os_vm_provision',PACKAGE/'scripts/provision-d09-lifecycle.py')
|
|
provision=importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(provision)
|
|
owner=provision.owner
|
|
assert not owner.BINDING.exists() and not (owner.ROOT/'ledger.json').exists(), 'fresh D09 authority required; use independent disk'
|
|
assert all(name in provision.MODULES for name in ('os_update_executor.py','os_update_plan.py','os_update_repository.py')), 'actual provisioner must package OS modules'
|
|
try:
|
|
account=pwd.getpwnam('archipelago')
|
|
except KeyError:
|
|
run('useradd','--uid','1000','--create-home','--home-dir','/home/archipelago','--shell','/usr/sbin/nologin','archipelago')
|
|
account=pwd.getpwnam('archipelago')
|
|
run('usermod','-G','','archipelago')
|
|
run('systemctl','stop','archipelago.service',check=False)
|
|
CONTROL.mkdir(mode=0o700)
|
|
os.chown(CONTROL,account.pw_uid,account.pw_gid)
|
|
cache=Path('/var/lib/archipelago/d09-requests')
|
|
cache.mkdir(mode=0o700)
|
|
os.chown(cache,account.pw_uid,account.pw_gid)
|
|
home=Path('/home/archipelago')
|
|
home.mkdir(exist_ok=True)
|
|
os.chown(home,account.pw_uid,account.pw_gid)
|
|
requester='''import json,pathlib,subprocess,time
|
|
p=pathlib.Path('/run/os-ota-d09-client')
|
|
while True:
|
|
for request in p.glob('request-*.json'):
|
|
output=request.with_name(request.name.replace('request-','result-'))
|
|
if output.exists():continue
|
|
value=json.loads(request.read_text())
|
|
r=subprocess.run(['/usr/libexec/archipelago-d09ctl',*value['args']],input=value.get('stdin',''),capture_output=True,text=True,timeout=25)
|
|
output.write_text(json.dumps({'exit':r.returncode,'stdout':r.stdout,'stderr':r.stderr}))
|
|
time.sleep(.1)
|
|
'''
|
|
write(PACKAGE/'d09-os-requester.py',requester)
|
|
python=Path(sys.executable).resolve().read_bytes()
|
|
write(owner.DESTINATIONS['backend'],python,0o755)
|
|
web=owner.DESTINATIONS['frontend']
|
|
assert not web.exists(), 'preserve unexpected frontend'
|
|
write(web/'index.html','root-appointed OS OTA requester fixture; not production frontend\n')
|
|
write(web/'archipelago-runtime/image-recipe/configs/nginx-archipelago.conf',
|
|
(PACKAGE/'image-recipe/configs/nginx-archipelago.conf').read_bytes())
|
|
write('/etc/systemd/system/archipelago.service','''[Unit]
|
|
Description=Private installed OS OTA updater requester fixture
|
|
[Service]
|
|
Type=exec
|
|
User=archipelago
|
|
ExecStart=/usr/local/bin/archipelago /opt/ota-source/d09-os-requester.py
|
|
NoNewPrivileges=yes
|
|
ProtectSystem=strict
|
|
ProtectHome=read-only
|
|
ReadWritePaths=/run/os-ota-d09-client /var/lib/archipelago/d09-requests
|
|
Restart=no
|
|
''')
|
|
run('systemctl','daemon-reload')
|
|
provision.install_package()
|
|
denials=[]
|
|
for command in (['touch','/usr/local/bin/os-ota-denied'],['touch',str(web/'denied')],['sudo','-n','true']):
|
|
result=run('runuser','-u','archipelago','--',*command,check=False)
|
|
assert result.returncode!=0
|
|
denials.append(dict(command=command,exit=result.returncode))
|
|
incoming=owner.ROOT/'incoming'
|
|
exclusion=wire.canonical(dict(scope='private VM fixture only',observed_denials=denials))
|
|
store.atomic(incoming/'exclusion.json',exclusion)
|
|
for generation in ('new','restore'):
|
|
shutil.copytree(web,incoming/(generation+'-frontend'))
|
|
write(incoming/(generation+'-backend'),python+(b'\nOS-OTA-D09-generation-two\n' if generation=='new' else b''),0o755)
|
|
write(incoming/'new-frontend/index.html','new fixture frontend; actual OS worker test\n')
|
|
config=(PACKAGE/'image-recipe/configs/60-terminal-workspace.conf').read_bytes()
|
|
write(incoming/'new-config',config)
|
|
old_effective=wire.digest(owner.effective_configuration().encode())
|
|
write(owner.DESTINATIONS['config'],config)
|
|
run('systemctl','daemon-reload')
|
|
new_effective=wire.digest(owner.effective_configuration().encode())
|
|
owner.DESTINATIONS['config'].unlink()
|
|
run('systemctl','daemon-reload')
|
|
binding=dict(schema_version=1,target_binding_digest=secrets.token_hex(32),composition_digest=secrets.token_hex(32),
|
|
admission_epoch_id=secrets.token_hex(16),transaction_id=secrets.token_hex(16),
|
|
grants={r:secrets.token_hex(32) for r in ('UPDATER','AUTOMATIC','MANUAL','GUARD','OBSERVER')},
|
|
objects={name:dict(old=owner.object_digest(destination),new=owner.object_digest(incoming/('new-'+name)),
|
|
restore=owner.object_digest(incoming/('restore-'+name))) for name,destination in owner.DESTINATIONS.items()},
|
|
artifacts=dict(backend=wire.digest((incoming/'new-backend').read_bytes()),frontend=owner.object_digest(incoming/'new-frontend')),
|
|
restore_artifacts=dict(backend=wire.digest(python),frontend=owner.object_digest(web)),
|
|
effective_config=dict(APPLY=new_effective,RESTORE=old_effective),
|
|
exclusion_evidence_digest=wire.digest(exclusion),helper_digest=wire.digest(Path('/usr/libexec/archipelago-d09-prepare-workspace').read_bytes()),
|
|
prerequisite=dict(grant_digest=secrets.token_hex(32),backend_sha256=wire.digest(python),effective_config_digest=old_effective))
|
|
return provision,binding
|
|
|
|
|
|
def request(args,stdin=''):
|
|
key=secrets.token_hex(8)
|
|
write(CONTROL/('request-'+key+'.json'),json.dumps(dict(args=args,stdin=stdin)))
|
|
output=CONTROL/('result-'+key+'.json')
|
|
wait(output.exists,'client reply',40)
|
|
value=json.loads(output.read_text())
|
|
print('ACTUAL_UPDATER_CLIENT',json.dumps(value),flush=True)
|
|
return value
|
|
|
|
|
|
def execute(context,os_binding,plan,scenario):
|
|
import os_update_executor as engine
|
|
import d09_wire as wire
|
|
import d09_store as store
|
|
provision,binding=context
|
|
owner=provision.owner
|
|
binding['os_update']=os_binding['os_update']
|
|
version='os-ota-disposable-20261103'
|
|
binding['release_identity']=[version,sorted([
|
|
['archipelago',version,binding['artifacts']['backend'],(owner.ROOT/'incoming/new-backend').stat().st_size],
|
|
['os-update-plan.json',version,binding['os_update']['plan_sha256'],len(wire.canonical(plan))]])]
|
|
source=engine.ROOT/'plans'/binding['os_update']['plan_sha256']
|
|
os_inputs=owner.ROOT/'incoming/os-update'
|
|
os_inputs.mkdir(mode=0o700)
|
|
for name in ('plan.json','manifest.json','preservation.json'):
|
|
shutil.copyfile(source/name,os_inputs/name)
|
|
shutil.copyfile(engine.VERIFIER,os_inputs/'verifier')
|
|
admission=owner.ROOT/'incoming/admission.json'
|
|
store.atomic(admission,wire.canonical(binding))
|
|
provision.provision(admission)
|
|
run('systemctl','start','archipelago-d09-lifecycle.service')
|
|
wait(lambda:owner.PRIVATE.exists() and owner.SOCKET.exists(),'live owner sockets',30)
|
|
wait(lambda:provision.activate_prerequisite()['prerequisite_started'],'one-use prerequisite start',60)
|
|
prior=owner.service()['InvocationID']
|
|
inspect=request(['inspect'])
|
|
assert inspect['exit']==0,inspect
|
|
observed=json.loads(inspect['stdout'])
|
|
assert observed['os_update']['supported'] and observed['os_update']['plan_sha256']==binding['os_update']['plan_sha256']
|
|
(EVIDENCE/'d09-binding.json').write_bytes(wire.canonical(binding))
|
|
before=engine.inventory()
|
|
canaries=Path('/var/lib/archipelago/ota-synthetic-canaries')
|
|
hashes=lambda:{p.name:engine.file_hash(p) for p in canaries.iterdir()}
|
|
preserved=hashes()
|
|
lock=None
|
|
result=dict(production_approved=False,native_application_qualified=False,actual_d09_client_owner_worker=True,scenario=scenario,passed=False)
|
|
try:
|
|
if scenario=='d09-locked-apt':
|
|
lock=open('/var/lib/dpkg/lock-frontend','a')
|
|
fcntl.lockf(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
|
|
response=request(['prepare','--selected-release',json.dumps(binding['release_identity']),
|
|
'--staging-generation',secrets.token_hex(16)])
|
|
assert response['exit']==0,response
|
|
def terminal():
|
|
value=json.loads((owner.ROOT/'ledger.json').read_text())
|
|
receipt=value.get('receipt') or {}
|
|
return value if receipt.get('phase') in ('CONFLICT','AWAITING_OBSERVATION','COMPLETE') else None
|
|
ledger=wait(terminal,'actual D09 worker terminal phase')
|
|
status=engine.Executor(binding['os_update']).status
|
|
result.update(d09_phase=ledger['receipt']['phase'],os_phase=status['phase'],
|
|
prior_invocation=prior,observed_invocation=owner.service()['InvocationID'])
|
|
logs='\n'.join(p.read_text() for p in engine.Executor(binding['os_update']).root.glob('apt-*.log'))
|
|
(EVIDENCE/'d09-apt.log').write_text(logs)
|
|
if scenario in ('d09-locked-apt','d09-missing-dependency'):
|
|
assert ledger['receipt']['phase']=='CONFLICT'
|
|
assert owner.service()['InvocationID']==prior and owner.service()['ActiveState']=='active'
|
|
assert engine.inventory()==before
|
|
if scenario=='d09-locked-apt':
|
|
assert 'lock-frontend' in logs and ('Could not' in logs or 'Unable' in logs),logs
|
|
else:
|
|
assert status['phase']=='DOWNLOAD'
|
|
# Preserve the actual executor exception from the real worker.
|
|
unit=ledger['execution']['unit']
|
|
journal=run('journalctl','-u',unit,'--no-pager').stdout
|
|
(EVIDENCE/'d09-worker.log').write_text(journal)
|
|
assert 'incomplete archive transaction' in journal,journal
|
|
result['manager_uninterrupted']=True
|
|
else:
|
|
assert ledger['receipt']['phase']=='AWAITING_OBSERVATION',ledger['receipt']
|
|
assert status['phase']=='REBOOT_REQUIRED'
|
|
assert owner.service()['InvocationID']!=prior
|
|
assert hashes()==preserved
|
|
with (canaries/'wallet.db').open('a') as stream:
|
|
stream.write('FORWARD-AFTER-D09-OS-APPLY-sequence-18\n')
|
|
(EVIDENCE/'canaries-after.json').write_bytes(wire.canonical(hashes()))
|
|
assert engine.stage(binding,'APPLY','apply')==status
|
|
accepted=next(v['request'] for v in ledger['requests'].values() if v['request']['operation']=='APPLY')
|
|
original=(owner.ROOT/'ledger.json').read_bytes()
|
|
replay=request(['submit'],wire.canonical(accepted).decode())
|
|
assert replay['exit']==0 and json.loads(replay['stdout'])['status']=='REFUSED',replay
|
|
assert (owner.ROOT/'ledger.json').read_bytes()==original
|
|
grub=Path('/boot/grub/grub.cfg').read_text()
|
|
assert plan['boot']['running_release'] in grub and plan['boot']['expected_release'] in grub
|
|
result.update(replay_refused=True,os_repeat_idempotent=True,retained_old_grub_entry=True,
|
|
forward_wallet_write_preserved=True,boot_id=engine.boot_id(),target_kernel=plan['boot']['expected_release'])
|
|
(EVIDENCE/'apply-result.json').write_bytes(wire.canonical(result))
|
|
result.update(passed=True,canaries_preserved=hashes()==preserved if scenario!='d09-apply' else True)
|
|
finally:
|
|
if lock:
|
|
lock.close()
|
|
(EVIDENCE/'d09-result.json').write_text(json.dumps(result,indent=2)+'\n')
|
|
run('journalctl','-u','archipelago-d09-lifecycle.service','--no-pager',check=False)
|
|
print('D09_INSTALLED_RESULT',json.dumps(result),flush=True)
|