secrets-engine/tests/test_session_revoke.py

import json
from types import SimpleNamespace

import pytest

from secrets_engine import cli
from secrets_engine.config import Config
from secrets_engine.errors import BackendError, DecisionError, ProvisioningError
from secrets_engine.openbao import OpenBaoClient, accessor_fingerprint


ACCESSOR = "test-known-accessor-value"


def _config(tmp_path):
    return Config(
        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",
    )


def _accessor_file(tmp_path, value=ACCESSOR, mode=0o600):
    path = tmp_path / "accessor.handle"
    path.write_text(value, encoding="utf-8")
    path.chmod(mode)
    return path


def test_session_revoke_uses_fingerprint_only(tmp_path, monkeypatch):
    seen = {}

    class _Client:
        def revoke_accessor(self, accessor):
            seen["accessor"] = accessor

    monkeypatch.setattr(
        cli, "_open_backend", lambda *_args, **_kwargs: _ctx(_Client())
    )
    args = SimpleNamespace(
        accessor_file=str(_accessor_file(tmp_path)),
        lease_id_file=None,
        stage="test",
        bootstrap_token_file=None,
        auth="auto",
    )
    rc = cli.cmd_session_revoke(_config(tmp_path), args)
    assert rc == 0
    assert seen["accessor"] == ACCESSOR
    records = [
        json.loads(line)
        for line in next((tmp_path / "evidence").glob("evidence-*.jsonl")).read_text().splitlines()
    ]
    dumped = json.dumps(records)
    assert ACCESSOR not in dumped
    assert records[-1]["detail"]["session_handle"] == accessor_fingerprint(ACCESSOR)
    assert records[-1]["result"] == "revoked"


class _ctx:
    def __init__(self, client):
        self.client = client

    def __enter__(self):
        return self.client

    def __exit__(self, *_args):
        return False


def test_session_revoke_production_fails_closed_before_backend(tmp_path, monkeypatch):
    monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False)
    monkeypatch.setattr(
        cli.OpenBaoClient,
        "resolve",
        lambda *_args, **_kwargs: pytest.fail("backend must not be reached"),
    )
    args = SimpleNamespace(
        accessor_file=str(_accessor_file(tmp_path)),
        stage="prod",
        bootstrap_token_file=None,
        auth="auto",
    )
    cfg = Config(
        catalog_dir=tmp_path,
        policy_dir=tmp_path,
        evidence_dir=tmp_path / "evidence",
        hub_url="http://127.0.0.1:8000",
        bao_addr="http://127.0.0.1:8200",
        topic_id="test-topic",
    )
    with pytest.raises(DecisionError, match="production action 'session-revoke'"):
        cli.cmd_session_revoke(cfg, args)


def test_session_revoke_lease_uses_fingerprint_only(tmp_path, monkeypatch):
    lease = "kv/data/test/team/thing/lease-id-secretish"
    seen = {}

    class _Client:
        def revoke_lease(self, lease_id):
            seen["lease"] = lease_id

        def revoke_accessor(self, _accessor):
            pytest.fail("must not revoke accessor")

    monkeypatch.setattr(cli, "_open_backend", lambda *_args, **_kwargs: _ctx(_Client()))
    path = tmp_path / "lease.handle"
    path.write_text(lease, encoding="utf-8")
    path.chmod(0o600)
    args = SimpleNamespace(
        accessor_file=None,
        lease_id_file=str(path),
        stage="test",
        bootstrap_token_file=None,
        auth="auto",
    )
    rc = cli.cmd_session_revoke(_config(tmp_path), args)
    assert rc == 0
    assert seen["lease"] == lease
    records = [
        json.loads(line)
        for line in next((tmp_path / "evidence").glob("evidence-*.jsonl")).read_text().splitlines()
    ]
    dumped = json.dumps(records)
    assert lease not in dumped
    assert records[-1]["detail"]["handle_kind"] == "lease"
    assert records[-1]["detail"]["session_handle"] == accessor_fingerprint(lease)


def test_session_revoke_requires_exactly_one_handle(tmp_path):
    args = SimpleNamespace(
        accessor_file=None,
        lease_id_file=None,
        stage="test",
        bootstrap_token_file=None,
        auth="auto",
    )
    with pytest.raises(ProvisioningError, match="exactly one"):
        cli.cmd_session_revoke(_config(tmp_path), args)


def test_revoke_accessor_rejects_blank(monkeypatch):
    client = OpenBaoClient(addr="http://example.invalid", token="t", bao_bin="bao")
    monkeypatch.setattr(
        client, "_run_ok", lambda *_args, **_kwargs: pytest.fail("must not call bao")
    )
    with pytest.raises(BackendError, match="missing or invalid"):
        client.revoke_accessor(" ")
    with pytest.raises(BackendError, match="missing or invalid"):
        client.revoke_lease("")