tegwick
c4504c6de9
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0217e-8c4c-7383-be6b-f50a6e485306
456 lines
17 KiB
Python
456 lines
17 KiB
Python
"""OpenBao backend adapter.
|
|
|
|
Thin wrapper over the `bao` CLI. Isolated here so the rest of the engine speaks
|
|
in lanes/plans, not in OpenBao endpoint quirks (FR: "isolate backend adapter").
|
|
|
|
Auth resolution order for the token:
|
|
1. explicit bootstrap token file (--bootstrap-token-file), mode-checked;
|
|
2. BAO_TOKEN / VAULT_TOKEN environment variable;
|
|
3. otherwise unauthenticated (only dry-run / read-health works).
|
|
|
|
This adapter NEVER returns a secret value to its callers except through the
|
|
narrow `read_field_present()` (boolean) and the exec-delivery path, which writes
|
|
straight into a child process and never logs.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from secrets_engine.errors import BackendError, ProvisioningError
|
|
from secrets_engine.safe_paths import containing_git_worktree
|
|
|
|
|
|
def read_strict_token_file(path: Path, *, purpose: str = "token") -> str:
|
|
"""Read token material only from a mode-0600 path outside Git worktrees."""
|
|
if not path.exists():
|
|
raise ProvisioningError(f"{purpose} file not found: {path}")
|
|
st = path.stat()
|
|
if st.st_mode & 0o077:
|
|
raise ProvisioningError(
|
|
f"{purpose} file {path} is group/other-accessible "
|
|
f"(mode {oct(st.st_mode & 0o777)}); must be 0600"
|
|
)
|
|
# Refuse a token file living inside a Git worktree.
|
|
worktree = containing_git_worktree(path)
|
|
if worktree is not None:
|
|
raise ProvisioningError(
|
|
f"{purpose} file {path} is inside a Git worktree ({worktree}); "
|
|
"store it outside any repo"
|
|
)
|
|
token = path.read_text(encoding="utf-8").strip()
|
|
if not token:
|
|
raise ProvisioningError(f"{purpose} file {path} is empty")
|
|
return token
|
|
|
|
|
|
def _check_token_file(path: Path) -> str:
|
|
"""Compatibility wrapper for bootstrap authentication input."""
|
|
return read_strict_token_file(path, purpose="bootstrap token")
|
|
|
|
|
|
@dataclass
|
|
class ScopedTokenSession:
|
|
"""One AppRole login token that revokes itself on close."""
|
|
|
|
client: "OpenBaoClient"
|
|
accessor_fingerprint: str
|
|
closed: bool = False
|
|
revocation_attempted: bool = False
|
|
revocation_succeeded: bool = False
|
|
|
|
def evidence(self) -> dict[str, object]:
|
|
"""Return non-secret lifecycle evidence for this issued session."""
|
|
return {
|
|
"session_handle": self.accessor_fingerprint,
|
|
"established": True,
|
|
"revocation_attempted": self.revocation_attempted,
|
|
"revocation_succeeded": self.revocation_succeeded,
|
|
}
|
|
|
|
def close(self) -> None:
|
|
if self.closed:
|
|
return
|
|
self.revocation_attempted = True
|
|
try:
|
|
self.client._run_ok(["token", "revoke", "-self"])
|
|
self.revocation_succeeded = True
|
|
finally:
|
|
# Drop the credential from the reusable client object even when
|
|
# backend cleanup fails; TTL/use limits remain the final backstop.
|
|
self.client.token = ""
|
|
self.closed = True
|
|
|
|
|
|
@dataclass
|
|
class OpenBaoClient:
|
|
addr: str
|
|
token: str = ""
|
|
bao_bin: str = ""
|
|
|
|
@classmethod
|
|
def resolve(
|
|
cls, addr: str, *, bootstrap_token_file: str | Path | None = None
|
|
) -> "OpenBaoClient":
|
|
token = ""
|
|
if bootstrap_token_file:
|
|
token = _check_token_file(Path(bootstrap_token_file))
|
|
else:
|
|
token = os.environ.get("BAO_TOKEN", os.environ.get("VAULT_TOKEN", ""))
|
|
bao_bin = shutil.which("bao") or shutil.which("vault") or ""
|
|
return cls(addr=addr, token=token, bao_bin=bao_bin)
|
|
|
|
# -- low level ---------------------------------------------------------
|
|
|
|
def _run(self, args: list[str], *, stdin: str | None = None) -> subprocess.CompletedProcess:
|
|
if not self.bao_bin:
|
|
raise BackendError(
|
|
"no 'bao' (or 'vault') CLI on PATH; cannot reach OpenBao backend"
|
|
)
|
|
env = dict(os.environ)
|
|
env["BAO_ADDR"] = self.addr
|
|
env["VAULT_ADDR"] = self.addr
|
|
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
|
|
|
|
def _run_ok(self, args: list[str], *, stdin: str | None = None) -> str:
|
|
proc = self._run(args, stdin=stdin)
|
|
if proc.returncode != 0:
|
|
# stderr from bao does not contain the secret value for these calls.
|
|
raise BackendError(
|
|
f"bao {' '.join(args[:2])} failed (exit {proc.returncode}): "
|
|
f"{proc.stderr.strip() or proc.stdout.strip()}"
|
|
)
|
|
return proc.stdout
|
|
|
|
def _run_ok_with_json_file(self, args: list[str], payload: dict) -> 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)
|
|
try:
|
|
os.fchmod(fd, 0o600)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
|
fd = -1
|
|
json.dump(payload, fh)
|
|
fh.write("\n")
|
|
return self._run_ok([*args, f"@{temp_path}"])
|
|
finally:
|
|
if fd >= 0:
|
|
os.close(fd)
|
|
try:
|
|
temp_path.unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
# -- health / capabilities --------------------------------------------
|
|
|
|
def is_reachable(self) -> bool:
|
|
if not self.bao_bin:
|
|
return False
|
|
proc = self._run(["status", "-format=json"])
|
|
# status returns non-zero when sealed but still reachable; treat any
|
|
# parseable JSON as reachable.
|
|
try:
|
|
json.loads(proc.stdout or "{}")
|
|
return True
|
|
except json.JSONDecodeError:
|
|
return proc.returncode == 0
|
|
|
|
# -- policies ----------------------------------------------------------
|
|
|
|
def write_policy(self, name: str, hcl: str) -> None:
|
|
self._run_ok(["policy", "write", name, "-"], stdin=hcl)
|
|
|
|
def read_policy(self, name: str) -> str | None:
|
|
proc = self._run(["policy", "read", name])
|
|
if proc.returncode != 0:
|
|
return None
|
|
return proc.stdout
|
|
|
|
# -- approle -----------------------------------------------------------
|
|
|
|
def ensure_approle_enabled(self) -> None:
|
|
proc = self._run(["auth", "list", "-format=json"])
|
|
if proc.returncode == 0:
|
|
try:
|
|
methods = json.loads(proc.stdout)
|
|
if "approle/" in methods:
|
|
return
|
|
except json.JSONDecodeError:
|
|
pass
|
|
enable = self._run(["auth", "enable", "approle"])
|
|
if enable.returncode == 0:
|
|
return
|
|
stderr = enable.stderr or ""
|
|
if "already in use" in stderr:
|
|
return
|
|
if "permission denied" in stderr.lower():
|
|
return
|
|
raise BackendError(f"could not enable approle: {stderr.strip()}")
|
|
|
|
def write_approle(
|
|
self,
|
|
role_name: str,
|
|
policies: list[str],
|
|
ttl: str = "30m",
|
|
*,
|
|
max_ttl: str | None = None,
|
|
secret_id_ttl: str | None = None,
|
|
secret_id_num_uses: int = 0,
|
|
token_num_uses: int = 0,
|
|
) -> None:
|
|
args = [
|
|
"write",
|
|
f"auth/approle/role/{role_name}",
|
|
f"token_policies={','.join(policies)}",
|
|
f"token_ttl={ttl}",
|
|
f"token_max_ttl={max_ttl or ttl}",
|
|
f"secret_id_num_uses={secret_id_num_uses}",
|
|
f"token_num_uses={token_num_uses}",
|
|
]
|
|
if secret_id_ttl:
|
|
args.append(f"secret_id_ttl={secret_id_ttl}")
|
|
self._run_ok(args)
|
|
|
|
def approle_exists(self, role_name: str) -> bool:
|
|
proc = self._run(["read", f"auth/approle/role/{role_name}"])
|
|
return proc.returncode == 0
|
|
|
|
def read_approle_role_id(self, role_name: str) -> str:
|
|
return self._run_ok(
|
|
["read", "-field=role_id", f"auth/approle/role/{role_name}/role-id"]
|
|
).strip()
|
|
|
|
def create_approle_secret_id(self, role_name: str) -> str:
|
|
return self._run_ok(
|
|
["write", "-field=secret_id", "-f", f"auth/approle/role/{role_name}/secret-id"]
|
|
).strip()
|
|
|
|
def create_approle_session(self, role_name: str) -> ScopedTokenSession:
|
|
"""Login through AppRole without putting role/secret ids in argv."""
|
|
role_id = self.read_approle_role_id(role_name)
|
|
secret_id = self.create_approle_secret_id(role_name)
|
|
try:
|
|
output = self._run_ok_with_json_file(
|
|
["write", "-format=json", "auth/approle/login"],
|
|
{"role_id": role_id, "secret_id": secret_id},
|
|
)
|
|
finally:
|
|
del secret_id
|
|
try:
|
|
auth = json.loads(output)["auth"]
|
|
token = str(auth["client_token"])
|
|
accessor = str(auth.get("accessor", ""))
|
|
except (json.JSONDecodeError, KeyError, TypeError) as e:
|
|
raise BackendError("malformed AppRole login response") from e
|
|
if not token:
|
|
raise BackendError("AppRole login returned an empty token")
|
|
import hashlib
|
|
|
|
fingerprint = hashlib.sha256(accessor.encode("utf-8")).hexdigest()[:12]
|
|
scoped = OpenBaoClient(addr=self.addr, token=token, bao_bin=self.bao_bin)
|
|
del token
|
|
return ScopedTokenSession(
|
|
client=scoped,
|
|
accessor_fingerprint=fingerprint,
|
|
)
|
|
|
|
@contextmanager
|
|
def approle_session(self, role_name: str):
|
|
"""Yield a scoped client and revoke its token on every exit path."""
|
|
session = self.create_approle_session(role_name)
|
|
try:
|
|
yield session
|
|
finally:
|
|
session.close()
|
|
|
|
def token_capabilities(self, path: str, *, token: str) -> list[str]:
|
|
"""Return token capabilities for a path without returning any secret value."""
|
|
client = OpenBaoClient(addr=self.addr, token=token, bao_bin=self.bao_bin)
|
|
out = client._run_ok(["token", "capabilities", "-format=json", path])
|
|
try:
|
|
data = json.loads(out)
|
|
except json.JSONDecodeError:
|
|
return [line.strip() for line in out.splitlines() if line.strip()]
|
|
if isinstance(data, list):
|
|
return [str(item) for item in data]
|
|
if isinstance(data, dict):
|
|
caps = data.get("capabilities", [])
|
|
if isinstance(caps, list):
|
|
return [str(item) for item in caps]
|
|
return []
|
|
|
|
def delete_policy(self, name: str) -> None:
|
|
self._run_ok(["policy", "delete", name])
|
|
|
|
def delete_approle(self, role_name: str) -> None:
|
|
self._run_ok(["delete", f"auth/approle/role/{role_name}"])
|
|
|
|
# -- KV v2 -------------------------------------------------------------
|
|
|
|
def kv_mount_exists(self, mount: str) -> bool:
|
|
proc = self._run(["secrets", "list", "-format=json"])
|
|
if proc.returncode != 0:
|
|
return False
|
|
try:
|
|
return f"{mount}/" in json.loads(proc.stdout)
|
|
except json.JSONDecodeError:
|
|
return False
|
|
|
|
def ensure_kv_mount(self, mount: str) -> None:
|
|
if self.kv_mount_exists(mount):
|
|
return
|
|
enable = self._run(["secrets", "enable", "-path", mount, "kv-v2"])
|
|
if enable.returncode == 0:
|
|
return
|
|
stderr = enable.stderr or ""
|
|
if "already in use" in stderr:
|
|
return
|
|
# Stage roles (e.g. secrets-engine-prod) cannot manage sys/mounts. When the
|
|
# caller also cannot list mounts, a permission-denied enable means the mount
|
|
# was operator-preprovisioned — continue to policy/approle apply.
|
|
if "permission denied" in stderr.lower():
|
|
return
|
|
raise BackendError(f"could not enable kv at {mount}: {stderr.strip()}")
|
|
|
|
def kv_current_version(self, mount: str, path: str) -> int:
|
|
"""Return current KV v2 version from metadata, or 0 when absent.
|
|
|
|
Metadata contains no secret values. Permission/backend errors fail closed
|
|
rather than being misread as a new path.
|
|
"""
|
|
proc = self._run(
|
|
["kv", "metadata", "get", "-format=json", f"-mount={mount}", path]
|
|
)
|
|
if proc.returncode != 0:
|
|
message = f"{proc.stderr}\n{proc.stdout}".lower()
|
|
if "no value found" in message or "not found" in message or "404" in message:
|
|
return 0
|
|
raise BackendError(
|
|
f"bao kv metadata get failed (exit {proc.returncode}): "
|
|
f"{proc.stderr.strip() or proc.stdout.strip()}"
|
|
)
|
|
try:
|
|
version = json.loads(proc.stdout).get("data", {}).get("current_version", 0)
|
|
return int(version)
|
|
except (json.JSONDecodeError, TypeError, ValueError) as e:
|
|
raise BackendError("malformed KV metadata response") from e
|
|
|
|
def kv_patch_fields(
|
|
self,
|
|
mount: str,
|
|
path: str,
|
|
values: dict[str, str],
|
|
*,
|
|
expected_version: int | None = None,
|
|
) -> int:
|
|
"""Atomically update declared fields without exposing values in argv.
|
|
|
|
Existing paths use server-side HTTP PATCH with CAS, preserving every
|
|
unmentioned sibling. New paths use CAS=0 put. Values are written to a
|
|
mode-0600 temporary JSON input file referenced by path and removed in a
|
|
finally block. The returned integer is the version used as the CAS base.
|
|
"""
|
|
if not values:
|
|
raise BackendError("refusing empty KV field update")
|
|
if any(not isinstance(k, str) or not k for k in values):
|
|
raise BackendError("KV field update contains an invalid field name")
|
|
version = (
|
|
self.kv_current_version(mount, path)
|
|
if expected_version is None
|
|
else expected_version
|
|
)
|
|
if version < 0:
|
|
raise BackendError("expected KV version must be non-negative")
|
|
|
|
if version == 0:
|
|
args = [
|
|
"kv",
|
|
"put",
|
|
f"-mount={mount}",
|
|
"-cas=0",
|
|
path,
|
|
]
|
|
else:
|
|
args = [
|
|
"kv",
|
|
"patch",
|
|
f"-mount={mount}",
|
|
"-method=patch",
|
|
f"-cas={version}",
|
|
path,
|
|
]
|
|
self._run_ok_with_json_file(args, values)
|
|
return version
|
|
|
|
def kv_metadata_exists(self, mount: str, path: str) -> bool:
|
|
proc = self._run(["kv", "metadata", "get", "-format=json", f"{mount}/{path}"])
|
|
return proc.returncode == 0
|
|
|
|
def kv_fields_present(
|
|
self,
|
|
mount: str,
|
|
path: str,
|
|
fields: list[str] | tuple[str, ...],
|
|
*,
|
|
token: str | None = None,
|
|
) -> dict[str, bool]:
|
|
"""Return declared-field presence after one read, never field values.
|
|
|
|
If ``token`` is given, the read is attempted as that scoped token. A
|
|
denied or malformed read reports every requested field as absent; no
|
|
response data is returned to the caller.
|
|
"""
|
|
requested = tuple(dict.fromkeys(fields))
|
|
if not requested:
|
|
return {}
|
|
client = self
|
|
if token is not None:
|
|
client = OpenBaoClient(addr=self.addr, token=token, bao_bin=self.bao_bin)
|
|
proc = client._run(["kv", "get", "-format=json", f"{mount}/{path}"])
|
|
if proc.returncode != 0:
|
|
return {field: False for field in requested}
|
|
try:
|
|
doc = json.loads(proc.stdout)
|
|
except json.JSONDecodeError:
|
|
return {field: False for field in requested}
|
|
data = doc.get("data", {}).get("data", {})
|
|
if not isinstance(data, dict):
|
|
return {field: False for field in requested}
|
|
return {field: field in data and bool(data[field]) for field in requested}
|
|
|
|
def kv_field_present(
|
|
self, mount: str, path: str, field: str, *, token: str | None = None
|
|
) -> bool:
|
|
"""Compatibility wrapper for a single non-secret presence result."""
|
|
return self.kv_fields_present(mount, path, [field], token=token)[field]
|
|
|
|
def kv_can_read(self, mount: str, path: str, *, token: str) -> bool:
|
|
"""True iff `token` is permitted to read the path at all (no value used)."""
|
|
client = OpenBaoClient(addr=self.addr, token=token, bao_bin=self.bao_bin)
|
|
proc = client._run(["kv", "get", "-format=json", f"{mount}/{path}"])
|
|
return proc.returncode == 0
|
|
|
|
def kv_delete_metadata(self, mount: str, path: str) -> None:
|
|
self._run_ok(["kv", "metadata", "delete", f"{mount}/{path}"])
|