mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-21 16:18:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

test(mcp-oauth): cover configurable redirect_uri + fix misleading SSH hint

Browse source 
The PR added a configurable `redirect_uri` (proxy/Funnel callbacks) but
shipped without tests, and the loopback SSH-tunnel hint stayed hardcoded —
actively misleading the exact proxy user the feature targets.

- Extract `_resolve_redirect_uri(cfg, port)` so the client-metadata and
  pre-registration paths derive an identical callback (a mismatch makes the
  authorization server reject the redirect).
- Make `_redirect_handler` redirect_uri-aware: a configured proxy callback
  reaches this machine on its own, so it no longer prints the `ssh -N -L`
  loopback guidance. Wired via `functools.partial` — no new global state.
- Document `redirect_uri` in the config block.
- 14 new tests (red/green TDD): helper resolution + empty-string fallback,
  metadata + pre-registration for configured/default, AnyUrl normalization,
  no-client_id skip, client_secret combo, and both SSH-hint branches.

ruff clean · 81 passed (tests/tools/test_mcp_oauth.py) · ty baseline unchanged

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Florian Burka 2026-06-17 18:12:08 +02:00 • committed by Teknium
parent 6297634d22
commit d0afcb125c
2 changed files with 317 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

275
tests/tools/test_mcp_oauth.py
View file

