secrets-engine/tests/test_approval_consume.py

import copy
import json
from io import BytesIO
from types import SimpleNamespace
from urllib.error import HTTPError, URLError
from urllib.request import Request

import pytest

from secrets_engine import cli
from secrets_engine.approval_consume import (
    ConsumeBinding,
    consume_approval,
    require_production_consume,
    resolve_consume_binding,
)
from secrets_engine.catalog import validate_entry
from secrets_engine.config import Config
from secrets_engine.errors import DecisionError, ProvisioningError
from secrets_engine.pep_stance import StanceApplication
from tests.test_catalog import VALID

DIGEST = "sha256:" + ("ab" * 32)
TOKEN = "test-approval-consume-credential"


def _binding(**overrides):
    payload = {
        "approval_id": "appr_test-1",
        "request_digest": DIGEST,
        "decision_id": "decision:test",
    }
    payload.update(overrides)
    return ConsumeBinding(**payload)


def _token_file(tmp_path, value=TOKEN, mode=0o600):
    path = tmp_path / "approval.token"
    path.write_text(value, encoding="utf-8")
    path.chmod(mode)
    return path


def _consumed_payload(**overrides):
    payload = {
        "approval_id": "appr_test-1",
        "status": "consumed",
        "request_digest": DIGEST,
        "decision_id": "decision:test",
        "consumed_at": "2026-09-02T12:00:00Z",
        "idempotent": False,
    }
    payload.update(overrides)
    return payload


class _Response:
    def __init__(self, status, payload):
        self._status = status
        self._raw = json.dumps(payload).encode("utf-8")

    def getcode(self):
        return self._status

    def read(self, _n=-1):
        return self._raw

    def close(self):
        return None


def _http_error(status):
    return HTTPError(
        "http://approval.test/v1/approvals/appr_test-1/consume",
        status,
        "error",
        hdrs=None,
        fp=BytesIO(b'{"error":"no"}'),
    )


def test_resolve_consume_binding_is_unserved():
    assert resolve_consume_binding(object(), object(), "apply", None) is None


def test_consume_success_and_same_digest_retry(tmp_path):
    seen = []

    def opener(request, timeout):
        seen.append((request.full_url, request.get_header("Authorization"), timeout))
        assert request.get_method() == "POST"
        body = json.loads(request.data.decode())
        assert body["request_digest"] == DIGEST
        assert body["decision_id"] == "decision:test"
        return _Response(200, _consumed_payload(idempotent=bool(seen[1:])))

    first = consume_approval(
        base_url="http://approval.test",
        token_file=_token_file(tmp_path),
        binding=_binding(),
        opener=opener,
    )
    retry = consume_approval(
        base_url="http://approval.test",
        token_file=_token_file(tmp_path),
        binding=_binding(),
        opener=opener,
    )
    assert first.request_digest == DIGEST
    assert first.idempotent is False
    assert retry.idempotent is True
    assert TOKEN not in repr(first)
    assert all(call[1] == f"Bearer {TOKEN}" for call in seen)


@pytest.mark.parametrize(
    "status,match",
    [
        (409, "conflict"),
        (404, "not found"),
        (401, "unauthorized"),
        (403, "unauthorized"),
        (503, "unavailable"),
        (500, "failed"),
    ],
)
def test_consume_http_errors_fail_closed(tmp_path, status, match):
    def opener(request, timeout):
        raise _http_error(status)

    with pytest.raises(DecisionError, match=match) as raised:
        consume_approval(
            base_url="http://approval.test",
            token_file=_token_file(tmp_path),
            binding=_binding(),
            opener=opener,
        )
    assert "OpenBao must not be called" in str(raised.value)
    assert TOKEN not in str(raised.value)


def test_consume_unreachable_fails_closed(tmp_path):
    def opener(request, timeout):
        raise URLError("down")

    with pytest.raises(DecisionError, match="unreachable"):
        consume_approval(
            base_url="http://approval.test",
            token_file=_token_file(tmp_path),
            binding=_binding(),
            opener=opener,
        )


def test_consume_rejects_digest_mismatch_and_non_consumed_status(tmp_path):
    def mismatch(request, timeout):
        return _Response(200, _consumed_payload(request_digest="sha256:" + ("cd" * 32)))

    def not_consumed(request, timeout):
        return _Response(200, _consumed_payload(status="approved"))

    with pytest.raises(DecisionError, match="digest does not match"):
        consume_approval(
            base_url="http://approval.test",
            token_file=_token_file(tmp_path),
            binding=_binding(),
            opener=mismatch,
        )
    with pytest.raises(DecisionError, match="not confirmed"):
        consume_approval(
            base_url="http://approval.test",
            token_file=_token_file(tmp_path),
            binding=_binding(),
            opener=not_consumed,
        )


def test_consume_does_not_treat_effect_as_permission(tmp_path):
    def opener(request, timeout):
        return _Response(
            200,
            _consumed_payload(effect="deny", allow=False, decision="no"),
        )

    consumed = consume_approval(
        base_url="http://approval.test",
        token_file=_token_file(tmp_path),
        binding=_binding(),
        opener=opener,
    )
    assert consumed.request_digest == DIGEST
    assert "effect" not in consumed.as_evidence()


