From db7e65e072d9764a2d7203c9b3a330cfaf428f52 Mon Sep 17 00:00:00 2001 From: arpan Date: Fri, 4 Sep 2026 00:22:29 +0530 Subject: [PATCH 1/2] item 6c: the gateway in the execution path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SPEC-v0.2 §6.1, §6.3, §6.5-§6.8, §6.10. Third of four stages for item 6: acceptance tests T19, T21, T22, T23, T25, and the end-to-end halves of T20 and T24 that 6b could only do as pure functions. The shape worth noticing is how §6.8 reaches the kernel. It is not a second outcome model: the executor translates the table into v0.1 §5.5's own vocabulary — return for COMMITTED, raise NotExecuted for FAILED, raise anything else for AMBIGUOUS — and Control then applies the rules it has always applied. The gateway does not get its own way of deciding what happened, and a lost response over HTTP blocks a retry for exactly the reason a lost response in-process does. §6.7's canonical forwarding is the other half of that: the request the gateway sends is built from the action's canonical arguments, never copied from the received body, so what was hashed, reserved and recorded is byte-for-byte what the upstream receives. A gateway relaying the original bytes would be binding an approval to one document and executing another. A fresh connection per intercepted call, because "the connection was never established" is the only claim in §6.8 that asserts non-execution and a pooled connection closed while idle fails on write, which is indistinguishable from a request that arrived. Two resolutions worth reading: An HTTP 401 is classified before the body is parsed. A peer may answer 401 with any payload it likes, including a well-formed JSON-RPC result; the status is the load-bearing part, because the resource server validated the token before dispatch and nothing reached the tool. §6.10 says a denied request must not be re-asked until it expires. The spec leaves open what happens when no approval exists and the effect key would refuse anyway; the fail-closed reading is to give that refusal now rather than trouble a human with a decision that could not be acted on. Marked `# SPEC:`. It only ever refuses more. CI installs the gateway extra. T30 still proves in a subprocess that `import ctrlrun` touches neither httpx nor ctrlrun.gateway, installed or not. --- .github/workflows/ci.yml | 6 +- pyproject.toml | 6 + src/ctrlrun/cli/main.py | 80 ++++ src/ctrlrun/gateway/__init__.py | 22 +- src/ctrlrun/gateway/server.py | 753 ++++++++++++++++++++++++++++++++ tests/test_demo.py | 3 +- tests/test_gateway_server.py | 637 +++++++++++++++++++++++++++ 7 files changed, 1498 insertions(+), 9 deletions(-) create mode 100644 src/ctrlrun/gateway/server.py create mode 100644 tests/test_gateway_server.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c3eccdf8..1379f1d4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -22,7 +22,9 @@ jobs: - name: Install run: | python -m pip install --upgrade pip - pip install -e ".[dev]" + # The gateway's tests need its extra; T30 still proves `import ctrlrun` does not + # touch it, in a subprocess, whether or not it happens to be installed here. + pip install -e ".[dev,gateway]" - name: pytest run: pytest @@ -62,7 +64,7 @@ jobs: run: | mkdir -p /tmp/sdist && tar xzf dist/*.tar.gz -C /tmp/sdist --strip-components=1 cd /tmp/sdist - pip install ".[dev]" + pip install ".[dev,gateway]" pytest -q - name: The quick start works from the wheel diff --git a/pyproject.toml b/pyproject.toml index d30ce5da..61c84e12 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -86,6 +86,12 @@ select = ["E", "F", "W", "I", "N", "UP", "B", "ANN", "SIM", "RUF"] # SPEC-v0.1 §5.3 freezes `commit_effect(effect_key, action_id, result: Any)`. An executor's # return value has no narrower type, and the store only serializes it. "src/ctrlrun/state.py" = ["ANN401"] +# The gateway's subject matter is untyped JSON off a socket: a JSON-RPC `params`, a tool's +# result, an HTTP client's response object behind a lazy import. Narrowing those to a lie +# would be worse than saying `Any` and validating at the boundary, which is what mcp.py +# does. N818: the error names follow this codebase's convention (DuplicateEffect, +# AmbiguousEffect), not the suffix rule. +"src/ctrlrun/gateway/*.py" = ["ANN401", "N818"] # Type hints are a src/ constraint; tests stay cheap to write. N802 lets acceptance # tests carry the SPEC §7 identifier verbatim (test_T7_...). "tests/*" = ["ANN", "N802"] diff --git a/src/ctrlrun/cli/main.py b/src/ctrlrun/cli/main.py index fc259f5d..f3f4ed5d 100644 --- a/src/ctrlrun/cli/main.py +++ b/src/ctrlrun/cli/main.py @@ -442,6 +442,86 @@ def _approval_dict(record: ApprovalRecord) -> dict[str, Any]: } +@main.command() +@click.option("--upstream", required=True, help="The MCP server this gateway fronts.") +@click.option("--alias", required=True, help="Names the upstream in 'mcp..'.") +@click.option("--listen", default="127.0.0.1:8900", show_default=True, help="HOST:PORT.") +@click.option("--path", default="/mcp", show_default=True, help="The MCP endpoint path.") +@click.option("--principal", default=None, help="A fixed agent name, for one tenant.") +@click.option("--principal-header", default=None, help="Take the agent from this header.") +@click.option( + "--principal-from-client-info", + is_flag=True, + help="Take it from clientInfo, which the protocol does not verify. Removed in v0.3.", +) +@click.option("--user-header", default=None, help="Take principal.user from this header.") +@click.option("--environment", default="production", show_default=True) +@click.option("--upstream-timeout", type=float, default=30.0, show_default=True) +@click.option("--max-body-bytes", type=int, default=1024 * 1024, show_default=True) +@click.option("--allow-origin", "allow_origins", multiple=True, help="Repeatable.") +@click.option("--allow-remote", is_flag=True, help="Permit a non-loopback --listen.") +def gateway( + upstream: str, + alias: str, + listen: str, + path: str, + principal: str | None, + principal_header: str | None, + principal_from_client_info: bool, + user_header: str | None, + environment: str, + upstream_timeout: float, + max_body_bytes: int, + allow_origins: tuple[str, ...], + allow_remote: bool, +) -> None: + """Front an MCP server, applying this directory's policy to every tools/call.""" + host, _, port = listen.rpartition(":") + try: + from ..gateway import serve + + _announce_actions_without_an_effect() + serve( + upstream=upstream, + alias=alias, + host=host or "127.0.0.1", + port=int(port), + path=path, + principal=principal, + principal_header=principal_header, + principal_from_client_info=principal_from_client_info, + user_header=user_header, + environment=environment, + upstream_timeout=upstream_timeout, + max_body_bytes=max_body_bytes, + allow_origins=tuple(allow_origins), + allow_remote=allow_remote, + ) + except CTRLRunError as exc: + raise _fail(exc) from exc + except KeyboardInterrupt: # pragma: no cover - an operator pressing ctrl-c + click.echo("") + + +def _announce_actions_without_an_effect() -> None: + """SPEC-v0.2 §3.2 — print the actions with no `effect:` at startup. + + A write with no effect key is exactly the configuration this product exists to prevent, + and it should be visible on the line that starts the process rather than discovered in a + receipt three weeks later. + """ + from ..policy import Policy + + policy = Policy.from_file() + keyless = sorted(name for name in policy.actions if policy.effect_template(name) is None) + if not keyless: + return + click.echo(f"{len(keyless)} action(s) have no effect: template and get no reservation:") + for name in keyless: + click.echo(f" {name}") + click.echo("That is right for a read, and wrong for anything that changes the world.") + + def _receipt_line(receipt: Receipt) -> str: return ( f"{iso_timestamp(receipt.finished_at)} {receipt.receipt_id} {receipt.action} " diff --git a/src/ctrlrun/gateway/__init__.py b/src/ctrlrun/gateway/__init__.py index 9e4887c4..43f1f022 100644 --- a/src/ctrlrun/gateway/__init__.py +++ b/src/ctrlrun/gateway/__init__.py @@ -14,7 +14,7 @@ import importlib from types import ModuleType -from typing import Final, NoReturn +from typing import Any, Final from ..errors import MissingDependency @@ -36,14 +36,24 @@ def http_client() -> ModuleType: raise MissingDependency(_HTTP_CLIENT, _EXTRA) from exc -def serve(*, upstream: str, alias: str, **options: object) -> NoReturn: +def serve(*, upstream: str, alias: str, **options: Any) -> None: """Run a gateway in front of one upstream MCP server (SPEC-v0.2 §6.1). The dependency check happens first and unconditionally, so an operator who has not installed the extra learns it from one clear line rather than from a stack trace. + + Blocks until interrupted. `Control.from_file()` finds the policy and the store, so a + gateway and a decorator-based worker sharing a policy share reservations (§6.1). """ http_client() - raise NotImplementedError( - "the gateway's request handling arrives with build-list items 6b-6d; " - "this release ships the StateStore half it stands on (SPEC-v0.2 §6.9.4, §6.10)" - ) + from ..control import Control + from .server import Gateway, GatewayConfig, httpx_forwarder, serve_forever + + config = GatewayConfig(upstream=upstream, alias=alias, **options) + control = Control.from_file() + forwarder = httpx_forwarder(config) + try: + serve_forever(Gateway(config, control, forwarder)) + finally: + forwarder.close() + control.store.close() diff --git a/src/ctrlrun/gateway/server.py b/src/ctrlrun/gateway/server.py new file mode 100644 index 00000000..a8186aab --- /dev/null +++ b/src/ctrlrun/gateway/server.py @@ -0,0 +1,753 @@ +"""The gateway's HTTP server. Build-list item 6c; SPEC-v0.2 §6.1, §6.3, §6.5-§6.8, §6.10. + +An MCP client points here instead of at the tool server. `tools/call` becomes an Action - +decided, reserved, executed and recorded exactly as `@protect` does it — and every other +method is relayed unchanged. No agent changes. + +The shape worth noticing is how §6.8 reaches the kernel. It is not a second outcome model: +the executor translates the table into v0.1 §5.5's own vocabulary — return for `COMMITTED`, +raise `NotExecuted` for `FAILED`, raise anything else for `AMBIGUOUS` — and `Control` then +applies the rules it has always applied. The gateway does not get its own way of deciding +what happened. +""" + +from __future__ import annotations + +import json +import logging +import threading +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Final, Protocol, TypeAlias + +from ..action import Action, Principal +from ..control import Control +from ..effect import EffectState, resolve_effect_key, resolve_resource +from ..errors import ( + ActionDenied, + AmbiguousEffect, + ApprovalMismatch, + ApprovalRequired, + CTRLRunError, + DuplicateEffect, + EffectKeyError, + InvalidArgument, + NotExecuted, +) +from ..receipt import Receipt +from .mcp import DEFAULT_MAX_BODY_BYTES, ParsedRequest, Refusal, parse_request +from .outcome import ( + GatewayOutcome, + Observed, + Transport, + UpstreamError, + UpstreamResult, + UpstreamStatus, + classify, +) + +_LOG = logging.getLogger("ctrlrun.gateway") + +#: A JSON-RPC id is a string, a number or null, and the gateway echoes back whatever it was +#: given rather than normalizing it. +JsonRpcId: TypeAlias = str | int | float | None + +#: SPEC-v0.2 §6.1 — loopback unless an operator says otherwise, per the transport's own +#: "when running locally, servers SHOULD bind only to localhost". +DEFAULT_LISTEN: Final = ("127.0.0.1", 8900) +DEFAULT_PATH: Final = "/mcp" + +#: Reserved for the webhook approval endpoint (§7.2). The MCP path may not overlap it. +CTRLRUN_PREFIX: Final = "/ctrlrun/" + +DEFAULT_UPSTREAM_TIMEOUT: Final = 30.0 +DEFAULT_APPROVAL_TIMEOUT: Final = 900.0 + +#: §6.6 — the alias boundary in an action name must be unambiguous even though tool names +#: may contain dots, so the alias may not. +ALIAS_PATTERN: Final = r"^[a-z0-9][a-z0-9_-]*$" + +#: SPEC-v0.2 §6.10, frozen in §11. +DENIED: Final = (-41001, "ctrlrun.denied", 403) +APPROVAL_REQUIRED: Final = (-41002, "ctrlrun.approval_required", 403) +APPROVAL_DENIED: Final = (-41003, "ctrlrun.approval_denied", 403) +DUPLICATE_EFFECT: Final = (-41004, "ctrlrun.duplicate_effect", 409) +AMBIGUOUS_EFFECT: Final = (-41005, "ctrlrun.ambiguous_effect", 409) +BLOCKED: Final = (-41006, "ctrlrun.blocked", 409) +NO_PRINCIPAL: Final = (-41007, "ctrlrun.no_principal", 403) +UNREPRESENTABLE: Final = (-41008, "ctrlrun.unrepresentable_argument", 400) + +#: §6.8 — the `_meta` key every intercepted response carries, so a client is not left +#: guessing what CTRLRun recorded. `com.ctrlrun/` is a legal prefix under the revision's +#: key-naming rules, and `_meta` on a result is not validated against a tool's outputSchema. +RECEIPT_META_KEY: Final = "com.ctrlrun/receipt" + +#: §6.6 — the decision reason recorded when an argument cannot be represented. +UNREPRESENTABLE_REASON: Final = "unrepresentable_argument" + + +class UpstreamAmbiguous(CTRLRunError): + """What the executor raises where §6.8 says the outcome is unknown. + + Any exception that is not `NotExecuted` is an `AMBIGUOUS` outcome to `Control` (v0.1 + §5.5), so this needs no special handling anywhere — it exists to carry the synthesized + response back to the client, not to change what the kernel does with it. + """ + + def __init__(self, outcome: GatewayOutcome) -> None: + super().__init__(str(outcome.token)) + self.outcome = outcome + + +@dataclass(frozen=True) +class GatewayConfig: + """Everything `ctrlrun gateway` was started with (SPEC-v0.2 §11).""" + + upstream: str + alias: str + host: str = DEFAULT_LISTEN[0] + port: int = DEFAULT_LISTEN[1] + path: str = DEFAULT_PATH + principal: str | None = None + principal_header: str | None = None + principal_from_client_info: bool = False + user_header: str | None = None + environment: str = "production" + wait_approvals: bool = False + approval_timeout: float = DEFAULT_APPROVAL_TIMEOUT + upstream_timeout: float = DEFAULT_UPSTREAM_TIMEOUT + max_body_bytes: int = DEFAULT_MAX_BODY_BYTES + allow_origins: tuple[str, ...] = () + allow_remote: bool = False + + def __post_init__(self) -> None: + import re + + if not re.match(ALIAS_PATTERN, self.alias): + raise InvalidArgument( + f"--alias {self.alias!r} must match {ALIAS_PATTERN} — no dots, so the alias " + "boundary in 'mcp..' stays unambiguous (SPEC-v0.2 §6.6)" + ) + if not self.path.startswith("/") or self.path.rstrip("/").startswith( + CTRLRUN_PREFIX.rstrip("/") + ): + raise InvalidArgument( + f"--path {self.path!r} must start with '/' and must not overlap " + f"{CTRLRUN_PREFIX!r}, which is reserved for the approvals endpoint" + ) + sources = [ + self.principal is not None, + self.principal_header is not None, + self.principal_from_client_info, + ] + if sum(sources) != 1: + raise InvalidArgument( + "exactly one of --principal, --principal-header or " + "--principal-from-client-info is required; there is no default, because an " + "Action cannot exist without a principal (SPEC-v0.2 §6.5)" + ) + if self.host not in ("127.0.0.1", "localhost", "::1") and not self.allow_remote: + raise InvalidArgument( + f"--listen {self.host} is not loopback; pass --allow-remote to mean it" + ) + + +@dataclass +class _Response: + """What goes back to the client.""" + + status: int + body: bytes = b"" + headers: Mapping[str, str] = field(default_factory=dict) + + +def json_rpc_error(rpc_id: Any, code: int, token: str, message: str, **data: Any) -> dict[str, Any]: + """One JSON-RPC error object in §6.10's shape. + + A refusal by CTRLRun is not an outcome of the tool; it is the statement that the tool did + not run. `isError: true` would be indistinguishable from the tool's own failure, and it + reaches the model as text — which is not where a policy denial belongs. + """ + return { + "jsonrpc": "2.0", + "id": rpc_id, + "error": {"code": code, "message": message, "data": {"error": token, **data}}, + } + + +class Forwarder(Protocol): + """How a gateway reaches its upstream. A Protocol so tests can substitute one.""" + + def __call__( + self, body: bytes, headers: Mapping[str, str], *, fresh: bool + ) -> tuple[Observed, bytes | None, int, Mapping[str, str]]: ... + + +class Gateway: + """One upstream, one alias, one `Control` (SPEC-v0.2 §6.1). + + Several upstreams means several processes: a single process multiplexing them would have + to choose an upstream per request from something in the request, and every candidate for + that something is agent-controlled. + """ + + def __init__(self, config: GatewayConfig, control: Control, forwarder: Forwarder) -> None: + self._config = config + self._control = control + self._forward = forwarder + + @property + def config(self) -> GatewayConfig: + return self._config + + # --- the request path --------------------------------------------------------------- + + def handle(self, body: bytes, headers: Mapping[str, str]) -> _Response: + """Decide one POST. Returns what the client gets, and records what happened.""" + origin = _header(headers, "origin") + if origin is not None and origin not in self._config.allow_origins: + # §6.1 — the transport requires Origin validation against DNS rebinding, and an + # empty allowlist that accepted every origin would be validation in name only. + _LOG.warning("refused a request from origin %r", origin) + return _Response(403) + + parsed = parse_request(body, headers, max_body_bytes=self._config.max_body_bytes) + if isinstance(parsed, Refusal): + _LOG.warning("refused a request: %s", parsed.message) + return self._refusal(parsed, _request_id(body)) + if not parsed.intercept: + # §6.3 — every other method is relayed, and has no CTRLRun outcome at all. No + # Action, no policy, no reservation, no receipt. `tools/list` is not an action. + return self._relay(parsed, headers) + return self._intercept(parsed, headers) + + def _refusal(self, refusal: Refusal, request_id: JsonRpcId) -> _Response: + if refusal.code is None: + return _Response(refusal.http_status) + document = { + "jsonrpc": "2.0", + "id": request_id, + "error": {"code": refusal.code, "message": refusal.message}, + } + return _json(refusal.http_status, document) + + def _relay(self, parsed: ParsedRequest, headers: Mapping[str, str]) -> _Response: + observed, payload, status, response_headers = self._forward( + json.dumps(parsed.document).encode(), headers, fresh=False + ) + if payload is None: + _LOG.warning("relaying %s failed: %s", parsed.method, observed) + return _Response(502) + return _Response(status, payload, response_headers) + + def _intercept(self, parsed: ParsedRequest, headers: Mapping[str, str]) -> _Response: + request_id = parsed.document.get("id") + principal = self._principal(parsed, headers) + if principal is None: + # §6.5 — no principal to attribute the refusal to, so it does not belong in the + # evidence log; and it must not be silent either. + _LOG.warning("refused %s: no principal derivable from the request", parsed.tool_name) + code, token, status = NO_PRINCIPAL + return _json( + status, + json_rpc_error(request_id, code, token, "no principal could be derived"), + ) + + action_name = f"mcp.{self._config.alias}.{parsed.tool_name}" + try: + action = self._action(action_name, parsed, principal) + except _Unrepresentable as refused: + self._record_unrepresentable(action_name, parsed, principal, refused.pointer) + code, token, status = UNREPRESENTABLE + return _json( + status, + json_rpc_error(request_id, code, token, str(refused), pointer=refused.pointer), + ) + except (InvalidArgument, EffectKeyError) as refused: + _LOG.warning("refused %s: %s", action_name, refused) + code, token, status = DENIED + return _json(status, json_rpc_error(request_id, code, token, str(refused))) + + return self._execute(action, parsed, headers, request_id) + + def _principal(self, parsed: ParsedRequest, headers: Mapping[str, str]) -> Principal | None: + """§6.5 — exactly one source, configured at startup, and no default.""" + agent: str | None = None + if self._config.principal is not None: + agent = self._config.principal + elif self._config.principal_header is not None: + agent = _header(headers, self._config.principal_header.lower()) + else: + meta = parsed.document.get("params", {}) + meta = meta.get("_meta", {}) if isinstance(meta, Mapping) else {} + info = ( + meta.get("io.modelcontextprotocol/clientInfo") + if isinstance(meta, Mapping) + else None + ) + agent = info.get("name") if isinstance(info, Mapping) else None + if not isinstance(agent, str) or not agent: + return None + user = ( + _header(headers, self._config.user_header.lower()) + if self._config.user_header is not None + else None + ) + return Principal(agent=agent, user=user or None) + + def _action(self, action_name: str, parsed: ParsedRequest, principal: Principal) -> Action: + """§6.6 — the Action a `tools/call` becomes. + + `params._meta` is deliberately not part of it: it carries the protocol version, + clientInfo, a progress token and trace context, all volatile, and including it would + change the action hash between two identical tool calls and void every approval. + """ + from .mcp import find_unrepresentable + + pointer = find_unrepresentable(dict(parsed.arguments)) + if pointer is not None: + raise _Unrepresentable(pointer) + resource_template = self._control.policy.resource_template(action_name) + return Action( + name=action_name, + arguments=dict(parsed.arguments), + principal=principal, + resource=( + None + if resource_template is None + else resolve_resource(resource_template, parsed.arguments) + ), + environment=self._config.environment, + ) + + def _record_unrepresentable( + self, action_name: str, parsed: ParsedRequest, principal: Principal, pointer: str + ) -> None: + """§6.6 — recorded like v0.1 §5.1's effect_key_error: there is a principal, so the + refusal belongs in the evidence log, and it never reaches the policy.""" + from ..policy import Decision, Evaluation + from ..receipt import EventType, ReceiptResult + + arguments = {key: str(value) for key, value in parsed.arguments.items()} + action = Action( + name=action_name, + arguments=arguments, + principal=principal, + environment=self._config.environment, + ) + control = self._control + control._append(EventType.ACTION_PROPOSED, action, {"action_hash": action.action_hash}) + control._append( + EventType.ACTION_DENIED, + action, + {"reason": UNREPRESENTABLE_REASON, "pointer": pointer}, + ) + control._record( + action, + Evaluation(Decision.DENY, UNREPRESENTABLE_REASON), + ReceiptResult.DENIED, + control._clock(), + error=f"{pointer} is a float, which an Action cannot represent", + ) + + def _execute( + self, + action: Action, + parsed: ParsedRequest, + headers: Mapping[str, str], + request_id: JsonRpcId, + ) -> _Response: + """Decide, forward and record — through `Control`, not around it.""" + effect_template = self._control.policy.effect_template(action.name) + try: + effect_key = ( + None if effect_template is None else resolve_effect_key(effect_template, action) + ) + except EffectKeyError as refused: + _LOG.warning("refused %s: %s", action.name, refused) + code, token, status = DENIED + return _json(status, json_rpc_error(request_id, code, token, str(refused))) + + options = self._control.policy.mcp_options(action.name) + held: dict[str, Any] = {} + + def executor() -> Any: + # §6.7 — the request the gateway sends is built from the action's *canonical* + # arguments, never copied from the received body: what a human approved, and + # what was hashed, reserved and recorded, is byte-for-byte what the upstream + # receives. Every other member is carried through unchanged. + forwarded = dict(parsed.document) + params = dict(forwarded.get("params", {})) + params["arguments"] = action.canonical_arguments + forwarded["params"] = params + observed, payload, status, response_headers = self._forward( + json.dumps(forwarded, separators=(",", ":")).encode(), headers, fresh=True + ) + outcome = classify(observed, not_executed_on_error=options.not_executed_on_error) + held["payload"] = payload + held["status"] = status + held["headers"] = response_headers + held["outcome"] = outcome + if outcome.effect is EffectState.COMMITTED: + return payload + if outcome.effect is EffectState.FAILED: + raise NotExecuted(str(outcome.token or observed)) + raise UpstreamAmbiguous(outcome) + + return self._through_control(action, executor, effect_key, held, request_id) + + def _through_control( + self, + action: Action, + executor: Callable[[], Any], + effect_key: str | None, + held: dict[str, Any], + request_id: JsonRpcId, + ) -> _Response: + from ..control import with_approval + + approval = None + if self._control.policy.evaluate(action).decision.value == "approve": + approval = self._control.store.find_granted_approval(action.action_hash) + if approval is None: + refusal = self._before_asking_a_human(action, effect_key, request_id) + if refusal is not None: + return refusal + try: + if approval is not None: + with with_approval(approval.approval_id): + receipt = self._control.execute(action, executor, effect_key) + else: + receipt = self._control.execute(action, executor, effect_key) + except ActionDenied as refused: + code, token, status = DENIED + return _json( + status, + json_rpc_error( + request_id, + code, + token, + str(refused), + reason=refused.reason, + action_id=action.action_id, + ), + ) + except ApprovalRequired as pending: + return self._awaiting(action, pending, request_id) + except DuplicateEffect as refused: + code, token, status = DUPLICATE_EFFECT + return _json( + status, + json_rpc_error( + request_id, + code, + token, + str(refused), + effect_key=refused.effect_key, + state=refused.state, + ), + ) + except AmbiguousEffect as refused: + code, token, status = AMBIGUOUS_EFFECT + return _json( + status, + json_rpc_error( + request_id, code, token, str(refused), effect_key=refused.effect_key + ), + ) + except ApprovalMismatch as refused: + code, token, status = BLOCKED + return _json( + status, + json_rpc_error(request_id, code, token, str(refused), reason=refused.reason), + ) + except (NotExecuted, UpstreamAmbiguous): + return self._upstream_response(action, held, effect_key) + return self._upstream_response(action, held, effect_key, receipt) + + def _before_asking_a_human( + self, action: Action, effect_key: str | None, request_id: JsonRpcId + ) -> _Response | None: + """Two reasons not to create another approval request (SPEC-v0.2 §6.10). + + The first is the spec's: an unexpired *denied* request for this hash already exists. + "No" is an answer, and without this an agent that dislikes a denial resends the call + and gets a fresh notification to a human, until one of them clicks the wrong button. + + The second the spec leaves open, and this is the fail-closed reading. `# SPEC: §6.10` + — if the effect key would refuse the action anyway, asking a human to approve it + wears the same approver down for the same reason, and the answer could not be acted + on if they said yes. So the refusal the reservation would give is given now, before + anyone is troubled by it. It only ever refuses more. + """ + denied = self._control.store.find_denied_request(action.action_hash) + if denied is not None: + code, token, status = APPROVAL_DENIED + return _json( + status, + json_rpc_error( + request_id, + code, + token, + "a human already refused this exact action; the denial holds until the " + "request expires", + request_id=denied.request_id, + ), + ) + if effect_key is None: + return None + record = self._control.store.get_effect(effect_key) + if record is None: + return None + if record.state is EffectState.COMMITTED: + code, token, status = DUPLICATE_EFFECT + return _json( + status, + json_rpc_error( + request_id, + code, + token, + f"effect {effect_key!r} was already committed by {record.action_id}", + effect_key=effect_key, + state="committed", + ), + ) + if record.state is EffectState.AMBIGUOUS: + code, token, status = AMBIGUOUS_EFFECT + return _json( + status, + json_rpc_error( + request_id, + code, + token, + f"effect {effect_key!r} has an unknown outcome from {record.action_id}", + effect_key=effect_key, + ), + ) + return None + + def _awaiting(self, action: Action, pending: ApprovalRequired, request_id: Any) -> _Response: + """§6.10 — "no" is an answer, and re-asking is not free.""" + record = self._control.store.get_approval(pending.request_id) + code, token, status = APPROVAL_REQUIRED + data: dict[str, Any] = {"request_id": pending.request_id, "action_hash": action.action_hash} + if record is not None: + data["expires_at"] = record.expires_at.isoformat() + return _json(status, json_rpc_error(request_id, code, token, str(pending), **data)) + + def _upstream_response( + self, + action: Action, + held: dict[str, Any], + effect_key: str | None, + receipt: Receipt | None = None, + ) -> _Response: + """Relay the upstream's own answer, or synthesize one where none arrived (§6.8).""" + outcome = held.get("outcome") + meta = { + "action_id": action.action_id, + "receipt_id": receipt.receipt_id if receipt is not None else None, + "effect_key": effect_key, + "result": str(outcome.effect) if outcome and outcome.effect else "ambiguous", + "attempt": receipt.attempt if receipt is not None else 1, + } + if outcome is not None and outcome.relay and held.get("payload") is not None: + document = json.loads(held["payload"]) + document.setdefault("_meta", {})[RECEIPT_META_KEY] = meta + return _Response(held["status"], _dump(document), held.get("headers", {})) + if outcome is None: + code, token, status = AMBIGUOUS_EFFECT + return _json(status, json_rpc_error(None, code, token, "no upstream outcome")) + document = json_rpc_error( + None, + outcome.code or -41010, + outcome.token or "ctrlrun.upstream_ambiguous", + "the upstream's outcome is not known to this gateway", + effect_key=effect_key, + action_id=action.action_id, + ) + document["_meta"] = {RECEIPT_META_KEY: meta} + return _Response(outcome.http_status or 502, _dump(document)) + + +class _Unrepresentable(CTRLRunError): + def __init__(self, pointer: str) -> None: + super().__init__(f"{pointer} is a float, which an Action cannot represent") + self.pointer = pointer + + +def _header(headers: Mapping[str, str], name: str) -> str | None: + for key, value in headers.items(): + if key.lower() == name.lower(): + return value + return None + + +def _request_id(body: bytes) -> JsonRpcId: + try: + document = json.loads(body) + except (ValueError, UnicodeDecodeError): + return None + return document.get("id") if isinstance(document, dict) else None + + +def _dump(document: Mapping[str, Any]) -> bytes: + return json.dumps(document, ensure_ascii=False, separators=(",", ":")).encode() + + +def _json(status: int, document: Mapping[str, Any]) -> _Response: + return _Response(status, _dump(document), {"Content-Type": "application/json"}) + + +# --- the transport ---------------------------------------------------------------------- + + +def httpx_forwarder(config: GatewayConfig) -> Any: + """Forward to the upstream, mapping the client's exceptions onto §6.8's `Transport`. + + **A fresh connection for every intercepted call**, and this is not an optimization + oversight. "The connection was never established" is the only claim in §6.8 that asserts + non-execution, and it is provable only if no request byte can have been written — a + pooled connection the upstream closed while idle fails on *write*, which is + indistinguishable from a request that arrived. It costs a handshake per consequential + action, and it buys the only `FAILED` in the table that comes from the transport. + """ + from . import http_client + + httpx = http_client() + pooled = httpx.Client(timeout=config.upstream_timeout) + + def forward( + body: bytes, headers: Mapping[str, str], *, fresh: bool + ) -> tuple[Any, bytes | None, int, Mapping[str, str]]: + relayed = { + key: value + for key, value in headers.items() + if key.lower() not in _HOP_BY_HOP and key.lower() != "content-length" + } + relayed["Content-Type"] = "application/json" + client = httpx.Client(timeout=config.upstream_timeout) if fresh else pooled + try: + response = client.post(config.upstream, content=body, headers=relayed) + except (httpx.ConnectError, httpx.ConnectTimeout): + return Transport.NEVER_CONNECTED, None, 502, {} + except httpx.TransportError: + return Transport.AFTER_REQUEST_SENT, None, 502, {} + finally: + if fresh: + client.close() + return (*_observe(response), response.status_code, dict(response.headers)) + + def _observe(response: Any) -> tuple[Observed, bytes | None]: + challenge = "www-authenticate" in response.headers + if response.status_code == 401 or (response.status_code == 403 and challenge): + # §6.8 — the resource server validates the token before dispatch, so nothing + # reached the tool. This is checked ahead of the body because a peer may answer + # 401 with any payload it likes, and the status is the part that is load-bearing. + return UpstreamStatus( + response.status_code, has_www_authenticate=challenge + ), response.content + try: + document = json.loads(response.content) + except (ValueError, UnicodeDecodeError): + document = None + if not isinstance(document, dict) or document.get("jsonrpc") != "2.0": + return UpstreamStatus(response.status_code, has_www_authenticate=challenge), None + if "error" in document: + code = document["error"].get("code") if isinstance(document["error"], dict) else None + if not isinstance(code, int): + return Transport.UNREADABLE_RESPONSE, None + return UpstreamError(code), response.content + result = document.get("result") + if not isinstance(result, Mapping): + return Transport.UNREADABLE_RESPONSE, None + return ( + UpstreamResult( + result_type=result.get("resultType"), is_error=bool(result.get("isError")) + ), + response.content, + ) + + forward.close = pooled.close # type: ignore[attr-defined] + return forward + + +_HOP_BY_HOP: Final = frozenset( + { + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailers", + "transfer-encoding", + "upgrade", + "host", + } +) + + +# --- the listening side (stdlib, per §6.11) --------------------------------------------- + + +def build_server(gateway: Gateway) -> ThreadingHTTPServer: + """A `ThreadingHTTPServer` bound to the configured address (SPEC-v0.2 §6.11).""" + config = gateway.config + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_POST(self) -> None: + if self.path.rstrip("/") != config.path.rstrip("/"): + self.send_error(404) + return + length = int(self.headers.get("Content-Length") or 0) + if length > config.max_body_bytes: + self._respond(_Response(413)) + return + body = self.rfile.read(length) + self._respond(gateway.handle(body, dict(self.headers.items()))) + + def _respond(self, response: _Response) -> None: + self.send_response(response.status) + for key, value in response.headers.items(): + if key.lower() in _HOP_BY_HOP or key.lower() == "content-length": + continue + self.send_header(key, value) + self.send_header("Content-Length", str(len(response.body))) + self.end_headers() + if response.body: + self.wfile.write(response.body) + + def log_message(self, format: str, *args: object) -> None: + _LOG.debug("%s - %s", self.address_string(), format % args) + + class Server(ThreadingHTTPServer): + daemon_threads = True + allow_reuse_address = True + + return Server((config.host, config.port), Handler) + + +def serve_forever(gateway: Gateway) -> None: + """Run until interrupted. `ctrlrun gateway` calls this.""" + server = build_server(gateway) + if gateway.config.host not in ("127.0.0.1", "localhost", "::1"): + _LOG.warning("listening on %s, which is not loopback", gateway.config.host) + if gateway.config.principal_from_client_info: + # §6.5 — the revision says clientInfo is self-reported and SHOULD NOT be relied on + # for security decisions. It is offerable in v0.2 only because a v0.1 policy cannot + # address the principal at all, so an unauthenticated one misattributes evidence and + # cannot widen an outcome. It is removed in v0.3. + _LOG.warning( + "--principal-from-client-info trusts a self-reported name; it is attribution on " + "a receipt, never an authorization input, and it is removed in v0.3" + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + thread.join() + finally: + server.shutdown() + server.server_close() diff --git a/tests/test_demo.py b/tests/test_demo.py index 898d7f9d..fc48f85d 100644 --- a/tests/test_demo.py +++ b/tests/test_demo.py @@ -942,7 +942,7 @@ def test_an_empty_CTRLRUN_CONFIG_is_refused(workspace, monkeypatch): def test_the_cli_offers_exactly_the_commands_the_spec_freezes(): - """SPEC-v0.1 §8, plus what SPEC-v0.2 §11 adds. `gateway` arrives with item 6.""" + """SPEC-v0.1 §8, plus what SPEC-v0.2 §11 adds.""" assert set(main.commands) == { "init", "demo", @@ -952,6 +952,7 @@ def test_the_cli_offers_exactly_the_commands_the_spec_freezes(): "effects", "resolve", "inspect", + "gateway", } diff --git a/tests/test_gateway_server.py b/tests/test_gateway_server.py new file mode 100644 index 00000000..0ad7a2bf --- /dev/null +++ b/tests/test_gateway_server.py @@ -0,0 +1,637 @@ +"""The gateway in the execution path. Build-list item 6c; SPEC-v0.2 §6.1, §6.3, §6.5-§6.10. + +Acceptance tests T19, T21, T22, T23, T25, and the end-to-end halves of T20 and T24 that +item 6b could only do as pure functions. + +The upstream is a real HTTP server on a real socket, because the guarantees under test are +about what happens between two processes: that the upstream is never called for a refused +request, that what it receives is byte-for-byte what was hashed, and that a lost response +blocks the retry. +""" + +from __future__ import annotations + +import json +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest + +from ctrlrun import Control, EffectState, InvalidArgument, Policy, SQLiteStateStore +from ctrlrun.gateway.server import ( + Gateway, + GatewayConfig, + build_server, + httpx_forwarder, +) +from ctrlrun.receipt import ReceiptResult + +httpx = pytest.importorskip("httpx", reason="the gateway extra is not installed") + +POLICY = """ +schema: ctrlrun.policy/v2 +actions: + mcp.acme.create_refund: + effect: "refund:{payment_id}" + resource: "payment:{payment_id}" + rules: + - when: { amount_gte: 0, amount_lte: 50000 } + decision: allow + - when: { amount_gte: 0, amount_lte: 1000000 } + decision: approve + - decision: deny + mcp.acme.read_only: + decision: allow + mcp.acme.reports_before_acting: + effect: "report:{payment_id}" + mcp: + not_executed_on_error: true + decision: allow +""" + +CURRENT = "2026-07-28" + + +def _call(tool="create_refund", arguments=None, request_id=1): + return { + "jsonrpc": "2.0", + "id": request_id, + "method": "tools/call", + "params": { + "name": tool, + "arguments": {"payment_id": "txn_1", "amount": 200} if arguments is None else arguments, + }, + } + + +def _headers(body_call=None, **extra): + call = body_call or _call() + headers = { + "MCP-Protocol-Version": CURRENT, + "Mcp-Method": "tools/call", + "Mcp-Name": call["params"]["name"], + "X-Agent": "refund-agent", + } + headers.update({key: value for key, value in extra.items() if value is not None}) + return headers + + +class Upstream: + """A fake MCP server. It records what it received, and answers however it was told to.""" + + def __init__(self) -> None: + self.calls: list[dict] = [] + self.headers: list[dict] = [] + self.reply: dict | None = None + self.status = 200 + self.extra_headers: dict[str, str] = {} + self.drop_after_request = False + self.body: bytes | None = None + + def respond(self, result=None, *, error=None, status=200, headers=None): + if error is not None: + self.reply = {"jsonrpc": "2.0", "id": 1, "error": error} + elif result is not None: + self.reply = {"jsonrpc": "2.0", "id": 1, "result": result} + self.status = status + self.extra_headers = headers or {} + + +@pytest.fixture +def upstream(): + state = Upstream() + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + raw = self.rfile.read(length) + state.body = raw + state.headers.append(dict(self.headers.items())) + try: + state.calls.append(json.loads(raw)) + except ValueError: + state.calls.append({"unparseable": raw.decode(errors="replace")}) + if state.drop_after_request: + self.close_connection = True + self.wfile.close() + return + payload = json.dumps(state.reply or {}).encode() if state.reply else b"not json" + self.send_response(state.status) + for key, value in state.extra_headers.items(): + self.send_header(key, value) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + state.url = f"http://127.0.0.1:{server.server_address[1]}/mcp" + yield state + server.shutdown() + server.server_close() + + +@pytest.fixture +def store(tmp_path): + store = SQLiteStateStore(tmp_path / "state.db") + yield store + store.close() + + +@pytest.fixture +def control(store): + return Control(Policy.from_yaml(POLICY), store) + + +@pytest.fixture +def gateway(upstream, control): + config = GatewayConfig(upstream=upstream.url, alias="acme", principal_header="X-Agent", port=0) + forwarder = httpx_forwarder(config) + yield Gateway(config, control, forwarder) + forwarder.close() + + +@pytest.fixture +def client(gateway): + """The gateway on a real socket, so a client speaks to it the way an agent would.""" + server = build_server(gateway) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + with httpx.Client(base_url=f"http://127.0.0.1:{port}", timeout=10) as opened: + yield opened + server.shutdown() + server.server_close() + + +def _post(client, body=None, headers=None, path="/mcp"): + body = _call() if body is None else body + return client.post(path, content=json.dumps(body).encode(), headers=headers or _headers(body)) + + +def _error(response): + return response.json()["error"] + + +# --- T19: the gateway forwards the canonical action ------------------------------------ + + +def test_T19_the_upstream_receives_the_canonical_arguments(client, upstream, store): + """§6.7 — what a human approved, and what was hashed, reserved and recorded, is + byte-for-byte what the upstream receives. A gateway relaying the original bytes would be + binding an approval to one document and executing another.""" + upstream.respond({"resultType": "complete", "content": []}) + scrambled = {"amount": 200, "payment_id": "txn_1"} + body = _call(arguments=scrambled) + body["params"]["arguments"] = {"payment_id": "txn_1", "amount": 200} + + _post(client, body) + + received = upstream.calls[0]["params"]["arguments"] + assert list(received) == ["amount", "payment_id"] # canonical: sorted keys + + +def test_T19_the_recorded_hash_is_the_hash_of_what_the_upstream_received(client, upstream, store): + upstream.respond({"resultType": "complete", "content": []}) + _post(client) + + from ctrlrun import Action, Principal, action_hash + + receipt = store.receipts()[-1] + rebuilt = Action( + name="mcp.acme.create_refund", + arguments=upstream.calls[0]["params"]["arguments"], + principal=Principal(agent="refund-agent"), + resource="payment:txn_1", + ) + assert receipt.action_hash == rebuilt.action_hash == action_hash(rebuilt) + + +def test_T19_the_action_is_named_for_the_alias_and_the_tool(client, upstream, store): + upstream.respond({"resultType": "complete", "content": []}) + _post(client) + + assert store.receipts()[-1].action == "mcp.acme.create_refund" + + +def test_T19_the_response_carries_the_ctrlrun_receipt_meta(client, upstream, store): + """§6.8 — so a client is not left guessing what CTRLRun recorded.""" + upstream.respond({"resultType": "complete", "content": []}) + + response = _post(client) + + meta = response.json()["_meta"]["com.ctrlrun/receipt"] + assert meta["effect_key"] == "refund:txn_1" + assert meta["result"] == "committed" + assert meta["attempt"] == 1 + assert meta["receipt_id"] == store.receipts()[-1].receipt_id + + +def test_T19_an_allowed_call_commits_the_effect(client, upstream, store): + upstream.respond({"resultType": "complete", "content": []}) + _post(client) + + assert store.get_effect("refund:txn_1").state is EffectState.COMMITTED + assert store.receipts()[-1].result is ReceiptResult.COMMITTED + + +# --- T20 end-to-end: the upstream is never called for a refused request ---------------- + + +@pytest.mark.parametrize( + ("body", "headers", "status", "code"), + [ + ("{not json", None, 400, -32700), + ([_call()], None, 400, -32600), + (None, {"MCP-Protocol-Version": "1999-01-01"}, 400, -32022), + (None, {"MCP-Protocol-Version": CURRENT, "Mcp-Name": "create_refund"}, 400, -32020), + ( + None, + {"MCP-Protocol-Version": CURRENT, "Mcp-Method": "tools/call", "Mcp-Name": "other"}, + 400, + -32020, + ), + ], +) +def test_T20_a_refused_request_never_reaches_the_upstream( + client, upstream, body, headers, status, code +): + content = body if isinstance(body, str) else json.dumps(body or _call()) + + response = client.post("/mcp", content=content.encode(), headers=headers or _headers()) + + assert response.status_code == status + assert _error(response)["code"] == code + assert upstream.calls == [] + + +def test_T20_an_unallowlisted_origin_is_refused_before_anything_else(client, upstream): + response = _post(client, headers={**_headers(), "Origin": "https://evil.example"}) + + assert response.status_code == 403 + assert upstream.calls == [] + + +def test_T20_a_body_over_the_limit_is_refused_with_413(client, upstream): + body = _call(arguments={"payment_id": "x" * (1024 * 1024 + 10), "amount": 200}) + + response = client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + + assert response.status_code == 413 + assert upstream.calls == [] + + +# --- T21: no principal, no action ------------------------------------------------------ + + +def test_T21_a_missing_principal_header_is_refused_with_41007(client, upstream, store): + headers = {key: value for key, value in _headers().items() if key != "X-Agent"} + + response = client.post("/mcp", content=json.dumps(_call()).encode(), headers=headers) + + assert response.status_code == 403 + assert _error(response)["code"] == -41007 + assert _error(response)["data"]["error"] == "ctrlrun.no_principal" + assert upstream.calls == [] + + +def test_T21_no_receipt_and_no_events_are_written_without_a_principal(client, store): + """§6.5 — there is no principal to attribute the refusal to, so it does not belong in + the evidence log. It must not be silent either, which is what the log line is for.""" + headers = {key: value for key, value in _headers().items() if key != "X-Agent"} + + client.post("/mcp", content=json.dumps(_call()).encode(), headers=headers) + + assert store.receipts() == () + assert store.events() == () + + +def test_T21_the_refusal_is_logged_with_the_action(client, caplog): + headers = {key: value for key, value in _headers().items() if key != "X-Agent"} + + with caplog.at_level("WARNING", logger="ctrlrun.gateway"): + client.post("/mcp", content=json.dumps(_call()).encode(), headers=headers) + + assert [r for r in caplog.records if "create_refund" in r.getMessage()] + + +def test_an_empty_principal_header_is_no_principal(client, upstream): + response = _post(client, headers={**_headers(), "X-Agent": ""}) + + assert _error(response)["code"] == -41007 + assert upstream.calls == [] + + +# --- T22: a float argument is refused, not coerced ------------------------------------- + + +def test_T22_a_float_argument_is_refused_with_41008_naming_the_pointer(client, upstream, store): + body = _call(arguments={"payment_id": "txn_1", "amount": 20.0}) + + response = client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + + assert response.status_code == 400 + assert _error(response)["code"] == -41008 + assert _error(response)["data"]["pointer"] == "/amount" + assert upstream.calls == [] + + +def test_T22_a_denied_receipt_is_written_with_no_policy_evaluated_event(client, upstream, store): + """§6.6 — recorded like v0.1 §5.1's effect_key_error: there is a principal, so it belongs + in the evidence log, and the policy is never consulted.""" + body = _call(arguments={"payment_id": "txn_1", "amount": 20.0}) + client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + + receipt = store.receipts()[-1] + assert receipt.result is ReceiptResult.DENIED + assert receipt.decision_reason == "unrepresentable_argument" + assert [event.type.value for event in store.events()] == [ + "ACTION_PROPOSED", + "ACTION_DENIED", + ] + + +# --- T23: a lost response blocks the retry (the signature test, over HTTP) ------------- + + +def test_T23_a_lost_response_leaves_the_effect_ambiguous(client, upstream, store): + upstream.drop_after_request = True + + response = _post(client) + + assert response.status_code == 502 + assert _error(response)["code"] == -41010 + assert store.get_effect("refund:txn_1").state is EffectState.AMBIGUOUS + + +def test_T23_the_meta_says_ambiguous(client, upstream, store): + upstream.drop_after_request = True + + response = _post(client) + + assert response.json()["_meta"]["com.ctrlrun/receipt"]["result"] == "ambiguous" + + +def test_T23_the_identical_call_sent_again_is_refused_and_the_upstream_called_once( + client, upstream, store +): + upstream.drop_after_request = True + _post(client) + upstream.drop_after_request = False + upstream.respond({"resultType": "complete", "content": []}) + + again = _post(client) + + assert again.status_code == 409 + assert _error(again)["code"] == -41005 + assert len(upstream.calls) == 1 + assert store.receipts()[-1].result is ReceiptResult.BLOCKED + + +# --- T24 end-to-end: the upstream's own answer reaches the client ---------------------- + + +def test_T24_connection_refused_is_FAILED_and_a_retry_is_permitted(control, store, upstream): + """Nothing was written, so the record is FAILED and the next attempt may run.""" + config = GatewayConfig( + upstream="http://127.0.0.1:1/mcp", alias="acme", principal_header="X-Agent", port=0 + ) + forwarder = httpx_forwarder(config) + gateway = Gateway(config, control, forwarder) + try: + response = gateway.handle(json.dumps(_call()).encode(), _headers()) + finally: + forwarder.close() + + assert json.loads(response.body)["error"]["code"] == -41011 + assert store.get_effect("refund:txn_1").state is EffectState.FAILED + + +def test_T24_is_error_true_is_AMBIGUOUS_and_relayed_unchanged(client, upstream, store): + upstream.respond({"resultType": "complete", "isError": True, "content": [{"x": 1}]}) + + response = _post(client) + + assert store.get_effect("refund:txn_1").state is EffectState.AMBIGUOUS + assert response.json()["result"]["content"] == [{"x": 1}] + + +def test_T24_is_error_true_is_FAILED_where_the_policy_says_so(client, upstream, store): + body = _call(tool="reports_before_acting") + upstream.respond({"resultType": "complete", "isError": True, "content": []}) + + client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + + assert store.get_effect("report:txn_1").state is EffectState.FAILED + + +@pytest.mark.parametrize(("code", "state"), [(-32602, "failed"), (-32603, "ambiguous")]) +def test_T24_jsonrpc_error_codes_map_as_specified(client, upstream, store, code, state): + upstream.respond(error={"code": code, "message": "no"}) + + response = _post(client) + + assert store.get_effect("refund:txn_1").state.value == state + assert response.json()["error"]["code"] == code # the upstream's own error, unchanged + + +def test_T24_a_401_is_FAILED_with_the_challenge_relayed_verbatim(client, upstream, store): + upstream.respond( + result={"resultType": "complete"}, + status=401, + headers={"WWW-Authenticate": 'Bearer realm="acme"'}, + ) + + response = _post(client) + + assert store.get_effect("refund:txn_1").state is EffectState.FAILED + assert response.headers["WWW-Authenticate"] == 'Bearer realm="acme"' + + +def test_T24_a_retry_is_permitted_after_a_401(client, upstream, store): + upstream.respond(result={"resultType": "complete"}, status=401) + _post(client) + upstream.respond(result={"resultType": "complete", "content": []}, status=200) + + _post(client) + + assert store.get_effect("refund:txn_1").state is EffectState.COMMITTED + assert store.get_effect("refund:txn_1").attempt == 2 + + +def test_T24_an_html_error_page_is_AMBIGUOUS(client, upstream, store): + upstream.reply = None # the handler writes b"not json" + upstream.status = 502 + + response = _post(client) + + assert store.get_effect("refund:txn_1").state is EffectState.AMBIGUOUS + assert response.json()["error"]["code"] == -41010 + + +@pytest.mark.parametrize("result_type", [None, "who-knows"]) +def test_T24_an_unrecognized_or_absent_result_type_is_AMBIGUOUS( + client, upstream, store, result_type +): + result = {"content": []} if result_type is None else {"resultType": result_type} + upstream.respond(result) + + _post(client) + + assert store.get_effect("refund:txn_1").state is EffectState.AMBIGUOUS + + +def test_T24_a_committed_response_reaches_the_client_otherwise_unchanged(client, upstream): + upstream.respond({"resultType": "complete", "content": [{"text": "ok"}], "extra": 7}) + + document = _post(client).json() + + assert document["result"] == { + "resultType": "complete", + "content": [{"text": "ok"}], + "extra": 7, + } + assert document["jsonrpc"] == "2.0" + + +# --- T25: approval over the gateway ---------------------------------------------------- + + +def _needs_approval(): + return _call(arguments={"payment_id": "txn_2", "amount": 200000}) + + +def test_T25_an_action_needing_a_human_returns_41002_and_writes_no_receipt(client, upstream, store): + body = _needs_approval() + + response = client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + + assert response.status_code == 403 + assert _error(response)["code"] == -41002 + assert _error(response)["data"]["request_id"].startswith("apr_") + assert store.receipts() == () + assert upstream.calls == [] + + +def test_T25_the_identical_call_executes_once_the_approval_is_granted(client, upstream, store): + """§6.10 — a client comes back by re-sending the identical `tools/call`, and the gateway + finds the granted approval by the action's hash.""" + upstream.respond({"resultType": "complete", "content": []}) + body = _needs_approval() + first = client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + request_id = _error(first)["data"]["request_id"] + store.grant_approval(request_id, "cli:local") + + second = client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + + assert second.status_code == 200 + assert store.get_effect("refund:txn_2").state is EffectState.COMMITTED + assert store.get_approval(request_id).status.value == "consumed" + assert len(upstream.calls) == 1 + + +def test_T25_a_third_identical_call_is_refused_and_the_upstream_called_once( + client, upstream, store +): + upstream.respond({"resultType": "complete", "content": []}) + body = _needs_approval() + first = client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + store.grant_approval(_error(first)["data"]["request_id"], "cli:local") + client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + + third = client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + + assert _error(third)["code"] in (-41004, -41006) + assert len(upstream.calls) == 1 + + +def test_T25_a_denied_request_is_not_re_asked_until_it_expires(client, upstream, store): + """§6.10 — "no" is an answer. Without this an agent that dislikes a denial resends the + call and gets a fresh notification to a human, until one of them clicks the wrong + button.""" + body = _needs_approval() + first = client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + request_id = _error(first)["data"]["request_id"] + store.deny_approval(request_id, "cli:local") + + second = client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + + assert _error(second)["code"] == -41003 + assert len(store.approvals_for(store.get_approval(request_id).action_hash)) == 1 + assert upstream.calls == [] + + +# --- §6.3: everything else is relayed -------------------------------------------------- + + +def test_a_non_intercepted_method_is_relayed_with_no_ctrlrun_outcome(client, upstream, store): + """`tools/list` is not an action; only calling a tool is.""" + upstream.respond({"tools": []}) + body = {"jsonrpc": "2.0", "id": 9, "method": "tools/list"} + + response = client.post( + "/mcp", + content=json.dumps(body).encode(), + headers={"MCP-Protocol-Version": CURRENT, "Mcp-Method": "tools/list"}, + ) + + assert response.status_code == 200 + assert len(upstream.calls) == 1 + assert store.receipts() == () + assert store.events() == () + assert "_meta" not in response.json() + + +def test_a_tool_with_no_effect_key_in_policy_gets_no_reservation(client, upstream, store): + body = _call(tool="read_only") + upstream.respond({"resultType": "complete", "content": []}) + + client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + + assert store.list_effects() == () + assert store.receipts()[-1].effect_key is None + + +# --- §6.1, §6.5, §6.6: what the config refuses at startup ------------------------------ + + +@pytest.mark.parametrize( + "overrides", + [ + {"alias": "ac.me"}, + {"alias": "ACME"}, + {"alias": ""}, + {"path": "/ctrlrun/approvals"}, + {"path": "mcp"}, + {"principal": None, "principal_header": None}, + {"principal": "a", "principal_header": "X-Agent"}, + {"host": "0.0.0.0"}, + ], +) +def test_the_config_refuses_what_it_cannot_run_safely(overrides): + base = { + "upstream": "http://localhost:8000", + "alias": "acme", + "principal_header": "X-Agent", + } + with pytest.raises(InvalidArgument): + GatewayConfig(**{**base, **overrides}) + + +def test_a_remote_bind_is_permitted_only_when_it_is_meant(): + config = GatewayConfig( + upstream="http://localhost:8000", + alias="acme", + principal_header="X-Agent", + host="0.0.0.0", + allow_remote=True, + ) + + assert config.host == "0.0.0.0" From 0cd3e80dd5c8281e3212c559c77e53675807e31f Mon Sep 17 00:00:00 2001 From: arpan Date: Fri, 4 Sep 2026 00:27:43 +0530 Subject: [PATCH 2/2] test: the DuplicateEffect path, which no approval test could reach MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mutation-testing §6.10 found it: the check that will not send a human a decision the effect key would refuse anyway catches an *approved* duplicate before Control ever raises, so the except-DuplicateEffect handler had no test. An allowed action goes all the way to the reservation, which is where v0.1 §5.4 refuses it, and that is the mapping under test. --- tests/test_gateway_server.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/tests/test_gateway_server.py b/tests/test_gateway_server.py index 0ad7a2bf..46c1c80e 100644 --- a/tests/test_gateway_server.py +++ b/tests/test_gateway_server.py @@ -501,6 +501,25 @@ def test_T24_a_committed_response_reaches_the_client_otherwise_unchanged(client, assert document["jsonrpc"] == "2.0" +def test_T24_a_second_identical_allowed_call_is_refused_as_a_duplicate(client, upstream, store): + """The `DuplicateEffect` path, which the approval tests never reach. + + An approved action is caught earlier, by the check that will not send a human a decision + the effect key would refuse anyway (§6.10). An *allowed* one goes all the way to the + reservation, which is where v0.1 §5.4 refuses it — and that is the mapping under test. + """ + upstream.respond({"resultType": "complete", "content": []}) + _post(client) + + again = _post(client) + + assert again.status_code == 409 + assert _error(again)["code"] == -41004 + assert _error(again)["data"]["state"] == "committed" + assert _error(again)["data"]["effect_key"] == "refund:txn_1" + assert len(upstream.calls) == 1 + + # --- T25: approval over the gateway ----------------------------------------------------