Andal.LND/edx-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move auth_exchange from common to openedx/core.

Browse Source 
Move oauth_dispatch from lms to openedx/core.
This commit is contained in:
Nimisha Asthagiri
2016-10-05 21:08:11 -04:00
parent ce1eb237d1
commit 4c0f85b4d9
34 changed files with 47 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
openedx/core/djangoapps/auth_exchange/__init__.py Normal file
View File

109
openedx/core/djangoapps/auth_exchange/forms.py Normal file
View File

@@ -0,0 +1,109 @@
"""
Forms to support third-party to first-party OAuth 2.0 access token exchange
"""
from django.contrib.auth.models import User
from django.forms import CharField
from edx_oauth2_provider.constants import SCOPE_NAMES
import provider.constants
from provider.forms import OAuthForm, OAuthValidationError
from provider.oauth2.forms import ScopeChoiceField, ScopeMixin
from provider.oauth2.models import Client
from oauth2_provider.models import Application
from requests import HTTPError
from social.backends import oauth as social_oauth
from social.exceptions import AuthException
from third_party_auth import pipeline
class AccessTokenExchangeForm(ScopeMixin, OAuthForm):
"""Form for access token exchange endpoint"""
access_token = CharField(required=False)
scope = ScopeChoiceField(choices=SCOPE_NAMES, required=False)
client_id = CharField(required=False)
def __init__(self, request, oauth2_adapter, *args, **kwargs):
super(AccessTokenExchangeForm, self).__init__(*args, **kwargs)
self.request = request
self.oauth2_adapter = oauth2_adapter
def _require_oauth_field(self, field_name):
"""
Raise an appropriate OAuthValidationError error if the field is missing
"""
field_val = self.cleaned_data.get(field_name)
if not field_val:
raise OAuthValidationError(
{
"error": "invalid_request",
"error_description": "{} is required".format(field_name),
}
)
return field_val
def clean_access_token(self):
"""
Validates and returns the "access_token" field.
"""
return self._require_oauth_field("access_token")
def clean_client_id(self):
"""
Validates and returns the "client_id" field.
"""
return self._require_oauth_field("client_id")
def clean(self):
if self._errors:
return {}
backend = self.request.backend
if not isinstance(backend, social_oauth.BaseOAuth2):
raise OAuthValidationError(
{
"error": "invalid_request",
"error_description": "{} is not a supported provider".format(backend.name),
}
)
self.request.session[pipeline.AUTH_ENTRY_KEY] = pipeline.AUTH_ENTRY_LOGIN_API
client_id = self.cleaned_data["client_id"]
try:
client = self.oauth2_adapter.get_client(client_id=client_id)
except (Client.DoesNotExist, Application.DoesNotExist):
raise OAuthValidationError(
{
"error": "invalid_client",
"error_description": "{} is not a valid client_id".format(client_id),
}
)
if client.client_type not in [provider.constants.PUBLIC, Application.CLIENT_PUBLIC]:
raise OAuthValidationError(
{
# invalid_client isn't really the right code, but this mirrors
# https://github.com/edx/django-oauth2-provider/blob/edx/provider/oauth2/forms.py#L331
"error": "invalid_client",
"error_description": "{} is not a public client".format(client_id),
}
)
self.cleaned_data["client"] = client
user = None
try:
user = backend.do_auth(self.cleaned_data.get("access_token"), allow_inactive_user=True)
except (HTTPError, AuthException):
pass
if user and isinstance(user, User):
self.cleaned_data["user"] = user
else:
# Ensure user does not re-enter the pipeline
self.request.social_strategy.clean_partial_pipeline()
raise OAuthValidationError(
{
"error": "invalid_grant",
"error_description": "access_token is not valid",
}
)
return self.cleaned_data

3
openedx/core/djangoapps/auth_exchange/models.py Normal file
View File

@@ -0,0 +1,3 @@
"""
A models.py is required to make this an app (until we move to Django 1.7)
"""

0
openedx/core/djangoapps/auth_exchange/tests/__init__.py Normal file
View File

111
openedx/core/djangoapps/auth_exchange/tests/mixins.py Normal file
View File

@@ -0,0 +1,111 @@
"""
Mixins to facilitate testing OAuth connections to Django-OAuth-Toolkit or
Django-OAuth2-Provider.
"""
# pylint: disable=protected-access
from unittest import skip, expectedFailure
from django.test.client import RequestFactory
from openedx.core.djangoapps.oauth_dispatch import adapters
from openedx.core.djangoapps.oauth_dispatch.tests.constants import DUMMY_REDIRECT_URL
from ..views import DOTAccessTokenExchangeView
class DOPAdapterMixin(object):
"""
Mixin to rewire existing tests to use django-oauth2-provider (DOP) backend
Overwrites self.client_id, self.access_token, self.oauth2_adapter
"""
client_id = 'dop_test_client_id'
access_token = 'dop_test_access_token'
oauth2_adapter = adapters.DOPAdapter()
def create_public_client(self, user, client_id=None):
"""
Create an oauth client application that is public.
"""
return self.oauth2_adapter.create_public_client(
name='Test Public Client',
user=user,
client_id=client_id,
redirect_uri=DUMMY_REDIRECT_URL,
)
def create_confidential_client(self, user, client_id=None):
"""
Create an oauth client application that is confidential.
"""
return self.oauth2_adapter.create_confidential_client(
name='Test Confidential Client',
user=user,
client_id=client_id,
redirect_uri=DUMMY_REDIRECT_URL,
)
def get_token_response_keys(self):
"""
Return the set of keys provided when requesting an access token
"""
return {'access_token', 'token_type', 'expires_in', 'scope'}
class DOTAdapterMixin(object):
"""
Mixin to rewire existing tests to use django-oauth-toolkit (DOT) backend
Overwrites self.client_id, self.access_token, self.oauth2_adapter
"""
client_id = 'dot_test_client_id'
access_token = 'dot_test_access_token'
oauth2_adapter = adapters.DOTAdapter()
def create_public_client(self, user, client_id=None):
"""
Create an oauth client application that is public.
"""
return self.oauth2_adapter.create_public_client(
name='Test Public Application',
user=user,
client_id=client_id,
redirect_uri=DUMMY_REDIRECT_URL,
)
def create_confidential_client(self, user, client_id=None):
"""
Create an oauth client application that is confidential.
"""
return self.oauth2_adapter.create_confidential_client(
name='Test Confidential Application',
user=user,
client_id=client_id,
redirect_uri=DUMMY_REDIRECT_URL,
)
def get_token_response_keys(self):
"""
Return the set of keys provided when requesting an access token
"""
return {'access_token', 'refresh_token', 'token_type', 'expires_in', 'scope'}
def test_get_method(self):
# Dispatch routes all get methods to DOP, so we test this on the view
request_factory = RequestFactory()
request = request_factory.get('/oauth2/exchange_access_token/')
request.session = {}
view = DOTAccessTokenExchangeView.as_view()
response = view(request, backend='facebook')
self.assertEqual(response.status_code, 400)
@expectedFailure
def test_single_access_token(self):
# TODO: Single access tokens not supported yet for DOT (See MA-2122)
super(DOTAdapterMixin, self).test_single_access_token()
@skip("Not supported yet (See MA-2123)")
def test_scopes(self):
super(DOTAdapterMixin, self).test_scopes()

112
openedx/core/djangoapps/auth_exchange/tests/test_forms.py Normal file
View File

@@ -0,0 +1,112 @@
# pylint: disable=no-member
"""
Tests for OAuth token exchange forms
"""
import unittest
from django.conf import settings
from django.contrib.sessions.middleware import SessionMiddleware
from django.test import TestCase
from django.test.client import RequestFactory
import httpretty
from provider import scope
import social.apps.django_app.utils as social_utils
from third_party_auth.tests.utils import ThirdPartyOAuthTestMixinFacebook, ThirdPartyOAuthTestMixinGoogle
from ..forms import AccessTokenExchangeForm
from .utils import AccessTokenExchangeTestMixin
from .mixins import DOPAdapterMixin, DOTAdapterMixin
class AccessTokenExchangeFormTest(AccessTokenExchangeTestMixin):
"""
Mixin that defines test cases for AccessTokenExchangeForm
"""
def setUp(self):
super(AccessTokenExchangeFormTest, self).setUp()
self.request = RequestFactory().post("dummy_url")
redirect_uri = 'dummy_redirect_url'
SessionMiddleware().process_request(self.request)
self.request.social_strategy = social_utils.load_strategy(self.request)
# pylint: disable=no-member
self.request.backend = social_utils.load_backend(self.request.social_strategy, self.BACKEND, redirect_uri)
def _assert_error(self, data, expected_error, expected_error_description):
form = AccessTokenExchangeForm(request=self.request, oauth2_adapter=self.oauth2_adapter, data=data)
self.assertEqual(
form.errors,
{"error": expected_error, "error_description": expected_error_description}
)
self.assertNotIn("partial_pipeline", self.request.session)
def _assert_success(self, data, expected_scopes):
form = AccessTokenExchangeForm(request=self.request, oauth2_adapter=self.oauth2_adapter, data=data)
self.assertTrue(form.is_valid())
self.assertEqual(form.cleaned_data["user"], self.user)
self.assertEqual(form.cleaned_data["client"], self.oauth_client)
self.assertEqual(scope.to_names(form.cleaned_data["scope"]), expected_scopes)
# This is necessary because cms does not implement third party auth
@unittest.skipUnless(settings.FEATURES.get("ENABLE_THIRD_PARTY_AUTH"), "third party auth not enabled")
@httpretty.activate
class DOPAccessTokenExchangeFormTestFacebook(
DOPAdapterMixin,
AccessTokenExchangeFormTest,
ThirdPartyOAuthTestMixinFacebook,
TestCase,
):
"""
Tests for AccessTokenExchangeForm used with Facebook, tested against
django-oauth2-provider (DOP).
"""
pass
# This is necessary because cms does not implement third party auth
@unittest.skipUnless(settings.FEATURES.get("ENABLE_THIRD_PARTY_AUTH"), "third party auth not enabled")
@httpretty.activate
class DOTAccessTokenExchangeFormTestFacebook(
DOTAdapterMixin,
AccessTokenExchangeFormTest,
ThirdPartyOAuthTestMixinFacebook,
TestCase,
):
"""
Tests for AccessTokenExchangeForm used with Facebook, tested against
django-oauth-toolkit (DOT).
"""
pass
# This is necessary because cms does not implement third party auth
@unittest.skipUnless(settings.FEATURES.get("ENABLE_THIRD_PARTY_AUTH"), "third party auth not enabled")
@httpretty.activate
class DOPAccessTokenExchangeFormTestGoogle(
DOPAdapterMixin,
AccessTokenExchangeFormTest,
ThirdPartyOAuthTestMixinGoogle,
TestCase,
):
"""
Tests for AccessTokenExchangeForm used with Google, tested against
django-oauth2-provider (DOP).
"""
pass
# This is necessary because cms does not implement third party auth
@unittest.skipUnless(settings.FEATURES.get("ENABLE_THIRD_PARTY_AUTH"), "third party auth not enabled")
@httpretty.activate
class DOTAccessTokenExchangeFormTestGoogle(
DOTAdapterMixin,
AccessTokenExchangeFormTest,
ThirdPartyOAuthTestMixinGoogle,
TestCase,
):
"""
Tests for AccessTokenExchangeForm used with Google, tested against
django-oauth-toolkit (DOT).
"""
pass

195
openedx/core/djangoapps/auth_exchange/tests/test_views.py Normal file
View File

@@ -0,0 +1,195 @@
"""
Tests for OAuth token exchange views
"""
# pylint: disable=no-member
from datetime import timedelta
import json
import mock
import unittest
import ddt
from django.conf import settings
from django.core.urlresolvers import reverse
from django.test import TestCase
import httpretty
import provider.constants
from provider.oauth2.models import AccessToken, Client
from rest_framework.test import APIClient
from student.tests.factories import UserFactory
from third_party_auth.tests.utils import ThirdPartyOAuthTestMixinFacebook, ThirdPartyOAuthTestMixinGoogle
from .mixins import DOPAdapterMixin, DOTAdapterMixin
from .utils import AccessTokenExchangeTestMixin
@ddt.ddt
class AccessTokenExchangeViewTest(AccessTokenExchangeTestMixin):
"""
Mixin that defines test cases for AccessTokenExchangeView
"""
def setUp(self):
super(AccessTokenExchangeViewTest, self).setUp()
self.url = reverse("exchange_access_token", kwargs={"backend": self.BACKEND})
self.csrf_client = APIClient(enforce_csrf_checks=True)
def _assert_error(self, data, expected_error, expected_error_description):
response = self.csrf_client.post(self.url, data)
self.assertEqual(response.status_code, 400)
self.assertEqual(response["Content-Type"], "application/json")
self.assertEqual(
json.loads(response.content),
{u"error": expected_error, u"error_description": expected_error_description}
)
self.assertNotIn("partial_pipeline", self.client.session)
def _assert_success(self, data, expected_scopes):
response = self.csrf_client.post(self.url, data)
self.assertEqual(response.status_code, 200)
self.assertEqual(response["Content-Type"], "application/json")
content = json.loads(response.content)
self.assertEqual(set(content.keys()), self.get_token_response_keys())
self.assertEqual(content["token_type"], "Bearer")
self.assertLessEqual(
timedelta(seconds=int(content["expires_in"])),
provider.constants.EXPIRE_DELTA_PUBLIC
)
self.assertEqual(content["scope"], self.oauth2_adapter.normalize_scopes(expected_scopes))
token = self.oauth2_adapter.get_access_token(token_string=content["access_token"])
self.assertEqual(token.user, self.user)
self.assertEqual(self.oauth2_adapter.get_client_for_token(token), self.oauth_client)
self.assertEqual(self.oauth2_adapter.get_token_scope_names(token), expected_scopes)
def test_single_access_token(self):
def extract_token(response):
"""
Returns the access token from the response payload.
"""
return json.loads(response.content)["access_token"]
self._setup_provider_response(success=True)
for single_access_token in [True, False]:
with mock.patch(
"openedx.core.djangoapps.auth_exchange.views.constants.SINGLE_ACCESS_TOKEN",
single_access_token,
):
first_response = self.client.post(self.url, self.data)
second_response = self.client.post(self.url, self.data)
self.assertEqual(first_response.status_code, 200)
self.assertEqual(second_response.status_code, 200)
self.assertEqual(
extract_token(first_response) == extract_token(second_response),
single_access_token
)
def test_get_method(self):
response = self.client.get(self.url, self.data)
self.assertEqual(response.status_code, 400)
self.assertEqual(
json.loads(response.content),
{
"error": "invalid_request",
"error_description": "Only POST requests allowed.",
}
)
def test_invalid_provider(self):
url = reverse("exchange_access_token", kwargs={"backend": "invalid"})
response = self.client.post(url, self.data)
self.assertEqual(response.status_code, 404)
# This is necessary because cms does not implement third party auth
@unittest.skipUnless(settings.FEATURES.get("ENABLE_THIRD_PARTY_AUTH"), "third party auth not enabled")
@httpretty.activate
class DOPAccessTokenExchangeViewTestFacebook(
DOPAdapterMixin,
AccessTokenExchangeViewTest,
ThirdPartyOAuthTestMixinFacebook,
TestCase,
):
"""
Tests for AccessTokenExchangeView used with Facebook
"""
pass
@unittest.skipUnless(settings.FEATURES.get("ENABLE_THIRD_PARTY_AUTH"), "third party auth not enabled")
@httpretty.activate
class DOTAccessTokenExchangeViewTestFacebook(
DOTAdapterMixin,
AccessTokenExchangeViewTest,
ThirdPartyOAuthTestMixinFacebook,
TestCase,
):
"""
Rerun AccessTokenExchangeViewTestFacebook tests against DOT backend
"""
pass
# This is necessary because cms does not implement third party auth
@unittest.skipUnless(settings.FEATURES.get("ENABLE_THIRD_PARTY_AUTH"), "third party auth not enabled")
@httpretty.activate
class DOPAccessTokenExchangeViewTestGoogle(
DOPAdapterMixin,
AccessTokenExchangeViewTest,
ThirdPartyOAuthTestMixinGoogle,
TestCase,
):
"""
Tests for AccessTokenExchangeView used with Google using
django-oauth2-provider backend.
"""
pass
# This is necessary because cms does not implement third party auth
@unittest.skipUnless(settings.FEATURES.get("ENABLE_THIRD_PARTY_AUTH"), "third party auth not enabled")
@httpretty.activate
class DOTAccessTokenExchangeViewTestGoogle(
DOTAdapterMixin,
AccessTokenExchangeViewTest,
ThirdPartyOAuthTestMixinGoogle,
TestCase,
):
"""
Tests for AccessTokenExchangeView used with Google using
django-oauth-toolkit backend.
"""
pass
@unittest.skipUnless(settings.FEATURES.get("ENABLE_OAUTH2_PROVIDER"), "OAuth2 not enabled")
class TestLoginWithAccessTokenView(TestCase):
"""
Tests for LoginWithAccessTokenView
"""
def setUp(self):
super(TestLoginWithAccessTokenView, self).setUp()
self.user = UserFactory()
self.oauth2_client = Client.objects.create(client_type=provider.constants.CONFIDENTIAL)
def _verify_response(self, access_token, expected_status_code, expected_cookie_name=None):
"""
Calls the login_with_access_token endpoint and verifies the response given the expected values.
"""
url = reverse("login_with_access_token")
response = self.client.post(url, HTTP_AUTHORIZATION="Bearer {0}".format(access_token))
self.assertEqual(response.status_code, expected_status_code)
if expected_cookie_name:
self.assertIn(expected_cookie_name, response.cookies)
def test_success(self):
access_token = AccessToken.objects.create(
token="test_access_token",
client=self.oauth2_client,
user=self.user,
)
self._verify_response(access_token, expected_status_code=204, expected_cookie_name='sessionid')
self.assertEqual(int(self.client.session['_auth_user_id']), self.user.id)
def test_unauthenticated(self):
self._verify_response("invalid_token", expected_status_code=401)
self.assertNotIn("session_key", self.client.session)

