fix: contain attended OpenBao login output

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a0290b-3241-74c3-b868-6049545af836

tests/test_proxy.py

@@ -13,6 +13,7 @@ from warden.proxy import (
     ProxyError,
     ResolvedFetch,
     caller_auth_present,
+    proxy_attended_login_exec,
     proxy_exec,
     proxy_fetch,
     resolve_fetch_command,
@@ -251,37 +252,143 @@ def test_cli_proxy_rejects_retired_no_policy_bypass(monkeypatch, tmp_path):
 
 # --- T4: login lane --------------------------------------------------------
 
-def test_cli_login_lane_runs_without_token_or_policy_ack(monkeypatch, tmp_path):
-    """Login lane skips the caller-auth precheck and the secret-read gate."""
+def test_cli_login_lane_contains_login_handoff_and_revocation(monkeypatch, tmp_path):
+    """Login and its reviewed child share a private, silent helper session."""
     _proxy_env(monkeypatch, tmp_path)
     monkeypatch.delenv("VAULT_TOKEN", raising=False)
     monkeypatch.delenv("BAO_TOKEN", raising=False)
     monkeypatch.setattr(Path, "home", lambda: tmp_path)  # no ~/.vault-token
 
-    ran = {}
+    calls = []
 
     def fake_run(argv, **kw):
-        ran["argv"] = argv
-        ran["stdout"] = kw.get("stdout")
-        return subprocess.CompletedProcess(argv, 0)
+        calls.append((argv, kw))
+        assert kw["stdout"] is subprocess.PIPE
+        assert kw["stderr"] is subprocess.PIPE
+        private_home = Path(kw["env"]["HOME"])
+        assert private_home != tmp_path
+        assert oct(private_home.stat().st_mode & 0o777) == "0o700"
+        helper = private_home / ".vault-token"
+        assert oct(helper.stat().st_mode & 0o777) == "0o600"
+        if argv[:2] == ["bao", "login"]:
+            helper.write_bytes(b"non-production-test-double")
+            helper.chmod(0o600)
+        return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")
 
     monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
-    r = runner.invoke(app, ["access", "login oidc", "--domain", "coulomb_social", "--fetch"])
+    r = runner.invoke(
+        app,
+        [
+            "access", "login oidc", "--domain", "coulomb_social",
+            "--exec", "--", "true",
+        ],
+    )
     assert r.exit_code == 0
-    assert ran["argv"][:2] == ["bao", "login"]   # interactive login ran
-    assert ran["stdout"] is None                  # inherited stdio — token not captured
+    assert [call[0][:2] for call in calls] == [
+        ["bao", "login"],
+        ["true"],
+        ["bao", "token"],
+    ]
+    assert not (tmp_path / ".warden-attended-login").exists()
+    assert "non-production-test-double" not in r.output
+    audit = (tmp_path / "state" / "access-audit.log").read_text()
+    assert "non-production-test-double" not in audit
 
 
-def test_cli_login_lane_rejects_exec(monkeypatch, tmp_path):
+def test_cli_login_lane_rejects_persistent_fetch(monkeypatch, tmp_path):
     _proxy_env(monkeypatch, tmp_path)
     monkeypatch.setattr(
         "warden.proxy.subprocess.run",
         lambda *a, **k: (_ for _ in ()).throw(AssertionError("should not run")),
     )
     r = runner.invoke(
-        app, ["access", "login oidc", "--domain", "coulomb_social", "--exec", "--", "true"]
+        app, ["access", "login oidc", "--domain", "coulomb_social", "--fetch"]
     )
     assert r.exit_code == 2
+    assert "requires --exec" in r.output
+
+
+def test_attended_login_refuses_read_only_home_before_auth(monkeypatch, tmp_path):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    monkeypatch.setattr(
+        "warden.proxy.subprocess.run",
+        lambda *a, **k: (_ for _ in ()).throw(AssertionError("OIDC started")),
+    )
+    tmp_path.chmod(0o555)
+    try:
+        with pytest.raises(ProxyError, match="writable default home"):
+            proxy_attended_login_exec(
+                ResolvedFetch(argv=["bao", "login", "-no-print"]),
+                child_argv=["true"],
+            )
+    finally:
+        tmp_path.chmod(0o700)
+
+
+def test_attended_login_persistence_failure_revokes_and_cleans(monkeypatch, tmp_path):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    calls = []
+
+    def fake_run(argv, **kw):
+        calls.append(argv)
+        # Login succeeds but the pre-created helper remains empty: persistence failed.
+        return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")
+
+    monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
+    with pytest.raises(ProxyError, match="failed closed before command handoff"):
+        proxy_attended_login_exec(
+            ResolvedFetch(argv=["bao", "login", "-no-print"]),
+            child_argv=["should-not-run"],
+        )
+    assert calls == [
+        ["bao", "login", "-no-print", "-format=json"],
+        ["bao", "token", "revoke", "-self"],
+    ]
+    assert not (tmp_path / ".warden-attended-login").exists()
+
+
+@pytest.mark.parametrize("stream", ["stdout", "stderr"])
+def test_attended_login_unexpected_output_is_contained_revoked_and_cleaned(
+    monkeypatch, tmp_path, capsys, stream
+):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    sentinel = "hvs.NONPRODUCTION_SENTINEL"
+    calls = []
+
+    def fake_run(argv, **kw):
+        calls.append((argv, dict(kw["env"])))
+        if argv[:2] == ["bao", "login"]:
+            output = sentinel.encode()
+            return subprocess.CompletedProcess(
+                argv,
+                0,
+                stdout=output if stream == "stdout" else b"",
+                stderr=output if stream == "stderr" else b"",
+            )
+        if argv[:3] == ["bao", "token", "revoke"]:
+            # The helper is empty. The second contained attempt uses the captured
+            # value only through BAO_TOKEN, never argv or visible output.
+            return subprocess.CompletedProcess(
+                argv,
+                0 if kw["env"].get("BAO_TOKEN") == sentinel else 1,
+                stdout=b"",
+                stderr=b"",
+            )
+        raise AssertionError("reviewed child ran after unexpected login output")
+
+    monkeypatch.setattr("warden.proxy.subprocess.run", fake_run)
+    with pytest.raises(ProxyError, match="failed closed before command handoff") as exc:
+        proxy_attended_login_exec(
+            ResolvedFetch(argv=["bao", "login", "-no-print"]),
+            child_argv=["should-not-run"],
+        )
+    captured = capsys.readouterr()
+    assert sentinel not in str(exc.value)
+    assert sentinel not in captured.out
+    assert sentinel not in captured.err
+    assert all(sentinel not in " ".join(argv) for argv, _ in calls)
+    assert calls[-1][1]["BAO_TOKEN"] == sentinel
+    assert not (tmp_path / ".warden-attended-login").exists()
 
 
 def test_real_catalog_login_entry_is_login_lane():