ref(typing): Type request.user with service accounts outside an allowlist

src/AGENTS.md

@@ -80,6 +80,9 @@ The `ViewerContext` contextvar carries the current viewer's identity. When the v
 - `requires_user = True` on an endpoint: the same check for the whole endpoint.
 - `set_request_principal(request, user, auth)`: the only place `request.user` and `request.auth` are set.
 
+mypy types DRF `request.user` as `User | AnonymousUser | ServiceAccountPrincipal`, except in files on `tools/mypy_helpers/request_user_allowlist.txt`.
+Once a listed file narrows `request.user`, remove it from the list; never add to it.
+
 A service account is a `ServiceAccountPrincipal`: no user; its access is `request.access`. Personal state (bookmarks, "is this mine?") needs a user; "who did this" fields record None. Never look rows up by a possibly-None user id: `filter(user_id=None)` matches rows with no user (S032).
 
 Query "mine" and "not mine" with `owned_by_acting_user`, never `user_id=request.user.id`:

src/sentry/service_accounts/principal.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import dataclasses
-from typing import NoReturn
+from typing import TYPE_CHECKING, NoReturn
 
 from sentry.api.exceptions import UserRequired
 
@@ -52,9 +52,12 @@ class ServiceAccountPrincipal:
     is_sentry_app = False
     is_suspended = False
 
-    @property
-    def id(self) -> ServiceAccountId:
-        return ServiceAccountId(self.service_account_id)
+    # Hidden from mypy so request.user.id needs narrowing; kept at runtime as a guard.
+    if not TYPE_CHECKING:
+
+        @property
+        def id(self) -> ServiceAccountId:
+            return ServiceAccountId(self.service_account_id)
 
     def get_display_name(self) -> str:
         return self.name

tests/sentry/auth/test_principal.py

@@ -234,7 +234,8 @@ def authenticate(self, request: Request) -> tuple[User, None] | None:
         return Request(RequestFactory().get("/"), authenticators=[Authentication()])
 
 
-SRC = Path(__file__).parents[3] / "src" / "sentry"
+ROOT = Path(__file__).parents[3]
+SRC = ROOT / "src" / "sentry"
 
 
 def _is_request_user(node: ast.AST) -> bool:
@@ -276,6 +277,7 @@ def _is_request_user_type_check(node: ast.AST) -> bool:
 # Lower these as callers move off. Exact, so the baseline can't drift back up.
 REQUEST_USER_ID_BASELINE = 135
 REQUEST_USER_TYPE_CHECK_BASELINE = 31
+REQUEST_USER_ALLOWLIST_BASELINE = 23
 
 
 def test_request_user_id_uses_only_go_down() -> None:
@@ -301,3 +303,19 @@ def test_request_user_type_checks_only_go_down() -> None:
         f"request.user type checks dropped to {count}; lower "
         "REQUEST_USER_TYPE_CHECK_BASELINE to match."
     )
+
+
+def test_request_user_mypy_allowlist_only_goes_down() -> None:
+    allowlist = "tools/mypy_helpers/request_user_allowlist.txt"
+    paths = (ROOT / allowlist).read_text().split()
+    missing = [path for path in paths if not (ROOT / path).is_file()]
+    assert not missing, f"Remove deleted files from {allowlist}: {missing}"
+    count = len(paths)
+    assert count <= REQUEST_USER_ALLOWLIST_BASELINE, (
+        f"{count - REQUEST_USER_ALLOWLIST_BASELINE} new file(s) in {allowlist}, where mypy "
+        "doesn't know request.user can be a service account. Don't add to it: narrow with "
+        "has_user(request.user), acting_user_id(request.user) or require_user(request)."
+    )
+    assert count == REQUEST_USER_ALLOWLIST_BASELINE, (
+        f"{allowlist} dropped to {count} files; lower REQUEST_USER_ALLOWLIST_BASELINE to match."
+    )

tests/sentry/service_accounts/test_principal.py

@@ -83,11 +83,11 @@ def test_orm_lookup_by_id_raises(self) -> None:
             service_account_id=self.user.id, organization_id=self.organization.id, name="a"
         )
         with pytest.raises(UserRequired):
-            OrganizationMember.objects.filter(user_id=principal.id).exists()  # type: ignore[misc]
+            OrganizationMember.objects.filter(user_id=principal.id).exists()  # type: ignore[attr-defined]
 
     def test_rpc_lookup_by_id_raises(self) -> None:
         principal = ServiceAccountPrincipal(
             service_account_id=self.user.id, organization_id=self.organization.id, name="a"
         )
         with pytest.raises(UserRequired):
-            user_service.get_user(user_id=principal.id)  # type: ignore[arg-type]
+            user_service.get_user(user_id=principal.id)  # type: ignore[attr-defined]

tests/tools/mypy_helpers/test_plugin.py

@@ -41,6 +41,14 @@ def call_mypy(src: str, *, plugins: list[str] | None = None) -> tuple[int, str]:
         with open(os.path.join(auth_dir, "model.pyi"), "w") as f:
             f.write("class AuthenticatedToken: ...")
 
+        # stubs for request.user types
+        users_dir = _fill_init_pyi(tmpdir, "sentry/users/models")
+        with open(os.path.join(users_dir, "user.pyi"), "w") as f:
+            f.write("class User: ...")
+        sa_dir = _fill_init_pyi(tmpdir, "sentry/service_accounts")
+        with open(os.path.join(sa_dir, "principal.pyi"), "w") as f:
+            f.write("class ServiceAccountPrincipal: ...")
+
         # stub for rest_framework.response.Response — make it Generic[T] with a
         # body-less overload, mirroring fixtures/stubs-for-mypy/rest_framework/
         # response.pyi. The Response-body-Any plugin hook needs this shape to
@@ -262,6 +270,21 @@ def test_adjusted_drf_request_auth() -> None:
     assert out == expected_plugins
 
 
