Change session_token decode signature

Author: NabeelAhsen

README.md

@@ -122,23 +122,23 @@ _Note: Your application must be public to test the billing process. To test on a
 
 ### Session tokens
 
-The Shopify Python API library provides helper methods to decode [session tokens](https://shopify.dev/concepts/apps/building-embedded-apps-using-session-tokens). You can use the `get_decoded_session_token` function to help extract and decode a session token from an HTTP Authorization header.
+The Shopify Python API library provides helper methods to decode [session tokens](https://shopify.dev/concepts/apps/building-embedded-apps-using-session-tokens). You can use the `decode_from_header` function to extract and decode a session token from an HTTP Authorization header.
 
 #### Basic usage
 
 ```python
 from shopify import session_token
 
-decoded_payload = session_token.get_decoded_session_token(
+decoded_payload = session_token.decode_from_header(
     authorization_header=your_auth_request_header,
     api_key=your_api_key,
     secret=your_api_secret,
 )
 ```
 
-#### Use session_token to create a decorator
+#### Create a decorator using `session_token`
 
-Here's a sample use-case of building a Django decorator using `session_token`:
+Here's a sample decorator that protects your app views/routes by requiring the presence of valid session tokens as part of a request's headers.
 
 ```python
 from shopify import session_token
@@ -148,27 +148,28 @@ def session_token_required(func):
     def wrapper(*args, **kwargs):
         request = args[0]  # Or flask.request if you use Flask
         try:
-            session_token = session_token.get_decoded_session_token(
+            decoded_session_token = session_token.decode_from_header(
                 authorization_header = request.headers.get('Authorization'),
                 api_key = SHOPIFY_API_KEY,
                 secret = SHOPIFY_API_SECRET
             )
-            with shopify_session(session_token):
+            with shopify_session(decoded_session_token):
                 return func(*args, **kwargs)
-        except session_token.SessionTokenError:
-            return generate_http_401_response()
+        except session_token.SessionTokenError as e:
+            # Log the error here
+            return unauthorized_401_response()
 
     return wrapper
 
 
-def shopify_session(session_token):
-    shopify_domain = session_token.get("dest")
+def shopify_session(decoded_session_token):
+    shopify_domain = decoded_session_token.get("dest")
     access_token = get_offline_access_token_by_shop_domain(shopify_domain)
 
     return shopify.Session.temp(shopify_domain, SHOPIFY_API_VERSION, access_token)
 
 
-@session_token_required
+@session_token_required  # Requests to /products require session tokens
 def products(request):
     products = shopify.Product.find()
     ...

shopify/session_token.py

@@ -33,7 +33,7 @@ class TokenAuthenticationError(SessionTokenError):
     pass
 
 
-def get_decoded_session_token(authorization_header, api_key, secret):
+def decode_from_header(authorization_header, api_key, secret):
     session_token = _extract_session_token(authorization_header)
     decoded_payload = _decode_session_token(session_token, api_key, secret)
     _validate_issuer(decoded_payload)

test/session_token_test.py

@@ -43,44 +43,62 @@ def test_raises_if_token_authentication_header_is_not_bearer(self):
         authorization_header = "Bad auth header"
 
         with self.assertRaises(session_token.TokenAuthenticationError) as cm:
-            session_token.get_decoded_session_token(authorization_header, api_key=self.api_key, secret=self.secret)
+            session_token.decode_from_header(authorization_header, api_key=self.api_key, secret=self.secret)
 
         self.assertEqual("The HTTP_AUTHORIZATION_HEADER provided does not contain a Bearer token", str(cm.exception))
 
     def test_raises_jwt_error_if_session_token_is_expired(self):
         self.payload["exp"] = timestamp((datetime.now() + timedelta(0, -10)))
 
-        with self.assertRaises(session_token.SessionTokenError, msg="Expird") as cm:
-            session_token.get_decoded_session_token(self.build_auth_header(), api_key=self.api_key, secret=self.secret)
+        with self.assertRaises(session_token.SessionTokenError) as cm:
+            session_token.decode_from_header(self.build_auth_header(), api_key=self.api_key, secret=self.secret)
 
         self.assertEqual("Signature has expired", str(cm.exception))
 
+    def test_raises_jwt_error_if_invalid_alg(self):
+        bad_session_token = jwt.encode(self.payload, None, algorithm="none")
+        invalid_header = "Bearer {session_token}".format(session_token=bad_session_token)
+
+        with self.assertRaises(session_token.SessionTokenError) as cm:
+            session_token.decode_from_header(invalid_header, api_key=self.api_key, secret=self.secret)
+
+        self.assertEqual("The specified alg value is not allowed", str(cm.exception))
+
+    def test_raises_jwt_error_if_invalid_signature(self):
+        bad_session_token = jwt.encode(self.payload, "bad_secret", algorithm="HS256")
+        invalid_header = "Bearer {session_token}".format(session_token=bad_session_token)
+
+        with self.assertRaises(session_token.SessionTokenError) as cm:
+            session_token.decode_from_header(invalid_header, api_key=self.api_key, secret=self.secret)
+
+        self.assertEqual("Signature verification failed", str(cm.exception))
+
     def test_raises_if_aud_doesnt_match_api_key(self):
         self.payload["aud"] = "bad audience"
 
         with self.assertRaises(session_token.SessionTokenError) as cm:
-            session_token.get_decoded_session_token(self.build_auth_header(), api_key=self.api_key, secret=self.secret)
+            session_token.decode_from_header(self.build_auth_header(), api_key=self.api_key, secret=self.secret)
 
         self.assertEqual("Invalid audience", str(cm.exception))
 
     def test_raises_if_issuer_hostname_is_invalid(self):
         self.payload["iss"] = "bad_shop_hostname"
 
         with self.assertRaises(session_token.InvalidIssuerError) as cm:
-            session_token.get_decoded_session_token(self.build_auth_header(), api_key=self.api_key, secret=self.secret)
+            session_token.decode_from_header(self.build_auth_header(), api_key=self.api_key, secret=self.secret)
 
         self.assertEqual("Invalid issuer", str(cm.exception))
 
     def test_raises_if_iss_and_dest_dont_match(self):
         self.payload["dest"] = "bad_shop.myshopify.com"
 
         with self.assertRaises(session_token.MismatchedHostsError) as cm:
-            session_token.get_decoded_session_token(self.build_auth_header(), api_key=self.api_key, secret=self.secret)
+            session_token.decode_from_header(self.build_auth_header(), api_key=self.api_key, secret=self.secret)
 
         self.assertEqual("The issuer and destination do not match", str(cm.exception))
 
     def test_returns_decoded_payload(self):
-        decoded_payload = session_token.get_decoded_session_token(
+        decoded_payload = session_token.decode_from_header(
             self.build_auth_header(), api_key=self.api_key, secret=self.secret
         )