tegwick
06dfe12c18
Planning stays off the store client. The process binds one stage credential at start, serve runs the same lifecycle library, and a local allow flag does not authorize a protected call. Assistant: grok Assistant-Session: 01a1609f-72b8-7222-9dff-fae300115e42
247 lines
8.4 KiB
Python
247 lines
8.4 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
|
|
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.errors import DecisionError
|
|
from secrets_engine.openbao import OpenBaoClient
|
|
from secrets_engine.plan import build_plan
|
|
from secrets_engine.service import invoke_protected, run_service
|
|
from secrets_engine.stage_binding import STAGE_TOKEN_FILES, StageBinding, 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 _isolate_authority(monkeypatch, tmp_path: Path):
|
|
"""No serving PDP, no published stance file, no stage credential."""
|
|
_clear_stage_env(monkeypatch)
|
|
monkeypatch.setenv("SECRETS_ENGINE_PEP_STANCE", str(tmp_path / "missing-stance.yaml"))
|
|
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 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 test_service_lifecycle_uses_the_cli_library_function(monkeypatch):
|
|
entry = validate_entry(copy.deepcopy(VALID))
|
|
binding = StageBinding(stage="test", credential=SENTINEL)
|
|
seen: list[tuple[str, str]] = []
|
|
real = lifecycle.build_lifecycle_plan
|
|
|
|
def spy(entry_arg, operation):
|
|
seen.append((entry_arg.id, operation))
|
|
return real(entry_arg, operation)
|
|
|
|
monkeypatch.setattr(lifecycle, "build_lifecycle_plan", spy)
|
|
plans = run_service(binding, [{"entry": entry, "operation": "suspend"}])
|
|
direct = real(entry, "suspend")
|
|
assert seen == [(entry.id, "suspend")]
|
|
assert plans[0].operation == direct.operation
|
|
assert plans[0].catalog_id == direct.catalog_id
|
|
assert [action.kind for action in plans[0].actions] == [
|
|
action.kind for action in direct.actions
|
|
]
|
|
rendered = plans[0].render()
|
|
assert SENTINEL not in rendered
|
|
|
|
|
|
def test_service_refuses_a_second_stage_credential():
|
|
entry = validate_entry(copy.deepcopy(VALID))
|
|
binding = StageBinding(stage="build", credential=SENTINEL)
|
|
with pytest.raises(StartupError, match="another stage credential"):
|
|
run_service(
|
|
binding,
|
|
[
|
|
{
|
|
"entry": entry,
|
|
"operation": "suspend",
|
|
"foreign_stage_credential": "present",
|
|
}
|
|
],
|
|
)
|
|
|
|
|
|
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_local_allow_does_not_authorize_protected_service_call(tmp_path, monkeypatch):
|
|
_isolate_authority(monkeypatch, tmp_path)
|
|
monkeypatch.setenv("SECRETS_ENGINE_LOCAL_ALLOW", "1")
|
|
entry = cli.get_entry(repo_root() / "catalog", "whynot-design-npm-publish")
|
|
with pytest.raises(DecisionError):
|
|
invoke_protected(Config.load(), entry, "apply", allow=True, auth_args=_Auth())
|
|
|
|
|
|
class _Auth:
|
|
auth = "env"
|
|
bootstrap_token_file = None
|
|
|
|
|
|
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
|