secrets-engine/tests/test_exec_owner_companions.py

import copy
from types import SimpleNamespace

import pytest

from secrets_engine import cli, exec_delivery
from secrets_engine.catalog import validate_entry
from secrets_engine.errors import CatalogError, DeliveryError
from secrets_engine.exec_owner import owner_digest, resolve_companions
from tests.test_catalog import VALID
from tests.test_exec_owner import bound  # noqa: F401  (fixture)
from tests.test_lane_state import _cfg


def _companion(primary_id="test-lane", **changes):
    data = copy.deepcopy(VALID)
    data.update(id="test-worker", path="test/team/worker", fields=["worker_token"])
    data["delivery_config"] = {"companion_of": [primary_id]}
    data.update(changes)
    return data


def _with_companion(data, env="WORKER_TOKEN", field="worker_token", catalog="test-worker"):
    data["delivery_config"]["exec_owner"]["companions"] = [
        {"catalog": catalog, "field": field, "env": env}
    ]
    return data


def _lookup(*entries):
    table = {e.id: e for e in entries}
    return lambda cid: table[cid]


def test_companion_value_reaches_bound_child_and_is_redacted(bound, monkeypatch, capsys):
    data, _, script = bound
    script.write_text('''test "$API_TOKEN" = synthetic-owner-value || exit 10
test "$WORKER_TOKEN" = synthetic-worker-value || exit 11
read ignored && exit 15
printf '%s %s\\n' "$API_TOKEN" "$WORKER_TOKEN"
printf 'companion-child-ok\\n'
''')
    import hashlib
    data["delivery_config"]["exec_owner"]["files"][str(script)]["sha256"] = hashlib.sha256(script.read_bytes()).hexdigest()
    entry = validate_entry(_with_companion(data))
    lane = validate_entry(_companion())
    companions = resolve_companions(entry, _lookup(lane))
    values = {"test-lane": "synthetic-owner-value", "test-worker": "synthetic-worker-value"}
    seen = []
    def fetch(client, e, field, **kwargs):
        seen.append((e.id, field))
        return values[e.id]
    monkeypatch.setattr(exec_delivery, "_fetch_value", fetch)
    sessions = {}
    rc = exec_delivery.exec_with_secret(
        object(), entry, "api_token", data["delivery_config"]["exec_owner"]["command"],
        mode="exec-env", expected_owner_digest=owner_digest(entry),
        companions=companions, companion_sessions=sessions,
    )
    out = capsys.readouterr().out
    assert rc == 0 and "companion-child-ok" in out
    assert "synthetic-owner-value" not in out and "synthetic-worker-value" not in out
    assert seen == [("test-lane", "api_token"), ("test-worker", "worker_token")]
    assert set(sessions) == {"test-worker"}


def test_companion_fetch_failure_starts_no_child(bound, monkeypatch):
    data, _, _ = bound
    entry = validate_entry(_with_companion(data))
    lane = validate_entry(_companion())
    def fetch(client, e, field, **kwargs):
        if e.id == "test-worker":
            raise DeliveryError("scoped read failed for lane 'test-worker' (denied or absent)")
        return "synthetic-owner-value"
    monkeypatch.setattr(exec_delivery, "_fetch_value", fetch)
    monkeypatch.setattr(exec_delivery, "_spawn", lambda *a, **k: pytest.fail("must not launch"))
    with pytest.raises(DeliveryError, match="test-worker"):
        exec_delivery.exec_with_secret(
            object(), entry, "api_token", data["delivery_config"]["exec_owner"]["command"],
            mode="exec-env", companions=resolve_companions(entry, _lookup(lane)),
        )


def test_companions_are_part_of_the_owner_digest(bound):
    data, _, _ = bound
    before = owner_digest(validate_entry(copy.deepcopy(data)))
    assert owner_digest(validate_entry(_with_companion(copy.deepcopy(data)))) != before
    changed = owner_digest(validate_entry(_with_companion(copy.deepcopy(data), env="OTHER_TOKEN")))
    assert changed != owner_digest(validate_entry(_with_companion(copy.deepcopy(data))))


@pytest.mark.parametrize("change", ["extra_key", "bad_env", "fixed_env", "forbidden_env", "duplicate", "not_list"])
def test_invalid_companion_spec_is_a_catalog_error(bound, change):
    data, _, _ = bound
    binding = _with_companion(data)["delivery_config"]["exec_owner"]
    spec = binding["companions"][0]
    if change == "extra_key": spec["note"] = "x"
    elif change == "bad_env": spec["env"] = "worker-token"
    elif change == "fixed_env": spec["env"] = "LANG"
    elif change == "forbidden_env": spec["env"] = "BAO_TOKEN"
    elif change == "duplicate": binding["companions"].append(dict(spec))
    else: binding["companions"] = {"catalog": "test-worker"}
    with pytest.raises(CatalogError):
        validate_entry(data)


