twilio · jingming · Dec 3, 2015 · Dec 3, 2015 · Dec 3, 2015 · Dec 3, 2015
diff --git a/CHANGES.md b/CHANGES.md
@@ -3,6 +3,13 @@ twilio-python Changelog
 
 Here you can see the full list of changes between each twilio-python release.
 
+Version 4.10.0
+-------------
+
+Released December 3, 2015:
+
+- Add Access Tokens
+
 Version 4.9.2
 -------------
 

diff --git a/setup.py b/setup.py
@@ -13,7 +13,7 @@
 #
 # You need to have the setuptools module installed. Try reading the setuptools
 # documentation: http://pypi.python.org/pypi/setuptools
-REQUIRES = ["httplib2 >= 0.7", "six", "pytz"]
+REQUIRES = ["httplib2 >= 0.7", "six", "pytz", "pyjwt"]
 
 if sys.version_info < (2, 6):
     REQUIRES.append('simplejson')

diff --git a/tests/test_access_token.py b/tests/test_access_token.py
@@ -0,0 +1,70 @@
+import unittest
+
+from nose.tools import assert_equal
+from twilio.jwt import decode
+from twilio.access_token import AccessToken, ConversationsGrant, IpMessagingGrant
+
+ACCOUNT_SID = 'AC123'
+SIGNING_KEY_SID = 'SK123'
+
+
+# python2.6 support
+def assert_is_not_none(obj):
+    assert obj is not None, '%r is None' % obj
+
+
+class AccessTokenTest(unittest.TestCase):
+    def _validate_claims(self, payload):
+        assert_equal(SIGNING_KEY_SID, payload['iss'])
+        assert_equal(ACCOUNT_SID, payload['sub'])
+        assert_is_not_none(payload['nbf'])
+        assert_is_not_none(payload['exp'])
+        assert_equal(payload['nbf'] + 3600, payload['exp'])
+        assert_is_not_none(payload['jti'])
+        assert_equal('{0}-{1}'.format(payload['iss'], payload['nbf']),
+                     payload['jti'])
+        assert_is_not_none(payload['grants'])
+
+    def test_empty_grants(self):
+        scat = AccessToken(ACCOUNT_SID, SIGNING_KEY_SID, 'secret')
+        token = str(scat)
+
+        assert_is_not_none(token)
+        payload = decode(token, 'secret')
+        self._validate_claims(payload)
+        assert_equal({}, payload['grants'])
+
+    def test_conversations_grant(self):
+        scat = AccessToken(ACCOUNT_SID, SIGNING_KEY_SID, 'secret')
+        scat.add_grant(ConversationsGrant())
+
+        token = str(scat)
+        assert_is_not_none(token)
+        payload = decode(token, 'secret')
+        self._validate_claims(payload)
+        assert_equal(1, len(payload['grants']))
+        assert_equal({}, payload['grants']['rtc'])
+
+    def test_ip_messaging_grant(self):
+        scat = AccessToken(ACCOUNT_SID, SIGNING_KEY_SID, 'secret')
+        scat.add_grant(IpMessagingGrant())
+
+        token = str(scat)
+        assert_is_not_none(token)
+        payload = decode(token, 'secret')
+        self._validate_claims(payload)
+        assert_equal(1, len(payload['grants']))
+        assert_equal({}, payload['grants']['ip_messaging'])
+
+    def test_grants(self):
+        scat = AccessToken(ACCOUNT_SID, SIGNING_KEY_SID, 'secret')
+        scat.add_grant(ConversationsGrant())
+        scat.add_grant(IpMessagingGrant())
+
+        token = str(scat)
+        assert_is_not_none(token)
+        payload = decode(token, 'secret')
+        self._validate_claims(payload)
+        assert_equal(2, len(payload['grants']))
+        assert_equal({}, payload['grants']['rtc'])
+        assert_equal({}, payload['grants']['ip_messaging'])
diff --git a/twilio/access_token.py b/twilio/access_token.py
@@ -0,0 +1,91 @@
+import time
+import jwt
+
+
+class IpMessagingGrant(object):
+    """ Grant to access Twilio IP Messaging """
+    def __init__(self, service_sid=None, endpoint_id=None,
+                 role_sid=None, credential_sid=None):
+        self.service_sid = service_sid
+        self.endpoint_id = endpoint_id
+        self.deployment_role_sid = role_sid
+        self.push_credential_sid = credential_sid
+
+    @property
+    def key(self):
+        return "ip_messaging"
+
+    def to_payload(self):
+        grant = {}
+        if self.service_sid:
+            grant['service_sid'] = self.service_sid
+        if self.endpoint_id:
+            grant['endpoint_id'] = self.endpoint_id
+        if self.deployment_role_sid:
+            grant['deployment_role_sid'] = self.deployment_role_sid
+        if self.push_credential_sid:
+            grant['push_credential_sid'] = self.push_credential_sid
+
+        return grant
+
+
+class ConversationsGrant(object):
+    """ Grant to access Twilio Conversations """
+    def __init__(self, configuration_profile_sid=None):
+        self.configuration_profile_sid = configuration_profile_sid
+
+    @property
+    def key(self):
+        return "rtc"
+
+    def to_payload(self):
+        grant = {}
+        if self.configuration_profile_sid:
+            grant['configuration_profile_sid'] = self.configuration_profile_sid
+
+        return grant
+
+
+class AccessToken(object):
+    """ Access Token used to access Twilio Resources """
+    def __init__(self, account_sid, signing_key_sid, secret,
+                 identity=None, ttl=3600):
+        self.account_sid = account_sid
+        self.signing_key_sid = signing_key_sid
+        self.secret = secret
+
+        self.identity = identity
+        self.ttl = ttl
+        self.grants = []
+
+    def add_grant(self, grant):
+        self.grants.append(grant)
+
+    def to_jwt(self, algorithm='HS256'):
+        now = int(time.time())
+        headers = {
+            "typ": "JWT",
+            "cty": "twilio-fpa;v=1"
+        }
+
+        grants = {}
+        if self.identity:
+            grants["identity"] = self.identity
+
+        for grant in self.grants:
+            grants[grant.key] = grant.to_payload()
+
+        payload = {
+            "jti": '{0}-{1}'.format(self.signing_key_sid, now),
+            "iss": self.signing_key_sid,
+            "sub": self.account_sid,
+            "nbf": now,
+            "exp": now + self.ttl,
+            "grants": grants
+        }
+

D5B8
+        return jwt.encode(payload, self.secret, headers=headers,
+                          algorithm=algorithm)
+
+    def __str__(self):
+        return self.to_jwt().decode('utf-8')
diff --git a/twilio/jwt/__init__.py b/twilio/jwt/__init__.py
@@ -41,9 +41,11 @@ def base64url_encode(input):
     return base64.urlsafe_b64encode(input).decode('utf-8').replace('=', '')
 
 
-def encode(payload, key, algorithm='HS256'):
+def encode(payload, key, algorithm='HS256', headers=None):
     segments = []
     header = {"typ": "JWT", "alg": algorithm}
+    if headers:
+        header.update(headers)
     segments.append(base64url_encode(binary(json.dumps(header))))
     segments.append(base64url_encode(binary(json.dumps(payload))))
     sign_input = '.'.join(segments)

diff --git a/twilio/version.py b/twilio/version.py
@@ -1,2 +1,2 @@
-__version_info__ = ('4', '9', '2')
+__version_info__ = ('4', '10', '0')
 __version__ = '.'.join(__version_info__)