feat(serve): HTTP transport + hardened Docker image for secure public serving

Add an opt-in streamable-HTTP transport alongside the default stdio, so
familiar can be reached from a home lab over HTTPS behind a TLS reverse proxy.

- config.http_settings(): env-driven transport/host/port/path/token settings.
- serve.run(): dispatch stdio vs http; fail closed on a non-loopback bind
  without a bearer token (FAMILIAR_ALLOW_INSECURE_BIND escape hatch).
- server.build_http_auth(): FastMCP StaticTokenVerifier bearer gate; add an
  unauthenticated /healthz route for container/proxy probes.
- `familiar serve` CLI alias; clean exit on InsecureBind.
- docker/Dockerfile (multi-stage, non-root) + hardened docker-compose.yml
  (read-only rootfs, cap_drop ALL, no-new-privileges, Traefik TLS labels),
  .dockerignore, .env.example. Token store mounts at /data; writes off by default.
- .gitea/workflows/docker.yml: build + smoke-test, opt-in image publish.
- docs/deployment.md + README/CLAUDE.md notes; tests for settings, fail-closed
  validation, and the bearer verifier.
This commit is contained in:
Latte committed 2026-07-03 19:55:20 +02:00
1 parent d65e9be5ff
commit f4d2dcb1d8
13 files changed
+727 -4

No files matched your search

+162
View File
@@ -0,0 +1,162 @@
"""Tests for transport selection, HTTP settings, and the bearer-auth gate."""
from __future__ import annotations
import asyncio
import pytest
from familiar import config, serve, server
_HTTP_ENV = [
"FAMILIAR_TRANSPORT",
"FAMILIAR_HTTP_HOST",
"FAMILIAR_HTTP_PORT",
"FAMILIAR_HTTP_PATH",
"FAMILIAR_HTTP_AUTH_TOKEN",
"FAMILIAR_ALLOW_INSECURE_BIND",
]
@pytest.fixture
def clean_env(monkeypatch):
"""Strip all transport-related env vars so defaults are exercised."""
for name in _HTTP_ENV:
monkeypatch.delenv(name, raising=False)
return monkeypatch
# --------------------------------------------------------------------------- #
# config.http_settings
# --------------------------------------------------------------------------- #
def test_http_settings_defaults(clean_env):
s = config.http_settings()
assert s.transport == "stdio"
assert s.host == "127.0.0.1"
assert s.port == 8000
assert s.path == "/mcp"
assert s.auth_token is None
assert s.allow_insecure_bind is False
assert s.is_loopback is True
def test_http_settings_overrides(clean_env):
clean_env.setenv("FAMILIAR_TRANSPORT", "HTTP") # case-insensitive
clean_env.setenv("FAMILIAR_HTTP_HOST", "0.0.0.0")
clean_env.setenv("FAMILIAR_HTTP_PORT", "9001")
clean_env.setenv("FAMILIAR_HTTP_PATH", "/eve")
clean_env.setenv("FAMILIAR_HTTP_AUTH_TOKEN", " secret ")
clean_env.setenv("FAMILIAR_ALLOW_INSECURE_BIND", "yes")
s = config.http_settings()
assert s.transport == "http"
assert s.host == "0.0.0.0"
assert s.port == 9001
assert s.path == "/eve"
assert s.auth_token == "secret" # trimmed
assert s.allow_insecure_bind is True
assert s.is_loopback is False
# --------------------------------------------------------------------------- #
# serve.validate — fail-closed on public binds
# --------------------------------------------------------------------------- #
def _settings(**kw) -> config.HttpSettings:
base = dict(
transport="http",
host="0.0.0.0",
port=8000,
path="/mcp",
auth_token=None,
allow_insecure_bind=False,
)
base.update(kw)
return config.HttpSettings(**base)
def test_validate_stdio_is_noop():
serve.validate(_settings(transport="stdio", host="0.0.0.0")) # no raise
def test_validate_loopback_needs_no_token():
serve.validate(_settings(host="127.0.0.1")) # no raise
def test_validate_public_without_token_raises():
with pytest.raises(serve.InsecureBind):
serve.validate(_settings(host="0.0.0.0"))
def test_validate_public_with_token_ok():
serve.validate(_settings(host="0.0.0.0", auth_token="secret")) # no raise
def test_validate_public_with_insecure_optout_ok():
serve.validate(_settings(host="0.0.0.0", allow_insecure_bind=True)) # no raise
# --------------------------------------------------------------------------- #
# server.build_http_auth — bearer token verifier
# --------------------------------------------------------------------------- #
def test_build_http_auth_none_without_token():
assert server.build_http_auth(None) is None
assert server.build_http_auth("") is None
def test_build_http_auth_verifies_matching_token():
verifier = server.build_http_auth("s3cret")
assert verifier is not None
good = asyncio.run(verifier.verify_token("s3cret"))
bad = asyncio.run(verifier.verify_token("wrong"))
assert good is not None
assert bad is None
# --------------------------------------------------------------------------- #
# serve.run — transport dispatch
# --------------------------------------------------------------------------- #
def test_run_stdio(clean_env, monkeypatch):
calls = {}
monkeypatch.setattr(server.mcp, "run", lambda *a, **k: calls.setdefault("args", (a, k)))
serve.run()
assert calls["args"] == ((), {}) # plain stdio run
def test_run_http_passes_transport_kwargs(clean_env, monkeypatch):
clean_env.setenv("FAMILIAR_TRANSPORT", "http")
clean_env.setenv("FAMILIAR_HTTP_AUTH_TOKEN", "secret")
clean_env.setenv("FAMILIAR_HTTP_PORT", "9100")
captured = {}
monkeypatch.setattr(server.mcp, "run", lambda *a, **k: captured.update(k))
serve.run()
assert captured["transport"] == "http"
assert captured["host"] == "127.0.0.1"
assert captured["port"] == 9100
assert captured["path"] == "/mcp"
def test_run_http_public_without_token_refuses(clean_env, monkeypatch):
clean_env.setenv("FAMILIAR_TRANSPORT", "http")
clean_env.setenv("FAMILIAR_HTTP_HOST", "0.0.0.0")
monkeypatch.setattr(server.mcp, "run", lambda *a, **k: pytest.fail("must not run"))
with pytest.raises(serve.InsecureBind):
serve.run()
def test_run_unknown_transport_exits(clean_env, monkeypatch):
clean_env.setenv("FAMILIAR_TRANSPORT", "carrier-pigeon")
monkeypatch.setattr(server.mcp, "run", lambda *a, **k: pytest.fail("must not run"))
with pytest.raises(SystemExit):
serve.run()
# --------------------------------------------------------------------------- #
# /healthz custom route
# --------------------------------------------------------------------------- #
def test_healthz_route():
resp = asyncio.run(server.healthz(None))
assert resp.status_code == 200
import json
body = json.loads(bytes(resp.body))
assert body["status"] == "ok"
assert body["server"] == "familiar"