Potential fix for username faking

oauth2_provider_jwt/views.py

@@ -5,12 +5,17 @@
 from django.conf import settings
 from django.utils.module_loading import import_string
 from oauth2_provider import views
+from oauth2_provider.models import get_access_token_model
 
 from .utils import generate_payload, encode_jwt
 
 logger = logging.getLogger(__name__)
 
 
+class WrongUsername(Exception):
+    pass
+
+
 class TokenView(views.TokenView):
     def _get_access_token_jwt(self, request, content):
         extra_data = {}
@@ -23,8 +28,17 @@ def _get_access_token_jwt(self, request, content):
         if 'scope' in content:
             extra_data['scope'] = content['scope']
 
-        if request.POST.get('username'):
-            extra_data['username'] = request.POST.get('username')
+        username = request.POST.get('username')
+        if username:
+            # HACK: The only way to verify the username is to check the token.
+            #       This means an extra wasted database call
+            token = get_access_token_model().objects.get(
+                token=content['access_token']
+            )
+            if token.user.get_username() != username:
+                raise WrongUsername()
+            extra_data['username'] = username
+
         payload = generate_payload(issuer, content['expires_in'], **extra_data)
         token = encode_jwt(payload)
         return token
@@ -47,11 +61,19 @@ def post(self, request, *args, **kwargs):
                 logger.warning(
                     'Missing JWT configuration, skipping token build')
             else:
-                content['access_token_jwt'] = self._get_access_token_jwt(
-                    request, content)
                 try:
-                    content = bytes(json.dumps(content), 'utf-8')
-                except TypeError:
-                    content = bytes(json.dumps(content).encode("utf-8"))
-                response.content = content
+                    content['access_token_jwt'] = self._get_access_token_jwt(
+                        request, content)
+                    try:
+                        content = bytes(json.dumps(content), 'utf-8')
+                    except TypeError:
+                        content = bytes(json.dumps(content).encode("utf-8"))
+                    response.content = content
+                except WrongUsername:
+                    response.status_code = 400
+                    response.content = json.dumps({
+                        "error": "invalid_request",
+                        "error_description": "Request username doesn't match "
+                                             "username in original authorize",
+                    })
         return response

tests/test_views.py

@@ -1,6 +1,7 @@
 import base64
 import datetime
 import json
+import re
 
 try:
     from urllib.parse import urlencode
@@ -119,6 +120,98 @@ def test_get_token(self):
         self.assertDictContainsSubset({'scope': 'read write'},
                                       self.decode_jwt(jwt_token))
 
+    def test_get_token_authorization_code(self):
+        """
+        Request an access token using Authorization Code Flow
+        """
+        Application.objects.create(
+            client_id='user_app_id',
+            client_secret='user_app_secret',
+            client_type=Application.CLIENT_CONFIDENTIAL,
+            authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+            name='user app',
+            skip_authorization=True,
+            redirect_uris='http://localhost:8002/callback',
+        )
+
+        self.client.force_login(self.test_user)
+
+        response = self.client.get(reverse("oauth2_provider_jwt:authorize") +
+                                   '?response_type=code&client_id=user_app_id')
+
+        self.assertEqual(302, response.status_code)
+        match = re.match(r'http://localhost:8002/callback\?code=(\w+)',
+                         response.url)
+        self.assertIsNotNone(match)
+        code = match.group(1)
+
+        # To simulate that the token call is normally made unauthenticated
+        self.client.logout()
+        data = {
+            'client_id': 'user_app_id',
+            'client_secret': 'user_app_secret',
+            'code': code,
+            'grant_type': 'authorization_code',
+            'redirect_uri': 'http://localhost:8002/callback',
+            'username': 'test_user',
+        }
+        response = self.client.post(reverse("oauth2_provider_jwt:token"), data)
+        self.assertEqual(200, response.status_code)
+        json_obj = response.json()
+        self.assertEqual('Bearer', json_obj['token_type'])
+        self.assertEqual('read write', json_obj['scope'])
+
+        access_token = json_obj['access_token']
+        self.assertTrue(access_token)
+        access_token_jwt = json_obj['access_token_jwt']
+        self.assertTrue(access_token_jwt)
+
+        payload_content = self.decode_jwt(access_token_jwt)
+        self.assertEqual('test_user', payload_content['username'])
+        self.assertEqual('read write', payload_content['scope'])
+
+    def test_get_token_authorization_code_wrong_user(self):
+        """
+        Fix for https://github.com/Humanitec/django-oauth-toolkit-jwt/issues/14
+        """
+        Application.objects.create(
+            client_id='user_app_id',
+            client_secret='user_app_secret',
+            client_type=Application.CLIENT_CONFIDENTIAL,
+            authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+            name='user app',
+            skip_authorization=True,
+            redirect_uris='http://localhost:8002/callback',
+        )
+
+        self.client.force_login(self.test_user)
+        response = self.client.get(reverse("oauth2_provider_jwt:authorize") +
+                                   '?response_type=code&client_id=user_app_id')
+
+        self.assertEqual(302, response.status_code)
+        match = re.match(r'http://localhost:8002/callback\?code=(\w+)',
+                         response.url)
+        self.assertIsNotNone(match)
+        code = match.group(1)
+
+        # To simulate that the token call is normally made unauthenticated
+        self.client.logout()
+        data = {
+            'client_id': 'user_app_id',
+            'client_secret': 'user_app_secret',
+            'code': code,
+            'grant_type': 'authorization_code',
+            'redirect_uri': 'http://localhost:8002/callback',
+            'username': 'some_fake_user',  # Pass in wrong user
+        }
+        response = self.client.post(reverse("oauth2_provider_jwt:token"), data)
+        self.assertEqual(400, response.status_code)
+        self.assertEqual({
+                        "error": "invalid_request",
+                        "error_description": "Request username doesn't match "
+                                             "username in original authorize",
+                        }, response.json())
+
     @patch('oauth2_provider_jwt.views.TokenView._is_jwt_config_set')
     def test_do_not_get_token_missing_conf(self, mock_is_jwt_config_set):
         """