Full-access profile switch retains managed proxy but drops named-profile domain allowlist

import json
import os
import pathlib
import queue
import re
import socket
import subprocess
import tempfile
import threading
import time
import sys


def toml(value):
    if isinstance(value, dict):
        return '{' + ', '.join(json.dumps(k) + ' = ' + toml(v) for k, v in value.items()) + '}'
    if isinstance(value, bool):
        return str(value).lower()
    return json.dumps(value)


diagnostic_home = pathlib.Path(tempfile.mkdtemp(prefix='codex-profile-transition-', dir='/private/tmp'))
work = diagnostic_home / 'work'
work.mkdir()
fixture = work / 'denied-test-file'
fixture.write_text('Harmless diagnostic fixture.\n')
denies = {str(fixture): 'deny'}
domains = {'github.com': 'allow'}
with socket.socket() as reserved:
    reserved.bind(('127.0.0.1', 0))
    port = reserved.getsockname()[1]
config = {
    'model': 'diagnostic-no-inference',
    'model_provider': 'diagnostic',
    'default_permissions': 'governed',
    'approval_policy': 'on-request',
    'features': {'network_proxy': {'enabled': True, 'proxy_url': f'http://127.0.0.1:{port}', 'enable_socks5': False}},
    'permissions': {'governed': {'extends': ':workspace', 'filesystem': denies, 'network': {'enabled': True, 'domains': domains}}},
    'model_providers': {'diagnostic': {'name': 'No inference diagnostic', 'base_url': 'http://127.0.0.1:1/v1', 'wire_api': 'responses', 'requires_openai_auth': False}},
}
(diagnostic_home / 'config.toml').write_text('\n'.join(k + ' = ' + toml(v) for k, v in config.items()) + '\n')
child_env = os.environ.copy()
child_env['CODEX_HOME'] = str(diagnostic_home)
inbox = queue.Queue()
events = []
results = {'diagnostic_home': str(diagnostic_home), 'proxy_port': port, 'deny_count': len(denies), 'steps': []}
stderr = (diagnostic_home / 'stderr.log').open('w')
process = subprocess.Popen([sys.argv[1] if len(sys.argv) > 1 else 'codex', 'app-server', '--stdio'], cwd=work, env=child_env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=stderr, text=True)


def reader():
    for line in process.stdout:
        try:
            inbox.put(json.loads(line))
        except ValueError:
            pass


threading.Thread(target=reader, daemon=True).start()
counter = 0


def rpc(method, params):
    global counter
    counter += 1
    request_id = counter
    process.stdin.write(json.dumps({'id': request_id, 'method': method, 'params': params}) + '\n')
    process.stdin.flush()
    deadline = time.monotonic() + 35
    while time.monotonic() < deadline:
        event = inbox.get(timeout=max(0.1, deadline - time.monotonic()))
        if event.get('id') == request_id:
            if 'error' in event:
                raise RuntimeError(json.dumps(event['error']))
            return event.get('result')
        events.append(event)
    raise TimeoutError(method)


def probe(label):
    deadline = time.monotonic() + 12
    while True:
        try:
            listeners = subprocess.run(['/usr/sbin/lsof', '-a', '-p', str(process.pid), '-iTCP', '-sTCP:LISTEN', '-Pn'], capture_output=True, text=True, timeout=5)
            candidates = re.findall(r'TCP 127\.0\.0\.1:(\d+) \(LISTEN\)', listeners.stdout)
            if len(candidates) != 1:
                raise ConnectionRefusedError('Expected exactly one diagnostic process loopback listener')
            active_port = int(candidates[0])
            connection = socket.create_connection(('127.0.0.1', active_port), timeout=2)
            break
        except ConnectionRefusedError:
            if time.monotonic() >= deadline:
                listeners = subprocess.run(['/usr/sbin/lsof', '-a', '-p', str(process.pid), '-iTCP', '-sTCP:LISTEN', '-Pn'], capture_output=True, text=True, timeout=5)
                results['listeners'] = listeners.stdout
                raise
            time.sleep(0.1)
    with connection:
        connection.settimeout(8)
        connection.sendall(b'CONNECT github.com:443 HTTP/1.1\r\nHost: github.com:443\r\n\r\n')
        response = connection.recv(4096).decode('utf-8', 'replace')
    result = {'label': label, 'http_status': response.splitlines()[0], 'actual_proxy_port': active_port}
    results['steps'].append(result)
    print(json.dumps(result), flush=True)


def wait_settings(profile):
    deadline = time.monotonic() + 20
    def matches(event):
        settings = event.get('params', {}).get('threadSettings', {})
        return event.get('method') == 'thread/settings/updated' and settings.get('activePermissionProfile', {}).get('id') == profile
    if any(matches(event) for event in events):
        return
    while time.monotonic() < deadline:
        event = inbox.get(timeout=max(0.1, deadline - time.monotonic()))
        events.append(event)
        if matches(event):
            return
    raise TimeoutError('settings notification ' + profile)


try:
    initialization = rpc('initialize', {'clientInfo': {'name': 'isolated_policy_diagnostic', 'version': '1'}, 'capabilities': {'experimentalApi': True, 'requestAttestation': False}})
    results['initialize'] = initialization
    process.stdin.write(json.dumps({'method': 'initialized'}) + '\n')
    process.stdin.flush()
    effective = rpc('config/read', {'cwd': str(work)})
    results['effective_config'] = {k: v for k, v in effective.get('config', {}).items() if k in ['features', 'permissions', 'default_permissions', 'sandbox_mode']}
    started = rpc('thread/start', {'cwd': str(work), 'permissions': 'governed', 'ephemeral': True, 'experimentalRawEvents': True})
    thread_id = started['thread']['id']
    results['start_policy'] = {k: started.get(k) for k in ['sandbox', 'activePermissionProfile', 'approvalPolicy']}
    rpc('thread/shellCommand', {'threadId': thread_id, 'command': '/usr/bin/true', 'timeoutMs': 1000})
    try:
        probe('governed_before')
    except ConnectionRefusedError:
        results['proxy_probe_unavailable'] = True
    rpc('thread/settings/update', {'threadId': thread_id, 'permissions': ':danger-full-access'})
    wait_settings(':danger-full-access')
    if not results.get('proxy_probe_unavailable'):
        probe('full_access_after_switch')
    rpc('thread/settings/update', {'threadId': thread_id, 'permissions': 'governed'})
    wait_settings('governed')
    if not results.get('proxy_probe_unavailable'):
        probe('governed_restored')
except Exception as error:
    results['error'] = str(error)
    print(json.dumps({'error': str(error)}), flush=True)
finally:
    process.terminate()
    try:
        process.wait(timeout=5)
    except subprocess.TimeoutExpired:
        process.kill()
        process.wait(timeout=5)
    stderr.close()
    while not inbox.empty():
        events.append(inbox.get_nowait())
    results['settings_events'] = [e for e in events if e.get('method') == 'thread/settings/updated']
    results['notification_methods'] = sorted({e.get('method', '') for e in events})
    (diagnostic_home / 'result.json').write_text(json.dumps(results, indent=2))
    print(json.dumps({'result_path': str(diagnostic_home / 'result.json')}), flush=True)