Experiment formal browser session fixture registration red plan 01 - kmosoti/FabricO11y GitHub Wiki
Explicit browser session states and cleanup, named red preflight01
Prospective: NOT RUN. C-069 admits AB/C state/lifecycle registration, implementation and source controls, not actual browser acceptance. Clean trust-only b1e7e83c74ba029c3b377ee047220db93853d578; production unchanged. Trust06241b2 registered literals; b1e7e83 preserves failed-control identity/process/marker evidence after independently confirmed cleanup.
Expected10 methods: four explicit two-state checker availability reds, one positive empty-finalizer control, five actual-finalizer cleanup counterexamples. Availability reds establish missing mechanism only. The actual main finalizer AST is executed with real tiny synthetic processes; no main setup, browser or server starts. Owned fork children are PIDFD/start-tick/PGID bound with45-second finite lifespan. Test-only wait shortens timeout to0.2s while checking original production15s/5s arguments; production timeouts remain unchanged.
Independent test cleanup uses subreaper, PIDFD SIGKILL and waitpid to stop/reap only bound owned processes, confirms private groups absent and sampler stopped, then restores prior subreaper setting. Exclusive<=32KiB cleanup-evidence.json records actual stop/identity/marker/SHA observations. Failed/uncertain bodies retain directories; positive confirmed controls remove scratch. Work already deleted by the flawed finalizer is recorded without reconstruction; stderr names retained paths.
Fixed contract
AB shares one fresh packaged-server state/PID; C gets a distinct exclusively created empty state/path/dev-inode/PID. Both bind identical candidate/package/executable/helper, actual process mapping/start/cgroup and empty-before-start witnesses. All32 original session observations/status/boundaries/hashes remain unchanged. AB groups must be confirmed empty before C starts; actual mode has no legacy integer-PID fallback or row rewriting. One global240-second deadline and550ms pacing cover setup, both states and stop. The later native adapter must bind actual unshifted monotonic time to the same REALTIME shim invocation PID/TID/sequence; controller receive time or wall subtraction is invalid. Prototypes/return semantics,<=28860 offset,256-byte record and1MiB ledger remain fixed. No native adapter is admitted here.
Pins
/run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-browser-fixture-red-01.pySHA2561053995be818f5814da3cddb67d776166423ada925efabf036b006c5218c2201./run/media/kmosoti/data/FabricO11y/scratch/console-access-acceptance/docs/experiments/formal/console-session-clock-protocol.mdSHA2566ce93cf267e16bd100c1132ec486f8597394c31c6f4b376d5884ca1788f5844e./run/media/kmosoti/data/FabricO11y/scratch/console-access-acceptance/tools/ui/test_session_clock.pySHA256440d73f6c3dded5f4ec6a6917b8ea641f9258f9917c8cad23095abe2bf85c88b./run/media/kmosoti/data/FabricO11y/scratch/console-access-acceptance/tools/ui/test_role_matrix.pySHA256e068f5669cf21cb2d2411e94689bb28518c8d8f8b08ba87eafaea97c51301432./run/media/kmosoti/data/FabricO11y/scratch/console-access-acceptance/tools/ui/session_clock_oracle.pySHA2564817a0d09c1c920a67242637cbbb3a6473d6bea42ce716d07a73e8d7122d1728./run/media/kmosoti/data/FabricO11y/scratch/console-access-acceptance/tools/ui/live_browser_check.pySHA256fa30e93a886f4f7224e2ab2c0bb770da153ab9ad18e426baf30ec341a4770e1a./run/media/kmosoti/data/FabricO11y/scratch/console-access-acceptance/tools/ui/session_clock.pySHA256f183ce8caf2f73fe16e11f9de3342b4368a5990a17964e75861cecb4c4e68a34./run/media/kmosoti/data/FabricO11y/scratch/console-access-acceptance/tools/ui/session_clock.cSHA25683b4b3d17044c98a84085050cf749a2df270d2298d626d0c4e1acd916cb04149.
Exact contained source cell
python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-browser-fixture-red-01.py browser-session-fixture-registration-red-01 1073741824 env PYTHONPATH=tools/ui python3 -B -m unittest -v test_session_clock.SessionFixtureBindingTests test_role_matrix.BrowserFixtureCleanupTests
Root alone executes serially with fresh strict95GB DATA census and full1GiB reservation, default16GiB high/20GiB max/no swap/1800s for the descendant tree. No browser, package launch, VM, Rust/C compilation. Imports launch only the stated synthetic control processes. After completion root reads #59, posts real methods/subcases/availability/setup classification and stopped receipts, retains failures, and confirms no live control groups before any production correction. Unexpected positive/setup/import/independent-cleanup failures stop/report this lane.
Literal session controls
"""Independent literal oracle controls and separate fixture lifecycle regressions."""
import copy
from contextlib import contextmanager
import ctypes
import hashlib
import inspect
import os
from pathlib import Path
import signal
import subprocess
import sys
import tempfile
import threading
from time import monotonic as real_monotonic, sleep
import unittest
from unittest.mock import Mock, patch
import session_clock_oracle as oracle
ROOT=Path(__file__).resolve().parents[2]
def clock_rows():
rows=[]
for sequence,offset in enumerate((0,1,1799,1800,28799,28800)):
for clock in (0,1,4):
raw=100_500_000_000 if clock==0 else 200_500_000_000
rows.append({'pid':700,'sequence':sequence,'offset_s':offset,'clock_id':clock,
'raw_ns':raw,'returned_ns':raw+(offset*1_000_000_000 if clock==0 else 0),
'syscall_before_ns':raw-1_000_000,'syscall_after_ns':raw+1_000_000})
return rows
def isolation():
ordinary={role:{'pid':pid,'helper_maps':[],'control_env_names':[],'control_fd_present':False,
'syscall_before_ns':100_000_000_000,'clock_ns':100_001_000_000,'syscall_after_ns':100_002_000_000}
for pid,role in enumerate(('controller','browser','driver','companion'),701)}
return {'target_pid':700,'target_executable_sha256':'a'*64,'target_helper_maps':['b'*64],
'target_control_cloexec':True,'target_control_type':'AF_UNIX/SOCK_SEQPACKET','ordinary_processes':ordinary,
'fork':{'pid':705,'target_offset_s':1,'raw_ns':100_000_000_000,'returned_ns':100_000_000_000,
'control_sequence_before':1,'control_sequence_after':1,'exec_helper_maps':[],
'exec_control_env_names':[],'exec_control_fd_present':False}}
def sessions():
metadata={'A':{'hash':'a'*64,'issued_s':1000,'expires_s':29800,'baseline_seen_s':1000},
'B':{'hash':'b'*64,'issued_s':1000,'expires_s':29800,'baseline_seen_s':1000},
'C':{'hash':'c'*64,'issued_s':10000,'expires_s':38800,'baseline_seen_s':10000}}
rows=[]
def add(cell,key,second,status,last_seen):
anchor=metadata[key]
rows.append({'cell':cell,'session_hash':anchor['hash'],'pid':700,'issued_s':anchor['issued_s'],
'expires_s':anchor['expires_s'],'last_seen_s':last_seen,'request_s':second,
'server_call_s':second,'status':status,'error':'sign in required' if status==401 else None})
add('idle_before','A',2799,200,1000);add('idle_at','B',2800,401,1000)
previous=10000
for second in range(11000,38001,1000):add('absolute_keepalive','C',second,200,previous);previous=second
add('absolute_before','C',38799,200,38000);add('absolute_at','C',38800,401,38799)
return rows,metadata
class SessionClockTests(unittest.TestCase):
def test_fixture_mechanism_sources_are_available_for_native_binding_admission(self):
# This initial red is mechanism absence, never a reproduced product bug.
for name in ('session_clock.py','session_clock.c','session_clock_probe.rs'):
self.assertTrue((ROOT/'tools/ui'/name).is_file(), 'missing fixture mechanism: '+name)
def test_literal_frames_and_bounds_reject_bad_wire_records(self):
frames=[bytes.fromhex(line) for line in (
'46434c4b7631000000000000000000000000000000000000',
'46434c4b7631000001000000000000000100000000000000',
'46434c4b7631000002000000000000000707000000000000',
'46434c4b7631000003000000000000000807000000000000',
'46434c4b7631000004000000000000007f70000000000000',
'46434c4b7631000005000000000000008070000000000000')]
oracle.check_frames(frames)
maximum=[b'FCLKv1\0\0'+i.to_bytes(8,'little')+i.to_bytes(8,'little') for i in range(64)]
oracle.check_frames(maximum)
for bad,error in [([], 'frame-count'),(maximum+[maximum[-1]],'frame-count'),
([frames[0][:-1]],'frame-length'),([frames[0]+b'x'],'frame-length'),
([b'wrong-v1'+frames[0][8:]],'frame-version'),
([frames[0][:16]+(1).to_bytes(8,'little')],'frame-offset'),
([frames[0],frames[0]],'frame-sequence'),([frames[0],frames[2]],'frame-sequence'),
([frames[0],frames[1],frames[2][:16]+bytes(8)],'frame-offset'),
([frames[0],frames[1][:16]+(28861).to_bytes(8,'little')],'frame-offset'),
([frames[0],frames[1][:16]+(2**64-1).to_bytes(8,'little')],'frame-offset')]:
with self.subTest(error=error),self.assertRaisesRegex(ValueError,error):oracle.check_frames(bad)
def test_independent_raw_clock_and_sleep_controls_reject_shifted_or_unbound_evidence(self):
oracle.check_probe_clocks(clock_rows(),700)
for index,field,value,error in [(3,'returned_ns',100_500_000_000,'clock-realtime-only'),
(1,'returned_ns',201_500_000_000,'clock-realtime-only'),
(3,'returned_ns',101_500_000_001,'clock-realtime-only'),
(3,'pid',701,'clock-pid'),(3,'sequence',0,'clock-binding'),
(0,'syscall_after_ns',102_000_000_000,'clock-bracket-inconclusive'),
(0,'raw_ns',100_000_000_000,'clock-raw-bracket')]:
changed=clock_rows();changed[index][field]=value
with self.subTest(field=field),self.assertRaisesRegex(ValueError,error):oracle.check_probe_clocks(changed,700)
with self.assertRaisesRegex(ValueError,'clock-completeness'):oracle.check_probe_clocks(clock_rows()[:-1],700)
sleep={'start_before_ns':200_000_000_000,'start_after_ns':200_001_000_000,
'end_before_ns':200_502_000_000,'end_after_ns':200_503_000_000,'instant_elapsed_ns':502_000_000}
oracle.check_sleep(sleep)
with self.assertRaisesRegex(ValueError,'sleep-monotonic'):oracle.check_sleep(dict(sleep,instant_elapsed_ns=1_502_000_000))
def test_target_maps_fork_and_child_escape_controls_are_independent(self):
def check(row):oracle.check_isolation(row,700,'b'*64,'a'*64)
check(isolation())
for field,value,error in [('target_pid',701,'isolation-target'),('target_helper_maps',[],'isolation-interception'),
('target_helper_maps',['c'*64],'isolation-interception'),('target_control_cloexec',False,'isolation-control'),
('target_executable_sha256','d'*64,'isolation-target')]:
changed=isolation();changed[field]=value
with self.subTest(field=field),self.assertRaisesRegex(ValueError,error):check(changed)
for role in ('controller','browser','driver','companion'):
for field,value in [('helper_maps',['b'*64]),('control_env_names',['FABRIC_CLOCK_FD']),('control_fd_present',True)]:
changed=isolation();changed['ordinary_processes'][role][field]=value
with self.subTest(role=role,field=field),self.assertRaisesRegex(ValueError,'isolation-escape'):check(changed)
changed=isolation();changed['ordinary_processes'][role]['clock_ns']+=1_000_000_000
with self.subTest(role=role),self.assertRaisesRegex(ValueError,'isolation-host-clock'):check(changed)
for field,value,error in [('returned_ns',101_000_000_000,'isolation-fork-consumption'),
('control_sequence_after',2,'isolation-fork-consumption'),
('exec_helper_maps',['b'*64],'isolation-exec-escape'),
('exec_control_env_names',['LD_PRELOAD'],'isolation-exec-escape'),
('exec_control_env_names',['FABRIC_CLOCK_FD'],'isolation-exec-escape'),
('exec_control_fd_present',True,'isolation-exec-escape')]:
changed=isolation();changed['fork'][field]=value
with self.subTest(field=field),self.assertRaisesRegex(ValueError,error):check(changed)
def test_literal_session_boundaries_reject_wrong_refusal_baseline_or_token(self):
rows,metadata=sessions();oracle.check_sessions(rows,metadata,700)
for index,field,value,error in [(1,'status',200,'session-refusal'),(1,'status',429,'session-refusal'),
(1,'status',403,'session-refusal'),(1,'status',None,'session-refusal'),
(1,'last_seen_s',1001,'session-baseline'),(1,'request_s',2801,'session-boundary'),
(-1,'server_call_s',38801,'session-boundary-inconclusive'),
(-1,'session_hash','d'*64,'session-binding'),(-1,'issued_s',10001,'session-baseline'),
(-1,'error','access denied','session-refusal')]:
changed=copy.deepcopy(rows);changed[index][field]=value
with self.subTest(field=field),self.assertRaisesRegex(ValueError,error):oracle.check_sessions(changed,metadata,700)
with self.assertRaisesRegex(ValueError,'session-completeness'):oracle.check_sessions(rows[:2]+rows[3:],metadata,700)
@contextmanager
def lifecycle_fixture():
libc=ctypes.CDLL(None,use_errno=True);previous=ctypes.c_int()
if libc.prctl(37,ctypes.byref(previous),0,0,0) or libc.prctl(36,1,0,0,0):raise RuntimeError('fixture subreaper unavailable')
try:
with tempfile.TemporaryDirectory(prefix='clock-lifecycle-',dir=os.environ['FABRIC_SCRATCH_ROOT']) as directory:yield Path(directory)
finally:
if libc.prctl(36,previous.value,0,0,0):raise RuntimeError('fixture subreaper restoration failed')
def marker_pid(marker):
deadline=real_monotonic()+2
while real_monotonic()<deadline:
try:
text=marker.read_text()
if text.isdigit():return int(text)
except FileNotFoundError:pass
sleep(.005)
raise RuntimeError('bounded fixture PID witness unavailable')
def reap_fixture(pid):
deadline=real_monotonic()+5
while real_monotonic()<deadline:
try:
if os.waitpid(pid,os.WNOHANG)[0]:return
except ChildProcessError:
if not Path(f'/proc/{pid}').exists():return
sleep(.005)
def stop_fixture(pid):
try:
if os.waitpid(pid,os.WNOHANG)[0]:return
os.kill(pid,signal.SIGKILL);os.waitpid(pid,0)
except (ProcessLookupError,ChildProcessError):pass
class SessionClockLifecycleTests(unittest.TestCase):
def test_exited_leader_does_not_leave_term_resistant_group_member(self):
import session_clock as sut
with lifecycle_fixture() as root:
marker=root/'child.pid'
program='import os,signal,sys,time;from pathlib import Path\nchild=os.fork()\nif not child:\n signal.signal(signal.SIGTERM,signal.SIG_IGN);Path(sys.argv[1]).write_text(str(os.getpid()))\n while True:signal.pause()\nwhile not Path(sys.argv[1]).exists():time.sleep(.005)\nos._exit(0)\n'
process=subprocess.Popen([sys.executable,'-c',program,str(marker)],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,start_new_session=True)
pid=marker_pid(marker);process.wait(timeout=2);receiver=threading.Thread(target=reap_fixture,args=(pid,),daemon=True);receiver.start()
fixture=sut.SessionClock.__new__(sut.SessionClock);fixture.process=process;fixture.receiver=receiver;fixture.parent=Mock();fixture.stderr=Mock()
try:
error=None
try:fixture.close()
except ValueError as failure:error=str(failure)
observed=(error,receiver.is_alive())
finally:
stop_fixture(pid);receiver.join(timeout=2);process.stdin.close();process.stdout.close()
self.assertEqual(observed,(None,False),'whole owned group must stop even after its leader has exited')
def test_compiler_timeout_stops_descendant_and_preserves_output_and_partial_build(self):
import session_clock as sut
with lifecycle_fixture() as root:
marker=root/'compiler-child.pid';compiler=root/'fixture-compiler';output=root/'receipt'
compiler.write_text('#!'+sys.executable+'\nimport os,signal,sys,time\nfrom pathlib import Path\n'
'if "-vV" in sys.argv:print("release: 1.99.0\\ncommit-hash: b940084d7");sys.exit(0)\n'
'if "--version" in sys.argv:print("literal fixture compiler");sys.exit(0)\n'
'Path(sys.argv[sys.argv.index("-o")+1]).write_bytes(b"partial-object-public-evidence")\n'
'print("compiler-timeout-public-output",file=sys.stderr,flush=True)\nchild=os.fork()\n'
'if not child:\n signal.signal(signal.SIGTERM,signal.SIG_IGN);Path(os.environ["FCLK_TEST_MARKER"]).write_text(str(os.getpid()))\n'
'while True:signal.pause()\n');compiler.chmod(0o700)
def watcher():
reap_fixture(marker_pid(marker))
receiver=threading.Thread(target=watcher,daemon=True);receiver.start();calls=0
def clock():
nonlocal calls
calls+=1;return real_monotonic()+(0 if calls==1 else 239.25)
try:
with patch.dict(os.environ,FCLK_TEST_MARKER=str(marker)),patch.object(sut.time,'monotonic',clock),patch.object(sys,'argv',['session-clock','--output',str(output),'--rustc',str(compiler),'--cc',str(compiler)]):
with self.assertRaises(subprocess.TimeoutExpired):sut.main()
receiver.join(timeout=1);pid=marker_pid(marker)
blob=b''.join(path.read_bytes() for path in output.rglob('*') if path.is_file() and path.stat().st_size<=2*1024**2)
observed=(Path(f'/proc/{pid}').exists(),b'compiler-timeout-public-output' in blob,b'partial-object-public-evidence' in blob)
finally:
if marker.exists():stop_fixture(marker_pid(marker))
receiver.join(timeout=2)
self.assertEqual(observed,(False,True,True),'timeout must stop descendants and archive captured output plus partial build evidence')
def test_replaced_images_are_rejected_against_the_original_frozen_witness(self):
import session_clock as sut
with lifecycle_fixture() as root:
executable=root/'probe';helper=root/'shim.so';libc=root/'libc.so.6';proc=root/'proc';proc.mkdir();libc.write_bytes(b'literal libc witness')
(proc/'exe').symlink_to(executable);(proc/'stat').write_text('700 (fixture) S '+'0 '*18+'123\n');(proc/'cgroup').write_text('0::/literal-contained-fixture\n')
for field,message in [('executable','isolation-target-frozen'),('helper','isolation-helper-frozen')]:
with self.subTest(field=field):
executable.write_bytes(b'literal frozen executable');helper.write_bytes(b'literal frozen helper')
expected={'executable_sha256':hashlib.sha256(b'literal frozen executable').hexdigest(),'helper_sha256':hashlib.sha256(b'literal frozen helper').hexdigest(),
'executable_identity':{'device':executable.stat().st_dev,'inode':executable.stat().st_ino},'helper_identity':{'device':helper.stat().st_dev,'inode':helper.stat().st_ino}}
stat=helper.stat();(proc/'maps').write_text(f'1000-2000 r-xp 00000000 {os.major(stat.st_dev):02x}:{os.minor(stat.st_dev):02x} {stat.st_ino} {helper}\n2000-3000 r-xp 00000000 00:00 1 {libc}\n')
(executable if field=='executable' else helper).write_bytes(b'literal replacement after freeze')
# Legacy checker lacks the witness parameter and admits the real changed files.
options={'expected':expected} if 'expected' in inspect.signature(sut.process_binding).parameters else {}
refusal=None
with patch.object(sut,'Path',side_effect=lambda value:proc if str(value)=='/proc/700' else Path(value)):
try:sut.process_binding(700,executable,helper,**options)
except ValueError as failure:refusal=str(failure)
self.assertEqual(refusal,message,'current path agreement cannot replace the pre-launch frozen hashes and identities')
def test_same_bytes_replacement_and_forged_mapping_reject_frozen_identity(self):
import session_clock as sut
with lifecycle_fixture() as root:
executable=root/'probe';helper=root/'shim.so';libc=root/'libc.so.6';proc=root/'proc';proc.mkdir();libc.write_bytes(b'literal libc witness')
(proc/'exe').symlink_to(executable);(proc/'stat').write_text('700 (fixture) S '+'0 '*18+'123\n');(proc/'cgroup').write_text('0::/literal-contained-fixture\n')
for case in ('executable-replaced','helper-replaced','mapping-device','mapping-inode'):
with self.subTest(case=case):
executable.write_bytes(b'literal frozen executable');helper.write_bytes(b'literal frozen helper')
expected={'executable_sha256':hashlib.sha256(b'literal frozen executable').hexdigest(),'helper_sha256':hashlib.sha256(b'literal frozen helper').hexdigest(),
'executable_identity':{'device':executable.stat().st_dev,'inode':executable.stat().st_ino},'helper_identity':{'device':helper.stat().st_dev,'inode':helper.stat().st_ino}}
if case.endswith('-replaced'):
selected=executable if case=='executable-replaced' else helper;frozen=selected.stat();replacement=root/'same-bytes-new-inode'
replacement.write_bytes(selected.read_bytes());self.assertNotEqual(replacement.stat().st_ino,frozen.st_ino)
replacement.replace(selected)
self.assertEqual(hashlib.sha256(selected.read_bytes()).hexdigest(),expected['executable_sha256' if selected==executable else 'helper_sha256'])
self.assertNotEqual(selected.stat().st_ino,frozen.st_ino)
stat=helper.stat();major=os.major(stat.st_dev)+(1 if case=='mapping-device' else 0);inode=stat.st_ino+(1 if case=='mapping-inode' else 0)
(proc/'maps').write_text(f'1000-2000 r-xp 00000000 {major:02x}:{os.minor(stat.st_dev):02x} {inode} {helper}\n2000-3000 r-xp 00000000 00:00 1 {libc}\n')
options={'expected':expected} if 'expected' in inspect.signature(sut.process_binding).parameters else {}
refusal=None
with patch.object(sut,'Path',side_effect=lambda value:proc if str(value)=='/proc/700' else Path(value)):
try:sut.process_binding(700,executable,helper,**options)
except ValueError as failure:refusal=str(failure)
self.assertEqual(refusal,'isolation-target-frozen' if case=='executable-replaced' else 'isolation-helper-frozen',
'unchanged bytes cannot replace frozen file or loaded mapping device/inode witnesses')
def session_fixtures():
# Independent literal receipts, not native process/state admission evidence.
rows, metadata = sessions()
fixtures = {}
for key, pid, inode, start, stop in (
('AB', 700, 101, 2_000_000_000, 10_000_000_000),
('C', 701, 102, 11_000_000_000, 31_000_000_000)):
state = {'path': '/owned/browser/state-' + key, 'device': 17, 'inode': inode}
fixtures[key] = {
'candidate_commit': '8fb6f413179b094288316e25845c137440ba23fa',
'package_sha256': 'd' * 64, 'executable_sha256': 'e' * 64,
'helper_sha256': 'f' * 64,
'process_binding': {
'pid': pid, 'start_ticks': pid + 100, 'cgroup': '0::/owned/browser-' + key + '\n',
'executable_sha256': 'e' * 64, 'helper_sha256': 'f' * 64,
'libc': '/owned/libc.so.6', 'libc_sha256': '1' * 64,
'helper_maps': ['08:01', '99', '/owned/session_clock.so'](/kmosoti/FabricO11y/wiki/'08:01',-'99',-'/owned/session_clock.so')},
'state_before': dict(state, created_monotonic_ns=start - 100_000_000,
observed_monotonic_ns=start - 50_000_000,
initial_entries=[], copy_source=None, descriptor_held=True),
'state_after': dict(state, observed_monotonic_ns=stop - 1_000_000),
'run_started_monotonic_ns': 1_000_000_000,
'deadline_monotonic_ns': 241_000_000_000,
'started_monotonic_ns': start, 'stopped_monotonic_ns': stop,
'owned_groups': [{'pgid': pid, 'start_ticks': pid + 100}],
'stop_observations': [{'pgid': pid, 'start_ticks': pid + 100,
'observed_monotonic_ns': stop,
'population': 0, 'confirmed': True}]}
for index, row in enumerate(rows):
row['pid'] = 700 if index < 2 else 701
before = 3_000_000_000 + index * 600_000_000 if index < 2 else 12_000_000_000 + (index - 2) * 600_000_000
row.update(request_before_monotonic_ns=before,
server_call_monotonic_ns=before + 1_000_000,
request_after_monotonic_ns=before + 2_000_000)
return rows, metadata, fixtures
class SessionFixtureBindingTests(unittest.TestCase):
def checker(self):
checker = getattr(oracle, 'check_session_fixtures', None)
self.assertTrue(callable(checker), 'missing explicit two-state session fixture checker')
return checker
def test_explicit_two_state_checker_accepts_literal_binding_without_rewriting_rows(self):
checker = self.checker()
inputs = session_fixtures()
original = copy.deepcopy(inputs)
self.assertIsNone(checker(*inputs))
self.assertEqual(inputs, original, 'checker must not rewrite observed rows or receipts')
def test_fixture_keys_images_and_process_receipts_are_independently_bound(self):
checker = self.checker()
self.assertIsNone(checker(*session_fixtures()))
for keys in ({'AB'}, {'AB', 'C', 'extra'}, {'A', 'B', 'C'}):
rows, metadata, fixtures = session_fixtures()
fixtures = {key: fixtures.get(key, fixtures['AB']) for key in keys}
with self.subTest(keys=keys), self.assertRaisesRegex(ValueError, '^session-fixture-keys$'):
checker(rows, metadata, fixtures)
cases = [
(('C', 'candidate_commit'), '0' * 40, 'images'),
(('C', 'package_sha256'), '0' * 64, 'images'),
(('C', 'executable_sha256'), '0' * 64, 'images'),
(('C', 'helper_sha256'), '0' * 64, 'images'),
(('C', 'process_binding', 'pid'), 700, 'pid'),
(('C', 'process_binding', 'pid'), 0, 'pid'),
(('C', 'process_binding', 'pid'), True, 'pid'),
(('C', 'process_binding', 'pid'), 701.0, 'pid'),
(('C', 'process_binding', 'start_ticks'), 0, 'process'),
(('C', 'process_binding', 'cgroup'), '', 'process'),
(('C', 'process_binding', 'executable_sha256'), '0' * 64, 'process'),
(('C', 'process_binding', 'helper_sha256'), '0' * 64, 'process'),
(('C', 'process_binding', 'helper_maps'), [], 'process')]
for path, value, error in cases:
rows, metadata, fixtures = session_fixtures()
target = fixtures
for part in path[:-1]:
target = target[part]
target[path[-1]] = value
with self.subTest(path=path), self.assertRaisesRegex(ValueError, '^session-fixture-' + error + '$'):
checker(rows, metadata, fixtures)
rows, metadata, fixtures = session_fixtures()
del fixtures['C']['process_binding']
with self.assertRaisesRegex(ValueError, '^session-fixture-process$'):
checker(rows, metadata, fixtures)
def test_fresh_distinct_state_and_serial_zero_population_stop_are_required(self):
checker = self.checker()
self.assertIsNone(checker(*session_fixtures()))
cases = [
(('C', 'state_before', 'path'), '/owned/browser/state-AB', 'state'),
(('C', 'state_before', 'inode'), 101, 'state'),
(('C', 'state_before', 'initial_entries'), ['access.state'], 'fresh'),
(('C', 'state_before', 'copy_source'), '/owned/browser/state-AB', 'fresh'),
(('C', 'state_before', 'descriptor_held'), False, 'fresh'),
(('C', 'state_before', 'created_monotonic_ns'), 11_000_000_000, 'fresh'),
(('C', 'state_before', 'observed_monotonic_ns'), 10_000_000_000, 'fresh'),
(('C', 'state_after', 'inode'), 103, 'state'),
(('C', 'state_after', 'observed_monotonic_ns'), 10_000_000_000, 'state'),
(('C', 'started_monotonic_ns'), 10_000_000_000, 'order'),
(('AB', 'stop_observations', 0, 'confirmed'), False, 'stop'),
(('AB', 'stop_observations', 0, 'confirmed'), 1, 'stop'),
(('AB', 'stop_observations', 0, 'population'), 1, 'stop'),
(('AB', 'stop_observations', 0, 'population'), False, 'stop'),
(('AB', 'stop_observations', 0, 'pgid'), 999, 'stop'),
(('AB', 'stop_observations', 0, 'start_ticks'), 801, 'stop'),
(('AB', 'stop_observations', 0, 'observed_monotonic_ns'), 11_000_000_000, 'stop'),
(('AB', 'stop_observations'), [], 'stop'),
(('AB', 'owned_groups'), [], 'stop'),
(('C', 'run_started_monotonic_ns'), 2_000_000_000, 'deadline'),
(('C', 'deadline_monotonic_ns'), 481_000_000_000, 'deadline'),
(('C', 'deadline_monotonic_ns'), True, 'deadline'),
(('C', 'stopped_monotonic_ns'), 241_000_000_001, 'deadline')]
for path, value, error in cases:
rows, metadata, fixtures = session_fixtures()
target = fixtures
for part in path[:-1]:
target = target[part]
target[path[-1]] = value
if path in (('C', 'state_before', 'path'), ('C', 'state_before', 'inode')):
fixtures['C']['state_after'][path[-1]] = value
if path == ('C', 'started_monotonic_ns'):
fixtures['C']['state_before']['created_monotonic_ns'] = value - 100_000_000
fixtures['C']['state_before']['observed_monotonic_ns'] = value - 50_000_000
with self.subTest(path=path), self.assertRaisesRegex(ValueError, '^session-fixture-' + error + '$'):
checker(rows, metadata, fixtures)
rows, metadata, fixtures = session_fixtures()
fixtures['AB']['owned_groups'].append({'pgid': 702, 'start_ticks': 802})
with self.assertRaisesRegex(ValueError, '^session-fixture-stop$'):
checker(rows, metadata, fixtures)
for field in ('owned_groups', 'stop_observations'):
rows, metadata, fixtures = session_fixtures()
fixtures['AB'][field].append(copy.deepcopy(fixtures['AB'][field][0]))
with self.subTest(field=field), self.assertRaisesRegex(ValueError, '^session-fixture-stop$'):
checker(rows, metadata, fixtures)
def test_exact_session_rows_keep_original_boundaries_and_actual_request_windows(self):
checker = self.checker()
self.assertIsNone(checker(*session_fixtures()))
cases = [
(0, 'pid', 701, 'session-fixture-row'),
(2, 'pid', 700, 'session-fixture-row'),
(2, 'pid', 999, 'session-fixture-row'),
(2, 'pid', True, 'session-fixture-row'),
(2, 'server_call_monotonic_ns', 11_999_999_999, 'session-fixture-window'),
(2, 'request_before_monotonic_ns', 10_999_999_999, 'session-fixture-window'),
(31, 'request_after_monotonic_ns', 31_000_000_001, 'session-fixture-window'),
(3, 'request_before_monotonic_ns', 12_549_999_999, 'session-fixture-pacing'),
(0, 'request_s', 2800, 'session-boundary'),
(1, 'status', 429, 'session-refusal'),
(1, 'error', 'rate limited', 'session-refusal'),
(31, 'status', 200, 'session-refusal'),
(31, 'last_seen_s', 38000, 'session-baseline'),
(31, 'session_hash', 'b' * 64, 'session-binding')]
for index, field, value, error in cases:
rows, metadata, fixtures = session_fixtures()
rows[index][field] = value
with self.subTest(index=index, field=field), self.assertRaisesRegex(ValueError, '^' + error + '$'):
checker(rows, metadata, fixtures)
rows, metadata, fixtures = session_fixtures()
with self.assertRaisesRegex(ValueError, '^session-completeness$'):
checker(rows[:5] + rows[6:], metadata, fixtures)
rows, metadata, fixtures = session_fixtures()
del rows[2]['server_call_monotonic_ns']
with self.assertRaisesRegex(ValueError, '^session-fixture-window$'):
checker(rows, metadata, fixtures)
rows, metadata, fixtures = session_fixtures()
fixtures['AB']['stopped_monotonic_ns'] = 4_000_000_000
fixtures['AB']['stop_observations'][0]['observed_monotonic_ns'] = 4_000_000_000
fixtures['C']['started_monotonic_ns'] = 4_100_000_000
fixtures['C']['state_before']['created_monotonic_ns'] = 4_050_000_000
fixtures['C']['state_before']['observed_monotonic_ns'] = 4_075_000_000
for index, row in enumerate(rows[2:]):
before = 4_149_999_999 + index * 600_000_000
row.update(request_before_monotonic_ns=before, server_call_monotonic_ns=before + 1_000_000,
request_after_monotonic_ns=before + 2_000_000)
with self.assertRaisesRegex(ValueError, '^session-fixture-pacing$'):
checker(rows, metadata, fixtures)
if __name__=='__main__':unittest.main()
Actual-finalizer controls and independent cleanup
"""ADR-derived route coverage; origin: issue45/C-027 and the missing legacy rows."""
import ast
import copy
from contextlib import contextmanager
import ctypes
import hashlib
import json
import os
from pathlib import Path
import re
import shutil
import signal
import socket
import stat
import subprocess
import sys
import tempfile
import threading
import time
from types import SimpleNamespace
import unittest
from unittest.mock import patch
import live_browser_check as browser
ROOT = Path(__file__).resolve().parents[2]
FIXTURE = ROOT / 'docs/experiments/formal/fixtures/console-route-roles.json'
def registered_routes(source):
routes = set()
for match in re.finditer(r'\.route\(\s*"([^"]+)"\s*,', source):
depth, end = 1, match.end()
while depth:
depth += (source[end] == '(') - (source[end] == ')')
end += 1
for method in re.findall(r'\b(get|post|put|patch|delete|head|options)\s*\(', source[match.end():end]):
routes.add((method.upper(), match.group(1)))
return routes
def witnessed_calls(request, check):
if hasattr(browser, 'run_role_matrix'):
browser.run_role_matrix(request, check)
return
# Execute ONLY the retained selected-matrix AST against synthetic transports.
# No main(), browser, sockets, subprocesses or native authority are invoked.
main = next(node for node in ast.parse(Path(browser.__file__).read_text()).body
if isinstance(node, ast.FunctionDef) and node.name == 'main')
statements = next(node.body for node in ast.walk(main) if isinstance(node, ast.Try)
and any(isinstance(n, ast.Assign) and any(isinstance(t, ast.Name) and t.id == 'owner_routes'
for t in n.targets) for n in node.body))
begin = next(i for i, node in enumerate(statements) if isinstance(node, ast.Assign)
and any(isinstance(t, ast.Name) and t.id == 'owner_routes' for t in node.targets))
end = next(i for i in range(begin + 1, len(statements)) if isinstance(statements[i], ast.Assign)
and any(isinstance(t, ast.Name) and t.id == 'read' for t in statements[i].targets))
namespace = {'credential': 'reader', 'spindle_token': 'spindle', 'check': check,
'bearer_api': lambda role, path, method='GET', body=None: request(role or 'anonymous', method, path, body),
'api': lambda path, method='GET', body=None: request('owner', method, path, body),
'expect': lambda answer, status, message, label: check(browser.semantic_rejection(answer, status, message), label)}
exec(compile(ast.Module(body=statements[begin:end], type_ignores=[]), '<retained-matrix>', 'exec'), namespace)
class RoleMatrixTests(unittest.TestCase):
def setUp(self):
self.fixture = json.loads(FIXTURE.read_text())
self.expected = {(role, method, path): outcome
for method, _, path, _, outcomes in self.fixture['rows']
for role, outcome in zip(self.fixture['roles'], outcomes)}
def request(self, role, method, path, body=None):
# The retained fixture targeted browser-edge; the successor owns matrix.
path = path.replace('/v1/console/nodes/browser-edge/', '/v1/console/nodes/matrix/')
status, error = self.expected.get((role, method, path), (200, None))
field = self.fixture['refusal_fields'].get(path, 'error')
return status, {field: error} if error else {}
def test_fixture_covers_both_registered_routers_and_all_control_actions(self):
rows = self.fixture['rows']
self.assertEqual(self.fixture['roles'], ['anonymous', 'spindle', 'reader', 'owner'])
self.assertEqual(len(rows), 36)
self.assertEqual(len(self.expected), 144)
registered = set().union(*(registered_routes((ROOT / 'crates/fabric-server/src' / name).read_text())
for name in ('console.rs', 'http.rs')))
self.assertEqual({(row[0], row[1]) for row in rows}, registered)
for prefix in ('/v1/console', '/v1/admin'):
self.assertEqual({row[2].rsplit('/', 1)[1] for row in rows
if row[1] == prefix + '/nodes/{name}/{action}'}, {'pause', 'resume', 'revoke'})
def test_live_matrix_calls_all_credentials_on_every_route_and_legacy_action(self):
calls = []
def request(role, method, path, body):
calls.append((role, method, path))
return self.request(role, method, path, body)
witnessed_calls(request, self.assertTrue)
self.assertEqual(set(calls), set(self.expected),
'live role matrix missing cells: ' + repr(sorted(set(self.expected) - set(calls))))
self.assertEqual(len(calls), 144, 'each registered role-route cell must run exactly once')
def test_reader_admission_mutation_is_rejected_by_the_live_checker(self):
def request(role, method, path, body):
if (role, method, path) == ('reader', 'GET', '/v1/console/audit'):
return 200, {'events': []}
return self.request(role, method, path, body)
with self.assertRaisesRegex(AssertionError, 'reader.*GET /v1/console/audit'):
witnessed_calls(request, self.assertTrue)
def test_independent_outcomes_never_accept_admission_refusal_or_role_crossover(self):
for role, path, expected in [
('spindle', '/v1/console/status', [401, 'sign in required']),
('reader', '/v1/console/audit', [403, 'role not permitted']),
('owner', '/v1/admin/nodes', [401, 'use a console credential']),
]:
self.assertEqual(self.expected[(role, 'GET', path)], expected)
self.assertFalse(browser.semantic_rejection((429, {'error': expected[1]}), *expected))
self.assertFalse(browser.semantic_rejection((200, {'error': expected[1]}), *expected))
self.assertFalse(browser.semantic_rejection((expected[0], {'error': 'wrong refusal'}), *expected))
missing = copy.deepcopy(self.fixture['rows'])
missing.remove(next(row for row in missing if row[0:3] ==
['POST', '/v1/admin/nodes/{name}/{action}', '/v1/admin/nodes/matrix/resume']))
self.assertEqual(len({(r[0], r[1]) for r in missing}), len({(r[0], r[1]) for r in self.fixture['rows']}))
self.assertNotEqual({(r[0], r[2]) for r in missing}, {(r[0], r[2]) for r in self.fixture['rows']})
def test_transports_refuse_expired_or_storage_limited_dispatch_and_progressing_body(self):
# Origin: Miso's independent review of0de4b46; no main/network/process.
source = ast.parse(Path(browser.__file__).read_text())
main = next(n for n in source.body if isinstance(n, ast.FunctionDef) and n.name == 'main')
functions = [n for n in source.body if isinstance(n, ast.FunctionDef) and n.name != 'main']
functions += [n for n in main.body if isinstance(n, ast.FunctionDef) and n.name in ('call', 'bearer_api')]
for transport in ('call', 'bearer_api'):
for case in ('expired', 'storage', 'paced_expiry', 'timeout', 'progressing'):
if transport == 'call' and case == 'paced_expiry':continue
with self.subTest(transport=transport, case=case):
now, opened, reads = [2.0 if case == 'expired' else 0.0], [], []
class Response:
status = 200
fp = SimpleNamespace(raw=SimpleNamespace(_sock=SimpleNamespace(settimeout=lambda _: None)))
def __enter__(self):return self
def __exit__(self, *_):pass
def read(self, size=None):
reads.append('whole')
if case == 'progressing':now[0] = 2.0
return b'{"value":{"ready":true}}'
def read1(self, size):
reads.append('chunk')
if case == 'progressing':now[0] = 2.0
return b'{"value":{"ready":true}}' if len(reads) == 1 else b''
def open_response(request, **kwargs):
opened.append(kwargs['timeout'])
return Response()
namespace = dict(vars(browser))
namespace.update(time=SimpleNamespace(monotonic=lambda:now[0], monotonic_ns=lambda:int(now[0]*1e9)),
deadline=1.0, storage_violation=SimpleNamespace(is_set=lambda:case == 'storage'),
endpoint='http://localhost:1', origin='https://localhost:1', receipt={},
args=SimpleNamespace(inject_role_defect=False), tls_opener=SimpleNamespace(open=open_response),
pace_api=lambda:now.__setitem__(0, 2.0 if case == 'paced_expiry' else now[0]),
cadence=SimpleNamespace(clock=lambda:now[0], last=None),
urllib=SimpleNamespace(request=SimpleNamespace(Request=browser.urllib.request.Request, urlopen=open_response),
error=browser.urllib.error))
exec(compile(ast.Module(body=functions, type_ignores=[]), '<transport-boundary>', 'exec'), namespace)
invoke = (lambda:namespace['call']('GET', '/status')) if transport == 'call' else (lambda:namespace['bearer_api'](None, '/v1/console/nodes'))
if case == 'timeout':
invoke()
self.assertEqual(len(opened), 1)
self.assertGreater(opened[0], 0)
self.assertLessEqual(opened[0], 1.0, 'socket timeout exceeds remaining inner deadline')
else:
with self.assertRaisesRegex(RuntimeError, 'deadline|storage|512 MiB'):
invoke()
if case == 'progressing':self.assertEqual(reads, ['chunk'], 'wall bound must stop a progressively readable body')
else:self.assertEqual(opened, [], 'expired/storage-limited dispatch reached network')
def test_actual_slow_headers_and_chunk_metadata_cannot_outlive_inner_io_bounds(self):
# Origin: CPython read1 still reads chunk metadata/header lines internally.
source = ast.parse(Path(browser.__file__).read_text())
main = next(n for n in source.body if isinstance(n, ast.FunctionDef) and n.name == 'main')
functions = [n for n in source.body if isinstance(n, ast.FunctionDef) and n.name != 'main']
functions += [n for n in main.body if isinstance(n, ast.FunctionDef) and n.name == 'call']
for case in ('headers', 'chunk_metadata', 'storage'):
with self.subTest(case=case):
storage, stop, errors = threading.Event(), threading.Event(), []
with socket.socket() as listener:
listener.bind(('127.0.0.1', 0));listener.listen(1);listener.settimeout(2)
def peer():
try:
connection, _ = listener.accept()
with connection:
connection.settimeout(1);request=b''
while b'\r\n\r\n' not in request:request+=connection.recv(4096)
body=b'{"value":{"ready":true}}'
prefix=(b'HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n'+f'{len(body):x};note='.encode()
if case == 'chunk_metadata' else b'HTTP/1.1 200 OK\r\nX-Progress: ')
connection.sendall(prefix);began=time.monotonic()
while time.monotonic()-began<.6 and not stop.is_set():
if case == 'storage' and time.monotonic()-began>=.12:storage.set()
connection.sendall(b'x');stop.wait(.01)
suffix=(b'\r\n'+body+b'\r\n0\r\n\r\n' if case == 'chunk_metadata'
else b'\r\nContent-Length: '+str(len(body)).encode()+b'\r\n\r\n'+body)
if not stop.is_set():connection.sendall(suffix)
except (BrokenPipeError, ConnectionResetError):pass
except Exception as error:errors.append(error)
worker=threading.Thread(target=peer, daemon=True);worker.start()
namespace=dict(vars(browser));began=time.monotonic()
namespace.update(endpoint=f'http://127.0.0.1:{listener.getsockname()[1]}',
deadline=began+(2 if case == 'storage' else .12),storage_violation=storage)
exec(compile(ast.Module(body=functions, type_ignores=[]), '<actual-io-boundary>', 'exec'), namespace)
previous_handler=browser.signal.getsignal(browser.signal.SIGALRM)
previous_timer=browser.signal.getitimer(browser.signal.ITIMER_REAL)
try:
with self.assertRaisesRegex(RuntimeError, 'storage|512 MiB' if case == 'storage' else 'deadline'):
namespace['call']('GET', '/status')
elapsed=time.monotonic()-began
finally:
stop.set();worker.join(1)
self.assertFalse(worker.is_alive(), 'synthetic peer did not stop')
self.assertEqual(errors, [], 'synthetic peer failed before exercising the bound')
self.assertEqual(browser.signal.getsignal(browser.signal.SIGALRM), previous_handler)
self.assertEqual(browser.signal.getitimer(browser.signal.ITIMER_REAL), previous_timer)
self.assertLessEqual(elapsed, .35, 'continuous metadata progress evaded the absolute I/O bound')
def test_actual_browser_dispatch_gap_survives_delayed_pace_continuation(self):
# Origin: Miso's 5ms continuation counterexample; execute actual injected JS.
source=ast.parse(Path(browser.__file__).read_text())
scripts=[n.args[0].value for n in ast.walk(source) if isinstance(n, ast.Call)
and isinstance(n.func, ast.Name) and n.func.id == 'js' and n.args
and isinstance(n.args[0], ast.Constant) and isinstance(n.args[0].value, str)
and 'window.fixturePaceQueue=Promise.resolve()' in n.args[0].value]
self.assertEqual(len(scripts), 1, 'one actual fetch scheduler must supply the control')
runner=browser.DATA/'tools/bun-1.4.0/bun'
if not runner.is_file():runner=Path(browser.shutil.which('bun') or str(runner))
self.assertTrue(runner.is_file(), 'Bun is required for the actual JavaScript control')
script='const setup='+json.dumps(scripts[0])+r''';
let now=0,delayed=false;const actual=[];
const window={fetch:async(url)=>{actual.push({path:url,started_ms:now});return {status:200,headers:{get:()=>null}}}};
new Function('window','performance','setTimeout','location','return function(){'+setup+'}')(
window,{now:()=>now},(resolve,delay)=>{now+=delay;resolve()},{href:'https://fixture.invalid/console/'})(550);
const pace=window.fixturePace;
window.fixturePace=async(...args)=>{const value=await pace(...args);if(!delayed){delayed=true;now+=5}return value};
await window.fetch('/v1/console/auth/login/start');await window.fetch('/v1/console/auth/login/finish');
window.fixtureUnpacedBurst=true;
await Promise.all(Array.from({length:16},()=>window.fetch('/v1/console/auth/login/start')));
console.log(JSON.stringify({actual,recorded:window.fixtureApiHeaders,truncated:window.fixtureApiHeadersTruncated}));
'''
result=subprocess.run([str(runner),'--eval',script],capture_output=True,text=True,timeout=5)
self.assertEqual(result.returncode,0,result.stderr)
witness=json.loads(result.stdout);actual=witness['actual'];recorded=witness['recorded']
self.assertEqual(len(actual),18,'two paced requests and the unchanged16-request unpaced control')
self.assertEqual([r['started_ms'] for r in actual],[r['started_ms'] for r in recorded], 'receipt stamps actual fetch dispatch')
self.assertEqual([r['status'] for r in recorded],[200]*18)
self.assertEqual([r['paced'] for r in recorded],[True,True]+[False]*16)
self.assertEqual(len({r['started_ms'] for r in actual[2:]}),1,'registered burst remains unpaced')
self.assertFalse(witness['truncated'])
self.assertGreaterEqual(actual[1]['started_ms']-actual[0]['started_ms'],550,
'5ms pace-continuation delay shrank actual dispatch gap: '+repr(actual[:2]))
def owned_browser_identity(pid):
process = Path('/proc') / str(pid)
fields = (process / 'stat').read_text().rsplit(')', 1)[1].split()
if process.stat().st_uid != os.getuid() or int(fields[19]) <= 0:
raise AssertionError('unowned browser control process')
return int(fields[19]), int(fields[2])
@contextmanager
def browser_cleanup_fixture(modes, blocked_sampler=False):
# Every process is real but synthetic, with a finite 45-second lifespan.
sys.path.insert(0, str(ROOT / 'tools'))
from resource_group import require_limits
require_limits()
libc, previous = ctypes.CDLL(None, use_errno=True), ctypes.c_int()
if libc.prctl(37, ctypes.byref(previous), 0, 0, 0) or libc.prctl(36, 1, 0, 0, 0):
raise RuntimeError('browser process controls require Linux subreaper')
directory = None
groups, identities, process_logs = [], [], []
sampler = None
body_failure = None
restore_attempted = False
stop_samples, release_sampler = threading.Event(), threading.Event()
try:
directory = Path(tempfile.mkdtemp(dir=os.environ['FABRIC_SCRATCH_ROOT'], prefix='browser-cleanup-control-'))
work, out = directory / 'work', directory / 'out'
work.mkdir()
out.mkdir()
script = directory / 'owned-process.py'
script.write_text('''import os, select, signal, sys, time
from pathlib import Path
directory, work, mode = Path(sys.argv[1]), Path(sys.argv[2]), sys.argv[3]
expires = time.monotonic() + 45
def atomic(name, text):
pending = directory / (name + '.pending')
with pending.open('x') as stream:
stream.write(text); stream.flush(); os.fsync(stream.fileno())
pending.replace(directory / name)
if mode == 'timeout': signal.signal(signal.SIGTERM, signal.SIG_IGN)
gate_read, gate_write = os.pipe()
child = os.fork()
if child == 0:
os.close(gate_write)
if not select.select([gate_read], [], [], 5)[0] or os.read(gate_read, 1) != b'1': os._exit(1)
os.close(gate_read)
signal.signal(signal.SIGTERM, signal.SIG_IGN)
held = (work / ('held-' + directory.name)).open('xb')
held.write(b'literal browser private fixture'); held.flush()
atomic('ready', 'ready')
time.sleep(max(0, expires - time.monotonic()))
os._exit(0)
os.close(gate_read)
try:
process = Path('/proc') / str(child)
fields = (process / 'stat').read_text().rsplit(')', 1)[1].split()
if process.stat().st_uid != os.getuid() or int(fields[19]) <= 0 or int(fields[1]) != os.getpid() or int(fields[2]) != os.getpid():
raise RuntimeError('fork child is outside owned private group')
atomic('child.pid', f'{child} {fields[19]}')
os.write(gate_write, b'1')
finally:
os.close(gate_write)
deadline = time.monotonic() + 5
while not (directory / 'ready').exists():
if time.monotonic() >= deadline: raise RuntimeError('child readiness deadline')
time.sleep(.01)
if mode != 'exited': time.sleep(max(0, expires - time.monotonic()))
''')
for index, mode in enumerate(modes):
group_dir = directory / ('group-' + str(index))
group_dir.mkdir()
log = (group_dir / 'process.log').open('xb')
process_logs.append(log)
process = subprocess.Popen([sys.executable, '-B', str(script), str(group_dir), str(work), mode],
start_new_session=True, stdout=log, stderr=log)
groups.append({'process': process, 'directory': group_dir})
start, pgid = owned_browser_identity(process.pid)
descriptor = os.pidfd_open(process.pid)
identities.append((process.pid, start, descriptor))
if owned_browser_identity(process.pid) != (start, process.pid) or pgid != process.pid:
raise AssertionError('owned leader identity changed')
deadline = time.monotonic() + 5
while not (group_dir / 'ready').exists():
if time.monotonic() >= deadline:
raise AssertionError('owned child did not publish readiness')
time.sleep(.01)
marker = group_dir / 'child.pid'
info = marker.lstat()
if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_nlink != 1 or info.st_size > 128:
raise AssertionError('invalid child identity marker')
child, child_start = map(int, marker.read_text().split())
if child <= 1 or child_start <= 0 or owned_browser_identity(child) != (child_start, process.pid):
raise AssertionError('unbound owned child identity')
child_descriptor = os.pidfd_open(child)
identities.append((child, child_start, child_descriptor))
if owned_browser_identity(child) != (child_start, process.pid):
raise AssertionError('child identity changed during pidfd binding')
groups[-1].update(child=child, child_start=child_start)
if mode == 'exited':
process.wait(timeout=5)
sampler = threading.Thread(target=lambda: release_sampler.wait(4) if blocked_sampler else stop_samples.wait(4))
sampler.start()
namespace = dict(vars(browser))
namespace.update(session=None, processes=[('group-' + str(i), item['process']) for i, item in enumerate(groups)],
receipt={'exit': 0}, stop_samples=stop_samples, sampler=sampler,
out=out, samples=[], logs=[], work=work)
yield namespace, groups
except BaseException as error:
body_failure = {'type': type(error).__name__, 'message': str(error)[:1024]}
raise
finally:
try:
errors = []
stop_evidence, group_evidence, file_evidence = [], [], []
cleanup_began_monotonic_ns = time.monotonic_ns()
sampler_alive_before_cleanup = sampler is not None and sampler.is_alive()
stop_samples.set()
release_sampler.set()
if sampler is not None:
try:
sampler.join(timeout=5)
if sampler.is_alive():
errors.append('sampler did not stop')
except RuntimeError as error:
errors.append(str(error))
for pid, start, descriptor in identities:
observed = {'pid': pid, 'expected_start_ticks': start,
'began_monotonic_ns': time.monotonic_ns(),
'pidfd_kill_attempted': False, 'stop_confirmed': False}
stop_evidence.append(observed)
try:
try:
current, pgid = owned_browser_identity(pid)
except FileNotFoundError:
observed.update(stop_confirmed=True, outcome='already absent')
continue
observed.update(observed_start_ticks=current, pgid=pgid)
if current != start:
raise AssertionError('refuse cleanup of reused PID')
fields = (Path('/proc') / str(pid) / 'stat').read_text().rsplit(')', 1)[1].split()
if int(fields[19]) != start:
raise AssertionError('process identity changed before state observation')
observed['state_before_signal'] = fields[0]
try:
observed['pidfd_kill_attempted'] = True
signal.pidfd_send_signal(descriptor, signal.SIGKILL)
except ProcessLookupError:
pass
deadline = time.monotonic() + 5
while True:
try:
reaped, status = os.waitpid(pid, os.WNOHANG)
except ChildProcessError:
reaped = 0
if reaped == pid:
observed.update(stop_confirmed=True, outcome='reaped by control', wait_status=status)
break
try:
current, _ = owned_browser_identity(pid)
except FileNotFoundError:
observed.update(stop_confirmed=True, outcome='confirmed absent')
break
if current != start or time.monotonic() >= deadline:
raise AssertionError('owned process identity/cleanup not confirmed')
time.sleep(.01)
except (OSError, ValueError, IndexError, AssertionError) as error:
errors.append(str(error))
observed['error'] = str(error)[:512]
finally:
observed['ended_monotonic_ns'] = time.monotonic_ns()
try:
os.close(descriptor)
except OSError as error:
errors.append(str(error))
for item in groups:
observed = {'pgid': item['process'].pid, 'confirmed_zero_population': False,
'before_monotonic_ns': time.monotonic_ns()}
group_evidence.append(observed)
try:
os.killpg(item['process'].pid, 0)
except ProcessLookupError:
observed['confirmed_zero_population'] = True
continue
except OSError as error:
errors.append(str(error))
observed['error'] = str(error)[:512]
else:
errors.append('owned group is still populated')
finally:
observed['after_monotonic_ns'] = time.monotonic_ns()
for log in process_logs:
try:
log.close()
except OSError as error:
errors.append(str(error))
restore_attempted = True
subreaper_restored = libc.prctl(36, previous.value, 0, 0, 0) == 0
if not subreaper_restored:
errors.append('restore previous test subreaper setting')
if directory is not None:
witnesses = [(directory / 'owned-process.py', 8192)]
for item in groups:
witnesses.extend((item['directory'] / name, cap)
for name, cap in (('child.pid', 128), ('ready', 128), ('process.log', 8192)))
for path, cap in witnesses:
record = {'path': str(path.relative_to(directory))}
file_evidence.append(record)
try:
descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
try:
before = os.fstat(descriptor)
if not stat.S_ISREG(before.st_mode) or before.st_uid != os.getuid() or before.st_nlink != 1 or before.st_size > cap:
raise AssertionError('invalid bounded evidence file')
raw = os.read(descriptor, cap + 1)
after = os.fstat(descriptor)
identity = lambda value: {key: getattr(value, 'st_' + key) for key in
('dev', 'ino', 'mode', 'uid', 'nlink', 'size', 'mtime_ns', 'ctime_ns')}
if len(raw) != before.st_size or identity(before) != identity(after):
raise AssertionError('evidence file changed during observation')
record.update(identity=identity(before), sha256=hashlib.sha256(raw).hexdigest())
if path.name == 'child.pid':
child, child_start = map(int, raw.split())
record.update(pid=child, start_ticks=child_start)
record['pidfd_bound'] = any((pid, start) == (child, child_start)
for pid, start, _ in identities)
if not record['pidfd_bound']:
raise AssertionError('marker lacks independently bound pidfd identity')
finally:
os.close(descriptor)
except (OSError, ValueError, AssertionError) as error:
record['error'] = str(error)[:512]
errors.append('evidence ' + record['path'] + ': ' + str(error)[:512])
retained = body_failure is not None or bool(errors)
evidence = {'body_failure': body_failure, 'scratch_retained': retained,
'cleanup_began_monotonic_ns': cleanup_began_monotonic_ns,
'processes': stop_evidence, 'groups': group_evidence, 'files': file_evidence,
'sampler_alive_before_cleanup': sampler_alive_before_cleanup,
'sampler_stopped': sampler is None or not sampler.is_alive(),
'subreaper_restored': subreaper_restored,
'work_exists_after_control': (directory / 'work').exists(),
'cleanup_errors': errors, 'recorded_monotonic_ns': time.monotonic_ns()}
try:
raw = json.dumps(evidence, indent=2) + '\n'
if len(raw.encode()) > 32768:
raise AssertionError('control cleanup evidence exceeds 32KiB')
with (directory / 'cleanup-evidence.json').open('x') as stream:
stream.write(raw)
stream.flush()
os.fsync(stream.fileno())
except (OSError, ValueError, AssertionError) as error:
errors.append('cleanup evidence: ' + str(error)[:512])
retained = True
if retained:
print('browser control evidence retained: ' + str(directory), file=sys.stderr)
if errors:
raise AssertionError('control scratch retained at ' + str(directory) + ': ' + '; '.join(errors))
if directory is not None and body_failure is None:
shutil.rmtree(directory)
finally:
if not restore_attempted and libc.prctl(36, previous.value, 0, 0, 0):
raise RuntimeError('restore previous test subreaper setting')
def actual_browser_finally(namespace):
# Execute the actual main finally; no browser/server or main setup is invoked.
source = ast.parse(Path(browser.__file__).read_text())
main = next(node for node in source.body if isinstance(node, ast.FunctionDef) and node.name == 'main')
finalizers = [node.finalbody for node in main.body if isinstance(node, ast.Try) and node.finalbody]
if len(finalizers) != 1:
raise AssertionError('one actual browser main finalizer must supply the control')
exec(compile(ast.Module(body=finalizers[0], type_ignores=[]), browser.__file__, 'exec'), namespace)
class BrowserFixtureCleanupTests(unittest.TestCase):
def assert_zero_population(self, groups):
for item in groups:
self.assertFalse((Path('/proc') / str(item['child'])).exists(), 'actual finalizer left an unreaped owned child')
with self.assertRaises(ProcessLookupError, msg='actual finalizer did not confirm zero private-group population'):
os.killpg(item['process'].pid, 0)
def test_actual_finally_normal_empty_fixture_has_confirmed_cleanup(self):
with browser_cleanup_fixture([]) as (namespace, groups):
actual_browser_finally(namespace)
self.assertFalse(namespace['sampler'].is_alive())
self.assertFalse(namespace['work'].exists())
self.assertTrue((namespace['out'] / 'browser.json').is_file())
self.assertTrue((namespace['out'] / 'resource-samples.json').is_file())
def test_actual_finally_stops_child_when_leader_already_exited(self):
with browser_cleanup_fixture(['exited']) as (namespace, groups):
self.assertEqual(groups[0]['process'].returncode, 0)
actual_browser_finally(namespace)
self.assert_zero_population(groups)
def test_actual_finally_stops_term_resistant_child_after_leader_termination(self):
with browser_cleanup_fixture(['normal']) as (namespace, groups):
actual_browser_finally(namespace)
self.assert_zero_population(groups)
def test_actual_finally_timeout_escalation_stops_whole_group(self):
with browser_cleanup_fixture(['timeout']) as (namespace, groups):
process = groups[0]['process']
original_wait, requested = process.wait, []
def shortened_wait(timeout=None):
requested.append(timeout)
return original_wait(timeout=min(timeout, .2) if timeout is not None else None)
# TEST ONLY real wait shortening; production 15s/5s limits stay fixed.
with patch.object(process, 'wait', side_effect=shortened_wait):
actual_browser_finally(namespace)
self.assertIn(15, requested, 'the actual production TERM wait supplied the timeout path')
self.assert_zero_population(groups)
def test_actual_finally_attempts_all_groups_and_retains_scratch_after_failed_stop(self):
with browser_cleanup_fixture(['normal', 'normal']) as (namespace, groups):
attempted, real_killpg = [], os.killpg
failed = groups[-1]['process'].pid
def fail_one_group(pgid, sig):
if sig:
attempted.append(pgid)
if pgid == failed and sig:
raise PermissionError('literal owned group stop failure')
return real_killpg(pgid, sig)
with patch.object(browser.os, 'killpg', side_effect=fail_one_group):
try:
actual_browser_finally(namespace)
except (OSError, RuntimeError):
pass
self.assertEqual(set(attempted), {item['process'].pid for item in groups},
'a first stop error must not suppress other owned group attempts')
self.assertTrue(namespace['work'].is_dir(), 'unconfirmed stop must retain private fixture scratch')
self.assertTrue((Path('/proc') / str(groups[-1]['child'])).exists(), 'injected refusal must reach a real living group')
self.assertNotIn('process groups stopped', namespace['receipt'].get('cleanup', ''), 'do not claim cleanup on refusal')
def test_actual_finally_preserves_scratch_until_bounded_sampler_is_stopped(self):
with browser_cleanup_fixture([], blocked_sampler=True) as (namespace, groups):
started = time.monotonic()
try:
actual_browser_finally(namespace)
except RuntimeError:
pass
self.assertTrue(namespace['stop_samples'].is_set(), 'actual finalizer must reach sampler stop')
self.assertLess(time.monotonic() - started, 3, 'sampler refusal must remain bounded')
self.assertTrue(namespace['sampler'].is_alive(), 'literal sampler must still be alive at refusal')
self.assertTrue(namespace['work'].is_dir(), 'scratch must not be removed while sampler lives')
self.assertNotIn('process groups stopped', namespace['receipt'].get('cleanup', ''), 'unconfirmed sampler stop is not cleanup')
if __name__ == '__main__':
unittest.main()