diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 1b49aabd5033..f11b21991f8b 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -4269,7 +4269,12 @@ def _prompt_reasoning_effort_selection(efforts, current_effort=""): -def _prompt_api_key(pconfig, existing_key: str, provider_id: str = "") -> tuple: +def _prompt_api_key( + pconfig, + existing_key: str, + provider_id: str = "", + existing_source: str = "", +) -> tuple: """Shared API-key entry point for ``hermes setup`` / ``hermes model``. Handles both first-time entry and the already-configured case. When a key @@ -4323,8 +4328,14 @@ def _prompt_api_key(pconfig, existing_key: str, provider_id: str = "") -> tuple: # Nothing we can rewrite; just acknowledge and move on. print() return existing_key, False + pool_backed = existing_source.startswith("credential_pool:") + menu = ( + " [K]eep / [R]eplace (default K): " + if pool_backed + else " [K]eep / [R]eplace / [C]lear (default K): " + ) try: - choice = input(" [K]eep / [R]eplace / [C]lear (default K): ").strip().lower() + choice = input(menu).strip().lower() except (KeyboardInterrupt, EOFError): print() choice = "k" @@ -4340,7 +4351,7 @@ def _prompt_api_key(pconfig, existing_key: str, provider_id: str = "") -> tuple: print() return new_key, False - if choice.startswith("c"): + if choice.startswith("c") and not pool_backed: save_env_value(key_env, "") print( f" API key cleared. Re-run `hermes setup` to configure {pconfig.name} again." diff --git a/hermes_cli/model_setup_flows.py b/hermes_cli/model_setup_flows.py index 90f628578a78..3878e5a4f46f 100644 --- a/hermes_cli/model_setup_flows.py +++ b/hermes_cli/model_setup_flows.py @@ -76,6 +76,13 @@ def bedrock_model_routable_from_region(model_id: str, region_name: str) -> bool: return matched_geo == geo +def _existing_api_key_for_model_flow(provider_id: str, pconfig) -> tuple[str, str]: + """Resolve an existing wizard credential without changing its storage.""" + from hermes_cli.auth import _resolve_api_key_provider_secret + + return _resolve_api_key_provider_secret(provider_id, pconfig) + + def _prune_replaced_custom_model_config_credentials( base_url: str, *, @@ -177,8 +184,6 @@ def _model_flow_openrouter(config, current_model=""): _save_model_choice, deactivate_provider, ) - from hermes_cli.config import get_env_value - # Route through _prompt_api_key so users can replace a stale/broken key # in-flow (K/R/C) instead of having to edit ~/.hermes/.env by hand. The # previous bypass-when-key-exists branch left no way to recover from a @@ -190,11 +195,16 @@ def _model_flow_openrouter(config, current_model=""): auth_type="api_key", api_key_env_vars=("OPENROUTER_API_KEY",), ) - existing_key = get_env_value("OPENROUTER_API_KEY") or "" + existing_key, existing_source = _existing_api_key_for_model_flow("openrouter", pconfig) if not existing_key: print("Get one at: https://openrouter.ai/keys") print() - _resolved, abort = _prompt_api_key(pconfig, existing_key, provider_id="openrouter") + _resolved, abort = _prompt_api_key( + pconfig, + existing_key, + provider_id="openrouter", + existing_source=existing_source, + ) if abort: return @@ -1975,14 +1985,13 @@ def _model_flow_kimi(config, current_model=""): base_url_env = pconfig.base_url_env_var or "" # Step 1: Check / prompt for API key - existing_key = "" - for ev in pconfig.api_key_env_vars: - existing_key = get_env_value(ev) or os.getenv(ev, "") - if existing_key: - break + existing_key, existing_source = _existing_api_key_for_model_flow(provider_id, pconfig) existing_key, abort = _prompt_api_key( - pconfig, existing_key, provider_id=provider_id + pconfig, + existing_key, + provider_id=provider_id, + existing_source=existing_source, ) if abort: return @@ -2060,14 +2069,13 @@ def _model_flow_stepfun(config, current_model=""): key_env = pconfig.api_key_env_vars[0] if pconfig.api_key_env_vars else "" base_url_env = pconfig.base_url_env_var or "" - existing_key = "" - for ev in pconfig.api_key_env_vars: - existing_key = get_env_value(ev) or os.getenv(ev, "") - if existing_key: - break + existing_key, existing_source = _existing_api_key_for_model_flow(provider_id, pconfig) existing_key, abort = _prompt_api_key( - pconfig, existing_key, provider_id=provider_id + pconfig, + existing_key, + provider_id=provider_id, + existing_source=existing_source, ) if abort: return @@ -2642,14 +2650,13 @@ def _model_flow_api_key_provider(config, provider_id, current_model=""): base_url_env = pconfig.base_url_env_var or "" # Check / prompt for API key - existing_key = "" - for ev in pconfig.api_key_env_vars: - existing_key = get_env_value(ev) or os.getenv(ev, "") - if existing_key: - break + existing_key, existing_source = _existing_api_key_for_model_flow(provider_id, pconfig) existing_key, abort = _prompt_api_key( - pconfig, existing_key, provider_id=provider_id + pconfig, + existing_key, + provider_id=provider_id, + existing_source=existing_source, ) if abort: return diff --git a/tests/hermes_cli/test_model_flow_pooled_credentials.py b/tests/hermes_cli/test_model_flow_pooled_credentials.py new file mode 100644 index 000000000000..6ec7a14d6729 --- /dev/null +++ b/tests/hermes_cli/test_model_flow_pooled_credentials.py @@ -0,0 +1,129 @@ +"""Pool-only credentials must be visible to interactive model setup flows.""" + +from __future__ import annotations + +from unittest.mock import patch + +from hermes_cli.auth import PROVIDER_REGISTRY +from hermes_cli.model_setup_flows import _existing_api_key_for_model_flow + + +class _PoolEntry: + access_token = "pool-secret" + runtime_api_key = "" + + +class _AvailablePool: + def has_credentials(self) -> bool: + return True + + def peek(self): + return _PoolEntry() + + +class _ExhaustedPool: + def has_credentials(self) -> bool: + return True + + def peek(self): + return None + + +def test_existing_key_precedence_is_dotenv_then_process_then_pool(tmp_path, monkeypatch): + pconfig = PROVIDER_REGISTRY["deepseek"] + hermes_home = tmp_path / "hermes" + hermes_home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + monkeypatch.setenv("DEEPSEEK_API_KEY", "process-secret") + (hermes_home / ".env").write_text("DEEPSEEK_API_KEY=dotenv-secret\n", encoding="utf-8") + + with patch("agent.credential_pool.load_pool", return_value=_AvailablePool()): + assert _existing_api_key_for_model_flow("deepseek", pconfig) == ( + "dotenv-secret", + "DEEPSEEK_API_KEY", + ) + + (hermes_home / ".env").write_text("", encoding="utf-8") + with patch("agent.credential_pool.load_pool", return_value=_AvailablePool()): + assert _existing_api_key_for_model_flow("deepseek", pconfig) == ( + "process-secret", + "DEEPSEEK_API_KEY", + ) + + monkeypatch.delenv("DEEPSEEK_API_KEY") + with patch("agent.credential_pool.load_pool", return_value=_AvailablePool()): + assert _existing_api_key_for_model_flow("deepseek", pconfig) == ( + "pool-secret", + "credential_pool:deepseek", + ) + + +def test_exhausted_pool_is_not_an_existing_key(monkeypatch): + pconfig = PROVIDER_REGISTRY["deepseek"] + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + with ( + patch("hermes_cli.config.get_env_value", return_value=""), + patch("agent.credential_pool.load_pool", return_value=_ExhaustedPool()), + ): + assert _existing_api_key_for_model_flow("deepseek", pconfig) == ("", "") + + +def test_generic_api_key_flow_passes_pool_key_to_existing_key_prompt(monkeypatch): + from hermes_cli.model_setup_flows import _model_flow_api_key_provider + + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + captured: dict[str, str] = {} + + def capture_prompt(_pconfig, existing_key, **_kwargs): + captured["existing_key"] = existing_key + return existing_key, True + + with ( + patch("hermes_cli.config.get_env_value", return_value=""), + patch("agent.credential_pool.load_pool", return_value=_AvailablePool()), + patch("hermes_cli.main._prompt_api_key", side_effect=capture_prompt), + ): + _model_flow_api_key_provider({}, "deepseek") + + assert captured["existing_key"] == "pool-secret" + + +def test_kimi_flow_passes_pool_key_to_existing_key_prompt(monkeypatch): + from hermes_cli.model_setup_flows import _model_flow_kimi + + monkeypatch.delenv("KIMI_API_KEY", raising=False) + monkeypatch.delenv("MOONSHOT_API_KEY", raising=False) + captured: dict[str, str] = {} + + def capture_prompt(_pconfig, existing_key, **_kwargs): + captured["existing_key"] = existing_key + return existing_key, True + + with ( + patch("hermes_cli.config.get_env_value", return_value=""), + patch("agent.credential_pool.load_pool", return_value=_AvailablePool()), + patch("hermes_cli.main._prompt_api_key", side_effect=capture_prompt), + ): + _model_flow_kimi({}) + + assert captured["existing_key"] == "pool-secret" + + +def test_exhausted_pool_still_uses_first_time_prompt(monkeypatch): + from hermes_cli.model_setup_flows import _model_flow_api_key_provider + + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + captured: dict[str, str] = {} + + def capture_prompt(_pconfig, existing_key, **_kwargs): + captured["existing_key"] = existing_key + return "", True + + with ( + patch("hermes_cli.config.get_env_value", return_value=""), + patch("agent.credential_pool.load_pool", return_value=_ExhaustedPool()), + patch("hermes_cli.main._prompt_api_key", side_effect=capture_prompt), + ): + _model_flow_api_key_provider({}, "deepseek") + + assert captured["existing_key"] == "" diff --git a/tests/hermes_cli/test_prompt_api_key.py b/tests/hermes_cli/test_prompt_api_key.py index 61da86523621..a243148f7779 100644 --- a/tests/hermes_cli/test_prompt_api_key.py +++ b/tests/hermes_cli/test_prompt_api_key.py @@ -37,6 +37,33 @@ def _run_prompt(existing_key, choice, new_key="", provider_id="", pconfig_name=" return m._prompt_api_key(pconfig, existing_key, provider_id=provider_id) +def test_pool_only_key_does_not_offer_or_execute_clear(profile_env, monkeypatch, capsys): + from hermes_cli import main as m + + pconfig = _pconfig("deepseek") + prompts = [] + + def choose_clear(prompt): + prompts.append(prompt) + return "c" + + monkeypatch.setattr("builtins.input", choose_clear) + with patch("hermes_cli.config.save_env_value") as save_env: + key, abort = m._prompt_api_key( + pconfig, + "pool-secret", + provider_id="deepseek", + existing_source="credential_pool:deepseek", + ) + + assert key == "pool-secret" + assert abort is False + assert prompts and "[C]lear" not in prompts[0] + assert "[K]eep / [R]eplace" in prompts[0] + save_env.assert_not_called() + assert "API key cleared" not in capsys.readouterr().out + + # First-time entry ──────────────────────────────────────────────────────────── def test_first_time_save_new_key(profile_env):