Improve auth handling of Locators

cms/djangoapps/auth/authz.py

@@ -1,3 +1,6 @@
+"""
+Studio authorization functions primarily for course creators, instructors, and staff
+"""
 #=======================================================================================================================
 #
 # This code is somewhat duplicative of access.py in the LMS. We will unify the code as a separate story
@@ -11,7 +14,8 @@
 from xmodule.modulestore import Location
 from xmodule.modulestore.locator import CourseLocator, Locator
 from xmodule.modulestore.django import loc_mapper
-from xmodule.modulestore.exceptions import InvalidLocationError
+from xmodule.modulestore.exceptions import InvalidLocationError, ItemNotFoundError
+import itertools
 
 
 # define a couple of simple roles, we just need ADMIN and EDITOR now for our purposes
@@ -26,7 +30,11 @@
 # of those two variables
 
 
-def get_course_groupname_for_role(location, role):
+def get_all_course_role_groupnames(location, role, use_filter=True):
+    '''
+    Get all of the possible groupnames for this role location pair. If use_filter==True,
+    only return the ones defined in the groups collection.
+    '''
     location = Locator.to_locator_or_location(location)
 
     # hack: check for existence of a group name in the legacy LMS format <role>_<course>
@@ -38,22 +46,46 @@ def get_course_groupname_for_role(location, role):
     except InvalidLocationError:  # will occur on old locations where location is not of category course
         pass
     if isinstance(location, Location):
+        # least preferred role_course format
         groupnames.append('{0}_{1}'.format(role, location.course))
+        try:
+            locator = loc_mapper().translate_location(location.course_id, location, False, False)
+            groupnames.append('{0}_{1}'.format(role, locator.course_id))
+        except (InvalidLocationError, ItemNotFoundError):
+            pass
     elif isinstance(location, CourseLocator):
         old_location = loc_mapper().translate_locator_to_location(location, get_course=True)
         if old_location:
+            # the slashified version of the course_id (myu/mycourse/myrun)
             groupnames.append('{0}_{1}'.format(role, old_location.course_id))
-
-    for groupname in groupnames:
-        if Group.objects.filter(name=groupname).exists():
-            return groupname
-    return groupnames[0]
+            # add the least desirable but sometimes occurring format.
+            groupnames.append('{0}_{1}'.format(role, old_location.course))
+    # filter to the ones which exist
+    default = groupnames[0]
+    if use_filter:
+        groupnames = [group for group in groupnames if Group.objects.filter(name=group).exists()]
+    return groupnames, default
 
 
-def get_users_in_course_group_by_role(location, role):
-    groupname = get_course_groupname_for_role(location, role)
-    (group, _created) = Group.objects.get_or_create(name=groupname)
-    return group.user_set.all()
+def get_course_groupname_for_role(location, role):
+    '''
+    Get the preferred used groupname for this role, location combo.
+    Preference order: 
+    * role_course_id (e.g., staff_myu.mycourse.myrun)
+    * role_old_course_id (e.g., staff_myu/mycourse/myrun)
+    * role_old_course (e.g., staff_mycourse)
+    '''
+    groupnames, default = get_all_course_role_groupnames(location, role)
+    return groupnames[0] if groupnames else default
+
+
+def get_course_role_users(course_locator, role):
+    '''
+    Get all of the users with the given role in the given course.
+    '''
+    groupnames, _ = get_all_course_role_groupnames(course_locator, role)
+    groups = [Group.objects.get(name=groupname) for groupname in groupnames]
+    return list(itertools.chain.from_iterable(group.user_set.all() for group in groups))
 
 
 def create_all_course_groups(creator, location):
@@ -65,11 +97,11 @@ def create_all_course_groups(creator, location):
 
 
 def create_new_course_group(creator, location, role):
