diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e93712c..38a6f90 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -41,6 +41,14 @@ jobs: test -f web-console/vendor/examples/financial-services/policy/allow.cedar python -m py_compile web-console/run.py web-console/webserver.py web-console/policy_variants.py + # Static checks first: they need no ports and fail in under a second. + # test_catalog_definition_drift catches a catalog that no longer matches + # its upstream server, which the gateway treats as a rug pull and denies. + - name: Tests + run: | + pip install pytest + python -m pytest tests/ -q + # Run in order: demo-01 produces workspace/trace-claim.json, which # demo-02 and demo-03 consume. - name: Demo 1 - cMCP in action diff --git a/demo-10-model-gateway/catalog.json b/demo-10-model-gateway/catalog.json index d2294bb..2559066 100644 --- a/demo-10-model-gateway/catalog.json +++ b/demo-10-model-gateway/catalog.json @@ -46,13 +46,25 @@ "output_schema": { "type": "object", "properties": { - "result": { + "model": { + "type": "string" + }, + "region": { + "type": "string" + }, + "cloud": { + "type": "string" + }, + "content": { + "type": "string" + }, + "upstream": { "type": "string" } } } }, - "definition_hash": "sha256:f59f39e06537b04c1acf7fa975b3f8a0367f3a468bf600b70bda38318c9f283c", + "definition_hash": "sha256:a1df74a4a40c99c90cdd565e0093e107aec863efc47a135e1a9940b2d7ff9384", "compliance_domain": "internal", "requires_baa": false, "sensitivity_level": "confidential", diff --git a/demo-10-model-gateway/client.py b/demo-10-model-gateway/client.py index 973a333..6e44c0a 100644 --- a/demo-10-model-gateway/client.py +++ b/demo-10-model-gateway/client.py @@ -7,6 +7,7 @@ """ import json import sys +import urllib.error import urllib.request from openai import OpenAI @@ -74,8 +75,16 @@ def main(): print(BAR) print("\nClosing the session and fetching the signed claim...\n") - with urllib.request.urlopen(f"{GATEWAY}/trust-record", timeout=20) as r: - record = json.loads(r.read()) + try: + with urllib.request.urlopen(f"{GATEWAY}/trust-record", timeout=20) as r: + record = json.loads(r.read()) + except urllib.error.HTTPError as exc: + detail = exc.read().decode(errors="replace")[:300] + print(f" could not fetch the signed claim: HTTP {exc.code} {detail}") + print(" Nothing to sign means no call was allowed. Check model-gateway.log for") + print(" UPSTREAM_CATALOG_DRIFT: catalog.json and model_server.py must describe") + print(" the same tool, or the gateway fail-closes on every call.") + return 1 trace = record.get("trace", {}) gw = record.get("gateway", {}) print(f" policy.bundle_hash {trace.get('policy', {}).get('bundle_hash', '')}") diff --git a/demo-10-model-gateway/model_server.py b/demo-10-model-gateway/model_server.py index 8623f4b..14cfa6f 100644 --- a/demo-10-model-gateway/model_server.py +++ b/demo-10-model-gateway/model_server.py @@ -68,6 +68,10 @@ def _err(id_, code: int, msg: str) -> dict: return {"jsonrpc": "2.0", "id": id_, "error": {"code": code, "message": msg}} +# The catalog's approved_definition is the contract. cMCP hashes description + +# input schema + output schema on both sides and fail-closes on any difference, +# so this dict and catalog.json's approved_definition must stay identical: an +# argument the endpoint sends but this schema omits reads as a rug pull. TOOL = { "name": "model.chat_completion", "description": "Run a chat completion against a model in the catalog", @@ -80,10 +84,21 @@ def _err(id_, code: int, msg: str) -> dict: "model_region": {"type": "string"}, "model_cloud": {"type": "string"}, "data_class": {"type": "string"}, + "contains_identifiers": {"type": "boolean"}, "redacted": {"type": "boolean"}, "redaction_count": {"type": "number"}, }, }, + "outputSchema": { + "type": "object", + "properties": { + "model": {"type": "string"}, + "region": {"type": "string"}, + "cloud": {"type": "string"}, + "content": {"type": "string"}, + "upstream": {"type": "string"}, + }, + }, } diff --git a/demo-10-model-gateway/run.py b/demo-10-model-gateway/run.py index be54aae..c3dfd00 100644 --- a/demo-10-model-gateway/run.py +++ b/demo-10-model-gateway/run.py @@ -93,7 +93,12 @@ def main(): while True: time.sleep(1) else: - subprocess.call([sys.executable, str(HERE / "client.py")], cwd=HERE) + # The client's exit code is this demo's exit code. Discarding it kept + # CI green through a gateway that denied every call. + rc = subprocess.call([sys.executable, str(HERE / "client.py")], cwd=HERE) + if rc != 0: + print(f"client.py exited with code {rc}. See the *.log files.", file=sys.stderr) + return rc except KeyboardInterrupt: pass finally: diff --git a/tests/test_catalog_definition_drift.py b/tests/test_catalog_definition_drift.py new file mode 100644 index 0000000..56949ab --- /dev/null +++ b/tests/test_catalog_definition_drift.py @@ -0,0 +1,94 @@ +"""Static guard against the drift that broke demo 10. + +cMCP hashes description + input schema + output schema on both sides of the +boundary and fail-closes when they differ. A catalog entry that describes an +argument the upstream server does not advertise is a `rug_pull` to the gateway, +so every call comes back 503 and the demo denies everything. Catching that here +costs no ports and no gateway start. +""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + +from cmcp_runtime.catalog.loader import definition_digest + +ROOT = Path(__file__).resolve().parent.parent + + +def _load(name: str, relative: str): + path = ROOT / relative + spec = importlib.util.spec_from_file_location(name, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.path.insert(0, str(path.parent)) + try: + spec.loader.exec_module(module) + finally: + sys.path.pop(0) + return module + + +def _catalog_entry(demo: str, tool_name: str) -> dict: + entries = json.loads((ROOT / demo / "catalog.json").read_text(encoding="utf-8")) + for entry in entries: + if entry["tool_name"] == tool_name: + return entry + pytest.fail(f"{demo}/catalog.json has no entry for {tool_name}") + + +def test_model_gateway_catalog_matches_the_advertised_tool(): + entry = _catalog_entry("demo-10-model-gateway", "model.chat_completion") + approved = entry["approved_definition"] + tool = _load("demo_model_server", "demo-10-model-gateway/model_server.py").TOOL + + assert definition_digest( + approved["description"], + approved.get("input_schema"), + approved.get("output_schema"), + ) == definition_digest( + str(tool.get("description", "")), + tool.get("inputSchema") or {}, + tool.get("outputSchema"), + ), ( + "catalog.json and model_server.py describe different tools. The gateway " + "reads that as a rug pull and denies every call." + ) + + +def test_model_gateway_endpoint_arguments_are_all_in_the_schema(): + """Every argument the endpoint sends must be an approved property. + + This is the specific gap that shipped: the endpoint sent + ``contains_identifiers``, the Cedar policy read it, and neither schema + declared it. + """ + entry = _catalog_entry("demo-10-model-gateway", "model.chat_completion") + properties = set(entry["approved_definition"]["input_schema"]["properties"]) + + source = (ROOT / "demo-10-model-gateway" / "endpoint.py").read_text(encoding="utf-8") + start = source.index(" args = {") + block = source[start:source.index("\n }", start)] + sent = {line.split('"')[1] for line in block.splitlines() if line.strip().startswith('"')} + + assert sent, "could not parse the endpoint's argument block" + assert sent <= properties, f"endpoint sends undeclared arguments: {sorted(sent - properties)}" + + +def test_stored_definition_hashes_are_current(): + """A hand-edited schema with a stale definition_hash fails at catalog load.""" + from cmcp_runtime.catalog.loader import _compute_definition_hash + + for catalog in sorted(ROOT.glob("demo-*/catalog.json")): + for entry in json.loads(catalog.read_text(encoding="utf-8")): + if "definition_hash" not in entry: + continue + computed = _compute_definition_hash(entry["approved_definition"]) + assert entry["definition_hash"] == computed, ( + f"{catalog.parent.name}/{entry['tool_name']}: stale definition_hash" + ) diff --git a/web-console/run.py b/web-console/run.py index 5cfc02e..633ad40 100644 --- a/web-console/run.py +++ b/web-console/run.py @@ -87,6 +87,28 @@ def _wait_for_port(port: int, what: str, process=None, timeout: float = 60.0) -> + _log_tail(what)) +def _assert_port_free(port: int, what: str) -> None: + """Refuse to start if something already owns the port. + + _wait_for_port() returns as soon as *anything* answers, so a gateway left + over from a demo run satisfies it instantly while this console's own + gateway dies on a bind error in cmcp.log. The console then scores every + assessment against the other gateway's policy bundle and still prints + plausible allow/deny lines. Same guard the terminal demos already carry. + """ + try: + with socket.create_connection(("127.0.0.1", port), timeout=1): + pass + except OSError: + return + sys.exit( + f"Port {port} is already in use, so {what} cannot start and this console " + f"would be scored against whatever is already listening. Stop it first " + f"(a cMCP gateway left over from another demo is the usual cause), then " + f"re-run." + ) + + def _ensure_submodule() -> None: if (EXAMPLE / "agent" / "credit_risk_agent.py").exists(): return @@ -129,6 +151,9 @@ def _write_gateway_config() -> None: def main() -> None: _ensure_submodule() + _assert_port_free(8080, "the EU credit-risk MCP server") + _assert_port_free(8443, "the cMCP gateway") + _assert_port_free(int(PORT), "the web console") _write_gateway_config() server_log = open(HERE / "server.log", "w") cmcp_log = open(HERE / "cmcp.log", "w")