Name the secrets-engine-openbao consumer custody coordinates.
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

Adds the consumer custody packet for KEY-WP-0037-T02 / RPF-WP-0035-T02,
strict value-free client-secret file checks with a check-secret-file CLI,
synthetic tests, and SECRETS-WP-0012 tracking founder decisions and the
attended delivery as needs_human wait tasks.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 132360@bnt-lap001
Assistant-Session: 769b1afd-4330-4c86-96b2-e510e26db42b
This commit is contained in:
tegwick 2026-10-06 13:39:29 +02:00
parent be8fab0332
commit a3897c2317
6 changed files with 475 additions and 6 deletions

View file

@ -957,6 +957,43 @@ def cmd_evidence_classify(cfg: Config, args) -> int:
# -- parser ----------------------------------------------------------------
def cmd_service_auth_check_secret_file(cfg: Config, args) -> int:
"""Value-free acceptance check of the KeyCape consumer secret file."""
import json as _json
from secrets_engine.service_auth import check_client_secret_file
raw = args.path or (
str(cfg.keycape_client_secret_file) if cfg.keycape_client_secret_file else ""
)
if not raw:
print(
"error: no path; pass --path or set SECRETS_ENGINE_KEYCAPE_CLIENT_SECRET_FILE",
file=sys.stderr,
)
return 2
checks = check_client_secret_file(Path(raw))
ok = bool(checks) and all(c.ok for c in checks)
if args.json:
print(
_json.dumps(
{
"path": raw,
"ok": ok,
"checks": [
{"name": c.name, "ok": c.ok, "detail": c.detail} for c in checks
],
},
sort_keys=True,
)
)
else:
for c in checks:
print(f"{'ok ' if c.ok else 'FAIL'} {c.name}{': ' + c.detail if c.detail else ''}")
print("accepted" if ok else "refused")
return 0 if ok else 1
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(prog="secrets-engine", description=__doc__.splitlines()[0])
p.add_argument("--version", action="version", version=f"secrets-engine {__version__}")
@ -1171,6 +1208,16 @@ def build_parser() -> argparse.ArgumentParser:
cl.add_argument("--json", action="store_true")
cl.set_defaults(func=cmd_evidence_classify)
sa = sub.add_parser("service-auth", help="KeyCape service-auth consumer checks")
sasub = sa.add_subparsers(dest="subcmd", required=True)
sacheck = sasub.add_parser(
"check-secret-file",
help="value-free check of the client-secret file (never reads it out)",
)
sacheck.add_argument("--path", default="")
sacheck.add_argument("--json", action="store_true")
sacheck.set_defaults(func=cmd_service_auth_check_secret_file)
return p

View file

@ -10,6 +10,8 @@ from __future__ import annotations
import base64
import binascii
import json
import os
import stat
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
@ -18,8 +20,8 @@ from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from secrets_engine.errors import BackendError
from secrets_engine.openbao import read_strict_token_file
from secrets_engine.errors import BackendError, ProvisioningError
from secrets_engine.safe_paths import containing_git_worktree
CLIENT_ID = "secrets-engine-openbao"
SUBJECT = "service:secrets-engine"
@ -154,6 +156,84 @@ def preflight_service_jwt(
Transport = Callable[..., Any]
@dataclass(frozen=True)
class SecretFileCheck:
"""One value-free consumer-file check. Never carries file contents."""
name: str
ok: bool
detail: str = ""
def check_client_secret_file(path: Path) -> list[SecretFileCheck]:
"""Inspect a confidential-client secret file without exposing its value.
The consumer copy must be an absolute, canonical (no symlink component),
regular, non-empty file owned by the effective user with no group/other
permission bits, in a parent directory owned by the same user with no
group/other permission bits, outside every Git worktree. Stops at the first
structural failure; later checks are then not reported as passed.
"""
checks: list[SecretFileCheck] = []
def add(name: str, ok: bool, detail: str = "") -> bool:
checks.append(SecretFileCheck(name, ok, detail))
return ok
raw = Path(path)
if not add("absolute", raw.is_absolute(), "" if raw.is_absolute() else "path must be absolute"):
return checks
try:
canonical = raw.resolve(strict=True)
except (OSError, RuntimeError):
add("exists", False, "file not found or unresolvable")
return checks
add("exists", True)
if not add(
"canonical",
canonical == raw,
"" if canonical == raw else "path has a symlink or non-canonical component",
):
return checks
st = os.lstat(raw)
if not add(
"regular_file",
stat.S_ISREG(st.st_mode),
"" if stat.S_ISREG(st.st_mode) else "not a regular file",
):
return checks
euid = os.geteuid()
add("file_owner", st.st_uid == euid, "" if st.st_uid == euid else "not owned by the running user")
mode = stat.S_IMODE(st.st_mode)
add("file_mode", not mode & 0o077, "" if not mode & 0o077 else f"group/other-accessible mode {oct(mode)}; must be 0600")
pst = os.lstat(raw.parent)
add("parent_owner", pst.st_uid == euid, "" if pst.st_uid == euid else "parent not owned by the running user")
pmode = stat.S_IMODE(pst.st_mode)
add(
"parent_mode",
not pmode & 0o077,
"" if not pmode & 0o077 else f"group/other-accessible parent mode {oct(pmode)}; must be 0700",
)
worktree = containing_git_worktree(raw)
add("outside_git_worktree", worktree is None, "" if worktree is None else "inside a Git worktree")
add("non_empty", st.st_size > 0, "" if st.st_size > 0 else "file is empty")
return checks
def read_client_secret_file(path: Path, *, purpose: str = "KeyCape client secret") -> str:
"""Read the consumer secret only after every value-free check passes."""
failed = [c for c in check_client_secret_file(path) if not c.ok]
if failed:
reasons = "; ".join(f"{c.name}: {c.detail}" for c in failed)
raise ProvisioningError(f"{purpose} file {path} refused ({reasons})")
value = Path(path).read_text(encoding="utf-8").strip()
if not value or any(ch.isspace() for ch in value):
raise ProvisioningError(
f"{purpose} file {path} refused (must hold exactly one non-empty token)"
)
return value
@dataclass(frozen=True)
class KeyCapeServiceAuthProvider:
config: KeyCapeServiceAuthConfig
@ -161,10 +241,7 @@ class KeyCapeServiceAuthProvider:
def exchange(self, *, now: datetime | None = None) -> ServiceJWT:
"""Perform one client-credentials exchange; never retry or fall back."""
secret = read_strict_token_file(
self.config.client_secret_file,
purpose="KeyCape client secret",
)
secret = read_client_secret_file(self.config.client_secret_file)
basic = base64.b64encode(
f"{self.config.client_id}:{secret}".encode("utf-8")
).decode("ascii")