From 07738f22378a29c1c6aeab0fbb3a9ee9be23fe00 Mon Sep 17 00:00:00 2001 From: Latte Date: Sat, 27 Jun 2026 19:37:37 +0200 Subject: [PATCH] Phase 4: write-action safety & guardrails (#22) Add writes.py: the single gate every write tool must pass. A write happens only when confirm=true AND writes are enabled (off by default); otherwise a dry-run preview is returned. Every executed write is appended to an audit log. New `familiar writes [enable|disable|status]` CLI and a read-only write_status MCP tool. 81 tests, coverage 92%. Closes #22. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/familiar/__main__.py | 28 +++++++++--- src/familiar/server.py | 5 ++- src/familiar/writes.py | 96 ++++++++++++++++++++++++++++++++++++++++ tests/test_writes.py | 65 +++++++++++++++++++++++++++ 4 files changed, 187 insertions(+), 7 deletions(-) create mode 100644 src/familiar/writes.py create mode 100644 tests/test_writes.py diff --git a/src/familiar/__main__.py b/src/familiar/__main__.py index c4a1009..e4a74b8 100644 --- a/src/familiar/__main__.py +++ b/src/familiar/__main__.py @@ -4,17 +4,20 @@ With no arguments it runs the MCP server over stdio (how Claude Desktop / Claude Code / Cowork launch it). The auth subcommands are meant to be run by hand in a terminal: - familiar login # add a character via EVE SSO (run once per character) - familiar status # list logged-in characters - familiar logout # forget ALL characters - familiar logout NAME # forget one character (by name or id) + familiar login # add a character via EVE SSO (run once per character) + familiar status # list logged-in characters + familiar logout # forget ALL characters + familiar logout NAME # forget one character (by name or id) + familiar writes # show whether write actions are enabled + familiar writes enable # allow write actions (mail, waypoints, ...) + familiar writes disable """ from __future__ import annotations import sys -from . import auth +from . import auth, writes def _login() -> int: @@ -48,9 +51,22 @@ def _logout() -> int: return 0 +def _writes() -> int: + sub = sys.argv[2] if len(sys.argv) > 2 else "status" + if sub == "enable": + writes.set_writes(True) + elif sub == "disable": + writes.set_writes(False) + elif sub != "status": + print("Usage: familiar writes [enable|disable|status]") + return 1 + print(f"Writes are {'ENABLED' if writes.writes_enabled() else 'disabled'}.") + return 0 + + def main() -> None: command = sys.argv[1] if len(sys.argv) > 1 else None - handlers = {"login": _login, "status": _status, "logout": _logout} + handlers = {"login": _login, "status": _status, "logout": _logout, "writes": _writes} if command in handlers: raise SystemExit(handlers[command]()) if command in {"-h", "--help", "help"}: diff --git a/src/familiar/server.py b/src/familiar/server.py index 081f762..de3a464 100644 --- a/src/familiar/server.py +++ b/src/familiar/server.py @@ -11,7 +11,7 @@ import datetime as _dt from fastmcp import FastMCP -from . import __version__, account, industry, market, structures, trade, universe +from . import __version__, account, industry, market, structures, trade, universe, writes mcp: FastMCP = FastMCP( name="familiar", @@ -51,3 +51,6 @@ account.register(mcp) industry.register(mcp) trade.register(mcp) structures.register(mcp) + +# Phase 4: write-action safety (the gate every write tool goes through). +writes.register(mcp) diff --git a/src/familiar/writes.py b/src/familiar/writes.py new file mode 100644 index 0000000..c910cb3 --- /dev/null +++ b/src/familiar/writes.py @@ -0,0 +1,96 @@ +"""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 (`familiar writes enable`, default off). + +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_enabled" + + +def writes_enabled() -> bool: + """True if writes are allowed (env override or the on-disk toggle).""" + env = os.environ.get("FAMILIAR_ENABLE_WRITES") + if env is not None: + return env.strip().lower() in {"1", "true", "yes", "on"} + return _flag_file().exists() + + +def set_writes(enabled: bool) -> None: + """Turn the on-disk write toggle on or off.""" + flag = _flag_file() + if enabled: + flag.write_text("enabled\n", encoding="utf-8") + elif flag.exists(): + flag.unlink() + + +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 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": "Enable with `familiar writes enable`; write tools also need confirm=true.", + } diff --git a/tests/test_writes.py b/tests/test_writes.py new file mode 100644 index 0000000..6be3508 --- /dev/null +++ b/tests/test_writes.py @@ -0,0 +1,65 @@ +"""Write-safety guardrails: toggle, dry-run, audit log.""" + +import asyncio +import json + +import pytest + +from familiar import writes + + +@pytest.fixture +def tmp_data(tmp_path, monkeypatch): + monkeypatch.setenv("FAMILIAR_DATA_DIR", str(tmp_path)) + monkeypatch.delenv("FAMILIAR_ENABLE_WRITES", raising=False) + return tmp_path + + +async def _exec(): + return {"ok": True} + + +def test_disabled_by_default(tmp_data): + assert writes.writes_enabled() is False + + +def test_toggle_on_disk(tmp_data): + writes.set_writes(True) + assert writes.writes_enabled() is True + writes.set_writes(False) + assert writes.writes_enabled() is False + + +def test_env_override(tmp_data, monkeypatch): + monkeypatch.setenv("FAMILIAR_ENABLE_WRITES", "true") + assert writes.writes_enabled() is True + monkeypatch.setenv("FAMILIAR_ENABLE_WRITES", "0") + assert writes.writes_enabled() is False + + +def test_dry_run_does_not_execute(tmp_data): + ran = [] + + async def execute(): + ran.append(1) + return "done" + + result = asyncio.run(writes.guard("send_mail", {"to": 1}, confirm=False, execute=execute)) + assert result["dry_run"] is True + assert not ran # never executed + + +def test_confirm_but_disabled_raises(tmp_data): + with pytest.raises(writes.WriteDisabled): + asyncio.run(writes.guard("send_mail", {}, confirm=True, execute=_exec)) + + +def test_confirm_enabled_executes_and_audits(tmp_data): + writes.set_writes(True) + result = asyncio.run(writes.guard("send_mail", {"subject": "hi"}, confirm=True, execute=_exec)) + assert result["executed"] is True + assert result["result"] == {"ok": True} + lines = writes.audit_log_path().read_text(encoding="utf-8").strip().splitlines() + entry = json.loads(lines[-1]) + assert entry["action"] == "send_mail" + assert "ts" in entry