Hiddenden/familiar
5
0 Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity
Files
dev
familiar/tests/test_serve.py
T
Latte 7bb99d94bf feat(serve): /healthz, fail-closed bind, and OAuth resource-server mode 
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.
2026-07-04 19:25:52 +02:00

394 lines
14 KiB
Python
Raw Permalink Blame History

"""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"
Reference in new issue View Git Blame Copy Permalink