WARDEN-WP-0036: tighten attended-login stdout/stderr allowlist
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

Reject non-null wrap_info (a wrapping token survives token revoke -self),
non-empty data, empty client_token, and leading whitespace before the OIDC
stderr prompt. Fail-closed and hand-off tests now run through the CLI and
assert sentinel absence from both audit files.

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-a6db62075883
This commit is contained in:
tegwick 2026-10-14 12:41:43 +02:00
parent 8a426fba69
commit 6e1430073c
3 changed files with 90 additions and 21 deletions

View file

@ -49,7 +49,9 @@ _TOKEN_HELPER_NAME = ".vault-token"
# values are never logged, returned, stored, or included in an exception.
#
# stdout: empty, or exactly one JSON object shaped like a ``bao login -format=json``
# response (api.Secret with an ``auth`` block).
# response (api.Secret with an ``auth`` block). ``wrap_info`` must be null or
# absent (a wrapping token would survive ``token revoke -self``), ``data`` null,
# absent or empty, and ``auth.client_token`` a non-empty string.
_LOGIN_JSON_TOP_TYPES: dict = {
"request_id": (str,),
"lease_id": (str,),
@ -57,7 +59,7 @@ _LOGIN_JSON_TOP_TYPES: dict = {
"renewable": (bool,),
"data": (dict, type(None)),
"warnings": (list, type(None)),
"wrap_info": (dict, type(None)),
"wrap_info": (type(None),),
"mount_type": (str,),
"auth": (dict,),
}
@ -78,7 +80,7 @@ _LOGIN_JSON_AUTH_TYPES: dict = {
# stderr: empty, or exactly the OpenBao OIDC CLI prompt (browser-launch or
# open-link variant, one https URL, optional wait line). Any other byte fails closed.
_LOGIN_OIDC_STDERR = re.compile(
r"\s*Complete the login via your OIDC provider\. "
r"Complete the login via your OIDC provider\. "
r"(?:Launching browser to:|Open the following link in your browser:)"
r"[ \t]*\n\s*https://[^\s]+[ \t]*\n"
r"\s*(?:Waiting for OIDC authentication to complete\.\.\.[ \t]*\n?\s*)?"
@ -428,8 +430,13 @@ def _login_stdout_allowed(raw: bytes) -> bool:
return False
if not _typed_keys_allowed(parsed, _LOGIN_JSON_TOP_TYPES):
return False
if parsed.get("data"):
return False
auth = parsed["auth"]
return "client_token" in auth and _typed_keys_allowed(auth, _LOGIN_JSON_AUTH_TYPES)
if not _typed_keys_allowed(auth, _LOGIN_JSON_AUTH_TYPES):
return False
token = auth.get("client_token")
return isinstance(token, str) and token != ""
def _login_stderr_allowed(raw: bytes) -> bool:

View file

@ -614,6 +614,19 @@ def _run_populated_login(monkeypatch, tmp_path, *, stdout: bytes, stderr: bytes)
_LOGIN_SENTINEL = "hvs.NONPRODUCTION_UNEXPECTED_LOGIN"
_OK_SENTINEL = "hvs.NONPRODUCTION_OK"
def _assert_login_audits(tmp_path, *, code: int, outcome: str) -> None:
for name in ("access-audit.log", "audit.jsonl"):
raw = (tmp_path / "state" / name).read_text()
assert _LOGIN_SENTINEL not in raw
assert _OK_SENTINEL not in raw
assert "NONPRODUCTION_HELPER_VALUE" not in raw
records = [json.loads(line) for line in raw.splitlines()]
login = next(record for record in records if record["action"] == "login")
assert login["exit_code"] == code
assert login["outcome"] == outcome
@pytest.mark.parametrize(
@ -638,6 +651,17 @@ _LOGIN_SENTINEL = "hvs.NONPRODUCTION_UNEXPECTED_LOGIN"
# JSON without auth block / non-object JSON
(json.dumps({"data": {"v": _LOGIN_SENTINEL}}).encode(), b""),
(json.dumps([_LOGIN_SENTINEL]).encode(), b""),
# wrapping token beside auth (would survive revoke -self)
(json.dumps({**_login_json("a"), "wrap_info": {"token": _LOGIN_SENTINEL}}).encode(), b""),
(json.dumps({**_login_json("a"), "wrap_info": {}}).encode(), b""),
# non-empty data block
(json.dumps({**_login_json("a"), "data": {"v": _LOGIN_SENTINEL}}).encode(), b""),
# empty or null client_token
(json.dumps({"auth": {"client_token": "", "accessor": _LOGIN_SENTINEL}}).encode(), b""),
(json.dumps({"auth": {"client_token": None, "accessor": _LOGIN_SENTINEL}}).encode(), b""),
# leading whitespace before the OIDC prompt
(b"", b" " + _OIDC_PROMPT),
(b"", b"\n" + _OIDC_PROMPT),
# duplicate keys
(b'{"auth": {"client_token": "a"}, "auth": {"client_token": "b"}}', b""),
# malformed JSON
@ -648,31 +672,53 @@ _LOGIN_SENTINEL = "hvs.NONPRODUCTION_UNEXPECTED_LOGIN"
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",
"no-auth", "json-array", "wrap-info-token", "wrap-info-empty",
"data-nonempty", "token-empty", "token-null", "stderr-leading-space",
"stderr-leading-newline", "duplicate-key", "malformed-json", "non-utf8",
],
)
def test_attended_login_non_allowlisted_output_with_populated_helper_fails_closed(
monkeypatch, tmp_path, capsys, stdout, stderr
):
_proxy_env(monkeypatch, tmp_path)
calls = _run_populated_login(monkeypatch, tmp_path, stdout=stdout, stderr=stderr)
result = runner.invoke(
app, ["access", "login oidc", "--domain", "coulomb_social",
"--exec", "--", "should-not-run"]
)
assert result.exit_code == 10, result.output
assert "non-allowlisted output" in result.output
assert _LOGIN_SENTINEL not in result.output
captured = capsys.readouterr()
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 len(calls) == 2
assert calls[0][:2] == ["bao", "login"]
assert calls[1] == ["bao", "token", "revoke", "-self"]
assert all(_LOGIN_SENTINEL not in " ".join(argv) for argv in calls)
assert not (tmp_path / ".warden-attended-login").exists()
_assert_login_audits(tmp_path, code=10, outcome="login_failed")
def test_attended_login_non_allowlisted_output_direct_raises_value_free(
monkeypatch, tmp_path
):
calls = _run_populated_login(
monkeypatch, tmp_path, stdout=_LOGIN_SENTINEL.encode(), stderr=b""
)
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 (exc.value.exit_code, exc.value.outcome) == (10, "login_failed")
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(
@ -680,19 +726,21 @@ def test_attended_login_non_allowlisted_output_with_populated_helper_fails_close
[
(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""),
(json.dumps(_login_json(_OK_SENTINEL)).encode() + b"\n", b""),
(json.dumps({"auth": {"client_token": _OK_SENTINEL}}).encode(), b""),
(json.dumps({"data": {}, "auth": {"client_token": _OK_SENTINEL}}).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"],
ids=["empty", "oidc-prompt", "full-json", "minimal-json", "data-empty", "open-link-prompt"],
)
def test_attended_login_allowlisted_output_hands_off(
monkeypatch, tmp_path, stdout, stderr
monkeypatch, tmp_path, capsys, stdout, stderr
):
_proxy_env(monkeypatch, tmp_path)
monkeypatch.setattr(Path, "home", lambda: tmp_path)
calls = []
@ -706,11 +754,18 @@ def test_attended_login_allowlisted_output_hands_off(
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
result = runner.invoke(
app, ["access", "login oidc", "--domain", "coulomb_social",
"--exec", "--", "reviewed-child"]
)
assert result.exit_code == 0, result.output
assert calls[1] == ["reviewed-child"]
assert calls[-1] == ["bao", "token", "revoke", "-self"]
assert _OK_SENTINEL not in result.output
captured = capsys.readouterr()
assert _OK_SENTINEL not in captured.out + captured.err
assert not (tmp_path / ".warden-attended-login").exists()
_assert_login_audits(tmp_path, code=0, outcome="ok")
def test_real_catalog_login_entry_is_login_lane():

View file

@ -104,3 +104,10 @@ test; the success test now uses a full JSON shape plus the OIDC prompt. The
September 1 T02 live operation was recorded in `9770d6a` (00:51 +0200), after
`b4c1d39` (00:13 +0200) and `8f01eef` (00:24 +0200); `proxy.py` at that time
contained the relaxed check (the installed CLI's exact revision was not recorded). No live OIDC, OpenBao call, or credential read was used for this fix.
Tightened the same day after an independent re-review (containment gap): stdout
`wrap_info` must be null or absent (a wrapping token would survive `token revoke
-self`), `data` null, absent or `{}`, and `auth.client_token` a non-empty string;
the stderr OIDC prompt must start at the first byte (no leading whitespace;
trailing newlines remain fine). The fail-closed and hand-off tests now run through
the CLI and assert sentinel absence from both `access-audit.log` and `audit.jsonl`.