From 63dd715d1587a99718c613175d9a3dbb6a161eca Mon Sep 17 00:00:00 2001 From: Aslaaen Date: Tue, 21 Apr 2026 02:07:13 +0300 Subject: [PATCH 1/3] fix: restrict provider URL detection to exact hostname matches --- hermes_cli/runtime_provider.py | 14 ++++++++-- run_agent.py | 23 +++++++++++++--- .../test_direct_provider_url_detection.py | 27 +++++++++++++++++++ .../test_detect_api_mode_for_url.py | 9 +++++++ 4 files changed, 67 insertions(+), 6 deletions(-) create mode 100644 tests/agent/test_direct_provider_url_detection.py diff --git a/hermes_cli/runtime_provider.py b/hermes_cli/runtime_provider.py index 392d7769dc44..57b6873d0475 100644 --- a/hermes_cli/runtime_provider.py +++ b/hermes_cli/runtime_provider.py @@ -6,6 +6,7 @@ import os import re from typing import Any, Dict, Optional +from urllib.parse import urlparse logger = logging.getLogger(__name__) @@ -35,6 +36,14 @@ def _normalize_custom_provider_name(value: str) -> str: return value.strip().lower().replace(" ", "-") +def _base_url_hostname(base_url: str) -> str: + raw = (base_url or "").strip() + if not raw: + return "" + parsed = urlparse(raw if "://" in raw else f"//{raw}") + return (parsed.hostname or "").lower().rstrip(".") + + def _detect_api_mode_for_url(base_url: str) -> Optional[str]: """Auto-detect api_mode from the resolved base URL. @@ -47,9 +56,10 @@ def _detect_api_mode_for_url(base_url: str) -> Optional[str]: ``chat_completions``. """ normalized = (base_url or "").strip().lower().rstrip("/") - if "api.x.ai" in normalized: + hostname = _base_url_hostname(base_url) + if hostname == "api.x.ai": return "codex_responses" - if "api.openai.com" in normalized and "openrouter" not in normalized: + if hostname == "api.openai.com": return "codex_responses" if normalized.endswith("/anthropic"): return "anthropic_messages" diff --git a/run_agent.py b/run_agent.py index 6dd28d11fe80..9da4bf93f355 100644 --- a/run_agent.py +++ b/run_agent.py @@ -38,6 +38,7 @@ from types import SimpleNamespace import uuid from typing import List, Dict, Any, Optional +from urllib.parse import urlparse from openai import OpenAI import fire from datetime import datetime @@ -127,6 +128,14 @@ from utils import atomic_json_write, env_var_enabled +def _base_url_hostname(base_url: str) -> str: + raw = (base_url or "").strip() + if not raw: + return "" + parsed = urlparse(raw if "://" in raw else f"//{raw}") + return (parsed.hostname or "").lower().rstrip(".") + + class _SafeWriter: """Transparent stdio wrapper that catches OSError/ValueError from broken pipes. @@ -703,6 +712,7 @@ def base_url(self) -> str: def base_url(self, value: str) -> None: self._base_url = value self._base_url_lower = value.lower() if value else "" + self._base_url_hostname = _base_url_hostname(value) def __init__( self, @@ -847,7 +857,7 @@ def __init__( elif (provider_name is None) and "chatgpt.com/backend-api/codex" in self._base_url_lower: self.api_mode = "codex_responses" self.provider = "openai-codex" - elif (provider_name is None) and "api.x.ai" in self._base_url_lower: + elif (provider_name is None) and self._base_url_hostname == "api.x.ai": self.api_mode = "codex_responses" self.provider = "xai" elif self.provider == "anthropic" or (provider_name is None and "api.anthropic.com" in self._base_url_lower): @@ -2259,8 +2269,13 @@ def _replay_compression_warning(self) -> None: def _is_direct_openai_url(self, base_url: str = None) -> bool: """Return True when a base URL targets OpenAI's native API.""" - url = (base_url or self._base_url_lower).lower() - return "api.openai.com" in url and "openrouter" not in url + if base_url is not None: + hostname = _base_url_hostname(base_url) + else: + hostname = getattr(self, "_base_url_hostname", "") or _base_url_hostname( + getattr(self, "_base_url_lower", "") + ) + return hostname == "api.openai.com" def _resolved_api_call_timeout(self) -> float: """Resolve the effective per-call request timeout in seconds. @@ -6747,7 +6762,7 @@ def _build_api_kwargs(self, api_messages: list) -> dict: if not is_github_responses: kwargs["prompt_cache_key"] = self.session_id - is_xai_responses = self.provider == "xai" or "api.x.ai" in (self.base_url or "").lower() + is_xai_responses = self.provider == "xai" or self._base_url_hostname == "api.x.ai" if reasoning_enabled and is_xai_responses: # xAI reasons automatically — no effort param, just include encrypted content diff --git a/tests/agent/test_direct_provider_url_detection.py b/tests/agent/test_direct_provider_url_detection.py new file mode 100644 index 000000000000..ed5dfab159fd --- /dev/null +++ b/tests/agent/test_direct_provider_url_detection.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +from run_agent import AIAgent + + +def _agent_with_base_url(base_url: str) -> AIAgent: + agent = object.__new__(AIAgent) + agent.base_url = base_url + return agent + + +def test_direct_openai_url_requires_openai_host(): + agent = _agent_with_base_url("https://api.openai.com.example/v1") + + assert agent._is_direct_openai_url() is False + + +def test_direct_openai_url_ignores_path_segment_match(): + agent = _agent_with_base_url("https://proxy.example.test/api.openai.com/v1") + + assert agent._is_direct_openai_url() is False + + +def test_direct_openai_url_accepts_native_host(): + agent = _agent_with_base_url("https://api.openai.com/v1") + + assert agent._is_direct_openai_url() is True diff --git a/tests/hermes_cli/test_detect_api_mode_for_url.py b/tests/hermes_cli/test_detect_api_mode_for_url.py index 4fc954032453..f758570ea582 100644 --- a/tests/hermes_cli/test_detect_api_mode_for_url.py +++ b/tests/hermes_cli/test_detect_api_mode_for_url.py @@ -28,6 +28,15 @@ def test_openrouter_is_not_codex_responses(self): # api.openai.com check must exclude openrouter (which routes to openai-hosted models). assert _detect_api_mode_for_url("https://openrouter.ai/api/v1") is None + def test_openai_host_suffix_does_not_match(self): + assert _detect_api_mode_for_url("https://api.openai.com.example/v1") is None + + def test_openai_path_segment_does_not_match(self): + assert _detect_api_mode_for_url("https://proxy.example.test/api.openai.com/v1") is None + + def test_xai_host_suffix_does_not_match(self): + assert _detect_api_mode_for_url("https://api.x.ai.example/v1") is None + class TestAnthropicMessagesDetection: """Third-party gateways that speak the Anthropic protocol under /anthropic.""" From 3d055a540d8810127470a37f167a0802ac4c0dff Mon Sep 17 00:00:00 2001 From: Teknium Date: Mon, 20 Apr 2026 20:58:01 -0700 Subject: [PATCH 2/3] fix: extend hostname-match provider detection across remaining call sites MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Aslaaen's fix in the original PR covered _detect_api_mode_for_url and the two openai/xai sites in run_agent.py. This finishes the sweep: the same substring-match false-positive class (e.g. https://api.openai.com.evil/v1, https://proxy/api.openai.com/v1, https://api.anthropic.com.example/v1) existed in eight more call sites, and the hostname helper was duplicated in two modules. - utils: add shared base_url_hostname() (single source of truth). - hermes_cli/runtime_provider, run_agent: drop local duplicates, import from utils. Reuse the cached AIAgent._base_url_hostname attribute everywhere it's already populated. - agent/auxiliary_client: switch codex-wrap auto-detect, max_completion_tokens gate (auxiliary_max_tokens_param), and custom-endpoint max_tokens kwarg selection to hostname equality. - run_agent: native-anthropic check in the Claude-style model branch and in the AIAgent init provider-auto-detect branch. - agent/model_metadata: Anthropic /v1/models context-length lookup. - hermes_cli/providers.determine_api_mode: anthropic / openai URL heuristics for custom/unknown providers (the /anthropic path-suffix convention for third-party gateways is preserved). - tools/delegate_tool: anthropic detection for delegated subagent runtimes. - hermes_cli/setup, hermes_cli/tools_config: setup-wizard vision-endpoint native-OpenAI detection (paired with deduping the repeated check into a single is_native_openai boolean per branch). Tests: - tests/test_base_url_hostname.py covers the helper directly (path-containing-host, host-suffix, trailing dot, port, case). - tests/hermes_cli/test_determine_api_mode_hostname.py adds the same regression class for determine_api_mode, plus a test that the /anthropic third-party gateway convention still wins. Also: add asslaenn5@gmail.com → Aslaaen to scripts/release.py AUTHOR_MAP. --- agent/auxiliary_client.py | 8 +-- agent/model_metadata.py | 4 +- hermes_cli/providers.py | 7 ++- hermes_cli/runtime_provider.py | 12 +--- hermes_cli/setup.py | 6 +- hermes_cli/tools_config.py | 6 +- run_agent.py | 21 ++----- scripts/release.py | 1 + .../test_determine_api_mode_hostname.py | 43 +++++++++++++++ tests/test_base_url_hostname.py | 55 +++++++++++++++++++ tools/delegate_tool.py | 3 +- utils.py | 22 ++++++++ 12 files changed, 151 insertions(+), 37 deletions(-) create mode 100644 tests/hermes_cli/test_determine_api_mode_hostname.py create mode 100644 tests/test_base_url_hostname.py diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index ea8702cb8172..55199e9b9191 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -48,6 +48,7 @@ from agent.credential_pool import load_pool from hermes_cli.config import get_hermes_home from hermes_constants import OPENROUTER_BASE_URL +from utils import base_url_hostname logger = logging.getLogger(__name__) @@ -1516,8 +1517,7 @@ def _needs_codex_wrap(client_obj, base_url_str: str, model_str: str) -> bool: # Auto-detect: api.openai.com + codex model name pattern if api_mode and api_mode != "codex_responses": return False # explicit non-codex mode - normalized_base = (base_url_str or "").strip().lower() - if "api.openai.com" in normalized_base and "openrouter" not in normalized_base: + if base_url_hostname(base_url_str) == "api.openai.com": model_lower = (model_str or "").lower() if "codex" in model_lower: return True @@ -2025,7 +2025,7 @@ def auxiliary_max_tokens_param(value: int) -> dict: # Only use max_completion_tokens for direct OpenAI custom endpoints if (not or_key and _read_nous_auth() is None - and "api.openai.com" in custom_base.lower()): + and base_url_hostname(custom_base) == "api.openai.com"): return {"max_completion_tokens": value} return {"max_tokens": value} @@ -2460,7 +2460,7 @@ def _build_call_kwargs( # Direct OpenAI api.openai.com with newer models needs max_completion_tokens. if provider == "custom": custom_base = base_url or _current_custom_base_url() - if "api.openai.com" in custom_base.lower(): + if base_url_hostname(custom_base) == "api.openai.com": kwargs["max_completion_tokens"] = max_tokens else: kwargs["max_tokens"] = max_tokens diff --git a/agent/model_metadata.py b/agent/model_metadata.py index c03c5e89cb38..84cd553c3976 100644 --- a/agent/model_metadata.py +++ b/agent/model_metadata.py @@ -14,6 +14,8 @@ import requests import yaml +from utils import base_url_hostname + from hermes_constants import OPENROUTER_MODELS_URL logger = logging.getLogger(__name__) @@ -1078,7 +1080,7 @@ def get_model_context_length( # 4. Anthropic /v1/models API (only for regular API keys, not OAuth) if provider == "anthropic" or ( - base_url and "api.anthropic.com" in base_url + base_url and base_url_hostname(base_url) == "api.anthropic.com" ): ctx = _query_anthropic_context_length(model, base_url or "https://api.anthropic.com", api_key) if ctx: diff --git a/hermes_cli/providers.py b/hermes_cli/providers.py index c701db4d505f..ca8b075f5e2d 100644 --- a/hermes_cli/providers.py +++ b/hermes_cli/providers.py @@ -23,6 +23,8 @@ from dataclasses import dataclass from typing import Any, Dict, List, Optional, Tuple +from utils import base_url_hostname + logger = logging.getLogger(__name__) @@ -434,9 +436,10 @@ def determine_api_mode(provider: str, base_url: str = "") -> str: # URL-based heuristics for custom / unknown providers if base_url: url_lower = base_url.rstrip("/").lower() - if url_lower.endswith("/anthropic") or "api.anthropic.com" in url_lower: + hostname = base_url_hostname(base_url) + if url_lower.endswith("/anthropic") or hostname == "api.anthropic.com": return "anthropic_messages" - if "api.openai.com" in url_lower: + if hostname == "api.openai.com": return "codex_responses" if "bedrock-runtime" in url_lower and "amazonaws.com" in url_lower: return "bedrock_converse" diff --git a/hermes_cli/runtime_provider.py b/hermes_cli/runtime_provider.py index 57b6873d0475..8a7b44fa40b1 100644 --- a/hermes_cli/runtime_provider.py +++ b/hermes_cli/runtime_provider.py @@ -6,7 +6,6 @@ import os import re from typing import Any, Dict, Optional -from urllib.parse import urlparse logger = logging.getLogger(__name__) @@ -30,20 +29,13 @@ ) from hermes_cli.config import get_compatible_custom_providers, load_config from hermes_constants import OPENROUTER_BASE_URL +from utils import base_url_hostname def _normalize_custom_provider_name(value: str) -> str: return value.strip().lower().replace(" ", "-") -def _base_url_hostname(base_url: str) -> str: - raw = (base_url or "").strip() - if not raw: - return "" - parsed = urlparse(raw if "://" in raw else f"//{raw}") - return (parsed.hostname or "").lower().rstrip(".") - - def _detect_api_mode_for_url(base_url: str) -> Optional[str]: """Auto-detect api_mode from the resolved base URL. @@ -56,7 +48,7 @@ def _detect_api_mode_for_url(base_url: str) -> Optional[str]: ``chat_completions``. """ normalized = (base_url or "").strip().lower().rstrip("/") - hostname = _base_url_hostname(base_url) + hostname = base_url_hostname(base_url) if hostname == "api.x.ai": return "codex_responses" if hostname == "api.openai.com": diff --git a/hermes_cli/setup.py b/hermes_cli/setup.py index b4fa877d8c8c..53b0c180aaa1 100644 --- a/hermes_cli/setup.py +++ b/hermes_cli/setup.py @@ -22,6 +22,7 @@ from hermes_cli.nous_subscription import get_nous_subscription_features from tools.tool_backend_helpers import managed_nous_tools_enabled +from utils import base_url_hostname from hermes_constants import get_optional_skills_dir logger = logging.getLogger(__name__) @@ -803,7 +804,8 @@ def setup_model_provider(config: dict, *, quick: bool = False): elif _vision_idx == 1: # OpenAI-compatible endpoint _base_url = prompt(" Base URL (blank for OpenAI)").strip() or "https://api.openai.com/v1" _api_key_label = " API key" - if "api.openai.com" in _base_url.lower(): + _is_native_openai = base_url_hostname(_base_url) == "api.openai.com" + if _is_native_openai: _api_key_label = " OpenAI API key" _oai_key = prompt(_api_key_label, password=True).strip() if _oai_key: @@ -811,7 +813,7 @@ def setup_model_provider(config: dict, *, quick: bool = False): # Save vision base URL to config (not .env — only secrets go there) _vaux = config.setdefault("auxiliary", {}).setdefault("vision", {}) _vaux["base_url"] = _base_url - if "api.openai.com" in _base_url.lower(): + if _is_native_openai: _oai_vision_models = ["gpt-4o", "gpt-4o-mini", "gpt-4.1", "gpt-4.1-mini", "gpt-4.1-nano"] _vm_choices = _oai_vision_models + ["Use default (gpt-4o-mini)"] _vm_idx = prompt_choice("Select vision model:", _vm_choices, 0) diff --git a/hermes_cli/tools_config.py b/hermes_cli/tools_config.py index ba8849e6faf1..23a03b3bd2b8 100644 --- a/hermes_cli/tools_config.py +++ b/hermes_cli/tools_config.py @@ -25,6 +25,7 @@ get_nous_subscription_features, ) from tools.tool_backend_helpers import managed_nous_tools_enabled +from utils import base_url_hostname logger = logging.getLogger(__name__) @@ -1179,7 +1180,8 @@ def _configure_simple_requirements(ts_key: str): _print_warning(" Skipped") elif idx == 1: base_url = _prompt(" OPENAI_BASE_URL (blank for OpenAI)").strip() or "https://api.openai.com/v1" - key_label = " OPENAI_API_KEY" if "api.openai.com" in base_url.lower() else " API key" + is_native_openai = base_url_hostname(base_url) == "api.openai.com" + key_label = " OPENAI_API_KEY" if is_native_openai else " API key" api_key = _prompt(key_label, password=True) if api_key and api_key.strip(): save_env_value("OPENAI_API_KEY", api_key.strip()) @@ -1189,7 +1191,7 @@ def _configure_simple_requirements(ts_key: str): _aux = _cfg.setdefault("auxiliary", {}).setdefault("vision", {}) _aux["base_url"] = base_url save_config(_cfg) - if "api.openai.com" in base_url.lower(): + if is_native_openai: save_env_value("AUXILIARY_VISION_MODEL", "gpt-4o-mini") _print_success(" Saved") else: diff --git a/run_agent.py b/run_agent.py index 9da4bf93f355..cbda3882e486 100644 --- a/run_agent.py +++ b/run_agent.py @@ -38,7 +38,6 @@ from types import SimpleNamespace import uuid from typing import List, Dict, Any, Optional -from urllib.parse import urlparse from openai import OpenAI import fire from datetime import datetime @@ -125,15 +124,7 @@ convert_scratchpad_to_think, has_incomplete_scratchpad, save_trajectory as _save_trajectory_to_file, ) -from utils import atomic_json_write, env_var_enabled - - -def _base_url_hostname(base_url: str) -> str: - raw = (base_url or "").strip() - if not raw: - return "" - parsed = urlparse(raw if "://" in raw else f"//{raw}") - return (parsed.hostname or "").lower().rstrip(".") +from utils import atomic_json_write, base_url_hostname, env_var_enabled @@ -712,7 +703,7 @@ def base_url(self) -> str: def base_url(self, value: str) -> None: self._base_url = value self._base_url_lower = value.lower() if value else "" - self._base_url_hostname = _base_url_hostname(value) + self._base_url_hostname = base_url_hostname(value) def __init__( self, @@ -860,7 +851,7 @@ def __init__( elif (provider_name is None) and self._base_url_hostname == "api.x.ai": self.api_mode = "codex_responses" self.provider = "xai" - elif self.provider == "anthropic" or (provider_name is None and "api.anthropic.com" in self._base_url_lower): + elif self.provider == "anthropic" or (provider_name is None and self._base_url_hostname == "api.anthropic.com"): self.api_mode = "anthropic_messages" self.provider = "anthropic" elif self._base_url_lower.rstrip("/").endswith("/anthropic"): @@ -2270,9 +2261,9 @@ def _replay_compression_warning(self) -> None: def _is_direct_openai_url(self, base_url: str = None) -> bool: """Return True when a base URL targets OpenAI's native API.""" if base_url is not None: - hostname = _base_url_hostname(base_url) + hostname = base_url_hostname(base_url) else: - hostname = getattr(self, "_base_url_hostname", "") or _base_url_hostname( + hostname = getattr(self, "_base_url_hostname", "") or base_url_hostname( getattr(self, "_base_url_lower", "") ) return hostname == "api.openai.com" @@ -2376,7 +2367,7 @@ def _anthropic_prompt_cache_policy( is_anthropic_wire = eff_api_mode == "anthropic_messages" is_native_anthropic = ( is_anthropic_wire - and (eff_provider == "anthropic" or "api.anthropic.com" in base_lower) + and (eff_provider == "anthropic" or base_url_hostname(eff_base_url) == "api.anthropic.com") ) if is_native_anthropic: diff --git a/scripts/release.py b/scripts/release.py index 4cd0c3064684..0b1fa8aa76f0 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -320,6 +320,7 @@ "haileymarshall005@gmail.com": "haileymarshall", "aniruddhaadak80@users.noreply.github.com": "aniruddhaadak80", "zheng.jerilyn@gmail.com": "jerilynzheng", + "asslaenn5@gmail.com": "Aslaaen", } diff --git a/tests/hermes_cli/test_determine_api_mode_hostname.py b/tests/hermes_cli/test_determine_api_mode_hostname.py new file mode 100644 index 000000000000..8b6cd042ce57 --- /dev/null +++ b/tests/hermes_cli/test_determine_api_mode_hostname.py @@ -0,0 +1,43 @@ +"""Regression tests for ``determine_api_mode`` hostname handling. + +Companion to tests/hermes_cli/test_detect_api_mode_for_url.py — the same +false-positive class (custom URLs containing ``api.openai.com`` / +``api.anthropic.com`` as a path segment or host suffix) must be rejected +by ``determine_api_mode`` as well, since it's the code path used by +custom/unknown providers in ``resolve_custom_provider``. +""" + +from __future__ import annotations + +from hermes_cli.providers import determine_api_mode + + +class TestOpenAIHostHardening: + def test_native_openai_url_is_codex_responses(self): + assert determine_api_mode("", "https://api.openai.com/v1") == "codex_responses" + + def test_openai_host_suffix_is_not_codex(self): + assert determine_api_mode("", "https://api.openai.com.example/v1") == "chat_completions" + + def test_openai_path_segment_is_not_codex(self): + assert determine_api_mode("", "https://proxy.example.test/api.openai.com/v1") == "chat_completions" + + +class TestAnthropicHostHardening: + def test_native_anthropic_url_is_anthropic_messages(self): + assert determine_api_mode("", "https://api.anthropic.com") == "anthropic_messages" + + def test_anthropic_host_suffix_is_not_anthropic(self): + assert determine_api_mode("", "https://api.anthropic.com.example/v1") == "chat_completions" + + def test_anthropic_path_segment_is_not_anthropic(self): + # A proxy whose path contains ``api.anthropic.com`` must not be misrouted. + # Note: the ``/anthropic`` convention for third-party gateways still wins + # via explicit path-suffix check — see test_anthropic_path_suffix_still_wins. + assert determine_api_mode("", "https://proxy.example.test/api.anthropic.com/v1") == "chat_completions" + + def test_anthropic_path_suffix_still_wins(self): + # Third-party Anthropic-compatible gateways (MiniMax, Zhipu GLM, LiteLLM + # proxies) expose the Anthropic protocol under a ``/anthropic`` suffix. + # That convention must still resolve to anthropic_messages. + assert determine_api_mode("", "https://api.minimax.io/anthropic") == "anthropic_messages" diff --git a/tests/test_base_url_hostname.py b/tests/test_base_url_hostname.py new file mode 100644 index 000000000000..89842cac2f23 --- /dev/null +++ b/tests/test_base_url_hostname.py @@ -0,0 +1,55 @@ +"""Targeted tests for ``utils.base_url_hostname``. + +The helper is used across provider routing, auxiliary client, and setup +wizards to avoid the substring-match false-positive class documented in +tests/agent/test_direct_provider_url_detection.py. +""" + +from __future__ import annotations + +from utils import base_url_hostname + + +def test_empty_returns_empty_string(): + assert base_url_hostname("") == "" + assert base_url_hostname(None) == "" # type: ignore[arg-type] + + +def test_plain_host_without_scheme(): + assert base_url_hostname("api.openai.com") == "api.openai.com" + assert base_url_hostname("api.openai.com/v1") == "api.openai.com" + + +def test_https_url_extracts_hostname_only(): + assert base_url_hostname("https://api.openai.com/v1") == "api.openai.com" + assert base_url_hostname("https://api.x.ai/v1") == "api.x.ai" + assert base_url_hostname("https://api.anthropic.com") == "api.anthropic.com" + + +def test_hostname_case_insensitive(): + assert base_url_hostname("https://API.OpenAI.com/v1") == "api.openai.com" + + +def test_trailing_dot_stripped(): + # Fully-qualified hostnames may include a trailing dot. + assert base_url_hostname("https://api.openai.com./v1") == "api.openai.com" + + +def test_path_containing_provider_host_is_not_the_hostname(): + # The key regression — proxy paths must never be misread as the host. + assert base_url_hostname("https://proxy.example.test/api.openai.com/v1") == "proxy.example.test" + assert base_url_hostname("https://proxy.example.test/api.anthropic.com/v1") == "proxy.example.test" + + +def test_host_suffix_is_not_the_provider(): + # A hostname that merely ends with the provider domain is not the provider. + assert base_url_hostname("https://api.openai.com.example/v1") == "api.openai.com.example" + assert base_url_hostname("https://api.x.ai.example/v1") == "api.x.ai.example" + + +def test_port_is_ignored(): + assert base_url_hostname("https://api.openai.com:443/v1") == "api.openai.com" + + +def test_whitespace_stripped(): + assert base_url_hostname(" https://api.openai.com/v1 ") == "api.openai.com" diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 2e6065245193..3851bad3fdb2 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -26,6 +26,7 @@ from typing import Any, Dict, List, Optional from toolsets import TOOLSETS +from utils import base_url_hostname # Tools that children must never have access to @@ -1027,7 +1028,7 @@ def _resolve_delegation_credentials(cfg: dict, parent_agent) -> dict: if "chatgpt.com/backend-api/codex" in base_lower: provider = "openai-codex" api_mode = "codex_responses" - elif "api.anthropic.com" in base_lower: + elif base_url_hostname(configured_base_url) == "api.anthropic.com": provider = "anthropic" api_mode = "anthropic_messages" diff --git a/utils.py b/utils.py index cf2582853f59..69a18d584edf 100644 --- a/utils.py +++ b/utils.py @@ -7,6 +7,7 @@ import tempfile from pathlib import Path from typing import Any, Union +from urllib.parse import urlparse import yaml @@ -194,3 +195,24 @@ def env_int(key: str, default: int = 0) -> int: def env_bool(key: str, default: bool = False) -> bool: """Read an environment variable as a boolean.""" return is_truthy_value(os.getenv(key, ""), default=default) + + +# ─── URL Parsing Helpers ────────────────────────────────────────────────────── + + +def base_url_hostname(base_url: str) -> str: + """Return the lowercased hostname for a base URL, or ``""`` if absent. + + Use exact-hostname comparisons against known provider hosts + (``api.openai.com``, ``api.x.ai``, ``api.anthropic.com``) instead of + substring matches on the raw URL. Substring checks treat attacker- or + proxy-controlled paths/hosts like ``https://api.openai.com.example/v1`` + or ``https://proxy.test/api.openai.com/v1`` as native endpoints, which + leads to wrong api_mode / auth routing. + """ + raw = (base_url or "").strip() + if not raw: + return "" + parsed = urlparse(raw if "://" in raw else f"//{raw}") + return (parsed.hostname or "").lower().rstrip(".") + From 5c0e695266e3d92a95769f0fcd5b7e8fa2b13fa4 Mon Sep 17 00:00:00 2001 From: Teknium Date: Mon, 20 Apr 2026 21:17:28 -0700 Subject: [PATCH 3/3] fix: sweep remaining provider-URL substring checks across codebase MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the hostname-hardening sweep — every substring check against a provider host in live-routing code is now hostname-based. This closes the same false-positive class for OpenRouter, GitHub Copilot, Kimi, Qwen, ChatGPT/Codex, Bedrock, GitHub Models, Vercel AI Gateway, Nous, Z.AI, Moonshot, Arcee, and MiniMax that the original PR closed for OpenAI, xAI, and Anthropic. New helper: - utils.base_url_host_matches(base_url, domain) — safe counterpart to 'domain in base_url'. Accepts hostname equality and subdomain matches; rejects path segments, host suffixes, and prefix collisions. Call sites converted (real-code only; tests, optional-skills, red-teaming scripts untouched): run_agent.py (10 sites): - AIAgent.__init__ Bedrock branch, ChatGPT/Codex branch (also path check) - header cascade for openrouter / copilot / kimi / qwen / chatgpt - interleaved-thinking trigger (openrouter + claude) - _is_openrouter_url(), _is_qwen_portal() - is_native_anthropic check - github-models-vs-copilot detection (3 sites) - reasoning-capable route gate (nousresearch, vercel, github) - codex-backend detection in API kwargs build - fallback api_mode Bedrock detection agent/auxiliary_client.py (7 sites): - extra-headers cascades in 4 distinct client-construction paths (resolve custom, resolve auto, OpenRouter-fallback-to-custom, _async_client_from_sync, resolve_provider_client explicit-custom, resolve_auto_with_codex) - _is_openrouter_client() base_url sniff agent/usage_pricing.py: - resolve_billing_route openrouter branch agent/model_metadata.py: - _is_openrouter_base_url(), Bedrock context-length lookup hermes_cli/providers.py: - determine_api_mode Bedrock heuristic hermes_cli/runtime_provider.py: - _is_openrouter_url flag for API-key preference (issues #420, #560) hermes_cli/doctor.py: - Kimi User-Agent header for /models probes tools/delegate_tool.py: - subagent Codex endpoint detection trajectory_compressor.py: - _detect_provider() cascade (8 providers: openrouter, nous, codex, zai, kimi-coding, arcee, minimax-cn, minimax) cli.py, gateway/run.py: - /model-switch cache-enabled hint (openrouter + claude) Bedrock detection tightened from 'bedrock-runtime in url' to 'hostname starts with bedrock-runtime. AND host is under amazonaws.com'. ChatGPT/Codex detection tightened from 'chatgpt.com/backend-api/codex in url' to 'hostname is chatgpt.com AND path contains /backend-api/codex'. Tests: - tests/test_base_url_hostname.py extended with a base_url_host_matches suite (exact match, subdomain, path-segment rejection, host-suffix rejection, host-prefix rejection, empty-input, case-insensitivity, trailing dot). Validation: 651 targeted tests pass (runtime_provider, minimax, bedrock, gemini, auxiliary, codex_cloudflare, usage_pricing, compressor_fallback, fallback_model, openai_client_lifecycle, provider_parity, cli_provider_resolution, delegate, credential_pool, context_compressor, plus the 4 hostname test modules). 26-assertion E2E call-site verification across 6 modules passes. --- agent/auxiliary_client.py | 30 ++++++------- agent/model_metadata.py | 10 +++-- agent/usage_pricing.py | 3 +- cli.py | 7 +-- gateway/run.py | 4 +- hermes_cli/doctor.py | 3 +- hermes_cli/providers.py | 4 +- hermes_cli/runtime_provider.py | 4 +- run_agent.py | 75 ++++++++++++++++++++------------- tests/test_base_url_hostname.py | 67 ++++++++++++++++++++++++++--- tools/delegate_tool.py | 5 ++- trajectory_compressor.py | 27 ++++++++---- utils.py | 21 +++++++++ 13 files changed, 184 insertions(+), 76 deletions(-) diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 55199e9b9191..50d4d86afb0d 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -48,7 +48,7 @@ from agent.credential_pool import load_pool from hermes_cli.config import get_hermes_home from hermes_constants import OPENROUTER_BASE_URL -from utils import base_url_hostname +from utils import base_url_host_matches, base_url_hostname logger = logging.getLogger(__name__) @@ -817,9 +817,9 @@ def _resolve_api_key_provider() -> Tuple[Optional[OpenAI], Optional[str]]: if is_native_gemini_base_url(base_url): return GeminiNativeClient(api_key=api_key, base_url=base_url), model extra = {} - if "api.kimi.com" in base_url.lower(): + if base_url_host_matches(base_url, "api.kimi.com"): extra["default_headers"] = {"User-Agent": "KimiCLI/1.30.0"} - elif "api.githubcopilot.com" in base_url.lower(): + elif base_url_host_matches(base_url, "api.githubcopilot.com"): from hermes_cli.models import copilot_default_headers extra["default_headers"] = copilot_default_headers() @@ -843,9 +843,9 @@ def _resolve_api_key_provider() -> Tuple[Optional[OpenAI], Optional[str]]: if is_native_gemini_base_url(base_url): return GeminiNativeClient(api_key=api_key, base_url=base_url), model extra = {} - if "api.kimi.com" in base_url.lower(): + if base_url_host_matches(base_url, "api.kimi.com"): extra["default_headers"] = {"User-Agent": "KimiCLI/1.30.0"} - elif "api.githubcopilot.com" in base_url.lower(): + elif base_url_host_matches(base_url, "api.githubcopilot.com"): from hermes_cli.models import copilot_default_headers extra["default_headers"] = copilot_default_headers() @@ -994,7 +994,7 @@ def _resolve_custom_runtime() -> Tuple[Optional[str], Optional[str], Optional[st return None, None, None custom_base = custom_base.strip().rstrip("/") - if "openrouter.ai" in custom_base.lower(): + if base_url_host_matches(custom_base, "openrouter.ai"): # requested='custom' falls back to OpenRouter when no custom endpoint is # configured. Treat that as "no custom endpoint" for auxiliary routing. return None, None, None @@ -1433,14 +1433,14 @@ def _to_async_client(sync_client, model: str): "api_key": sync_client.api_key, "base_url": str(sync_client.base_url), } - base_lower = str(sync_client.base_url).lower() - if "openrouter" in base_lower: + sync_base_url = str(sync_client.base_url) + if base_url_host_matches(sync_base_url, "openrouter.ai"): async_kwargs["default_headers"] = dict(_OR_HEADERS) - elif "api.githubcopilot.com" in base_lower: + elif base_url_host_matches(sync_base_url, "api.githubcopilot.com"): from hermes_cli.models import copilot_default_headers async_kwargs["default_headers"] = copilot_default_headers() - elif "api.kimi.com" in base_lower: + elif base_url_host_matches(sync_base_url, "api.kimi.com"): async_kwargs["default_headers"] = {"User-Agent": "KimiCLI/1.30.0"} return AsyncOpenAI(**async_kwargs), model @@ -1621,9 +1621,9 @@ def _wrap_if_needed(client_obj, final_model_str: str, base_url_str: str = ""): provider, ) extra = {} - if "api.kimi.com" in custom_base.lower(): + if base_url_host_matches(custom_base, "api.kimi.com"): extra["default_headers"] = {"User-Agent": "KimiCLI/1.30.0"} - elif "api.githubcopilot.com" in custom_base.lower(): + elif base_url_host_matches(custom_base, "api.githubcopilot.com"): from hermes_cli.models import copilot_default_headers extra["default_headers"] = copilot_default_headers() client = OpenAI(api_key=custom_key, base_url=custom_base, **extra) @@ -1728,9 +1728,9 @@ def _wrap_if_needed(client_obj, final_model_str: str, base_url_str: str = ""): # Provider-specific headers headers = {} - if "api.kimi.com" in base_url.lower(): + if base_url_host_matches(base_url, "api.kimi.com"): headers["User-Agent"] = "KimiCLI/1.30.0" - elif "api.githubcopilot.com" in base_url.lower(): + elif base_url_host_matches(base_url, "api.githubcopilot.com"): from hermes_cli.models import copilot_default_headers headers.update(copilot_default_headers()) @@ -2154,7 +2154,7 @@ def cleanup_stale_async_clients() -> None: def _is_openrouter_client(client: Any) -> bool: for obj in (client, getattr(client, "_client", None), getattr(client, "client", None)): - if obj and "openrouter" in str(getattr(obj, "base_url", "") or "").lower(): + if obj and base_url_host_matches(str(getattr(obj, "base_url", "") or ""), "openrouter.ai"): return True return False diff --git a/agent/model_metadata.py b/agent/model_metadata.py index 84cd553c3976..47f9bba94fdd 100644 --- a/agent/model_metadata.py +++ b/agent/model_metadata.py @@ -14,7 +14,7 @@ import requests import yaml -from utils import base_url_hostname +from utils import base_url_host_matches, base_url_hostname from hermes_constants import OPENROUTER_MODELS_URL @@ -220,7 +220,7 @@ def _auth_headers(api_key: str = "") -> Dict[str, str]: def _is_openrouter_base_url(base_url: str) -> bool: - return "openrouter.ai" in _normalize_base_url(base_url).lower() + return base_url_host_matches(base_url, "openrouter.ai") def _is_custom_endpoint(base_url: str) -> bool: @@ -1089,7 +1089,11 @@ def get_model_context_length( # 4b. AWS Bedrock — use static context length table. # Bedrock's ListFoundationModels doesn't expose context window sizes, # so we maintain a curated table in bedrock_adapter.py. - if provider == "bedrock" or (base_url and "bedrock-runtime" in base_url): + if provider == "bedrock" or ( + base_url + and base_url_hostname(base_url).startswith("bedrock-runtime.") + and base_url_host_matches(base_url, "amazonaws.com") + ): try: from agent.bedrock_adapter import get_bedrock_context_length return get_bedrock_context_length(model) diff --git a/agent/usage_pricing.py b/agent/usage_pricing.py index 29c75b172ad7..3554c5b9914b 100644 --- a/agent/usage_pricing.py +++ b/agent/usage_pricing.py @@ -6,6 +6,7 @@ from typing import Any, Dict, Literal, Optional from agent.model_metadata import fetch_endpoint_model_metadata, fetch_model_metadata +from utils import base_url_host_matches DEFAULT_PRICING = {"input": 0.0, "output": 0.0} @@ -393,7 +394,7 @@ def resolve_billing_route( if provider_name == "openai-codex": return BillingRoute(provider="openai-codex", model=model, base_url=base_url or "", billing_mode="subscription_included") - if provider_name == "openrouter" or "openrouter.ai" in base: + if provider_name == "openrouter" or base_url_host_matches(base_url or "", "openrouter.ai"): return BillingRoute(provider="openrouter", model=model, base_url=base_url or "", billing_mode="official_models_api") if provider_name == "anthropic": return BillingRoute(provider="anthropic", model=model.split("/")[-1], base_url=base_url or "", billing_mode="official_docs_snapshot") diff --git a/cli.py b/cli.py index 15f60aa307c0..68243946f4db 100644 --- a/cli.py +++ b/cli.py @@ -74,6 +74,7 @@ # User-managed env files should override stale shell exports on restart. from hermes_constants import get_hermes_home, display_hermes_home from hermes_cli.env_loader import load_hermes_dotenv +from utils import base_url_host_matches _hermes_home = get_hermes_home() _project_env = Path(__file__).parent / '.env' @@ -1836,7 +1837,7 @@ def __init__( # Match key to resolved base_url: OpenRouter URL → prefer OPENROUTER_API_KEY, # custom endpoint → prefer OPENAI_API_KEY (issue #560). # Note: _ensure_runtime_credentials() re-resolves this before first use. - if self.base_url and "openrouter.ai" in self.base_url: + if self.base_url and base_url_host_matches(self.base_url, "openrouter.ai"): self.api_key = api_key or os.getenv("OPENROUTER_API_KEY") or os.getenv("OPENAI_API_KEY") else: self.api_key = api_key or os.getenv("OPENAI_API_KEY") or os.getenv("OPENROUTER_API_KEY") @@ -4996,7 +4997,7 @@ def _apply_model_switch_result(self, result, persist_global: bool) -> None: pass cache_enabled = ( - ("openrouter" in (result.base_url or "").lower() and "claude" in result.new_model.lower()) + (base_url_host_matches(result.base_url or "", "openrouter.ai") and "claude" in result.new_model.lower()) or result.api_mode == "anthropic_messages" ) if cache_enabled: @@ -5224,7 +5225,7 @@ def _handle_model_switch(self, cmd_original: str): # Cache notice cache_enabled = ( - ("openrouter" in (result.base_url or "").lower() and "claude" in result.new_model.lower()) + (base_url_host_matches(result.base_url or "", "openrouter.ai") and "claude" in result.new_model.lower()) or result.api_mode == "anthropic_messages" ) if cache_enabled: diff --git a/gateway/run.py b/gateway/run.py index 3fba1d8d9965..6ce409ff1be3 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -86,7 +86,7 @@ def _ensure_ssl_certs() -> None: # Resolve Hermes home directory (respects HERMES_HOME override) from hermes_constants import get_hermes_home -from utils import atomic_yaml_write, is_truthy_value +from utils import atomic_yaml_write, base_url_host_matches, is_truthy_value _hermes_home = get_hermes_home() # Load environment variables from ~/.hermes/.env first. @@ -5661,7 +5661,7 @@ async def _on_model_selected( # Cache notice cache_enabled = ( - ("openrouter" in (result.base_url or "").lower() and "claude" in result.new_model.lower()) + (base_url_host_matches(result.base_url or "", "openrouter.ai") and "claude" in result.new_model.lower()) or result.api_mode == "anthropic_messages" ) if cache_enabled: diff --git a/hermes_cli/doctor.py b/hermes_cli/doctor.py index 8247d25913ef..e16f0bf5e64b 100644 --- a/hermes_cli/doctor.py +++ b/hermes_cli/doctor.py @@ -30,6 +30,7 @@ from hermes_cli.colors import Colors, color from hermes_constants import OPENROUTER_MODELS_URL +from utils import base_url_host_matches _PROVIDER_ENV_HINTS = ( @@ -952,7 +953,7 @@ def run_doctor(args): _base = _to_openai_base_url(_base) _url = (_base.rstrip("/") + "/models") if _base else _default_url _headers = {"Authorization": f"Bearer {_key}"} - if "api.kimi.com" in _url.lower(): + if base_url_host_matches(_base, "api.kimi.com"): _headers["User-Agent"] = "KimiCLI/1.30.0" _resp = httpx.get( _url, diff --git a/hermes_cli/providers.py b/hermes_cli/providers.py index ca8b075f5e2d..1764474aa9a5 100644 --- a/hermes_cli/providers.py +++ b/hermes_cli/providers.py @@ -23,7 +23,7 @@ from dataclasses import dataclass from typing import Any, Dict, List, Optional, Tuple -from utils import base_url_hostname +from utils import base_url_host_matches, base_url_hostname logger = logging.getLogger(__name__) @@ -441,7 +441,7 @@ def determine_api_mode(provider: str, base_url: str = "") -> str: return "anthropic_messages" if hostname == "api.openai.com": return "codex_responses" - if "bedrock-runtime" in url_lower and "amazonaws.com" in url_lower: + if hostname.startswith("bedrock-runtime.") and base_url_host_matches(base_url, "amazonaws.com"): return "bedrock_converse" return "chat_completions" diff --git a/hermes_cli/runtime_provider.py b/hermes_cli/runtime_provider.py index 8a7b44fa40b1..3b2b4cab3cfd 100644 --- a/hermes_cli/runtime_provider.py +++ b/hermes_cli/runtime_provider.py @@ -29,7 +29,7 @@ ) from hermes_cli.config import get_compatible_custom_providers, load_config from hermes_constants import OPENROUTER_BASE_URL -from utils import base_url_hostname +from utils import base_url_host_matches, base_url_hostname def _normalize_custom_provider_name(value: str) -> str: @@ -482,7 +482,7 @@ def _resolve_openrouter_runtime( # When hitting a custom endpoint (e.g. Z.ai, local LLM), prefer # OPENAI_API_KEY so the OpenRouter key doesn't leak to an unrelated # provider (issues #420, #560). - _is_openrouter_url = "openrouter.ai" in base_url + _is_openrouter_url = base_url_host_matches(base_url, "openrouter.ai") if _is_openrouter_url: api_key_candidates = [ explicit_api_key, diff --git a/run_agent.py b/run_agent.py index cbda3882e486..5ec62a06a282 100644 --- a/run_agent.py +++ b/run_agent.py @@ -124,7 +124,7 @@ convert_scratchpad_to_think, has_incomplete_scratchpad, save_trajectory as _save_trajectory_to_file, ) -from utils import atomic_json_write, base_url_hostname, env_var_enabled +from utils import atomic_json_write, base_url_host_matches, base_url_hostname, env_var_enabled @@ -845,7 +845,10 @@ def __init__( self.api_mode = "codex_responses" elif self.provider == "xai": self.api_mode = "codex_responses" - elif (provider_name is None) and "chatgpt.com/backend-api/codex" in self._base_url_lower: + elif (provider_name is None) and ( + self._base_url_hostname == "chatgpt.com" + and "/backend-api/codex" in self._base_url_lower + ): self.api_mode = "codex_responses" self.provider = "openai-codex" elif (provider_name is None) and self._base_url_hostname == "api.x.ai": @@ -859,8 +862,12 @@ def __init__( # use a URL convention ending in /anthropic. Auto-detect these so the # Anthropic Messages API adapter is used instead of chat completions. self.api_mode = "anthropic_messages" - elif self.provider == "bedrock" or "bedrock-runtime" in self._base_url_lower: - # AWS Bedrock — auto-detect from provider name or base URL. + elif self.provider == "bedrock" or ( + self._base_url_hostname.startswith("bedrock-runtime.") + and base_url_host_matches(self._base_url_lower, "amazonaws.com") + ): + # AWS Bedrock — auto-detect from provider name or base URL + # (bedrock-runtime..amazonaws.com). self.api_mode = "bedrock_converse" else: self.api_mode = "chat_completions" @@ -1158,23 +1165,23 @@ def __init__( client_kwargs["command"] = self.acp_command client_kwargs["args"] = self.acp_args effective_base = base_url - if "openrouter" in effective_base.lower(): + if base_url_host_matches(effective_base, "openrouter.ai"): client_kwargs["default_headers"] = { "HTTP-Referer": "https://hermes-agent.nousresearch.com", "X-OpenRouter-Title": "Hermes Agent", "X-OpenRouter-Categories": "productivity,cli-agent", } - elif "api.githubcopilot.com" in effective_base.lower(): + elif base_url_host_matches(effective_base, "api.githubcopilot.com"): from hermes_cli.models import copilot_default_headers client_kwargs["default_headers"] = copilot_default_headers() - elif "api.kimi.com" in effective_base.lower(): + elif base_url_host_matches(effective_base, "api.kimi.com"): client_kwargs["default_headers"] = { "User-Agent": "KimiCLI/1.30.0", } - elif "portal.qwen.ai" in effective_base.lower(): + elif base_url_host_matches(effective_base, "portal.qwen.ai"): client_kwargs["default_headers"] = _qwen_portal_headers() - elif "chatgpt.com" in effective_base.lower(): + elif base_url_host_matches(effective_base, "chatgpt.com"): from agent.auxiliary_client import _codex_cloudflare_headers client_kwargs["default_headers"] = _codex_cloudflare_headers(api_key) else: @@ -1230,7 +1237,7 @@ def __init__( # stream tool call arguments token-by-token, keeping the # connection alive. _effective_base = str(client_kwargs.get("base_url", "")).lower() - if "openrouter" in _effective_base and "claude" in (self.model or "").lower(): + if base_url_host_matches(_effective_base, "openrouter.ai") and "claude" in (self.model or "").lower(): headers = client_kwargs.get("default_headers") or {} existing_beta = headers.get("x-anthropic-beta", "") _FINE_GRAINED = "fine-grained-tool-streaming-2025-05-14" @@ -2328,7 +2335,7 @@ def _compute_non_stream_stale_timeout(self, messages: list[dict[str, Any]]) -> f def _is_openrouter_url(self) -> bool: """Return True when the base URL targets OpenRouter.""" - return "openrouter" in self._base_url_lower + return base_url_host_matches(self._base_url_lower, "openrouter.ai") def _anthropic_prompt_cache_policy( self, @@ -2363,7 +2370,7 @@ def _anthropic_prompt_cache_policy( base_lower = eff_base_url.lower() is_claude = "claude" in eff_model.lower() - is_openrouter = "openrouter" in base_lower + is_openrouter = base_url_host_matches(eff_base_url, "openrouter.ai") is_anthropic_wire = eff_api_mode == "anthropic_messages" is_native_anthropic = ( is_anthropic_wire @@ -5002,20 +5009,19 @@ def _try_refresh_anthropic_client_credentials(self) -> bool: def _apply_client_headers_for_base_url(self, base_url: str) -> None: from agent.auxiliary_client import _AI_GATEWAY_HEADERS, _OR_HEADERS - normalized = (base_url or "").lower() - if "openrouter" in normalized: + if base_url_host_matches(base_url, "openrouter.ai"): self._client_kwargs["default_headers"] = dict(_OR_HEADERS) - elif "ai-gateway.vercel.sh" in normalized: + elif base_url_host_matches(base_url, "ai-gateway.vercel.sh"): self._client_kwargs["default_headers"] = dict(_AI_GATEWAY_HEADERS) - elif "api.githubcopilot.com" in normalized: + elif base_url_host_matches(base_url, "api.githubcopilot.com"): from hermes_cli.models import copilot_default_headers self._client_kwargs["default_headers"] = copilot_default_headers() - elif "api.kimi.com" in normalized: + elif base_url_host_matches(base_url, "api.kimi.com"): self._client_kwargs["default_headers"] = {"User-Agent": "KimiCLI/1.30.0"} - elif "portal.qwen.ai" in normalized: + elif base_url_host_matches(base_url, "portal.qwen.ai"): self._client_kwargs["default_headers"] = _qwen_portal_headers() - elif "chatgpt.com" in normalized: + elif base_url_host_matches(base_url, "chatgpt.com"): from agent.auxiliary_client import _codex_cloudflare_headers self._client_kwargs["default_headers"] = _codex_cloudflare_headers( self._client_kwargs.get("api_key", "") @@ -6163,7 +6169,10 @@ def _try_activate_fallback(self) -> bool: # provider-specific exceptions like Copilot gpt-5-mini on # chat completions. fb_api_mode = "codex_responses" - elif fb_provider == "bedrock" or "bedrock-runtime" in fb_base_url.lower(): + elif fb_provider == "bedrock" or ( + base_url_hostname(fb_base_url).startswith("bedrock-runtime.") + and base_url_host_matches(fb_base_url, "amazonaws.com") + ): fb_api_mode = "bedrock_converse" old_model = self.model @@ -6596,7 +6605,7 @@ def _anthropic_preserve_dots(self) -> bool: def _is_qwen_portal(self) -> bool: """Return True when the base URL targets Qwen Portal.""" - return "portal.qwen.ai" in self._base_url_lower + return base_url_host_matches(self._base_url_lower, "portal.qwen.ai") def _qwen_prepare_chat_messages(self, api_messages: list) -> list: prepared = copy.deepcopy(api_messages) @@ -6717,12 +6726,15 @@ def _build_api_kwargs(self, api_messages: list) -> dict: instructions = DEFAULT_AGENT_IDENTITY is_github_responses = ( - "models.github.ai" in self.base_url.lower() - or "api.githubcopilot.com" in self.base_url.lower() + base_url_host_matches(self.base_url, "models.github.ai") + or base_url_host_matches(self.base_url, "api.githubcopilot.com") ) is_codex_backend = ( self.provider == "openai-codex" - or "chatgpt.com/backend-api/codex" in self.base_url.lower() + or ( + self._base_url_hostname == "chatgpt.com" + and "/backend-api/codex" in self._base_url_lower + ) ) # Resolve reasoning effort: config > default (medium) @@ -6923,8 +6935,8 @@ def _build_api_kwargs(self, api_messages: list) -> dict: _is_openrouter = self._is_openrouter_url() _is_github_models = ( - "models.github.ai" in self._base_url_lower - or "api.githubcopilot.com" in self._base_url_lower + base_url_host_matches(self._base_url_lower, "models.github.ai") + or base_url_host_matches(self._base_url_lower, "api.githubcopilot.com") ) # Provider preferences (only, ignore, order, sort) are OpenRouter- @@ -7000,11 +7012,14 @@ def _supports_reasoning_extra_body(self) -> bool: Some providers/routes reject `reasoning` with 400s, so gate it to known reasoning-capable model families and direct Nous Portal. """ - if "nousresearch" in self._base_url_lower: + if base_url_host_matches(self._base_url_lower, "nousresearch.com"): return True - if "ai-gateway.vercel.sh" in self._base_url_lower: + if base_url_host_matches(self._base_url_lower, "ai-gateway.vercel.sh"): return True - if "models.github.ai" in self._base_url_lower or "api.githubcopilot.com" in self._base_url_lower: + if ( + base_url_host_matches(self._base_url_lower, "models.github.ai") + or base_url_host_matches(self._base_url_lower, "api.githubcopilot.com") + ): try: from hermes_cli.models import github_model_reasoning_efforts @@ -10566,7 +10581,7 @@ def _stop_spinner(): self._vprint(f"{self.log_prefix} 💡 Your API key was rejected by the provider. Check:", force=True) self._vprint(f"{self.log_prefix} • Is the key valid? Run: hermes setup", force=True) self._vprint(f"{self.log_prefix} • Does your account have access to {_model}?", force=True) - if "openrouter" in str(_base).lower(): + if base_url_host_matches(str(_base), "openrouter.ai"): self._vprint(f"{self.log_prefix} • Check credits: https://openrouter.ai/settings/credits", force=True) else: self._vprint(f"{self.log_prefix} 💡 This type of error won't be fixed by retrying.", force=True) diff --git a/tests/test_base_url_hostname.py b/tests/test_base_url_hostname.py index 89842cac2f23..54aca08c0279 100644 --- a/tests/test_base_url_hostname.py +++ b/tests/test_base_url_hostname.py @@ -1,13 +1,17 @@ -"""Targeted tests for ``utils.base_url_hostname``. +"""Targeted tests for ``utils.base_url_hostname`` and ``base_url_host_matches``. -The helper is used across provider routing, auxiliary client, and setup -wizards to avoid the substring-match false-positive class documented in +These helpers are used across provider routing, auxiliary client, setup +wizards, billing routes, and the trajectory compressor to avoid the +substring-match false-positive class documented in tests/agent/test_direct_provider_url_detection.py. """ from __future__ import annotations -from utils import base_url_hostname +from utils import base_url_hostname, base_url_host_matches + + +# ─── base_url_hostname ──────────────────────────────────────────────────── def test_empty_returns_empty_string(): @@ -31,18 +35,15 @@ def test_hostname_case_insensitive(): def test_trailing_dot_stripped(): - # Fully-qualified hostnames may include a trailing dot. assert base_url_hostname("https://api.openai.com./v1") == "api.openai.com" def test_path_containing_provider_host_is_not_the_hostname(): - # The key regression — proxy paths must never be misread as the host. assert base_url_hostname("https://proxy.example.test/api.openai.com/v1") == "proxy.example.test" assert base_url_hostname("https://proxy.example.test/api.anthropic.com/v1") == "proxy.example.test" def test_host_suffix_is_not_the_provider(): - # A hostname that merely ends with the provider domain is not the provider. assert base_url_hostname("https://api.openai.com.example/v1") == "api.openai.com.example" assert base_url_hostname("https://api.x.ai.example/v1") == "api.x.ai.example" @@ -53,3 +54,55 @@ def test_port_is_ignored(): def test_whitespace_stripped(): assert base_url_hostname(" https://api.openai.com/v1 ") == "api.openai.com" + + +# ─── base_url_host_matches ──────────────────────────────────────────────── + + +class TestBaseUrlHostMatchesExact: + def test_exact_domain_matches(self): + assert base_url_host_matches("https://openrouter.ai/api/v1", "openrouter.ai") is True + assert base_url_host_matches("https://moonshot.ai", "moonshot.ai") is True + + def test_subdomain_matches(self): + # A subdomain of the registered domain should match — needed for + # api.moonshot.ai / api.kimi.com / portal.qwen.ai lookups that + # accept both the bare registrable domain and any subdomain under it. + assert base_url_host_matches("https://api.moonshot.ai/v1", "moonshot.ai") is True + assert base_url_host_matches("https://api.kimi.com/v1", "api.kimi.com") is True + assert base_url_host_matches("https://portal.qwen.ai/v1", "portal.qwen.ai") is True + + +class TestBaseUrlHostMatchesNegatives: + """The reason this helper exists — defend against substring collisions.""" + + def test_path_segment_containing_domain_does_not_match(self): + assert base_url_host_matches("https://evil.test/moonshot.ai/v1", "moonshot.ai") is False + assert base_url_host_matches("https://proxy.example.test/openrouter.ai/v1", "openrouter.ai") is False + assert base_url_host_matches("https://proxy/api.kimi.com/v1", "api.kimi.com") is False + + def test_host_suffix_does_not_match(self): + # Attacker-controlled hosts that end with the domain string are not + # the domain. + assert base_url_host_matches("https://moonshot.ai.evil/v1", "moonshot.ai") is False + assert base_url_host_matches("https://openrouter.ai.example/v1", "openrouter.ai") is False + + def test_host_prefix_does_not_match(self): + # "fake-openrouter.ai" is not a subdomain of openrouter.ai. + assert base_url_host_matches("https://fake-openrouter.ai/v1", "openrouter.ai") is False + + +class TestBaseUrlHostMatchesEdgeCases: + def test_empty_base_url_returns_false(self): + assert base_url_host_matches("", "openrouter.ai") is False + assert base_url_host_matches(None, "openrouter.ai") is False # type: ignore[arg-type] + + def test_empty_domain_returns_false(self): + assert base_url_host_matches("https://openrouter.ai/v1", "") is False + + def test_case_insensitive(self): + assert base_url_host_matches("https://OpenRouter.AI/v1", "openrouter.ai") is True + assert base_url_host_matches("https://openrouter.ai/v1", "OPENROUTER.AI") is True + + def test_trailing_dot_on_domain_stripped(self): + assert base_url_host_matches("https://openrouter.ai/v1", "openrouter.ai.") is True diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 3851bad3fdb2..7065e129acaa 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -1025,7 +1025,10 @@ def _resolve_delegation_credentials(cfg: dict, parent_agent) -> dict: base_lower = configured_base_url.lower() provider = "custom" api_mode = "chat_completions" - if "chatgpt.com/backend-api/codex" in base_lower: + if ( + base_url_hostname(configured_base_url) == "chatgpt.com" + and "/backend-api/codex" in base_lower + ): provider = "openai-codex" api_mode = "codex_responses" elif base_url_hostname(configured_base_url) == "api.anthropic.com": diff --git a/trajectory_compressor.py b/trajectory_compressor.py index b0fec6041eea..ff2dcc6266f2 100644 --- a/trajectory_compressor.py +++ b/trajectory_compressor.py @@ -40,6 +40,8 @@ from typing import List, Dict, Any, Optional, Tuple, Callable from dataclasses import dataclass, field from datetime import datetime + +from utils import base_url_host_matches, base_url_hostname import fire from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn, TimeElapsedColumn, TimeRemainingColumn from rich.console import Console @@ -432,22 +434,29 @@ def _get_async_client(self): def _detect_provider(self) -> str: """Detect the provider name from the configured base_url.""" - url = (self.config.base_url or "").lower() - if "openrouter" in url: + url = self.config.base_url or "" + if base_url_host_matches(url, "openrouter.ai"): return "openrouter" - if "nousresearch.com" in url: + if base_url_host_matches(url, "nousresearch.com"): return "nous" - if "chatgpt.com/backend-api/codex" in url: + if ( + base_url_hostname(url) == "chatgpt.com" + and "/backend-api/codex" in url.lower() + ): return "codex" - if "api.z.ai" in url: + if base_url_host_matches(url, "z.ai"): return "zai" - if "moonshot.ai" in url or "moonshot.cn" in url or "api.kimi.com" in url: + if ( + base_url_host_matches(url, "moonshot.ai") + or base_url_host_matches(url, "moonshot.cn") + or base_url_host_matches(url, "api.kimi.com") + ): return "kimi-coding" - if "arcee.ai" in url: + if base_url_host_matches(url, "arcee.ai"): return "arcee" - if "minimaxi.com" in url: + if base_url_host_matches(url, "minimaxi.com"): return "minimax-cn" - if "minimax.io" in url: + if base_url_host_matches(url, "minimax.io"): return "minimax" # Unknown base_url — not a known provider return "" diff --git a/utils.py b/utils.py index 69a18d584edf..6b998e223084 100644 --- a/utils.py +++ b/utils.py @@ -216,3 +216,24 @@ def base_url_hostname(base_url: str) -> str: parsed = urlparse(raw if "://" in raw else f"//{raw}") return (parsed.hostname or "").lower().rstrip(".") + +def base_url_host_matches(base_url: str, domain: str) -> bool: + """Return True when the base URL's hostname is ``domain`` or a subdomain. + + Safer counterpart to ``domain in base_url``, which is the substring + false-positive class documented on ``base_url_hostname``. Accepts bare + hosts, full URLs, and URLs with paths. + + base_url_host_matches("https://api.moonshot.ai/v1", "moonshot.ai") == True + base_url_host_matches("https://moonshot.ai", "moonshot.ai") == True + base_url_host_matches("https://evil.com/moonshot.ai/v1", "moonshot.ai") == False + base_url_host_matches("https://moonshot.ai.evil/v1", "moonshot.ai") == False + """ + hostname = base_url_hostname(base_url) + if not hostname: + return False + domain = (domain or "").strip().lower().rstrip(".") + if not domain: + return False + return hostname == domain or hostname.endswith("." + domain) +