feat(seer): Add principal-aware agent token minting

src/sentry/api/authentication.py

@@ -628,10 +628,12 @@ def accepts_auth(self, auth: list[bytes]) -> bool:
     def authenticate_token(self, request: Request, token_str: str) -> tuple[Any, Any]:
         try:
             claims = agent_token.decode_agent_token(token_str)
-            user_id = int(claims["sub"])
             # Building the token casts org and scopes too, so any missing/mis-typed claim
             # in a signed token is a clean 401 here, not a 500 downstream.
             auth_token = agent_token.build_authenticated_token(claims)
+            user_id = auth_token.user_id
+            if user_id is None:
+                raise ValueError("Agent token has no user principal")
         except (PyJWTError, KeyError, ValueError, TypeError):
             raise AuthenticationFailed("Invalid agent token")
 

src/sentry/seer/agent_token.py

@@ -7,7 +7,9 @@
 from __future__ import annotations
 
 from collections.abc import Iterable
+from dataclasses import dataclass
 from datetime import datetime, timedelta
+from enum import StrEnum
 from typing import Any, TypedDict
 
 from django.conf import settings
@@ -34,9 +36,34 @@
 DEFAULT_TOKEN_TTL = timedelta(minutes=5)
 
 AGENT_TOKEN_KIND = "agent_token"
+AGENT_TOKEN_VERSION = 1
+
+
+class AgentPrincipalType(StrEnum):
+    USER = "user"
+
+
+SUPPORTED_AGENT_PRINCIPAL_TYPES = frozenset({AgentPrincipalType.USER})
+MINTABLE_AGENT_PRINCIPAL_TYPES = frozenset({AgentPrincipalType.USER})
+
+
+class MintingPrincipalRejection(StrEnum):
+    INTEGRATION = "integration"
+    AGENT = "agent"
+    UNSUPPORTED = "unsupported"
+
+
+@dataclass(frozen=True)
+class AgentPrincipal:
+    type: AgentPrincipalType
+    id: int
+
+
+type MintingPrincipal = AgentPrincipal | MintingPrincipalRejection
 
 
 class AgentTokenClaims(TypedDict):
+    ver: int
     aud: str
     sub: str
     org: int
@@ -58,6 +85,43 @@ def readonly_scopes() -> frozenset[str]:
     return frozenset(settings.SENTRY_READONLY_SCOPES)
 
 
+def resolve_minting_principal(user: Any, auth: Any) -> MintingPrincipal:
+    if is_agent_auth(auth):
+        return MintingPrincipalRejection.AGENT
+    if getattr(user, "is_sentry_app", False):
+        return MintingPrincipalRejection.INTEGRATION
+    if not getattr(user, "is_authenticated", False):
+        return MintingPrincipalRejection.UNSUPPORTED
+
+    user_id = getattr(user, "id", None)
+    if not isinstance(user_id, int) or isinstance(user_id, bool):
+        return MintingPrincipalRejection.UNSUPPORTED
+    return AgentPrincipal(AgentPrincipalType.USER, user_id)
+
+
+def encode_principal_subject(principal: AgentPrincipal) -> str:
+    return f"{principal.type.value}:{principal.id}"
+
+
+def decode_principal_subject(subject: str) -> AgentPrincipal:
+    principal_type, separator, principal_id = subject.partition(":")
+    if separator != ":":
+        raise jwt.DecodeError("invalid agent token subject")
+    try:
+        parsed_type = AgentPrincipalType(principal_type)
+    except ValueError:
+        raise jwt.DecodeError("unsupported agent token principal")
+    if parsed_type not in SUPPORTED_AGENT_PRINCIPAL_TYPES:
+        raise jwt.DecodeError("unsupported agent token principal")
+    if not principal_id.isdigit():
+        raise jwt.DecodeError("invalid agent token subject")
+    return AgentPrincipal(parsed_type, int(principal_id))
+
+
+def principal_from_claims(claims: AgentTokenClaims) -> AgentPrincipal:
+    return decode_principal_subject(claims["sub"])
+
+
 def active_grant_scopes(organization_id: int, user_id: int, session_id: str) -> set[str]:
     """Unexpired scopes the user approved for the agent in this org + session. Keyed on
     authenticated identity, never client input."""