-    groupname = get_course_groupname_for_role(location, role)
-    (group, created) = Group.objects.get_or_create(name=groupname)
-    if created:
-        group.save()
-
+    '''
+    Create the new course group always using the preferred name even if another form already exists.
+    '''
+    groupnames, __ = get_all_course_role_groupnames(location, role, use_filter=False)
+    group, __ = Group.objects.get_or_create(name=groupnames[0])
     creator.groups.add(group)
     creator.save()
 
@@ -82,41 +114,39 @@ def _delete_course_group(location):
     asserted permissions
     """
     # remove all memberships
-    instructors = Group.objects.get(name=get_course_groupname_for_role(location, INSTRUCTOR_ROLE_NAME))
-    for user in instructors.user_set.all():
-        user.groups.remove(instructors)
-        user.save()
-
-    staff = Group.objects.get(name=get_course_groupname_for_role(location, STAFF_ROLE_NAME))
-    for user in staff.user_set.all():
-        user.groups.remove(staff)
-        user.save()
+    for role in [INSTRUCTOR_ROLE_NAME, STAFF_ROLE_NAME]:
+        groupnames, _ = get_all_course_role_groupnames(location, role)
+        for groupname in groupnames:
+            group = Group.objects.get(name=groupname)
+            for user in group.user_set.all():
+                user.groups.remove(group)
+                user.save()
 
 
 def _copy_course_group(source, dest):
     """
     This is to be called only by either a command line code path or through an app which has already
     asserted permissions to do this action
     """
-    instructors = Group.objects.get(name=get_course_groupname_for_role(source, INSTRUCTOR_ROLE_NAME))
-    new_instructors_group = Group.objects.get(name=get_course_groupname_for_role(dest, INSTRUCTOR_ROLE_NAME))
-    for user in instructors.user_set.all():
-        user.groups.add(new_instructors_group)
-        user.save()
-
-    staff = Group.objects.get(name=get_course_groupname_for_role(source, STAFF_ROLE_NAME))
-    new_staff_group = Group.objects.get(name=get_course_groupname_for_role(dest, STAFF_ROLE_NAME))
-    for user in staff.user_set.all():
-        user.groups.add(new_staff_group)
-        user.save()
+    for role in [INSTRUCTOR_ROLE_NAME, STAFF_ROLE_NAME]:
+        groupnames, _ = get_all_course_role_groupnames(source, role)
+        for groupname in groupnames:
+            group = Group.objects.get(name=groupname)
+            new_group, _ = Group.objects.get_or_create(name=get_course_groupname_for_role(dest, INSTRUCTOR_ROLE_NAME))
+            for user in group.user_set.all():
+                user.groups.add(new_group)
+                user.save()
 
 
 def add_user_to_course_group(caller, user, location, role):
+    """
+    If caller is authorized, add the given user to the given course's role
+    """
     # only admins can add/remove other users
     if not is_user_in_course_group_role(caller, location, INSTRUCTOR_ROLE_NAME):
         raise PermissionDenied
 
-    group = Group.objects.get(name=get_course_groupname_for_role(location, role))
+    group, _ = Group.objects.get_or_create(name=get_course_groupname_for_role(location, role))
     return _add_user_to_group(user, group)
 
 
@@ -132,9 +162,7 @@ def add_user_to_creator_group(caller, user):
     if not caller.is_active or not caller.is_authenticated or not caller.is_staff:
         raise PermissionDenied
 
-    (group, created) = Group.objects.get_or_create(name=COURSE_CREATOR_GROUP_NAME)
-    if created:
-        group.save()
+    (group, _) = Group.objects.get_or_create(name=COURSE_CREATOR_GROUP_NAME)
     return _add_user_to_group(user, group)
 
 
@@ -152,6 +180,9 @@ def _add_user_to_group(user, group):
 
 
 def get_user_by_email(email):
+    """
+    Get the user whose email is the arg. Return None if no such user exists.
+    """
     user = None
     # try to look up user, return None if not found
     try:
@@ -163,13 +194,21 @@ def get_user_by_email(email):
 
 
 def remove_user_from_course_group(caller, user, location, role):
+    """
+    If caller is authorized, remove the given course x role authorization for user
+    """
     # only admins can add/remove other users
     if not is_user_in_course_group_role(caller, location, INSTRUCTOR_ROLE_NAME):
         raise PermissionDenied
 
     # see if the user is actually in that role, if not then we don't have to do anything
-    if is_user_in_course_group_role(user, location, role):
-        _remove_user_from_group(user, get_course_groupname_for_role(location, role))
+    groupnames, _ = get_all_course_role_groupnames(location, role)
+    for groupname in groupnames:
+        groups = user.groups.filter(name=groupname)
+        if groups:
+            # will only be one with that name
+            user.groups.remove(groups[0])
+            user.save()
 
 
 def remove_user_from_creator_group(caller, user):
@@ -195,11 +234,16 @@ def _remove_user_from_group(user, group_name):
 
 
 def is_user_in_course_group_role(user, location, role, check_staff=True):
+    """
+    Check whether the given user has the given role in this course. If check_staff
+    then give permission if the user is staff without doing a course-role query.
+    """
     if user.is_active and user.is_authenticated:
         # all "is_staff" flagged accounts belong to all groups
         if check_staff and user.is_staff:
             return True
-        return user.groups.filter(name=get_course_groupname_for_role(location, role)).exists()
+        groupnames, _ = get_all_course_role_groupnames(location, role)
+        return any(user.groups.filter(name=groupname).exists() for groupname in groupnames)
 
     return False
 

cms/djangoapps/contentstore/tests/test_permissions.py

@@ -0,0 +1,127 @@
+"""
+Test CRUD for authorization.
+"""
+from django.test.utils import override_settings
+from django.contrib.auth.models import User, Group
+
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+from contentstore.tests.modulestore_config import TEST_MODULESTORE
+from contentstore.tests.utils import AjaxEnabledTestClient
+from xmodule.modulestore.django import loc_mapper
+from xmodule.modulestore import Location
+from auth.authz import INSTRUCTOR_ROLE_NAME, STAFF_ROLE_NAME
+from auth import authz
+import copy
+from contentstore.views.access import has_access
+
+
+@override_settings(MODULESTORE=TEST_MODULESTORE)
+class TestCourseAccess(ModuleStoreTestCase):
+    """
+    Course-based access (as opposed to access of a non-course xblock)
+    """
+    def setUp(self):
+        """
+        Create a staff user and log them in (creating the client).
+
+        Create a pool of users w/o granting them any permissions
+        """
+        super(TestCourseAccess, self).setUp()
+        uname = 'testuser'
+        email = 'test+courses@edx.org'
+        password = 'foo'
+
+        # Create the use so we can log them in.
+        self.user = User.objects.create_user(uname, email, password)
+
+        # Note that we do not actually need to do anything
+        # for registration if we directly mark them active.
+        self.user.is_active = True
+        # Staff has access to view all courses
+        self.user.is_staff = True
+        self.user.save()
+
+        self.client = AjaxEnabledTestClient()
+        self.client.login(username=uname, password=password)
+
+        # create a course via the view handler which has a different strategy for permissions than the factory
+        self.course_location = Location(['i4x', 'myu', 'mydept.mycourse', 'course', 'myrun'])
+        self.course_locator = loc_mapper().translate_location(
+            self.course_location.course_id, self.course_location, False, True
+        )
+        self.client.ajax_post(
+            self.course_locator.url_reverse('course'),
+            {
+                'org': self.course_location.org, 
+                'number': self.course_location.course,
+                'display_name': 'My favorite course',
+                'run': self.course_location.name,
+            }
+        )
+
+        self.users = self._create_users()
+
+    def _create_users(self):
+        """
+        Create 8 users and return them
+        """
+        users = []
+        for i in range(8):
+            username = "user{}".format(i)
+            email = "test+user{}@edx.org".format(i)
+            user = User.objects.create_user(username, email, 'foo')
+            user.is_active = True
+            user.save()
+            users.append(user)
+        return users
+
+    def tearDown(self):
+        """
+        Reverse the setup
+        """
+        self.client.logout()
+        ModuleStoreTestCase.tearDown(self)
+
+    def test_get_all_users(self):
+        """
+        Test getting all authors for a course where their permissions run the gamut of allowed group
+        types.
+        """
+        # first check the groupname for the course creator.
+        self.assertTrue(
+            self.user.groups.filter(
+                name="{}_{}".format(INSTRUCTOR_ROLE_NAME, self.course_locator.course_id)
+            ).exists(),
+            "Didn't add creator as instructor."
+        )
+        users = copy.copy(self.users)
+        user_by_role = {}
+        # add the misc users to the course in different groups
+        for role in [INSTRUCTOR_ROLE_NAME, STAFF_ROLE_NAME]:
+            user_by_role[role] = []
+            groupnames, _ = authz.get_all_course_role_groupnames(self.course_locator, role)
+            for groupname in groupnames:
+                group, _ = Group.objects.get_or_create(name=groupname)
+                user = users.pop()
+                user_by_role[role].append(user)
+                user.groups.add(group)
+                user.save()
+                self.assertTrue(has_access(user, self.course_locator), "{} does not have access".format(user))
+                self.assertTrue(has_access(user, self.course_location), "{} does not have access".format(user))
+
+        response = self.client.get_html(self.course_locator.url_reverse('course_team'))
+        for role in [INSTRUCTOR_ROLE_NAME, STAFF_ROLE_NAME]:
+            for user in user_by_role[role]:
+                self.assertContains(response, user.email)
+
+        # test copying course permissions
+        copy_course_location = Location(['i4x', 'copyu', 'copydept.mycourse', 'course', 'myrun'])
+        copy_course_locator = loc_mapper().translate_location(
+            copy_course_location.course_id, copy_course_location, False, True
+        )
+        # pylint: disable=protected-access
+        authz._copy_course_group(self.course_locator, copy_course_locator)
+        for role in [INSTRUCTOR_ROLE_NAME, STAFF_ROLE_NAME]:
+            for user in user_by_role[role]:
+                self.assertTrue(has_access(user, copy_course_locator), "{} no copy access".format(user))
+                self.assertTrue(has_access(user, copy_course_location), "{} no copy access".format(user))

cms/djangoapps/contentstore/tests/test_users.py

@@ -29,8 +29,8 @@ def setUp(self):
         self.detail_url = self.location.url_reverse('course_team', self.ext_user.email)
         self.inactive_detail_url = self.location.url_reverse('course_team', self.inactive_user.email)
         self.invalid_detail_url = self.location.url_reverse('course_team', "nonexistent@user.com")
-        self.staff_groupname = get_course_groupname_for_role(self.course.location, "staff")
-        self.inst_groupname = get_course_groupname_for_role(self.course.location, "instructor")
+        self.staff_groupname = get_course_groupname_for_role(self.course_locator, "staff")
+        self.inst_groupname = get_course_groupname_for_role(self.course_locator, "instructor")
 
     def test_index(self):
         resp = self.client.get(self.index_url, HTTP_ACCEPT='text/html')
@@ -145,18 +145,6 @@ def test_detail_post_missing_role(self):
         self.assertIn("error", result)
         self.assert_not_enrolled()
 
-    def test_detail_post_bad_json(self):
-        resp = self.client.post(
-            self.detail_url,
-            data="{foo}",
-            content_type="application/json",
-            HTTP_ACCEPT="application/json",
-        )
-        self.assertEqual(resp.status_code, 400)
-        result = json.loads(resp.content)
-        self.assertIn("error", result)
-        self.assert_not_enrolled()
-
     def test_detail_post_no_json(self):
         resp = self.client.post(
             self.detail_url,