tegwick
afd1c8e593
secrets-engine wrap writes a single-use OpenBao wrap token to a mode-0600 out-of-repo file and never prints it. KV reads and AppRole secret_ids are wrapped with a 15m TTL cap. Unwrapped secret payloads fail closed. Production wrap remains fail-closed. Assistant: grok Assistant-Session: 01a05f07-ae72-7781-9fcb-19efd61add00
106 lines
3.2 KiB
Python
106 lines
3.2 KiB
Python
"""Response-wrapped operator handoff.
|
|
|
|
Writes a single-use wrap token to a mode-0600 file outside Git. The wrap token
|
|
is secret material and is never printed or recorded in evidence.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from secrets_engine.catalog import CatalogEntry
|
|
from secrets_engine.errors import ProvisioningError
|
|
from secrets_engine.openbao import OpenBaoClient, WrappedResponse
|
|
from secrets_engine.safe_paths import containing_git_worktree
|
|
|
|
_TTL_RE = re.compile(r"^([1-9][0-9]*)([smh])$")
|
|
MAX_WRAP_SECONDS = 15 * 60
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WrapHandoffResult:
|
|
catalog_id: str
|
|
kind: str
|
|
out_file: str
|
|
ttl: str
|
|
wrap_handle: str
|
|
creation_path: str
|
|
|
|
|
|
def normalize_wrap_ttl(value: str) -> str:
|
|
text = (value or "").strip()
|
|
match = _TTL_RE.fullmatch(text)
|
|
if not match:
|
|
raise ProvisioningError("wrap ttl must be like 60s or 15m")
|
|
amount = int(match.group(1))
|
|
unit = match.group(2)
|
|
seconds = amount * {"s": 1, "m": 60, "h": 3600}[unit]
|
|
if seconds > MAX_WRAP_SECONDS:
|
|
raise ProvisioningError("wrap ttl must not exceed 15m")
|
|
return text
|
|
|
|
|
|
def _validate_output_path(path: Path) -> Path:
|
|
resolved = path.expanduser().resolve()
|
|
worktree = containing_git_worktree(resolved)
|
|
if worktree is not None:
|
|
raise ProvisioningError(
|
|
f"wrap file {resolved} is inside a Git worktree ({worktree}); "
|
|
"keep wrap tokens outside repos"
|
|
)
|
|
if resolved.exists() and resolved.stat().st_mode & 0o077:
|
|
raise ProvisioningError(
|
|
f"wrap file {resolved} is group/other-accessible "
|
|
f"(mode {oct(resolved.stat().st_mode & 0o777)}); must be 0600"
|
|
)
|
|
return resolved
|
|
|
|
|
|
def _write_mode_0600(path: Path, value: str) -> None:
|
|
path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
fd: int | None = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
try:
|
|
os.fchmod(fd, 0o600)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
|
fd = None
|
|
fh.write(value)
|
|
fh.write("\n")
|
|
finally:
|
|
if fd is not None:
|
|
os.close(fd)
|
|
|
|
|
|
def write_wrapped_handoff(
|
|
client: OpenBaoClient,
|
|
entry: CatalogEntry,
|
|
*,
|
|
out_file: Path,
|
|
ttl: str,
|
|
) -> WrapHandoffResult:
|
|
"""Mint a wrap token and write it without printing it."""
|
|
ttl = normalize_wrap_ttl(ttl)
|
|
out_path = _validate_output_path(out_file)
|
|
wrapped: WrappedResponse
|
|
if entry.kind == "auth-capability":
|
|
wrapped = client.wrap_approle_secret_id(entry.role_name, ttl=ttl)
|
|
elif entry.stores_kv_value():
|
|
wrapped = client.wrap_kv_get(entry.mount, entry.path, ttl=ttl)
|
|
else:
|
|
raise ProvisioningError(
|
|
f"lane '{entry.id}' kind {entry.kind} has no wrapped handoff"
|
|
)
|
|
token = wrapped.wrap_token
|
|
try:
|
|
_write_mode_0600(out_path, token)
|
|
finally:
|
|
token = ""
|
|
return WrapHandoffResult(
|
|
catalog_id=entry.id,
|
|
kind=entry.kind,
|
|
out_file=str(out_path),
|
|
ttl=ttl,
|
|
wrap_handle=wrapped.accessor_fingerprint,
|
|
creation_path=wrapped.creation_path,
|
|
)
|