security(owners): enforce OWNERS files read from PR base branch using worktrees

Critical security fix to prevent PRs from modifying their own approval requirements.

Previously, OWNERS files were read from the main repository clone, which could be
influenced by PR content. This created a security vulnerability where a malicious
PR could:
- Modify OWNERS files in changed directories
- Change who can approve the PR
- Potentially bypass approval requirements

Solution:
- Use git worktree to create isolated checkout of PR's base branch (target branch)
- Always read OWNERS files from base branch, not PR head branch
- Ensures approval requirements cannot be modified by PR content
- Maintains performance by using existing worktree helper

Changes:
- Modified get_all_repository_approvers_and_reviewers() to accept branch parameter
- Integrated git_worktree_checkout for isolated base branch access
- Updated _get_file_content_from_local() to support custom base paths
- Enhanced file discovery to skip hidden directories in worktree
- Updated all tests to mock worktree creation

Impact:
- Security: PRs cannot modify their own approval requirements
- Performance: Minimal overhead (worktree creation is fast)
- Reliability: Fail-fast on worktree creation failures

Author: myakove

webhook_server/libs/handlers/owners_files_handler.py

@@ -12,6 +12,7 @@
 from github.PullRequest import PullRequest
 from github.Repository import Repository
 
+from webhook_server.utils import helpers as helpers_module
 from webhook_server.utils.constants import COMMAND_ADD_ALLOWED_USER_STR, ROOT_APPROVERS_KEY
 from webhook_server.utils.helpers import format_task_fields
 
@@ -32,10 +33,13 @@ async def initialize(self, pull_request: PullRequest) -> "OwnersFileHandler":
         Phase 1: Fetch independent data in parallel (changed files + OWNERS data)
         Phase 2: Process derived data in parallel (approvers + reviewers)
         """
+        # Extract base branch for OWNERS files (PR target branch)
+        base_branch = await asyncio.to_thread(lambda: pull_request.base.ref)
+
         # Phase 1: Parallel data fetching - independent GitHub API operations
         self.changed_files, self.all_repository_approvers_and_reviewers = await asyncio.gather(
             self.list_changed_files(pull_request=pull_request),
-            self.get_all_repository_approvers_and_reviewers(pull_request=pull_request),
+            self.get_all_repository_approvers_and_reviewers(branch=base_branch),
         )
 
         # Phase 2: Parallel data processing - all depend on phase 1 but independent of each other
@@ -106,16 +110,20 @@ def _validate_owners_content(self, content: Any, path: str) -> bool:
             self.logger.error(f"{self.log_prefix} Invalid OWNERS file {path}: {e}")
             return False
 
-    async def _get_file_content_from_local(self, content_path: Path) -> tuple[str, str] | None:
+    async def _get_file_content_from_local(
+        self, content_path: Path, base_path: Path | None = None
+    ) -> tuple[str, str] | None:
         """Read OWNERS file from local cloned repository.
 
         Args:
             content_path: Path object pointing to OWNERS file in clone_repo_dir
+            base_path: Base path to compute relative path from (defaults to clone_repo_dir)
 
         Returns:
             Tuple of (file_content, relative_path_str) or None if file is unreadable
         """
-        relative_path = content_path.relative_to(Path(self.github_webhook.clone_repo_dir))
+        _base_path = base_path if base_path else Path(self.github_webhook.clone_repo_dir)
+        relative_path = content_path.relative_to(_base_path)
         self.logger.debug(f"{self.log_prefix} Reading OWNERS file from local clone: {relative_path}")
 
         try:
@@ -137,62 +145,85 @@ async def _get_file_content_from_local(self, content_path: Path) -> tuple[str, s
             )
             return None
 
-    async def get_all_repository_approvers_and_reviewers(self, pull_request: PullRequest) -> dict[str, dict[str, Any]]:
+    async def get_all_repository_approvers_and_reviewers(self, branch: str) -> dict[str, dict[str, Any]]:
         """Get all repository approvers and reviewers from OWNERS files.
 
