secrets-engine/tests/test_consumer_secret_file.py

"""Consumer custody file checks for secrets-engine-openbao (synthetic values only)."""
from __future__ import annotations

import json
import os
from pathlib import Path

import pytest

from secrets_engine import cli
from secrets_engine.errors import ProvisioningError
from secrets_engine.service_auth import check_client_secret_file, read_client_secret_file

SYNTHETIC = "synthetic-not-a-real-client-secret"


def _private_file(tmp_path: Path, value: str = SYNTHETIC) -> Path:
    d = tmp_path / "keycape"
    d.mkdir(mode=0o700)
    d.chmod(0o700)
    f = d / "secrets-engine-openbao.client-secret"
    f.write_text(value + "\n", encoding="utf-8")
    f.chmod(0o600)
    return f


def _failed(path: Path) -> set[str]:
    return {c.name for c in check_client_secret_file(path) if not c.ok}


def test_accepts_private_file_and_reads_value(tmp_path):
    f = _private_file(tmp_path)
    assert _failed(f) == set()
    assert read_client_secret_file(f) == SYNTHETIC


def test_checks_never_carry_the_value(tmp_path):
    f = _private_file(tmp_path)
    f.chmod(0o644)
    for c in check_client_secret_file(f):
        assert SYNTHETIC not in c.detail
    with pytest.raises(ProvisioningError) as exc:
        read_client_secret_file(f)
    assert SYNTHETIC not in str(exc.value)


@pytest.mark.parametrize("mode", [0o640, 0o604, 0o660])
def test_rejects_group_or_other_file_mode(tmp_path, mode):
    f = _private_file(tmp_path)
    f.chmod(mode)
    assert "file_mode" in _failed(f)


def test_rejects_shared_parent_directory(tmp_path):
    f = _private_file(tmp_path)
    f.parent.chmod(0o750)
    assert "parent_mode" in _failed(f)


def test_rejects_symlink(tmp_path):
    f = _private_file(tmp_path)
    link = f.parent / "link"
    link.symlink_to(f)
    assert "canonical" in _failed(link)


def test_rejects_relative_missing_and_empty(tmp_path):
    assert "absolute" in _failed(Path("relative.secret"))
    assert "exists" in _failed(tmp_path / "absent")
    f = _private_file(tmp_path, value="")
    f.write_text("", encoding="utf-8")
    assert "non_empty" in _failed(f)


def test_rejects_multi_token_content(tmp_path):
    f = _private_file(tmp_path, value="two tokens")
    with pytest.raises(ProvisioningError, match="exactly one"):
        read_client_secret_file(f)


def test_rejects_file_inside_git_worktree(tmp_path):
    repo = tmp_path / "repo"
    (repo / ".git").mkdir(parents=True)
    (repo / ".git" / "HEAD").write_text("ref: refs/heads/main\n")
    f = _private_file(repo)
    assert "outside_git_worktree" in _failed(f)


@pytest.mark.skipif(os.geteuid() != 0, reason="ownership change needs root")
def test_rejects_foreign_owner(tmp_path):  # pragma: no cover - root only
    f = _private_file(tmp_path)
    os.chown(f, 65534, -1)
    assert "file_owner" in _failed(f)


def test_cli_check_reports_without_value(tmp_path, capsys):
    f = _private_file(tmp_path)
    rc = cli.main(["service-auth", "check-secret-file", "--path", str(f), "--json"])
    out = capsys.readouterr().out
    assert rc == 0
    assert SYNTHETIC not in out
    assert json.loads(out)["ok"] is True
    f.chmod(0o640)
    rc = cli.main(["service-auth", "check-secret-file", "--path", str(f)])
    out = capsys.readouterr().out
    assert rc == 1 and "FAIL file_mode" in out and SYNTHETIC not in out


def test_cli_check_uses_configured_env_path(tmp_path, capsys, monkeypatch):
    f = _private_file(tmp_path)
    monkeypatch.setenv("SECRETS_ENGINE_APPROVAL_CLIENT_SECRET_FILE", str(f))
    assert cli.main(["service-auth", "check-secret-file"]) == 0
    assert "accepted" in capsys.readouterr().out