Generate API client credentials.
ECOM-3946
This commit is contained in:
@@ -1,17 +1,30 @@
|
||||
"""Decorators for API access management."""
|
||||
from functools import wraps
|
||||
|
||||
from django.core.urlresolvers import reverse
|
||||
from django.http import HttpResponseNotFound
|
||||
from django.shortcuts import redirect
|
||||
|
||||
from openedx.core.djangoapps.api_admin.models import ApiAccessConfig
|
||||
from openedx.core.djangoapps.api_admin.models import ApiAccessRequest, ApiAccessConfig
|
||||
|
||||
|
||||
def api_access_enabled_or_404(view):
|
||||
def api_access_enabled_or_404(view_func):
|
||||
"""If API access management feature is not enabled, return a 404."""
|
||||
@wraps(view)
|
||||
def wrapped_view(request, *args, **kwargs):
|
||||
@wraps(view_func)
|
||||
def wrapped_view(view_obj, *args, **kwargs):
|
||||
"""Wrapper for the view function."""
|
||||
if ApiAccessConfig.current().enabled:
|
||||
return view(request, *args, **kwargs)
|
||||
return view_func(view_obj, *args, **kwargs)
|
||||
return HttpResponseNotFound()
|
||||
return wrapped_view
|
||||
|
||||
|
||||
def require_api_access(view_func):
|
||||
"""If the requesting user does not have API access, bounce them to the request form."""
|
||||
@wraps(view_func)
|
||||
def wrapped_view(view_obj, *args, **kwargs):
|
||||
"""Wrapper for the view function."""
|
||||
if ApiAccessRequest.has_api_access(args[0].user):
|
||||
return view_func(view_obj, *args, **kwargs)
|
||||
return redirect(reverse('api_admin:api-request'))
|
||||
return wrapped_view
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from django.db import migrations, models
|
||||
from django.conf import settings
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('api_admin', '0004_auto_20160412_1506'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.AlterField(
|
||||
model_name='apiaccessrequest',
|
||||
name='user',
|
||||
field=models.OneToOneField(related_name='api_access_request', to=settings.AUTH_USER_MODEL),
|
||||
),
|
||||
]
|
||||
@@ -32,7 +32,7 @@ class ApiAccessRequest(TimeStampedModel):
|
||||
(DENIED, _('Denied')),
|
||||
(APPROVED, _('Approved')),
|
||||
)
|
||||
user = models.OneToOneField(User)
|
||||
user = models.OneToOneField(User, related_name='api_access_request')
|
||||
status = models.CharField(
|
||||
max_length=255,
|
||||
choices=STATUS_CHOICES,
|
||||
|
||||
@@ -1,12 +1,16 @@
|
||||
"""Factories for API management."""
|
||||
import factory
|
||||
from factory.django import DjangoModelFactory
|
||||
from oauth2_provider.models import get_application_model
|
||||
|
||||
from microsite_configuration.tests.factories import SiteFactory
|
||||
from openedx.core.djangoapps.api_admin.models import ApiAccessRequest
|
||||
from student.tests.factories import UserFactory
|
||||
|
||||
|
||||
Application = get_application_model() # pylint: disable=invalid-name
|
||||
|
||||
|
||||
class ApiAccessRequestFactory(DjangoModelFactory):
|
||||
"""Factory for ApiAccessRequest objects."""
|
||||
class Meta(object):
|
||||
@@ -14,3 +18,12 @@ class ApiAccessRequestFactory(DjangoModelFactory):
|
||||
|
||||
user = factory.SubFactory(UserFactory)
|
||||
site = factory.SubFactory(SiteFactory)
|
||||
|
||||
|
||||
class ApplicationFactory(DjangoModelFactory):
|
||||
"""Factory for OAuth Application objects."""
|
||||
class Meta(object):
|
||||
model = Application
|
||||
|
||||
authorization_grant_type = Application.GRANT_CLIENT_CREDENTIALS
|
||||
client_type = Application.CLIENT_CONFIDENTIAL
|
||||
|
||||
@@ -1,16 +1,22 @@
|
||||
#pylint: disable=missing-docstring
|
||||
import unittest
|
||||
|
||||
import ddt
|
||||
from django.conf import settings
|
||||
from django.core.urlresolvers import reverse
|
||||
from django.test import TestCase
|
||||
from django.test.utils import override_settings
|
||||
from oauth2_provider.models import get_application_model
|
||||
|
||||
from openedx.core.djangoapps.api_admin.models import ApiAccessRequest, ApiAccessConfig
|
||||
from openedx.core.djangoapps.api_admin.tests.factories import ApiAccessRequestFactory
|
||||
from openedx.core.djangoapps.api_admin.tests.factories import ApiAccessRequestFactory, ApplicationFactory
|
||||
from openedx.core.djangoapps.api_admin.tests.utils import VALID_DATA
|
||||
from student.tests.factories import UserFactory
|
||||
|
||||
|
||||
Application = get_application_model() # pylint: disable=invalid-name
|
||||
|
||||
|
||||
class ApiAdminTest(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
@@ -86,6 +92,8 @@ class ApiRequestViewTest(ApiAdminTest):
|
||||
|
||||
|
||||
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
|
||||
@override_settings(PLATFORM_NAME='edX')
|
||||
@ddt.ddt
|
||||
class ApiRequestStatusViewTest(ApiAdminTest):
|
||||
|
||||
def setUp(self):
|
||||
@@ -103,14 +111,35 @@ class ApiRequestStatusViewTest(ApiAdminTest):
|
||||
response = self.client.get(self.url)
|
||||
self.assertRedirects(response, reverse('api_admin:api-request'))
|
||||
|
||||
def test_get_with_request(self):
|
||||
@ddt.data(
|
||||
(ApiAccessRequest.APPROVED, 'Your request to access the edX Course Catalog API has been approved.'),
|
||||
(ApiAccessRequest.PENDING, 'Your request to access the edX Course Catalog API is being processed.'),
|
||||
(ApiAccessRequest.DENIED, 'Your request to access the edX Course Catalog API has been denied.'),
|
||||
)
|
||||
@ddt.unpack
|
||||
def test_get_with_request(self, status, expected):
|
||||
"""
|
||||
Verify that users who have requested access can see a message
|
||||
regarding their request status.
|
||||
"""
|
||||
ApiAccessRequestFactory(user=self.user)
|
||||
ApiAccessRequestFactory(user=self.user, status=status)
|
||||
response = self.client.get(self.url)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn(expected, response.content)
|
||||
|
||||
def test_get_with_existing_application(self):
|
||||
"""
|
||||
Verify that if the user has created their client credentials, they
|
||||
are shown on the status page.
|
||||
"""
|
||||
ApiAccessRequestFactory(user=self.user, status=ApiAccessRequest.APPROVED)
|
||||
application = ApplicationFactory(user=self.user)
|
||||
response = self.client.get(self.url)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
unicode_content = response.content.decode('utf-8')
|
||||
self.assertIn(application.client_secret, unicode_content) # pylint: disable=no-member
|
||||
self.assertIn(application.client_id, unicode_content) # pylint: disable=no-member
|
||||
self.assertIn(application.redirect_uris, unicode_content) # pylint: disable=no-member
|
||||
|
||||
def test_get_anonymous(self):
|
||||
"""Verify that users must be logged in to see the page."""
|
||||
@@ -124,6 +153,49 @@ class ApiRequestStatusViewTest(ApiAdminTest):
|
||||
response = self.client.get(self.url)
|
||||
self.assertEqual(response.status_code, 404)
|
||||
|
||||
@ddt.data(
|
||||
(ApiAccessRequest.APPROVED, True, True),
|
||||
(ApiAccessRequest.DENIED, True, False),
|
||||
(ApiAccessRequest.PENDING, True, False),
|
||||
(ApiAccessRequest.APPROVED, False, True),
|
||||
(ApiAccessRequest.DENIED, False, False),
|
||||
(ApiAccessRequest.PENDING, False, False),
|
||||
)
|
||||
@ddt.unpack
|
||||
def test_post(self, status, application_exists, new_application_created):
|
||||
"""
|
||||
Verify that posting the form creates an application if the user is
|
||||
approved, and does not otherwise. Also ensure that if the user
|
||||
already has an application, it is deleted before a new
|
||||
application is created.
|
||||
"""
|
||||
if application_exists:
|
||||
old_application = ApplicationFactory(user=self.user)
|
||||
ApiAccessRequestFactory(user=self.user, status=status)
|
||||
self.client.post(self.url, {
|
||||
'name': 'test.com',
|
||||
'redirect_uris': 'http://example.com'
|
||||
})
|
||||
applications = Application.objects.filter(user=self.user)
|
||||
if application_exists and new_application_created:
|
||||
self.assertEqual(applications.count(), 1)
|
||||
self.assertNotEqual(old_application, applications[0])
|
||||
elif application_exists:
|
||||
self.assertEqual(applications.count(), 1)
|
||||
self.assertEqual(old_application, applications[0])
|
||||
elif new_application_created:
|
||||
self.assertEqual(applications.count(), 1)
|
||||
else:
|
||||
self.assertEqual(applications.count(), 0)
|
||||
|
||||
def test_post_with_errors(self):
|
||||
ApiAccessRequestFactory(user=self.user, status=ApiAccessRequest.APPROVED)
|
||||
response = self.client.post(self.url, {
|
||||
'name': 'test.com',
|
||||
'redirect_uris': 'not a url'
|
||||
})
|
||||
self.assertIn('Enter a valid URL.', response.content)
|
||||
|
||||
|
||||
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
|
||||
class ApiTosViewTest(ApiAdminTest):
|
||||
|
||||
@@ -9,13 +9,19 @@ from django.utils.translation import ugettext as _
|
||||
from django.views.generic import View
|
||||
from django.views.generic.base import TemplateView
|
||||
from django.views.generic.edit import CreateView
|
||||
from edxmako.shortcuts import render_to_response
|
||||
from oauth2_provider.generators import generate_client_secret, generate_client_id
|
||||
from oauth2_provider.models import get_application_model
|
||||
from oauth2_provider.views import ApplicationRegistration
|
||||
|
||||
from edxmako.shortcuts import render_to_response
|
||||
from openedx.core.djangoapps.api_admin.decorators import require_api_access
|
||||
from openedx.core.djangoapps.api_admin.forms import ApiAccessRequestForm
|
||||
from openedx.core.djangoapps.api_admin.models import ApiAccessRequest
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
Application = get_application_model() # pylint: disable=invalid-name
|
||||
|
||||
|
||||
class ApiRequestView(CreateView):
|
||||
"""Form view for requesting API access."""
|
||||
@@ -38,23 +44,72 @@ class ApiRequestView(CreateView):
|
||||
return super(ApiRequestView, self).form_valid(form)
|
||||
|
||||
|
||||
class ApiRequestStatusView(View):
|
||||
class ApiRequestStatusView(ApplicationRegistration):
|
||||
"""View for confirming our receipt of an API request."""
|
||||
|
||||
def get(self, request):
|
||||
success_url = reverse_lazy('api_admin:api-status')
|
||||
|
||||
def get(self, request, form=None): # pylint: disable=arguments-differ
|
||||
"""
|
||||
If the user has not created an API request, redirect them to the
|
||||
request form. Otherwise, display the status of their API request.
|
||||
request form. Otherwise, display the status of their API
|
||||
request. We take `form` as an optional argument so that we can
|
||||
display validation errors correctly on the page.
|
||||
"""
|
||||
status = ApiAccessRequest.api_access_status(request.user)
|
||||
if status is None:
|
||||
if form is None:
|
||||
form = self.get_form_class()()
|
||||
|
||||
user = request.user
|
||||
try:
|
||||
api_request = ApiAccessRequest.objects.get(user=user)
|
||||
except ApiAccessRequest.DoesNotExist:
|
||||
return redirect(reverse('api_admin:api-request'))
|
||||
try:
|
||||
application = Application.objects.get(user=user)
|
||||
except Application.DoesNotExist:
|
||||
application = None
|
||||
|
||||
# We want to fill in a few fields ourselves, so remove them
|
||||
# from the form so that the user doesn't see them.
|
||||
for field in ('client_type', 'client_secret', 'client_id', 'authorization_grant_type'):
|
||||
form.fields.pop(field)
|
||||
|
||||
return render_to_response('api_admin/status.html', {
|
||||
'status': status,
|
||||
'api_support_link': _('TODO'),
|
||||
'status': api_request.status,
|
||||
'api_support_link': settings.API_DOCUMENTATION_URL,
|
||||
'api_support_email': settings.API_ACCESS_MANAGER_EMAIL,
|
||||
'form': form,
|
||||
'application': application,
|
||||
})
|
||||
|
||||
def get_form(self, form_class=None):
|
||||
form = super(ApiRequestStatusView, self).get_form(form_class)
|
||||
# Copy the data, since it's an immutable QueryDict.
|
||||
copied_data = form.data.copy()
|
||||
# Now set the fields that were removed earlier. We give them
|
||||
# confidential client credentials, and generate their client
|
||||
# ID and secret.
|
||||
copied_data.update({
|
||||
'authorization_grant_type': Application.GRANT_CLIENT_CREDENTIALS,
|
||||
'client_type': Application.CLIENT_CONFIDENTIAL,
|
||||
'client_secret': generate_client_secret(),
|
||||
'client_id': generate_client_id(),
|
||||
})
|
||||
form.data = copied_data
|
||||
return form
|
||||
|
||||
def form_valid(self, form):
|
||||
# Delete any existing applications if the user has decided to regenerate their credentials
|
||||
Application.objects.filter(user=self.request.user).delete()
|
||||
return super(ApiRequestStatusView, self).form_valid(form)
|
||||
|
||||
def form_invalid(self, form):
|
||||
return self.get(self.request, form)
|
||||
|
||||
@require_api_access
|
||||
def post(self, request):
|
||||
return super(ApiRequestStatusView, self).post(request)
|
||||
|
||||
|
||||
class ApiTosView(TemplateView):
|
||||
"""View to show the API Terms of Service."""
|
||||
|
||||
Reference in New Issue
Block a user