b4fbd7930e
database_url, migration_database_url, mcp_auth_token and the mathpix app id/key are now pydantic SecretStr, so they render as '**********' in any repr/log/traceback instead of leaking the plaintext credential. Consumers (db.get_conn, cli.migrate, mcp._require_http_token, mathpix.extract_formulas) call .get_secret_value() at the point of use. Tests updated to the SecretStr type. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
123 lines
4.1 KiB
Python
123 lines
4.1 KiB
Python
"""test_http_auth.py — HTTP transport token-auth enforcement.
|
|
|
|
Tests:
|
|
1. _require_http_token() raises RuntimeError when MCP_AUTH_TOKEN is absent.
|
|
2. _require_http_token() returns the token when it is set.
|
|
3. main() with transport='http' raises without token (refuses to start).
|
|
4. main() with transport='stdio' runs mcp.run (happy path, no token needed).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from pydantic import SecretStr
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _require_http_token
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_require_http_token_raises_without_token() -> None:
|
|
"""_require_http_token() must raise RuntimeError when mcp_auth_token is None."""
|
|
from codex.mcp_server import _require_http_token
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.mcp_auth_token = None
|
|
|
|
with (
|
|
patch("codex.config.get_settings", return_value=mock_settings),
|
|
pytest.raises(RuntimeError, match="MCP_AUTH_TOKEN"),
|
|
):
|
|
_require_http_token()
|
|
|
|
|
|
def test_require_http_token_returns_token_when_set() -> None:
|
|
"""_require_http_token() must return the token string when it is configured."""
|
|
from codex.mcp_server import _require_http_token
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.mcp_auth_token = SecretStr("super-secret-token")
|
|
|
|
with patch("codex.config.get_settings", return_value=mock_settings):
|
|
result = _require_http_token()
|
|
|
|
assert result == "super-secret-token"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# main() — HTTP transport without token → RuntimeError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_main_http_without_token_raises() -> None:
|
|
"""main() with MCP_TRANSPORT=http and no token must raise RuntimeError."""
|
|
from codex.mcp_server import main
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.mcp_transport = "http"
|
|
mock_settings.mcp_auth_token = None
|
|
|
|
with (
|
|
patch("codex.config.get_settings", return_value=mock_settings),
|
|
pytest.raises(RuntimeError, match="MCP_AUTH_TOKEN"),
|
|
):
|
|
main()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# main() — stdio transport runs mcp.run
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_main_stdio_calls_mcp_run() -> None:
|
|
"""main() with stdio transport must call mcp.run(transport='stdio')."""
|
|
from codex import mcp_server
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.mcp_transport = "stdio"
|
|
|
|
with (
|
|
patch("codex.config.get_settings", return_value=mock_settings),
|
|
patch.object(mcp_server.mcp, "run") as mock_run,
|
|
):
|
|
mcp_server.main()
|
|
|
|
mock_run.assert_called_once_with(transport="stdio")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# main() — HTTP transport with valid token attempts uvicorn.run
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_main_http_with_token_calls_uvicorn() -> None:
|
|
"""main() with http transport + valid token must invoke uvicorn.run."""
|
|
from codex import mcp_server
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.mcp_transport = "http"
|
|
mock_settings.mcp_auth_token = SecretStr("test-token")
|
|
mock_settings.mcp_host = "127.0.0.1"
|
|
mock_settings.mcp_port = 8765
|
|
|
|
# Stub the ASGI app returned by streamable_http_app
|
|
mock_app = MagicMock()
|
|
mock_app.add_middleware = MagicMock()
|
|
|
|
with (
|
|
patch("codex.config.get_settings", return_value=mock_settings),
|
|
patch.object(mcp_server.mcp, "streamable_http_app", return_value=mock_app),
|
|
patch("uvicorn.run") as mock_uvicorn,
|
|
):
|
|
mcp_server.main()
|
|
|
|
mock_uvicorn.assert_called_once()
|
|
call_kwargs = mock_uvicorn.call_args
|
|
assert call_kwargs is not None
|
|
# host and port must match settings
|
|
_, kwargs = call_kwargs
|
|
assert kwargs.get("host") == "127.0.0.1"
|
|
assert kwargs.get("port") == 8765
|