Add authenticate_user method

Author: jonathanlloyd

pusher_push_notifications/__init__.py

@@ -1,16 +1,23 @@
 """Pusher Push Notifications Python server SDK"""
 
+import datetime
 import json
 import re
+import time
 
+import jwt
 import requests
 import six
 
 SDK_VERSION = '1.0.3'
+
 INTEREST_MAX_LENGTH = 164
 INTEREST_REGEX = re.compile('^(_|-|=|@|,|\\.|;|[A-Z]|[a-z]|[0-9])*$')
 MAX_NUMBER_OF_INTERESTS = 100
 
+USER_ID_MAX_LENGTH = 164
+AUTH_TOKEN_DURATION = datetime.timedelta(days=1)
+
 
 class PusherError(Exception):
     """Base class for all Pusher push notifications errors"""
@@ -174,3 +181,39 @@ def publish(self, interests, publish_body):
             handle_http_error(response_body, response.status_code)
 
         return response_body
+
+    def authenticate_user(self, user_id):
+        """Generate an auth token which will allow devices to associate
+            themselves with the given user id
+
+        Args:
+            user_id (string): user id for which the token will be valid
+
+        Returns:
+            auth token for the requested user id (string)
+
+        Raises:
+            ValueError: if user_id is not a string
+            ValueError: is user_id is longer than the maximum of 164 chars
+
+        """
+        if not isinstance(user_id, six.string_types):
+            raise ValueError('user_id must be a string')
+        if len(user_id) > USER_ID_MAX_LENGTH:
+            raise ValueError('user_id longer than the maximum of 164 chars')
+
+        issuer = 'https://{}.pushnotifications.pusher.com'.format(self.instance_id)
+
+        now = datetime.datetime.utcnow()
+        expiry_datetime = now + AUTH_TOKEN_DURATION
+        expiry_timestamp = int(time.mktime(expiry_datetime.timetuple()))
+
+        return jwt.encode(
+            {
+                'iss': issuer,
+                'sub': user_id,
+                'exp': expiry_timestamp,
+            },
+            self.secret_key,
+            algorithm='HS256',
+        ).decode('utf-8')

requirements.txt

@@ -1,2 +1,3 @@
 six>1.4.0,<2
 requests>2.5.0,<3
+pyjwt==1.7.1

tests/test_push_notifications.py

@@ -1,10 +1,12 @@
 """Unit tests for Pusher Push Notifications Python server SDK"""
 
+import six
+import time
 import unittest
 
+import jwt
 import requests_mock
 
-
 from pusher_push_notifications import (
     PushNotifications,
     PusherAuthError,
@@ -456,3 +458,49 @@ def test_publish_should_error_correctly_if_error_not_json(self):
                 )
             self.assertIn('Unknown error: no description', str(e.exception))
 
+    def test_authenticate_user_should_return_token(self):
+        user_id = 'user-0001'
+        pn_client = PushNotifications(
+            'INSTANCE_ID',
+            'SECRET_KEY'
+        )
+
+        token_string = pn_client.authenticate_user(user_id)
+
+        self.assertIsInstance(token_string, six.string_types)
+        self.assertTrue(len(token_string) > 0)
+
+        decoded_token = jwt.decode(
+            token_string,
+            'SECRET_KEY',
+            algorithm='HS256',
+        )
+
+        expected_issuer = 'https://INSTANCE_ID.pushnotifications.pusher.com'
+        expected_subject = user_id
+
+        self.assertEquals(decoded_token.get('iss'), expected_issuer)
+        self.assertEquals(decoded_token.get('sub'), expected_subject)
+        self.assertIsNotNone(decoded_token.get('exp'))
+        self.assertTrue(decoded_token.get('exp') > time.time())
+
+
+    def test_authenticate_user_should_fail_if_user_id_not_a_string(self):
+        user_id = False
+        pn_client = PushNotifications(
+            'INSTANCE_ID',
+            'SECRET_KEY'
+        )
+        with self.assertRaises(ValueError) as e:
+            pn_client.authenticate_user(user_id)
+        self.assertIn('user_id must be a string', str(e.exception))
+
+    def test_authenticate_user_should_fail_if_user_id_too_long(self):
+        user_id = 'A' * 165
+        pn_client = PushNotifications(
+            'INSTANCE_ID',
+            'SECRET_KEY'
+        )
+        with self.assertRaises(ValueError) as e:
+            pn_client.authenticate_user(user_id)
+        self.assertIn('longer than the maximum of 164 chars', str(e.exception))