fix(mcp): invalidate OAuth tokens when the configured client changes (port of cline/cline#12983)

tests/tools/test_mcp_oauth.py

@@ -402,6 +402,101 @@ def test_removes_files(self, tmp_path, monkeypatch):
         assert not (d / "myserver.client.json").exists()
 
 
+# ---------------------------------------------------------------------------
+# Client-change token invalidation (port of cline/cline#12983)
+# ---------------------------------------------------------------------------
+
+class TestInvalidateTokensOnClientChange:
+    """Editing oauth.client_id/client_secret must discard tokens minted
+    under the previous client identity (they can only fail with
+    invalid_client), while an unchanged identity preserves them."""
+
+    def _seed(self, tmp_path, monkeypatch, client_id="client-a",
+              client_secret=None):
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        storage = HermesTokenStorage("chg-server")
+        d = tmp_path / "mcp-tokens"
+        d.mkdir(parents=True, exist_ok=True)
+        info = {"client_id": client_id, "redirect_uris": ["http://localhost:1455/callback"]}
+        if client_secret:
+            info["client_secret"] = client_secret
+        (d / "chg-server.client.json").write_text(json.dumps(info))
+        (d / "chg-server.json").write_text(json.dumps({
+            "access_token": "old-token", "token_type": "Bearer",
+        }))
+        (d / "chg-server.meta.json").write_text(json.dumps({
+            "issuer": "https://idp.example",
+            "authorization_endpoint": "https://idp.example/auth",
+            "token_endpoint": "https://idp.example/token",
+        }))
+        return storage, d
+
+    def test_changed_client_id_drops_tokens(self, tmp_path, monkeypatch):
+        from tools.mcp_oauth import _invalidate_tokens_on_client_change
+        storage, d = self._seed(tmp_path, monkeypatch)
+        _invalidate_tokens_on_client_change(storage, "client-b", None)
+        assert not (d / "chg-server.json").exists()
+        assert not (d / "chg-server.meta.json").exists()
+        # client.json is left for _maybe_preregister_client to overwrite
+        assert (d / "chg-server.client.json").exists()
+
+    def test_changed_secret_drops_tokens(self, tmp_path, monkeypatch):
+        from tools.mcp_oauth import _invalidate_tokens_on_client_change
+        storage, d = self._seed(tmp_path, monkeypatch,
+                                client_secret="old-secret")
+        _invalidate_tokens_on_client_change(storage, "client-a", "new-secret")
+        assert not (d / "chg-server.json").exists()
+
+    def test_same_client_preserves_tokens(self, tmp_path, monkeypatch):
+        from tools.mcp_oauth import _invalidate_tokens_on_client_change
+        storage, d = self._seed(tmp_path, monkeypatch,
+                                client_secret="sec")
+        _invalidate_tokens_on_client_change(storage, "client-a", "sec")
+        assert (d / "chg-server.json").exists()
+        assert (d / "chg-server.meta.json").exists()
+
+    def test_no_prior_client_info_is_noop(self, tmp_path, monkeypatch):
+        from tools.mcp_oauth import _invalidate_tokens_on_client_change
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        storage = HermesTokenStorage("fresh-server")
+        d = tmp_path / "mcp-tokens"
+        d.mkdir(parents=True, exist_ok=True)
+        (d / "fresh-server.json").write_text(json.dumps({
+            "access_token": "tok", "token_type": "Bearer",
+        }))
+        _invalidate_tokens_on_client_change(storage, "client-x", None)
+        # No recorded client identity -> nothing provably stale.
+        assert (d / "fresh-server.json").exists()
+
+    def test_preregister_flow_invalidates_end_to_end(self, tmp_path, monkeypatch):
+        """_maybe_preregister_client wires the check in before overwriting
+        client.json — the full config-edit flow drops stale tokens."""
+        pytest.importorskip("mcp")
+        from tools.mcp_oauth import (
+            _build_client_metadata, _maybe_preregister_client,
+        )
+        storage, d = self._seed(tmp_path, monkeypatch)
+        cfg = {"client_id": "client-b", "_resolved_port": 1455}
+        meta = _build_client_metadata(dict(cfg))
+        _maybe_preregister_client(storage, cfg, meta)
+        assert not (d / "chg-server.json").exists(), (
+            "tokens minted under client-a must not survive switch to client-b"
+        )
+        info = json.loads((d / "chg-server.client.json").read_text())
+        assert info["client_id"] == "client-b"
+
+    def test_preregister_flow_same_client_keeps_tokens(self, tmp_path, monkeypatch):
+        pytest.importorskip("mcp")
+        from tools.mcp_oauth import (
+            _build_client_metadata, _maybe_preregister_client,
+        )
+        storage, d = self._seed(tmp_path, monkeypatch)
+        cfg = {"client_id": "client-a", "_resolved_port": 1455}
+        meta = _build_client_metadata(dict(cfg))
+        _maybe_preregister_client(storage, cfg, meta)
+        assert (d / "chg-server.json").exists()
+
+
 # ---------------------------------------------------------------------------
 # Non-interactive / startup-safety tests
 # ---------------------------------------------------------------------------

tools/mcp_oauth.py

@@ -1220,6 +1220,61 @@ def _build_client_metadata(cfg: dict) -> "OAuthClientMetadata":
     return OAuthClientMetadata.model_validate(metadata_kwargs)
 
 
+def _invalidate_tokens_on_client_change(
+    storage: "HermesTokenStorage",
+    new_client_id: str,
+    new_client_secret: str | None,
+) -> None:
+    """Drop cached tokens when the configured OAuth client identity changes.
+
+    Tokens are minted for a specific ``client_id``: after the user edits
+    ``oauth.client_id`` / ``oauth.client_secret`` in config.yaml (or switches
+    from dynamic registration to a pre-registered client), the old tokens are
+    unusable — the token endpoint rejects their refresh with
+    ``invalid_client``. Pre-registered clients are deliberately exempt from
+    the ``invalid_client`` auto-poison path (config-supplied identity can't
+    be healed by re-registration), so without this check the stale tokens
+    wedge every request until the user manually wipes
+    ``~/.hermes/mcp-tokens/<server>.*``.
+
+    Compares the on-disk ``client.json`` identity against the incoming
+    config identity BEFORE the new client info overwrites it. Matching
+    identity is a no-op so live sessions and valid tokens are preserved.
+    Port of cline/cline#12983's "invalidate tokens when OAuth client
+    changes" invariant.
+    """
+    existing = _read_json(storage._client_info_path())
+    if not isinstance(existing, dict):
+        return
+    old_client_id = existing.get("client_id")
+    if not old_client_id:
+        return
+    old_client_secret = existing.get("client_secret") or None
+    if old_client_id == new_client_id and old_client_secret == (
+        new_client_secret or None
+    ):
+        return
+    removed = False
+    for path in (storage._tokens_path(), storage._meta_path()):
+        try:
+            if path.exists():
+                path.unlink()
+                removed = True
+        except OSError as exc:  # non-fatal — stale tokens fail later anyway
+            logger.warning(
+                "MCP OAuth '%s': could not remove stale %s after client "
+                "change: %s", storage._server_name, path.name, exc,
+            )
+    if removed:
+        logger.warning(
+            "MCP OAuth '%s': configured OAuth client changed (client_id %r "
+            "-> %r); discarded tokens minted under the previous client. "
+            "Re-authorize with: hermes mcp login %s",
+            storage._server_name, old_client_id, new_client_id,
+            storage._server_name,
+        )
+
+
 def _maybe_preregister_client(
     storage: "HermesTokenStorage",
     cfg: dict,
@@ -1231,6 +1286,9 @@ def _maybe_preregister_client(
         return
     if OAuthClientInformationFull is None:
         _ensure_sdk_loaded()
+    _invalidate_tokens_on_client_change(
+        storage, client_id, cfg.get("client_secret")
+    )
     port = cfg["_resolved_port"]
     redirect_uri = _resolve_redirect_uri(cfg, port)