102
openedx/core/djangoapps/auth_exchange/tests/utils.py Normal file
View File

@@ -0,0 +1,102 @@
"""
Test utilities for OAuth access token exchange
"""
from social.apps.django_app.default.models import UserSocialAuth
from third_party_auth.tests.utils import ThirdPartyOAuthTestMixin
class AccessTokenExchangeTestMixin(ThirdPartyOAuthTestMixin):
"""
A mixin to define test cases for access token exchange. The following
methods must be implemented by subclasses:
* _assert_error(data, expected_error, expected_error_description)
* _assert_success(data, expected_scopes)
"""
def setUp(self): # pylint: disable=arguments-differ
super(AccessTokenExchangeTestMixin, self).setUp()
# Initialize to minimal data
self.data = {
"access_token": self.access_token,
"client_id": self.client_id,
}
def _assert_error(self, _data, _expected_error, _expected_error_description):
"""
Given request data, execute a test and check that the expected error
was returned (along with any other appropriate assertions).
"""
raise NotImplementedError()
def _assert_success(self, data, expected_scopes):
"""
Given request data, execute a test and check that the expected scopes
were returned (along with any other appropriate assertions).
"""
raise NotImplementedError()
def _create_client(self):
"""
Create an oauth2 client application using class defaults.
"""
return self.create_public_client(self.user, self.client_id)
def test_minimal(self):
self._setup_provider_response(success=True)
self._assert_success(self.data, expected_scopes=[])
def test_scopes(self):
self._setup_provider_response(success=True)
self.data["scope"] = "profile email"
self._assert_success(self.data, expected_scopes=["profile", "email"])
def test_missing_fields(self):
for field in ["access_token", "client_id"]:
data = dict(self.data)
del data[field]
self._assert_error(data, "invalid_request", "{} is required".format(field))
def test_invalid_client(self):
self.data["client_id"] = "nonexistent_client"
self._assert_error(
self.data,
"invalid_client",
"nonexistent_client is not a valid client_id"
)
def test_confidential_client(self):
self.data['client_id'] += '_confidential'
self.oauth_client = self.create_confidential_client(self.user, self.data['client_id'])
self._assert_error(
self.data,
"invalid_client",
"{}_confidential is not a public client".format(self.client_id),
)
def test_inactive_user(self):
self.user.is_active = False
self.user.save() # pylint: disable=no-member
self._setup_provider_response(success=True)
self._assert_success(self.data, expected_scopes=[])
def test_invalid_acess_token(self):
self._setup_provider_response(success=False)
self._assert_error(self.data, "invalid_grant", "access_token is not valid")
def test_no_linked_user(self):
UserSocialAuth.objects.all().delete()
self._setup_provider_response(success=True)
self._assert_error(self.data, "invalid_grant", "access_token is not valid")
def test_user_automatically_linked_by_email(self):
UserSocialAuth.objects.all().delete()
self._setup_provider_response(success=True, email=self.user.email)
self._assert_success(self.data, expected_scopes=[])
def test_inactive_user_not_automatically_linked(self):
UserSocialAuth.objects.all().delete()
self._setup_provider_response(success=True, email=self.user.email)
self.user.is_active = False
self.user.save() # pylint: disable=no-member
self._assert_error(self.data, "invalid_grant", "access_token is not valid")

177
openedx/core/djangoapps/auth_exchange/views.py Normal file
View File

@@ -0,0 +1,177 @@
"""
Views to support exchange of authentication credentials.
The following are currently implemented:
1. AccessTokenExchangeView:
3rd party (social-auth) OAuth 2.0 access token -> 1st party (open-edx) OAuth 2.0 access token
2. LoginWithAccessTokenView:
1st party (open-edx) OAuth 2.0 access token -> session cookie
"""
# pylint: disable=abstract-method
from django.conf import settings
from django.contrib.auth import login
import django.contrib.auth as auth
from django.http import HttpResponse
from django.utils.decorators import method_decorator
from django.views.decorators.csrf import csrf_exempt
from edx_oauth2_provider.constants import SCOPE_VALUE_DICT
from oauth2_provider.settings import oauth2_settings
from oauth2_provider.views.base import TokenView as DOTAccessTokenView
from oauthlib.oauth2.rfc6749.tokens import BearerToken
from provider import constants
from provider.oauth2.views import AccessTokenView as DOPAccessTokenView
from rest_framework import permissions
from rest_framework.response import Response
from rest_framework.views import APIView
import social.apps.django_app.utils as social_utils
from openedx.core.djangoapps.auth_exchange.forms import AccessTokenExchangeForm
from openedx.core.djangoapps.oauth_dispatch import adapters
from openedx.core.lib.api.authentication import OAuth2AuthenticationAllowInactiveUser
class AccessTokenExchangeBase(APIView):
"""
View for token exchange from 3rd party OAuth access token to 1st party
OAuth access token.
"""
@method_decorator(csrf_exempt)
@method_decorator(social_utils.strategy("social:complete"))
def dispatch(self, *args, **kwargs):
return super(AccessTokenExchangeBase, self).dispatch(*args, **kwargs)
def get(self, request, _backend): # pylint: disable=arguments-differ
"""
Pass through GET requests without the _backend
"""
return super(AccessTokenExchangeBase, self).get(request)
def post(self, request, _backend): # pylint: disable=arguments-differ
"""
Handle POST requests to get a first-party access token.
"""
form = AccessTokenExchangeForm(request=request, oauth2_adapter=self.oauth2_adapter, data=request.POST) # pylint: disable=no-member
if not form.is_valid():
return self.error_response(form.errors) # pylint: disable=no-member
user = form.cleaned_data["user"]
scope = form.cleaned_data["scope"]
client = form.cleaned_data["client"]
return self.exchange_access_token(request, user, scope, client)
def exchange_access_token(self, request, user, scope, client):
"""
Exchange third party credentials for an edx access token, and return a
serialized access token response.
"""
if constants.SINGLE_ACCESS_TOKEN:
edx_access_token = self.get_access_token(request, user, scope, client) # pylint: disable=no-member
else:
edx_access_token = self.create_access_token(request, user, scope, client)
return self.access_token_response(edx_access_token) # pylint: disable=no-member
class DOPAccessTokenExchangeView(AccessTokenExchangeBase, DOPAccessTokenView):
"""
View for token exchange from 3rd party OAuth access token to 1st party
OAuth access token. Uses django-oauth2-provider (DOP) to manage access
tokens.
"""
oauth2_adapter = adapters.DOPAdapter()
class DOTAccessTokenExchangeView(AccessTokenExchangeBase, DOTAccessTokenView):
"""
View for token exchange from 3rd party OAuth access token to 1st party
OAuth access token. Uses django-oauth-toolkit (DOT) to manage access
tokens.
"""
oauth2_adapter = adapters.DOTAdapter()
def get(self, request, _backend):
return Response(status=400, data={
'error': 'invalid_request',
'error_description': 'Only POST requests allowed.',
})
def get_access_token(self, request, user, scope, client):
"""
TODO: MA-2122: Reusing access tokens is not yet supported for DOT.
Just return a new access token.
"""
return self.create_access_token(request, user, scope, client)
def create_access_token(self, request, user, scope, client):
"""
Create and return a new access token.
"""
_days = 24 * 60 * 60
token_generator = BearerToken(
expires_in=settings.OAUTH_EXPIRE_PUBLIC_CLIENT_DAYS * _days,
request_validator=oauth2_settings.OAUTH2_VALIDATOR_CLASS(),
)
self._populate_create_access_token_request(request, user, scope, client)
return token_generator.create_token(request, refresh_token=True)
def access_token_response(self, token):
"""
Wrap an access token in an appropriate response
"""
return Response(data=token)
def _populate_create_access_token_request(self, request, user, scope, client):
"""
django-oauth-toolkit expects certain non-standard attributes to
be present on the request object. This function modifies the
request object to match these expectations
"""
request.user = user
request.scopes = [SCOPE_VALUE_DICT[scope]]
request.client = client
request.state = None
request.refresh_token = None
request.extra_credentials = None
request.grant_type = client.authorization_grant_type
def error_response(self, form_errors):
"""
Return an error response consisting of the errors in the form
"""
return Response(status=400, data=form_errors)
class LoginWithAccessTokenView(APIView):
"""
View for exchanging an access token for session cookies
"""
authentication_classes = (OAuth2AuthenticationAllowInactiveUser,)
permission_classes = (permissions.IsAuthenticated,)
@staticmethod
def _get_path_of_arbitrary_backend_for_user(user):
"""
Return the path to the first found authentication backend that recognizes the given user.
"""
for backend_path in settings.AUTHENTICATION_BACKENDS:
backend = auth.load_backend(backend_path)
if backend.get_user(user.id):
return backend_path
@method_decorator(csrf_exempt)
def post(self, request):
"""
Handler for the POST method to this view.
"""
# The django login method stores the user's id in request.session[SESSION_KEY] and the
# path to the user's authentication backend in request.session[BACKEND_SESSION_KEY].
# The login method assumes the backend path had been previously stored in request.user.backend
# in the 'authenticate' call. However, not all authentication providers do so.
# So we explicitly populate the request.user.backend field here.
if not hasattr(request.user, 'backend'):
request.user.backend = self._get_path_of_arbitrary_backend_for_user(request.user)
login(request, request.user) # login generates and stores the user's cookies in the session
return HttpResponse(status=204) # cookies stored in the session are returned with the response

