Keep ambient tokens out of the service-JWT OpenBao login
The anonymous JWT login inherited BAO_TOKEN/VAULT_TOKEN and the bao CLI token helper, so it presented an env/operator token alongside the JWT. Run that login with no env token and a token helper that returns none, proven against the real bao CLI. This closes secrets-engine's part of the CCR-2026-0038 no-fallback check. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 544698@bnt-lap001 Assistant-Session: 279b8520-335f-4041-ba2a-9491335473db
This commit is contained in:
parent
1732c419d2
commit
9ec4e0bea5
4 changed files with 112 additions and 19 deletions
|
|
@ -115,6 +115,28 @@ class ScopedTokenSession:
|
|||
self.closed = True
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _unchanged(env: dict[str, str]):
|
||||
yield env
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _no_ambient_token(env: dict[str, str]):
|
||||
"""Strip env tokens and point the CLI at a token helper that returns none."""
|
||||
helper = shutil.which("true")
|
||||
if not helper:
|
||||
raise BackendError("cannot disable the bao token helper; refusing login")
|
||||
run_env = {
|
||||
k: v for k, v in env.items() if k not in ("BAO_TOKEN", "VAULT_TOKEN")
|
||||
}
|
||||
with tempfile.TemporaryDirectory(prefix="se-bao-anon-") as tmp:
|
||||
config = Path(tmp) / "cli.hcl"
|
||||
config.write_text(f'token_helper = "{helper}"\n', encoding="utf-8")
|
||||
run_env["BAO_CONFIG_PATH"] = str(config)
|
||||
run_env["VAULT_CONFIG_PATH"] = str(config)
|
||||
yield run_env
|
||||
|
||||
|
||||
@dataclass
|
||||
class OpenBaoClient:
|
||||
addr: str
|
||||
|
|
@ -135,7 +157,9 @@ class OpenBaoClient:
|
|||
|
||||
# -- low level ---------------------------------------------------------
|
||||
|
||||
def _run(self, args: list[str], *, stdin: str | None = None) -> subprocess.CompletedProcess:
|
||||
def _run(
|
||||
self, args: list[str], *, stdin: str | None = None, anonymous: bool = False
|
||||
) -> subprocess.CompletedProcess:
|
||||
if not self.bao_bin:
|
||||
raise BackendError(
|
||||
"no 'bao' (or 'vault') CLI on PATH; cannot reach OpenBao backend"
|
||||
|
|
@ -146,22 +170,25 @@ class OpenBaoClient:
|
|||
if self.token:
|
||||
env["BAO_TOKEN"] = self.token
|
||||
env["VAULT_TOKEN"] = self.token
|
||||
try:
|
||||
return subprocess.run(
|
||||
[self.bao_bin, *args],
|
||||
input=stdin,
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
except FileNotFoundError as e:
|
||||
raise BackendError(f"backend binary not runnable: {e}") from e
|
||||
except subprocess.TimeoutExpired as e:
|
||||
raise BackendError(f"backend call timed out: {' '.join(args)}") from e
|
||||
with _no_ambient_token(env) if anonymous else _unchanged(env) as run_env:
|
||||
try:
|
||||
return subprocess.run(
|
||||
[self.bao_bin, *args],
|
||||
input=stdin,
|
||||
env=run_env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
except FileNotFoundError as e:
|
||||
raise BackendError(f"backend binary not runnable: {e}") from e
|
||||
except subprocess.TimeoutExpired as e:
|
||||
raise BackendError(f"backend call timed out: {' '.join(args)}") from e
|
||||
|
||||
def _run_ok(self, args: list[str], *, stdin: str | None = None) -> str:
|
||||
proc = self._run(args, stdin=stdin)
|
||||
def _run_ok(
|
||||
self, args: list[str], *, stdin: str | None = None, anonymous: bool = False
|
||||
) -> str:
|
||||
proc = self._run(args, stdin=stdin, anonymous=anonymous)
|
||||
if proc.returncode != 0:
|
||||
# stderr from bao does not contain the secret value for these calls.
|
||||
raise BackendError(
|
||||
|
|
@ -170,7 +197,9 @@ class OpenBaoClient:
|
|||
)
|
||||
return proc.stdout
|
||||
|
||||
def _run_ok_with_json_file(self, args: list[str], payload: dict) -> str:
|
||||
def _run_ok_with_json_file(
|
||||
self, args: list[str], payload: dict, *, anonymous: bool = False
|
||||
) -> str:
|
||||
"""Run a CLI operation with JSON from a strict temporary file reference."""
|
||||
fd, temp_name = tempfile.mkstemp(prefix="se-bao-input-", suffix=".json")
|
||||
temp_path = Path(temp_name)
|
||||
|
|
@ -180,6 +209,8 @@ class OpenBaoClient:
|
|||
fd = -1
|
||||
json.dump(payload, fh)
|
||||
fh.write("\n")
|
||||
if anonymous:
|
||||
return self._run_ok([*args, f"@{temp_path}"], anonymous=True)
|
||||
return self._run_ok([*args, f"@{temp_path}"])
|
||||
finally:
|
||||
if fd >= 0:
|
||||
|
|
@ -332,10 +363,16 @@ class OpenBaoClient:
|
|||
self._run_ok(["delete", f"auth/approle/role/{role_name}"])
|
||||
|
||||
def login_jwt(self, mount: str, role: str, jwt: str) -> ScopedTokenSession:
|
||||
"""Login through JWT auth without putting the JWT in argv."""
|
||||
"""Login through JWT auth without putting the JWT in argv.
|
||||
|
||||
The login carries no ambient token: neither BAO_TOKEN/VAULT_TOKEN nor a
|
||||
CLI token helper is consulted, so a service-jwt login never presents an
|
||||
env or bootstrap credential alongside the JWT.
|
||||
"""
|
||||
output = self._run_ok_with_json_file(
|
||||
["write", "-format=json", f"auth/{mount}/login"],
|
||||
{"role": role, "jwt": jwt},
|
||||
anonymous=True,
|
||||
)
|
||||
try:
|
||||
auth = json.loads(output)["auth"]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue