ref: fix types for sentry.api.permissions

pyproject.toml

@@ -131,7 +131,6 @@ module = [
     "sentry.api.endpoints.project_rules_configuration",
     "sentry.api.invite_helper",
     "sentry.api.paginator",
-    "sentry.api.permissions",
     "sentry.auth.helper",
     "sentry.auth.provider",
     "sentry.db.mixin",
@@ -237,9 +236,11 @@ module = [
     "sentry.api.helpers.deprecation",
     "sentry.api.helpers.group_index.update",
     "sentry.api.helpers.source_map_helper",
+    "sentry.api.permissions",
     "sentry.api.serializers.models.organization_member.*",
     "sentry.api.serializers.rest_framework.group_notes",
     "sentry.audit_log.services.*",
+    "sentry.auth.access",
     "sentry.auth.manager",
     "sentry.auth.services.*",
     "sentry.auth.view",

src/sentry/api/bases/organization.py

@@ -10,6 +10,7 @@
 from rest_framework.exceptions import ParseError, PermissionDenied
 from rest_framework.permissions import BasePermission
 from rest_framework.request import Request
+from rest_framework.views import APIView
 
 from sentry.api.base import Endpoint
 from sentry.api.exceptions import ResourceDoesNotExist
@@ -79,7 +80,7 @@ def needs_sso(self, request: Request, organization: Organization | RpcOrganizati
     def has_object_permission(
         self,
         request: Request,
-        view: object,
+        view: APIView,
         organization: Organization | RpcOrganization | RpcUserOrganizationContext,
     ) -> bool:
         self.determine_access(request, organization)
@@ -106,7 +107,7 @@ class OrganizationAuditPermission(OrganizationPermission):
     def has_object_permission(
         self,
         request: Request,
-        view: object,
+        view: APIView,
         organization: Organization | RpcOrganization | RpcUserOrganizationContext,
     ) -> bool:
         if super().has_object_permission(request, view, organization):

src/sentry/api/bases/organizationmember.py

@@ -5,6 +5,7 @@
 from rest_framework import serializers
 from rest_framework.fields import empty
 from rest_framework.request import Request
+from rest_framework.views import APIView
 
 from sentry.api.exceptions import ResourceDoesNotExist
 from sentry.api.permissions import StaffPermissionMixin
@@ -30,7 +31,7 @@ class MemberPermission(OrganizationPermission):
     def has_object_permission(
         self,
         request: Request,
-        view: object,
+        view: APIView,
         organization: Organization | RpcOrganization | RpcUserOrganizationContext,
     ) -> bool:
         if not super().has_object_permission(request, view, organization):

src/sentry/api/bases/project.py

@@ -7,6 +7,7 @@
 from rest_framework.permissions import BasePermission
 from rest_framework.request import Request
 from rest_framework.response import Response
+from rest_framework.views import APIView
 
 from sentry.api.base import Endpoint
 from sentry.api.exceptions import ResourceDoesNotExist
@@ -41,7 +42,7 @@ class ProjectPermission(OrganizationPermission):
         "DELETE": ["project:admin"],
     }
 
-    def has_object_permission(self, request: Request, view, project):
+    def has_object_permission(self, request: Request, view: APIView, project: Project) -> bool:  # type: ignore[override]  # XXX: inheritance-for-convenience
         has_org_scope = super().has_object_permission(request, view, project.organization)
 
         # If allow_joinleave is False, some org-roles will not have project:read for all projects

src/sentry/api/permissions.py

@@ -66,56 +66,6 @@ def has_permission(self, request: Request, view: object) -> bool:
         return is_active_staff(request)
 
 
-class StaffPermissionMixin:
-    """
-    Sentry endpoints that should be accessible by staff but have an existing permission
-    class (that is not StaffPermission) require this mixin because staff does not give
-    any scopes.
-    NOTE: This mixin MUST be the leftmost parent class in the child class declaration in
-    order to work properly. See 'OrganizationAndStaffPermission' for an example of this or
-    https://www.python.org/download/releases/2.3/mro/ to learn more.
-    """
-
-    staff_allowed_methods = {"GET", "POST", "PUT", "DELETE"}
-
-    def has_permission(self, request, *args, **kwargs) -> bool:
-        """
-        Calls the parent class's has_permission method. If it returns False or
-        raises an exception and the method is allowed by the mixin, we then check
-        if the request is from an active staff. Raised exceptions are not caught
-        if the request is not allowed by the mixin or from an active staff.
-        """
-        try:
-            if super().has_permission(request, *args, **kwargs):
-                return True
-        except Exception:
-            if not (request.method in self.staff_allowed_methods and is_active_staff(request)):
-                raise
-            return True
-        return request.method in self.staff_allowed_methods and is_active_staff(request)
-
-    def has_object_permission(self, request, *args, **kwargs) -> bool:
-        """
-        Calls the parent class's has_object_permission method. If it returns False or
-        raises an exception and the method is allowed by the mixin, we then check
-        if the request is from an active staff. Raised exceptions are not caught
-        if the request is not allowed by the mixin or from an active staff.
-        """
-        try:
-            if super().has_object_permission(request, *args, **kwargs):
-                return True
-        except Exception:
-            if not (request.method in self.staff_allowed_methods and is_active_staff(request)):
-                raise
-            return True
-        return request.method in self.staff_allowed_methods and is_active_staff(request)
-
-    def is_not_2fa_compliant(self, request, *args, **kwargs) -> bool:
-        return super().is_not_2fa_compliant(request, *args, **kwargs) and not is_active_staff(
-            request
-        )
-
-
 # NOTE(schew2381): This is a temporary permission that does NOT perform an OR
 # between SuperuserPermission and StaffPermission. Instead, it uses StaffPermission
 # if the option is enabled for the user, and otherwise checks SuperuserPermission. We
@@ -154,7 +104,7 @@ class ScopedPermission(BasePermission):
 
     def has_permission(self, request: Request, view: APIView) -> bool:
         # session-based auth has all scopes for a logged in user
-        if not getattr(request, "auth", None):
+        if not request.auth:
             return request.user.is_authenticated
 
         if is_org_auth_token_auth(request.auth):
@@ -164,7 +114,8 @@ def has_permission(self, request: Request, view: APIView) -> bool:
             # where a project is available to update the project_last_used_id.
             update_org_auth_token_last_used(request.auth, [])
 
-        allowed_scopes: set[str] = set(self.scope_map.get(request.method, []))
+        assert request.method is not None
+        allowed_scopes = set(self.scope_map.get(request.method, []))
         current_scopes = request.auth.get_scopes()
         return any(s in allowed_scopes for s in current_scopes)
 
@@ -276,6 +227,56 @@ def determine_access(
                 raise MemberDisabledOverLimit(organization)
 
 
+class StaffPermissionMixin(SentryPermission):
+    """
+    Sentry endpoints that should be accessible by staff but have an existing permission
+    class (that is not StaffPermission) require this mixin because staff does not give
+    any scopes.
+    NOTE: This mixin MUST be the leftmost parent class in the child class declaration in
+    order to work properly. See 'OrganizationAndStaffPermission' for an example of this or
+    https://www.python.org/download/releases/2.3/mro/ to learn more.
+    """
+
+    staff_allowed_methods = {"GET", "POST", "PUT", "DELETE"}
+
+    def has_permission(self, request: Request, view: APIView) -> bool:
+        """
+        Calls the parent class's has_permission method. If it returns False or
+        raises an exception and the method is allowed by the mixin, we then check
+        if the request is from an active staff. Raised exceptions are not caught
+        if the request is not allowed by the mixin or from an active staff.
+        """
+        try:
+            if super().has_permission(request, view):
+                return True
+        except Exception:
+            if not (request.method in self.staff_allowed_methods and is_active_staff(request)):
+                raise
+            return True
+        return request.method in self.staff_allowed_methods and is_active_staff(request)
+
+    def has_object_permission(self, request: Request, view: APIView, obj: Any) -> bool:
+        """
+        Calls the parent class's has_object_permission method. If it returns False or
+        raises an exception and the method is allowed by the mixin, we then check
+        if the request is from an active staff. Raised exceptions are not caught
+        if the request is not allowed by the mixin or from an active staff.
+        """
+        try:
+            if super().has_object_permission(request, view, obj):
+                return True
+        except Exception:
+            if not (request.method in self.staff_allowed_methods and is_active_staff(request)):
+                raise
+            return True
+        return request.method in self.staff_allowed_methods and is_active_staff(request)
+
+    def is_not_2fa_compliant(
+        self, request: Request, organization: RpcOrganization | Organization
+    ) -> bool:
+        return super().is_not_2fa_compliant(request, organization) and not is_active_staff(request)
+
+
 class DemoSafePermission(SentryPermission):
     """
     A permission class that extends `SentryPermission` to provide read-only access for users
@@ -297,13 +298,12 @@ def determine_access(
                 id=extract_id_from(organization), user_id=request.user.id if request.user else None
             )
 
-        if org_context is None:
-            assert False, "Failed to fetch organization in determine_access"
+        assert org_context is not None, "Failed to fetch organization in determine_access"
 
         if demo_mode.is_demo_user(request.user):
             if org_context.member and demo_mode.is_demo_mode_enabled():
                 readonly_scopes = demo_mode.get_readonly_scopes()
-                org_context.member.scopes = readonly_scopes
+                org_context.member.scopes = sorted(readonly_scopes)
                 request.access = access.from_request_org_and_scopes(
                     request=request,
                     rpc_user_org_context=org_context,

src/sentry/auth/access.py

@@ -1,17 +1,5 @@
 from __future__ import annotations
 
-from sentry.constants import ObjectStatus
-
-__all__ = [
-    "from_user",
-    "from_member",
-    "DEFAULT",
-    "from_user_and_rpc_user_org_context",
-    "from_user_and_api_user_org_context",
-    "from_rpc_member",
-    "from_api_member",
-]
-
 import abc
 from collections.abc import Collection, Iterable, Mapping
 from dataclasses import dataclass
@@ -21,14 +9,15 @@
 import sentry_sdk
 from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
+from rest_framework.request import Request
 
 from sentry import features, roles
 from sentry.auth.services.access.service import access_service
-from sentry.auth.services.auth import RpcAuthState, RpcMemberSsoState
+from sentry.auth.services.auth import AuthenticatedToken, RpcAuthState, RpcMemberSsoState
 from sentry.auth.staff import is_active_staff
 from sentry.auth.superuser import get_superuser_scopes, is_active_superuser
-from sentry.auth.system import SystemToken, is_system_auth
-from sentry.models.apikey import ApiKey
+from sentry.auth.system import is_system_auth
+from sentry.constants import ObjectStatus
 from sentry.models.organization import Organization
 from sentry.models.organizationmember import OrganizationMember
 from sentry.models.organizationmemberteam import OrganizationMemberTeam
@@ -43,6 +32,16 @@
 from sentry.users.services.user import RpcUser
 from sentry.utils import metrics
 
+__all__ = (
+    "from_user",
+    "from_member",
+    "DEFAULT",
+    "from_user_and_rpc_user_org_context",
+    "from_user_and_api_user_org_context",
+    "from_rpc_member",
+    "from_api_member",
+)
+
 
 def has_role_in_organization(role: str, organization: Organization, user_id: int) -> bool:
     query = OrganizationMember.objects.filter(
@@ -897,7 +896,7 @@ def __init__(self) -> None:
 
 def from_request_org_and_scopes(
     *,
-    request: Any,
+    request: Request,
     rpc_user_org_context: RpcUserOrganizationContext | None = None,
     scopes: Iterable[str] | None = None,
 ) -> Access:
@@ -936,7 +935,7 @@ def from_request_org_and_scopes(
             scopes=superuser_scopes,
         )
 
-    if hasattr(request, "auth") and not request.user.is_authenticated:
+    if request.auth is not None and not request.user.is_authenticated:
         return from_rpc_auth(request.auth, rpc_user_org_context)
 
     return from_user_and_rpc_user_org_context(
@@ -970,7 +969,7 @@ def normalize_valid_user(user: User | RpcUser | AnonymousUser | None) -> User |
 
 def from_user_and_rpc_user_org_context(
     *,
-    user: User | RpcUser | None,
+    user: User | AnonymousUser | RpcUser | None,
     rpc_user_org_context: RpcUserOrganizationContext | None = None,
     is_superuser: bool = False,
     is_staff: bool = False,
@@ -996,7 +995,7 @@ def from_user_and_rpc_user_org_context(
 
 
 def from_request(
-    request: Any, organization: Organization | None = None, scopes: Iterable[str] | None = None
+    request: Request, organization: Organization | None = None, scopes: Iterable[str] | None = None
 ) -> Access:
     is_superuser = is_active_superuser(request)
     is_staff = is_active_staff(request)
@@ -1043,7 +1042,7 @@ def from_request(
             permissions=access_service.get_permissions_for_user(request.user.id),
         )
 
-    if hasattr(request, "auth") and not request.user.is_authenticated:
+    if request.auth is not None and not request.user.is_authenticated:
         return from_auth(request.auth, organization)
 
     return from_user(user=request.user, organization=organization, scopes=scopes, is_staff=is_staff)
@@ -1166,11 +1165,10 @@ def from_rpc_member(
 from_api_member = from_rpc_member
 
 
-def from_auth(auth: ApiKey | SystemToken, organization: Organization) -> Access:
+def from_auth(auth: AuthenticatedToken, organization: Organization) -> Access:
     if is_system_auth(auth):
         return SystemAccess()
-    assert not isinstance(auth, SystemToken)
-    if auth.organization_id == organization.id:
+    elif auth.organization_id == organization.id:
         return OrganizationGlobalAccess(
             auth.organization_id, settings.SENTRY_SCOPES, sso_is_valid=True
         )
@@ -1179,7 +1177,7 @@ def from_auth(auth: ApiKey | SystemToken, organization: Organization) -> Access:
 
 
 def from_rpc_auth(
-    auth: ApiKey | SystemToken, rpc_user_org_context: RpcUserOrganizationContext
+    auth: AuthenticatedToken, rpc_user_org_context: RpcUserOrganizationContext
 ) -> Access:
     if is_system_auth(auth):
         return SystemAccess()

src/sentry/users/api/bases/user.py

@@ -25,7 +25,7 @@
 class UserPermission(DemoSafePermission):
 
     def has_object_permission(
-        self, request: Request, view: APIView, user: User | RpcUser | None = None
+        self, request: Request, view: APIView, user: User | RpcUser | None
     ) -> bool:
 
         if user is None or request.user.id == user.id:

tests/sentry/api/bases/test_organization.py

@@ -281,7 +281,7 @@ def build_request(self, user=None, active_superuser=False, **params):
             user = self.user
         request.user = user
         request.auth = None
-        request.access = from_request(request, self.org)
+        request.access = from_request(drf_request_from_request(request), self.org)
         return request
 
 

tests/sentry/api/test_permissions.py

@@ -163,7 +163,7 @@ def test_determine_access(self):
             organization=readonly_rpc_context,
         )
 
-        assert readonly_rpc_context.member.scopes == READONLY_SCOPES
+        assert readonly_rpc_context.member.scopes == sorted(READONLY_SCOPES)
 
     @override_options({"demo-mode.enabled": False, "demo-mode.users": []})
     def test_determine_access_no_demo_users(self):