Files

63 lines
2.8 KiB
Python

#!/usr/bin/env python3
"""Real C01 Children FD sweep -> credential drop -> native FD14 remap.
No hardware and no storage keys. The disposable native probe checks the channel
and absence of authority FD13; full native protocol checks run separately.
"""
import os
from pathlib import Path
import socket
import subprocess
import sys
import tempfile
import time
assert os.geteuid()==0 and os.environ.get('ARCHY_RADIO_FIXTURE')=='1'
repo=Path(__file__).resolve().parents[2];sys.path.insert(0,str(repo/'scripts'))
import c01_children
import c01_radio
import c01_radio_owner
import c01_supervisor
with tempfile.TemporaryDirectory() as temporary:
root=Path(temporary);root.chmod(0o755)
binary=root/'probe'
subprocess.run(['/usr/bin/cc',str(repo/'scripts/tests/radio_handoff_fixture.c'),'-o',str(binary)],check=True)
executable=os.open(binary,os.O_RDONLY|os.O_CLOEXEC)
address=str(root/'radio.sock')
peer=os.fork()
if peer==0:
listener=socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET)
listener.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1)
listener.bind(address);listener.listen(1)
channel,_=listener.accept();channel.settimeout(5)
import struct
data,ancillary,_,_=channel.recvmsg(16,socket.CMSG_SPACE(12))
credentials=[struct.unpack('3i',value) for level,kind,value in ancillary
if (level,kind)==(socket.SOL_SOCKET,socket.SCM_CREDENTIALS)]
assert data==b'FD14' and len(credentials)==1 and credentials[0][1:]==(53001,53001)
channel.send(b'OK');os._exit(0)
end=time.monotonic()+5
while not Path(address).exists():
assert time.monotonic()<end;time.sleep(.01)
channel=socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET);channel.connect(address)
children=c01_children.Children()
parent=os.getpid()
def body(control):
c01_supervisor.close_except({control.fileno(),executable,channel.fileno()})
c01_children.drop_credentials(53001,53001,parent)
c01_supervisor.native(executable,'c01-managed-v7',{14:channel.fileno()},{'PATH':'/nonexistent'})
try:
assert c01_radio.application_fds((executable,),None)==(executable,), 'v6 ABI retention changed'
children.spawn('application',body,c01_radio.application_fds((executable,),channel))
channel.close()
while children.children:
assert not children.poll()
assert time.monotonic()<end;time.sleep(.01)
assert children.results['application']==0,children.results
assert os.waitstatus_to_exitcode(os.waitpid(peer,0)[1])==0
children.close()
print('PASS actual Children pre-callback sweep retains only appointed radio capability; credential drop/native mapping supplies FD14, not FD13',flush=True)
finally:
os.close(executable)