Show the OpenBao identity warden access evaluates

reuse-surface-hub-write-token does not select a caller. The data plane
uses the ambient token, which can be the coding-agent role denied by
agent-high-risk-boundary while an attended platform-admin lookup still
shows read. Advisory and proxy output now report that entity, role, and
policies without the token. CCR-2026-0005 still has no OIDC binding for
workload-kv-read-reuse-surface-runtime.

Assistant: grok
Assistant-Session: 01a13b2d-0bc1-7a91-ab1d-325bcfb09bf0

tests/test_access.py

@@ -93,6 +93,31 @@ def test_operator_subject_does_not_override_legacy_agent_marker():
 
 # --- CLI -------------------------------------------------------------------
 
+def test_access_advisory_shows_ambient_openbao_caller(monkeypatch):
+    monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog()))
+
+    def described(env=None, **kwargs):
+        return {
+            "source": "BAO_TOKEN",
+            "entity_id": "4a1f663c-0621-0483-edd0-f852274f4b6b",
+            "display_name": "platform-root",
+            "auth_path": "auth/netkingdom/oidc/callback",
+            "policies": ["agent-high-risk-boundary"],
+            "oidc_role": "coding-agent",
+            "ttl": 900,
+            "error": None,
+        }
+
+    monkeypatch.setattr("warden.openbao_caller.describe_ambient_openbao_caller", described)
+    r = runner.invoke(app, ["access", "reuse-surface hub write token", "--json"])
+    assert r.exit_code == 0
+    payload = json.loads(r.stdout)
+    caller = payload["openbao_caller"]
+    assert caller["entity_id"] == "4a1f663c-0621-0483-edd0-f852274f4b6b"
+    assert "agent-high-risk-boundary" in caller["boundary_note"]
+    assert "id" not in caller
+
+
 def test_access_advisory_output(monkeypatch):
     monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog()))
     r = runner.invoke(app, ["access", "npm token", "--domain", "coulomb_social"])

tests/test_openbao_caller.py

@@ -0,0 +1,125 @@
+"""Sanitized ambient OpenBao caller identity."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from warden.openbao_caller import (
+    ambient_token_source,
+    caller_boundary_note,
+    describe_ambient_openbao_caller,
+)
+
+
+def test_source_prefers_bao_token_over_vault_and_file(monkeypatch, tmp_path):
+    monkeypatch.setenv("BAO_TOKEN", "hvs.bao")
+    monkeypatch.setenv("VAULT_TOKEN", "hvs.vault")
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    (tmp_path / ".vault-token").write_text("file-token")
+    assert ambient_token_source() == "BAO_TOKEN"
+
+
+def test_source_uses_vault_token_file_when_env_absent(monkeypatch, tmp_path):
+    monkeypatch.delenv("BAO_TOKEN", raising=False)
+    monkeypatch.delenv("VAULT_TOKEN", raising=False)
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    (tmp_path / ".vault-token").write_text("file-token")
+    assert ambient_token_source() == "vault-token-file"
+
+
+def test_lookup_drops_token_fields(monkeypatch):
+    secret = "hvs.this-must-not-appear"
+    monkeypatch.setenv("BAO_TOKEN", secret)
+    payload = {
+        "data": {
+            "id": secret,
+            "accessor": "also-secret",
+            "entity_id": "4a1f663c-0621-0483-edd0-f852274f4b6b",
+            "display_name": "netkingdom-uid=platform-root",
+            "path": "auth/netkingdom/oidc/callback",
+            "policies": ["default", "operator-custody", "platform-admin"],
+            "meta": {"role": "platform-admin", "email": "hidden@example.invalid"},
+            "ttl": 3001,
+        }
+    }
+
+    def runner(argv, **kwargs):
+        class Completed:
+            returncode = 0
+            stdout = json.dumps(payload)
+            stderr = ""
+
+        assert kwargs["env"]["BAO_TOKEN"] == secret
+        return Completed()
+
+    monkeypatch.setattr("warden.openbao_caller.subprocess.run", runner)
+    described = describe_ambient_openbao_caller()
+    encoded = json.dumps(described)
+    assert secret not in encoded
+    assert "accessor" not in described
+    assert "email" not in encoded
+    assert described["entity_id"].startswith("4a1f663c-")
+    assert described["oidc_role"] == "platform-admin"
+    assert described["policies"] == ["default", "operator-custody", "platform-admin"]
+    assert described["source"] == "BAO_TOKEN"
+    assert described["error"] is None
+
+
+def test_lookup_failure_has_no_raw_output(monkeypatch):
+    monkeypatch.setenv("VAULT_TOKEN", "hvs.caller")
+    monkeypatch.delenv("BAO_TOKEN", raising=False)
+
+    def runner(argv, **kwargs):
+        class Completed:
+            returncode = 2
+            stdout = ""
+            stderr = "Error: permission denied hvs.caller"
+
+        return Completed()
+
+    monkeypatch.setattr("warden.openbao_caller.subprocess.run", runner)
+    described = describe_ambient_openbao_caller()
+    assert described["error"] == "token lookup failed"
+    assert described["entity_id"] is None
+    assert "hvs.caller" not in json.dumps(described)
+
+
+def test_missing_client(monkeypatch):
+    monkeypatch.setenv("BAO_TOKEN", "hvs.bao")
+
+    def runner(argv, **kwargs):
+        raise FileNotFoundError(argv[0])
+
+    monkeypatch.setattr("warden.openbao_caller.subprocess.run", runner)
+    assert describe_ambient_openbao_caller()["error"] == "bao client not found"
+
+
+def test_no_token(monkeypatch, tmp_path):
+    monkeypatch.delenv("BAO_TOKEN", raising=False)
+    monkeypatch.delenv("VAULT_TOKEN", raising=False)
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    described = describe_ambient_openbao_caller()
+    assert described["source"] == "none"
+    assert described["error"] == "no ambient OpenBao token"
+
+
+@pytest.mark.parametrize(
+    ("policies", "high_risk", "fragment"),
+    [
+        (
+            ["agent-high-risk-boundary", "workload-kv-read-issue-core-runtime"],
+            True,
+            "agent-high-risk-boundary",
+        ),
+        (["default", "platform-admin"], True, "platform-admin"),
+        (["workload-kv-read-reuse-surface-runtime"], True, None),
+    ],
+)
+def test_boundary_note(policies, high_risk, fragment):
+    note = caller_boundary_note({"policies": policies}, high_risk=high_risk)
+    if fragment is None:
+        assert note is None
+    else:
+        assert fragment in note