Integrate permission checking into mention decorator

Author: joshuadavidthomas

src/django_github_app/permissions.py

@@ -5,17 +5,12 @@
 
 import cachetools
 import gidgethub
+from gidgethub import sansio
 
 from django_github_app.github import AsyncGitHubAPI
 from django_github_app.github import SyncGitHubAPI
 
 
-class PermissionCacheKey(NamedTuple):
-    owner: str
-    repo: str
-    username: str
-
-
 class Permission(int, Enum):
     NONE = 0
     READ = 1
@@ -47,6 +42,12 @@ def from_string(cls, permission: str) -> Permission:
 )
 
 
+class PermissionCacheKey(NamedTuple):
+    owner: str
+    repo: str
+    username: str
+
+
 async def aget_user_permission(
     gh: AsyncGitHubAPI, owner: str, repo: str, username: str
 ) -> Permission:
@@ -92,18 +93,121 @@ def get_user_permission(
     try:
         # Check if user is a collaborator and get their permission
         data = gh.getitem(f"/repos/{owner}/{repo}/collaborators/{username}/permission")
-        permission_str = data.get("permission", "none")
+        permission_str = data.get("permission", "none")  # type: ignore[attr-defined]
         permission = Permission.from_string(permission_str)
     except gidgethub.HTTPException as e:
         if e.status_code == 404:
             # User is not a collaborator, they have read permission if repo is public
             # Check if repo is public
             try:
                 repo_data = gh.getitem(f"/repos/{owner}/{repo}")
-                if not repo_data.get("private", True):
+                if not repo_data.get("private", True):  # type: ignore[attr-defined]
                     permission = Permission.READ
             except gidgethub.HTTPException:
                 pass
 
     cache[cache_key] = permission
     return permission
+
+
+class EventInfo(NamedTuple):
+    comment_author: str | None
+    owner: str | None
+    repo: str | None
+
+    @classmethod
+    def from_event(cls, event: sansio.Event) -> EventInfo:
+        comment_author = None
+        owner = None
+        repo = None
+
+        if "comment" in event.data:
+            comment_author = event.data["comment"]["user"]["login"]
+
+        if "repository" in event.data:
+            owner = event.data["repository"]["owner"]["login"]
+            repo = event.data["repository"]["name"]
+
+        return cls(comment_author=comment_author, owner=owner, repo=repo)
+
+
+class PermissionCheck(NamedTuple):
+    has_permission: bool
+    error_message: str | None
+
+
+PERMISSION_CHECK_ERROR_MESSAGE = """
+❌ **Permission Denied**
+
+@{comment_author}, you need at least **{required_permission}** permission to use this command.
+
+Your current permission level: **{user_permission}**
+"""
+
+
+async def acheck_mention_permission(
+    event: sansio.Event, gh: AsyncGitHubAPI, required_permission: Permission
+) -> PermissionCheck:
+    comment_author, owner, repo = EventInfo.from_event(event)
+
+    if not (comment_author and owner and repo):
+        return PermissionCheck(has_permission=False, error_message=None)
+
+    user_permission = await aget_user_permission(gh, owner, repo, comment_author)
+
+    if user_permission >= required_permission:
+        return PermissionCheck(has_permission=True, error_message=None)
+
+    return PermissionCheck(
+        has_permission=False,
+        error_message=PERMISSION_CHECK_ERROR_MESSAGE.format(
+            comment_author=comment_author,
+            required_permission=required_permission.name.lower(),
+            user_permission=user_permission.name.lower(),
+        ),
+    )
+
+
+def check_mention_permission(
+    event: sansio.Event, gh: SyncGitHubAPI, required_permission: Permission
+) -> PermissionCheck:
+    comment_author, owner, repo = EventInfo.from_event(event)
+
+    if not (comment_author and owner and repo):
+        return PermissionCheck(has_permission=False, error_message=None)
+
+    user_permission = get_user_permission(gh, owner, repo, comment_author)
+
+    if user_permission >= required_permission:
+        return PermissionCheck(has_permission=True, error_message=None)
+
+    return PermissionCheck(
+        has_permission=False,
+        error_message=PERMISSION_CHECK_ERROR_MESSAGE.format(
+            comment_author=comment_author,
+            required_permission=required_permission.name.lower(),
+            user_permission=user_permission.name.lower(),
+        ),
+    )
+
+
+def get_comment_post_url(event: sansio.Event) -> str | None:
+    if event.data.get("action") != "created":
+        return None
+
+    _, owner, repo = EventInfo.from_event(event)
+
+    if not (owner and repo):
+        return None
+
+    if "issue" in event.data:
+        issue_number = event.data["issue"]["number"]
+        return f"/repos/{owner}/{repo}/issues/{issue_number}/comments"
+    elif "pull_request" in event.data:
+        pr_number = event.data["pull_request"]["number"]
+        return f"/repos/{owner}/{repo}/issues/{pr_number}/comments"
+    elif "commit_sha" in event.data:
+        commit_sha = event.data["commit_sha"]
+        return f"/repos/{owner}/{repo}/commits/{commit_sha}/comments"
+
+    return None

src/django_github_app/routing.py

@@ -14,9 +14,15 @@
 from gidgethub.routing import Router as GidgetHubRouter
 
 from ._typing import override
+from .github import AsyncGitHubAPI
+from .github import SyncGitHubAPI
 from .mentions import MentionScope
 from .mentions import check_event_for_mention
 from .mentions import check_event_scope
+from .permissions import Permission
+from .permissions import acheck_mention_permission
+from .permissions import check_mention_permission
+from .permissions import get_comment_post_url
 
 AsyncCallback = Callable[..., Awaitable[None]]
 SyncCallback = Callable[..., None]
@@ -76,7 +82,7 @@ def decorator(func: CB) -> CB:
 
             @wraps(func)
             async def async_wrapper(
-                event: sansio.Event, *args: Any, **wrapper_kwargs: Any
+                event: sansio.Event, gh: AsyncGitHubAPI, *args: Any, **kwargs: Any
             ) -> None:
                 # TODO: Get actual bot username from installation/app data
                 username = "bot"  # Placeholder
@@ -87,12 +93,29 @@ async def async_wrapper(
                 if not check_event_scope(event, scope):
                     return
 
-                # TODO: Check permissions. For now, just call through.
-                await func(event, *args, **wrapper_kwargs)  # type: ignore[func-returns-value]
+                # Check permissions if required
+                if permission is not None:
+                    required_perm = Permission.from_string(permission)
+                    permission_check = await acheck_mention_permission(
+                        event, gh, required_perm
+                    )
+
+                    if not permission_check.has_permission:
+                        # Post error comment if we have an error message
+                        if permission_check.error_message:
+                            comment_url = get_comment_post_url(event)
+                            if comment_url:
+                                await gh.post(
+                                    comment_url,
+                                    data={"body": permission_check.error_message},
+                                )
+                        return
+
+                await func(event, gh, *args, **kwargs)  # type: ignore[func-returns-value]
 
             @wraps(func)
             def sync_wrapper(
-                event: sansio.Event, *args: Any, **wrapper_kwargs: Any
+                event: sansio.Event, gh: SyncGitHubAPI, *args: Any, **kwargs: Any
             ) -> None:
                 # TODO: Get actual bot username from installation/app data
                 username = "bot"  # Placeholder
@@ -103,8 +126,25 @@ def sync_wrapper(
                 if not check_event_scope(event, scope):
                     return
 
-                # TODO: Check permissions. For now, just call through.
-                func(event, *args, **wrapper_kwargs)
+                # Check permissions if required
+                if permission is not None:
+                    required_perm = Permission.from_string(permission)
+                    permission_check = check_mention_permission(
+                        event, gh, required_perm
+                    )
+
+                    if not permission_check.has_permission:
+                        # Post error comment if we have an error message
+                        if permission_check.error_message:
+                            comment_url = get_comment_post_url(event)
+                            if comment_url:
+                                gh.post(  # type: ignore[unused-coroutine]
+                                    comment_url,
+                                    data={"body": permission_check.error_message},
+                                )
+                        return
+
+                func(event, gh, *args, **kwargs)
 
             wrapper: MentionHandler
             if iscoroutinefunction(func):

tests/conftest.py

@@ -150,6 +150,31 @@ async def mock_getiter(*args, **kwargs):
     return _get_mock_github_api
 
 
+@pytest.fixture
+def get_mock_github_api_sync():
+    def _get_mock_github_api_sync(return_data):
+        from django_github_app.github import SyncGitHubAPI
+
+        mock_api = MagicMock(spec=SyncGitHubAPI)
+
+        def mock_getitem(*args, **kwargs):
+            return return_data
+
+        def mock_getiter(*args, **kwargs):
+            yield from return_data
+
+        def mock_post(*args, **kwargs):
+            pass
+
+        mock_api.getitem = mock_getitem
+        mock_api.getiter = mock_getiter
+        mock_api.post = mock_post
+
+        return mock_api
+
+    return _get_mock_github_api_sync
+
+
 @pytest.fixture
 def installation(get_mock_github_api, baker):
     installation = baker.make(