secrets-engine/tests/test_wrap.py

import copy
import json
from types import SimpleNamespace

import pytest

from secrets_engine.catalog import validate_entry
from secrets_engine.config import Config
from secrets_engine.errors import BackendError, DecisionError, ProvisioningError
from secrets_engine.openbao import OpenBaoClient
from secrets_engine.wrap import normalize_wrap_ttl, write_wrapped_handoff
from tests.test_catalog import VALID

WRAP_TOKEN = "wrap-token-SUPER-SECRET"
WRAP_ACCESSOR = "wrap-accessor-value"


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

    def wrap_kv_get(self, mount, path, *, ttl):
        from secrets_engine.openbao import WrappedResponse, accessor_fingerprint

        self.calls.append(("kv", mount, path, ttl))
        return WrappedResponse(
            wrap_token=WRAP_TOKEN,
            accessor_fingerprint=accessor_fingerprint(WRAP_ACCESSOR),
            ttl=ttl,
            creation_path=f"{mount}/{path}",
        )

    def wrap_approle_secret_id(self, role_name, *, ttl):
        from secrets_engine.openbao import WrappedResponse, accessor_fingerprint

        self.calls.append(("approle", role_name, ttl))
        return WrappedResponse(
            wrap_token=WRAP_TOKEN,
            accessor_fingerprint=accessor_fingerprint(WRAP_ACCESSOR),
            ttl=ttl,
            creation_path=f"auth/approle/role/{role_name}/secret-id",
        )


def test_normalize_wrap_ttl_bounds():
    assert normalize_wrap_ttl("15m") == "15m"
    assert normalize_wrap_ttl("60s") == "60s"
    with pytest.raises(ProvisioningError, match="15m"):
        normalize_wrap_ttl("16m")
    with pytest.raises(ProvisioningError, match="like"):
        normalize_wrap_ttl("15")


def test_wrap_kv_writes_0600_file_and_omits_token_from_result(tmp_path):
    out = tmp_path / "wrap.token"
    entry = validate_entry(copy.deepcopy(VALID))
    result = write_wrapped_handoff(
        FakeWrapClient(), entry, out_file=out, ttl="15m"
    )
    assert out.read_text().strip() == WRAP_TOKEN
    assert (out.stat().st_mode & 0o077) == 0
    assert result.wrap_handle
    assert WRAP_TOKEN not in result.wrap_handle
    assert WRAP_TOKEN not in result.out_file
    assert WRAP_ACCESSOR not in result.wrap_handle


def test_wrap_rejects_repo_paths(tmp_path):
    import subprocess
    repository = tmp_path / "repository"
    repository.mkdir()
    subprocess.run(["git", "init", str(repository)], check=True, capture_output=True)
    entry = validate_entry(copy.deepcopy(VALID))
    with pytest.raises(ProvisioningError, match="Git worktree"):
        write_wrapped_handoff(
            FakeWrapClient(),
            entry,
            out_file=repository / "wrap.token",
            ttl="15m",
        )


def test_parse_wrap_rejects_unwrapped_secret_payload():
    client = OpenBaoClient(addr="http://example.invalid", token="t", bao_bin="bao")
    secret_json = json.dumps(
        {"data": {"data": {"api_token": "npm_SHOULDNEVERPARSE"}}}
    )
    with pytest.raises(BackendError, match="was not applied") as raised:
        client._parse_wrap_response(
            secret_json, ttl="15m", creation_path="secret/x"
        )
    assert "npm_SHOULDNEVERPARSE" not in str(raised.value)


def test_parse_wrap_accepts_wrap_info_only():
    client = OpenBaoClient(addr="http://example.invalid", token="t", bao_bin="bao")
    payload = json.dumps(
        {
            "wrap_info": {
                "token": WRAP_TOKEN,
                "accessor": WRAP_ACCESSOR,
                "ttl": 900,
                "creation_path": "secret/x",
            }
        }
    )
    wrapped = client._parse_wrap_response(
        payload, ttl="15m", creation_path="secret/x"
    )
    assert wrapped.wrap_token == WRAP_TOKEN
    assert WRAP_ACCESSOR not in wrapped.accessor_fingerprint
    assert WRAP_TOKEN not in repr(wrapped)


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

    def fake_run(args, stdin=None):
        seen["args"] = list(args)
        return SimpleNamespace(
            returncode=0,
            stdout=json.dumps(
                {"wrap_info": {"token": WRAP_TOKEN, "accessor": WRAP_ACCESSOR, "ttl": 900, "creation_path": "secret/data/test/team/thing"}}
            ),
            stderr="",
        )

    monkeypatch.setattr(client, "_run", fake_run)
    wrapped = client.wrap_kv_get("secret", "test/team/thing", ttl="15m")
    assert "-wrap-ttl=15m" in seen["args"]
    assert WRAP_TOKEN not in " ".join(seen["args"])
    assert wrapped.wrap_token == WRAP_TOKEN


def test_production_wrap_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"),
    )
    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",
    )
    args = SimpleNamespace(
        catalog_id=entry.id,
        out=str(tmp_path / "wrap.token"),
        ttl="15m",
        bootstrap_token_file=None,
        auth="auto",
    )
    with pytest.raises(DecisionError, match="production action 'wrap'"):
        cli.cmd_wrap(cfg, args)

