feat: enforce restricted trust-domain directory decisions

README.md

@@ -160,6 +160,17 @@ Set `ClientConfig(consumer_auth_mode="required")` when a workload must not
 silently fall back to anonymous dispatch if consumer-token acquisition fails.
 The default remains `"optional"`; `"disabled"` skips token acquisition.
 
+### Restricted trust-domain directories
+
+Restricted operation is opt-in and fail closed. Supply a
+`RestrictedDirectoryContext` whose membership credential is referenced through
+an environment variable or an owner-only regular file. The client sends it only
+to the configured directory, refuses redirects and legacy discovery fallback,
+and accepts candidates or tokens only when the directory returns a matching,
+current operation decision. Public behavior is unchanged when the context is
+absent. This client does not expose peer-gossip or CIP-worker participation, so
+those capabilities are not implied by restricted directory support.
+
 ### Experimental local candidate rankers
 
 Library users may attach a `CandidateRanker` to test a learned or
@@ -672,6 +683,13 @@ distributed store. None of these APIs is used by normal discovery, submission
 or node serving. They are unstable research surfaces and must not be treated as
 deployed protocol guarantees.
 
+`iicp_client.restricted_membership` verifies the shared restricted trust-domain
+membership and authenticated-gossip fixtures with the same bounded refusal
+reasons as the Rust reference. It is a verification foundation only: Python
+bootstrap, peer storage, relay selection and CIP do not claim restricted-domain
+support until issue #103 and the cross-implementation black-box gate are
+complete. Public behavior is unchanged.
+
 ---
 
 ## Development

src/iicp_client/__init__.py

@@ -90,6 +90,17 @@
     qualify_service,
     qualify_service_async,
 )