@ -314,6 +314,48 @@ class TestRedirectHandlerSshHint:
err = capsys.readouterr().err
assert "ssh -N -L" not in err
def test_configured_redirect_uri_shows_proxy_hint_not_tunnel(self, monkeypatch, capsys):
"""With a proxy redirect_uri, the SSH hint must not push the loopback tunnel.
The Funnel/proxy callback reaches this machine on its own, so the
``ssh -N -L`` guidance would be actively misleading.
"""
import tools.mcp_oauth as mco
monkeypatch.setattr(mco, "_oauth_port", 49203)
monkeypatch.setattr(mco, "_is_interactive", lambda: True)
monkeypatch.setenv("SSH_CLIENT", "1.2.3.4 1234 22")
monkeypatch.setattr(mco, "_can_open_browser", lambda: False)
handler = _make_redirect_handler(
49203, redirect_uri="https://oauth.example.ts.net/callback"
)
self._run(handler("https://example.com/auth"))
err = capsys.readouterr().err
assert "https://oauth.example.ts.net/callback" in err
assert "no SSH tunnel needed" in err
assert "ssh -N -L" not in err
assert "127.0.0.1" not in err
def test_configured_redirect_uri_no_hint_when_local(self, monkeypatch, capsys):
"""Off SSH, a configured redirect_uri prints no remote-session hint."""
import tools.mcp_oauth as mco
monkeypatch.setattr(mco, "_oauth_port", 49204)
monkeypatch.setattr(mco, "_is_interactive", lambda: True)
monkeypatch.delenv("SSH_CLIENT", raising=False)
monkeypatch.delenv("SSH_TTY", raising=False)
monkeypatch.setattr(mco, "_can_open_browser", lambda: True)
monkeypatch.setattr("webbrowser.open", lambda url, **kw: True)
handler = _make_redirect_handler(
49204, redirect_uri="https://oauth.example.ts.net/callback"
)
self._run(handler("https://example.com/auth"))
err = capsys.readouterr().err
assert "Remote session detected" not in err
assert "no SSH tunnel needed" not in err
# ---------------------------------------------------------------------------
# Path traversal protection
@ -862,6 +904,106 @@ def test_configure_callback_port_uses_explicit_port():
assert cfg["_resolved_port"] == 54321
_PROXY_REDIRECT = "https://oauth.example.ts.net/callback"
def test_resolve_redirect_uri_prefers_configured_value():
"""An explicit redirect_uri in cfg overrides the localhost default."""
from tools.mcp_oauth import _resolve_redirect_uri
assert _resolve_redirect_uri({"redirect_uri": _PROXY_REDIRECT}, 1234) == _PROXY_REDIRECT
def test_resolve_redirect_uri_falls_back_to_localhost():
"""No redirect_uri → the loopback callback derived from the port."""
from tools.mcp_oauth import _resolve_redirect_uri
assert _resolve_redirect_uri({}, 1234) == "http://127.0.0.1:1234/callback"
def test_resolve_redirect_uri_empty_string_falls_back():
"""An empty redirect_uri is treated as unset (YAML ``redirect_uri:``)."""
from tools.mcp_oauth import _resolve_redirect_uri
assert _resolve_redirect_uri({"redirect_uri": ""}, 5678) == "http://127.0.0.1:5678/callback"
def test_build_client_metadata_uses_configured_redirect_uri():
"""A proxied redirect_uri (e.g. Tailscale Funnel) flows into the metadata.
Without this the redirect_uri is pinned to ``http://127.0.0.1:<port>/callback``,
which a public HTTPS proxy cannot reach.
"""
pytest.importorskip("mcp")
from tools.mcp_oauth import _build_client_metadata, _configure_callback_port
cfg = {"redirect_uri": _PROXY_REDIRECT}
_configure_callback_port(cfg)
md = _build_client_metadata(cfg)
assert [str(u).rstrip("/") for u in md.redirect_uris] == [_PROXY_REDIRECT]
def test_build_client_metadata_redirect_uri_defaults_to_localhost():
"""Without redirect_uri, metadata keeps the loopback callback default."""
pytest.importorskip("mcp")
from tools.mcp_oauth import _build_client_metadata, _configure_callback_port
cfg: dict = {}
port = _configure_callback_port(cfg)
md = _build_client_metadata(cfg)
assert [str(u).rstrip("/") for u in md.redirect_uris] == [
f"http://127.0.0.1:{port}/callback"
]
def test_maybe_preregister_client_persists_configured_redirect_uri(tmp_path, monkeypatch):
"""Pre-registered client info records the configured redirect_uri verbatim.
The redirect_uri on the stored client_info MUST match the one in the
authorization request, or the provider rejects the callback.
"""
pytest.importorskip("mcp")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
from tools.mcp_oauth import (
HermesTokenStorage,
_build_client_metadata,
_configure_callback_port,
_maybe_preregister_client,
)
cfg = {"client_id": "preset-client", "redirect_uri": _PROXY_REDIRECT}
_configure_callback_port(cfg)
storage = HermesTokenStorage("proxy-srv")
_maybe_preregister_client(storage, cfg, _build_client_metadata(cfg))
written = json.loads(storage._client_info_path().read_text())
assert [u.rstrip("/") for u in written["redirect_uris"]] == [_PROXY_REDIRECT]
def test_maybe_preregister_client_redirect_uri_defaults_to_localhost(tmp_path, monkeypatch):
"""Without redirect_uri, pre-registration falls back to the loopback callback."""
pytest.importorskip("mcp")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
from tools.mcp_oauth import (
HermesTokenStorage,
_build_client_metadata,
_configure_callback_port,
_maybe_preregister_client,
)
cfg = {"client_id": "preset-client"}
port = _configure_callback_port(cfg)
storage = HermesTokenStorage("loopback-srv")
_maybe_preregister_client(storage, cfg, _build_client_metadata(cfg))
written = json.loads(storage._client_info_path().read_text())
assert [u.rstrip("/") for u in written["redirect_uris"]] == [
f"http://127.0.0.1:{port}/callback"
]
def test_build_oauth_auth_preserves_server_url_path():
"""server_url with path is forwarded to OAuthClientProvider unmodified.
@ -895,6 +1037,139 @@ def test_build_oauth_auth_preserves_server_url_path():
assert captured["server_url"] == "https://mcp.notion.com/mcp"
def test_build_oauth_auth_wires_configured_redirect_uri_into_handler(monkeypatch, capsys):
"""The configured redirect_uri is bound into the redirect_handler closure so
the remote-session hint stays accurate for proxied callbacks."""
from tools import mcp_oauth
captured: dict = {}
class _FakeProvider:
def __init__(self, **kwargs):
captured.update(kwargs)
with patch.object(mcp_oauth, "_OAUTH_AVAILABLE", True), \
patch.object(mcp_oauth, "OAuthClientProvider", _FakeProvider), \
patch.object(mcp_oauth, "_is_interactive", return_value=True), \
patch.object(mcp_oauth, "_maybe_preregister_client"), \
patch.object(mcp_oauth, "HermesTokenStorage") as mock_storage_cls:
mock_storage_cls.return_value = MagicMock(has_cached_tokens=lambda: True)
build_oauth_auth(
server_name="proxy",
server_url="https://mcp.example.com/mcp",
oauth_config={"redirect_uri": _PROXY_REDIRECT},
)
# Behavior check: on a remote session, the bound handler prints the proxy
# callback hint (not the loopback SSH-tunnel guidance).
monkeypatch.setenv("SSH_CLIENT", "1.2.3.4 1234 22")
monkeypatch.setattr(mcp_oauth, "_is_interactive", lambda: True)
monkeypatch.setattr(mcp_oauth, "_can_open_browser", lambda: False)
asyncio.get_event_loop().run_until_complete(
captured["redirect_handler"]("https://example.com/auth")
)
err = capsys.readouterr().err
assert _PROXY_REDIRECT in err
assert "no SSH tunnel needed" in err
assert "ssh -N -L" not in err
def test_build_oauth_auth_handler_redirect_uri_none_when_unset(monkeypatch, capsys):
"""Without a configured redirect_uri, the bound handler falls back to the
loopback SSH-tunnel hint."""
from tools import mcp_oauth
captured: dict = {}
class _FakeProvider:
def __init__(self, **kwargs):
captured.update(kwargs)
with patch.object(mcp_oauth, "_OAUTH_AVAILABLE", True), \
patch.object(mcp_oauth, "OAuthClientProvider", _FakeProvider), \
patch.object(mcp_oauth, "_is_interactive", return_value=True), \
patch.object(mcp_oauth, "_maybe_preregister_client"), \
patch.object(mcp_oauth, "HermesTokenStorage") as mock_storage_cls:
mock_storage_cls.return_value = MagicMock(has_cached_tokens=lambda: True)
build_oauth_auth(
server_name="loopback",
server_url="https://mcp.example.com/mcp",
oauth_config={"redirect_port": 49299},
)
monkeypatch.setenv("SSH_CLIENT", "1.2.3.4 1234 22")
monkeypatch.setattr(mcp_oauth, "_is_interactive", lambda: True)
monkeypatch.setattr(mcp_oauth, "_can_open_browser", lambda: False)
asyncio.get_event_loop().run_until_complete(
captured["redirect_handler"]("https://example.com/auth")
)
err = capsys.readouterr().err
assert "ssh -N -L" in err
assert "no SSH tunnel needed" not in err
def test_build_client_metadata_redirect_uri_without_path_is_normalized():
"""pydantic AnyUrl appends a trailing slash to a bare-hostname redirect_uri.
Both _build_client_metadata and _maybe_preregister_client run the value
through AnyUrl, so they normalize identically and stay consistent — this
pins that behavior so a future pydantic change is caught.
"""
pytest.importorskip("mcp")
from tools.mcp_oauth import _build_client_metadata, _configure_callback_port
cfg = {"redirect_uri": "https://oauth.example.ts.net"}
_configure_callback_port(cfg)
md = _build_client_metadata(cfg)
assert str(md.redirect_uris[0]) == "https://oauth.example.ts.net/"
def test_maybe_preregister_client_skips_when_no_client_id(tmp_path, monkeypatch):
"""No client_id → pre-registration is a no-op even with a configured redirect_uri."""
pytest.importorskip("mcp")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
from tools.mcp_oauth import (
HermesTokenStorage,
_build_client_metadata,
_configure_callback_port,
_maybe_preregister_client,
)
cfg = {"redirect_uri": _PROXY_REDIRECT} # no client_id
_configure_callback_port(cfg)
storage = HermesTokenStorage("no-client-id-srv")
_maybe_preregister_client(storage, cfg, _build_client_metadata(cfg))
assert not storage._client_info_path().exists()
def test_maybe_preregister_client_redirect_uri_with_secret(tmp_path, monkeypatch):
"""redirect_uri + client_secret: callback stored verbatim, auth method confidential."""
pytest.importorskip("mcp")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
from tools.mcp_oauth import (
HermesTokenStorage,
_build_client_metadata,
_configure_callback_port,
_maybe_preregister_client,
)
cfg = {
"client_id": "my-client",
"client_secret": "shhh",
"redirect_uri": _PROXY_REDIRECT,
}
_configure_callback_port(cfg)
storage = HermesTokenStorage("secret-proxy-srv")
_maybe_preregister_client(storage, cfg, _build_client_metadata(cfg))
written = json.loads(storage._client_info_path().read_text())
assert [u.rstrip("/") for u in written["redirect_uris"]] == [_PROXY_REDIRECT]
assert written["client_secret"] == "shhh"
assert written["token_endpoint_auth_method"] == "client_secret_post"
class TestPasteCallbackReader:
"""_paste_callback_reader parses redirect URLs / query strings from stdin."""

53
tools/mcp_oauth.py
View file

@ -29,6 +29,7 @@ Configuration in config.yaml::
client_secret: "secret" # confidential clients only
scope: "read write" # default: server-provided
redirect_port: 0 # 0 = auto-pick free port
redirect_uri: "https://proxy/callback" # default: loopback callback
client_name: "My Custom Client" # default: "Hermes Agent"
"""
@ -575,12 +576,17 @@ def _make_callback_handler() -> tuple[type, dict]:
# ---------------------------------------------------------------------------
def _make_redirect_handler(port: int):
def _make_redirect_handler(port: int, redirect_uri: str | None = None):
"""Return a redirect handler closure that closes over the given port.
Using a closure instead of reading the module-level ``_oauth_port`` avoids
cross-server state pollution when multiple MCP servers run OAuth
concurrently (fixes #44588).
``redirect_uri`` is the configured proxy callback (e.g. a Tailscale Funnel
URL), or ``None`` for the loopback default. It tailors the remote-session
hint: a proxied callback reaches this machine on its own, so the loopback
SSH-tunnel guidance would be misleading.
"""
async def _redirect_handler(authorization_url: str) -> None:
"""Show the authorization URL to the user.
@ -610,13 +616,23 @@ def _make_redirect_handler(port: int):
)
print(msg, file=sys.stderr)
# On a remote SSH session the OAuth provider redirects to
# http://127.0.0.1:<port>/callback, which reaches the callback server on
# the *remote* machine — not the user's local machine where the browser
# opened. Two ways out: paste the redirect URL back (default fallback,
# offered by _wait_for_callback on interactive TTYs), or set up an SSH
# port forward so the redirect tunnels through.
if port and (os.getenv("SSH_CLIENT") or os.getenv("SSH_TTY")):
on_ssh = bool(os.getenv("SSH_CLIENT") or os.getenv("SSH_TTY"))
if on_ssh and redirect_uri:
# A configured proxy callback (e.g. Tailscale Funnel) forwards the
# redirect to the listener on this machine, so no tunnel/paste is needed.
print(
f" Remote session detected. After you authorize, the provider redirects to\n"
f" {redirect_uri}\n"
f" which forwards to the callback listener on this machine — no SSH tunnel needed.\n",
file=sys.stderr,
)
elif on_ssh and port:
# Loopback default: the provider redirects to
# http://127.0.0.1:<port>/callback, which reaches the callback server on
# the *remote* machine — not the user's local machine where the browser
# opened. Two ways out: paste the redirect URL back (default fallback,
# offered by _wait_for_callback on interactive TTYs), or set up an SSH
# port forward so the redirect tunnels through.
print(
f" Remote session detected. After you authorize, the provider redirects to\n"
f" http://127.0.0.1:{port}/callback\n"
@ -902,6 +918,19 @@ def _configure_callback_port(cfg: dict) -> int:
return port
def _resolve_redirect_uri(cfg: dict, port: int) -> str:
"""Resolve the OAuth callback URL: configured ``redirect_uri`` or loopback.
A configured ``redirect_uri`` lets the callback go through a proxy (e.g. a
Tailscale Funnel exposing a public HTTPS URL that forwards to localhost);
otherwise we default to ``http://127.0.0.1:<port>/callback``. An empty value
is treated as unset. Both the client metadata and any pre-registered client
info must derive the redirect_uri here so they stay identical — a mismatch
makes the authorization server reject the callback.
"""
return cfg.get("redirect_uri") or f"http://127.0.0.1:{port}/callback"
def _build_client_metadata(cfg: dict) -> "OAuthClientMetadata":
"""Build OAuthClientMetadata from the oauth config dict.
@ -915,7 +944,7 @@ def _build_client_metadata(cfg: dict) -> "OAuthClientMetadata":
)
client_name = cfg.get("client_name", "Hermes Agent")
scope = cfg.get("scope")
redirect_uri = cfg.get("redirect_uri") or f"http://127.0.0.1:{port}/callback"
redirect_uri = _resolve_redirect_uri(cfg, port)
metadata_kwargs: dict[str, Any] = {
"client_name": client_name,
@ -942,7 +971,7 @@ def _maybe_preregister_client(
if not client_id:
return
port = cfg["_resolved_port"]
redirect_uri = cfg.get("redirect_uri") or f"http://127.0.0.1:{port}/callback"
redirect_uri = _resolve_redirect_uri(cfg, port)
info_dict: dict[str, Any] = {
"client_id": client_id,
@ -1009,7 +1038,9 @@ def build_oauth_auth(
# Use closure factories to avoid global state pollution (#44588).
resolved_port = cfg.get("_resolved_port", _oauth_port)
redirect_handler = _make_redirect_handler(resolved_port)
redirect_handler = _make_redirect_handler(
resolved_port, redirect_uri=cfg.get("redirect_uri") or None
)
return OAuthClientProvider(
server_url=server_url,