fix(seer): Honor approved superuser access in agent tokens (#126377)

Agent tokens minted from an approved superuser ViewerContext can read
the selected organization when the delegating user is not a member.
Sentry carries the nested `superuser: {expires_at}` claim into the token
and always limits elevated access to read-only, including when approved
agent write grants exist. Token expiry is bounded by the original
deadline. Token use rechecks current user eligibility, customer access
policy, and organization binding.

Org members keep membership-based minting, including approved write
grants, even when browser superuser mode is active. Their tokens do not
carry a superuser claim or inherit its deadline.

This PR is stacked on #126126,
which issues and accepts the ViewerContext claim with a maximum
five-minute deadline. getsentry/seer#8606
carries it through Seer.

Author: gricha

src/sentry/api/authentication.py

@@ -618,7 +618,7 @@ class AgentTokenAuthentication(StandardAuthentication):
     The agent credential remains the authorization authority. A non-authoritative,
     ephemeral copy of the delegating user is returned only for compatibility with
     callsites that still require ``request.user``; API access is derived from the
-    delegating member and capped by the token in ``access.from_agent_auth``."""
+    delegating member or approved superuser context, capped by the token."""
 
     token_name = b"bearer"
 
@@ -661,7 +661,12 @@ def fail(reason: str, **extra: Any) -> NoReturn:
             fail("no_user_principal", org_id=auth_token.organization_id)
 
         # The delegating user must still be valid even though they are not the request user.
-        user = user_service.get_user(user_id=user_id)
+        if auth_token.superuser is not None:
+            # Non-members do not necessarily receive this cell's user-cache invalidations.
+            users = user_service.get_many(filter={"user_ids": [user_id]})
+            user = users[0] if users else None
+        else:
+            user = user_service.get_user(user_id=user_id)
         if user is None:
             fail("user_not_found", user_id=user_id)
         if not user.is_active:
@@ -677,7 +682,10 @@ def fail(reason: str, **extra: Any) -> NoReturn:
         )
         if org_context is None:
             fail("org_context_missing", user_id=user_id, org_id=auth_token.organization_id)
-        if org_context.member is None:
+        if auth_token.superuser is not None:
+            if resolve_superuser_access(auth_token.superuser, user, org_context) is None:
+                fail("superuser_access_invalid", user_id=user_id)
+        elif org_context.member is None:
             fail("org_membership_missing", user_id=user_id, org_id=auth_token.organization_id)
         if not features.has(
             agent_token.FEATURE_FLAG,

src/sentry/core/endpoints/project_index.py

@@ -63,7 +63,7 @@ def get(self, request: Request) -> Response:
                 if org_context is None:
                     queryset = queryset.none()
                 else:
-                    agent_access = access.from_agent_auth(request.auth, org_context)
+                    agent_access = access.from_rpc_auth(request.auth, org_context)
                     queryset = queryset.filter(organization_id=request.auth.organization_id)
                     if not agent_access.has_global_access:
                         queryset = queryset.filter(id__in=agent_access.accessible_project_ids)

src/sentry/middleware/viewer_context.py

@@ -7,6 +7,7 @@
 from django.http.request import HttpRequest
 from django.http.response import HttpResponseBase
 
+from sentry.auth.services.auth import AuthenticatedToken
 from sentry.seer.agent_token import is_agent_auth
 from sentry.viewer_context import (
     ActorType,
@@ -103,4 +104,5 @@ def _viewer_context_from_request(request: HttpRequest) -> ViewerContext:
         organization_id=organization_id,
         actor_type=actor_type,
         token=auth,
+        superuser=auth.superuser if isinstance(auth, AuthenticatedToken) else None,
     )

src/sentry/seer/agent_token.py

@@ -10,20 +10,22 @@
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from enum import StrEnum
-from typing import TypedDict, TypeGuard
+from typing import NotRequired, TypedDict, TypeGuard
 
 from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
 from django.db import router, transaction
 from django.utils import timezone
 from jwt import PyJWTError
+from pydantic import ValidationError
 
 from sentry.auth.services.auth import AuthenticatedToken
 from sentry.seer.models.agent_write_grant import (
     AGENT_SESSION_ID_MAX_LENGTH,
     DEFAULT_EXPIRATION,
     SeerAgentWriteGrant,
 )
+from sentry.types.superuser import SuperuserAccess
 from sentry.users.models.user import User
 from sentry.users.services.user import RpcUser
 from sentry.utils import jwt
@@ -76,6 +78,7 @@ class AgentTokenClaims(TypedDict):
     sid: str
     iat: int
     exp: int
+    superuser: NotRequired[dict[str, object]]
 
 
 def _signing_key() -> str:
@@ -170,6 +173,7 @@ def encode_agent_token(
     scopes: Iterable[str],
     session_id: str,
     ttl: timedelta = DEFAULT_TOKEN_TTL,
+    superuser: SuperuserAccess | None = None,
 ) -> tuple[str, datetime]:
     """Mint a signed agent token. Returns the JWT and its expiry. No DB write."""
     now = timezone.now()
@@ -186,6 +190,8 @@ def encode_agent_token(
         "iat": int(now.timestamp()),
         "exp": int(expires_at.timestamp()),
     }
+    if superuser is not None:
+        payload["superuser"] = superuser.dict()
     token = jwt.encode(
         payload,
         _signing_key(),
@@ -274,7 +280,7 @@ def _validate_claims(claims: Mapping[str, object]) -> AgentTokenClaims:
     if expires_at <= issued_at:
         raise jwt.DecodeError("invalid agent token lifetime")
 
-    return AgentTokenClaims(
+    result = AgentTokenClaims(
         ver=version,
         aud=audience,
         sub=subject,
@@ -284,6 +290,12 @@ def _validate_claims(claims: Mapping[str, object]) -> AgentTokenClaims:
         iat=issued_at,
         exp=expires_at,
     )
+    if "superuser" in claims:
+        try:
+            result["superuser"] = SuperuserAccess.parse_obj(claims["superuser"]).dict()
+        except ValidationError as exc:
+            raise jwt.DecodeError("invalid agent superuser access") from exc
+    return result
 
 
 def decode_agent_token(token_str: str) -> AgentTokenClaims:
@@ -313,6 +325,7 @@ def build_authenticated_token(claims: AgentTokenClaims) -> AuthenticatedToken:
         scopes=claims["scopes"],
         user_id=principal.id,
         organization_id=claims["org"],
+        superuser=SuperuserAccess.parse_obj(claims["superuser"]) if "superuser" in claims else None,
     )
 
 

src/sentry/seer/endpoints/organization_agent_token.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
+from datetime import datetime
+from datetime import timezone as datetime_timezone
 from typing import Any, cast
 
+from django.utils import timezone
 from drf_spectacular.utils import extend_schema
 from rest_framework import serializers
 from rest_framework.exceptions import PermissionDenied
@@ -25,6 +28,7 @@
 from sentry.models.organization import Organization
 from sentry.seer import agent_token
 from sentry.seer.endpoints.agent_request import AgentTokenRequestData, AgentTokenRequestSerializer
+from sentry.viewer_context import get_viewer_context
 
 
 class AgentTokenResponseSerializer(serializers.Serializer):
@@ -68,6 +72,7 @@ def post(self, request: Request, organization: Organization) -> Response:
         Body: ``{"sessionId": str, "requestedScopes"?: [str]}``. The token's scopes are the
         caller's own scopes intersected with read-only plus any approved grants for this
         org and session; ``requestedScopes`` can only narrow further. No token is stored.
+        Propagated superuser elevation always remains read-only, including with write grants.
         """
         if not features.has(agent_token.FEATURE_FLAG, organization, actor=request.user):
             raise ResourceDoesNotExist
@@ -100,11 +105,31 @@ def post(self, request: Request, organization: Organization) -> Response:
             requested_scopes=requested_scopes,
         )
 
+        viewer = get_viewer_context()
+        superuser = (
+            viewer.superuser
+            if viewer is not None
+            and viewer.user_id == user_id
+            and viewer.organization_id == organization.id
+            else None
+        )
+        ttl = agent_token.DEFAULT_TOKEN_TTL
+        if superuser is not None:
+            scopes = sorted(set(scopes) & agent_token.readonly_scopes())
+            ttl = min(
+                ttl,
+                datetime.fromtimestamp(superuser.expires_at, datetime_timezone.utc)
+                - timezone.now(),
+            )
+            if ttl.total_seconds() <= 0:
+                raise PermissionDenied("Elevated session expired.")
         token, expires_at = agent_token.encode_agent_token(
             user_id=user_id,
             organization_id=organization.id,
             scopes=scopes,
             session_id=session_id,
+            superuser=superuser,
+            ttl=ttl,
         )
         return Response(
             {