fix: contain attended OpenBao login output

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

registry/routing/catalog.yaml

@@ -147,8 +147,9 @@ entries:
   - id: openbao-platform-admin-login
     title: Attended OpenBao platform administration login
     # This is an identity bootstrap, not a secret value lane. The authority it
-    # establishes is high-risk, but the command prints no token and writes only
-    # to the caller's normal OpenBao token helper.
+    # establishes is high-risk. Safety does not rely on -no-print: Warden must
+    # preflight a private writable helper, contain both output streams, run only
+    # the reviewed child command, self-revoke, and remove the helper.
     risk: high
     workload_ref:
       applicability: not-applicable

src/warden/cli.py

@@ -1169,13 +1169,20 @@ def _access_json(entry, expanded, gate: str, domain: Optional[str]) -> dict:
             "ops-warden holds no token."
         )
     elif expanded.exec_capable:
-        verb = "fetch" if entry.lane != "login" else "login"
-        payload["next_action"] = (
-            f"ops-warden can proxy this {verb} as the caller: "
-            f"`warden access <need> --fetch`"
-            + ("" if entry.lane == "login" else " (or `--exec -- <cmd>`)")
-            + f". Runs {entry.owner_repo}'s tool with your identity; ops-warden holds no value."
-        )
+        if entry.lane == "login":
+            payload["next_action"] = (
+                "Run the attended login and reviewed operation only inside the "
+                "contained envelope: `warden access <need> --exec -- "
+                "<reviewed-command>`. The private helper is preflighted, all output "
+                "is suppressed, and the session is revoked and removed afterward."
+            )
+        else:
+            payload["next_action"] = (
+                "ops-warden can proxy this fetch as the caller: "
+                "`warden access <need> --fetch` (or `--exec -- <cmd>`). "
+                f"Runs {entry.owner_repo}'s tool with your identity; "
+                "ops-warden holds no value."
+            )
     else:
         payload["next_action"] = (
             f"obtain from {entry.owner_repo} ({entry.subsystem}); "
@@ -1210,6 +1217,7 @@ def _access_proxy(
         build_wrapped_fetch,
         caller_auth_present,
         proxy_exec,
+        proxy_attended_login_exec,
         proxy_fetch,
         proxy_fetch_fingerprint,
         proxy_fetch_to_file,
@@ -1240,18 +1248,17 @@ def _access_proxy(
     decision_id = None
 
     if is_login:
-        # Login lane: interactive auth bootstrap. No caller-auth precheck (you have no
-        # token yet — that's the point) and no secret-read gate (it needs an identity
-        # this flow establishes). --exec is meaningless here.
-        if do_exec:
+        # Login lane: the authentication and reviewed child command share one
+        # isolated token-helper home. No credential may persist beyond that child.
+        if not do_exec or not child_argv:
             err.print(
-                "[red]--exec is not valid for a login lane[/red] "
-                f"({entry.id!r} is interactive auth). Use --fetch."
+                "[red]A login lane requires --exec -- <reviewed-command>[/red] "
+                f"({entry.id!r} cannot create a persistent login-only handoff)."
             )
             raise typer.Exit(2)
         err.print(
-            "[dim]login lane — interactive auth bootstrap; no secret-read gate, "
-            "token stays in the caller's own store.[/dim]"
+            "[dim]login lane — contained OIDC and reviewed command; private helper, "
+            "suppressed output, deterministic self-revocation.[/dim]"
         )
     else:
         if no_policy:
@@ -1343,7 +1350,9 @@ def _access_proxy(
         f"(caller identity; value not persisted)[/dim]"
     )
     try:
-        if do_exec:
+        if is_login:
+            rc = proxy_attended_login_exec(resolved, child_argv=child_argv)
+        elif do_exec:
             if not child_argv:
                 err.print("[red]--exec needs a command after `--`[/red], e.g. `-- npm publish`.")
                 raise typer.Exit(2)
@@ -1524,8 +1533,13 @@ def access(
         console.print(f"  auth     : {expanded.auth_method}")
     if expanded.path_template:
         console.print(f"  path     : {expanded.path_template}")
-    if expanded.fetch_command:
+    if expanded.fetch_command and entry.lane != "login":
         console.print(f"  fetch    : {expanded.fetch_command}")
+    elif expanded.fetch_command:
+        console.print(
+            "  login    : [dim]internal to the contained --exec envelope; "
+            "do not invoke separately[/dim]"
+        )
     if expanded.policy_ref:
         console.print(f"  policy   : {expanded.policy_ref}   [dim]({gate})[/dim]")
     console.print(f"  wiki     : {entry.wiki_ref}")
@@ -1544,12 +1558,16 @@ def access(
             console.print(f"  pointer  : [dim]{entry.pointer_command}[/dim]")
     if expanded.exec_capable:
         label = "fallback" if entry.has_native_exec else "proxy"
-        hint = (
-            "transparent conduit — fetches as you"
-            if entry.lane != "login"
-            else "runs the interactive login as you"
-        )
-        console.print(f"  {label:<8} : [dim]{proxy} --fetch[/dim]  [yellow]({hint})[/yellow]")
+        if entry.lane == "login":
+            console.print(
+                f"  {label:<8} : [dim]{proxy} --exec -- <reviewed-command>[/dim]  "
+                "[yellow](contained login + command; output suppressed)[/yellow]"
+            )
+        else:
+            console.print(
+                f"  {label:<8} : [dim]{proxy} --fetch[/dim]  "
+                "[yellow](transparent conduit — fetches as you)[/yellow]"
+            )
     if expanded.path_template and "<" in expanded.path_template:
         console.print(
             "  note     : remaining <…> placeholders are owner-confirmed names "
@@ -1564,14 +1582,21 @@ def access(
             "conduit (runs the fetch as you, holds nothing)."
         )
     elif expanded.exec_capable:
-        verb = "fetch this for you" if entry.lane != "login" else "run this login for you"
-        console.print(
-            f"\n[green]ops-warden can {verb}[/green] as the caller — "
-            f"[bold]{proxy} --fetch[/bold]"
-            + ("" if entry.lane == "login" else f" (or [bold]{proxy} --exec -- <cmd>[/bold])")
-            + f". It runs {entry.owner_repo}'s tool with [bold]your[/bold] identity; the "
-            "value streams to you and ops-warden never holds, caches, or logs it."
-        )
+        if entry.lane == "login":
+            console.print(
+                "\n[green]Contained attended login[/green] — "
+                f"[bold]{proxy} --exec -- <reviewed-command>[/bold]. The login, "
+                "command, and revocation use a private helper with suppressed output; "
+                "the session is removed afterward."
+            )
+        else:
+            console.print(
+                "\n[green]ops-warden can fetch this for you[/green] as the caller — "
+                f"[bold]{proxy} --fetch[/bold] (or "
+                f"[bold]{proxy} --exec -- <cmd>[/bold]). It runs "
+                f"{entry.owner_repo}'s tool with [bold]your[/bold] identity; the "
+                "value streams to you and ops-warden never holds, caches, or logs it."
+            )
     else:
         console.print(
             f"\n[yellow]ops-warden does not hold this secret.[/yellow] "

src/warden/plan.py

@@ -142,9 +142,13 @@ def _autonomous_commands(entry: RouteEntry, domain: Optional[str]) -> List[str]:
 
     expanded = expand_handoff(entry, domain=domain)
     if entry.lane == "login":
-        cmds.append(f"warden access {entry.id} --fetch")
+        cmds.append(
+            f"warden access {entry.id} --exec -- <reviewed-command>"
+        )
         if expanded.fetch_command:
-            cmds.append(f"# attended owner login: {expanded.fetch_command}")
+            cmds.append(
+                "# owner login is contained by warden; do not invoke it separately"
+            )
         return cmds
     if entry.has_native_exec and entry.exec_command:
         cmds.append(entry.exec_command)
@@ -175,14 +179,20 @@ def _autonomous_commands(entry: RouteEntry, domain: Optional[str]) -> List[str]:
 def _founder_for_entry(entry: RouteEntry, need: str, domain: Optional[str]) -> FounderAct:
     expanded = expand_handoff(entry, domain=domain)
     if entry.lane == "login":
+        contained_command = (
+            f"warden access {entry.id} --exec -- <reviewed-command>"
+        )
         return FounderAct(
             kind="oidc_login",
             summary=f"Interactive OIDC/MFA login via {entry.owner_repo}",
             details={
                 "lane_id": entry.id,
                 "auth_method": expanded.auth_method,
-                "fetch_command": expanded.fetch_command,
-                "desk_hint": f"warden desk --from-plan (act=oidc_login) or: {expanded.fetch_command}",
+                "fetch_command": contained_command,
+                "desk_hint": (
+                    "warden desk --from-plan (act=oidc_login); execute only through: "
+                    + contained_command
+                ),
             },
         )
     if entry.lane == "ceremony":

src/warden/proxy.py

@@ -8,11 +8,12 @@ intact. Three guardrails are enforced here in code:
   caller's own environment. ops-warden injects no token of its own; if the caller has
   no credential, the underlying tool fails and we surface the auth pointer. We never
   add a `*_TOKEN` warden owns to the child environment.
-* **G2 — transit only, no persistence/logging of values.** ``proxy_fetch`` runs the
-  tool with **inherited** stdout/stderr (never a pipe), so the value streams to the
-  caller and never enters warden's memory. ``proxy_exec`` reads the value solely to
-  place it in a child process's environment (the accepted proxy tradeoff) and never
-  writes it to disk or log. The audit record is metadata only.
+* **G2 — bounded transports, no logging of values.** Ordinary ``proxy_fetch`` runs
+  the tool with inherited stdout/stderr so the value never enters warden's memory;
+  sanctioned exec/file transports hold it only for their bounded handoff. The
+  high-risk attended-login lane is stricter: it captures every client byte inside
+  an isolated helper session, permits no output, self-revokes, and cleans up. Audit
+  records are metadata only.
 * **G3 — policy gate before fetch.** The CLI runs ``check_fetch_policy`` before
   calling anything here; this module refuses to run an unresolved command template.
 
@@ -24,7 +25,10 @@ import json
 import os
 import re
 import shlex
+import shutil
+import stat
 import subprocess
+import tempfile
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
@@ -33,6 +37,9 @@ from typing import List, Optional
 from warden.routing.models import RouteEntry
 
 _PLACEHOLDER = re.compile(r"<[^>]+>")
+_OPENBAO_TOKEN = re.compile(rb"\b(?:hvs|hvb|hvr)\.[A-Za-z0-9_-]{8,}\b")
+_ATTENDED_LOGIN_ROOT = ".warden-attended-login"
+_TOKEN_HELPER_NAME = ".vault-token"
 
 
 @dataclass(frozen=True)
@@ -205,6 +212,267 @@ def proxy_fetch(resolved: ResolvedFetch) -> int:
     return completed.returncode
 
 
+def _assert_owned_mode(path: Path, *, mode: int, directory: bool) -> None:
+    """Require a caller-owned, non-symlink path with an exact private mode."""
+    info = path.lstat()
+    expected_type = stat.S_ISDIR if directory else stat.S_ISREG
+    if not expected_type(info.st_mode) or stat.S_ISLNK(info.st_mode):
+        raise ProxyError("attended login private storage has an unsafe path type")
+    if hasattr(os, "getuid") and info.st_uid != os.getuid():
+        raise ProxyError("attended login private storage is not caller-owned")
+    if stat.S_IMODE(info.st_mode) != mode:
+        raise ProxyError("attended login private storage has an unsafe mode")
+
+
+def _prepare_attended_login_home() -> tuple[Path, Path, bool]:
+    """Create and prove an isolated token-helper home before authentication."""
+    home = Path.home()
+    try:
+        home_info = home.lstat()
+    except OSError as exc:
+        raise ProxyError(
+            "attended login requires a usable writable default home before OIDC"
+        ) from exc
+    if (
+        not stat.S_ISDIR(home_info.st_mode)
+        or stat.S_ISLNK(home_info.st_mode)
+        or stat.S_IMODE(home_info.st_mode) & 0o222 == 0
+    ):
+        raise ProxyError(
+            "attended login requires a usable writable default home before OIDC"
+        )
+
+    # Prove the default home itself is writable. A pre-existing writable child must
+    # not let a newly read-only HOME reach the OIDC process.
+    probe_fd = -1
+    probe_path: Path | None = None
+    probe_cleanup_error: OSError | None = None
+    try:
+        probe_fd, probe_name = tempfile.mkstemp(prefix=".warden-home-probe-", dir=home)
+        probe_path = Path(probe_name)
+        os.write(probe_fd, b"preflight")
+        os.fsync(probe_fd)
+    except OSError as exc:
+        raise ProxyError(
+            "attended login requires a usable writable default home before OIDC"
+        ) from exc
+    finally:
+        if probe_fd >= 0:
+            os.close(probe_fd)
+        if probe_path is not None:
+            try:
+                probe_path.unlink()
+            except OSError as exc:
+                probe_cleanup_error = exc
+    if probe_cleanup_error is not None:
+        raise ProxyError("attended login home preflight cleanup failed") from probe_cleanup_error
+
+    root = home / _ATTENDED_LOGIN_ROOT
+    root_created = False
+    try:
+        root.mkdir(mode=0o700)
+        root_created = True
+    except FileExistsError:
+        pass
+    except OSError as exc:
+        raise ProxyError("could not establish attended login private storage") from exc
+    _assert_owned_mode(root, mode=0o700, directory=True)
+
+    try:
+        session = Path(tempfile.mkdtemp(prefix="session-", dir=root))
+        session.chmod(0o700)
+        _assert_owned_mode(session, mode=0o700, directory=True)
+        helper = session / _TOKEN_HELPER_NAME
+        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
+        if hasattr(os, "O_NOFOLLOW"):
+            flags |= os.O_NOFOLLOW
+        fd = os.open(helper, flags, 0o600)
+        try:
+            # Exercise persistence before auth, then leave the helper empty for bao.
+            os.write(fd, b"preflight")
+            os.fsync(fd)
+            os.ftruncate(fd, 0)
+        finally:
+            os.close(fd)
+        _assert_owned_mode(helper, mode=0o600, directory=False)
+    except (OSError, ProxyError) as exc:
+        if "session" in locals():
+            shutil.rmtree(session, ignore_errors=True)
+        if root_created:
+            try:
+                root.rmdir()
+            except OSError:
+                pass
+        if isinstance(exc, ProxyError):
+            raise
+        raise ProxyError("could not establish attended login private storage") from exc
+    return root, session, root_created
+
+
+def _contained_run(argv: List[str], *, env: dict) -> subprocess.CompletedProcess:
+    """Run with both output streams captured and never forwarded."""
+    return subprocess.run(  # noqa: S603
+        argv,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        stdin=None,
+        env=env,
+        check=False,
+    )
+
+
+def _output_bytes(completed: subprocess.CompletedProcess) -> bytes:
+    stdout = completed.stdout
+    stderr = completed.stderr
+    if isinstance(stdout, str):
+        stdout = stdout.encode("utf-8", errors="replace")
+    if isinstance(stderr, str):
+        stderr = stderr.encode("utf-8", errors="replace")
+    stdout = stdout if isinstance(stdout, bytes) else b""
+    stderr = stderr if isinstance(stderr, bytes) else b""
+    return stdout + b"\n" + stderr
+
+
+def _revoke_contained(
+    bao_binary: str,
+    *,
+    env: dict,
+    possible_output: bytes,
+) -> bool:
+    """Attempt self-revocation without exposing helper or captured output."""
+    revoke_env = dict(env)
+    try:
+        first = _contained_run(
+            [bao_binary, "token", "revoke", "-self"], env=revoke_env
+        )
+    except OSError:
+        return False
+    if first.returncode == 0:
+        return True
+
+    # A helper-persistence failure can leave the issued token only in contained
+    # client output. Use it solely for immediate self-revocation, never for a log,
+    # return value, hash, fingerprint, file, or argv.
+    match = _OPENBAO_TOKEN.search(possible_output)
+    if match is None:
+        return False
+    token = match.group(0).decode("ascii")
+    revoke_env["BAO_TOKEN"] = token
+    revoke_env.pop("VAULT_TOKEN", None)
+    try:
+        try:
+            second = _contained_run(
+                [bao_binary, "token", "revoke", "-self"], env=revoke_env
+            )
+        except OSError:
+            return False
+        return second.returncode == 0
+    finally:
+        revoke_env.pop("BAO_TOKEN", None)
+        token = ""  # noqa: F841 - best-effort release of the credential reference
+
+
+def proxy_attended_login_exec(
+    resolved: ResolvedFetch,
+    *,
+    child_argv: List[str],
+) -> int:
+    """Run an attended login and one silent child inside a private helper home.
+
+    The default home is proven writable before the OIDC client starts. Login,
+    child, and revocation output are captured and discarded. Any non-empty output,
+    persistence defect, or non-zero result fails closed; any possibly issued token
+    is revoked before the isolated helper directory is removed.
+    """
+    if not child_argv:
+        raise ProxyError(
+            "attended login requires --exec -- <reviewed-command>; a persistent "
+            "login-only handoff is not permitted"
+        )
+    if (
+        resolved.argv is None
+        or len(resolved.argv) < 2
+        or Path(resolved.argv[0]).name != "bao"
+        or resolved.argv[1] != "login"
+    ):
+        raise ProxyError("attended login requires a direct bao login argv")
+
+    root, session, root_created = _prepare_attended_login_home()
+    helper = session / _TOKEN_HELPER_NAME
+    env = _caller_env()
+    env["HOME"] = str(session)
+    env.pop("BAO_TOKEN", None)
+    env.pop("VAULT_TOKEN", None)
+    for key in (
+        "BAO_LOG_LEVEL",
+        "BAO_LOG_FORMAT",
+        "VAULT_LOG_LEVEL",
+        "VAULT_LOG_FORMAT",
+    ):
+        env.pop(key, None)
+
+    login_argv = list(resolved.argv)
+    if not any(arg == "-format" or arg.startswith("-format=") for arg in login_argv):
+        login_argv.append("-format=json")
+
+    try:
+        try:
+            login = _contained_run(login_argv, env=env)
+        except OSError as exc:
+            raise ProxyError("attended login client could not start before OIDC") from exc
+        login_output = _output_bytes(login)
+        helper_valid = False
+        try:
+            _assert_owned_mode(helper, mode=0o600, directory=False)
+            helper_valid = helper.stat().st_size > 0
+        except (OSError, ProxyError):
+            helper_valid = False
+
+        if login.returncode != 0 or login_output.strip() or not helper_valid:
+            revoked = _revoke_contained(
+                resolved.argv[0], env=env, possible_output=login_output
+            )
+            status = "revoked" if revoked else "revocation could not be confirmed"
+            raise ProxyError(
+                "attended login failed closed before command handoff; any possible "
+                f"issued session was contained and {status}"
+            )
+
+        try:
+            child = _contained_run(child_argv, env=env)
+        except OSError as exc:
+            revoked = _revoke_contained(
+                resolved.argv[0], env=env, possible_output=b""
+            )
+            status = "revoked" if revoked else "revocation could not be confirmed"
+            raise ProxyError(
+                "attended command could not start; the login session was " + status
+            ) from exc
+
+        child_output = _output_bytes(child)
+        revoked = _revoke_contained(
+            resolved.argv[0], env=env, possible_output=child_output
+        )
+        if child.returncode != 0 or child_output.strip():
+            status = "revoked" if revoked else "revocation could not be confirmed"
+            raise ProxyError(
+                "attended command failed closed because it returned a failure or "
+                f"unexpected output; the login session was {status}"
+            )
+        if not revoked:
+            raise ProxyError(
+                "attended command completed but session revocation could not be confirmed"
+            )
+        return 0
+    finally:
+        try:
+            shutil.rmtree(session)
+            if root_created:
+                root.rmdir()
+        except OSError as exc:
+            raise ProxyError("attended login private storage cleanup failed") from exc
+
+
 def _capture_value(resolved: ResolvedFetch) -> str:
     """Run the fetch and return its stdout (the value) minus one trailing newline.
 

src/warden/worker.py

@@ -588,7 +588,13 @@ def draft_route_answer(query: str) -> str:
     elif e.has_native_exec:
         parts.append(f"Primary: {e.exec_command}.")
     elif e.exec_capable:
-        parts.append(f"Proxy: warden access {e.id} --fetch (as the caller).")
+        if e.lane == "login":
+            parts.append(
+                f"Contained login: warden access {e.id} --exec -- "
+                "<reviewed-command>."
+            )
+        else:
+            parts.append(f"Proxy: warden access {e.id} --fetch (as the caller).")
     parts.append(f"See {e.wiki_ref}.")
     return " ".join(parts)
 

tests/test_plan.py

@@ -65,19 +65,17 @@ def test_plan_first_time_openbao_database_admin_uses_platform_admin_login():
     assert plan.founder_act.kind == "oidc_login"
     command = plan.founder_act.details["fetch_command"]
     assert command == (
-        "bao login -no-print -method=oidc -path=netkingdom role=platform-admin"
+        "warden access openbao-platform-admin-login --exec -- <reviewed-command>"
     )
     assert "financials" not in command
     assert "paste_once" not in plan.founder_act.details["desk_hint"]
     assert any(
-        item == "warden access openbao-platform-admin-login --fetch"
+        item
+        == "warden access openbao-platform-admin-login --exec -- <reviewed-command>"
         for item in plan.commands
     )
-    assert not any(
-        flag in item
-        for item in plan.commands
-        for flag in ("--exec", "--out", "--wrap")
-    )
+    assert not any("--fetch" in item for item in plan.commands)
+    assert not any("--out" in item or "--wrap" in item for item in plan.commands)
 
 
 def test_plan_openbao_shamir_recovery_uses_approval_ceremony_not_secret_provision():

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():

tests/test_routing.py

@@ -962,11 +962,10 @@ def test_cli_route_gaps_fail_on_stale_exits_3(repo_catalog_env):
     assert result.exit_code == 3
     rows = json.loads(result.stdout)
     assert any(r["stale"] for r in rows)
-    # A freshly reviewed lane can still be stale because it was never verified;
-    # asked-and-waiting must not silently pass the gate as the calendar moves.
+    # An asked-and-waiting lane stays stale until it is verified, regardless of
+    # how many calendar days have elapsed since the request.
     assert any(
         r["stale"]
-        and r["days_since_review"] <= 1
         and r["verified"] == "asked-and-waiting"
         for r in rows
     )

wiki/playbooks/openbao-platform-admin-login.md

@@ -18,27 +18,35 @@ workload KV-read lane and it does not provision a secret value.
    `openbao-api-key`, a workload role, paste-once provisioning, or root, stop and
    report a routing defect.
 
-2. The operator performs the one identity act through KeyCape OIDC/MFA:
+2. The operator performs the identity act and the separately reviewed owner
+   command through one contained envelope:
 
    ```bash
-   bao login -no-print -method=oidc -path=netkingdom role=platform-admin
+   warden access openbao-platform-admin-login --exec -- <reviewed-owner-command>
    ```
 
-   `-no-print` is mandatory. Do not paste a token into chat, State Hub, a shell
-   argument, or a temporary handoff file. Root is offline break-glass authority,
-   not a fallback for an OIDC or callback failure.
+   Warden refuses a login-only `--fetch`. Before OIDC it proves the caller's
+   default home is usable, creates a caller-owned `0700` isolated home and a
+   `0600` token helper, and then runs `bao login -no-print` with both stdout and
+   stderr captured. The reviewed command runs in the same contained home with
+   both streams captured; it must persist any permitted metadata evidence itself
+   and remain silent. Warden self-revokes the session and removes the helper on
+   every success or failure path.
+
+   Safety does not rely on `-no-print`. Any client or child output, helper
+   persistence defect, non-zero exit, or revocation/cleanup defect fails closed.
+   Captured bytes are never returned, logged, excerpted, hashed, or fingerprinted.
+   Do not paste a token into chat, State Hub, a shell argument, or a handoff file.
+   Root is offline break-glass authority, not a fallback for failure.
 
 3. Verify authority using metadata or capabilities only, never by reading a
    secret value. Then run only the separately reviewed owner procedure. For the
    database engine this procedure lives in `rapp-postgres`; the login does not
    itself approve configuration changes.
 
-4. Revoke the attended token when the reviewed operation and its non-secret
-   verification are complete:
-
-   ```bash
-   bao token revoke -self
-   ```
+4. Confirm the contained command exits successfully. Warden performs and checks
+   `bao token revoke -self` inside the contained environment before cleanup; do
+   not retain or reuse the helper.
 
 If browser login fails before authentication, confirm the `netkingdom` auth
 mount, `platform-admin` role, and allowed callback with `railiance-platform` and