feat(mcp-oauth): accept 'skip' at paste prompt to bypass auth without disabling server (#32069)

When an MCP server triggers OAuth at startup, the user can now type 'skip'
(or 'cancel', 's', 'n', 'no', 'q', 'quit') at the paste prompt + Enter to
exit the flow cleanly and continue agent startup without that server.

Previously the only ways to bypass an unwanted OAuth prompt were:
  - Wait the full 5-minute paste timeout
  - Ctrl+C (also kills the whole reload, may leave half-state)
  - Edit config.yaml to set 'enabled: false' on the server

Skip writes a sentinel to result['error'] which _wait_for_callback maps to
OAuthNonInteractiveError('user_skipped'). mcp_tool already classifies that
as an auth error in _is_auth_error() and the reconnect loop logs it as
'not retrying automatically' — server stays disconnected for the session,
other MCP servers continue normally, no infinite retry burn.

The skip message tells users how to re-auth later ('hermes mcp login') or
disable persistently ('enabled: false'), so they don't have to remember.

14 new tests covering: case-insensitive skip parsing, all 7 skip tokens,
skip not stomping an HTTP-listener win, skip routed to skip path rather
than URL-parse path, sentinel mapped to OAuthNonInteractiveError, prompt
mentions the skip option.
This commit is contained in:
Teknium 2026-05-25 05:37:30 -07:00 committed by GitHub
parent bdf3696705
commit 8191f663dd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 109 additions and 1 deletions

View file

@ -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()

View file

@ -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 <server>` 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: