blob: b5f986ada5b4b4f2fd366fa2f6d93205328c68c2 [file] [log] [blame]
# Copyright 2011 the Melange authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Module containing the AccessChecker class that contains helper functions
for checking access.
"""
import urllib
from google.appengine.ext import ndb
from django.core.urlresolvers import reverse
from django.utils.translation import ugettext
from codein.logic import task as task_logic
from codein.logic import timeline as timeline_logic
from codein.views.helper import urls
from melange.models import profile as profile_model
from melange.request import exception
from melange.request import links
from melange.utils import time
from soc.logic import dicts
from soc.models.org_app_record import OrgAppRecord
from soc.views.helper import access_checker
from soc.modules.gci.logic import conversation as gciconversation_logic
from soc.modules.gci.logic import task as gci_task_logic
from soc.modules.gci.models import conversation as gciconversation_model
from soc.modules.gci.models import task as task_model
DEF_ALREADY_PARTICIPATING_AS_NON_STUDENT = ugettext(
'You cannot register as a student since you are already a '
'mentor or organization administrator in %s.')
DEF_ALL_WORK_STOPPED = ugettext(
'All work on tasks has stopped. You can no longer place comments, '
'submit work or make any changes to existing tasks.')
DEF_COMMENTING_NOT_ALLOWED = ugettext(
"No more comments can be placed at this time.")
DEF_NO_TASK_CREATE_PRIV = ugettext(
'You do not have sufficient privileges to create a new task for %s.')
DEF_NO_TASK_EDIT_PRIV = ugettext(
'You do not have sufficient privileges to edit a new task for %s.')
DEF_NO_PREV_ORG_MEMBER = ugettext(
'To apply as an organization for GCI you must have you must have '
'participated as a member of an organization in a prior instance '
'of Google Summer of Code or Google Code-in and have a profile '
'registered in Melange')
DEF_NOT_ORG_ADMIN_FOR_ORG_APP = ugettext(
"You should be listed as one of the administrators on %(org_name)s's "
"organization application to create a new organization profile for "
"%(org_name)s.")
DEF_TASK_UNEDITABLE_STATUS = ugettext(
'This task is already published and published tasks cannot be edited.')
DEF_TASK_MUST_BE_IN_STATES = ugettext(
'The task must be in one of the followings states %s')
DEF_TASK_MAY_NOT_BE_IN_STATES = ugettext(
'The task may not be in one of the followings states %s')
DEF_ORG_APP_REJECTED = ugettext(
'This org application has been rejected')
DEF_NOT_IN_CONVERSATION = ugettext(
'You do not have sufficient privileges to view this conversation.')
class Mutator(access_checker.Mutator):
"""Helper class for access checking.
Mutates the data object as requested.
"""
def __init__(self, data):
super(Mutator, self).__init__(data)
self.unsetAll()
def unsetAll(self):
self.data.task = access_checker.unset
self.data.comments = access_checker.unset
self.data.work_submissions = access_checker.unset
self.data.is_visible = access_checker.unset
self.data.full_edit = access_checker.unset
self.data.conversation = access_checker.unset
def taskFromKwargs(self, comments=False, work_submissions=True):
"""Sets the GCITask entity in RequestData object.
The entity that is set will always be in a valid state and for the program
that is set in the RequestData.
Args:
comments: If true the comments on this task are added to RequestData
work_submissions: If true the work submissions on this task are added to
RequestData
"""
task_id = long(self.data.kwargs['id'])
task = task_model.GCITask.get_by_id(task_id)
if not task or (task.program.key() != self.data.program.key()) or \
task.status == 'invalid':
error_msg = access_checker.DEF_ID_BASED_ENTITY_NOT_EXISTS % {
'model': 'GCITask',
'id': task_id,
}
raise exception.NotFound(message=error_msg)
self.data.task = task
if comments:
self.data.comments = task.comments()
if work_submissions:
self.data.work_submissions = task.workSubmissions()
def taskFromKwargsIfId(self):
"""Sets the GCITask entity in RequestData object if ID exists or None."""
if 'id' in self.data.kwargs:
self.taskFromKwargs()
else:
self.data.task = None
def conversationFromKwargs(self):
"""Sets the GCIConversation entity in the RequestData object.
Args:
messages: If true, the messages for this conversation are added to
RequestData
"""
task_id = long(self.data.kwargs['id'])
conversation = gciconversation_model.GCIConversation.get_by_id(task_id)
if (not conversation or
ndb.Key.to_old_key(conversation.program) != self.data.program.key()):
error_msg = access_checker.DEF_ID_BASED_ENTITY_NOT_EXISTS % {
'model': 'GCIConversation',
'id': task_id
}
raise exception.NotFound(message=error_msg)
self.data.conversation = conversation
def orgAppFromOrgId(self):
org_id = self.data.GET.get('org_id')
if not org_id:
raise exception.BadRequest(message='Missing org_id')
q = OrgAppRecord.all()
q.filter('survey', self.data.org_app)
q.filter('org_id', org_id)
self.data.org_app_record = q.get()
if not self.data.org_app_record:
raise exception.NotFound(
message="There is no org_app for the org_id %s" % org_id)
def fullEdit(self, full_edit=False):
"""Sets full_edit to True/False depending on the status of the task."""
self.data.full_edit = full_edit
class AccessChecker(access_checker.AccessChecker):
"""Access checker for GCI specific methods."""
def isTaskVisible(self):
"""Checks if the task is visible to the public.
Returns: True if the task is visible, if the task is not visible
but the user can edit the task, False.
"""
assert access_checker.isSet(self.data.task)
# TODO(nathaniel): Yep, this is weird.
can_edit = False
try:
self.checkCanUserEditTask()
self.checkHasTaskEditableStatus()
self.checkTimelineAllowsTaskEditing()
can_edit = True
except exception.UserError:
pass
if not self.data.timeline.tasksPubliclyVisible():
if can_edit:
return False
period = self.data.timeline.tasksPubliclyVisibleOn()
raise exception.Forbidden(
message=access_checker.DEF_PAGE_INACTIVE_BEFORE % period)
if not self.data.task.isAvailable():
if can_edit:
return False
raise exception.Forbidden(message=access_checker.DEF_PAGE_INACTIVE)
return True
def isTaskInState(self, states):
"""Checks if the task is in any of the given states.
Args:
states: List of states in which a task may be for this check to pass.
"""
assert access_checker.isSet(self.data.task)
if self.data.task.status not in states:
raise exception.Forbidden(message=DEF_TASK_MUST_BE_IN_STATES % states)
def isTaskNotInStates(self, states):
"""Checks if the task is not in any of the given states.
Args:
states: List of states in which a task may not be for this check to pass.
"""
assert access_checker.isSet(self.data.task)
if self.data.task.status in states:
raise exception.Forbidden(message=DEF_TASK_MAY_NOT_BE_IN_STATES % states)
def canApplyStudent(self, edit_url):
"""Checks if a user may apply as a student to the program."""
if self.data.ndb_profile:
if self.data.ndb_profile.is_student:
raise exception.Redirect(edit_url)
else:
raise exception.Forbidden(
message=DEF_ALREADY_PARTICIPATING_AS_NON_STUDENT % (
self.data.program.name))
self.studentSignupActive()
# custom pre-registration age check for GCI students
age_check = self.data.request.COOKIES.get('age_check', None)
if not age_check or age_check == '0':
# no age check done or it failed
kwargs = dicts.filter(self.data.kwargs, ['sponsor', 'program'])
age_check_url = reverse('gci_age_check', kwargs=kwargs)
raise exception.Redirect(age_check_url)
else:
self.isLoggedIn()
def hasNonStudentProfileInAProgram(self):
"""Check if the user has participated in the previous programs.
This checks if the user has at least one non-student profile in previous
programs.
"""
if not self.data.ndb_user:
raise exception.Forbidden(message=DEF_NO_PREV_ORG_MEMBER)
query = profile_model.Profile.query(
profile_model.Profile.is_mentor == True,
profile_model.Profile.status == profile_model.Status.ACTIVE,
ancestor=self.data.ndb_user.key)
if not query.get():
raise exception.Forbidden(message=DEF_NO_PREV_ORG_MEMBER)
def canTakeOrgApp(self):
"""Check if the user can take the org app.
A user can take the GCI org app if he/she participated in GSoC or GCI as
a non-student and has a non-student profile for the current program.
"""
# TODO(daniel): make this a program setting - sometimes it may be possible
# to accept organizations which have not participated before
self.hasNonStudentProfileInAProgram()
self.hasProfileOrRedirectToCreate('org_admin')
def isOrgAppAccepted(self):
"""Checks if the org app stored in request data is accepted."""
assert self.data.org_app_record
if self.data.org_app_record.status != 'accepted':
raise exception.Forbidden(message=DEF_ORG_APP_REJECTED)
def isUserAdminForOrgApp(self):
"""Checks if the user is listed as an admin for the org app in RequestData.
"""
assert self.data.org_app
assert self.data.org_app_record
if not self.data.ndb_user or self.data.ndb_user.key.to_old_key() not in [
OrgAppRecord.main_admin.get_value_for_datastore(
self.data.org_app_record),
OrgAppRecord.backup_admin.get_value_for_datastore(
self.data.org_app_record)]:
raise exception.Forbidden(message=DEF_NOT_ORG_ADMIN_FOR_ORG_APP % {
'org_name': self.data.org_app_record.name})
def hasProfileOrRedirectToCreate(self, role, get_params=None):
"""Checks if user has a profile and redirects to "Create Profile" page
if a profile is not present.
Args:
role: type of profile that should potentially be created. May be one
of: org_admin, mentor, student.
get_params: optional dictionary with GET parameters that should be
appended to the redirect URL
"""
if not self.data.ndb_profile:
get_params = get_params or {}
profile_url = links.ABSOLUTE_LINKER.program(
self.data.program, urls.UrlNames.PROFILE_REGISTER_AS_ORG_MEMBER,
secure=True)
if get_params:
profile_url += '?' + urllib.urlencode(get_params)
raise exception.Redirect(profile_url)
def isBeforeAllWorkStopped(self):
"""Raises exception.UserError if all work on tasks has stopped."""
if timeline_logic.isAfterStopAllWorkDeadline(self.data.program_timeline):
raise exception.Forbidden(message=DEF_ALL_WORK_STOPPED)
def isCommentingAllowed(self):
"""Raises exception.UserError if commenting is not allowed."""
if not task_logic.isCommentingAllowed(
self.data.task, self.data.ndb_profile, self.data.program):
raise exception.Forbidden(message=DEF_COMMENTING_NOT_ALLOWED)
def canCreateTask(self):
"""Checks whether the currently logged in user can edit the task."""
return self.canCreateTaskWithRequiredRole('mentor')
def canBulkCreateTask(self):
"""Checks whether the currently logged in user can bulk create tasks."""
return self.canCreateTaskWithRequiredRole('org_admin')
def canCreateTaskWithRequiredRole(self, required_role):
"""Checks whether the currently logged in user can create or edit
a task, when the specified role is required.
"""
if required_role == 'mentor':
valid_org_keys = self.data.ndb_profile.mentor_for
elif required_role == 'org_admin':
valid_org_keys = self.data.ndb_profile.admin_for
else:
raise ValueError('Invalid required_role argument ' + str(required_role))
if self.data.url_ndb_org.key not in valid_org_keys:
raise exception.Forbidden(message=DEF_NO_TASK_CREATE_PRIV % (
self.data.url_ndb_org.name))
if (time.isBefore(self.data.timeline.orgsAnnouncedOn()) \
or self.data.timeline.tasksClaimEnded()):
raise exception.Forbidden(message=access_checker.DEF_PAGE_INACTIVE)
def canUserEditTask(self):
"""Returns True/False depending on whether the currently logged in user
can edit the task.
"""
org_key = task_logic.getOrganizationKey(self.data.task)
return (
self.data.ndb_profile and
org_key in self.data.ndb_profile.mentor_for)
def checkCanUserEditTask(self):
"""Checks whether the currently logged in user can edit the task."""
assert access_checker.isSet(self.data.task)
if not self.canUserEditTask():
org_key = task_logic.getOrganizationKey(self.data.task)
raise exception.Forbidden(
message=DEF_NO_TASK_EDIT_PRIV % org_key.get().name)
def checkHasTaskEditableStatus(self):
"""Checks whether the task is in one of the editable states.
We specifically do not allow editing of tasks which are already claimed.
"""
if not gci_task_logic.hasTaskEditableStatus(self.data.task):
raise exception.Forbidden(message=DEF_TASK_UNEDITABLE_STATUS)
def timelineAllowsTaskEditing(self):
"""Returns True/False depending on whether orgs can edit task depending
on where in the program timeline we are currently in.
"""
return not (time.isBefore(self.data.timeline.orgsAnnouncedOn()) or
self.data.timeline.tasksClaimEnded())
def checkTimelineAllowsTaskEditing(self):
"""Checks if organizations can edit tasks at the current time in
the program.
"""
if not self.timelineAllowsTaskEditing():
raise exception.Forbidden(message=access_checker.DEF_PAGE_INACTIVE)
def isUserInConversation(self):
"""Checks if the user is part of a conversation."""
assert access_checker.isSet(self.data.conversation)
assert access_checker.isSet(self.data.user)
query = gciconversation_logic.queryConversationUserForConversationAndUser(
self.data.conversation.key, ndb.Key.from_old_key(self.data.user.key()))
if query.count() == 0:
raise exception.Forbidden(message=DEF_NOT_IN_CONVERSATION)
class DeveloperAccessChecker(access_checker.DeveloperAccessChecker):
"""Developer access checker for GCI specific methods."""
def isTaskVisible(self):
return True