tegwick
8dbf62515a
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
125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
"""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
|