openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-08-23 13:09:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add CIMD support to MCP OAuth registration (#38089)

Browse Source 
## What changed

- Make automatic MCP OAuth registration prefer Client ID Metadata Documents
  (CIMD) when the authorization server advertises support for public clients and
  Codex is using its native loopback callback. Fall back to advertised Dynamic
  Client Registration (DCR) otherwise.
- Add explicit `cimd` and `dcr` registration overrides to the CLI and app-server
  OAuth login API. Validate CIMD metadata and callback URLs before starting the
  authorization flow.
- Use a callback-specific Codex client metadata URL for CIMD and retain the
  exact redirect URI through authorization and token exchange.

## Testing

- Cover automatic and forced CIMD selection, DCR fallback, invalid metadata and
  redirects, token refresh, authenticated MCP requests, and conformance
  regression checks.

GitOrigin-RevId: 4238372ca53b0f38e781e141ab5da97e0a6ddf45
This commit is contained in:
stevenlee-oai
2026-08-11 23:55:22 +00:00
committed by copyberry
parent 85f331772f
commit 4c89139da9
21 changed files with 822 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
scripts/mcp_conformance/codex_conformance_adapter.py
View File

@@ -43,6 +43,7 @@ class AdapterFailure(RuntimeError):
CIMD_CLIENT_METADATA_URL = "https://conformance-test.local/client-metadata.json"
PRE_REGISTERED_CLIENT_SECRET_ENV_VAR = "MCP_CONFORMANCE_CLIENT_SECRET"
CLIENT_REGISTRATION_OVERRIDE_ENV_VAR = "CODEX_CONFORMANCE_CLIENT_REGISTRATION"
AUTH_COMPLETION_METHOD = "mcpServer/oauthLogin/completed"
EXPECTED_AUTH_REJECTION_SCENARIOS = frozenset(
{
@@ -303,11 +304,9 @@ def _drive_headless_authorization(
def _oauth_client_id(
scenario: str,
context: Mapping[str, object],
*,
require_production_client_identity: bool = False,
) -> str | None:
if scenario == "auth/basic-cimd":
return None if require_production_client_identity else CIMD_CLIENT_METADATA_URL
return CIMD_CLIENT_METADATA_URL
if scenario == "auth/pre-registration":
client_id = context.get("client_id")
if not isinstance(client_id, str) or not client_id:
@@ -316,6 +315,23 @@ def _oauth_client_id(
return None
def _client_registration_override(
scenario: str,
*,
require_automatic_auth: bool,
) -> str | None:
requested = os.environ.get(CLIENT_REGISTRATION_OVERRIDE_ENV_VAR)
if requested:
if requested not in {"auto", "cimd", "dcr"}:
raise AdapterFailure(
f"{CLIENT_REGISTRATION_OVERRIDE_ENV_VAR} must be auto, cimd, or dcr"
)
return requested
if scenario == "auth/offline-access-scope" and not require_automatic_auth:
return "dcr"
return None
def _write_auth_registration(
config_path: Path,
*,
@@ -359,6 +375,7 @@ def _oauth_login(
*,
scopes: Sequence[str] | None,
timeout_seconds: float,
client_registration: str | None = None,
) -> tuple[bool, str | None]:
event_index = len(client.events)
params: dict[str, object] = {
@@ -367,6 +384,8 @@ def _oauth_login(
}
if scopes is not None:
params["scopes"] = list(scopes)
if client_registration is not None:
params["clientRegistration"] = client_registration
response = client.request("mcpServer/oauth/login", params)
result, detail = _response_result(response)
if result is None:
@@ -424,11 +443,13 @@ def _login_reload_and_call(
workspace: Path,
timeout_seconds: float,
scopes: Sequence[str] | None = None,
client_registration: str | None = None,
) -> None:
success, error = _oauth_login(
client,
scopes=scopes,
timeout_seconds=timeout_seconds,
client_registration=client_registration,
)
if not success:
raise AdapterFailure(f"OAuth login failed: {error or 'unknown error'}")
@@ -444,6 +465,7 @@ def _exercise_auth_scenario(
workspace: Path,
timeout_seconds: float,
require_automatic_auth: bool = False,
client_registration: str | None = None,
) -> str:
if scenario in EXPECTED_AUTH_REJECTION_SCENARIOS:
success, error = _oauth_login(
@@ -566,6 +588,7 @@ def _exercise_auth_scenario(
client,
workspace=workspace,
timeout_seconds=timeout_seconds,
client_registration=client_registration,
)
return "completed OAuth login, authenticated discovery, and tool call"
@@ -741,11 +764,7 @@ def run_adapter(server_url: str) -> dict[str, object]:
if not feature_configured:
raise AdapterFailure("could not configure the modern MCP feature")
oauth_client_id = _oauth_client_id(
scenario,
context,
require_production_client_identity=require_automatic_auth,
)
oauth_client_id = _oauth_client_id(scenario, context)
oauth_client_secret = context.get("client_secret")
oauth_client_secret_env_var = None
if (
@@ -880,12 +899,17 @@ def run_adapter(server_url: str) -> dict[str, object]:
raise AdapterFailure("could not enable the modern MCP feature")
if scenario.startswith("auth/"):
client_registration = _client_registration_override(
scenario,
require_automatic_auth=require_automatic_auth,
)
detail = _exercise_auth_scenario(
client,
scenario=scenario,
workspace=workspace,
timeout_seconds=timeout_seconds,
require_automatic_auth=require_automatic_auth,
client_registration=client_registration,
)
if scenario == "auth/pre-registration" and isinstance(
oauth_client_secret,

14
scripts/mcp_conformance/regression-baseline-v1.json
View File

@@ -4030,6 +4030,20 @@
"source": "official",
"transport": "official-http"
},
{
"check_id": "auto_cimd",
"mode": "2026-07-28",
"scenario": "auth/offline-access-scope",
"source": "supplemental",
"transport": "official-http"
},
{
"check_id": "forced_cimd",
"mode": "2026-07-28",
"scenario": "auth/offline-access-scope",
"source": "supplemental",
"transport": "official-http"
},
{
"check_id": "app_server_initialize",
"mode": "2026-07-28",

64
scripts/mcp_conformance/run_codex_compliance.py
View File

@@ -1233,12 +1233,76 @@ def _run_official_case(
source="harness",
)
)
if mode == MODERN_VERSION and "auth/offline-access-scope" in scenarios:
case.checks.extend(
_cimd_registration_checks(
codex_binary,
adapter_script,
conformance_command=conformance_command,
case_home=case_home / "cimd-registration",
timeout_seconds=timeout_seconds,
enable_modern_feature=enable_modern_feature,
)
)
if diagnostics:
case.diagnostics = "\n\n".join(diagnostics)[-16_000:]
case.finish(started_at)
return case
def _cimd_registration_checks(
codex_binary: Path,
adapter_script: Path,
*,
conformance_command: Sequence[str],
case_home: Path,
timeout_seconds: float,
enable_modern_feature: bool,
) -> list[CheckResult]:
checks: list[CheckResult] = []
case_home.mkdir(parents=True)
for check_name, client_registration in (
("auto_cimd", None),
("forced_cimd", "cimd"),
):
env = _isolated_environment(case_home / check_name)
env["CODEX_CONFORMANCE_TIMEOUT"] = str(timeout_seconds)
env["CODEX_CONFORMANCE_ENABLE_MODERN_FEATURE"] = (
"1" if enable_modern_feature else "0"
)
env["CODEX_CONFORMANCE_REQUIRE_AUTOMATIC_AUTH"] = "1"
if client_registration is not None:
env["CODEX_CONFORMANCE_CLIENT_REGISTRATION"] = client_registration
results = run_official_mode(
conformance_command=conformance_command,
adapter_script=adapter_script,
codex_binary=codex_binary,
mode=MODERN_VERSION,
scenarios=("auth/offline-access-scope",),
output_dir=case_home / check_name / "official-results",
timeout_seconds=timeout_seconds,
base_env=env,
)
result = results[0] if len(results) == 1 else None
success = result is not None and result.success
checks.append(
CheckResult(
name=f"supplemental/auth/offline-access-scope/{check_name}",
success=success,
detail=(
result.adapter_detail or result.runner_detail
if result is not None
else "supplemental offline-access-scope scenario did not run"
),
source="supplemental",
scenario="auth/offline-access-scope",
check_id=check_name,
category="auth",
)
)
return checks
def run_compliance(
codex_binary: Path,
*,

88
scripts/mcp_conformance/test_codex_compliance.py
View File

@@ -529,6 +529,55 @@ def test_official_case_preserves_independent_runner_failure_with_successful_adap
assert case.diagnostics == f"{scenario}:\nofficial runner failed"
def test_official_case_keeps_cimd_registration_checks_in_regression_gated_http_case(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[dict[str, object]] = []
def run_official_mode(**kwargs: object) -> list[OfficialScenarioResult]:
calls.append(kwargs)
scenarios = kwargs["scenarios"]
assert isinstance(scenarios, (tuple, list))
scenario = scenarios[0]
assert isinstance(scenario, str)
return [
OfficialScenarioResult(
scenario=scenario,
success=True,
adapter_success=True,
adapter_detail="adapter succeeded",
)
]
monkeypatch.setattr(
"run_codex_compliance.run_official_mode",
run_official_mode,
)
case = _run_official_case(
Path("/opt/codex"),
Path("/src/codex_conformance_adapter.py"),
conformance_command=["conformance"],
mode=MODERN_VERSION,
scenarios=["auth/offline-access-scope"],
case_home=tmp_path / "case",
timeout_seconds=1.0,
enable_modern_feature=True,
)
assert case.transport == "official-http"
supplemental = [check for check in case.checks if check.source == "supplemental"]
assert [check.check_id for check in supplemental] == ["auto_cimd", "forced_cimd"]
assert len(calls) == 3
auto_env = calls[1]["base_env"]
forced_env = calls[2]["base_env"]
assert isinstance(auto_env, dict)
assert isinstance(forced_env, dict)
assert auto_env.get("CODEX_CONFORMANCE_CLIENT_REGISTRATION") is None
assert forced_env["CODEX_CONFORMANCE_CLIENT_REGISTRATION"] == "cimd"
def _regression_report() -> dict[str, object]:
cases: list[dict[str, object]] = []
known_modern = {
@@ -599,6 +648,18 @@ def _regression_report() -> dict[str, object]:
"check_id": None,
}
)
if mode == MODERN_VERSION:
official_checks.extend(
{
"name": f"supplemental/auth/offline-access-scope/{check_id}",
"success": True,
"status": "PASS",
"source": "supplemental",
"scenario": "auth/offline-access-scope",
"check_id": check_id,
}
for check_id in ("auto_cimd", "forced_cimd")
)
cases.append(
{
"mode": mode,
@@ -690,6 +751,33 @@ def test_regression_gate_rejects_a_new_modern_failure() -> None:
assert any(item["scenario"] == "tools_call" for item in gate["newFailures"])
@pytest.mark.parametrize("check_id", ["auto_cimd", "forced_cimd"])
def test_regression_gate_rejects_cimd_registration_failures(
check_id: str,
) -> None:
baseline = _compact_regression_baseline(_regression_report())
candidate = _regression_report()
checks = _regression_case(candidate, MODERN_VERSION, "official-http")["checks"]
assert isinstance(checks, list)
check = next(
item
for item in checks
if isinstance(item, dict)
and item.get("source") == "supplemental"
and item.get("check_id") == check_id
)
check["success"] = False
check["status"] = "FAIL"
gate = _evaluate_regression_gate(candidate, baseline)
assert gate["success"] is False
assert any(
item["source"] == "supplemental" and item["check_id"] == check_id
for item in gate["newFailures"]
)
def test_regression_gate_rejects_a_new_intermediate_oauth_failure() -> None:
baseline = _regression_report()
candidate = deepcopy(baseline)

50
scripts/mcp_conformance/test_official_conformance.py
View File

@@ -10,7 +10,9 @@ import pytest
import run_codex_compliance
from codex_conformance_adapter import (
CIMD_CLIENT_METADATA_URL,
CLIENT_REGISTRATION_OVERRIDE_ENV_VAR,
AdapterFailure,
_client_registration_override,
_exercise_auth_scenario,
_oauth_client_id,
_validate_oauth_secret_not_persisted,
@@ -303,17 +305,6 @@ def test_strict_auth_accepts_product_owned_reauthentication(
assert manual_reauthorizations == []
def test_strict_auth_does_not_invent_a_production_client_metadata_url() -> None:
assert (
_oauth_client_id(
"auth/basic-cimd",
{},
require_production_client_identity=True,
)
is None
)
@pytest.mark.parametrize("filename", ["config.toml", ".credentials.json"])
def test_oauth_client_secret_persistence_is_detected_without_disclosing_it(
tmp_path: Path,
@@ -347,14 +338,6 @@ def test_oauth_client_secret_environment_reference_is_not_a_persisted_secret(
def test_auth_adapter_selects_only_scenario_provided_client_ids() -> None:
assert _oauth_client_id("auth/basic-cimd", {}) == CIMD_CLIENT_METADATA_URL
assert (
_oauth_client_id(
"auth/basic-cimd",
{},
require_production_client_identity=True,
)
is None
)
assert (
_oauth_client_id(
"auth/pre-registration",
@@ -367,6 +350,35 @@ def test_auth_adapter_selects_only_scenario_provided_client_ids() -> None:
_oauth_client_id("auth/pre-registration", {})
@pytest.mark.parametrize(
("scenario", "require_automatic_auth", "override", "expected"),
[
("auth/offline-access-not-supported", False, None, None),
("auth/offline-access-scope", False, None, "dcr"),
("auth/offline-access-scope", True, None, None),
("auth/offline-access-scope", False, "cimd", "cimd"),
],
)
def test_auth_adapter_scopes_client_registration_overrides(
monkeypatch: pytest.MonkeyPatch,
scenario: str,
require_automatic_auth: bool,
override: str | None,
expected: str | None,
) -> None:
if override is None:
monkeypatch.delenv(CLIENT_REGISTRATION_OVERRIDE_ENV_VAR, raising=False)
else:
monkeypatch.setenv(CLIENT_REGISTRATION_OVERRIDE_ENV_VAR, override)
assert (
_client_registration_override(
scenario,
require_automatic_auth=require_automatic_auth,
)
== expected
)
def test_auth_registration_does_not_persist_context_secret(tmp_path: Path) -> None:
config_path = tmp_path / "config.toml"
config_path.write_text('mcp_oauth_credentials_store = "file"\n', encoding="utf-8")