275 lines
16 KiB
Python
275 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Native crypto + real bootstrap/D09/OS test. Not candidate qualification.
|
|
|
|
Reuses the peer lane's guest fixture scaffolding with explicit, checked fixture-
|
|
only substitutions. Production helpers/crypto/executor are never monkeypatched.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import socket
|
|
import struct
|
|
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.insert(0,str(PACKAGE/'scripts'))
|
|
import os_update_executor as executor
|
|
import os_updater_activate as activation
|
|
import os_updater_bootstrap as bootstrap
|
|
assert executor.ROOT==activation.OS_ROOT==Path('/var/lib/archipelago-os-update'), 'correct root custody prerequisite'
|
|
assert 'var/lib/archipelago-os-update/transactions' in bootstrap.OS_TRANSACTION_ROOTS
|
|
assert not (PACKAGE/'accepted-old-archipelago').exists(), 'isolated fixture must not claim accepted native candidate'
|
|
fixture=Path('/opt/os-ota-crypto-native')
|
|
proof=json.loads(Path('/opt/os-ota-crypto-fixture.json').read_text())
|
|
assert proof['scope']=='ISOLATED_COMPILETIME_TEST_ROOT_NOT_CANDIDATE'
|
|
assert executor.file_hash(fixture)==proof['binary_sha256']
|
|
fixture.chmod(0o755)
|
|
keys=Path('/root/os-ota-crypto-fixture')
|
|
keys.mkdir(mode=0o700)
|
|
Path('/root/os-ota-crypto.seed').rename(keys/'test.seed')
|
|
(keys/'test.seed').chmod(0o600)
|
|
(keys/'wrong.seed').write_bytes(os.urandom(32))
|
|
(keys/'wrong.seed').chmod(0o600)
|
|
|
|
def preservation_probes(context):
|
|
account=context['account']
|
|
home=Path('/home/archipelago')
|
|
probes=[home/'Work/operator-fixture.txt',home/'.local/state/archipelago/sessions/history-fixture.txt',
|
|
Path('/etc/systemd/system/archipelago.service.d/90-operator-fixture.conf')]
|
|
for path in probes[:2]:
|
|
path.parent.mkdir(parents=True,exist_ok=True)
|
|
path.write_text('preserve exact fixture bytes, identity and intent\n')
|
|
path.chmod(0o600)
|
|
for parent in (path,*path.parents):
|
|
if parent==home.parent:break
|
|
os.chown(parent,account.pw_uid,account.pw_gid)
|
|
probes[2].parent.mkdir(parents=True,exist_ok=True)
|
|
probes[2].write_text('[Service]\nEnvironment=OS_OTA_OPERATOR_CANARY=retained\n')
|
|
context['crypto_preservation_paths']=probes
|
|
context['crypto_preservation_before']=[path_identity(p) for p in probes]
|
|
|
|
def path_identity(path):
|
|
st=path.stat()
|
|
return [str(path),st.st_ino,st.st_mode,st.st_uid,st.st_gid,executor.file_hash(path)]
|
|
|
|
def capture_owner(context):
|
|
pid=int(subprocess.check_output(['systemctl','show','archipelago-d09-lifecycle.service','-p','MainPID','--value'],text=True))
|
|
context['crypto_owner_before']=context['owner'].process(pid)
|
|
context['crypto_epoch_before']=json.loads((context['owner'].ROOT/'ledger.json').read_text())['epoch']
|
|
context['crypto_root_uuid']=subprocess.check_output(['blkid','-s','UUID','-o','value','/dev/vda'],text=True).strip()
|
|
|
|
def crypto_appointment(context):
|
|
binding=context['binding']
|
|
source=context['context']['source']
|
|
manifest=source/'manifest.json'
|
|
unsigned=source/'unsigned-fixture-manifest.json'
|
|
manifest.rename(unsigned)
|
|
subprocess.run([str(fixture),'fixture-sign',str(keys/'test.seed'),str(unsigned),str(manifest)],check=True)
|
|
checks=[]
|
|
def verify(name,path,accept,env=None):
|
|
r=subprocess.run([str(fixture),'ceremony','verify',str(path)],capture_output=True,text=True,env=env)
|
|
assert (r.returncode==0)==accept,(name,r.stdout,r.stderr)
|
|
checks.append(dict(name=name,exit=r.returncode,stdout=r.stdout,stderr=r.stderr))
|
|
verify('genuine-fixture-signature',manifest,True)
|
|
verify('unsigned-refused',unsigned,False)
|
|
tamper=json.loads(manifest.read_text());tamper['qualification']='tampered after signature'
|
|
(source/'tampered.json').write_text(json.dumps(tamper))
|
|
verify('tampered-signature-refused',source/'tampered.json',False)
|
|
subprocess.run([str(fixture),'fixture-sign',str(keys/'wrong.seed'),str(unsigned),str(source/'wrong-key.json')],check=True)
|
|
verify('valid-signature-wrong-root-refused',source/'wrong-key.json',False)
|
|
env=dict(os.environ,ARCHY_RELEASE_ROOT_PUBKEY='00'*32)
|
|
verify('runtime-env-cannot-replace-compiled-test-pin',manifest,True,env)
|
|
production=Path('/opt/os-ota-native')
|
|
if production.exists():
|
|
production.chmod(0o755)
|
|
r=subprocess.run([str(production),'ceremony','verify',str(manifest)],capture_output=True,text=True)
|
|
assert r.returncode!=0 and 'pinned release-root anchor' in r.stderr,r.stderr
|
|
checks.append(dict(name='production-root-refuses-test-signature',exit=r.returncode,stderr=r.stderr))
|
|
binding['os_update']['manifest_sha256']=executor.file_hash(manifest)
|
|
# Planner adapter is retained as evidence, replaced only on this fresh VM by
|
|
# genuine native verification using the isolated test anchor.
|
|
if executor.VERIFIER.exists():
|
|
assert 'ROOT_APPOINTED_EXACT_FIXTURE_ONLY_NOT_PRODUCTION_SIGNATURE' in executor.VERIFIER.read_text()
|
|
executor.VERIFIER.rename(source/'retained-planner-exact-pin-adapter.py')
|
|
import shutil
|
|
shutil.copyfile(fixture,executor.VERIFIER)
|
|
executor.VERIFIER.chmod(0o755)
|
|
binding['os_update']['verifier_sha256']=executor.file_hash(executor.VERIFIER)
|
|
assert binding['os_update']['verifier_sha256']==binding['prerequisite']['backend_sha256']
|
|
assert not (executor.ROOT/'verifier-custody.json').exists(), 'custody must be established by actual provisioner'
|
|
(Path('/root/fixture-verifier-appointment.json')).write_text(json.dumps(dict(
|
|
fixture_only=True,cryptographic_verification=True,production_signature_approval=False,proof=proof)))
|
|
(Path('/var/log/os-ota-stage')/'native-crypto-checks.json').write_text(json.dumps(checks,indent=2)+'\n')
|
|
print('GENUINE_NATIVE_CRYPTO_CHECKS',json.dumps(checks),flush=True)
|
|
|
|
source=PACKAGE/'scripts/tests/os_updater_activation_guest.py'
|
|
raw=source.read_text()
|
|
original_hash=hashlib.sha256(raw.encode()).hexdigest()
|
|
old='python = Path(sys.executable).resolve().read_bytes()'
|
|
assert raw.count(old)==1
|
|
raw=raw.replace(old,"python = Path('/opt/os-ota-crypto-native').read_bytes()")
|
|
old='ExecStart=/usr/local/bin/archipelago /opt/ota-source/requester.py'
|
|
assert old in raw
|
|
# NATIVE=False; replacing the base unit command does not activate the separate
|
|
# accepted-candidate branch of the peer fixture.
|
|
raw=raw.replace(old,'ExecStart=/usr/local/bin/archipelago')
|
|
old='NoNewPrivileges=yes\nProtectSystem=strict'
|
|
assert raw.count(old)==1
|
|
raw=raw.replace(old,'RuntimeDirectory=d09-fixture-client\nRuntimeDirectoryMode=0700\nRuntimeDirectoryPreserve=yes\n'+old)
|
|
start=raw.index('# A tiny native exact-pin adapter in the disposable guest')
|
|
end=raw.index("write(incoming / 'admission.json', wire.canonical(binding))",start)
|
|
raw=raw[:start]+'crypto_appointment(globals())\n'+raw[end:]
|
|
raw=raw.replace('ROOT-APPOINTED-EXACT-FIXTURE-NOT-PRODUCTION-SIGNATURE','NATIVE_CRYPTO_COMPILETIME_TEST_ROOT_NOT_CANDIDATE')
|
|
marker='# A separate pre-existing fixture admin represents the independently verified'
|
|
assert raw.count(marker)==1
|
|
raw=raw.replace(marker,'preservation_probes(globals())\n'+marker)
|
|
marker="assert result['state'] == 'D09_PREREQUISITE_ACTIVATED'"
|
|
assert raw.count(marker)==1
|
|
raw=raw.replace(marker,marker+'\ncapture_owner(globals())')
|
|
context=dict(__name__='isolated_crypto_bootstrap',__file__=str(source),crypto_appointment=crypto_appointment,
|
|
preservation_probes=preservation_probes,capture_owner=capture_owner)
|
|
transformed=Path('/root/os-ota-crypto-transformed-guest.py')
|
|
transformed.write_text(raw)
|
|
Path('/var/log/journal').mkdir(exist_ok=True)
|
|
subprocess.run(['journalctl','--flush'],check=True)
|
|
try:
|
|
exec(compile(raw,str(transformed),'exec'),context)
|
|
finally:
|
|
evidence=Path('/var/log/os-ota-stage')
|
|
evidence.mkdir(exist_ok=True)
|
|
units=['archipelago-d09-lifecycle.service','archipelago.service']
|
|
ledger=Path('/var/lib/archipelago-d09/ledger.json')
|
|
if ledger.exists():
|
|
value=json.loads(ledger.read_text())
|
|
(evidence/'retained-d09-ledger.json').write_bytes(ledger.read_bytes())
|
|
if value.get('execution'):
|
|
units.append(value['execution']['unit'])
|
|
command=['journalctl','--no-pager','-o','short-monotonic']
|
|
for unit in units:command+=['-u',unit]
|
|
r=subprocess.run(command,capture_output=True,text=True)
|
|
(evidence/'crypto-d09-journal.log').write_text(r.stdout+r.stderr)
|
|
r=subprocess.run(['systemctl','show',*units],capture_output=True,text=True)
|
|
(evidence/'crypto-d09-units.log').write_text(r.stdout+r.stderr)
|
|
owner=context['owner']
|
|
deadline=time.monotonic()+120
|
|
while time.monotonic()<deadline:
|
|
value=json.loads((owner.ROOT/'ledger.json').read_text())
|
|
phase=(value.get('receipt') or {}).get('phase')
|
|
if phase in ('AWAITING_OBSERVATION','COMPLETE'):
|
|
break
|
|
if phase=='CONFLICT':
|
|
raise AssertionError(value['receipt'])
|
|
time.sleep(.2)
|
|
else:
|
|
raise TimeoutError('D09 worker did not finish app/config activation after OS apply')
|
|
# Actual namespace/PTY and exact preexisting file observations, then actual
|
|
# root-observer ACK. COMPLETE remains distinct from the pending OS reboot.
|
|
import d09_wire as wire
|
|
import d09_store as store
|
|
receipt=value['receipt']
|
|
pid=owner.service()['MainPID']
|
|
subprocess.run(['nsenter','-t',pid,'-m','--','runuser','-u','archipelago','--','bash','-c',
|
|
'cd /home/archipelago/Work && script -q -c pwd /dev/null'],check=True)
|
|
for directory in ('/home/archipelago/Work','/home/archipelago/.local/state/archipelago/sessions'):
|
|
subprocess.run(['nsenter','-t',pid,'-m','--','runuser','-u','archipelago','--','touch',directory+'/os-ota-observed-write'],check=True)
|
|
for path in ('/usr/local/bin/os-ota-observed-denied','/opt/archipelago/web-ui/os-ota-observed-denied'):
|
|
assert subprocess.run(['nsenter','-t',pid,'-m','--','runuser','-u','archipelago','--','touch',path],capture_output=True).returncode!=0
|
|
after=[path_identity(p) for p in context['crypto_preservation_paths']]
|
|
assert after==context['crypto_preservation_before']
|
|
base=dict(transaction_id=receipt['transaction_id'],observed_invocation_id=receipt['observed_invocation_id'])
|
|
namespace=dict(base,qualification='isolated native requester and real OS PTY; not product terminal/C01 UAT',
|
|
default_terminal_created=True,default_terminal_workspace='/home/archipelago/Work',
|
|
namespace_writable=['/home/archipelago/Work','/home/archipelago/.local/state/archipelago/sessions'])
|
|
preservation=dict(base,preserved={name:dict(before=wire.digest(wire.canonical(before)),after=wire.digest(wire.canonical(now)))
|
|
for name,before,now in zip(('work','history','custom_fragments'),context['crypto_preservation_before'],after)})
|
|
def publish(value):
|
|
raw=wire.canonical(value);hashed=wire.digest(raw)
|
|
store.atomic(owner.ROOT/'observations'/(hashed+'.json'),raw)
|
|
return hashed
|
|
observation={key:receipt[key] for key in ('transaction_id','target_binding_digest','composition_digest',
|
|
'observed_invocation_id','effective_config_observation_digest','config_current_digest')}
|
|
observation.update(namespace_observation_digest=publish(namespace),preservation_observation_digest=publish(preservation))
|
|
binding=context['binding']
|
|
request=dict(schema_version=1,operation='ACK_OBSERVATION',route='OBSERVER',reason='OBSERVATION_READY',
|
|
transaction_id=receipt['transaction_id'],request_id=os.urandom(16).hex(),target_binding_digest=receipt['target_binding_digest'],
|
|
composition_digest=receipt['composition_digest'],grant_digest=binding['grants']['OBSERVER'],
|
|
expected_sequence=value['sequence'],requester_invocation_id=None,observer_receipt_digest=publish(observation),
|
|
selected_release_digest=None,staging_generation=None)
|
|
def exchange():
|
|
with socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as conn:
|
|
conn.settimeout(150);conn.connect(str(owner.SOCKET))
|
|
assert struct.unpack('3i',conn.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12))[1]==0
|
|
conn.sendall(wire.canonical(request));conn.shutdown(socket.SHUT_WR)
|
|
raw=b''
|
|
while part:=conn.recv(8192):raw+=part
|
|
return wire.parse(raw)
|
|
assert exchange()['status']=='ACCEPTED'
|
|
deadline=time.monotonic()+90
|
|
while time.monotonic()<deadline:
|
|
value=json.loads((owner.ROOT/'ledger.json').read_text())
|
|
if value['receipt']['phase']=='COMPLETE':break
|
|
assert value['receipt']['phase']!='CONFLICT',value['receipt']
|
|
time.sleep(.2)
|
|
else:raise TimeoutError('same-boot userspace COMPLETE')
|
|
assert value['receipt']['outcome']=='COMPLETE'
|
|
deadline=time.monotonic()+30
|
|
while time.monotonic()<deadline:
|
|
active=subprocess.check_output(['systemctl','show',value['execution']['unit'],'-p','ActiveState','--value'],text=True).strip()
|
|
if active in ('inactive','failed'):break
|
|
time.sleep(.1)
|
|
else:raise TimeoutError('completed observation worker drain')
|
|
head=(owner.ROOT/'ledger.json').read_bytes()
|
|
assert exchange()['status']=='ACCEPTED'
|
|
assert (owner.ROOT/'ledger.json').read_bytes()==head
|
|
phase='COMPLETE'
|
|
owner_after=owner.process(context['crypto_owner_before']['pid'])
|
|
assert owner_after==context['crypto_owner_before']
|
|
assert value['epoch']==context['crypto_epoch_before']
|
|
assert os.stat('/var/lib/archipelago').st_uid==1000
|
|
assert executor.ROOT.stat().st_uid==0 and executor.ROOT.stat().st_mode & 0o022==0
|
|
subprocess.run(['runuser','-u','archipelago','--','touch','/var/lib/archipelago/os-ota-custody-positive-probe'],check=True)
|
|
denials=[]
|
|
for args in (['touch',str(executor.ROOT/'app-write-denied')],
|
|
['python3','-c',"import os;os.rename('/var/lib/archipelago-os-update','/var/lib/app-rename-denied')"],
|
|
['python3','-c',"open('/root/os-ota-crypto-fixture/test.seed','rb').read()"]):
|
|
r=subprocess.run(['runuser','-u','archipelago','--',*args],capture_output=True,text=True)
|
|
assert r.returncode!=0,args
|
|
denials.append(dict(command=args,exit=r.returncode))
|
|
evidence=Path('/var/log/os-ota-stage')
|
|
before=executor.inventory()
|
|
binding=context['binding']
|
|
status=executor.Executor(binding['os_update']).status
|
|
assert status['phase']=='REBOOT_REQUIRED'
|
|
assert executor.stage(binding,'APPLY','apply')==status
|
|
assert executor.inventory()==before
|
|
wallet=Path('/var/lib/archipelago/ota-synthetic-canaries/wallet.db')
|
|
with wallet.open('a') as f:f.write('FORWARD-AFTER-CRYPTO-D09-sequence-18\n')
|
|
forward={path:executor.file_hash(Path(path)) for path in context['preserved']}
|
|
result=dict(passed=True,production_approved=False,candidate_artifact_qualified=False,
|
|
genuine_native_crypto=True,fixture_proof=proof,bootstrap_fixture_source_sha256=original_hash,
|
|
transformed_fixture_sha256=executor.file_hash(transformed),
|
|
production_helpers_unmodified=True,os_root=str(executor.ROOT),app_data_uid=1000,
|
|
denied_app_mutations=denials,d09_phase=phase,os_phase=status['phase'],
|
|
same_boot_userspace_complete=True,observer_ack_idempotent=True,owner_lifetime_before=context['crypto_owner_before'],
|
|
owner_lifetime_after=owner_after,manager_events=value.get('manager_events',[]),
|
|
signed_manager_reexec=context['context']['plan']['manager_reexec'],
|
|
workspace_preservation=context['crypto_preservation_before'],
|
|
root_filesystem_uuid=context['crypto_root_uuid'],in_place_package_route=True,
|
|
target_kernel=context['context']['plan']['boot']['expected_release'],forward_preserved=forward)
|
|
import importlib.util
|
|
provision=context['provision']
|
|
installed={name:executor.file_hash(Path('/usr/libexec')/name) for name in provision.MODULES}
|
|
for name,hashed in installed.items():assert hashed==executor.file_hash(PACKAGE/'scripts'/name),name
|
|
result['installed_module_sha256']=installed
|
|
result['old_boot_retained']=all(executor.file_hash(path)==sha for path,sha in context['context']['plan']['boot']['retained_files'].items() if path!='/boot/grub/grub.cfg')
|
|
assert result['old_boot_retained']
|
|
(evidence/'same-boot-complete-ledger.json').write_bytes(head)
|
|
(evidence/'same-boot-complete-observation.json').write_bytes(wire.canonical(observation))
|
|
(evidence/'crypto-result.json').write_text(json.dumps(result,indent=2)+'\n')
|
|
print('CRYPTO_BOOTSTRAP_OS_PASS',json.dumps(result),flush=True)
|