"""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)