Add companion lanes to catalog-bound exec owners (SECRETS-WP-0011 T01-T03)

A configured exec owner may receive fields from other consenting kv lanes.
Each lane is gated, consumed and read through its own AppRole; any refusal
starts no child. Companions are part of the owner digest.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 226514@bnt-lap001
Assistant-Session: 26ba103d-05fe-45a1-9cd7-9475bf239df6

src/secrets_engine/catalog.py

@@ -385,6 +385,19 @@ def validate_entry(data: dict[str, Any], *, source: str = "<memory>") -> Catalog
         if "exec-env" not in modes or set(modes) - {"exec-env", "read-check"}:
             raise CatalogError(f"{source}: exec_owner permits only exec-env and read-check")
 
+    companion_of = delivery_config.get("companion_of")
+    if companion_of is not None and (
+        not isinstance(companion_of, list)
+        or not companion_of
+        or not all(isinstance(x, str) and x.strip() for x in companion_of)
+        or len(set(companion_of)) != len(companion_of)
+        or data["kind"] != "kv"
+        or "exec-env" not in modes
+    ):
+        raise CatalogError(
+            f"{source}: delivery_config.companion_of must list primary lane ids on a kv exec-env lane"
+        )
+
     # npm-config delivery must declare WHERE it publishes (registry + scope), so
     # the registry is catalog data, never hardcoded in the engine.
     if "npm-config" in modes:

src/secrets_engine/cli.py

@@ -546,32 +546,61 @@ def cmd_wrap(cfg: Config, args) -> int:
 
 
 def cmd_exec(cfg: Config, args) -> int:
+    from contextlib import ExitStack
+
     from secrets_engine.exec_delivery import exec_with_secret
-    from secrets_engine.exec_owner import validate_delivery_target
+    from secrets_engine.exec_owner import resolve_companions, validate_delivery_target
     entry = get_entry(cfg.catalog_dir, args.catalog)
     field = args.field or (entry.fields[0] if entry.fields else "")
     session_detail: dict[str, object] = {}
+    companion_sessions: dict[str, dict[str, object]] = {}
     # Refuse a substituted recipient before consuming approval or opening Bao.
     owner_digest = validate_delivery_target(entry, field, args.command, args.mode)
+    companions = resolve_companions(entry, lambda cid: get_entry(cfg.catalog_dir, cid))
     command_name = args.command[0] if args.command else ""
-    with _privileged_evidence(
-        cfg,
-        entry,
-        "exec",
-        detail={
-            "command": command_name,
-            "mode": args.mode,
-            "field": field,
-            "session": session_detail,
-            "exec_owner_sha256": owner_digest,
-        },
-    ) as evidence:
+    with ExitStack() as stack:
+        evidence = stack.enter_context(_privileged_evidence(
+            cfg,
+            entry,
+            "exec",
+            detail={
+                "command": command_name,
+                "mode": args.mode,
+                "field": field,
+                "session": session_detail,
+                "exec_owner_sha256": owner_digest,
+                "companions": [lane.id for lane, _, _ in companions],
+            },
+        ))
         # require approval + readiness before running.
         decision = _require_lane_approval(
             cfg, entry, "exec", evidence, fields=(field,) if field else ()
         )
         evidence.mark_approved(decision)
         require_delivery_state(cfg.evidence_dir, entry.id, "exec")
+        # Each companion read is its own protected action: its own evidence,
+        # stance, approval and consume. No lane's decision covers another.
+        companion_evidence = []
+        for lane, lane_field, _env in companions:
+            lane_evidence = stack.enter_context(_privileged_evidence(
+                cfg,
+                lane,
+                "exec",
+                detail={
+                    "command": command_name,
+                    "mode": "exec-env",
+                    "field": lane_field,
+                    "companion_of": entry.id,
+                    "session": companion_sessions.setdefault(lane.id, {}),
+                    "exec_owner_sha256": owner_digest,
+                },
+            ))
+            lane_decision = _require_lane_approval(
+                cfg, lane, "exec", lane_evidence, fields=(lane_field,)
+            )
+            lane_evidence.mark_approved(lane_decision)
+            require_delivery_state(cfg.evidence_dir, lane.id, "exec")
+            companion_evidence.append(lane_evidence)
         if not args.command:
             from secrets_engine.errors import DeliveryError
 
