secrets-engine/tests/test_rotate.py

import copy
from types import SimpleNamespace

import pytest

from secrets_engine.catalog import validate_entry
from secrets_engine.config import Config
from secrets_engine.errors import DecisionError, ProvisioningError
from secrets_engine.lane_state import load_lane_state, save_lane_state
from secrets_engine.rotate import render_rotate_plan, rotate_from_file
from tests.test_catalog import VALID


class RecordingPatchClient:
    def __init__(self):
        self.calls = []

    def ensure_kv_mount(self, mount):
        self.calls.append(("ensure", mount))

    def kv_patch_fields(self, mount, path, values):
        self.calls.append(("patch", mount, path, list(values)))
        return 1


def _cfg(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 test_rotate_plan_is_non_secret():
    entry = validate_entry(copy.deepcopy(VALID))
    text = render_rotate_plan(entry, "api_token")
    assert "api_token" in text
    assert "test/team/thing" in text
    assert "SUPER" not in text


def test_rotate_patches_declared_field_only(tmp_path):
    entry = validate_entry(copy.deepcopy(VALID))
    value = tmp_path / "new.value"
    value.write_text("replacement-value", encoding="utf-8")
    value.chmod(0o600)
    client = RecordingPatchClient()
    field = rotate_from_file(client, entry, "api_token", value)
    assert field == "api_token"
    assert client.calls[-1][0] == "patch"
    assert client.calls[-1][3] == ["api_token"]


def test_rotate_rejects_auth_capability():
    from tests.test_auth_capability import AUTH

    entry = validate_entry(copy.deepcopy(AUTH))
    with pytest.raises(ProvisioningError, match="auth-capability"):
        rotate_from_file(object(), entry, "api_token", __import__("pathlib").Path("/tmp/x"))


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

    entry = validate_entry(copy.deepcopy(VALID))
    save_lane_state(
        tmp_path / "evidence", entry.id, "compromised", operation="compromise"
    )
    monkeypatch.setattr(cli, "get_entry", lambda *_args: entry)
    monkeypatch.setattr(cli, "_require_lane_approval", lambda *_args, **_kwargs: None)
    monkeypatch.setattr(
        "secrets_engine.rotate.rotate_from_file",
        lambda *_args, **_kwargs: "api_token",
    )

    class _Client:
        pass

    monkeypatch.setattr(
        cli, "_open_backend", lambda *_a, **_k: __import__("contextlib").nullcontext(_Client())
    )
    args = SimpleNamespace(
        catalog_id=entry.id,
        stage=entry.stage,
        field="api_token",
        from_file="/tmp/new.value",
        dry_run=False,
        bootstrap_token_file=None,
        auth="auto",
    )
    rc = cli.cmd_rotate(_cfg(tmp_path), args)
    assert rc == 0
    assert load_lane_state(tmp_path / "evidence", entry.id).state == "compromised"
    assert load_lane_state(tmp_path / "evidence", entry.id).last_operation == "rotate"


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

    data = copy.deepcopy(VALID)
    data.update(stage="prod", approval={"model": "decision", "decision_ref": "x"})
    entry = validate_entry(data)
    monkeypatch.setattr(cli, "get_entry", lambda *_args: entry)
    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(
        catalog_id=entry.id,
        stage="prod",
        field="api_token",
        from_file="/tmp/new.value",
        dry_run=False,
        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 'rotate'"):
        cli.cmd_rotate(cfg, args)