@@ -101,9 +165,12 @@ def encode_agent_token(
     """Mint a signed agent token. Returns the JWT and its expiry. No DB write."""
     now = timezone.now()
     expires_at = now + ttl
+    principal = AgentPrincipal(AgentPrincipalType.USER, user_id)
     payload: AgentTokenClaims = {
+        "ver": AGENT_TOKEN_VERSION,
         "aud": AGENT_TOKEN_AUDIENCE,
-        "sub": str(user_id),
+        # Keep the numeric wire subject until typed-subject decoding has reached every instance.
+        "sub": str(principal.id),
         "org": organization_id,
         "scopes": sorted(scopes),
         "sid": session_id,
@@ -137,8 +204,23 @@ def _validate_claims(claims: dict[str, Any]) -> AgentTokenClaims:
         raise jwt.DecodeError("missing agent token claim")
     if claims["aud"] != AGENT_TOKEN_AUDIENCE:
         raise jwt.DecodeError("invalid agent token audience")
-    if not isinstance(claims["sub"], str) or not claims["sub"].isdigit():
+    version = claims.get("ver")
+    subject = claims["sub"]
+    if version is None:
+        if not isinstance(subject, str) or not subject.isdigit():
+            raise jwt.DecodeError("invalid agent token subject")
+        subject = encode_principal_subject(AgentPrincipal(AgentPrincipalType.USER, int(subject)))
+        version = AGENT_TOKEN_VERSION
+    elif (
+        not isinstance(version, int) or isinstance(version, bool) or version != AGENT_TOKEN_VERSION
+    ):
+        raise jwt.DecodeError("unsupported agent token version")
+    if not isinstance(subject, str):
         raise jwt.DecodeError("invalid agent token subject")
+    if subject.isdigit():
+        # Rolling deployments may still emit numeric subjects. Their five-minute expiry bounds
+        # this compatibility path until typed-subject emission is enabled separately.
+        subject = encode_principal_subject(AgentPrincipal(AgentPrincipalType.USER, int(subject)))
     if not isinstance(claims["org"], int) or isinstance(claims["org"], bool):
         raise jwt.DecodeError("invalid agent token organization")
     scopes = claims["scopes"]
@@ -157,15 +239,18 @@ def _validate_claims(claims: dict[str, Any]) -> AgentTokenClaims:
             raise jwt.DecodeError(f"invalid agent token {claim_name}")
     if claims["exp"] <= claims["iat"]:
         raise jwt.DecodeError("invalid agent token lifetime")
-    return {
+    validated_claims: AgentTokenClaims = {
+        "ver": version,
         "aud": claims["aud"],
-        "sub": claims["sub"],
+        "sub": subject,
         "org": claims["org"],
         "scopes": scopes,
         "sid": session_id,
         "iat": claims["iat"],
         "exp": claims["exp"],
     }
+    principal_from_claims(validated_claims)
+    return validated_claims
 
 
 def decode_agent_token(token_str: str) -> AgentTokenClaims:
@@ -189,10 +274,11 @@ def is_agent_auth(auth: Any) -> bool:
 
 def build_authenticated_token(claims: AgentTokenClaims) -> AuthenticatedToken:
     """Build a delegated-user credential from claims verified by ``decode_agent_token``."""
+    principal = principal_from_claims(claims)
     return AuthenticatedToken(
         kind=AGENT_TOKEN_KIND,
         scopes=claims["scopes"],
-        user_id=int(claims["sub"]),
+        user_id=principal.id,
         organization_id=claims["org"],
     )
 

src/sentry/seer/endpoints/organization_agent_token.py

@@ -72,11 +72,12 @@ def post(self, request: Request, organization: Organization) -> Response:
         if not features.has(agent_token.FEATURE_FLAG, organization, actor=request.user):
             raise ResourceDoesNotExist
 
-        # Minting is a user-initiated action (direct session or Seer's X-Viewer-Context on
-        # the user's behalf). A non-user actor -- including an agent token itself -- must not
-        # mint, so identity is always a real user, never anonymous.
-        if not request.user.is_authenticated:
-            raise PermissionDenied("Minting requires a user session.")
+        minting_principal = agent_token.resolve_minting_principal(request.user, request.auth)
+        if (
+            not isinstance(minting_principal, agent_token.AgentPrincipal)
+            or minting_principal.type not in agent_token.MINTABLE_AGENT_PRINCIPAL_TYPES
+        ):
+            raise PermissionDenied("Minting requires a user principal.")
 
         data: Any = request.data
         if not isinstance(data, Mapping):
@@ -89,8 +90,7 @@ def post(self, request: Request, organization: Organization) -> Response:
         session_id = validated_data["sessionId"]
         requested_scopes = validated_data.get("requestedScopes")
 
-        user_id = request.user.id
-        assert user_id is not None  # guaranteed by the user-session requirement above
+        user_id = minting_principal.id
 
         # request.access.scopes is already the caller's role scopes intersected with any
         # OAuth token scopes, so it is the correct upper bound for de-escalation. Identity

tests/sentry/seer/endpoints/test_organization_agent_token.py

@@ -40,7 +40,8 @@ def test_mint_defaults_to_readonly(self) -> None:
             resp = self._mint(sessionId="s1")
         assert resp.status_code == 200, resp.content
         claims = agent_token.decode_agent_token(resp.data["token"])
-        assert claims["sub"] == str(self.owner.id)
+        assert claims["ver"] == agent_token.AGENT_TOKEN_VERSION
+        assert claims["sub"] == f"user:{self.owner.id}"
         assert claims["org"] == self.org.id
         assert claims["sid"] == "s1"
         assert "org:write" not in claims["scopes"]
@@ -84,9 +85,29 @@ def test_identity_comes_from_request_not_body(self) -> None:
         with self.feature(FLAG):
             resp = self._mint(sessionId="s1", userId=other.id, org=999999)
         claims = agent_token.decode_agent_token(resp.data["token"])
-        assert claims["sub"] == str(self.owner.id)
+        assert claims["sub"] == f"user:{self.owner.id}"
         assert claims["org"] == self.org.id
 
+    def test_sentry_app_installation_token_cannot_mint(self) -> None:
+        integration = self.create_internal_integration(
+            organization=self.org,
+            scopes=("org:read",),
+        )
+        token = self.create_internal_integration_token(
+            user=self.owner,
+            internal_integration=integration,
+        )
+
+        with self.feature(FLAG):
+            response = self.client.post(
+                f"/api/0/organizations/{self.org.slug}/agent/token/",
+                data={"sessionId": "s1"},
+                format="json",
+                HTTP_AUTHORIZATION=f"Bearer {token.token}",
+            )
+
+        assert response.status_code == 403, response.content
+
     def test_approved_grant_is_folded_into_token(self) -> None:
         self._grant(session_id="s1", scopes=["org:write"])
         self.login_as(self.owner)

tests/sentry/seer/test_agent_token.py

@@ -3,9 +3,11 @@
 import contextlib
 from collections.abc import Iterable
 from datetime import datetime, timedelta
+from types import SimpleNamespace
 from typing import Any
 
 import pytest
+from django.contrib.auth.models import AnonymousUser
 from django.contrib.sessions.backends.base import SessionBase
 from django.core.cache import cache
 from django.test import RequestFactory, override_settings
@@ -89,6 +91,39 @@ def _has_object_perm(self, drf_request: Request) -> bool:
 
     # ----- authentication -----
 
+    def test_minting_principal_resolves_only_users(self) -> None:
+        resolved = agent_token.resolve_minting_principal(self.owner, None)
+        assert resolved == agent_token.AgentPrincipal(
+            agent_token.AgentPrincipalType.USER,
+            self.owner.id,
+        )
+
+        unsupported = agent_token.resolve_minting_principal(AnonymousUser(), None)
+        assert unsupported is agent_token.MintingPrincipalRejection.UNSUPPORTED
+
+    def test_minting_principal_rejects_integrations_and_agents(self) -> None:
+        integration_user = SimpleNamespace(is_authenticated=True, is_sentry_app=True)
+        integration = agent_token.resolve_minting_principal(integration_user, None)
+        assert integration is agent_token.MintingPrincipalRejection.INTEGRATION
+
+        agent = agent_token.resolve_minting_principal(
+            AnonymousUser(),
+            SimpleNamespace(kind=agent_token.AGENT_TOKEN_KIND),
+        )
+        assert agent is agent_token.MintingPrincipalRejection.AGENT
+
+    def test_principal_subject_round_trips(self) -> None:
+        principal = agent_token.AgentPrincipal(agent_token.AgentPrincipalType.USER, self.owner.id)
+        subject = agent_token.encode_principal_subject(principal)
+
+        assert subject == f"user:{self.owner.id}"
+        assert agent_token.decode_principal_subject(subject) == principal
+
+    def test_principal_subject_rejects_invalid_or_unsupported_types(self) -> None:
+        for subject in ("1", "user:", "user:not-an-id", "service_account:1"):
+            with pytest.raises(jwt.DecodeError):
+                agent_token.decode_principal_subject(subject)
+
     def test_valid_token_authenticates_as_non_user_actor(self) -> None:
         # The agent is a non-user actor: the request user is anonymous and the credential
         # records the delegating user it acts on behalf of.
@@ -123,6 +158,60 @@ def test_minted_token_is_raw_typed_jwt(self) -> None:
 
         assert token.count(".") == 2
         assert jwt.peek_header(token)["typ"] == agent_token.AGENT_TOKEN_TYPE
+        wire_claims = jwt.decode(
+            token,
+            SECRET,
+            audience=agent_token.AGENT_TOKEN_AUDIENCE,
+            algorithms=["HS256"],
+        )
+        assert wire_claims["sub"] == str(self.owner.id)
+        claims = agent_token.decode_agent_token(token)
+        assert claims["ver"] == agent_token.AGENT_TOKEN_VERSION
+        assert claims["sub"] == f"user:{self.owner.id}"
+
+    def test_legacy_numeric_user_subject_is_normalized(self) -> None:
+        now = int(timezone.now().timestamp())
+        token = self._typed_token(
+            {
+                "aud": agent_token.AGENT_TOKEN_AUDIENCE,
+                "sub": str(self.owner.id),
+                "org": self.org.id,
+                "scopes": ["org:read"],
+                "sid": "s1",
+                "iat": now,
+                "exp": now + 300,
+            }
+        )
+
+        claims = agent_token.decode_agent_token(token)
+        assert claims["ver"] == agent_token.AGENT_TOKEN_VERSION
+        assert claims["sub"] == f"user:{self.owner.id}"
+        result = self._auth(token)
+        assert result is not None
+        assert result[1].user_id == self.owner.id
+
+    def test_unsupported_token_principal_is_rejected(self) -> None:
+        now = int(timezone.now().timestamp())
+        for version, subject in (
+            (agent_token.AGENT_TOKEN_VERSION + 1, "user:1"),
+            (True, "user:1"),
+            (agent_token.AGENT_TOKEN_VERSION, "service_account:1"),
+        ):
+            token = self._typed_token(
+                {
+                    "ver": version,
+                    "aud": agent_token.AGENT_TOKEN_AUDIENCE,
+                    "sub": subject,
+                    "org": self.org.id,
+                    "scopes": ["org:read"],
+                    "sid": "s1",
+                    "iat": now,
+                    "exp": now + 300,
+                }
+            )
+
+            with pytest.raises(AuthenticationFailed):
+                self._auth(token)
 
     def test_non_agent_bearer_is_deferred(self) -> None:
         # An ordinary database-backed token stays with the existing authenticator.