Read the secrets-engine-openbao client secret in memory from OpenBao

Founder direction 2026-10-06 and ops-mason plan
keycape-secrets-engine-openbao-client-custody (658cdbc): no workstation copy.
service-jwt now gets CLIENT_SECRET from platform/workloads/keycape/
secrets-engine-openbao through one attended OIDC login on
keycape-secrets-engine-openbao-operator-workload-kv-read (-no-store, token from
a pipe, header-only GET, unconditional revoke-self). The consumer file is
refused for this profile and kept for the approval client.
service-auth probe-openbao prints value-free first-use evidence.

SECRETS-WP-0012: T02 cancelled (D1-D4 superseded), T04 done, T03 is the
founder-run first live use. Synthetic tests only; 524 passed, 1 skipped.

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

tests/test_consumer_secret_file.py

@@ -108,6 +108,6 @@ def test_cli_check_reports_without_value(tmp_path, capsys):
 
 def test_cli_check_uses_configured_env_path(tmp_path, capsys, monkeypatch):
     f = _private_file(tmp_path)
-    monkeypatch.setenv("SECRETS_ENGINE_KEYCAPE_CLIENT_SECRET_FILE", str(f))
+    monkeypatch.setenv("SECRETS_ENGINE_APPROVAL_CLIENT_SECRET_FILE", str(f))
     assert cli.main(["service-auth", "check-secret-file"]) == 0
     assert "accepted" in capsys.readouterr().out

tests/test_engine_auth.py

