BOM-1264: add third-party-auth scope and usage

common/djangoapps/third_party_auth/api/permissions.py

@@ -2,32 +2,90 @@
 Third party auth API related permissions
 """
 
+import logging
 
-from rest_framework import permissions
-
+from edx_django_utils.monitoring import set_custom_metric
+from edx_rest_framework_extensions.auth.jwt.decoder import decode_jwt_filters
+from edx_rest_framework_extensions.permissions import (
+    IsSuperuser,
+    JwtHasScope,
+    JwtRestrictedApplication,
+    NotJwtRestrictedApplication
+)
+from rest_condition import C
+from rest_framework.permissions import BasePermission
 from third_party_auth.models import ProviderApiPermissions
 
+from openedx.core.lib.api.permissions import ApiKeyHeaderPermission
+
+log = logging.getLogger(__name__)
+
 
-class ThirdPartyAuthProviderApiPermission(permissions.BasePermission):
+class ThirdPartyAuthProviderApiPermission(BasePermission):
     """
     Allow someone to access the view if they have valid OAuth client credential.
-    """
-    def __init__(self, provider_id):
-        """ Initialize the class with a provider_id """
-        self.provider_id = provider_id
 
+    Deprecated: Only works for DOP oauth applications. To be removed as part of DOPrecation.
+
+    """
     def has_permission(self, request, view):
         """
         Check if the OAuth client associated with auth token in current request has permission to access
         the information for provider
         """
-        if not request.auth or not self.provider_id:
+        provider_id = view.kwargs.get('provider_id')
+        if not request.auth or not provider_id:
             # doesn't have access token or no provider_id specified
             return False
 
         try:
-            ProviderApiPermissions.objects.get(client__pk=request.auth.client_id, provider_id=self.provider_id)
+            ProviderApiPermissions.objects.get(client__pk=request.auth.client_id, provider_id=provider_id)
         except ProviderApiPermissions.DoesNotExist:
             return False
 
+        set_custom_metric('deprecated_ThirdPartyAuthProviderApiPermission', True)
         return True
+
+
+class JwtHasTpaProviderFilterForRequestedProvider(BasePermission):
+    """
+    Ensures the JWT used to authenticate contains the appropriate tpa_provider
+    filter for the provider_id requested in the view.
+    """
+    message = 'JWT missing required tpa_provider filter.'
+
+    def has_permission(self, request, view):
+        """
+        Ensure that the provider_id kwarg provided to the view exists exists
+        in the tpa_provider filters in the JWT used to authenticate.
+        """
+        provider_id = view.kwargs.get('provider_id')
+        if not provider_id:
+            log.warning("Permission JwtHasTpaProviderFilterForRequestedProvider requires a view with provider_id.")
+            return False
+
+        jwt_filters = decode_jwt_filters(request.auth)
+        for filter_type, filter_value in jwt_filters:
+            if filter_type == 'tpa_provider' and filter_value == provider_id:
+                return True
+
+        log.warning(
+            "Permission JwtHasTpaProviderFilterForRequestedProvider: required filter tpa_provider:%s was not found.",
+            provider_id,
+        )
+        return False
+
+
+# TODO: Remove ApiKeyHeaderPermission. Check deprecated_api_key_header custom metric for active usage.
+_NOT_JWT_RESTRICTED_TPA_PERMISSIONS = (
+    C(NotJwtRestrictedApplication) &
+    (C(IsSuperuser) | ApiKeyHeaderPermission | ThirdPartyAuthProviderApiPermission)
+)
+_JWT_RESTRICTED_TPA_PERMISSIONS = (
+    C(JwtRestrictedApplication) &
+    JwtHasScope &
+    JwtHasTpaProviderFilterForRequestedProvider
+)
+TPA_PERMISSIONS = (
+    (_NOT_JWT_RESTRICTED_TPA_PERMISSIONS | _JWT_RESTRICTED_TPA_PERMISSIONS)
+)

