feat(codex): inject host credentials via egress

tests/unit/test_codex_auth.py

@@ -0,0 +1,83 @@
+"""Unit: host Codex auth extraction."""
+
+from __future__ import annotations
+
+import base64
+import json
+import tempfile
+import unittest
+from datetime import datetime, timezone
+from pathlib import Path
+
+from bot_bottle.codex_auth import codex_auth_path, codex_host_access_token
+from bot_bottle.log import Die
+
+
+def _jwt(exp: int) -> str:
+    def enc(obj: dict) -> str:
+        raw = json.dumps(obj, separators=(",", ":")).encode()
+        return base64.urlsafe_b64encode(raw).decode().rstrip("=")
+    return f"{enc({'alg': 'none'})}.{enc({'exp': exp})}.sig"
+
+
+class TestCodexHostAccessToken(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="cb-codex-auth.")
+        self.home = Path(self.tmp.name)
+        self.auth_path = self.home / "auth.json"
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _write(self, payload: dict) -> None:
+        self.auth_path.write_text(json.dumps(payload))
+
+    def test_auth_path_uses_codex_home(self):
+        self.assertEqual(
+            self.auth_path,
+            codex_auth_path({"CODEX_HOME": str(self.home)}),
+        )
+
+    def test_returns_fresh_chatgpt_access_token(self):
+        token = _jwt(2000000000)
+        self._write({
+            "auth_mode": "chatgpt",
+            "tokens": {"access_token": token, "refresh_token": "hidden"},
+        })
+        out = codex_host_access_token(
+            {"CODEX_HOME": str(self.home)},
+            now=datetime(2026, 1, 1, tzinfo=timezone.utc),
+        )
+        self.assertEqual(token, out)
+
+    def test_missing_auth_file_dies(self):
+        with self.assertRaises(Die):
+            codex_host_access_token({"CODEX_HOME": str(self.home)})
+
+    def test_non_chatgpt_auth_dies(self):
+        self._write({"auth_mode": "api_key", "tokens": {"access_token": _jwt(2)}})
+        with self.assertRaises(Die):
+            codex_host_access_token({"CODEX_HOME": str(self.home)})
+
+    def test_expired_token_dies(self):
+        self._write({
+            "auth_mode": "chatgpt",
+            "tokens": {"access_token": _jwt(1000)},
+        })
+        with self.assertRaises(Die):
+            codex_host_access_token(
+                {"CODEX_HOME": str(self.home)},
+                now=datetime(2026, 1, 1, tzinfo=timezone.utc),
+            )
+
+    def test_non_jwt_token_dies(self):
+        self._write({
+            "auth_mode": "chatgpt",
+            "tokens": {"access_token": "not-a-jwt"},
+        })
+        with self.assertRaises(Die):
+            codex_host_access_token({"CODEX_HOME": str(self.home)})
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_egress.py

