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()