def test_primary_field_env_cannot_collide_with_companion(bound, monkeypatch):
    data, _, _ = bound
    entry = validate_entry(_with_companion(data, env="API_TOKEN"))
    monkeypatch.setattr(exec_delivery, "_fetch_value", lambda *a, **k: pytest.fail("must refuse before read"))
    with pytest.raises(DeliveryError, match="conflicts"):
        exec_delivery.exec_with_secret(object(), entry, "api_token", data["delivery_config"]["exec_owner"]["command"], mode="exec-env")


@pytest.mark.parametrize("change", ["no_consent", "other_primary", "stage", "field", "mode", "own_owner", "self"])
def test_companion_resolution_refuses(bound, change):
    data, _, _ = bound
    lane_data = _companion()
    catalog = "test-worker"
    if change == "no_consent": lane_data["delivery_config"] = {}
    elif change == "other_primary": lane_data["delivery_config"] = {"companion_of": ["another-lane"]}
    elif change == "stage": lane_data.update(stage="build", path="build/team/worker")
    elif change == "field": lane_data["fields"] = ["other"]
    elif change == "mode": lane_data["delivery_modes"] = ["read-check"]; lane_data["delivery_config"] = {}
    elif change == "own_owner": lane_data["delivery_config"]["exec_owner"] = {"status": "pending", "owner": "x", "reason": "y"}
    else: catalog = "test-lane"
    entry = validate_entry(_with_companion(data, catalog=catalog))
    lane = validate_entry(lane_data)
    with pytest.raises(DeliveryError):
        resolve_companions(entry, _lookup(lane, entry))


@pytest.mark.parametrize("value", [[], "test-lane", ["test-lane", "test-lane"], [""]])
def test_companion_of_must_be_a_nonempty_unique_list(value):
    lane_data = _companion()
    lane_data["delivery_config"] = {"companion_of": value}
    with pytest.raises(CatalogError):
        validate_entry(lane_data)


def test_companion_of_requires_exec_env():
    lane_data = _companion()
    lane_data["delivery_modes"] = ["read-check"]
    with pytest.raises(CatalogError):
        validate_entry(lane_data)


def test_each_lane_is_gated_separately_before_backend(bound, monkeypatch, tmp_path):
    data, _, _ = bound
    entry = validate_entry(_with_companion(data))
    lane = validate_entry(_companion())
    table = {entry.id: entry, lane.id: lane}
    monkeypatch.setattr(cli, "get_entry", lambda _dir, cid: table[cid])
    gated = []
    def gate(cfg, e, action, evidence, *, fields=()):
        gated.append((e.id, action, fields))
        if e.id == "test-worker":
            raise DeliveryError("companion lane denied")
        return SimpleNamespace(status="approved")
    monkeypatch.setattr(cli, "_require_lane_approval", gate)
    monkeypatch.setattr(cli, "require_delivery_state", lambda *a: None)
    monkeypatch.setattr(cli, "_open_backend", lambda *a, **k: pytest.fail("no backend after a lane refusal"))
    command = data["delivery_config"]["exec_owner"]["command"]
    with pytest.raises(DeliveryError, match="companion lane denied"):
        cli.cmd_exec(_cfg(tmp_path), SimpleNamespace(field=None, catalog=entry.id, command=command, mode="exec-env"))
    assert gated == [("test-lane", "exec", ("api_token",)), ("test-worker", "exec", ("worker_token",))]


def test_unresolvable_companion_refuses_before_any_gate(bound, monkeypatch, tmp_path):
    data, _, _ = bound
    entry = validate_entry(_with_companion(data))
    def get(_dir, cid):
        if cid == entry.id:
            return entry
        raise KeyError(cid)
    monkeypatch.setattr(cli, "get_entry", get)
    monkeypatch.setattr(cli, "_require_lane_approval", lambda *a, **k: pytest.fail("no gate"))
    monkeypatch.setattr(cli, "_open_backend", lambda *a, **k: pytest.fail("no backend"))
    command = data["delivery_config"]["exec_owner"]["command"]
    with pytest.raises(DeliveryError, match="unavailable"):
        cli.cmd_exec(_cfg(tmp_path), SimpleNamespace(field=None, catalog=entry.id, command=command, mode="exec-env"))