fix(auth): enforce credential pool provider boundaries (#63048)

Retain the provider-boundary core of #52799 while reusing the pool reload and handoff paths already landed in #53591 and #62417.

Co-authored-by: Flownium <157689911+itsflownium@users.noreply.github.com>

agent/agent_init.py

@@ -412,13 +412,25 @@ def init_agent(
     agent.skip_context_files = skip_context_files
     agent.load_soul_identity = load_soul_identity
     agent.pass_session_id = pass_session_id
-    agent._credential_pool = credential_pool
     agent.log_prefix_chars = log_prefix_chars
     agent.log_prefix = f"{log_prefix} " if log_prefix else ""
     # Store effective base URL for feature detection (prompt caching, reasoning, etc.)
     agent.base_url = base_url or ""
     provider_name = provider.strip().lower() if isinstance(provider, str) and provider.strip() else None
     agent.provider = provider_name or ""
+    if credential_pool is not None:
+        try:
+            from agent.credential_pool import credential_pool_matches_provider
+
+            if not credential_pool_matches_provider(
+                credential_pool,
+                agent.provider,
+                base_url=agent.base_url,
+            ):
+                credential_pool = None
+        except Exception:
+            credential_pool = None
+    agent._credential_pool = credential_pool
     agent.acp_command = acp_command or command
     agent.acp_args = list(acp_args or args or [])
     if api_mode in {"chat_completions", "codex_responses", "anthropic_messages", "bedrock_converse", "codex_app_server"}:

agent/credential_pool.py

@@ -445,6 +445,44 @@ def get_pool_strategy(provider: str) -> str:
     return STRATEGY_FILL_FIRST
 
 
+def credential_pool_matches_provider(
+    pool_or_provider: Any,
+    provider: Optional[str],
+    *,
+    base_url: Optional[str] = None,
+) -> bool:
+    """Return whether a pool belongs to the requested runtime provider.
+
+    Named custom endpoints intentionally use two identities: the live agent is
+    ``custom`` while its pool is keyed ``custom:<name>``. Accept that pair only
+    when the runtime base URL resolves to the exact same custom pool key.
+    Empty string identities fail closed. Legacy pool adapters without a
+    ``provider`` attribute remain compatible; production pools are scoped.
+    """
+    raw_pool_provider = getattr(pool_or_provider, "provider", None)
+    if raw_pool_provider is None:
+        if isinstance(pool_or_provider, str):
+            raw_pool_provider = pool_or_provider
+        else:
+            # Backward compatibility for lightweight/unscoped pool adapters.
+            # Production CredentialPool instances always carry ``provider``;
+            # old plugins and tests may expose only select()/has_credentials().
+            return True
+    pool_provider = str(raw_pool_provider or "").strip().lower()
+    provider_norm = str(provider or "").strip().lower()
+    if not pool_provider or not provider_norm:
+        return False
+    if pool_provider == provider_norm:
+        return True
+    if provider_norm != "custom" or not pool_provider.startswith(CUSTOM_POOL_PREFIX):
+        return False
+    try:
+        matched_pool = get_custom_provider_pool_key(base_url or "")
+    except Exception:
+        return False
+    return str(matched_pool or "").strip().lower() == pool_provider
+
+
 DEFAULT_MAX_CONCURRENT_PER_CREDENTIAL = 1
 
 

hermes_cli/runtime_provider.py

@@ -11,7 +11,13 @@ from typing import Any, Dict, Optional
 logger = logging.getLogger(__name__)
 
 from hermes_cli import auth as auth_mod
-from agent.credential_pool import CredentialPool, PooledCredential, get_custom_provider_pool_key, load_pool
+from agent.credential_pool import (
+    CredentialPool,
+    PooledCredential,
+    credential_pool_matches_provider,
+    get_custom_provider_pool_key,
+    load_pool,
+)
 from agent.secret_scope import get_secret as _get_secret
 from hermes_cli.auth import (
     AuthError,
@@ -1731,7 +1737,19 @@ def resolve_runtime_provider(
                 if not pool_api_key or not _agent_key_is_usable(nous_state, min_ttl):
                     logger.debug("Nous pool entry agent_key still unavailable, falling through to runtime resolution")
                     pool_api_key = ""
-        if entry is not None and pool_api_key:
+        if (
+            entry is not None
+            and pool_api_key
+            and credential_pool_matches_provider(
+                pool,
+                provider,
+                base_url=(
+                    getattr(entry, "runtime_base_url", None)
+                    or getattr(entry, "base_url", None)
+                    or ""
+                ),
+            )
+        ):
             return _resolve_runtime_from_pool_entry(
                 provider=provider,
                 entry=entry,

tests/agent/test_credential_pool_provider_boundary.py

@@ -0,0 +1,64 @@
+"""Credential pools must never cross provider or custom-endpoint boundaries."""
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from agent.credential_pool import credential_pool_matches_provider
+from hermes_cli import runtime_provider as rp
+
+
+def test_provider_match_requires_exact_non_custom_identity():
+    assert credential_pool_matches_provider("deepseek", "deepseek")
+    assert not credential_pool_matches_provider("openai-codex", "deepseek")
+    assert not credential_pool_matches_provider("", "deepseek")
+
+
+def test_custom_pool_match_is_scoped_by_endpoint():
+    with patch(
+        "agent.credential_pool.get_custom_provider_pool_key",
+        return_value="custom:lab",
+    ):
+        assert credential_pool_matches_provider(
+            "custom:lab", "custom", base_url="https://lab.example/v1"
+        )
+        assert not credential_pool_matches_provider(
+            "custom:other", "custom", base_url="https://lab.example/v1"
+        )
+
+
+def test_runtime_ignores_pool_loaded_for_different_provider(monkeypatch):
+    entry = SimpleNamespace(
+        provider="openai-codex",
+        access_token="wrong-token",
+        runtime_api_key="wrong-token",
+        runtime_base_url="https://chatgpt.com/backend-api/codex",
+        base_url="https://chatgpt.com/backend-api/codex",
+    )
+    pool = SimpleNamespace(
+        provider="openai-codex",
+        has_credentials=lambda: True,
+        select=lambda: entry,
+    )
+    monkeypatch.setattr(rp, "load_pool", lambda _provider: pool)
+    monkeypatch.setattr(rp, "resolve_provider", lambda *_a, **_kw: "deepseek")
+    monkeypatch.setattr(
+        rp,
+        "_get_model_config",
+        lambda: {"provider": "deepseek", "default": "deepseek-chat"},
+    )
+    monkeypatch.setattr(
+        rp,
+        "resolve_api_key_provider_credentials",
+        lambda _provider: {
+            "provider": "deepseek",
+            "api_key": "deepseek-key",
+            "base_url": "https://api.deepseek.com/v1",
+            "source": "env",
+        },
+    )
+
+    resolved = rp.resolve_runtime_provider(requested="deepseek")
+
+    assert resolved["provider"] == "deepseek"
+    assert resolved["api_key"] == "deepseek-key"
+    assert resolved["base_url"] == "https://api.deepseek.com/v1"