@@ -80,15 +80,11 @@ def test_explicit_service_jwt_fails_closed_without_contract(tmp_path, monkeypatc
 
 def test_jwt_login_failure_does_not_fall_back_to_env(tmp_path, monkeypatch):
     monkeypatch.setenv("BAO_TOKEN", "must-not-be-used")
-    secret = tmp_path / "client.secret"
-    secret.write_text("client-secret-value", encoding="utf-8")
-    secret.chmod(0o600)
     cfg = _cfg(
         tmp_path,
         openbao_jwt_login_file=_jwt_contract(tmp_path),
         keycape_token_url="https://keycape.example.test/token",
         keycape_issuer="https://keycape.example.test",
-        keycape_client_secret_file=secret,
     )
     monkeypatch.setattr(
         "secrets_engine.engine_auth.KeyCapeServiceAuthProvider.exchange",

tests/test_openbao_secret_source.py

@@ -0,0 +1,236 @@
+"""In-memory OpenBao source for the secrets-engine-openbao client secret.
+
+Synthetic values only. Proves the operator token and the client secret never
+reach argv, env, stdout/stderr or files, that revoke-self always runs, and that
+the file source and ambient tokens are refused for this profile.
+"""
+import functools
+import json
+import os
+from types import SimpleNamespace
+from urllib.error import HTTPError
+
+import pytest
+
+from secrets_engine import cli, engine_auth
+from secrets_engine.errors import BackendError
+from secrets_engine.openbao import OpenBaoClient, ScopedTokenSession
+from secrets_engine.service_auth import (
+    KeyCapeServiceAuthProvider,
+    OpenBaoOidcClientSecretSource,
+)
+from tests.test_engine_auth import _cfg, _jwt_contract
+from tests.test_service_auth import _jwt
+
+SECRET = "synthetic-client-secret-0123456789"
+OP_TOKEN = "synthetic-operator-token-abcdef"
+OP_ACCESSOR = "synthetic-operator-accessor"
+AMBIENT = "ambient-token-must-not-be-used"
+ADDR = "https://bao.example.test"
+
+
+class _Resp:
+    def __init__(self, body: bytes, status: int = 200):
+        self.status, self._body = status, body
+
+    def read(self, _n=-1):
+        return self._body
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_):
+        return False
+
+
+class FakeBao:
+    def __init__(self, kv_status=200, revoke_status=204):
+        self.kv_status, self.revoke_status = kv_status, revoke_status
+        self.argv, self.env, self.requests = [], [], []
+
+    def runner(self, argv, env):
+        self.argv.append(list(argv))
+        self.env.append(dict(env))
+        out = {"auth": {"client_token": OP_TOKEN, "accessor": OP_ACCESSOR}}
+        return 0, json.dumps(out).encode()
+
+    def transport(self, request, *, timeout):
+        self.requests.append(
+            (request.get_method(), request.full_url, dict(request.header_items()))
+        )
+        if request.full_url.endswith("/v1/platform/data/workloads/keycape/secrets-engine-openbao"):
+            if self.kv_status != 200:
+                raise HTTPError(request.full_url, self.kv_status, "denied", {}, None)
+            return _Resp(json.dumps({"data": {"data": {"CLIENT_SECRET": SECRET}}}).encode())
+        if request.full_url.endswith("/v1/auth/token/revoke-self"):
+            return _Resp(b"", self.revoke_status)
+        pytest.fail(f"unexpected request {request.full_url}")
+
+
+def _source(fake, **kw):
+    return OpenBaoOidcClientSecretSource(
+        bao_addr=ADDR, runner=fake.runner, transport=fake.transport, **kw
+    )
+
+
+def _assert_no_leak(fake, *texts):
+    for value in (SECRET, OP_TOKEN, OP_ACCESSOR):
+        for argv in fake.argv:
+            assert value not in " ".join(argv)
+        for env in fake.env:
+            assert value not in " ".join(f"{k}={v}" for k, v in env.items())
+        assert value not in " ".join(f"{k}={v}" for k, v in os.environ.items())
+        for text in texts:
+            assert value not in text
+
+
+def _assert_no_files(root):
+    for dirpath, _dirs, files in os.walk(root):
+        for name in files:
+            data = open(os.path.join(dirpath, name), "rb").read()
+            assert SECRET.encode() not in data and OP_TOKEN.encode() not in data
+
+
+def test_reads_in_memory_with_header_only_and_revokes(monkeypatch, tmp_path, capsys):
+    monkeypatch.setenv("BAO_TOKEN", AMBIENT)
+    monkeypatch.setenv("VAULT_TOKEN", AMBIENT)
+    monkeypatch.setenv("HOME", str(tmp_path))
+    fake = FakeBao()
+    src = _source(fake)
+    assert src() == SECRET
+    assert fake.argv == [[
+        "bao", "login", "-method=oidc", "-path=netkingdom", "-no-store",
+        "-format=json", "role=keycape-secrets-engine-openbao-operator-workload-kv-read",
+    ]]
+    env = fake.env[0]
+    assert "BAO_TOKEN" not in env and "VAULT_TOKEN" not in env
+    assert env["BAO_ADDR"] == ADDR
+    (get, kv_url, kv_headers), (post, rv_url, rv_headers) = fake.requests
+    assert (get, post) == ("GET", "POST")
+    assert kv_url == ADDR + "/v1/platform/data/workloads/keycape/secrets-engine-openbao"
+    assert rv_url == ADDR + "/v1/auth/token/revoke-self"
+    for headers in (kv_headers, rv_headers):
+        assert headers["X-vault-token"] == OP_TOKEN
+        assert AMBIENT not in json.dumps(headers)
+    ev = src.evidence()
+    assert ev == {
+        "secret_source": "openbao-oidc-in-memory",
+        "operator_session_handle": ev["operator_session_handle"],
+        "operator_revocation_attempted": True,
+        "operator_revocation_succeeded": True,
+    }
+    assert len(ev["operator_session_handle"]) == 12
+    out, err = capsys.readouterr()
+    _assert_no_leak(fake, out, err, repr(src), json.dumps(ev))
+    _assert_no_files(tmp_path)
+
+
+def test_revoke_self_runs_when_kv_read_is_denied(capsys):
+    fake = FakeBao(kv_status=403)
+    src = _source(fake)
+    with pytest.raises(BackendError, match="HTTP 403") as error:
+        src()
+    assert [r[0] for r in fake.requests] == ["GET", "POST"]
+    assert src.revocation_attempted and src.revocation_succeeded
+    _assert_no_leak(fake, str(error.value), *capsys.readouterr())
+
+
+def test_failed_login_makes_no_request():
+    fake = FakeBao()
+    src = _source(fake)
+    src.runner = lambda argv, env: (2, b"")
+    with pytest.raises(BackendError, match="exit 2"):
+        src()
+    assert fake.requests == []
+
+
+@pytest.mark.parametrize(
+    "override",
+    [
+        {"oidc_role": "platform-admin"},
+        {"oidc_mount": "oidc"},
+        {"kv_path": "workloads/keycape/other"},
+        {"kv_field": "OTHER"},
+    ],
+)
+def test_coordinates_are_pinned(override):
+    with pytest.raises(BackendError, match="reviewed plan"):
+        _source(FakeBao(), **override)
+
+
+@pytest.mark.parametrize("addr", ["http://bao.example.test", "ftp://x", "", "https://u:p@bao"])
+def test_address_must_be_https_or_loopback(addr):
+    with pytest.raises(BackendError):
+        OpenBaoOidcClientSecretSource(bao_addr=addr)
+    OpenBaoOidcClientSecretSource(bao_addr="http://127.0.0.1:8200")
+
+
+def test_engine_refuses_consumer_file_for_openbao_profile(tmp_path):
+    secret = tmp_path / "client.secret"
+    secret.write_text(SECRET)
+    secret.chmod(0o600)
+    cfg = _cfg(
+        tmp_path,
+        openbao_jwt_login_file=_jwt_contract(tmp_path),
+        keycape_token_url="https://keycape.example.test/token",
+        keycape_issuer="https://keycape.example.test",
+        keycape_client_secret_file=secret,
+    )
+    with pytest.raises(BackendError, match="refused for the secrets-engine-openbao") as error:
+        engine_auth.login_service_jwt(cfg)
+    assert SECRET not in str(error.value)
+
+
+def test_probe_end_to_end_is_value_free(tmp_path, monkeypatch, capsys):
+    monkeypatch.setenv("BAO_TOKEN", AMBIENT)
+    monkeypatch.setenv("HOME", str(tmp_path))
+    fake = FakeBao()
+    keycape_auth = []
+
+    def keycape_transport(request, *, timeout):
+        keycape_auth.append(request.get_header("Authorization"))
+        return _Resp(json.dumps(
+            {"access_token": _jwt(), "token_type": "Bearer", "expires_in": 900}
+        ).encode())
+
+    monkeypatch.setattr(
+        engine_auth, "OpenBaoOidcClientSecretSource",
+        functools.partial(
+            OpenBaoOidcClientSecretSource, runner=fake.runner, transport=fake.transport
+        ),
+    )
+    monkeypatch.setattr(
+        engine_auth, "KeyCapeServiceAuthProvider",
+        functools.partial(KeyCapeServiceAuthProvider, transport=keycape_transport),
+    )
+    monkeypatch.setattr("secrets_engine.service_auth.preflight_service_jwt",
+                        lambda token, config, now=None: SimpleNamespace(token=token))
+    jwt_logins = []
+
+    def fake_login_jwt(self, mount, role, jwt):
+        jwt_logins.append((mount, role))
+        client = OpenBaoClient(addr=self.addr, token="engine-token", bao_bin="bao")
+        monkeypatch.setattr(client, "_run_ok", lambda args, **_k: "")
+        return ScopedTokenSession(client=client, accessor_fingerprint="0123456789ab")
+
+    monkeypatch.setattr(OpenBaoClient, "login_jwt", fake_login_jwt)
+    cfg = _cfg(
+        tmp_path,
+        bao_addr=ADDR,
+        openbao_jwt_login_file=_jwt_contract(tmp_path),
+        keycape_token_url="https://keycape.example.test/token",
+        keycape_issuer="https://keycape.example.test",
+    )
+    monkeypatch.setattr(cli.Config, "load", classmethod(lambda _cls: cfg))
+    rc = cli.main(["service-auth", "probe-openbao", "--json"])
+    out, err = capsys.readouterr()
+    assert rc == 0, err
+    payload = json.loads(out)
+    assert payload["ok"] is True
+    assert payload["secret_source"] == "openbao-oidc-in-memory"
+    assert payload["operator_revocation_succeeded"] is True
+    assert payload["revocation_succeeded"] is True
+    assert len(jwt_logins) == 1
+    assert len(keycape_auth) == 1 and keycape_auth[0].startswith("Basic ")
+    _assert_no_leak(fake, out, err)
+    _assert_no_files(tmp_path)

tests/test_service_auth.py

@@ -45,11 +45,10 @@ def _jwt(**overrides):
     return f"{_segment({'alg': 'RS256', 'typ': 'JWT'})}.{_segment(claims)}.signature"
 
 
-def _config(secret_file=Path("/tmp/keycape-client-secret")):
+def _config(_unused=None):
     return KeyCapeServiceAuthConfig(
         token_url="https://keycape.example.test/token",
         issuer="https://keycape.example.test",
-        client_secret_file=secret_file,
     )
 
 
@@ -114,7 +113,9 @@ def test_exchange_uses_basic_auth_and_never_places_secret_in_body(tmp_path):
             }
         )
 