@@ -585,7 +614,11 @@ def cmd_exec(cfg: Config, args) -> int:
                 mode=args.mode,
                 session_evidence=session_detail,
                 expected_owner_digest=owner_digest,
+                companions=companions,
+                companion_sessions=companion_sessions,
             )
+        for lane_evidence in companion_evidence:
+            lane_evidence.finish(f"exit-{rc}")
         evidence.finish(f"exit-{rc}")
     return rc
 

src/secrets_engine/exec_delivery.py

@@ -164,11 +164,11 @@ def _secret_file(value: str) -> Iterator[Path]:
         _unlink_secret_file(path)
 
 
-def _stream_redacted(proc: subprocess.Popen, secret: str) -> None:
+def _stream_redacted(proc: subprocess.Popen, secrets: list[str]) -> None:
     """Stream child output through the redactor (backstop)."""
     assert proc.stdout is not None
     for line in proc.stdout:
-        sys.stdout.write(redact_text(line, extra=[secret]))
+        sys.stdout.write(redact_text(line, extra=secrets))
         sys.stdout.flush()
 
 
@@ -182,6 +182,8 @@ def exec_with_secret(
     policy_dir=None,
     session_evidence: dict[str, object] | None = None,
     expected_owner_digest: str | None = None,
+    companions: list[tuple[CatalogEntry, str, str]] = (),
+    companion_sessions: dict[str, dict[str, object]] | None = None,
 ) -> int:
     """Run `command` with the lane's secret injected for the child only.
 
@@ -225,6 +227,16 @@ def exec_with_secret(
         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] = {}
+    for lane, lane_field, env_name in companions:
+        lane_session: dict[str, object] = (
+            companion_sessions.setdefault(lane.id, {}) if companion_sessions is not None else {}
+        )
+        extra[env_name] = _fetch_value(client, lane, lane_field, session_evidence=lane_session)
     # Recheck after retrieval too: a config changed during auth/read cannot be
     # launched with a value authorized for the previous recipient.
     if validate_delivery_target(entry, field, command, mode) != binding_digest:
@@ -245,27 +257,28 @@ def exec_with_secret(
         with _npm_userconfig(registry, scope, token_env) as npmrc:
             child_env["NPM_CONFIG_USERCONFIG"] = str(npmrc)
             child_env[token_env] = value
-            rc = _spawn(command, child_env, value)
+            rc = _spawn(command, child_env, [value])
         return rc
 
     if mode == "exec-env":
         # Inject under a conventional name derived from the field.
         env_name = field.upper()
         child_env[env_name] = value
+        child_env.update(extra)
         if binding is not None:
-            return _spawn(command, child_env, value, cwd=binding["cwd"])
-        return _spawn(command, child_env, value)
+            return _spawn(command, child_env, [value, *extra.values()], cwd=binding["cwd"])
+        return _spawn(command, child_env, [value])
 
     if mode == "exec-file":
         env_name = f"{field.upper()}_FILE"
         with _secret_file(value) as secret_path:
             child_env[env_name] = str(secret_path)
-            return _spawn(command, child_env, value)
+            return _spawn(command, child_env, [value])
 
     raise DeliveryError(f"unsupported delivery mode '{mode}'")
 
 
-def _spawn(command: list[str], env: dict[str, str], secret: str, *, cwd: str | None = None) -> int:
+def _spawn(command: list[str], env: dict[str, str], secrets: list[str], *, cwd: str | None = None) -> int:
     """Spawn the child, stream redacted output, propagate signals, ensure cleanup."""
     try:
         proc = subprocess.Popen(
@@ -286,7 +299,7 @@ def _spawn(command: list[str], env: dict[str, str], secret: str, *, cwd: str | N
     old_int = signal.signal(signal.SIGINT, _forward)
     old_term = signal.signal(signal.SIGTERM, _forward)
     try:
-        _stream_redacted(proc, secret)
+        _stream_redacted(proc, secrets)
         return proc.wait()
     finally:
         signal.signal(signal.SIGINT, old_int)

src/secrets_engine/exec_owner.py

@@ -32,7 +32,7 @@ def validate_exec_owner(value: object) -> dict[str, Any]:
             raise CatalogError("pending exec_owner requires only status, owner and reason")
         return value
     required = {"status", "owner", "command", "cwd", "environment", "files"}
-    if value.get("status") != "configured" or set(value) != required:
+    if value.get("status") != "configured" or not required <= set(value) <= required | {"companions"}:
         raise CatalogError("configured exec_owner requires exact command/cwd/environment/files")
     command = value["command"]
     if not isinstance(command, list) or not command or not all(_text(x) for x in command) or not _absolute(command[0]):
@@ -60,9 +60,66 @@ def validate_exec_owner(value: object) -> dict[str, Any]:
     for arg in command[1:]:
         if arg.startswith("/") and arg not in files:
             raise CatalogError("exec_owner absolute file arguments must have file pins")
+    _validate_companions(value.get("companions", []), env)
     return value
 
 
+def _validate_companions(companions: object, env: dict[str, str]) -> None:
+    """Structural check only; cross-lane checks need the catalog (resolve_companions)."""
+    if not isinstance(companions, list):
+        raise CatalogError("exec_owner companions must be a list")
+    names: set[str] = set()
+    lanes: set[tuple[str, str]] = set()
+    for spec in companions:
+        if not isinstance(spec, dict) or set(spec) != {"catalog", "field", "env"} or not all(
+            _text(spec[k]) for k in ("catalog", "field", "env")
+        ):
+            raise CatalogError("exec_owner companion requires exactly catalog, field and env")
+        name = spec["env"]
+        if not re.fullmatch(r"[A-Z_][A-Z0-9_]*", name) or name.startswith(
+            ("LD_", "DYLD_", "PYTHON", "BAO_", "VAULT_", "SECRETS_ENGINE_")
+        ):
+            raise CatalogError("exec_owner companion env name is invalid or forbidden")
+        if name in env or name in names or (spec["catalog"], spec["field"]) in lanes:
+            raise CatalogError("exec_owner companion env or lane field is duplicated")
+        names.add(name)
+        lanes.add((spec["catalog"], spec["field"]))
+
+
+def companion_specs(entry) -> list[dict[str, str]]:
+    binding = owner_binding(entry)
+    if binding is None or binding.get("status") != "configured":
+        return []
+    return list(binding.get("companions", []))
+
+
+def resolve_companions(entry, lookup) -> list[tuple[Any, str, str]]:
+    """Resolve each companion to (lane, field, env). ``lookup`` maps id -> entry.
+
+    A companion lane must consent by naming this primary in its own
+    ``delivery_config.companion_of``, so no owner can list another team's lane
+    by itself. Runs before any approval is consumed or OpenBao is opened.
+    """
+    resolved = []
+    for spec in companion_specs(entry):
+        if spec["catalog"] == entry.id:
+            raise DeliveryError("exec owner companion must be a different lane")
+        try:
+            lane = lookup(spec["catalog"])
+        except Exception as exc:
+            raise DeliveryError(f"exec owner companion lane '{spec['catalog']}' is unavailable") from exc
+        if lane.kind != "kv" or lane.stage != entry.stage:
+            raise DeliveryError(f"companion lane '{lane.id}' must be a kv lane in stage '{entry.stage}'")
+        if spec["field"] not in lane.fields or "exec-env" not in lane.delivery_modes:
+            raise DeliveryError(f"companion lane '{lane.id}' does not declare exec-env delivery of that field")
+        if entry.id not in (lane.delivery_config.get("companion_of") or []):
+            raise DeliveryError(f"companion lane '{lane.id}' does not consent to delivery with '{entry.id}'")
+        if "exec_owner" in lane.delivery_config:
+            raise DeliveryError(f"companion lane '{lane.id}' must not bind its own exec owner")
+        resolved.append((lane, spec["field"], spec["env"]))
+    return resolved
+
+
 def owner_binding(entry) -> dict[str, Any] | None:
     config = entry.delivery_config
     if "exec_owner" not in config:
@@ -102,7 +159,9 @@ def validate_delivery_target(entry, field: str, command: list[str], mode: str) -
         raise DeliveryError("exec owner binding is pending; no delivery is admitted")
     if mode not in {"auto", "exec-env"} or "exec-env" not in entry.delivery_modes:
         raise DeliveryError("exec owner requires exec-env delivery")
-    if field not in entry.fields or field.upper() in binding["environment"]:
+    if field not in entry.fields or field.upper() in binding["environment"] or field.upper() in {
+        spec["env"] for spec in binding.get("companions", [])
+    }:
         raise DeliveryError("exec owner field is undeclared or conflicts with fixed environment")
     if command != binding["command"]:
         raise DeliveryError("command differs from the catalog-bound exec owner")