diff --git a/SCOPE.md b/SCOPE.md index 2cadd77..0f19b85 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -204,7 +204,8 @@ secrets-engine evidence heartbeat|drain|classify The implemented exec adapters are `exec-env` and `npm-config`. `read-check` is verification, `approle-login` is auth-capability handoff metadata, and -`exec-file`/`wrapped` are reserved schema names without executable adapters. +`exec-file` remains a reserved schema name without an exec adapter. +`secrets-engine wrap` implements response-wrapped operator handoff. ## Proven Operationally @@ -223,7 +224,7 @@ verification, `approle-login` is auth-capability handoff metadata, and - A service API, daemon, UI, queue, scheduler, or remote multi-user service. - OpenBao JWT login and platform materialization for the implemented KeyCape service-auth provider. -- Native `exec-file` or response-wrapped delivery. +- Native `exec-file` delivery. - Provider-side rotation or coordinated multi-consumer rollout. - First-class rotate, compromise, reactivate, lease-status, or audit report commands; lifecycle operations currently execute plans without persistent diff --git a/docs/catalog-admission.md b/docs/catalog-admission.md index a61a0fe..cd22cf8 100644 --- a/docs/catalog-admission.md +++ b/docs/catalog-admission.md @@ -43,8 +43,8 @@ delivery_auth: The current native implementation supports AppRole. An entry that declares `exec-env`, `exec-file`, `npm-config`, `read-check`, or `wrapped` must therefore -declare delivery auth. `exec-file` and `wrapped` remain reserved schema modes; -they are not implemented by `secrets-engine exec` yet. +declare delivery auth. `exec-file` remains a reserved exec schema mode. +`wrapped` operator handoff is `secrets-engine wrap`; it is not an exec adapter. Engine-managed AppRoles may bound `token_ttl`, `token_max_ttl`, `secret_id_ttl`, `secret_id_num_uses`, and `token_num_uses`. The admitted diff --git a/docs/cli.md b/docs/cli.md index e0bfa86..c14456e 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -55,6 +55,7 @@ secrets-engine apply --stage [--dry-run] [--bootstrap-token-file F secrets-engine provision --stage --field NAME (--from-file F | --generate) secrets-engine verify [--field NAME] [--positive] [--negative] [--negative-token-file F] secrets-engine handoff --stage --role-id-file F --secret-id-file F +secrets-engine wrap --out F [--ttl 15m] secrets-engine exec --catalog [--field NAME] [--mode auto|npm-config|exec-env] -- CMD... secrets-engine policy publication secrets-engine route [--json] @@ -80,6 +81,10 @@ local decision is accepted for a prod-labeled lane only with `SECRETS_ENGINE_UNSAFE_DEMO=1`, an empty Hub URL, and loopback OpenBao; the demo scripts set those three conditions themselves. +`wrap` writes a single-use OpenBao response-wrap token to `--out` (mode 0600, +outside Git) and never prints it. KV lanes wrap a path read; auth-capability +lanes wrap a secret_id. TTL max 15m. Production live wrap stays fail-closed. + `handoff` is for `kind: auth-capability` lanes such as `warden-sign`. It mints a fresh AppRole `secret_id` and writes `role_id` plus `secret_id` to caller-chosen mode-0600 files outside Git worktrees. It never prints the `secret_id`; use the diff --git a/docs/hardening-backlog.md b/docs/hardening-backlog.md index 5553042..efecda0 100644 --- a/docs/hardening-backlog.md +++ b/docs/hardening-backlog.md @@ -46,8 +46,12 @@ contents in this repo. ## H2 — Response-wrapped handoff -- Add a `wrapped` delivery mode using OpenBao response wrapping for operator - handoff flows where exec-time injection does not fit. +- Implemented: `secrets-engine wrap --out F [--ttl 15m]` writes a + single-use wrap token to a mode-0600 out-of-repo file. KV lanes wrap a read; + auth-capability lanes wrap a secret_id. The wrap token is never printed. + Evidence is wrap-handle fingerprint, ttl, and path only. TTL max 15m. + Unwrapped secret payloads fail closed. Production remains fail-closed. +- `exec --mode wrapped` is still not an exec adapter; this is operator handoff. ## H3 — Production dual-control diff --git a/evidence-classification.yaml b/evidence-classification.yaml index 6935517..b499744 100644 --- a/evidence-classification.yaml +++ b/evidence-classification.yaml @@ -45,7 +45,7 @@ rules: - id: production-control-mutation kind: load-bearing - actions: [revoke, lifecycle-suspend, lifecycle-deactivate, provision, session-revoke] + actions: [revoke, lifecycle-suspend, lifecycle-deactivate, provision, session-revoke, wrap] stages: [prod] emission: local-outbox note: >- diff --git a/layer.yaml b/layer.yaml index fed1ba5..c18662f 100644 --- a/layer.yaml +++ b/layer.yaml @@ -34,6 +34,7 @@ owned_tooling: - "bao policy/auth/kv subprocess adapter" - "CAS-aware KV create/patch via JSON input files, never argv values" - "JWT login via JSON input files; issued engine tokens self-revoke" + - "response wrapping via -wrap-ttl; wrap token never in argv evidence" note: >- This is the owned Lifecycle contact, not a Staff §5 shape. A new direct OpenBao client outside the listed modules is a finding. @@ -58,6 +59,7 @@ protected_actions: - lifecycle-deactivate - lifecycle-destroy - session-revoke + - wrap # §13 proposed capabilities. Owner status is proposed, not assented, until # the surface exists in this repository's own contract. diff --git a/src/secrets_engine/cli.py b/src/secrets_engine/cli.py index 6523960..4ff39d0 100644 --- a/src/secrets_engine/cli.py +++ b/src/secrets_engine/cli.py @@ -9,6 +9,7 @@ Command surface (FR7): provision --stage (--from-file F | --generate) --field NAME verify [--positive] [--negative] [--field NAME] [--negative-token-file F] handoff --stage --role-id-file F --secret-id-file F + wrap --out F [--ttl 15m] exec --catalog [--field NAME] [--mode auto|npm-config|exec-env] -- CMD... route [--json] revoke @@ -435,6 +436,37 @@ def cmd_handoff(cfg: Config, args) -> int: return 0 +def cmd_wrap(cfg: Config, args) -> int: + from secrets_engine.wrap import write_wrapped_handoff + + entry = get_entry(cfg.catalog_dir, args.catalog_id) + with _privileged_evidence( + cfg, entry, "wrap", detail={"ttl": args.ttl, "out_file": args.out} + ) as evidence: + decision = _require_lane_approval(cfg, entry, "wrap", evidence) + evidence.mark_approved(decision) + with _open_backend(cfg, args, evidence) as client: + result = write_wrapped_handoff( + client, entry, out_file=Path(args.out), ttl=args.ttl + ) + print( + f"wrote wrap token for lane '{result.catalog_id}' — token not displayed" + ) + print(f" out: {result.out_file}") + print(f" ttl: {result.ttl}") + print(f" handle: {result.wrap_handle or '-'}") + evidence.finish( + "wrap-token-written", + detail={ + "out_file": result.out_file, + "ttl": result.ttl, + "wrap_handle": result.wrap_handle, + "creation_path": result.creation_path, + }, + ) + return 0 + + def cmd_exec(cfg: Config, args) -> int: from secrets_engine.exec_delivery import exec_with_secret entry = get_entry(cfg.catalog_dir, args.catalog) @@ -795,6 +827,16 @@ def build_parser() -> argparse.ArgumentParser: add_token_arg(ha) ha.set_defaults(func=cmd_handoff) + wr = sub.add_parser( + "wrap", + help="write a single-use OpenBao wrap token to a file (never printed)", + ) + wr.add_argument("catalog_id") + wr.add_argument("--out", required=True, help="mode-0600 wrap-token file outside Git") + wr.add_argument("--ttl", default="15m", help="wrap TTL, max 15m (default 15m)") + add_token_arg(wr) + wr.set_defaults(func=cmd_wrap) + ex = sub.add_parser("exec", help="run a command with the secret injected for the child only") ex.add_argument("--catalog", required=True) ex.add_argument("--field", default=None) diff --git a/src/secrets_engine/evidence_class.py b/src/secrets_engine/evidence_class.py index 50c17b9..51acaa3 100644 --- a/src/secrets_engine/evidence_class.py +++ b/src/secrets_engine/evidence_class.py @@ -44,6 +44,7 @@ SHIPPED_RULES = ( "lifecycle-deactivate", "provision", "session-revoke", + "wrap", ), "stages": ("prod",), }, diff --git a/src/secrets_engine/openbao.py b/src/secrets_engine/openbao.py index a883168..1d9bf76 100644 --- a/src/secrets_engine/openbao.py +++ b/src/secrets_engine/openbao.py @@ -21,7 +21,7 @@ import stat import subprocess import tempfile from contextlib import contextmanager -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from secrets_engine.errors import BackendError, ProvisioningError @@ -61,6 +61,24 @@ def accessor_fingerprint(accessor: str) -> str: return hashlib.sha256(accessor.encode("utf-8")).hexdigest()[:12] +@dataclass(frozen=True) +class WrappedResponse: + """One OpenBao response-wrapped payload. The wrap token is secret.""" + + wrap_token: str = field(repr=False) + accessor_fingerprint: str + ttl: str + creation_path: str + + def evidence(self) -> dict[str, object]: + payload: dict[str, object] = { + "wrap_handle": self.accessor_fingerprint, + "wrap_ttl": self.ttl, + "creation_path": self.creation_path, + } + return payload + + @dataclass class ScopedTokenSession: """One AppRole login token that revokes itself on close.""" @@ -335,6 +353,44 @@ class OpenBaoClient: raise BackendError("token accessor is missing or invalid") self._run_ok(["token", "revoke", "-accessor", accessor]) + def _parse_wrap_response(self, stdout: str, *, ttl: str, creation_path: str) -> WrappedResponse: + try: + payload = json.loads(stdout) + except json.JSONDecodeError as exc: + raise BackendError("wrapped response is not JSON") from exc + if not isinstance(payload, dict): + raise BackendError("wrapped response is invalid") + wrap = payload.get("wrap_info") + if not isinstance(wrap, dict) or not wrap.get("token"): + raise BackendError("response wrapping was not applied") + token = str(wrap["token"]) + accessor = str(wrap.get("accessor") or "") + wrap_ttl = wrap.get("ttl") + return WrappedResponse( + wrap_token=token, + accessor_fingerprint=accessor_fingerprint(accessor) if accessor else "", + ttl=str(wrap_ttl if wrap_ttl is not None else ttl), + creation_path=creation_path, + ) + + def wrap_kv_get(self, mount: str, path: str, *, ttl: str) -> WrappedResponse: + """Wrap a KV read. Fail if OpenBao returns the secret unwrapped.""" + target = f"{mount}/{path}" + proc = self._run(["kv", "get", f"-wrap-ttl={ttl}", "-format=json", target]) + if proc.returncode != 0: + raise BackendError("wrapped KV read failed") + return self._parse_wrap_response(proc.stdout, ttl=ttl, creation_path=target) + + def wrap_approle_secret_id(self, role_name: str, *, ttl: str) -> WrappedResponse: + """Wrap a single-use AppRole secret_id. Fail if returned unwrapped.""" + target = f"auth/approle/role/{role_name}/secret-id" + proc = self._run( + ["write", f"-wrap-ttl={ttl}", "-format=json", "-force", target] + ) + if proc.returncode != 0: + raise BackendError("wrapped AppRole secret-id mint failed") + return self._parse_wrap_response(proc.stdout, ttl=ttl, creation_path=target) + # -- KV v2 ------------------------------------------------------------- def kv_mount_exists(self, mount: str) -> bool: diff --git a/src/secrets_engine/wrap.py b/src/secrets_engine/wrap.py new file mode 100644 index 0000000..9b72c02 --- /dev/null +++ b/src/secrets_engine/wrap.py @@ -0,0 +1,106 @@ +"""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, + ) diff --git a/tests/test_layer_conformance.py b/tests/test_layer_conformance.py index ba56a62..d2226c4 100644 --- a/tests/test_layer_conformance.py +++ b/tests/test_layer_conformance.py @@ -172,12 +172,14 @@ def test_classify_does_not_grant_permission(): apply_prod = classify("apply", "prod") heartbeat = classify("evidence-heartbeat", "prod") session_revoke = classify("session-revoke", "prod") + wrap = classify("wrap", "prod") assert prod_provision.kind == "load-bearing" assert test_provision.kind == "attributive" assert destroy.kind == "load-bearing" assert apply_prod.kind == "attributive" assert heartbeat.kind == "heartbeat" assert session_revoke.kind == "load-bearing" + assert wrap.kind == "load-bearing" assert prod_provision.completeness_claimed is False assert CLASSIFICATION.exists() diff --git a/tests/test_wrap.py b/tests/test_wrap.py new file mode 100644 index 0000000..670eae6 --- /dev/null +++ b/tests/test_wrap.py @@ -0,0 +1,160 @@ +import copy +import json +from types import SimpleNamespace + +import pytest + +from secrets_engine.catalog import validate_entry +from secrets_engine.config import Config +from secrets_engine.errors import BackendError, DecisionError, ProvisioningError +from secrets_engine.openbao import OpenBaoClient +from secrets_engine.wrap import normalize_wrap_ttl, write_wrapped_handoff +from tests.test_catalog import VALID + +WRAP_TOKEN = "wrap-token-SUPER-SECRET" +WRAP_ACCESSOR = "wrap-accessor-value" + + +class FakeWrapClient: + def __init__(self): + self.calls = [] + + def wrap_kv_get(self, mount, path, *, ttl): + from secrets_engine.openbao import WrappedResponse, accessor_fingerprint + + self.calls.append(("kv", mount, path, ttl)) + return WrappedResponse( + wrap_token=WRAP_TOKEN, + accessor_fingerprint=accessor_fingerprint(WRAP_ACCESSOR), + ttl=ttl, + creation_path=f"{mount}/{path}", + ) + + def wrap_approle_secret_id(self, role_name, *, ttl): + from secrets_engine.openbao import WrappedResponse, accessor_fingerprint + + self.calls.append(("approle", role_name, ttl)) + return WrappedResponse( + wrap_token=WRAP_TOKEN, + accessor_fingerprint=accessor_fingerprint(WRAP_ACCESSOR), + ttl=ttl, + creation_path=f"auth/approle/role/{role_name}/secret-id", + ) + + +def test_normalize_wrap_ttl_bounds(): + assert normalize_wrap_ttl("15m") == "15m" + assert normalize_wrap_ttl("60s") == "60s" + with pytest.raises(ProvisioningError, match="15m"): + normalize_wrap_ttl("16m") + with pytest.raises(ProvisioningError, match="like"): + normalize_wrap_ttl("15") + + +def test_wrap_kv_writes_0600_file_and_omits_token_from_result(tmp_path): + out = tmp_path / "wrap.token" + entry = validate_entry(copy.deepcopy(VALID)) + result = write_wrapped_handoff( + FakeWrapClient(), entry, out_file=out, ttl="15m" + ) + assert out.read_text().strip() == WRAP_TOKEN + assert (out.stat().st_mode & 0o077) == 0 + assert result.wrap_handle + assert WRAP_TOKEN not in result.wrap_handle + assert WRAP_TOKEN not in result.out_file + assert WRAP_ACCESSOR not in result.wrap_handle + + +def test_wrap_rejects_repo_paths(): + entry = validate_entry(copy.deepcopy(VALID)) + with pytest.raises(ProvisioningError, match="Git worktree"): + write_wrapped_handoff( + FakeWrapClient(), + entry, + out_file=__import__("pathlib").Path("wrap.token"), + ttl="15m", + ) + + +def test_parse_wrap_rejects_unwrapped_secret_payload(): + client = OpenBaoClient(addr="http://example.invalid", token="t", bao_bin="bao") + secret_json = json.dumps( + {"data": {"data": {"api_token": "npm_SHOULDNEVERPARSE"}}} + ) + with pytest.raises(BackendError, match="was not applied") as raised: + client._parse_wrap_response( + secret_json, ttl="15m", creation_path="secret/x" + ) + assert "npm_SHOULDNEVERPARSE" not in str(raised.value) + + +def test_parse_wrap_accepts_wrap_info_only(): + client = OpenBaoClient(addr="http://example.invalid", token="t", bao_bin="bao") + payload = json.dumps( + { + "wrap_info": { + "token": WRAP_TOKEN, + "accessor": WRAP_ACCESSOR, + "ttl": 900, + } + } + ) + wrapped = client._parse_wrap_response( + payload, ttl="15m", creation_path="secret/x" + ) + assert wrapped.wrap_token == WRAP_TOKEN + assert WRAP_ACCESSOR not in wrapped.accessor_fingerprint + assert WRAP_TOKEN not in repr(wrapped) + + +def test_wrap_kv_get_uses_wrap_ttl_flag(monkeypatch): + client = OpenBaoClient(addr="http://example.invalid", token="t", bao_bin="bao") + seen = {} + + def fake_run(args, stdin=None): + seen["args"] = list(args) + return SimpleNamespace( + returncode=0, + stdout=json.dumps( + {"wrap_info": {"token": WRAP_TOKEN, "accessor": WRAP_ACCESSOR}} + ), + stderr="", + ) + + monkeypatch.setattr(client, "_run", fake_run) + wrapped = client.wrap_kv_get("secret", "test/team/thing", ttl="15m") + assert "-wrap-ttl=15m" in seen["args"] + assert WRAP_TOKEN not in " ".join(seen["args"]) + assert wrapped.wrap_token == WRAP_TOKEN + + +def test_production_wrap_fails_closed(tmp_path, monkeypatch): + from secrets_engine import cli + + data = copy.deepcopy(VALID) + data.update(stage="prod", approval={"model": "decision", "decision_ref": "x"}) + entry = validate_entry(data) + monkeypatch.setattr(cli, "get_entry", lambda *_args: entry) + monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False) + monkeypatch.setattr( + cli.OpenBaoClient, + "resolve", + lambda *_args, **_kwargs: pytest.fail("backend must not be reached"), + ) + cfg = Config( + catalog_dir=tmp_path, + policy_dir=tmp_path, + evidence_dir=tmp_path / "evidence", + hub_url="http://127.0.0.1:8000", + bao_addr="http://127.0.0.1:8200", + topic_id="test-topic", + ) + args = SimpleNamespace( + catalog_id=entry.id, + out=str(tmp_path / "wrap.token"), + ttl="15m", + bootstrap_token_file=None, + auth="auto", + ) + with pytest.raises(DecisionError, match="production action 'wrap'"): + cli.cmd_wrap(cfg, args)