-    provider = KeyCapeServiceAuthProvider(_config(secret_file), transport=transport)
+    provider = KeyCapeServiceAuthProvider(
+        _config(), transport=transport, secret_source=lambda: "client-secret-value"
+    )
     result = provider.exchange(now=NOW)
     assert result.expires_at == int(NOW.timestamp()) + 900
     assert captured["authorization"].startswith("Basic ")
@@ -138,7 +139,9 @@ def test_exchange_rejects_refresh_or_id_tokens_without_exposing_secret(tmp_path)
             }
         )
 
-    provider = KeyCapeServiceAuthProvider(_config(secret_file), transport=transport)
+    provider = KeyCapeServiceAuthProvider(
+        _config(), transport=transport, secret_source=lambda: secret
+    )
     with pytest.raises(BackendError, match="forbidden extra token") as error:
         provider.exchange(now=NOW)
     assert secret not in str(error.value)
@@ -149,12 +152,24 @@ def test_provider_config_is_exact_and_has_no_bootstrap_fallback():
         KeyCapeServiceAuthConfig(
             token_url="http://keycape.example.test/token",
             issuer="https://keycape.example.test",
-            client_secret_file=Path("/tmp/unused"),
         )
     with pytest.raises(BackendError, match="accepted contract"):
         KeyCapeServiceAuthConfig(
             token_url="https://keycape.example.test/token",
             issuer="https://keycape.example.test",
-            client_secret_file=Path("/tmp/unused"),
             client_id="operator",
         )
+
+
+def test_openbao_profile_refuses_file_source(tmp_path):
+    with pytest.raises(BackendError, match="file source is refused"):
+        KeyCapeServiceAuthConfig(
+            token_url="https://keycape.example.test/token",
+            issuer="https://keycape.example.test",
+            client_secret_file=tmp_path / "client.secret",
+        )
+    provider = KeyCapeServiceAuthProvider(
+        _config(), transport=lambda *_a, **_k: pytest.fail("no exchange")
+    )
+    with pytest.raises(BackendError, match="no file fallback"):
+        provider.exchange(now=NOW)