0
openedx/core/djangoapps/oauth_dispatch/__init__.py Normal file
View File

7
openedx/core/djangoapps/oauth_dispatch/adapters/__init__.py Normal file
View File

@@ -0,0 +1,7 @@
"""
Adapters to provide a common interface to django-oauth2-provider (DOP) and
django-oauth-toolkit (DOT).
"""
from .dop import DOPAdapter
from .dot import DOTAdapter

70
openedx/core/djangoapps/oauth_dispatch/adapters/dop.py Normal file
View File

@@ -0,0 +1,70 @@
"""
Adapter to isolate django-oauth2-provider dependencies
"""
from provider.oauth2 import models
from provider import constants, scope
class DOPAdapter(object):
"""
Standard interface for working with django-oauth2-provider
"""
backend = object()
def create_confidential_client(self, name, user, redirect_uri, client_id=None):
"""
Create an oauth client application that is confidential.
"""
return models.Client.objects.create(
name=name,
user=user,
client_id=client_id,
redirect_uri=redirect_uri,
client_type=constants.CONFIDENTIAL,
)
def create_public_client(self, name, user, redirect_uri, client_id=None):
"""
Create an oauth client application that is public.
"""
return models.Client.objects.create(
name=name,
user=user,
client_id=client_id,
redirect_uri=redirect_uri,
client_type=constants.PUBLIC,
)
def get_client(self, **filters):
"""
Get the oauth client application with the specified filters.
Wraps django's queryset.get() method.
"""
return models.Client.objects.get(**filters)
def get_client_for_token(self, token):
"""
Given an AccessToken object, return the associated client application.
"""
return token.client
def get_access_token(self, token_string):
"""
Given a token string, return the matching AccessToken object.
"""
return models.AccessToken.objects.get(token=token_string)
def normalize_scopes(self, scopes):
"""
Given a list of scopes, return a space-separated list of those scopes.
"""
return ' '.join(scopes)
def get_token_scope_names(self, token):
"""
Given an access token object, return its scopes.
"""
return scope.to_names(token.scope)

78
openedx/core/djangoapps/oauth_dispatch/adapters/dot.py Normal file
View File

@@ -0,0 +1,78 @@
"""
Adapter to isolate django-oauth-toolkit dependencies
"""
from oauth2_provider import models
class DOTAdapter(object):
"""
Standard interface for working with django-oauth-toolkit
"""
backend = object()
def create_confidential_client(self,
name,
user,
redirect_uri,
client_id=None,
authorization_grant_type=models.Application.GRANT_AUTHORIZATION_CODE):
"""
Create an oauth client application that is confidential.
"""
return models.Application.objects.create(
name=name,
user=user,
client_id=client_id,
client_type=models.Application.CLIENT_CONFIDENTIAL,
authorization_grant_type=authorization_grant_type,
redirect_uris=redirect_uri,
)
def create_public_client(self, name, user, redirect_uri, client_id=None):
"""
Create an oauth client application that is public.
"""
return models.Application.objects.create(
name=name,
user=user,
client_id=client_id,
client_type=models.Application.CLIENT_PUBLIC,
authorization_grant_type=models.Application.GRANT_PASSWORD,
redirect_uris=redirect_uri,
)
def get_client(self, **filters):
"""
Get the oauth client application with the specified filters.
Wraps django's queryset.get() method.
"""
return models.Application.objects.get(**filters)
def get_client_for_token(self, token):
"""
Given an AccessToken object, return the associated client application.
"""
return token.application
def get_access_token(self, token_string):
"""
Given a token string, return the matching AccessToken object.
"""
return models.AccessToken.objects.get(token=token_string)
def normalize_scopes(self, scopes):
"""
Given a list of scopes, return a space-separated list of those scopes.
"""
if not scopes:
scopes = ['default']
return ' '.join(scopes)
def get_token_scope_names(self, token):
"""
Given an access token object, return its scopes.
"""
return list(token.scopes)

73
openedx/core/djangoapps/oauth_dispatch/admin.py Normal file
View File

@@ -0,0 +1,73 @@
"""
Override admin configuration for django-oauth-toolkit
"""
from django.contrib.admin import ModelAdmin, site
from oauth2_provider import models
def reregister(model_class):
"""
Remove the existing admin, and register it anew with the given ModelAdmin
Usage:
@reregister(ModelClass)
class ModelClassAdmin(ModelAdmin):
pass
"""
def decorator(cls):
"""
The actual decorator that does the work.
"""
site.unregister(model_class)
site.register(model_class, cls)
return cls
return decorator
@reregister(models.AccessToken)
class DOTAccessTokenAdmin(ModelAdmin):
"""
Custom AccessToken Admin
"""
date_hierarchy = u'expires'
list_display = [u'token', u'user', u'application', u'expires']
list_filter = [u'application']
raw_id_fields = [u'user']
search_fields = [u'token', u'user__username']
@reregister(models.RefreshToken)
class DOTRefreshTokenAdmin(ModelAdmin):
"""
Custom AccessToken Admin
"""
list_display = [u'token', u'user', u'application', u'access_token']
list_filter = [u'application']
raw_id_fields = [u'user', u'access_token']
search_fields = [u'token', u'user__username', u'access_token__token']
@reregister(models.Application)
class DOTApplicationAdmin(ModelAdmin):
"""
Custom Application Admin
"""
list_display = [u'name', u'user', u'client_type', u'authorization_grant_type', u'client_id']
list_filter = [u'client_type', u'authorization_grant_type']
raw_id_fields = [u'user']
search_fields = [u'name', u'user__username', u'client_id']
@reregister(models.Grant)
class DOTGrantAdmin(ModelAdmin):
"""
Custom Grant Admin
"""
date_hierarchy = u'expires'
list_display = [u'code', u'user', u'application', u'expires']
list_filter = [u'application']
raw_id_fields = [u'user']
search_fields = [u'code', u'user__username']

14
openedx/core/djangoapps/oauth_dispatch/apps.py Normal file
View File

@@ -0,0 +1,14 @@
"""
Configure OAuthDispatch App
"""
from __future__ import absolute_import
from django.apps import AppConfig
class OAuthDispatchAppConfig(AppConfig):
"""
OAuthDispatch Configuration
"""
name = u'openedx.core.djangoapps.oauth_dispatch'

66
openedx/core/djangoapps/oauth_dispatch/dot_overrides.py Normal file
View File

@@ -0,0 +1,66 @@
"""
Classes that override default django-oauth-toolkit behavior
"""
from __future__ import unicode_literals
from django.contrib.auth import authenticate, get_user_model
from oauth2_provider.oauth2_validators import OAuth2Validator
class EdxOAuth2Validator(OAuth2Validator):
"""
Validator class that implements edX-specific custom behavior:
* It allows users to log in with their email or username.
* It does not require users to be active before logging in.
"""
def validate_user(self, username, password, client, request, *args, **kwargs):
"""
Authenticate users, but allow inactive users (with u.is_active == False)
to authenticate.
"""
user = self._authenticate(username=username, password=password)
if user is not None:
request.user = user
return True
return False
def _authenticate(self, username, password):
"""
Authenticate the user, allowing the user to identify themself either by
username or email
"""
authenticated_user = authenticate(username=username, password=password)
if authenticated_user is None:
UserModel = get_user_model() # pylint: disable=invalid-name
try:
email_user = UserModel.objects.get(email=username)
except UserModel.DoesNotExist:
authenticated_user = None
else:
authenticated_user = authenticate(username=email_user.username, password=password)
return authenticated_user
def save_bearer_token(self, token, request, *args, **kwargs):
"""
Ensure that access tokens issued via client credentials grant are associated with the owner of the
``Application``.
"""
grant_type = request.grant_type
user = request.user
if grant_type == 'client_credentials':
# Temporarily remove the grant type to avoid triggering the super method's code that removes request.user.
request.grant_type = None
# Ensure the tokens get associated with the correct user since DOT does not normally
# associate access tokens issued with the client_credentials grant to users.
request.user = request.client.user
super(EdxOAuth2Validator, self).save_bearer_token(token, request, *args, **kwargs)
# Restore the original request attributes
request.grant_type = grant_type
request.user = user

0
openedx/core/djangoapps/oauth_dispatch/tests/__init__.py Normal file
View File

5
openedx/core/djangoapps/oauth_dispatch/tests/constants.py Normal file
View File

@@ -0,0 +1,5 @@
"""
Constants for testing purposes
"""
DUMMY_REDIRECT_URL = u'https://example.com/edx/redirect'

40
openedx/core/djangoapps/oauth_dispatch/tests/factories.py Normal file
View File

@@ -0,0 +1,40 @@
# pylint: disable=missing-docstring
from datetime import datetime, timedelta
import factory
from factory.django import DjangoModelFactory
from factory.fuzzy import FuzzyText
import pytz
from oauth2_provider.models import Application, AccessToken, RefreshToken
from student.tests.factories import UserFactory
class ApplicationFactory(DjangoModelFactory):
class Meta(object):
model = Application
user = factory.SubFactory(UserFactory)
client_id = factory.Sequence(u'client_{0}'.format)
client_secret = 'some_secret'
client_type = 'confidential'
authorization_grant_type = 'Client credentials'
class AccessTokenFactory(DjangoModelFactory):
class Meta(object):
model = AccessToken
django_get_or_create = ('user', 'application')
token = FuzzyText(length=32)
expires = datetime.now(pytz.UTC) + timedelta(days=1)
class RefreshTokenFactory(DjangoModelFactory):
class Meta(object):
model = RefreshToken
django_get_or_create = ('user', 'application')
token = FuzzyText(length=32)

57
openedx/core/djangoapps/oauth_dispatch/tests/mixins.py Normal file
View File

@@ -0,0 +1,57 @@
"""
OAuth Dispatch test mixins
"""
import jwt
from django.conf import settings
from student.models import UserProfile, anonymous_id_for_user
class AccessTokenMixin(object):
""" Mixin for tests dealing with OAuth 2 access tokens. """
def assert_valid_jwt_access_token(self, access_token, user, scopes=None):
"""
Verify the specified JWT access token is valid, and belongs to the specified user.
Args:
access_token (str): JWT
user (User): User whose information is contained in the JWT payload.
Returns:
dict: Decoded JWT payload
"""
scopes = scopes or []
audience = settings.JWT_AUTH['JWT_AUDIENCE']
issuer = settings.JWT_AUTH['JWT_ISSUER']
payload = jwt.decode(
access_token,
settings.JWT_AUTH['JWT_SECRET_KEY'],
algorithms=[settings.JWT_AUTH['JWT_ALGORITHM']],
audience=audience,
issuer=issuer
)
expected = {
'aud': audience,
'iss': issuer,
'preferred_username': user.username,
'scopes': scopes,
'sub': anonymous_id_for_user(user, None),
}
if 'email' in scopes:
expected['email'] = user.email
if 'profile' in scopes:
try:
name = UserProfile.objects.get(user=user).name
except UserProfile.DoesNotExist:
name = None
expected['name'] = name
expected['administrator'] = user.is_staff
self.assertDictContainsSubset(expected, payload)
return payload

75
openedx/core/djangoapps/oauth_dispatch/tests/test_client_credentials.py Normal file
View File

@@ -0,0 +1,75 @@
""" Tests for OAuth 2.0 client credentials support. """
from __future__ import unicode_literals
import json
import unittest
from django.conf import settings
from django.core.urlresolvers import reverse
from django.test import TestCase
from edx_oauth2_provider.tests.factories import ClientFactory
from oauth2_provider.models import Application
from provider.oauth2.models import AccessToken
from student.tests.factories import UserFactory
from . import mixins
from .constants import DUMMY_REDIRECT_URL
from ..adapters import DOTAdapter
@unittest.skipUnless(settings.FEATURES.get("ENABLE_OAUTH2_PROVIDER"), "OAuth2 not enabled")
class ClientCredentialsTest(mixins.AccessTokenMixin, TestCase):
""" Tests validating the client credentials grant behavior. """
def setUp(self):
super(ClientCredentialsTest, self).setUp()
self.user = UserFactory()
def test_access_token(self):
""" Verify the client credentials grant can be used to obtain an access token whose default scopes allow access
to the user info endpoint.
"""
oauth_client = ClientFactory(user=self.user)
data = {
'grant_type': 'client_credentials',
'client_id': oauth_client.client_id,
'client_secret': oauth_client.client_secret
}
response = self.client.post(reverse('oauth2:access_token'), data)
self.assertEqual(response.status_code, 200)
access_token = json.loads(response.content)['access_token']
expected = AccessToken.objects.filter(client=oauth_client, user=self.user).first().token
self.assertEqual(access_token, expected)
headers = {
'HTTP_AUTHORIZATION': 'Bearer ' + access_token
}
response = self.client.get(reverse('oauth2:user_info'), **headers)
self.assertEqual(response.status_code, 200)
def test_jwt_access_token(self):
""" Verify the client credentials grant can be used to obtain a JWT access token. """
application = DOTAdapter().create_confidential_client(
name='test dot application',
user=self.user,
authorization_grant_type=Application.GRANT_CLIENT_CREDENTIALS,
redirect_uri=DUMMY_REDIRECT_URL,
client_id='dot-app-client-id',
)
scopes = ['read', 'write', 'email']
data = {
'grant_type': 'client_credentials',
'client_id': application.client_id,
'client_secret': application.client_secret,
'scope': ' '.join(scopes),
'token_type': 'jwt'
}
response = self.client.post(reverse('access_token'), data)
self.assertEqual(response.status_code, 200)
content = json.loads(response.content)
access_token = content['access_token']
self.assertEqual(content['scope'], data['scope'])
self.assert_valid_jwt_access_token(access_token, self.user, scopes)

77
openedx/core/djangoapps/oauth_dispatch/tests/test_dop_adapter.py Normal file
View File

@@ -0,0 +1,77 @@
"""
Tests for DOP Adapter
"""
from datetime import timedelta
import ddt
from django.test import TestCase
from django.utils.timezone import now
from provider.oauth2 import models
from provider import constants
from student.tests.factories import UserFactory
from ..adapters import DOPAdapter
from .constants import DUMMY_REDIRECT_URL
@ddt.ddt
class DOPAdapterTestCase(TestCase):
"""
Test class for DOPAdapter.
"""
adapter = DOPAdapter()
def setUp(self):
super(DOPAdapterTestCase, self).setUp()
self.user = UserFactory()
self.public_client = self.adapter.create_public_client(
name='public client',
user=self.user,
redirect_uri=DUMMY_REDIRECT_URL,
client_id='public-client-id',
)
self.confidential_client = self.adapter.create_confidential_client(
name='confidential client',
user=self.user,
redirect_uri=DUMMY_REDIRECT_URL,
client_id='confidential-client-id',
)
@ddt.data(
('confidential', constants.CONFIDENTIAL),
('public', constants.PUBLIC),
)
@ddt.unpack
def test_create_client(self, client_name, client_type):
client = getattr(self, '{}_client'.format(client_name))
self.assertIsInstance(client, models.Client)
self.assertEqual(client.client_id, '{}-client-id'.format(client_name))
self.assertEqual(client.client_type, client_type)
def test_get_client(self):
client = self.adapter.get_client(client_type=constants.CONFIDENTIAL)
self.assertIsInstance(client, models.Client)
self.assertEqual(client.client_type, constants.CONFIDENTIAL)
def test_get_client_not_found(self):
with self.assertRaises(models.Client.DoesNotExist):
self.adapter.get_client(client_id='not-found')
def test_get_client_for_token(self):
token = models.AccessToken(
user=self.user,
client=self.public_client,
)
self.assertEqual(self.adapter.get_client_for_token(token), self.public_client)
def test_get_access_token(self):
token = models.AccessToken.objects.create(
token='token-id',
client=self.public_client,
user=self.user,
expires=now() + timedelta(days=30),
)
self.assertEqual(self.adapter.get_access_token(token_string='token-id'), token)

76
openedx/core/djangoapps/oauth_dispatch/tests/test_dot_adapter.py Normal file
View File

@@ -0,0 +1,76 @@
"""
Tests for DOT Adapter
"""
from datetime import timedelta
import ddt
from django.test import TestCase
from django.utils.timezone import now
from oauth2_provider import models
from student.tests.factories import UserFactory
from ..adapters import DOTAdapter
from .constants import DUMMY_REDIRECT_URL
@ddt.ddt
class DOTAdapterTestCase(TestCase):
"""
Test class for DOTAdapter.
"""
adapter = DOTAdapter()
def setUp(self):
super(DOTAdapterTestCase, self).setUp()
self.user = UserFactory()
self.public_client = self.adapter.create_public_client(
name='public app',
user=self.user,
redirect_uri=DUMMY_REDIRECT_URL,
client_id='public-client-id',
)
self.confidential_client = self.adapter.create_confidential_client(
name='confidential app',
user=self.user,
redirect_uri=DUMMY_REDIRECT_URL,
client_id='confidential-client-id',
)
@ddt.data(
('confidential', models.Application.CLIENT_CONFIDENTIAL),
('public', models.Application.CLIENT_PUBLIC),
)
@ddt.unpack
def test_create_client(self, client_name, client_type):
client = getattr(self, '{}_client'.format(client_name))
self.assertIsInstance(client, models.Application)
self.assertEqual(client.client_id, '{}-client-id'.format(client_name))
self.assertEqual(client.client_type, client_type)
def test_get_client(self):
client = self.adapter.get_client(client_type=models.Application.CLIENT_CONFIDENTIAL)
self.assertIsInstance(client, models.Application)
self.assertEqual(client.client_type, models.Application.CLIENT_CONFIDENTIAL)
def test_get_client_not_found(self):
with self.assertRaises(models.Application.DoesNotExist):
self.adapter.get_client(client_id='not-found')
def test_get_client_for_token(self):
token = models.AccessToken(
user=self.user,
application=self.public_client,
)
self.assertEqual(self.adapter.get_client_for_token(token), self.public_client)
def test_get_access_token(self):
token = models.AccessToken.objects.create(
token='token-id',
application=self.public_client,
user=self.user,
expires=now() + timedelta(days=30),
)
self.assertEqual(self.adapter.get_access_token(token_string='token-id'), token)

66
openedx/core/djangoapps/oauth_dispatch/tests/test_dot_overrides.py Normal file
View File

@@ -0,0 +1,66 @@
"""
Test of custom django-oauth-toolkit behavior
"""
# pylint: disable=protected-access
from django.contrib.auth.models import User
from django.test import TestCase, RequestFactory
from ..dot_overrides import EdxOAuth2Validator
class AuthenticateTestCase(TestCase):
"""
Test that users can authenticate with either username or email
"""
def setUp(self):
super(AuthenticateTestCase, self).setUp()
self.user = User.objects.create_user(
username='darkhelmet',
password='12345',
email='darkhelmet@spaceball_one.org',
)
self.validator = EdxOAuth2Validator()
def test_authenticate_with_username(self):
user = self.validator._authenticate(username='darkhelmet', password='12345')
self.assertEqual(
self.user,
user
)
def test_authenticate_with_email(self):
user = self.validator._authenticate(username='darkhelmet@spaceball_one.org', password='12345')
self.assertEqual(
self.user,
user
)
class CustomValidationTestCase(TestCase):
"""
Test custom user validation works.
In particular, inactive users should be able to validate.
"""
def setUp(self):
super(CustomValidationTestCase, self).setUp()
self.user = User.objects.create_user(
username='darkhelmet',
password='12345',
email='darkhelmet@spaceball_one.org',
)
self.validator = EdxOAuth2Validator()
self.request_factory = RequestFactory()
def test_active_user_validates(self):
self.assertTrue(self.user.is_active)
request = self.request_factory.get('/')
self.assertTrue(self.validator.validate_user('darkhelmet', '12345', client=None, request=request))
def test_inactive_user_validates(self):
self.user.is_active = False
self.user.save()
request = self.request_factory.get('/')
self.assertTrue(self.validator.validate_user('darkhelmet', '12345', client=None, request=request))

45
openedx/core/djangoapps/oauth_dispatch/tests/test_factories.py Normal file
View File

@@ -0,0 +1,45 @@
# pylint: disable=missing-docstring
from django.test import TestCase
from oauth2_provider.models import Application, AccessToken, RefreshToken
from openedx.core.djangoapps.oauth_dispatch.tests import factories
from student.tests.factories import UserFactory
class TestClientFactory(TestCase):
def setUp(self):
super(TestClientFactory, self).setUp()
self.user = UserFactory.create()
def test_client_factory(self):
actual_application = factories.ApplicationFactory(user=self.user)
expected_application = Application.objects.get(user=self.user)
self.assertEqual(actual_application, expected_application)
class TestAccessTokenFactory(TestCase):
def setUp(self):
super(TestAccessTokenFactory, self).setUp()
self.user = UserFactory.create()
def test_access_token_client_factory(self):
application = factories.ApplicationFactory(user=self.user)
actual_access_token = factories.AccessTokenFactory(user=self.user, application=application)
expected_access_token = AccessToken.objects.get(user=self.user)
self.assertEqual(actual_access_token, expected_access_token)
class TestRefreshTokenFactory(TestCase):
def setUp(self):
super(TestRefreshTokenFactory, self).setUp()
self.user = UserFactory.create()
def test_refresh_token_factory(self):
application = factories.ApplicationFactory(user=self.user)
access_token = factories.AccessTokenFactory(user=self.user, application=application)
actual_refresh_token = factories.RefreshTokenFactory(
user=self.user, application=application, access_token=access_token
)
expected_refresh_token = RefreshToken.objects.get(user=self.user, access_token=access_token)
self.assertEqual(actual_refresh_token, expected_refresh_token)

430
openedx/core/djangoapps/oauth_dispatch/tests/test_views.py Normal file
View File

@@ -0,0 +1,430 @@
"""
Tests for Blocks Views
"""
import json
import ddt
from django.conf import settings
from django.test import RequestFactory, TestCase
from django.core.urlresolvers import reverse
import httpretty
from provider import constants
import unittest
from student.tests.factories import UserFactory
from third_party_auth.tests.utils import ThirdPartyOAuthTestMixin, ThirdPartyOAuthTestMixinGoogle
from .constants import DUMMY_REDIRECT_URL
from .. import adapters
if settings.FEATURES.get("ENABLE_OAUTH2_PROVIDER"):
from .. import views
from . import mixins
@unittest.skipUnless(settings.FEATURES.get("ENABLE_OAUTH2_PROVIDER"), "OAuth2 not enabled")
class _DispatchingViewTestCase(TestCase):
"""
Base class for tests that exercise DispatchingViews.
Subclasses need to define self.url.
"""
dop_adapter = adapters.DOPAdapter()
dot_adapter = adapters.DOTAdapter()
def setUp(self):
super(_DispatchingViewTestCase, self).setUp()
self.user = UserFactory()
self.dot_app = self.dot_adapter.create_public_client(
name='test dot application',
user=self.user,
redirect_uri=DUMMY_REDIRECT_URL,
client_id='dot-app-client-id',
)
self.dop_app = self.dop_adapter.create_public_client(
name='test dop client',
user=self.user,
redirect_uri=DUMMY_REDIRECT_URL,
client_id='dop-app-client-id',
)
def _post_request(self, user, client, token_type=None):
"""
Call the view with a POST request objectwith the appropriate format,
returning the response object.
"""
return self.client.post(self.url, self._post_body(user, client, token_type)) # pylint: disable=no-member
def _post_body(self, user, client, token_type=None):
"""
Return a dictionary to be used as the body of the POST request
"""
raise NotImplementedError()
@ddt.ddt
class TestAccessTokenView(mixins.AccessTokenMixin, _DispatchingViewTestCase):
"""
Test class for AccessTokenView
"""
def setUp(self):
self.url = reverse('access_token')
self.view_class = views.AccessTokenView
super(TestAccessTokenView, self).setUp()
def _post_body(self, user, client, token_type=None):
"""
Return a dictionary to be used as the body of the POST request
"""
body = {
'client_id': client.client_id,
'grant_type': 'password',
'username': user.username,
'password': 'test',
}
if token_type:
body['token_type'] = token_type
return body
@ddt.data('dop_app', 'dot_app')
def test_access_token_fields(self, client_attr):
client = getattr(self, client_attr)
response = self._post_request(self.user, client)
self.assertEqual(response.status_code, 200)
data = json.loads(response.content)
self.assertIn('access_token', data)
self.assertIn('expires_in', data)
self.assertIn('scope', data)
self.assertIn('token_type', data)
@ddt.data('dop_app', 'dot_app')
def test_jwt_access_token(self, client_attr):
client = getattr(self, client_attr)
response = self._post_request(self.user, client, token_type='jwt')
self.assertEqual(response.status_code, 200)
data = json.loads(response.content)
self.assertIn('expires_in', data)
self.assertEqual(data['token_type'], 'JWT')
self.assert_valid_jwt_access_token(data['access_token'], self.user, data['scope'].split(' '))
def test_dot_access_token_provides_refresh_token(self):
response = self._post_request(self.user, self.dot_app)
self.assertEqual(response.status_code, 200)
data = json.loads(response.content)
self.assertIn('refresh_token', data)
def test_dop_public_client_access_token(self):
response = self._post_request(self.user, self.dop_app)
self.assertEqual(response.status_code, 200)
data = json.loads(response.content)
self.assertNotIn('refresh_token', data)
@ddt.ddt
@httpretty.activate
class TestAccessTokenExchangeView(ThirdPartyOAuthTestMixinGoogle, ThirdPartyOAuthTestMixin, _DispatchingViewTestCase):
"""
Test class for AccessTokenExchangeView
"""
def setUp(self):
self.url = reverse('exchange_access_token', kwargs={'backend': 'google-oauth2'})
self.view_class = views.AccessTokenExchangeView
super(TestAccessTokenExchangeView, self).setUp()
def _post_body(self, user, client, token_type=None):
return {
'client_id': client.client_id,
'access_token': self.access_token,
}
@ddt.data('dop_app', 'dot_app')
def test_access_token_exchange_calls_dispatched_view(self, client_attr):
client = getattr(self, client_attr)
self.oauth_client = client
self._setup_provider_response(success=True)
response = self._post_request(self.user, client)
self.assertEqual(response.status_code, 200)
# pylint: disable=abstract-method
@ddt.ddt
class TestAuthorizationView(_DispatchingViewTestCase):
"""
Test class for AuthorizationView
"""
dop_adapter = adapters.DOPAdapter()
def setUp(self):
super(TestAuthorizationView, self).setUp()
self.user = UserFactory()
self.dot_app = self.dot_adapter.create_confidential_client(
name='test dot application',
user=self.user,
redirect_uri=DUMMY_REDIRECT_URL,
client_id='confidential-dot-app-client-id',
)
self.dop_app = self.dop_adapter.create_confidential_client(
name='test dop client',
user=self.user,
redirect_uri=DUMMY_REDIRECT_URL,
client_id='confidential-dop-app-client-id',
)
@ddt.data(
('dop', 'authorize'),
('dot', 'allow')
)
@ddt.unpack
def test_post_authorization_view(self, client_type, allow_field):
oauth_application = getattr(self, '{}_app'.format(client_type))
self.client.login(username=self.user.username, password='test')
response = self.client.post(
'/oauth2/authorize/',
{
'client_id': oauth_application.client_id,
'response_type': 'code',
'state': 'random_state_string',
'redirect_uri': DUMMY_REDIRECT_URL,
'scope': 'profile email',
allow_field: True,
},
follow=True,
)
check_response = getattr(self, '_check_{}_response'.format(client_type))
check_response(response)
def _check_dot_response(self, response):
"""
Check that django-oauth-toolkit gives an appropriate authorization response.
"""
# django-oauth-toolkit tries to redirect to the user's redirect URL
self.assertEqual(response.status_code, 404) # We used a non-existent redirect url.
expected_redirect_prefix = u'{}?'.format(DUMMY_REDIRECT_URL)
self._assert_startswith(self._redirect_destination(response), expected_redirect_prefix)
def _check_dop_response(self, response):
"""
Check that django-oauth2-provider gives an appropriate authorization response.
"""
# django-oauth-provider redirects to a confirmation page
self.assertRedirects(response, u'http://testserver/oauth2/authorize/confirm', target_status_code=200)
context = response.context_data
form = context['form']
self.assertIsNone(form['authorize'].value())
oauth_data = context['oauth_data']
self.assertEqual(oauth_data['redirect_uri'], DUMMY_REDIRECT_URL)
self.assertEqual(oauth_data['state'], 'random_state_string')
# TODO: figure out why it chooses this scope.
self.assertEqual(oauth_data['scope'], constants.READ_WRITE)
def _assert_startswith(self, string, prefix):
"""
Assert that the string starts with the specified prefix.
"""
self.assertTrue(string.startswith(prefix), u'{} does not start with {}'.format(string, prefix))
@staticmethod
def _redirect_destination(response):
"""
Return the final destination of the redirect chain in the response object
"""
return response.redirect_chain[-1][0]
@unittest.skipUnless(settings.FEATURES.get("ENABLE_OAUTH2_PROVIDER"), "OAuth2 not enabled")
class TestViewDispatch(TestCase):
"""
Test that the DispatchingView dispatches the right way.
"""
dop_adapter = adapters.DOPAdapter()
dot_adapter = adapters.DOTAdapter()
def setUp(self):
super(TestViewDispatch, self).setUp()
self.user = UserFactory()
self.view = views._DispatchingView() # pylint: disable=protected-access
self.dop_adapter.create_public_client(
name='',
user=self.user,
client_id='dop-id',
redirect_uri=DUMMY_REDIRECT_URL
)
self.dot_adapter.create_public_client(
name='',
user=self.user,
client_id='dot-id',
redirect_uri=DUMMY_REDIRECT_URL
)
def assert_is_view(self, view_candidate):
"""
Assert that a given object is a view. That is, it is callable, and
takes a request argument. Note: while technically, the request argument
could take any name, this assertion requires the argument to be named
`request`. This is good practice. You should do it anyway.
"""
_msg_base = u'{view} is not a view: {reason}'
msg_not_callable = _msg_base.format(view=view_candidate, reason=u'it is not callable')
msg_no_request = _msg_base.format(view=view_candidate, reason=u'it has no request argument')
self.assertTrue(hasattr(view_candidate, '__call__'), msg_not_callable)
args = view_candidate.func_code.co_varnames
self.assertTrue(args, msg_no_request)
self.assertEqual(args[0], 'request')
def _post_request(self, client_id):
"""
Return a request with the specified client_id in the body
"""
return RequestFactory().post('/', {'client_id': client_id})
def _get_request(self, client_id):
"""
Return a request with the specified client_id in the get parameters
"""
return RequestFactory().get('/?client_id={}'.format(client_id))
def test_dispatching_post_to_dot(self):
request = self._post_request('dot-id')
self.assertEqual(self.view.select_backend(request), self.dot_adapter.backend)
def test_dispatching_post_to_dop(self):
request = self._post_request('dop-id')
self.assertEqual(self.view.select_backend(request), self.dop_adapter.backend)
def test_dispatching_get_to_dot(self):
request = self._get_request('dot-id')
self.assertEqual(self.view.select_backend(request), self.dot_adapter.backend)
def test_dispatching_get_to_dop(self):
request = self._get_request('dop-id')
self.assertEqual(self.view.select_backend(request), self.dop_adapter.backend)
def test_dispatching_with_no_client(self):
request = self._post_request(None)
self.assertEqual(self.view.select_backend(request), self.dop_adapter.backend)
def test_dispatching_with_invalid_client(self):
request = self._post_request('abcesdfljh')
self.assertEqual(self.view.select_backend(request), self.dop_adapter.backend)
def test_get_view_for_dot(self):
view_object = views.AccessTokenView()
self.assert_is_view(view_object.get_view_for_backend(self.dot_adapter.backend))
def test_get_view_for_dop(self):
view_object = views.AccessTokenView()
self.assert_is_view(view_object.get_view_for_backend(self.dop_adapter.backend))
def test_get_view_for_no_backend(self):
view_object = views.AccessTokenView()
self.assertRaises(KeyError, view_object.get_view_for_backend, None)
class TestRevokeTokenView(_DispatchingViewTestCase): # pylint: disable=abstract-method
"""
Test class for RevokeTokenView
"""
def setUp(self):
self.login_with_access_token_url = reverse("login_with_access_token")
self.revoke_token_url = reverse('revoke_token')
self.access_token_url = reverse('access_token')
super(TestRevokeTokenView, self).setUp()
response = self.client.post(self.access_token_url, self.access_token_post_body_with_password())
access_token_data = json.loads(response.content)
self.access_token = access_token_data['access_token']
self.refresh_token = access_token_data['refresh_token']
def access_token_post_body_with_password(self):
"""
Returns a dictionary to be used as the body of the access_token
POST request with 'password' grant
"""
return {
'client_id': self.dot_app.client_id,
'grant_type': 'password',
'username': self.user.username,
'password': 'test',
}
def access_token_post_body_with_refresh_token(self, refresh_token):
"""
Returns a dictionary to be used as the body of the access_token
POST request with 'refresh_token' grant
"""
return {
'client_id': self.dot_app.client_id,
'grant_type': 'refresh_token',
'refresh_token': refresh_token,
}
def revoke_token_post_body(self, token):
"""
Returns a dictionary to be used as the body of the revoke_token POST request
"""
return {
'client_id': self.dot_app.client_id,
'token': token,
}
def login_with_access_token(self):
"""
Login with access token and return response
"""
return self.client.post(
self.login_with_access_token_url,
HTTP_AUTHORIZATION="Bearer {0}".format(self.access_token)
)
def _assert_access_token_is_valid(self):
"""
Asserts that oauth assigned access_token is valid and usable
"""
self.assertEqual(self.login_with_access_token().status_code, 204)
def _assert_access_token_invalidated(self):
"""
Asserts that oauth assigned access_token is not valid
"""
self.assertEqual(self.login_with_access_token().status_code, 401)
def _assert_refresh_token_invalidated(self):
"""
Asserts that oauth assigned refresh_token is not valid
"""
response = self.client.post(
self.access_token_url,
self.access_token_post_body_with_refresh_token(self.refresh_token)
)
self.assertEqual(response.status_code, 401)
def verify_revoke_token(self, token):
"""
Verifies access of token before and after revoking
"""
self._assert_access_token_is_valid()
response = self.client.post(self.revoke_token_url, self.revoke_token_post_body(token))
self.assertEqual(response.status_code, 200)
self._assert_access_token_invalidated()
self._assert_refresh_token_invalidated()
def test_revoke_refresh_token_dot(self):
"""
Tests invalidation/revoke of user tokens against refresh token for django-oauth-toolkit
"""
self.verify_revoke_token(self.refresh_token)
def test_revoke_access_token_dot(self):
"""
Tests invalidation/revoke of user access token for django-oauth-toolkit
"""
self.verify_revoke_token(self.access_token)

