Credential pool checkout ignoring provider boundaries

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

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

  merged to main

  files: agent/agent_init.py, agent/credential_pool.py, hermes_cli/runtime_provider.py, tests/agent/test_credential_pool_provider_boundary.py

  Show the diff

  • 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 = [redacted:secret-assignment]
         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 = [redacted:secret-assignment]
         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 @@
     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="[redacted:secret-assignment]",
    +        runtime_api_key="[redacted:secret-assignment]",
    +        runtime_base_url="https://chatgpt.com/backend-api/codex",
    +        base_url="https://chatgpt.com/backend-api/codex",
    +    )
    +    pool = SimpleNamespace(
    +        provider="openai-codex",
    +        has_credentials=[redacted:secret-assignment] 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": "[redacted:secret-assignment]",
    +            "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"
    \ No newline at end of file