diff --git a/tests/tools/test_mcp_oauth.py b/tests/tools/test_mcp_oauth.py index 53e0abfd615..b858127cd07 100644 --- a/tests/tools/test_mcp_oauth.py +++ b/tests/tools/test_mcp_oauth.py @@ -754,3 +754,78 @@ class TestWaitForCallbackPasteIntegration: asyncio.run(_wait_for_callback()) err = capsys.readouterr().err assert "paste the redirect URL" not in err + + +class TestPasteCallbackSkipToken: + """User can type `skip` (or similar) at the paste prompt to bail out.""" + + def _empty_result(self): + return {"auth_code": None, "state": None, "error": None} + + @pytest.mark.parametrize("token", ["skip", "SKIP", "Skip", "cancel", "s", "n", "no", "q", "quit"]) + def test_skip_tokens_set_sentinel(self, monkeypatch, token): + from tools.mcp_oauth import _USER_SKIPPED_SENTINEL + result = self._empty_result() + monkeypatch.setattr("sys.stdin", MagicMock(readline=lambda: token + "\n")) + _paste_callback_reader(result) + assert result["error"] == _USER_SKIPPED_SENTINEL + assert result["auth_code"] is None + + def test_skip_message_printed(self, monkeypatch, capsys): + result = self._empty_result() + monkeypatch.setattr("sys.stdin", MagicMock(readline=lambda: "skip\n")) + _paste_callback_reader(result) + err = capsys.readouterr().err + assert "OAuth skipped" in err + assert "hermes mcp login" in err + + def test_skip_does_not_overwrite_http_winner(self, monkeypatch): + """If HTTP listener already wrote a code, `skip` must not stomp it.""" + result = {"auth_code": "from_http", "state": "x", "error": None} + monkeypatch.setattr("sys.stdin", MagicMock(readline=lambda: "skip\n")) + _paste_callback_reader(result) + assert result["auth_code"] == "from_http" + assert result["error"] is None + + def test_skip_token_not_parsed_as_url(self, monkeypatch, capsys): + """`skip` must NOT fall through to URL parsing (which would silently no-op).""" + from tools.mcp_oauth import _USER_SKIPPED_SENTINEL + result = self._empty_result() + monkeypatch.setattr("sys.stdin", MagicMock(readline=lambda: "skip\n")) + _paste_callback_reader(result) + # Must take skip path, not the "did not contain code=" path + assert result["error"] == _USER_SKIPPED_SENTINEL + err = capsys.readouterr().err + assert "did not contain" not in err + + +class TestWaitForCallbackSkipIntegration: + """_wait_for_callback maps the skip sentinel to OAuthNonInteractiveError.""" + + def test_skip_raises_non_interactive_error(self, monkeypatch): + """Skip token must raise OAuthNonInteractiveError (mcp_tool handles as non-fatal).""" + import tools.mcp_oauth as mod + mod._oauth_port = _find_free_port() + monkeypatch.setattr(mod, "_is_interactive", lambda: True) + monkeypatch.setattr("sys.stdin", MagicMock(readline=lambda: "skip\n")) + + async def instant_sleep(_): + pass + with patch.object(mod.asyncio, "sleep", instant_sleep): + with pytest.raises(OAuthNonInteractiveError, match="user_skipped"): + asyncio.run(_wait_for_callback()) + + def test_paste_prompt_mentions_skip(self, monkeypatch, capsys): + """The interactive prompt must tell users about the skip option.""" + import tools.mcp_oauth as mod + mod._oauth_port = _find_free_port() + monkeypatch.setattr(mod, "_is_interactive", lambda: True) + monkeypatch.setattr("sys.stdin", MagicMock(readline=lambda: "skip\n")) + + async def instant_sleep(_): + pass + with patch.object(mod.asyncio, "sleep", instant_sleep): + with pytest.raises(OAuthNonInteractiveError): + asyncio.run(_wait_for_callback()) + err = capsys.readouterr().err + assert "skip" in err.lower() diff --git a/tools/mcp_oauth.py b/tools/mcp_oauth.py index c79d999cd93..832a6f5945f 100644 --- a/tools/mcp_oauth.py +++ b/tools/mcp_oauth.py @@ -94,6 +94,16 @@ class OAuthNonInteractiveError(RuntimeError): _oauth_port: int | None = None +# Skip tokens accepted at the paste prompt — exit OAuth without auth. +_SKIP_TOKENS = frozenset({"skip", "cancel", "s", "n", "no", "q", "quit"}) + +# Sentinel value written to result["error"] when the user skipped via stdin. +# _wait_for_callback maps this to OAuthNonInteractiveError ("user_skipped") +# so the MCP setup path treats it as a non-fatal "continue without this +# server" rather than a hard failure. +_USER_SKIPPED_SENTINEL = "__hermes_user_skipped__" + + # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- @@ -490,7 +500,8 @@ async def _wait_for_callback() -> tuple[str, str | None]: if _is_interactive(): print( "\n Or paste the redirect URL here (or the ``?code=...&state=...`` " - "portion) and press Enter:", + "portion) and press Enter. Type ``skip`` + Enter to continue " + "without this server:", file=sys.stderr, flush=True, ) @@ -511,6 +522,8 @@ async def _wait_for_callback() -> tuple[str, str | None]: finally: server.server_close() + if result["error"] == _USER_SKIPPED_SENTINEL: + raise OAuthNonInteractiveError("user_skipped") if result["error"]: raise RuntimeError(f"OAuth authorization failed: {result['error']}") if result["auth_code"] is None: @@ -529,6 +542,10 @@ def _paste_callback_reader(result: dict) -> None: - Full redirect URL: ``http://127.0.0.1:37949/callback?code=...&state=...`` - The provider's own callback URL: ``https://mcp.example.com/callback?code=...&state=...`` - Just the query string: ``?code=...&state=...`` or ``code=...&state=...`` + - A skip token (``skip``, ``cancel``, ``s``, ``n``, ``no``, ``q``, ``quit``) + — exits the OAuth flow cleanly without auth. Caller raises + :class:`OAuthNonInteractiveError` so MCP connection setup treats this + as a non-fatal "user opted out" and continues without that server. Failures to parse, EOF, or interrupts are swallowed — this is best-effort fallback alongside the HTTP listener, which remains the primary path. @@ -547,6 +564,22 @@ def _paste_callback_reader(result: dict) -> None: if result.get("auth_code") is not None or result.get("error") is not None: return + # Skip token: user explicitly opted out of authorization. Mark the + # result with a sentinel error string that _wait_for_callback maps + # to OAuthNonInteractiveError (already handled by mcp_tool.py as a + # non-fatal "skip this server and continue startup" path). + if line.lower() in _SKIP_TOKENS: + if result.get("auth_code") is not None or result.get("error") is not None: + return + result["error"] = _USER_SKIPPED_SENTINEL + print( + " OAuth skipped. Run `hermes mcp login ` later to " + "authenticate, or set ``enabled: false`` on that server in " + "config.yaml to disable persistently.", + file=sys.stderr, + ) + return + # Strip a leading "?" if user pasted just a query string. query = line if "?" in line: