fix(appconnect): JWT token duration (#27310)

Turns out the JWT token is only allowed to be valid for maximum 20
mins and it seems Apple decided to start enforcing this.

This includes some more JWT-related fixes:

- Appstore connect was still using PyJWT directly instead of the
  sentry wrapper module.  That's wasn't the reason to create a wrapper
  module.

- The sentry.util.jwt module was not hooked up to mypy correctly.
  Hooking this up correctly surfaced some more typing errors.

This change carefully avoids making any runtime changes to the JWT
wrapper, so there should be no risk regarding this there.  The typing
information is only used during the mypy linting.

mypy.ini

@@ -30,13 +30,14 @@ files = src/sentry/api/bases/external_actor.py,
         src/sentry/snuba/query_subscription_consumer.py,
         src/sentry/tasks/app_store_connect.py,
         src/sentry/tasks/update_user_reports.py,
+        src/sentry/unmerge.py,
         src/sentry/utils/appleconnect/,
         src/sentry/utils/avatar.py,
         src/sentry/utils/codecs.py,
         src/sentry/utils/dates.py,
+        src/sentry/utils/jwt.py,
         src/sentry/utils/kvstore,
         src/sentry/utils/snql.py,
-        src/sentry/unmerge.py,
         src/sentry/web/decorators.py
 
 ; Enable all options used with --strict

src/sentry/utils/appleconnect/appstore_connect.py

@@ -3,10 +3,9 @@ import time
 from collections import namedtuple
 from typing import Any, Dict, Generator, List, Mapping, Optional, Union
 
-import jwt
 from requests import Session
 
-from sentry.utils import safe
+from sentry.utils import jwt, safe
 from sentry.utils.json import JSONData
 
 logger = logging.getLogger(__name__)
@@ -16,19 +15,18 @@ AppConnectCredentials = namedtuple("AppConnectCredentials", ["key_id", "key", "i
 
 def _get_authorization_header(
     credentials: AppConnectCredentials, expiry_sec: Optional[int] = None
-) -> str:
-    """
-    Creates a JWT (javascript web token) for use with app store connect API
+) -> Mapping[str, str]:
+    """Creates a JWT (javascript web token) for use with app store connect API
 
     All requests to app store connect require an "Authorization" header build as below.
 
-    Note: Setting a very large expiry for the token will cause the authorization to fail,
-    the default is one hour, which should be suitable for most cases.
+    Note: The maximum allowed expiry time is 20 minutes.  The default is somewhat shorter
+    than that to avoid running into the limit.
 
     :return: the Bearer auth token to be added as the  "Authorization" header
     """
     if expiry_sec is None:
-        expiry_sec = 60 * 60  # default one hour
+        expiry_sec = 60 * 10  # default to 10 mins
     token = jwt.encode(
         {
             "iss": credentials.issuer_id,
@@ -39,7 +37,7 @@ def _get_authorization_header(
         algorithm="ES256",
         headers={"kid": credentials.key_id, "alg": "ES256", "typ": "JWT"},
     )
-    return f"Bearer {token}"
+    return jwt.authorization_header(token)
 
 
 def _get_appstore_json(
@@ -54,7 +52,7 @@ def _get_appstore_json(
     :raises ValueError: if the request failed or the response body could not be parsed as
        JSON.
     """
-    headers = {"Authorization": _get_authorization_header(credentials)}
+    headers = _get_authorization_header(credentials)
 
     if not url.startswith("https://"):
         full_url = "https://api.appstoreconnect.apple.com"

src/sentry/utils/jwt.py

@@ -16,10 +16,12 @@ from cryptography.hazmat.primitives.serialization import (
 )
 from jwt import DecodeError
 
+from sentry.utils.json import JSONData
+
 __all__ = ["peek_claims", "decode", "encode", "authorization_header", "DecodeError"]
 
 
-def peek_header(token: str) -> Mapping[str, str]:
+def peek_header(token: str) -> JSONData:
     """Returns the headers in the JWT token without validation.
 
     Headers are not signed and can thus be spoofed.  You can use these to decide on what
@@ -28,18 +30,17 @@ def peek_header(token: str) -> Mapping[str, str]:
 
     :param token: The JWT token to extract the headers from.
     """
-    return pyjwt.get_unverified_header(token.encode("UTF-8"))  # type: ignore
+    return pyjwt.get_unverified_header(token.encode("UTF-8"))
 
 
-def peek_claims(token: str) -> Mapping[str, str]:
+def peek_claims(token: str) -> JSONData:
     """Returns the claims (payload) in the JWT token without validation.
 
     These claims can be used to look up the correct key to use in :func:`decode`.
 
     :param token: The JWT token to extract the payload from.
     """
-    # This type is checked in the tests so this is fine.
-    return pyjwt.decode(token, options={"verify_signature": False})  # type: ignore
+    return pyjwt.decode(token, options={"verify_signature": False})
 
 
 def decode(
@@ -47,8 +48,8 @@ def decode(
     key: str,
     *,  # Force passing optional arguments by keyword
     audience: Optional[Union[str, bool]] = None,
-    algorithms: List[str] = ["HS256"],
-) -> Mapping[str, str]:
+    algorithms: Optional[List[str]] = ["HS256"],
+) -> JSONData:
     """Returns the claims (payload) in the JWT token.
 
     This will raise an exception if the claims can not be validated with the provided key.
@@ -63,6 +64,11 @@ def decode(
     """
     # TODO: We do not currently have type-safety for keys suitable for decoding *and*
     # encoding vs those only suitable for decoding.
+    # TODO(flub): The algorithms parameter really does not need to be Optional and should be
+    # a straight List[str].  However this is used by some unclear code in
+    # sentry.integrations.msteams.webook.verify_signature which isn't checked by mypy yet,
+    # and I am too afraid to change this.  One day (hah!) all will be checked by mypy and
+    # this can be safely fixed.
     options = {"verify": True}
     kwargs = dict()
     if audience is False:
@@ -73,16 +79,15 @@ def decode(
         kwargs["audience"] = audience
     if algorithms is None:
         algorithms = ["HS256"]
-    # This type is checked in the tests so this is fine.
-    return pyjwt.decode(token, key, options=options, algorithms=algorithms, **kwargs)  # type: ignore
+    return pyjwt.decode(token, key, options=options, algorithms=algorithms, **kwargs)
 
 
 def encode(
-    payload: Mapping[str, str],
+    payload: JSONData,
     key: str,
     *,  # Force passing optional arguments by keyword
     algorithm: str = "HS256",
-    headers: Optional[Mapping[str, str]] = None,
+    headers: Optional[JSONData] = None,
 ) -> str:
     """Encode a JWT token containing the provided payload/claims.