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..17da8885 --- /dev/null +++ b/src/ctrlrun/webhook.py @@ -0,0 +1,372 @@ +"""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(): + # 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" + ) + 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()) + # 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: + """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..95f59a08 --- /dev/null +++ b/tests/test_webhook.py @@ -0,0 +1,538 @@ +"""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 + self.redirect_status = 302 + + +@pytest.fixture +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() + + 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 + + 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 +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 + + +@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, **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 + + +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 --------------------------------------------------- + + +@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 + + +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", "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 + + 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