diff --git a/SCOPE.md b/SCOPE.md index 3d379ef..2314042 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -236,8 +236,10 @@ credentials. - `instance` prints the one bound stage. The injected credential is not printed, returned, or written to Git, State Hub, logs, or the workplan. - A second stage credential in the same process refuses startup. -- `serve` keeps the lifecycle library the CLI uses running for that one stage. - It is the in-process manifestation of this library. +- `serve` keeps that same lifecycle path running for the one injected stage. + It applies the first job, then waits on stdin for further jobs until the + stream ends. Each job uses `build_lifecycle_plan` and the live lifecycle + apply path. This is the in-process manifestation of the library. - `--allow` and `SECRETS_ENGINE_LOCAL_ALLOW` do not authorize apply or the service. Permission to use an instance is the access-engine claim in `docs/engine-use-claim.md`. Protected side effects still require a decision diff --git a/src/secrets_engine/cli.py b/src/secrets_engine/cli.py index 6483f2b..6e62462 100644 --- a/src/secrets_engine/cli.py +++ b/src/secrets_engine/cli.py @@ -292,30 +292,62 @@ def cmd_instance(cfg: Config, args) -> int: return 0 +def execute_lifecycle_plan(cfg: Config, args, entry, plan, evidence): + """Live lifecycle path shared by `lifecycle` and `serve`. + + Builds nothing itself: the caller already called ``build_lifecycle_plan``. + Approval, the OpenBao client, and ``apply_lifecycle_plan`` stay here. + """ + from secrets_engine.errors import PolicyGuardError + from secrets_engine.lifecycle import apply_lifecycle_plan, require_destroy_confirmation + + if args.operation == "destroy": + require_destroy_confirmation(entry, getattr(args, "confirm_destroy", "") or "") + raise PolicyGuardError( + "live destroy is disabled until an exact-action destruction " + "approval contract is available; use --dry-run to inspect targets" + ) + decision = _require_lane_approval( + cfg, entry, args.operation, evidence, auth_args=args + ) + evidence.mark_approved(decision) + with _open_backend(cfg, args, evidence) as client: + result = apply_lifecycle_plan(client, plan) + mapped = operation_state(args.operation) + record = None + if mapped: + record = save_lane_state( + cfg.evidence_dir, entry.id, mapped, operation=args.operation + ) + detail = { + "applied": list(result.applied), + "preserved": list(result.preserved), + } + if record is not None: + detail["lane_state"] = record.state + evidence.finish("applied", detail=detail) + return result + + def cmd_serve(cfg: Config, args) -> int: - """Keep the lifecycle library running for the one injected stage.""" - from secrets_engine.catalog import get_entry - from secrets_engine.service import local_allow_requested, run_service + """Keep the lifecycle library running for the one injected stage. + + The argv job runs first. The process then waits on stdin for more jobs + and leaves only when that stream ends. Each job uses the live lifecycle + path. ``--allow`` is not authority. + """ + from secrets_engine.service import local_allow_requested, run_service, service_jobs local_allow_requested(bool(getattr(args, "allow", False))) - binding = getattr(args, "stage_binding", None) - entry = get_entry(cfg.catalog_dir, args.catalog_id) - foreign = getattr(args, "foreign_stage_credential_file", None) or "" - foreign_value = "" - if foreign: - foreign_value = "present" - plans = run_service( - binding, - [ - { - "entry": entry, - "operation": args.operation, - "foreign_stage_credential": foreign_value, - } - ], + outcomes = run_service( + cfg, + service_jobs(args.catalog_id, args.operation, sys.stdin), + allow=bool(getattr(args, "allow", False)), + auth_args=args, ) - for plan in plans: + for plan, result in outcomes: print(plan.render()) + print(result.render()) return 0 @@ -853,11 +885,7 @@ def cmd_state_show(cfg: Config, args) -> int: def cmd_lifecycle(cfg: Config, args) -> int: - from secrets_engine.lifecycle import ( - apply_lifecycle_plan, - build_lifecycle_plan, - require_destroy_confirmation, - ) + from secrets_engine.lifecycle import build_lifecycle_plan entry = get_entry(cfg.catalog_dir, args.catalog_id) if args.operation in {"compromise", "reactivate"}: @@ -873,36 +901,9 @@ def cmd_lifecycle(cfg: Config, args) -> int: f"lifecycle-{args.operation}", detail={"operation": plan.operation}, ) as evidence: - if args.operation == "destroy": - require_destroy_confirmation(entry, args.confirm_destroy) - # A lane-level approval is not an authorization to erase custody. - # Keep the destructive live path closed until T04 supplies a - # canonical exact-action and dual-control-capable decision contract. - from secrets_engine.errors import PolicyGuardError - - raise PolicyGuardError( - "live destroy is disabled until an exact-action destruction " - "approval contract is available; use --dry-run to inspect targets" - ) - decision = _require_lane_approval(cfg, entry, args.operation, evidence, auth_args=args) - evidence.mark_approved(decision) - with _open_backend(cfg, args, evidence) as client: - result = apply_lifecycle_plan(client, plan) - mapped = operation_state(args.operation) - record = None - if mapped: - record = save_lane_state( - cfg.evidence_dir, entry.id, mapped, operation=args.operation - ) - print(plan.render()) - print(result.render()) - detail = { - "applied": list(result.applied), - "preserved": list(result.preserved), - } - if record is not None: - detail["lane_state"] = record.state - evidence.finish("applied", detail=detail) + result = execute_lifecycle_plan(cfg, args, entry, plan, evidence) + print(plan.render()) + print(result.render()) return 0 @@ -1126,11 +1127,7 @@ def build_parser() -> argparse.ArgumentParser: action="store_true", help="ignored; a local flag does not authorize this process", ) - srv.add_argument( - "--foreign-stage-credential-file", - default="", - help=argparse.SUPPRESS, - ) + add_token_arg(srv) srv.set_defaults(func=cmd_serve) pl = sub.add_parser("plan", help="render a guarded plan (no mutation)") diff --git a/src/secrets_engine/service.py b/src/secrets_engine/service.py index 1b6f6ae..c045f96 100644 --- a/src/secrets_engine/service.py +++ b/src/secrets_engine/service.py @@ -1,20 +1,21 @@ """Kept-running manifestation of the lifecycle library. -This is the same library the CLI uses, in a process that stays up for one -injected stage. It is not a second implementation and not a Deployment. +This is the same library the CLI uses. The process stays in the job loop +until the job stream ends. It is not a second implementation and not a +cluster workload manifest. """ from __future__ import annotations +import json import os -from collections.abc import Iterable, Mapping +from collections.abc import Iterable, Iterator, Mapping from typing import Any from secrets_engine import lifecycle as lifecycle_lib -from secrets_engine.catalog import CatalogEntry -from secrets_engine.lifecycle import LifecyclePlan -from secrets_engine.stage_binding import StageBinding, StartupError +from secrets_engine.stage_binding import StartupError, load_stage_binding LOCAL_ALLOW_ENV = "SECRETS_ENGINE_LOCAL_ALLOW" +CATALOG_STAGE = {"build": "build", "test": "test", "production": "prod"} def local_allow_requested(flag: bool = False) -> bool: @@ -23,49 +24,77 @@ def local_allow_requested(flag: bool = False) -> bool: return bool(flag) or env +def service_jobs( + catalog_id: str, operation: str, lines: Iterable[str] +) -> Iterator[dict[str, str]]: + """Yield the argv job, then block on further lines until the stream ends.""" + yield {"catalog_id": catalog_id, "operation": operation, "confirm_destroy": ""} + for line in lines: + text = line.strip() + if not text: + continue + try: + payload = json.loads(text) + except json.JSONDecodeError as exc: + raise StartupError("service job is not a JSON object") from exc + if not isinstance(payload, dict): + raise StartupError("service job must be a mapping") + next_id = str(payload.get("catalog_id") or "") + next_operation = str(payload.get("operation") or "") + if not next_id or not next_operation: + raise StartupError("service job is missing catalog_id or operation") + yield { + "catalog_id": next_id, + "operation": next_operation, + "confirm_destroy": str(payload.get("confirm_destroy") or ""), + } + + def run_service( - binding: StageBinding | None, - jobs: Iterable[Mapping[str, Any]], -) -> list[LifecyclePlan]: - """Run lifecycle jobs for one stage until the job iterable ends. - - A job may carry ``foreign_stage_credential`` only as a refusal test. A - present value refuses the job. The binding credential is never rendered. - """ - if binding is None or not binding.credential or not binding.stage: - raise StartupError("startup refused: no stage credential is present") - plans: list[LifecyclePlan] = [] - for job in jobs: - foreign = str(job.get("foreign_stage_credential") or "") - if foreign: - raise StartupError( - "startup refused: another stage credential is present" - ) - entry = job.get("entry") - if not isinstance(entry, CatalogEntry): - raise StartupError("service job is missing a catalog entry") - operation = str(job.get("operation") or "") - plans.append(lifecycle_lib.build_lifecycle_plan(entry, operation)) - return plans - - -def invoke_protected( cfg: Any, - entry: CatalogEntry, - action: str, + jobs: Iterable[Mapping[str, str]], *, allow: bool = False, auth_args: Any = None, -) -> Any: - """Gate a protected call the way the CLI does. +) -> list[tuple[Any, Any]]: + """Run lifecycle jobs until the iterator ends. - ``allow`` and ``SECRETS_ENGINE_LOCAL_ALLOW`` are recorded as not authority - and then discarded. The call still requires a decision record or the - published unreachable-engine stance. + Startup reads the stage credential files. Two files refuse startup before + any job is pulled. Each job calls ``build_lifecycle_plan`` and then the + live lifecycle function the CLI uses, which applies the plan. ``allow`` + is discarded and does not authorize that call. """ local_allow_requested(allow) - from secrets_engine.cli import _require_lane_approval + binding = load_stage_binding() + if binding is None or not binding.credential or not binding.stage: + raise StartupError("startup refused: no stage credential is present") + catalog_stage = CATALOG_STAGE.get(binding.stage, "") + from types import SimpleNamespace - return _require_lane_approval( - cfg, entry, action, None, auth_args=auth_args - ) + from secrets_engine.catalog import get_entry + from secrets_engine.cli import _privileged_evidence, execute_lifecycle_plan + + outcomes: list[tuple[Any, Any]] = [] + for job in jobs: + entry = get_entry(cfg.catalog_dir, str(job.get("catalog_id") or "")) + if entry.stage != catalog_stage: + raise StartupError( + "startup refused: lane stage does not match the injected stage" + ) + operation = str(job.get("operation") or "") + plan = lifecycle_lib.build_lifecycle_plan(entry, operation) + call = SimpleNamespace( + operation=operation, + auth=getattr(auth_args, "auth", "env"), + bootstrap_token_file=getattr(auth_args, "bootstrap_token_file", None), + confirm_destroy=str(job.get("confirm_destroy") or ""), + ) + with _privileged_evidence( + cfg, + entry, + f"lifecycle-{operation}", + detail={"operation": operation}, + ) as evidence: + result = execute_lifecycle_plan(cfg, call, entry, plan, evidence) + outcomes.append((plan, result)) + return outcomes diff --git a/tests/test_wp0016_manifestations.py b/tests/test_wp0016_manifestations.py index 5fec55a..8fe2e3f 100644 --- a/tests/test_wp0016_manifestations.py +++ b/tests/test_wp0016_manifestations.py @@ -6,6 +6,9 @@ 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 @@ -14,11 +17,10 @@ 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 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" @@ -62,10 +64,10 @@ def _clear_stage_env(monkeypatch): 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")) +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", @@ -80,6 +82,32 @@ def _isolate_authority(monkeypatch, tmp_path: Path): 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] = [] @@ -150,43 +178,121 @@ def test_cli_refuses_two_stage_credentials(tmp_path, monkeypatch, capsys): _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 _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 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 +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", ] - rendered = plans[0].render() + 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_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_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(): @@ -224,17 +330,31 @@ def test_local_allow_does_not_authorize_protected_apply(tmp_path, monkeypatch, c _assert_sentinel_absent(captured) -def test_local_allow_does_not_authorize_protected_service_call(tmp_path, monkeypatch): +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") - 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 + 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():