Add crypto token capability

* Add a method to generate crypto tokens for use as Bearer tokens
* Add a method to verify an incoming crypto token and unpack the header and
  claims
* Uses the PyJWT library

This is not JWT token support, merely a way to generate Bearer tokens that
won't have to be stored in a database but can self-validate and store
additional information that you see fit.

1 files changed, 134 insertions, 0 deletions

diff --git a/tests/oauth2/rfc6749/test_server.py b/tests/oauth2/rfc6749/test_server.py
index b6ad6c9..f523136 100644
--- a/tests/oauth2/rfc6749/test_server.py
+++ b/tests/oauth2/rfc6749/test_server.py
@@ -1,9 +1,12 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import, unicode_literals
 from ...unittest import TestCase
+import Crypto.PublicKey.RSA as RSA
 import json
+import jwt
 import mock
 
+from oauthlib import common
 from oauthlib.oauth2.rfc6749.endpoints.authorization import AuthorizationEndpoint
 from oauthlib.oauth2.rfc6749.endpoints.token import TokenEndpoint
 from oauthlib.oauth2.rfc6749.endpoints.resource import ResourceEndpoint
@@ -158,6 +161,137 @@ class TokenEndpointTest(TestCase):
         self.assertEqual(json.loads(body), token)
 
 
+class CryptoTokenEndpointTest(TestCase):
+
+    def setUp(self):
+        self.expires_in = 1800
+
+        def set_user(request):
+            request.user = mock.MagicMock()
+            request.client = mock.MagicMock()
+            request.client.client_id = 'mocked_client_id'
+            request.expires_in = self.expires_in
+            request.payload = {}
+            return True
+
+        self.mock_validator = mock.MagicMock()
+        self.mock_validator.authenticate_client.side_effect = set_user
+        self.addCleanup(setattr, self, 'mock_validator', mock.MagicMock())
+        auth_code = AuthorizationCodeGrant(
+                request_validator=self.mock_validator)
+        password = ResourceOwnerPasswordCredentialsGrant(
+                request_validator=self.mock_validator)
+        client = ClientCredentialsGrant(
+                request_validator=self.mock_validator)
+        supported_types = {
+                'authorization_code': auth_code,
+                'password': password,
+                'client_credentials': client,
+        }
+
+        self.private_pem = (
+            "-----BEGIN RSA PRIVATE KEY-----\n"
+            "MIIEpAIBAAKCAQEA6TtDhWGwzEOWZP6m/zHoZnAPLABfetvoMPmxPGjFjtDuMRPv\n"
+            "EvI1sbixZBjBtdnc5rTtHUUQ25Am3JzwPRGo5laMGbj1pPyCPxlVi9LK82HQNX0B\n"
+            "YK7tZtVfDHElQA7F4v3j9d3rad4O9/n+lyGIQ0tT7yQcBm2A8FEaP0bZYCLMjwMN\n"
+            "WfaVLE8eXHyv+MfpNNLI9wttLxygKYM48I3NwsFuJgOa/KuodXaAmf8pJnx8t1Wn\n"
+            "nxvaYXFiUn/TxmhM/qhemPa6+0nqq+aWV5eT7xn4K/ghLgNs09v6Yge0pmPl9Oz+\n"
+            "+bjJ+aKRnAmwCOY8/5U5EilAiUOeBoO9+8OXtwIDAQABAoIBAGFTTbXXMkPK4HN8\n"
+            "oItVdDlrAanG7hECuz3UtFUVE3upS/xG6TjqweVLwRqYCh2ssDXFwjy4mXRGDzF4\n"
+            "e/e/6s9Txlrlh/w1MtTJ6ZzTdcViR9RKOczysjZ7S5KRlI3KnGFAuWPcG2SuOWjZ\n"
+            "dZfzcj1Crd/ZHajBAVFHRsCo/ATVNKbTRprFfb27xKpQ2BwH/GG781sLE3ZVNIhs\n"
+            "aRRaED4622kI1E/WXws2qQMqbFKzo0m1tPbLb3Z89WgZJ/tRQwuDype1Vfm7k6oX\n"
+            "xfbp3948qSe/yWKRlMoPkleji/WxPkSIalzWSAi9ziN/0Uzhe65FURgrfHL3XR1A\n"
+            "B8UR+aECgYEA7NPQZV4cAikk02Hv65JgISofqV49P8MbLXk8sdnI1n7Mj10TgzU3\n"
+            "lyQGDEX4hqvT0bTXe4KAOxQZx9wumu05ejfzhdtSsEm6ptGHyCdmYDQeV0C/pxDX\n"
+            "JNCK8XgMku2370XG0AnyBCT7NGlgtDcNCQufcesF2gEuoKiXg6Zjo7sCgYEA/Bzs\n"
+            "9fWGZZnSsMSBSW2OYbFuhF3Fne0HcxXQHipl0Rujc/9g0nccwqKGizn4fGOE7a8F\n"
+            "usQgJoeGcinL7E9OEP/uQ9VX1C9RNVjIxP1O5/Guw1zjxQQYetOvbPhN2QhD1Ye7\n"
+            "0TRKrW1BapcjwLpFQlVg1ZeTPOi5lv24W/wX9jUCgYEAkrMSX/hPuTbrTNVZ3L6r\n"
+            "NV/2hN+PaTPeXei/pBuXwOaCqDurnpcUfFcgN/IP5LwDVd+Dq0pHTFFDNv45EFbq\n"
+            "R77o5n3ZVsIVEMiyJ1XgoK8oLDw7e61+15smtjT69Piz+09pu+ytMcwGn4y3Dmsb\n"
+            "dALzHYnL8iLRU0ubrz0ec4kCgYAJiVKRTzNBPptQom49h85d9ac3jJCAE8o3WTjh\n"
+            "Gzt0uHXrWlqgO280EY/DTnMOyXjqwLcXxHlu26uDP/99tdY/IF8z46sJ1KxetzgI\n"
+            "84f7kBHLRAU9m5UNeFpnZdEUB5MBTbwWAsNcYgiabpMkpCcghjg+fBhOsoLqqjhC\n"
+            "CnwhjQKBgQDkv0QTdyBU84TE8J0XY3eLQwXbrvG2yD5A2ntN3PyxGEneX5WTJGMZ\n"
+            "xJxwaFYQiDS3b9E7b8Q5dg8qa5Y1+epdhx3cuQAWPm+AoHKshDfbRve4txBDQAqh\n"
+            "c6MxSWgsa+2Ld5SWSNbGtpPcmEM3Fl5ttMCNCKtNc0UE16oHwaPAIw==\n"
+            "-----END RSA PRIVATE KEY-----"
+        )
+        token = tokens.BearerToken(self.mock_validator,
+                token_generator=tokens.crypto_token_generator(self.private_pem),
+                expires_in=self.expires_in)
+        self.endpoint = TokenEndpoint('authorization_code',
+                default_token_type=token, grant_types=supported_types)
+
+    @mock.patch('oauthlib.common.generate_token', new=lambda: 'abc')
+    def test_authorization_grant(self):
+        body = 'grant_type=authorization_code&code=abc&scope=all+of+them&state=xyz'
+        headers, body, status_code = self.endpoint.create_token_response(
+                '', body=body)
+        body = json.loads(body)
+        token = {
+            'token_type': 'Bearer',
+            'expires_in': self.expires_in,
+            'access_token': body['access_token'],
+            'refresh_token': 'abc',
+            'state': 'xyz'
+        }
+        self.assertEqual(body, token)
+
+    @mock.patch('oauthlib.common.generate_token', new=lambda: 'abc')
+    def test_password_grant(self):
+        body = 'grant_type=password&username=a&password=hello&scope=all+of+them'
+        headers, body, status_code = self.endpoint.create_token_response(
+                '', body=body)
+        body = json.loads(body)
+        token = {
+            'token_type': 'Bearer',
+            'expires_in': self.expires_in,
+            'access_token': body['access_token'],
+            'refresh_token': 'abc',
+            'scope': 'all of them',
+        }
+        self.assertEqual(body, token)
+
+    @mock.patch('oauthlib.common.generate_token', new=lambda: 'abc')
+    def test_scopes_stored_in_access_token(self):
+        body = 'grant_type=password&username=a&password=hello&scope=all+of+them'
+        headers, body, status_code = self.endpoint.create_token_response(
+                '', body=body)
+
+        access_token = json.loads(body)['access_token']
+
+        claims = common.verify_crypto_token(self.private_pem, access_token)
+
+        self.assertEqual(claims['scope'], 'all of them')
+
+    @mock.patch('oauthlib.common.generate_token', new=lambda: 'abc')
+    def test_client_grant(self):
+        body = 'grant_type=client_credentials&scope=all+of+them'
+        headers, body, status_code = self.endpoint.create_token_response(
+                '', body=body)
+        body = json.loads(body)
+        token = {
+            'token_type': 'Bearer',
+            'expires_in': self.expires_in,
+            'access_token': body['access_token'],
+            'scope': 'all of them',
+        }
+        self.assertEqual(body, token)
+
+    def test_missing_type(self):
+        _, body, _ = self.endpoint.create_token_response('', body='')
+        token = {'error': 'unsupported_grant_type'}
+        self.assertEqual(json.loads(body), token)
+
+    def test_invalid_type(self):
+        body = 'grant_type=invalid'
+        _, body, _ = self.endpoint.create_token_response('', body=body)
+        token = {'error': 'unsupported_grant_type'}
+        self.assertEqual(json.loads(body), token)
+
+
 class ResourceEndpointTest(TestCase):
 
     def setUp(self):