Files
ARCADIA-Studio/harness/experiments/navigation-prerelease-native-20260911/payload/close_readonly.py
T

88 lines
4.7 KiB
Python

"""Separate read-only process/package closure; never launches or stops anything.
Original helper.py remains unchanged. This avoids its post-exit image-path query.
Run only after root has observed the relevant GUI window absent.
"""
import argparse
import ctypes
from ctypes import wintypes
import hashlib
from pathlib import Path
import sys
BASE = Path(__file__).resolve().parent
EXPECTED_HELPER = '11e1ac5379c28e17b8a9f5b883ddf0640994eea34de257f4e66efe6033190877'
def process_state(pid, creation):
api = ctypes.WinDLL('kernel32', use_last_error=True)
api.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
api.OpenProcess.restype = wintypes.HANDLE
api.CloseHandle.argtypes = [wintypes.HANDLE]
api.CloseHandle.restype = wintypes.BOOL
api.GetProcessTimes.argtypes = [wintypes.HANDLE] + [ctypes.POINTER(wintypes.FILETIME)] * 4
api.GetProcessTimes.restype = wintypes.BOOL
api.GetExitCodeProcess.argtypes = [wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD)]
api.GetExitCodeProcess.restype = wintypes.BOOL
handle = api.OpenProcess(0x1000, False, pid)
if not handle:
error = ctypes.get_last_error()
return {'originalProcessAbsent': error == 87, 'openProcessError': error,
'reason': 'PID is absent' if error == 87 else 'Process query refused; absence unproven'}
try:
created, exited, kernel, user = (wintypes.FILETIME() for _ in range(4))
if not api.GetProcessTimes(handle, ctypes.byref(created), ctypes.byref(exited), ctypes.byref(kernel), ctypes.byref(user)):
return {'originalProcessAbsent': False, 'getProcessTimesError': ctypes.get_last_error()}
actual_creation = (created.dwHighDateTime << 32) | created.dwLowDateTime
if actual_creation != creation:
return {'originalProcessAbsent': True, 'pidReused': True, 'currentCreationFiletime': actual_creation,
'reason': 'PID now belongs to a different process creation'}
code = wintypes.DWORD()
if not api.GetExitCodeProcess(handle, ctypes.byref(code)):
return {'originalProcessAbsent': False, 'getExitCodeProcessError': ctypes.get_last_error()}
return {'originalProcessAbsent': code.value != 259, 'pidReused': False,
'currentCreationFiletime': actual_creation, 'exitCodeQueried': code.value,
'reason': 'Original process is still active' if code.value == 259 else 'Original process has terminated'}
finally:
api.CloseHandle(handle)
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--observed-absent', action='store_true')
args = parser.parse_args()
assert args.observed_absent, 'Root must observe the GUI window absent first.'
# Import only the exact reviewed, guarded module. Its main action dispatch is
# not executed by import; we call no launch/oracle/original closure function.
assert hashlib.sha256((BASE / 'helper.py').read_bytes()).hexdigest() == EXPECTED_HELPER
sys.dont_write_bytecode = True
import helper as original
result_path = original.TRIAL / 'closed-readonly.json'
process_path = original.TRIAL / 'closed-readonly-process.json'
package_path = original.TRIAL / 'after-close-readonly-package.json'
assert not any(p.exists() for p in (result_path, process_path, package_path)), 'Preserve previous closure evidence.'
result = {'started': original.now(), 'status': 'PENDING', 'original_helper_sha256': EXPECTED_HELPER,
'scope': 'Root-observed window absence, read-only process state and exact bound ZIP/package map; no GUI or oracle rerun.'}
try:
launch = original.load(original.TRIAL / 'launch.json')
identity = original.load(original.TRIAL / 'process-identity.json')
state = process_state(launch['pid'], identity['creation_filetime'])
original.record(process_path, {'atUtc': original.now(), 'pid': launch['pid'], 'originalCreationFiletime': identity['creation_filetime'], **state})
assert state['originalProcessAbsent'], state
original.package_guard('after-close-readonly')
result.update(status='PASS', pid=launch['pid'], originalProcessAbsent=True, completePackageUnchanged=True,
process_receipt_sha256=original.sha(process_path), package_receipt_sha256=original.sha(package_path))
except BaseException as error:
result.update(status='FAIL', error=repr(error))
raise
finally:
result['finished'] = original.now()
original.record(result_path, result)
print('PASS: original process absent; bound ZIP and complete package unchanged.')
if __name__ == '__main__':
sys.stdout.reconfigure(encoding='utf-8')
sys.stderr.reconfigure(encoding='utf-8')
main()