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

Merge pull request #1799 from edx/dhm/bug-1003

Browse Source 
Improve auth handling of Locators
This commit is contained in:
Don Mitchell
2013-12-02 12:42:51 -08:00
parent 45d373c278 9dc68b03a0
commit 7fa7641c0a
5 changed files with 246 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

130
cms/djangoapps/auth/authz.py
View File

@@ -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 django.conf import settings
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 @@ COURSE_CREATOR_GROUP_NAME = "course_creator_group"
# 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,15 +114,13 @@ 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):
@@ -98,25 +128,25 @@ 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

127
cms/djangoapps/contentstore/tests/test_permissions.py Normal file
View File

@@ -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))

16
cms/djangoapps/contentstore/tests/test_users.py
View File

@@ -29,8 +29,8 @@ class UsersTestCase(CourseTestCase):
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 @@ class UsersTestCase(CourseTestCase):
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,

4
cms/djangoapps/contentstore/views/course.py
View File

@@ -292,7 +292,8 @@ def create_new_course(request):
initialize_course_tabs(new_course)
create_all_course_groups(request.user, new_course.location)
new_location = loc_mapper().translate_location(new_course.location.course_id, new_course.location, False, True)
create_all_course_groups(request.user, new_location)
# seed the forums
seed_permissions_roles(new_course.location.course_id)
@@ -301,7 +302,6 @@ def create_new_course(request):
# work.
CourseEnrollment.enroll(request.user, new_course.location.course_id)
new_location = loc_mapper().translate_location(new_course.location.course_id, new_course.location, False, True)
return JsonResponse({'url': new_location.url_reverse("course/", "")})

63
cms/djangoapps/contentstore/views/user.py
View File

@@ -1,5 +1,4 @@
import json
from django.conf import settings
from django.core.exceptions import PermissionDenied
from django.contrib.auth.models import User, Group
from django.contrib.auth.decorators import login_required
@@ -10,9 +9,11 @@ from django_future.csrf import ensure_csrf_cookie
from mitxmako.shortcuts import render_to_response
from xmodule.modulestore.django import modulestore, loc_mapper
from util.json_request import JsonResponse
from util.json_request import JsonResponse, expect_json
from auth.authz import (
STAFF_ROLE_NAME, INSTRUCTOR_ROLE_NAME, get_course_groupname_for_role)
STAFF_ROLE_NAME, INSTRUCTOR_ROLE_NAME, get_course_groupname_for_role,
get_course_role_users
)
from course_creators.views import user_requested_access
from .access import has_access
@@ -35,6 +36,7 @@ def request_course_creator(request):
return JsonResponse({"Status": "OK"})
# pylint: disable=unused-argument
@login_required
@ensure_csrf_cookie
@require_http_methods(("GET", "POST", "PUT", "DELETE"))
@@ -62,38 +64,39 @@ def course_team_handler(request, tag=None, course_id=None, branch=None, version_
return HttpResponseNotFound()
def _manage_users(request, location):
def _manage_users(request, locator):
"""
This view will return all CMS users who are editors for the specified course
"""
old_location = loc_mapper().translate_locator_to_location(location)
old_location = loc_mapper().translate_locator_to_location(locator)
# check that logged in user has permissions to this item
if not has_access(request.user, location, role=INSTRUCTOR_ROLE_NAME) and not has_access(request.user, location, role=STAFF_ROLE_NAME):
if not has_access(request.user, locator):
raise PermissionDenied()
course_module = modulestore().get_item(old_location)
staff_groupname = get_course_groupname_for_role(location, "staff")
staff_group, __ = Group.objects.get_or_create(name=staff_groupname)
inst_groupname = get_course_groupname_for_role(location, "instructor")
inst_group, __ = Group.objects.get_or_create(name=inst_groupname)
instructors = get_course_role_users(locator, INSTRUCTOR_ROLE_NAME)
# the page only lists staff and assumes they're a superset of instructors. Do a union to ensure.
staff = set(get_course_role_users(locator, STAFF_ROLE_NAME)).union(instructors)
return render_to_response('manage_users.html', {
'context_course': course_module,
'staff': staff_group.user_set.all(),
'instructors': inst_group.user_set.all(),
'allow_actions': has_access(request.user, location, role=INSTRUCTOR_ROLE_NAME),
'staff': staff,
'instructors': instructors,
'allow_actions': has_access(request.user, locator, role=INSTRUCTOR_ROLE_NAME),
})
def _course_team_user(request, location, email):
old_location = loc_mapper().translate_locator_to_location(location)
@expect_json
def _course_team_user(request, locator, email):
"""
Handle the add, remove, promote, demote requests ensuring the requester has authority
"""
# check that logged in user has permissions to this item
if has_access(request.user, location, role=INSTRUCTOR_ROLE_NAME):
if has_access(request.user, locator, role=INSTRUCTOR_ROLE_NAME):
# instructors have full permissions
pass
elif has_access(request.user, location, role=STAFF_ROLE_NAME) and email == request.user.email:
elif has_access(request.user, locator, role=STAFF_ROLE_NAME) and email == request.user.email:
# staff can only affect themselves
pass
else:
@@ -123,7 +126,7 @@ def _course_team_user(request, location, email):
# what's the highest role that this user has?
groupnames = set(g.name for g in user.groups.all())
for role in roles:
role_groupname = get_course_groupname_for_role(old_location, role)
role_groupname = get_course_groupname_for_role(locator, role)
if role_groupname in groupnames:
msg["role"] = role
break
@@ -139,7 +142,7 @@ def _course_team_user(request, location, email):
# make sure that the role groups exist
groups = {}
for role in roles:
groupname = get_course_groupname_for_role(old_location, role)
groupname = get_course_groupname_for_role(locator, role)
group, __ = Group.objects.get_or_create(name=groupname)
groups[role] = group
@@ -162,22 +165,13 @@ def _course_team_user(request, location, email):
return JsonResponse()
# all other operations require the requesting user to specify a role
if request.META.get("CONTENT_TYPE", "").startswith("application/json") and request.body:
try:
payload = json.loads(request.body)
except:
return JsonResponse({"error": _("malformed JSON")}, 400)
try:
role = payload["role"]
except KeyError:
return JsonResponse({"error": _("`role` is required")}, 400)
else:
if not "role" in request.POST:
return JsonResponse({"error": _("`role` is required")}, 400)
role = request.POST["role"]
role = request.json.get("role", request.POST.get("role"))
if role is None:
return JsonResponse({"error": _("`role` is required")}, 400)
old_location = loc_mapper().translate_locator_to_location(locator)
if role == "instructor":
if not has_access(request.user, location, role=INSTRUCTOR_ROLE_NAME):
if not has_access(request.user, locator, role=INSTRUCTOR_ROLE_NAME):
msg = {
"error": _("Only instructors may create other instructors")
}
@@ -203,4 +197,3 @@ def _course_team_user(request, location, email):
CourseEnrollment.enroll(user, old_location.course_id)
return JsonResponse()