+def test_drf_request_user_includes_service_accounts() -> None:
+    src = """\
+from rest_framework.request import Request
+x: Request
+reveal_type(x.user)
+"""
+    expected = """\
+<string>:3: note: Revealed type is "sentry.users.models.user.User | django.contrib.auth.models.AnonymousUser | sentry.service_accounts.principal.ServiceAccountPrincipal"
+Success: no issues found in 1 source file
+"""
+    ret, out = call_mypy(src)
+    assert ret == 0
+    assert out == expected
+
+
 def test_csp_response_attribute() -> None:
     # technically undocumented -- django-csp's decorators usually do this
     src = """\

tools/mypy_helpers/plugin.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
 import functools
+import os.path
 from collections.abc import Callable
 from typing import Any
 
 from mypy.build import PRI_MYPY
 from mypy.errorcodes import ATTR_DEFINED
+from mypy.lookup import lookup_fully_qualified
 from mypy.messages import format_type
 from mypy.nodes import ARG_POS, DictExpr, ListExpr, MypyFile, StrExpr, TypeInfo
 from mypy.plugin import (
@@ -16,6 +18,7 @@
     MethodContext,
     MethodSigContext,
     Plugin,
+    ReportConfigContext,
     SemanticAnalyzerPluginInterface,
 )
 from mypy.plugins.common import add_attribute_to_class, add_method_to_class
@@ -160,6 +163,40 @@ def _adjust_request_members(ctx: ClassDefContext) -> None:
         add_attribute_to_class(ctx.api, ctx.cls, "auth", _request_auth_tp(ctx.api))
 
 
+_REQUEST_USER_FULLNAME = "rest_framework.request.Request.user"
+_REQUEST_USER_TPS = (
+    "sentry.users.models.user.User",
+    "django.contrib.auth.models.AnonymousUser",
+    "sentry.service_accounts.principal.ServiceAccountPrincipal",
+)
+_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+_REQUEST_USER_ALLOWLIST = os.path.join(os.path.dirname(__file__), "request_user_allowlist.txt")
+
+
+@functools.cache
+def _request_user_allowlist() -> frozenset[str]:
+    with open(_REQUEST_USER_ALLOWLIST) as f:
+        return frozenset(line.strip() for line in f if line.strip())
+
+
+def _is_request_user_allowlisted(path: str) -> bool:
+    return os.path.relpath(os.path.abspath(path), _REPO_ROOT) in _request_user_allowlist()
+
+
+def _request_user_attribute(ctx: AttributeContext) -> Type:
+    # sentry.auth.principal.set_request_principal can set a ServiceAccountPrincipal
+    if _is_request_user_allowlisted(ctx.api.path):
+        return ctx.default_attr_type
+    # named_generic_type walks parent packages, which don't always list these submodules
+    modules = ctx.api.modules  # type: ignore[attr-defined]
+    items: list[Type] = []
+    for tp in _REQUEST_USER_TPS:
+        st = lookup_fully_qualified(tp, modules, raise_on_missing=True)
+        assert st is not None and isinstance(st.node, TypeInfo), tp
+        items.append(Instance(st.node, ()))
+    return UnionType(items)
+
+
 def _add_principal_members_to_anonymous_user(ctx: ClassDefContext) -> None:
     if ctx.cls.name == "AnonymousUser":
         # added by sentry.auth.principal.patch_request_principal
@@ -550,12 +587,24 @@ def get_attribute_hook(self, fullname: str) -> Callable[[AttributeContext], Type
         if fullname.startswith("sentry.utils.lazy_service_wrapper.LazyServiceWrapper."):
             _, attr = fullname.rsplit(".", 1)
             return functools.partial(_lazy_service_wrapper_attribute, attr=attr)
+        elif fullname == _REQUEST_USER_FULLNAME:
+            return _request_user_attribute
         else:
             return None
 
+    def report_config_data(self, ctx: ReportConfigContext) -> bool:
+        # recheck a cached module when it joins or leaves the request.user allowlist
+        return _is_request_user_allowlisted(ctx.path)
+
     def get_additional_deps(self, file: MypyFile) -> list[tuple[int, str, int]]:
-        if file.fullname in {"django.http", "django.http.request", "rest_framework.request"}:
+        if file.fullname in {"django.http", "django.http.request"}:
             return [(PRI_MYPY, "sentry.auth.services.auth.model", -1)]
+        elif file.fullname == "rest_framework.request":
+            return [
+                (PRI_MYPY, "sentry.auth.services.auth.model", -1),
+                (PRI_MYPY, "sentry.users.models.user", -1),
+                (PRI_MYPY, "sentry.service_accounts.principal", -1),
+            ]
         else:
             return []
 

tools/mypy_helpers/request_user_allowlist.txt

@@ -0,0 +1,23 @@
+src/sentry/api/bases/organizationmember.py
+src/sentry/api/endpoints/organization_auth_tokens.py
+src/sentry/api/endpoints/organization_events.py
+src/sentry/api/endpoints/organization_events_stats.py
+src/sentry/api/endpoints/organization_releases.py
+src/sentry/api/endpoints/organization_trace_item_attributes.py
+src/sentry/api/endpoints/prompts_activity.py
+src/sentry/api/helpers/group_index/index.py
+src/sentry/core/endpoints/organization_member_details.py
+src/sentry/core/endpoints/organization_member_index.py
+src/sentry/core/endpoints/organization_member_requests_invite_index.py
+src/sentry/core/endpoints/organization_member_utils.py
+src/sentry/dashboards/endpoints/organization_dashboards.py
+src/sentry/data_export/endpoints/data_export.py
+src/sentry/issues/endpoints/group_event_details.py
+src/sentry/issues/endpoints/group_events.py
+src/sentry/issues/endpoints/group_integration_details.py
+src/sentry/issues/endpoints/organization_issues_count.py
+src/sentry/notebooks/endpoints/base.py
+src/sentry/seer/endpoints/organization_agent_token.py
+src/sentry/users/api/endpoints/user_authenticator_details.py
+src/sentry/web/client_config.py
+src/sentry/workflow_engine/endpoints/organization_detector_index.py