WARDEN-WP-0026 T02: safe access transports (no secret values on stdout)
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 2s

- proxy.py: proxy_fetch_to_file (mode-0600 file), build_wrapped_fetch +
  proxy_fetch_wrapped (single-use OpenBao response-wrapping token), _capture_value
  helper, is_bao_kv_fetch.
- warden access: --out FILE, --wrap [--wrap-ttl], --unsafe-stdout. Raw --fetch to a
  non-TTY stdout is refused (exit 6) — captured/piped output is the disclosure risk;
  sanctioned transports are --out / --exec / --wrap.
- canon: anti-pattern (secret value onto captured stdout) + transport table in
  .claude/rules/credential-routing.md; OperatorAccessAssist.md examples + G2 updated.
- tests: file/wrap/build + stdout-guard in tests/test_proxy.py. 293 pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-07-16 14:51:56 +02:00
parent c749561b75
commit 359ca1bd0e
6 changed files with 281 additions and 11 deletions

View file

@ -287,3 +287,80 @@ def test_invalid_lane_rejected(tmp_path):
import pytest
with pytest.raises(CatalogError, match="invalid lane"):
load_catalog(p)
# ---------------------------------------------------------------------------
# Safe access transports (WARDEN-WP-0026 T02) — no secret values on stdout
# ---------------------------------------------------------------------------
from warden.proxy import ( # noqa: E402
build_wrapped_fetch,
is_bao_kv_fetch,
proxy_fetch_to_file,
proxy_fetch_wrapped,
)
def test_fetch_to_file_writes_mode_0600_and_no_stdout(tmp_path, capsys):
out = tmp_path / "secret.out"
rc = proxy_fetch_to_file(ResolvedFetch(shell_cmd="printf 'sekret'"), out)
assert rc == 0
assert out.read_text() == "sekret"
assert oct(out.stat().st_mode & 0o777) == "0o600"
# nothing printed to stdout/stderr by the transport itself
captured = capsys.readouterr()
assert "sekret" not in captured.out and "sekret" not in captured.err
def test_fetch_to_file_forces_0600_on_preexisting_loose_file(tmp_path):
out = tmp_path / "pre.out"
out.write_text("old")
out.chmod(0o644)
proxy_fetch_to_file(ResolvedFetch(shell_cmd="printf 'new'"), out)
assert out.read_text() == "new"
assert oct(out.stat().st_mode & 0o777) == "0o600"
def test_wrapped_fetch_returns_token_not_value():
payload = '{"wrap_info":{"token":"hvs.WRAP"}}'
token = proxy_fetch_wrapped(ResolvedFetch(shell_cmd=f"printf '%s' '{payload}'"))
assert token == "hvs.WRAP"
def test_wrapped_fetch_bad_output_raises():
with pytest.raises(ProxyError, match="wrapping token"):
proxy_fetch_wrapped(ResolvedFetch(shell_cmd="printf 'not-json'"))
def test_build_wrapped_fetch_only_for_bao_kv():
bao = _entry(fetch_command="bao kv get -field=API_TOKEN platform/x", path_template="platform/x")
assert is_bao_kv_fetch(bao)
argv = build_wrapped_fetch(bao, ttl="9m").argv
assert argv == ["bao", "kv", "get", "-wrap-ttl=9m", "-format=json", "platform/x"]
piped = _entry(fetch_command="kubectl get secret x -o json | base64 -d", path_template="x")
assert not is_bao_kv_fetch(piped)
with pytest.raises(ProxyError, match="response wrapping"):
build_wrapped_fetch(piped)
def test_build_wrapped_fetch_refuses_placeholder_path():
e = _entry(fetch_command="bao kv get -field=<FIELD> <path_template>",
path_template="platform/workloads/<domain>/x")
with pytest.raises(ProxyError, match="concrete path"):
build_wrapped_fetch(e)
def test_access_fetch_to_nonterminal_stdout_is_refused(tmp_path, monkeypatch):
"""The anti-pattern: streaming a value to captured stdout is refused (exit 6)."""
_proxy_env(monkeypatch, tmp_path)
monkeypatch.setenv("VAULT_TOKEN", "caller-token") # G1 caller-auth precheck
# The guard trips before the fetch runs; make a real bao call fail loudly if reached.
monkeypatch.setattr(
"warden.proxy.subprocess.run",
lambda *a, **k: (_ for _ in ()).throw(AssertionError("fetch ran despite stdout guard")),
)
# CliRunner captures stdout (not a tty), so the guard trips without --unsafe-stdout.
r = runner.invoke(app, ["access", "whynot-design-npm-publish", "--fetch", "--no-policy"])
assert r.exit_code == 6
assert "sanctioned transport" in r.output.lower() or "refusing" in r.output.lower()