Create a session_token file with helper methods

Author: NabeelAhsen

shopify/session_token.py

@@ -0,0 +1,81 @@
+import jwt
+import re
+import six
+import sys
+
+if sys.version_info[0] < 3:  # Backwards compatibility for python < v3.0.0
+    from urlparse import urljoin
+else:
+    from urllib.parse import urljoin
+
+
+HOSTNAME_PATTERN = r"[a-z0-9][a-z0-9-]*[a-z0-9]"
+SHOP_DOMAIN_RE = re.compile(r"^https://{h}\.myshopify\.com/$".format(h=HOSTNAME_PATTERN))
+
+ALGORITHM = "HS256"
+PREFIX = "Bearer "
+REQUIRED_FIELDS = ["iss", "dest", "sub", "jti", "sid"]
+
+
+class SessionTokenError(Exception):
+    pass
+
+
+class InvalidIssuerError(SessionTokenError):
+    pass
+
+
+class MismatchedHostsError(SessionTokenError):
+    pass
+
+
+class TokenAuthenticationError(SessionTokenError):
+    pass
+
+
+def get_decoded_session_token(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)
+
+    return decoded_payload
+
+
+def _extract_session_token(authorization_header):
+    if not authorization_header.startswith(PREFIX):
+        raise TokenAuthenticationError("The HTTP_AUTHORIZATION_HEADER provided does not contain a Bearer token")
+
+    return authorization_header[len(PREFIX) :]
+
+
+def _decode_session_token(session_token, api_key, secret):
+    try:
+        return jwt.decode(
+            session_token,
+            secret,
+            audience=api_key,
+            algorithms=[ALGORITHM],
+            options={"require": REQUIRED_FIELDS},
+        )
+    except jwt.exceptions.PyJWTError as exception:
+        six.raise_from(SessionTokenError(str(exception)), exception)
+
+
+def _validate_issuer(decoded_payload):
+    _validate_issuer_hostname(decoded_payload)
+    _validate_issuer_and_dest_match(decoded_payload)
+
+
+def _validate_issuer_hostname(decoded_payload):
+    issuer_root = urljoin(decoded_payload["iss"], "/")
+
+    if not SHOP_DOMAIN_RE.match(issuer_root):
+        raise InvalidIssuerError("Invalid issuer")
+
+
+def _validate_issuer_and_dest_match(decoded_payload):
+    issuer_root = urljoin(decoded_payload["iss"], "/")
+    dest_root = urljoin(decoded_payload["dest"], "/")
+
+    if issuer_root != dest_root:
+        raise MismatchedHostsError("The issuer and destination do not match")

shopify/utils/__init__.py

@@ -1,4 +1,4 @@
-from shopify.utils import *
+from shopify import session_token
 from test.test_helper import TestCase
 from datetime import datetime, timedelta
 
@@ -13,7 +13,7 @@ def timestamp(date):
     return time.mktime(date.timetuple()) if sys.version_info[0] < 3 else date.timestamp()
 
 
-class TestSessionTokenUtilityGetDecodedSessionToken(TestCase):
+class TestSessionTokenGetDecodedSessionToken(TestCase):
     @classmethod
     def setUpClass(self):
         self.secret = "API Secret"
@@ -42,45 +42,45 @@ def build_auth_header(self):
     def test_raises_if_token_authentication_header_is_not_bearer(self):
         authorization_header = "Bad auth header"
 
-        with self.assertRaises(TokenAuthenticationError):
-            SessionTokenUtility.get_decoded_session_token(
-                authorization_header, api_key=self.api_key, secret=self.secret
-            )
+        with self.assertRaises(session_token.TokenAuthenticationError) as cm:
+            session_token.get_decoded_session_token(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(jwt.exceptions.ExpiredSignatureError):
-            SessionTokenUtility.get_decoded_session_token(
-                self.build_auth_header(), api_key=self.api_key, secret=self.secret
-            )
+        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)
+
+        self.assertEqual("Signature has expired", str(cm.exception))
 
     def test_raises_if_aud_doesnt_match_api_key(self):
         self.payload["aud"] = "bad audience"
 
-        with self.assertRaises(jwt.exceptions.InvalidAudienceError):
-            SessionTokenUtility.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.get_decoded_session_token(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(InvalidIssuerError):
-            SessionTokenUtility.get_decoded_session_token(
-                self.build_auth_header(), api_key=self.api_key, secret=self.secret
-            )
+        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)
+
+        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(MismatchedHostsError):
-            SessionTokenUtility.get_decoded_session_token(
-                self.build_auth_header(), api_key=self.api_key, secret=self.secret
-            )
+        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)
+
+        self.assertEqual("The issuer and destination do not match", str(cm.exception))
 
     def test_returns_decoded_payload(self):
-        decoded_payload = SessionTokenUtility.get_decoded_session_token(
+        decoded_payload = session_token.get_decoded_session_token(
             self.build_auth_header(), api_key=self.api_key, secret=self.secret
         )