From 9ebacffed0f0b6e476d8bd7c24e758209230daaf Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:20:46 -0700 Subject: [PATCH 1/3] refactor(hermes_cli): auth.py inline single-use prefix/logout/base-url wrappers, partial terminal-refresh predicates, compact dict literals --- hermes_cli/auth.py | 191 +++++++++++++++------------------------------ 1 file changed, 64 insertions(+), 127 deletions(-) diff --git a/hermes_cli/auth.py b/hermes_cli/auth.py index 30f3624d23..1cc03394ff 100644 --- a/hermes_cli/auth.py +++ b/hermes_cli/auth.py @@ -24,6 +24,7 @@ import webbrowser # noqa: F401 (tests patch auth_mod.webbrowser.open; same mod 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 @@ -327,28 +328,20 @@ KNOWN_PROVIDER_KEY_PREFIXES: Dict[str, tuple] = { } -def _secret_matches_declared_prefix(provider_id: str, value: str) -> bool: - """False only when the provider declares key prefixes and none match (fail-open otherwise).""" - prefixes = KNOWN_PROVIDER_KEY_PREFIXES.get(provider_id) - return not prefixes or any(value.startswith(p) for p in prefixes) - - -def _warn_malformed_secret(provider_id: str, source: str) -> None: - 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(KNOWN_PROVIDER_KEY_PREFIXES.get(provider_id, ()))) - - 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.""" + 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 - if not _secret_matches_declared_prefix(provider_id, val): - _warn_malformed_secret(provider_id, source) + 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 @@ -384,10 +377,8 @@ def _resolve_api_key_provider_secret(provider_id: str, pconfig: ProviderConfig) from agent.credential_pool import load_pool pool = load_pool(provider_id) if pool and pool.has_credentials(): - candidates = [] entry = pool.peek() - if entry is not None: - candidates.append(entry) + 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): @@ -410,9 +401,8 @@ def _resolve_api_key_provider_secret(provider_id: str, pconfig: ProviderConfig) 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 - ) + 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 @@ -498,16 +488,14 @@ def _load_global_auth_store() -> Dict[str, Any]: cached_path, cached_mtime, cached_store = _global_auth_store_cache if (cached_path, cached_mtime) == cache_key: return cached_store - if os.environ.get("PYTEST_CURRENT_TEST"): - real_home_env = os.environ.get("HOME", "") - if real_home_env: - real_root = Path(real_home_env) / ".hermes" / "auth.json" - try: - if global_path.resolve(strict=False) == real_root.resolve(strict=False): - _global_auth_store_cache = None - return {} - except Exception: - pass + 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: @@ -648,9 +636,8 @@ def _load_auth_store(auth_file: Optional[Path] = None) -> Dict[str, Any]: 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.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, @@ -668,18 +655,14 @@ def _load_auth_store(auth_file: Optional[Path] = None) -> Dict[str, Any]: 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(), "providers": providers, + "active_provider": "nous" if providers else None} return _empty_auth_store() def _write_private_file_atomic( - target: Path, - payload: str, - *, - replace: Optional[Callable[[Any, Any], Any]] = None, + target: Path, payload: str, *, replace: Optional[Callable[[Any, Any], Any]] = None, fsync_dir: bool = False) -> None: """Write *payload* to *target* via a 0o600 temp file + atomic rename. @@ -700,8 +683,8 @@ def _write_private_file_atomic( try: dir_fd = os.open(str(target.parent), os.O_RDONLY) except OSError: - dir_fd = None - if dir_fd is not None: + pass + else: try: os.fsync(dir_fd) finally: @@ -1223,8 +1206,7 @@ def _refuse_env_adoption_if_config_corrupt() -> None: different one. Fires ONLY on the auto path and clears itself as soon as the file parses again. """ try: - from hermes_cli.config import get_active_config_parse_failure, get_config_path - + from hermes_cli.config import get_active_config_parse_failure err = get_active_config_parse_failure() if not err: return @@ -1402,13 +1384,11 @@ def resolve_provider( if normalized in ("openrouter", "custom") or normalized in PROVIDER_REGISTRY: return normalized if normalized != "auto": - _config_hint = _get_config_hint_for_unknown_provider(normalized) - msg = f"Unknown provider '{normalized}'." - if _config_hint: - msg += f"\n\n{_config_hint}" - else: - msg += " Check 'hermes model' for available providers, or run 'hermes doctor' to diagnose config issues." - raise AuthError(msg, code="invalid_provider") + hint = _get_config_hint_for_unknown_provider(normalized) + raise AuthError( + f"Unknown provider '{normalized}'." + (f"\n\n{hint}" if hint else ( + " Check 'hermes model' for available providers, or run 'hermes doctor' to diagnose config issues.")), + code="invalid_provider") if explicit_api_key or explicit_base_url: # one-off CLI creds always mean openrouter/custom return "openrouter" @@ -1493,11 +1473,8 @@ def _last_auth_error_marker( ) -> Dict[str, Any]: """The ``last_auth_error`` record persisted when dead OAuth material is quarantined.""" return { - "provider": provider, - "code": error.code if default_code is None else (error.code or default_code), - "message": str(error), - "reason": reason, - "relogin_required": True, + "provider": provider, "code": error.code if default_code is None else (error.code or default_code), + "message": str(error), "reason": reason, "relogin_required": True, "at": datetime.now(timezone.utc).isoformat()} @@ -1609,10 +1586,8 @@ def resolve_nous_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) - timeout = httpx.Timeout(timeout_seconds if timeout_seconds else 15.0) - with httpx.Client( - timeout=timeout, headers={"Accept": "application/json"}, verify=verify, - ) as client: + 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, @@ -1735,16 +1710,9 @@ def _is_terminal_refresh_error(exc: Exception, provider: str) -> bool: return OAUTH_PROVIDER_FLOWS[provider].is_terminal_refresh_error(exc) -def _is_terminal_nous_refresh_error(exc: Exception) -> bool: - return _is_terminal_refresh_error(exc, "nous") - - -def _is_terminal_xai_oauth_refresh_error(exc: Exception) -> bool: - return _is_terminal_refresh_error(exc, "xai-oauth") - - -def _is_terminal_codex_oauth_refresh_error(exc: Exception) -> bool: - return _is_terminal_refresh_error(exc, "openai-codex") +_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]]: @@ -1752,16 +1720,11 @@ def _codex_pool_rate_limited_status() -> Optional[Dict[str, Any]]: 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, + "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."), + "error": (rate_limit.get("message") + or "Codex provider quota exhausted; retry after the usage limit resets."), "reset_at": rate_limit.get("reset_at")} @@ -1817,11 +1780,9 @@ def get_api_key_provider_status(provider_id: str) -> Dict[str, Any]: 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( - configured=configured, base_url=base_url, - key_source=key_source or ("local-offline" if actual_local_noauth else ""), - logged_in=configured, # compat with the OAuth status shape - ) + 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 @@ -1967,22 +1928,18 @@ def _get_azure_foundry_auth_status() -> Dict[str, Any]: cfg = {} model_cfg = cfg.get("model") if isinstance(cfg, dict) else None - auth_mode = "api_key" - base_url = "" - if isinstance(model_cfg, dict): - auth_mode = str(model_cfg.get("auth_mode") or "api_key").strip().lower() or "api_key" - base_url = str(model_cfg.get("base_url") or "").strip() + 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"] = base_url + 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 = {} - if isinstance(model_cfg, dict) and isinstance(model_cfg.get("entra"), dict): - entra_cfg = model_cfg["entra"] + 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", @@ -2027,14 +1984,6 @@ def _copilot_runtime_base_url(api_key: str, default: str, env_url: str) -> str: return base_url -def _lmstudio_runtime_base_url(api_key: str, default: str, env_url: str) -> str: - return _normalize_lmstudio_runtime_base_url(_default_api_key_base_url(api_key, default, env_url)) - - -def _actual_runtime_base_url(api_key: str, default: str, env_url: str) -> str: - return normalize_actual_base_url(_default_api_key_base_url(api_key, default, env_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]] = { @@ -2042,8 +1991,8 @@ _API_KEY_BASE_URL_RESOLVERS: Dict[str, Callable[[str, str, str], str]] = { "kimi-coding-cn": _resolve_kimi_base_url, "zai": _resolve_zai_base_url, "copilot": _copilot_runtime_base_url, - "lmstudio": _lmstudio_runtime_base_url, - "actual": _actual_runtime_base_url} + "lmstudio": lambda k, d, e: _normalize_lmstudio_runtime_base_url(_default_api_key_base_url(k, d, e)), + "actual": lambda k, d, e: normalize_actual_base_url(_default_api_key_base_url(k, d, e))} def resolve_api_key_provider_credentials(provider_id: str) -> Dict[str, Any]: @@ -2094,15 +2043,11 @@ def resolve_external_process_provider_credentials(provider_id: str) -> Dict[str, 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, - # Placeholder credential: the subprocess owns real auth. Keyed on the provider id so each - # external-process provider gets a distinct value. - "api_key": pconfig.id or provider_id, - "base_url": base_url.rstrip("/"), - "command": resolved_command or command, - "args": args, - "source": "process"} + "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 ─────────────────────────────────────────────────────────────────── @@ -2166,17 +2111,10 @@ def _get_config_provider() -> Optional[str]: return (provider.strip().lower() or None) if isinstance(provider, str) else None -def _config_provider_matches(provider_id: Optional[str]) -> bool: - """Return True when config.yaml currently selects *provider_id*.""" - return bool(provider_id) and _get_config_provider() == provider_id.strip().lower() - - def _should_reset_config_provider_on_logout(provider_id: Optional[str]) -> bool: - """Return True when logout should reset the model provider config.""" - if not provider_id: - return False - normalized = provider_id.strip().lower() - return normalized in PROVIDER_REGISTRY and _config_provider_matches(normalized) + """True when logout should reset model.provider: a registry provider that 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]: @@ -2208,9 +2146,8 @@ def _reset_config_provider() -> Path: def login_command(args) -> None: """Deprecated: use 'hermes model' or 'hermes setup' instead.""" - print("The 'hermes login' command has been removed.") - print("Use 'hermes auth' to manage credentials,") - print("'hermes model' to select a provider, or 'hermes setup' for full setup.") + 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) From 83cdfdcaf56ab10fda3a2d127c4678ace64581ed Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:25:15 -0700 Subject: [PATCH 2/3] refactor(hermes_cli): auth.py collapse resolve/cache/config defensive ladders, shared _resolved_key helper --- hermes_cli/auth.py | 158 ++++++++++++++++++++------------------------- 1 file changed, 70 insertions(+), 88 deletions(-) diff --git a/hermes_cli/auth.py b/hermes_cli/auth.py index 1cc03394ff..cdd5e2dab7 100644 --- a/hermes_cli/auth.py +++ b/hermes_cli/auth.py @@ -443,17 +443,12 @@ def _auth_file_path() -> Path: # Seat belt: under pytest, refuse to touch the real user's auth store (catches tests that forgot # to monkeypatch HERMES_HOME or escaped the hermetic conftest). In production this is one dict # lookup. - if os.environ.get("PYTEST_CURRENT_TEST"): - real_home_auth = (Path.home() / ".hermes" / "auth.json").resolve(strict=False) - try: - resolved = path.resolve(strict=False) - except Exception: - resolved = path - if resolved == real_home_auth: - 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.") + 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 @@ -484,10 +479,9 @@ def _load_global_auth_store() -> Dict[str, Any]: str(global_path.resolve(strict=False)), global_path.stat().st_mtime_ns) except Exception: cache_key = None - if cache_key is not None and _global_auth_store_cache is not None: - cached_path, cached_mtime, cached_store = _global_auth_store_cache - if (cached_path, cached_mtime) == cache_key: - return cached_store + 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: @@ -502,7 +496,7 @@ def _load_global_auth_store() -> Dict[str, Any]: _global_auth_store_cache = None return {} if cache_key is not None: - _global_auth_store_cache = (cache_key[0], cache_key[1], store) + _global_auth_store_cache = (*cache_key, store) return store @@ -517,14 +511,18 @@ def _same_path(left: Path, right: Path) -> bool: return left == right +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.""" - try: - key = str(target_path.resolve(strict=False)) - except Exception: - key = str(target_path) with _auth_target_lock_holders_guard: - return _auth_target_lock_holders.setdefault(key, threading.local()) + return _auth_target_lock_holders.setdefault(_resolved_key(target_path), threading.local()) def _kernel_lock(lock_file: Any, acquire: bool) -> None: @@ -853,19 +851,17 @@ def read_credential_pool(provider_id: Optional[str] = None) -> Dict[str, Any]: the profile fully shadows global on the next read.""" pool = _load_auth_store().get("credential_pool") pool = pool if isinstance(pool, dict) else {} - global_store = _load_global_auth_store() - global_pool = global_store.get("credential_pool") if global_store else None + 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) or not gp_entries: + if not (isinstance(gp_entries, list) and gp_entries): continue - if isinstance(existing, list) and existing: # profile wins whenever it has ANY entries - continue - merged[gp_key] = list(gp_entries) + 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) @@ -939,8 +935,8 @@ def write_credential_pool( sanitized = [ sanitize_borrowed_credential_payload(e, provider_id) if isinstance(e, dict) else e for e in entries] - existing = pool.get(provider_id) - existing_list = existing if isinstance(existing, list) else [] + 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)) merged: List[Dict[str, Any]] = [ @@ -1047,8 +1043,8 @@ def _config_selects_provider(normalized: str) -> bool: if not isinstance(moa_cfg, dict): return False presets = moa_cfg.get("presets") - return _moa_block_matches(moa_cfg) or ( - isinstance(presets, dict) and any(_moa_block_matches(p) for p in presets.values())) + 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: @@ -1079,8 +1075,8 @@ def _explicit_env_credentials_present(normalized: str) -> bool: if pconfig is None: from hermes_cli.providers import get_provider pconfig = get_provider(normalized) - if not pconfig: - return False + 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": @@ -1113,8 +1109,8 @@ def _keyless_provider_has_explicit_config(normalized: str) -> bool: 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 as _load_cfg - bedrock_cfg = _load_cfg().get("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 @@ -1189,9 +1185,8 @@ def _get_config_hint_for_unknown_provider(provider_name: str) -> str: 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}") - first_hint = ci.hint.splitlines()[0] if ci.hint else "" - if first_hint: - lines.append(f" → {first_hint}") + if ci.hint and ci.hint.splitlines()[0]: + lines.append(f" → {ci.hint.splitlines()[0]}") return "\n".join(lines) except Exception: return "" @@ -1329,11 +1324,9 @@ def _config_model_provider() -> Tuple[Any, Optional[str]]: from hermes_cli.config import load_config model_cfg = (load_config() or {}).get("model") - if isinstance(model_cfg, dict): - _cfg_provider = model_cfg.get("provider") - if isinstance(_cfg_provider, str) and _cfg_provider.strip().lower() in PROVIDER_REGISTRY: - return model_cfg, _cfg_provider.strip().lower() - return model_cfg, None + provider = model_cfg.get("provider") if isinstance(model_cfg, dict) else None + provider = provider.strip().lower() if isinstance(provider, str) else "" + return model_cfg, (provider if provider in PROVIDER_REGISTRY else None) except Exception as e: logger.debug("Could not read config.yaml model.provider for auto-resolution: %s", e) return None, None @@ -1385,10 +1378,9 @@ def resolve_provider( return normalized if normalized != "auto": hint = _get_config_hint_for_unknown_provider(normalized) - raise AuthError( - f"Unknown provider '{normalized}'." + (f"\n\n{hint}" if hint else ( - " Check 'hermes model' for available providers, or run 'hermes doctor' to diagnose config issues.")), - code="invalid_provider") + 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" @@ -1473,8 +1465,8 @@ def _last_auth_error_marker( ) -> Dict[str, Any]: """The ``last_auth_error`` record persisted when dead OAuth material is quarantined.""" return { - "provider": provider, "code": error.code if default_code is None else (error.code or default_code), - "message": str(error), "reason": reason, "relogin_required": True, + "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()} @@ -1546,10 +1538,9 @@ def resolve_nous_access_token( memoable = not insecure and ca_bundle is None if memoable: with _RESOLVE_TOKEN_CACHE_LOCK: - if _RESOLVE_TOKEN_CACHE is not None: - cached_at, cached_token = _RESOLVE_TOKEN_CACHE - if (time.monotonic() - cached_at) < _RESOLVE_TOKEN_CACHE_TTL_S: - return cached_token + cached = _RESOLVE_TOKEN_CACHE + if cached is not None and (time.monotonic() - cached[0]) < _RESOLVE_TOKEN_CACHE_TTL_S: + return cached[1] def _memo(token: str) -> str: global _RESOLVE_TOKEN_CACHE @@ -1620,13 +1611,9 @@ _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: - auth_file_key = str(auth_file.resolve(strict=False)) - except Exception: - auth_file_key = str(auth_file) - try: - return auth_file_key, auth_file.stat().st_mtime + return _resolved_key(auth_file), auth_file.stat().st_mtime except Exception: # missing file included: key without an mtime - return auth_file_key, None + return _resolved_key(auth_file), None def invalidate_nous_auth_status_cache() -> None: @@ -1645,12 +1632,9 @@ def get_nous_auth_status() -> Dict[str, Any]: now = time.monotonic() auth_file_key, mtime = _auth_file_cache_key() cached = _nous_auth_status_cache - if cached is not None: - cached_at, cached_auth_file_key, cached_mtime, cached_status = cached - if ( - (cached_auth_file_key, cached_mtime) == (auth_file_key, mtime) - and (now - cached_at) < _NOUS_AUTH_STATUS_CACHE_TTL): - return dict(cached_status) + 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 @@ -1712,7 +1696,8 @@ def _is_terminal_refresh_error(exc: Exception, provider: str) -> bool: _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") +_is_terminal_codex_oauth_refresh_error = partial( + _is_terminal_refresh_error, provider="openai-codex") def _codex_pool_rate_limited_status() -> Optional[Dict[str, Any]]: @@ -1720,8 +1705,9 @@ def _codex_pool_rate_limited_status() -> Optional[Dict[str, Any]]: 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, + "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."), @@ -1848,9 +1834,8 @@ def _external_process_spec( 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), "") - if not command: - command = str(getattr(profile, "process_command", "") 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 @@ -1991,8 +1976,8 @@ _API_KEY_BASE_URL_RESOLVERS: Dict[str, Callable[[str, str, str], str]] = { "kimi-coding-cn": _resolve_kimi_base_url, "zai": _resolve_zai_base_url, "copilot": _copilot_runtime_base_url, - "lmstudio": lambda k, d, e: _normalize_lmstudio_runtime_base_url(_default_api_key_base_url(k, d, e)), - "actual": lambda k, d, e: normalize_actual_base_url(_default_api_key_base_url(k, d, e))} + "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]: @@ -2046,8 +2031,9 @@ def resolve_external_process_provider_credentials(provider_id: str) -> Dict[str, # 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"} + "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 ─────────────────────────────────────────────────────────────────── @@ -2072,11 +2058,8 @@ def _update_config_for_provider( current_model = config.get("model") if isinstance(current_model, dict): model_cfg = dict(current_model) - elif _nonempty_str(current_model): - model_cfg = {"default": current_model.strip()} else: - model_cfg = {} - + 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("/") @@ -2112,7 +2095,7 @@ def _get_config_provider() -> Optional[str]: def _should_reset_config_provider_on_logout(provider_id: Optional[str]) -> bool: - """True when logout should reset model.provider: a registry provider that config.yaml selects.""" + """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 @@ -2157,10 +2140,9 @@ def get_minimax_oauth_auth_status() -> Dict[str, Any]: if not state or not state.get("access_token"): return {"logged_in": False, "provider": "minimax-oauth"} try: - expires_at = datetime.fromisoformat(state.get("expires_at", "")).timestamp() - token_valid = (expires_at - time.time()) > 0 + token_valid = datetime.fromisoformat(state.get("expires_at", "")).timestamp() > time.time() except Exception: - token_valid = bool(state.get("access_token")) + 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")} @@ -2186,9 +2168,9 @@ def logout_command(args) -> None: if should_reset_config: _reset_config_provider() print(f"Logged out of {provider_name}.") - if should_reset_config and os.getenv("OPENROUTER_API_KEY"): - print("Hermes will use OpenRouter for inference.") - elif should_reset_config: - print("Run `hermes model` or configure an API key to use Hermes.") - else: + 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.") From 8d4e21c1f0d3cd4e9170e4ec094ec0f659540419 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:38:30 -0700 Subject: [PATCH 3/3] refactor(hermes_cli): auth_commands.py _ask prompt helper replaces 7 EOF/Ctrl-C ladders, compact imports/tables --- hermes_cli/auth_commands.py | 145 ++++++++++++------------------------ 1 file changed, 49 insertions(+), 96 deletions(-) diff --git a/hermes_cli/auth_commands.py b/hermes_cli/auth_commands.py index ab84b105dd..7b6b7f0cd1 100644 --- a/hermes_cli/auth_commands.py +++ b/hermes_cli/auth_commands.py @@ -12,22 +12,10 @@ from typing import Any, Callable import uuid from agent.credential_pool import ( - AUTH_TYPE_API_KEY, - AUTH_TYPE_OAUTH, - CUSTOM_POOL_PREFIX, - SOURCE_MANUAL, - SOURCE_MANUAL_DEVICE_CODE, - STATUS_EXHAUSTED, - STRATEGY_FILL_FIRST, - STRATEGY_ROUND_ROBIN, - STRATEGY_RANDOM, - STRATEGY_LEAST_USED, - PooledCredential, - _exhausted_until, - _normalize_custom_pool_name, - get_pool_strategy, - label_from_token, - list_custom_pool_providers, + AUTH_TYPE_API_KEY, AUTH_TYPE_OAUTH, CUSTOM_POOL_PREFIX, SOURCE_MANUAL, + SOURCE_MANUAL_DEVICE_CODE, STATUS_EXHAUSTED, STRATEGY_FILL_FIRST, STRATEGY_ROUND_ROBIN, + STRATEGY_RANDOM, STRATEGY_LEAST_USED, PooledCredential, _exhausted_until, + _normalize_custom_pool_name, get_pool_strategy, label_from_token, list_custom_pool_providers, load_pool) import hermes_cli.auth as auth_mod from hermes_cli.auth import PROVIDER_REGISTRY @@ -43,20 +31,17 @@ def _get_custom_provider_entries() -> list[dict]: """Return configured provider entries with legacy and canonical pool IDs.""" try: from hermes_cli.config import get_compatible_custom_providers, load_config - config = load_config() except Exception: return [] result: list[dict] = [] for entry in get_compatible_custom_providers(config): name = entry.get("name") if isinstance(entry, dict) else None - if not isinstance(name, str) or not name.strip(): - continue - result.append({ - **entry, - "name": name.strip(), - "pool_key": f"{CUSTOM_POOL_PREFIX}{_normalize_custom_pool_name(name)}", - "provider_key": str(entry.get("provider_key", "") or "").strip()}) + if isinstance(name, str) and name.strip(): + result.append({ + **entry, "name": name.strip(), + "pool_key": f"{CUSTOM_POOL_PREFIX}{_normalize_custom_pool_name(name)}", + "provider_key": str(entry.get("provider_key", "") or "").strip()}) return result @@ -87,9 +72,8 @@ def _resolve_custom_provider_input(raw: str) -> str | None: _PROVIDER_ALIASES = { - "or": "openrouter", "open-router": "openrouter", - "grok-oauth": "xai-oauth", "xai-oauth": "xai-oauth", "x-ai-oauth": "xai-oauth", "xai-grok-oauth": "xai-oauth", -} + "or": "openrouter", "open-router": "openrouter", "grok-oauth": "xai-oauth", + "xai-oauth": "xai-oauth", "x-ai-oauth": "xai-oauth", "xai-grok-oauth": "xai-oauth"} def _normalize_provider(provider: str) -> str: @@ -107,25 +91,20 @@ def _migrate_legacy_custom_pool_key(provider: str, legacy_key: str) -> None: legacy_entries = credential_pool.get(legacy_key) if not isinstance(legacy_entries, list) or not legacy_entries: return - current_entries = credential_pool.get(provider) merged = list(current_entries) if isinstance(current_entries, list) else [] known_ids = {e.get("id") for e in merged if isinstance(e, dict) and e.get("id")} for entry in legacy_entries: entry_id = entry.get("id") if isinstance(entry, dict) else None - if entry_id and entry_id in known_ids: - continue - merged.append(entry) - if entry_id: - known_ids.add(entry_id) - + if not entry_id or entry_id not in known_ids: + merged.append(entry) + if entry_id: + known_ids.add(entry_id) credential_pool[provider] = merged del credential_pool[legacy_key] auth_mod._save_auth_store(auth_store) - try: from hermes_cli.models import clear_provider_models_cache - clear_provider_models_cache(legacy_key) except Exception: pass @@ -136,7 +115,6 @@ def _provider_base_url(provider: str) -> str: return OPENROUTER_BASE_URL if provider.startswith(CUSTOM_POOL_PREFIX): from agent.credential_pool import _get_custom_provider_config - return str((_get_custom_provider_config(provider) or {}).get("base_url") or "").strip() configured = _configured_provider_entry(provider) if configured is not None: @@ -146,11 +124,8 @@ def _provider_base_url(provider: str) -> str: def _is_known_provider(provider: str, configured_provider: dict | None) -> bool: - return ( - provider in PROVIDER_REGISTRY - or provider == "openrouter" - or provider.startswith(CUSTOM_POOL_PREFIX) - or configured_provider is not None) + return (provider in PROVIDER_REGISTRY or provider == "openrouter" + or provider.startswith(CUSTOM_POOL_PREFIX) or configured_provider is not None) def _display_source(source: str) -> str: @@ -172,7 +147,8 @@ def _classify_exhausted_status(entry) -> tuple[str, bool]: reason = str(getattr(entry, "last_error_reason", "") or "").strip().lower() message = str(getattr(entry, "last_error_message", "") or "").strip().lower() for label, retry_window, codes, reason_tokens, message_tokens in _EXHAUSTED_CLASSES: - if code in codes or any(t in reason for t in reason_tokens) or any(t in message for t in message_tokens): + if (code in codes or any(t in reason for t in reason_tokens) + or any(t in message for t in message_tokens)): return label, retry_window return "exhausted", True @@ -204,7 +180,6 @@ def _format_exhausted_status(entry) -> str: def _anthropic_oauth_login(args) -> dict: from agent import anthropic_adapter as anthropic_mod - creds = anthropic_mod.run_hermes_oauth_login_pure() if not creds: raise SystemExit("Anthropic OAuth login did not return credentials.") @@ -273,6 +248,14 @@ _OAUTH_ADD_SPECS: dict[str, _OAuthAddSpec] = { } +def _ask(prompt: str, reader: Callable[[str], str] | None = None) -> str | None: + """Stripped answer from *reader* (default ``input``); None when the user hits EOF / Ctrl-C.""" + try: + return (reader or input)(prompt).strip() + except (EOFError, KeyboardInterrupt): + return None + + def _add_nous_oauth_credential(args, provider: str) -> None: """``hermes auth add nous --type oauth``: shared-credential import, else device-code login.""" custom_label = (getattr(args, "label", None) or "").strip() or None @@ -296,11 +279,8 @@ def _add_nous_oauth_credential(args, provider: str) -> None: found = "Found existing shared Nous OAuth credentials" print() print(found) - try: - do_import = input("Import these credentials? [Y/n]: ").strip().lower() - except (EOFError, KeyboardInterrupt): - do_import = "y" - if do_import in {"", "y", "yes"}: + do_import = _ask("Import these credentials? [Y/n]: ") + if do_import is None or do_import.lower() in {"", "y", "yes"}: print("Rehydrating Nous session from shared credentials...") rehydrated = auth_mod._try_import_shared_nous_state(timeout_seconds=timeout) if rehydrated is not None: @@ -331,9 +311,8 @@ def _unsuppress_provider_sources(provider: str) -> None: def _add_api_key_credential(args, provider: str, pool) -> None: - token = (getattr(args, "api_key", None) or "").strip() - if not token: - token = masked_secret_prompt("Paste your API key: ").strip() + token = ((getattr(args, "api_key", None) or "").strip() + or masked_secret_prompt("Paste your API key: ").strip()) if not token: raise SystemExit("No API key provided.") default_label = f"api-key-{len(pool.entries()) + 1}" @@ -358,9 +337,9 @@ def auth_add_command(args) -> None: is_custom = provider.startswith(CUSTOM_POOL_PREFIX) requested_type = str(getattr(args, "auth_type", "") or "").strip().lower() - if requested_type in {AUTH_TYPE_API_KEY, "api-key"}: + if requested_type == "api-key": requested_type = AUTH_TYPE_API_KEY - if not requested_type: + elif not requested_type: oauth_default = provider in _OAUTH_CAPABLE_PROVIDERS and not is_custom requested_type = AUTH_TYPE_OAUTH if oauth_default else AUTH_TYPE_API_KEY @@ -405,9 +384,7 @@ def auth_list_command(args) -> None: else: credential_pool = auth_mod._load_auth_store().get("credential_pool") providers = sorted({ - *PROVIDER_REGISTRY.keys(), - "openrouter", - *list_custom_pool_providers(), + *PROVIDER_REGISTRY.keys(), "openrouter", *list_custom_pool_providers(), *(e["provider_key"] for e in _get_custom_provider_entries() if e["provider_key"]), *(credential_pool.keys() if isinstance(credential_pool, dict) else ())}) for provider in providers: @@ -449,7 +426,6 @@ def auth_remove_command(args) -> None: # custom config) has a RemovalStep in agent.credential_sources; it does the source-specific # cleanup while suppression + user-facing output are centralised here. from agent.credential_sources import find_removal_step - step = find_removal_step(provider, removed.source) if step is None: # unregistered source, e.g. "manual": nothing external to clean up return @@ -574,15 +550,9 @@ def _interactive_auth() -> None: print("What would you like to do?") for i, choice in enumerate(choices, 1): print(f" {i}. {choice}") - try: - raw = input("\nChoice: ").strip() - except (EOFError, KeyboardInterrupt): - return - if not raw or raw == str(len(choices)): - return - handler = { - "1": _interactive_add, "2": _interactive_remove, "3": _interactive_reset, "4": _interactive_strategy, - }.get(raw) + raw = _ask("\nChoice: ") + handler = {"1": _interactive_add, "2": _interactive_remove, "3": _interactive_reset, + "4": _interactive_strategy}.get(raw) if handler is not None: handler() @@ -594,9 +564,8 @@ def _pick_provider(prompt: str = "Provider") -> str: print(f"\nKnown providers: {', '.join(known)}") if custom_display: print(f"Custom endpoints: {', '.join(custom_display)}") - try: - raw = line_input(f"{prompt}: ").strip() - except (EOFError, KeyboardInterrupt): + raw = _ask(f"{prompt}: ", line_input) + if raw is None: raise SystemExit() return _normalize_provider(raw) @@ -612,20 +581,16 @@ def _interactive_add() -> None: print(f"\n{provider} supports both API keys and OAuth login.") print(" 1. API key (paste a key from the provider dashboard)") print(" 2. OAuth login (authenticate via browser)") - try: - type_choice = input("Type [1/2]: ").strip() - except (EOFError, KeyboardInterrupt): + type_choice = _ask("Type [1/2]: ") + if type_choice is None: return if type_choice == "2": auth_type = "oauth" - - try: - label = line_input("Label / account name (optional): ").strip() or None - except (EOFError, KeyboardInterrupt): + label = _ask("Label / account name (optional): ", line_input) + if label is None: return - auth_add_command(SimpleNamespace( - provider=provider, auth_type=auth_type, label=label, api_key=None, + provider=provider, auth_type=auth_type, label=label or None, api_key=None, portal_url=None, inference_url=None, client_id=None, scope=None, no_browser=False, timeout=None, insecure=False, ca_bundle=None)) @@ -638,10 +603,7 @@ def _interactive_remove() -> None: return for i, e in enumerate(pool.entries(), 1): print(f" #{i} {e.label:25s} {e.auth_type:10s} {e.source}{_format_exhausted_status(e)} [id:{e.id}]") - try: - raw = line_input("Remove #, id, or label (blank to cancel): ").strip() - except (EOFError, KeyboardInterrupt): - return + raw = _ask("Remove #, id, or label (blank to cancel): ", line_input) if raw: auth_remove_command(SimpleNamespace(provider=provider, target=raw)) @@ -665,12 +627,8 @@ def _interactive_strategy() -> None: print(f"\nCurrent strategy for {provider}: {current}") print() for i, s in enumerate(strategies, 1): - marker = " ←" if s == current else "" - print(f" {i}. {s:15s} — {_STRATEGY_DESCRIPTIONS[s]}{marker}") - try: - raw = input("\nStrategy [1-4]: ").strip() - except (EOFError, KeyboardInterrupt): - return + print(f" {i}. {s:15s} — {_STRATEGY_DESCRIPTIONS[s]}{' ←' if s == current else ''}") + raw = _ask("\nStrategy [1-4]: ") if not raw: return try: @@ -678,7 +636,6 @@ def _interactive_strategy() -> None: except (ValueError, IndexError): print("Invalid choice.") return - from hermes_cli.config import load_config, save_config cfg = load_config() pool_strategies = cfg.get("credential_pool_strategies") @@ -691,12 +648,8 @@ def _interactive_strategy() -> None: _AUTH_ACTIONS = { - "add": auth_add_command, - "list": auth_list_command, - "remove": auth_remove_command, - "reset": auth_reset_command, - "status": auth_status_command, - "logout": auth_logout_command, + "add": auth_add_command, "list": auth_list_command, "remove": auth_remove_command, + "reset": auth_reset_command, "status": auth_status_command, "logout": auth_logout_command, "spotify": auth_spotify_command}