Latte
458e31685f
Add a `familiar serve` subcommand that runs the FastMCP streamable-HTTP transport for use as a remote MCP server (e.g. a Mistral custom connector), while stdio stays the default and is completely unchanged. - config: FAMILIAR_HTTP_HOST/PORT/PATH (default 127.0.0.1:8080/mcp) and a live http_token() reader for FAMILIAR_HTTP_TOKEN. - server: configure_http()/serve_http() attach a StaticTokenVerifier so a missing/wrong bearer token gets 401 + WWW-Authenticate: Bearer and a valid one passes the handshake. Refuses to start (SystemExit) without a token. - writes: in HTTP mode the write default flips to disabled unless FAMILIAR_ENABLE_WRITES is explicitly truthy; guard() logic is untouched. - auth: manual_login() + parse_callback() power `login --no-browser`, printing the SSO URL and reading the pasted callback from stdin (no local browser). - bump fastmcp pin to >=3.0.0 for the streamable-HTTP + verifier API.
120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
"""Guardrails for write/command actions.
|
|
|
|
familiar is safe by default: a write only happens when **all** of these hold:
|
|
|
|
1. the caller passes ``confirm=True`` (otherwise it gets a dry-run preview), and
|
|
2. writes are enabled globally (on by default; opt out via `familiar writes disable`).
|
|
|
|
Every executed write is appended to an audit log on disk. This module is the
|
|
single choke point every write tool must go through.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
|
|
from . import config
|
|
|
|
|
|
class WriteDisabled(RuntimeError):
|
|
"""A write was confirmed but writes are globally disabled."""
|
|
|
|
|
|
def _flag_file():
|
|
return config.data_dir() / "writes_disabled"
|
|
|
|
|
|
# Flipped on by `familiar serve` (HTTP transport). It only changes what the
|
|
# *default* is when FAMILIAR_ENABLE_WRITES is unset: a network-reachable,
|
|
# LLM-driven endpoint should not send mail or set waypoints unless the operator
|
|
# opted in explicitly, so in HTTP mode the default is disabled rather than
|
|
# enabled. An explicit env value (either way) still wins.
|
|
_http_context = False
|
|
|
|
|
|
def set_http_context(enabled: bool) -> None:
|
|
"""Mark the process as serving over HTTP so writes default to disabled."""
|
|
global _http_context
|
|
_http_context = enabled
|
|
|
|
|
|
def writes_enabled() -> bool:
|
|
"""True if writes are allowed (env override, HTTP default, or on-disk toggle).
|
|
|
|
Precedence: an explicit ``FAMILIAR_ENABLE_WRITES`` wins. With it unset,
|
|
writes are **disabled by default in HTTP mode** (see :func:`set_http_context`)
|
|
and **enabled by default** otherwise (the local stdio case), unless the user
|
|
opted out with the disable-flag file (`familiar writes disable`).
|
|
"""
|
|
env = os.environ.get("FAMILIAR_ENABLE_WRITES")
|
|
if env is not None:
|
|
return env.strip().lower() in {"1", "true", "yes", "on"}
|
|
if _http_context:
|
|
return False
|
|
return not _flag_file().exists()
|
|
|
|
|
|
def set_writes(enabled: bool) -> None:
|
|
"""Turn the on-disk write toggle on or off (writes are on by default)."""
|
|
flag = _flag_file()
|
|
if enabled:
|
|
if flag.exists():
|
|
flag.unlink()
|
|
else:
|
|
flag.write_text("disabled\n", encoding="utf-8")
|
|
|
|
|
|
def audit_log_path():
|
|
return config.data_dir() / "audit.log"
|
|
|
|
|
|
def _audit(entry: dict[str, Any]) -> None:
|
|
line = json.dumps({"ts": time.time(), **entry})
|
|
with audit_log_path().open("a", encoding="utf-8") as fh:
|
|
fh.write(line + "\n")
|
|
|
|
|
|
async def guard(
|
|
action: str,
|
|
params: dict[str, Any],
|
|
confirm: bool,
|
|
execute: Callable[[], Awaitable[Any]],
|
|
) -> dict[str, Any]:
|
|
"""Run a write through the safety gate.
|
|
|
|
* ``confirm=False`` -> return a dry-run preview, execute nothing.
|
|
* ``confirm=True`` but writes disabled -> raise :class:`WriteDisabled`.
|
|
* ``confirm=True`` and writes enabled -> execute, audit, return the result.
|
|
"""
|
|
if not confirm:
|
|
return {
|
|
"dry_run": True,
|
|
"action": action,
|
|
"params": params,
|
|
"note": "Nothing was changed. Pass confirm=true to execute (writes must be enabled).",
|
|
}
|
|
if not writes_enabled():
|
|
raise WriteDisabled(
|
|
f"Writes are disabled, so '{action}' was not performed. "
|
|
"Run `familiar writes enable` to re-allow write actions."
|
|
)
|
|
result = await execute()
|
|
_audit({"action": action, "params": params})
|
|
return {"executed": True, "action": action, "result": result}
|
|
|
|
|
|
def register(mcp) -> None:
|
|
"""Register a read-only tool reporting whether writes are enabled."""
|
|
|
|
@mcp.tool
|
|
async def write_status() -> dict[str, Any]:
|
|
"""Whether familiar is currently allowed to perform write actions."""
|
|
return {
|
|
"writes_enabled": writes_enabled(),
|
|
"note": "On by default; disable with `familiar writes disable`. Write tools also need confirm=true.",
|
|
}
|