Merge pull request #479 from Shopify/add-session-token-utilites

[Feature] Add session token support

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:

README.md

@@ -120,6 +120,61 @@ _Note: Your application must be public to test the billing process. To test on a
     has_been_billed = activated_charge.status == 'active'
     ```
 
+### 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 `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.decode_from_header(
+    authorization_header=your_auth_request_header,
+    api_key=your_api_key,
+    secret=your_api_secret,
+)
+```
+
+#### Create a 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
+
+
+def session_token_required(func):
+    def wrapper(*args, **kwargs):
+        request = args[0]  # Or flask.request if you use Flask
+        try:
+            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(decoded_session_token):
+                return func(*args, **kwargs)
+        except session_token.SessionTokenError as e:
+            # Log the error here
+            return unauthorized_401_response()
+
+    return wrapper
+
+
+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  # Requests to /products require session tokens
+def products(request):
+    products = shopify.Product.find()
+    ...
+```
+
 ### Advanced Usage
 It is recommended to have at least a basic grasp on the principles of the [pyactiveresource](https://github.com/Shopify/pyactiveresource) library, which is a port of rails/ActiveResource to Python and upon which this package relies heavily.
 

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/session_token.py

@@ -0,0 +1,80 @@
+import jwt
+import re
+import six
+import sys
+
+from shopify.utils import shop_url
+
+if sys.version_info[0] < 3:  # Backwards compatibility for python < v3.0.0
+    from urlparse import urljoin
+else:
+    from urllib.parse import urljoin
+
+
+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 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)
+
+    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_url.sanitize_shop_domain(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

@@ -0,0 +1,20 @@
+import re
+import sys
+
+if sys.version_info[0] < 3:  # Backwards compatibility for python < v3.0.0
+    from urlparse import urlparse
+else:
+    from urllib.parse import urlparse
+
+HOSTNAME_PATTERN = r"[a-z0-9][a-z0-9-]*[a-z0-9]"
+
+
+def sanitize_shop_domain(shop_domain, myshopify_domain="myshopify.com"):
+    name = str(shop_domain).lower().strip()
+    if myshopify_domain not in name and "." not in name:
+        name += ".{domain}".format(domain=myshopify_domain)
+    name = re.sub(r"https?://", "", name)
+
+    uri = urlparse("http://{hostname}".format(hostname=name))
+    if re.match(r"{h}\.{d}$".format(h=HOSTNAME_PATTERN, d=re.escape(myshopify_domain)), uri.netloc):
+        return uri.netloc

shopify/utils/shop_url.py

@@ -0,0 +1,105 @@
+from shopify import session_token
+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 TestSessionTokenGetDecodedSessionToken(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(session_token.TokenAuthenticationError) as cm:
+            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) 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.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.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.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.decode_from_header(
+            self.build_auth_header(), api_key=self.api_key, secret=self.secret
+        )
+
+        self.assertEqual(self.payload, decoded_payload)

test/session_token_test.py

@@ -0,0 +1,44 @@
+from shopify.utils import shop_url
+from test.test_helper import TestCase
+
+
+class TestSanitizeShopDomain(TestCase):
+    def test_returns_hostname_for_good_shop_domains(self):
+        good_shop_domains = [
+            "my-shop",
+            "my-shop.myshopify.com",
+            "http://my-shop.myshopify.com",
+            "https://my-shop.myshopify.com",
+        ]
+        sanitized_shops = [shop_url.sanitize_shop_domain(shop_domain) for shop_domain in good_shop_domains]
+
+        self.assertTrue(all(shop == "my-shop.myshopify.com" for shop in sanitized_shops))
+
+    def test_returns_none_for_bad_shop_domains(self):
+        bad_shop_domains = [
+            "myshop.com",
+            "myshopify.com",
+            "shopify.com",
+            "two words",
+            "store.myshopify.com.evil.com",
+            "/foo/bar",
+            "/foo.myshopify.io.evil.ru",
+            "%0a123.myshopify.io ",
+            "foo.bar.myshopify.io",
+        ]
+        sanitized_shops = [shop_url.sanitize_shop_domain(shop_domain) for shop_domain in bad_shop_domains]
+
+        self.assertTrue(all(shop_domain is None for shop_domain in sanitized_shops))
+
+    def test_returns_hostname_for_custom_shop_domains(self):
+        custom_shop_domains = [
+            "my-shop",
+            "my-shop.myshopify.io",
+            "http://my-shop.myshopify.io",
+            "https://my-shop.myshopify.io",
+        ]
+        sanitized_shops = [
+            shop_url.sanitize_shop_domain(shop_domain, "myshopify.io") for shop_domain in custom_shop_domains
+        ]
+
+        self.assertTrue(all(shop == "my-shop.myshopify.io" for shop in sanitized_shops))