def test_token_file_guards(tmp_path):
    with pytest.raises(ProvisioningError, match="must be 0600"):
        consume_approval(
            base_url="http://approval.test",
            token_file=_token_file(tmp_path, mode=0o644),
            binding=_binding(),
            opener=lambda *_args, **_kwargs: pytest.fail("must not HTTP"),
        )
    with pytest.raises(DecisionError, match="canonical request digest"):
        consume_approval(
            base_url="http://approval.test",
            token_file=_token_file(tmp_path),
            binding=_binding(request_digest="not-a-digest"),
            opener=lambda *_args, **_kwargs: pytest.fail("must not HTTP"),
        )


def test_require_production_consume_skips_non_prod_and_demo(tmp_path, monkeypatch):
    test_entry = validate_entry(copy.deepcopy(VALID))
    prod_entry = validate_entry(dict(VALID, stage="prod"))
    cfg = SimpleNamespace(
        hub_url="",
        bao_addr="http://127.0.0.1:8200",
        approval_url="http://approval.test",
        approval_token_file=_token_file(tmp_path),
    )
    assert require_production_consume(cfg, test_entry, binding=_binding()) is None
    monkeypatch.setenv("SECRETS_ENGINE_UNSAFE_DEMO", "1")
    assert require_production_consume(cfg, prod_entry, binding=_binding()) is None


def test_require_production_consume_fails_closed_without_binding():
    entry = validate_entry(dict(VALID, stage="prod"))
    cfg = SimpleNamespace(
        hub_url="http://127.0.0.1:8000",
        bao_addr="http://127.0.0.1:8200",
        approval_url="http://approval.test",
        approval_token_file=None,
    )
    with pytest.raises(DecisionError, match="no durable consume binding"):
        require_production_consume(cfg, entry, binding=None)


def _config(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",
        approval_url="http://approval.test",
        approval_token_file=_token_file(tmp_path),
    )
    values.update(overrides)
    return Config(**values)


def _provision_args(entry):
    return SimpleNamespace(
        catalog_id=entry.id,
        stage=entry.stage,
        field="api_token",
        generate=False,
        from_file="/tmp/test-value-file",
        bootstrap_token_file=None,
    )


def _allow_prod_stance(_cfg, _entry, action, **_kwargs):
    return StanceApplication(
        stage="prod",
        failure_mode="fail_closed",
        action=action,
        demo_exception=False,
    )


def test_stance_bypass_still_cannot_reach_openbao_without_consume(
    tmp_path, monkeypatch
):
    entry = validate_entry(dict(VALID, stage="prod", approval={"model": "bootstrap-only"}))
    monkeypatch.setattr(cli, "get_entry", lambda *_args: entry)
    monkeypatch.setattr(cli, "apply_unreachable_engine_stance", _allow_prod_stance)
    monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False)
    monkeypatch.setattr(
        cli.OpenBaoClient,
        "resolve",
        lambda *_args, **_kwargs: pytest.fail("backend must not be reached"),
    )
    with pytest.raises(DecisionError, match="no durable consume binding"):
        cli.cmd_provision(_config(tmp_path), _provision_args(entry))


def test_consume_conflict_prevents_openbao(tmp_path, monkeypatch):
    entry = validate_entry(dict(VALID, stage="prod", approval={"model": "bootstrap-only"}))
    monkeypatch.setattr(cli, "get_entry", lambda *_args: entry)
    monkeypatch.setattr(cli, "apply_unreachable_engine_stance", _allow_prod_stance)
    monkeypatch.setattr(
        cli,
        "resolve_consume_binding",
        lambda *_args: _binding(),
    )
    monkeypatch.setattr(
        "secrets_engine.approval_consume.urlopen",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(_http_error(409)),
    )
    monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False)
    monkeypatch.setattr(
        cli.OpenBaoClient,
        "resolve",
        lambda *_args, **_kwargs: pytest.fail("backend must not be reached"),
    )
    with pytest.raises(DecisionError, match="conflict"):
        cli.cmd_provision(_config(tmp_path), _provision_args(entry))
    records = [
        json.loads(line)
        for line in next((tmp_path / "evidence").glob("evidence-*.jsonl")).read_text().splitlines()
    ]
    assert [record["result"] for record in records] == [
        "attempt",
        "failed-DecisionError",
    ]
    assert TOKEN not in json.dumps(records)


def test_confirmed_consume_allows_openbao_resolve(tmp_path, monkeypatch):
    entry = validate_entry(dict(VALID, stage="prod", approval={"model": "bootstrap-only"}))
    reached = {}

    def opener(request, timeout):
        assert isinstance(request, Request)
        return _Response(200, _consumed_payload())

    monkeypatch.setattr(cli, "get_entry", lambda *_args: entry)
    monkeypatch.setattr(cli, "apply_unreachable_engine_stance", _allow_prod_stance)
    monkeypatch.setattr(cli, "resolve_consume_binding", lambda *_args: _binding())
    monkeypatch.setattr("secrets_engine.approval_consume.urlopen", opener)
    monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False)
    monkeypatch.setattr(
        cli.OpenBaoClient,
        "resolve",
        lambda *_args, **_kwargs: reached.setdefault("openbao", True) or object(),
    )
    monkeypatch.setattr(
        cli,
        "provision_from_file",
        lambda *_args, **_kwargs: "api_token",
    )
    rc = cli.cmd_provision(_config(tmp_path), _provision_args(entry))
    assert rc == 0
    assert reached.get("openbao") is True
    records = [
        json.loads(line)
        for line in next((tmp_path / "evidence").glob("evidence-*.jsonl")).read_text().splitlines()
    ]
    assert records[-1]["detail"]["approval_consumed"] is True
    assert records[-1]["detail"]["approval_id"] == "appr_test-1"
    assert TOKEN not in json.dumps(records)