Refuse login-only protected execution before consuming approval
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 2s

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
This commit is contained in:
tegwick 2026-10-11 14:19:30 +02:00
parent 4378ec1fc2
commit 97dfa5ea8f
5 changed files with 426 additions and 11 deletions

View file

@ -43,7 +43,11 @@ from secrets_engine.config import Config, repo_root
from secrets_engine.decisions import require_approved, resolve_decision
from secrets_engine.errors import BackendError, DecisionError, SecretsEngineError
from secrets_engine.evidence import EvidenceWriter, PrivilegedActionEvidence
from secrets_engine.engine_auth import login_service_jwt, select_engine_auth
from secrets_engine.engine_auth import (
login_service_jwt,
select_engine_auth,
require_backend_execution_ready,
)
from secrets_engine.lane_state import (
load_lane_state,
operation_state,
@ -111,6 +115,7 @@ def _require_lane_approval(
evidence: PrivilegedActionEvidence | None = None,
*,
fields: tuple[str, ...] = (),
auth_args: object | None = None,
):
"""Apply published PEP stance, resolve lane approval, then CAS-consume.
@ -125,6 +130,8 @@ def _require_lane_approval(
# Steps 1 and 2 first: a validated claim and decision are what make the
# unreachable-engine residue inapplicable. Absent configuration returns
# None and production stays closed exactly as before.
if auth_args is not None:
require_backend_execution_ready(cfg, auth_args, action or "unknown")
authorization = authorize_action(
cfg, entry, action or "unknown", None, fields=fields,
policy_targets=(entry.policy_name,) if entry.has_delivery_auth else (),
@ -303,7 +310,7 @@ def cmd_apply(cfg: Config, args) -> int:
return 0
with _privileged_evidence(cfg, entry, "apply") as evidence:
decision = _require_lane_approval(cfg, entry, "apply", evidence)
decision = _require_lane_approval(cfg, entry, "apply", evidence, auth_args=args)
evidence.mark_approved(decision)
plan = build_plan(
entry, args.stage, decision_id=decision.id if decision else ""
@ -331,7 +338,8 @@ def cmd_provision(cfg: Config, args) -> int:
f"lane '{entry.id}' is stage '{entry.stage}', not '{args.stage}'"
)
decision = _require_lane_approval(
cfg, entry, "provision", evidence, fields=(field,) if field else ()
cfg, entry, "provision", evidence, fields=(field,) if field else (),
auth_args=args,
)
evidence.mark_approved(decision)
require_provision_state(cfg.evidence_dir, entry.id)
@ -368,7 +376,8 @@ def cmd_rotate(cfg: Config, args) -> int:
f"lane '{entry.id}' is stage '{entry.stage}', not '{args.stage}'"
)
decision = _require_lane_approval(
cfg, entry, "rotate", evidence, fields=(field,) if field else ()
cfg, entry, "rotate", evidence, fields=(field,) if field else (),
auth_args=args,
)
evidence.mark_approved(decision)
with _open_backend(cfg, args, evidence) as client:
@ -410,7 +419,8 @@ def cmd_verify(cfg: Config, args) -> int:
},
) as evidence:
decision = _require_lane_approval(
cfg, entry, "verify", evidence, fields=tuple(fields)
cfg, entry, "verify", evidence, fields=tuple(fields),
auth_args=args,
)
evidence.mark_approved(decision)
with _open_backend(cfg, args, evidence) as client:
@ -482,7 +492,7 @@ def cmd_handoff(cfg: Config, args) -> int:
raise ProvisioningError(
f"lane '{entry.id}' is {entry.kind}; handoff needs auth-capability"
)
decision = _require_lane_approval(cfg, entry, "handoff", evidence)
decision = _require_lane_approval(cfg, entry, "handoff", evidence, auth_args=args)
evidence.mark_approved(decision)
require_delivery_state(cfg.evidence_dir, entry.id, "handoff")
with _open_backend(cfg, args, evidence) as client:
@ -522,7 +532,7 @@ def cmd_wrap(cfg: Config, args) -> int:
with _privileged_evidence(
cfg, entry, "wrap", detail={"ttl": args.ttl, "out_file": args.out}
) as evidence:
decision = _require_lane_approval(cfg, entry, "wrap", evidence)
decision = _require_lane_approval(cfg, entry, "wrap", evidence, auth_args=args)
evidence.mark_approved(decision)
require_delivery_state(cfg.evidence_dir, entry.id, "wrap")
with _open_backend(cfg, args, evidence) as client:
@ -576,7 +586,8 @@ def cmd_exec(cfg: Config, args) -> int:
))
# require approval + readiness before running.
decision = _require_lane_approval(
cfg, entry, "exec", evidence, fields=(field,) if field else ()
cfg, entry, "exec", evidence, fields=(field,) if field else (),
auth_args=args,
)
evidence.mark_approved(decision)
require_delivery_state(cfg.evidence_dir, entry.id, "exec")
@ -598,7 +609,8 @@ def cmd_exec(cfg: Config, args) -> int:
},
))
lane_decision = _require_lane_approval(
cfg, lane, "exec", lane_evidence, fields=(lane_field,)
cfg, lane, "exec", lane_evidence, fields=(lane_field,),
auth_args=args,
)
lane_evidence.mark_approved(lane_decision)
require_delivery_state(cfg.evidence_dir, lane.id, "exec")
@ -646,6 +658,7 @@ def cmd_session_revoke(cfg: Config, args) -> int:
detail={"handle_kind": kind},
)
with evidence:
require_backend_execution_ready(cfg, args, "session-revoke")
stance = apply_unreachable_engine_stance(cfg, stance_entry, "session-revoke")
evidence.mark_stance(stance)
handle = read_strict_token_file(
@ -726,7 +739,7 @@ def cmd_revoke(cfg: Config, args) -> int:
with _privileged_evidence(
cfg, entry, "revoke", detail={"operation": plan.operation}
) as evidence:
decision = _require_lane_approval(cfg, entry, "deactivate", evidence)
decision = _require_lane_approval(cfg, entry, "deactivate", evidence, auth_args=args)
evidence.mark_approved(decision)
with _open_backend(cfg, args, evidence) as client:
result = apply_lifecycle_plan(client, plan)
@ -831,7 +844,7 @@ def cmd_lifecycle(cfg: Config, args) -> int:
"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)
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)

View file

@ -156,3 +156,21 @@ def login_service_jwt(cfg: Any) -> ScopedTokenSession:
finally:
# Drop the KeyCape JWT from this frame; OpenBao verifies the signature.
del service_jwt
def require_backend_execution_ready(cfg: Any, args: Any, action: str) -> AuthSelection:
"""Reject a login-only provider before approval consume or backend work.
CCR-2026-0038 authenticates identity; it grants no protected lane action.
A configured login contract is not an execution entitlement. No mediated
execution backend is shipped, so no local file/flag may declare it ready.
Named bootstrap/env providers retain their existing separate gates.
"""
selection = select_engine_auth(cfg, args)
if selection.provider == "service-jwt":
raise BackendError(
"service-jwt protected execution is not ready: CCR-2026-0038 is "
"login-only; no reviewed task-scoped execution backend is shipped. "
"Approval was not consumed and no OpenBao call was made"
)
return selection