Send a caller identity to flex-auth so policy.enabled can flip

flex-auth's flex-auth-ops-warden pin (FLEX-WP-0016) TokenReviews the caller and
binds resource.system: ops-warden to system:serviceaccount:ops-warden:ops-warden.
policy.py posted /v1/check with no Authorization header, so the pin logs
"caller authentication warning" and can only run callerAuth.mode: warn — which,
under ADHOC-2026-08-17-T01, is exactly what blocks policy.enabled: true.

- policy.caller_auth (none | file | env | command) + src/warden/caller_identity.py:
  token resolved per call, never cached, written, or logged (ADR-0002)
- both check_sign_policy and check_fetch_policy attach the bearer header; an
  unobtainable token fails closed rather than retrying anonymously
- scripts/check_policy_caller_identity.py: read-only gate, prints length and a
  truncated fingerprint only, distinguishes 401 (audience/binding) from 403
- example config: caller_auth block, and flex_auth_url corrected — it pointed at
  flex-auth.flex-auth.svc, a Service that does not exist
- WARDEN-WP-0031, PolicyGatedSigning caller-identity section and flip sequence

Default stays mode: none, so behaviour is unchanged until an operator opts in.

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

tests/test_policy.py

@@ -1,4 +1,5 @@
 """Tests for warden.policy — flex-auth gate."""
+import subprocess
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -6,7 +7,12 @@ import httpx
 import pytest
 
 from warden.ca import CAError
-from warden.config import PolicyConfig
+from warden.caller_identity import (
+    CallerIdentityError,
+    caller_auth_headers,
+    resolve_caller_token,
+)
+from warden.config import CallerAuthConfig, PolicyConfig
 from warden.models import ActorType, CertSpec
 from warden.policy import check_sign_policy, pubkey_fingerprint
 
@@ -137,4 +143,123 @@ def test_subject_from_env(tmp_path, monkeypatch):
         check_sign_policy(cfg, _spec(pubkey))
 
     body = post.call_args[1]["json"]
-    assert body["subject"]["id"] == "iam:bernd"
+    assert body["subject"]["id"] == "iam:bernd"
+
+# --- caller identity (FLEX-WP-0016 / WARDEN-WP-0031) -----------------------
+
+def test_caller_auth_none_sends_no_header():
+    assert caller_auth_headers(CallerAuthConfig()) == {}
+
+
+def test_caller_auth_file_reads_projected_token(tmp_path):
+    token_file = tmp_path / "token"
+    token_file.write_text("sa-token-value\n")
+    cfg = CallerAuthConfig(mode="file", token_path=token_file)
+    assert caller_auth_headers(cfg) == {"Authorization": "Bearer sa-token-value"}
+
+
+def test_caller_auth_file_missing_raises(tmp_path):
+    cfg = CallerAuthConfig(mode="file", token_path=tmp_path / "absent")
+    with pytest.raises(CallerIdentityError, match="unreadable"):
+        resolve_caller_token(cfg)
+
+
+def test_caller_auth_env_mode(monkeypatch):
+    monkeypatch.setenv("WARDEN_POLICY_CALLER_TOKEN", " env-token ")
+    assert resolve_caller_token(CallerAuthConfig(mode="env")) == "env-token"
+    monkeypatch.setenv("WARDEN_POLICY_CALLER_TOKEN", "")
+    with pytest.raises(CallerIdentityError, match="unset or empty"):
+        resolve_caller_token(CallerAuthConfig(mode="env"))
+
+
+def test_caller_auth_command_mode_uses_stdout(monkeypatch):
+    cfg = CallerAuthConfig(mode="command", command=["kubectl", "create", "token"])
+
+    def fake_run(cmd, **kwargs):
+        assert cmd == cfg.command
+        return subprocess.CompletedProcess(cmd, 0, stdout="minted-token\n", stderr="")
+
+    monkeypatch.setattr(subprocess, "run", fake_run)
+    assert resolve_caller_token(cfg) == "minted-token"
+
+
+def test_caller_auth_command_failure_message_excludes_token(monkeypatch):
+    cfg = CallerAuthConfig(mode="command", command=["kubectl", "create", "token"])
+
+    def fake_run(cmd, **kwargs):
+        return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="error: forbidden\n")
+
+    monkeypatch.setattr(subprocess, "run", fake_run)
+    with pytest.raises(CallerIdentityError, match="error: forbidden"):
+        resolve_caller_token(cfg)
+
+
+def test_caller_auth_rejects_whitespace_token(tmp_path):
+    token_file = tmp_path / "token"
+    token_file.write_text("two words")
+    cfg = CallerAuthConfig(mode="file", token_path=token_file)
+    with pytest.raises(CallerIdentityError, match="whitespace"):
+        resolve_caller_token(cfg)
+
+
+def test_sign_policy_sends_authorization_header(tmp_path, monkeypatch):
+    """The header flex-auth's ops-warden pin needs to leave warn mode."""
+    from warden import policy as policy_mod
+
+    token_file = tmp_path / "token"
+    token_file.write_text("sa-token-value")
+    pubkey = tmp_path / "id.pub"
+    pubkey.write_text("ssh-ed25519 AAAA test\n")
+
+    cfg = PolicyConfig(
+        enabled=True,
+        caller_auth=CallerAuthConfig(mode="file", token_path=token_file),
+    )
+    spec = CertSpec(
+        actor_name="agt-state-hub-bridge",
+        actor_type=ActorType.AGT,
+        principals=["agt"],
+        ttl_hours=24,
+        pubkey_path=pubkey,
+    )
+
+    seen = {}
+
+    class _Response:
+        status_code = 200
+
+        def raise_for_status(self):
+            return None
+
+        def json(self):
+            return {"effect": "allow", "id": "decision:49350f1064f674d7"}
+
+    def fake_post(url, json=None, headers=None, timeout=None):
+        seen["headers"] = headers
+        return _Response()
+
+    monkeypatch.setattr(policy_mod.httpx, "post", fake_post)
+    assert policy_mod.check_sign_policy(cfg, spec) == "decision:49350f1064f674d7"
+    assert seen["headers"] == {"Authorization": "Bearer sa-token-value"}
+
+
+def test_sign_policy_fail_closed_when_caller_token_unavailable(tmp_path):
+    from warden.ca import CAError
+    from warden import policy as policy_mod
+
+    pubkey = tmp_path / "id.pub"
+    pubkey.write_text("ssh-ed25519 AAAA test\n")
+    cfg = PolicyConfig(
+        enabled=True,
+        fail_closed=True,
+        caller_auth=CallerAuthConfig(mode="file", token_path=tmp_path / "absent"),
+    )
+    spec = CertSpec(
+        actor_name="agt-state-hub-bridge",
+        actor_type=ActorType.AGT,
+        principals=["agt"],
+        ttl_hours=24,
+        pubkey_path=pubkey,
+    )
+    with pytest.raises(CAError, match="caller identity unavailable"):
+        policy_mod.check_sign_policy(cfg, spec)