OSVauco/opax-mcp/test_mcp_tools.py

import os
import re
import pytest
import subprocess
import tempfile
from pathlib import Path
from unittest.mock import patch, MagicMock, ANY, AsyncMock

import json
from httpx import ASGITransport, AsyncClient

# Importer alt som skal testes
from server import app, TOOLS, WEB_AGENT_ALLOWED_TOOLS


# --- Fixtures ---

@pytest.fixture
def mock_repo_root(monkeypatch):
    """Setter REPO_ROOT til en trygg, midlertidig mappe for unit-tester."""
    with tempfile.TemporaryDirectory() as tmpdir:
        monkeypatch.setattr("server.REPO_ROOT", Path(tmpdir))
        yield Path(tmpdir)

@pytest.fixture
def isolated_git_repo(monkeypatch):
    """Setter opp et isolert Git-miljø og patcher REPO_ROOT for integrasjonstester."""
    with tempfile.TemporaryDirectory() as tmpdir_str:
        tmpdir = Path(tmpdir_str)
        remote_repo_path = tmpdir / "remote.git"
        local_repo_path = tmpdir / "local"

        subprocess.run(["git", "init", "--bare", str(remote_repo_path)], check=True)
        subprocess.run(["git", "clone", str(remote_repo_path), str(local_repo_path)], check=True)
        subprocess.run(["git", "config", "user.name", "Test User"], cwd=local_repo_path, check=True)
        subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=local_repo_path, check=True)
        # Add an initial commit so the repo is not empty
        (local_repo_path / ".gitkeep").touch()
        subprocess.run(["git", "add", ".gitkeep"], cwd=local_repo_path, check=True)
        subprocess.run(["git", "commit", "-m", "Initial commit"], cwd=local_repo_path, check=True)
        branch_name = subprocess.run(["git", "rev-parse", "--abbrev-ref", "HEAD"], cwd=local_repo_path, check=True, capture_output=True, text=True).stdout.strip()
        subprocess.run(["git", "push", "origin", branch_name], cwd=local_repo_path, check=True)

        monkeypatch.setattr("server.REPO_ROOT", local_repo_path)
        yield local_repo_path, remote_repo_path

@pytest.fixture
def mcp_secret(monkeypatch):
    """Sets the MCP_SECRET environment variable for testing."""
    secret = "test-secret-for-dispatch"
    monkeypatch.setenv("MCP_SECRET", secret)
    yield secret

# --- Unit-tester for Hjelpefunksjoner ---

def test_validate_path_inside_repo_success(mock_repo_root):
    (mock_repo_root / "testdir").mkdir()
    (mock_repo_root / "testdir" / "test.txt").touch()
    assert _validate_path_inside_repo("testdir/test.txt") == (mock_repo_root / "testdir" / "test.txt").resolve()

@pytest.mark.parametrize(
    "invalid_path, match_error",
    [
        (None, "Path must be a non-empty string."),
        (123, "Path must be a non-empty string."),
        ("", "Path must be a non-empty string."),
        ("   ", "Path must be a non-empty string."),
        ('a\x00b', "Path cannot contain control characters."),
        ('a\rb', "Path cannot contain control characters."),
        ('a\nb', "Path cannot contain control characters."),
        ('a\\b', "Path cannot contain backslashes."),
        ("/etc/passwd", "Path cannot be absolute."),
        ("C:boot.ini", "Path cannot be absolute."),
        ("//server/share", "Path cannot be absolute."),
        ("./a", "Path contains a disallowed segment"),
        ("a/./b", "Path contains a disallowed segment"),
        ("../a", "Path contains a disallowed segment"),
        ("a/../b", "Path contains a disallowed segment"),
        (".git/config", "Path contains a disallowed segment"),
        ("a/.git/b", "Path contains a disallowed segment"),
    ],
    ids=[
        "non-string", "non-string-int", "empty", "whitespace",
        "nul-char", "cr-char", "lf-char", "backslash",
        "absolute-unix", "windows-drive-prefix", "unc",
        "dot-segment", "dot-segment-internal", "traversal", "traversal-internal",
        "dot-git-segment", "dot-git-segment-internal",
    ]
)
def test_validate_path_rejects_invalid_inputs(mock_repo_root, invalid_path, match_error):
    with pytest.raises(ValueError, match=re.escape(match_error)):
        _validate_path_inside_repo(invalid_path)


def test_validate_path_rejects_symlink_escape(mock_repo_root):
    """Tests that a symlink pointing outside the repository root is rejected."""
    outside_file = mock_repo_root.parent / "sensitive_file.txt"
    outside_file.write_text("secret")

    symlink_in_repo = mock_repo_root / "link_to_secret"

    try:
        os.symlink(outside_file, symlink_in_repo)
    except (OSError, NotImplementedError) as e:
        pytest.skip(f"Symlink creation skipped: {e}")

    with pytest.raises(ValueError, match=re.escape("Path resolves outside the repository root.")):
        _validate_path_inside_repo("link_to_secret")

@pytest.mark.parametrize("url, expected", [
    ("https://git.vauco.no/chris/OSVauco.git", True),
    ("git@git.vauco.no:chris/OSVauco.git", True),
    ("https://github.com/evil/repo.git", False),
])
def test_is_allowed_remote(url, expected):
    assert _is_allowed_remote(url) is expected