26
openedx/core/djangoapps/oauth_dispatch/urls.py Normal file
View File

@@ -0,0 +1,26 @@
"""
OAuth2 wrapper urls
"""
from django.conf import settings
from django.conf.urls import patterns, url
from django.views.decorators.csrf import csrf_exempt
from . import views
urlpatterns = patterns(
'',
url(r'^authorize/?$', csrf_exempt(views.AuthorizationView.as_view()), name='authorize'),
url(r'^access_token/?$', csrf_exempt(views.AccessTokenView.as_view()), name='access_token'),
url(r'^revoke_token/?$', csrf_exempt(views.RevokeTokenView.as_view()), name="revoke_token"),
)
if settings.FEATURES.get('ENABLE_THIRD_PARTY_AUTH'):
urlpatterns += (
url(
r'^exchange_access_token/(?P<backend>[^/]+)/$',
csrf_exempt(views.AccessTokenExchangeView.as_view()),
name='exchange_access_token',
),
)

134
openedx/core/djangoapps/oauth_dispatch/views.py Normal file
View File

@@ -0,0 +1,134 @@
"""
Views that dispatch processing of OAuth requests to django-oauth2-provider or
django-oauth-toolkit as appropriate.
"""
from __future__ import unicode_literals
import json
from django.views.generic import View
from edx_oauth2_provider import views as dop_views # django-oauth2-provider views
from oauth2_provider import models as dot_models, views as dot_views # django-oauth-toolkit
from openedx.core.djangoapps.auth_exchange import views as auth_exchange_views
from openedx.core.lib.token_utils import JwtBuilder
from . import adapters
class _DispatchingView(View):
"""
Base class that route views to the appropriate provider view. The default
behavior routes based on client_id, but this can be overridden by redefining
`select_backend()` if particular views need different behavior.
"""
# pylint: disable=no-member
dot_adapter = adapters.DOTAdapter()
dop_adapter = adapters.DOPAdapter()
def get_adapter(self, request):
"""
Returns the appropriate adapter based on the OAuth client linked to the request.
"""
if dot_models.Application.objects.filter(client_id=self._get_client_id(request)).exists():
return self.dot_adapter
else:
return self.dop_adapter
def dispatch(self, request, *args, **kwargs):
"""
Dispatch the request to the selected backend's view.
"""
backend = self.select_backend(request)
view = self.get_view_for_backend(backend)
return view(request, *args, **kwargs)
def select_backend(self, request):
"""
Given a request that specifies an oauth `client_id`, return the adapter
for the appropriate OAuth handling library. If the client_id is found
in a django-oauth-toolkit (DOT) Application, use the DOT adapter,
otherwise use the django-oauth2-provider (DOP) adapter, and allow the
calls to fail normally if the client does not exist.
"""
return self.get_adapter(request).backend
def get_view_for_backend(self, backend):
"""
Return the appropriate view from the requested backend.
"""
if backend == self.dot_adapter.backend:
return self.dot_view.as_view()
elif backend == self.dop_adapter.backend:
return self.dop_view.as_view()
else:
raise KeyError('Failed to dispatch view. Invalid backend {}'.format(backend))
def _get_client_id(self, request):
"""
Return the client_id from the provided request
"""
if request.method == u'GET':
return request.GET.get('client_id')
else:
return request.POST.get('client_id')
class AccessTokenView(_DispatchingView):
"""
Handle access token requests.
"""
dot_view = dot_views.TokenView
dop_view = dop_views.AccessTokenView
def dispatch(self, request, *args, **kwargs):
response = super(AccessTokenView, self).dispatch(request, *args, **kwargs)
if response.status_code == 200 and request.POST.get('token_type', '').lower() == 'jwt':
expires_in, scopes, user = self._decompose_access_token_response(request, response)
content = {
'access_token': JwtBuilder(user).build_token(scopes, expires_in),
'expires_in': expires_in,
'token_type': 'JWT',
'scope': ' '.join(scopes),
}
response.content = json.dumps(content)
return response
def _decompose_access_token_response(self, request, response):
""" Decomposes the access token in the request to an expiration date, scopes, and User. """
content = json.loads(response.content)
access_token = content['access_token']
scope = content['scope']
access_token_obj = self.get_adapter(request).get_access_token(access_token)
user = access_token_obj.user
scopes = scope.split(' ')
expires_in = content['expires_in']
return expires_in, scopes, user
class AuthorizationView(_DispatchingView):
"""
Part of the authorization flow.
"""
dop_view = dop_views.Capture
dot_view = dot_views.AuthorizationView
class AccessTokenExchangeView(_DispatchingView):
"""
Exchange a third party auth token.
"""
dop_view = auth_exchange_views.DOPAccessTokenExchangeView
dot_view = auth_exchange_views.DOTAccessTokenExchangeView
class RevokeTokenView(_DispatchingView):
"""
Dispatch to the RevokeTokenView of django-oauth-toolkit
"""
dot_view = dot_views.RevokeTokenView

2
openedx/core/lib/api/tests/test_authentication.py
View File

@@ -27,7 +27,7 @@ from rest_framework.views import APIView
from rest_framework_oauth import permissions
from rest_framework_oauth.compat import oauth2_provider, oauth2_provider_scope
from lms.djangoapps.oauth_dispatch import adapters
from openedx.core.djangoapps.oauth_dispatch import adapters
from openedx.core.lib.api import authentication
factory = APIRequestFactory() # pylint: disable=invalid-name

2
openedx/core/lib/tests/test_token_utils.py
View File

@@ -4,7 +4,7 @@ from django.test import TestCase
import jwt
from nose.plugins.attrib import attr
from lms.djangoapps.oauth_dispatch.tests import mixins
from openedx.core.djangoapps.oauth_dispatch.tests import mixins
from openedx.core.lib.token_utils import JwtBuilder
from student.tests.factories import UserFactory, UserProfileFactory