tegwick
bc58184d71
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
67 lines
3 KiB
Python
67 lines
3 KiB
Python
"""Two cataloged KV lanes delivered to one bound owner on a throwaway OpenBao.
|
|
|
|
Each lane gets its own policy and AppRole, and each value is read through that
|
|
lane's own AppRole session. The child sees both values; the output does not.
|
|
"""
|
|
import copy
|
|
import hashlib
|
|
import os
|
|
|
|
from secrets_engine.apply import apply_plan
|
|
from secrets_engine.catalog import validate_entry
|
|
from secrets_engine.exec_delivery import exec_with_secret
|
|
from secrets_engine.exec_owner import owner_digest, resolve_companions
|
|
from secrets_engine.plan import build_plan
|
|
from secrets_engine.provision import provision_from_file
|
|
from tests.test_exec_owner import bound # noqa: F401 (fixture)
|
|
from tests.test_exec_owner_companions import _companion, _lookup, _with_companion
|
|
from tests.test_integration_bao import bao_dev, pytestmark # noqa: F401
|
|
|
|
|
|
def _provision(client, entry, field, value, tmp_path):
|
|
source = tmp_path / f"{entry.id}.value"
|
|
source.write_text(value)
|
|
os.chmod(source, 0o600)
|
|
apply_plan(client, entry, build_plan(entry, entry.stage, decision_id="test"))
|
|
provision_from_file(client, entry, field, source)
|
|
|
|
|
|
def test_two_lanes_reach_one_bound_owner(bao_dev, bound, tmp_path, capsys):
|
|
data, _, script = bound
|
|
script.write_text('''test "$API_TOKEN" = primary-integration-value || exit 10
|
|
test "$WORKER_TOKEN" = worker-integration-value || exit 11
|
|
test -z "$BAO_TOKEN" || exit 12
|
|
read ignored && exit 15
|
|
printf '%s %s\\n' "$API_TOKEN" "$WORKER_TOKEN"
|
|
printf 'two-lane-child-ok\\n'
|
|
''')
|
|
data["delivery_config"]["exec_owner"]["files"][str(script)]["sha256"] = hashlib.sha256(
|
|
script.read_bytes()
|
|
).hexdigest()
|
|
entry = validate_entry(_with_companion(copy.deepcopy(data)))
|
|
lane = validate_entry(_companion())
|
|
_provision(bao_dev, entry, "api_token", "primary-integration-value", tmp_path)
|
|
_provision(bao_dev, lane, "worker_token", "worker-integration-value", tmp_path)
|
|
|
|
sessions, primary_session = {}, {}
|
|
rc = exec_with_secret(
|
|
bao_dev, entry, "api_token", data["delivery_config"]["exec_owner"]["command"],
|
|
mode="exec-env", session_evidence=primary_session,
|
|
expected_owner_digest=owner_digest(entry),
|
|
companions=resolve_companions(entry, _lookup(lane)), companion_sessions=sessions,
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert rc == 0 and "two-lane-child-ok" in out
|
|
assert "primary-integration-value" not in out and "worker-integration-value" not in out
|
|
assert primary_session.get("established") and sessions["test-worker"].get("established")
|
|
|
|
|
|
def test_primary_approle_cannot_read_companion_path(bao_dev, bound, tmp_path):
|
|
data, _, _ = bound
|
|
entry = validate_entry(_with_companion(copy.deepcopy(data)))
|
|
lane = validate_entry(_companion())
|
|
_provision(bao_dev, entry, "api_token", "primary-integration-value", tmp_path)
|
|
_provision(bao_dev, lane, "worker_token", "worker-integration-value", tmp_path)
|
|
with bao_dev.approle_session(entry.role_name) as session:
|
|
proc = session.client._run(["kv", "get", "-format=json", f"{lane.mount}/{lane.path}"])
|
|
assert proc.returncode != 0
|