secrets-engine/tests/test_engine_auth.py

import copy
from types import SimpleNamespace

import pytest
import yaml

from secrets_engine.catalog import validate_entry
from secrets_engine.config import Config
from secrets_engine.engine_auth import (
    login_service_jwt,
    select_engine_auth,
)
from secrets_engine.errors import BackendError
from secrets_engine.openbao import OpenBaoClient
from tests.test_catalog import VALID


def _cfg(tmp_path, **overrides):
    values = dict(
        catalog_dir=tmp_path,
        policy_dir=tmp_path,
        evidence_dir=tmp_path / "evidence",
        hub_url="",
        bao_addr="http://127.0.0.1:8200",
        topic_id="test-topic",
    )
    values.update(overrides)
    return Config(**values)


def _jwt_contract(tmp_path, issuer="https://keycape.example.test"):
    path = tmp_path / "jwt-login.yaml"
    path.write_text(
        yaml.safe_dump({"mount": "jwt", "role": "secrets-engine", "bound_issuer": issuer}),
        encoding="utf-8",
    )
    return path


def test_auto_without_jwt_contract_keeps_bootstrap_and_env(tmp_path):
    cfg = _cfg(tmp_path)
    env = select_engine_auth(cfg, SimpleNamespace(auth="auto", bootstrap_token_file=None))
    assert env.provider == "env"
    boot = select_engine_auth(
        cfg, SimpleNamespace(auth="auto", bootstrap_token_file="/tmp/bootstrap.token")
    )
    assert boot.provider == "bootstrap"
    assert boot.break_glass is True


def test_service_jwt_refuses_bootstrap_file_and_does_not_read_env(tmp_path, monkeypatch):
    monkeypatch.setenv("BAO_TOKEN", "must-not-be-used")
    cfg = _cfg(tmp_path, openbao_jwt_login_file=_jwt_contract(tmp_path))
    with pytest.raises(BackendError, match="no fallback"):
        select_engine_auth(
            cfg,
            SimpleNamespace(auth="auto", bootstrap_token_file="/tmp/bootstrap.token"),
        )
    with pytest.raises(BackendError, match="no fallback"):
        select_engine_auth(
            cfg,
            SimpleNamespace(auth="service-jwt", bootstrap_token_file="/tmp/bootstrap.token"),
        )
    selected = select_engine_auth(
        cfg, SimpleNamespace(auth="auto", bootstrap_token_file=None)
    )
    assert selected.provider == "service-jwt"


def test_explicit_service_jwt_fails_closed_without_contract(tmp_path, monkeypatch):
    monkeypatch.setenv("BAO_TOKEN", "must-not-be-used")
    cfg = _cfg(tmp_path)
    selected = select_engine_auth(
        cfg, SimpleNamespace(auth="service-jwt", bootstrap_token_file=None)
    )
    assert selected.provider == "service-jwt"
    with pytest.raises(BackendError, match="JWT mount/role contract is not published"):
        login_service_jwt(cfg)


def test_jwt_login_failure_does_not_fall_back_to_env(tmp_path, monkeypatch):
    monkeypatch.setenv("BAO_TOKEN", "must-not-be-used")
    cfg = _cfg(
        tmp_path,
        openbao_jwt_login_file=_jwt_contract(tmp_path),
        keycape_token_url="https://keycape.example.test/token",
        keycape_issuer="https://keycape.example.test",
    )
    monkeypatch.setattr(
        "secrets_engine.engine_auth.KeyCapeServiceAuthProvider.exchange",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(BackendError("exchange failed")),
    )
    resolved = []
    monkeypatch.setattr(
        OpenBaoClient,
        "resolve",
        lambda *_args, **_kwargs: resolved.append("used") or pytest.fail("fallback"),
    )
    with pytest.raises(BackendError, match="exchange failed"):
        login_service_jwt(cfg)
    assert resolved == []


def test_login_jwt_keeps_jwt_out_of_argv_and_revokes(monkeypatch):
    client = OpenBaoClient(addr="http://example.invalid", token="", bao_bin="bao")
    captured = {}

    def fake_json_call(args, payload, *, anonymous=False):
        captured["anonymous"] = anonymous
        captured["args"] = list(args)
        captured["payload"] = dict(payload)
        return '{"auth":{"client_token":"jwt-child-token","accessor":"jwt-accessor"}}'

    monkeypatch.setattr(client, "_run_ok_with_json_file", fake_json_call)
    session = client.login_jwt("jwt", "secrets-engine", "header.payload.sig")
    assert "header.payload.sig" not in " ".join(captured["args"])
    assert captured["payload"]["jwt"] == "header.payload.sig"
    assert captured["anonymous"] is True
    assert "jwt-accessor" not in session.accessor_fingerprint
    revoke = []
    monkeypatch.setattr(
        session.client,
        "_run_ok",
        lambda args, **_kwargs: revoke.append(list(args)) or "",
    )
    session.close()
    assert revoke == [["token", "revoke", "-self"]]
    assert session.client.token == ""


def test_provision_jwt_auth_never_reaches_openbao_on_missing_contract(
    tmp_path, monkeypatch
):
    from secrets_engine import cli

    entry = validate_entry(copy.deepcopy(VALID))
    monkeypatch.setattr(cli, "get_entry", lambda *_args: entry)
    monkeypatch.setattr(cli, "_require_lane_approval", lambda *_args, **_kwargs: None)
    monkeypatch.setattr(
        cli.OpenBaoClient,
        "resolve",
        lambda *_args, **_kwargs: pytest.fail("backend must not be reached"),
    )
    args = SimpleNamespace(
        catalog_id=entry.id,
        stage=entry.stage,
        field="api_token",
        generate=False,
        from_file="/tmp/test-value-file",
        bootstrap_token_file=None,
        auth="service-jwt",
    )
    cfg = _cfg(tmp_path)
    with pytest.raises(BackendError, match="JWT mount/role contract is not published"):
        cli.cmd_provision(cfg, args)


def test_login_jwt_presents_no_ambient_env_or_helper_token(monkeypatch):
    monkeypatch.setenv("BAO_TOKEN", "env-token-must-not-be-sent")
    monkeypatch.setenv("VAULT_TOKEN", "env-token-must-not-be-sent")
    seen = {}

    def fake_run(argv, **kwargs):
        env = kwargs["env"]
        seen["env"] = dict(env)
        seen["helper_config"] = open(env["BAO_CONFIG_PATH"], encoding="utf-8").read()
        return SimpleNamespace(
            returncode=0,
            stdout='{"auth":{"client_token":"jwt-child-token","accessor":"jwt-accessor"}}',
            stderr="",
        )

    monkeypatch.setattr("secrets_engine.openbao.subprocess.run", fake_run)
    client = OpenBaoClient(addr="http://example.invalid", token="", bao_bin="bao")
    client.login_jwt("jwt", "secrets-engine", "header.payload.sig")
    assert "BAO_TOKEN" not in seen["env"]
    assert "VAULT_TOKEN" not in seen["env"]
    assert seen["env"]["VAULT_CONFIG_PATH"] == seen["env"]["BAO_CONFIG_PATH"]
    assert seen["helper_config"].startswith('token_helper = "')