fix(config): prevent custom endpoint slug collisions from cross-routing API credentials

hermes_cli/config.py

@@ -15,6 +15,7 @@
 """
 
 import copy
+import hashlib
 import json
 import logging
 import os
@@ -8219,11 +8220,26 @@ def custom_endpoint_key_env(identity: str) -> str:
     - It keys off the endpoint's own identity, not just its hostname, so two
       endpoints on one host (``127.0.0.1:8000`` and ``:8001``) get separate
       slots instead of the second save clobbering the first's credential.
+    - A stable digest preserves distinctions the readable slug cannot. For
+      example, ``acme-prod`` and ``acme_prod`` both slug to ``ACME_PROD``;
+      without the digest, saving either endpoint silently overwrites the
+      other's credential.
     - The fixed ``HERMES_CUSTOM_`` prefix keeps the result a valid POSIX name
       even when the slug starts with a digit, which every IP-based local
       endpoint does (``127.0.0.1`` → ``127_0_0_1``). ``save_env_value``
       rejects digit-leading names outright.
     """
+    canonical = str(identity or "").strip().upper()
+    if not canonical:
+        return "HERMES_CUSTOM_API_KEY"
+    slug = re.sub(r"[^A-Z0-9]+", "_", canonical).strip("_")
+    digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest().upper()
+    readable = f"{slug}_" if slug else ""
+    return f"HERMES_CUSTOM_{readable}{digest}_API_KEY"
+
+
+def _legacy_custom_endpoint_key_env(identity: str) -> str:
+    """Return the pre-digest custom-endpoint slot for migration cleanup."""
     slug = re.sub(r"[^A-Z0-9]+", "_", str(identity or "").upper()).strip("_")
     return f"HERMES_CUSTOM_{slug}_API_KEY" if slug else "HERMES_CUSTOM_API_KEY"
 

hermes_cli/web_server.py

@@ -73,6 +73,7 @@
     save_env_value,
     remove_env_value,
     custom_endpoint_key_env,
+    _legacy_custom_endpoint_key_env,
     check_config_version,
     detect_install_method,
     format_docker_update_message,
@@ -7709,7 +7710,30 @@ def _detach_main_model_from_provider(cfg: Dict[str, Any], provider_key: str) ->
     cfg["model"] = model_cfg
 
 
-def _write_custom_endpoint(cfg: Dict[str, Any], body: CustomEndpointUpdate) -> Tuple[str, Dict[str, Any]]:
+def _config_references_key_env(cfg: Dict[str, Any], key_env: str) -> bool:
+    """Return whether another configured route still owns ``key_env``."""
+    candidates: List[Any] = [cfg.get("model")]
+    providers = cfg.get("providers")
+    if isinstance(providers, dict):
+        candidates.extend(providers.values())
+    custom_providers = cfg.get("custom_providers")
+    if isinstance(custom_providers, list):
+        candidates.extend(custom_providers)
+
+    template = f"${{{key_env}}}"
+    for entry in candidates:
+        if not isinstance(entry, dict):
+            continue
+        if str(entry.get("key_env") or "").strip() == key_env:
+            return True
+        if str(entry.get("api_key") or "").strip() == template:
+            return True
+    return False
+
+
+def _write_custom_endpoint(
+    cfg: Dict[str, Any], body: CustomEndpointUpdate
+) -> Tuple[str, Dict[str, Any], Tuple[str, ...]]:
     endpoint_id = _custom_endpoint_id(body.id or body.name)
     name = (body.name or "").strip()
     base_url = (body.base_url or "").strip().rstrip("/")
@@ -7731,6 +7755,8 @@ def _write_custom_endpoint(cfg: Dict[str, Any], body: CustomEndpointUpdate) -> T
     existing = providers.get(endpoint_id)
     if not isinstance(existing, dict):
         existing = {}
+    previous_key_env = str(existing.get("key_env") or "").strip()
+    obsolete_key_envs: set[str] = set()
 
     # Merge onto the existing entry rather than replacing it. A providers.<name>
     # block is not owned by this panel: it can carry hand-written keys the
@@ -7769,14 +7795,17 @@ def _write_custom_endpoint(cfg: Dict[str, Any], body: CustomEndpointUpdate) -> T
     # reference it via ``key_env`` — the same indirection built-in providers
     # use and that runtime_provider.py already resolves at load time.
     env_var = custom_endpoint_key_env(endpoint_id)
+    legacy_env_var = _legacy_custom_endpoint_key_env(endpoint_id)
     submitted_key = body.api_key.strip() if body.api_key is not None else None
     if submitted_key:
         save_env_value(env_var, submitted_key)
         entry["key_env"] = env_var
         entry.pop("api_key", None)
+        if previous_key_env == legacy_env_var and previous_key_env != env_var:
+            obsolete_key_envs.add(previous_key_env)
     elif submitted_key is not None:
         # Blank field means "clear the key", not "leave it alone".
