Add appropriate messages to OAuth2 401 responses.
Format responses as a JSON dict containing an error_code and a developer_message. user_message is not necessary, as these are API level errors, not seen by end users. MA-1900
This commit is contained in:
J. Cliff Dyer
@@ -1,10 +1,15 @@
|
||||
"""
|
||||
Tests for OAuth2. This module is copied from django-rest-framework-oauth (tests/test_authentication.py)
|
||||
and updated to use our subclass of OAuth2Authentication.
|
||||
Tests for OAuth2. This module is copied from django-rest-framework-oauth
|
||||
(tests/test_authentication.py) and updated to use our subclass of OAuth2Authentication.
|
||||
"""
|
||||
|
||||
from __future__ import unicode_literals
|
||||
import datetime
|
||||
from collections import namedtuple
|
||||
from datetime import datetime, timedelta
|
||||
import itertools
|
||||
import json
|
||||
|
||||
import ddt
|
||||
|
||||
from django.conf.urls import patterns, url, include
|
||||
from django.contrib.auth.models import User
|
||||
@@ -22,7 +27,7 @@ from rest_framework.views import APIView
|
||||
|
||||
from provider import scope, constants
|
||||
|
||||
from ..authentication import OAuth2AuthenticationAllowInactiveUser
|
||||
from .. import authentication
|
||||
|
||||
factory = APIRequestFactory() # pylint: disable=invalid-name
|
||||
|
||||
@@ -43,30 +48,35 @@ class MockView(APIView): # pylint: disable=missing-docstring
|
||||
# This is the a change we've made from the django-rest-framework-oauth version
|
||||
# of these tests. We're subclassing our custom OAuth2AuthenticationAllowInactiveUser
|
||||
# instead of OAuth2Authentication.
|
||||
class OAuth2AuthenticationDebug(OAuth2AuthenticationAllowInactiveUser): # pylint: disable=missing-docstring
|
||||
class OAuth2AuthenticationDebug(authentication.OAuth2AuthenticationAllowInactiveUser): # pylint: disable=missing-docstring
|
||||
allow_query_params_token = True
|
||||
|
||||
|
||||
urlpatterns = patterns(
|
||||
'',
|
||||
url(r'^oauth2/', include('provider.oauth2.urls', namespace='oauth2')),
|
||||
url(r'^oauth2-test/$', MockView.as_view(authentication_classes=[OAuth2AuthenticationAllowInactiveUser])),
|
||||
url(
|
||||
r'^oauth2-test/$',
|
||||
MockView.as_view(authentication_classes=[authentication.OAuth2AuthenticationAllowInactiveUser])
|
||||
),
|
||||
url(r'^oauth2-test-debug/$', MockView.as_view(authentication_classes=[OAuth2AuthenticationDebug])),
|
||||
url(
|
||||
r'^oauth2-with-scope-test/$',
|
||||
MockView.as_view(
|
||||
authentication_classes=[OAuth2AuthenticationAllowInactiveUser],
|
||||
authentication_classes=[authentication.OAuth2AuthenticationAllowInactiveUser],
|
||||
permission_classes=[permissions.TokenHasReadWriteScope]
|
||||
)
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@ddt.ddt
|
||||
class OAuth2Tests(TestCase):
|
||||
"""OAuth 2.0 authentication"""
|
||||
urls = 'openedx.core.lib.api.tests.test_authentication'
|
||||
|
||||
def setUp(self):
|
||||
super(OAuth2Tests, self).setUp()
|
||||
self.csrf_client = APIClient(enforce_csrf_checks=True)
|
||||
self.username = 'john'
|
||||
self.email = 'lennon@thebeatles.com'
|
||||
@@ -75,8 +85,8 @@ class OAuth2Tests(TestCase):
|
||||
|
||||
self.CLIENT_ID = 'client_key' # pylint: disable=invalid-name
|
||||
self.CLIENT_SECRET = 'client_secret' # pylint: disable=invalid-name
|
||||
self.ACCESS_TOKEN = "access_token" # pylint: disable=invalid-name
|
||||
self.REFRESH_TOKEN = "refresh_token" # pylint: disable=invalid-name
|
||||
self.ACCESS_TOKEN = 'access_token' # pylint: disable=invalid-name
|
||||
self.REFRESH_TOKEN = 'refresh_token' # pylint: disable=invalid-name
|
||||
|
||||
self.oauth2_client = oauth2_provider.oauth2.models.Client.objects.create(
|
||||
client_id=self.CLIENT_ID,
|
||||
@@ -111,51 +121,59 @@ class OAuth2Tests(TestCase):
|
||||
# edx-auth2-provider.
|
||||
scope.SCOPE_NAME_DICT = {'read': constants.READ, 'write': constants.WRITE}
|
||||
|
||||
def get_with_bearer_token(self, target_url, params=None, token=None):
|
||||
"""
|
||||
Make a GET request to the specified URL with an OAuth2 bearer token. If
|
||||
no token is provided, a valid token will be used. Query parameters can
|
||||
also be passed in if desired.
|
||||
"""
|
||||
auth = self._create_authorization_header(token)
|
||||
return self.csrf_client.get(target_url, params, HTTP_AUTHORIZATION=auth)
|
||||
|
||||
def post_with_bearer_token(self, target_url, token=None):
|
||||
"""
|
||||
Make a POST request to the specified URL with an OAuth2 bearer token. If
|
||||
no token is provided, a valid token will be used.
|
||||
"""
|
||||
auth = self._create_authorization_header(token)
|
||||
return self.csrf_client.post(target_url, HTTP_AUTHORIZATION=auth)
|
||||
|
||||
def check_error_codes(self, response, status_code, error_code):
|
||||
"""
|
||||
Ensure that the response has the appropriate HTTP status, and provides
|
||||
the expected error_code in the JSON response body.
|
||||
"""
|
||||
response_dict = json.loads(response.content)
|
||||
self.assertEqual(response.status_code, status_code)
|
||||
self.assertEqual(response_dict['error_code'], error_code)
|
||||
|
||||
def _create_authorization_header(self, token=None): # pylint: disable=missing-docstring
|
||||
return "Bearer {0}".format(token or self.access_token.token)
|
||||
if token is None:
|
||||
token = self.access_token.token
|
||||
return "Bearer {0}".format(token)
|
||||
|
||||
@ddt.data(None, {})
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_with_wrong_authorization_header_token_type_failing(self):
|
||||
def test_get_form_with_wrong_authorization_header_token_type_failing(self, params):
|
||||
"""Ensure that a wrong token type lead to the correct HTTP error status code"""
|
||||
auth = "Wrong token-type-obviously"
|
||||
response = self.csrf_client.get('/oauth2-test/', {}, HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
response = self.csrf_client.get('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
response = self.csrf_client.get(
|
||||
'/oauth2-test/',
|
||||
params,
|
||||
HTTP_AUTHORIZATION='Wrong token-type-obviously'
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_with_wrong_authorization_header_token_format_failing(self):
|
||||
"""Ensure that a wrong token format lead to the correct HTTP error status code"""
|
||||
auth = "Bearer wrong token format"
|
||||
response = self.csrf_client.get('/oauth2-test/', {}, HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
response = self.csrf_client.get('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_with_wrong_authorization_header_token_failing(self):
|
||||
"""Ensure that a wrong token lead to the correct HTTP error status code"""
|
||||
auth = "Bearer wrong-token"
|
||||
response = self.csrf_client.get('/oauth2-test/', {}, HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
response = self.csrf_client.get('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_with_wrong_authorization_header_token_missing(self):
|
||||
"""Ensure that a missing token lead to the correct HTTP error status code"""
|
||||
auth = "Bearer"
|
||||
response = self.csrf_client.get('/oauth2-test/', {}, HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
response = self.csrf_client.get('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 401)
|
||||
# If no Authorization header is provided that contains a bearer token,
|
||||
# authorization passes to the next registered authorization class, or
|
||||
# (in this case) to standard DRF fallback code, so no error_code is
|
||||
# provided (yet).
|
||||
self.assertNotIn('error_code', json.loads(response.content))
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_passing_auth(self):
|
||||
"""Ensure GETing form over OAuth with correct client credentials succeed"""
|
||||
auth = self._create_authorization_header()
|
||||
response = self.csrf_client.get('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
response = self.get_with_bearer_token('/oauth2-test/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_passing_auth_url_transport(self):
|
||||
@@ -164,72 +182,106 @@ class OAuth2Tests(TestCase):
|
||||
'/oauth2-test/',
|
||||
data={'access_token': self.access_token.token}
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_passing_auth_url_transport(self):
|
||||
"""Ensure GETing form over OAuth with correct client credentials in query succeed when DEBUG is True"""
|
||||
query = urlencode({'access_token': self.access_token.token})
|
||||
response = self.csrf_client.get('/oauth2-test-debug/?%s' % query)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_get_form_failing_auth_url_transport(self):
|
||||
"""Ensure GETing form over OAuth with correct client credentials in query fails when DEBUG is False"""
|
||||
query = urlencode({'access_token': self.access_token.token})
|
||||
response = self.csrf_client.get('/oauth2-test/?%s' % query)
|
||||
self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
|
||||
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||||
|
||||
# This case is handled directly by DRF so no error_code is provided (yet).
|
||||
self.assertNotIn('error_code', json.loads(response.content))
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_passing_auth(self):
|
||||
"""Ensure POSTing form over OAuth with correct credentials passes and does not require CSRF"""
|
||||
auth = self._create_authorization_header()
|
||||
response = self.csrf_client.post('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
response = self.post_with_bearer_token('/oauth2-test/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_token_removed_failing_auth(self):
|
||||
"""Ensure POSTing when there is no OAuth access token in db fails"""
|
||||
self.access_token.delete()
|
||||
auth = self._create_authorization_header()
|
||||
response = self.csrf_client.post('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
|
||||
response = self.post_with_bearer_token('/oauth2-test/')
|
||||
self.check_error_codes(
|
||||
response,
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
error_code=authentication.OAUTH2_TOKEN_ERROR_NONEXISTENT
|
||||
)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_with_refresh_token_failing_auth(self):
|
||||
"""Ensure POSTing with refresh token instead of access token fails"""
|
||||
auth = self._create_authorization_header(token=self.refresh_token.token)
|
||||
response = self.csrf_client.post('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
|
||||
response = self.post_with_bearer_token('/oauth2-test/', token=self.refresh_token.token)
|
||||
self.check_error_codes(
|
||||
response,
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
error_code=authentication.OAUTH2_TOKEN_ERROR_NONEXISTENT
|
||||
)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_with_expired_access_token_failing_auth(self):
|
||||
"""Ensure POSTing with expired access token fails with an 'Invalid token' error"""
|
||||
self.access_token.expires = datetime.datetime.now() - datetime.timedelta(seconds=10) # 10 seconds late
|
||||
"""Ensure POSTing with expired access token fails with a 'token_expired' error"""
|
||||
self.access_token.expires = datetime.now() - timedelta(seconds=10) # 10 seconds late
|
||||
self.access_token.save()
|
||||
auth = self._create_authorization_header()
|
||||
response = self.csrf_client.post('/oauth2-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
|
||||
self.assertIn('Invalid token', response.content)
|
||||
response = self.post_with_bearer_token('/oauth2-test/')
|
||||
self.check_error_codes(
|
||||
response,
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
error_code=authentication.OAUTH2_TOKEN_ERROR_EXPIRED
|
||||
)
|
||||
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_with_invalid_scope_failing_auth(self):
|
||||
"""Ensure POSTing with a readonly scope instead of a write scope fails"""
|
||||
read_only_access_token = self.access_token
|
||||
read_only_access_token.scope = oauth2_provider_scope.SCOPE_NAME_DICT['read']
|
||||
read_only_access_token.save()
|
||||
auth = self._create_authorization_header(token=read_only_access_token.token)
|
||||
response = self.csrf_client.get('/oauth2-with-scope-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
response = self.csrf_client.post('/oauth2-with-scope-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||
TokenErrorDDT = namedtuple('TokenErrorDDT', ['token', 'error_code'])
|
||||
|
||||
@ddt.data(
|
||||
*itertools.product(
|
||||
[None, {}],
|
||||
[
|
||||
TokenErrorDDT('wrong format', authentication.OAUTH2_TOKEN_ERROR_MALFORMED),
|
||||
TokenErrorDDT('wrong-token', authentication.OAUTH2_TOKEN_ERROR_NONEXISTENT),
|
||||
TokenErrorDDT('', authentication.OAUTH2_TOKEN_ERROR_NOT_PROVIDED),
|
||||
]
|
||||
)
|
||||
)
|
||||
@ddt.unpack
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_post_form_with_valid_scope_passing_auth(self):
|
||||
"""Ensure POSTing with a write scope succeed"""
|
||||
read_write_access_token = self.access_token
|
||||
read_write_access_token.scope = oauth2_provider_scope.SCOPE_NAME_DICT['write']
|
||||
read_write_access_token.save()
|
||||
auth = self._create_authorization_header(token=read_write_access_token.token)
|
||||
response = self.csrf_client.post('/oauth2-with-scope-test/', HTTP_AUTHORIZATION=auth)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
def test_response_for_get_request_with_bad_auth_token(self, http_params, token_error):
|
||||
response = self.get_with_bearer_token('/oauth2-test/', http_params, token=token_error.token)
|
||||
self.check_error_codes(
|
||||
response,
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
error_code=token_error.error_code
|
||||
)
|
||||
|
||||
@ddt.data(
|
||||
TokenErrorDDT('notatoken', authentication.OAUTH2_TOKEN_ERROR_NONEXISTENT),
|
||||
TokenErrorDDT('malformed token', authentication.OAUTH2_TOKEN_ERROR_MALFORMED),
|
||||
TokenErrorDDT('', authentication.OAUTH2_TOKEN_ERROR_NOT_PROVIDED),
|
||||
)
|
||||
def test_response_for_post_request_with_bad_auth_token(self, token_error):
|
||||
response = self.post_with_bearer_token('/oauth2-test/', token=token_error.token)
|
||||
self.check_error_codes(response, status_code=status.HTTP_401_UNAUTHORIZED, error_code=token_error.error_code)
|
||||
|
||||
ScopeStatusDDT = namedtuple('ScopeStatusDDT', ['scope', 'read_status', 'write_status'])
|
||||
|
||||
@ddt.data(
|
||||
ScopeStatusDDT('read', read_status=status.HTTP_200_OK, write_status=status.HTTP_403_FORBIDDEN),
|
||||
ScopeStatusDDT('write', status.HTTP_403_FORBIDDEN, status.HTTP_200_OK),
|
||||
)
|
||||
@unittest.skipUnless(oauth2_provider, 'django-oauth2-provider not installed')
|
||||
def test_responses_to_scoped_requests(self, scope_statuses):
|
||||
self.access_token.scope = oauth2_provider_scope.SCOPE_NAME_DICT[scope_statuses.scope]
|
||||
self.access_token.save()
|
||||
response = self.get_with_bearer_token('/oauth2-with-scope-test/', token=self.access_token.token)
|
||||
self.assertEqual(response.status_code, scope_statuses.read_status)
|
||||
response = self.post_with_bearer_token('/oauth2-with-scope-test/', token=self.access_token.token)
|
||||
self.assertEqual(response.status_code, scope_statuses.write_status)
|
||||
|
||||
69
openedx/core/lib/api/tests/test_exceptions.py
Normal file
69
openedx/core/lib/api/tests/test_exceptions.py
Normal file
@@ -0,0 +1,69 @@
|
||||
"""
|
||||
Test Custom Exceptions
|
||||
"""
|
||||
import ddt
|
||||
from django.test import TestCase
|
||||
from rest_framework import exceptions as drf_exceptions
|
||||
|
||||
from .. import exceptions
|
||||
|
||||
|
||||
@ddt.ddt
|
||||
class TestDictExceptionsAllowDictDetails(TestCase):
|
||||
"""
|
||||
Standard DRF exceptions coerce detail inputs to strings. We want to use
|
||||
dicts to allow better customization of error messages. Demonstrate that
|
||||
we can provide dictionaries as exception details, and that custom
|
||||
classes subclass the relevant DRF exceptions, to provide consistent
|
||||
exception catching behavior.
|
||||
"""
|
||||
|
||||
def test_drf_errors_coerce_strings(self):
|
||||
# Demonstrate the base issue we are trying to solve.
|
||||
exc = drf_exceptions.AuthenticationFailed({u'error_code': -1})
|
||||
self.assertEqual(exc.detail, u"{u'error_code': -1}")
|
||||
|
||||
@ddt.data(
|
||||
exceptions.AuthenticationFailed,
|
||||
exceptions.NotAuthenticated,
|
||||
exceptions.NotFound,
|
||||
exceptions.ParseError,
|
||||
exceptions.PermissionDenied,
|
||||
)
|
||||
def test_exceptions_allows_dict_detail(self, exception_class):
|
||||
exc = exception_class({u'error_code': -1})
|
||||
self.assertEqual(exc.detail, {u'error_code': -1})
|
||||
|
||||
def test_method_not_allowed_allows_dict_detail(self):
|
||||
exc = exceptions.MethodNotAllowed(u'POST', {u'error_code': -1})
|
||||
self.assertEqual(exc.detail, {u'error_code': -1})
|
||||
|
||||
def test_not_acceptable_allows_dict_detail(self):
|
||||
exc = exceptions.NotAcceptable({u'error_code': -1}, available_renderers=['application/json'])
|
||||
self.assertEqual(exc.detail, {u'error_code': -1})
|
||||
self.assertEqual(exc.available_renderers, ['application/json'])
|
||||
|
||||
|
||||
@ddt.ddt
|
||||
class TestDictExceptionSubclassing(TestCase):
|
||||
"""
|
||||
Custom exceptions should subclass standard DRF exceptions, so code that
|
||||
catches the DRF exceptions also catches ours.
|
||||
"""
|
||||
|
||||
@ddt.data(
|
||||
(exceptions.AuthenticationFailed, drf_exceptions.AuthenticationFailed),
|
||||
(exceptions.NotAcceptable, drf_exceptions.NotAcceptable),
|
||||
(exceptions.NotAuthenticated, drf_exceptions.NotAuthenticated),
|
||||
(exceptions.NotFound, drf_exceptions.NotFound),
|
||||
(exceptions.ParseError, drf_exceptions.ParseError),
|
||||
(exceptions.PermissionDenied, drf_exceptions.PermissionDenied),
|
||||
)
|
||||
@ddt.unpack
|
||||
def test_exceptions_subclass_drf_exceptions(self, exception_class, drf_exception_class):
|
||||
exc = exception_class({u'error_code': -1})
|
||||
self.assertIsInstance(exc, drf_exception_class)
|
||||
|
||||
def test_method_not_allowed_subclasses_drf_exception(self):
|
||||
exc = exceptions.MethodNotAllowed(u'POST', {u'error_code': -1})
|
||||
self.assertIsInstance(exc, drf_exceptions.MethodNotAllowed)
|
||||
Reference in New Issue
Block a user