common/djangoapps/third_party_auth/api/tests/test_permissions.py

@@ -7,10 +7,17 @@
 
 import ddt
 from django.conf import settings
-from mock import Mock
+from django.test import RequestFactory, TestCase
+from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
+from edx_rest_framework_extensions.auth.jwt.tests.utils import generate_jwt
+from mock import Mock, patch
+from rest_framework.authentication import SessionAuthentication
+from rest_framework.response import Response
 from rest_framework.test import APITestCase
+from rest_framework.views import APIView
+from student.tests.factories import UserFactory
 
-from third_party_auth.api.permissions import ThirdPartyAuthProviderApiPermission
+from third_party_auth.api.permissions import ThirdPartyAuthProviderApiPermission, TPA_PERMISSIONS
 from third_party_auth.tests.testutil import ThirdPartyAuthTestMixin
 
 IDP_SLUG_TESTSHIB = 'testshib'
@@ -21,12 +28,6 @@
 @unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
 class ThirdPartyAuthApiPermissionTest(ThirdPartyAuthTestMixin, APITestCase):
     """ Tests for third party auth API permission """
-    def setUp(self):
-        """ Create users and oauth client for use in the tests """
-        super(ThirdPartyAuthApiPermissionTest, self).setUp()
-
-        client = self.configure_oauth_client()
-        self.configure_api_permission(client, PROVIDER_ID_TESTSHIB)
 
     @ddt.data(
         (1, PROVIDER_ID_TESTSHIB, True),
@@ -37,20 +38,183 @@ def setUp(self):
     )
     @ddt.unpack
     def test_api_permission(self, client_pk, provider_id, expect):
+        dop_client = self.configure_oauth_dop_client()
+        self.configure_api_permission(dop_client, PROVIDER_ID_TESTSHIB)
+
         request = Mock()
         request.auth = Mock()
         request.auth.client_id = client_pk
+        view = Mock(kwargs={'provider_id': provider_id})
 
-        result = ThirdPartyAuthProviderApiPermission(provider_id).has_permission(request, None)
+        result = ThirdPartyAuthProviderApiPermission().has_permission(request, view)
         self.assertEqual(result, expect)
 
     def test_api_permission_unauthorized_client(self):
-        client = self.configure_oauth_client()
-        self.configure_api_permission(client, 'saml-anotherprovider')
+        dop_client = self.configure_oauth_dop_client()
+        self.configure_api_permission(dop_client, 'saml-anotherprovider')
 
         request = Mock()
         request.auth = Mock()
-        request.auth.client_id = client.pk
+        request.auth.client_id = dop_client.pk
+        view = Mock(kwargs={'provider_id': PROVIDER_ID_TESTSHIB})
 
-        result = ThirdPartyAuthProviderApiPermission(PROVIDER_ID_TESTSHIB).has_permission(request, None)
+        result = ThirdPartyAuthProviderApiPermission().has_permission(request, view)
         self.assertEqual(result, False)
