From 6e1430073c64ed79204cdf676f18bfa41812d5d2 Mon Sep 17 00:00:00 2001 From: tegwick Date: Wed, 14 Oct 2026 12:41:43 +0200 Subject: [PATCH] WARDEN-WP-0036: tighten attended-login stdout/stderr allowlist 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 Assistant: claude-code Assistant-Model: opus Assistant-Process: 680842@bnt-lap001 Assistant-Session: 54eb9e8e-7d6b-4f0c-b7a6-a6db62075883 --- src/warden/proxy.py | 15 +++- tests/test_proxy.py | 89 +++++++++++++++---- ...N-WP-0036-attended-login-openbao-output.md | 7 ++ 3 files changed, 90 insertions(+), 21 deletions(-) diff --git a/src/warden/proxy.py b/src/warden/proxy.py index 1f7433f..1747487 100644 --- a/src/warden/proxy.py +++ b/src/warden/proxy.py @@ -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: diff --git a/tests/test_proxy.py b/tests/test_proxy.py index 4088788..35cbcee 100644 --- a/tests/test_proxy.py +++ b/tests/test_proxy.py @@ -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(): diff --git a/workplans/WARDEN-WP-0036-attended-login-openbao-output.md b/workplans/WARDEN-WP-0036-attended-login-openbao-output.md index aba808c..84b1440 100644 --- a/workplans/WARDEN-WP-0036-attended-login-openbao-output.md +++ b/workplans/WARDEN-WP-0036-attended-login-openbao-output.md @@ -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`.