diff --git a/.gitignore b/.gitignore index 0f65165..289c95e 100644 --- a/.gitignore +++ b/.gitignore @@ -18,6 +18,9 @@ artifacts/gate-decisions/*.json # Live mesh telemetry heartbeats (emitted by nodes at runtime, not committed; dir kept via .gitkeep) artifacts/mesh-heartbeats/*.json +# MCP ops-surface receipts (sealed per tool call at runtime; dir kept via .gitkeep) +artifacts/mcp-receipts/*.json + # Porter control-plane build outputs (compiled locally, not committed) artifacts/porter-shim/shim artifacts/cloudshell-hardened-pack/culler/culler diff --git a/Makefile b/Makefile index 8fa51dc..433106a 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ # SourceOS Continuum — lifecycle entry points. # Control-plane targets delegate to Makefile.porter (the rehomed Porter control plane). -.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute +.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute mesh-demo grant commons mcp validate: ## repo hygiene + CapD validity python3 tools/validate.py @@ -48,3 +48,6 @@ promotion-gate: ## rollout gate: require an APPROVE review verdict (fail-closed, rollout: promotion-gate ## promote local → scale-up cluster (hyperswarm), gated on an APPROVE review verdict @echo "[continuum] rollout — promote via caps.infra.cluster-scaleup.hyperswarm (promotion gate passed)" + +mcp: ## run the governed MCP ops surface (stdio; add ADD an agent client, e.g. Claude Code). Guarded tools fail-closed unless CONTINUUM_MCP_ALLOW_GUARDED=1 + python3 tools/mcp_ops_server.py diff --git a/artifacts/mcp-receipts/.gitkeep b/artifacts/mcp-receipts/.gitkeep new file mode 100644 index 0000000..ac30fbf --- /dev/null +++ b/artifacts/mcp-receipts/.gitkeep @@ -0,0 +1 @@ +# Sealed per-tool-call receipts from the MCP ops surface (tools/mcp_ops_server.py) land here at runtime. diff --git a/tools/mcp_ops_server.py b/tools/mcp_ops_server.py new file mode 100644 index 0000000..caf0646 --- /dev/null +++ b/tools/mcp_ops_server.py @@ -0,0 +1,233 @@ +#!/usr/bin/env python3 +"""Governed MCP ops surface for sourceos-continuum — agent-native, fail-closed, hash-sealed. + +A minimal stdio JSON-RPC (MCP) server, no external dependencies, exposing continuum's OWNED +control-plane surface as agent-callable tools so an agent (Claude Code, Cursor, ...) can drive +the platform conversationally. This is superiority-march move #1: it closes the #1 competitive +gap (agent-native MCP ops surface) while keeping the properties the SaaS incumbents lack. + +What makes this different from Qovery/Port/Render MCP surfaces (all SaaS, audit-logged): + - FAIL-CLOSED by default. Read tools run. A `guarded` (mutating/decisioning) tool is REFUSED + unless an explicit policy grant is present (CONTINUUM_MCP_ALLOW_GUARDED) — it never + executes-then-audits. Refusal is the default, not the exception. + - A HASH-SEALED receipt on EVERY tool call — allow or refuse — emitted to the evidence + bundle. A tamper-evident ledger, not a mutable audit log. + - FULLY OPEN: MIT, self-hosted, stdio transport, zero external deps, scale-to-zero (spawned + on demand, exits when the client disconnects). + +Boundary (CONTINUUM_SCOPE.md): this orchestrates + governs continuum's owned surface (CapD +capabilities, lifecycle, evidence, the promotion gate). It consumes — never reimplements — the +review verdict (prophet-platform) and the source/registry (Gitea/zot). +""" +from __future__ import annotations + +import hashlib +import importlib.util +import json +import os +import sys +from datetime import datetime, timezone +from pathlib import Path + +_ROOT = Path(__file__).resolve().parent.parent +PROTOCOL_VERSION = "2025-06-18" +SERVER = {"name": "sourceos-continuum-ops", "version": "0.1.0"} +EVIDENCE_DIR = _ROOT / "artifacts" / "mcp-receipts" +GUARDED_GRANT_ENV = "CONTINUUM_MCP_ALLOW_GUARDED" + + +def _load(mod_name: str, rel: str): + spec = importlib.util.spec_from_file_location(mod_name, _ROOT / rel) + mod = importlib.util.module_from_spec(spec) + sys.modules[mod_name] = mod + spec.loader.exec_module(mod) + return mod + + +def _seal(body: dict) -> str: + return "sha256:" + hashlib.sha256( + json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() + + +def _digest(obj) -> str: + return "sha256:" + hashlib.sha256( + json.dumps(obj, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")).hexdigest() + + +def _emit_receipt(tool: str, arguments: dict, mode: str, decision: str, result) -> dict: + """Seal and persist a receipt for one tool call. Every call is recorded — this is the + ledger that makes the surface auditable-by-construction, not by opt-in logging.""" + receipt = { + "surface": "sourceos-continuum.mcp_ops.v1", + "tool": tool, + "mode": mode, + "decision": decision, # "allow" | "refuse" + "arguments_digest": _digest(arguments), + "result_digest": _digest(result), + "decided_at": datetime.now(timezone.utc).isoformat(), + } + receipt["receipt_digest"] = _seal(receipt) + try: + EVIDENCE_DIR.mkdir(parents=True, exist_ok=True) + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S_%f") + (EVIDENCE_DIR / f"{tool}.{decision}.{stamp}.json").write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n") + except OSError: + pass # a read-only estate must not break the call; the receipt is still returned inline + return receipt + + +# ── tool handlers (pure reads over continuum's owned surface) ───────────────────────── + +def _tool_list_capabilities(_args: dict) -> dict: + caps = [] + capd_dir = _ROOT / "capd" + for f in sorted(capd_dir.glob("*.capd.json")) if capd_dir.is_dir() else []: + try: + data = json.loads(f.read_text()) + caps.append({"file": f.name, "capability_id": data.get("capability_id"), + "kind": data.get("kind"), "status": data.get("status"), + "composes_with": data.get("composes_with", {})}) + except (OSError, json.JSONDecodeError): + continue + return {"capabilities": caps} + + +def _tool_lifecycle_status(_args: dict) -> dict: + return {"lifecycle": [ + {"stage": "onboard", "owned": True}, + {"stage": "develop", "owned": True}, + {"stage": "cloud-native-test", "owned": True}, + {"stage": "rollout", "owned": True, "gate": "promotion_gate (fail-closed on APPROVE verdict)"}, + ], "source": "docs/LIFECYCLE.md"} + + +def _tool_list_evidence(args: dict) -> dict: + limit = int(args.get("limit", 20)) + bundles = [] + for d in ("gate-decisions", "mcp-receipts"): + p = _ROOT / "artifacts" / d + if p.is_dir(): + for f in sorted(p.glob("*.json"), reverse=True)[:limit]: + bundles.append({"bundle": d, "name": f.name}) + return {"evidence": bundles[:limit]} + + +def _tool_promotion_gate(args: dict) -> dict: + """Guarded: run the promotion gate on a consumed review verdict. Emits a gate decision, so + it is policy-gated. Reuses tools/promotion_gate.py (single source; not reimplemented).""" + verdict = args.get("verdict") + if not isinstance(verdict, dict): + return {"error": "verdict (a sealed review receipt object) is required"} + pg = _load("promotion_gate", "tools/promotion_gate.py") + promote, decision = pg.gate(verdict, _ROOT / "artifacts" / "gate-decisions") + return {"promotion": decision["promotion"], "decision": decision} + + +TOOLS = { + "continuum_list_capabilities": { + "mode": "read", + "description": "List the platform's CapD capability contracts (id, kind, status, composition).", + "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}, + "handler": _tool_list_capabilities, + }, + "continuum_lifecycle_status": { + "mode": "read", + "description": "The continuum lifecycle stages (onboard -> develop -> cloud-native-test -> rollout) and their gates.", + "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}, + "handler": _tool_lifecycle_status, + }, + "continuum_list_evidence": { + "mode": "read", + "description": "List the most recent sealed evidence bundles (gate decisions + MCP receipts).", + "inputSchema": {"type": "object", "properties": {"limit": {"type": "integer"}}, "additionalProperties": False}, + "handler": _tool_list_evidence, + }, + "continuum_promotion_gate": { + "mode": "guarded", + "description": "Run the fail-closed rollout promotion gate over a consumed review verdict. GUARDED: refused unless policy-granted.", + "inputSchema": {"type": "object", "properties": {"verdict": {"type": "object"}}, + "required": ["verdict"], "additionalProperties": False}, + "handler": _tool_promotion_gate, + }, +} + + +def _guarded_allowed() -> bool: + return os.environ.get(GUARDED_GRANT_ENV, "") not in ("", "0", "false", "no") + + +def call_tool(name: str, arguments: dict) -> dict: + """Dispatch one tool call through the governance gate, sealing a receipt either way. + Returns an MCP tools/call result ({content, isError}).""" + arguments = arguments or {} + tool = TOOLS.get(name) + if tool is None: + result = {"error": f"unknown tool {name!r}"} + _emit_receipt(name, arguments, "unknown", "refuse", result) + return {"content": [{"type": "text", "text": json.dumps(result)}], "isError": True} + + mode = tool["mode"] + if mode == "guarded" and not _guarded_allowed(): + result = {"refused": True, + "reason": f"guarded tool {name!r} is fail-closed; set {GUARDED_GRANT_ENV}=1 " + f"(an explicit, audited policy grant) to permit it"} + _emit_receipt(name, arguments, mode, "refuse", result) + return {"content": [{"type": "text", "text": json.dumps(result)}], "isError": True} + + try: + result = tool["handler"](arguments) + except Exception as exc: # a tool fault must not crash the server + result = {"error": f"{type(exc).__name__}: {exc}"} + _emit_receipt(name, arguments, mode, "error", result) + return {"content": [{"type": "text", "text": json.dumps(result)}], "isError": True} + + receipt = _emit_receipt(name, arguments, mode, "allow", result) + result["_receipt"] = receipt["receipt_digest"] # the sealed evidence for this call + return {"content": [{"type": "text", "text": json.dumps(result, indent=2, sort_keys=True)}]} + + +def handle(request: dict) -> dict | None: + """Handle one JSON-RPC request. Returns a response dict, or None for notifications.""" + method = request.get("method") + rid = request.get("id") + if method == "initialize": + return {"jsonrpc": "2.0", "id": rid, "result": { + "protocolVersion": PROTOCOL_VERSION, + "capabilities": {"tools": {}}, + "serverInfo": SERVER, + "instructions": "Governed continuum ops surface. Read tools run; guarded tools are fail-closed. Every call is sealed.", + }} + if method in ("notifications/initialized", "notifications/cancelled"): + return None + if method == "ping": + return {"jsonrpc": "2.0", "id": rid, "result": {}} + if method == "tools/list": + return {"jsonrpc": "2.0", "id": rid, "result": {"tools": [ + {"name": n, "description": t["description"], "inputSchema": t["inputSchema"], + "annotations": {"readOnlyHint": t["mode"] == "read"}} + for n, t in TOOLS.items()]}} + if method == "tools/call": + params = request.get("params") or {} + return {"jsonrpc": "2.0", "id": rid, "result": call_tool(params.get("name"), params.get("arguments"))} + return {"jsonrpc": "2.0", "id": rid, + "error": {"code": -32601, "message": f"method not found: {method}"}} + + +def main() -> int: + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + request = json.loads(line) + except json.JSONDecodeError: + continue + response = handle(request) + if response is not None: + sys.stdout.write(json.dumps(response) + "\n") + sys.stdout.flush() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/test_mcp_ops_server.py b/tools/test_mcp_ops_server.py new file mode 100644 index 0000000..8951eca --- /dev/null +++ b/tools/test_mcp_ops_server.py @@ -0,0 +1,105 @@ +"""Coverage for tools/mcp_ops_server.py — the governed MCP ops surface. + +Drives the real MCP JSON-RPC methods (initialize / tools/list / tools/call) and proves the +governance model: read tools run and seal a receipt; a guarded tool is refused fail-closed +without a policy grant and runs with one; every call — allow or refuse — lands a tamper-evident +receipt in the evidence bundle. +""" +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] + + +def _load(name: str, rel: str): + spec = importlib.util.spec_from_file_location(name, ROOT / rel) + mod = importlib.util.module_from_spec(spec) + sys.modules[name] = mod + spec.loader.exec_module(mod) + return mod + + +srv = _load("mcp_ops_server", "tools/mcp_ops_server.py") + + +@pytest.fixture(autouse=True) +def _isolate(tmp_path, monkeypatch): + monkeypatch.setattr(srv, "EVIDENCE_DIR", tmp_path / "mcp-receipts") + monkeypatch.delenv(srv.GUARDED_GRANT_ENV, raising=False) + + +def _call(name, arguments=None): + return srv.handle({"jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": {"name": name, "arguments": arguments or {}}}) + + +def test_initialize_advertises_the_server(): + r = srv.handle({"jsonrpc": "2.0", "id": 1, "method": "initialize"}) + assert r["result"]["protocolVersion"] == srv.PROTOCOL_VERSION + assert r["result"]["serverInfo"]["name"] == "sourceos-continuum-ops" + assert "tools" in r["result"]["capabilities"] + + +def test_notifications_get_no_response(): + assert srv.handle({"jsonrpc": "2.0", "method": "notifications/initialized"}) is None + + +def test_tools_list_flags_read_vs_guarded(): + r = srv.handle({"jsonrpc": "2.0", "id": 1, "method": "tools/list"}) + by = {t["name"]: t for t in r["result"]["tools"]} + assert set(by) == {"continuum_list_capabilities", "continuum_lifecycle_status", + "continuum_list_evidence", "continuum_promotion_gate"} + assert by["continuum_list_capabilities"]["annotations"]["readOnlyHint"] is True + assert by["continuum_promotion_gate"]["annotations"]["readOnlyHint"] is False + + +def test_read_tool_runs_and_seals_a_receipt(): + r = _call("continuum_list_capabilities") + assert "isError" not in r["result"] + payload = json.loads(r["result"]["content"][0]["text"]) + assert "capabilities" in payload and payload["_receipt"].startswith("sha256:") + files = list(srv.EVIDENCE_DIR.glob("*.json")) + assert len(files) == 1 + rec = json.loads(files[0].read_text()) + assert rec["decision"] == "allow" + body = {k: v for k, v in rec.items() if k != "receipt_digest"} + assert srv._seal(body) == rec["receipt_digest"] # tamper-evident + + +def test_guarded_tool_is_fail_closed_without_a_grant(): + r = _call("continuum_promotion_gate", {"verdict": {}}) + assert r["result"]["isError"] is True + payload = json.loads(r["result"]["content"][0]["text"]) + assert payload["refused"] is True and "fail-closed" in payload["reason"] + # the refusal is itself sealed + refusals = list(srv.EVIDENCE_DIR.glob("*refuse*.json")) + assert len(refusals) == 1 and json.loads(refusals[0].read_text())["decision"] == "refuse" + + +def test_guarded_tool_runs_with_an_explicit_grant(monkeypatch): + monkeypatch.setenv(srv.GUARDED_GRANT_ENV, "1") + pg = _load("promotion_gate", "tools/promotion_gate.py") + verdict = {"tool": "prophet-platform.review_gate.v1", + "reviewed": {"idempotency_key": "engine@0.4.45"}, "verdict": "APPROVE", "checks": []} + verdict["review_digest"] = pg._recompute_seal(verdict) + r = _call("continuum_promotion_gate", {"verdict": verdict}) + assert "isError" not in r["result"], r["result"] + payload = json.loads(r["result"]["content"][0]["text"]) + assert payload["promotion"] == "allow" + + +def test_unknown_tool_is_refused_and_sealed(): + r = _call("does_not_exist") + assert r["result"]["isError"] is True + assert list(srv.EVIDENCE_DIR.glob("*.json")) # even the unknown-tool refusal is recorded + + +def test_unknown_method_returns_jsonrpc_error(): + r = srv.handle({"jsonrpc": "2.0", "id": 1, "method": "bogus/method"}) + assert r["error"]["code"] == -32601 diff --git a/tools/validate.py b/tools/validate.py index af0ffa4..a8b17fe 100644 --- a/tools/validate.py +++ b/tools/validate.py @@ -28,6 +28,7 @@ "tools/mesh_telemetry.py", "tools/mcp_a2a_grant.py", "tools/commons.py", + "tools/mcp_ops_server.py", ] CAPD_KEYS = ("capability_id", "kind", "status", "links", "composes_with", "policy") # Every CapD in capd/ must carry the core keys and parse — not just the flagship control-plane one.