Latte
7bb99d94bf
Extend the HTTP transport (from the remote-MCP work) so it's production-ready.
- /healthz: an unauthenticated GET liveness route that returns only
{status, server, version} — no account data, no token state. For Docker and
reverse-proxy probes; on stdio the route is never served.
- Fail-closed bind: refuse to start without authentication, and make the
message distinguish a loopback bind from a public (0.0.0.0) one. The only
override is the explicit FAMILIAR_ALLOW_INSECURE_BIND=true escape hatch, which
permits an unauthenticated bind (with a loud warning).
- FAMILIAR_AUTH_MODE=oauth: run as an OAuth 2.1 resource server. A JWTVerifier
validates external-issuer JWTs (JWKS from FAMILIAR_OAUTH_JWKS_URI, else
discovered from FAMILIAR_OAUTH_ISSUER's OIDC metadata) against issuer +
audience, wrapped in a RemoteAuthProvider that serves
/.well-known/oauth-protected-resource so OAuth-only clients (Claude web) can
discover the authorization server. familiar is a resource server only — no
authorization server, no DCR. bearer mode is unchanged and stays the default.
Tests cover healthz (200, no leak), fail-closed vs escape hatch, and oauth
(valid JWT 200; missing/expired/wrong-audience 401 + WWW-Authenticate; the
protected-resource metadata), plus the new config readers.
394 lines
14 KiB
Python
394 lines
14 KiB
Python
"""Tests for the opt-in HTTP transport, bearer auth, and headless login.
|
|
|
|
The stdio path and every existing tool stay untouched; these only exercise the
|
|
new `familiar serve` surface: the bearer verifier (401 without / with a bad
|
|
token, through with a good one), the start-refusal when no token is set, the
|
|
FAMILIAR_HTTP_* config defaults, the write-default flip in HTTP mode, and the
|
|
headless-login callback parsing.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import importlib
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from conftest import make_jwt
|
|
|
|
from familiar import auth, config, server, writes
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Bearer auth on the HTTP transport
|
|
# --------------------------------------------------------------------------- #
|
|
def _initialize_request() -> dict:
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "initialize",
|
|
"params": {
|
|
"protocolVersion": "2025-06-18",
|
|
"capabilities": {},
|
|
"clientInfo": {"name": "test", "version": "1"},
|
|
},
|
|
}
|
|
|
|
|
|
async def _status_for(app, authorization: str | None) -> int:
|
|
"""POST an MCP initialize to the ASGI app and return the HTTP status."""
|
|
headers = {"Accept": "application/json, text/event-stream"}
|
|
if authorization is not None:
|
|
headers["Authorization"] = authorization
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with app.router.lifespan_context(app):
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
resp = await client.post("/mcp", json=_initialize_request(), headers=headers)
|
|
return resp.status_code, resp.headers.get("www-authenticate")
|
|
|
|
|
|
@pytest.fixture
|
|
def http_server(monkeypatch):
|
|
"""Configure the shared mcp for HTTP with a known token, then reset it."""
|
|
monkeypatch.setenv("FAMILIAR_HTTP_TOKEN", "s3cret-token")
|
|
original_auth = server.mcp.auth
|
|
token = server.configure_http()
|
|
app = server.mcp.http_app(path="/mcp")
|
|
yield token, app
|
|
server.mcp.auth = original_auth
|
|
writes.set_http_context(False)
|
|
|
|
|
|
def test_missing_token_is_401_with_www_authenticate(http_server):
|
|
_, app = http_server
|
|
status, www_auth = asyncio.run(_status_for(app, None))
|
|
assert status == 401
|
|
assert www_auth and www_auth.lower().startswith("bearer")
|
|
|
|
|
|
def test_wrong_token_is_401(http_server):
|
|
_, app = http_server
|
|
status, _ = asyncio.run(_status_for(app, "Bearer nope"))
|
|
assert status == 401
|
|
|
|
|
|
def test_valid_token_passes_handshake(http_server):
|
|
token, app = http_server
|
|
status, _ = asyncio.run(_status_for(app, f"Bearer {token}"))
|
|
assert status == 200
|
|
|
|
|
|
def test_healthz_needs_no_token(http_server):
|
|
_, app = http_server
|
|
|
|
async def _get_health():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with app.router.lifespan_context(app):
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
return await client.get("/healthz")
|
|
|
|
resp = asyncio.run(_get_health())
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Start refusal without a token
|
|
# --------------------------------------------------------------------------- #
|
|
def test_configure_http_refuses_without_token(monkeypatch):
|
|
monkeypatch.delenv("FAMILIAR_HTTP_TOKEN", raising=False)
|
|
original_auth = server.mcp.auth
|
|
try:
|
|
with pytest.raises(SystemExit) as excinfo:
|
|
server.configure_http()
|
|
assert "FAMILIAR_HTTP_TOKEN" in str(excinfo.value)
|
|
finally:
|
|
server.mcp.auth = original_auth
|
|
writes.set_http_context(False)
|
|
|
|
|
|
def test_configure_http_refuses_on_blank_token(monkeypatch):
|
|
monkeypatch.setenv("FAMILIAR_HTTP_TOKEN", " ")
|
|
original_auth = server.mcp.auth
|
|
try:
|
|
with pytest.raises(SystemExit):
|
|
server.configure_http()
|
|
finally:
|
|
server.mcp.auth = original_auth
|
|
writes.set_http_context(False)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Config defaults for the new env vars
|
|
# --------------------------------------------------------------------------- #
|
|
def test_http_config_defaults(monkeypatch):
|
|
for var in ("FAMILIAR_HTTP_HOST", "FAMILIAR_HTTP_PORT", "FAMILIAR_HTTP_PATH"):
|
|
monkeypatch.delenv(var, raising=False)
|
|
try:
|
|
importlib.reload(config)
|
|
assert config.HTTP_HOST == "127.0.0.1"
|
|
assert config.HTTP_PORT == 8080
|
|
assert config.HTTP_PATH == "/mcp"
|
|
finally:
|
|
importlib.reload(config)
|
|
|
|
|
|
def test_http_config_overrides(monkeypatch):
|
|
monkeypatch.setenv("FAMILIAR_HTTP_HOST", "0.0.0.0")
|
|
monkeypatch.setenv("FAMILIAR_HTTP_PORT", "9001")
|
|
monkeypatch.setenv("FAMILIAR_HTTP_PATH", "/eve")
|
|
try:
|
|
importlib.reload(config)
|
|
assert config.HTTP_HOST == "0.0.0.0"
|
|
assert config.HTTP_PORT == 9001
|
|
assert config.HTTP_PATH == "/eve"
|
|
finally:
|
|
for var in ("FAMILIAR_HTTP_HOST", "FAMILIAR_HTTP_PORT", "FAMILIAR_HTTP_PATH"):
|
|
monkeypatch.delenv(var, raising=False)
|
|
importlib.reload(config)
|
|
|
|
|
|
def test_http_token_reads_and_strips_env(monkeypatch):
|
|
monkeypatch.setenv("FAMILIAR_HTTP_TOKEN", " padded ")
|
|
assert config.http_token() == "padded"
|
|
monkeypatch.delenv("FAMILIAR_HTTP_TOKEN", raising=False)
|
|
assert config.http_token() == ""
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Write default in the HTTP context
|
|
# --------------------------------------------------------------------------- #
|
|
def test_http_context_disables_writes_by_default(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("FAMILIAR_DATA_DIR", str(tmp_path))
|
|
monkeypatch.delenv("FAMILIAR_ENABLE_WRITES", raising=False)
|
|
try:
|
|
writes.set_http_context(True)
|
|
assert writes.writes_enabled() is False
|
|
finally:
|
|
writes.set_http_context(False)
|
|
|
|
|
|
def test_http_context_write_env_still_wins(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("FAMILIAR_DATA_DIR", str(tmp_path))
|
|
monkeypatch.setenv("FAMILIAR_ENABLE_WRITES", "1")
|
|
try:
|
|
writes.set_http_context(True)
|
|
assert writes.writes_enabled() is True
|
|
finally:
|
|
writes.set_http_context(False)
|
|
|
|
|
|
def test_stdio_writes_stay_enabled_by_default(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("FAMILIAR_DATA_DIR", str(tmp_path))
|
|
monkeypatch.delenv("FAMILIAR_ENABLE_WRITES", raising=False)
|
|
writes.set_http_context(False)
|
|
assert writes.writes_enabled() is True
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Headless login callback parsing + flow
|
|
# --------------------------------------------------------------------------- #
|
|
def test_parse_callback_full_url():
|
|
code, state = auth.parse_callback("http://localhost:8765/callback?code=abc123&state=xyz")
|
|
assert code == "abc123"
|
|
assert state == "xyz"
|
|
|
|
|
|
def test_parse_callback_bare_code():
|
|
code, state = auth.parse_callback(" justthecode ")
|
|
assert code == "justthecode"
|
|
assert state is None
|
|
|
|
|
|
def test_parse_callback_rejects_empty():
|
|
with pytest.raises(auth.AuthError):
|
|
auth.parse_callback("")
|
|
|
|
|
|
def test_manual_login_exchanges_and_saves(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("FAMILIAR_DATA_DIR", str(tmp_path))
|
|
saved: dict = {}
|
|
|
|
async def fake_exchange(code, verifier, **kwargs):
|
|
assert code == "the-code"
|
|
return {
|
|
"access_token": make_jwt({"sub": "CHARACTER:EVE:42", "name": "Pilot", "scp": []}),
|
|
"refresh_token": "refresh-xyz",
|
|
"expires_in": 1200,
|
|
}
|
|
|
|
monkeypatch.setattr(auth, "exchange_code", fake_exchange)
|
|
monkeypatch.setattr(auth, "_save_character", lambda bundle: saved.update(bundle))
|
|
|
|
bundle = auth.manual_login(
|
|
read_input=lambda _prompt: "http://localhost:8765/callback?code=the-code"
|
|
)
|
|
|
|
assert bundle["character_id"] == 42
|
|
assert bundle["character_name"] == "Pilot"
|
|
assert saved["character_id"] == 42
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Fail-closed bind + insecure escape hatch
|
|
# --------------------------------------------------------------------------- #
|
|
def _reset_auth():
|
|
server.mcp.auth = None
|
|
writes.set_http_context(False)
|
|
|
|
|
|
def test_public_bind_without_token_refused(monkeypatch):
|
|
monkeypatch.delenv("FAMILIAR_HTTP_TOKEN", raising=False)
|
|
monkeypatch.delenv("FAMILIAR_ALLOW_INSECURE_BIND", raising=False)
|
|
monkeypatch.delenv("FAMILIAR_AUTH_MODE", raising=False)
|
|
try:
|
|
with pytest.raises(SystemExit) as excinfo:
|
|
server.configure_http(host="0.0.0.0")
|
|
assert "public interface" in str(excinfo.value)
|
|
finally:
|
|
_reset_auth()
|
|
|
|
|
|
def test_loopback_without_token_still_refused(monkeypatch):
|
|
monkeypatch.delenv("FAMILIAR_HTTP_TOKEN", raising=False)
|
|
monkeypatch.delenv("FAMILIAR_ALLOW_INSECURE_BIND", raising=False)
|
|
monkeypatch.delenv("FAMILIAR_AUTH_MODE", raising=False)
|
|
try:
|
|
with pytest.raises(SystemExit):
|
|
server.configure_http(host="127.0.0.1")
|
|
finally:
|
|
_reset_auth()
|
|
|
|
|
|
def test_insecure_bind_escape_hatch_allows_tokenless(monkeypatch):
|
|
monkeypatch.delenv("FAMILIAR_HTTP_TOKEN", raising=False)
|
|
monkeypatch.setenv("FAMILIAR_ALLOW_INSECURE_BIND", "true")
|
|
monkeypatch.delenv("FAMILIAR_AUTH_MODE", raising=False)
|
|
try:
|
|
result = server.configure_http(host="0.0.0.0")
|
|
assert result == ""
|
|
assert server.mcp.auth is None
|
|
finally:
|
|
_reset_auth()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# OAuth resource-server mode
|
|
# --------------------------------------------------------------------------- #
|
|
_ISS = "https://idp.example.com"
|
|
_RES = "https://familiar.example.com"
|
|
|
|
|
|
@pytest.fixture
|
|
def oauth_server(monkeypatch):
|
|
"""Configure oauth mode with an injected JWTVerifier over a local RSA key."""
|
|
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
|
|
|
|
monkeypatch.setenv("FAMILIAR_AUTH_MODE", "oauth")
|
|
monkeypatch.setenv("FAMILIAR_OAUTH_ISSUER", _ISS)
|
|
monkeypatch.setenv("FAMILIAR_OAUTH_RESOURCE", _RES)
|
|
monkeypatch.setenv("FAMILIAR_OAUTH_AUDIENCE", _RES)
|
|
original = server.mcp.auth
|
|
kp = RSAKeyPair.generate()
|
|
verifier = JWTVerifier(public_key=kp.public_key, issuer=_ISS, audience=_RES)
|
|
server.configure_http(host="0.0.0.0", verifier=verifier)
|
|
app = server.mcp.http_app(path="/mcp")
|
|
yield kp, app
|
|
server.mcp.auth = original
|
|
writes.set_http_context(False)
|
|
|
|
|
|
def test_oauth_valid_jwt_passes(oauth_server):
|
|
kp, app = oauth_server
|
|
token = kp.create_token(issuer=_ISS, audience=_RES, subject="user1")
|
|
status, _ = asyncio.run(_status_for(app, f"Bearer {token}"))
|
|
assert status == 200
|
|
|
|
|
|
def test_oauth_missing_token_401(oauth_server):
|
|
_, app = oauth_server
|
|
status, www_auth = asyncio.run(_status_for(app, None))
|
|
assert status == 401
|
|
assert www_auth and www_auth.lower().startswith("bearer")
|
|
|
|
|
|
def test_oauth_wrong_audience_401(oauth_server):
|
|
kp, app = oauth_server
|
|
token = kp.create_token(issuer=_ISS, audience="https://other.example.com", subject="user1")
|
|
status, _ = asyncio.run(_status_for(app, f"Bearer {token}"))
|
|
assert status == 401
|
|
|
|
|
|
def test_oauth_expired_token_401(oauth_server):
|
|
kp, app = oauth_server
|
|
token = kp.create_token(issuer=_ISS, audience=_RES, subject="user1", expires_in_seconds=-60)
|
|
status, _ = asyncio.run(_status_for(app, f"Bearer {token}"))
|
|
assert status == 401
|
|
|
|
|
|
def test_oauth_protected_resource_metadata(oauth_server):
|
|
_, app = oauth_server
|
|
|
|
async def _get_metadata():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with app.router.lifespan_context(app):
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
return await client.get("/.well-known/oauth-protected-resource/mcp")
|
|
|
|
resp = asyncio.run(_get_metadata())
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["resource"] == f"{_RES}/mcp"
|
|
assert body["authorization_servers"] == [f"{_ISS}/"]
|
|
|
|
|
|
def test_oauth_requires_issuer_and_resource(monkeypatch):
|
|
monkeypatch.setenv("FAMILIAR_AUTH_MODE", "oauth")
|
|
monkeypatch.delenv("FAMILIAR_OAUTH_ISSUER", raising=False)
|
|
monkeypatch.delenv("FAMILIAR_OAUTH_RESOURCE", raising=False)
|
|
try:
|
|
with pytest.raises(SystemExit) as excinfo:
|
|
server.configure_http(host="0.0.0.0")
|
|
assert "FAMILIAR_OAUTH_ISSUER" in str(excinfo.value)
|
|
finally:
|
|
_reset_auth()
|
|
|
|
|
|
def test_resolve_jwks_uri_passthrough():
|
|
assert server._resolve_jwks_uri(_ISS, "https://idp/keys") == "https://idp/keys"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Config readers for the new env vars
|
|
# --------------------------------------------------------------------------- #
|
|
def test_is_loopback():
|
|
assert config.is_loopback("127.0.0.1")
|
|
assert config.is_loopback("localhost")
|
|
assert not config.is_loopback("0.0.0.0")
|
|
assert not config.is_loopback("192.168.1.10")
|
|
|
|
|
|
def test_auth_mode_default_and_values(monkeypatch):
|
|
monkeypatch.delenv("FAMILIAR_AUTH_MODE", raising=False)
|
|
assert config.auth_mode() == "bearer"
|
|
monkeypatch.setenv("FAMILIAR_AUTH_MODE", "OAuth")
|
|
assert config.auth_mode() == "oauth"
|
|
monkeypatch.setenv("FAMILIAR_AUTH_MODE", "nonsense")
|
|
assert config.auth_mode() == "bearer"
|
|
|
|
|
|
def test_allow_insecure_bind(monkeypatch):
|
|
monkeypatch.delenv("FAMILIAR_ALLOW_INSECURE_BIND", raising=False)
|
|
assert config.allow_insecure_bind() is False
|
|
monkeypatch.setenv("FAMILIAR_ALLOW_INSECURE_BIND", "yes")
|
|
assert config.allow_insecure_bind() is True
|
|
|
|
|
|
def test_oauth_audience_defaults_to_resource(monkeypatch):
|
|
monkeypatch.setenv("FAMILIAR_OAUTH_RESOURCE", _RES)
|
|
monkeypatch.delenv("FAMILIAR_OAUTH_AUDIENCE", raising=False)
|
|
assert config.oauth_audience() == _RES
|
|
monkeypatch.setenv("FAMILIAR_OAUTH_AUDIENCE", "custom-aud")
|
|
assert config.oauth_audience() == "custom-aud"
|