Create SessionTokenUtility object to decode a session token

Add PyJWT dependency

Support python 2.7

Remove contradictory autopep8 formatter

Create custom exceptions for SessionTokenUtility

Address comments

Restructure utils into a subpackage

Author: NabeelAhsen

.pre-commit-config.yaml

@@ -6,10 +6,6 @@ repos:
     hooks:
     -   id: end-of-file-fixer
     -   id: trailing-whitespace
--   repo: https://github.com/pre-commit/mirrors-autopep8
-    rev: v1.5.4
-    hooks:
-    -   id: autopep8
 -   repo: https://github.com/psf/black
     rev: 20.8b1
     hooks:

setup.py

@@ -23,6 +23,8 @@
     license="MIT License",
     install_requires=[
         "pyactiveresource>=2.2.2",
+        "PyJWT <= 1.7.1; python_version == '2.7'",
+        "PyJWT >= 2.0.0; python_version >= '3.6'",
         "PyYAML",
         "six",
     ],

shopify/utils/__init__.py

@@ -0,0 +1,2 @@
+from .exceptions import *
+from .utilities import SessionTokenUtility

shopify/utils/exceptions.py

@@ -0,0 +1,10 @@
+class InvalidIssuerError(Exception):
+    pass
+
+
+class MismatchedHostsError(Exception):
+    pass
+
+
+class TokenAuthenticationError(Exception):
+    pass

shopify/utils/utilities.py

@@ -0,0 +1,64 @@
+from .exceptions import *
+
+import jwt
+import re
+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
+
+
+class SessionTokenUtility:
+    ALGORITHM = "HS256"
+    PREFIX = "Bearer "
+    REQUIRED_FIELDS = ["iss", "dest", "sub", "jti", "sid"]
+
+    @classmethod
+    def get_decoded_session_token(cls, authorization_header, api_key, secret):
+        session_token = cls.__extract_session_token(authorization_header)
+        decoded_payload = cls.__decode_session_token(session_token, api_key, secret)
+        cls.__validate_issuer(decoded_payload)
+
+        return decoded_payload
+
+    @classmethod
+    def __extract_session_token(cls, authorization_header):
+        if not authorization_header.startswith(cls.PREFIX):
+            raise TokenAuthenticationError("The HTTP_AUTHORIZATION_HEADER provided does not contain a Bearer token")
+
+        return authorization_header[len(cls.PREFIX) :]
+
+    @classmethod
+    def __decode_session_token(cls, session_token, api_key, secret):
+        return jwt.decode(
+            session_token,
+            secret,
+            audience=api_key,
+            algorithms=[cls.ALGORITHM],
+            options={"require": cls.REQUIRED_FIELDS},
+        )
+
+    @classmethod
+    def __validate_issuer(cls, decoded_payload):
+        cls.__validate_issuer_hostname(decoded_payload)
+        cls.__validate_issuer_and_dest_match(decoded_payload)
+
+    @classmethod
+    def __validate_issuer_hostname(cls, decoded_payload):
+        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))
+
+        issuer_root = urljoin(decoded_payload["iss"], "/")
+
+        if not shop_domain_re.match(issuer_root):
+            raise InvalidIssuerError()
+
+    @classmethod
+    def __validate_issuer_and_dest_match(cls, decoded_payload):
+        issuer_root = urljoin(decoded_payload["iss"], "/")
+        dest_root = urljoin(decoded_payload["dest"], "/")
+
+        if issuer_root != dest_root:
+            raise MismatchedHostsError()

test/utils/utilities_test.py

@@ -0,0 +1,87 @@
+from shopify.utils import *
+from test.test_helper import TestCase
+from datetime import datetime, timedelta
+
+import jwt
+import sys
+
+if sys.version_info[0] < 3:  # Backwards compatibility for python < v3.0.0
+    import time
+
+
+def timestamp(date):
+    return time.mktime(date.timetuple()) if sys.version_info[0] < 3 else date.timestamp()
+
+
+class TestSessionTokenUtilityGetDecodedSessionToken(TestCase):
+    @classmethod
+    def setUpClass(self):
+        self.secret = "API Secret"
+        self.api_key = "API key"
+
+    @classmethod
+    def setUp(self):
+        current_time = datetime.now()
+        self.payload = {
+            "iss": "https://test-shop.myshopify.com/admin",
+            "dest": "https://test-shop.myshopify.com",
+            "aud": self.api_key,
+            "sub": "1",
+            "exp": timestamp((current_time + timedelta(0, 60))),
+            "nbf": timestamp(current_time),
+            "iat": timestamp(current_time),
+            "jti": "4321",
+            "sid": "abc123",
+        }
+
+    @classmethod
+    def build_auth_header(self):
+        mock_session_token = jwt.encode(self.payload, self.secret, algorithm="HS256")
+        return "Bearer {session_token}".format(session_token=mock_session_token)
+
+    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
+            )
+
+    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
+            )
+
+    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
+            )
+
+    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
+            )
+
+    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
+            )
+
+    def test_returns_decoded_payload(self):
+        decoded_payload = SessionTokenUtility.get_decoded_session_token(
+            self.build_auth_header(), api_key=self.api_key, secret=self.secret
+        )
+
+        self.assertEqual(self.payload, decoded_payload)