fix(cli): honour model_aliases api_key, stop cross-provider key leak (#83612)
Salvaged from PR #84199 by @RickyYii. DirectAlias gains api_key/key_env; the direct-alias override re-resolves credentials against the alias endpoint (host-gated, #28660) and reuses the pre-alias key only on an origin match; oneshot -m <alias> passes the alias key as explicit_api_key; direct-alias branch gains the OLLAMA_API_KEY host gate. Fixes #83612.
This commit is contained in:
@@ -1773,6 +1773,13 @@ display:
|
||||
# Aliases are checked BEFORE the models.dev catalog, so they can route
|
||||
# to endpoints not in the catalog (e.g. Ollama Cloud, local servers).
|
||||
#
|
||||
# An alias pointing at its own endpoint can carry that endpoint's
|
||||
# credential with `api_key` (a literal, or a "${VAR}" reference) or
|
||||
# `key_env` (an env var name). `api_key` wins if both are set. When
|
||||
# neither is set the key is resolved from the alias HOST — never from
|
||||
# whatever provider was active before the switch, which would send that
|
||||
# provider's secret to an unrelated third party.
|
||||
#
|
||||
# model_aliases:
|
||||
# opus:
|
||||
# model: claude-opus-4-6
|
||||
@@ -1785,6 +1792,11 @@ display:
|
||||
# model: glm-4.7
|
||||
# provider: custom
|
||||
# base_url: "https://ollama.com/v1"
|
||||
# theta:
|
||||
# model: theta-1
|
||||
# provider: custom
|
||||
# base_url: "https://theta.example.com/v1"
|
||||
# key_env: THETA_API_KEY # or: api_key: "${THETA_API_KEY}"
|
||||
|
||||
# =============================================================================
|
||||
# Privacy
|
||||
|
||||
@@ -48,7 +48,7 @@ from agent.models_dev import (
|
||||
get_model_info,
|
||||
list_provider_models,
|
||||
)
|
||||
from utils import base_url_host_matches, base_url_hostname
|
||||
from utils import base_url_host_matches, base_url_hostname, base_url_origin
|
||||
|
||||
# Providers whose picker model list should NOT be capped by max_models.
|
||||
# OpenCode Zen / Go are aggregators whose full catalogs (70+ models each) must
|
||||
@@ -513,10 +513,21 @@ MODEL_ALIASES: dict[str, ModelIdentity] = {
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class DirectAlias(NamedTuple):
|
||||
"""Exact model mapping that bypasses catalog resolution."""
|
||||
"""Exact model mapping that bypasses catalog resolution.
|
||||
|
||||
``api_key`` / ``key_env`` carry the alias endpoint's OWN credential.
|
||||
Without them the switch keeps whatever key the *default* provider
|
||||
resolved, which 401s against the alias host and sends that provider's
|
||||
secret to an unrelated third party (#83612).
|
||||
"""
|
||||
model: str
|
||||
provider: str
|
||||
base_url: str
|
||||
# Defaulted so existing positional construction —
|
||||
# ``DirectAlias(model, provider, base_url)`` — keeps working for callers
|
||||
# and for the string-format aliases built below.
|
||||
api_key: str = ""
|
||||
key_env: str = ""
|
||||
|
||||
|
||||
# Built-in direct aliases (can be extended via config.yaml model_aliases:)
|
||||
@@ -540,6 +551,16 @@ def _load_direct_aliases() -> dict[str, DirectAlias]:
|
||||
model: "minimax-m2.7"
|
||||
provider: custom
|
||||
base_url: "https://ollama.com/v1"
|
||||
theta:
|
||||
model: "theta-1"
|
||||
provider: custom
|
||||
base_url: "https://theta.example.com/v1"
|
||||
api_key: "sk-..." # literal, or "${THETA_API_KEY}"
|
||||
key_env: "THETA_API_KEY" # read from the environment instead
|
||||
|
||||
``api_key``/``key_env`` are the alias endpoint's own credential. When
|
||||
neither is set the key is resolved from the alias HOST, never from the
|
||||
previously active provider (#83612).
|
||||
|
||||
Also reads ``model.aliases`` (set by ``hermes config set model.aliases.xxx``)
|
||||
and converts simple string entries (``ds-flash: deepseek/deepseek-v4-flash``)
|
||||
@@ -563,6 +584,8 @@ def _load_direct_aliases() -> dict[str, DirectAlias]:
|
||||
if model:
|
||||
merged[name.strip().lower()] = DirectAlias(
|
||||
model=model, provider=provider, base_url=base_url,
|
||||
api_key=str(entry.get("api_key", "") or "").strip(),
|
||||
key_env=str(entry.get("key_env", "") or "").strip(),
|
||||
)
|
||||
|
||||
# --- model.aliases (string-based format, from config set) ---
|
||||
@@ -593,16 +616,132 @@ def _load_direct_aliases() -> dict[str, DirectAlias]:
|
||||
return merged
|
||||
|
||||
|
||||
# Identity of the config the cached aliases were built from. The cache is
|
||||
# process-global but its source is profile-local, so it must be keyed or the
|
||||
# first profile to resolve an alias pins its definitions — and, since entries
|
||||
# carry `api_key`, its credentials — for every later profile in the process.
|
||||
# Same shape `load_config()` already keys its own cache on, so a profile
|
||||
# switch (HERMES_HOME moves, so the path moves) and a config/key rotation
|
||||
# (mtime/size move) both invalidate.
|
||||
_DIRECT_ALIAS_IDENTITY: Optional[tuple] = None
|
||||
# A copy of what this loader last produced. Callers and tests seed
|
||||
# DIRECT_ALIASES both by rebinding the module attribute AND by editing it in
|
||||
# place, so neither the object's identity nor a "did we load" flag can tell
|
||||
# our own stale cache from someone else's contents. Comparing against what we
|
||||
# actually wrote does: if the dict no longer holds it, the entries are not
|
||||
# ours to discard.
|
||||
_DIRECT_ALIAS_LOADED: Optional[dict] = None
|
||||
|
||||
|
||||
def _direct_alias_source_identity() -> Optional[tuple]:
|
||||
"""Identity of the active profile's alias source, or None if unknowable.
|
||||
|
||||
None means "do not reuse the cache" — a source we cannot identify must
|
||||
not be assumed to be the one already loaded.
|
||||
"""
|
||||
try:
|
||||
from hermes_constants import get_config_path
|
||||
|
||||
path = get_config_path()
|
||||
try:
|
||||
stat = path.stat()
|
||||
except OSError:
|
||||
# A missing config is still a definite identity for this profile.
|
||||
return (str(path), None, None)
|
||||
return (str(path), stat.st_mtime_ns, stat.st_size)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _ensure_direct_aliases() -> None:
|
||||
"""Lazy-load direct aliases on first use.
|
||||
"""Load direct aliases for the ACTIVE profile, caching per config identity.
|
||||
|
||||
Mutates the existing DIRECT_ALIASES dict in place rather than rebinding
|
||||
the module attribute. This keeps `from hermes_cli.model_switch import
|
||||
DIRECT_ALIASES` references valid in callers — rebinding would leave them
|
||||
pointing at a stale empty dict.
|
||||
"""
|
||||
if not DIRECT_ALIASES:
|
||||
DIRECT_ALIASES.update(_load_direct_aliases())
|
||||
global _DIRECT_ALIAS_IDENTITY, _DIRECT_ALIAS_LOADED
|
||||
identity = _direct_alias_source_identity()
|
||||
if DIRECT_ALIASES and (
|
||||
# Contents are not what we loaded — seeded or edited by a caller.
|
||||
# Not ours to discard.
|
||||
DIRECT_ALIASES != _DIRECT_ALIAS_LOADED
|
||||
# Ours, and still the same config file at the same signature.
|
||||
or (identity is not None and identity == _DIRECT_ALIAS_IDENTITY)
|
||||
):
|
||||
return
|
||||
loaded = _load_direct_aliases()
|
||||
# clear()+update() rather than a rebind: callers hold this exact dict.
|
||||
DIRECT_ALIASES.clear()
|
||||
DIRECT_ALIASES.update(loaded)
|
||||
_DIRECT_ALIAS_IDENTITY = identity
|
||||
_DIRECT_ALIAS_LOADED = dict(loaded)
|
||||
|
||||
|
||||
def direct_alias_api_key(alias: DirectAlias) -> str:
|
||||
"""Resolve a direct alias's own credential, or "" when it has none.
|
||||
|
||||
Precedence, highest first — ``api_key`` always wins over ``key_env``, so
|
||||
an entry carrying both is not ambiguous:
|
||||
|
||||
1. ``api_key: "${VAR}"`` — indirection, read from the environment.
|
||||
2. ``api_key: "sk-..."`` — literal.
|
||||
3. ``key_env: VAR`` — read from the environment.
|
||||
4. otherwise "" — the caller resolves from the alias host instead.
|
||||
Environment reads go through the per-profile secret scope for the same
|
||||
reason the user-provider branch does: a raw ``os.environ`` read hands
|
||||
this profile whatever key the process env holds — another profile's,
|
||||
under the multiplexed gateway.
|
||||
"""
|
||||
raw = (alias.api_key or "").strip()
|
||||
if raw.startswith("${") and raw.endswith("}"):
|
||||
return _scoped_key_env(raw[2:-1].strip())
|
||||
if raw:
|
||||
return raw
|
||||
return _scoped_key_env((alias.key_env or "").strip())
|
||||
|
||||
|
||||
def direct_alias_runtime_request(alias: DirectAlias) -> tuple[str, Optional[str]]:
|
||||
"""Return ``(requested_provider, explicit_api_key)`` for resolving *alias*.
|
||||
|
||||
Single owner of the invariant that a URL-bearing direct alias resolves its
|
||||
credential for the alias HOST, never for its provider label. A label like
|
||||
``anthropic`` on an unrelated URL would otherwise reach that provider's
|
||||
explicit-runtime branch, keep the foreign URL, and fall back to the live
|
||||
vendor token. Bare ``custom`` is host-gated (#28660), so an authoritative
|
||||
URL still resolves its vendor key and a foreign one resolves none.
|
||||
|
||||
An alias with no base_url keeps its label: there is no foreign host to
|
||||
protect against, and the label is the only routing information there is.
|
||||
"""
|
||||
key = direct_alias_api_key(alias) or None
|
||||
if alias.base_url:
|
||||
return "custom", key
|
||||
return (alias.provider or "custom"), key
|
||||
|
||||
|
||||
# Hosts where plaintext HTTP is not a downgrade — a local server has no
|
||||
# network hop to intercept.
|
||||
_LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1", "0.0.0.0"})
|
||||
|
||||
|
||||
def _may_reuse_session_credential(session_base_url: str, alias_base_url: str) -> bool:
|
||||
"""Whether the session's key may follow a switch to *alias_base_url*.
|
||||
|
||||
Same hostname is NOT sufficient to authorise handing a bearer secret to a
|
||||
new URL. ``http://h`` and ``https://h:8443`` are different origins and
|
||||
different trust boundaries, so an alias that keeps the hostname but drops
|
||||
the scheme would otherwise put a live session credential on the wire in
|
||||
the clear. Require an identical (scheme, host, port), and refuse plaintext
|
||||
outside loopback.
|
||||
"""
|
||||
session = base_url_origin(session_base_url)
|
||||
alias = base_url_origin(alias_base_url)
|
||||
if not session[1] or session != alias:
|
||||
return False
|
||||
scheme, hostname, _ = alias
|
||||
return scheme == "https" or hostname in _LOOPBACK_HOSTS
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -1985,9 +2124,70 @@ def switch_model(
|
||||
_ensure_direct_aliases()
|
||||
_da = DIRECT_ALIASES.get(resolved_alias)
|
||||
if _da is not None and _da.base_url:
|
||||
base_url = _da.base_url
|
||||
# Credentials above were resolved against the DEFAULT provider.
|
||||
# Carrying that key onto the alias's endpoint both 401s and ships
|
||||
# the default provider's secret to an unrelated third-party host
|
||||
# (#83612). The alias's own endpoint decides the credential
|
||||
# instead: its declared key when it has one, otherwise a fresh
|
||||
# resolution against the alias base_url, whose env-key fallbacks
|
||||
# are gated on authoritative hosts (#28660) — so OLLAMA_API_KEY
|
||||
# still resolves for an ollama.com alias while OPENROUTER_API_KEY
|
||||
# never reaches an unrelated host.
|
||||
_alias_key = direct_alias_api_key(_da)
|
||||
if _alias_key:
|
||||
# The alias states its own credential: nothing left to
|
||||
# resolve, and re-entering the resolver would only risk a
|
||||
# second local-endpoint model probe.
|
||||
base_url = _da.base_url
|
||||
api_key = _alias_key
|
||||
elif api_key and api_key != "no-key-required" and (
|
||||
_may_reuse_session_credential(base_url, _da.base_url)
|
||||
):
|
||||
# The alias points at the very origin the resolution above
|
||||
# already produced a key for, so that key is the
|
||||
# host-appropriate one and re-entering the resolver would only
|
||||
# repeat the work — including, for a local endpoint with no
|
||||
# configured model, a second bounded /models probe.
|
||||
base_url = _da.base_url
|
||||
else:
|
||||
try:
|
||||
# Shared owner of the label-vs-host invariant; the one-shot
|
||||
# path resolves through the same helper.
|
||||
_req, _explicit = direct_alias_runtime_request(_da)
|
||||
_alias_runtime = resolve_runtime_provider(
|
||||
requested=_req,
|
||||
explicit_api_key=_explicit,
|
||||
explicit_base_url=_da.base_url,
|
||||
target_model=new_model,
|
||||
)
|
||||
except Exception:
|
||||
_alias_runtime = {}
|
||||
# The already-resolved key is reusable only when the alias
|
||||
# points at the SAME ORIGIN it was resolved for (an alias that
|
||||
# just pins a model on the endpoint already in use). Across
|
||||
# origins it is the leak, so it is dropped, not carried.
|
||||
_same_host = _may_reuse_session_credential(base_url, _da.base_url)
|
||||
base_url = _alias_runtime.get("base_url", "") or _da.base_url
|
||||
# The resolver reports "no key found" with the
|
||||
# `no-key-required` placeholder rather than "". Normalise it
|
||||
# so a same-host credential still outranks the placeholder.
|
||||
_resolved_key = _alias_runtime.get("api_key", "")
|
||||
if _resolved_key == "no-key-required":
|
||||
_resolved_key = ""
|
||||
api_key = (
|
||||
_resolved_key
|
||||
or (api_key if _same_host else "")
|
||||
or "no-key-required"
|
||||
)
|
||||
api_mode = "" # clear so determine_api_mode re-detects from URL
|
||||
if target_provider.strip().lower() == "ollama":
|
||||
# Upstream's providers.ollama refinement: pick up the
|
||||
# configured key only for the configured native root, and drop
|
||||
# both the key and the provider-level headers for any other
|
||||
# origin. Orthogonal to the resolution above and kept as-is —
|
||||
# except that it is skipped when the alias declared its own
|
||||
# credential, since an explicit api_key/key_env outranks a
|
||||
# provider-level config key (this PR's documented precedence).
|
||||
if not _alias_key and target_provider.strip().lower() == "ollama":
|
||||
_ollama_cfg = _get_provider_config_dict("ollama")
|
||||
_ollama_cfg_base = str(
|
||||
_ollama_cfg.get("base_url")
|
||||
|
||||
@@ -399,6 +399,7 @@ def _run_agent(
|
||||
# the caller just asked for.
|
||||
effective_provider = (provider or "").strip() or None
|
||||
explicit_base_url_from_alias: Optional[str] = None
|
||||
explicit_api_key_from_alias: Optional[str] = None
|
||||
if effective_provider is None and (model or env_model):
|
||||
# Only auto-detect when the model was explicitly requested via arg or
|
||||
# env var (not when it came from config — that's the "use my defaults"
|
||||
@@ -417,6 +418,20 @@ def _run_agent(
|
||||
if direct is not None:
|
||||
effective_model = direct.model
|
||||
effective_provider = direct.provider
|
||||
# Resolve the alias through the SAME owner the interactive
|
||||
# `/model` path uses. Passing `direct.provider` alongside a
|
||||
# URL-bearing alias would let a label like `anthropic` reach
|
||||
# that provider's explicit-runtime branch, keep the alias's
|
||||
# unrelated base_url, and fall back to the live vendor token —
|
||||
# a bearer credential crossing an origin boundary. The helper
|
||||
# forces bare `custom` for URL-bearing aliases (host-gated,
|
||||
# #28660) and carries the alias's own key when it declares one.
|
||||
try:
|
||||
effective_provider, explicit_api_key_from_alias = (
|
||||
_ms.direct_alias_runtime_request(direct)
|
||||
)
|
||||
except Exception:
|
||||
explicit_api_key_from_alias = None
|
||||
if direct.base_url:
|
||||
explicit_base_url_from_alias = direct.base_url.rstrip("/")
|
||||
else:
|
||||
@@ -436,6 +451,7 @@ def _run_agent(
|
||||
requested=effective_provider,
|
||||
target_model=effective_model or None,
|
||||
explicit_base_url=explicit_base_url_from_alias,
|
||||
explicit_api_key=explicit_api_key_from_alias,
|
||||
)
|
||||
|
||||
# Pull in explicit toolsets when provided; otherwise use whatever the user
|
||||
|
||||
@@ -1226,9 +1226,15 @@ def _resolve_named_custom_runtime(
|
||||
return pool_result
|
||||
_da_is_openai_url = base_url_host_matches(base_url, "openai.com") or base_url_host_matches(base_url, "openai.azure.com")
|
||||
_da_is_openrouter = base_url_host_matches(base_url, "openrouter.ai")
|
||||
_da_is_ollama_url = base_url_host_matches(base_url, "ollama.com")
|
||||
api_key_candidates = [
|
||||
(explicit_api_key or "").strip(),
|
||||
# Gate env key fallbacks on authoritative hosts (#28660)
|
||||
# OLLAMA_API_KEY needs its own gate here: _host_derived_api_key
|
||||
# deliberately skips it, expecting an explicit host-matched path
|
||||
# like this one (GHSA-76xc-57q6-vm5m). Without it a `model_aliases:`
|
||||
# entry pointing at Ollama Cloud resolved no key at all.
|
||||
(_getenv("OLLAMA_API_KEY", "").strip() if _da_is_ollama_url else ""),
|
||||
(_getenv("OPENAI_API_KEY", "").strip() if _da_is_openai_url else ""),
|
||||
(_getenv("OPENROUTER_API_KEY", "").strip() if _da_is_openrouter else ""),
|
||||
# Bonus (#28660): derive `<VENDOR>_API_KEY` from the host so users
|
||||
|
||||
@@ -48,6 +48,11 @@ model_aliases:
|
||||
model: qwen3.5:397b
|
||||
provider: custom
|
||||
base_url: "https://ollama.com/v1"
|
||||
theta:
|
||||
model: theta-1
|
||||
provider: custom
|
||||
base_url: "https://theta.example.com/v1"
|
||||
key_env: THETA_API_KEY # or: api_key: "${THETA_API_KEY}"
|
||||
|
||||
# Short form ("provider/model"), also via CLI:
|
||||
# hermes config set model.aliases.fav openrouter/anthropic/claude-sonnet-4.6
|
||||
@@ -58,6 +63,11 @@ model:
|
||||
|
||||
`/model fav` — session-scoped; add `--global` to persist as default.
|
||||
|
||||
An alias with its own `base_url` authenticates with its own credential
|
||||
(`api_key`, which also accepts a `"${VAR}"` reference, or `key_env`). With
|
||||
neither set the key is resolved from the alias HOST, never carried over from
|
||||
the provider that was active before the switch.
|
||||
|
||||
Built-in aliases (catalog-resolved against the active provider): `sonnet`,
|
||||
`opus`, `haiku`, `claude`, `gpt5`, `gpt`, `codex`, `o3`, `o4`, `gemini`,
|
||||
`deepseek`, `grok`, `llama`, `qwen`, `minimax`, `nemotron`, `kimi`, `glm`,
|
||||
|
||||
850
tests/hermes_cli/test_model_alias_credentials_83612.py
Normal file
850
tests/hermes_cli/test_model_alias_credentials_83612.py
Normal file
@@ -0,0 +1,850 @@
|
||||
"""Direct-alias (``model_aliases:``) credential resolution (#83612).
|
||||
|
||||
An alias that points at a custom endpoint must authenticate with **its own**
|
||||
credential. Before the fix ``DirectAlias`` had no ``api_key`` field at all, so
|
||||
a configured key was silently dropped and the alias inherited whatever key the
|
||||
*default* provider had already resolved — a 401 against the alias host and a
|
||||
cross-provider credential leak to an unrelated third party.
|
||||
|
||||
The regression that matters most is the leak: assert on the credential the
|
||||
endpoint probe is actually handed, not just on the returned struct.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
ALIAS_HOST = "https://theta.example.com/v1"
|
||||
DEFAULT_PROVIDER_SECRET = "sk-or-DEFAULT-PROVIDER-SECRET"
|
||||
|
||||
|
||||
def _install_config(monkeypatch, alias_entry):
|
||||
"""Point every config reader at a single-alias config."""
|
||||
cfg = {
|
||||
"model": {"default": "gpt-4", "provider": "openrouter"},
|
||||
"model_aliases": {"theta": alias_entry},
|
||||
}
|
||||
monkeypatch.setattr("hermes_cli.config.load_config", lambda *a, **k: cfg)
|
||||
monkeypatch.setattr("hermes_cli.runtime_provider.load_config", lambda *a, **k: cfg)
|
||||
return cfg
|
||||
|
||||
|
||||
def _switch_to_alias(monkeypatch, alias_entry):
|
||||
"""Run ``/model theta`` and capture what the endpoint probe was given.
|
||||
|
||||
Returns ``(result, probed)`` where ``probed`` holds the api_key/base_url
|
||||
handed to ``validate_requested_model`` — i.e. the credential that goes out
|
||||
on the wire to the alias host.
|
||||
"""
|
||||
_install_config(monkeypatch, alias_entry)
|
||||
monkeypatch.setenv("OPENROUTER_API_KEY", DEFAULT_PROVIDER_SECRET)
|
||||
|
||||
probed = {}
|
||||
|
||||
def _fake_validate(model_name, provider, *, api_key=None, base_url=None,
|
||||
api_mode=None, **_kwargs):
|
||||
probed["api_key"] = api_key
|
||||
probed["base_url"] = base_url
|
||||
return {"accepted": True, "persist": True, "recognized": True, "message": ""}
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.models.validate_requested_model", _fake_validate
|
||||
)
|
||||
|
||||
import hermes_cli.model_switch as ms
|
||||
|
||||
monkeypatch.setattr(ms, "DIRECT_ALIASES", {})
|
||||
result = ms.switch_model(
|
||||
raw_input="theta",
|
||||
current_provider="openrouter",
|
||||
current_model="gpt-4",
|
||||
current_base_url="https://openrouter.ai/api/v1",
|
||||
current_api_key=DEFAULT_PROVIDER_SECRET,
|
||||
)
|
||||
return result, probed
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Loading
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestDirectAliasCredentialLoading:
|
||||
def test_api_key_and_key_env_are_loaded_from_config(self, monkeypatch):
|
||||
"""``api_key``/``key_env`` survive into the DirectAlias (were dropped)."""
|
||||
_install_config(
|
||||
monkeypatch,
|
||||
{
|
||||
"model": "theta-1",
|
||||
"provider": "custom",
|
||||
"base_url": ALIAS_HOST,
|
||||
"api_key": "sk-literal",
|
||||
"key_env": "THETA_API_KEY",
|
||||
},
|
||||
)
|
||||
from hermes_cli.model_switch import _load_direct_aliases
|
||||
|
||||
alias = _load_direct_aliases()["theta"]
|
||||
assert alias.api_key == "sk-literal"
|
||||
assert alias.key_env == "THETA_API_KEY"
|
||||
|
||||
def test_credential_fields_default_to_empty(self, monkeypatch):
|
||||
"""Aliases without credentials keep working (positional construction)."""
|
||||
from hermes_cli.model_switch import DirectAlias
|
||||
|
||||
alias = DirectAlias("theta-1", "custom", ALIAS_HOST)
|
||||
assert alias.api_key == ""
|
||||
assert alias.key_env == ""
|
||||
|
||||
|
||||
class TestDirectAliasApiKeyHelper:
|
||||
@pytest.mark.parametrize(
|
||||
"entry, expected",
|
||||
[
|
||||
({"api_key": "sk-literal"}, "sk-literal"),
|
||||
({"api_key": "${THETA_API_KEY}"}, "sk-from-env"),
|
||||
({"key_env": "THETA_API_KEY"}, "sk-from-env"),
|
||||
({}, ""),
|
||||
],
|
||||
)
|
||||
def test_resolves_literal_env_template_and_key_env(
|
||||
self, monkeypatch, entry, expected
|
||||
):
|
||||
monkeypatch.setenv("THETA_API_KEY", "sk-from-env")
|
||||
from hermes_cli.model_switch import DirectAlias, direct_alias_api_key
|
||||
|
||||
alias = DirectAlias("theta-1", "custom", ALIAS_HOST, **entry)
|
||||
assert direct_alias_api_key(alias) == expected
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# /model <alias> — the switch path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestModelSwitchUsesAliasCredential:
|
||||
def test_alias_api_key_is_sent_to_alias_host(self, monkeypatch):
|
||||
result, probed = _switch_to_alias(
|
||||
monkeypatch,
|
||||
{
|
||||
"model": "theta-1",
|
||||
"provider": "custom",
|
||||
"base_url": ALIAS_HOST,
|
||||
"api_key": "sk-theta-ALIAS-SECRET",
|
||||
},
|
||||
)
|
||||
assert result.base_url == ALIAS_HOST
|
||||
assert result.api_key == "sk-theta-ALIAS-SECRET"
|
||||
assert probed["api_key"] == "sk-theta-ALIAS-SECRET"
|
||||
|
||||
def test_alias_key_env_is_sent_to_alias_host(self, monkeypatch):
|
||||
monkeypatch.setenv("THETA_API_KEY", "sk-theta-FROM-ENV")
|
||||
result, _ = _switch_to_alias(
|
||||
monkeypatch,
|
||||
{
|
||||
"model": "theta-1",
|
||||
"provider": "custom",
|
||||
"base_url": ALIAS_HOST,
|
||||
"key_env": "THETA_API_KEY",
|
||||
},
|
||||
)
|
||||
assert result.api_key == "sk-theta-FROM-ENV"
|
||||
|
||||
def test_default_provider_key_never_reaches_the_alias_host(self, monkeypatch):
|
||||
"""The leak: an alias with no credential of its own must NOT inherit
|
||||
the default provider's key just because that key was resolved first."""
|
||||
result, probed = _switch_to_alias(
|
||||
monkeypatch,
|
||||
{"model": "theta-1", "provider": "custom", "base_url": ALIAS_HOST},
|
||||
)
|
||||
assert result.base_url == ALIAS_HOST
|
||||
assert result.api_key != DEFAULT_PROVIDER_SECRET
|
||||
assert probed["api_key"] != DEFAULT_PROVIDER_SECRET
|
||||
assert probed["base_url"] == ALIAS_HOST
|
||||
|
||||
def test_same_host_alias_still_resolves_that_host_key(self, monkeypatch):
|
||||
"""Host-gated resolution keeps working: an openrouter.ai alias still
|
||||
gets OPENROUTER_API_KEY — this is not a blanket "drop the key"."""
|
||||
result, _ = _switch_to_alias(
|
||||
monkeypatch,
|
||||
{
|
||||
"model": "theta-1",
|
||||
"provider": "custom",
|
||||
"base_url": "https://openrouter.ai/api/v1",
|
||||
},
|
||||
)
|
||||
assert result.api_key == DEFAULT_PROVIDER_SECRET
|
||||
|
||||
def test_ollama_cloud_alias_resolves_ollama_api_key(self, monkeypatch):
|
||||
"""Ollama Cloud aliases authenticate with OLLAMA_API_KEY, not the
|
||||
previously active provider's key."""
|
||||
monkeypatch.setenv("OLLAMA_API_KEY", "sk-ollama-KEY")
|
||||
result, _ = _switch_to_alias(
|
||||
monkeypatch,
|
||||
{
|
||||
"model": "qwen3.5:397b",
|
||||
"provider": "custom",
|
||||
"base_url": "https://ollama.com/v1",
|
||||
},
|
||||
)
|
||||
assert result.api_key == "sk-ollama-KEY"
|
||||
|
||||
|
||||
class TestSessionKeyIsHostScoped:
|
||||
"""The key already resolved for the session is reusable only on the same
|
||||
host. This is what keeps a user pinned to a custom endpoint working while
|
||||
still closing the cross-host leak."""
|
||||
|
||||
def _switch(self, monkeypatch, session_base_url):
|
||||
cfg = {
|
||||
"model": {"default": "m", "provider": "ollama-launch"},
|
||||
"model_aliases": {
|
||||
"theta": {
|
||||
"model": "theta-1",
|
||||
"provider": "custom",
|
||||
"base_url": "https://myhost.test/v1",
|
||||
}
|
||||
},
|
||||
}
|
||||
monkeypatch.setattr("hermes_cli.config.load_config", lambda *a, **k: cfg)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.runtime_provider.load_config", lambda *a, **k: cfg
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.models.validate_requested_model",
|
||||
lambda *a, **k: {
|
||||
"accepted": True,
|
||||
"persist": True,
|
||||
"recognized": True,
|
||||
"message": "",
|
||||
},
|
||||
)
|
||||
import hermes_cli.model_switch as ms
|
||||
|
||||
monkeypatch.setattr(ms, "DIRECT_ALIASES", {})
|
||||
return ms.switch_model(
|
||||
raw_input="theta",
|
||||
current_provider="ollama-launch",
|
||||
current_model="m",
|
||||
current_base_url=session_base_url,
|
||||
current_api_key="sk-session-KEY",
|
||||
)
|
||||
|
||||
def test_same_host_alias_keeps_the_session_key(self, monkeypatch):
|
||||
result = self._switch(monkeypatch, "https://myhost.test/v1")
|
||||
assert result.api_key == "sk-session-KEY"
|
||||
|
||||
def test_different_host_alias_drops_the_session_key(self, monkeypatch):
|
||||
result = self._switch(monkeypatch, "https://elsewhere.test/v1")
|
||||
assert result.api_key != "sk-session-KEY"
|
||||
|
||||
|
||||
class TestBuiltinProviderKeysDoNotLeak:
|
||||
"""The leak is not specific to custom providers. A session on a built-in
|
||||
provider (Anthropic, OpenAI, ...) must not forward its key either when the
|
||||
alias resolves to an unrelated host — the credential is dropped on host
|
||||
mismatch regardless of which branch resolved it."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"provider, env_var, session_base_url",
|
||||
[
|
||||
("anthropic", "ANTHROPIC_API_KEY", "https://api.anthropic.com"),
|
||||
("openai", "OPENAI_API_KEY", "https://api.openai.com/v1"),
|
||||
],
|
||||
)
|
||||
def test_builtin_provider_key_not_forwarded_to_alias_host(
|
||||
self, monkeypatch, provider, env_var, session_base_url
|
||||
):
|
||||
secret = f"sk-{provider}-SECRET"
|
||||
cfg = {
|
||||
"model": {"default": "m", "provider": provider},
|
||||
"model_aliases": {
|
||||
"theta": {
|
||||
"model": "theta-1",
|
||||
"provider": "custom",
|
||||
"base_url": ALIAS_HOST,
|
||||
}
|
||||
},
|
||||
}
|
||||
monkeypatch.setattr("hermes_cli.config.load_config", lambda *a, **k: cfg)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.runtime_provider.load_config", lambda *a, **k: cfg
|
||||
)
|
||||
monkeypatch.setenv(env_var, secret)
|
||||
|
||||
probed = {}
|
||||
|
||||
def _fake_validate(model_name, prov, *, api_key=None, base_url=None,
|
||||
api_mode=None, **_kwargs):
|
||||
probed["api_key"] = api_key
|
||||
probed["base_url"] = base_url
|
||||
return {"accepted": True, "persist": True, "recognized": True, "message": ""}
|
||||
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.models.validate_requested_model", _fake_validate
|
||||
)
|
||||
import hermes_cli.model_switch as ms
|
||||
|
||||
monkeypatch.setattr(ms, "DIRECT_ALIASES", {})
|
||||
result = ms.switch_model(
|
||||
raw_input="theta",
|
||||
current_provider=provider,
|
||||
current_model="m",
|
||||
current_base_url=session_base_url,
|
||||
current_api_key=secret,
|
||||
)
|
||||
assert result.base_url == ALIAS_HOST
|
||||
assert result.api_key != secret
|
||||
assert probed["api_key"] != secret
|
||||
|
||||
|
||||
class TestProviderLabelCannotSelectAKeyForAnArbitraryHost:
|
||||
"""A direct alias's `provider:` label must not route credential resolution.
|
||||
|
||||
With a base_url but no declared credential, a label like `anthropic` used
|
||||
to reach that provider's own resolver, which picks ANTHROPIC_API_KEY out
|
||||
of the environment while keeping the alias's unrelated base_url — a
|
||||
built-in provider's bearer secret handed to a third-party host. The alias
|
||||
endpoint is resolved as bare `custom` instead, which is host-gated.
|
||||
"""
|
||||
|
||||
def _switch(self, monkeypatch, alias, session_provider="openrouter",
|
||||
session_base_url="https://openrouter.ai/api/v1"):
|
||||
cfg = {
|
||||
"model": {"default": "m", "provider": session_provider},
|
||||
"model_aliases": {"theta": alias},
|
||||
}
|
||||
monkeypatch.setattr("hermes_cli.config.load_config", lambda *a, **k: cfg)
|
||||
monkeypatch.setattr("hermes_cli.runtime_provider.load_config", lambda *a, **k: cfg)
|
||||
probed = {}
|
||||
|
||||
def _fake_validate(model_name, prov, *, api_key=None, base_url=None,
|
||||
api_mode=None, **_kwargs):
|
||||
probed["api_key"] = api_key
|
||||
return {"accepted": True, "persist": True, "recognized": True, "message": ""}
|
||||
|
||||
monkeypatch.setattr("hermes_cli.models.validate_requested_model", _fake_validate)
|
||||
import hermes_cli.model_switch as ms
|
||||
|
||||
monkeypatch.setattr(ms, "DIRECT_ALIASES", {})
|
||||
result = ms.switch_model(
|
||||
raw_input="theta", current_provider=session_provider, current_model="m",
|
||||
current_base_url=session_base_url, current_api_key="sk-session",
|
||||
)
|
||||
return result, probed
|
||||
|
||||
@pytest.mark.parametrize("provider, env_var", [
|
||||
("anthropic", "ANTHROPIC_API_KEY"),
|
||||
("openai", "OPENAI_API_KEY"),
|
||||
("openrouter", "OPENROUTER_API_KEY"),
|
||||
])
|
||||
def test_builtin_label_does_not_pull_that_providers_key_to_a_foreign_host(
|
||||
self, monkeypatch, provider, env_var
|
||||
):
|
||||
secret = f"sk-{provider}-SECRET"
|
||||
monkeypatch.setenv(env_var, secret)
|
||||
result, probed = self._switch(
|
||||
monkeypatch,
|
||||
{"model": "c", "provider": provider, "base_url": "https://evil.test/v1"},
|
||||
)
|
||||
# Either the switch resolves no key for the foreign host, or it fails
|
||||
# outright — never the built-in provider's secret.
|
||||
assert result.api_key != secret
|
||||
assert probed.get("api_key") != secret
|
||||
|
||||
def test_authoritative_host_still_resolves_its_vendor_key(self, monkeypatch):
|
||||
"""Host gating is the point, not a blanket refusal: an alias whose URL
|
||||
IS authoritative for the vendor still authenticates."""
|
||||
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-anthropic-SECRET")
|
||||
result, _ = self._switch(
|
||||
monkeypatch,
|
||||
{"model": "c", "provider": "anthropic",
|
||||
"base_url": "https://api.anthropic.com/v1"},
|
||||
)
|
||||
assert result.api_key == "sk-anthropic-SECRET"
|
||||
|
||||
|
||||
class TestSessionCredentialIsScopedToTheOrigin:
|
||||
"""Reusing the session key across a scheme or port change is a downgrade.
|
||||
|
||||
The override compared hostnames only, so an alias could keep the host and
|
||||
move an HTTPS session to `http://` (or another port) while the live
|
||||
credential followed it onto the new, untrusted origin.
|
||||
"""
|
||||
|
||||
def _switch(self, monkeypatch, alias_base_url, session_base_url):
|
||||
cfg = {
|
||||
"model": {"default": "m", "provider": "my-endpoint"},
|
||||
"model_aliases": {"theta": {
|
||||
"model": "m2", "provider": "custom", "base_url": alias_base_url}},
|
||||
}
|
||||
monkeypatch.setattr("hermes_cli.config.load_config", lambda *a, **k: cfg)
|
||||
monkeypatch.setattr("hermes_cli.runtime_provider.load_config", lambda *a, **k: cfg)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.models.validate_requested_model",
|
||||
lambda *a, **k: {"accepted": True, "persist": True,
|
||||
"recognized": True, "message": ""},
|
||||
)
|
||||
import hermes_cli.model_switch as ms
|
||||
|
||||
monkeypatch.setattr(ms, "DIRECT_ALIASES", {})
|
||||
return ms.switch_model(
|
||||
raw_input="theta", current_provider="my-endpoint", current_model="m",
|
||||
current_base_url=session_base_url, current_api_key="sk-SESSION-SECRET",
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("alias_url, session_url, why", [
|
||||
("http://api.example.com/v1", "https://api.example.com/v1", "https->http"),
|
||||
("https://api.example.com:8443/v1", "https://api.example.com/v1", "port change"),
|
||||
("http://api.example.com:8080/v1", "https://api.example.com/v1", "scheme+port"),
|
||||
("https://other.example.com/v1", "https://api.example.com/v1", "cross-host"),
|
||||
])
|
||||
def test_origin_change_drops_the_session_credential(
|
||||
self, monkeypatch, alias_url, session_url, why
|
||||
):
|
||||
assert self._switch(monkeypatch, alias_url, session_url).api_key != "sk-SESSION-SECRET"
|
||||
|
||||
@pytest.mark.parametrize("url", [
|
||||
"https://api.example.com/v1",
|
||||
"http://127.0.0.1:11434/v1", # loopback plaintext is not a downgrade
|
||||
"http://localhost:8080/v1",
|
||||
])
|
||||
def test_same_origin_keeps_the_session_credential(self, monkeypatch, url):
|
||||
assert self._switch(monkeypatch, url, url).api_key == "sk-SESSION-SECRET"
|
||||
|
||||
def test_default_port_and_explicit_port_are_the_same_origin(self, monkeypatch):
|
||||
result = self._switch(
|
||||
monkeypatch, "https://api.example.com:443/v1", "https://api.example.com/v1"
|
||||
)
|
||||
assert result.api_key == "sk-SESSION-SECRET"
|
||||
|
||||
|
||||
class TestCredentialPrecedenceIsExplicit:
|
||||
"""`api_key` outranks `key_env`, so an entry carrying both is unambiguous."""
|
||||
|
||||
def test_api_key_wins_over_key_env(self, monkeypatch):
|
||||
monkeypatch.setenv("THETA_API_KEY", "sk-from-key-env")
|
||||
from hermes_cli.model_switch import DirectAlias, direct_alias_api_key
|
||||
|
||||
alias = DirectAlias("theta-1", "custom", ALIAS_HOST,
|
||||
"sk-literal", "THETA_API_KEY")
|
||||
assert direct_alias_api_key(alias) == "sk-literal"
|
||||
|
||||
def test_env_template_api_key_also_wins_over_key_env(self, monkeypatch):
|
||||
monkeypatch.setenv("PRIMARY", "sk-from-template")
|
||||
monkeypatch.setenv("FALLBACK", "sk-from-key-env")
|
||||
from hermes_cli.model_switch import DirectAlias, direct_alias_api_key
|
||||
|
||||
alias = DirectAlias("theta-1", "custom", ALIAS_HOST,
|
||||
"${PRIMARY}", "FALLBACK")
|
||||
assert direct_alias_api_key(alias) == "sk-from-template"
|
||||
|
||||
def test_key_env_used_when_api_key_is_blank(self, monkeypatch):
|
||||
monkeypatch.setenv("FALLBACK", "sk-from-key-env")
|
||||
from hermes_cli.model_switch import DirectAlias, direct_alias_api_key
|
||||
|
||||
alias = DirectAlias("theta-1", "custom", ALIAS_HOST, " ", "FALLBACK")
|
||||
assert direct_alias_api_key(alias) == "sk-from-key-env"
|
||||
|
||||
|
||||
class TestSchemelessBaseUrls:
|
||||
"""Scheme-less base URLs.
|
||||
|
||||
A loopback alias written without a scheme keeps working — the loopback
|
||||
exemption does not depend on the scheme. A scheme-less URL that also
|
||||
changes origin is refused, which costs nothing in practice: httpx cannot
|
||||
build a client from one at all (`localhost:11434/v1` parses as
|
||||
scheme='localhost', host=''), so such a base_url is non-functional
|
||||
regardless of what this comparison decides.
|
||||
"""
|
||||
|
||||
def test_hostname_and_port_survive_without_a_scheme(self):
|
||||
from utils import base_url_origin
|
||||
|
||||
assert base_url_origin("localhost:11434/v1") == ("", "localhost", 11434)
|
||||
assert base_url_origin("127.0.0.1:11434") == ("", "127.0.0.1", 11434)
|
||||
|
||||
@pytest.mark.parametrize("url", ["localhost:11434/v1", "127.0.0.1:11434/v1"])
|
||||
def test_schemeless_loopback_alias_keeps_the_session_credential(self, url):
|
||||
from hermes_cli.model_switch import _may_reuse_session_credential
|
||||
|
||||
assert _may_reuse_session_credential(url, url) is True
|
||||
|
||||
def test_schemeless_is_not_treated_as_the_schemed_origin(self):
|
||||
"""`http://h` and `h` are not asserted equal — an unknown scheme is
|
||||
not evidence that the origin is unchanged."""
|
||||
from hermes_cli.model_switch import _may_reuse_session_credential
|
||||
|
||||
assert _may_reuse_session_credential(
|
||||
"http://localhost:11434/v1", "localhost:11434/v1"
|
||||
) is False
|
||||
|
||||
def test_httpx_cannot_use_a_schemeless_base_url(self):
|
||||
"""Pins the premise above: this is why the strict answer is harmless."""
|
||||
httpx = pytest.importorskip("httpx")
|
||||
|
||||
assert httpx.URL("localhost:11434/v1").host == ""
|
||||
assert httpx.URL("api.example.com/v1").host == ""
|
||||
assert httpx.URL("http://localhost:11434/v1").host == "localhost"
|
||||
|
||||
|
||||
class TestAliasCacheIsProfileScoped:
|
||||
"""DIRECT_ALIASES is process-global; its source is profile-local.
|
||||
|
||||
Entries carry `api_key`, so an unkeyed cache lets the first profile to
|
||||
resolve an alias pin its definitions AND its credentials for every later
|
||||
profile in the process. These tests switch profiles inside one process
|
||||
and never clear the cache by hand — clearing it would hide the bug.
|
||||
"""
|
||||
|
||||
def _profile(self, tmp_path, name, body):
|
||||
home = tmp_path / name
|
||||
home.mkdir(parents=True, exist_ok=True)
|
||||
(home / "config.yaml").write_text(body, encoding="utf-8")
|
||||
return home
|
||||
|
||||
def _load(self, monkeypatch, home):
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
import hermes_cli.model_switch as ms
|
||||
|
||||
ms._ensure_direct_aliases()
|
||||
return ms.DIRECT_ALIASES
|
||||
|
||||
def test_second_profile_does_not_inherit_the_first_profiles_key(
|
||||
self, tmp_path, monkeypatch
|
||||
):
|
||||
a = self._profile(tmp_path, "a", (
|
||||
'model_aliases:\n'
|
||||
' theta:\n model: a-model\n provider: custom\n'
|
||||
' base_url: "https://a.example.com/v1"\n'
|
||||
' api_key: "sk-PROFILE-A-SECRET"\n'
|
||||
))
|
||||
b = self._profile(tmp_path, "b", (
|
||||
'model_aliases:\n'
|
||||
' theta:\n model: b-model\n provider: custom\n'
|
||||
' base_url: "https://b.example.com/v1"\n'
|
||||
' api_key: "sk-PROFILE-B-SECRET"\n'
|
||||
))
|
||||
assert self._load(monkeypatch, a)["theta"].api_key == "sk-PROFILE-A-SECRET"
|
||||
|
||||
theta = self._load(monkeypatch, b)["theta"]
|
||||
assert theta.api_key == "sk-PROFILE-B-SECRET"
|
||||
assert theta.model == "b-model"
|
||||
assert theta.base_url == "https://b.example.com/v1"
|
||||
|
||||
def test_alias_absent_from_the_second_profile_does_not_persist(
|
||||
self, tmp_path, monkeypatch
|
||||
):
|
||||
a = self._profile(tmp_path, "a2", (
|
||||
'model_aliases:\n'
|
||||
' only-in-a:\n model: x\n provider: custom\n'
|
||||
' base_url: "https://a.example.com/v1"\n'
|
||||
))
|
||||
b = self._profile(tmp_path, "b2", (
|
||||
'model_aliases:\n'
|
||||
' only-in-b:\n model: y\n provider: custom\n'
|
||||
' base_url: "https://b.example.com/v1"\n'
|
||||
))
|
||||
assert "only-in-a" in self._load(monkeypatch, a)
|
||||
|
||||
loaded = self._load(monkeypatch, b)
|
||||
assert "only-in-a" not in loaded
|
||||
assert "only-in-b" in loaded
|
||||
|
||||
def test_key_rotation_in_place_is_picked_up(self, tmp_path, monkeypatch):
|
||||
home = self._profile(tmp_path, "rot", (
|
||||
'model_aliases:\n'
|
||||
' theta:\n model: m\n provider: custom\n'
|
||||
' base_url: "https://h.example.com/v1"\n'
|
||||
' api_key: "sk-BEFORE-ROTATION"\n'
|
||||
))
|
||||
assert self._load(monkeypatch, home)["theta"].api_key == "sk-BEFORE-ROTATION"
|
||||
|
||||
(home / "config.yaml").write_text((
|
||||
'model_aliases:\n'
|
||||
' theta:\n model: m\n provider: custom\n'
|
||||
' base_url: "https://h.example.com/v1"\n'
|
||||
' api_key: "sk-AFTER-ROTATION-XYZ"\n'
|
||||
), encoding="utf-8")
|
||||
assert self._load(monkeypatch, home)["theta"].api_key == "sk-AFTER-ROTATION-XYZ"
|
||||
|
||||
def test_cache_is_still_mutated_in_place(self, tmp_path, monkeypatch):
|
||||
"""Callers hold this exact dict (#16767) — reloading must not rebind."""
|
||||
home = self._profile(tmp_path, "inplace", (
|
||||
'model_aliases:\n'
|
||||
' theta:\n model: m\n provider: custom\n'
|
||||
' base_url: "https://h.example.com/v1"\n'
|
||||
))
|
||||
import hermes_cli.model_switch as ms
|
||||
|
||||
before = id(ms.DIRECT_ALIASES)
|
||||
self._load(monkeypatch, home)
|
||||
assert id(ms.DIRECT_ALIASES) == before
|
||||
|
||||
|
||||
class TestOneShotUsesTheSameHostInvariant:
|
||||
"""`hermes chat -m <alias>` must not trust the alias's provider label.
|
||||
|
||||
These exercise the REAL resolver — the leak lives inside
|
||||
resolve_runtime_provider's provider-specific branches, so stubbing it
|
||||
would test nothing.
|
||||
"""
|
||||
|
||||
@pytest.mark.parametrize("provider, env_var", [
|
||||
("anthropic", "ANTHROPIC_API_KEY"),
|
||||
("deepseek", "DEEPSEEK_API_KEY"),
|
||||
("xai", "XAI_API_KEY"),
|
||||
])
|
||||
def test_no_key_alias_on_a_foreign_host_gets_no_provider_token(
|
||||
self, monkeypatch, provider, env_var
|
||||
):
|
||||
secret = f"sk-{provider}-LIVE-TOKEN"
|
||||
monkeypatch.setenv(env_var, secret)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.runtime_provider.load_config",
|
||||
lambda *a, **k: {"model": {"default": "m", "provider": "openrouter"}},
|
||||
)
|
||||
from hermes_cli.model_switch import DirectAlias, direct_alias_runtime_request
|
||||
from hermes_cli.runtime_provider import resolve_runtime_provider
|
||||
|
||||
alias = DirectAlias("c", provider, "https://evil.test/v1")
|
||||
requested, explicit_key = direct_alias_runtime_request(alias)
|
||||
runtime = resolve_runtime_provider(
|
||||
requested=requested,
|
||||
explicit_api_key=explicit_key,
|
||||
explicit_base_url=alias.base_url,
|
||||
)
|
||||
assert runtime.get("api_key") != secret
|
||||
|
||||
def test_label_is_kept_when_the_alias_has_no_url(self):
|
||||
"""Nothing to protect against without a foreign host, and the label is
|
||||
the only routing information there is."""
|
||||
from hermes_cli.model_switch import DirectAlias, direct_alias_runtime_request
|
||||
|
||||
assert direct_alias_runtime_request(
|
||||
DirectAlias("c", "anthropic", "")
|
||||
) == ("anthropic", None)
|
||||
|
||||
def test_url_bearing_alias_is_forced_to_custom(self):
|
||||
from hermes_cli.model_switch import DirectAlias, direct_alias_runtime_request
|
||||
|
||||
assert direct_alias_runtime_request(
|
||||
DirectAlias("c", "anthropic", "https://evil.test/v1")
|
||||
) == ("custom", None)
|
||||
|
||||
def test_declared_key_is_carried_through(self, monkeypatch):
|
||||
from hermes_cli.model_switch import DirectAlias, direct_alias_runtime_request
|
||||
|
||||
assert direct_alias_runtime_request(
|
||||
DirectAlias("c", "anthropic", "https://evil.test/v1", "sk-own")
|
||||
) == ("custom", "sk-own")
|
||||
|
||||
|
||||
class TestBaseUrlOrigin:
|
||||
"""The origin helper the reuse decision is built on."""
|
||||
|
||||
@pytest.mark.parametrize("url, expected", [
|
||||
("https://h/v1", ("https", "h", 443)),
|
||||
("https://h:443/v1", ("https", "h", 443)),
|
||||
("http://h/v1", ("http", "h", 80)),
|
||||
("https://h:8443/v1", ("https", "h", 8443)),
|
||||
("https://H./v1", ("https", "h", 443)),
|
||||
("", ("", "", 0)),
|
||||
("https://h:99999/v1", ("", "", 0)),
|
||||
])
|
||||
def test_origin_normalisation(self, url, expected):
|
||||
from utils import base_url_origin
|
||||
|
||||
assert base_url_origin(url) == expected
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Host gating in the direct-alias runtime branch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestDirectAliasHostGating:
|
||||
@pytest.mark.parametrize(
|
||||
"base_url, expect_key",
|
||||
[
|
||||
("https://ollama.com/v1", True),
|
||||
# Look-alike and path-embedded hosts must NOT get the credential
|
||||
# (GHSA-76xc-57q6-vm5m).
|
||||
("https://ollama.com.attacker.test/v1", False),
|
||||
("http://127.0.0.1/ollama.com/v1", False),
|
||||
],
|
||||
)
|
||||
def test_ollama_key_is_host_matched_not_substring_matched(
|
||||
self, monkeypatch, base_url, expect_key
|
||||
):
|
||||
monkeypatch.setenv("OLLAMA_API_KEY", "sk-ollama-KEY")
|
||||
from hermes_cli.runtime_provider import _resolve_named_custom_runtime
|
||||
|
||||
runtime = _resolve_named_custom_runtime(
|
||||
requested_provider="custom", explicit_base_url=base_url
|
||||
)
|
||||
assert (runtime["api_key"] == "sk-ollama-KEY") is expect_key
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# hermes chat -m <alias> — the oneshot path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestOneshotPassesAliasCredential:
|
||||
def test_alias_api_key_is_passed_to_the_resolver(self, monkeypatch):
|
||||
"""``hermes chat -m theta`` must hand the alias's key to
|
||||
resolve_runtime_provider, not leave it to env fallbacks."""
|
||||
from hermes_cli.model_switch import DirectAlias
|
||||
import hermes_cli.model_switch as ms
|
||||
|
||||
monkeypatch.setattr(
|
||||
ms,
|
||||
"DIRECT_ALIASES",
|
||||
{"theta": DirectAlias("theta-1", "custom", ALIAS_HOST, "sk-theta-ALIAS")},
|
||||
)
|
||||
monkeypatch.setattr(ms, "_ensure_direct_aliases", lambda: None)
|
||||
|
||||
captured = {}
|
||||
|
||||
def _fake_resolve(**kwargs):
|
||||
captured.update(kwargs)
|
||||
raise RuntimeError("stop after credential resolution")
|
||||
|
||||
# oneshot imports the resolver inside the function, so patch it at
|
||||
# its source module.
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.runtime_provider.resolve_runtime_provider", _fake_resolve
|
||||
)
|
||||
monkeypatch.setattr("hermes_cli.config.load_config", lambda *a, **k: {})
|
||||
import hermes_cli.oneshot as oneshot
|
||||
|
||||
# _run_agent holds the alias wiring; run_oneshot() wraps it in a
|
||||
# catch-all that would swallow the sentinel.
|
||||
with pytest.raises(RuntimeError, match="stop after credential resolution"):
|
||||
oneshot._run_agent(prompt="hi", model="theta")
|
||||
|
||||
assert captured["explicit_base_url"] == ALIAS_HOST
|
||||
assert captured["explicit_api_key"] == "sk-theta-ALIAS"
|
||||
|
||||
|
||||
class TestNoProductionCodeMutatesTheAliasCacheInPlace:
|
||||
"""The profile-isolation property depends on an unwritten rule.
|
||||
|
||||
``_ensure_direct_aliases`` keeps a copy of what it loaded and treats any
|
||||
divergence as a caller's data, not its own stale cache — that is what lets
|
||||
tests seed ``DIRECT_ALIASES`` in place without the loader wiping them
|
||||
(#16767). The cost is that a *production* in-place mutation would pin the
|
||||
cache: contents would never again match the copy, so the config-identity
|
||||
check that reloads on a profile switch would stop being consulted, and one
|
||||
profile's aliases and credentials would be served to the next.
|
||||
|
||||
No production code mutates it today — only the loader itself, and every
|
||||
other reference is a read. This pins that, because the failure mode is
|
||||
silent: nothing raises, nothing logs, and the leak only shows up as one
|
||||
profile answering with another profile's key.
|
||||
"""
|
||||
|
||||
#: The only place allowed to write the cache.
|
||||
OWNER = ("hermes_cli/model_switch.py", "_ensure_direct_aliases")
|
||||
|
||||
MUTATORS = frozenset(
|
||||
{"update", "clear", "pop", "popitem", "setdefault", "__setitem__"}
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _production_sources():
|
||||
import pathlib
|
||||
|
||||
repo = pathlib.Path(__file__).resolve().parents[2]
|
||||
skip = {".git", "node_modules", "tests", "build", "dist", ".venv"}
|
||||
for path in repo.rglob("*.py"):
|
||||
if any(part in skip for part in path.parts):
|
||||
continue
|
||||
yield path, path.relative_to(repo).as_posix()
|
||||
|
||||
@classmethod
|
||||
def _violations(cls, source: str, rel: str):
|
||||
"""Yield (function, description) for each in-place write."""
|
||||
import ast
|
||||
|
||||
tree = ast.parse(source)
|
||||
enclosing = {}
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
for child in ast.walk(node):
|
||||
enclosing[id(child)] = node.name
|
||||
|
||||
def _names(node):
|
||||
"""DIRECT_ALIASES, whether bare or attribute-qualified."""
|
||||
if isinstance(node, ast.Name):
|
||||
return node.id
|
||||
if isinstance(node, ast.Attribute):
|
||||
return node.attr
|
||||
return None
|
||||
|
||||
for node in ast.walk(tree):
|
||||
where = enclosing.get(id(node), "<module>")
|
||||
if (rel, where) == cls.OWNER:
|
||||
continue
|
||||
if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
|
||||
if (
|
||||
_names(node.func.value) == "DIRECT_ALIASES"
|
||||
and node.func.attr in cls.MUTATORS
|
||||
):
|
||||
yield where, f"DIRECT_ALIASES.{node.func.attr}()"
|
||||
if isinstance(node, (ast.Assign, ast.AugAssign)):
|
||||
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
|
||||
for target in targets:
|
||||
if (
|
||||
isinstance(target, ast.Subscript)
|
||||
and _names(target.value) == "DIRECT_ALIASES"
|
||||
):
|
||||
yield where, "DIRECT_ALIASES[...] = ..."
|
||||
|
||||
def test_only_the_loader_writes_the_cache(self):
|
||||
found = []
|
||||
for path, rel in self._production_sources():
|
||||
try:
|
||||
source = path.read_text(encoding="utf-8")
|
||||
except (OSError, UnicodeDecodeError):
|
||||
continue
|
||||
if "DIRECT_ALIASES" not in source:
|
||||
continue
|
||||
try:
|
||||
found.extend(
|
||||
f"{rel}:{where}: {what}"
|
||||
for where, what in self._violations(source, rel)
|
||||
)
|
||||
except SyntaxError:
|
||||
continue
|
||||
|
||||
assert not found, (
|
||||
"In-place writes to DIRECT_ALIASES outside "
|
||||
f"{self.OWNER[0]}::{self.OWNER[1]} pin the alias cache and break "
|
||||
"per-profile isolation:\n " + "\n ".join(found)
|
||||
)
|
||||
|
||||
def test_the_scan_actually_detects_a_violation(self):
|
||||
"""Negative control — an always-passing scanner would prove nothing."""
|
||||
offending = (
|
||||
"from hermes_cli.model_switch import DIRECT_ALIASES\n"
|
||||
"def warm():\n"
|
||||
" DIRECT_ALIASES.update({'x': 1})\n"
|
||||
" DIRECT_ALIASES['y'] = 2\n"
|
||||
)
|
||||
|
||||
hits = list(self._violations(offending, "some/other_module.py"))
|
||||
|
||||
assert {what for _, what in hits} == {
|
||||
"DIRECT_ALIASES.update()",
|
||||
"DIRECT_ALIASES[...] = ...",
|
||||
}
|
||||
assert all(where == "warm" for where, _ in hits)
|
||||
|
||||
def test_the_owner_itself_is_exempt(self):
|
||||
"""...and an exemption that swallowed everything would prove nothing."""
|
||||
import pathlib
|
||||
|
||||
repo = pathlib.Path(__file__).resolve().parents[2]
|
||||
source = (repo / self.OWNER[0]).read_text(encoding="utf-8")
|
||||
|
||||
assert list(self._violations(source, self.OWNER[0])) == []
|
||||
# The loader really does write in place, so the exemption is load-bearing.
|
||||
assert "DIRECT_ALIASES.clear()" in source
|
||||
assert "DIRECT_ALIASES.update(loaded)" in source
|
||||
30
utils.py
30
utils.py
@@ -903,6 +903,36 @@ def model_forces_max_completion_tokens(model: str) -> bool:
|
||||
)
|
||||
|
||||
|
||||
def base_url_origin(base_url: str) -> tuple[str, str, int]:
|
||||
"""Return ``(scheme, hostname, effective_port)`` for a base URL.
|
||||
|
||||
Origin, not just host. ``https://h/v1`` and ``http://h/v1`` are different
|
||||
trust boundaries, and so are two ports on the same host, so any decision
|
||||
about handing a bearer secret to a new URL has to compare all three —
|
||||
hostname equality alone would authorise an HTTPS→HTTP downgrade.
|
||||
|
||||
The port is normalised to the scheme default (443/80) when absent, so
|
||||
``https://h`` and ``https://h:443`` compare equal. Returns
|
||||
``("", "", 0)`` when the URL yields no usable hostname or a bad port.
|
||||
"""
|
||||
raw = (base_url or "").strip()
|
||||
if not raw:
|
||||
return ("", "", 0)
|
||||
parsed = urlparse(raw if "://" in raw else f"//{raw}")
|
||||
scheme = (parsed.scheme or "").lower()
|
||||
hostname = (parsed.hostname or "").lower().rstrip(".")
|
||||
if not hostname:
|
||||
return ("", "", 0)
|
||||
try:
|
||||
port = parsed.port
|
||||
except ValueError:
|
||||
# Out-of-range or non-numeric port — not a usable origin.
|
||||
return ("", "", 0)
|
||||
if port is None:
|
||||
port = {"https": 443, "http": 80}.get(scheme, 0)
|
||||
return (scheme, hostname, port)
|
||||
|
||||
|
||||
def base_url_host_matches(base_url: str, domain: str) -> bool:
|
||||
"""Return True when the base URL's hostname is ``domain`` or a subdomain.
|
||||
|
||||
|
||||
@@ -197,8 +197,19 @@ model_aliases:
|
||||
model: qwen3-coder:30b
|
||||
provider: custom
|
||||
base_url: http://localhost:11434/v1
|
||||
theta:
|
||||
model: theta-1
|
||||
provider: custom
|
||||
base_url: https://theta.example.com/v1
|
||||
key_env: THETA_API_KEY # or: api_key: "${THETA_API_KEY}"
|
||||
```
|
||||
|
||||
An alias with its own `base_url` can carry that endpoint's credential via
|
||||
`api_key` (a literal, or a `"${VAR}"` reference) or `key_env` (an environment
|
||||
variable name); `api_key` wins if both are set. With neither set, the key is
|
||||
resolved from the alias **host**
|
||||
and never inherited from the provider that was active before the switch.
|
||||
|
||||
**Short form** — `provider/model` in one string. Set from the shell without editing YAML:
|
||||
|
||||
```bash
|
||||
|
||||
@@ -301,6 +301,26 @@ model_aliases:
|
||||
provider: x-ai
|
||||
```
|
||||
|
||||
An alias that points at its own endpoint can also carry that endpoint's
|
||||
credential, with either `api_key` (a literal, or a `"${VAR}"` reference) or
|
||||
`key_env` (the name of an environment variable). If both are set, `api_key`
|
||||
wins:
|
||||
|
||||
```yaml
|
||||
model_aliases:
|
||||
theta:
|
||||
model: theta-1
|
||||
provider: custom
|
||||
base_url: "https://theta.example.com/v1"
|
||||
key_env: THETA_API_KEY # or: api_key: "${THETA_API_KEY}"
|
||||
```
|
||||
|
||||
When an alias sets neither, the key is resolved from the alias **host** —
|
||||
`OLLAMA_API_KEY` for an `ollama.com` endpoint, `DEEPSEEK_API_KEY` for
|
||||
`api.deepseek.com`, and so on. It is never inherited from whichever provider
|
||||
happened to be active before the switch, so switching to an alias cannot send
|
||||
one provider's secret to another provider's host.
|
||||
|
||||
**Short string form (`model.aliases.<name>: provider/model`)** — convenient from the shell because `hermes config set` writes scalars and now also parses inline list/mapping literals, though this short alias form still can't carry a custom `base_url`:
|
||||
|
||||
```bash
|
||||
|
||||
Reference in New Issue
Block a user