tegwick
f211525f00
The process reads one stage credential file at startup, applies each job through the CLI lifecycle functions, and waits for further jobs. A second stage credential file refuses startup. A local allow flag does not authorize the call. Assistant: grok Assistant-Session: 01a1609f-72b8-7222-9dff-fae300115e42
367 lines
12 KiB
Python
367 lines
12 KiB
Python
"""SECRETS-WP-0016: plan stays off the store client; apply still calls it.
|
|
|
|
Later cases cover one-stage process binding, the in-process service, and the
|
|
refusal of a local allow flag. Fixture credentials stay out of captured output.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import io
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from secrets_engine import cli, lifecycle
|
|
from secrets_engine.apply import apply_plan
|
|
from secrets_engine.catalog import validate_entry
|
|
from secrets_engine.config import Config, repo_root
|
|
from secrets_engine.openbao import OpenBaoClient
|
|
from secrets_engine.plan import build_plan
|
|
from secrets_engine.service import run_service
|
|
from secrets_engine.stage_binding import STAGE_TOKEN_FILES, StartupError
|
|
from tests.test_catalog import VALID
|
|
|
|
SENTINEL = "fixture-stage-credential-wp0016"
|
|
STAGE_ENV = tuple(STAGE_TOKEN_FILES.values())
|
|
|
|
|
|
class RecordingApplyClient:
|
|
def __init__(self):
|
|
self.calls: list[str] = []
|
|
|
|
def kv_mount_exists(self, mount):
|
|
self.calls.append("kv_mount_exists")
|
|
assert mount
|
|
return False
|
|
|
|
def ensure_kv_mount(self, mount):
|
|
self.calls.append("ensure_kv_mount")
|
|
assert mount
|
|
|
|
def read_policy(self, name):
|
|
self.calls.append("read_policy")
|
|
assert name
|
|
return None
|
|
|
|
def write_policy(self, name, hcl):
|
|
self.calls.append("write_policy")
|
|
assert name and hcl
|
|
|
|
def ensure_approle_enabled(self):
|
|
self.calls.append("ensure_approle_enabled")
|
|
|
|
def write_approle(self, role_name, policies, ttl="30m", **kwargs):
|
|
self.calls.append("write_approle")
|
|
assert role_name and policies and ttl
|
|
del kwargs
|
|
|
|
|
|
def _clear_stage_env(monkeypatch):
|
|
for key in STAGE_ENV:
|
|
monkeypatch.delenv(key, raising=False)
|
|
monkeypatch.delenv("SECRETS_ENGINE_LOCAL_ALLOW", raising=False)
|
|
|
|
|
|
def _quiet_runtime(monkeypatch, tmp_path: Path):
|
|
"""Keep the test off the hub, the PDP, and the repo evidence directory."""
|
|
monkeypatch.setenv("SECRETS_ENGINE_HUB_URL", "")
|
|
monkeypatch.setenv("SECRETS_ENGINE_EVIDENCE", str(tmp_path / "evidence"))
|
|
for key in (
|
|
"SECRETS_ENGINE_APPROVAL_URL",
|
|
"SECRETS_ENGINE_APPROVAL_TOKEN_FILE",
|
|
"SECRETS_ENGINE_PDP_URL",
|
|
"SECRETS_ENGINE_PDP_TOKEN_FILE",
|
|
"SECRETS_ENGINE_PDP_TRUST_FILE",
|
|
"SECRETS_ENGINE_PDP_VERIFIER",
|
|
"SECRETS_ENGINE_OPENBAO_JWT_LOGIN",
|
|
"BAO_TOKEN",
|
|
"VAULT_TOKEN",
|
|
):
|
|
monkeypatch.delenv(key, raising=False)
|
|
|
|
|
|
def _isolate_authority(monkeypatch, tmp_path: Path):
|
|
"""No serving PDP, no published stance file, no stage credential."""
|
|
_clear_stage_env(monkeypatch)
|
|
_quiet_runtime(monkeypatch, tmp_path)
|
|
monkeypatch.setenv("SECRETS_ENGINE_PEP_STANCE", str(tmp_path / "missing-stance.yaml"))
|
|
|
|
|
|
class _Auth:
|
|
auth = "env"
|
|
bootstrap_token_file = None
|
|
|
|
|
|
class _LifecycleClient:
|
|
def __init__(self):
|
|
self.calls: list[str] = []
|
|
|
|
def delete_approle(self, name):
|
|
self.calls.append(f"delete_approle:{name}")
|
|
|
|
def delete_policy(self, name):
|
|
raise AssertionError(f"unexpected policy delete {name}")
|
|
|
|
def kv_delete_metadata(self, mount, path):
|
|
raise AssertionError(f"unexpected kv delete {mount}/{path}")
|
|
|
|
|
|
def test_plan_path_constructs_no_openbao_client(monkeypatch):
|
|
constructed: list[str] = []
|
|
|
|
def _refuse_init(self, *args, **kwargs):
|
|
constructed.append("init")
|
|
raise AssertionError("plan path constructed an OpenBao client")
|
|
|
|
def _refuse_resolve(cls, *args, **kwargs):
|
|
constructed.append("resolve")
|
|
raise AssertionError("plan path resolved an OpenBao client")
|
|
|
|
monkeypatch.setattr(OpenBaoClient, "__init__", _refuse_init)
|
|
monkeypatch.setattr(OpenBaoClient, "resolve", _refuse_resolve)
|
|
entry = validate_entry(copy.deepcopy(VALID))
|
|
plan = build_plan(entry, "test", decision_id="d1")
|
|
assert plan.catalog_id == entry.id
|
|
assert plan.actions
|
|
assert constructed == []
|
|
|
|
|
|
def test_apply_path_issues_openbao_calls():
|
|
entry = validate_entry(copy.deepcopy(VALID))
|
|
plan = build_plan(entry, "test", decision_id="d1")
|
|
client = RecordingApplyClient()
|
|
result = apply_plan(client, entry, plan)
|
|
for name in ("ensure_kv_mount", "write_policy", "write_approle"):
|
|
assert name in client.calls
|
|
assert any(name.startswith("kv-mount") for name in result.applied)
|
|
assert any(name.startswith("policy") for name in result.applied)
|
|
assert any(name.startswith("approle") for name in result.applied)
|
|
|
|
|
|
def _write_stage_file(tmp_path: Path, name: str) -> Path:
|
|
path = tmp_path / name
|
|
path.write_text(SENTINEL + "\n", encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def _assert_sentinel_absent(captured):
|
|
assert SENTINEL not in captured.out
|
|
assert SENTINEL not in captured.err
|
|
|
|
|
|
@pytest.mark.parametrize("stage", ("build", "test", "production"))
|
|
def test_cli_binds_one_stage_and_hides_the_credential(stage, tmp_path, monkeypatch, capsys):
|
|
_clear_stage_env(monkeypatch)
|
|
path = _write_stage_file(tmp_path, stage)
|
|
monkeypatch.setenv(STAGE_TOKEN_FILES[stage], str(path))
|
|
for _ in range(2):
|
|
capsys.readouterr()
|
|
code = cli.main(["instance"])
|
|
captured = capsys.readouterr()
|
|
assert code == 0
|
|
assert captured.out.strip() == f"stage={stage}"
|
|
_assert_sentinel_absent(captured)
|
|
|
|
|
|
def test_cli_refuses_two_stage_credentials(tmp_path, monkeypatch, capsys):
|
|
_clear_stage_env(monkeypatch)
|
|
monkeypatch.setenv(STAGE_TOKEN_FILES["build"], str(_write_stage_file(tmp_path, "build")))
|
|
monkeypatch.setenv(STAGE_TOKEN_FILES["test"], str(_write_stage_file(tmp_path, "test")))
|
|
for _ in range(2):
|
|
capsys.readouterr()
|
|
code = cli.main(["instance"])
|
|
captured = capsys.readouterr()
|
|
assert code != 0
|
|
assert "more than one stage credential is present" in captured.err
|
|
_assert_sentinel_absent(captured)
|
|
|
|
|
|
def _bind_stage(monkeypatch, tmp_path: Path, stage: str) -> None:
|
|
_clear_stage_env(monkeypatch)
|
|
monkeypatch.setenv(STAGE_TOKEN_FILES[stage], str(_write_stage_file(tmp_path, stage)))
|
|
|
|
|
|
def test_service_lifecycle_uses_the_cli_library_and_waits(tmp_path, monkeypatch):
|
|
_quiet_runtime(monkeypatch, tmp_path)
|
|
_bind_stage(monkeypatch, tmp_path, "build")
|
|
client = _LifecycleClient()
|
|
monkeypatch.setattr(OpenBaoClient, "resolve", lambda *_a, **_k: client)
|
|
order: list[str] = []
|
|
real_build = lifecycle.build_lifecycle_plan
|
|
real_apply = lifecycle.apply_lifecycle_plan
|
|
|
|
def spy_build(entry, operation):
|
|
order.append(f"build:{operation}")
|
|
return real_build(entry, operation)
|
|
|
|
def spy_apply(bound_client, plan):
|
|
order.append(f"apply:{plan.operation}")
|
|
return real_apply(bound_client, plan)
|
|
|
|
monkeypatch.setattr(lifecycle, "build_lifecycle_plan", spy_build)
|
|
monkeypatch.setattr(lifecycle, "apply_lifecycle_plan", spy_apply)
|
|
|
|
def jobs():
|
|
order.append("yield-1")
|
|
yield {
|
|
"catalog_id": "example-build-test-token",
|
|
"operation": "suspend",
|
|
"confirm_destroy": "",
|
|
}
|
|
order.append("yield-2")
|
|
yield {
|
|
"catalog_id": "example-build-test-token",
|
|
"operation": "suspend",
|
|
"confirm_destroy": "",
|
|
}
|
|
|
|
outcomes = run_service(
|
|
Config.load(), jobs(), allow=True, auth_args=_Auth()
|
|
)
|
|
assert order == [
|
|
"yield-1",
|
|
"build:suspend",
|
|
"apply:suspend",
|
|
"yield-2",
|
|
"build:suspend",
|
|
"apply:suspend",
|
|
]
|
|
assert len(client.calls) == 2
|
|
assert all(call.startswith("delete_approle:") for call in client.calls)
|
|
rendered = "\n".join(plan.render() + result.render() for plan, result in outcomes)
|
|
assert SENTINEL not in rendered
|
|
assert "applied:" in rendered
|
|
|
|
|
|
def test_serve_reads_a_second_job_from_stdin(tmp_path, monkeypatch, capsys):
|
|
_quiet_runtime(monkeypatch, tmp_path)
|
|
_bind_stage(monkeypatch, tmp_path, "build")
|
|
client = _LifecycleClient()
|
|
monkeypatch.setattr(OpenBaoClient, "resolve", lambda *_a, **_k: client)
|
|
extra = json.dumps(
|
|
{"catalog_id": "example-build-test-token", "operation": "suspend"}
|
|
)
|
|
monkeypatch.setattr(sys, "stdin", io.StringIO(extra + "\n"))
|
|
code = cli.main(
|
|
[
|
|
"serve",
|
|
"example-build-test-token",
|
|
"--operation",
|
|
"suspend",
|
|
"--auth",
|
|
"env",
|
|
"--allow",
|
|
]
|
|
)
|
|
captured = capsys.readouterr()
|
|
assert code == 0
|
|
assert len(client.calls) == 2
|
|
assert captured.out.count("applied:") == 2
|
|
_assert_sentinel_absent(captured)
|
|
|
|
|
|
def test_service_refuses_a_second_stage_credential_file(tmp_path, monkeypatch, capsys):
|
|
_clear_stage_env(monkeypatch)
|
|
monkeypatch.setenv(STAGE_TOKEN_FILES["build"], str(_write_stage_file(tmp_path, "build")))
|
|
monkeypatch.setenv(
|
|
STAGE_TOKEN_FILES["production"],
|
|
str(_write_stage_file(tmp_path, "production")),
|
|
)
|
|
|
|
pulled: list[str] = []
|
|
|
|
def jobs():
|
|
pulled.append("pulled")
|
|
yield {
|
|
"catalog_id": "example-build-test-token",
|
|
"operation": "suspend",
|
|
"confirm_destroy": "",
|
|
}
|
|
|
|
with pytest.raises(StartupError, match="more than one stage credential is present") as raised:
|
|
run_service(Config.load(), jobs(), allow=True, auth_args=_Auth())
|
|
assert pulled == []
|
|
assert SENTINEL not in str(raised.value)
|
|
|
|
capsys.readouterr()
|
|
code = cli.main(
|
|
["serve", "example-build-test-token", "--operation", "suspend", "--auth", "env"]
|
|
)
|
|
captured = capsys.readouterr()
|
|
assert code != 0
|
|
assert "more than one stage credential is present" in captured.err
|
|
_assert_sentinel_absent(captured)
|
|
|
|
|
|
def test_no_deployment_manifest_was_added():
|
|
root = repo_root()
|
|
for path in root.rglob("*"):
|
|
if ".git" in path.parts or path.suffix not in {".yaml", ".yml"}:
|
|
continue
|
|
text = path.read_text(encoding="utf-8", errors="ignore")
|
|
assert "kind: Deployment" not in text
|
|
|
|
|
|
def test_local_allow_does_not_authorize_protected_apply(tmp_path, monkeypatch, capsys):
|
|
_isolate_authority(monkeypatch, tmp_path)
|
|
monkeypatch.setenv("SECRETS_ENGINE_LOCAL_ALLOW", "1")
|
|
monkeypatch.setattr(
|
|
OpenBaoClient,
|
|
"resolve",
|
|
lambda *_a, **_k: pytest.fail("protected apply opened a backend"),
|
|
)
|
|
code = cli.main(
|
|
[
|
|
"apply",
|
|
"whynot-design-npm-publish",
|
|
"--stage",
|
|
"prod",
|
|
"--auth",
|
|
"env",
|
|
"--allow",
|
|
]
|
|
)
|
|
captured = capsys.readouterr()
|
|
assert code != 0
|
|
assert "applied:" not in captured.out
|
|
assert "unable to load PEP stance" in captured.err or "DecisionError" in captured.err or "error:" in captured.err
|
|
_assert_sentinel_absent(captured)
|
|
|
|
|
|
def test_serve_allow_flag_does_not_authorize(tmp_path, monkeypatch, capsys):
|
|
_isolate_authority(monkeypatch, tmp_path)
|
|
_bind_stage(monkeypatch, tmp_path, "production")
|
|
monkeypatch.setenv("SECRETS_ENGINE_LOCAL_ALLOW", "1")
|
|
monkeypatch.setattr(
|
|
OpenBaoClient,
|
|
"resolve",
|
|
lambda *_a, **_k: pytest.fail("serve opened a backend"),
|
|
)
|
|
code = cli.main(
|
|
[
|
|
"serve",
|
|
"whynot-design-npm-publish",
|
|
"--operation",
|
|
"suspend",
|
|
"--auth",
|
|
"env",
|
|
"--allow",
|
|
]
|
|
)
|
|
captured = capsys.readouterr()
|
|
assert code != 0
|
|
assert "applied:" not in captured.out
|
|
assert "unable to load PEP stance" in captured.err
|
|
_assert_sentinel_absent(captured)
|
|
|
|
|
|
def test_use_claim_names_access_engine_and_stores_no_credential():
|
|
path = repo_root() / "docs" / "engine-use-claim.md"
|
|
text = path.read_text(encoding="utf-8")
|
|
assert "access-engine" in text
|
|
assert SENTINEL not in text
|
|
lowered = text.lower()
|
|
assert "token=" not in lowered
|
|
assert "bao_token" not in lowered
|