"""Native Codex authentication with synthetic transports; never import/start the server.""" from __future__ import annotations import ast import asyncio from contextlib import asynccontextmanager import json from pathlib import Path import threading from types import SimpleNamespace import unittest from unittest.mock import AsyncMock from fastapi import FastAPI, HTTPException, Request from fastapi.testclient import TestClient import codex_auth from codex_app_server import CodexAppServerClient, CodexAppServerRequestError, CodexAppServerTimeout import test_codex_subagents_admin_isolated as admin_fixture from tests.test_codex_app_server import FakeProcessFactory, NO_RESPONSE, wait_until SOURCE = Path(__file__).with_name("agent_server.py") NAMES = {"codex_auth_operation", "reserve_codex_goals_reconfiguration", "release_codex_goals_reconfiguration", "active_codex_work_labels", "queued_turn_backend", "acquire_codex_control_thread"} nodes = [node for node in ast.parse(SOURCE.read_text()).body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in NAMES] assert {node.name for node in nodes} == NAMES OPERATION_CODE = compile(ast.Module(body=nodes, type_ignores=[]), str(SOURCE), "exec") NATIVE = {"X-AgentsDock-Token": "synthetic-native-token"} SECRET = "sk-synthetic-auth-test-only-not-a-real-key" class CodexAuthTests(unittest.IsolatedAsyncioTestCase): def setUp(self): fixture = admin_fixture.CodexSubagentsAdminTests() fixture.setUp() self.addCleanup(fixture.doCleanups) self.ns = fixture.ns self.manager = SimpleNamespace(request=AsyncMock(return_value={"type": "apiKey"}), client=SimpleNamespace(_turns_by_thread={})) self.ns.update({"asynccontextmanager": asynccontextmanager, "codex_auth": codex_auth, "CodexAppServerManager": object, "ensure_session_not_deleting": lambda session_id: None, "CODEX_AUTH_LOCK": asyncio.Lock(), "ACTIVE_LOCK": asyncio.Lock(), "QUEUE_LOCK": asyncio.Lock(), "managed_server_update_admission_blocker": lambda: None, "DEFAULT_BACKEND": "codex", "BACKEND_CODEX": "codex", "STORE": SimpleNamespace(sessions={}), "UNSAFE_HTTP_MUTATION_TASKS": {}, "UNSAFE_HTTP_MUTATION_ADMISSION_LOCK": asyncio.Lock(), "request_route_parses_body": lambda request: False, "unsafe_http_mutation_blocked_response": lambda **kwargs: None, "register_unsafe_http_mutation_locked": lambda task: ("synthetic-lease", task, {}), "release_unsafe_http_mutation": lambda *args: None, "UnsafeMutationAdmissionRejected": type("UnsafeMutationAdmissionRejected", (RuntimeError,), {}), "BUSY_SESSIONS": set(), "CURRENT_TURNS": {}, "ACTIVE": {}, "QUEUED_TURNS": {}, "RUN_NOW_TURNS": {}, "SERVER_MAINTENANCE_SESSIONS": set(), "CODEX_NATIVE_ACTION_TASKS": {}, "CODEX_PENDING_INTERACTIONS": {}, "CODEX_SUBAGENT_INDEX_LOCK": threading.RLock(), "CODEX_SUBAGENT_STATE": {}, "codex_subagent_has_live_owner": lambda thread, state: state.get("live") is True, "SIDE_QUESTIONS": SimpleNamespace(receipts={}), "codex_app_server_manager": AsyncMock(return_value=self.manager)}) self.ns["codex_app_server_managers"] = lambda: (self.manager,) exec(OPERATION_CODE, self.ns) self.app = FastAPI() self.app.middleware("http")(self.ns["require_agent_token"]) self.auth_router = codex_auth.create_router(authorize=self.ns["require_native_admin_control"], operation=self.ns["codex_auth_operation"], available=lambda: self.ns["CODEX_TRANSPORT"] != "exec") self.app.include_router(self.auth_router) self.client = TestClient(self.app) self.addCleanup(self.client.close) def login(self, body=None, headers=None): return self.client.post("/api/admin/codex/auth/api-key", headers=NATIVE if headers is None else headers, json={"api_key": SECRET} if body is None else body) def test_legacy_login_rejected_without_native_access_or_restart(self): result = self.login() self.assertEqual(result.status_code, 409, result.text) self.assertIn("Custom endpoint", result.json()["detail"]) self.assertIn("shared Codex CLI login", result.json()["detail"]) self.assertNotIn(SECRET, result.text) self.assertEqual(result.headers["cache-control"], "no-store") self.manager.request.assert_not_awaited() self.ns["codex_app_server_manager"].assert_not_awaited() self.assertFalse(self.ns["CODEX_GOALS_RECONFIGURING"]) self.ns["close_codex_app_server_manager"].assert_not_called() self.assertFalse(codex_auth.capability(available=True)["api_key_login"]) def test_status_projects_api_key_chatgpt_signed_out_and_future_mode(self): cases = [(None, "none", None), ({"type": "apiKey", "apiKey": SECRET, "email": SECRET}, "apiKey", None), ({"type": "chatgpt", "email": "test@example.invalid", "planType": "pro", "accessToken": SECRET}, "chatgpt", "test@example.invalid"), ({"type": "future", "token": SECRET}, "other", None)] for account, mode, email in cases: with self.subTest(mode=mode): self.manager.request.reset_mock() self.manager.request.return_value = {"account": account, "requiresOpenaiAuth": True, "secret": SECRET} response = self.client.get("/api/admin/codex/auth", headers=NATIVE) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.json()["auth_mode"], mode) self.assertEqual(response.json()["email"], email) self.assertNotIn(SECRET, response.text) self.manager.request.assert_awaited_once_with("account/read", {"refreshToken": False}, timeout=30.0) def test_rejected_body_never_echoes_secret_or_reaches_native(self): cases = [{}, {"api_key": SECRET, "extra": SECRET}, {"api_key": [SECRET]}, {"api_key": 123}, {"api_key": True}, {"api_key": ""}, {"api_key": SECRET + " "}, {"api_key": SECRET + "\n"}, {"api_key": SECRET + "é"}, {"api_key": SECRET * 150}, [SECRET]] for body in cases: response = self.login(body) self.assertIn(response.status_code, (409, 413), response.text) self.assertNotIn(SECRET, response.text) self.manager.request.assert_not_awaited() response = self.client.post("/api/admin/codex/auth/api-key", headers={**NATIVE, "Content-Type": "application/json"}, content='{"api_key":"' + SECRET) self.assertEqual(response.status_code, 409) self.assertNotIn(SECRET, response.text) async def test_rejected_route_never_reads_submitted_key(self): request = Request({"type": "http", "method": "POST", "path": "/api/admin/codex/auth/api-key", "headers": [(b"x-agentsdock-token", b"synthetic-native-token")], "query_string": b""}, receive=AsyncMock(side_effect=AssertionError("body read"))) endpoint = next(route.endpoint for route in self.auth_router.routes if getattr(route, "path", None) == "/api/admin/codex/auth/api-key") with self.assertRaises(HTTPException) as rejected: await endpoint(request) self.assertEqual(rejected.exception.status_code, 409) self.manager.request.assert_not_awaited() self.ns["codex_app_server_manager"].assert_not_awaited() def test_authentication_and_transport_rejected_before_body_or_provider(self): cases = [({}, 401), ({"Authorization": "Bearer synthetic-native-token"}, 401), ({**NATIVE, "Origin": "https://example.invalid"}, 403), ({**NATIVE, "Cookie": "x=1"}, 403), ({**NATIVE, "Sec-Fetch-Mode": "cors"}, 403)] for headers, status in cases: response = self.login(headers=headers) self.assertEqual(response.status_code, status, response.text) self.manager.request.reset_mock() self.assertEqual(self.client.options("/api/admin/codex/auth", headers=NATIVE).status_code, 403) self.assertEqual(self.client.get("/api/admin/codex/auth?token=synthetic-native-token").status_code, 401) response = self.client.post("/api/admin/codex/auth/api-key", headers=NATIVE, content=SECRET) self.assertEqual(response.status_code, 415) self.manager.request.assert_not_awaited() self.ns["AGENT_TOKEN"] = "" self.assertEqual(self.login().status_code, 503) async def test_framing_and_duplicate_tokens_rejected_without_body_read(self): native = (b"x-agentsdock-token", b"synthetic-native-token") json_type = (b"content-type", b"application/json") cases = [([native, native], 401), ([native, json_type], 411), ([native, json_type, (b"content-length", b"8193")], 413), ([native, json_type, (b"content-length", b"2"), (b"transfer-encoding", b"chunked")], 400)] for headers, expected in cases: request = Request({"type": "http", "method": "POST", "path": "/api/admin/codex/auth/api-key", "scheme": "http", "server": ("127.0.0.1", 1), "client": ("127.0.0.1", 2), "headers": headers, "query_string": b""}, receive=AsyncMock(side_effect=AssertionError("body read"))) next_handler = AsyncMock() response = await self.ns["require_agent_token"](request, next_handler) self.assertEqual(response.status_code, expected) next_handler.assert_not_called() def test_status_failures_are_sanitized_without_retry(self): for failure in (RuntimeError(SECRET), asyncio.TimeoutError(SECRET)): self.manager.request.reset_mock(side_effect=True) self.manager.request.side_effect = failure response = self.client.get("/api/admin/codex/auth", headers=NATIVE) self.assertEqual(response.status_code, 503) self.assertNotIn(SECRET, response.text) self.manager.request.assert_awaited_once() self.assertFalse(self.ns["CODEX_GOALS_RECONFIGURING"]) self.manager.request.side_effect = None for ack in (None, {}, {"account": SECRET, "requiresOpenaiAuth": True}): self.manager.request.return_value = ack response = self.client.get("/api/admin/codex/auth", headers=NATIVE) self.assertEqual(response.status_code, 502) self.assertNotIn(SECRET, response.text) def test_unavailable_transport_never_starts_provider(self): self.ns["CODEX_TRANSPORT"] = "exec" response = self.client.get("/api/admin/codex/auth", headers=NATIVE) self.assertFalse(response.json()["available"]) self.assertEqual(self.login().status_code, 409) self.ns["codex_app_server_manager"].assert_not_awaited() async def test_busy_active_queue_goal_side_chat_subagent_and_native_turn_are_preserved(self): session = {"backend": "codex"} self.ns["STORE"].sessions["chat"] = session busy_cases = [(self.ns["BUSY_SESSIONS"], "chat"), (self.ns["QUEUED_TURNS"], ("chat", [{"queued_id": "queued"}])), (self.ns["RUN_NOW_TURNS"], ("chat", {})), (session, ("codex_goal", {"status": "active"})), (self.ns["CODEX_SUBAGENT_STATE"], ("child", {"live": True})), (self.manager.client._turns_by_thread, ("thread", SimpleNamespace(_completed=False)))] for container, value in busy_cases: if isinstance(container, set): container.add(value) else: container[value[0]] = value[1] with self.assertRaises(HTTPException) as raised: async with self.ns["codex_auth_operation"](mutate=True): self.fail("busy operation admitted") self.assertEqual(raised.exception.status_code, 409) self.assertFalse(self.ns["CODEX_GOALS_RECONFIGURING"]) if isinstance(container, set): self.assertIn(value, container) container.remove(value) else: self.assertIs(container.pop(value[0]), value[1]) waiting = asyncio.create_task(asyncio.Event().wait()) self.ns["SIDE_QUESTIONS"].receipts[("owner", "chat", "request")] = SimpleNamespace(task=waiting) try: with self.assertRaises(HTTPException) as raised: async with self.ns["codex_auth_operation"](mutate=True): self.fail("side chat admitted") self.assertEqual(raised.exception.status_code, 409) self.assertFalse(waiting.done()) finally: waiting.cancel() await asyncio.gather(waiting, return_exceptions=True) self.manager.request.assert_not_awaited() async def test_admission_remains_reserved_until_login_settles_and_releases_on_cancellation(self): self.ns["STORE"].sessions["chat"] = {"backend": "codex"} started = asyncio.Event() async def work(): async with self.ns["codex_auth_operation"](mutate=True): started.set() await asyncio.Event().wait() task = asyncio.create_task(work()) await started.wait() self.assertTrue(self.ns["CODEX_GOALS_RECONFIGURING"]) with self.assertRaises(HTTPException) as blocked_control: await self.ns["acquire_codex_control_thread"]("chat") self.assertEqual(blocked_control.exception.status_code, 409) with self.assertRaises(HTTPException): await self.ns["reserve_codex_goals_reconfiguration"]() task.cancel() await asyncio.gather(task, return_exceptions=True) self.assertFalse(self.ns["CODEX_GOALS_RECONFIGURING"]) class CodexAuthTransportTests(unittest.IsolatedAsyncioTestCase): async def test_auth_write_timeout_does_not_retire_shared_provider_or_repeat_request(self): for method, params in (("account/read", {"refreshToken": False}), ("account/login/start", {"type": "apiKey", "apiKey": SECRET})): factory = FakeProcessFactory() client = CodexAppServerClient("synthetic-codex", cwd="/tmp", env_factory=lambda: {}, process_factory=factory, request_timeout=1) self.addAsyncCleanup(client.close) await client.start() async def blocked_drain(): await asyncio.Event().wait() factory.process.stdin.drain = blocked_drain with self.assertRaises(CodexAppServerTimeout): await client.request(method, params, timeout=0.01) self.assertIsNone(factory.process.returncode) self.assertTrue(client.ready) self.assertEqual(len(factory.calls), 1) self.assertEqual(len([item for item in factory.process.messages if item.get("method") == method]), 1) async def test_auth_errors_notifications_and_delayed_stderr_do_not_retain_secret(self): factory = FakeProcessFactory() client = CodexAppServerClient("synthetic-codex", cwd="/tmp", env_factory=lambda: {}, process_factory=factory, request_timeout=1) self.addAsyncCleanup(client.close) seen = [] client.add_notification_handler(lambda item: seen.append(item)) def reject(message): factory.process.feed_stderr(SECRET) factory.process.feed({"method": "account/login/completed", "params": {"error": SECRET}}) factory.process.feed({"id": message["id"], "error": {"code": -1, "message": SECRET, "data": SECRET}}) return NO_RESPONSE factory.process.responders["account/login/start"] = reject with self.assertRaises(CodexAppServerRequestError) as raised: await client.request("account/login/start", {"type": "apiKey", "apiKey": SECRET}) self.assertNotIn(SECRET, str(raised.exception)) self.assertNotIn(SECRET, repr(raised.exception.error)) factory.process.feed_stderr("delayed " + SECRET) await asyncio.sleep(0) await asyncio.sleep(0) self.assertEqual(client.stderr_tail, []) self.assertEqual(seen, []) rate_limits = {"method": "account/rateLimits/updated", "params": {"rateLimits": {"limitId": "codex"}}} client._route_notification(rate_limits) self.assertEqual(seen, [rate_limits]) self.assertEqual(len([item for item in factory.process.messages if item.get("method") == "account/login/start"]), 1) if __name__ == "__main__": unittest.main()