[REF] survey: adapt controllers and portal flow to new ACLs

Purpose of this commit is to adapt controllers and routes to new ACLs. As
survey access is now controlled routes have to use survey ID and browse
records as sudo when access is granted.

This commit is linked to task ID 1911586 and PR #28986.
This commit is contained in:
Thibault Delavallée
2019-01-14 15:22:18 +00:00
parent 5878579b10
commit dfc37ecede
4 changed files with 260 additions and 136 deletions
+197 -125
View File
@@ -8,6 +8,7 @@ from datetime import datetime
from math import ceil
from odoo import fields, http, SUPERUSER_ID
from odoo.exceptions import UserError
from odoo.http import request
from odoo.tools import ustr
@@ -16,145 +17,206 @@ _logger = logging.getLogger(__name__)
class Survey(http.Controller):
def _check_bad_cases(self, survey, token=None):
# In case of bad survey, redirect to surveys list
if not survey.sudo().exists():
return werkzeug.utils.redirect("/survey/")
def _fetch_from_access_token(self, survey_id, access_token):
""" Check that given token matches an answer from the given survey_id.
Returns a sudo-ed browse record of survey in order to avoid access rights
issues now that access is granted through token. """
survey_sudo = request.env['survey.survey'].with_context(active_test=False).sudo().browse(survey_id)
if not access_token:
answer_sudo = request.env['survey.user_input'].sudo()
else:
answer_sudo = request.env['survey.user_input'].sudo().search([
('survey_id', '=', survey_sudo.id),
('token', '=', access_token)
], limit=1)
return survey_sudo, answer_sudo
# In case of auth required, block public user
if survey.auth_required and request.env.user._is_public():
return request.render("survey.auth_required", {'survey': survey, 'token': token})
def _check_validity(self, survey_id, access_token, ensure_token=True):
""" Check survey is open and can be taken. This does not checks for
security rules, only functional / business rules. It returns a string key
allowing further manipulation of validity issues
# In case of non open surveys
if survey.stage_id.closed:
* survey_wrong: survey does not exist;
* survey_auth: authentication is required;
* survey_closed: survey is closed and does not accept input anymore;
* survey_void: survey is void and should not be taken;
* token_wrong: given token not recognized;
* token_required: no token given although it is necessary to access the
survey;
* answer_deadline: token linked to an expired answer;
:param ensure_token: whether user input existence based on given access token
should be enforced or not, depending on the route requesting a token or
allowing external world calls;
"""
survey_sudo, answer_sudo = self._fetch_from_access_token(survey_id, access_token)
if not survey_sudo.exists():
return 'survey_wrong'
if access_token and not answer_sudo:
return 'token_wrong'
if not answer_sudo and ensure_token:
return 'token_required'
if survey_sudo.auth_required and request.env.user._is_public():
return 'survey_auth'
if survey_sudo.is_closed or not survey_sudo.active:
return 'survey_closed'
if not survey_sudo.page_ids:
return 'survey_void'
# In case of delayed deadline # TDE FIXME
if answer_sudo and answer_sudo.deadline:
dt_now = datetime.now()
if dt_now > answer_sudo.deadline:
return 'answer_deadline'
return True
def _get_access_data(self, survey_id, access_token, ensure_token=True):
""" Get back data related to survey and user input, given the ID and access
token provided by the route.
: param ensure_token: whether user input existence should be enforced or not(see ``_check_validity``)
"""
survey_sudo, answer_sudo = request.env['survey.survey'].sudo(), request.env['survey.user_input'].sudo()
has_survey_access, can_answer = False, False
validity_code = self._check_validity(survey_id, access_token, ensure_token=ensure_token)
if validity_code != 'survey_wrong':
survey_sudo, answer_sudo = self._fetch_from_access_token(survey_id, access_token)
try:
survey_user = survey_sudo.sudo(request.env.user)
survey_user.check_access_rights(self, 'read', raise_exception=True)
survey_user.check_access_rule(self, 'read')
except:
pass
else:
has_survey_access = True
can_answer = bool(answer_sudo)
if not can_answer:
can_answer = survey_sudo.auth_required is False
return {
'survey_sudo': survey_sudo,
'answer_sudo': answer_sudo,
'has_survey_access': has_survey_access,
'can_answer': can_answer,
'validity_code': validity_code,
}
def _redirect_with_error(self, access_data, error_key):
survey_sudo = access_data['survey_sudo']
answer_sudo = access_data['answer_sudo']
if error_key == 'survey_void' and access_data['can_answer']:
return request.render("survey.nopages", {'survey': survey_sudo})
elif error_key == 'survey_closed' and access_data['can_answer']:
return request.render("survey.notopen")
elif error_key == 'survey_auth' and answer_sudo.token:
return request.render("survey.auth_required", {'survey': survey_sudo, 'token': answer_sudo.token})
elif error_key == 'answer_deadline' and answer_sudo.token:
return request.render("survey.notopen")
# If there is no pages
if not survey.page_ids:
return request.render("survey.nopages", {'survey': survey})
return werkzeug.utils.redirect("/")
# Everything seems to be ok
return None
def _check_deadline(self, user_input):
'''Prevent opening of the survey if the deadline has turned out
! This will NOT disallow access to users who have already partially filled the survey !'''
deadline = user_input.deadline
if deadline:
dt_deadline = fields.Datetime.from_string(deadline)
dt_now = datetime.now()
if dt_now > dt_deadline: # survey is not open anymore
return request.render("survey.notopen")
return None
@http.route('/survey/test/<model("survey.survey"):survey>', type='http', auth='user', website=True)
def survey_test(self, survey, token=None, **kwargs):
@http.route('/survey/test/<int:survey_id>', type='http', auth='user', website=True)
def survey_test(self, survey_id, token=None, **kwargs):
""" Test mode for surveys: create a test answer, only for managers or officers
testing their surveys """
if request.env.user.has_group('survey.group_survey_manager') or \
request.env.user.has_group('survey.group_survey_user') and survey.create_uid == request.env.user:
user_input = request.env['survey.user_input'].create({
'survey_id': survey.id,
'test_entry': True
})
return request.redirect('/survey/start/%s?token=%s' % (survey.id, user_input.token))
return werkzeug.utils.redirect('/')
survey_sudo = request.env['survey.survey'].sudo().browse(survey_id)
try:
answer_sudo = survey_sudo._create_answer(user=request.env.user, test_entry=True)
except:
return werkzeug.utils.redirect('/')
return request.redirect('/survey/start/%s?token=%s' % (survey_sudo.id, answer_sudo.token))
@http.route('/survey/start/<model("survey.survey"):survey>', type='http', auth='public', website=True)
def survey_start(self, survey, token=None):
@http.route('/survey/start/<int:survey_id>', type='http', auth='public', website=True)
def survey_start(self, survey_id, token=None, email=False):
""" Start a survey by providing a token linked to an answer or generate
a new token if access is allowed """
UserInput = request.env['survey.user_input']
access_data = self._get_access_data(survey_id, token, ensure_token=False)
if access_data['validity_code'] is not True:
return self._redirect_with_error(access_data, access_data['validity_code'])
# Controls if the survey can be displayed
errpage = self._check_bad_cases(survey, token=token)
if errpage:
return errpage
survey_sudo, answer_sudo = access_data['survey_sudo'], access_data['answer_sudo']
if not answer_sudo:
try:
answer_sudo = survey_sudo._create_answer(user=request.env.user, email=email)
except UserError:
answer_sudo = False
# Manual surveying
if not token:
vals = {'survey_id': survey.id}
if not request.env.user._is_public():
vals['partner_id'] = request.env.user.partner_id.id
user_input = UserInput.create(vals)
else:
user_input = UserInput.sudo().search([('token', '=', token)], limit=1)
if not user_input:
return request.render("survey.403", {'survey': survey})
# Do not open expired survey
errpage = self._check_deadline(user_input)
if errpage:
return errpage
if not answer_sudo:
try:
survey_sudo.sudo(request.env.user).check_access_rights('read')
survey_sudo.sudo(request.env.user).check_access_rule('read')
except:
return werkzeug.utils.redirect("/")
else:
return request.render("survey.403", {'survey': survey_sudo})
# Select the right page
if user_input.state == 'new': # Intro page
data = {'survey': survey, 'page': None, 'token': user_input.token}
if answer_sudo.state == 'new': # Intro page
data = {'survey': survey_sudo, 'page': None, 'token': answer_sudo.token}
return request.render('survey.survey_init', data)
else:
return request.redirect('/survey/fill/%s/%s' % (survey.id, user_input.token))
return request.redirect('/survey/fill/%s/%s' % (survey_sudo.id, answer_sudo.token))
@http.route('/survey/fill/<model("survey.survey"):survey>/<string:token>', type='http', auth='public', website=True)
def survey_display_page(self, survey, token, prev=None):
Survey = request.env['survey.survey']
UserInput = request.env['survey.user_input']
@http.route('/survey/fill/<int:survey_id>/<string:token>', type='http', auth='public', website=True)
def survey_display_page(self, survey_id, token, prev=None):
access_data = self._get_access_data(survey_id, token, ensure_token=True)
if access_data['validity_code'] is not True:
return self._redirect_with_error(access_data, access_data['validity_code'])
# Controls if the survey can be displayed
errpage = self._check_bad_cases(survey)
if errpage:
return errpage
# Load the user_input
user_input = UserInput.sudo().search([('token', '=', token)], limit=1)
if not user_input: # Invalid token
return request.render("survey.403", {'survey': survey})
# Do not display expired survey (even if some pages have already been
# displayed -- There's a time for everything!)
errpage = self._check_deadline(user_input)
if errpage:
return errpage
survey_sudo, answer_sudo = access_data['survey_sudo'], access_data['answer_sudo']
# Select the right page
if user_input.state == 'new': # First page
page, page_nr, last = Survey.next_page(user_input, 0, go_back=False)
data = {'survey': survey, 'page': page, 'page_nr': page_nr, 'token': user_input.token}
if answer_sudo.state == 'new': # First page
page, page_nr, last = survey_sudo.next_page(answer_sudo, 0, go_back=False)
data = {'survey': survey_sudo, 'page': page, 'page_nr': page_nr, 'token': answer_sudo.token}
if last:
data.update({'last': True})
return request.render('survey.survey', data)
elif user_input.state == 'done': # Display success message
return request.render('survey.sfinished', {'survey': survey,
elif answer_sudo.state == 'done': # Display success message
return request.render('survey.sfinished', {'survey': survey_sudo,
'token': token,
'user_input': user_input})
elif user_input.state == 'skip':
'user_input': answer_sudo})
elif answer_sudo.state == 'skip':
flag = (True if prev and prev == 'prev' else False)
page, page_nr, last = Survey.next_page(user_input, user_input.last_displayed_page_id.id, go_back=flag)
page, page_nr, last = survey_sudo.next_page(answer_sudo, answer_sudo.last_displayed_page_id.id, go_back=flag)
#special case if you click "previous" from the last page, then leave the survey, then reopen it from the URL, avoid crash
if not page:
page, page_nr, last = Survey.next_page(user_input, user_input.last_displayed_page_id.id, go_back=True)
page, page_nr, last = survey_sudo.next_page(answer_sudo, answer_sudo.last_displayed_page_id.id, go_back=True)
data = {'survey': survey, 'page': page, 'page_nr': page_nr, 'token': user_input.token}
data = {'survey': survey_sudo, 'page': page, 'page_nr': page_nr, 'token': answer_sudo.token}
if last:
data.update({'last': True})
return request.render('survey.survey', data)
else:
return request.render("survey.403", {'survey': survey})
return request.render("survey.403", {'survey': survey_sudo})
@http.route('/survey/prefill/<model("survey.survey"):survey>/<string:token>', type='http', auth='public', website=True)
def survey_get_answers(self, survey, token, page_id=None):
@http.route('/survey/prefill/<int:survey_id>/<string:token>', type='http', auth='public', website=True)
def survey_get_answers(self, survey_id, token, page_id=None):
""" TDE NOTE: original comment: # AJAX prefilling of a survey -> AJAX / http ?? """
UserInputLine = request.env['survey.user_input_line']
ret = {}
access_data = self._get_access_data(survey_id, token, ensure_token=True)
if access_data['validity_code'] is not True:
return {}
survey_sudo, answer_sudo = access_data['survey_sudo'], access_data['answer_sudo']
# Fetch previous answers
if page_id:
previous_answers = UserInputLine.sudo().search([('user_input_id.token', '=', token), ('page_id', '=', page_id)])
previous_answers = answer_sudo.user_input_line_ids.filtered(lambda line: line.page_id.id == page_id)
else:
previous_answers = UserInputLine.sudo().search([('user_input_id.token', '=', token)])
previous_answers = answer_sudo.user_input_line_ids
# Return non empty answers in a JSON compatible format
ret = {}
for answer in previous_answers:
if not answer.skipped:
answer_tag = '%s_%s_%s' % (answer.survey_id.id, answer.page_id.id, answer.question_id.id)
@@ -182,31 +244,41 @@ class Survey(http.Controller):
_logger.warning("[survey] No answer has been found for question %s marked as non skipped" % answer_tag)
return json.dumps(ret, default=str)
@http.route('/survey/scores/<model("survey.survey"):survey>/<string:token>', type='http', auth='public', website=True)
def survey_get_scores(self, survey, token, page_id=None):
@http.route('/survey/scores/<int:survey_id>/<string:token>', type='http', auth='public', website=True)
def survey_get_scores(self, survey_id, token, page_id=None):
""" TDE NOTE: original comment: # AJAX scores loading for quiz correction mode -> AJAX / http ?? """
ret = {}
access_data = self._get_access_data(survey_id, token, ensure_token=True)
if access_data['validity_code'] is not True:
return {}
# Fetch answers
previous_answers = request.env['survey.user_input_line'].sudo().search([('user_input_id.token', '=', token)])
survey_sudo, answer_sudo = access_data['survey_sudo'], access_data['answer_sudo']
# Compute score for each question
for answer in previous_answers:
ret = {}
for answer in answer_sudo.user_input_line_ids:
tmp_score = ret.get(answer.question_id.id, 0.0)
ret.update({answer.question_id.id: tmp_score + answer.quizz_mark})
return json.dumps(ret)
@http.route('/survey/submit/<model("survey.survey"):survey>/<string:token>', type='http', methods=['POST'], auth='public', website=True)
def survey_submit(self, survey, token, **post):
@http.route('/survey/submit/<int:survey_id>/<string:token>', type='http', methods=['POST'], auth='public', website=True)
def survey_submit(self, survey_id, token, **post):
""" TDE NOTE: original comment: # AJAX submission of a page -> AJAX / http ?? """
_logger.debug('Incoming data: %s', post)
access_data = self._get_access_data(survey_id, token, ensure_token=True)
if access_data['validity_code'] is not True:
return {}
survey_sudo, answer_sudo = access_data['survey_sudo'], access_data['answer_sudo']
page_id = int(post['page_id'])
questions = request.env['survey.question'].search([('page_id', '=', page_id)])
questions = request.env['survey.question'].sudo().search([
('survey_id', '=', survey_id),
('page_id', '=', page_id)
])
# Answer validation
errors = {}
for question in questions:
answer_tag = "%s_%s_%s" % (survey.id, page_id, question.id)
answer_tag = "%s_%s_%s" % (survey_id, page_id, question.id)
errors.update(question.validate_question(post, answer_tag))
ret = {}
@@ -214,39 +286,39 @@ class Survey(http.Controller):
# Return errors messages to webpage
ret['errors'] = errors
else:
# Store answers into database
try:
user_input = request.env['survey.user_input'].sudo().search([('token', '=', token)], limit=1)
except KeyError: # Invalid token
return request.render("survey.403", {'survey': survey})
user_id = request.env.user.id if user_input.input_type != 'link' else SUPERUSER_ID
for question in questions:
answer_tag = "%s_%s_%s" % (survey.id, page_id, question.id)
request.env['survey.user_input_line'].sudo(user=user_id).save_lines(user_input.id, question, post, answer_tag)
answer_tag = "%s_%s_%s" % (survey_id, page_id, question.id)
request.env['survey.user_input_line'].sudo().save_lines(answer_sudo.id, question, post, answer_tag)
go_back = post['button_submit'] == 'previous'
next_page, _, last = request.env['survey.survey'].next_page(user_input, page_id, go_back=go_back)
next_page, _, last = request.env['survey.survey'].next_page(answer_sudo, page_id, go_back=go_back)
vals = {'last_displayed_page_id': page_id}
if next_page is None and not go_back:
vals.update({'state': 'done'})
else:
vals.update({'state': 'skip'})
user_input.sudo(user=user_id).write(vals)
ret['redirect'] = '/survey/fill/%s/%s' % (survey.id, token)
answer_sudo.write(vals)
ret['redirect'] = '/survey/fill/%s/%s' % (survey_sudo.id, token)
if go_back:
ret['redirect'] += '?prev=prev'
return json.dumps(ret)
@http.route('/survey/print/<model("survey.survey"):survey>', type='http', auth='public', website=True)
def survey_print(self, survey, token=None, **post):
@http.route('/survey/print/<int:survey_id>', type='http', auth='public', website=True)
def survey_print(self, survey_id, token=None, **post):
'''Display an survey in printable view; if <token> is set, it will
grab the answers of the user_input_id that has <token>.'''
access_data = self._get_access_data(survey_id, token, ensure_token=False)
if access_data['validity_code'] is not True and (
not access_data['has_survey_access'] or access_data['validity_code'] not in ['token_required', 'survey_closed', 'survey_void']):
return self._redirect_with_error(access_data, access_data['validity_code'])
survey_sudo, answer_sudo = access_data['survey_sudo'], access_data['answer_sudo']
return request.render('survey.survey_print', {
'survey': survey,
'survey': survey_sudo,
'token': token,
'page_nr': 0,
'quizz_correction': True if survey.quizz_mode and token else False})
'quizz_correction': True if survey_sudo.quizz_mode and token else False})
@http.route('/survey/results/<model("survey.survey"):survey>', type='http', auth='user', website=True)
def survey_report(self, survey, token=None, **post):
+53 -1
View File
@@ -60,7 +60,7 @@ class Survey(models.Model):
""" Computes a public URL for the survey """
base_url = self.env['ir.config_parameter'].sudo().get_param('web.base.url')
for survey in self:
survey.public_url = urls.url_join(base_url, "survey/start/%s" % (slug(survey)))
survey.public_url = urls.url_join(base_url, "survey/start/%s" % (survey.id))
@api.model
def _read_group_stage_ids(self, stages, domain, order):
@@ -76,6 +76,58 @@ class Survey(models.Model):
default = dict(default or {}, title=title)
return super(Survey, self).copy_data(default)
@api.multi
def _create_answer(self, user=False, partner=False, email=False, test_entry=False, **additional_vals):
""" Main entry point to get a token back or create a new one. This method
does check for current user access in order to explicitely validate
security.
:param user: target user asking for a token; it might be void or a
public user in which case an email is welcomed;
:param email: email of the person asking the token is no user exists;
"""
self.check_access_rights('read')
self.check_access_rule('read')
tokens = self.env['survey.user_input']
for survey in self:
if partner and not user and partner.user_ids:
user = partner.user_ids[0]
survey._check_answer_creation(user, partner, email, test_entry=test_entry)
answer_vals = {
'survey_id': survey.id,
'test_entry': test_entry,
}
if user and not user._is_public:
answer_vals['partner_id'] = user.partner_id.id
answer_vals['email'] = user.email
elif partner:
answer_vals['partner_id'] = partner.id
answer_vals['email'] = partner.email
else:
answer_vals['email'] = email
answer_vals.update(additional_vals)
tokens += tokens.create(answer_vals)
return tokens
@api.multi
def _check_answer_creation(self, user, partner, email, test_entry=False):
""" Ensure conditions to create new tokens are met. """
self.ensure_one()
if test_entry:
if not user.has_group('survey.group_survey_manager') or not user.has_group('survey.group_survey_user'):
raise UserError(_('Creating test token is not allowed for you.'))
else:
if not self.active:
raise UserError(_('Creating token for archived surveys is not allowed.'))
elif self.is_closed:
raise UserError(_('Creating token for closed surveys is not allowed.'))
if self.auth_required and (not user or user._is_public()):
raise UserError(_('Creating token for external users is not allowed for surveys requesting authentication.'))
@api.model
def next_page(self, user_input, page_id, go_back=False):
""" The next page to display to the user, knowing that page_id is the id
+3 -3
View File
@@ -6,10 +6,10 @@ import odoo.tests
class TestUi(odoo.tests.HttpCase):
def test_01_admin_survey_tour(self):
self.phantom_js("/survey/start/user-feedback-form-1", "odoo.__DEBUG__.services['web_tour.tour'].run('test_survey')", "odoo.__DEBUG__.services['web_tour.tour'].tours.test_survey.ready", login="admin")
self.phantom_js("/survey/start/1", "odoo.__DEBUG__.services['web_tour.tour'].run('test_survey')", "odoo.__DEBUG__.services['web_tour.tour'].tours.test_survey.ready", login="admin")
def test_02_demo_survey_tour(self):
self.phantom_js("/survey/start/user-feedback-form-1", "odoo.__DEBUG__.services['web_tour.tour'].run('test_survey')", "odoo.__DEBUG__.services['web_tour.tour'].tours.test_survey.ready", login="demo")
self.phantom_js("/survey/start/1", "odoo.__DEBUG__.services['web_tour.tour'].run('test_survey')", "odoo.__DEBUG__.services['web_tour.tour'].tours.test_survey.ready", login="demo")
def test_03_public_survey_tour(self):
self.phantom_js("/survey/start/user-feedback-form-1", "odoo.__DEBUG__.services['web_tour.tour'].run('test_survey')", "odoo.__DEBUG__.services['web_tour.tour'].tours.test_survey.ready")
self.phantom_js("/survey/start/1", "odoo.__DEBUG__.services['web_tour.tour'].run('test_survey')", "odoo.__DEBUG__.services['web_tour.tour'].tours.test_survey.ready")
+7 -7
View File
@@ -26,7 +26,7 @@
<h1>Thank you!</h1>
<div t-field="survey.thank_you_message" class="oe_no_empty" />
<div t-if='survey.quizz_mode'>You scored <t t-esc="user_input.quizz_score" /> points.</div>
<div>If you wish, you can <a t-att-href="'/survey/print/%s?token=%s' % (slug(survey), token)">review your answers</a>.</div>
<div>If you wish, you can <a t-att-href="'/survey/print/%s?token=%s' % (survey.id, token)">review your answers</a>.</div>
</div>
</div>
</div>
@@ -137,7 +137,7 @@
<div class='jumbotron mt32'>
<h1 t-field='survey.title' />
<div t-field='survey.description' class="oe_no_empty"/>
<a role="button" class="btn btn-primary btn-lg" t-att-href="'/survey/fill/%s/%s' % (slug(survey), token)">
<a role="button" class="btn btn-primary btn-lg" t-att-href="'/survey/fill/%s/%s' % (survey.id, token)">
Start Survey
</a>
</div>
@@ -170,10 +170,10 @@
</div>
<form role="form" method="post" class="js_surveyform" t-att-name="'%s_%s' % (survey.id, page.id)"
t-att-action="'/survey/fill/%s/%s' % (slug(survey), token)"
t-att-data-prefill="'/survey/prefill/%s/%s?page_id=%s' % (slug(survey), token, page.id)"
t-att-data-validate="'/survey/validate/%s/%s' % (slug(survey), token)"
t-att-data-submit="'/survey/submit/%s/%s' % (slug(survey), token)">
t-att-action="'/survey/fill/%s/%s' % (survey.id, token)"
t-att-data-prefill="'/survey/prefill/%s/%s?page_id=%s' % (survey.id, token, page.id)"
t-att-data-validate="'/survey/validate/%s/%s' % (survey.id, token)"
t-att-data-submit="'/survey/submit/%s/%s' % (survey.id, token)">
<input type="hidden" name="csrf_token" t-att-value="request.csrf_token()"/>
<input type="hidden" name="page_id" t-att-value="page.id" />
<input type="hidden" name="token" t-att-value="token" />
@@ -342,7 +342,7 @@
<h1><span t-field='survey.title'/></h1>
<t t-if="survey.description"><div t-field='survey.description' class="oe_no_empty"/></t>
</div>
<div role="form" class="js_surveyform" t-att-name="'%s' % (survey.id)" t-att-data-prefill="'/survey/prefill/%s/%s' % (slug(survey), token)">
<div role="form" class="js_surveyform" t-att-name="'%s' % (survey.id)" t-att-data-prefill="'/survey/prefill/%s/%s' % (survey.id, token)">
<t t-foreach="survey.page_ids" t-as="page">
<div class="o_page_header">
<h1 t-field='page.title' />