97 lines
5.4 KiB
Python
97 lines
5.4 KiB
Python
#!/usr/bin/python3
|
|
"""Actual PID1/epoch/caller retirement component; never invokes kernel sleep."""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
assert os.geteuid()==0 and Path('/run/.containerenv').exists()
|
|
assert socket.gethostname()=='c01-prerequisite-qualification' and os.environ.get('C01_POWER_COMPONENT')=='1'
|
|
sys.path.insert(0,'/opt/c01-fixture/scripts')
|
|
import c01_power
|
|
|
|
OUT=Path('/var/lib/c01-normal')
|
|
def run(*args,check=True):
|
|
p=subprocess.run(args,capture_output=True,text=True,timeout=90)
|
|
if check and p.returncode:raise AssertionError((args,p.returncode,p.stdout,p.stderr))
|
|
return p
|
|
def record(name,value):(OUT/name).write_text(json.dumps(value,indent=2)+'\n')
|
|
def wait(body):
|
|
for _ in range(300):
|
|
if body():return
|
|
time.sleep(.1)
|
|
raise AssertionError('power component deadline')
|
|
|
|
run('/usr/libexec/archipelago-d09-provision','--activate-prerequisite')
|
|
wait(lambda:'C01_NON_ADMITTING_EPOCH_LIFETIME_HELD' in run('journalctl','-u','archipelago.service','--no-pager').stdout)
|
|
before=c01_power.show('archipelago.service')
|
|
assert int(before['MainPID'])>1
|
|
before_epoch=Path('/var/lib/c01-composed/private-authority/epoch/anchor').read_bytes()
|
|
work=Path('/var/lib/c01-fixture/home/Work/operator.txt')
|
|
meta=work.stat();content=work.read_bytes()
|
|
initial=c01_power.read(c01_power.STATE)
|
|
assert run('systemctl','start','archipelago-c01-power-barrier@systemd-hibernate.service',check=False).returncode!=0
|
|
assert c01_power.read(c01_power.STATE)==initial, 'unsupported hibernation changed power authority'
|
|
|
|
# Actual polkit answers for a real untrusted process; never invoke suspend to
|
|
# test denial on a container sharing the host kernel.
|
|
child=subprocess.Popen(['/usr/bin/setpriv','--reuid=65534','--regid=65534','--clear-groups','/usr/bin/sleep','60'])
|
|
try:
|
|
wait(lambda:'65534' in Path(f'/proc/{child.pid}/status').read_text().split('Uid:',1)[1].splitlines()[0])
|
|
identity=c01_power.identity(child.pid)
|
|
answers={}
|
|
for action in ('org.freedesktop.login1.suspend','org.freedesktop.login1.suspend-ignore-inhibit',
|
|
'org.freedesktop.login1.hibernate','org.freedesktop.systemd1.manage-units',
|
|
'org.freedesktop.policykit.exec','org.freedesktop.timedate1.set-time'):
|
|
reply=run('busctl','call','org.freedesktop.PolicyKit1','/org/freedesktop/PolicyKit1/Authority',
|
|
'org.freedesktop.PolicyKit1.Authority','CheckAuthorization','(sa{sv})sa{ss}us',
|
|
'unix-process','3','pid','u',str(child.pid),'start-time','t',str(identity['start']),
|
|
'uid','i','65534',action,'0','0','').stdout.strip()
|
|
assert 'false false' in reply, (action,reply)
|
|
answers[action]=reply
|
|
record('power-untrusted-denials.json',answers)
|
|
finally:
|
|
child.terminate();child.wait(timeout=5)
|
|
|
|
barrier='archipelago-c01-power-barrier@systemd-suspend.service'
|
|
run('systemctl','start',barrier)
|
|
state=json.loads((c01_power.ROOT/'retirement-1.json').read_bytes())
|
|
assert state['phase']=='SleepPermitted' and state['supervisor']['pid']==int(before['MainPID'])
|
|
assert Path('/var/lib/c01-composed/private-authority/epoch/anchor').read_bytes()!=before_epoch
|
|
assert work.stat().st_ino==meta.st_ino and work.read_bytes()==content
|
|
record('power-prepared.json',state)
|
|
|
|
# The post-boundary observer automatically treats this unexecuted kernel action
|
|
# as an abort, issues ONE fresh scoped D09 activation and starts a new data-only
|
|
# lifetime. It must never claim APP_READY for this deliberately non-admitting role.
|
|
wait(lambda:(c01_power.ROOT/'reactivation-2.submitted').exists())
|
|
wait(lambda:int(c01_power.show('archipelago.service')['MainPID']) not in (0,int(before['MainPID'])))
|
|
after=c01_power.read(c01_power.STATE)
|
|
assert after['phase']=='RecoveryRequired' and after['sequence']==state['sequence']+1
|
|
assert after['barrier_boottime_ns']>state['barrier_boottime_ns']
|
|
ledger=json.loads(Path('/var/lib/archipelago-d09/ledger.json').read_bytes())
|
|
assert len(ledger['grants'])==3 and len(set(ledger['grants']))==3
|
|
assert len(ledger['readmissions'])==1 and ledger['prerequisite']['state']=='START_INTENT'
|
|
assert not (ledger['prerequisite'].get('observation'))
|
|
record('automatic-power-reactivation.json',ledger)
|
|
record('power-return.json',after)
|
|
run('systemctl','stop','archipelago.service')
|
|
|
|
# Real PID1 dependency failure must not run even this harmless sentinel: the
|
|
# manager is now deliberately stopped and cannot attest epoch retirement.
|
|
sentinel=OUT/'power-effect-would-have-run'
|
|
Path('/run/systemd/system/c01-power-effect-test.service').write_text(
|
|
f'[Unit]\nRequires={barrier}\nAfter={barrier}\n[Service]\nType=oneshot\nExecStart=/usr/bin/touch {sentinel}\n')
|
|
run('systemctl','daemon-reload')
|
|
assert run('systemctl','start','c01-power-effect-test.service',check=False).returncode!=0
|
|
assert not sentinel.exists() and c01_power.read(c01_power.STATE)==after
|
|
record('result.json',{'exit':0,'native':True,'epoch_revocation':True,'caller_hierarchies_retired':True,
|
|
'dependency_failure_blocked_effect':True,'automatic_time_policy':'fresh delegated NTS only',
|
|
'kernel_sleep_executed':False,'application_admitted':False,'awake_bound_qualified':False,
|
|
'automatic_d09_reactivation':True,'remaining':'qualified awake bound and real kernel power-route acceptance'})
|
|
print('PASS native epoch revocation, actual caller retirement, untrusted power denial, mandatory PID1 dependency and abort barrier',flush=True)
|