secrets-engine/tests/test_exec_delivery.py

import copy

import pytest

from secrets_engine.catalog import validate_entry
from secrets_engine.errors import DeliveryError
from secrets_engine.exec_delivery import (
    _fetch_value,
    _npm_userconfig,
    _registry_authkey,
    exec_with_secret,
)
from tests.test_catalog import VALID


def test_registry_authkey_strips_scheme_and_trails_slash():
    assert (
        _registry_authkey("https://forgejo.coulomb.social/api/packages/coulomb/npm/")
        == "//forgejo.coulomb.social/api/packages/coulomb/npm/"
    )
    # missing trailing slash is added
    assert _registry_authkey("https://host/api/npm") == "//host/api/npm/"


def test_npm_userconfig_writes_registry_and_token_ref_not_value():
    registry = "https://forgejo.coulomb.social/api/packages/coulomb/npm/"
    with _npm_userconfig(registry, "@whynot", "NPM_AUTH_TOKEN") as path:
        body = path.read_text()
        assert f"@whynot:registry={registry}" in body
        # token is referenced via env expansion, never written literally
        assert "${NPM_AUTH_TOKEN}" in body
        assert "//forgejo.coulomb.social/api/packages/coulomb/npm/:_authToken" in body
        # file is mode 0600
        assert (path.stat().st_mode & 0o077) == 0
    # cleaned up on context exit
    assert not path.exists()


def test_exec_env_injects_only_selected_declared_field(monkeypatch):
    data = copy.deepcopy(VALID)
    data["fields"] = ["primary", "selected_value"]
    entry = validate_entry(data)

    def fake_fetch(_client, got_entry, field):
        assert got_entry == entry
        assert field == "selected_value"
        return "test-secret-value"

    def fake_spawn(command, env, secret):
        assert command == ["probe"]
        assert secret == "test-secret-value"
        assert env["SELECTED_VALUE"] == "test-secret-value"
        assert "PRIMARY" not in env
        return 0

    monkeypatch.setattr("secrets_engine.exec_delivery._fetch_value", fake_fetch)
    monkeypatch.setattr("secrets_engine.exec_delivery._spawn", fake_spawn)
    assert (
        exec_with_secret(
            object(), entry, "selected_value", ["probe"], mode="exec-env"
        )
        == 0
    )


def test_exec_file_injects_path_not_value_and_unlinks(monkeypatch, tmp_path):
    data = copy.deepcopy(VALID)
    data["delivery_modes"] = ["exec-file"]
    entry = validate_entry(data)
    seen = {}

    def fake_fetch(_client, _entry, field):
        assert field == "api_token"
        return "test-secret-value"

    def fake_spawn(command, env, secret):
        path = env["API_TOKEN_FILE"]
        seen["path"] = path
        seen["exists_during"] = __import__("pathlib").Path(path).is_file()
        seen["mode"] = __import__("pathlib").Path(path).stat().st_mode & 0o777
        seen["contents"] = __import__("pathlib").Path(path).read_text()
        seen["has_value_env"] = "API_TOKEN" in env
        assert command == ["probe"]
        assert secret == "test-secret-value"
        return 0

    monkeypatch.setattr("secrets_engine.exec_delivery._fetch_value", fake_fetch)
    monkeypatch.setattr("secrets_engine.exec_delivery._spawn", fake_spawn)
    assert exec_with_secret(object(), entry, "api_token", ["probe"], mode="exec-file") == 0
    assert seen["exists_during"] is True
    assert seen["mode"] == 0o600
    assert seen["contents"].strip() == "test-secret-value"
    assert seen["has_value_env"] is False
    assert not __import__("pathlib").Path(seen["path"]).exists()


def test_exec_file_unlinks_after_child_failure(monkeypatch):
    data = copy.deepcopy(VALID)
    data["delivery_modes"] = ["exec-file"]
    entry = validate_entry(data)
    seen = {}

    monkeypatch.setattr(
        "secrets_engine.exec_delivery._fetch_value",
        lambda *_args, **_kwargs: "test-secret-value",
    )

    def fake_spawn(command, env, secret):
        seen["path"] = env["API_TOKEN_FILE"]
        raise RuntimeError("child exploded")

    monkeypatch.setattr("secrets_engine.exec_delivery._spawn", fake_spawn)
    with pytest.raises(RuntimeError, match="child exploded"):
        exec_with_secret(object(), entry, "api_token", ["probe"], mode="exec-file")
    assert seen["path"]
    assert not __import__("pathlib").Path(seen["path"]).exists()


def test_exec_rejects_undeclared_field_before_fetch(monkeypatch):
    entry = validate_entry(VALID)
    monkeypatch.setattr(
        "secrets_engine.exec_delivery._fetch_value",
        lambda *_args, **_kwargs: pytest.fail("must not fetch undeclared field"),
    )
    with pytest.raises(DeliveryError):
        exec_with_secret(object(), entry, "other_field", ["probe"], mode="exec-env")


def test_fetch_records_non_secret_session_cleanup_before_child(monkeypatch):
    entry = validate_entry(VALID)

    class Session:
        def __init__(self):
            self.client = self
            self.closed = False

        def _run(self, _args):
            from types import SimpleNamespace
            import json

            return SimpleNamespace(
                returncode=0,
                stdout=json.dumps({"data": {"data": {"api_token": "test-value"}}}),
            )

        def evidence(self):
            return {
                "session_handle": "safe-handle",
                "established": True,
                "revocation_attempted": self.closed,
                "revocation_succeeded": self.closed,
            }

    session = Session()

    class Client:
        from contextlib import contextmanager

        @contextmanager
        def approle_session(self, _role):
            try:
                yield session
            finally:
                session.closed = True

    evidence = {}
    value = _fetch_value(Client(), entry, "api_token", session_evidence=evidence)

    assert value == "test-value"
    assert evidence == {
        "session_handle": "safe-handle",
        "established": True,
        "revocation_attempted": True,
        "revocation_succeeded": True,
    }
    assert "test-value" not in repr(evidence)