-        Reads OWNERS files from local cloned repository instead of GitHub API.
-        Saves ~1,200-2,000 API calls per webhook.
-        """
-        # Ensure repository is cloned first
-        await self.github_webhook._clone_repository_for_pr(pull_request)
+        Reads OWNERS files from local cloned repository at specified branch using worktree.
+
+        Args:
+            branch: Branch name to read OWNERS files from (e.g., 'main', 'develop')
 
+        Returns:
+            Dictionary mapping OWNERS file paths to their approvers and reviewers
+        """
         # Dictionary mapping OWNERS file paths to their approvers and reviewers
         _owners: dict[str, dict[str, Any]] = {}
         tasks: list[Coroutine[Any, Any, Any]] = []
 
         max_owners_files = 1000  # Intentionally hardcoded limit to prevent runaway processing
         owners_count = 0
 
-        # Find all OWNERS files via filesystem walk
-        self.logger.debug(f"{self.log_prefix} Finding OWNERS files in local clone")
-        clone_path = Path(self.github_webhook.clone_repo_dir)
+        # Use existing worktree helper to create isolated checkout of specified branch
+        async with helpers_module.git_worktree_checkout(
+            repo_dir=self.github_webhook.clone_repo_dir,
+            checkout=branch,
+            log_prefix=self.log_prefix,
+            mask_sensitive=self.github_webhook.mask_sensitive,
+        ) as (success, worktree_path, _stdout, stderr):
+            if not success:
+                self.logger.error(f"{self.log_prefix} Failed to create worktree for branch {branch}: {stderr}")
+                raise RuntimeError(f"Failed to create worktree for branch {branch}")
+
+            self.logger.debug(
+                f"{self.log_prefix} Reading OWNERS files from branch '{branch}' using worktree at {worktree_path}"
+            )
 
-        # Use rglob to recursively find all OWNERS files
-        def find_owners_files() -> list[Path]:
-            return list(clone_path.rglob("OWNERS"))
+            # Read OWNERS files from worktree (not main clone)
+            clone_path = Path(worktree_path)
 
-        owners_files = await asyncio.to_thread(find_owners_files)
+            # Find all OWNERS files via filesystem walk
+            self.logger.debug(f"{self.log_prefix} Finding OWNERS files in worktree")
 
-        for owners_file_path in owners_files:
-            owners_count += 1
-            if owners_count > max_owners_files:
-                self.logger.error(f"{self.log_prefix} Too many OWNERS files (>{max_owners_files})")
-                break
+            # Use rglob to recursively find all OWNERS files
+            def find_owners_files() -> list[Path]:
+                return [
+                    p
+                    for p in clone_path.rglob("OWNERS")
+                    if not any(part.startswith(".") for part in p.relative_to(clone_path).parts)
+                ]
 
-            relative_path = owners_file_path.relative_to(clone_path)
-            self.logger.debug(f"{self.log_prefix} Found OWNERS file: {relative_path}")
-            tasks.append(self._get_file_content_from_local(owners_file_path))
+            owners_files = await asyncio.to_thread(find_owners_files)
 
-        results = await asyncio.gather(*tasks)
+            for owners_file_path in owners_files:
+                owners_count += 1
+                if owners_count > max_owners_files:
+                    self.logger.error(f"{self.log_prefix} Too many OWNERS files (>{max_owners_files})")
+                    break
 
-        for result in results:
-            # Skip files that couldn't be read (deleted or unreadable)
-            if result is None:
-                continue
+                relative_path = owners_file_path.relative_to(clone_path)
+                self.logger.debug(f"{self.log_prefix} Found OWNERS file: {relative_path}")
+                tasks.append(self._get_file_content_from_local(owners_file_path, base_path=clone_path))
 
-            file_content, relative_path_str = result
+            results = await asyncio.gather(*tasks)
 
