fix(mcp): bind stored OAuth tokens to their server URL and clean on remove

hermes_cli/web_routers/mcp.py

@@ -152,6 +152,18 @@ def _run():
     removed = await asyncio.to_thread(_run)
     if not removed:
         raise HTTPException(status_code=404, detail=f"Server '{name}' not found")
+    try:
+        # Same rationale as the TUI mcp.servers.remove path: a removed
+        # server's OAuth tokens (incl. the refresh token) must not survive
+        # on disk (#90703). Best-effort — removal already succeeded.
+        from tools.mcp_oauth import remove_oauth_tokens
+
+        remove_oauth_tokens(name)
+    except Exception as cleanup_err:
+        _log.warning(
+            "Removed MCP server '%s' but its OAuth tokens could not be "
+            "cleaned up: %s", name, cleanup_err,
+        )
     return {"ok": True}
 
 

tests/tools/test_mcp_oauth_cold_load_expiry.py

@@ -476,3 +476,119 @@ def patched(*args, **kwargs):
     assert calls == [], (
         f"Pre-flight must not fire when no tokens are stored, but got {calls}"
     )
+
+
+class TestServerUrlBinding:
+    """#90703: tokens are bound to the server URL they were minted for."""
+
+    def _write_tokens(self, tmp_path, payload):
+        from tools.mcp_oauth import _get_token_dir
+
+        token_dir = _get_token_dir()
+        token_dir.mkdir(parents=True, exist_ok=True)
+        (token_dir / "srv.json").write_text(json.dumps(payload))
+
+    def test_mismatched_url_refuses_tokens(self, tmp_path, monkeypatch):
+        """A token minted for endpoint A must not be served for endpoint B."""
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        from tools.mcp_oauth import HermesTokenStorage
+
+        self._write_tokens(
+            tmp_path,
+            {
+                "access_token": "a",
+                "token_type": "Bearer",
+                "expires_in": 3600,
+                "refresh_token": "r",
+                "hermes_server_url": "https://old.example.com/mcp",
+            },
+        )
+
+        storage = HermesTokenStorage("srv", server_url="https://new.example.com/mcp")
+        assert asyncio.run(storage.get_tokens()) is None
+
+    def test_matching_url_loads_tokens(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        from tools.mcp_oauth import HermesTokenStorage
+
+        self._write_tokens(
+            tmp_path,
+            {
+                "access_token": "a",
+                "token_type": "Bearer",
+                "expires_in": 3600,
+                "refresh_token": "r",
+                "hermes_server_url": "https://same.example.com/mcp",
+            },
+        )
+
+        storage = HermesTokenStorage(
+            "srv", server_url="https://same.example.com/mcp/"
+        )
+        reloaded = asyncio.run(storage.get_tokens())
+        assert reloaded is not None
+        assert reloaded.access_token == "a"
+
+    def test_legacy_file_without_url_lazily_stamps_it(self, tmp_path, monkeypatch):
+        """Pre-binding files keep working (no forced re-login) and are
+        stamped with the current URL so the *next* read is protected."""
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        from tools.mcp_oauth import HermesTokenStorage, _get_token_dir
+
+        self._write_tokens(
+            tmp_path,
+            {
+                "access_token": "a",
+                "token_type": "Bearer",
+                "expires_in": 3600,
+                "refresh_token": "r",
+            },
+        )
+
+        storage = HermesTokenStorage("srv", server_url="https://now.example.com/mcp")
+        reloaded = asyncio.run(storage.get_tokens())
+        assert reloaded is not None  # legacy passes through
+
+        on_disk = json.loads((_get_token_dir() / "srv.json").read_text())
+        assert on_disk.get("hermes_server_url") == "https://now.example.com/mcp"
+
+    def test_set_tokens_records_url(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        from mcp.shared.auth import OAuthToken
+        from tools.mcp_oauth import HermesTokenStorage, _get_token_dir
+
+        storage = HermesTokenStorage("srv", server_url="https://svc.example.com/mcp")
+        asyncio.run(
+            storage.set_tokens(
+                OAuthToken(
+                    access_token="a",
+                    token_type="Bearer",
+                    expires_in=3600,
+                    refresh_token="r",
+                )
+            )
+        )
+
+        on_disk = json.loads((_get_token_dir() / "srv.json").read_text())
+        assert on_disk.get("hermes_server_url") == "https://svc.example.com/mcp"
+
+    def test_unbound_storage_ignores_url_field(self, tmp_path, monkeypatch):
+        """Storage constructed without a URL (CLI login flows) still loads
+        bound files — the binding only filters when the reader knows its
+        endpoint."""
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        from tools.mcp_oauth import HermesTokenStorage
+
+        self._write_tokens(
+            tmp_path,
+            {
+                "access_token": "a",
+                "token_type": "Bearer",
+                "expires_in": 3600,
+                "refresh_token": "r",
+                "hermes_server_url": "https://some.example.com/mcp",
+            },
+        )
+
+        storage = HermesTokenStorage("srv")
+        assert asyncio.run(storage.get_tokens()) is not None

tests/tui_gateway/test_mcp_profile_rpcs.py

@@ -246,3 +246,31 @@ def test_default_profile_add_when_profile_omitted(hermes_root):
     assert "rootsvc" not in _read_yaml(root / "profiles" / "work" / "config.yaml").get(
         "mcp_servers", {}
     )
+
+
+def test_remove_also_deletes_stored_oauth_tokens(hermes_root):
+    """#90703: mcp.servers.remove must not leave the removed server's OAuth
+    tokens (incl. refresh token) on disk — a server re-added under the same
+    name would silently resume the old OAuth session."""
+    import json as _json
+
+    root = hermes_root
+    _result(
+        _call(
+            "mcp.servers.add",
+            {"profile": "work", "name": "temp", "config": {"command": "temp-bin"}},
+        )
+    )
+    # Simulate a stored token file in the PROFILE's home — the remove runs
+    # under the profile's HERMES_HOME override, so that is where the
+    # cleanup must land.
+    tokens_dir = root / "profiles" / "work" / "mcp-tokens"
+    tokens_dir.mkdir(parents=True, exist_ok=True)
+    (tokens_dir / "temp.json").write_text(
+        _json.dumps({"access_token": "a", "refresh_token": "r"})
+    )
+
+    resp = _result(_call("mcp.servers.remove", {"profile": "work", "name": "temp"}))
+    assert resp["removed"] is True
+
+    assert not (tokens_dir / "temp.json").exists()

tools/mcp_oauth.py

@@ -464,9 +464,21 @@ class HermesTokenStorage:
         HERMES_HOME/mcp-tokens/<server_name>.cimd-off      -- CIMD refused here
     """
 
-    def __init__(self, server_name: str, *, hermes_home: str | Path | None = None):
+    def __init__(
+        self,
+        server_name: str,
+        *,
+        hermes_home: str | Path | None = None,
+        server_url: str | None = None,
+    ):
         self._server_name = _safe_filename(server_name)
         self._hermes_home = Path(hermes_home) if hermes_home is not None else None
+        # Optional endpoint binding (#90703): when the caller knows the
+        # server's configured URL, tokens minted for a DIFFERENT endpoint
+        # (server renamed, url edited, or another server whose name
+        # sanitizes to the same filename) are refused instead of silently
+        # sent to the wrong authorization server.
+        self._server_url = (str(server_url).strip().rstrip("/") or None) if server_url else None
 
     def _tokens_path(self) -> Path:
         return _get_token_dir(self._hermes_home) / f"{self._server_name}.json"
@@ -488,6 +500,28 @@ async def get_tokens(self) -> "OAuthToken | None":
             return None
         if OAuthToken is None and not _ensure_sdk_loaded():
             return None
+        # Endpoint binding (#90703): refuse tokens minted for a different
+        # server URL instead of sending them to the wrong endpoint. Legacy
+        # files predate the binding field — pass them through and lazily
+        # stamp the current URL so the next read (or a later url edit) is
+        # protected without forcing every existing user to re-login.
+        stored_url = data.pop("hermes_server_url", None)
+        if self._server_url is not None:
+            if stored_url is None:
+                try:
+                    _write_json(self._tokens_path(), {
+                        **data, "hermes_server_url": self._server_url,
+                    })
+                except Exception:
+                    pass
+            elif str(stored_url).strip().rstrip("/") != self._server_url:
+                logger.warning(
+                    "OAuth tokens at %s were minted for %s, not %s -- "
+                    "ignoring (server url changed or name collision); "
+                    "re-run the OAuth flow for this server",
+                    self._tokens_path(), stored_url, self._server_url,
+                )
+                return None
         # Hermes records an absolute wall-clock ``expires_at`` alongside the
         # SDK's serialized token (see ``set_tokens``). On read we rewrite
         # ``expires_in`` to the remaining seconds so the SDK's downstream
@@ -540,6 +574,10 @@ async def set_tokens(self, tokens: "OAuthToken") -> None:
                 # Mock tokens or unusual shapes: skip the expires_at write
                 # rather than fail persistence.
                 pass
+        if self._server_url is not None:
+            # Endpoint binding (#90703): get_tokens refuses this file if it
+            # is later read for a different server URL.
+            payload["hermes_server_url"] = self._server_url
         _write_json(self._tokens_path(), payload)
         logger.debug("OAuth tokens saved for %s", self._server_name)
 
@@ -1910,7 +1948,7 @@ def build_oauth_auth(
     apply_oauth_provider_defaults(
         cfg, server_name=server_name, server_url=server_url
     )
-    storage = HermesTokenStorage(server_name)
+    storage = HermesTokenStorage(server_name, server_url=server_url)
 
     if not _is_interactive() and not storage.has_cached_tokens():
         raise OAuthNonInteractiveError(

tools/mcp_oauth_manager.py

@@ -667,7 +667,7 @@ def _build_provider(
         apply_oauth_provider_defaults(
             cfg, server_name=server_name, server_url=entry.server_url
         )
-        storage = HermesTokenStorage(server_name)
+        storage = HermesTokenStorage(server_name, server_url=entry.server_url)
 
         from tools.mcp_dashboard_oauth import get_dashboard_oauth_flow
 

tui_gateway/methods_tools.py

@@ -2259,6 +2259,19 @@ def _(rid, params: dict) -> dict:
         removed = _remove_mcp_server(name)
         if not removed:
             return _err(rid, 4064, f"server '{name}' not found")
+        try:
+            # A removed server's stored OAuth tokens (incl. the refresh
+            # token) must not survive on disk — otherwise a server re-added
+            # under the same name silently resumes the old OAuth session
+            # (#90703). Best-effort: removal already succeeded.
+            from tools.mcp_oauth import remove_oauth_tokens
+
+            remove_oauth_tokens(name)
+        except Exception as cleanup_err:
+            logger.warning(
+                "Removed MCP server '%s' but its OAuth tokens could not be "
+                "cleaned up: %s", name, cleanup_err,
+            )
         return _ok(rid, {"ok": True, "removed": True})
     except Exception as e:
         return _err(rid, 5024, str(e))