diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 0bc82e4..beb6084 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -440,10 +440,48 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)" ) - if worktree_path: - dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path))) + workspace = author_mutation_worktree.resolve_mutation_workspace( + worktree_path, + PROJECT_ROOT, + active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV), + author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV), + ) + real_workspace = os.path.realpath(workspace) + real_root = os.path.realpath(PROJECT_ROOT) + + if real_workspace != real_root: + if not _preflight_in_test_mode(): + if not os.path.exists(real_workspace): + raise RuntimeError( + f"Branches-only mutation guard (#274): worktree path '{workspace}' does not exist (fail closed)" + ) + if not os.path.isdir(real_workspace): + raise RuntimeError( + f"Branches-only mutation guard (#274): worktree path '{workspace}' is not a directory (fail closed)" + ) + try: + res = subprocess.run( + ["git", "-C", real_workspace, "rev-parse", "--git-common-dir"], + capture_output=True, + text=True, + check=True, + ) + common_dir = os.path.realpath(res.stdout.strip()) + expected_dir = os.path.realpath(os.path.join(real_root, ".git")) + if common_dir != expected_dir: + raise RuntimeError( + f"Branches-only mutation guard (#274): worktree '{workspace}' does not belong to the target repository '{PROJECT_ROOT}' (fail closed)" + ) + except Exception as e: + if isinstance(e, RuntimeError): + raise e + raise RuntimeError( + f"Branches-only mutation guard (#274): worktree '{workspace}' is not a valid git repository (fail closed)" + ) + + dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace))) if dirty_files: - details = _preflight_workspace_details(worktree_path, dirty_files) + details = _preflight_workspace_details(workspace, dirty_files) raise RuntimeError( "Pre-flight order violation: Active task workspace has tracked " "file edits before mutation (fail closed). " @@ -972,6 +1010,7 @@ def gitea_create_issue( repo: str | None = None, allow_duplicate_override: bool = False, split_from_issue: int | None = None, + worktree_path: str | None = None, ) -> dict: """Create a new issue on a Gitea repository. @@ -984,6 +1023,7 @@ def gitea_create_issue( repo: Override the repository name. allow_duplicate_override: Operator-approved split after duplicate found. split_from_issue: Existing duplicate issue number when overriding. + worktree_path: Optional path to verify branches-only guard. Returns: dict with 'number' of the created issue ('url' only with the reveal opt-in). @@ -1010,7 +1050,7 @@ def gitea_create_issue( ) if blocked: return blocked - verify_preflight_purity(remote) + verify_preflight_purity(remote, worktree_path=worktree_path) base = repo_api_url(h, o, r) open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth) closed_issues = api_get_all( diff --git a/tests/test_create_issue_workspace_guard.py b/tests/test_create_issue_workspace_guard.py new file mode 100644 index 0000000..2cd9f41 --- /dev/null +++ b/tests/test_create_issue_workspace_guard.py @@ -0,0 +1,145 @@ +import os +import sys +import unittest +from pathlib import Path +from unittest.mock import patch, MagicMock + +# Ensure we import from the repo root +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import gitea_mcp_server as srv + +FAKE_AUTH = {"Authorization": "token test-token"} +# Stable control checkout (parent of branches/), not the MCP server worktree root. +CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3]) +PROJECT_ROOT = srv.PROJECT_ROOT + + +class TestCreateIssueWorkspaceGuard(unittest.TestCase): + + def setUp(self): + # Reset preflight flags + srv._preflight_whoami_called = True + srv._preflight_capability_called = True + srv._preflight_resolved_role = "author" + srv._preflight_whoami_violation = False + srv._preflight_capability_violation = False + + # Disable early return in verify_preflight_purity for testing + self._orig_in_test = srv._preflight_in_test_mode + srv._preflight_in_test_mode = lambda: False + + def tearDown(self): + srv._preflight_in_test_mode = self._orig_in_test + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._profile_permission_block", return_value=None) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server.api_get_all", return_value=[]) + @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"}) + def test_create_issue_stable_checkout_rejected(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth): + # Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that + # path is the stable control checkout (not under branches/), mutation must fail. + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with self.assertRaises(RuntimeError) as ctx: + srv.gitea_create_issue(title="Test issue", body="body text") + self.assertIn("stable control checkout", str(ctx.exception)) + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._profile_permission_block", return_value=None) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server.api_get_all", return_value=[]) + @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"}) + @patch("os.path.exists", return_value=True) + @patch("os.path.isdir", return_value=True) + @patch("subprocess.run") + def test_create_issue_valid_worktree_succeeds(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth): + # Mock subprocess.run for git --git-common-dir to return PROJECT_ROOT/.git + mock_res = MagicMock() + mock_res.stdout = f"{CONTROL_CHECKOUT_ROOT}/.git\n" + mock_run.return_value = mock_res + + mock_api.return_value = {"number": 42, "html_url": "https://gitea.example.com/issues/42"} + + # Provide a valid branches path under the control checkout root + valid_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1") + + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""): + res = srv.gitea_create_issue( + title="Test issue", body="body", worktree_path=valid_path + ) + + self.assertEqual(res["number"], 42) + mock_api.assert_called_once() + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._profile_permission_block", return_value=None) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server.api_get_all", return_value=[]) + @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"}) + def test_create_issue_missing_worktree_fails_closed(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth): + # Path under branches/ but doesn't exist + missing_path = os.path.join( + CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-worktree-path-999" + ) + + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with self.assertRaises(RuntimeError) as ctx: + srv.gitea_create_issue( + title="Test issue", body="body", worktree_path=missing_path + ) + self.assertIn("does not exist (fail closed)", str(ctx.exception)) + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._profile_permission_block", return_value=None) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server.api_get_all", return_value=[]) + @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"}) + @patch("os.path.exists", return_value=True) + @patch("os.path.isdir", return_value=True) + @patch("subprocess.run") + def test_create_issue_wrong_repo_fails_closed(self, mock_run, mock_isdir, mock_exists, _git, _get_all, mock_api, _role, _ns, _prof, _auth): + # Mock subprocess.run for git --git-common-dir to return a different path + mock_res = MagicMock() + mock_res.stdout = "/Users/jasonwalker/Development/some-other-repo/.git\n" + mock_run.return_value = mock_res + + wrong_repo_path = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "feat-issue-1") + + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with self.assertRaises(RuntimeError) as ctx: + srv.gitea_create_issue( + title="Test issue", body="body", worktree_path=wrong_repo_path + ) + self.assertIn("does not belong to the target repository", str(ctx.exception)) + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._profile_permission_block", return_value=None) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + def test_create_issue_fails_without_whoami_preflight(self, _ns, _prof, _auth): + srv._preflight_whoami_called = False + with self.assertRaises(RuntimeError) as ctx: + srv.gitea_create_issue(title="Test issue", body="body") + self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception)) + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._profile_permission_block", return_value=None) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + def test_create_issue_fails_without_capability_preflight(self, _ns, _prof, _auth): + srv._preflight_capability_called = False + with self.assertRaises(RuntimeError) as ctx: + srv.gitea_create_issue(title="Test issue", body="body") + self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main()