def test_mask_secrets(mock_repo_root):
    from server import SECRET_MASK_PATTERNS, _mask_secrets
    test_secret = "TEST_SECRET_VALUE_123"
    test_pattern = re.compile(re.escape(test_secret))
    SECRET_MASK_PATTERNS.append(test_pattern)
    try:
        text = f"Token: gitea_abc123def456, Secret: {test_secret}"
        masked = _mask_secrets(text)
        assert "[MASKED_TOKEN]" in masked
        assert "[MASKED_SECRET]" in masked
        assert test_secret not in masked
    finally:
        SECRET_MASK_PATTERNS.remove(test_pattern)

@patch("server.subprocess.run")
def test_run_git_success(mock_run, mock_repo_root):
    mock_run.return_value = MagicMock(stdout="OK", stderr="", returncode=0)
    _run_git(["status", "--short"], timeout=30)
    mock_run.assert_called_once_with(
        ["git", "status", "--short"],
        cwd=mock_repo_root,
        capture_output=True, text=True, timeout=30, check=True
    )

def test_run_git_invalid_args():
    with pytest.raises(ValueError):
        _run_git(["status\nls"], timeout=30)

# --- Unit-tester for Read-only Verktøy ---


# --- Tests for MCP Handler Dispatch Policy ---

@pytest.mark.asyncio
class TestMcpHandlerDispatch:

    async def make_rpc_call(self, tool_name, token, params=None):
        headers = {"Authorization": f"Bearer {token}"}
        request_body = {
            "jsonrpc": "2.0",
            "id": "test-id-123",
            "method": "tools/call",
            "params": {"name": tool_name, "arguments": params or {}}
        }
        transport = ASGITransport(app=app)
        async with AsyncClient(transport=transport, base_url="http://testserver") as client:
            response = await client.post("/", json=request_body, headers=headers)
        return response

    @pytest.mark.parametrize(
        "blocked_tool",
        [
            "commit_and_push_files",
            "send_email",
            "trigger_build",
            "set_billing_budget",
            "create_email_alias",
            "get_emails",
            "preview_git_change",
        ]
    )
    async def test_rejects_existing_but_blocked_tools(self, monkeypatch, mcp_secret, blocked_tool):
        assert blocked_tool in TOOLS
        assert blocked_tool not in WEB_AGENT_ALLOWED_TOOLS

        mock_tool_func = AsyncMock()
        _, description, schema = TOOLS[blocked_tool]
        monkeypatch.setitem(TOOLS, blocked_tool, (mock_tool_func, description, schema))

        response = await self.make_rpc_call(blocked_tool, mcp_secret)

        assert response.status_code == 200
        json_res = response.json()
        assert json_res["error"]["code"] == -32601
        assert json_res["error"]["message"] == "Tool not allowed or not found."
        mock_tool_func.assert_not_awaited()

    async def test_rejects_unknown_tool(self, mcp_secret):
        response = await self.make_rpc_call("non_existent_tool", mcp_secret)

        assert response.status_code == 200
        json_res = response.json()
        assert json_res["error"]["code"] == -32601
        assert json_res["error"]["message"] == "Tool not allowed or not found."

    async def test_rejects_allowed_tool_with_malicious_path_traversal(self, mcp_secret):
        """
        Confirms that an ALLOWED tool (read_repo_file) still rejects a request
        with a malicious path traversal argument at the handler level. This tests
        that the tool's internal validation is triggered correctly.
        """
        response = await self.make_rpc_call(
            "read_repo_file",
            mcp_secret,
            params={"path": "../../../etc/passwd"}
        )

        assert response.status_code == 200
        json_res = response.json()
        assert "result" not in json_res
        assert "error" in json_res, "The response should contain a JSON-RPC error object"
        assert json_res["error"]["code"] == -32000, "Expected a generic server error for a validation failure"

    @pytest.mark.parametrize(
        "allowed_tool", sorted(WEB_AGENT_ALLOWED_TOOLS)
    )
    async def test_allows_and_dispatches_web_agent_tools(self, monkeypatch, mcp_secret, allowed_tool):
        mock_tool_func = AsyncMock(return_value={"status": "mocked_success"})
        _, description, schema = TOOLS[allowed_tool]
        monkeypatch.setitem(TOOLS, allowed_tool, (mock_tool_func, description, schema))

        test_args = {"param": "value"}
        response = await self.make_rpc_call(allowed_tool, mcp_secret, params=test_args)

        assert response.status_code == 200
        json_res = response.json()
        assert "result" in json_res
        inner_result = json.loads(json_res["result"]["content"][0]["text"])
        assert inner_result["status"] == "mocked_success"

        mock_tool_func.assert_awaited_once_with(test_args)

    async def test_auth_is_checked_before_policy(self, monkeypatch, mcp_secret):
        mock_tool_func = AsyncMock()
        allowed_tool = "get_health"
        monkeypatch.setitem(TOOLS, allowed_tool, (mock_tool_func, "", {}))

        response = await self.make_rpc_call(allowed_tool, "wrong-secret")

        assert response.status_code == 401
        mock_tool_func.assert_not_awaited()