-            try:
-                content = yaml.safe_load(file_content)
-                if self._validate_owners_content(content, relative_path_str):
-                    parent_path = str(Path(relative_path_str).parent)
-                    if not parent_path or parent_path == ".":
-                        parent_path = "."
-                    _owners[parent_path] = content
+            for result in results:
+                # Skip files that couldn't be read (deleted or unreadable)
+                if result is None:
+                    continue
 
-            except yaml.YAMLError:
-                self.logger.exception(f"{self.log_prefix} Invalid OWNERS file {relative_path_str}")
-                continue
+                file_content, relative_path_str = result
+
+                try:
+                    content = yaml.safe_load(file_content)
+                    if self._validate_owners_content(content, relative_path_str):
+                        parent_path = str(Path(relative_path_str).parent)
+                        if not parent_path or parent_path == ".":
+                            parent_path = "."
+                        _owners[parent_path] = content
+
+                except yaml.YAMLError:
+                    self.logger.exception(f"{self.log_prefix} Invalid OWNERS file {relative_path_str}")
+                    continue
 
         return _owners
 

webhook_server/tests/test_github_api.py

@@ -1,6 +1,8 @@
 import asyncio
 import os
 import tempfile
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
 from typing import Any
 from unittest.mock import AsyncMock, Mock, patch
 
@@ -74,6 +76,21 @@ def minimal_headers(self) -> dict[str, str]:
     def logger(self):
         return get_logger(name="test")
 
+    @pytest.fixture
+    def mock_git_worktree(self, tmp_path):
+        """Mock git_worktree_checkout context manager."""
+
+        @asynccontextmanager
+        async def mock_worktree(
+            repo_dir: str, checkout: str, log_prefix: str, mask_sensitive: bool = True
+        ) -> AsyncGenerator[tuple[bool, str, str, str], None]:
+            # Create temp directory to simulate worktree
+            worktree_path = str(tmp_path / "worktree")
+            os.makedirs(worktree_path, exist_ok=True)
+            yield (True, worktree_path, "", "")  # success, path, stdout, stderr
+
+        return mock_worktree
+
     @patch("webhook_server.libs.github_api.Config")
     @patch("webhook_server.libs.github_api.get_api_with_highest_rate_limit")
     @patch("webhook_server.libs.github_api.get_github_repo_api")
@@ -218,6 +235,7 @@ async def test_process_pull_request_event(
         mock_repo_api: Mock,
         pull_request_payload: dict[str, Any],
         webhook_headers: Headers,
+        mock_git_worktree,
     ) -> None:
         """Test processing pull_request event."""
         # Mock GitHub API to prevent network calls
@@ -262,6 +280,10 @@ async def test_process_pull_request_event(
                 return_value=Mock(decoded_content=b"approvers:\n  - user1\nreviewers:\n  - user2"),
             ),
             patch.object(webhook, "_clone_repository_for_pr", new=AsyncMock(return_value=None)),
+            patch(
+                "webhook_server.libs.handlers.owners_files_handler.helpers_module.git_worktree_checkout",
+                new=mock_git_worktree,
+            ),
         ):
             await webhook.process()
             mock_process_pr.assert_called_once()
@@ -319,6 +341,7 @@ async def test_process_issue_comment_event(
         mock_api_rate_limit: Mock,
         mock_repo_api: Mock,
         issue_comment_payload: dict[str, Any],
+        mock_git_worktree,
     ) -> None:
         """Test processing issue_comment event."""
         # Mock GitHub API to prevent network calls
@@ -364,6 +387,10 @@ async def test_process_issue_comment_event(
                 return_value=Mock(decoded_content=b"approvers:\n  - user1\nreviewers:\n  - user2"),
             ),
             patch.object(webhook, "_clone_repository_for_pr", new=AsyncMock(return_value=None)),
+            patch(
+                "webhook_server.libs.handlers.owners_files_handler.helpers_module.git_worktree_checkout",
+                new=mock_git_worktree,
+            ),
         ):
             await webhook.process()
             mock_process_comment.assert_called_once()
