Enforce companion-only credential delivery and refresh activation handoff
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 7s

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e324-abce-7e51-bb2b-496f097afdb0
This commit is contained in:
tegwick 2026-09-27 15:58:53 +02:00
parent 303cbf652b
commit 5c6f2b319d
8 changed files with 150 additions and 24 deletions

View file

@ -221,14 +221,28 @@ def exec_with_secret(
f"(allowed {sorted(declared)})"
)
# The caller's resolved lanes must exactly match the pinned recipient.
# Refuse missing/substituted companions before reading even the primary.
from secrets_engine.exec_owner import companion_specs, resolve_companions
supplied = [
{"catalog": lane.id, "field": lane_field, "env": env_name}
for lane, lane_field, env_name in companions
]
if supplied != companion_specs(entry):
raise DeliveryError("companion delivery differs from the catalog-bound exec owner")
if companions:
if binding_digest is None or mode != "exec-env":
raise DeliveryError("companion delivery requires a configured exec owner and exec-env")
lanes = {lane.id: lane for lane, _, _ in companions}
resolve_companions(entry, lanes.__getitem__)
if session_evidence is None:
value = _fetch_value(client, entry, field)
else:
value = _fetch_value(
client, entry, field, session_evidence=session_evidence
)
if companions and (binding_digest is None or mode != "exec-env"):
raise DeliveryError("companion delivery requires a configured exec owner and exec-env")
# Every lane is read through its own AppRole session; any failure raises
# before a child exists, and the values already read go out of scope.
extra: dict[str, str] = {}

View file

@ -152,6 +152,8 @@ def _check_path(path: Path, *, directory: bool = False, private: bool = False) -
def validate_delivery_target(entry, field: str, command: list[str], mode: str) -> str | None:
if entry.delivery_config.get("companion_of"):
raise DeliveryError("companion-only lane requires delivery through its bound primary owner")
binding = owner_binding(entry)
if binding is None:
return None