Item 3: upstream identity pinning

.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

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

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**.
 

src/ctrlrun/gateway/__init__.py

@@ -102,8 +102,12 @@ 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)
+    forwarder = httpx_forwarder(config, control.policy)
     gateway = Gateway(config, control, forwarder)
     _announce(control, config, gateway.identity, authority_path)
     try:

src/ctrlrun/gateway/server.py

@@ -48,7 +48,7 @@
     IdentityProvider,
     StaticIdentityProvider,
 )
-from ..policy import OBSERVE
+from ..policy import OBSERVE, UPSTREAM_MISMATCH, UPSTREAM_UNVERIFIED, Policy
 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(
@@ -1078,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) ---------------------------------------------

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: