diff options
Diffstat (limited to 'rest_framework/tests/authentication.py')
| -rw-r--r-- | rest_framework/tests/authentication.py | 177 |
1 files changed, 166 insertions, 11 deletions
diff --git a/rest_framework/tests/authentication.py b/rest_framework/tests/authentication.py index a02aef55..ddd61b63 100644 --- a/rest_framework/tests/authentication.py +++ b/rest_framework/tests/authentication.py @@ -1,5 +1,4 @@ from __future__ import unicode_literals -from django.core.urlresolvers import reverse from django.contrib.auth.models import User from django.http import HttpResponse from django.test import Client, TestCase @@ -8,23 +7,25 @@ from rest_framework import HTTP_HEADER_ENCODING from rest_framework import exceptions from rest_framework import permissions from rest_framework import status -from rest_framework.authtoken.models import Token from rest_framework.authentication import ( BaseAuthentication, TokenAuthentication, BasicAuthentication, SessionAuthentication, + OAuthAuthentication, OAuth2Authentication ) +from rest_framework.authtoken.models import Token from rest_framework.compat import patterns, url, include -from rest_framework.compat import oauth2_provider +from rest_framework.compat import oauth2_provider, oauth2_provider_models +from rest_framework.compat import oauth, oauth_provider from rest_framework.tests.utils import RequestFactory from rest_framework.views import APIView import json import base64 +import time import datetime - factory = RequestFactory() @@ -46,11 +47,12 @@ urlpatterns = patterns('', (r'^basic/$', MockView.as_view(authentication_classes=[BasicAuthentication])), (r'^token/$', MockView.as_view(authentication_classes=[TokenAuthentication])), (r'^auth-token/$', 'rest_framework.authtoken.views.obtain_auth_token'), + (r'^oauth/$', MockView.as_view(authentication_classes=[OAuthAuthentication])) ) if oauth2_provider is not None: - urlpatterns += patterns('', - url(r'^oauth2/', include('provider.oauth2.urls', namespace = 'oauth2')), + urlpatterns += patterns('', + url(r'^oauth2/', include('provider.oauth2.urls', namespace='oauth2')), url(r'^oauth2-test/$', MockView.as_view(authentication_classes=[OAuth2Authentication])), ) @@ -235,6 +237,159 @@ class IncorrectCredentialsTests(TestCase): self.assertEqual(response.data, {'detail': 'Bad credentials'}) +class OAuthTests(TestCase): + """OAuth 1.0a authentication""" + urls = 'rest_framework.tests.authentication' + + def setUp(self): + # these imports are here because oauth is optional and hiding them in try..except block or compat + # could obscure problems if something breaks + from oauth_provider.models import Consumer, Resource + from oauth_provider.models import Token as OAuthToken + from oauth_provider import consts + + self.consts = consts + + self.csrf_client = Client(enforce_csrf_checks=True) + self.username = 'john' + self.email = 'lennon@thebeatles.com' + self.password = 'password' + self.user = User.objects.create_user(self.username, self.email, self.password) + + self.CONSUMER_KEY = 'consumer_key' + self.CONSUMER_SECRET = 'consumer_secret' + self.TOKEN_KEY = "token_key" + self.TOKEN_SECRET = "token_secret" + + self.consumer = Consumer.objects.create(key=self.CONSUMER_KEY, secret=self.CONSUMER_SECRET, + name='example', user=self.user, status=self.consts.ACCEPTED) + + self.resource = Resource.objects.create(name="resource name", url="api/") + self.token = OAuthToken.objects.create(user=self.user, consumer=self.consumer, resource=self.resource, + token_type=OAuthToken.ACCESS, key=self.TOKEN_KEY, secret=self.TOKEN_SECRET, is_approved=True + ) + + def _create_authorization_header(self): + params = { + 'oauth_version': "1.0", + 'oauth_nonce': oauth.generate_nonce(), + 'oauth_timestamp': int(time.time()), + 'oauth_token': self.token.key, + 'oauth_consumer_key': self.consumer.key + } + + req = oauth.Request(method="GET", url="http://example.com", parameters=params) + + signature_method = oauth.SignatureMethod_PLAINTEXT() + req.sign_request(signature_method, self.consumer, self.token) + + return req.to_header()["Authorization"] + + def _create_authorization_url_parameters(self): + params = { + 'oauth_version': "1.0", + 'oauth_nonce': oauth.generate_nonce(), + 'oauth_timestamp': int(time.time()), + 'oauth_token': self.token.key, + 'oauth_consumer_key': self.consumer.key + } + + req = oauth.Request(method="GET", url="http://example.com", parameters=params) + + signature_method = oauth.SignatureMethod_PLAINTEXT() + req.sign_request(signature_method, self.consumer, self.token) + return dict(req) + + @unittest.skipUnless(oauth_provider, 'django-oauth-plus not installed') + @unittest.skipUnless(oauth, 'oauth2 not installed') + def test_post_form_passing_oauth(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('/oauth/', {'example': 'example'}, HTTP_AUTHORIZATION=auth) + self.assertEqual(response.status_code, 200) + + @unittest.skipUnless(oauth_provider, 'django-oauth-plus not installed') + @unittest.skipUnless(oauth, 'oauth2 not installed') + def test_post_form_repeated_nonce_failing_oauth(self): + """Ensure POSTing form over OAuth with repeated auth (same nonces and timestamp) credentials fails""" + auth = self._create_authorization_header() + response = self.csrf_client.post('/oauth/', {'example': 'example'}, HTTP_AUTHORIZATION=auth) + self.assertEqual(response.status_code, 200) + + # simulate reply attack auth header containes already used (nonce, timestamp) pair + response = self.csrf_client.post('/oauth/', {'example': 'example'}, HTTP_AUTHORIZATION=auth) + self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN)) + + @unittest.skipUnless(oauth_provider, 'django-oauth-plus not installed') + @unittest.skipUnless(oauth, 'oauth2 not installed') + def test_post_form_token_removed_failing_oauth(self): + """Ensure POSTing when there is no OAuth access token in db fails""" + self.token.delete() + auth = self._create_authorization_header() + response = self.csrf_client.post('/oauth/', {'example': 'example'}, HTTP_AUTHORIZATION=auth) + self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN)) + + @unittest.skipUnless(oauth_provider, 'django-oauth-plus not installed') + @unittest.skipUnless(oauth, 'oauth2 not installed') + def test_post_form_consumer_status_not_accepted_failing_oauth(self): + """Ensure POSTing when consumer status is anything other than ACCEPTED fails""" + for consumer_status in (self.consts.CANCELED, self.consts.PENDING, self.consts.REJECTED): + self.consumer.status = consumer_status + self.consumer.save() + + auth = self._create_authorization_header() + response = self.csrf_client.post('/oauth/', {'example': 'example'}, HTTP_AUTHORIZATION=auth) + self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN)) + + @unittest.skipUnless(oauth_provider, 'django-oauth-plus not installed') + @unittest.skipUnless(oauth, 'oauth2 not installed') + def test_post_form_with_request_token_failing_oauth(self): + """Ensure POSTing with unauthorized request token instead of access token fails""" + self.token.token_type = self.token.REQUEST + self.token.save() + + auth = self._create_authorization_header() + response = self.csrf_client.post('/oauth/', {'example': 'example'}, HTTP_AUTHORIZATION=auth) + self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN)) + + @unittest.skipUnless(oauth_provider, 'django-oauth-plus not installed') + @unittest.skipUnless(oauth, 'oauth2 not installed') + def test_post_form_with_urlencoded_parameters(self): + """Ensure POSTing with x-www-form-urlencoded auth parameters passes""" + params = self._create_authorization_url_parameters() + response = self.csrf_client.post('/oauth/', params) + self.assertEqual(response.status_code, 200) + + @unittest.skipUnless(oauth_provider, 'django-oauth-plus not installed') + @unittest.skipUnless(oauth, 'oauth2 not installed') + def test_get_form_with_url_parameters(self): + """Ensure GETing with auth in url parameters passes""" + params = self._create_authorization_url_parameters() + response = self.csrf_client.get('/oauth/', params) + self.assertEqual(response.status_code, 200) + + @unittest.skipUnless(oauth_provider, 'django-oauth-plus not installed') + @unittest.skipUnless(oauth, 'oauth2 not installed') + def test_post_hmac_sha1_signature_passes(self): + """Ensure POSTing using HMAC_SHA1 signature method passes""" + params = { + 'oauth_version': "1.0", + 'oauth_nonce': oauth.generate_nonce(), + 'oauth_timestamp': int(time.time()), + 'oauth_token': self.token.key, + 'oauth_consumer_key': self.consumer.key + } + + req = oauth.Request(method="POST", url="http://testserver/oauth/", parameters=params) + + signature_method = oauth.SignatureMethod_HMAC_SHA1() + req.sign_request(signature_method, self.consumer, self.token) + auth = req.to_header()["Authorization"] + + response = self.csrf_client.post('/oauth/', HTTP_AUTHORIZATION=auth) + self.assertEqual(response.status_code, 200) + + class OAuth2Tests(TestCase): """OAuth 2.0 authentication""" urls = 'rest_framework.tests.authentication' @@ -251,21 +406,21 @@ class OAuth2Tests(TestCase): self.ACCESS_TOKEN = "access_token" self.REFRESH_TOKEN = "refresh_token" - self.oauth2_client = oauth2_provider.models.Client.objects.create( - client_id=self.CLIENT_ID, + self.oauth2_client = oauth2_provider_models.Client.objects.create( + client_id=self.CLIENT_ID, client_secret=self.CLIENT_SECRET, redirect_uri='', client_type=0, - name='example', + name='example', user=None, ) - self.access_token = oauth2_provider.models.AccessToken.objects.create( + self.access_token = oauth2_provider_models.AccessToken.objects.create( token=self.ACCESS_TOKEN, client=self.oauth2_client, user=self.user, ) - self.refresh_token = oauth2_provider.models.RefreshToken.objects.create( + self.refresh_token = oauth2_provider_models.RefreshToken.objects.create( user=self.user, access_token=self.access_token, client=self.oauth2_client |