From ec86791b1e15f34a7a05bc4c923725be77329332 Mon Sep 17 00:00:00 2001 From: arpan Date: Fri, 4 Sep 2026 01:23:49 +0530 Subject: [PATCH 1/2] item 7: the webhook approval provider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SPEC-v0.2 §7, acceptance tests T27 and T28. Outbound is core because it needs neither an HTTP server nor an SDK: one signed POST over stdlib urllib.request. The inbound endpoint is the gateway's, because that is the one server this release ships. The signature is over the exact bytes sent. Two encoders that agree on meaning and differ on whitespace would produce a MAC the receiver cannot verify, and a receiver that re-serialized the payload to check would be verifying its own guess rather than what arrived. Every inbound condition is checked before anything is written, and the refusal message has the same shape for all of them: telling a caller which condition it missed is telling it what to try next. A replayed answer is idempotent only where it asks for the answer the record already carries — a grant replayed onto a denied or consumed record is a different question, and is refused. Single use stays where v0.1 §4.2 A2 put it, at consumption. An undelivered notification is not an approval. If every attempt fails the request stays pending, ApprovalRequired is raised with a request_id that `ctrlrun approve` can still answer, and the failure is logged without the secret or the payload — the payload carries the arguments, which are the customer identifiers and amounts. Redirects are never followed: one to another host would deliver the signed payload, and the action inside it, somewhere the operator did not name. An http:// url is refused unless it is loopback and meant. urllib.request is imported inside the methods that use it, not at module scope. It drags in `ssl`, and §1.1's rule — a core install pays only for what it uses — reads the same against the stdlib as against the extras. The examples' no-network harness caught this: it had replaced socket.socket with a function, which `ssl` cannot subclass. The harness now replaces the type with one that refuses to connect, which is what a cut cable actually looks like. --- src/ctrlrun/__init__.py | 2 + src/ctrlrun/cli/main.py | 21 ++ src/ctrlrun/gateway/__init__.py | 24 +- src/ctrlrun/gateway/server.py | 28 ++ src/ctrlrun/webhook.py | 367 +++++++++++++++++++++++++ tests/test_examples.py | 20 +- tests/test_packaging.py | 13 + tests/test_webhook.py | 462 ++++++++++++++++++++++++++++++++ 8 files changed, 934 insertions(+), 3 deletions(-) create mode 100644 src/ctrlrun/webhook.py create mode 100644 tests/test_webhook.py diff --git a/src/ctrlrun/__init__.py b/src/ctrlrun/__init__.py index 1f204959..3fa7616e 100644 --- a/src/ctrlrun/__init__.py +++ b/src/ctrlrun/__init__.py @@ -31,6 +31,7 @@ from .policy import Decision, Policy from .receipt import Event, EventSink, JSONLEventSink, Receipt from .state import InMemoryStateStore, SQLiteStateStore, StateStore +from .webhook import WebhookApprovalProvider __all__ = [ "Action", @@ -66,6 +67,7 @@ "ScriptedApprovalProvider", "StateStore", "Suspended", + "WebhookApprovalProvider", "action_hash", "canonicalize", "context", diff --git a/src/ctrlrun/cli/main.py b/src/ctrlrun/cli/main.py index f3f4ed5d..df2c7dc5 100644 --- a/src/ctrlrun/cli/main.py +++ b/src/ctrlrun/cli/main.py @@ -460,6 +460,19 @@ def _approval_dict(record: ApprovalRecord) -> dict[str, Any]: @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.") +@click.option("--public-url", default=None, help="Where the gateway is reachable, for respond_to.") +@click.option("--webhook-url", default=None, help="Notify this endpoint on APPROVAL_REQUESTED.") +@click.option( + "--webhook-secret-file", + default=None, + type=click.Path(exists=True, dir_okay=False), + help="Read the shared secret from here instead of $CTRLRUN_WEBHOOK_SECRET.", +) +@click.option( + "--allow-insecure-webhook", + is_flag=True, + help="Permit an http:// webhook url, loopback only.", +) def gateway( upstream: str, alias: str, @@ -474,6 +487,10 @@ def gateway( max_body_bytes: int, allow_origins: tuple[str, ...], allow_remote: bool, + public_url: str | None, + webhook_url: str | None, + webhook_secret_file: str | None, + allow_insecure_webhook: bool, ) -> None: """Front an MCP server, applying this directory's policy to every tools/call.""" host, _, port = listen.rpartition(":") @@ -496,6 +513,10 @@ def gateway( max_body_bytes=max_body_bytes, allow_origins=tuple(allow_origins), allow_remote=allow_remote, + public_url=public_url, + webhook_url=webhook_url, + webhook_secret_file=webhook_secret_file, + allow_insecure_webhook=allow_insecure_webhook, ) except CTRLRunError as exc: raise _fail(exc) from exc diff --git a/src/ctrlrun/gateway/__init__.py b/src/ctrlrun/gateway/__init__.py index 43f1f022..36d1aba3 100644 --- a/src/ctrlrun/gateway/__init__.py +++ b/src/ctrlrun/gateway/__init__.py @@ -47,10 +47,32 @@ def serve(*, upstream: str, alias: str, **options: Any) -> None: """ http_client() from ..control import Control + from ..webhook import WebhookApprovalProvider, webhook_secret from .server import Gateway, GatewayConfig, httpx_forwarder, serve_forever - config = GatewayConfig(upstream=upstream, alias=alias, **options) + public_url = options.pop("public_url", None) + webhook_url = options.pop("webhook_url", None) + secret_file = options.pop("webhook_secret_file", None) + allow_insecure = bool(options.pop("allow_insecure_webhook", False)) + secret = webhook_secret(secret_file) if secret_file else webhook_secret() + + config = GatewayConfig(upstream=upstream, alias=alias, webhook_secret=secret, **options) control = Control.from_file() + if webhook_url: + # §7 — the provider notifies; the gateway's endpoint receives. Both need the same + # secret, and neither takes it from a command-line argument (§7.3). + control = Control( + control.policy, + control.store, + WebhookApprovalProvider( + control.store, + url=webhook_url, + secret=secret, + public_url=public_url, + allow_insecure=allow_insecure, + ), + sinks=[], + ) forwarder = httpx_forwarder(config) try: serve_forever(Gateway(config, control, forwarder)) diff --git a/src/ctrlrun/gateway/server.py b/src/ctrlrun/gateway/server.py index bbac1886..afda860f 100644 --- a/src/ctrlrun/gateway/server.py +++ b/src/ctrlrun/gateway/server.py @@ -18,6 +18,7 @@ import threading from collections.abc import Callable, Mapping from dataclasses import dataclass, field +from datetime import timedelta from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any, Final, Protocol, TypeAlias @@ -61,6 +62,7 @@ #: Reserved for the webhook approval endpoint (§7.2). The MCP path may not overlap it. CTRLRUN_PREFIX: Final = "/ctrlrun/" +APPROVALS_PATH: Final = "/ctrlrun/approvals/" DEFAULT_UPSTREAM_TIMEOUT: Final = 30.0 DEFAULT_APPROVAL_TIMEOUT: Final = 900.0 @@ -151,6 +153,8 @@ class GatewayConfig: allow_remote: bool = False elicitation_timeout: float = DEFAULT_ELICITATION_TIMEOUT max_elicitation_rounds: int = DEFAULT_MAX_ELICITATION_ROUNDS + webhook_secret: str | None = None + replay_window: float = 300.0 def __post_init__(self) -> None: import re @@ -234,6 +238,30 @@ def config(self) -> GatewayConfig: # --- the request path --------------------------------------------------------------- + def handle_approval( + self, request_id: str, body: bytes, headers: Mapping[str, str] + ) -> _Response: + """`POST /ctrlrun/approvals/` (SPEC-v0.2 §7.2). + + The gateway serves it because it is the one server this release ships. Everything the + endpoint decides lives in `webhook.handle_inbound`, which is core: this is the socket + and nothing else. + """ + from ..webhook import SIGNATURE_HEADER, handle_inbound + + if self._config.webhook_secret is None: + _LOG.warning("an inbound approval arrived but no webhook secret is configured") + return _Response(404) + status, message = handle_inbound( + self._control.store, + request_id, + body, + _header(headers, SIGNATURE_HEADER) or "", + secret=self._config.webhook_secret, + replay_window=timedelta(seconds=self._config.replay_window), + ) + return _json(status, {"status": "ok" if status == 200 else "refused", "detail": message}) + 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") diff --git a/src/ctrlrun/webhook.py b/src/ctrlrun/webhook.py new file mode 100644 index 00000000..fded4d96 --- /dev/null +++ b/src/ctrlrun/webhook.py @@ -0,0 +1,367 @@ +"""The webhook approval provider. Build-list item 7; SPEC-v0.2 §7. + +Core, not an extra, because the outbound half needs neither an HTTP server nor a third-party +SDK: it is one signed POST over stdlib `urllib.request`. Only the inbound endpoint needs a +server, and that is the gateway's (§7.2). + +The signature is over the **exact bytes sent**. Two encoders that agree on meaning and differ +on whitespace would produce a MAC the receiver cannot verify, and a receiver that +re-serialized the payload to check would be verifying its own guess rather than what arrived. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import logging +import os +import time +from collections.abc import Callable +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import TYPE_CHECKING, Any, Final + +if TYPE_CHECKING: # `urllib.request` drags in `ssl`; see `_deliver`. + import urllib.request + +from .action import Action +from .approval import ( + DEFAULT_APPROVAL_TTL, + Approval, + ApprovalRequest, + ApprovalStatus, + ApprovalStore, + LocalApprovalProvider, +) +from .errors import CTRLRunError, InvalidArgument +from .receipt import iso_timestamp + +_LOG = logging.getLogger(__name__) + +#: SPEC-v0.2 §7.1 — the schema of one outbound notification. +REQUEST_SCHEMA: Final = "ctrlrun.approval_request/v1" + +#: And the header both directions carry it in. +SIGNATURE_HEADER: Final = "CTRLRun-Signature" + +#: §7.2 — how far a timestamp may be from now. No nonce store is needed: inside the window a +#: replayed grant is idempotent because the record is already granted, and outside it this +#: check rejects. +DEFAULT_REPLAY_WINDOW: Final = timedelta(seconds=300) + +DEFAULT_TIMEOUT: Final = timedelta(seconds=10) +DEFAULT_RETRIES: Final = 2 +DEFAULT_BACKOFF: Final = timedelta(seconds=1) + +#: §7.3 — read from here or from a file, never from a command-line argument, which every +#: process listing on the host can read. +SECRET_ENV_VAR: Final = "CTRLRUN_WEBHOOK_SECRET" + +#: Shorter than this is not a secret. HMAC-SHA256's block is 64 bytes; 32 is the floor below +#: which the MAC is weaker than the hash it is built on. +MIN_SECRET_BYTES: Final = 32 + +#: §7.2 — the endpoint the gateway serves, and the prefix its path may not overlap. +APPROVALS_PATH: Final = "/ctrlrun/approvals/" + +_LOOPBACK: Final = frozenset({"127.0.0.1", "localhost", "::1", "[::1]"}) + + +def sign(body: bytes, secret: str, *, at: datetime | None = None) -> str: + """`t=,v1=` over `f"{t}.{body}"` (SPEC-v0.2 §7.1).""" + moment = at if at is not None else datetime.now(UTC) + stamp = int(moment.timestamp()) + mac = hmac.new( + secret.encode("utf-8"), f"{stamp}.{body.decode('utf-8')}".encode(), hashlib.sha256 + ) + return f"t={stamp},v1={mac.hexdigest()}" + + +def verify(header: str, body: bytes, secret: str, window: timedelta) -> bool: + """Whether `header` signs `body` and is inside `window` (SPEC-v0.2 §7.2). + + `hmac.compare_digest`, because a byte-by-byte comparison of a MAC leaks its prefix to + anyone who can time the answer — and the caller here is by definition untrusted. + """ + parts: dict[str, str] = {} + for part in header.split(","): + name, _, value = part.partition("=") + parts[name.strip()] = value.strip() + if "t" not in parts or "v1" not in parts: + return False + try: + stamp = int(parts["t"]) + except ValueError: + return False + if abs(time.time() - stamp) > window.total_seconds(): + return False + expected = hmac.new( + secret.encode("utf-8"), f"{stamp}.{body.decode('utf-8')}".encode(), hashlib.sha256 + ).hexdigest() + return hmac.compare_digest(expected, parts["v1"]) + + +def webhook_secret(path: str | os.PathLike[str] | None = None) -> str | None: + """The shared secret, from a file or `$CTRLRUN_WEBHOOK_SECRET` (SPEC-v0.2 §7.3). + + Set but empty is a misconfiguration, not a licence to run unsigned — the same reading + `$CTRLRUN_CONFIG` gets in v0.1 §3.4. + """ + if path is not None: + return _checked_secret(Path(path).read_text(encoding="utf-8").strip()) + configured = os.environ.get(SECRET_ENV_VAR) + if configured is None: + return None + return _checked_secret(configured.strip()) + + +def _checked_secret(secret: object) -> str: + if not isinstance(secret, str) or not secret.strip(): + raise InvalidArgument( + f"the webhook secret is empty; set {SECRET_ENV_VAR} or --webhook-secret-file" + ) + if len(secret.encode("utf-8")) < MIN_SECRET_BYTES: + raise InvalidArgument( + f"the webhook secret is shorter than {MIN_SECRET_BYTES} bytes; a MAC is only " + "worth the key behind it" + ) + return secret + + +class WebhookApprovalProvider: + """Notify a human system on `APPROVAL_REQUESTED`, and let it answer (SPEC-v0.2 §7). + + An `ApprovalProvider` (v0.1 §4.3): `request()` records the request and posts it, `wait()` + polls the store exactly as `LocalApprovalProvider` does — the answer arrives out of band, + whether from the gateway's inbound endpoint or from `ctrlrun approve`. + + **An undelivered notification is not an approval.** If every attempt fails the request + stays `pending`, `ApprovalRequired` is raised as usual with a real `request_id` that + `ctrlrun approve` can still answer, and the failure is logged. The action is refused + either way, which is the only outcome that is safe when nobody was told. + """ + + def __init__( + self, + store: ApprovalStore, + *, + url: str, + secret: object, + timeout: timedelta = DEFAULT_TIMEOUT, + retries: int = DEFAULT_RETRIES, + backoff: timedelta = DEFAULT_BACKOFF, + replay_window: timedelta = DEFAULT_REPLAY_WINDOW, + public_url: str | None = None, + allow_insecure: bool = False, + clock: Callable[[], datetime] = lambda: datetime.now(UTC), + ) -> None: + self._secret = _checked_secret(secret) + self._url = _checked_url(url, allow_insecure) + self._timeout = timeout + self._retries = max(0, retries) + self._backoff = backoff + self._replay_window = replay_window + self._public_url = public_url.rstrip("/") if public_url else None + self._store = store + self._clock = clock + self._local = LocalApprovalProvider(store, clock=clock) + + @property + def replay_window(self) -> timedelta: + return self._replay_window + + def request(self, action: Action, ttl: timedelta = DEFAULT_APPROVAL_TTL) -> ApprovalRequest: + """Record the request, then tell somebody. The record is written first. + + A notification for a request the store never accepted would point at nothing, and a + human who answered it would be answering a question CTRLRun cannot connect to an + action. + """ + request = self._local.request(action, ttl) + self._deliver(request) + return request + + def wait(self, request_id: str, timeout: timedelta | None = None) -> Approval | None: + return self._local.wait(request_id, timeout) + + def _deliver(self, request: ApprovalRequest) -> None: + # Imported here, not at module scope: `urllib.request` pulls in `ssl`, and nothing + # that never sends a webhook should pay for a TLS stack at `import ctrlrun` (§1.1's + # dependency rule, applied to the stdlib as well as to the extras). + import urllib.error + + body = _payload(request, self._public_url) + signature = sign(body, self._secret, at=self._clock()) + headers = {"Content-Type": "application/json", SIGNATURE_HEADER: signature} + for attempt in range(self._retries + 1): + try: + self._post(body, headers) + return + except (urllib.error.URLError, OSError, CTRLRunError) as exc: + # Never the secret, and never the body: the payload carries the action's + # arguments, which are the customer identifiers and amounts (§8's rule for + # the OTel sink, applied to a log line). + _LOG.warning( + "webhook delivery for %s failed (attempt %d of %d): %s", + request.request_id, + attempt + 1, + self._retries + 1, + exc, + ) + if attempt < self._retries and self._backoff.total_seconds() > 0: + time.sleep(self._backoff.total_seconds() * (2**attempt)) + _LOG.warning( + "webhook delivery for %s gave up; the request is still pending and " + "'ctrlrun approve %s' can answer it", + request.request_id, + request.request_id, + ) + + def _post(self, body: bytes, headers: dict[str, str]) -> None: + import urllib.request + + request = urllib.request.Request(self._url, data=body, headers=headers, method="POST") + # `_NoRedirects` because a redirect to another host would deliver the signed payload, + # and the action inside it, somewhere the operator did not name. + opener = urllib.request.build_opener(_no_redirects()) + with opener.open(request, timeout=self._timeout.total_seconds()) as response: + if response.status >= 400: + raise CTRLRunError(f"the webhook endpoint answered {response.status}") + + +def _no_redirects() -> urllib.request.BaseHandler: + """A handler that refuses every redirect (SPEC-v0.2 §7.1). + + A function rather than a class so `urllib.request` stays out of module scope: see + `_deliver`. A redirect to another host would deliver the signed payload, and the action + inside it, somewhere the operator did not name. + """ + import urllib.request + + class Refuse(urllib.request.HTTPRedirectHandler): + def redirect_request(self, *args: object, **kwargs: object) -> None: + return None + + return Refuse() + + +def _checked_url(url: str, allow_insecure: bool) -> str: + from urllib.parse import urlparse + + parsed = urlparse(url) + if parsed.scheme == "https": + return url + if parsed.scheme != "http": + raise InvalidArgument(f"webhook url {url!r} must be http or https") + host = (parsed.hostname or "").lower() + if host in _LOOPBACK and allow_insecure: + return url + raise InvalidArgument( + f"webhook url {url!r} is not https; an unencrypted hop would put the action and its " + "signature on the wire. Loopback is permitted with --allow-insecure-webhook." + ) + + +def _payload(request: ApprovalRequest, public_url: str | None) -> bytes: + document: dict[str, Any] = { + "schema": REQUEST_SCHEMA, + "request_id": request.request_id, + "action_hash": request.action_hash, + "action": _action_document(request.action), + "created_at": iso_timestamp(request.created_at), + "expires_at": iso_timestamp(request.expires_at), + } + if public_url is not None: + # §7.2 — present only when there is somewhere to POST back to. Its absence is how the + # receiving system knows the answer has to come from the CLI. + document["respond_to"] = f"{public_url}{APPROVALS_PATH}{request.request_id}" + return json.dumps(document, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + + +def _action_document(action: Action) -> dict[str, Any]: + return { + "name": action.name, + "arguments": dict(action.canonical_arguments), + "principal": {"agent": action.principal.agent, "user": action.principal.user}, + "resource": action.resource, + "environment": action.environment, + } + + +def handle_inbound( + store: ApprovalStore, + path_request_id: str, + body: bytes, + signature: str, + *, + secret: str, + replay_window: timedelta = DEFAULT_REPLAY_WINDOW, +) -> tuple[int, str]: + """Answer one `POST /ctrlrun/approvals/` (SPEC-v0.2 §7.2). + + Every condition must hold or the request is rejected with HTTP 400, **no state change**, + and a warning. They are checked before anything is written, and the failure message is + the same shape for all of them: telling a caller which condition it missed is telling it + what to try next. + """ + if not verify(signature, body, secret, replay_window): + return _refused(path_request_id, "the signature did not verify inside the replay window") + try: + document = json.loads(body) + except (ValueError, UnicodeDecodeError): + return _refused(path_request_id, "the body is not valid JSON") + if not isinstance(document, dict): + return _refused(path_request_id, "the body is not an object") + + if document.get("request_id") != path_request_id: + return _refused(path_request_id, "the path and the body name different requests") + record = store.get_approval(path_request_id) + if record is None: + return _refused(path_request_id, "there is no such request") + if document.get("action_hash") != record.action_hash: + return _refused(path_request_id, "the action hash does not match the stored request") + + decision = document.get("decision") + if decision not in ("grant", "deny"): + return _refused(path_request_id, "decision must be 'grant' or 'deny'") + approver = document.get("approver") + if not isinstance(approver, str) or not approver.strip(): + return _refused(path_request_id, "approver must be a non-empty string") + + already = _already_answered(record.status, decision) + if already is not None: + # §7.2 — inside the replay window a replayed answer is idempotent, which is why no + # nonce store is needed: the record already says what this message is asking for. + # Outside the window the timestamp check above rejected it. + return 200, already + + try: + # Single use is enforced at consumption (v0.1 §4.2 A2), and this endpoint does not + # get to weaken it: `grant_approval` refuses a record that is consumed, denied or + # expired, so a replay arriving *after* consumption cannot resurrect anything. + if decision == "grant": + store.grant_approval(path_request_id, approver) + else: + store.deny_approval(path_request_id, approver) + except CTRLRunError as refused: + return _refused(path_request_id, str(refused)) + return 200, "ok" + + +def _already_answered(status: ApprovalStatus, decision: str) -> str | None: + """Whether this message asks for the answer the record already carries (§7.2). + + Only the *same* answer is idempotent. A grant replayed onto a denied record, or onto a + consumed one, is not a replay — it is a different question, and it is refused below. + """ + if decision == "grant" and status is ApprovalStatus.GRANTED: + return "already granted" + if decision == "deny" and status is ApprovalStatus.DENIED: + return "already denied" + return None + + +def _refused(request_id: str, why: str) -> tuple[int, str]: + _LOG.warning("inbound approval for %s refused: %s", request_id, why) + return 400, why diff --git a/tests/test_examples.py b/tests/test_examples.py index 1e420f14..bfa167fb 100644 --- a/tests/test_examples.py +++ b/tests/test_examples.py @@ -81,12 +81,28 @@ import socket +_real = socket.socket + + +class _Refusing(_real): + """A socket that exists but will not connect, which is what a cut cable looks like. + + Replacing the *type* with a function breaks anything that subclasses it — `ssl` does — + so the refusal goes on the operations instead. + """ + + def connect(self, *args, **kwargs): + raise RuntimeError("an example tried to connect; examples must run with no network") + + def connect_ex(self, *args, **kwargs): + raise RuntimeError("an example tried to connect; examples must run with no network") + def _refuse(*args, **kwargs): - raise RuntimeError("an example opened a socket; examples must run with no network") + raise RuntimeError("an example tried to resolve a name; examples run with no network") -socket.socket = _refuse +socket.socket = _Refusing socket.create_connection = _refuse socket.getaddrinfo = _refuse ''' diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 59f255ca..d621c4ed 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -75,6 +75,19 @@ def test_importing_ctrlrun_does_not_import_the_gateway_extra(): assert finished.stdout.strip() == "False" +def test_importing_ctrlrun_does_not_import_a_tls_stack(): + """§7's outbound half is stdlib, but `urllib.request` drags in `ssl`, and §1.1's rule is + that a core install pays only for what it uses. The import is inside the method.""" + finished = subprocess.run( + [sys.executable, "-c", "import ctrlrun, sys; print('ssl' in sys.modules)"], + capture_output=True, + text=True, + check=True, + ) + + assert finished.stdout.strip() == "False" + + def test_importing_ctrlrun_does_not_import_the_gateway_package(): finished = subprocess.run( [sys.executable, "-c", "import ctrlrun, sys; print('ctrlrun.gateway' in sys.modules)"], diff --git a/tests/test_webhook.py b/tests/test_webhook.py new file mode 100644 index 00000000..b7ac208c --- /dev/null +++ b/tests/test_webhook.py @@ -0,0 +1,462 @@ +"""The webhook approval provider. Build-list item 7; SPEC-v0.2 §7, tests T27 and T28. + +Outbound is core: one signed POST over stdlib `urllib.request`, so `pip install ctrlrun` does +not grow for it. The inbound endpoint is the gateway's, because that is the one server this +release ships. + +The signature is over the **exact bytes sent**, and every inbound condition must hold or the +request is rejected with no state change. An undelivered notification is not an approval, and +a replayed one must not resurrect anything. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import threading +import time +from datetime import UTC, datetime, timedelta +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest + +from ctrlrun import ( + Action, + ApprovalRequired, + Control, + InvalidArgument, + Policy, + Principal, + SQLiteStateStore, + WebhookApprovalProvider, + context, + protect, +) +from ctrlrun.approval import ApprovalStatus +from ctrlrun.webhook import ( + REQUEST_SCHEMA, + SIGNATURE_HEADER, + sign, + verify, +) + +SECRET = "a" * 32 + +POLICY = """ +schema: ctrlrun.policy/v1 +actions: + stripe.refund: + decision: approve +""" + + +class Receiver: + """The other end: it records what it was sent, and answers however it was told to.""" + + def __init__(self) -> None: + self.posts: list[tuple[dict[str, str], bytes]] = [] + self.status = 200 + self.redirect_to: str | None = None + + +@pytest.fixture +def receiver(): + state = Receiver() + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + state.posts.append((dict(self.headers.items()), self.rfile.read(length))) + if state.redirect_to is not None: + self.send_response(307) + self.send_header("Location", state.redirect_to) + self.send_header("Content-Length", "0") + self.end_headers() + return + self.send_response(state.status) + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + state.url = f"http://127.0.0.1:{server.server_address[1]}/hook" + yield state + server.shutdown() + server.server_close() + + +@pytest.fixture +def store(tmp_path): + store = SQLiteStateStore(tmp_path / "state.db") + yield store + store.close() + + +def _provider(store, receiver, **overrides): + return WebhookApprovalProvider( + store, + url=receiver.url, + secret=SECRET, + allow_insecure=True, + **overrides, + ) + + +def _action(amount: int = 2000) -> Action: + return Action( + name="stripe.refund", + arguments={"payment_id": "txn_1", "amount": amount}, + principal=Principal(agent="refund-agent"), + ) + + +def _sent(receiver): + """HTTP header names are case-insensitive, and urllib normalizes what it sends.""" + headers, body = receiver.posts[-1] + folded = {name.lower(): value for name, value in headers.items()} + return folded, body, json.loads(body) + + +# --- T27: the outbound signature ------------------------------------------------------- + + +def test_T27_the_outbound_post_carries_a_signature_over_the_exact_bytes_sent(store, receiver): + """Over the bytes, not over a re-serialization of the payload: two encoders that agree + on meaning and differ on whitespace would produce a signature the receiver cannot + verify, and a receiver that re-serialized to check would be verifying its own guess.""" + _provider(store, receiver).request(_action()) + + headers, body, _ = _sent(receiver) + + assert verify(headers[SIGNATURE_HEADER.lower()], body, SECRET, timedelta(seconds=300)) + + +def test_T27_a_signature_over_different_bytes_does_not_verify(store, receiver): + _provider(store, receiver).request(_action()) + headers, body, _ = _sent(receiver) + + assert not verify( + headers[SIGNATURE_HEADER.lower()], body + b" ", SECRET, timedelta(seconds=300) + ) + + +def test_T27_the_signature_is_hmac_sha256_over_timestamp_dot_body(store, receiver): + """§7.1's construction, spelled out so a receiving system can be written against it.""" + _provider(store, receiver).request(_action()) + headers, body, _ = _sent(receiver) + + parts = dict(part.split("=", 1) for part in headers[SIGNATURE_HEADER.lower()].split(",")) + expected = hmac.new( + SECRET.encode(), f"{parts['t']}.{body.decode()}".encode(), hashlib.sha256 + ).hexdigest() + assert parts["v1"] == expected + + +def test_T27_the_payload_carries_what_the_spec_names(store, receiver): + request = _provider(store, receiver).request(_action()) + _, _, payload = _sent(receiver) + + assert payload["schema"] == REQUEST_SCHEMA == "ctrlrun.approval_request/v1" + assert payload["request_id"] == request.request_id + assert payload["action_hash"] == request.action_hash + assert payload["action"]["name"] == "stripe.refund" + assert payload["created_at"] and payload["expires_at"] + + +def test_respond_to_is_absent_without_a_public_url(store, receiver): + """How the receiving system knows the answer has to come from the CLI (§7.2).""" + _provider(store, receiver).request(_action()) + + assert "respond_to" not in _sent(receiver)[2] + + +def test_respond_to_names_the_gateways_endpoint_when_a_public_url_is_given(store, receiver): + request = _provider(store, receiver, public_url="https://gw.example").request(_action()) + + respond_to = _sent(receiver)[2]["respond_to"] + assert respond_to == f"https://gw.example/ctrlrun/approvals/{request.request_id}" + + +def test_the_content_type_is_json(store, receiver): + _provider(store, receiver).request(_action()) + + assert _sent(receiver)[0]["content-type"] == "application/json" + + +# --- T27: the inbound conditions ------------------------------------------------------- + + +def _inbound(store, request, decision="grant", *, approver="slack:U123", **overrides): + from ctrlrun.webhook import handle_inbound + + body = { + "request_id": overrides.get("body_request_id", request.request_id), + "action_hash": overrides.get("action_hash", request.action_hash), + "decision": decision, + "approver": approver, + } + raw = json.dumps(body).encode() + signature = overrides.get( + "signature", sign(raw, SECRET, at=overrides.get("at", datetime.now(UTC))) + ) + return handle_inbound( + store, + overrides.get("path_request_id", request.request_id), + raw, + signature, + secret=SECRET, + replay_window=timedelta(seconds=300), + ) + + +def _pending(store, receiver): + return _provider(store, receiver).request(_action()) + + +def test_T27_a_valid_grant_moves_the_request_to_granted(store, receiver): + request = _pending(store, receiver) + + status, _ = _inbound(store, request) + + assert status == 200 + record = store.get_approval(request.request_id) + assert record.status is ApprovalStatus.GRANTED + assert record.approver == "slack:U123" + + +def test_T27_a_valid_deny_moves_the_request_to_denied(store, receiver): + request = _pending(store, receiver) + + status, _ = _inbound(store, request, "deny") + + assert status == 200 + assert store.get_approval(request.request_id).status is ApprovalStatus.DENIED + + +@pytest.mark.parametrize( + "overrides", + [ + {"signature": "t=1,v1=deadbeef"}, + {"signature": "nonsense"}, + {"signature": ""}, + {"at": datetime.now(UTC) - timedelta(seconds=400)}, + {"at": datetime.now(UTC) + timedelta(seconds=400)}, + {"body_request_id": "apr_somethingelse"}, + {"action_hash": "sha256:" + "0" * 64}, + ], +) +def test_T27_every_broken_condition_is_400_and_changes_nothing(store, receiver, overrides): + request = _pending(store, receiver) + + status, _ = _inbound(store, request, **overrides) + + assert status == 400 + assert store.get_approval(request.request_id).status is ApprovalStatus.PENDING + + +def test_T27_a_path_body_request_id_disagreement_is_refused(store, receiver): + request = _pending(store, receiver) + + status, _ = _inbound(store, request, path_request_id="apr_other") + + assert status == 400 + assert store.get_approval(request.request_id).status is ApprovalStatus.PENDING + + +def test_T27_an_unknown_request_is_refused(store, receiver): + request = _pending(store, receiver) + store.deny_approval(request.request_id, "cli:local") + + status, _ = _inbound(store, request) + + assert status == 400 + assert store.get_approval(request.request_id).status is ApprovalStatus.DENIED + + +def test_T27_a_replayed_valid_grant_is_idempotent(store, receiver): + """No nonce store is needed: inside the window the record is already granted, and + outside it the timestamp check rejects (§7.2).""" + request = _pending(store, receiver) + at = datetime.now(UTC) + first, _ = _inbound(store, request, at=at) + second, _ = _inbound(store, request, at=at) + + assert first == 200 + assert second == 200 + assert store.get_approval(request.request_id).status is ApprovalStatus.GRANTED + + +def test_T27_a_grant_replayed_after_consumption_does_not_resurrect_it(store, receiver): + """Single use is enforced at consumption (v0.1 §4.2 A2), and this endpoint does not get + to weaken it.""" + action = _action() + request = _provider(store, receiver).request(action) + _inbound(store, request) + store.consume_approval(request.request_id, action.action_hash) + + status, _ = _inbound(store, request) + + assert status == 400 + assert store.get_approval(request.request_id).status is ApprovalStatus.CONSUMED + + +# --- T28: an undelivered webhook approves nothing -------------------------------------- + + +def test_T28_the_provider_gives_up_after_its_bounded_retries(store, receiver, caplog): + receiver.status = 500 + + with caplog.at_level("WARNING", logger="ctrlrun"): + request = _provider(store, receiver, retries=2, backoff=timedelta(0)).request(_action()) + + assert len(receiver.posts) == 3 # the first attempt, then two retries + assert store.get_approval(request.request_id).status is ApprovalStatus.PENDING + assert [r for r in caplog.records if "webhook" in r.getMessage().lower()] + + +def test_T28_the_action_is_still_refused_and_the_request_id_still_answerable( + store, receiver, caplog +): + """The action is refused either way — an undelivered request is not an approval — and + `ctrlrun approve` can still answer the request that was written.""" + receiver.status = 500 + control = Control( + Policy.from_yaml(POLICY), + store, + _provider(store, receiver, retries=1, backoff=timedelta(0)), + ) + + @protect("stripe.refund", effect="refund:{payment_id}", control=control) + def refund(payment_id: str, amount: int) -> str: + return "re_1" + + with ( + caplog.at_level("WARNING", logger="ctrlrun"), + context(agent="refund-agent"), + pytest.raises(ApprovalRequired) as pending, + ): + refund(payment_id="txn_1", amount=2000) + + assert store.get_approval(pending.value.request_id).status is ApprovalStatus.PENDING + assert store.receipts() == () + store.grant_approval(pending.value.request_id, "cli:local") + assert store.get_approval(pending.value.request_id).status is ApprovalStatus.GRANTED + + +def test_T28_an_unreachable_endpoint_does_not_raise_into_the_kernel(store, tmp_path): + provider = WebhookApprovalProvider( + store, + url="http://127.0.0.1:1/hook", + secret=SECRET, + allow_insecure=True, + retries=0, + backoff=timedelta(0), + ) + + request = provider.request(_action()) + + assert store.get_approval(request.request_id).status is ApprovalStatus.PENDING + + +# --- §7.1: what the provider refuses --------------------------------------------------- + + +def test_a_redirect_is_never_followed(store, receiver): + """A redirect to another host would deliver the signed payload, and the action, somewhere + the operator did not name.""" + receiver.redirect_to = "http://127.0.0.1:1/elsewhere" + + request = _provider(store, receiver, retries=0, backoff=timedelta(0)).request(_action()) + + assert len(receiver.posts) == 1 + assert store.get_approval(request.request_id).status is ApprovalStatus.PENDING + + +def test_an_http_url_is_refused_unless_it_is_loopback_and_meant(store): + with pytest.raises(InvalidArgument): + WebhookApprovalProvider(store, url="http://example.com/hook", secret=SECRET) + + with pytest.raises(InvalidArgument): + WebhookApprovalProvider( + store, url="http://127.0.0.1:9/hook", secret=SECRET, allow_insecure=False + ) + + WebhookApprovalProvider(store, url="https://example.com/hook", secret=SECRET) + + +# --- §7.3: the secret ------------------------------------------------------------------ + + +@pytest.mark.parametrize("secret", ["", " ", "a" * 31, None]) +def test_a_weak_or_missing_secret_is_refused(store, secret): + with pytest.raises(InvalidArgument): + WebhookApprovalProvider(store, url="https://example.com/hook", secret=secret) + + +def test_the_secret_never_appears_in_a_log_line(store, receiver, caplog): + receiver.status = 500 + + with caplog.at_level("DEBUG", logger="ctrlrun"): + _provider(store, receiver, retries=1, backoff=timedelta(0)).request(_action()) + + assert SECRET not in "\n".join(record.getMessage() for record in caplog.records) + + +def test_the_secret_is_read_from_the_environment(store, monkeypatch): + from ctrlrun.webhook import webhook_secret + + monkeypatch.setenv("CTRLRUN_WEBHOOK_SECRET", SECRET) + assert webhook_secret() == SECRET + + monkeypatch.setenv("CTRLRUN_WEBHOOK_SECRET", "") + with pytest.raises(InvalidArgument): + webhook_secret() + + monkeypatch.delenv("CTRLRUN_WEBHOOK_SECRET") + assert webhook_secret() is None + + +def test_the_secret_can_come_from_a_file(store, tmp_path, monkeypatch): + from ctrlrun.webhook import webhook_secret + + path = tmp_path / "secret" + path.write_text(f" {SECRET}\n", encoding="utf-8") + monkeypatch.delenv("CTRLRUN_WEBHOOK_SECRET", raising=False) + + assert webhook_secret(path) == SECRET + + +def test_signing_is_stable_and_verification_is_not_a_prefix_match(): + body = b'{"a":1}' + at = datetime.now(UTC) + + signature = sign(body, SECRET, at=at) + + assert verify(signature, body, SECRET, timedelta(seconds=300)) + assert not verify(signature[:-1], body, SECRET, timedelta(seconds=300)) + assert not verify(signature, body, "b" * 32, timedelta(seconds=300)) + + +def test_a_timestamp_outside_the_window_is_refused_in_both_directions(): + body = b'{"a":1}' + window = timedelta(seconds=300) + + old = sign(body, SECRET, at=datetime.now(UTC) - timedelta(seconds=400)) + future = sign(body, SECRET, at=datetime.now(UTC) + timedelta(seconds=400)) + + assert not verify(old, body, SECRET, window) + assert not verify(future, body, SECRET, window) + + +def test_verification_tolerates_a_little_clock_skew(): + body = b'{"a":1}' + signature = sign(body, SECRET, at=datetime.now(UTC) - timedelta(seconds=5)) + + assert verify(signature, body, SECRET, timedelta(seconds=300)) + assert time.time() > 0 From d09955c89d12e1f4a4e71933d34420451d599fa6 Mon Sep 17 00:00:00 2001 From: arpan Date: Fri, 4 Sep 2026 01:40:54 +0530 Subject: [PATCH 2/2] fix: six webhook checks that no test could distinguish, and one dead branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mutation-testing §7 found more than the usual crop. Two were untested rather than subsumed: no test sent a request id the store had never seen, and none sent a decision that was neither grant nor deny. Both now exist. The path/body request-id comparison had a test that passed for the wrong reason: with only the path wrong, the lookup finds nothing and the refusal is right by accident. Parametrized over both directions now — with only the body wrong, the lookup finds the *real* record and the comparison is the only thing between it and a grant. `if response.status >= 400` was dead: urlopen raises HTTPError for anything at or above 400, and HTTPError is a URLError, which _deliver already treats as a failed delivery. Removed. The redirect test was the interesting one. It used a 307 — the one code urllib refuses on a POST by itself — so the no-redirect handler was never exercised. urllib follows 301, 302 and 303, converting the POST to a GET, and the test's second server only recorded POSTs, so even a followed redirect was invisible to it. Parametrized over the three codes urllib actually follows, and the second server records GETs. The empty-secret check is subsumed by the length check and stays for its message: "empty" points at the configuration, "too short" at the value. The test asserts which one it got. --- src/ctrlrun/webhook.py | 11 +++- tests/test_webhook.py | 142 +++++++++++++++++++++++++++++++---------- 2 files changed, 117 insertions(+), 36 deletions(-) diff --git a/src/ctrlrun/webhook.py b/src/ctrlrun/webhook.py index fded4d96..17da8885 100644 --- a/src/ctrlrun/webhook.py +++ b/src/ctrlrun/webhook.py @@ -118,6 +118,9 @@ def webhook_secret(path: str | os.PathLike[str] | None = None) -> str | None: def _checked_secret(secret: object) -> str: if not isinstance(secret, str) or not secret.strip(): + # The length check below would refuse this too. The message is the difference, and + # it is the one an operator needs: "empty" points at the configuration, "too short" + # points at the value. raise InvalidArgument( f"the webhook secret is empty; set {SECRET_ENV_VAR} or --webhook-secret-file" ) @@ -225,9 +228,11 @@ def _post(self, body: bytes, headers: dict[str, str]) -> None: # `_NoRedirects` because a redirect to another host would deliver the signed payload, # and the action inside it, somewhere the operator did not name. opener = urllib.request.build_opener(_no_redirects()) - with opener.open(request, timeout=self._timeout.total_seconds()) as response: - if response.status >= 400: - raise CTRLRunError(f"the webhook endpoint answered {response.status}") + # No status check here: `urlopen` raises `HTTPError` for anything >= 400, and + # `HTTPError` is a `URLError`, which `_deliver` already treats as a failed delivery. + # A second check would be a branch no test could reach. + with opener.open(request, timeout=self._timeout.total_seconds()): + return def _no_redirects() -> urllib.request.BaseHandler: diff --git a/tests/test_webhook.py b/tests/test_webhook.py index b7ac208c..95f59a08 100644 --- a/tests/test_webhook.py +++ b/tests/test_webhook.py @@ -58,37 +58,64 @@ def __init__(self) -> None: self.posts: list[tuple[dict[str, str], bytes]] = [] self.status = 200 self.redirect_to: str | None = None + self.redirect_status = 302 @pytest.fixture -def receiver(): - state = Receiver() - - class Handler(BaseHTTPRequestHandler): - protocol_version = "HTTP/1.1" - - def do_POST(self): - length = int(self.headers.get("Content-Length") or 0) - state.posts.append((dict(self.headers.items()), self.rfile.read(length))) - if state.redirect_to is not None: - self.send_response(307) - self.send_header("Location", state.redirect_to) +def receiver_factory(): + servers = [] + + def make() -> Receiver: + state = Receiver() + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + state.posts.append((dict(self.headers.items()), self.rfile.read(length))) + if state.redirect_to is not None: + self.send_response(state.redirect_status) + self.send_header("Location", state.redirect_to) + self.send_header("Content-Length", "0") + self.end_headers() + return + self.send_response(state.status) self.send_header("Content-Length", "0") self.end_headers() - return - self.send_response(state.status) - self.send_header("Content-Length", "0") - self.end_headers() - def log_message(self, *args): - pass + def do_GET(self): + # urllib converts a POST to a GET when it follows 301/302/303, so a server + # that only recorded POSTs could not see a redirect it was sent. + state.posts.append((dict(self.headers.items()), b"")) + self.send_response(200) + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + state.url = f"http://127.0.0.1:{server.server_address[1]}/hook" + servers.append(server) + return state - server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) - threading.Thread(target=server.serve_forever, daemon=True).start() - state.url = f"http://127.0.0.1:{server.server_address[1]}/hook" - yield state - server.shutdown() - server.server_close() + yield make + for server in servers: + server.shutdown() + server.server_close() + + +@pytest.fixture +def receiver(receiver_factory): + return receiver_factory() + + +@pytest.fixture +def elsewhere(receiver_factory): + """Somewhere the operator did not name, for the redirect test.""" + return receiver_factory() @pytest.fixture @@ -260,10 +287,43 @@ def test_T27_every_broken_condition_is_400_and_changes_nothing(store, receiver, assert store.get_approval(request.request_id).status is ApprovalStatus.PENDING -def test_T27_a_path_body_request_id_disagreement_is_refused(store, receiver): +@pytest.mark.parametrize("side", ["path", "body"]) +def test_T27_a_path_body_request_id_disagreement_is_refused(store, receiver, side): + """Both directions. With only the path wrong, the lookup finds nothing and the refusal + is right for the wrong reason; with only the body wrong, the lookup finds the *real* + record and the comparison is the only thing standing between it and a grant.""" request = _pending(store, receiver) + where = {"path": {"path_request_id": "apr_other"}, "body": {"body_request_id": "apr_other"}} - status, _ = _inbound(store, request, path_request_id="apr_other") + status, _ = _inbound(store, request, **where[side]) + + assert status == 400 + assert store.get_approval(request.request_id).status is ApprovalStatus.PENDING + + +def test_T27_a_request_the_store_never_saw_is_refused(store, receiver): + """Not the same as a request in the wrong state: there is no record at all.""" + request = _pending(store, receiver) + unknown = type(request)( + request_id="apr_nothing", + action_hash=request.action_hash, + action=request.action, + created_at=request.created_at, + expires_at=request.expires_at, + ) + + status, _ = _inbound(store, unknown) + + assert status == 400 + assert store.get_approval("apr_nothing") is None + + +@pytest.mark.parametrize("decision", ["approve", "GRANT", "", None, 1]) +def test_T27_a_decision_that_is_not_grant_or_deny_is_refused(store, receiver, decision): + """A body the endpoint half-understands is a body it must not act on.""" + request = _pending(store, receiver) + + status, _ = _inbound(store, request, decision) assert status == 400 assert store.get_approval(request.request_id).status is ApprovalStatus.PENDING @@ -367,14 +427,23 @@ def test_T28_an_unreachable_endpoint_does_not_raise_into_the_kernel(store, tmp_p # --- §7.1: what the provider refuses --------------------------------------------------- -def test_a_redirect_is_never_followed(store, receiver): - """A redirect to another host would deliver the signed payload, and the action, somewhere - the operator did not name.""" - receiver.redirect_to = "http://127.0.0.1:1/elsewhere" +@pytest.mark.parametrize("status", [301, 302, 303]) +def test_a_redirect_is_never_followed(store, receiver, elsewhere, status): + """A redirect to another host would deliver the request, and the signature header with + it, somewhere the operator did not name. + + Parametrized over 301/302/303 deliberately: those are the codes `urllib` follows on a + POST — converting it to a GET against the new host — and therefore the only ones where + the handler is what stops it. It refuses 307 and 308 on its own, so a test using those + would pass with the handler removed and prove nothing. + """ + receiver.redirect_status = status + receiver.redirect_to = elsewhere.url request = _provider(store, receiver, retries=0, backoff=timedelta(0)).request(_action()) assert len(receiver.posts) == 1 + assert elsewhere.posts == [] assert store.get_approval(request.request_id).status is ApprovalStatus.PENDING @@ -393,11 +462,18 @@ def test_an_http_url_is_refused_unless_it_is_loopback_and_meant(store): # --- §7.3: the secret ------------------------------------------------------------------ -@pytest.mark.parametrize("secret", ["", " ", "a" * 31, None]) -def test_a_weak_or_missing_secret_is_refused(store, secret): - with pytest.raises(InvalidArgument): +@pytest.mark.parametrize( + ("secret", "why"), [("", "empty"), (" ", "empty"), ("a" * 31, "shorter"), (None, "empty")] +) +def test_a_weak_or_missing_secret_is_refused(store, secret, why): + """The length check would refuse an empty one too; the message is the difference, and it + is the one an operator needs — "empty" points at the configuration, "too short" at the + value.""" + with pytest.raises(InvalidArgument) as refused: WebhookApprovalProvider(store, url="https://example.com/hook", secret=secret) + assert why in str(refused.value) + def test_the_secret_never_appears_in_a_log_line(store, receiver, caplog): receiver.status = 500