+
+
+@ddt.ddt
+@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
+class ThirdPartyAuthPermissionTest(TestCase):
+    """ Tests for third party auth TPA_PERMISSIONS """
+
+    class SomeTpaClassView(APIView):
+        """view used to test TPA_permissions"""
+        authentication_classes = (JwtAuthentication, SessionAuthentication)
+        permission_classes = (TPA_PERMISSIONS,)
+        required_scopes = ['tpa:read']
+
+        def get(self, request, provider_id=None):
+            return Response(data="Success")
+
+    def _create_user(self, is_superuser=False):
+        return UserFactory(username='this_user', is_superuser=is_superuser)
+
+    def _create_request(self, auth_header=None):
+        url = '/'
+        extra = dict(HTTP_AUTHORIZATION=auth_header) if auth_header else dict()
+        return RequestFactory().get(url, **extra)
+
+    def _create_session(self, request, user):
+        request.user = user
+
+    def _create_jwt_header(self, user, is_restricted=False, scopes=None, filters=None):
+        token = generate_jwt(user, is_restricted=is_restricted, scopes=scopes, filters=filters)
+        return "JWT {}".format(token)
+
+    def test_anonymous_fails(self):
+        request = self._create_request()
+        response = self.SomeTpaClassView().dispatch(request)
+        self.assertEqual(response.status_code, 401)
+
+    def test_session_superuser_succeeds(self):
+        user = self._create_user(is_superuser=True)
+        request = self._create_request()
+        self._create_session(request, user)
+
+        response = self.SomeTpaClassView().dispatch(request)
+        self.assertEqual(response.status_code, 200)
+
+    def test_session_user_fails(self):
+        user = self._create_user()
+        request = self._create_request()
+        self._create_session(request, user)
+
+        response = self.SomeTpaClassView().dispatch(request)
+        self.assertEqual(response.status_code, 403)
+
+    @ddt.data(
+        # **** Unenforced ****
+        # unrestricted
+        dict(
+            is_enforced=False,
+            is_restricted=False,
+            expected_response=403,
+        ),
+
+        # restricted
+        dict(
+            is_enforced=False,
+            is_restricted=True,
+            expected_response=403,
+        ),
+
+        # **** Enforced ****
+        # unrestricted (for example, jwt cookies)
+        dict(
+            is_enforced=True,
+            is_restricted=False,
+            expected_response=403,
+        ),
+
+        # restricted (note: further test cases for scopes and filters are in tests below)
+        dict(
+            is_enforced=True,
+            is_restricted=True,
+            expected_response=403,
+        ),
+    )
+    @ddt.unpack
+    def test_jwt_without_scopes_and_filters(
+            self,
+            is_enforced,
+            is_restricted,
+            expected_response,
+    ):
+        # pylint: disable=line-too-long
+        # Note: Unenforced tests can be retired when rollout waffle switch `oauth2.enforce_jwt_scopes` is retired.
+        # See https://github.com/edx/edx-drf-extensions/blob/609e1dbaa98f476b36e50143de97732f2f6a9b4f/edx_rest_framework_extensions/config.py#L5
+        # pylint: enable=line-too-long
+        with patch('edx_rest_framework_extensions.permissions.waffle.switch_is_active') as mock_toggle:
+            mock_toggle.return_value = is_enforced
+            user = self._create_user()
+
+            auth_header = self._create_jwt_header(user, is_restricted=is_restricted)
+            request = self._create_request(
+                auth_header=auth_header,
+            )
+
+            response = self.SomeTpaClassView().dispatch(request)
+            self.assertEqual(response.status_code, expected_response)
+
+    @ddt.data(
+        # valid scopes
+        dict(scopes=['tpa:read'], expected_response=200),
+        dict(scopes=['tpa:read', 'another_scope'], expected_response=200),
+
+        # invalid scopes
+        dict(scopes=[], expected_response=403),
+        dict(scopes=['another_scope'], expected_response=403),
+    )
+    @ddt.unpack
+    def test_jwt_scopes(self, scopes, expected_response):
+        self._assert_jwt_enforced_restricted_case(
+            scopes=scopes,
+            filters=['tpa_provider:some_tpa_provider'],
+            expected_response=expected_response,
+        )
+
+    @ddt.data(
+        # valid provider filters
+        dict(
+            filters=['tpa_provider:some_tpa_provider', 'tpa_provider:another_tpa_provider'],
+            expected_response=200,
+        ),
+
+        # invalid provider filters
+        dict(
+            filters=['tpa_provider:another_tpa_provider'],
+            expected_response=403,
+        ),
+        dict(
+            filters=[],
+            expected_response=403,
+        ),
+    )
+    @ddt.unpack
+    def test_jwt_org_filters(self, filters, expected_response):
+        self._assert_jwt_enforced_restricted_case(
+            scopes=['tpa:read'],
+            filters=filters,
+            expected_response=expected_response,
+        )
+
+    def _assert_jwt_enforced_restricted_case(self, scopes, filters, expected_response):
+        with patch('edx_rest_framework_extensions.permissions.waffle.switch_is_active') as mock_toggle:
+            mock_toggle.return_value = True
+            user = self._create_user()
+
+            auth_header = self._create_jwt_header(user, is_restricted=True, scopes=scopes, filters=filters)
+            request = self._create_request(auth_header=auth_header)
+
+            response = self.SomeTpaClassView().dispatch(request, provider_id='some_tpa_provider')
+            self.assertEqual(response.status_code, expected_response)