@pytest.mark.parametrize('change', [
    {'creation_path':'secret/data/other-recipient'}, {'creation_path':None},
    {'ttl':901}, {'ttl':60}, {'ttl':True}, {'ttl':None},
    {'token':42}, {'token':'value with whitespace'}, {'accessor':None},
])
def test_native_wrap_binding_cannot_be_inferred_from_request(change):
    client=OpenBaoClient(addr='http://example.invalid',token='t',bao_bin='bao')
    value={'token':WRAP_TOKEN,'accessor':WRAP_ACCESSOR,'ttl':900,'creation_path':'secret/data/test/team/thing'}
    value.update(change)
    with pytest.raises(BackendError) as raised:
        client._parse_wrap_response(json.dumps({'wrap_info':value}),ttl='15m',creation_path='secret/data/test/team/thing')
    assert WRAP_TOKEN not in str(raised.value)


def test_valid_wrap_info_cannot_mask_unwrapped_payload():
    client=OpenBaoClient(addr='http://example.invalid',token='t',bao_bin='bao')
    value={'wrap_info':{'token':WRAP_TOKEN,'accessor':WRAP_ACCESSOR,'ttl':900,'creation_path':'secret/x'},'data':{'secret':'NEVER-LOG-ME'}}
    with pytest.raises(BackendError) as raised:
        client._parse_wrap_response(json.dumps(value),ttl='15m',creation_path='secret/x')
    assert 'NEVER-LOG-ME' not in str(raised.value)


def test_existing_output_prevents_mint_and_preserves_original(tmp_path):
    out=tmp_path/'wrap.token';out.write_text('previous trial');out.chmod(0o600)
    client=FakeWrapClient()
    with pytest.raises(ProvisioningError,match='already exists'):
        write_wrapped_handoff(client,validate_entry(copy.deepcopy(VALID)),out_file=out,ttl='60s')
    assert client.calls==[] and out.read_text()=='previous trial'


def test_leaf_and_parent_symlink_cannot_redirect_handoff(tmp_path):
    from pathlib import Path
    target=tmp_path/'retained';target.write_text('private existing');target.chmod(0o600)
    out=tmp_path/'wrap.token';out.symlink_to(target)
    client=FakeWrapClient();entry=validate_entry(copy.deepcopy(VALID))
    with pytest.raises(ProvisioningError):write_wrapped_handoff(client,entry,out_file=out,ttl='60s')
    folder=tmp_path/'private';folder.mkdir(mode=0o700)
    parent=tmp_path/'redirect';parent.symlink_to(folder,target_is_directory=True)
    with pytest.raises(ProvisioningError):write_wrapped_handoff(client,entry,out_file=parent/'wrap.token',ttl='60s')
    assert client.calls==[] and target.read_text()=='private existing'


def test_backend_failure_retains_private_reservation_and_prevents_remint(tmp_path):
    class Failed(FakeWrapClient):
        def wrap_kv_get(self,*args,**kwargs):
            self.calls.append('attempt');raise BackendError('unverified native mint')
    client=Failed();out=tmp_path/'wrap.token';entry=validate_entry(copy.deepcopy(VALID))
    with pytest.raises(BackendError):write_wrapped_handoff(client,entry,out_file=out,ttl='60s')
    assert out.exists() and out.read_bytes()==b'' and out.stat().st_mode&0o077==0
    with pytest.raises(ProvisioningError):write_wrapped_handoff(client,entry,out_file=out,ttl='60s')
    assert client.calls==['attempt']


def test_shared_output_directory_refuses_before_backend(tmp_path):
    parent=tmp_path/'shared';parent.mkdir(mode=0o755);client=FakeWrapClient()
    with pytest.raises(ProvisioningError,match='owned and private'):
        write_wrapped_handoff(client,validate_entry(copy.deepcopy(VALID)),out_file=parent/'wrap.token',ttl='60s')
    assert client.calls==[]


def test_concurrent_handoff_reserves_before_only_one_backend_mint(tmp_path):
    from concurrent.futures import ThreadPoolExecutor
    import threading
    entered,release=threading.Event(),threading.Event()
    class Paused(FakeWrapClient):
        def wrap_kv_get(self,*args,**kwargs):
            entered.set();assert release.wait(5)
            return super().wrap_kv_get(*args,**kwargs)
    client=Paused();entry=validate_entry(copy.deepcopy(VALID));out=tmp_path/'wrap.token'
    with ThreadPoolExecutor(max_workers=2) as pool:
        first=pool.submit(write_wrapped_handoff,client,entry,out_file=out,ttl='60s')
        assert entered.wait(5)
        try:
            with pytest.raises(ProvisioningError):write_wrapped_handoff(client,entry,out_file=out,ttl='60s')
        finally:release.set()
        first.result()
    assert len(client.calls)==1 and out.read_text().strip()==WRAP_TOKEN


def test_output_replacement_during_backend_call_does_not_overwrite_other_recipient(tmp_path):
    out=tmp_path/'wrap.token';other=tmp_path/'other';other.write_text('retained recipient')
    class Replaced(FakeWrapClient):
        def wrap_kv_get(self,*args,**kwargs):
            out.unlink();out.symlink_to(other)
            return super().wrap_kv_get(*args,**kwargs)
    client=Replaced()
    with pytest.raises(ProvisioningError,match='reservation changed'):
        write_wrapped_handoff(client,validate_entry(copy.deepcopy(VALID)),out_file=out,ttl='60s')
    assert other.read_text()=='retained recipient' and len(client.calls)==1