Check against session oidc_state on OIDC callback.

Author: johngian

mozilla_django_oidc/views.py

@@ -3,6 +3,7 @@
 except ImportError:
     from urllib.parse import urlencode
 
+from django.core.exceptions import SuspiciousOperation
 from django.core.urlresolvers import reverse
 from django.contrib import auth
 from django.http import HttpResponseRedirect
@@ -40,6 +41,14 @@ def get(self, request):
                 'code': request.GET['code'],
                 'state': request.GET['state']
             }
+
+            if 'oidc_state' not in request.session:
+                return self.login_failure()
+
+            if request.GET['state'] != request.session['oidc_state']:
+                msg = 'Session `oidc_state` does not match the OIDC callback state'
+                raise SuspiciousOperation(msg)
+
             self.user = auth.authenticate(**kwargs)
 
             if self.user and self.user.is_active:
@@ -60,14 +69,18 @@ def __init__(self, *args, **kwargs):
 
     def get(self, request):
         """OIDC client authentication initialization HTTP endpoint"""
+        state = get_random_string(import_from_settings('OIDC_STATE_SIZE', 32))
+
         params = {
             'response_type': 'code',
             'scope': 'openid',
             'client_id': self.OIDC_OP_CLIENT_ID,
             'redirect_uri': absolutify(reverse('oidc_authentication_callback')),
-            'state': get_random_string(import_from_settings('OIDC_STATE_SIZE', 32))
+            'state': state,
         }
 
+        request.session['oidc_state'] = state
+
         query = urlencode(params)
         redirect_url = '{url}?{query}'.format(url=self.OIDC_OP_AUTH_ENDPOINT, query=query)
         return HttpResponseRedirect(redirect_url)

tests/test_views.py

@@ -5,6 +5,7 @@
 
 from mock import patch
 
+from django.core.exceptions import SuspiciousOperation
 from django.contrib.auth import get_user_model
 from django.core.urlresolvers import reverse
 from django.test import RequestFactory, TestCase, override_settings
@@ -32,6 +33,9 @@ def test_get_auth_success(self):
         }
         url = reverse('oidc_authentication_callback')
         request = self.factory.get(url, get_data)
+        request.session = {
+            'oidc_state': 'example_state'
+        }
         callback_view = views.OIDCAuthenticationCallbackView.as_view()
 
         with patch('mozilla_django_oidc.views.auth.authenticate') as mock_auth:
@@ -55,6 +59,9 @@ def test_get_auth_failure_nonexisting_user(self):
 
         url = reverse('oidc_authentication_callback')
         request = self.factory.get(url, get_data)
+        request.session = {
+            'oidc_state': 'example_state'
+        }
         callback_view = views.OIDCAuthenticationCallbackView.as_view()
 
         with patch('mozilla_django_oidc.views.auth.authenticate') as mock_auth:
@@ -80,6 +87,9 @@ def test_get_auth_failure_inactive_user(self):
 
         url = reverse('oidc_authentication_callback')
         request = self.factory.get(url, get_data)
+        request.session = {
+            'oidc_state': 'example_state'
+        }
         callback_view = views.OIDCAuthenticationCallbackView.as_view()
 
         with patch('mozilla_django_oidc.views.auth.authenticate') as mock_auth:
@@ -105,6 +115,54 @@ def test_get_auth_dirty_data(self):
         self.assertEqual(response.status_code, 302)
         self.assertEqual(response.url, '/failure')
 
+    @override_settings(LOGIN_REDIRECT_URL_FAILURE='/failure')
+    def test_get_auth_failure_missing_session_state(self):
+        """Test authentication failure attempt for an inactive user."""
+        user = User.objects.create_user('example_username')
+        user.is_active = False
+        user.save()
+
+        get_data = {
+            'code': 'example_code',
+            'state': 'example_state'
+        }
+
+        url = reverse('oidc_authentication_callback')
+        request = self.factory.get(url, get_data)
+        request.session = {
+        }
+        callback_view = views.OIDCAuthenticationCallbackView.as_view()
+
+        response = callback_view(request)
+
+        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.url, '/failure')
+
+    @override_settings(LOGIN_REDIRECT_URL_FAILURE='/failure')
+    def test_get_auth_failure_tampered_session_state(self):
+        """Test authentication failure attempt for an inactive user."""
+        user = User.objects.create_user('example_username')
+        user.is_active = False
+        user.save()
+
+        get_data = {
+            'code': 'example_code',
+            'state': 'example_state'
+        }
+
+        url = reverse('oidc_authentication_callback')
+        request = self.factory.get(url, get_data)
+        request.session = {
+            'oidc_state': 'tampered_state'
+        }
+        callback_view = views.OIDCAuthenticationCallbackView.as_view()
+
+        with self.assertRaises(SuspiciousOperation) as context:
+            callback_view(request)
+
+        expected_error_message = 'Session `oidc_state` does not match the OIDC callback state'
+        self.assertEqual(context.exception.args, (expected_error_message,))
+
 
 class OIDCAuthorizationRequestViewTestCase(TestCase):
     def setUp(self):
@@ -119,6 +177,7 @@ def test_get(self, mock_random_string):
         mock_random_string.return_value = 'examplestring'
         url = reverse('oidc_authentication_init')
         request = self.factory.get(url)
+        request.session = dict()
         login_view = views.OIDCAuthenticationRequestView.as_view()
         response = login_view(request)
         self.assertEqual(response.status_code, 302)