fix(mcp): fail fast for noninteractive oauth without tokens

This commit is contained in:
helix4u 2026-06-14 17:19:48 -06:00 committed by Teknium
parent aca11c227e
commit dcc3216955
6 changed files with 72 additions and 28 deletions

View file

@ -11,6 +11,14 @@ from pathlib import Path
import pytest
def _set_interactive_stdin(monkeypatch, *, is_tty: bool = True) -> None:
from unittest.mock import MagicMock
mock_stdin = MagicMock()
mock_stdin.isatty.return_value = is_tty
monkeypatch.setattr("tools.mcp_oauth.sys.stdin", mock_stdin)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@ -649,6 +657,7 @@ class TestMcpRemoveEvictsManager:
"hermes_cli.mcp_config.get_hermes_home", lambda: tmp_path
)
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_set_interactive_stdin(monkeypatch)
from tools.mcp_oauth_manager import get_manager, reset_manager_for_tests
reset_manager_for_tests()
@ -745,4 +754,3 @@ class TestMcpLogin:
assert "Authenticated — 3 tool(s) available" in out
assert "no OAuth token" not in out

View file

@ -26,6 +26,12 @@ from tools.mcp_oauth import (
)
def _set_interactive_stdin(monkeypatch, *, is_tty: bool = True) -> None:
mock_stdin = MagicMock()
mock_stdin.isatty.return_value = is_tty
monkeypatch.setattr("tools.mcp_oauth.sys.stdin", mock_stdin)
# ---------------------------------------------------------------------------
# HermesTokenStorage
# ---------------------------------------------------------------------------
@ -164,6 +170,7 @@ class TestBuildOAuthAuth:
pytest.skip("MCP SDK auth not available")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_set_interactive_stdin(monkeypatch)
auth = build_oauth_auth("test", "https://example.com/mcp")
assert isinstance(auth, OAuthClientProvider)
@ -180,6 +187,7 @@ class TestBuildOAuthAuth:
pytest.skip("MCP SDK auth not available")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_set_interactive_stdin(monkeypatch)
build_oauth_auth("slack", "https://slack.example.com/mcp", {
"client_id": "my-app-id",
"client_secret": "my-secret",
@ -199,6 +207,7 @@ class TestBuildOAuthAuth:
pytest.skip("MCP SDK auth not available")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_set_interactive_stdin(monkeypatch)
provider = build_oauth_auth("scoped", "https://example.com/mcp", {
"scope": "read write admin",
})
@ -403,6 +412,7 @@ class TestOAuthPortSharing:
pytest.skip("MCP SDK auth not available")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_set_interactive_stdin(monkeypatch)
build_oauth_auth("test-port", "https://example.com/mcp")
assert mod._oauth_port is not None
assert isinstance(mod._oauth_port, int)
@ -479,32 +489,21 @@ class TestWaitForCallbackNoBlocking:
class TestBuildOAuthAuthNonInteractive:
"""build_oauth_auth() in non-interactive mode."""
def test_noninteractive_without_cached_tokens_warns(self, tmp_path, monkeypatch, caplog):
"""Without cached tokens, non-interactive mode logs a clear warning."""
try:
from mcp.client.auth import OAuthClientProvider
except ImportError:
pytest.skip("MCP SDK auth not available")
def test_noninteractive_without_cached_tokens_fails_fast(self, tmp_path, monkeypatch):
"""Without cached tokens, non-interactive mode skips browser auth."""
pytest.importorskip("mcp.client.auth")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
mock_stdin = MagicMock()
mock_stdin.isatty.return_value = False
monkeypatch.setattr("tools.mcp_oauth.sys.stdin", mock_stdin)
import logging
with caplog.at_level(logging.WARNING, logger="tools.mcp_oauth"):
auth = build_oauth_auth("atlassian", "https://mcp.atlassian.com/v1/mcp")
assert auth is not None
assert "no cached tokens found" in caplog.text.lower()
assert "non-interactive" in caplog.text.lower()
with pytest.raises(OAuthNonInteractiveError, match="non-interactive"):
build_oauth_auth("atlassian", "https://mcp.atlassian.com/v1/mcp")
def test_noninteractive_with_cached_tokens_no_warning(self, tmp_path, monkeypatch, caplog):
"""With cached tokens, non-interactive mode logs no 'no cached tokens' warning."""
try:
from mcp.client.auth import OAuthClientProvider
except ImportError:
pytest.skip("MCP SDK auth not available")
pytest.importorskip("mcp.client.auth")
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
mock_stdin = MagicMock()

View file

@ -18,6 +18,14 @@ import pytest
pytest.importorskip("mcp.client.auth.oauth2", reason="MCP SDK 1.26.0+ required")
def _set_interactive_stdin(monkeypatch, *, is_tty: bool = True) -> None:
from unittest.mock import MagicMock
mock_stdin = MagicMock()
mock_stdin.isatty.return_value = is_tty
monkeypatch.setattr("tools.mcp_oauth.sys.stdin", mock_stdin)
@pytest.mark.asyncio
async def test_external_refresh_picked_up_without_restart(tmp_path, monkeypatch):
"""Simulate Cthulhu's cron workflow end-to-end.
@ -160,6 +168,7 @@ async def test_handle_401_returns_false_when_no_provider(tmp_path, monkeypatch):
async def test_invalidate_if_disk_changed_handles_missing_file(tmp_path, monkeypatch):
"""invalidate_if_disk_changed returns False when tokens file doesn't exist."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_set_interactive_stdin(monkeypatch)
from tools.mcp_oauth_manager import MCPOAuthManager, reset_manager_for_tests
reset_manager_for_tests()
@ -181,6 +190,7 @@ async def test_provider_is_reused_across_reconnects(tmp_path, monkeypatch):
first post-reconnect auth flow would spuriously "detect" a change.
"""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_set_interactive_stdin(monkeypatch)
from tools.mcp_oauth_manager import MCPOAuthManager, reset_manager_for_tests
reset_manager_for_tests()

View file

@ -7,6 +7,7 @@ cache. See `tools/mcp_oauth_manager.py` for design rationale.
import json
import os
import time
from unittest.mock import MagicMock
import pytest
@ -16,6 +17,12 @@ pytest.importorskip(
)
def _set_interactive_stdin(monkeypatch, *, is_tty: bool = True) -> None:
mock_stdin = MagicMock()
mock_stdin.isatty.return_value = is_tty
monkeypatch.setattr("tools.mcp_oauth.sys.stdin", mock_stdin)
def test_manager_is_singleton():
"""get_manager() returns the same instance across calls."""
from tools.mcp_oauth_manager import get_manager, reset_manager_for_tests
@ -28,6 +35,7 @@ def test_manager_is_singleton():
def test_manager_get_or_build_provider_caches(tmp_path, monkeypatch):
"""Calling get_or_build_provider twice with same name returns same provider."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_set_interactive_stdin(monkeypatch)
from tools.mcp_oauth_manager import MCPOAuthManager
mgr = MCPOAuthManager()
@ -39,6 +47,7 @@ def test_manager_get_or_build_provider_caches(tmp_path, monkeypatch):
def test_manager_get_or_build_rebuilds_on_url_change(tmp_path, monkeypatch):
"""Changing the URL discards the cached provider."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_set_interactive_stdin(monkeypatch)
from tools.mcp_oauth_manager import MCPOAuthManager
mgr = MCPOAuthManager()
@ -50,6 +59,7 @@ def test_manager_get_or_build_rebuilds_on_url_change(tmp_path, monkeypatch):
def test_manager_remove_evicts_cache(tmp_path, monkeypatch):
"""remove(name) evicts the provider from cache AND deletes disk files."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_set_interactive_stdin(monkeypatch)
from tools.mcp_oauth_manager import MCPOAuthManager
# Pre-seed tokens on disk
@ -131,6 +141,7 @@ def test_manager_builds_hermes_provider_subclass(tmp_path, monkeypatch):
)
reset_manager_for_tests()
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_set_interactive_stdin(monkeypatch)
mgr = MCPOAuthManager()
provider = mgr.get_or_build_provider("srv", "https://example.com/mcp", None)
@ -139,3 +150,17 @@ def test_manager_builds_hermes_provider_subclass(tmp_path, monkeypatch):
assert isinstance(provider, _HERMES_PROVIDER_CLS)
assert provider._hermes_server_name == "srv"
def test_manager_fails_fast_noninteractive_without_cached_tokens(tmp_path, monkeypatch):
"""A daemon without cached MCP OAuth tokens must not enter browser auth."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_set_interactive_stdin(monkeypatch, is_tty=False)
from tools.mcp_oauth import OAuthNonInteractiveError
from tools.mcp_oauth_manager import MCPOAuthManager
mgr = MCPOAuthManager()
with pytest.raises(OAuthNonInteractiveError, match="non-interactive"):
mgr.get_or_build_provider("linear", "https://mcp.linear.app/mcp", None)
assert mgr._entries["linear"].provider is None

View file

@ -754,12 +754,12 @@ def build_oauth_auth(
storage = HermesTokenStorage(server_name)
if not _is_interactive() and not storage.has_cached_tokens():
logger.warning(
"MCP OAuth for '%s': non-interactive environment and no cached tokens "
raise OAuthNonInteractiveError(
"MCP OAuth for "
f"'{server_name}': non-interactive environment and no cached tokens "
"found. The OAuth flow requires browser authorization. Run "
"interactively first to complete the initial authorization, then "
"cached tokens will be reused.",
server_name,
f"`hermes mcp login {server_name}` interactively first to complete "
"initial authorization, then cached tokens will be reused."
)
_configure_callback_port(cfg)

View file

@ -408,6 +408,7 @@ class MCPOAuthManager:
# Local imports avoid circular deps at module import time.
from tools.mcp_oauth import (
HermesTokenStorage,
OAuthNonInteractiveError,
_OAUTH_AVAILABLE,
_build_client_metadata,
_configure_callback_port,
@ -424,11 +425,12 @@ class MCPOAuthManager:
storage = HermesTokenStorage(server_name)
if not _is_interactive() and not storage.has_cached_tokens():
logger.warning(
"MCP OAuth for '%s': non-interactive environment and no "
"cached tokens found. Run interactively first to complete "
"initial authorization.",
server_name,
raise OAuthNonInteractiveError(
"MCP OAuth for "
f"'{server_name}': non-interactive environment and no "
"cached tokens found. Run `hermes mcp login "
f"{server_name}` interactively first to complete initial "
"authorization."
)
_configure_callback_port(cfg)