FabledSteward/tests/core/test_git_askpass.py

"""Unit tests for GIT_ASKPASS injection in git_pull()."""
import os
import stat
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest


def test_get_sources_includes_http_token(tmp_path):
    """get_sources() must pass http_token through from config."""
    from steward.ansible.sources import get_sources

    cfg = {
        "cache_dir": str(tmp_path),
        "sources": [
            {
                "name": "repo",
                "type": "git",
                "url": "https://git.example.com/org/repo.git",
                "http_token": "mytoken123",
            }
        ],
    }
    sources = get_sources(cfg)
    src = next(s for s in sources if s["name"] == "repo")
    assert src["http_token"] == "mytoken123"


def test_get_sources_missing_http_token_defaults_empty(tmp_path):
    """http_token defaults to empty string when not in source config."""
    from steward.ansible.sources import get_sources

    cfg = {
        "cache_dir": str(tmp_path),
        "sources": [
            {"name": "r", "type": "git", "url": "https://example.com/r.git"}
        ],
    }
    sources = get_sources(cfg)
    src = next(s for s in sources if s["name"] == "r")
    assert src["http_token"] == ""


@pytest.mark.asyncio
async def test_git_pull_creates_askpass_when_token_set(tmp_path):
    """git_pull() sets GIT_ASKPASS env when source has http_token."""
    from steward.ansible.sources import git_pull

    # Simulate repo already cloned (has .git dir) so we hit the pull path.
    repo_path = tmp_path / "repo"
    (repo_path / ".git").mkdir(parents=True)

    source = {
        "name": "repo",
        "path": str(repo_path),
        "url": "https://git.example.com/org/repo.git",
        "branch": "main",
        "http_token": "secret-token",
    }

    captured_env = {}

    async def fake_subprocess(*args, **kwargs):
        captured_env.update(kwargs.get("env", {}))
        proc = MagicMock()
        proc.returncode = 0
        proc.communicate = AsyncMock(return_value=(b"", b""))
        return proc

    with patch("asyncio.create_subprocess_exec", side_effect=fake_subprocess):
        await git_pull(source)

    assert "GIT_ASKPASS" in captured_env
    askpass_path = captured_env["GIT_ASKPASS"]
    # Script should be cleaned up after the call
    assert not os.path.exists(askpass_path), "askpass script should be deleted in finally"


@pytest.mark.asyncio
async def test_git_pull_askpass_reads_the_token_from_its_environment(tmp_path):
    """The askpass script holds no token; git's environment carries it."""
    from steward.ansible.sources import ASKPASS_TOKEN_ENV, git_pull

    repo_path = tmp_path / "repo"
    (repo_path / ".git").mkdir(parents=True)

    source = {
        "name": "repo",
        "path": str(repo_path),
        "url": "https://git.example.com/org/repo.git",
        "branch": "main",
        "http_token": 'tok"$(id)`x`',
    }

    captured = {}

    async def fake_subprocess(*args, **kwargs):
        env = kwargs.get("env", {})
        askpass = env.get("GIT_ASKPASS", "")
        if askpass and os.path.exists(askpass):
            captured["content"] = Path(askpass).read_text()
            mode = stat.S_IMODE(os.stat(askpass).st_mode)
            captured["executable"] = bool(mode & stat.S_IXUSR)
        captured["token"] = env.get(ASKPASS_TOKEN_ENV)
        proc = MagicMock()
        proc.returncode = 0
        proc.communicate = AsyncMock(return_value=(b"", b""))
        return proc

    with patch("asyncio.create_subprocess_exec", side_effect=fake_subprocess):
        await git_pull(source)

    assert "tok" not in captured["content"]
    assert "oauth2" in captured["content"]
    assert captured["executable"] is True
    assert captured["token"] == 'tok"$(id)`x`'


def test_askpass_prints_a_hostile_token_as_data(tmp_path):
    """Run the real script: shell syntax in a token comes back verbatim."""
    import subprocess
    from steward.ansible.sources import ASKPASS_SCRIPT, ASKPASS_TOKEN_ENV

    script = tmp_path / "askpass.sh"
    script.write_text(ASKPASS_SCRIPT)
    script.chmod(0o700)
    token = 'a"b$(touch pwned)`id` c\\'
    env = {"PATH": os.environ.get("PATH", ""), ASKPASS_TOKEN_ENV: token}
    out = subprocess.run([str(script), "Password for 'https://x'"], env=env, cwd=tmp_path,
                         capture_output=True, text=True, check=True).stdout
    assert out == token + "\n"
    assert not (tmp_path / "pwned").exists()
    user = subprocess.run([str(script), "Username for 'https://x'"], env=env,
                          capture_output=True, text=True, check=True).stdout
    assert user == "oauth2\n"


def test_sealed_tokens_open_in_get_sources(tmp_path, monkeypatch):
    from steward.ansible.sources import get_sources
    from steward.core import crypto
    from steward.core.settings import seal_source_tokens

    monkeypatch.setattr(crypto, "_fernet", crypto._make_fernet("unit-test-key"))
    stored, count = seal_source_tokens([
        {"name": "repo", "type": "git", "url": "https://x/r.git", "http_token": "mytoken123"},
        {"name": "plain", "type": "git", "url": "https://x/p.git"},
    ])
    assert count == 1 and crypto.is_encrypted(stored[0]["http_token"])
    assert seal_source_tokens(stored) == (stored, 0)  # sealing twice changes nothing
    src = next(s for s in get_sources({"cache_dir": str(tmp_path), "sources": stored})
               if s["name"] == "repo")
    assert src["http_token"] == "mytoken123"


@pytest.mark.asyncio
async def test_git_pull_no_askpass_without_token(tmp_path):
    """git_pull() does NOT set GIT_ASKPASS when http_token is empty."""
    from steward.ansible.sources import git_pull

    repo_path = tmp_path / "repo"
    (repo_path / ".git").mkdir(parents=True)

    source = {
        "name": "repo",
        "path": str(repo_path),
        "url": "https://git.example.com/org/repo.git",
        "branch": "main",
        "http_token": "",
    }

    captured_env = {}

    async def fake_subprocess(*args, **kwargs):
        captured_env.update(kwargs.get("env") or {})
        proc = MagicMock()
        proc.returncode = 0
        proc.communicate = AsyncMock(return_value=(b"", b""))
        return proc

    with patch("asyncio.create_subprocess_exec", side_effect=fake_subprocess):
        await git_pull(source)

    assert "GIT_ASKPASS" not in captured_env