common/djangoapps/third_party_auth/api/tests/test_views.py

@@ -23,6 +23,9 @@
 from third_party_auth.api.permissions import ThirdPartyAuthProviderApiPermission
 from third_party_auth.models import ProviderApiPermissions
 from third_party_auth.tests.testutil import ThirdPartyAuthTestMixin
+from third_party_auth.api.permissions import (JwtRestrictedApplication,
+                                              JwtHasScope,
+                                              JwtHasTpaProviderFilterForRequestedProvider)
 
 VALID_API_KEY = "i am a key"
 IDP_SLUG_TESTSHIB = 'testshib'
@@ -46,7 +49,7 @@ def get_mapping_data_by_usernames(usernames):
 class TpaAPITestCase(ThirdPartyAuthTestMixin, APITestCase):
     """ Base test class """
 
-    def setUp(self):
+    def setUp(self):  # pylint: disable=arguments-differ
         """ Create users for use in the tests """
         super(TpaAPITestCase, self).setUp()
 
@@ -234,8 +237,8 @@ class UserMappingViewAPITests(TpaAPITestCase):
     """
     @ddt.data(
         (VALID_API_KEY, PROVIDER_ID_TESTSHIB, 200, get_mapping_data_by_usernames(LINKED_USERS)),
-        ("i am an invalid key", PROVIDER_ID_TESTSHIB, 403, None),
-        (None, PROVIDER_ID_TESTSHIB, 403, None),
+        ("i am an invalid key", PROVIDER_ID_TESTSHIB, 401, None),
+        (None, PROVIDER_ID_TESTSHIB, 401, None),
         (VALID_API_KEY, 'non-existing-id', 404, []),
     )
     @ddt.unpack
@@ -336,7 +339,7 @@ def test_user_mappings_only_return_requested_idp_mapping_by_provider_id(self):
         (True, True, 200),
         (False, True, 200),
         (True, False, 200),
-        (False, False, 403)
+        (False, False, 401)
     )
     @ddt.unpack
     def test_user_mapping_permission_logic(self, api_key_permission, token_permission, expect):
@@ -346,6 +349,19 @@ def test_user_mapping_permission_logic(self, api_key_permission, token_permissio
                 response = self.client.get(url)
                 self.assertEqual(response.status_code, expect)
 
+    @ddt.data(
+        (True, 200),
+        (False, 401),
+    )
+    @ddt.unpack
+    def test_list_all_user_mappings_tpa_permission_logic(self, has_permission, expect):
+        url = reverse('third_party_auth_user_mapping_api', kwargs={'provider_id': PROVIDER_ID_TESTSHIB})
+        with patch.object(JwtHasTpaProviderFilterForRequestedProvider, 'has_permission', return_value=has_permission):
+            with patch.object(JwtRestrictedApplication, 'has_permission', return_value=has_permission):
+                with patch.object(JwtHasScope, 'has_permission', return_value=has_permission):
+                    response = self.client.get(url)
+                    self.assertEqual(response.status_code, expect)
+
     def _verify_response(self, response, expect_code, expect_result):
         """ verify the items in data_list exists in response and data_results matches results in response """
         self.assertEqual(response.status_code, expect_code)