Approach for adding a CORS-middleware

Author: cheif

oauth2_provider/middleware.py

@@ -1,6 +1,9 @@
+from django import http
 from django.contrib.auth import authenticate
 from django.utils.cache import patch_vary_headers
 
+from .models import Application
+
 
 class OAuth2TokenMiddleware(object):
     """
@@ -32,3 +35,46 @@ def process_request(self, request):
     def process_response(self, request, response):
         patch_vary_headers(response, ('Authorization',))
         return response
+
+HEADERS = ('x-requested-with', 'content-type', 'accept', 'origin',
+           'authorization', 'x-csrftoken')
+METHODS = ('GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS')
+
+
+class CorsMiddleware(object):
+    def process_request(self, request):
+        '''If this is a preflight-request, we must always return 200'''
+        if (request.method == 'OPTIONS' and
+                'HTTP_ACCESS_CONTROL_REQUEST_METHOD' in request.META):
+            return http.HttpResponse()
+        return None
+
+    def process_response(self, request, response):
+        '''Add cors-headers to request if they can be derived correctly'''
+        try:
+            cors_allow_origin = _get_cors_allow_origin_header(request)
+        except Application.NoSuitableOriginFoundError:
+            pass
+        else:
+            response['Access-Control-Allow-Origin'] = cors_allow_origin
+            response['Access-Control-Allow-Credentials'] = 'true'
+            if request.method == 'OPTIONS':
+                response['Access-Control-Allow-Headers'] = ', '.join(HEADERS)
+                response['Access-Control-Allow-Methods'] = ', '.join(METHODS)
+        return response
+
+
+def _get_cors_allow_origin_header(request):
+    '''Fetch the oauth-application that is responsible for making the
+    request and return a sutible cors-header, or None
+    '''
+    origin = request.META.get('HTTP_ORIGIN')
+    if origin:
+        try:
+            app = Application.objects.filter(redirect_uris__contains=origin)[0]
+        except IndexError:
+            # No application for this origin found
+            pass
+        else:
+            return app.get_cors_header(origin)
+    raise Application.NoSuitableOriginFoundError

oauth2_provider/models.py

@@ -120,12 +120,31 @@ def clean(self):
             error = _('Redirect_uris could not be empty with {0} grant_type')
             raise ValidationError(error.format(self.authorization_grant_type))
 
+    def get_cors_header(self, origin):
+        '''Return a proper cors-header for this origin, in the context of this
+        application.
+
+        :param origin: Origin-url from HTTP-request.
+        :raises: Application.NoSuitableOriginFoundError
+        '''
+        parsed_origin = urlparse(origin)
+        for allowed_uri in self.redirect_uris.split():
+            parsed_allowed_uri = urlparse(allowed_uri)
+            if (parsed_allowed_uri.scheme == parsed_origin.scheme and
+                    parsed_allowed_uri.netloc == parsed_origin.netloc and
+                    parsed_allowed_uri.port == parsed_origin.port):
+                return origin
+        raise Application.NoSuitableOriginFoundError
+
     def get_absolute_url(self):
         return reverse('oauth2_provider:detail', args=[str(self.id)])
 
     def __str__(self):
         return self.name or self.client_id
 
+    class NoSuitableOriginFoundError(Exception):
+        pass
+
 
 class Application(AbstractApplication):
     class Meta(AbstractApplication.Meta):

oauth2_provider/tests/test_cors_middleware.py

@@ -0,0 +1,85 @@
+from datetime import timedelta
+
+from django.test import TestCase, Client, override_settings
+from django.utils import timezone
+from django.conf.urls import patterns, url
+from django.http import HttpResponse
+from django.views.generic import View
+
+from ..models import AccessToken, get_application_model
+from django.contrib.auth import get_user_model
+
+
+Application = get_application_model()
+UserModel = get_user_model()
+
+
+class MockView(View):
+    def post(self, request):
+        return HttpResponse()
+
+urlpatterns = patterns(
+    '',
+    url(r'^cors-test/$', MockView.as_view()),
+)
+
+
+@override_settings(
+    ROOT_URLCONF='oauth2_provider.tests.test_cors_middleware',
+    AUTHENTICATION_BACKENDS=('oauth2_provider.backends.OAuth2Backend',),
+    MIDDLEWARE_CLASSES=(
+        'oauth2_provider.middleware.OAuth2TokenMiddleware',
+        'oauth2_provider.middleware.CorsMiddleware',
+    ))
+class TestCORSMiddleware(TestCase):
+    def setUp(self):
+        self.user = UserModel.objects.create_user('test_user', '[email protected]')
+        self.application = Application.objects.create(
+            name='Test Application',
+            redirect_uris='https://foo.bar',
+            user=self.user,
+            client_type=Application.CLIENT_CONFIDENTIAL,
+            authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+        )
+
+        self.access_token = AccessToken.objects.create(
+            user=self.user,
+            scope='read write',
+            expires=timezone.now() + timedelta(seconds=300),
+            token='secret-access-token-key',
+            application=self.application
+        )
+
+        auth_header = "Bearer {0}".format(self.access_token.token)
+        self.client = Client(HTTP_AUTHORIZATION=auth_header)
+
+    def test_cors_successful(self):
+        '''Ensure that we get cors-headers according to our oauth-app'''
+        resp = self.client.post('/cors-test/', HTTP_ORIGIN='https://foo.bar')
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp['Access-Control-Allow-Origin'], 'https://foo.bar')
+        self.assertEqual(resp['Access-Control-Allow-Credentials'], 'true')
+
+    def test_cors_no_auth(self):
+        '''Ensure that CORS-headers are sent non-authenticated requests'''
+        client = Client()
+        resp = client.post('/cors-test/', HTTP_ORIGIN='https://foo.bar')
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp['Access-Control-Allow-Origin'], 'https://foo.bar')
+        self.assertEqual(resp['Access-Control-Allow-Credentials'], 'true')
+
+    def test_cors_wrong_origin(self):
+        '''Ensure that CORS-headers aren't sent to requests from wrong origin'''
+        resp = self.client.post('/cors-test/', HTTP_ORIGIN='https://bar.foo')
+        self.assertEqual(resp.status_code, 200)
+        self.assertFalse(resp.has_header('Access-Control-Allow-Origin'))
+
+    def test_cors_200_preflight(self):
+        '''Ensure that preflight always get 200 responses'''
+        resp = self.client.options('/cors-test/',
+                                   HTTP_ACCESS_CONTROL_REQUEST_METHOD='GET',
+                                   HTTP_ORIGIN='https://foo.bar')
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(resp['Access-Control-Allow-Origin'], 'https://foo.bar')
+        self.assertTrue(resp.has_header('Access-Control-Allow-Headers'))
+        self.assertTrue(resp.has_header('Access-Control-Allow-Methods'))