@@ -688,7 +715,9 @@ def test_prepare_log_prefix_with_color_file(
             assert result2 is not None
 
     @pytest.mark.asyncio
-    async def test_process_check_run_event(self, minimal_hook_data: dict, minimal_headers: dict, logger: Mock) -> None:
+    async def test_process_check_run_event(
+        self, minimal_hook_data: dict, minimal_headers: dict, logger: Mock, mock_git_worktree
+    ) -> None:
         """Test processing check run event."""
         check_run_data = {
             "repository": {"name": "test-repo", "full_name": "org/test-repo"},
@@ -748,8 +777,14 @@ async def test_process_check_run_event(self, minimal_hook_data: dict, minimal_he
                                     mock_pr_handler.return_value.check_if_can_be_merged = AsyncMock(return_value=None)
 
                                     webhook = GithubWebhook(check_run_data, headers, logger)
-                                    with patch.object(
-                                        webhook, "_clone_repository_for_pr", new=AsyncMock(return_value=None)
+                                    with (
+                                        patch.object(
+                                            webhook, "_clone_repository_for_pr", new=AsyncMock(return_value=None)
+                                        ),
+                                        patch(
+                                            "webhook_server.libs.handlers.owners_files_handler.helpers_module.git_worktree_checkout",
+                                            new=mock_git_worktree,
+                                        ),
                                     ):
                                         await webhook.process()
 

webhook_server/tests/test_owners_files_handler.py

@@ -1,3 +1,5 @@
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
 from pathlib import Path
 from unittest.mock import AsyncMock, Mock, call, patch
 
@@ -184,9 +186,17 @@ async def test_get_all_repository_approvers_and_reviewers(
         # Set clone_repo_dir to tmp_path
         owners_file_handler.github_webhook.clone_repo_dir = str(tmp_path)
 
-        # Mock _clone_repository_for_pr to do nothing (already "cloned" to tmp_path)
-        with patch.object(owners_file_handler.github_webhook, "_clone_repository_for_pr", new=AsyncMock()):
-            result = await owners_file_handler.get_all_repository_approvers_and_reviewers(mock_pull_request)
+        # Mock git_worktree_checkout to use tmp_path directly (simulate successful worktree)
+        @asynccontextmanager
+        async def mock_worktree(
+            repo_dir: str, checkout: str, log_prefix: str, mask_sensitive: bool = True
+        ) -> AsyncGenerator[tuple[bool, str, str, str], None]:
+            yield (True, str(tmp_path), "", "")
+
+        with patch(
+            "webhook_server.libs.handlers.owners_files_handler.helpers_module.git_worktree_checkout", mock_worktree
+        ):
+            result = await owners_file_handler.get_all_repository_approvers_and_reviewers(branch="main")
 
         expected = {
             ".": {"approvers": ["root_approver1", "root_approver2"], "reviewers": ["root_reviewer1", "root_reviewer2"]},
@@ -222,9 +232,17 @@ async def test_get_all_repository_approvers_and_reviewers_too_many_files(
         owners_file_handler.github_webhook.clone_repo_dir = str(tmp_path)
         owners_file_handler.logger.error = Mock()
 
-        # Mock _clone_repository_for_pr
-        with patch.object(owners_file_handler.github_webhook, "_clone_repository_for_pr", new=AsyncMock()):
-            result = await owners_file_handler.get_all_repository_approvers_and_reviewers(mock_pull_request)
+        # Mock git_worktree_checkout to use tmp_path directly
+        @asynccontextmanager
+        async def mock_worktree(
+            repo_dir: str, checkout: str, log_prefix: str, mask_sensitive: bool = True
+        ) -> AsyncGenerator[tuple[bool, str, str, str], None]:
+            yield (True, str(tmp_path), "", "")
+
+        with patch(
+            "webhook_server.libs.handlers.owners_files_handler.helpers_module.git_worktree_checkout", mock_worktree
+        ):
+            result = await owners_file_handler.get_all_repository_approvers_and_reviewers(branch="main")
 
         assert len(result) == 1000
         owners_file_handler.logger.error.assert_called_once()
@@ -242,9 +260,17 @@ async def test_get_all_repository_approvers_and_reviewers_invalid_yaml(
         owners_file_handler.github_webhook.clone_repo_dir = str(tmp_path)
         owners_file_handler.logger.exception = Mock()
 
-        # Mock _clone_repository_for_pr
-        with patch.object(owners_file_handler.github_webhook, "_clone_repository_for_pr", new=AsyncMock()):
-            result = await owners_file_handler.get_all_repository_approvers_and_reviewers(mock_pull_request)
+        # Mock git_worktree_checkout to use tmp_path directly
+        @asynccontextmanager
+        async def mock_worktree(
+            repo_dir: str, checkout: str, log_prefix: str, mask_sensitive: bool = True
+        ) -> AsyncGenerator[tuple[bool, str, str, str], None]:
+            yield (True, str(tmp_path), "", "")
+
+        with patch(
+            "webhook_server.libs.handlers.owners_files_handler.helpers_module.git_worktree_checkout", mock_worktree
+        ):
+            result = await owners_file_handler.get_all_repository_approvers_and_reviewers(branch="main")
 
         assert result == {}
         owners_file_handler.logger.exception.assert_called_once()
@@ -262,9 +288,17 @@ async def test_get_all_repository_approvers_and_reviewers_invalid_content(
         owners_file_handler.github_webhook.clone_repo_dir = str(tmp_path)
         owners_file_handler.logger.error = Mock()
 
-        # Mock _clone_repository_for_pr
-        with patch.object(owners_file_handler.github_webhook, "_clone_repository_for_pr", new=AsyncMock()):
-            result = await owners_file_handler.get_all_repository_approvers_and_reviewers(mock_pull_request)
+        # Mock git_worktree_checkout to use tmp_path directly
+        @asynccontextmanager
+        async def mock_worktree(
+            repo_dir: str, checkout: str, log_prefix: str, mask_sensitive: bool = True
+        ) -> AsyncGenerator[tuple[bool, str, str, str], None]:
+            yield (True, str(tmp_path), "", "")
+
+        with patch(
+            "webhook_server.libs.handlers.owners_files_handler.helpers_module.git_worktree_checkout", mock_worktree
+        ):
+            result = await owners_file_handler.get_all_repository_approvers_and_reviewers(branch="main")
 
         assert result == {}
         owners_file_handler.logger.error.assert_called_once()

webhook_server/tests/test_pull_request_owners.py

@@ -1,3 +1,5 @@
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
 from unittest.mock import AsyncMock, patch
 
 import pytest
@@ -144,11 +146,19 @@ async def test_get_all_repository_approvers_and_reviewers(
     # Set clone_repo_dir to tmp_path
     process_github_webhook.clone_repo_dir = str(tmp_path)
 
+    # Mock git_worktree_checkout to use tmp_path directly (simulate successful worktree)
+    @asynccontextmanager
+    async def mock_worktree(
+        repo_dir: str, checkout: str, log_prefix: str, mask_sensitive: bool = True
+    ) -> AsyncGenerator[tuple[bool, str, str, str], None]:
+        yield (True, str(tmp_path), "", "")
+
     # Mock _clone_repository_for_pr to do nothing (already "cloned" to tmp_path)
-    with patch.object(process_github_webhook, "_clone_repository_for_pr", new=AsyncMock()):
-        read_owners_result = await owners_file_handler.get_all_repository_approvers_and_reviewers(
-            pull_request=pull_request
-        )
+    with (
+        patch.object(process_github_webhook, "_clone_repository_for_pr", new=AsyncMock()),
+        patch("webhook_server.libs.handlers.owners_files_handler.helpers_module.git_worktree_checkout", mock_worktree),
+    ):
+        read_owners_result = await owners_file_handler.get_all_repository_approvers_and_reviewers(branch="main")
 
     assert read_owners_result == owners_file_handler.all_repository_approvers_and_reviewers