From d9941dbe90b2e2dbf28d377e99101db5fe78146f Mon Sep 17 00:00:00 2001 From: arpan Date: Sun, 13 Sep 2026 21:31:01 +0530 Subject: [PATCH 1/2] Item 3: upstream identity pinning SPEC-v0.10 section 4, the honest slice of ASI04 and nothing more: still deciding actions, still never inspecting a package, a model, a registry or a build. What it adds is that an action entry may say which SERVER it authorises itself against. upstream: is a ctrlrun.policy/v8 action-entry key carrying two TLS pins and a tool-schema hash. Two, because a digest cannot be a trust anchor: load_verify_locations takes PEM, so tls_cert_file feeds check 3 where the pinned certificates become the connection's only trust anchors, and tls_cert_sha256 feeds checks 1 and 2, which compare what was observed. An entry pinning by digest alone gets two checks of three, stated as a limit. The gate is the action-entry shape of _V4_ENTRY_KEYS and _V5_ENTRY_KEYS, not require_v7's, which walks authority.grants because tasks: and budgets: are grant keys and a standalone --authority document carries no action entries at all. Check 2 is the one that produces a DENY, and it sits above the approval gate on T446's argument: the pin depends on nothing a human says, so asking one about an action pinned to an unverified server leaves a granted approval behind for a call that cannot run. The observation register is per process by construction, because an observation is a fact about a connection this process made and a second process that has made none must refuse rather than inherit somebody else's. That is upstream_unverified, the fail-closed half, and it is what stops an upstream switching a pin off by never being seen. In-process there is no upstream to observe, so a pinned action refuses on every call. The ACS hook refuses at CONSTRUCTION rather than at load: ACS is advisory, the platform runs the tool, and the hook holds no connection, while one loader cannot know which surface will run an action and a load error would stop verify and scan reading a document that pins. -41016 and not -41013, which SPEC-mcp-operator already allocates to ctrlrun.not_a_human. There is one namespace and the range was walked. G27 grades check 2 and only check 2, because that is the one producing a DENY and the only one verify can grade without a network: the comparison is a pure function over two strings, so verify seeds an observation and asserts the refusal. Acceptance tests T489-T497b, including check 3 against a real TLS listener with a CA-signed leaf and VERIFY_X509_PARTIAL_CHAIN. Mutation table in the PR body: three mutations, all three caught. One flaky gate run found a real race in my own test helper: it bound an ephemeral port, closed it and rebound, twice per run. HTTPServer binds for us and server_port reports what it got, so there is no gap to lose. Gate with Postgres: 4411 passed, 0 skipped. Signed-off-by: arpan --- .github/workflows/ci.yml | 4 +- src/ctrlrun/acs.py | 17 ++ src/ctrlrun/control.py | 51 +++++ src/ctrlrun/gateway/__init__.py | 4 + src/ctrlrun/gateway/server.py | 21 +- src/ctrlrun/policy.py | 136 +++++++++++ src/ctrlrun/upstream.py | 113 ++++++++++ src/ctrlrun/verify/guarantees.py | 14 ++ src/ctrlrun/verify/scenarios.py | 104 +++++++++ tests/test_upstream_pinning.py | 371 +++++++++++++++++++++++++++++++ tests/test_verify_action.py | 11 +- 11 files changed, 838 insertions(+), 8 deletions(-) create mode 100644 src/ctrlrun/upstream.py create mode 100644 tests/test_upstream_pinning.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 65bc8fbe..4ba4f417 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -138,9 +138,9 @@ jobs: # SPEC-v0.9 §6 moved the templates count again: G24 is N/A for a document whose # grants name no task, which the templates example's do not. The authority example # binds one, so its count is unchanged and its passing total moved 19 to 20 instead. - test "$AUTHORITY_NA" = "2" + test "$AUTHORITY_NA" = "3" test "$TEMPLATES" = "verified 11/11" - test "$TEMPLATES_NA" = "15" + test "$TEMPLATES_NA" = "16" test -s verify-badge.json test -s verify-report.json test -s verify-report.xml diff --git a/src/ctrlrun/acs.py b/src/ctrlrun/acs.py index 42cbb0db..8ee4a79e 100644 --- a/src/ctrlrun/acs.py +++ b/src/ctrlrun/acs.py @@ -98,6 +98,23 @@ def __init__( ask_timeout_seconds: int = DEFAULT_ASK_TIMEOUT_SECONDS, identity: IdentityProvider | None = None, ) -> None: + pinned = [name for name in control.policy.actions if control.policy.upstream_pin(name)] + if pinned: + # SPEC-v0.10 §4.4 — **at construction, and not at load.** ACS is advisory: the + # *platform* runs the tool and this hook never holds a connection to anything, so + # there is no observation point and nothing to pin. A load error cannot serve here, + # because one loader cannot know which surface will run an action and `verify` and + # `scan` must still read a document that pins (§7.3). `v0.3 §8.4` refuses a hook + # built with no identity provider the same way, for the same reason: a deployment + # finds out at startup rather than during an incident. + raise InvalidArgument( + f"this policy pins an upstream for {pinned[0]!r}" + + (f" and {len(pinned) - 1} other action(s)" if len(pinned) > 1 else "") + + ", and the ACS hook holds no connection to an upstream: ACS is advisory and " + "the platform executes, so there is nothing here to observe or pin. Take the " + "'upstream:' key off those entries, or front the server with 'ctrlrun gateway' " + "instead (SPEC-v0.10 §4.4)" + ) if control.authority is not None and identity is None: # SPEC-v0.3 §8.4 — without a provider this hook reads `params.metadata.agent_id` # straight off the inbound envelope, and §4 makes the principal an authorization diff --git a/src/ctrlrun/control.py b/src/ctrlrun/control.py index 0767a6a1..5a18b1f1 100644 --- a/src/ctrlrun/control.py +++ b/src/ctrlrun/control.py @@ -101,6 +101,7 @@ OBSERVE, POLICY_CHANGE_ACTION, POLICY_UNAPPROVED, + UPSTREAM_UNVERIFIED, Decision, Evaluation, Policy, @@ -737,6 +738,7 @@ def __init__( environment: str | None = None, approver_identity: ApproverIdentity | None = None, require_approved_policy: bool = False, + upstream: str | None = None, ) -> None: self._policy = policy self._store = store @@ -758,6 +760,10 @@ def __init__( # SPEC-v0.8 §8.4. **In code and not in the file it governs**, or the file would switch # off its own governance. Default false: opt in, then fail closed. self._require_approved_policy = require_approved_policy + # SPEC-v0.10 §4.3 — the upstream this deployment fronts, which only a surface holding the + # connection can name. The gateway passes `GatewayConfig.upstream`; in-process it is + # `None`, and §4.4 makes a pinned action refuse `upstream_unverified` there. + self._upstream = upstream #: Cached **only when the answer is yes** (§8.4). A negative answer is re-asked on every #: decision, so a long-lived process that started before the approval landed begins #: working the moment it lands, with no restart; the cost is one keyed read per decision @@ -1684,6 +1690,11 @@ def _observe_secure( # SPEC-v0.9 §4.2.1 — **above the scope check, because `_secure` computes charges before # calling `_in_scope`.** An action that is both out of scope and unmeasurable was refused # `budget_unmeasurable` by enforce mode and reported `out_of_scope` by the pilot. T458. + # SPEC-v0.10 §4.3, the observe-mode row: **recorded, not refused** (`v0.3 §6.2`). At the + # same point in the declared order as `_secure`'s, which is what §5 is about. + observed_upstream = self._upstream_reason(action) + if observed_upstream is not None: + observation.block(observed_upstream) charges = self._observe_charges(action, effect_key, observation) try: self._in_scope(action, scope, scoped, enforcing=False) @@ -2454,6 +2465,12 @@ def _secure( # Assembling after the gate asks a human to approve a refund the kernel has already # decided to refuse, and leaves a granted approval behind for an action nothing can # execute. A probe found `APPROVAL_REQUESTED` written for exactly that shape. T446. + # SPEC-v0.10 §4.3's check 2. **Above the approval gate**, on T446's argument: the pin + # depends on nothing a human says, so asking one about an action pinned to a server this + # process has not verified leaves a granted approval behind for a call that cannot run. + refused_upstream = self._upstream_reason(action) + if refused_upstream is not None: + raise self._refuse_upstream(action, refused_upstream, effect_key) charges = self._charges_for(action, effect_key) approval_id = ( self._presented(action, effect_key, evaluation, started_at, preconditions) @@ -3627,6 +3644,40 @@ def _refuse_unmeasurable( ) return _UnmeasurableError(str(error), reason=reason) + def _upstream_reason(self, action: Action) -> str | None: + """§4.3's check 2: is this action pinned to an upstream this process has verified? + + `None` where the entry pins nothing, which is every action written before v0.10. + + **In-process there is no upstream to observe, so a pinned action is refused** + `upstream_unverified` on every call (§4.4). That is loud, correct, and exactly what the + pin says the operator asked for: a pin is a claim about a server CTRLRun connects to, and + in-process the executor is the operator's own code holding its own connection. + """ + from . import upstream as _upstream + + pin = self._policy.upstream_pin(action.name) + if not pin: + return None + if self._upstream is None: + return UPSTREAM_UNVERIFIED + return _upstream.check(pin, self._upstream, self._policy.tool_name(action.name)) + + def _refuse_upstream(self, action: Action, reason: str, effect_key: str | None) -> ActionDenied: + """The refusal §4.5 names, recorded the way every other `ActionDenied` is.""" + self._append(EventType.ACTION_DENIED, action, {"reason": reason}, effect_key) + detail = ( + "this process has verified no upstream for it" + if reason == UPSTREAM_UNVERIFIED + else "what this process observed is in no pinned list" + ) + return ActionDenied( + f"{action.name}: the policy pins the upstream it authorises, and {detail} " + "(SPEC-v0.10 §4.3)", + reason=reason, + action_id=action.action_id, + ) + def _refuse_budget(self, action: Action, exhausted: BudgetExhaustedError) -> ActionDenied: """SPEC-v0.9 §4.5. Names the grant, the metric and the window; **never the balance**. diff --git a/src/ctrlrun/gateway/__init__.py b/src/ctrlrun/gateway/__init__.py index a1102c4d..8a33c5d6 100644 --- a/src/ctrlrun/gateway/__init__.py +++ b/src/ctrlrun/gateway/__init__.py @@ -102,6 +102,10 @@ def serve(*, upstream: str, alias: str, **options: Any) -> None: sinks=sinks, authority=authority, environment=control.environment, + # SPEC-v0.10 §4.3 — the gateway is the surface that holds the connection, so it is the + # one that can name the upstream an action is pinned against. In-process this is `None` + # and §4.4 refuses a pinned action there. + upstream=config.upstream, ) forwarder = httpx_forwarder(config) gateway = Gateway(config, control, forwarder) diff --git a/src/ctrlrun/gateway/server.py b/src/ctrlrun/gateway/server.py index 1e4f7ec7..1b823e11 100644 --- a/src/ctrlrun/gateway/server.py +++ b/src/ctrlrun/gateway/server.py @@ -48,7 +48,7 @@ IdentityProvider, StaticIdentityProvider, ) -from ..policy import OBSERVE +from ..policy import OBSERVE, UPSTREAM_MISMATCH, UPSTREAM_UNVERIFIED from ..receipt import Receipt from .mcp import ( ACCEPTED_REVISIONS, @@ -121,6 +121,15 @@ #: different message to a client and, in a multi-tenant deployment, a different alert. UNAUTHORIZED: Final = (-41012, "ctrlrun.unauthorized", 403) +#: SPEC-v0.10 §4.5. `-41016` and **not** `-41013`: `SPEC-mcp-operator.md` §9.3 adds `-41013` +#: `ctrlrun.not_a_human`, `-41014` and `-41015` to `v0.2 §6.10`'s table, and there is one +#: namespace. `-41001` to `-41015` are allocated; this is the first free one. +#: +#: A distinct code earns its keep on `v0.3 §8.4`'s test: `-41001` means this action is not +#: permitted to anyone, `-41012` means not to **you**, and this means not against **that +#: server**, which a client answers differently from either. +UPSTREAM_UNPINNED: Final = (-41016, "ctrlrun.upstream_unpinned", 403) + #: §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. @@ -783,7 +792,15 @@ def _through_control( ), ) except ActionDenied as refused: - code, token, status = DENIED + # SPEC-v0.10 §4.5 — the two upstream reasons get their own code, and this branch is + # **inside** the `ActionDenied` clause rather than above it, because they are + # `ActionDenied` reasons and not a new exception type. `v0.3 §8.4`'s ordering hazard + # does not arise: one type, discriminated on the reason it carries. + code, token, status = ( + UPSTREAM_UNPINNED + if refused.reason in (UPSTREAM_MISMATCH, UPSTREAM_UNVERIFIED) + else DENIED + ) return _json( status, json_rpc_error( diff --git a/src/ctrlrun/policy.py b/src/ctrlrun/policy.py index b1a6a051..889d8ee1 100644 --- a/src/ctrlrun/policy.py +++ b/src/ctrlrun/policy.py @@ -14,6 +14,7 @@ import logging import operator import os +import re import unicodedata from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass, field @@ -64,6 +65,8 @@ #: version moves once, here, with item 1, and item 3 fills it under the version already in #: place: two branches racing a schema bump is how a catalogue ends up with a stub row. POLICY_SCHEMA_V7: Final = "ctrlrun.policy/v7" +#: SPEC-v0.10 §4.6 — `v8` adds one action-entry key, `upstream:`. Bumped once, by item 3. +POLICY_SCHEMA_V8: Final = "ctrlrun.policy/v8" #: All of them, newest last, for the message an unknown schema produces. **In version order**, #: which `_at_least` reads: a version added out of order would make every gate below lie. @@ -75,6 +78,7 @@ POLICY_SCHEMA_V5, POLICY_SCHEMA_V6, POLICY_SCHEMA_V7, + POLICY_SCHEMA_V8, ) @@ -185,6 +189,13 @@ def _at_least(schema: str, minimum: str) -> bool: #: them differently. POLICY_UNAPPROVED: Final = "policy_unapproved" +#: SPEC-v0.10 §4.5 — the two upstream refusals, separately observable because "the server +#: changed" and "nobody has checked" are different findings an operator fixes differently. +#: `UPSTREAM_UNVERIFIED` is the fail-closed half and the one to get right: a pin that does +#: nothing when nothing was observed is a pin an upstream can switch off by never being seen. +UPSTREAM_MISMATCH: Final = "upstream_mismatch" +UPSTREAM_UNVERIFIED: Final = "upstream_unverified" + _V6_ENTRY_KEYS: Final[Mapping[str, str]] = { "approvals_required": ( "an older reader would ignore the threshold and consume on the first grant, which is a " @@ -192,6 +203,17 @@ def _at_least(schema: str, minimum: str) -> bool: ), } +#: SPEC-v0.10 §4.6 — the action-entry key `ctrlrun.policy/v8` adds, gated in the shape +#: `_V4_ENTRY_KEYS` and `_V5_ENTRY_KEYS` use and **not** `require_v7`'s: that one walks +#: `authority.grants`, because `tasks:` and `budgets:` are grant keys, and a standalone +#: `--authority` document carries no action entries at all. +_V8_ENTRY_KEYS: Final[Mapping[str, str]] = { + "upstream": ( + "an older reader would ignore the pin and authorise the action against any server at " + "all, which is the whole of what the key restricts" + ), +} + _RULE_KEYS: Final = frozenset({"when", "decision", "controls"}) #: SPEC-v0.2 §3.1 — the keys `ctrlrun.policy/v2` adds to an action entry. The gateway has no @@ -202,11 +224,16 @@ def _at_least(schema: str, minimum: str) -> bool: | _V2_ENTRY_KEYS | frozenset(_V5_ENTRY_KEYS) | frozenset(_V6_ENTRY_KEYS) + | frozenset(_V8_ENTRY_KEYS) ) #: And the closed key set of the `mcp` mapping, which is one key wide. _MCP_KEYS: Final = frozenset({"not_executed_on_error"}) +#: SPEC-v0.10 §4.2 — the pin's closed key set, and the shape of a `sha256:` digest. +_UPSTREAM_KEYS: Final = frozenset({"tls_cert_sha256", "tls_cert_file", "tool_schema_sha256"}) +_SHA256: Final = re.compile(r"^sha256:[0-9a-f]{64}$") + #: Names of `Action` fields (SPEC-v0.1 §2.1), which a condition cannot address: conditions #: see the action's *arguments* and nothing else (§3.2). Writing one reads like it scopes a #: rule — `when: { environment_eq: production }` — and matches nothing, so it is refused at @@ -539,6 +566,36 @@ def _in_registry_order(cited: tuple[str, ...], order: tuple[str, ...]) -> tuple[ return tuple(known) + tuple(item for item in cited if item not in rank) +@dataclass(frozen=True) +class UpstreamPin: + """Which server an action entry authorises itself against (SPEC-v0.10 §4.2). + + **Not folded into `McpOptions`**, which is the closest existing name: that one carries + claims an operator makes about their upstream's *behaviour* (`not_executed_on_error` is a + `NotExecuted` hint), and this carries a claim about its *identity*, which is an + authorization input. Merging them would put a pin inside a structure whose documented job + is a classifier hint. + + **Two TLS keys, because a digest cannot be a trust anchor.** `certs` feeds §4.3's check 3, + where the pinned certificates become the connection's only trust anchors and a swapped + server fails the handshake; `SSLContext.load_verify_locations` takes PEM, and there is no + way to hand OpenSSL a hash and have it validate a chain. `cert_sha256` feeds checks 1 and 2, + which compare what was observed. An entry pinning by digest alone gets the first two checks + and not the third, which §4.2 states as a limit rather than leaving to be discovered. + + The two must agree: every certificate `certs` holds hashes to a digest `cert_sha256` names, + checked at load. A rotation that moved only one half would fail at the handshake on a day + an operator believed they had prepared for. + """ + + cert_sha256: tuple[str, ...] = () + certs: tuple[str, ...] = () + tool_schema_sha256: str | None = None + + def __bool__(self) -> bool: + return bool(self.cert_sha256 or self.certs or self.tool_schema_sha256) + + @dataclass(frozen=True) class McpOptions: """Per-tool assertions an operator makes about their upstream (SPEC-v0.2 §3.1, §6.8). @@ -562,6 +619,8 @@ class _ActionPolicy: effect: str | None = None resource: str | None = None mcp: McpOptions = _DEFAULT_MCP_OPTIONS + #: SPEC-v0.10 §4.2 — which upstream this entry authorises itself against, or an empty pin. + upstream: UpstreamPin = field(default_factory=UpstreamPin) #: §7.3 — the control ids this action cites, which govern every rule under it. controls: tuple[str, ...] = () #: §7.4 — which of this action's arguments carry which class of data. @@ -898,6 +957,25 @@ def data_scope(self, action: Action) -> frozenset[str]: entry = self.actions.get(action.name) return frozenset() if entry is None else entry.data_scope(action.canonical_arguments) + def upstream_pin(self, action_name: str) -> UpstreamPin: + """This action's upstream pin, or an empty one (SPEC-v0.10 §4.2). + + An empty pin is satisfied by anything, which is every action entry written before v0.10 + and why they all upgrade untouched. + """ + entry = self.actions.get(action_name) + return UpstreamPin() if entry is None else entry.upstream + + def tool_name(self, action_name: str) -> str | None: + """The upstream tool this action routes to, for §4.2's tool-schema pin. + + The action name **is** the tool name at the gateway (`v0.2 §6.6` builds the Action from + `params.name`), so this is the identity today and exists as a name rather than as an + inlined assumption: a deployment that ever mapped one to the other would change here and + nowhere else. + """ + return action_name if action_name in self.actions else None + def effect_template(self, action_name: str) -> str | None: """This action's `effect:` template, or `None` (SPEC-v0.2 §3.1, §11). @@ -1468,6 +1546,55 @@ def _reject_reserved_elsewhere(actions: Mapping[str, _ActionPolicy], source: str ) +def _parse_upstream(value: object, where: str) -> UpstreamPin: + """SPEC-v0.10 §4.2. Refuse what the pin cannot mean, at load, where an operator is present. + + A malformed pin is a `PolicyError` and never a pin that quietly matches nothing: a key whose + typo turns it off is the fail-open direction, and §4.5's whole point is that an unverified + upstream refuses rather than passes. + """ + if value is None: + return UpstreamPin() + if not isinstance(value, Mapping): + raise PolicyError(f"{where}: 'upstream' must be a mapping") + unknown = set(value) - _UPSTREAM_KEYS + if unknown: + raise PolicyError( + f"{where}: unknown 'upstream' key(s) {sorted(unknown)!r}; the pin's keys are " + f"{sorted(_UPSTREAM_KEYS)!r} (SPEC-v0.10 §4.2)" + ) + digests = value.get("tls_cert_sha256", []) + if isinstance(digests, str): + digests = [digests] + if not isinstance(digests, list) or not all(isinstance(item, str) for item in digests): + raise PolicyError( + f"{where}: 'upstream.tls_cert_sha256' must be a list of 'sha256:…' strings; it is a " + "LIST so an operator can carry the current and the next certificate across a " + "rotation without an outage (SPEC-v0.10 §4.2)" + ) + for digest in digests: + if not _SHA256.match(digest): + raise PolicyError( + f"{where}: 'upstream.tls_cert_sha256' entry {digest!r} is not 'sha256:' " + "followed by 64 hex characters" + ) + files = value.get("tls_cert_file", []) + if isinstance(files, str): + files = [files] + if not isinstance(files, list) or not all(isinstance(item, str) for item in files): + raise PolicyError(f"{where}: 'upstream.tls_cert_file' must be a path or a list of paths") + schema_hash = value.get("tool_schema_sha256") + if schema_hash is not None and ( + not isinstance(schema_hash, str) or not _SHA256.match(schema_hash) + ): + raise PolicyError( + f"{where}: 'upstream.tool_schema_sha256' must be 'sha256:' followed by 64 hex chars" + ) + return UpstreamPin( + cert_sha256=tuple(digests), certs=tuple(files), tool_schema_sha256=schema_hash + ) + + def _parse_entry( entry: object, where: str, @@ -1511,7 +1638,14 @@ def _parse_entry( f"{where}: {key!r}{_at_line(line_of(key))} needs 'schema: {POLICY_SCHEMA_V6}'; " f"this document declares {schema!r}, and {consequence}" ) + for key, consequence in _V8_ENTRY_KEYS.items(): + if key in entry and not _at_least(schema, POLICY_SCHEMA_V8): + raise PolicyError( + f"{where}: {key!r}{_at_line(line_of(key))} needs 'schema: {POLICY_SCHEMA_V8}'; " + f"this document declares {schema!r}, and {consequence}" + ) labels = _parse_data(entry.get("data"), where) + pin = _parse_upstream(entry.get("upstream"), where) ceiling = _parse_max_attempts(entry, where, line_of) required = _parse_approvals_required(entry, where, line_of) @@ -1522,6 +1656,7 @@ def _parse_entry( effect=effect, resource=resource, mcp=mcp, + upstream=pin, controls=cited, data=MappingProxyType(labels), max_attempts=ceiling, @@ -1539,6 +1674,7 @@ def _parse_entry( effect=effect, resource=resource, mcp=mcp, + upstream=pin, controls=cited, data=MappingProxyType(labels), max_attempts=ceiling, diff --git a/src/ctrlrun/upstream.py b/src/ctrlrun/upstream.py new file mode 100644 index 00000000..eccae2d9 --- /dev/null +++ b/src/ctrlrun/upstream.py @@ -0,0 +1,113 @@ +"""What was observed about an upstream, and whether it matches the pin. SPEC-v0.10 §4. + +**Per process, by construction.** §4.3's check 2 answers from what *this* process has seen, which +is why the register is a module-level mapping and not a store table: an observation is a fact about +a connection this process made, and a second process that has made none must refuse rather than +inherit somebody else's. That is `UPSTREAM_UNVERIFIED`, and it is the fail-closed half of §4.5. + +**The comparison is a pure function over two strings**, which is what lets `ctrlrun verify` grade +G27 with no TLS listener and no certificate to generate (§7): it seeds an observation and asserts +the refusal. +""" + +from __future__ import annotations + +import hashlib +import threading +from collections.abc import Mapping +from typing import Any, Final + +from .action import canonical_bytes +from .policy import UPSTREAM_MISMATCH, UPSTREAM_UNVERIFIED, UpstreamPin + +#: SPEC-v0.10 §4.2 — the tool-schema hash's domain tag. `canonical_bytes` is the one +#: canonicalizer (`v0.9 §5.5` uses it for the scope hash), and a domain tag is what stops a hash +#: over a tool's advertised entry ever equalling one over some other mapping with the same keys. +_TOOL_SCHEMA_DOMAIN: Final = "ctrlrun.upstream.tool_schema/v1" + +_LOCK: Final = threading.Lock() +#: upstream name -> the SHA-256 of the leaf certificate last observed for it. +_CERTS: dict[str, str] = {} +#: (upstream name, tool name) -> the hash of the tool's last advertised schema. +_TOOLS: dict[tuple[str, str], str] = {} + + +def tool_schema_hash(entry: Mapping[str, Any]) -> str: + """`"sha256:" + hex(SHA-256(canonical_bytes({domain, entry})))` (SPEC-v0.10 §4.2). + + Over the **whole** advertised entry: name, description and input schema together, because a + description that changed is a tool whose behaviour an operator has not reviewed. + """ + payload = canonical_bytes({"schema": _TOOL_SCHEMA_DOMAIN, "entry": entry}) + return "sha256:" + hashlib.sha256(payload).hexdigest() + + +def cert_hash(der: bytes) -> str: + """The digest §4.2 pins: SHA-256 over the leaf certificate's DER bytes.""" + return "sha256:" + hashlib.sha256(der).hexdigest() + + +def observe_certificate(upstream: str, der: bytes) -> str: + """Record the leaf certificate this process saw for `upstream`, and return its digest.""" + digest = cert_hash(der) + with _LOCK: + _CERTS[upstream] = digest + return digest + + +def observe_tool_schema(upstream: str, tool: str, entry: Mapping[str, Any]) -> str: + """Record the schema `upstream` advertised for `tool`, and return its hash.""" + digest = tool_schema_hash(entry) + with _LOCK: + _TOOLS[upstream, tool] = digest + return digest + + +def forget(upstream: str | None = None) -> None: + """Drop observations. Whole-register with no argument, which tests and `verify` use.""" + with _LOCK: + if upstream is None: + _CERTS.clear() + _TOOLS.clear() + return + _CERTS.pop(upstream, None) + for key in [key for key in _TOOLS if key[0] == upstream]: + del _TOOLS[key] + + +def check(pin: UpstreamPin, upstream: str, tool: str | None = None) -> str | None: + """§4.3's check 2: the reason this action is refused, or `None` where the pin is satisfied. + + **An empty pin is satisfied by anything**, which is every action entry written before v0.10 + and why they all upgrade untouched. + + **A pin with nothing observed is `upstream_unverified`, never admitted.** That is the row the + whole section turns on: an upstream that is never observed would otherwise switch the pin off + by being absent, and §4.5's fail-closed half exists to stop exactly that. + """ + if not pin: + return None + with _LOCK: + seen_cert = _CERTS.get(upstream) + seen_tool = _TOOLS.get((upstream, tool)) if tool is not None else None + if pin.cert_sha256: + if seen_cert is None: + return UPSTREAM_UNVERIFIED + if seen_cert not in pin.cert_sha256: + return UPSTREAM_MISMATCH + if pin.tool_schema_sha256 is not None: + if tool is None or seen_tool is None: + return UPSTREAM_UNVERIFIED + if seen_tool != pin.tool_schema_sha256: + return UPSTREAM_MISMATCH + return None + + +__all__ = [ + "cert_hash", + "check", + "forget", + "observe_certificate", + "observe_tool_schema", + "tool_schema_hash", +] diff --git a/src/ctrlrun/verify/guarantees.py b/src/ctrlrun/verify/guarantees.py index abf911cf..dc542de2 100644 --- a/src/ctrlrun/verify/guarantees.py +++ b/src/ctrlrun/verify/guarantees.py @@ -197,6 +197,16 @@ class Guarantee: "a hop is named on both sides", ("v0.10 §3.4", "v0.10 §3.6 T478", "v0.10 §3.6 T488"), ), + Guarantee( + "G27", + # 28 characters against `report._TITLE_WIDTH`'s 32. It grades §4.3's **check 2** and only + # check 2: that is the one producing a DENY, which is what this title promises. Check 3 + # refuses at the handshake and produces `NotExecuted` with the effect `FAILED`, a + # different outcome under a different name, and a scenario allowed to grade either would + # report PASS without anybody knowing which (SPEC-v0.10 §7). + "a swapped upstream is denied", + ("v0.10 §4.3", "v0.10 §4.7 T490", "v0.10 §4.7 T493"), + ), ) #: By id, for `--only` and for the report. Insertion order is catalogue order. @@ -291,6 +301,9 @@ class Guarantee: #: spelling. This one is added because a reason that was silently unreachable would be the false #: `N/A` §7 opens by forbidding. NO_HOP_ACTION: Final = "the document's delegable grant admits no action verify can drive" +#: SPEC-v0.10 §7, G27. A statement about the operator's **document**: whether any action entry +#: declares an `upstream:` pin at all. +NO_UPSTREAM_PIN: Final = "no action entry pins an upstream" #: SPEC-v0.9 §8, G22. A statement about the operator's **document**: whether any grant it declares #: carries a budget at all. NO_BUDGET: Final = "no grant carries a budget" @@ -448,6 +461,7 @@ class Guarantee: "NO_METRIC_TO_MEASURE", "NO_RESOURCE_TO_SCOPE", "NO_TASKS", + "NO_UPSTREAM_PIN", "PER_CONNECTION_BACKEND", "PROCESSES", "SCOPE_PROVIDER_NOTE", diff --git a/src/ctrlrun/verify/scenarios.py b/src/ctrlrun/verify/scenarios.py index bd91abf3..5f9ff9a8 100644 --- a/src/ctrlrun/verify/scenarios.py +++ b/src/ctrlrun/verify/scenarios.py @@ -96,6 +96,8 @@ from ..identity import IdentityContext from ..policy import ( POLICY_CHANGE_ACTION, + UPSTREAM_MISMATCH, + UPSTREAM_UNVERIFIED, Condition, Decision, Policy, @@ -778,6 +780,9 @@ def select( needs_threshold: bool = False, ceiling_bound: int | None = None, grant_filter: Callable[[Grant], bool] | None = None, + #: SPEC-v0.10 §7, G27: the action must be one the document pins an upstream for, which is + #: a property of the action rather than of a grant, so `grant_filter` cannot express it. + action_filter: Callable[[str], bool] | None = None, mutation: Mapping[str, Any] | None = None, ) -> _Selection | None: """§3.2 — the first action, sorted by codepoint, that satisfies the requirements. @@ -797,6 +802,8 @@ def select( self._budget_miss = None self._metric_miss = None for name in sorted(self.policy.actions): + if action_filter is not None and not action_filter(name): + continue if needs_effect and self.policy.effect_template(name) is None: continue ceiling = self.policy.max_attempts(name) @@ -4461,6 +4468,103 @@ def hand_on() -> Any: finally: store.close() + def g27(self) -> GuaranteeResult: + """SPEC-v0.10 §4.3, §7. A swapped upstream is denied, at check 2. + + **Check 2 and only check 2**, which is the one that produces a `DENY`. Check 3 refuses at + the handshake and produces `NotExecuted` with the effect `FAILED`; a scenario allowed to + grade either would report `PASS` for a guarantee whose title promises a denial that never + happened. + + **And check 2 is the only one verify can grade without a network.** The comparison is a + pure function over two strings, so verify seeds an observation and asserts the refusal, + with no TLS listener to stand up and no certificate to generate. §8.1 made the same move + for G23's scope provider and said why: a guarantee about a code surface is graded against + a scenario verify constructs, rather than reporting `N/A` about something it never saw. + """ + from .. import upstream as _upstream + + pinned = [name for name in sorted(self.policy.actions) if self.policy.upstream_pin(name)] + if not pinned: + return self.na("G27", reg.NO_UPSTREAM_PIN) + selection = self.select(action_filter=lambda name: name in pinned) + if selection is None: + return self.na("G27", reg.NO_GRANT_MATCHES) + pin = self.policy.upstream_pin(selection.action) + if not pin.cert_sha256: + return self.na("G27", reg.NO_UPSTREAM_PIN) + control, store, recorder, _ = self._control_for("G27", selection) + control._upstream = f"{reg.SYNTHETIC_PREFIX}-upstream" + + def body(detail: dict[str, Any]) -> None: + detail["action"] = selection.action + detail["pinned"] = list(pin.cert_sha256) + here = f"{reg.SYNTHETIC_PREFIX}-upstream" + + # The positive control first: the pinned certificate admits the action. Without it a + # kernel that refused every pinned action whatever would grade PASS, which is what + # `v0.4 §2.2` means by a guarantee that could not have failed. + _upstream.forget(here) + _upstream._CERTS[here] = pin.cert_sha256[0] + executor = _Executor() + receipt = self.execute( + control, + selection.build(), + executor, + selection.effect_key, + self.approve(control, store, selection.build(), selection), + ) + _expect_control( + receipt.result is ReceiptResult.COMMITTED and executor.calls == 1, + "against the pinned certificate the action runs", + f"it ended {receipt.result} after {executor.calls} executor calls", + ) + + # A swapped server: a different certificate behind the same name. + _upstream._CERTS[here] = "sha256:" + "f" * 64 + swapped_executor = _Executor() + refusal = self.refused( + lambda: self.execute( + control, selection.build(), swapped_executor, selection.effect_key, None + ), + (ActionDenied,), + "ActionDenied(reason='upstream_mismatch')", + "an action pinned to one server ran against another", + ) + _expect( + getattr(refusal, "reason", "") == UPSTREAM_MISMATCH, + "ActionDenied(reason='upstream_mismatch')", + f"ActionDenied(reason={getattr(refusal, 'reason', '')!r})", + ) + _expect( + swapped_executor.calls == 0, + "the upstream is never called", + f"the executor ran {swapped_executor.calls} times", + ) + + # And nothing observed at all, which is the fail-closed half (§4.5). + _upstream.forget(here) + unverified_executor = _Executor() + missing = self.refused( + lambda: self.execute( + control, selection.build(), unverified_executor, selection.effect_key, None + ), + (ActionDenied,), + "ActionDenied(reason='upstream_unverified')", + "an action pinned to a server nothing has verified ran anyway", + ) + _expect( + getattr(missing, "reason", "") == UPSTREAM_UNVERIFIED, + "ActionDenied(reason='upstream_unverified')", + f"ActionDenied(reason={getattr(missing, 'reason', '')!r})", + ) + + try: + return self.graded("G27", selection, store, recorder, body) + finally: + _upstream.forget(f"{reg.SYNTHETIC_PREFIX}-upstream") + store.close() + #: SPEC-v0.8 §3.4, §11.7 — the claim verify's own approver principals carry their roles in. #: Named for what it is, and `SYNTHETIC_PREFIX`ed nowhere, because it is a claim **name** and a diff --git a/tests/test_upstream_pinning.py b/tests/test_upstream_pinning.py new file mode 100644 index 00000000..312a135a --- /dev/null +++ b/tests/test_upstream_pinning.py @@ -0,0 +1,371 @@ +"""SPEC-v0.10 §4, item 3: upstream identity pinning. + +The honest slice of `ASI04` and nothing more: this decides actions, and it never inspects a +package, a model, a registry or a build. What it adds is that an action entry may say **which +server** it authorises itself against. + +Three checks, one rule (§4.3). These tests cover check 2, the one that produces a `DENY`, and +check 3's mechanism against a real TLS listener. Check 1 is the gateway's startup refusal. +""" + +from __future__ import annotations + +import http.server +import socket +import ssl +import subprocess +import threading +from pathlib import Path + +import pytest + +from ctrlrun.action import Action, Principal +from ctrlrun.control import Control +from ctrlrun.errors import ActionDenied, InvalidArgument, PolicyError +from ctrlrun.policy import UPSTREAM_MISMATCH, UPSTREAM_UNVERIFIED, Policy +from ctrlrun.state import SQLiteStateStore +from ctrlrun.upstream import ( + cert_hash, + check, + forget, + observe_certificate, + observe_tool_schema, + tool_schema_hash, +) + +DIGEST_A = "sha256:" + "a" * 64 +DIGEST_B = "sha256:" + "b" * 64 + +DOC = """ +schema: ctrlrun.policy/v8 +actions: + stripe.refund: + decision: allow + upstream: + tls_cert_sha256: {pins} +""" + + +def _policy(pins: str = f'["{DIGEST_A}"]') -> Policy: + return Policy.from_yaml(DOC.format(pins=pins), source="test_upstream") + + +def _action() -> Action: + return Action( + name="stripe.refund", + resource=None, + arguments={"amount": 10}, + principal=Principal(agent="worker"), + environment="production", + ) + + +@pytest.fixture(autouse=True) +def _clean_register(): + forget() + yield + forget() + + +# --- T489, T490, T492, T493: check 2 ------------------------------------------------------- + + +def test_T489_the_pinned_upstream_admits_the_action(tmp_path): + """The negative control for every row below. Without it a kernel that refused every pinned + action whatever would pass them all, which is `v0.4 §2.2`'s guarantee that could not fail.""" + store = SQLiteStateStore(str(tmp_path / "s.db")) + control = Control(_policy(), store, upstream="mcp.example") + observe_certificate("mcp.example", b"the-pinned-cert") + pinned = _policy(f'["{cert_hash(b"the-pinned-cert")}"]') + control = Control(pinned, store, upstream="mcp.example") + + receipt = control.execute(_action(), lambda: "ok") + + assert receipt.result.value == "committed" + + +def test_T490_a_swapped_server_behind_the_same_name_is_refused(tmp_path): + """§4.3's check 2, and the sharp case of §4.1: everything else still matches. The grant + matches, the constraints hold, the receipt would still say `stripe.refund`.""" + store = SQLiteStateStore(str(tmp_path / "s.db")) + pinned = _policy(f'["{cert_hash(b"the-pinned-cert")}"]') + control = Control(pinned, store, upstream="mcp.example") + observe_certificate("mcp.example", b"a-different-cert") + ran = [] + + with pytest.raises(ActionDenied) as refused: + control.execute(_action(), lambda: ran.append(1)) + + assert refused.value.reason == UPSTREAM_MISMATCH + assert ran == [], "the upstream was called for an action pinned to another server" + + +def test_T493_a_pin_with_nothing_observed_is_refused_and_never_admitted(tmp_path): + """**The fail-closed half, and the one to get right** (§4.5). A pin that does nothing when + nothing was observed is a pin an upstream can switch off by never being seen.""" + store = SQLiteStateStore(str(tmp_path / "s.db")) + control = Control(_policy(), store, upstream="mcp.example") + ran = [] + + with pytest.raises(ActionDenied) as refused: + control.execute(_action(), lambda: ran.append(1)) + + assert refused.value.reason == UPSTREAM_UNVERIFIED + assert ran == [] + + +def test_T492_rotation_admits_either_certificate_and_refuses_a_third(tmp_path): + """§4.2. The key is a **list** so an operator can carry the current and the next certificate + across a rotation; a single-valued pin makes every renewal an outage, which is how a pin gets + switched off permanently.""" + store = SQLiteStateStore(str(tmp_path / "s.db")) + current, following = cert_hash(b"current"), cert_hash(b"next") + control = Control(_policy(f'["{current}", "{following}"]'), store, upstream="mcp.example") + + for blob in (b"current", b"next"): + observe_certificate("mcp.example", blob) + assert control.execute(_action(), lambda: "ok").result.value == "committed" + + observe_certificate("mcp.example", b"a-third") + with pytest.raises(ActionDenied) as refused: + control.execute(_action(), lambda: "ok") + assert refused.value.reason == UPSTREAM_MISMATCH + + +def test_T489b_an_entry_that_pins_nothing_is_unchanged(tmp_path): + """Opt in, then fail closed. Every action entry written before v0.10 pins nothing, and an + empty pin is satisfied by anything, which is why they all upgrade untouched.""" + store = SQLiteStateStore(str(tmp_path / "s.db")) + plain = Policy.from_yaml( + "schema: ctrlrun.policy/v8\nactions:\n stripe.refund:\n decision: allow\n", + source="t", + ) + control = Control(plain, store, upstream="mcp.example") + + assert control.execute(_action(), lambda: "ok").result.value == "committed" + + +# --- T491: check 3, against a real listener ------------------------------------------------ + + +def _ca_signed(tmp: Path, name: str) -> tuple[Path, Path]: + """A tiny CA and one leaf it signs, so the leaf is NOT self-signed: the realistic shape, and + the one that needs `VERIFY_X509_PARTIAL_CHAIN` to be a trust anchor at all.""" + ca_key, ca_crt = tmp / f"{name}-ca.key", tmp / f"{name}-ca.crt" + subprocess.run( + [ + "openssl", + "req", + "-x509", + "-newkey", + "rsa:2048", + "-nodes", + "-keyout", + str(ca_key), + "-out", + str(ca_crt), + "-days", + "1", + "-subj", + f"/CN={name}-ca", + ], + check=True, + capture_output=True, + ) + key, csr, crt = tmp / f"{name}.key", tmp / f"{name}.csr", tmp / f"{name}.crt" + subprocess.run( + [ + "openssl", + "req", + "-newkey", + "rsa:2048", + "-nodes", + "-keyout", + str(key), + "-out", + str(csr), + "-subj", + "/CN=localhost", + ], + check=True, + capture_output=True, + ) + ext = tmp / f"{name}.ext" + ext.write_text("subjectAltName=DNS:localhost\n") + subprocess.run( + [ + "openssl", + "x509", + "-req", + "-in", + str(csr), + "-CA", + str(ca_crt), + "-CAkey", + str(ca_key), + "-CAcreateserial", + "-out", + str(crt), + "-days", + "1", + "-extfile", + str(ext), + ], + check=True, + capture_output=True, + ) + return key, crt + + +def _serve(key: Path, crt: Path) -> int: + """A TLS listener on a port the OS picks **and we never let go of**. + + Binding an ephemeral port, closing it, and rebinding is a race: another process can take the + port in the gap, and this file stands up two servers so it runs the gap twice. One flaky gate + run is what found it. `HTTPServer` binds for us and `server_port` reports what it got, so + there is no gap to lose. + """ + + class Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self) -> None: + self.send_response(200) + self.send_header("Content-Length", "2") + self.end_headers() + self.wfile.write(b"ok") + + def log_message(self, *args: object) -> None: + pass + + server = http.server.HTTPServer(("127.0.0.1", 0), Handler) + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(str(crt), str(key)) + server.socket = context.wrap_socket(server.socket, server_side=True) + threading.Thread(target=server.serve_forever, daemon=True).start() + return int(server.server_port) + + +@pytest.mark.serial +def test_T491_the_pinned_certificate_is_the_connections_only_trust_anchor(tmp_path): + """§4.3's check 3, the one that **prevents** rather than attributing. + + A digest cannot be a trust anchor, which is why §4.2 carries `tls_cert_file` beside + `tls_cert_sha256`: `load_verify_locations` takes PEM. A CA-signed leaf becomes a valid anchor + with `VERIFY_X509_PARTIAL_CHAIN`, and a swapped server then fails the handshake **before any + request byte**, which is what makes `SPEC-v0.7 §2.3`'s `NotExecuted` claim true of it. + """ + good_key, good_crt = _ca_signed(tmp_path, "good") + evil_key, evil_crt = _ca_signed(tmp_path, "evil") + good_port, evil_port = _serve(good_key, good_crt), _serve(evil_key, evil_crt) + + def context() -> ssl.SSLContext: + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.load_verify_locations(cadata=good_crt.read_text()) + ctx.verify_flags |= ssl.VERIFY_X509_PARTIAL_CHAIN + return ctx + + def reach(port: int) -> str: + try: + with ( + socket.create_connection(("127.0.0.1", port), timeout=5) as raw, + context().wrap_socket(raw, server_hostname="localhost") as tls, + ): + tls.send(b"GET / HTTP/1.0\r\nHost: localhost\r\n\r\n") + tls.recv(16) + return "handshake ok" + except ssl.SSLCertVerificationError: + return "refused" + + assert reach(good_port) == "handshake ok", "the pinned server was refused" + assert reach(evil_port) == "refused", "a swapped server completed the handshake" + + +# --- T494, T495, T496, T497: the key, the gate, the surfaces ------------------------------- + + +def test_T494_the_tool_schema_hash_covers_the_whole_advertised_entry(): + """§4.2. Name, description and input schema together, because a description that changed is + a tool whose behaviour an operator has not reviewed.""" + entry = {"name": "refund", "description": "issue a refund", "inputSchema": {"type": "object"}} + moved = {**entry, "description": "issue a refund, or a payout"} + + assert tool_schema_hash(entry) == tool_schema_hash(dict(reversed(list(entry.items())))) + assert tool_schema_hash(entry) != tool_schema_hash(moved) + + +def test_T494b_a_tool_whose_schema_moved_under_an_approved_name_is_refused(tmp_path): + """The second half of §4.1's sharp case: the server did not change, the tool's schema did.""" + entry = {"name": "refund", "inputSchema": {"type": "object"}} + pinned = Policy.from_yaml( + "schema: ctrlrun.policy/v8\nactions:\n stripe.refund:\n decision: allow\n" + f' upstream: {{ tool_schema_sha256: "{tool_schema_hash(entry)}" }}\n', + source="t", + ) + store = SQLiteStateStore(str(tmp_path / "s.db")) + control = Control(pinned, store, upstream="mcp.example") + + observe_tool_schema("mcp.example", "stripe.refund", entry) + assert control.execute(_action(), lambda: "ok").result.value == "committed" + + observe_tool_schema("mcp.example", "stripe.refund", {**entry, "inputSchema": {"type": "array"}}) + with pytest.raises(ActionDenied) as refused: + control.execute(_action(), lambda: "ok") + assert refused.value.reason == UPSTREAM_MISMATCH + + +def test_T496_upstream_in_a_v7_document_is_a_load_error(): + """§4.6. An older reader that ignored the key would authorise the action against any server + at all, which is the whole of what the key restricts.""" + with pytest.raises(PolicyError) as refused: + Policy.from_yaml( + "schema: ctrlrun.policy/v7\nactions:\n stripe.refund:\n decision: allow\n" + f' upstream: {{ tls_cert_sha256: ["{DIGEST_A}"] }}\n', + source="t", + ) + + assert "ctrlrun.policy/v8" in str(refused.value) + assert "upstream" in str(refused.value) + + +def test_T497_the_acs_hook_refuses_a_pin_at_construction(tmp_path): + """§4.4. ACS is advisory: the platform runs the tool and this hook holds no connection, so + there is nothing to observe and nothing to pin. + + **At construction and not at load**, which §4.4 argues at length: one loader cannot know + which surface will run an action, and a load error would stop `verify` and `scan` reading a + document that pins, which §7.3's exit criterion requires them to do. + """ + pytest.importorskip("httpx") + from ctrlrun.acs import AcsControlHook + + store = SQLiteStateStore(str(tmp_path / "s.db")) + control = Control(_policy(), store) + + with pytest.raises(InvalidArgument) as refused: + AcsControlHook(control) + + assert "upstream" in str(refused.value) + assert "ctrlrun gateway" in str(refused.value) + + +def test_T497b_a_pinned_document_still_loads_in_process(tmp_path): + """The other half of §4.4, and the reason the refusal is not a load error: `verify` and + `scan` load through the in-process path, so a document that pins must **read**. What refuses + is the action, at decision time, under `upstream_unverified`.""" + store = SQLiteStateStore(str(tmp_path / "s.db")) + control = Control(_policy(), store) # no upstream: in-process + + assert control.policy.upstream_pin("stripe.refund").cert_sha256 == (DIGEST_A,) + with pytest.raises(ActionDenied) as refused: + control.execute(_action(), lambda: "ok") + assert refused.value.reason == UPSTREAM_UNVERIFIED + + +def test_the_check_is_a_pure_function_over_two_strings(): + """Which is what lets `verify` grade G27 with no TLS listener and no certificate (§7).""" + from ctrlrun.policy import UpstreamPin + + assert check(UpstreamPin(), "anything") is None + assert check(UpstreamPin(cert_sha256=(DIGEST_A,)), "u") == UPSTREAM_UNVERIFIED + observe_certificate("u", b"x") + assert check(UpstreamPin(cert_sha256=(cert_hash(b"x"),)), "u") is None + assert check(UpstreamPin(cert_sha256=(DIGEST_B,)), "u") == UPSTREAM_MISMATCH diff --git a/tests/test_verify_action.py b/tests/test_verify_action.py index 33f76f9c..41396eae 100644 --- a/tests/test_verify_action.py +++ b/tests/test_verify_action.py @@ -139,8 +139,8 @@ def test_T118_ci_asserts_the_two_shapes_the_specification_names(): assert 'test "$AUTHORITY" = "verified 24/24"' in script assert 'test "$TEMPLATES" = "verified 11/11"' in script - assert 'test "$AUTHORITY_NA" = "2"' in script - assert 'test "$TEMPLATES_NA" = "15"' in script + assert 'test "$AUTHORITY_NA" = "3"' in script + assert 'test "$TEMPLATES_NA" = "16"' in script @pytest.mark.authority @@ -158,8 +158,11 @@ def test_T118_the_two_configurations_really_do_report_those_shapes(): # when a shape changes) while the N/A counts above are derived. assert authority.badge["message"] == "verified 24/24" # G13 and G15: SQLite has no clock of its own to diverge from, and the document declares - # no `max_attempts` (SPEC-v0.7 §8.9). - assert authority.not_applicable == 2 + # no `max_attempts` (SPEC-v0.7 §8.9). **And G27**, because no action entry in this document + # pins an upstream: SPEC-v0.10 §7.3's exit criterion wants a shipped example that does, and + # §4.4 makes a pinned action refuse on every in-process call, so the example that satisfies + # it demonstrates the refusal rather than a working call. That example is the release item's. + assert authority.not_applicable == 3 assert templates.badge is not None assert templates.badge["message"] == "verified 11/11" assert templates.applicable + templates.not_applicable == len(reg.GUARANTEES) From 516372adeb2b014a110743d5e113fdd9264dda53 Mon Sep 17 00:00:00 2001 From: arpan Date: Sun, 13 Sep 2026 23:06:43 +0530 Subject: [PATCH 2/2] Wire check 3, and answer three CodeQL alerts CodeQL reported two HIGH security alerts on this PR, both in my own test file: ssl.SSLContext(PROTOCOL_TLS_SERVER) and PROTOCOL_TLS_CLIENT still admit TLS 1.0 and 1.1 unless a floor is set. A listener in a test for a pinning feature that negotiates a protocol the product would refuse is testing something the product does not do, so both contexts now set minimum_version explicitly. The third alert was a variable assigned twice in T489, which was sloppiness. Production had no exposure: jwt_identity and revocation both use ssl.create_default_context(), which floors at TLS 1.2. But checking that turned up a real gap in my own scope claim. The PR body said T491 proves check 3, and it proved the MECHANISM while nothing wired it: the gateway's forwarder built its client with httpx's ordinary verification whatever the policy pinned. Check 3 is the only one of the three that PREVENTS rather than attributing, so shipping section 4 with it unimplemented would have overclaimed the section. upstream.pinned_context builds the context: PARTIAL_CHAIN, because a real upstream's leaf is CA-signed and OpenSSL wants a chain terminating at a self-signed certificate unless told a trusted non-root may end it, and without the flag a pinned leaf refuses every connection including the right one. Plus the TLS floor the alert above taught me to set, here rather than inherited. httpx_forwarder takes the policy and builds one where any entry pins a certificate file. T491b asserts the wiring, the floor and the flag. Gate with Postgres: 4412 passed, 0 skipped. Signed-off-by: arpan --- src/ctrlrun/gateway/__init__.py | 2 +- src/ctrlrun/gateway/server.py | 24 +++++++++++++++++---- src/ctrlrun/gateway/transport.py | 19 +++++++++++++--- src/ctrlrun/upstream.py | 34 ++++++++++++++++++++++++++++- tests/test_upstream_pinning.py | 37 +++++++++++++++++++++++++++++++- 5 files changed, 106 insertions(+), 10 deletions(-) diff --git a/src/ctrlrun/gateway/__init__.py b/src/ctrlrun/gateway/__init__.py index 8a33c5d6..bbd90ddc 100644 --- a/src/ctrlrun/gateway/__init__.py +++ b/src/ctrlrun/gateway/__init__.py @@ -107,7 +107,7 @@ def serve(*, upstream: str, alias: str, **options: Any) -> None: # and §4.4 refuses a pinned action there. upstream=config.upstream, ) - forwarder = httpx_forwarder(config) + forwarder = httpx_forwarder(config, control.policy) gateway = Gateway(config, control, forwarder) _announce(control, config, gateway.identity, authority_path) try: diff --git a/src/ctrlrun/gateway/server.py b/src/ctrlrun/gateway/server.py index 1b823e11..0d75e011 100644 --- a/src/ctrlrun/gateway/server.py +++ b/src/ctrlrun/gateway/server.py @@ -48,7 +48,7 @@ IdentityProvider, StaticIdentityProvider, ) -from ..policy import OBSERVE, UPSTREAM_MISMATCH, UPSTREAM_UNVERIFIED +from ..policy import OBSERVE, UPSTREAM_MISMATCH, UPSTREAM_UNVERIFIED, Policy from ..receipt import Receipt from .mcp import ( ACCEPTED_REVISIONS, @@ -1095,12 +1095,28 @@ def _request_id(body: bytes) -> JsonRpcId: # --- the transport ---------------------------------------------------------------------- -def httpx_forwarder(config: GatewayConfig) -> Any: - """Forward HTTP and SSE, using a fresh connection for every intercepted action.""" +def httpx_forwarder(config: GatewayConfig, policy: Policy | None = None) -> Any: + """Forward HTTP and SSE, using a fresh connection for every intercepted action. + + **SPEC-v0.10 §4.3's check 3**, where `policy` is given and any entry pins a certificate file: + every pinned certificate becomes a trust anchor for this gateway's one outbound connection, + so a swapped server fails the handshake **before the first request byte**. That is the only + one of §4.3's three checks that prevents rather than attributing, and it needs nothing new to + make `v0.7 §2.3`'s `NotExecuted` claim true of it. + + An entry pinning by digest alone contributes nothing here and gets checks 1 and 2 only, which + §4.2 states as a limit rather than leaving to be discovered. + """ + from ..upstream import pinned_context from . import http_client from .transport import HTTPForwarder - return HTTPForwarder(config.upstream, config.upstream_timeout, http_client()) + pinned: set[str] = set() + if policy is not None: + for name in policy.actions: + pinned.update(policy.upstream_pin(name).certs) + verify = pinned_context(tuple(sorted(pinned))) if pinned else None + return HTTPForwarder(config.upstream, config.upstream_timeout, http_client(), verify) # --- the listening side (stdlib, per §6.11) --------------------------------------------- diff --git a/src/ctrlrun/gateway/transport.py b/src/ctrlrun/gateway/transport.py index 399f2bc7..a740b211 100644 --- a/src/ctrlrun/gateway/transport.py +++ b/src/ctrlrun/gateway/transport.py @@ -297,11 +297,24 @@ def _observe(document: Any, expected_id: Any, revision: str) -> Observed: class HTTPForwarder: - def __init__(self, upstream: str, timeout: float, httpx: Any) -> None: + def __init__( + self, upstream: str, timeout: float, httpx: Any, verify: Any | None = None + ) -> None: self.upstream = upstream self.timeout = timeout self.httpx = httpx - self.pooled = httpx.Client(timeout=timeout) + #: SPEC-v0.10 §4.3's check 3, and the only one of the three that **prevents** rather than + #: attributing. Where an operator pinned certificates, this is an `ssl.SSLContext` whose + #: only trust anchors are those certificates, so a swapped server fails the handshake + #: **before the first request byte**, which is what makes `v0.7 §2.3`'s `NotExecuted` + #: claim true of it. `None` is httpx's ordinary verification, unchanged. + self.verify = verify + self.pooled = self._client() + + def _client(self) -> Any: + if self.verify is None: + return self.httpx.Client(timeout=self.timeout) + return self.httpx.Client(timeout=self.timeout, verify=self.verify) def close(self) -> None: self.pooled.close() @@ -322,7 +335,7 @@ def request( # may clear the environment while this one is in flight (§12.2.14). proxied = _through_a_proxy() run = _EXECUTOR_RUN.get() - client = self.httpx.Client(timeout=self.timeout) if owned else self.pooled + client = self._client() if owned else self.pooled try: with client.stream(method, self.upstream, content=body, headers=relayed) as response: if run is not None: diff --git a/src/ctrlrun/upstream.py b/src/ctrlrun/upstream.py index eccae2d9..d5661a93 100644 --- a/src/ctrlrun/upstream.py +++ b/src/ctrlrun/upstream.py @@ -15,7 +15,10 @@ import hashlib import threading from collections.abc import Mapping -from typing import Any, Final +from typing import TYPE_CHECKING, Any, Final + +if TYPE_CHECKING: # `ssl` is stdlib but not needed unless a deployment pins (SPEC-v0.10 §4.3). + import ssl from .action import canonical_bytes from .policy import UPSTREAM_MISMATCH, UPSTREAM_UNVERIFIED, UpstreamPin @@ -103,11 +106,40 @@ def check(pin: UpstreamPin, upstream: str, tool: str | None = None) -> str | Non return None +def pinned_context(certs: tuple[str, ...]) -> ssl.SSLContext: + """An `ssl.SSLContext` whose only trust anchors are the pinned certificates (§4.3, check 3). + + **The check that prevents.** Checks 1 and 2 compare an observation, which is a decision about + the past; this one refuses the handshake, so a swapped server never receives a request byte + and `v0.7 §2.3`'s `NotExecuted` claim is true of it without anything new. + + `VERIFY_X509_PARTIAL_CHAIN` is what makes a **leaf** a valid anchor. A real upstream's leaf is + signed by a CA, so loading it into the trust store is not enough on its own: OpenSSL wants the + chain to terminate at a self-signed certificate unless told that a trusted non-root may end + it. Without the flag a pinned CA-signed leaf refuses every connection, including the right + one. + + **A TLS floor, set explicitly.** `PROTOCOL_TLS_CLIENT` still admits TLS 1.0 and 1.1, and + CodeQL flags that high on a test file of this feature's own. A context built for a pinning + check that then negotiates a protocol the rest of the product would not is the wrong shape to + ship from a security library, so the floor is here rather than inherited. + """ + import ssl + + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context.minimum_version = ssl.TLSVersion.TLSv1_2 + context.verify_flags |= ssl.VERIFY_X509_PARTIAL_CHAIN + for path in certs: + context.load_verify_locations(cafile=path) + return context + + __all__ = [ "cert_hash", "check", "forget", "observe_certificate", "observe_tool_schema", + "pinned_context", "tool_schema_hash", ] diff --git a/tests/test_upstream_pinning.py b/tests/test_upstream_pinning.py index 312a135a..193e7c34 100644 --- a/tests/test_upstream_pinning.py +++ b/tests/test_upstream_pinning.py @@ -74,7 +74,6 @@ def test_T489_the_pinned_upstream_admits_the_action(tmp_path): """The negative control for every row below. Without it a kernel that refused every pinned action whatever would pass them all, which is `v0.4 §2.2`'s guarantee that could not fail.""" store = SQLiteStateStore(str(tmp_path / "s.db")) - control = Control(_policy(), store, upstream="mcp.example") observe_certificate("mcp.example", b"the-pinned-cert") pinned = _policy(f'["{cert_hash(b"the-pinned-cert")}"]') control = Control(pinned, store, upstream="mcp.example") @@ -238,6 +237,10 @@ def log_message(self, *args: object) -> None: server = http.server.HTTPServer(("127.0.0.1", 0), Handler) context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + # `PROTOCOL_TLS_SERVER` still admits TLS 1.0 and 1.1, which CodeQL flags high and is right + # to: a listener in a test for a *pinning* feature that negotiates a protocol the product + # would refuse is testing something the product does not do. + context.minimum_version = ssl.TLSVersion.TLSv1_2 context.load_cert_chain(str(crt), str(key)) server.socket = context.wrap_socket(server.socket, server_side=True) threading.Thread(target=server.serve_forever, daemon=True).start() @@ -259,6 +262,7 @@ def test_T491_the_pinned_certificate_is_the_connections_only_trust_anchor(tmp_pa def context() -> ssl.SSLContext: ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.minimum_version = ssl.TLSVersion.TLSv1_2 ctx.load_verify_locations(cadata=good_crt.read_text()) ctx.verify_flags |= ssl.VERIFY_X509_PARTIAL_CHAIN return ctx @@ -369,3 +373,34 @@ def test_the_check_is_a_pure_function_over_two_strings(): observe_certificate("u", b"x") assert check(UpstreamPin(cert_sha256=(cert_hash(b"x"),)), "u") is None assert check(UpstreamPin(cert_sha256=(DIGEST_B,)), "u") == UPSTREAM_MISMATCH + + +def test_T491b_the_gateways_forwarder_pins_when_the_policy_does(tmp_path): + """§4.3's check 3, wired: the forwarder's verification context is built from the certificates + the policy pins, and is httpx's ordinary verification where it pins none.""" + pytest.importorskip("httpx") + from ctrlrun.gateway.server import GatewayConfig, httpx_forwarder + + _, crt = _ca_signed(tmp_path, "pinned") + config = GatewayConfig(upstream="https://mcp.example", alias="x", principal="worker") + + plain = httpx_forwarder(config, _policy()) + assert plain.verify is None, ( + "a pin by digest alone contributes no trust anchor; §4.2 states that as a limit" + ) + + with_file = Policy.from_yaml( + "schema: ctrlrun.policy/v8\nactions:\n stripe.refund:\n decision: allow\n" + f' upstream: {{ tls_cert_file: "{crt}" }}\n', + source="t", + ) + pinning = httpx_forwarder(config, with_file) + assert pinning.verify is not None + assert pinning.verify.minimum_version is ssl.TLSVersion.TLSv1_2, ( + "a context built for a pinning check must not negotiate a protocol the product would " + "refuse; CodeQL flagged exactly this, high, on this file's own listener" + ) + assert pinning.verify.verify_flags & ssl.VERIFY_X509_PARTIAL_CHAIN, ( + "without PARTIAL_CHAIN a pinned CA-signed leaf refuses every connection, the right one " + "included" + )