@@ -4,6 +4,7 @@ resolution (PRD 0017)."""
 import unittest
 
 from bot_bottle.egress import (
+    CODEX_HOST_CREDENTIAL_TOKEN_REF,
     egress_manifest_routes,
     egress_render_routes,
     egress_resolve_token_values,
@@ -22,6 +23,21 @@ def _bottle(routes):
     }).bottles["dev"]
 
 
+def _codex_bottle(*, forward_host_credentials: bool, routes):
+    return Manifest.from_json_obj({
+        "bottles": {
+            "dev": {
+                "agent_provider": {
+                    "template": "codex",
+                    "forward_host_credentials": forward_host_credentials,
+                },
+                "egress": {"routes": routes},
+            }
+        },
+        "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+    }).bottles["dev"]
+
+
 class TestRoutesForBottle(unittest.TestCase):
     def test_authenticated_route_gets_slot(self):
         b = _bottle([{
@@ -107,6 +123,38 @@ class TestRoutesForBottleUsesManifestOnly(unittest.TestCase):
         effective = [r.host for r in egress_routes_for_bottle(b)]
         self.assertEqual(["x.example"], effective)
 
+    def test_codex_forward_host_credentials_adds_chatgpt_route(self):
+        b = _codex_bottle(forward_host_credentials=True, routes=[])
+        routes = egress_routes_for_bottle(b)
+        self.assertEqual(["chatgpt.com"], [r.host for r in routes])
+        self.assertEqual("Bearer", routes[0].auth_scheme)
+        self.assertEqual("EGRESS_TOKEN_0", routes[0].token_env)
+        self.assertEqual(CODEX_HOST_CREDENTIAL_TOKEN_REF, routes[0].token_ref)
+
+    def test_codex_forward_host_credentials_upgrades_bare_chatgpt_route(self):
+        b = _codex_bottle(
+            forward_host_credentials=True,
+            routes=[{"host": "chatgpt.com", "path_allowlist": ["/backend-api/"]}],
+        )
+        routes = egress_routes_for_bottle(b)
+        self.assertEqual(1, len(routes))
+        self.assertEqual("chatgpt.com", routes[0].host)
+        self.assertEqual("Bearer", routes[0].auth_scheme)
+        self.assertEqual("EGRESS_TOKEN_0", routes[0].token_env)
+        self.assertEqual(CODEX_HOST_CREDENTIAL_TOKEN_REF, routes[0].token_ref)
+        self.assertEqual(("/backend-api/",), routes[0].path_allowlist)
+
+    def test_codex_forward_host_credentials_conflicts_with_authed_route(self):
+        b = _codex_bottle(
+            forward_host_credentials=True,
+            routes=[{
+                "host": "chatgpt.com",
+                "auth": {"scheme": "Bearer", "token_ref": "OTHER"},
+            }],
+        )
+        with self.assertRaises(Die):
+            egress_routes_for_bottle(b)
+
 
 class TestTokenEnvMap(unittest.TestCase):
     def test_only_authenticated_routes_contribute(self):
@@ -217,6 +265,13 @@ class TestResolveTokenValues(unittest.TestCase):
                 {"GH_PAT": ""},
             )
 
+    def test_codex_host_credential_ref_is_resolved_by_launch(self):
+        out = egress_resolve_token_values(
+            {"EGRESS_TOKEN_0": CODEX_HOST_CREDENTIAL_TOKEN_REF},
+            {},
+        )
+        self.assertEqual({}, out)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_manifest_egress.py

@@ -29,6 +29,13 @@ def _provider_bottle(provider, routes):
     }).bottles["dev"]
 
 
+def _provider_config_bottle(agent_provider):
+    return Manifest.from_json_obj({
+        "bottles": {"dev": {"agent_provider": agent_provider}},
+        "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+    }).bottles["dev"]
+
+
 class TestMinimalRoute(unittest.TestCase):
     def test_host_only(self):
         b = _bottle([{"host": "api.example.com"}])
@@ -52,6 +59,33 @@ class TestMinimalRoute(unittest.TestCase):
             _bottle([{"host": "x.example", "wat": "yes"}])
 
 
+class TestAgentProviderHostCredentials(unittest.TestCase):
+    def test_forward_host_credentials_defaults_false(self):
+        b = _provider_config_bottle({"template": "codex"})
+        self.assertFalse(b.agent_provider.forward_host_credentials)
+
+    def test_forward_host_credentials_allowed_for_codex(self):
+        b = _provider_config_bottle({
+            "template": "codex",
+            "forward_host_credentials": True,
+        })
+        self.assertTrue(b.agent_provider.forward_host_credentials)
+
+    def test_forward_host_credentials_must_be_boolean(self):
+        with self.assertRaises(ManifestError):
+            _provider_config_bottle({
+                "template": "codex",
+                "forward_host_credentials": "yes",
+            })
+
+    def test_forward_host_credentials_rejected_for_claude(self):
+        with self.assertRaises(ManifestError):
+            _provider_config_bottle({
+                "template": "claude",
+                "forward_host_credentials": True,
+            })
+
+
 class TestPathAllowlist(unittest.TestCase):
     def test_optional(self):
         b = _bottle([{"host": "x.example"}])