Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/client/cli/commands/codex_settings.py: 0%
186 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1import hashlib
2import json
3import re
4import subprocess
5from collections.abc import Callable, Mapping
6from dataclasses import dataclass
7from functools import reduce
8from pathlib import Path
9from types import MappingProxyType
10from typing import Final, Literal, TypeAlias
12import tomlkit
13from pydantic import BaseModel, ConfigDict, ValidationError
14from tomlkit.container import OutOfOrderTableProxy
15from tomlkit.exceptions import TOMLKitError
16from tomlkit.items import InlineTable, Table
17from tomlkit.toml_document import TOMLDocument
19from litellm.litellm_core_utils.private_json import (
20 commit_staged_json,
21 discard_staged_json,
22 ensure_private_dir,
23 stage_private_bytes,
24 stage_private_json,
25)
27from .agents import CODEX_PROXY_PROVIDER, codex_proxy_provider
29_PROVIDER_PATH: Final = f"model_providers.{CODEX_PROXY_PROVIDER}"
30_OWNED_PATHS: Final = ("model_provider", "model", "profile", _PROVIDER_PATH)
31_Table: TypeAlias = TOMLDocument | Table | InlineTable | OutOfOrderTableProxy
32_EMPTY: Final[Mapping[str, object]] = MappingProxyType({})
33_MIN_CODEX_VERSION: Final = (0, 129, 0)
36class CodexSettingsError(Exception):
37 pass
40class _Receipt(BaseModel):
41 model_config = ConfigDict(frozen=True, extra="forbid")
43 version: Literal[1] = 1
44 settings_path: str
45 file_existed: bool
46 providers_existed: bool
47 previous: Mapping[str, str | None]
48 written: Mapping[str, str]
51@dataclass(frozen=True, slots=True)
52class CodexUnconfigureOutcome:
53 restored: tuple[str, ...]
54 kept: tuple[str, ...]
55 file_removed: bool
58def codex_configure_state_path(settings_path: Path) -> Path:
59 target: Final = settings_path.resolve()
60 digest: Final = hashlib.sha256(str(target).encode()).hexdigest()
61 return target.parent / ".litellm" / f"codex_configure_{digest}.json"
64def _read(settings_path: Path) -> TOMLDocument:
65 try:
66 document: Final = tomlkit.parse(settings_path.read_bytes()) if settings_path.exists() else tomlkit.document()
67 except (OSError, UnicodeError, TOMLKitError) as error:
68 raise CodexSettingsError(
69 f"Could not read Codex settings at {settings_path}; no settings were changed"
70 ) from error
71 providers: Final = _mapping(document).get("model_providers")
72 parent: Final = _table(providers)
73 if providers is not None and parent is None:
74 raise CodexSettingsError("Codex model_providers must be a TOML table; no settings were changed")
75 entries: Final = _mapping(parent) if parent is not None else _EMPTY
76 configured: Final = entries.get(CODEX_PROXY_PROVIDER)
77 if configured is not None and _table(configured) is None:
78 raise CodexSettingsError("Codex model_providers.litellm must be a TOML table; no settings were changed")
79 return document
82def _mapping(value: Mapping[str, object]) -> Mapping[str, object]:
83 return value
86def _table(value: object) -> _Table | None:
87 return value if isinstance(value, (TOMLDocument, Table, InlineTable, OutOfOrderTableProxy)) else None
90def _snapshot(document: TOMLDocument, path: str) -> str | None:
91 section, _, key = path.rpartition(".")
92 parent: Final = _table(_mapping(document).get(section)) if section else document
93 if parent is None or key not in parent:
94 return None
95 values: Final = _mapping(parent)
96 return tomlkit.dumps(MappingProxyType({"value": values[key]}))
99def _fingerprint(value: str | None) -> str:
100 normalized: Final = "missing" if value is None else json.dumps(tomlkit.parse(value), sort_keys=True, default=str)
101 return hashlib.sha256(normalized.encode()).hexdigest()
104def _with(document: TOMLDocument, path: str, snapshot: str | None) -> TOMLDocument:
105 section, _, key = path.rpartition(".")
106 if section and section not in document and snapshot is not None:
107 contents: Final = tomlkit.parse(tomlkit.dumps(MappingProxyType({key: tomlkit.parse(snapshot).item("value")})))
108 return tomlkit.parse(document.as_string() + "\n" + tomlkit.dumps(MappingProxyType({section: contents})))
109 updated: Final = tomlkit.parse(document.as_string())
110 parent: Final = _table(_mapping(updated).get(section)) if section else updated
111 if parent is None:
112 return updated
113 if snapshot is None:
114 if key in parent:
115 del parent[key]
116 else:
117 parent[key] = tomlkit.parse(snapshot).item("value")
118 return updated
121def _receipt(settings_path: Path) -> _Receipt | None:
122 path: Final = codex_configure_state_path(settings_path)
123 if not path.exists():
124 return None
125 try:
126 receipt: Final = _Receipt.model_validate_json(path.read_bytes())
127 if receipt.settings_path != str(settings_path.resolve()) or frozenset(receipt.previous) != frozenset(
128 receipt.written
129 ):
130 raise ValueError("invalid receipt scope")
131 if not frozenset(receipt.written) <= frozenset(_OWNED_PATHS):
132 raise ValueError("invalid receipt ownership")
133 for snapshot in receipt.previous.values():
134 if snapshot is not None and tuple(tomlkit.parse(snapshot)) != ("value",):
135 raise ValueError("invalid receipt snapshot")
136 except (OSError, UnicodeError, TOMLKitError, ValidationError, ValueError) as error:
137 raise CodexSettingsError(
138 f"Could not read the Codex configure receipt at {path}; no settings were changed"
139 ) from error
140 return receipt
143def _codex_version() -> str | None:
144 try:
145 result: Final = subprocess.run(("codex", "--version"), capture_output=True, text=True, timeout=5, check=False)
146 except (OSError, subprocess.SubprocessError, UnicodeError):
147 return None
148 return result.stdout if result.returncode == 0 else None
151def require_safe_codex(*, version: Callable[[], str | None] = _codex_version) -> None:
152 output: Final = version()
153 matched: Final = re.fullmatch(r"codex-cli (\d+)\.(\d+)\.(\d+)", output.strip()) if output is not None else None
154 if matched is not None and tuple(int(part) for part in matched.groups()) >= _MIN_CODEX_VERSION:
155 return
156 raise CodexSettingsError(
157 "Codex 0.129.0 or newer (stable) must be installed before saving a gateway key. "
158 "Older versions allow repository settings to redirect authenticated requests. "
159 "Install or update Codex, check `codex --version`, then retry."
160 )
163def preflight_codex_settings(settings_path: Path) -> None:
164 require_safe_codex()
165 _read(settings_path)
166 _receipt(settings_path)
169def _ours(document: TOMLDocument, path: str, receipt: _Receipt) -> bool:
170 return receipt.written.get(path) == _fingerprint(_snapshot(document, path))
173def _stage_settings(path: Path, document: TOMLDocument) -> str:
174 try:
175 return stage_private_bytes(str(path), document.as_string().encode())
176 except OSError as error:
177 raise CodexSettingsError(f"Could not stage Codex settings at {path}; no settings were changed") from error
180def _commit(path: Path, staged: str | None, commit: Callable[[str, str], None]) -> None:
181 if staged is None:
182 path.unlink(missing_ok=True)
183 else:
184 commit(staged, str(path))
187def configure_codex_settings(
188 base_url: str,
189 api_key: str,
190 model: str,
191 settings_path: Path,
192 *,
193 commit: Callable[[str, str], None] = commit_staged_json,
194) -> None:
195 require_safe_codex()
196 current: Final = _read(settings_path)
197 earlier: Final = _receipt(settings_path)
198 headers: Final = tomlkit.parse(tomlkit.dumps(MappingProxyType({"Authorization": f"Bearer {api_key}"})))
199 provider_table: Final = tomlkit.parse(
200 tomlkit.dumps(MappingProxyType({**codex_proxy_provider(base_url), "http_headers": headers}))
201 )
202 provider: Final = tomlkit.dumps(MappingProxyType({"value": provider_table}))
203 selections: Final = tomlkit.parse(
204 tomlkit.dumps(MappingProxyType({"model_provider": CODEX_PROXY_PROVIDER, "model": model}))
205 )
206 merged: Final = _with(
207 _with(
208 _with(_with(current, "profile", None), "model", _snapshot(selections, "model")),
209 "model_provider",
210 _snapshot(selections, "model_provider"),
211 ),
212 _PROVIDER_PATH,
213 provider,
214 )
215 owned: Final = tuple(
216 path
217 for path in _OWNED_PATHS
218 if _fingerprint(_snapshot(current, path)) != _fingerprint(_snapshot(merged, path))
219 or (earlier is not None and _ours(current, path, earlier))
220 )
221 receipt: Final = _Receipt(
222 settings_path=str(settings_path.resolve()),
223 file_existed=settings_path.exists() if earlier is None else earlier.file_existed,
224 providers_existed="model_providers" in current if earlier is None else earlier.providers_existed,
225 previous=MappingProxyType(
226 {
227 path: earlier.previous[path]
228 if earlier is not None and _ours(current, path, earlier)
229 else _snapshot(current, path)
230 for path in owned
231 }
232 ),
233 written=MappingProxyType({path: _fingerprint(_snapshot(merged, path)) for path in owned}),
234 )
235 target: Final = settings_path.resolve()
236 state_path: Final = codex_configure_state_path(settings_path)
237 try:
238 ensure_private_dir(state_path.parent)
239 staged_receipt: Final = stage_private_json(str(state_path), receipt.model_dump(mode="json"))
240 except OSError as error:
241 raise CodexSettingsError(f"Could not stage the Codex configure receipt at {state_path}") from error
242 try:
243 staged_settings: Final = _stage_settings(target, merged)
244 except CodexSettingsError:
245 discard_staged_json(staged_receipt)
246 raise
247 try:
248 commit(staged_receipt, str(state_path))
249 except OSError as error:
250 discard_staged_json(staged_receipt)
251 discard_staged_json(staged_settings)
252 raise CodexSettingsError(
253 f"Could not write the Codex configure receipt at {state_path}; no settings were changed"
254 ) from error
255 try:
256 commit(staged_settings, str(target))
257 except OSError as error:
258 discard_staged_json(staged_settings)
259 try:
260 _commit(
261 state_path,
262 None if earlier is None else stage_private_json(str(state_path), earlier.model_dump(mode="json")),
263 commit_staged_json,
264 )
265 except OSError as rollback_error:
266 raise CodexSettingsError(
267 f"Codex settings were not written and its receipt at {state_path} could not be restored"
268 ) from rollback_error
269 raise CodexSettingsError(
270 f"Could not write Codex settings at {settings_path}; the earlier receipt was restored"
271 ) from error
274def unconfigure_codex_settings(
275 settings_path: Path, *, commit: Callable[[str, str], None] = commit_staged_json
276) -> CodexUnconfigureOutcome:
277 current: Final = _read(settings_path)
278 receipt: Final = _receipt(settings_path)
279 if receipt is None:
280 raise CodexSettingsError("Codex is not configured by `lite configure codex`; nothing to undo")
281 ours: Final = tuple(path for path in receipt.written if settings_path.exists() and _ours(current, path, receipt))
282 restored_owned: Final = reduce(lambda document, path: _with(document, path, receipt.previous[path]), ours, current)
283 providers: Final = _table(_mapping(restored_owned).get("model_providers"))
284 restored: Final = (
285 _with(restored_owned, "model_providers", None)
286 if providers is not None and not providers and not receipt.providers_existed
287 else restored_owned
288 )
289 target: Final = settings_path.resolve()
290 file_removed: Final = not restored.as_string().strip() and not (receipt.file_existed and target.exists())
291 staged: Final = None if file_removed else _stage_settings(target, restored)
292 state_path: Final = codex_configure_state_path(settings_path)
293 try:
294 _commit(target, staged, commit)
295 state_path.unlink()
296 except OSError as error:
297 if staged is not None:
298 discard_staged_json(staged)
299 raise CodexSettingsError(
300 "Could not finish undoing Codex configuration; the receipt was kept for retry"
301 ) from error
302 return CodexUnconfigureOutcome(
303 restored=tuple(path for path in ours if _snapshot(current, path) != _snapshot(restored, path)),
304 kept=tuple(path for path in receipt.written if path not in ours and _snapshot(current, path) is not None),
305 file_removed=file_removed,
306 )