diff --git "a/hermes_cli/auth.py" "b/hermes_cli/auth.py" new file mode 100644--- /dev/null +++ "b/hermes_cli/auth.py" @@ -0,0 +1,2328 @@ +"""Multi-provider authentication system for Hermes Agent. + +- ``ProviderConfig`` / ``PROVIDER_REGISTRY`` describe every known inference provider. +- The auth store (``~/.hermes/auth.json``) holds per-provider state, the credential pool and + suppression markers; ``_auth_store_lock`` / ``_load_auth_store`` / ``_save_auth_store`` are the + only I/O primitives (cross-process flock, atomic 0o600 writes). +- ``resolve_provider()`` picks the active provider via the documented priority chain. +- ``OAUTH_PROVIDER_FLOWS`` maps each OAuth provider to its resolver/status builder; the flows live in + ``auth_nous``/``auth_codex``/``auth_xai``/``auth_qwen``/``auth_minimax``/``auth_spotify``/``auth_openrouter`` and are + re-imported here so ``hermes_cli.auth.`` stays the public/patchable surface.""" + +from __future__ import annotations + +import json +import logging +import os +import shutil +import shlex +import threading +import time +import webbrowser # noqa: F401 (tests patch auth_mod.webbrowser.open; same module object) + +from contextlib import ExitStack, contextmanager +from dataclasses import dataclass, field +from functools import partial +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Dict, FrozenSet, Iterable, List, Optional, Tuple +from urllib.parse import urlparse + +from hermes_cli.config import ( + get_hermes_home, get_config_path, read_raw_config, require_readable_config_before_write) +from hermes_constants import OPENROUTER_BASE_URL, hermes_home_key, secure_parent_dir +from agent.credential_persistence import sanitize_borrowed_credential_payload +from utils import atomic_json_write, atomic_yaml_write, env_float, is_truthy_value # noqa: F401 (env_float: agent.credential_pool reads auth_mod.env_float) +from hermes_cli.auth_zai_kimi import ( # noqa: F401 re-exported + KIMI_CODE_BASE_URL, ZAI_ENDPOINTS, _normalize_lmstudio_runtime_base_url, _resolve_kimi_base_url, + _resolve_zai_base_url, detect_zai_endpoint) +from hermes_cli.auth_model_picker import ( # noqa: F401 re-exported + _prompt_model_selection, _save_model_choice) +from hermes_cli.auth_device_flow import ( # noqa: F401 re-exported + _can_open_graphical_browser, _default_verify, _is_remote_session, + _nous_device_auth_timeout_message, _offer_existing_oauth_credentials, + _poll_device_token_generic, _poll_for_token, _print_device_code_instructions, + _print_login_success, _print_loopback_ssh_hint, _prompt_yes_no, _request_device_code, + _resolve_verify, _ssh_user_at_host) +from hermes_cli.auth_oauth_grants import ( # noqa: F401 re-exported + SINGLE_USE_REFRESH_POOL_PROVIDERS, _oauth_heal_clean_marks, _oauth_heal_notices, + consume_oauth_heal_notices, heal_forked_single_use_oauth_grants, + strip_cloned_single_use_oauth_grants) +from hermes_cli.auth_nous import ( # noqa: F401 re-exported + NOUS_SESSION_TERMINAL, NOUS_SESSION_UNKNOWN, NOUS_SESSION_VALID, _ALLOWED_NOUS_INFERENCE_HOSTS, + _agent_key_is_usable, _apply_nous_refreshed_tokens, _assert_nous_inference_jwt_usable, + _compute_nous_auth_status, _format_nous_entitlement_auth_error, _healed_nous_inference_url, + _login_nous, _merge_shared_nous_oauth_state, _migrate_stale_nous_portal_url, + _nous_device_code_login, _nous_inference_env_override, _nous_invoke_jwt_is_usable, + _nous_invoke_jwt_status, _nous_portal_env_override, _nous_shared_store_lock, + _nous_shared_store_path, _pool_first_oauth_status, _quarantine_nous_oauth_state, + _quarantine_nous_pool_entries, _read_shared_nous_state, _refresh_access_token, + _refresh_nous_or_quarantine, _select_nous_invoke_jwt, _sync_nous_pool_from_auth_store, + _token_fingerprint, _try_import_shared_nous_state, _validate_nous_inference_url_from_network, + _write_shared_nous_state, fetch_nous_models, get_nous_auth_status_local, + get_nous_session_validity, persist_nous_credentials, refresh_nous_oauth_from_state, + resolve_nous_runtime_credentials, step_up_nous_billing_scope) +from hermes_cli.auth_minimax import ( # noqa: F401 re-exported + _MINIMAX_OAUTH_ERROR_BODY_LIMIT, _login_minimax_oauth, _minimax_oauth_login, _minimax_pkce_pair, + _minimax_poll_token, _minimax_post_form, _minimax_request_user_code, + _minimax_resolve_token_expiry_unix, _minimax_response_error_text, _minimax_save_auth_state, + _refresh_minimax_oauth_state, build_minimax_oauth_token_provider, + resolve_minimax_oauth_runtime_credentials) +from hermes_cli.auth_xai import ( # noqa: F401 re-exported + _login_xai_oauth, _read_xai_oauth_tokens, _refresh_xai_oauth_tokens, _save_xai_oauth_tokens, + _write_through_xai_oauth_to_global_root, _xai_access_token_is_expiring, + _xai_oauth_device_code_login, _xai_oauth_discovery, _xai_oauth_poll_device_token, + _xai_oauth_request_device_code, _xai_proactive_refresh_skew_seconds, + _xai_validate_inference_base_url, refresh_xai_oauth_pure, resolve_xai_oauth_runtime_credentials) +from hermes_cli.auth_codex import ( # noqa: F401 re-exported + _codex_access_token_is_expiring, _codex_device_code_login, _codex_http_client, + _codex_pool_rate_limit_status, _codex_quota_probe_cache, _codex_usage_probe_url, + _import_codex_cli_tokens, _is_codex_rate_limit_shaped, _login_openai_codex, + _probe_codex_quota_restored, _read_codex_tokens, _refresh_codex_auth_tokens, _save_codex_tokens, + clear_codex_pool_quota_cooldowns, refresh_codex_oauth_pure, resolve_codex_runtime_credentials) +from hermes_cli.auth_spotify import ( # noqa: F401 re-exported + _refresh_spotify_oauth_state, get_spotify_auth_status, login_spotify_command, + resolve_spotify_runtime_credentials) +from hermes_cli.auth_openrouter import _openrouter_pkce_login # noqa: F401 re-exported +from hermes_cli.auth_qwen import ( # noqa: F401 re-exported + _qwen_access_token_is_expiring, _qwen_cli_auth_path, _read_qwen_cli_tokens, + _refresh_qwen_cli_tokens, _save_qwen_cli_tokens, get_qwen_auth_status, + resolve_qwen_runtime_credentials) +from hermes_cli.auth_constants import ( # noqa: F401 re-exported + _decode_jwt_claims, AUTH_STORE_VERSION, AUTH_LOCK_TIMEOUT_SECONDS, DEFAULT_NOUS_PORTAL_URL, + DEFAULT_NOUS_INFERENCE_URL, DEFAULT_NOUS_CLIENT_ID, NOUS_BILLING_MANAGE_SCOPE, + DEFAULT_NOUS_SCOPE, NOUS_DEVICE_CODE_SOURCE, NOUS_AUTH_PATH_INVOKE_JWT, + ACCESS_TOKEN_REFRESH_SKEW_SECONDS, NOUS_INVOKE_JWT_MIN_TTL_SECONDS, DEFAULT_CODEX_BASE_URL, + DEFAULT_XAI_OAUTH_BASE_URL, MINIMAX_OAUTH_CLIENT_ID, MINIMAX_OAUTH_SCOPE, + MINIMAX_OAUTH_GLOBAL_BASE, MINIMAX_OAUTH_CN_BASE, MINIMAX_OAUTH_GLOBAL_INFERENCE, + MINIMAX_OAUTH_CN_INFERENCE, MINIMAX_OAUTH_REFRESH_SKEW_SECONDS, DEFAULT_QWEN_BASE_URL, + DEFAULT_GITHUB_MODELS_BASE_URL, DEFAULT_COPILOT_ACP_BASE_URL, DEFAULT_OLLAMA_CLOUD_BASE_URL, + DEFAULT_ACTUAL_BASE_URL, DEFAULT_ACTUAL_LOCAL_BASE_URL, STEPFUN_STEP_PLAN_INTL_BASE_URL, + STEPFUN_STEP_PLAN_CN_BASE_URL, CODEX_OAUTH_CLIENT_ID, CODEX_OAUTH_TOKEN_URL, + CODEX_ACCESS_TOKEN_REFRESH_SKEW_SECONDS, XAI_OAUTH_CLIENT_ID, XAI_OAUTH_SCOPE, + XAI_ACCESS_TOKEN_REFRESH_SKEW_SECONDS, QWEN_ACCESS_TOKEN_REFRESH_SKEW_SECONDS, + DEFAULT_SPOTIFY_ACCOUNTS_BASE_URL, DEFAULT_SPOTIFY_API_BASE_URL, SPOTIFY_DOCS_URL, + DEFAULT_SPOTIFY_SCOPE, SERVICE_PROVIDER_NAMES, LMSTUDIO_NOAUTH_PLACEHOLDER, + ACTUAL_LOCAL_NOAUTH_PLACEHOLDER, CODEX_RATE_LIMITED_CODE, AuthError, _nous_err, httpx) + +logger = logging.getLogger(__name__) + +try: + import fcntl +except Exception: + fcntl = None +try: + import msvcrt +except Exception: + msvcrt = None + +def is_actual_local_base_url(base_url: str) -> bool: + """Return True for Actual's loopback local API endpoint.""" + try: + host = (urlparse(base_url or "").hostname or "").lower().rstrip(".") + except Exception: + return False + return host in {"localhost", "127.0.0.1", "::1", "0.0.0.0"} + + +def normalize_actual_base_url(base_url: str) -> str: + """Return Actual's OpenAI-compatible base URL (hosted api.actual.inc or the loopback local server; + both expose a /v1 surface for the selected OpenAI-compatible transport).""" + url = str(base_url or "").strip().rstrip("/") + if not url: + return DEFAULT_ACTUAL_BASE_URL + try: + parsed = urlparse(url) + host = (parsed.hostname or "").lower().rstrip(".") + path = parsed.path.rstrip("/") + except Exception: + return url + if path in {"", "/"} and (host == "api.actual.inc" or is_actual_local_base_url(url)): + return url + "/v1" + return url + + +# ── Provider Registry ─────────────────────────────────────────────────────────────────────────────── + +@dataclass +class ProviderConfig: + """Describes a known inference provider.""" + id: str + name: str + auth_type: str # "oauth_device_code", "oauth_external", "oauth_minimax", "api_key", ... + portal_base_url: str = "" + inference_base_url: str = "" + client_id: str = "" + scope: str = "" + extra: Dict[str, Any] = field(default_factory=dict) + api_key_env_vars: tuple = () # API-key providers: env vars to check, in priority order + base_url_env_var: str = "" # optional env var overriding the base URL + + +def _api_key_provider( + id: str, name: str, inference_base_url: str, api_key_env_vars: tuple, + base_url_env_var: str = "", auth_type: str = "api_key") -> ProviderConfig: + """Compact constructor for the common env-var-keyed provider shape.""" + return ProviderConfig( + id=id, name=name, auth_type=auth_type, inference_base_url=inference_base_url, + api_key_env_vars=api_key_env_vars, base_url_env_var=base_url_env_var) + + +# Registry rows in priority order (resolve_provider() scans api_key rows in this order). A tuple +# row is ``_api_key_provider(id, name, inference_base_url, api_key_env_vars[, base_url_env_var +# [, auth_type]])``; OAuth / bespoke rows are full ``ProviderConfig`` objects. +_REGISTRY_ROWS: Tuple[Any, ...] = ( + ProviderConfig( + "nous", "Nous Portal", "oauth_device_code", portal_base_url=DEFAULT_NOUS_PORTAL_URL, + inference_base_url=DEFAULT_NOUS_INFERENCE_URL, client_id=DEFAULT_NOUS_CLIENT_ID, + scope=DEFAULT_NOUS_SCOPE), + ProviderConfig("openai-codex", "OpenAI Codex", "oauth_external", inference_base_url=DEFAULT_CODEX_BASE_URL), + ("openai-api", "OpenAI API", "https://api.openai.com/v1", ("OPENAI_API_KEY",), "OPENAI_BASE_URL"), + ProviderConfig( + "xai-oauth", "xAI Grok OAuth (SuperGrok / Premium+)", "oauth_external", + inference_base_url=DEFAULT_XAI_OAUTH_BASE_URL), + ProviderConfig("qwen-oauth", "Qwen OAuth", "oauth_external", inference_base_url=DEFAULT_QWEN_BASE_URL), + ("lmstudio", "LM Studio", "http://127.0.0.1:1234/v1", ("LM_API_KEY",), "LM_BASE_URL"), + ("copilot", "GitHub Copilot", DEFAULT_GITHUB_MODELS_BASE_URL, + ("COPILOT_GITHUB_TOKEN", "GH_TOKEN", "GITHUB_TOKEN"), "COPILOT_API_BASE_URL"), + ProviderConfig( + "copilot-acp", "GitHub Copilot ACP", "external_process", + inference_base_url=DEFAULT_COPILOT_ACP_BASE_URL, base_url_env_var="COPILOT_ACP_BASE_URL"), + ("gemini", "Google AI Studio", "https://generativelanguage.googleapis.com/v1beta", + ("GOOGLE_API_KEY", "GEMINI_API_KEY"), "GEMINI_BASE_URL"), + ("zai", "Z.AI / GLM", "https://api.z.ai/api/paas/v4", + ("GLM_API_KEY", "ZAI_API_KEY", "Z_AI_API_KEY"), "GLM_BASE_URL"), + # Legacy platform.moonshot.ai keys use this endpoint (OpenAI-compat); sk-kimi- (Kimi Code) + # keys are auto-redirected to api.kimi.com/coding by _resolve_kimi_base_url(). + ("kimi-coding", "Kimi / Moonshot", "https://api.moonshot.ai/v1", + ("KIMI_API_KEY", "KIMI_CODING_API_KEY"), "KIMI_BASE_URL"), + ("kimi-coding-cn", "Kimi / Moonshot (China)", "https://api.moonshot.cn/v1", ("KIMI_CN_API_KEY",)), + ("stepfun", "StepFun Step Plan", STEPFUN_STEP_PLAN_INTL_BASE_URL, ("STEPFUN_API_KEY",), "STEPFUN_BASE_URL"), + ("arcee", "Arcee AI", "https://api.arcee.ai/api/v1", ("ARCEEAI_API_KEY",), "ARCEE_BASE_URL"), + ("gmi", "GMI Cloud", "https://api.gmi-serving.com/v1", ("GMI_API_KEY",), "GMI_BASE_URL"), + ("actual", "Actual Computer", DEFAULT_ACTUAL_BASE_URL, ("ACTUAL_API_KEY",), "ACTUAL_BASE_URL"), + ("minimax", "MiniMax", "https://api.minimax.io/anthropic", ("MINIMAX_API_KEY",), "MINIMAX_BASE_URL"), + ProviderConfig( + "minimax-oauth", "MiniMax (OAuth \u00b7 minimax.io)", "oauth_minimax", + portal_base_url=MINIMAX_OAUTH_GLOBAL_BASE, inference_base_url=MINIMAX_OAUTH_GLOBAL_INFERENCE, + client_id=MINIMAX_OAUTH_CLIENT_ID, scope=MINIMAX_OAUTH_SCOPE, + extra={"region": "global", "cn_portal_base_url": MINIMAX_OAUTH_CN_BASE, + "cn_inference_base_url": MINIMAX_OAUTH_CN_INFERENCE}), + # CLAUDE_CODE_OAUTH_TOKEN is NOT an API key despite auth_type="api_key": `claude setup-token` + # yields an `sk-ant-oat01…` OAuth token (401s as x-api-key, 429s as bare Bearer). It stays in + # this tuple because the tuple doubles as the credential-DISCOVERY list + # (agent/credential_pool.py builds its env scan from it); the adapter routes it down the OAuth + # path by prefix. Only ANTHROPIC_API_KEY and ANTHROPIC_TOKEN are usable as literal API keys. + ("anthropic", "Anthropic", "https://api.anthropic.com", + ("ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN"), "ANTHROPIC_BASE_URL"), + ("alibaba", "Qwen Cloud", "https://dashscope-intl.aliyuncs.com/compatible-mode/v1", + ("DASHSCOPE_API_KEY",), "DASHSCOPE_BASE_URL"), + ("alibaba-coding-plan", "Alibaba Cloud (Coding Plan)", "https://coding-intl.dashscope.aliyuncs.com/v1", + ("ALIBABA_CODING_PLAN_API_KEY", "DASHSCOPE_API_KEY"), "ALIBABA_CODING_PLAN_BASE_URL"), + ("minimax-cn", "MiniMax (China)", "https://api.minimaxi.com/anthropic", ("MINIMAX_CN_API_KEY",), + "MINIMAX_CN_BASE_URL"), + ("deepseek", "DeepSeek", "https://api.deepseek.com/v1", ("DEEPSEEK_API_KEY",), "DEEPSEEK_BASE_URL"), + ("xai", "xAI", "https://api.x.ai/v1", ("XAI_API_KEY",), "XAI_BASE_URL"), + ("nvidia", "NVIDIA NIM", "https://integrate.api.nvidia.com/v1", ("NVIDIA_API_KEY",), "NVIDIA_BASE_URL"), + ("ai-gateway", "Vercel AI Gateway", "https://ai-gateway.vercel.sh/v1", ("AI_GATEWAY_API_KEY",), + "AI_GATEWAY_BASE_URL"), + ("opencode-zen", "OpenCode Zen", "https://opencode.ai/zen/v1", ("OPENCODE_ZEN_API_KEY",), + "OPENCODE_ZEN_BASE_URL"), + # OpenCode Go mixes API surfaces by model (GLM/Kimi: OpenAI chat under /v1; MiniMax and + # Qwen 3.7: Anthropic Messages under /v1/messages). Keep the base at /v1; api_mode is per-model. + ("opencode-go", "OpenCode Go", "https://opencode.ai/zen/go/v1", ("OPENCODE_GO_API_KEY",), + "OPENCODE_GO_BASE_URL"), + # Deliberately NO api_key_env_vars: the free tier is served anonymously (any unrecognized bearer + # is a 401), so there is no secret to configure. Select via `hermes model` / `/model free`. + ("opencode-free", "OpenCode Free", "https://opencode.ai/zen/v1", ()), + ("kilocode", "Kilo Code", "https://api.kilo.ai/api/gateway", ("KILOCODE_API_KEY",), "KILOCODE_BASE_URL"), + ("huggingface", "Hugging Face", "https://router.huggingface.co/v1", ("HF_TOKEN",), "HF_BASE_URL"), + ("xiaomi", "Xiaomi MiMo", "https://api.xiaomimimo.com/v1", ("XIAOMI_API_KEY",), "XIAOMI_BASE_URL"), + ("tencent-tokenhub", "Tencent TokenHub", "https://tokenhub.tencentmaas.com/v1", ("TOKENHUB_API_KEY",), + "TOKENHUB_BASE_URL"), + ("tencent-tokenplan", "Tencent TokenPlan", "https://api.lkeap.cloud.tencent.com/plan/anthropic", + ("TOKENPLAN_API_KEY",), "TOKENPLAN_BASE_URL"), + ("ollama-cloud", "Ollama Cloud", DEFAULT_OLLAMA_CLOUD_BASE_URL, ("OLLAMA_API_KEY",), "OLLAMA_BASE_URL"), + ("bedrock", "AWS Bedrock", "https://bedrock-runtime.us-east-1.amazonaws.com", (), "BEDROCK_BASE_URL", + "aws_sdk"), + # No static inference_base_url: Vertex's endpoint is computed per request from project_id + + # region (agent/vertex_adapter.py build_vertex_base_url), not a fixed host. + ("vertex", "Google Vertex AI", "", (), "", "vertex"), + ("azure-foundry", "Azure Foundry", "", ("AZURE_FOUNDRY_API_KEY",), "AZURE_FOUNDRY_BASE_URL")) +PROVIDER_REGISTRY: Dict[str, ProviderConfig] = { + p.id: p for p in (r if isinstance(r, ProviderConfig) else _api_key_provider(*r) for r in _REGISTRY_ROWS) +} + +# Providers handled outside the registry: copilot/kimi/zai have bespoke token refresh here; +# openrouter/custom are aggregator/user-supplied and runtime_provider relies on +# ``openrouter not in PROVIDER_REGISTRY``. +_REGISTRY_PLUGIN_SKIP = frozenset({"copilot", "kimi-coding", "kimi-coding-cn", "zai", "openrouter", "custom"}) + + +def _register_plugin_provider(pp: Any) -> None: + """Auto-register one providers/ profile (plugins/model-providers//) not declared above. + + External-process (ACP) providers have no API-key env vars; registering them is what lets an + out-of-tree provider pass ``resolve_provider()``'s known-provider gate ("Unknown provider").""" + if pp.auth_type == "external_process": + pconfig = ProviderConfig( + pp.name, pp.display_name or pp.name, "external_process", inference_base_url=pp.base_url) + elif pp.auth_type == "api_key" and pp.env_vars and pp.name not in _REGISTRY_PLUGIN_SKIP: + is_url = lambda v: v.endswith("_BASE_URL") or v.endswith("_URL") # noqa: E731 + pconfig = _api_key_provider( + pp.name, pp.display_name or pp.name, pp.base_url, + tuple(v for v in pp.env_vars if not is_url(v)) or pp.env_vars, + next((v for v in pp.env_vars if is_url(v)), None) or "") + else: + return + PROVIDER_REGISTRY[pp.name] = pconfig + for alias in pp.aliases: # so resolve_provider() resolves them too + PROVIDER_REGISTRY.setdefault(alias, pconfig) + + +try: + from providers import list_providers as _list_providers_for_registry + for _pp in _list_providers_for_registry(): + if _pp.name not in PROVIDER_REGISTRY: + _register_plugin_provider(_pp) +except Exception: + pass + + +def get_anthropic_key() -> str: + """First usable Anthropic credential (``.env`` preferred over a stale shell export), or ``""``. + + Order mirrors ``PROVIDER_REGISTRY["anthropic"].api_key_env_vars``. + + Checks both the ``.env`` file and the process environment, preferring ``~/.hermes/.env`` so a deliberate + key rotation isn't shadowed by a stale shell export (matches the api-key resolution path — see #20591). + """ + from hermes_cli.config import get_env_value_prefer_dotenv + env_vars = PROVIDER_REGISTRY["anthropic"].api_key_env_vars + return next((v for v in (get_env_value_prefer_dotenv(var) or "" for var in env_vars) if v), "") + + +# ── Secret validation ─────────────────────────────────────────────────────────────────────────────── + +_PLACEHOLDER_SECRET_VALUES = { + "*", "**", "***", "changeme", "your_api_key", "your_api_key_here", "your-api-key", + "placeholder", "example", "dummy", "null", "none"} + + +def has_usable_secret(value: Any, *, min_length: int = 4) -> bool: + """Return True when a configured secret looks usable, not empty/placeholder.""" + if not isinstance(value, str): + return False + cleaned = value.strip() + return len(cleaned) >= min_length and cleaned.lower() not in _PLACEHOLDER_SECRET_VALUES + + +# Known API-key prefixes per provider. Only listed providers get prefix validation; everyone else +# is fail-open. Keeps an obviously malformed key in .env (truncated paste, wrong provider's key) +# from silently shadowing a valid credential-pool entry and producing opaque 401s. +# See #93593. +KNOWN_PROVIDER_KEY_PREFIXES: Dict[str, tuple] = { + "openrouter": ("sk-or-",), # all OpenRouter keys are sk-or-... (currently sk-or-v1-) +} + + +def _usable_declared_secret(provider_id: str, value: Any, source: str) -> Optional[str]: + """*value* stripped when it is a usable, prefix-valid secret; None (after warning on a provable + prefix mismatch, so it never shadows a later credential source) otherwise. Providers without a + declared prefix are fail-open.""" + val = str(value or "").strip() + if not has_usable_secret(val): + return None + prefixes = KNOWN_PROVIDER_KEY_PREFIXES.get(provider_id) + if prefixes and not any(val.startswith(p) for p in prefixes): + logger.warning( + "Ignoring %s for provider %r: value does not match the expected key " + "prefix (%s). Falling back to the next credential source. Fix or " + "remove the malformed key to silence this warning.", + source, provider_id, " or ".join(prefixes)) + return None + return val + + +def _model_level_key_env(provider_id: str) -> str: + """``model.key_env`` when config.yaml's main model targets *provider_id*, else ``""``. + + The Desktop settings UI saves registry-provider keys as a credential pointer + (``model.key_env`` → ``$HERMES_HOME/.env``) instead of the registry's canonical env var, + so credential resolution must consult it (#106336). + """ + try: + from hermes_cli.config import load_config + model_cfg = (load_config() or {}).get("model") + except Exception: + return "" + if not isinstance(model_cfg, dict): + return "" + if str(model_cfg.get("provider") or "").strip().lower() != provider_id: + return "" + return str(model_cfg.get("key_env") or model_cfg.get("api_key_env") or "").strip() + + +def _resolve_api_key_provider_secret(provider_id: str, pconfig: ProviderConfig) -> tuple[str, str]: + """Resolve an API-key provider's token and indicate where it came from.""" + if provider_id == "copilot": + # The dedicated copilot auth module does proper token validation/exchange. + try: + from hermes_cli.copilot_auth import resolve_copilot_token, get_copilot_api_token + token, source = resolve_copilot_token() + if token: + api_token, _base_url = get_copilot_api_token(token) + return api_token, source + except ValueError as exc: + logger.warning("Copilot token validation failed: %s", exc) + except Exception: + pass + return "", "" + + # Prefer ~/.hermes/.env over os.environ so a deliberate key rotation in .env isn't shadowed by + # a stale shell export inherited from a parent process (Codex CLI, test runners, etc.). + from hermes_cli.config import get_env_value_prefer_dotenv + + # Desktop-saved credential pointer: the settings UI persists registry-provider keys as + # model.key_env → $HERMES_HOME/.env (e.g. HERMES_CUSTOM_LMSTUDIO_API_KEY) while keeping + # model.provider on the registry id, so the pointer must be honored here or the UI-saved + # key is silently ignored and lmstudio falls through to its no-auth placeholder (#106336). + key_env = _model_level_key_env(provider_id) + if key_env: + val = _usable_declared_secret(provider_id, get_env_value_prefer_dotenv(key_env), key_env) + if val: + return val, key_env + + for env_var in pconfig.api_key_env_vars: + val = _usable_declared_secret(provider_id, get_env_value_prefer_dotenv(env_var), env_var) + if val: + # A provably malformed key (declared prefix mismatch) must not shadow a valid credential-pool + # entry (#93593). Warn and keep looking instead of returning it. + return val, env_var + + # Fallback: credential pool (e.g. zai key stored via auth.json). Prefer the pool's own + # selection (peek) but try the rest too so one malformed entry doesn't block a valid one. + pool_source = f"credential_pool:{provider_id}" + try: + from agent.credential_pool import load_pool + pool = load_pool(provider_id) + if pool and pool.has_credentials(): + entry = pool.peek() + candidates = [entry] if entry is not None else [] + try: + for extra in pool.entries(): + if extra is not None and all(extra is not c for c in candidates): + candidates.append(extra) + except Exception: + pass + for entry in candidates: + key = getattr(entry, "access_token", "") or getattr(entry, "runtime_api_key", "") + val = _usable_declared_secret(provider_id, key, pool_source) + if val: + return val, pool_source + except Exception: + pass + return "", "" + + +# ── Error formatting (AuthError itself lives in auth_constants) ───────────────────────────────────── + +def is_rate_limited_auth_error(error: Exception) -> bool: + """True when an :class:`AuthError` is upstream rate-limiting / quota: transient, and + re-authenticating cannot fix it, so callers should say "retry later", not ``hermes auth``.""" + return (isinstance(error, AuthError) and not error.relogin_required + and error.code == CODEX_RATE_LIMITED_CODE) + + +# Entitlement failures: Nous gets a Portal-aware message; other providers a fixed generic one (or +# the raw error when no generic text exists for the code). +_GENERIC_ENTITLEMENT_MESSAGES = { + "subscription_required": "No active paid subscription found. Please purchase/activate a subscription, then retry.", + "insufficient_credits": "Subscription credits are exhausted. Top up/renew credits, then retry."} +_ENTITLEMENT_ERROR_CODES = frozenset(_GENERIC_ENTITLEMENT_MESSAGES) | { + "subscription_expired", "no_usable_credits", "account_missing", "member_spend_cap_exceeded"} + + +def format_auth_error(error: Exception) -> str: + """Map auth failures to concise user-facing guidance.""" + if not isinstance(error, AuthError) or is_rate_limited_auth_error(error): + # Rate-limit / quota errors are not credential problems: never append "re-authenticate". + return str(error) + if error.relogin_required: + return f"{error} Run `hermes model` to re-authenticate." + if error.code in _ENTITLEMENT_ERROR_CODES: + if error.provider == "nous": + return _format_nous_entitlement_auth_error(error) + generic = _GENERIC_ENTITLEMENT_MESSAGES.get(error.code) + if generic: + return generic + if error.code == "temporarily_unavailable": + return f"{error} Please retry in a few seconds." + return str(error) + + +def _nonempty_str(value: Any) -> bool: + return isinstance(value, str) and bool(value.strip()) + + +# ── Auth Store — persistence layer for ~/.hermes/auth.json ────────────────────────────────────────── + +def _auth_file_path() -> Path: + path = get_hermes_home() / "auth.json" + # Seat belt: under pytest, refuse to touch the real user's auth store (tests that forgot to + # monkeypatch HERMES_HOME or escaped the hermetic conftest). In production: one dict lookup. + if (os.environ.get("PYTEST_CURRENT_TEST") + and _same_path(path, Path.home() / ".hermes" / "auth.json")): + raise RuntimeError( + f"Refusing to touch real user auth store during test run: {path}. " + "Set HERMES_HOME to a tmp_path in your test fixture, or run " + "via scripts/run_tests.sh for hermetic CI-parity env.") + return path + + +def _global_auth_file_path() -> Optional[Path]: + """Global-root auth.json in profile mode; None when profile and global root are the same dir. + + Read-only fallback path, so no pytest seat belt here (it lives on ``_auth_file_path()``).""" + try: + from hermes_constants import get_default_hermes_root + global_root = get_default_hermes_root() + except Exception: + return None + return None if _same_path(get_hermes_home(), global_root) else global_root / "auth.json" + + +def _load_global_auth_store() -> Dict[str, Any]: + """Load the global-root auth store (read-only fallback, mtime-memoised); ``{}`` when absent or + unreadable — a malformed global store must never break profile reads.""" + global _global_auth_store_cache + global_path = _global_auth_file_path() + if global_path is None or not global_path.exists(): + _global_auth_store_cache = None + return {} + try: + cache_key: Optional[Tuple[str, int]] = ( + str(global_path.resolve(strict=False)), global_path.stat().st_mtime_ns) + except Exception: + cache_key = None + cached = _global_auth_store_cache + if cache_key is not None and cached is not None and cached[:2] == cache_key: + return cached[2] + if os.environ.get("PYTEST_CURRENT_TEST") and os.environ.get("HOME"): + real_root = Path(os.environ["HOME"]) / ".hermes" / "auth.json" + try: + if global_path.resolve(strict=False) == real_root.resolve(strict=False): + _global_auth_store_cache = None + return {} + except Exception: + pass + try: + store = _load_auth_store(global_path) + except Exception: + _global_auth_store_cache = None + return {} + if cache_key is not None: + _global_auth_store_cache = (*cache_key, store) + return store + + +_auth_target_lock_holders: Dict[str, threading.local] = {} +_auth_target_lock_holders_guard = threading.Lock() + + +def _same_path(left: Path, right: Path) -> bool: + try: + return left.resolve(strict=False) == right.resolve(strict=False) + except Exception: + return left == right + + +def _is_same_auth_store(left: Path, right: Path) -> bool: + """True when two auth paths name ONE store rather than two copies. + ``_same_path`` resolves symlinks and ``..``; ``samefile`` adds hardlinks and bind-mounts + (same inode under two resolved names). Used by the forked-grant heal: a shared store has + no "other side" to consolidate. + + See #101356. + """ + if _same_path(left, right): + return True + try: + return left.samefile(right) + except OSError: + return False + + +def _resolved_key(path: Path) -> str: + """Canonical string for *path* (resolved when possible) used as a cache / lock-holder key.""" + try: + return str(path.resolve(strict=False)) + except Exception: + return str(path) + + +def _auth_lock_holder_for(target_path: Path) -> threading.local: + """Return a reentrancy tracker keyed to one canonical auth-store path.""" + with _auth_target_lock_holders_guard: + return _auth_target_lock_holders.setdefault(_resolved_key(target_path), threading.local()) + + +def _kernel_lock(lock_file: Any, acquire: bool) -> None: + """Non-blocking exclusive flock (fcntl) or 1-byte msvcrt lock at offset 0; ``acquire=False`` releases.""" + if fcntl: + fcntl.flock(lock_file.fileno(), (fcntl.LOCK_EX | fcntl.LOCK_NB) if acquire else fcntl.LOCK_UN) + else: + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_NBLCK if acquire else msvcrt.LK_UNLCK, 1) + + +@contextmanager +def _file_lock( + lock_path: Path, holder: threading.local, timeout_seconds: float, timeout_message: str): + """Cross-process advisory flock helper, reentrant per-thread via ``holder.depth``. + + Falls back to a depth-only guard when neither ``fcntl`` nor ``msvcrt`` is available. Callers + supply their own ``threading.local`` so independent locks (profile store vs global root vs the + shared Nous store) track reentrancy separately.""" + if getattr(holder, "depth", 0) > 0: + holder.depth += 1 + try: + yield + finally: + holder.depth -= 1 + return + + lock_path.parent.mkdir(parents=True, exist_ok=True) + with ExitStack() as stack: + lock_file = None + if fcntl is not None or msvcrt is not None: + # msvcrt.locking needs a non-empty file with the pointer at 0. This convenience write can + # race another holder's byte-range lock and raise PermissionError (reproduced with 20 + # concurrent processes on Windows); losing the race just means the file already has + # content, so swallow it. + if msvcrt and (not lock_path.exists() or lock_path.stat().st_size == 0): + try: + lock_path.write_text(" ", encoding="utf-8") + except (OSError, PermissionError): + pass + lock_file = stack.enter_context(lock_path.open("r+" if msvcrt else "a+", encoding="utf-8")) + deadline = time.monotonic() + max(1.0, timeout_seconds) + while True: + try: + _kernel_lock(lock_file, True) + break + except (BlockingIOError, OSError, PermissionError): + if time.monotonic() >= deadline: + raise TimeoutError(timeout_message) + time.sleep(0.05) + + holder.depth = 1 + try: + yield + finally: + holder.depth = 0 + if lock_file is not None: + try: + _kernel_lock(lock_file, False) + except (OSError, IOError): + pass + + +@contextmanager +def _auth_store_lock( + timeout_seconds: float = AUTH_LOCK_TIMEOUT_SECONDS, *, target_path: Optional[Path] = None): + """Cross-process advisory lock for one auth.json read/write transaction. + + ``target_path`` is required for profile-to-global write-throughs: each path has its own + reentrancy tracker and kernel lock. Lock ordering invariant: ``_auth_store_lock`` FIRST (outer), + ``_nous_shared_store_lock`` SECOND (inner), else deadlock against a concurrent shared import.""" + auth_path = target_path if target_path is not None else _auth_file_path() + with _file_lock( + auth_path.with_suffix(".lock"), _auth_lock_holder_for(auth_path), timeout_seconds, + "Timed out waiting for auth store lock"): + yield + + +def _empty_auth_store() -> Dict[str, Any]: + return {"version": AUTH_STORE_VERSION, "providers": {}} + + +def _load_auth_store(auth_file: Optional[Path] = None) -> Dict[str, Any]: + auth_file = auth_file or _auth_file_path() + if not auth_file.exists(): + return _empty_auth_store() + try: + raw = json.loads(auth_file.read_text(encoding="utf-8-sig")) + except OSError: + # Exists but unreadable (EMFILE, EACCES, EIO, stalled mount): contents are not bad, and this + # module read-modify-writes everywhere, so an empty store here is one _save_auth_store() + # away from erasing every credential. Fail loudly. + logger.warning( + "auth: could not read %s, leaving the store on disk untouched " + "rather than degrading to an empty one", + auth_file, exc_info=True) + raise + except Exception as exc: + # Genuine corruption: unparseable JSON or non-UTF-8 bytes. Preserve a copy, but never + # advertise a backup that was not written. + corrupt_path = auth_file.with_suffix(".json.corrupt") + try: + shutil.copy2(auth_file, corrupt_path) + preserved = True + except Exception: + preserved = False + logger.debug("auth: could not preserve a copy of the corrupt store at %s", corrupt_path, + exc_info=True) + logger.warning( + "auth: failed to parse %s (%s), starting with empty store. %s %s", + auth_file, exc, + "Corrupt file preserved at" if preserved else "A copy could NOT be preserved at", + corrupt_path) + return _empty_auth_store() + + if isinstance(raw, dict) and ( + isinstance(raw.get("providers"), dict) or isinstance(raw.get("credential_pool"), dict)): + raw.setdefault("providers", {}) + if isinstance(raw.get("providers"), dict): + _migrate_stale_nous_portal_url(raw["providers"]) + return raw + + if isinstance(raw, dict) and isinstance(raw.get("systems"), dict): # legacy "systems" format + systems = raw["systems"] + providers = {"nous": systems["nous_portal"]} if "nous_portal" in systems else {} + return {**_empty_auth_store(), "providers": providers, + "active_provider": "nous" if providers else None} + return _empty_auth_store() + + +def _save_private_json(target: Path, data: Any, *, fsync_dir: bool = False, **dump_kwargs: Any) -> None: + """0600 credential JSON under a 0700 parent (``secure_parent_dir`` refuses ``/``, top-level dirs + and the install tree). ``atomic_json_write`` creates the temp file 0600 before any byte lands.""" + target.parent.mkdir(parents=True, exist_ok=True) + secure_parent_dir(target) + atomic_json_write(target, data, mode=0o600, fsync_dir=fsync_dir, **dump_kwargs) + + +def _save_auth_store(auth_store: Dict[str, Any], target_path: Optional[Path] = None) -> Path: + """Atomically persist *auth_store* (0o600, parent tightened to 0o700) to the active store, or to + an explicit *target_path* (e.g. the global-root write-through for rotating xAI OAuth grants).""" + auth_file = target_path if target_path is not None else _auth_file_path() + auth_store["version"] = AUTH_STORE_VERSION + auth_store["updated_at"] = datetime.now(timezone.utc).isoformat() + _save_private_json(auth_file, auth_store, fsync_dir=True) + if target_path is not None: + # A write-through to the global root must not be masked by the mtime memo: on coarse-mtime + # filesystems a read-after-write in the same tick would keep serving the pre-write store. + global _global_auth_store_cache + _global_auth_store_cache = None + return auth_file + + +def _store_section(auth_store: Dict[str, Any], key: str) -> Dict[str, Any]: + """Return ``auth_store[key]`` as a dict, replacing a missing/non-dict value in place.""" + section = auth_store.get(key) + if not isinstance(section, dict): + section = auth_store[key] = {} + return section + + +def _provider_state_in(store: Dict[str, Any], provider_id: str) -> Optional[Dict[str, Any]]: + """Shallow copy of ``store["providers"][provider_id]`` when it is a dict, else None.""" + providers = store.get("providers") if store else None + state = providers.get(provider_id) if isinstance(providers, dict) else None + return dict(state) if isinstance(state, dict) else None + + +def _load_provider_state_with_source( + auth_store: Dict[str, Any], provider_id: str, +) -> tuple[Optional[Dict[str, Any]], Optional[Path]]: + """Provider state plus the auth.json path it came from (profile first, then the global root). + + Refresh paths that rotate single-use OAuth refresh tokens must write the updated chain back to + the same store they read.""" + state = _provider_state_in(auth_store, provider_id) + if state is not None: + return state, _auth_file_path() + global_state = _provider_state_in(_load_global_auth_store(), provider_id) + return (global_state, _global_auth_file_path()) if global_state is not None else (None, None) + + +def _load_provider_state(auth_store: Dict[str, Any], provider_id: str) -> Optional[Dict[str, Any]]: + """Provider state; in profile mode falls back to the global-root ``auth.json`` per provider (same + shadowing as ``read_credential_pool``), so profile workers see globally-authed providers.""" + return _load_provider_state_with_source(auth_store, provider_id)[0] + + +@contextmanager +def _provider_state_transaction( + provider_id: str, timeout_seconds: float = AUTH_LOCK_TIMEOUT_SECONDS): + """Lock the active auth store and any global fallback source, in that order. + + Re-reading the source after its lock is acquired prevents stale refreshes and whole-file lost + updates without inverting the documented auth -> shared lock order. ``timeout_seconds`` applies + to BOTH locks: a transaction that spans a network call must let waiters outlive that call.""" + with _auth_store_lock(timeout_seconds): + auth_store = _load_auth_store() + state, source_path = _load_provider_state_with_source(auth_store, provider_id) + if source_path is None or _same_path(source_path, _auth_file_path()): + yield auth_store, state, source_path + return + with _auth_store_lock(timeout_seconds, target_path=source_path): + yield auth_store, _provider_state_in(_load_auth_store(source_path), provider_id), source_path + + +def _store_provider_state( + auth_store: Dict[str, Any], provider_id: str, state: Dict[str, Any], *, set_active: bool = True, +) -> None: + _store_section(auth_store, "providers")[provider_id] = state + if set_active: + auth_store["active_provider"] = provider_id + + +def _save_provider_state(auth_store: Dict[str, Any], provider_id: str, state: Dict[str, Any]) -> None: + """Write *state* under ``providers`` and make *provider_id* the active provider.""" + _store_provider_state(auth_store, provider_id, state, set_active=True) + + +def _save_active_provider_state(provider_id: str, state: Dict[str, Any]) -> Path: + """Lock, load, write *state* as the active provider, save. Returns the auth store path.""" + with _auth_store_lock(): + auth_store = _load_auth_store() + _save_provider_state(auth_store, provider_id, state) + return _save_auth_store(auth_store) + + +def _persist_provider_state_to_store( + provider_id: str, state: Dict[str, Any], target_path: Path, *, set_active: bool = False, +) -> Path: + """Merge one provider into a specific auth store under that store's lock.""" + with _auth_store_lock(target_path=target_path): + auth_store = _load_auth_store(target_path) + _store_provider_state(auth_store, provider_id, dict(state), set_active=set_active) + return _save_auth_store(auth_store, target_path=target_path) + + +def _save_provider_state_to_source( + auth_store: Dict[str, Any], provider_id: str, state: Dict[str, Any], source_path: Optional[Path], +) -> None: + """Persist provider state back to the auth store it was read from. + + A token refresh rewrites credentials, not the user's choice of provider: ``active_provider`` is + left as it is (a Nous free-tier identity refreshed for a connector call must not become the + inference provider of an install that has its own key).""" + if source_path is None or _same_path(source_path, _auth_file_path()): + _store_provider_state(auth_store, provider_id, state, set_active=False) + _save_auth_store(auth_store) + else: + _persist_provider_state_to_store(provider_id, state, source_path, set_active=False) + + +def mark_provider_active_if_unset(provider_id: str) -> None: + """Set ``active_provider`` only when none is set yet: the first ``hermes auth add`` credential must + make its provider active (else setup reports "No inference provider configured"); later adds + leave the user's choice untouched.""" + with _auth_store_lock(): + auth_store = _load_auth_store() + if not (auth_store.get("active_provider") or "").strip(): + auth_store["active_provider"] = provider_id + _save_auth_store(auth_store) + + +def is_known_auth_provider(provider_id: str) -> bool: + normalized = (provider_id or "").strip().lower() + return normalized in PROVIDER_REGISTRY or normalized in SERVICE_PROVIDER_NAMES + + +def get_auth_provider_display_name(provider_id: str) -> str: + normalized = (provider_id or "").strip().lower() + if normalized in PROVIDER_REGISTRY: + return PROVIDER_REGISTRY[normalized].name + return SERVICE_PROVIDER_NAMES.get(normalized, provider_id) + + +def is_runtime_provider_routable(provider_id: str) -> bool: + """Whether runtime resolution recognizes a provider identity (a capability check, not a credential + check): ``resolve_provider`` normalization plus the special runtime identities outside the registry.""" + normalized = (provider_id or "").strip().lower() + if not normalized: + return False + if normalized in {"auto", "openrouter", "custom", "moa"} or normalized.startswith("custom:"): + return True + try: + resolve_provider(normalized) + except AuthError: + return False + return True + + +def read_credential_pool(provider_id: Optional[str] = None) -> Dict[str, Any]: + """Return the persisted credential pool, or one provider slice. + + In profile mode the global-root ``auth.json`` is a read-only fallback applied per provider ONLY + when the profile has zero entries for it (``hermes auth add`` in the profile shadows global).""" + pool = _load_auth_store().get("credential_pool") + pool = pool if isinstance(pool, dict) else {} + global_pool = _load_global_auth_store().get("credential_pool") + global_pool = global_pool if isinstance(global_pool, dict) else {} + + if provider_id is None: + merged = dict(pool) + for gp_key, gp_entries in global_pool.items(): + existing = merged.get(gp_key) + if not (isinstance(gp_entries, list) and gp_entries): + continue + if not (isinstance(existing, list) and existing): # profile wins when it has ANY entries + merged[gp_key] = list(gp_entries) + return merged + + provider_entries = pool.get(provider_id) + if isinstance(provider_entries, list) and provider_entries: + return list(provider_entries) + global_entries = global_pool.get(provider_id) + return list(global_entries) if isinstance(global_entries, list) else [] + + +_POOL_STATUS_FIELDS = ( + "last_status", "last_status_at", "last_error_code", "last_error_reason", "last_error_message", + "last_error_reset_at") + + +def _merge_disk_cooldown_state( + entry: Dict[str, Any], disk_entry: Optional[Dict[str, Any]], provider_id: str, +) -> Dict[str, Any]: + """Keep a newer on-disk cooldown/quarantine over a stale in-memory one. + + ``write_credential_pool`` persists an in-memory snapshot that may predate another process + marking the same credential exhausted/dead; without this merge the later rewrite resurrects a + rate-limited key as healthy and both processes resume hammering it.""" + if not isinstance(disk_entry, dict): + return entry + try: + from agent.credential_pool import ( + PooledCredential, STATUS_DEAD, STATUS_EXHAUSTED, _exhausted_until, _parse_absolute_timestamp, + ) + + disk_status = disk_entry.get("last_status") + if disk_status not in (STATUS_DEAD, STATUS_EXHAUSTED): + return entry + # A token change means the caller re-authed this entry and intentionally cleared its status: + # never resurrect the old cooldown onto fresh credentials. + mem_access = entry.get("access_token") or "" + disk_access = disk_entry.get("access_token") or "" + if mem_access and disk_access and mem_access != disk_access: + return entry + disk_ts = _parse_absolute_timestamp(disk_entry.get("last_status_at")) or 0.0 + mem_ts = _parse_absolute_timestamp(entry.get("last_status_at")) or 0.0 + if disk_ts <= mem_ts: + return entry + if disk_status == STATUS_EXHAUSTED: + until = _exhausted_until(PooledCredential.from_dict(provider_id, disk_entry)) + if until is None or until <= time.time(): + return entry + return {**entry, **{f: disk_entry.get(f) for f in _POOL_STATUS_FIELDS}} + except Exception: # pragma: no cover - best-effort merge + return entry + + +def _entry_ids(entries: Iterable[Any]) -> Dict[str, Dict[str, Any]]: + return {e.get("id"): e for e in entries if isinstance(e, dict) and e.get("id")} + + +def write_credential_pool( + provider_id: str, entries: List[Dict[str, Any]], *, + removed_ids: Optional[Iterable[str]] = None, + status_cleared_ids: Optional[Iterable[str]] = None, +) -> Path: + """Persist one provider's credential pool under auth.json. + + Final disk-boundary sanitizer for borrowed credentials (callers may pass raw dicts). Entries on + disk but missing from *entries* (added concurrently) are merged back unless in *removed_ids*, + so a rotation/exhaustion rewrite never drops a concurrent credential. Entries in + *status_cleared_ids* were cleared deliberately (``hermes auth reset``) and skip the + recency merge, which would otherwise read their cleared ``last_status_at`` (None -> + epoch 0) as a stale snapshot and copy a still-binding cooldown back.""" + removed = {rid for rid in (removed_ids or ()) if rid} + with _auth_store_lock(): + auth_store = _load_auth_store() + pool = _store_section(auth_store, "credential_pool") + sanitized = [ + sanitize_borrowed_credential_payload(e, provider_id) if isinstance(e, dict) else e + for e in entries] + existing_list = pool.get(provider_id) + existing_list = existing_list if isinstance(existing_list, list) else [] + existing_by_id = _entry_ids(existing_list) + new_ids = set(_entry_ids(sanitized)) + status_cleared = {cid for cid in (status_cleared_ids or ()) if cid} + merged: List[Dict[str, Any]] = [ + _merge_disk_cooldown_state( + e, None if e.get("id") in status_cleared else existing_by_id.get(e.get("id")), provider_id, + ) + if isinstance(e, dict) else e + for e in sanitized] + for disk_entry in existing_list: + disk_id = disk_entry.get("id") if isinstance(disk_entry, dict) else None + if disk_id and disk_id not in new_ids and disk_id not in removed: + merged.append(sanitize_borrowed_credential_payload(disk_entry, provider_id)) + pool[provider_id] = merged + return _save_auth_store(auth_store) + + +def _suppressed_source_list(suppressed: Dict[str, Any], provider_id: str) -> Optional[List[str]]: + """Canonical (list-form) suppressed sources for *provider_id*; a legacy mapping (keys = source + names) is migrated to the list form in place.""" + raw_sources = suppressed.get(provider_id) + if isinstance(raw_sources, list): + return raw_sources + if isinstance(raw_sources, dict): + suppressed[provider_id] = [str(name) for name in raw_sources] + return suppressed[provider_id] + return None + + +def suppress_credential_source(provider_id: str, source: str) -> None: + """Mark a credential source as suppressed so it won't be re-seeded.""" + with _auth_store_lock(): + auth_store = _load_auth_store() + suppressed = _store_section(auth_store, "suppressed_sources") + provider_list = _suppressed_source_list(suppressed, provider_id) + if provider_list is None: + provider_list = suppressed[provider_id] = [] + if source not in provider_list: + provider_list.append(source) + _save_auth_store(auth_store) + + +def is_source_suppressed(provider_id: str, source: str) -> bool: + """Check if a credential source has been suppressed by the user.""" + try: + return source in _load_auth_store().get("suppressed_sources", {}).get(provider_id, []) + except Exception: + return False + + +def unsuppress_credential_source(provider_id: str, source: str) -> bool: + """Clear a suppression marker so the source will be re-seeded on the next load.""" + with _auth_store_lock(): + auth_store = _load_auth_store() + suppressed = auth_store.get("suppressed_sources") + if not isinstance(suppressed, dict): + return False + provider_list = _suppressed_source_list(suppressed, provider_id) + if provider_list is None or source not in provider_list: + return False + provider_list.remove(source) + if not provider_list: + suppressed.pop(provider_id, None) + if not suppressed: + auth_store.pop("suppressed_sources", None) + _save_auth_store(auth_store) + return True + + +def get_provider_auth_state(provider_id: str) -> Optional[Dict[str, Any]]: + """Persisted auth state for a provider (profile first, global-root fallback), or None.""" + return _load_provider_state(_load_auth_store(), provider_id) + + +def nous_token_has_billing_scope() -> bool: + """Return True if the currently-held Nous token carries ``billing:manage``. + + Reads the persisted ``scope`` string saved at login (``_save_provider_state`` + stores ``token_data.get("scope") or scope``). A space-delimited match. Used by + the lazy step-up: if False, the first billing call will 403 ``insufficient_scope`` + anyway, but checking up front lets a surface skip a doomed round-trip. + """ + try: + state = get_provider_auth_state("nous") or {} + except Exception: + return False + scope = state.get("scope") + if not isinstance(scope, str): + return False + return NOUS_BILLING_MANAGE_SCOPE in scope.split() + + +def get_active_provider() -> Optional[str]: + """Return the currently active provider ID from auth store.""" + return _load_auth_store().get("active_provider") + + +def _active_provider_is(normalized: str) -> bool: + active = (_load_auth_store().get("active_provider") or "").strip().lower() + return bool(active) and active == normalized + + +def _slot_selects(slot: Any, normalized: str) -> bool: + return isinstance(slot, dict) and (slot.get("provider") or "").strip().lower() == normalized + + +def _config_selects_provider(normalized: str) -> bool: + """config.yaml ``model.provider``, or a MoA advisor/aggregator slot naming the provider. + + MoA presets are explicit model selections too: ``provider: anthropic`` in a MoA slot opts into + Anthropic credentials for that slot even when the main model is another provider; otherwise + Claude Code OAuth entries get pruned by ``load_pool("anthropic")`` and MoA advisors fail with + "no ANTHROPIC_API_KEY" while the picker says Anthropic is logged in.""" + from hermes_cli.config import load_config + cfg = load_config() + if _slot_selects(cfg.get("model"), normalized): + return True + + def _moa_block_matches(block: Any) -> bool: + return isinstance(block, dict) and ( + any(_slot_selects(s, normalized) for s in block.get("reference_models") or []) + or _slot_selects(block.get("aggregator"), normalized)) + + moa_cfg = cfg.get("moa") + if not isinstance(moa_cfg, dict): + return False + presets = moa_cfg.get("presets") + presets = presets.values() if isinstance(presets, dict) else () + return _moa_block_matches(moa_cfg) or any(_moa_block_matches(p) for p in presets) + + +def _explicit_pool_entry_present(normalized: str) -> bool: + """Pool rows from EXPLICIT Hermes flows (manual add / device-code / PKCE) or live env keys; + ambient borrowed sources (gh_cli / claude_code / qwen-cli) are deliberately excluded.""" + return any(_pool_entry_is_explicit(entry) for entry in read_credential_pool(normalized)) + + +# Set by Claude Code itself, not by the user explicitly configuring anthropic in Hermes. +_IMPLICIT_ENV_VARS = frozenset({"CLAUDE_CODE_OAUTH_TOKEN"}) +_EXPLICIT_POOL_SOURCES = frozenset({"device_code", "loopback_pkce", "hermes_pkce", "manual"}) +_VERTEX_PROVIDER_IDS = ("vertex", "google-vertex", "vertex-ai", "gcp-vertex", "vertexai") + + +def _env_secret(name: str) -> bool: + return has_usable_secret(os.getenv(name, "")) + + +def _explicit_env_credentials_present(normalized: str) -> bool: + """True when the user has pasted an explicit credential env var for *normalized*. + + Falls back to the models.dev ``ProviderDef`` (same shape) for non-registry providers such as + openrouter. AWS SDK providers are checked via explicit env vars only — NOT boto3's chain, so + ambient EC2 IMDS / SSO profiles never auto-surface.""" + pconfig = PROVIDER_REGISTRY.get(normalized) + if pconfig is None: + from hermes_cli.providers import get_provider + pconfig = get_provider(normalized) + if not pconfig: + return False + if pconfig.auth_type == "api_key": + return any(_env_secret(v) for v in pconfig.api_key_env_vars if v not in _IMPLICIT_ENV_VARS) + if pconfig.auth_type == "aws_sdk": + return _env_secret("AWS_BEARER_TOKEN_BEDROCK") or ( + _env_secret("AWS_ACCESS_KEY_ID") and _env_secret("AWS_SECRET_ACCESS_KEY")) + return False + + +def _pool_entry_is_explicit(entry: Any) -> bool: + """True for pool rows the user created via an explicit Hermes flow (or a still-live env key).""" + if not isinstance(entry, dict): + return False + source = str(entry.get("source") or "").strip().lower() + if source.startswith("env:"): + # A stale env-seeded entry survives in auth.json after the user deletes the env var: only + # count it when the referenced var still resolves to a usable secret NOW. + # See #55790. + env_var = entry.get("source", "").split(":", 1)[1].strip() + return bool(env_var and _env_secret(env_var)) + return bool(source) and (source in _EXPLICIT_POOL_SOURCES or source.startswith("manual:")) + + +def _keyless_provider_has_explicit_config(normalized: str) -> bool: + """Vertex / Bedrock count as explicit when Hermes-scoped routing config is present. + + Uses has_explicit_vertex_config(), NOT has_vertex_credentials(): the latter also counts an + ambient GOOGLE_APPLICATION_CREDENTIALS path (commonly set for unrelated GCP work). Only + Hermes-scoped signals (VERTEX_PROJECT_ID / vertex.project_id / VERTEX_CREDENTIALS_PATH) count + here.""" + if normalized in _VERTEX_PROVIDER_IDS: + from agent.vertex_adapter import has_explicit_vertex_config + return bool(has_explicit_vertex_config()) + if normalized == "bedrock": + from hermes_cli.config import load_config + bedrock_cfg = load_config().get("bedrock") + return isinstance(bedrock_cfg, dict) and bool(str(bedrock_cfg.get("region") or "").strip()) + return False + + +# Ordered explicit-configuration checks: ``(check, best_effort)``. Best-effort checks treat an +# exception as "no"; the env-var check is NOT best-effort — a failure there must surface rather +# than let a later, weaker signal decide. +_EXPLICIT_CONFIG_CHECKS: Tuple[Tuple[Callable[[str], bool], bool], ...] = ( + (_active_provider_is, True), (_config_selects_provider, True), + (_explicit_env_credentials_present, False), (_explicit_pool_entry_present, True), + (_keyless_provider_has_explicit_config, True)) + + +def is_provider_explicitly_configured(provider_id: str) -> bool: + """True only if the user explicitly configured this provider: auth.json ``active_provider``, + config.yaml ``model.provider`` / MoA slots, a pasted provider env var, a pool entry from a + Hermes-initiated flow, or Hermes-scoped routing config for keyless cloud-SDK providers. Ambient + borrowed credentials (gh CLI, qwen-cli, ~/.claude/.credentials.json) never count.""" + normalized = (provider_id or "").strip().lower() + for check, best_effort in _EXPLICIT_CONFIG_CHECKS: + try: + if check(normalized): + return True + except Exception as exc: + if not best_effort: + raise + logger.debug("explicit-config check %s failed for %s: %s", check.__name__, provider_id, exc) + return False + + +def clear_provider_auth(provider_id: Optional[str] = None) -> bool: + """Clear auth state for a provider (the active one when *provider_id* is None). Used by + ``hermes logout``. Returns True if something was cleared.""" + with _auth_store_lock(): + auth_store = _load_auth_store() + target = provider_id or auth_store.get("active_provider") + if not target: + return False + cleared = False + for section in ("providers", "credential_pool"): + entries = _store_section(auth_store, section) + if target in entries: + del entries[target] + cleared = True + if auth_store.get("active_provider") == target: + auth_store["active_provider"] = None + cleared = True + if cleared: + _save_auth_store(auth_store) + return cleared + + +def deactivate_provider() -> None: + """Clear active_provider without deleting credentials: used when the user switches to a non-OAuth + provider (OpenRouter, custom) so auto-resolution doesn't keep picking the OAuth provider.""" + with _auth_store_lock(): + auth_store = _load_auth_store() + auth_store["active_provider"] = None + _save_auth_store(auth_store) + + +# ── Provider Resolution — picks which provider to use ─────────────────────────────────────────────── + + +def _get_config_hint_for_unknown_provider(provider_name: str) -> str: + """Return a helpful hint string when provider resolution fails.""" + try: + from hermes_cli.config import validate_config_structure + issues = validate_config_structure() + if not issues: + return "" + lines = ["Config issue detected — run 'hermes doctor' for full diagnostics:"] + for ci in issues: + lines.append(f" [{'ERROR' if ci.severity == 'error' else 'WARNING'}] {ci.message}") + if ci.hint and ci.hint.splitlines()[0]: + lines.append(f" → {ci.hint.splitlines()[0]}") + return "\n".join(lines) + except Exception: + return "" + + +def _refuse_env_adoption_if_config_corrupt() -> None: + """Refuse env-key/pool auto-adoption of openrouter while config.yaml is corrupt. + + A corrupt config loads as ``DEFAULT_CONFIG`` (no ``model.provider``), so the env sniff would + silently adopt the PAID openrouter provider over whatever the broken config really names. + Fires ONLY on the auto path and clears itself once the file parses again.""" + try: + from hermes_cli.config import get_active_config_parse_failure + err = get_active_config_parse_failure() + if not err: + return + path = get_config_path() + except Exception as e: + logger.debug("Could not probe config parse-failure state: %s", e) + return + raise AuthError( + f"config.yaml at {path} is corrupt ({err}) — refusing to auto-select " + f"an inference provider from environment keys. Fix the YAML (a backup " + f"was saved next to it) or run hermes setup.", + code="corrupt_config") + + +# Provider aliases accepted by resolve_provider(). Plugin-declared aliases +# (plugins/model-providers//) are layered on at call time; this hardcoded +# table remains authoritative for existing names. +_PROVIDER_ALIASES: Dict[str, str] = { + "glm": "zai", "z-ai": "zai", "z.ai": "zai", "zhipu": "zai", + "google": "gemini", "google-gemini": "gemini", "google-ai-studio": "gemini", + "x-ai": "xai", "x.ai": "xai", "grok": "xai", + "xai-oauth": "xai-oauth", "x-ai-oauth": "xai-oauth", + "grok-oauth": "xai-oauth", "xai-grok-oauth": "xai-oauth", + "kimi": "kimi-coding", "kimi-for-coding": "kimi-coding", "moonshot": "kimi-coding", + "kimi-cn": "kimi-coding-cn", "moonshot-cn": "kimi-coding-cn", + "step": "stepfun", "stepfun-coding-plan": "stepfun", + "arcee-ai": "arcee", "arceeai": "arcee", + "gmi-cloud": "gmi", "gmicloud": "gmi", + "actual-computer": "actual", "actualcomputer": "actual", "aci": "actual", + "minimax-china": "minimax-cn", "minimax_cn": "minimax-cn", + "minimax-portal": "minimax-oauth", "minimax-global": "minimax-oauth", "minimax_oauth": "minimax-oauth", + "alibaba_coding": "alibaba-coding-plan", "alibaba-coding": "alibaba-coding-plan", + "alibaba_coding_plan": "alibaba-coding-plan", + "claude": "anthropic", "claude-code": "anthropic", + "github": "copilot", "github-copilot": "copilot", + "github-models": "copilot", "github-model": "copilot", + "github-copilot-acp": "copilot-acp", "copilot-acp-agent": "copilot-acp", + "aigateway": "ai-gateway", "vercel": "ai-gateway", "vercel-ai-gateway": "ai-gateway", + "opencode": "opencode-zen", "zen": "opencode-zen", + "free": "opencode-free", "opencode_free": "opencode-free", + "qwen-portal": "qwen-oauth", "qwen-cli": "qwen-oauth", "qwen-oauth": "qwen-oauth", + "hf": "huggingface", "hugging-face": "huggingface", "huggingface-hub": "huggingface", + "mimo": "xiaomi", "xiaomi-mimo": "xiaomi", + "tencent": "tencent-tokenhub", "tokenhub": "tencent-tokenhub", + "tencent-cloud": "tencent-tokenhub", "tencentmaas": "tencent-tokenhub", + "tokenplan": "tencent-tokenplan", "tencent-lkeap": "tencent-tokenplan", + "aws": "bedrock", "aws-bedrock": "bedrock", "amazon-bedrock": "bedrock", "amazon": "bedrock", + "go": "opencode-go", "opencode-go-sub": "opencode-go", + "kilo": "kilocode", "kilo-code": "kilocode", "kilo-gateway": "kilocode", + "lmstudio": "lmstudio", "lm-studio": "lmstudio", "lm_studio": "lmstudio", + # Local server aliases — route through the generic custom provider + "ollama": "custom", "ollama_cloud": "ollama-cloud", + "vllm": "custom", "llamacpp": "custom", + "llama.cpp": "custom", "llama-cpp": "custom"} + + +def _plugin_aliases() -> Dict[str, str]: + """``_PROVIDER_ALIASES`` extended with aliases declared in plugins/model-providers//.""" + aliases = dict(_PROVIDER_ALIASES) + try: + from providers import list_providers as _lp + for _pp in _lp(): + for _alias in _pp.aliases: + aliases.setdefault(_alias, _pp.name) + except Exception: + pass + return aliases + + +def _scoped_key_env_reader() -> Callable[[str], str]: + """Scope-aware key reader for provider auto-detection. + + Under multiplex a secondary profile's keys live only in its secret scope, not os.environ. Catch + ONLY ImportError: any other auxiliary_client failure must propagate rather than silently + falling back to os.getenv (a traceless fail-open).""" + try: + # Scope-aware key reads: under multiplex a secondary profile's API keys live only in its secret + # scope, not os.environ — a bare getenv here would find nothing and auto-resolution would report "No + # LLM provider configured" for every secondary profile (same class as #86905). + from agent.auxiliary_client import _scoped_key_env + return _scoped_key_env + except ImportError: + logger.warning( + "agent.auxiliary_client unavailable (%s); provider auto-detection " + "will read keys from the process environment only — under " + "multiplex, secondary profiles may report 'No LLM provider'.", + "import failed") + return lambda name: os.getenv(name) or "" + + +def _openrouter_auto_detected(scoped_key_env: Callable[[str], str]) -> bool: + """True when an OpenRouter credential exists via env key or the credential pool (a key added via + `hermes auth add openrouter` has no env var; without the pool check it is invisible to + auto-detection and requests go out with no Authorization header).""" + if any(has_usable_secret(scoped_key_env(v)) for v in ("OPENAI_API_KEY", "OPENROUTER_API_KEY")): + return True + try: + # Auto-detect an OpenRouter credential added via `hermes auth add openrouter` (manual pool entry, no + # env var). Without this, a key that only lives in the credential pool is invisible to + # auto-detection — the user sees `hermes auth list` showing the credential while requests go out + # with no Authorization header ("HTTP 401: Missing Authentication header"). The env-var check above + # only covers keys exported as OPENROUTER_API_KEY / OPENAI_API_KEY. See issue #42130. + from agent.credential_pool import load_pool as _load_pool + return bool(_load_pool("openrouter").has_credentials()) + except Exception as e: + logger.debug("Could not check OpenRouter credential pool: %s", e) + return False + + +def _logged_in_oauth_active_provider(*, skip_free_tier: bool = False) -> Optional[str]: + """auth.json ``active_provider`` when it is a registry provider that reports logged in.""" + try: + _maybe = _load_auth_store().get("active_provider") + if _maybe == "nous": + from hermes_cli.anon_auth import guest_enabled, has_guest + if has_guest() and (skip_free_tier or not guest_enabled()): + return None # the free tier is off (or being discounted), so a guest is not a login + if _maybe and _maybe in PROVIDER_REGISTRY and get_auth_status(_maybe).get("logged_in"): + return _maybe + except Exception as e: + logger.debug("Could not pre-read active auth provider: %s", e) + return None + + +def _config_model_provider() -> Tuple[Any, Optional[str]]: + """``(model_cfg, provider)`` from config.yaml when ``model.provider`` names a registry provider + or a custom OpenAI-compatible endpoint (``custom``, ``custom:``, ``vllm``/``ollama``/...). + + The normal chat/gateway path resolves config.provider upstream in resolve_requested_provider(); + this is the safety net for the direct ``resolve_provider("auto")`` callers. A configured custom + endpoint is explicit intent like any registry pin: without this rung the boot inventory + (``free_tier_bootstrap``) read a llama.cpp/vLLM install as "nothing configured" and the + dashboard's Ink chat parked every session on Setup Required while ``hermes chat`` worked + (#108383).""" + try: + from hermes_cli.config import load_config + model_cfg = (load_config() or {}).get("model") + provider = model_cfg.get("provider") if isinstance(model_cfg, dict) else None + provider = provider.strip().lower() if isinstance(provider, str) else "" + provider = _plugin_aliases().get(provider, provider) + if provider == "custom" or provider.startswith("custom:"): + return model_cfg, "custom" + if provider in PROVIDER_REGISTRY: + return model_cfg, provider + # No provider pin but a base_url the bare-custom runtime rung would honour (a loopback + # llama.cpp/vLLM/ollama server) — same explicit intent, spelled by URL. + base_url = str(model_cfg.get("base_url") or "").strip() if isinstance(model_cfg, dict) else "" + if base_url: + from hermes_cli.runtime_provider import _config_base_url_trustworthy_for_bare_custom + if _config_base_url_trustworthy_for_bare_custom(base_url, provider): + return model_cfg, "custom" + return model_cfg, None + except Exception as e: + logger.debug("Could not read config.yaml model.provider for auto-resolution: %s", e) + return None, None + + +# API-key providers never auto-selected from env: GitHub tokens are commonly present for repo/tool +# access and must not hijack inference; LM Studio is a local server whose availability isn't +# implied by LM_API_KEY (may be offline; no-auth setup uses a placeholder). Both need an explicit +# choice. +_NO_AUTO_DETECT_PROVIDERS = frozenset({"copilot", "lmstudio"}) + + +def _env_key_auto_detected( + scoped_key_env: Callable[[str], str], oauth_active: Optional[str]) -> Optional[str]: + """First registry api_key provider (registry order) with a usable env key, warning when it + preempts a logged-in OAuth provider so a stale key in ~/.hermes/.env never switches silently.""" + for pid, pconfig in PROVIDER_REGISTRY.items(): + if pconfig.auth_type != "api_key" or pid in _NO_AUTO_DETECT_PROVIDERS: + continue + for env_var in pconfig.api_key_env_vars: + if has_usable_secret(scoped_key_env(env_var)): + if oauth_active and oauth_active != pid: + logger.warning( + # An exported API key now wins over a logged-in OAuth provider (the #29285 fix). + # Surface that so a user who deliberately uses OAuth but has a stale key in + # ~/.hermes/.env isn't silently switched without knowing why. + "Provider resolved to %r via %s, preempting your " + "logged-in OAuth provider %r. If you meant to use the " + "OAuth login, unset %s or set `model.provider` " + "explicitly.", + pid, env_var, oauth_active, env_var) + return pid + return None + + +def resolve_provider( + requested: Optional[str] = None, + *, + explicit_api_key: Optional[str] = None, + explicit_base_url: Optional[str] = None, + skip_free_tier: bool = False) -> str: + """Determine which inference provider to use. + + "auto" priority (explicit intent beats a stale OAuth login): 1. CLI api_key/base_url -> + "openrouter"; 2. config.yaml ``model.provider``; 3. OPENAI_API_KEY / OPENROUTER_API_KEY -> + "openrouter"; 4. OpenRouter pool; 5. provider env keys; 6. auth.json ``active_provider``; + 7. Nous free tier when it is on and its identity exists (never created here); + 8. AWS Bedrock chain; 9. AuthError(no_provider_configured). + + ``skip_free_tier`` hides rungs 6-for-a-free-tier-identity and 7: the boot bootstrap asks + "what would carry inference if the free tier did not exist?" to decide whether a fresh identity + may become ``active_provider``. + + 1. 3. 4. 5. Provider-specific API keys (GLM, Kimi, MiniMax, ...) -> that provider 7. 8. Error (no + provider configured) See #29285. + """ + normalized = (requested or "auto").strip().lower() + normalized = _plugin_aliases().get(normalized, normalized) + + if normalized in ("openrouter", "custom") or normalized in PROVIDER_REGISTRY: + return normalized + if normalized != "auto": + hint = _get_config_hint_for_unknown_provider(normalized) + tail = (f"\n\n{hint}" if hint else " Check 'hermes model' for available providers, " + "or run 'hermes doctor' to diagnose config issues.") + raise AuthError(f"Unknown provider '{normalized}'." + tail, code="invalid_provider") + + if explicit_api_key or explicit_base_url: # one-off CLI creds always mean openrouter/custom + return "openrouter" + + _model_cfg, cfg_provider = _config_model_provider() + if cfg_provider: + return cfg_provider + + _scoped_key_env = _scoped_key_env_reader() + if _openrouter_auto_detected(_scoped_key_env): + _refuse_env_adoption_if_config_corrupt() + return "openrouter" + + # Determined up front so the env-key tier can warn when an exported key preempts it; the actual + # OAuth fallback still happens after the env-key tier. + _oauth_active = _logged_in_oauth_active_provider(skip_free_tier=skip_free_tier) + env_pid = _env_key_auto_detected(_scoped_key_env, _oauth_active) + if env_pid: + return env_pid + + # Logged-in OAuth provider is a LAST-RESORT fallback (it used to sit above the env/config + # checks, so a stale login silently overrode explicit intent). + # Logged-in OAuth provider (auth.json `active_provider`) — a LAST-RESORT fallback, chosen only when the + # user expressed no other preference above. Demoted here so explicit intent always wins. See #29285. + if _oauth_active: + if isinstance(_model_cfg, dict) and _model_cfg and not _model_cfg.get("provider"): + logger.warning( + "Provider resolved to logged-in OAuth provider %r because " + "config.yaml `model` has no `provider` key. If you meant a " + "different provider, set `model.provider` explicitly.", + _oauth_active) + return _oauth_active + + # Nous free tier, when it is on and its identity already exists. This rung sits ABOVE the Bedrock + # chain on purpose: every rung above this line is explicit user intent (CLI creds, config, env + # keys, a sign-in); the boto chain below is implicit host state, and a leftover ~/.aws profile + # used to win the first turn of a fresh install (NS-829). The rung never CREATES the identity: + # that is the boot bootstrap's job (free_tier_bootstrap), so provider resolution stays free of + # network and a fresh install without the bootstrap resolves exactly as upstream does. + if not skip_free_tier: + try: + from hermes_cli.anon_auth import guest_enabled, has_guest + if guest_enabled() and has_guest(): + return "nous" + except Exception as exc: + logger.debug("free tier check during provider resolution skipped: %s", exc) + # AWS Bedrock via the boto3 credential chain (IAM roles, SSO, env vars): implicit host state, + # below explicit keys and below the free tier. + try: + from agent.bedrock_adapter import has_aws_credentials + if has_aws_credentials(): + return "bedrock" + except ImportError: + pass # boto3 not installed + raise AuthError( + "No inference provider configured. Run 'hermes model' to choose a " + "provider and model, or set an API key (OPENROUTER_API_KEY, " + "OPENAI_API_KEY, etc.) in ~/.hermes/.env.", + code="no_provider_configured") + + +# ── Timestamp / TTL helpers ───────────────────────────────────────────────────────────────────────── + +def _utc_now_z() -> str: + """Current UTC time as an ISO-8601 string with a ``Z`` suffix (last_refresh format).""" + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _parse_iso_timestamp(value: Any) -> Optional[float]: + text = value.strip() if isinstance(value, str) else "" + if not text: + return None + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except Exception: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.timestamp() + + +def _is_expiring(expires_at_iso: Any, skew_seconds: int) -> bool: + expires_epoch = _parse_iso_timestamp(expires_at_iso) + return expires_epoch is None or expires_epoch <= (time.time() + skew_seconds) + + +def _tls_state_from_verify(verify: Any) -> Dict[str, Any]: + """Persistable ``tls`` block derived from an httpx ``verify`` value.""" + return {"insecure": verify is False, "ca_bundle": verify if isinstance(verify, str) else None} + + +def _last_auth_error_marker( + provider: str, error: "AuthError", *, reason: str, default_code: Optional[str] = None, +) -> Dict[str, Any]: + """The ``last_auth_error`` record persisted when dead OAuth material is quarantined.""" + return { + "provider": provider, "message": str(error), "reason": reason, "relogin_required": True, + "code": error.code if default_code is None else (error.code or default_code), + "at": datetime.now(timezone.utc).isoformat()} + + +_FLAT_OAUTH_TOKEN_KEYS = ("access_token", "refresh_token", "expires_at", "expires_in", "obtained_at") + + +def _quarantine_flat_oauth_state(state: Dict[str, Any], provider: str, exc: "AuthError") -> None: + """Strip dead tokens from a flat OAuth state after a terminal runtime refresh failure so + subsequent calls fail fast without a network retry (mirrors the Nous / xAI / Codex pattern).""" + for _k in _FLAT_OAUTH_TOKEN_KEYS: + state.pop(_k, None) + state["last_auth_error"] = _last_auth_error_marker( + provider, exc, reason="runtime_refresh_failure", default_code="refresh_failed") + + +def _coerce_ttl_seconds(expires_in: Any) -> int: + try: + return max(0, int(expires_in)) + except Exception: + return 0 + + +def _optional_base_url(value: Any) -> Optional[str]: + cleaned = value.strip().rstrip("/") if isinstance(value, str) else "" + return cleaned or None + + +# Valid Nous Portal hosts; a stored portal_base_url outside this set is a misconfiguration and falls +# back to the default. localhost / 127.0.0.1 are for local development and testing. +_NOUS_PORTAL_ALLOWED_HOSTS: FrozenSet[str] = frozenset({ + "portal.nousresearch.com", "localhost", "127.0.0.1"}) + +# Per-process memo for resolve_nous_access_token: startup runs one check_fn per managed tool and +# each would trigger its own ~15s blocking refresh of an expired token; a short-TTL memo collapses +# the burst into one round-trip. Callers needing freshness use force_fresh/refresh_nous_oauth_pure. +# Keyed by hermes_home_key(): the resolution itself is profile-scoped (_auth_file_path reads the +# per-turn HERMES_HOME override a multiplex gateway sets), so a single slot would hand profile A's +# Portal bearer to profile B for up to the TTL. +_RESOLVE_TOKEN_CACHE_LOCK = threading.Lock() +_RESOLVE_TOKEN_CACHE: "dict[str, tuple[float, str]]" = {} +_RESOLVE_TOKEN_CACHE_TTL_S = 5.0 + + +def _nous_portal_base_url(state: Dict[str, Any]) -> str: + """HERMES_PORTAL_BASE_URL / NOUS_PORTAL_BASE_URL is the trusted operator override and wins + OUTRIGHT, bypassing the host allowlist (which exists to reject an untrusted network-provided + value, not one the operator configured). Otherwise the stored/default value, allowlist-gated.""" + env_portal_override = _nous_portal_env_override() + if env_portal_override: + return env_portal_override.rstrip("/") + portal_base_url = _optional_base_url(state.get("portal_base_url")) or DEFAULT_NOUS_PORTAL_URL + portal_base_url = portal_base_url.rstrip("/") + host = urlparse(portal_base_url).hostname + if host and host not in _NOUS_PORTAL_ALLOWED_HOSTS: + logger.warning( + "auth: ignoring invalid portal_base_url %r (host %r not in allowlist), using default", + portal_base_url, host) + return DEFAULT_NOUS_PORTAL_URL + return portal_base_url + + +def resolve_nous_access_token( + *, + timeout_seconds: float = 15.0, + insecure: Optional[bool] = None, + ca_bundle: Optional[str] = None, + refresh_skew_seconds: int = ACCESS_TOKEN_REFRESH_SKEW_SECONDS) -> str: + """Resolve a refresh-aware Nous Portal access token for managed tool gateways.""" + # Only a default-TLS resolution is memoised; error paths never populate the memo. + memoable = not insecure and ca_bundle is None + cache_key = hermes_home_key() + if memoable: + with _RESOLVE_TOKEN_CACHE_LOCK: + cached = _RESOLVE_TOKEN_CACHE.get(cache_key) + if cached is not None and (time.monotonic() - cached[0]) < _RESOLVE_TOKEN_CACHE_TTL_S: + return cached[1] + + def _memo(token: str) -> str: + if memoable: + with _RESOLVE_TOKEN_CACHE_LOCK: + _RESOLVE_TOKEN_CACHE[cache_key] = (time.monotonic(), token) + return token + + with _provider_state_transaction("nous") as (auth_store, state, state_source_path): + if not state: + raise _nous_err("Hermes is not logged into Nous Portal.", relogin=True) + portal_base_url = _nous_portal_base_url(state) + client_id = str(state.get("client_id") or DEFAULT_NOUS_CLIENT_ID) + verify = _resolve_verify(insecure=insecure, ca_bundle=ca_bundle, auth_state=state) + persist = lambda: _save_provider_state_to_source( # noqa: E731 + auth_store, "nous", state, state_source_path) + + lock_timeout = max(timeout_seconds + 5.0, AUTH_LOCK_TIMEOUT_SECONDS) + with _nous_shared_store_lock(timeout_seconds=lock_timeout): + from hermes_cli.anon_auth import is_guest_state, refresh_guest_state + if is_guest_state(state): + # Guest seam: the anon_ credential is the identity; a first use has no access token + # yet and an expired one is re-exchanged. No refresh token, no quarantine. + access_token = state.get("access_token") + if isinstance(access_token, str) and access_token and not _is_expiring( + state.get("expires_at"), refresh_skew_seconds): + return _memo(access_token) + with httpx.Client(timeout=httpx.Timeout(timeout_seconds or 15.0), + headers={"Accept": "application/json"}, verify=verify) as client: + refresh_guest_state(state, client) + persist() + _write_shared_nous_state(state) + return _memo(state["access_token"]) + + merged_shared = _merge_shared_nous_oauth_state(state) + access_token = state.get("access_token") + refresh_token = state.get("refresh_token") + if not isinstance(access_token, str) or not access_token: + raise _nous_err("No access token found for Nous Portal login.", relogin=True) + + if not _is_expiring(state.get("expires_at"), refresh_skew_seconds): + if merged_shared: + persist() + # Memoise the valid-token fast path too: each check_fn otherwise pays two + # cross-process file locks to get here. The token has >= refresh_skew_seconds (>= + # 120s) of life, so a 5s memo can never serve an expired token. + return _memo(access_token) + + if not isinstance(refresh_token, str) or not refresh_token: + raise _nous_err("Session expired and no refresh token is available.", relogin=True) + + with httpx.Client(timeout=httpx.Timeout(timeout_seconds or 15.0), + headers={"Accept": "application/json"}, verify=verify) as client: + refreshed = _refresh_nous_or_quarantine( + client=client, auth_store=auth_store, state=state, portal_base_url=portal_base_url, + client_id=client_id, refresh_token=refresh_token, + reason="managed_access_token_refresh_failure", persist=persist) + + _apply_nous_refreshed_tokens(state, refreshed, refresh_token) + state["portal_base_url"] = portal_base_url + state["client_id"] = client_id + state["tls"] = _tls_state_from_verify(verify) + persist() + _write_shared_nous_state(state) + return _memo(state["access_token"]) + + +# ── Status helpers ────────────────────────────────────────────────────────────────────────────────── + +# Process-level memo for get_nous_auth_status(): it validates via a synchronous refresh POST +# (~350ms) and read-only UI surfaces call it many times per render (~31x per menu paint), burning +# single-use refresh tokens. Keyed on auth.json path + mtime so profile switches don't share a memo +# and login/logout/add/remove invalidate naturally. +_NOUS_AUTH_STATUS_CACHE_TTL = 15.0 # seconds +_nous_auth_status_cache: Optional[Tuple[float, str, Optional[float], Dict[str, Any]]] = None + +# mtime-keyed memo for _load_global_auth_store(): (path, mtime_ns, store); same invalidation rule. +_global_auth_store_cache: Optional[Tuple[str, int, Dict[str, Any]]] = None + + +def _auth_file_cache_key() -> Tuple[str, Optional[float]]: + auth_file = _auth_file_path() + try: + return _resolved_key(auth_file), auth_file.stat().st_mtime + except Exception: # missing file included: key without an mtime + return _resolved_key(auth_file), None + + +def invalidate_nous_auth_status_cache() -> None: + """Clear the get_nous_auth_status() memo (for code paths that mutate Nous auth state without + touching auth.json, e.g. tests; login/logout invalidate via the mtime check automatically).""" + global _nous_auth_status_cache + _nous_auth_status_cache = None + + +def get_nous_auth_status() -> Dict[str, Any]: + """Status snapshot for Nous auth, memoised ~15s keyed on the auth.json mtime. + + Prefers the auth-store provider state (the live source of truth for refresh) and validates it by + resolving runtime credentials so revoked refresh sessions do not show up as a healthy login.""" + global _nous_auth_status_cache + now = time.monotonic() + auth_file_key, mtime = _auth_file_cache_key() + cached = _nous_auth_status_cache + if (cached is not None and cached[1:3] == (auth_file_key, mtime) + and (now - cached[0]) < _NOUS_AUTH_STATUS_CACHE_TTL): + return dict(cached[3]) + status = _compute_nous_auth_status() + _nous_auth_status_cache = (now, auth_file_key, mtime, dict(status)) + return status + + +@dataclass(frozen=True) +class OAuthProviderFlow: + """Per-provider OAuth plumbing, keyed by provider id in ``OAUTH_PROVIDER_FLOWS``. + + Callables are named (strings) and looked up in this module at call time so + ``monkeypatch.setattr("hermes_cli.auth.resolve_codex_runtime_credentials", ...)`` applies.""" + provider_id: str + resolve_fn: str + status_fn: str + terminal_refresh_codes: FrozenSet[str] = frozenset() # retrying the same refresh token cannot succeed + # ``hermes logout`` with no active provider falls back to config.yaml ``model.provider`` only + # for providers whose credentials live in auth.json. + logout_from_config: bool = False + + def resolve(self, **kwargs: Any) -> Dict[str, Any]: + return globals()[self.resolve_fn](**kwargs) + + def status(self) -> Dict[str, Any]: + return globals()[self.status_fn]() + + def is_terminal_refresh_error(self, exc: Exception) -> bool: + return ( + isinstance(exc, AuthError) and exc.provider == self.provider_id + and exc.code in self.terminal_refresh_codes and bool(exc.relogin_required)) + + +_OAUTH_GRANT_DEAD_CODES = frozenset({"invalid_grant", "invalid_token", "refresh_token_reused"}) + +OAUTH_PROVIDER_FLOWS: Dict[str, OAuthProviderFlow] = { + "nous": OAuthProviderFlow( + "nous", "resolve_nous_runtime_credentials", "get_nous_auth_status", + terminal_refresh_codes=_OAUTH_GRANT_DEAD_CODES, logout_from_config=True), + "openai-codex": OAuthProviderFlow( + "openai-codex", "resolve_codex_runtime_credentials", "get_codex_auth_status", + terminal_refresh_codes=_OAUTH_GRANT_DEAD_CODES | {"codex_refresh_failed", "codex_auth_missing_refresh_token"}, + logout_from_config=True), + "xai-oauth": OAuthProviderFlow( + "xai-oauth", "resolve_xai_oauth_runtime_credentials", "get_xai_oauth_auth_status", + terminal_refresh_codes=frozenset({"xai_refresh_failed", "xai_auth_missing_refresh_token"}), + logout_from_config=True), + "qwen-oauth": OAuthProviderFlow( + "qwen-oauth", "resolve_qwen_runtime_credentials", "get_qwen_auth_status"), + "minimax-oauth": OAuthProviderFlow( + "minimax-oauth", "resolve_minimax_oauth_runtime_credentials", "get_minimax_oauth_auth_status"), +} + + +def _is_terminal_refresh_error(exc: Exception, provider: str) -> bool: + """True when retrying the same *provider* refresh token cannot succeed.""" + return OAUTH_PROVIDER_FLOWS[provider].is_terminal_refresh_error(exc) + + +_is_terminal_nous_refresh_error = partial(_is_terminal_refresh_error, provider="nous") +_is_terminal_xai_oauth_refresh_error = partial(_is_terminal_refresh_error, provider="xai-oauth") +_is_terminal_codex_oauth_refresh_error = partial( + _is_terminal_refresh_error, provider="openai-codex") + + +def _codex_pool_rate_limited_status() -> Optional[Dict[str, Any]]: + rate_limit = _codex_pool_rate_limit_status() + if not rate_limit: + return None + return { + "logged_in": True, "auth_store": str(_auth_file_path()), + "last_refresh": rate_limit.get("last_refresh"), "auth_mode": "chatgpt", + "source": f"pool:{rate_limit.get('label') or 'unknown'}", "rate_limited": True, + "error_code": CODEX_RATE_LIMITED_CODE, + "error": (rate_limit.get("message") + or "Codex provider quota exhausted; retry after the usage limit resets."), + "reset_at": rate_limit.get("reset_at")} + + +def get_codex_auth_status() -> Dict[str, Any]: + """Status snapshot for Codex auth (pool first, then legacy provider state).""" + return _pool_first_oauth_status( + "openai-codex", is_expiring=_codex_access_token_is_expiring, auth_mode="chatgpt", + resolve=resolve_codex_runtime_credentials, on_pool_miss=_codex_pool_rate_limited_status) + + +def get_xai_oauth_auth_status() -> Dict[str, Any]: + # auth_mode is display/telemetry only; device-code is the only xAI OAuth flow, so report it + # unconditionally (auth.json may still carry a legacy ``oauth_pkce`` label). + return _pool_first_oauth_status( + "xai-oauth", is_expiring=_xai_access_token_is_expiring, auth_mode="oauth_device_code", + resolve=resolve_xai_oauth_runtime_credentials) + + +def _provider_env_base_url(pconfig: ProviderConfig) -> str: + if pconfig.id == "actual": + from hermes_cli.providers import normalize_provider + + model = read_raw_config().get("model") + if isinstance(model, dict) and normalize_provider(str(model.get("provider") or "")) == "actual": + configured_url = str(model.get("base_url") or "").strip() + if configured_url: + return configured_url + return os.getenv(pconfig.base_url_env_var, "").strip() if pconfig.base_url_env_var else "" + + +def _provider_is_keyless(provider_id: str) -> bool: + """HermesOverlay keyless flag — the same source the provider catalog and GUI contract tests use.""" + try: + from hermes_cli.providers import HERMES_OVERLAYS + return bool(getattr(HERMES_OVERLAYS.get(provider_id), "keyless", False)) + except Exception: + return False + + +def get_api_key_provider_status(provider_id: str) -> Dict[str, Any]: + """Status snapshot for API-key providers (z.ai, Kimi, MiniMax).""" + pconfig = PROVIDER_REGISTRY.get(provider_id) + if not pconfig or pconfig.auth_type != "api_key": + return {"configured": False} + status = { + "configured": True, "provider": provider_id, "name": pconfig.name, "key_source": "keyless", + "base_url": pconfig.inference_base_url, "logged_in": True} + if _provider_is_keyless(provider_id): + # Keyless providers (opencode-free) are served anonymously: every install counts as + # configured. + return status + + api_key, key_source = _resolve_api_key_provider_secret(provider_id, pconfig) + env_url = _provider_env_base_url(pconfig) + if provider_id in {"kimi-coding", "kimi-coding-cn"}: + base_url = _resolve_kimi_base_url(api_key, pconfig.inference_base_url, env_url) + else: + base_url = env_url or pconfig.inference_base_url + actual_local_noauth = False + if provider_id == "actual": + base_url = normalize_actual_base_url(base_url) + actual_local_noauth = not api_key and is_actual_local_base_url(base_url) + configured = bool(api_key) or actual_local_noauth + status.update( # logged_in mirrors configured for compat with the OAuth status shape + configured=configured, base_url=base_url, logged_in=configured, + key_source=key_source or ("local-offline" if actual_local_noauth else "")) + return status + + +def _external_process_auth_evidence(provider_id: str) -> tuple[bool, Optional[str]]: + """Best-effort POSITIVE evidence ``(verified, source)`` that an external-process CLI is authed. + + False means "not verifiable from here", NOT "signed out" (the Copilot CLI may use an OS keychain + Hermes can't read). Deliberately subprocess-free: spawning ``gh auth token`` from status + endpoints/pickers re-creates the cold-start stall copilot_auth.py avoids.""" + if provider_id != "copilot-acp": + return False, None + # 1. Supported env tokens — the same vars the Copilot CLI itself honors. + try: + from hermes_cli.copilot_auth import COPILOT_ENV_VARS, validate_copilot_token + for env_var in COPILOT_ENV_VARS: + val = os.getenv(env_var, "").strip() + if val and validate_copilot_token(val)[0]: + return True, f"env: {env_var}" + except Exception as exc: + logger.debug("copilot-acp env token evidence check failed: %s", exc) + # 2. The Copilot CLI's own plaintext token store (written by `copilot login` when no OS keychain + # is available). The file is JSONC — strip //-comment lines before parsing. + try: + cli_config = os.path.expanduser("~/.copilot/config.json") + if os.path.isfile(cli_config): + with open(cli_config, "r", encoding="utf-8", errors="ignore") as fh: + raw = "\n".join( + line for line in fh.read().splitlines() if not line.lstrip().startswith("//")) + tokens = (json.loads(raw) if raw.strip() else {}).get("copilotTokens") + if isinstance(tokens, dict) and any( + isinstance(v, str) and v.strip() for v in tokens.values()): + return True, "~/.copilot/config.json" + except Exception as exc: + logger.debug("copilot-acp CLI config evidence check failed: %s", exc) + # 3. Known on-disk GitHub Copilot credential stores (the same files models.py fingerprints). + for cred_path in ("~/.config/github-copilot/hosts.json", "~/.config/github-copilot/apps.json"): + try: + expanded = os.path.expanduser(cred_path) + if os.path.isfile(expanded) and os.path.getsize(expanded) > 2: + return True, cred_path + except OSError: + continue + return False, None + + +def _external_process_spec( + pconfig: ProviderConfig) -> tuple[str, List[str], str, Optional[str], tuple[str, ...]]: + """``(command, args, base_url, resolved_command, command_env_vars)`` for an ACP provider. + + Launch details come from the provider's own profile (copilot-acp: HERMES_COPILOT_ACP_COMMAND / + COPILOT_CLI_PATH / HERMES_COPILOT_ACP_ARGS), so out-of-tree providers describe their binary.""" + base_url = _provider_env_base_url(pconfig) or pconfig.inference_base_url + try: + from providers import get_provider_profile as _get_provider_profile + profile = _get_provider_profile(pconfig.id) + except Exception: + profile = None + command_env_vars = tuple(getattr(profile, "process_command_env_vars", ()) or ()) + args_env_var = str(getattr(profile, "process_args_env_var", "") or "") + command = (next((v for v in (os.getenv(var, "").strip() for var in command_env_vars) if v), "") + or str(getattr(profile, "process_command", "") or "")) + raw_args = os.getenv(args_env_var, "").strip() if args_env_var else "" + args = shlex.split(raw_args) if raw_args else list(getattr(profile, "process_args", ()) or []) + return command, args, base_url, shutil.which(command) if command else None, command_env_vars + + +def get_external_process_provider_status(provider_id: str) -> Dict[str, Any]: + """Status snapshot for providers that run a local subprocess. + + ``configured``/``logged_in`` are structural (executable resolves or TCP endpoint set): the + subprocess owns real auth. ``auth_verified``/``auth_source`` carry positive evidence only.""" + pconfig = PROVIDER_REGISTRY.get(provider_id) + if not pconfig or pconfig.auth_type != "external_process": + return {"configured": False} + command, args, base_url, resolved_command, _ = _external_process_spec(pconfig) + available = bool(resolved_command or base_url.startswith("acp+tcp://")) + auth_verified, auth_source = _external_process_auth_evidence(provider_id) + return { + "configured": available, "provider": provider_id, "name": pconfig.name, "command": command, + "args": args, "resolved_command": resolved_command, "base_url": base_url, + "logged_in": available, "auth_verified": auth_verified, "auth_source": auth_source} + + +def _get_aws_sdk_auth_status(target: str) -> Dict[str, Any]: + """AWS SDK providers (Bedrock) — check via boto3 credential chain.""" + try: + from agent.bedrock_adapter import has_aws_credentials + return {"logged_in": has_aws_credentials(), "provider": target} + except ImportError: + return {"logged_in": False, "provider": target, "error": "boto3 not installed"} + + +def get_auth_status(provider_id: Optional[str] = None) -> Dict[str, Any]: + """Generic auth status dispatcher: bespoke builders (``OAUTH_PROVIDER_FLOWS`` plus Spotify / + Azure Foundry) first, then the registry ``auth_type`` so a whole provider class (e.g. every + external-process ACP backend) gets a real status. Builders are looked up by NAME at call time so + tests that patch ``hermes_cli.auth.get_*_auth_status`` still apply.""" + target = (provider_id or get_active_provider() or "").strip().lower() + if not target: + return {"logged_in": False} + status_fn_name = _BESPOKE_STATUS_FUNCTIONS.get(target) + if status_fn_name: + return globals()[status_fn_name]() + pconfig = PROVIDER_REGISTRY.get(target) + if pconfig and pconfig.auth_type in _STATUS_BY_AUTH_TYPE: + return globals()[_STATUS_BY_AUTH_TYPE[pconfig.auth_type]](target) + return {"logged_in": False} + + +# Bespoke status builders (name -> looked up in this module at call time) win over the +# auth_type-keyed fallbacks below. +_BESPOKE_STATUS_FUNCTIONS: Dict[str, str] = { + **{pid: flow.status_fn for pid, flow in OAUTH_PROVIDER_FLOWS.items()}, + "spotify": "get_spotify_auth_status", + "azure-foundry": "_get_azure_foundry_auth_status"} +_STATUS_BY_AUTH_TYPE: Dict[str, str] = { + "external_process": "get_external_process_provider_status", + "api_key": "get_api_key_provider_status", + "aws_sdk": "_get_aws_sdk_auth_status"} + + +def _get_azure_foundry_auth_status() -> Dict[str, Any]: + """Structural auth status for Azure Foundry. + + ``entra_id``: ``azure-identity`` importable — never invokes the Entra credential chain (keeps + CLI startup flat; ``hermes doctor`` runs the live probe). ``api_key`` (default): usable + ``AZURE_FOUNDRY_API_KEY``.""" + info: Dict[str, Any] = {"provider": "azure-foundry"} + try: + from hermes_cli.config import load_config, get_env_value_prefer_dotenv + cfg = load_config() + except Exception: + cfg = {} + model_cfg = cfg.get("model") if isinstance(cfg, dict) else None + if not isinstance(model_cfg, dict): + model_cfg = {} + auth_mode = str(model_cfg.get("auth_mode") or "api_key").strip().lower() or "api_key" + info["auth_mode"] = auth_mode + info["base_url"] = str(model_cfg.get("base_url") or "").strip() + + if auth_mode == "entra_id": + try: + from agent.azure_identity_adapter import ( + EntraIdentityConfig, SCOPE_AI_AZURE_DEFAULT, has_azure_identity_installed) + installed = has_azure_identity_installed() + entra_cfg = model_cfg["entra"] if isinstance(model_cfg.get("entra"), dict) else {} + identity_config = EntraIdentityConfig.from_dict(entra_cfg, default_scope=SCOPE_AI_AZURE_DEFAULT) + info.update( + azure_identity_installed=installed, scope=identity_config.scope, credential_probe="not_run", + credential_verified=False, logged_in=bool(installed), + hint=( + "azure-identity is installed; live credential validation " + "is skipped here. Run `hermes doctor` to verify token acquisition." + ) if installed else ( + "azure-identity not installed. Install with: " + "pip install azure-identity (or rely on Hermes' " + "lazy-install at first use).")) + except Exception as exc: + info["logged_in"] = False + info["error"] = f"azure-identity check failed: {exc}" + return info + + try: + api_key = get_env_value_prefer_dotenv("AZURE_FOUNDRY_API_KEY") or "" + except Exception: + api_key = os.getenv("AZURE_FOUNDRY_API_KEY", "") + info["logged_in"] = has_usable_secret(api_key) + return info + + +def _default_api_key_base_url(api_key: str, default: str, env_url: str) -> str: + return env_url.rstrip("/") if env_url else default + + +def _copilot_runtime_base_url(api_key: str, default: str, env_url: str) -> str: + """Copilot's API base comes from the token-exchange response (endpoints.api, proxy-ep fallback), + authoritative for Enterprise / proxied accounts; falls back to the registry default.""" + base_url = _default_api_key_base_url(api_key, default, env_url) + try: + from hermes_cli.copilot_auth import resolve_copilot_token, get_copilot_api_token + raw_token, _ = resolve_copilot_token() + if raw_token: + resolved = (get_copilot_api_token(raw_token)[1] or "").strip() + if resolved: + base_url = resolved + except Exception as exc: + logger.debug("Copilot base URL resolution fell back to default: %s", exc) + return base_url + + +# Providers whose runtime base URL is not simply env-override-or-registry-default: +# ``(api_key, registry_default, env_override) -> base_url``. +_API_KEY_BASE_URL_RESOLVERS: Dict[str, Callable[[str, str, str], str]] = { + "kimi-coding": _resolve_kimi_base_url, + "kimi-coding-cn": _resolve_kimi_base_url, + "zai": _resolve_zai_base_url, + "copilot": _copilot_runtime_base_url, + "lmstudio": lambda *a: _normalize_lmstudio_runtime_base_url(_default_api_key_base_url(*a)), + "actual": lambda *a: normalize_actual_base_url(_default_api_key_base_url(*a))} + + +def resolve_api_key_provider_credentials(provider_id: str) -> Dict[str, Any]: + """Resolve API key and base URL for an API-key provider.""" + pconfig = PROVIDER_REGISTRY.get(provider_id) + if not pconfig or pconfig.auth_type != "api_key": + raise AuthError( + f"Provider '{provider_id}' is not an API-key provider.", + provider=provider_id, code="invalid_provider") + + api_key, key_source = _resolve_api_key_provider_secret(provider_id, pconfig) + # No-auth LM Studio: a placeholder so runtime / auxiliary_client see the local server as + # configured. doctor still reports unconfigured because the status path uses the raw secret. + if not api_key and provider_id == "lmstudio": + api_key = LMSTUDIO_NOAUTH_PLACEHOLDER + key_source = key_source or "default" + + env_url = _provider_env_base_url(pconfig) + resolve_url = _API_KEY_BASE_URL_RESOLVERS.get(provider_id, _default_api_key_base_url) + base_url = resolve_url(api_key, pconfig.inference_base_url, env_url) + # An API-key provider must never hand back an empty base URL (a set-but-empty + # COPILOT_API_BASE_URL or similar env override otherwise wedges chat inference). + if not _nonempty_str(base_url): + base_url = pconfig.inference_base_url + + if not api_key and provider_id == "actual" and is_actual_local_base_url(base_url): + api_key = ACTUAL_LOCAL_NOAUTH_PLACEHOLDER + key_source = key_source or "local-offline" + return { + "provider": provider_id, "api_key": api_key, "base_url": base_url.rstrip("/"), + "source": key_source or "default"} + + +def resolve_external_process_provider_credentials(provider_id: str) -> Dict[str, Any]: + """Resolve runtime details for local subprocess-backed providers.""" + pconfig = PROVIDER_REGISTRY.get(provider_id) + if not pconfig or pconfig.auth_type != "external_process": + raise AuthError( + f"Provider '{provider_id}' is not an external-process provider.", + provider=provider_id, code="invalid_provider") + + command, args, base_url, resolved_command, command_env_vars = _external_process_spec(pconfig) + if not resolved_command and not base_url.startswith("acp+tcp://"): + _hint = " or set " + "/".join(command_env_vars) if command_env_vars else "" + raise AuthError( + f"Could not find the '{provider_id}' CLI command " + f"'{command or '(none configured)'}'. Install it{_hint}.", + provider=provider_id, + code="missing_external_process_cli") + # api_key is a placeholder: the subprocess owns real auth. Keyed on the provider id so each + # external-process provider gets a distinct value. + return { + "provider": provider_id, "api_key": pconfig.id or provider_id, + "base_url": base_url.rstrip("/"), "command": resolved_command or command, "args": args, + "source": "process"} + + +# ── CLI Commands — login / logout ─────────────────────────────────────────────────────────────────── + +def _update_config_for_provider( + provider_id: str, inference_base_url: str, default_model: Optional[str] = None, + *, clear_default: bool = False) -> Path: + """Update config.yaml and auth.json to reflect the active provider. + + *default_model*, when given, is written as ``model.default`` in the same step so the gateway + (which re-reads config per message) can't pick up the new provider before model selection + finishes and send an OpenRouter-style ``vendor/model`` name to a direct API. *clear_default* + removes ``model.default`` in that same write, for a caller that has no model to offer and must + not leave the previous provider's model paired with the new host.""" + with _auth_store_lock(): # so auto-resolution picks this provider + auth_store = _load_auth_store() + auth_store["active_provider"] = provider_id + _save_auth_store(auth_store) + + config_path = get_config_path() + config_path.parent.mkdir(parents=True, exist_ok=True) + require_readable_config_before_write(config_path) + config = read_raw_config() + current_model = config.get("model") + if isinstance(current_model, dict): + model_cfg = dict(current_model) + else: + model_cfg = {"default": current_model.strip()} if _nonempty_str(current_model) else {} + model_cfg["provider"] = provider_id + if inference_base_url and inference_base_url.strip(): + model_cfg["base_url"] = inference_base_url.rstrip("/") + else: + model_cfg.pop("base_url", None) # clear stale base_url when switching providers + + # Built-in providers resolve credentials from env/auth state, not inline model.api_key left + # over from a previous custom provider. + from hermes_cli.config import clear_model_endpoint_credentials + clear_model_endpoint_credentials(model_cfg) + + # An OpenRouter-formatted default like "anthropic/claude-opus-4.6" fails on direct-API + # providers. + if default_model: + cur_default = model_cfg.get("default", "") + if not cur_default or "/" in cur_default: + model_cfg["default"] = default_model + elif clear_default: + model_cfg.pop("default", None) + config["model"] = model_cfg + atomic_yaml_write(config_path, config, sort_keys=False) + return config_path + + +def _get_config_provider() -> Optional[str]: + """Return model.provider from config.yaml, normalized, if present.""" + try: + config = read_raw_config() + except Exception: + return None + model = config.get("model") if config else None + provider = model.get("provider") if isinstance(model, dict) else None + return (provider.strip().lower() or None) if isinstance(provider, str) else None + + +def _should_reset_config_provider_on_logout(provider_id: Optional[str]) -> bool: + """True when logout should reset model.provider (a registry provider config.yaml selects).""" + normalized = (provider_id or "").strip().lower() + return normalized in PROVIDER_REGISTRY and _get_config_provider() == normalized + + +def _logout_default_provider_from_config() -> Optional[str]: + """Fallback logout target when auth.json has no active provider but config.yaml still selects an + OAuth provider (e.g. openai-codex) — otherwise logout said "No provider is currently logged in" + and never reset model.provider.""" + provider = _get_config_provider() + flow = OAUTH_PROVIDER_FLOWS.get(provider or "") + return provider if flow and flow.logout_from_config else None + + +def _reset_config_provider() -> Path: + """Reset config.yaml provider back to auto after logout.""" + config_path = get_config_path() + if not config_path.exists(): + return config_path + require_readable_config_before_write(config_path) + config = read_raw_config() + if not config: + return config_path + model = config.get("model") + if isinstance(model, dict): + model["provider"] = "auto" + if "base_url" in model: + model["base_url"] = OPENROUTER_BASE_URL + atomic_yaml_write(config_path, config, sort_keys=False) + return config_path + + +def login_command(args) -> None: + """Deprecated: use 'hermes model' or 'hermes setup' instead.""" + print("The 'hermes login' command has been removed.\nUse 'hermes auth' to manage credentials,\n" + "'hermes model' to select a provider, or 'hermes setup' for full setup.") + raise SystemExit(0) + + +def get_minimax_oauth_auth_status() -> Dict[str, Any]: + """Return auth status dict for MiniMax OAuth provider.""" + state = get_provider_auth_state("minimax-oauth") + if not state or not state.get("access_token"): + return {"logged_in": False, "provider": "minimax-oauth"} + try: + token_valid = datetime.fromisoformat(state.get("expires_at", "")).timestamp() > time.time() + except Exception: + token_valid = True # access_token is known non-empty here + return { + "logged_in": token_valid, "provider": "minimax-oauth", + "region": state.get("region", "global"), "expires_at": state.get("expires_at")} + + +def logout_command(args) -> None: + """Clear auth state for a provider.""" + provider_id = getattr(args, "provider", None) + if provider_id and not is_known_auth_provider(provider_id): + print(f"Unknown provider: {provider_id}") + raise SystemExit(1) + target = provider_id or get_active_provider() or _logout_default_provider_from_config() + if not target: + print("No provider is currently logged in.") + return + if target == "nous": + from hermes_cli.anon_auth import FREE_TIER_NOT_SIGNED_IN, is_guest_state + if is_guest_state(get_provider_auth_state("nous")): + # Free tier is not a login; there is nothing to log out of and nothing is cleared. + print(FREE_TIER_NOT_SIGNED_IN) + return + should_reset_config = _should_reset_config_provider_on_logout(target) + provider_name = get_auth_provider_display_name(target) + if not (clear_provider_auth(target) or should_reset_config): + print(f"No auth state found for {provider_name}.") + return + if target == "nous": + # A profile logout must not be re-adopted from the cross-profile store on the next boot. + from hermes_cli.auth_nous import _clear_shared_nous_state + _clear_shared_nous_state("logout") + if should_reset_config: + _reset_config_provider() + print(f"Logged out of {provider_name}.") + if not should_reset_config: + print("Model provider configuration was unchanged.") + elif os.getenv("OPENROUTER_API_KEY"): + print("Hermes will use OpenRouter for inference.") + else: + print("Run `hermes model` or configure an API key to use Hermes.") + + +# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ---- +# Names external plugins imported from this module before the Sep 2026 decomposition. +# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does). +# The whole block is removed by reverting the commit that added it. +from http.server import BaseHTTPRequestHandler # noqa: F401,E402 +from http.server import HTTPServer # noqa: F401,E402 +from typing import TYPE_CHECKING # noqa: F401,E402 +import base64 # noqa: F401,E402 +import hashlib # noqa: F401,E402 +from urllib.parse import parse_qs # noqa: F401,E402 +import ssl # noqa: F401,E402 +import subprocess # noqa: F401,E402 +import sys # noqa: F401,E402 +from urllib.parse import urlencode # noqa: F401,E402 + + +_PLUGIN_COMPAT_LAZY = { + 'CODEX_OAUTH_USER_AGENT': ('hermes_cli.auth_constants', 'CODEX_OAUTH_USER_AGENT'), + 'CODEX_QUOTA_PROBE_MIN_INTERVAL_SECONDS': ('hermes_cli.auth_codex', 'CODEX_QUOTA_PROBE_MIN_INTERVAL_SECONDS'), + 'DEFAULT_SPOTIFY_REDIRECT_URI': ('hermes_cli.auth_constants', 'DEFAULT_SPOTIFY_REDIRECT_URI'), + 'DEVICE_AUTH_POLL_INTERVAL_CAP_SECONDS': ('hermes_cli.auth_constants', 'DEVICE_AUTH_POLL_INTERVAL_CAP_SECONDS'), + 'MINIMAX_OAUTH_GRANT_TYPE': ('hermes_cli.auth_constants', 'MINIMAX_OAUTH_GRANT_TYPE'), + 'NOUS_INFERENCE_INVOKE_SCOPE': ('hermes_cli.auth_constants', 'NOUS_INFERENCE_INVOKE_SCOPE'), + 'NOUS_SHARED_STORE_FILENAME': ('hermes_cli.auth_nous', 'NOUS_SHARED_STORE_FILENAME'), + 'OAUTH_OVER_SSH_DOCS_URL': ('hermes_cli.auth_constants', 'OAUTH_OVER_SSH_DOCS_URL'), + 'QWEN_OAUTH_CLIENT_ID': ('hermes_cli.auth_constants', 'QWEN_OAUTH_CLIENT_ID'), + 'QWEN_OAUTH_TOKEN_URL': ('hermes_cli.auth_constants', 'QWEN_OAUTH_TOKEN_URL'), + 'SINGLE_USE_OAUTH_SINGLETON_FILES': ('hermes_cli.auth_oauth_grants', 'SINGLE_USE_OAUTH_SINGLETON_FILES'), + 'SPOTIFY_ACCESS_TOKEN_REFRESH_SKEW_SECONDS': ('hermes_cli.auth_constants', 'SPOTIFY_ACCESS_TOKEN_REFRESH_SKEW_SECONDS'), + 'SPOTIFY_DASHBOARD_URL': ('hermes_cli.auth_constants', 'SPOTIFY_DASHBOARD_URL'), + 'XAI_OAUTH_DEVICE_CODE_URL': ('hermes_cli.auth_constants', 'XAI_OAUTH_DEVICE_CODE_URL'), + 'XAI_OAUTH_DISCOVERY_URL': ('hermes_cli.auth_constants', 'XAI_OAUTH_DISCOVERY_URL'), + 'XAI_OAUTH_ISSUER': ('hermes_cli.auth_constants', 'XAI_OAUTH_ISSUER'), + 'refresh_nous_oauth_pure': ('hermes_cli.auth_nous', 'refresh_nous_oauth_pure'), +} + + +def __getattr__(name): # PEP 562 — lazy so no import cycles + target = _PLUGIN_COMPAT_LAZY.get(name) + if target is None: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + import importlib + from hermes_cli.plugin_compat import warn_once + warn_once(__name__, name, *target) + return getattr(importlib.import_module(target[0]), target[1]) +# ---- END PLUGIN-COMPAT ----