IDP token introspection

stytch/b2b/api/sessions.py

@@ -27,6 +27,8 @@
 from stytch.core.http.client import AsyncClient, SyncClient
 from stytch.shared import jwt_helpers, rbac_local
 from stytch.shared.policy_cache import PolicyCache
+from stytch.b2b.models.sessions import AccessTokenJWTClaims, AccessTokenJWTResponse
+from pydantic import ValidationError
 
 
 class Sessions:
@@ -757,3 +759,90 @@ async def authenticate_jwt_local_async(
         return local_resp.member_session
 
     # ENDMANUAL(authenticate_jwt_local)
+
+    # MANUAL(introspect_idp_access_token)(SERVICE_METHOD)
+    # ADDIMPORT: from typing import Optional
+    # ADDIMPORT: from stytch.b2b.models.idp import AccessTokenJWTResponse
+    def introspect_idp_access_token(
+        self,
+        access_token: str,
+        client_id: str,
+        client_secret: Optional[str] = None,
+        grant_type: str = 'authorization_code',
+        token_type_hint: str = 'access_token'
+    ) -> Optional[AccessTokenJWTResponse]:
+        return self.introspect_idp_access_token_local(access_token, client_id) or self.introspect_idp_access_token_network(access_token, client_id, client_secret, grant_type, token_type_hint)
+    # ENDMANUAL(introspect_idp_access_token)
+
+
+    # MANUAL(introspect_idp_access_token_network)(SERVICE_METHOD)
+    # ADDIMPORT: from typing import Optional
+    # ADDIMPORT: from stytch.b2b.models.idp import AccessTokenJWTClaims, AccessTokenJWTResponse
+    # ADDIMPORT: from stytch.shared import jwt_helpers
+    # ADDIMPORT: from stytch.shared import rbac_local
+    # ADDIMPORT: from pydantic import ValidationError
+    def introspect_idp_access_token_network(
+        self,
+        access_token: str,
+        client_id: str,
+        client_secret: Optional[str] = None,
+        grant_type: str = 'authorization_code',
+        token_type_hint: str = 'access_token'
+    ) -> Optional[Dict[str, any]]:
+        headers: Dict[str, str] = {
+            "Content-Type": "application/x-www-form-urlencoded"
+        }
+        data: Dict[str, Any] = {
+            "token": access_token,
+            "client_id": client_id,
+            "grant_type": grant_type,
+            "token_type_hint": token_type_hint
+        }
+        if client_secret is not None:
+            data["client_secret"] = client_secret
+
+        url = self.api_base.url_for(f"/v1/public/{self.project_id}/oauth2/introspect", data)
+        res = self.sync_client.postForm(url, data, headers)
+        try:
+            jwtResponse = AccessTokenJWTResponse.from_json(res.response.status_code, res.json)
+            return AccessTokenJWTClaims(
+                subject=jwtResponse.sub,
+                scopes=jwtResponse.scope,
+                custom_claims=None
+            )
+        except ValidationError as e:
+            return None
+
+    # ENDMANUAL(introspect_idp_access_token_network)
+
+    # MANUAL(introspect_idp_access_token_local)(SERVICE_METHOD)
+    # ADDIMPORT: from typing import Optional
+    # ADDIMPORT: from stytch.b2b.models.sessions import AccessTokenJWTClaims
+    # ADDIMPORT: from stytch.shared import jwt_helpers
+    def introspect_idp_access_token_local(
+        self,
+        access_token: str,
+        client_id: str,
+    ) -> Optional[AccessTokenJWTClaims]:
+        _scope_claim = "scope"
+        generic_claims = jwt_helpers.authenticate_jwt_local(
+            project_id=self.project_id,
+            jwks_client=self.jwks_client,
+            jwt=access_token,
+            custom_audience=client_id,
+            custom_issuer=f"https://stytch.com/{self.project_id}"
+        )
+        if generic_claims is None:
+            return None
+
+        custom_claims = {
+            k: v for k, v in generic_claims.untyped_claims.items() if k != _scope_claim
+        }
+
+        return AccessTokenJWTClaims(
+            subject=generic_claims.reserved_claims["sub"],
+            scopes=generic_claims.untyped_claims[_scope_claim],
+            custom_claims=custom_claims
+        )
+
+    # ENDMANUAL(introspect_idp_access_token_local)

stytch/b2b/models/sessions.py

@@ -201,3 +201,40 @@ class LocalJWTResponse(pydantic.BaseModel):
 
 
 # ENDMANUAL(LocalJWTResponse)
+
+
+# MANUAL(AccessTokenJWTResponse)(TYPES)
+# ADDIMPORT: from typing import Any, Dict, List, Optional
+# ADDIMPORT: import pydantic
+class AccessTokenJWTResponse(ResponseBase):
+    """Response type for `Sessions.introspect_idp_access_token`.
+    Fields:
+      - active: Whether or not this token is active.
+      - sub: Subject of this JWT.
+      - scope: Scopes that this JWT is granted.
+    """  # noqa
+
+    active: bool
+    sub: Optional[str]
+    scope: Optional[str]
+
+
+# ENDMANUAL(AccessTokenJWTResponse)
+
+# MANUAL(AccessTokenJWTClaims)(TYPES)
+# ADDIMPORT: from typing import Any, Dict, List, Optional
+# ADDIMPORT: import pydantic
+class AccessTokenJWTClaims(pydantic.BaseModel):
+    """Response type for `Sessions.introspect_idp_access_token`.
+    Fields:
+      - subject: The subject (either user_id or member_id) that the JWT is intended for.
+      - scopes: A list of scopes granted, separated by spaces.
+      - custom_claims: A dict of custom claims of the JWT.
+    """  # noqa
+
+    subject: str
+    scopes: Optional[str]
+    custom_claims: Optional[Dict[str, Any]] = None
+
+
+# ENDMANUAL(AccessTokenJWTClaims)

stytch/core/http/client.py

@@ -65,6 +65,17 @@ def post(
         final_headers.update(headers or {})
         resp = requests.post(url, json=json, headers=final_headers, auth=self.auth)
         return self._response_from_request(resp)
+
+    def postForm(
+        self,
+        url: str,
+        form: Optional[Dict[str, Any]],
+        headers: Optional[Dict[str, str]] = None,
+    ) -> ResponseWithJson:
+        final_headers = self.headers.copy()
+        final_headers.update(headers or {})
+        resp = requests.post(url, data=form, headers=final_headers, auth=self.auth)
+        return self._response_from_request(resp)
 
     def put(
         self,

stytch/shared/jwt_helpers.py

@@ -19,6 +19,8 @@ def authenticate_jwt_local(
     jwt: str,
     max_token_age_seconds: Optional[int] = None,
     leeway: int = 0,
+    custom_audience: Optional[str] = None,
+    custom_issuer: Optional[str] = False,
 ) -> Optional[GenericClaims]:
     """Parse a JWT and verify the signature locally
     (without calling /authenticate in the API).
@@ -32,13 +34,13 @@ def authenticate_jwt_local(
     The value for leeway is the maximum allowable difference in seconds when
     comparing timestamps. It defaults to zero.
     """
-    jwt_audience = project_id
-    jwt_issuer = f"stytch.com/{project_id}"
+    jwt_audience = custom_audience if custom_audience else project_id
+    jwt_issuer = custom_issuer if custom_issuer else f"stytch.com/{project_id}"
 
     now = time.time()
 
     signing_key = jwks_client.get_signing_key_from_jwt(jwt)
-
+    
     try:
         # NOTE: The max_token_age_seconds value is applied after decoding.
         payload = pyjwt.decode(