WARDEN-WP-0036: restore fail-closed on non-allowlisted attended-login output
b4c1d39 dropped login output from the attended-login fail-closed condition,
so any stdout/stderr was ignored once the helper was populated (RPF-WP-0017
criterion 4 regression). Login output stays captured and never forwarded and
is now accepted only when empty or allowlisted by structure: one bao login
-format=json object on stdout, the OpenBao OIDC prompt on stderr. Anything
else self-revokes, cleans up, and exits 10/login_failed without retry.
Adds populated-helper tests for unexpected stdout, stderr, malformed and
multi-object JSON, and unexpected keys.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 680842@bnt-lap001
Assistant-Session: 54eb9e8e-7d6b-4f0c-b7a6-a6db62075883This commit is contained in:
parent
11b2a9d7bc
commit
244f1ae2ca
3 changed files with 311 additions and 11 deletions
|
|
@ -10,6 +10,7 @@ from typer.testing import CliRunner
|
|||
|
||||
from warden.cli import app
|
||||
from warden.proxy import (
|
||||
AttendedLoginError,
|
||||
ProxyError,
|
||||
ResolvedFetch,
|
||||
caller_auth_present,
|
||||
|
|
@ -370,8 +371,9 @@ def test_cli_attended_result_and_both_audits(monkeypatch, tmp_path, phase, code,
|
|||
raise OSError("client unavailable")
|
||||
if phase != "missing_helper":
|
||||
helper.write_bytes(sentinel)
|
||||
login_json = json.dumps({"auth": {"client_token": sentinel.decode()}})
|
||||
return subprocess.CompletedProcess(
|
||||
argv, 1 if phase == "login_failed" else 0, sentinel, b""
|
||||
argv, 1 if phase == "login_failed" else 0, login_json.encode(), b""
|
||||
)
|
||||
if argv == ["reviewed-child"]:
|
||||
if phase == "child_start":
|
||||
|
|
@ -508,8 +510,8 @@ def test_attended_login_contained_success_output_never_escapes(monkeypatch, tmp_
|
|||
return subprocess.CompletedProcess(
|
||||
argv,
|
||||
0,
|
||||
stdout=json.dumps({"auth": {"client_token": sentinel}}).encode(),
|
||||
stderr=b"",
|
||||
stdout=json.dumps(_login_json(sentinel)).encode(),
|
||||
stderr=_OIDC_PROMPT,
|
||||
)
|
||||
if argv == ["reviewed-child"]:
|
||||
child_ran = True
|
||||
|
|
@ -558,6 +560,159 @@ def test_attended_login_preserves_caller_warden_config_for_reviewed_child(
|
|||
assert seen_config == str(caller_config)
|
||||
|
||||
|
||||
_OIDC_PROMPT = (
|
||||
b"Complete the login via your OIDC provider. Launching browser to:\n\n"
|
||||
b" https://keycape.example.invalid/auth?state=x&nonce=y\n\n\n"
|
||||
b"Waiting for OIDC authentication to complete...\n"
|
||||
)
|
||||
|
||||
|
||||
def _login_json(token: str) -> dict:
|
||||
return {
|
||||
"request_id": "req",
|
||||
"lease_id": "",
|
||||
"lease_duration": 0,
|
||||
"renewable": False,
|
||||
"data": None,
|
||||
"warnings": None,
|
||||
"mount_type": "",
|
||||
"auth": {
|
||||
"client_token": token,
|
||||
"accessor": "acc",
|
||||
"policies": ["default"],
|
||||
"token_policies": ["default"],
|
||||
"identity_policies": None,
|
||||
"metadata": {"role": "r"},
|
||||
"orphan": True,
|
||||
"entity_id": "ent",
|
||||
"lease_duration": 3600,
|
||||
"renewable": True,
|
||||
"mfa_requirement": None,
|
||||
"num_uses": 0,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _run_populated_login(monkeypatch, tmp_path, *, stdout: bytes, stderr: bytes):
|
||||
"""Login exits 0 with a populated private helper but the given output."""
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
calls = []
|
||||
|
||||
def fake_run(argv, **kw):
|
||||
calls.append(argv)
|
||||
if argv[:2] == ["bao", "login"]:
|
||||
helper = Path(kw["env"]["HOME"]) / ".vault-token"
|
||||
helper.write_text("hvs.NONPRODUCTION_HELPER_VALUE")
|
||||
helper.chmod(0o600)
|
||||
return subprocess.CompletedProcess(argv, 0, stdout=stdout, stderr=stderr)
|
||||
if argv[:3] == ["bao", "token", "revoke"]:
|
||||
return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")
|
||||
raise AssertionError("reviewed child ran after non-allowlisted login output")
|
||||
|
||||
monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
|
||||
return calls
|
||||
|
||||
|
||||
_LOGIN_SENTINEL = "hvs.NONPRODUCTION_UNEXPECTED_LOGIN"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"stdout,stderr",
|
||||
[
|
||||
# unexpected stdout text
|
||||
(_LOGIN_SENTINEL.encode(), b""),
|
||||
# unexpected stderr text
|
||||
(b"", _LOGIN_SENTINEL.encode()),
|
||||
# unexpected stderr appended to an otherwise valid OIDC prompt
|
||||
(b"", _OIDC_PROMPT + _LOGIN_SENTINEL.encode()),
|
||||
# valid JSON followed by extra text
|
||||
(json.dumps(_login_json(_LOGIN_SENTINEL)).encode() + b"\ntrailing", b""),
|
||||
# two JSON objects
|
||||
((json.dumps(_login_json(_LOGIN_SENTINEL)) * 2).encode(), b""),
|
||||
# unexpected top-level key
|
||||
(json.dumps({**_login_json(_LOGIN_SENTINEL), "extra": _LOGIN_SENTINEL}).encode(), b""),
|
||||
# unexpected auth key
|
||||
(json.dumps({"auth": {"client_token": _LOGIN_SENTINEL, "x": 1}}).encode(), b""),
|
||||
# wrong value type
|
||||
(json.dumps({"auth": {"client_token": _LOGIN_SENTINEL, "num_uses": True}}).encode(), b""),
|
||||
# JSON without auth block / non-object JSON
|
||||
(json.dumps({"data": {"v": _LOGIN_SENTINEL}}).encode(), b""),
|
||||
(json.dumps([_LOGIN_SENTINEL]).encode(), b""),
|
||||
# duplicate keys
|
||||
(b'{"auth": {"client_token": "a"}, "auth": {"client_token": "b"}}', b""),
|
||||
# malformed JSON
|
||||
(b'{"auth": {"client_token": "' + _LOGIN_SENTINEL.encode() + b'"', b""),
|
||||
# undecodable bytes
|
||||
(b"\xff\xfe", b""),
|
||||
],
|
||||
ids=[
|
||||
"stdout-text", "stderr-text", "stderr-after-prompt", "json-trailing",
|
||||
"json-twice", "top-extra-key", "auth-extra-key", "wrong-type",
|
||||
"no-auth", "json-array", "duplicate-key", "malformed-json", "non-utf8",
|
||||
],
|
||||
)
|
||||
def test_attended_login_non_allowlisted_output_with_populated_helper_fails_closed(
|
||||
monkeypatch, tmp_path, capsys, stdout, stderr
|
||||
):
|
||||
calls = _run_populated_login(monkeypatch, tmp_path, stdout=stdout, stderr=stderr)
|
||||
with pytest.raises(AttendedLoginError, match="non-allowlisted output") as exc:
|
||||
proxy_attended_login_exec(
|
||||
ResolvedFetch(argv=["bao", "login", "-no-print"]),
|
||||
child_argv=["should-not-run"],
|
||||
)
|
||||
assert exc.value.exit_code == 10
|
||||
assert exc.value.outcome == "login_failed"
|
||||
captured = capsys.readouterr()
|
||||
assert _LOGIN_SENTINEL not in str(exc.value)
|
||||
assert _LOGIN_SENTINEL not in captured.out
|
||||
assert _LOGIN_SENTINEL not in captured.err
|
||||
# No retry: exactly one login, then self-revocation through the populated
|
||||
# helper; the reviewed child never runs.
|
||||
assert calls == [
|
||||
["bao", "login", "-no-print", "-format=json"],
|
||||
["bao", "token", "revoke", "-self"],
|
||||
]
|
||||
assert not (tmp_path / ".warden-attended-login").exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"stdout,stderr",
|
||||
[
|
||||
(b"", b""),
|
||||
(b"", _OIDC_PROMPT),
|
||||
(json.dumps(_login_json("hvs.NONPRODUCTION_OK")).encode() + b"\n", b""),
|
||||
(json.dumps({"auth": {"client_token": "hvs.NONPRODUCTION_OK"}}).encode(), b""),
|
||||
(
|
||||
b"",
|
||||
b"Complete the login via your OIDC provider. Open the following link "
|
||||
b"in your browser:\n\n https://keycape.example.invalid/auth?s=1\n\n\n",
|
||||
),
|
||||
],
|
||||
ids=["empty", "oidc-prompt", "full-json", "minimal-json", "open-link-prompt"],
|
||||
)
|
||||
def test_attended_login_allowlisted_output_hands_off(
|
||||
monkeypatch, tmp_path, stdout, stderr
|
||||
):
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
calls = []
|
||||
|
||||
def fake_run(argv, **kw):
|
||||
calls.append(argv)
|
||||
if argv[:2] == ["bao", "login"]:
|
||||
helper = Path(kw["env"]["HOME"]) / ".vault-token"
|
||||
helper.write_text("hvs.NONPRODUCTION_HELPER_VALUE")
|
||||
helper.chmod(0o600)
|
||||
return subprocess.CompletedProcess(argv, 0, stdout=stdout, stderr=stderr)
|
||||
return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")
|
||||
|
||||
monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
|
||||
assert proxy_attended_login_exec(
|
||||
ResolvedFetch(argv=["bao", "login", "-no-print"]),
|
||||
child_argv=["reviewed-child"],
|
||||
) == 0
|
||||
assert calls[1] == ["reviewed-child"]
|
||||
|
||||
|
||||
def test_real_catalog_login_entry_is_login_lane():
|
||||
from warden.routing import load_catalog
|
||||
e = load_catalog(_repo_catalog()).get("key-cape-oidc-login")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue