Access Control Backends

cms/envs/common.py

@@ -82,6 +82,8 @@
     # Theme to use when no site or site theme is defined,
     DEFAULT_SITE_THEME,
 
+    ACCESS_CONTROL_BACKENDS,
+
     # Default site to use if no site exists matching request headers
     SITE_ID,
 

lms/djangoapps/courseware/access.py

@@ -37,6 +37,7 @@
 from courseware.masquerade import get_masquerade_role, is_masquerading_as_student
 from lms.djangoapps.ccx.custom_exception import CCXLocatorValidationException
 from lms.djangoapps.ccx.models import CustomCourseForEdX
+from lms.lib.access_control_backends import access_control_backends
 from mobile_api.models import IgnoreMobileAvailableFlagConfig
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
 from openedx.core.djangoapps.external_auth.models import ExternalAuthMap
@@ -376,7 +377,12 @@ def can_see_about_page():
         'see_about_page': can_see_about_page,
     }
 
-    return _dispatch(checkers, action, user, courselike)
+    return access_control_backends.query(
+        action='course.{action}'.format(action=action),
+        user=user,
+        resource=courselike,
+        default_has_access=_dispatch(checkers, action, user, courselike),
+    )
 
 
 def _has_access_error_desc(user, action, descriptor, course_key):

lms/djangoapps/courseware/tests/test_access_control_backends_integration.py

@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+"""
+Integration tests for the access control framework with the Access Control Backends plugins.
+"""
+import datetime
+
+import ddt
+import pytz
+from mock import patch, Mock
+from opaque_keys.edx.locator import CourseLocator
+
+import courseware.access as access
+from lms.lib.access_control_backends import access_control_backends
+from student.tests.factories import CourseEnrollmentAllowedFactory, UserFactory
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+
+
+@ddt.ddt
+class AccessWithACLBackendsTestCase(ModuleStoreTestCase):
+    """
+    Integration tests for `access._has_access_course`.
+    """
+
+    def setUp(self):
+        """
+        Set up tests environment.
+        """
+        tomorrow = datetime.datetime.now(pytz.utc) + datetime.timedelta(days=1)
+        self.user = UserFactory.create()
+        self.course = Mock(
+            enrollment_domain='',
+            enrollment_end=tomorrow,
+            enrollment_start=tomorrow,
+            id=CourseLocator('edX', 'test', '2012_Fall'),
+        )
+        CourseEnrollmentAllowedFactory(email=self.user.email, course_id=self.course.id)
+
+    def test_has_access_with_no_acl_backends(self):
+        """
+        Ensure that the `access._has_access_course` queries the Access Control Backends.
+        """
+        assert access._has_access_course(self.user, 'enroll', self.course)
+
+    @ddt.data(False, True)
+    def test_has_access_with_acl_backends(self, backend_access):
+        """
+        Ensure that the `access._has_access_course` queries the Access Control Backends.
+        """
+        with patch.object(access_control_backends, 'query', Mock(return_value=backend_access)):
+            assert backend_access == access._has_access_course(self.user, 'enroll', self.course)

lms/envs/aws.py

@@ -378,6 +378,8 @@
     "/C=US/ST=Massachusetts/O=Massachusetts Institute of Technology/OU=Client CA v1/CN={0}/emailAddress={1}"
 )
 
+ACCESS_CONTROL_BACKENDS = ENV_TOKENS.get('ACCESS_CONTROL_BACKENDS', {})
+
 # Django CAS external authentication settings
 CAS_EXTRA_LOGIN_PARAMS = ENV_TOKENS.get("CAS_EXTRA_LOGIN_PARAMS", None)
 if FEATURES.get('AUTH_USE_CAS'):

lms/envs/common.py

@@ -653,6 +653,8 @@ def _add_microsite_dirs_to_default_template_engine(settings):
 STUDENT_FILEUPLOAD_MAX_SIZE = 4 * 1000 * 1000  # 4 MB
 MAX_FILEUPLOADS_PER_INPUT = 20
 
+ACCESS_CONTROL_BACKENDS = {}
+
 # Set request limits for maximum size of a request body and maximum number of GET/POST parameters. (>=Django 1.10)
 # Limits are currently disabled - but can be used for finer-grained denial-of-service protection.
 DATA_UPLOAD_MAX_MEMORY_SIZE = None

lms/lib/access_control_backends.py

@@ -0,0 +1,106 @@
+"""
+A plugin system for customizing access control in the platform.
+"""
+from importlib import import_module
+from lazy import lazy
+import logging
+import six
+
+from django.conf import settings
+
+log = logging.getLogger(__name__)
+
+
+class AccessControlBackends(object):
+    """
+    The access control backend service object.
+
+    Meant to be instantiated by this module, so use the `access_control_backends` object.
+    """
+    SUPPORTED_ACTIONS = {
+        'course.load',
+        'course.load_mobile',
+        'course.enroll',
+        'course.see_exists',
+        'course.staff',
+        'course.instructor',
+        'course.see_in_catalog',
+        'course.see_about_page',
+    }
+    UNSUPPORTED_ERROR_FMT = '`AccessControlBackends` does not support the action `{action}` yet'.format
+
+    @lazy
+    def backends(self):
+        """
+        Parse the access control backends settings and resolve the backend functions.
+
+        :return: a dictionary with the following format: {
+            ACTION_NAME: {
+                "FUNC": BACKEND_FUNCTION,
+                "OPTIONS": BACKEND_OPTIONS_DICT,
+            },
+            ANOTHER_ACTION_NAME: ...,
+        }
+        """
+        backends = settings.ACCESS_CONTROL_BACKENDS
+        resolved_backends = {}
+        for action, backend in six.iteritems(backends):
+            if action not in self.SUPPORTED_ACTIONS:
+                raise NotImplementedError(self.UNSUPPORTED_ERROR_FMT(action=action))
+
+            path = backend['NAME']
+            options = backend.get('OPTIONS', {})
+            try:
+                module, func_name = path.split(':', 1)
+                module = import_module(module)
+                func = getattr(module, func_name)
+                resolved_backends[action] = {
+                    'FUNC': func,
+                    'OPTIONS': options,
+                }
+            except Exception:
+                log.exception(
+                    'Something went wrong in reading the ACCESS_CONTROL_BACKENDS settings for `{action}`.'.format(
+                        action=action,
+                    )
+                )
+                raise
+
+        return resolved_backends
+
+    def query(self, action, user, resource, default_has_access):
+        """
+        Invoke an Access Control Backend.
+
+        :param action: currently supporting the course access actions in SUPPORTED_ACTIONS.
+        :param user: The User model object.
+        :param resource: The course/resource ID.
+        :param default_has_access: True/False What's the default Open edX access control.
+        :return: True/False whether the `user` can perform the `action` on the `resource` or not.
+        """
+        if action not in self.SUPPORTED_ACTIONS:
+            raise NotImplementedError(self.UNSUPPORTED_ERROR_FMT(action=action))
+
+        backend = self.backends.get(action)
+
+        if backend:
+            try:
+                backend_func = backend['FUNC']
+                return backend_func(
+                    user=user,
+                    resource=resource,
+                    default_has_access=default_has_access,
+                    options=backend['OPTIONS'],
+                )
+            except Exception:
+                log.exception(
+                    'Something went wrong in querying the access control backend for `{action}`.'.format(
+                        action=action,
+                    )
+                )
+                raise
+
+        return default_has_access
+
+
+access_control_backends = AccessControlBackends()

lms/lib/tests/test_access_control_backends.py

@@ -0,0 +1,177 @@
+"""
+Test cases for the pluggable access control system.
+"""
+import ddt
+import pytest
+
+from django.conf import settings
+from django.test.utils import override_settings
+from django.test import TestCase
+from mock import patch, Mock
+
+from lms.lib.access_control_backends import AccessControlBackends
+
+
+@ddt.ddt
+class AccessControlBackendsTests(TestCase):
+    """
+    Tests for the AccessControlBackends class.
+    """
+
+    def setUp(self):
+        """
+        Instantiate a new AccessControlBackends object.
+        """
+        self.acl_backends = AccessControlBackends()
+
+    def test_sanity_check(self):
+        """
+        Check that the settings are empty by default as well as no default backends are found.
+        """
+        assert settings.ACCESS_CONTROL_BACKENDS == {}
+        assert self.acl_backends.backends == {}
+
+    @ddt.data(
+        {
+            'first_config': {
+                'course.see_in_catalog': {
+                    'NAME': 'lms.lib:see_in_catalog_backend',
+                }
+            },
+            'second_config': {},
+            'expected_count': 1,
+        },
+        {
+            'first_config': {},
+            'second_config': {
+                'course.see_in_catalog': {
+                    'NAME': 'lms.lib:see_in_catalog_backend',
+                }
+            },
+            'expected_count': 0,
+        }
+    )
+    @ddt.unpack
+    @patch('lms.lib.see_in_catalog_backend', Mock(), create=True)
+    def test_backends_cache(self, first_config, second_config, expected_count):
+        """
+        Check the `@lazy` attribute behaviour.
+
+        Ensures that the first use backend property loads the configuration.
+        The second use of the property should use the cached results instead of re-reading the configs.
+        """
+        with override_settings(ACCESS_CONTROL_BACKENDS=first_config):
+            assert len(self.acl_backends.backends) == expected_count, 'Should read the correct configs'
+
+        with override_settings(ACCESS_CONTROL_BACKENDS=second_config):
+            assert len(self.acl_backends.backends) == expected_count, 'Should not read the configs but use the cache'
+
+    @override_settings(ACCESS_CONTROL_BACKENDS={
+        'course.see_in_catalog': {
+            'NAME': 'lms.lib:see_in_catalog_backend',
+            'OPTIONS': {
+                'dummy_option': 500,
+            },
+        }
+    })
+    @patch('lms.lib.see_in_catalog_backend', create=True)
+    def test_settings_with_options(self, mock_backend):
+        """
+        Test the happy scenario for a backend with options.
+        """
+        assert self.acl_backends.backends == {
+            'course.see_in_catalog': {
+                'FUNC': mock_backend,
+                'OPTIONS': {
+                    'dummy_option': 500,
+                },
+            }
+        }
+
+    @override_settings(ACCESS_CONTROL_BACKENDS={
+        'course.see_in_catalog': {
+            'NAME': 'lms.lib:see_in_catalog_backend',
+        }
+    })
+    @patch('lms.lib.access_control_backends.log')
+    def test_settings_with_missing_function(self, mock_log):
+        """
+        Check that the system fails explicitly on a missing function.
+        """
+        with pytest.raises(AttributeError):
+            _ = self.acl_backends.backends
+        mock_log.exception.assert_called_with(
+            'Something went wrong in reading the ACCESS_CONTROL_BACKENDS settings for `course.see_in_catalog`.'
+        )
+
+    @override_settings(ACCESS_CONTROL_BACKENDS={
+        'studio.create_course': {
+            'NAME': 'lms.lib:see_in_catalog_backend',
+        }
+    })
+    def test_settings_with_unknown_actions(self):
+        """
+        Ensure only supported actions can be used.
+
+        SUPPORTED_ACTIONS can be extended whenever needed.
+        """
+        with pytest.raises(NotImplementedError) as e:
+            _ = self.acl_backends.backends
+        assert e.match('`AccessControlBackends` does not support the action `studio.create_course` yet')
+
+    @override_settings(ACCESS_CONTROL_BACKENDS={
+        'course.enroll': {
+            'NAME': 'lms.lib:enroll_backend',
+            'OPTIONS': {
+                'dummy_option': 500,
+            },
+        }
+    })
+    @ddt.data(True, False)
+    def test_query_existing_backend(self, return_value):
+        """
+        Test a correctly working backend.
+        """
+        with patch('lms.lib.enroll_backend', create=True, return_value=return_value) as mock_backend:
+            assert not mock_backend.call_count
+            course = Mock()
+            user = Mock()
+            has_access = self.acl_backends.query('course.enroll', user, course, True)
+            assert has_access == return_value
+            mock_backend.assert_called_once_with(
+                user=user,
+                resource=course,
+                default_has_access=True,
+                options={
+                    'dummy_option': 500,
+                },
+            )
+
+    @override_settings(ACCESS_CONTROL_BACKENDS={})
+    @ddt.data(True, False)
+    def test_query_missing_backend(self, default_has_access):
+        """
+        Ensure that the `default_has_access` is used when querying an action without a plugged-in backend.
+        """
+        course = Mock()
+        user = Mock()
+        assert default_has_access == self.acl_backends.query('course.enroll', user, course, default_has_access)
+
+    @override_settings(ACCESS_CONTROL_BACKENDS={
+        'course.load': {
+            'NAME': 'lms.lib:load_backend',
+        }
+    })
+    @patch('lms.lib.load_backend', Mock(side_effect=ArithmeticError('Dividing by zero!')), create=True)
+    @patch('lms.lib.access_control_backends.log')
+    def test_query_broken_backend(self, mock_log):
+        """
+        Ensure a broken backend fails explicitly.
+        """
+        course = Mock()
+        user = Mock()
+        with pytest.raises(ArithmeticError):
+            self.acl_backends.query('course.load', user, course, True)
+        mock_log.exception.assert_called_once_with(
+            'Something went wrong in querying the access control backend for `course.load`.'
+        )

lms/templates/dashboard/_dashboard_certificate_information.html

@@ -45,7 +45,7 @@
   % else:
     <div class="message message-status ${status_css_class} is-shown">
       <p class="message-copy">${_("Your final grade:")}
-        <span class="grade-value">${"{0:.0f}%".format(float(cert_status['grade'])*100)}</span>.
+        <span class="grade-value">${"{0:.0f}%".format(float(cert_status.get('grade', 0))*100)}</span>.
 
         % if cert_status['status'] == 'notpassing':
           % if enrollment.mode != 'audit':