+from iicp_client.restricted_directory import (
+    PROFILE_ID as RESTRICTED_TRUST_DOMAIN_PROFILE_ID,
+)
+from iicp_client.restricted_directory import (
+    RestrictedDirectoryContext,
+    RestrictedEligibility,
+    SecretRef,
+)
+from iicp_client.restricted_directory import (
+    validate_decision as validate_restricted_directory_decision,
+)
 from iicp_client.routing_policy import (
     ROUTING_POLICY_REFUSAL_CODE,
     filter_nodes_for_routing_policy,
@@ -142,6 +153,11 @@
 __all__ = [
     "IicpClient",
     "IicpError",
+    "RestrictedDirectoryContext",
+    "RestrictedEligibility",
+    "SecretRef",
+    "RESTRICTED_TRUST_DOMAIN_PROFILE_ID",
+    "validate_restricted_directory_decision",
     "EFFECTIVE_CAPABILITY_PROFILE_ID",
     "EFFECTIVE_CAPABILITY_SCHEMA_VERSION",
     "CapabilityClaimProvenance",

src/iicp_client/_http.py

@@ -43,11 +43,15 @@ async def get_json(
     component: str = "directory",
     tls_verify: bool = True,
     traceparent: str | None = None,
+    extra_headers: dict[str, str] | None = None,
+    follow_redirects: bool = False,
 ) -> dict[str, Any]:
     timeout = timeout_ms / 1000.0
     headers = {"traceparent": traceparent or _traceparent()}
+    if extra_headers:
+        headers.update(extra_headers)
     try:
-        async with httpx.AsyncClient(timeout=timeout, verify=_tls_context(tls_verify)) as client:
+        async with httpx.AsyncClient(timeout=timeout, verify=_tls_context(tls_verify), follow_redirects=follow_redirects) as client:
             resp = await client.get(url, params=params, headers=headers)
     except httpx.TimeoutException:
         raise IicpError(

src/iicp_client/client.py

@@ -21,6 +21,7 @@
 from iicp_client.errors import IicpError
 from iicp_client.policy import ensure_intent_allowed
 from iicp_client.request_projection import project_execution_constraints, project_route_options
+from iicp_client.restricted_directory import validate_decision
 from iicp_client.routing_policy import (
     ROUTING_POLICY_REFUSAL_CODE,
     filter_nodes_for_routing_policy,
@@ -148,6 +149,11 @@ def __init__(self, config: ClientConfig | None = None) -> None:
         self._ct_cache: dict[tuple[str, str], tuple[str, int]] = {}
         self._dispatch_ticket_key: str | None = None
         self._candidate_ranker: CandidateRanker | None = None
+        self._restricted_eligibility: dict[int, object] = {}
+        if self._cfg.restricted_directory is not None:
+            self._cfg.restricted_directory.validate()
+            if self._cfg.route_discovery_mode == "legacy":
+                raise ValueError("restricted directory mode cannot use legacy route discovery")
 
     def with_candidate_ranker(self, ranker: CandidateRanker) -> IicpClient:
         """Attach an optional ranker for candidates that already passed eligibility."""
@@ -171,22 +177,29 @@ async def _acquire_consumer_token(self, target_node_id: str, intent: str, timeou
                 return tok
         base = self._cfg.directory_url.rstrip("/").removesuffix("/api")
         url = f"{base}/api/v1/consumer-token"
+        restricted = self._cfg.restricted_directory
         try:
-            async with httpx.AsyncClient(timeout=timeout_s) as client:
+            async with httpx.AsyncClient(timeout=timeout_s, follow_redirects=False) as client:
+                headers = {"Authorization": f"Bearer {node_token}"}
+                if restricted:
+                    headers.update(restricted.headers())
                 r = await client.post(
                     url,
                     json={"target_node_id": target_node_id, "intent": intent},
-                    headers={"Authorization": f"Bearer {node_token}"},
+                    headers=headers,
                 )
                 if r.status_code == 201:
                     data = r.json()
+                    if restricted:
+                        validate_decision(data, restricted, "consumer_token")
                     token: str = data.get("token", "")
                     exp_unix: int = int(data.get("expires_at", 0))
                     if token:
                         self._ct_cache[cache_key] = (token, exp_unix)
                         return token
         except Exception:
-            pass
+            if restricted:
+                raise
         return None
 
     def _select_candidates(self, all_nodes: list[Node], top_n: int) -> list[Node]:
@@ -281,6 +294,8 @@ async def _ticketed_candidates(
         excluded: list[str] = []
         candidates: list[Node] = []
         headers = {"Accept": "application/json", "Content-Type": "application/json"}
+        if self._cfg.restricted_directory is not None:
+            headers.update(self._cfg.restricted_directory.headers())
         if traceparent:
             headers["traceparent"] = traceparent
 
@@ -298,6 +313,11 @@ async def _ticketed_candidates(
                 error_code = data.get("error", {}).get("code") if isinstance(data, dict) else None
 
                 if response.status_code == 201:
+                    eligibility = (
+                        validate_decision(data, self._cfg.restricted_directory, "dispatch_ticket")
+                        if self._cfg.restricted_directory is not None
+                        else None
+                    )
                     ticket = data.get("ticket") if isinstance(data, dict) else None
                     node_id = data.get("node_id") if isinstance(data, dict) else None
                     if self._dispatch_ticket_key is None:
@@ -334,6 +354,8 @@ async def _ticketed_candidates(
                         route = {**route, "node_id": data.get("node_id", route.get("node_id"))}
                         node = self._node_from_route(route, ticket_id_prefix=data.get("ticket_id_prefix"))
                         if node is not None:
+                            if eligibility is not None:
+                                self._restricted_eligibility[id(node)] = eligibility
                             candidates.append(node)
                             excluded.append(node.node_id[:8])
                             continue
@@ -349,6 +371,8 @@ async def _ticketed_candidates(
                 if response.status_code in {404, 405, 501} or (
                     response.status_code == 503 and error_code == "not_configured"
                 ):
+                    if self._cfg.restricted_directory is not None:
+                        raise IicpError("restricted_directory_fallback_refused", "restricted mode cannot fall back to legacy discovery", "directory")
                     raise _LegacyDiscoveryRequired
                 raise IicpError(
                     code=f"IICP-DISPATCH-TICKET-{response.status_code}",
@@ -398,6 +422,13 @@ async def discover_async(
             component="directory",
             tls_verify=self._cfg.tls_verify,
             traceparent=traceparent,
+            extra_headers=self._cfg.restricted_directory.headers() if self._cfg.restricted_directory else None,
+            follow_redirects=self._cfg.restricted_directory is None,
+        )
+        eligibility = (
+            validate_decision(data, self._cfg.restricted_directory, "discovery")
+            if self._cfg.restricted_directory is not None
+            else None
         )
         elapsed = int((time.monotonic() - t0) * 1000)
 
@@ -438,6 +469,8 @@ async def discover_async(
             ):
                 continue
             nodes.append(node)
+            if eligibility is not None:
+                self._restricted_eligibility[id(node)] = eligibility
         diversity = data.get("diversity_evidence")
         return NodeList(
             nodes=nodes,

src/iicp_client/restricted_directory.py

@@ -0,0 +1,76 @@
+"""Fail-closed restricted trust-domain directory operation boundary."""
+from __future__ import annotations
+
+import os
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Literal
+
+from iicp_client.errors import IicpError
+
+PROFILE_ID = "urn:iicp:profile:restricted-trust-domain:v1"
+DECISION_SCHEMA = "iicp.restricted-trust-domain.directory-decision.v0"
+
+def _refused(message: str) -> IicpError:
+    return IicpError("restricted_directory_decision_refused", message, "directory", retryable=False)
+
+@dataclass(frozen=True)
+class SecretRef:
+    """Reference to membership material; the value is never serialized."""
+    kind: Literal["environment", "file"]
+    value: str
+
+    def resolve(self) -> str:
+        secret = ""
+        if self.kind == "environment":
+            secret = os.environ.get(self.value, "")
+        elif self.kind == "file":
+            path = Path(self.value)
+            if not path.is_symlink() and path.is_file() and not path.stat().st_mode & 0o077:
+                secret = path.read_text(encoding="utf-8").strip()
+        if not secret:
+            raise _refused("restricted directory membership credential is unavailable")
+        return secret
+
+@dataclass(frozen=True)
+class RestrictedDirectoryContext:
+    domain_id: str
+    authority_id: str
+    subject_id: str
+    subject_kind: Literal["node", "client", "directory"]
+    minimum_membership_generation: int
+    membership_credential: SecretRef
+
+    def validate(self) -> None:
+        if (not self.domain_id.strip() or not self.authority_id.strip() or not self.subject_id.strip()
+                or self.subject_kind not in {"node", "client", "directory"}
+                or self.minimum_membership_generation < 1):
+            raise _refused("restricted directory context is incomplete")
+        self.membership_credential.resolve()
+
+    def headers(self) -> dict[str, str]:
+        return {"X-IICP-Membership": self.membership_credential.resolve(), "X-IICP-Subject-Id": self.subject_id}
+
+@dataclass(frozen=True)
+class RestrictedEligibility:
+    domain_id: str
+    authority_id: str
+    membership_generation: int
+    membership_expires_at: int
+
+def validate_decision(body: dict[str, Any], context: RestrictedDirectoryContext, operation: str) -> RestrictedEligibility:
+    raw = body.get("restricted_domain_decision")
+    expected = {"schema", "profile", "decision", "operation", "domain_id", "authority_id", "subject_kind", "membership_generation", "membership_expires_at"}
+    if not isinstance(raw, dict) or set(raw) != expected:
+        raise _refused("restricted directory decision is missing or malformed")
+    generation, expiry = raw["membership_generation"], raw["membership_expires_at"]
+    if (not isinstance(generation, int) or isinstance(generation, bool)
+            or not isinstance(expiry, int) or isinstance(expiry, bool)):
+        raise _refused("restricted directory decision is malformed")
+    if (raw["schema"] != DECISION_SCHEMA or raw["profile"] != PROFILE_ID or raw["decision"] != "eligible"
+            or raw["operation"] != operation or raw["domain_id"] != context.domain_id
+            or raw["authority_id"] != context.authority_id or raw["subject_kind"] != context.subject_kind
+            or generation < context.minimum_membership_generation or expiry <= int(time.time())):
+        raise _refused("restricted directory decision does not match the request context")
+    return RestrictedEligibility(raw["domain_id"], raw["authority_id"], generation, expiry)