166 lines
9.4 KiB
Python
166 lines
9.4 KiB
Python
#!/usr/bin/python3
|
|
"""Normal caller/Bridge/authority-loop integration under the isolated runner.
|
|
|
|
The root coordinator is the production clock_refresh function. NTS measurement,
|
|
collector namespace transitions and outer cgroup management are fixture seams; peer credentials,
|
|
processes, issuer handoff, owner queue, storage, expiry and control protocol are real.
|
|
This is clock mechanism qualification, not actual namespace isolation coverage.
|
|
"""
|
|
import json,os,select,signal,socket,subprocess,sys,time
|
|
from dataclasses import replace
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[2]/'scripts'))
|
|
import c01_supervisor as s
|
|
import c01_nts as nts
|
|
import c01_recovery as recovery
|
|
import c01_children as lifecycle
|
|
|
|
assert os.geteuid()==0 and os.environ.get('ARCHY_TEST_ISOLATED')=='1'
|
|
binary,case=sys.argv[1:];cfg=json.loads(Path(case).read_text());root=Path(cfg['root'])
|
|
APP=61021;AUTH=61022
|
|
def drop(uid):
|
|
os.setgroups([]);os.setgid(uid);os.setuid(uid)
|
|
def native(role,uid,channel):
|
|
name='c01_launcher::clock_refresh::tests::normal_refresh_'+role+'_process'
|
|
env={**os.environ,'C01_NORMAL_REFRESH_CASE':case,'C01_NORMAL_REFRESH_FD':str(channel.fileno())}
|
|
return subprocess.Popen([binary,'--exact',name,'--ignored','--test-threads=1','--nocapture'],
|
|
env=env,pass_fds=(channel.fileno(),),preexec_fn=lambda:drop(uid),stdout=subprocess.DEVNULL)
|
|
|
|
def synthetic_measurement(policy,raw=False):
|
|
# Explicit synthetic issuer input. No live NTS claim or local-clock fallback
|
|
# exists in production: this replacement is confined to this isolated driver.
|
|
begin=nts.clock_reading(raw=raw);end=nts.clock_reading(raw=raw)
|
|
assessment=nts.Assessment('01020304','127.0.0.1',Decimal(end.wall)/nts.NANO,Decimal(0),1_000_000,(),())
|
|
return nts.bind_observation(assessment,begin,end,recovery.boot_id(),recovery.boot_id(),policy)
|
|
nts.collect=synthetic_measurement
|
|
|
|
def fixture_collector_namespace():
|
|
# The backend runner deliberately drops SYS_ADMIN. Do not expand its host
|
|
# capabilities for a synthetic NTS test. Install this seam ONLY in the
|
|
# forked fixture collector, never in production or the parent coordinator.
|
|
network=s.c01_network;libc=s.LIBC
|
|
expected=network.process_namespace(os.getpid())
|
|
def same_namespace(pid,role):
|
|
assert role=='managed' and pid==os.getpid()
|
|
assert network.process_namespace(pid)==expected
|
|
return expected
|
|
def enter_host_worker(actual_libc):
|
|
assert actual_libc is s.LIBC
|
|
same_namespace(os.getpid(),'managed')
|
|
class CollectorLibc:
|
|
def __getattr__(self,name):return getattr(libc,name)
|
|
def setns(self,fd,kind):
|
|
assert kind==network.CLONE_NEWNET and network.identity(fd)==expected
|
|
same_namespace(os.getpid(),'managed')
|
|
return 0
|
|
s.LIBC=CollectorLibc()
|
|
s.c01_network=SimpleNamespace(identity=network.identity,CLONE_NEWNET=network.CLONE_NEWNET,
|
|
enter_host_worker=enter_host_worker,require_private=same_namespace)
|
|
|
|
class FixtureChildren:
|
|
def __init__(self):self.children={};self.results={};self.reserved={};self.stopping=False;self.spawned=0
|
|
def spawn(self,role,body,keep=()):
|
|
assert role=='collector' and role not in self.children and role not in self.results
|
|
parent,child=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET)
|
|
parent.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1)
|
|
pid=os.fork()
|
|
if pid==0:
|
|
try:
|
|
parent.close();s.close_except({child.fileno()})
|
|
fixture_collector_namespace();body(child);os._exit(0)
|
|
except BaseException:
|
|
import traceback;traceback.print_exc();os._exit(111)
|
|
child.close();record=SimpleNamespace(role=role,pid=pid,channel=parent,pidfd=os.pidfd_open(pid),process=None)
|
|
self.children[role]=record;self.spawned+=1;return record
|
|
def poll(self):
|
|
record=self.children.get('collector')
|
|
if record:
|
|
pid,status=os.waitpid(record.pid,os.WNOHANG)
|
|
if pid:
|
|
self.results['collector']=os.waitstatus_to_exitcode(status)
|
|
record.channel.close();os.close(record.pidfd);del self.children['collector']
|
|
rearm_collector=lifecycle.Children.rearm_collector
|
|
|
|
class Owner:
|
|
receive=s.Supervisor.receive;wait=s.Supervisor.wait;relay_collector=s.Supervisor.relay_collector
|
|
def __init__(self):
|
|
self.children=FixtureChildren();self.clock_sequence=0;self.power_state=None
|
|
self.completed_one_shots={'collector'};self.wake_read,self.wake_write=os.pipe2(os.O_NONBLOCK|os.O_CLOEXEC)
|
|
def tick(self):
|
|
self.children.poll()
|
|
assert self.children.results.get('collector',0)==0,'collector failed'
|
|
return True
|
|
|
|
owner=Owner();records=[]
|
|
try:
|
|
parent,child=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET);parent.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1)
|
|
authority_process=native('authority',AUTH,child);child.close()
|
|
authority=SimpleNamespace(role='authority',pid=authority_process.pid,channel=parent,process=authority_process)
|
|
owner.children.children['authority']=authority;records.append(authority)
|
|
end=s.now()+120*10**9;owner.wait(authority,b'LISTENING',AUTH,AUTH,end)
|
|
parent,child=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET);parent.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1)
|
|
app_process=native('application',APP,child);child.close()
|
|
app=SimpleNamespace(role='application',pid=app_process.pid,channel=parent,process=app_process)
|
|
owner.children.children['application']=app;records.append(app)
|
|
app_peer=dict(pid=app.pid,uid=APP,gid=APP);authority_peer=dict(pid=authority.pid,uid=AUTH,gid=AUTH)
|
|
s.packet(authority.channel,app_peer);s.packet(app.channel,authority_peer)
|
|
owner.wait(app,b'CONNECTED',APP,APP,end)
|
|
context=json.loads(owner.receive(authority,AUTH,AUTH,end));assert context['status']=='EPOCH_CONTEXT'
|
|
maximum=cfg['delegation']['max_awake'];maximum=maximum['secs']*10**9+maximum['nanos']
|
|
q=recovery.PlatformQualification(os.uname().release,recovery.clocksource(),'ab'*32,10**9,250_000_000,0,maximum,cfg['delegation']['characterized'])
|
|
prepared=SimpleNamespace(epoch=dict(delegation=cfg['delegation']),value=dict(binding=cfg['binding'],clock_domain='01'*16))
|
|
collector=s.start_collector(owner,q,cfg['binding'],bytes([1])*16,authority_peer,'\0'+cfg['address'],end,prepared.epoch,context['challenge'])
|
|
owner.relay_collector(collector,authority,AUTH,AUTH,end)
|
|
owner.wait(authority,b'AUTHORIZED',AUTH,AUTH,end);owner.wait(collector,b'ADMIN_SENT',0,0,end);s.packet(collector.channel,b'RELEASE')
|
|
s.packet(app.channel,b'INITIAL_AUTHORIZED')
|
|
ready=False;failed=False;deadline=time.monotonic()+25
|
|
result_file=root/'data/refresh-result.json'
|
|
while time.monotonic()<deadline:
|
|
owner.tick()
|
|
if result_file.exists() and app_process.poll() is not None:break
|
|
if select.select([app.channel],[],[],.05)[0]:
|
|
# Exit can race the poll above. EOF is not an authenticated status
|
|
# packet; accept it only with the completed fixture result and a
|
|
# successful child exit (checked below).
|
|
if not app.channel.recv(1,socket.MSG_PEEK):
|
|
app_process.wait(timeout=5)
|
|
assert result_file.exists(),'application exited without a fixture result'
|
|
break
|
|
packet=owner.receive(app,APP,APP,s.now()+120*10**9)
|
|
if packet==b'FAILED':failed=True;break
|
|
msg=json.loads(packet)
|
|
if msg['status']=='APP_READY':ready=True
|
|
elif msg['status']=='CLOCK_REFRESH':
|
|
s.clock_refresh(owner,prepared,q,app,authority,app_peer,authority_peer,'\0'+cfg['address'],msg)
|
|
else:raise AssertionError('unexpected normal app status '+repr(msg))
|
|
if result_file.exists() and app_process.poll() is not None:break
|
|
assert ready
|
|
app_process.wait(timeout=5)
|
|
assert app_process.returncode==0
|
|
result=json.loads(result_file.read_text())
|
|
if cfg['uncertain']:
|
|
assert failed and owner.clock_sequence==0 and owner.children.spawned==1 and result['uncertain_effect']
|
|
else:
|
|
assert not failed and owner.clock_sequence==2 and owner.children.spawned==3
|
|
assert result['pid']==app.pid and len(result['domains'])==3 and result['old_consumed_and_expired_refused'] and result['new_request_succeeded']
|
|
print(json.dumps(dict(uncertain=cfg['uncertain'],application_pid=app.pid,authority_pid=authority.pid,
|
|
refreshes=owner.clock_sequence,collectors=owner.children.spawned,normal_caller=True,normal_authority_loop=True,
|
|
application_or_container_restart=False,qualification='clock-mechanism-only',
|
|
namespace_transition='fixture-seam-no-setns',actual_namespace_isolation=False,
|
|
source_root=str(Path(__file__).resolve().parents[2]),result=result)))
|
|
finally:
|
|
for record in records:
|
|
if record.process.poll() is None:record.process.terminate()
|
|
for record in records:
|
|
try:record.process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:record.process.kill();record.process.wait()
|
|
record.channel.close()
|
|
collector=owner.children.children.get('collector')
|
|
if collector:
|
|
try:signal.pidfd_send_signal(collector.pidfd,signal.SIGKILL)
|
|
except ProcessLookupError:pass
|
|
os.waitpid(collector.pid,0);collector.channel.close();os.close(collector.pidfd)
|
|
os.close(owner.wake_read);os.close(owner.wake_write)
|