-        remove_env_value(env_var)
+        obsolete_key_envs.update((env_var, legacy_env_var))
         entry.pop("key_env", None)
         entry.pop("api_key", None)
     elif str(entry.get("api_key") or "").strip() and not _config_api_key_is_env_ref(endpoint_id):
@@ -7791,6 +7820,23 @@ def _write_custom_endpoint(cfg: Dict[str, Any], body: CustomEndpointUpdate) -> T
     providers[endpoint_id] = entry
     cfg["providers"] = providers
 
+    # ``activate`` mirrors the endpoint credential onto ``model``. Rotating or
+    # clearing an already-active endpoint must update that mirror before the
+    # legacy env slot is removed, otherwise the next request loses auth or
+    # keeps following the stale credential reference.
+    model_cfg = cfg.get("model")
+    if (
+        submitted_key is not None
+        and isinstance(model_cfg, dict)
+        and str(model_cfg.get("provider") or "").strip().lower() == endpoint_id
+    ):
+        if submitted_key:
+            model_cfg["key_env"] = env_var
+            model_cfg.pop("api_key", None)
+        else:
+            model_cfg.pop("key_env", None)
+            model_cfg.pop("api_key", None)
+
     if body.make_default:
         cfg["model"] = _apply_main_model_assignment(
             cfg.get("model", {}), endpoint_id, model, base_url
@@ -7799,7 +7845,7 @@ def _write_custom_endpoint(cfg: Dict[str, Any], body: CustomEndpointUpdate) -> T
             cfg["model"]["key_env"] = entry["key_env"]
             cfg["model"].pop("api_key", None)
 
-    return endpoint_id, entry
+    return endpoint_id, entry, tuple(sorted(obsolete_key_envs))
 
 
 @app.get("/api/providers/custom-endpoints")
@@ -7817,8 +7863,13 @@ def upsert_custom_endpoint(body: CustomEndpointUpdate):
     """Create or update a v12+ ``providers`` custom endpoint entry."""
     try:
         cfg = load_config()
-        endpoint_id, _entry = _write_custom_endpoint(cfg, body)
+        endpoint_id, _entry, obsolete_key_envs = _write_custom_endpoint(cfg, body)
         save_config(cfg)
+        # Commit the new config reference before removing the old secret. If
+        # config persistence fails, the previous endpoint remains usable.
+        for key_env in obsolete_key_envs:
+            if not _config_references_key_env(cfg, key_env):
+                remove_env_value(key_env)
         response = _custom_endpoint_response(cfg)
         response["ok"] = True
         response["id"] = endpoint_id
@@ -7875,8 +7926,15 @@ def delete_custom_endpoint(endpoint_id: str):
         providers.pop(provider_key, None)
         cfg["providers"] = providers
         _detach_main_model_from_provider(cfg, provider_key)
-        remove_env_value(custom_endpoint_key_env(provider_key))
         save_config(cfg)
+        # Remove both the current collision-resistant slot and the legacy
+        # slug-only slot used by endpoints saved before this fix.
+        for key_env in {
+            custom_endpoint_key_env(provider_key),
+            _legacy_custom_endpoint_key_env(provider_key),
+        }:
+            if not _config_references_key_env(cfg, key_env):
+                remove_env_value(key_env)
         response = _custom_endpoint_response(cfg)
         response["ok"] = True
         return response

tests/cli/test_cli_provider_resolution.py

@@ -1011,3 +1011,11 @@ def test_custom_endpoint_key_env_separates_ports_on_one_host():
 
     assert custom_endpoint_key_env("127.0.0.1_8000") != custom_endpoint_key_env("127.0.0.1_8001")
     assert custom_endpoint_key_env("acme") == custom_endpoint_key_env("ACME")
+
+
+def test_custom_endpoint_key_env_preserves_punctuation_distinctions():
+    """Readable slugs that match must still use different credential slots."""
+    from hermes_cli.config import custom_endpoint_key_env
+
+    identities = ("acme-prod", "acme_prod", "acme.prod", "acme prod")
+    assert len({custom_endpoint_key_env(identity) for identity in identities}) == len(identities)

tests/hermes_cli/test_web_server.py

@@ -4952,6 +4952,147 @@ def test_two_endpoints_on_one_host_keep_separate_credentials(self):
         assert get_env_value(custom_endpoint_key_env("local-8000")) == "sk-first"
         assert get_env_value(custom_endpoint_key_env("local-8001")) == "sk-second"
 
+    def test_slug_colliding_endpoints_route_only_their_own_credentials(self):
+        """Distinct endpoint ids must never cross-route one another's secrets.
+
+        ``acme-prod`` and ``acme_prod`` are both valid provider ids, but the
+        old slug-only env name collapsed both to ``ACME_PROD``. The second
+        save then made runtime send its key to the first endpoint's URL.
+        """
+        from hermes_cli.config import get_env_value, load_config
+        from hermes_cli.runtime_provider import _get_named_custom_provider
+
+        endpoints = (
+            ("acme-prod", "https://endpoint-a.invalid/v1", "sk-only-a"),
+            ("acme_prod", "https://endpoint-b.invalid/v1", "sk-only-b"),
+        )
+        for endpoint_id, base_url, api_key in endpoints:
+            resp = self.client.post(
+                "/api/providers/custom-endpoints",
+                json={
+                    "id": endpoint_id,
+                    "name": endpoint_id,
+                    "base_url": base_url,
+                    "model": "m",
+                    "api_key": api_key,
+                },
+            )
+            assert resp.status_code == 200
+
+        providers = load_config()["providers"]
+        first_env = providers["acme-prod"]["key_env"]
+        second_env = providers["acme_prod"]["key_env"]
+        assert first_env != second_env
+        assert get_env_value(first_env) == "sk-only-a"
+        assert get_env_value(second_env) == "sk-only-b"
+
+        for endpoint_id, base_url, api_key in endpoints:
+            resolved = _get_named_custom_provider(endpoint_id)
+            assert resolved is not None
+            assert resolved["base_url"] == base_url
+            assert resolved["api_key"] == api_key
+
+    def test_rotating_a_legacy_custom_endpoint_key_migrates_active_mirror(self):
+        """A pre-digest slot stays live until config points at its replacement."""
+        from hermes_cli.config import (
+            _legacy_custom_endpoint_key_env,
+            custom_endpoint_key_env,
+            get_env_value,
+            load_config,
+            save_config,
+            save_env_value,
+        )
+
+        endpoint_id = "acme-prod"
+        legacy_env = _legacy_custom_endpoint_key_env(endpoint_id)
+        save_env_value(legacy_env, "sk-legacy")
+        cfg = load_config()
+        cfg["providers"] = {
+            endpoint_id: {
+                "name": "Acme",
+                "base_url": "https://endpoint-a.invalid/v1",
+                "model": "m",
+                "models": {"m": {}},
+                "key_env": legacy_env,
+            }
+        }
+        cfg["model"] = {
+            "provider": endpoint_id,
+            "default": "m",
+            "base_url": "https://endpoint-a.invalid/v1",
+            "key_env": legacy_env,
+        }
+        save_config(cfg)
+
+        resp = self.client.post(
+            "/api/providers/custom-endpoints",
+            json={
+                "id": endpoint_id,
+                "name": "Acme",
+                "base_url": "https://endpoint-a.invalid/v1",
+                "model": "m",
+                "api_key": "sk-rotated",
+            },
+        )
+        assert resp.status_code == 200
+
+        new_env = custom_endpoint_key_env(endpoint_id)
+        cfg = load_config()
+        assert cfg["providers"][endpoint_id]["key_env"] == new_env
+        assert cfg["model"]["key_env"] == new_env
+        assert get_env_value(new_env) == "sk-rotated"
+        assert not get_env_value(legacy_env)
+
+    def test_legacy_collision_slot_survives_until_last_reference_migrates(self):
+        """Repairing one pre-fix endpoint must not erase its sibling's key."""
+        from hermes_cli.config import (
+            _legacy_custom_endpoint_key_env,
+            custom_endpoint_key_env,
+            get_env_value,
+            load_config,
+            save_config,
+            save_env_value,
+        )
+
+        legacy_env = _legacy_custom_endpoint_key_env("acme-prod")
+        assert legacy_env == _legacy_custom_endpoint_key_env("acme_prod")
+        save_env_value(legacy_env, "sk-shared-last-value")
+        cfg = load_config()
+        cfg["providers"] = {
+            endpoint_id: {
+                "name": endpoint_id,
+                "base_url": base_url,
+                "model": "m",
+                "models": {"m": {}},
+                "key_env": legacy_env,
+            }
+            for endpoint_id, base_url in (
+                ("acme-prod", "https://endpoint-a.invalid/v1"),
+                ("acme_prod", "https://endpoint-b.invalid/v1"),
+            )
+        }
+        save_config(cfg)
+
+        resp = self.client.post(
+            "/api/providers/custom-endpoints",
+            json={
+                "id": "acme-prod",
+                "name": "acme-prod",
+                "base_url": "https://endpoint-a.invalid/v1",
+                "model": "m",
+                "api_key": "sk-only-a",
+            },
+        )
+        assert resp.status_code == 200
+
+        cfg = load_config()
+        assert cfg["providers"]["acme-prod"]["key_env"] == custom_endpoint_key_env(
+            "acme-prod"
+        )
+        assert cfg["providers"]["acme_prod"]["key_env"] == legacy_env
+        assert get_env_value(custom_endpoint_key_env("acme-prod")) == "sk-only-a"
+        assert get_env_value(legacy_env) == "sk-shared-last-value"
+
     def test_custom_endpoint_response_reports_a_key_held_in_env(self):
         """has_api_key must follow key_env, not just a plaintext api_key.