WARDEN: separate fetch evaluator URL for secret-read checks (CUST-WP-0075)

check_fetch_policy now posts only to policy.fetch_flex_auth_url (access-engine
pin flex-auth-ops-warden-fetch, workstation 127.0.0.1:19091). Signing keeps
policy.flex_auth_url. Unset or unreachable fetch evaluator fails closed with a
message naming it; no fallback to the signing pin or to allow. warden access
advisory output names the fetch evaluator. Recorded under WARDEN-WP-0039-T03.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 3775@bnt-lap001
Assistant-Session: cfacb6a5-ad5b-4f79-ac67-053e3b0d34a3

src/warden/access.py

@@ -98,6 +98,10 @@ def policy_gate_status() -> str:
         cfg = load_config()
     except ConfigError:
         return "advisory — no warden.yaml (caller identity; gate not enforced)"
-    if cfg.policy.flex_auth_url:
-        return f"zone-aware — flex-auth at {cfg.policy.flex_auth_url}"
-    return "zone-aware — evaluator unconfigured; unknown-zone fail_open applies"
+    fetch_url = cfg.policy.fetch_flex_auth_url
+    if fetch_url:
+        return f"zone-aware — fetches gated by flex-auth-ops-warden-fetch at {fetch_url}"
+    return (
+        "fail-closed — secret-read evaluator unconfigured "
+        "(policy.fetch_flex_auth_url unset); fetches are refused"
+    )

src/warden/cli.py

@@ -1551,6 +1551,8 @@ def access(
         )
     if expanded.policy_ref:
         console.print(f"  policy   : {expanded.policy_ref}   [dim]({gate})[/dim]")
+    elif entry.exec_capable:
+        console.print(f"  gate     : [dim]{gate}[/dim]")
     console.print(f"  wiki     : {entry.wiki_ref}")
     console.print(f"  canon    : {entry.canon_ref}")
 

src/warden/config.py

@@ -45,7 +45,11 @@ class CallerAuthConfig:
 
 @dataclass
 class PolicyConfig:
+    # SSH signing pin (flex-auth-ops-warden). Admits only resource.system ops-warden.
     flex_auth_url: Optional[str] = None
+    # Secret-read pin (flex-auth-ops-warden-fetch, CUST-WP-0075). Used only by
+    # check_fetch_policy; there is no fallback to flex_auth_url.
+    fetch_flex_auth_url: Optional[str] = None
     zone_registry_path: Optional[Path] = None
     failure_modes: Dict[str, str] = field(
         default_factory=lambda: {
@@ -211,8 +215,12 @@ def load_config(path: Optional[Path] = None) -> WardenConfig:
         )
     zone_registry_path = policy_raw.get("zone_registry_path")
     flex_auth_url = str(policy_raw.get("flex_auth_url", "")).strip() or None
+    fetch_flex_auth_url = (
+        str(policy_raw.get("fetch_flex_auth_url", "") or "").strip() or None
+    )
     policy_cfg = PolicyConfig(
         flex_auth_url=flex_auth_url,
+        fetch_flex_auth_url=fetch_flex_auth_url,
         zone_registry_path=(
             Path(os.path.expanduser(str(zone_registry_path)))
             if zone_registry_path

src/warden/policy.py

@@ -191,15 +191,21 @@ def check_fetch_policy(
     The action is ``read`` on a ``secret`` resource owned by another subsystem —
     ops-warden is the conduit, not the owner. Unresolved target workload identity
     selects the explicit ``unknown`` profile; no secret value enters the request.
+
+    The check goes only to ``cfg.fetch_flex_auth_url`` — the secret-read pin
+    (flex-auth-ops-warden-fetch, CUST-WP-0075). It never falls back to the SSH
+    signing pin (``cfg.flex_auth_url``), whose caller binding admits only the
+    ``ops-warden`` system. An unset or unreachable fetch evaluator always fails
+    closed, whatever the zone failure mode.
     """
     zone = "unknown"
     fail_closed = _is_fail_closed(cfg, zone)
-    if cfg.flex_auth_url is None:
-        _evaluator_failure(
-            "flex-auth URL is not configured for security zone 'unknown'",
-            fail_closed=fail_closed,
+    if cfg.fetch_flex_auth_url is None:
+        raise CAError(
+            "secret-read evaluator not configured: set policy.fetch_flex_auth_url "
+            "(flex-auth-ops-warden-fetch pin); fetch checks never use the SSH "
+            "signing pin policy.flex_auth_url"
         )
-        return None
     subject_id = os.environ.get(cfg.subject_env, "").strip() or "operator"
     request = {
         "subject": {"id": subject_id, "type": "operator", "tenant": cfg.tenant},
@@ -213,8 +219,8 @@ def check_fetch_policy(
         "context": {"need_id": need_id, "owner_repo": owner_repo, "domain": domain},
     }
 
-    url = cfg.flex_auth_url.rstrip("/") + "/v1/check"
-    headers = _caller_headers(cfg, fail_closed=fail_closed)
+    url = cfg.fetch_flex_auth_url.rstrip("/") + "/v1/check"
+    headers = _caller_headers(cfg, fail_closed=True)
     try:
         response = httpx.post(url, json=request, headers=headers, timeout=10.0)
         response.raise_for_status()
@@ -231,12 +237,10 @@ def check_fetch_policy(
         )
         return None
     except httpx.RequestError as e:
-        _evaluator_failure(
-            f"flex-auth unreachable at {cfg.flex_auth_url!r} for security zone 'unknown'",
-            fail_closed=fail_closed,
-            cause=e,
-        )
-        return None
+        raise CAError(
+            "secret-read evaluator (flex-auth-ops-warden-fetch) unreachable at "
+            f"{cfg.fetch_flex_auth_url!r}; fetch refused (no fallback to the signing pin)"
+        ) from e
 
     try:
         decision = response.json()