Kevin Veen-BirkenbachandClaude Opus 5.5
58187e8323
The inbox shows the console output of a command under its "Ausführen" row, kept to the same tail the job list shows. A deny picks its radio channel (post by default, call, newsletter, broadcast, pan-pan, security) and an optional reason; the passkey signs denial, channel and reason, the hub verifies the proof before it stores or sends anything, and the PermissionRequest hook quotes a reason only when it verifies. Agents ask for an auto-approval rule through the MCP tool rule_request; the request shows in the inbox with a dry run over the last seven days and is accepted only with a passkey assertion the hub verifies against the pinned keys and the hub origin, for every rule kind. Requests lapse with the ask lifetime and are capped per agent and in total. The image runs as nobody by default and carries the health check command; compose keeps the timing. 🎯 Fix: 93 % 💥 Break: 4 % 💸 Cost: 98 % ⏱️ Time: 75 % 🛡️ Security: 100 % 🏆 Score: 100 🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 % Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
128 lines
7.9 KiB
Python
128 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
import http.client
|
|
import json
|
|
import unittest
|
|
|
|
from hub_case import HubCase
|
|
from hub_support import PAGE, SCORE, TREE
|
|
from kvb_orchestrator import handler, mcp
|
|
|
|
TOOLS = {"review_request", "review_status", "review_wait", "review_list", "decision_request", "decision_wait", "rule_request", "rule_request_wait", "operator_message"}
|
|
|
|
|
|
class TestHubMcp(HubCase):
|
|
def test_mcp_round_trip(self):
|
|
init = self.rpc("initialize", {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "t", "version": "1"}})
|
|
self.assertEqual(init["result"]["protocolVersion"], "2025-06-18")
|
|
self.assertIn("tools", init["result"]["capabilities"])
|
|
self.assertEqual(init["result"]["serverInfo"]["name"], "kvb-orchestrator")
|
|
self.assertEqual(self.rpc("initialize", {"protocolVersion": "1999-01-01"})["result"]["protocolVersion"], mcp.PROTOCOLS[0])
|
|
status, _, body = self.post_mcp({"jsonrpc": "2.0", "method": "notifications/initialized"}, {})
|
|
self.assertEqual((status, body), (202, b""))
|
|
self.assertEqual(self.rpc("ping", {})["result"], {})
|
|
tools = self.rpc("tools/list", {})["result"]["tools"]
|
|
self.assertEqual({t["name"] for t in tools}, TOOLS)
|
|
for entry in tools:
|
|
with self.subTest(tool=entry["name"]):
|
|
schema = entry["inputSchema"]
|
|
self.assertEqual((schema["type"], schema["additionalProperties"]), ("object", False))
|
|
self.assertEqual(sorted(schema["required"]), sorted(schema["properties"]))
|
|
review_id = self.request("feat: hub", "table")
|
|
self.assertRegex(review_id, r"^[0-9a-f]{16}$")
|
|
state = self.tool("review_status", {"id": review_id})
|
|
self.assertEqual(json.loads(state["content"][0]["text"]), state["structuredContent"])
|
|
self.assertEqual(state["structuredContent"], {"id": review_id, "status": "pending", "tree": TREE, "decided": None})
|
|
listed = self.tool("review_list", {})["structuredContent"]["reviews"]
|
|
self.assertEqual([(r["id"], r["title"], r["tree"]) for r in listed], [(review_id, "feat: hub", TREE)])
|
|
self.assertEqual(self.rpc("resources/list", {})["error"]["code"], -32601)
|
|
self.assertEqual(self.rpc("tools/call", {"name": "review_approve", "arguments": {"id": review_id}})["error"]["code"], -32602)
|
|
status, _, body = self.call("POST", "/mcp", b"{not json", {"Host": self.host, "Content-Type": "application/json"})
|
|
self.assertEqual((status, json.loads(body)["error"]["code"]), (400, -32700))
|
|
|
|
def test_tool_arguments_are_validated(self):
|
|
good = self.arguments("feat: x", "", self.page("bundle.html", PAGE))
|
|
for name, arguments in (
|
|
("review_request", {k: v for k, v in good.items() if k != "notes"}),
|
|
("review_request", {k: v for k, v in good.items() if k != "repo"}),
|
|
("review_request", {**good, "repo": "relative/.git"}),
|
|
("review_request", {**good, "extra": "1"}),
|
|
("review_request", {**good, "tree": "HEAD"}),
|
|
("review_request", {**good, "title": " "}),
|
|
("review_request", {**good, "title": "two\nlines"}),
|
|
("review_request", {**good, "worktree": "relative/path"}),
|
|
("review_request", {**good, "branch": "has space"}),
|
|
("review_request", {**good, "notes": 5}),
|
|
("review_request", {k: v for k, v in good.items() if k != "gate"}),
|
|
("review_request", {k: v for k, v in good.items() if k != "score"}),
|
|
("review_request", {**good, "gate": "maybe"}),
|
|
("review_request", {**good, "score": 5}),
|
|
("review_request", {**good, "score": "x" * 501}),
|
|
("review_status", {"id": "../../etc"}),
|
|
("review_status", {"id": "0123456789abcdef"}),
|
|
("review_wait", {"id": "0123456789abcdef", "timeout_seconds": 1}),
|
|
("review_list", {"extra": 1}),
|
|
):
|
|
with self.subTest(name=name, arguments=arguments):
|
|
self.assertTrue(self.tool(name, arguments)["isError"])
|
|
review_id = self.request("feat: x", "")
|
|
for timeout in (-1, True, "5", None, float("nan"), float("inf")):
|
|
with self.subTest(timeout=timeout):
|
|
self.assertTrue(self.tool("review_wait", {"id": review_id, "timeout_seconds": timeout})["isError"])
|
|
bad = self.rpc("tools/call", {"name": "review_status", "arguments": ["x"]})
|
|
self.assertEqual(bad["error"]["code"], -32602)
|
|
|
|
def test_mcp_rejects_foreign_origins_hosts_and_simple_requests(self):
|
|
ping = {"jsonrpc": "2.0", "id": 1, "method": "ping"}
|
|
self.assertEqual(self.post_mcp(ping, {"Origin": "http://evil.example"})[0], 403)
|
|
self.assertEqual(self.post_mcp(ping, {"Host": "evil.example"})[0], 403)
|
|
self.assertEqual(self.post_mcp(ping, {"Content-Type": "text/plain"})[0], 415)
|
|
self.assertEqual(self.post_mcp(ping, {"Origin": f"http://{self.host}"})[0], 200)
|
|
self.assertEqual(self.get("/mcp")[0], 405)
|
|
|
|
def headers_only(self, path: str, headers: dict[str, str]) -> tuple[int, bytes]:
|
|
connection = http.client.HTTPConnection("127.0.0.1", self.port, timeout=3)
|
|
try:
|
|
connection.putrequest("POST", path, skip_host=True)
|
|
for name, value in {"Content-Length": "1024", **headers}.items():
|
|
connection.putheader(name, value)
|
|
connection.endheaders()
|
|
response = connection.getresponse()
|
|
return response.status, response.read()
|
|
finally:
|
|
connection.close()
|
|
|
|
def test_every_post_guard_answers_before_the_body_arrives(self):
|
|
guard = self.guard(self.host)
|
|
for path, headers, status, text in (
|
|
("/api/asks", {"Host": "evil.example", "X-Kvb-Orchestrator": "1"}, 403, "forbidden host"),
|
|
("/api/asks", {"Host": self.host}, 403, "forbidden: local POST with the X-Kvb-Orchestrator header and no foreign origin only"),
|
|
("/mcp", {"Host": self.host, "Origin": "http://evil.example", "Content-Type": "application/json"}, 403, "foreign origin"),
|
|
("/mcp", {"Host": self.host, "Content-Type": "text/plain"}, 415, "use application/json"),
|
|
("/api/runner/0123456789abcdef/report", guard, 403, "forbidden: the runner calls with the X-Kvb-Orchestrator header and no origin only"),
|
|
("/api/messages", self.client(), 403, "forbidden: same-origin POST with the X-Kvb-Orchestrator header only"),
|
|
("/api/stt", {**guard, "Content-Type": "text/plain"}, 415, "use audio/wav"),
|
|
("/api/state", guard, 404, "not found"),
|
|
("/api/uploads", {**guard, "Content-Length": str(handler.UPLOAD_LIMIT + 1)}, 413, "body too large"),
|
|
("/api/messages", {**guard, "Content-Length": "many"}, 411, "Content-Length required"),
|
|
):
|
|
with self.subTest(path=path, headers=headers):
|
|
self.assertEqual(self.headers_only(path, headers), (status, f"{text}\n".encode()))
|
|
|
|
def test_mcp_requires_gate_score_and_recommended(self):
|
|
tools = {entry["name"]: entry["inputSchema"] for entry in self.rpc("tools/list", {})["result"]["tools"]}
|
|
self.assertEqual(tools["review_request"]["properties"]["gate"]["enum"], ["pass", "fail"])
|
|
self.assertTrue({"gate", "score"} <= set(tools["review_request"]["required"]))
|
|
self.assertIn("recommended", tools["decision_request"]["required"])
|
|
review_id = self.request("feat: gated", "")
|
|
decision_id = self.decision("Wait")
|
|
inbox = {item["id"]: item for item in self.snapshot()["inbox"]}
|
|
self.assertEqual((inbox[review_id]["gate"], inbox[review_id]["score"], inbox[review_id]["green"]), ("pass", SCORE, True))
|
|
self.assertEqual((inbox[decision_id]["recommended"], inbox[decision_id]["green"]), ("Wait", True))
|
|
plain = self.decision("")
|
|
self.assertFalse(next(item for item in self.snapshot()["inbox"] if item["id"] == plain)["green"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|