secrets-engine/tests/test_integration_bao.py

"""Live integration test against a throwaway OpenBao dev server.

Skipped automatically if the `bao` CLI is not on PATH. Boots an in-memory dev
server on a private port, then drives apply -> provision -> verify(+/-) ->
exec-delivery and asserts the value is reachable by the child but not the parent.
"""
import json
import os
import shutil
import socket
import subprocess
import time
from pathlib import Path

import pytest

from secrets_engine.apply import apply_plan
from secrets_engine.catalog import get_entry
from secrets_engine.config import repo_root
from secrets_engine.errors import BackendError
from secrets_engine.exec_delivery import exec_with_secret
from secrets_engine.openbao import OpenBaoClient
from secrets_engine.plan import build_plan
from secrets_engine.provision import provision_from_file
from secrets_engine.verify import verify_negative, verify_positive

pytestmark = pytest.mark.skipif(
    shutil.which("bao") is None and shutil.which("vault") is None,
    reason="no OpenBao/Vault CLI on PATH",
)


def _free_port() -> int:
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return port


@pytest.fixture()
def bao_dev():
    bao = shutil.which("bao") or shutil.which("vault")
    port = _free_port()
    addr = f"http://127.0.0.1:{port}"
    token = "se-test-root"
    proc = subprocess.Popen(
        [bao, "server", "-dev", "-dev-no-store-token", f"-dev-root-token-id={token}",
         f"-dev-listen-address=127.0.0.1:{port}"],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )
    client = OpenBaoClient(addr=addr, token=token, bao_bin=bao)
    for _ in range(50):
        if client.is_reachable():
            break
        time.sleep(0.2)
    else:
        proc.kill()
        pytest.fail("dev OpenBao did not become reachable")
    try:
        yield client
    finally:
        proc.kill()


def test_full_chain(bao_dev, tmp_path):
    client = bao_dev
    entry = get_entry(repo_root() / "catalog", "whynot-design-npm-publish")

    plan = build_plan(entry, "prod", decision_id="test")
    apply_plan(client, entry, plan)

    # provision from a mode-0600 file outside the repo (tmp_path is outside)
    tokenfile = tmp_path / "tok"
    tokenfile.write_text("npm_integrationTESTvalue1234567890")
    os.chmod(tokenfile, 0o600)
    provision_from_file(client, entry, "npm_token", tokenfile)

    # Server-side patch preserves an unmentioned sibling. Provisioning the
    # declared field again must not replace that sibling.
    client.kv_patch_fields(
        entry.mount, entry.path, {"integration_sibling": "still-present"}
    )
    tokenfile.write_text("npm_integrationTESTvalue0987654321")
    os.chmod(tokenfile, 0o600)
    provision_from_file(client, entry, "npm_token", tokenfile)
    assert client.kv_field_present(
        entry.mount, entry.path, "integration_sibling", token=client.token
    )

    pos = verify_positive(client, entry, "npm_token")
    assert pos.passed, pos.detail
    client.write_policy(
        "se-unrelated-denied",
        f'path "{entry.mount}/data/{entry.path}" {{ capabilities = ["deny"] }}\n',
    )
    unrelated = json.loads(
        client._run_ok(
            ["token", "create", "-format=json", "-policy=se-unrelated-denied"]
        )
    )["auth"]["client_token"]
    neg = verify_negative(client, entry, unrelated_token=unrelated)
    assert neg.passed, neg.detail

    # A real unrelated identity with accidental policy overlap must make the
    # negative check fail; an invalid/garbage token could not detect this.
    client.write_policy(
        "se-unrelated-overlap",
        f'path "{entry.mount}/data/{entry.path}" {{ capabilities = ["read"] }}\n',
    )
    overlapping = json.loads(
        client._run_ok(
            ["token", "create", "-format=json", "-policy=se-unrelated-overlap"]
        )
    )["auth"]["client_token"]
    leaked = verify_negative(client, entry, unrelated_token=overlapping)
    assert not leaked.passed, leaked.detail

    # exec delivery: child can resolve token via npmrc; assert via a probe script
    probe = tmp_path / "probe.sh"
    probe.write_text(
        "#!/usr/bin/env bash\n"
        'grep -q _authToken "$NPM_CONFIG_USERCONFIG" && echo CHILD_HAS_TOKEN\n'
    )
    os.chmod(probe, 0o755)
    rc = exec_with_secret(client, entry, "npm_token", [str(probe)], mode="npm-config")
    assert rc == 0
    # the parent process never received the value as an env var
    assert "SE_NPM_TOKEN" not in os.environ


def test_merge_safe_patch_rejects_stale_cas(bao_dev):
    client = bao_dev
    mount = "cas-test"
    path = "build/example"
    client.ensure_kv_mount(mount)
    client.kv_patch_fields(mount, path, {"first": "one", "second": "two"})
    stale = client.kv_current_version(mount, path)
    client.kv_patch_fields(mount, path, {"first": "new"}, expected_version=stale)

    with pytest.raises(BackendError):
        client.kv_patch_fields(
            mount, path, {"second": "stale-write"}, expected_version=stale
        )

    assert client.kv_field_present(mount, path, "first", token=client.token)
    assert client.kv_field_present(mount, path, "second", token=client.token)


def test_idempotent_apply(bao_dev):
    client = bao_dev
    entry = get_entry(repo_root() / "catalog", "whynot-design-npm-publish")
    plan = build_plan(entry, "prod", decision_id="test")
    first = apply_plan(client, entry, plan)
    second = apply_plan(client, entry, plan)
    # policy should be reported unchanged on the second apply
    assert any("unchanged" in s for s in second.skipped)