296 lines
14 KiB
Python
296 lines
14 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
||
|
|
|
||
|
|
import base64
|
||
|
|
import json
|
||
|
|
|
||
|
|
from markupsafe import Markup
|
||
|
|
from psycopg2 import IntegrityError
|
||
|
|
from werkzeug.exceptions import BadRequest
|
||
|
|
|
||
|
|
from odoo import http, SUPERUSER_ID, _, _lt
|
||
|
|
from odoo.addons.base.models.ir_qweb_fields import nl2br, nl2br_enclose
|
||
|
|
from odoo.http import request
|
||
|
|
from odoo.tools import plaintext2html
|
||
|
|
from odoo.exceptions import AccessDenied, ValidationError, UserError
|
||
|
|
from odoo.tools.misc import hmac, consteq
|
||
|
|
|
||
|
|
|
||
|
|
class WebsiteForm(http.Controller):
|
||
|
|
|
||
|
|
@http.route('/website/form', type='http', auth="public", methods=['POST'], multilang=False)
|
||
|
|
def website_form_empty(self, **kwargs):
|
||
|
|
# This is a workaround to don't add language prefix to <form action="/website/form/" ...>
|
||
|
|
return ""
|
||
|
|
|
||
|
|
# Check and insert values from the form on the model <model>
|
||
|
|
@http.route('/website/form/<string:model_name>', type='http', auth="public", methods=['POST'], website=True, csrf=False)
|
||
|
|
def website_form(self, model_name, **kwargs):
|
||
|
|
# Partial CSRF check, only performed when session is authenticated, as there
|
||
|
|
# is no real risk for unauthenticated sessions here. It's a common case for
|
||
|
|
# embedded forms now: SameSite policy rejects the cookies, so the session
|
||
|
|
# is lost, and the CSRF check fails, breaking the post for no good reason.
|
||
|
|
csrf_token = request.params.pop('csrf_token', None)
|
||
|
|
if request.session.uid and not request.validate_csrf(csrf_token):
|
||
|
|
raise BadRequest('Session expired (invalid CSRF token)')
|
||
|
|
|
||
|
|
try:
|
||
|
|
# The except clause below should not let what has been done inside
|
||
|
|
# here be committed. It should not either roll back everything in
|
||
|
|
# this controller method. Instead, we use a savepoint to roll back
|
||
|
|
# what has been done inside the try clause.
|
||
|
|
with request.env.cr.savepoint():
|
||
|
|
if request.env['ir.http']._verify_request_recaptcha_token('website_form'):
|
||
|
|
# request.params was modified, update kwargs to reflect the changes
|
||
|
|
kwargs = dict(request.params)
|
||
|
|
kwargs.pop('model_name')
|
||
|
|
return self._handle_website_form(model_name, **kwargs)
|
||
|
|
error = _("Suspicious activity detected by Google reCaptcha.")
|
||
|
|
except (ValidationError, UserError) as e:
|
||
|
|
error = e.args[0]
|
||
|
|
return json.dumps({
|
||
|
|
'error': error,
|
||
|
|
})
|
||
|
|
|
||
|
|
def _handle_website_form(self, model_name, **kwargs):
|
||
|
|
model_record = request.env['ir.model'].sudo().search([('model', '=', model_name), ('website_form_access', '=', True)])
|
||
|
|
if not model_record:
|
||
|
|
return json.dumps({
|
||
|
|
'error': _("The form's specified model does not exist")
|
||
|
|
})
|
||
|
|
|
||
|
|
try:
|
||
|
|
data = self.extract_data(model_record, kwargs)
|
||
|
|
# If we encounter an issue while extracting data
|
||
|
|
except ValidationError as e:
|
||
|
|
# I couldn't find a cleaner way to pass data to an exception
|
||
|
|
return json.dumps({'error_fields': e.args[0]})
|
||
|
|
|
||
|
|
try:
|
||
|
|
id_record = self.insert_record(request, model_record, data['record'], data['custom'], data.get('meta'))
|
||
|
|
if id_record:
|
||
|
|
self.insert_attachment(model_record, id_record, data['attachments'])
|
||
|
|
# in case of an email, we want to send it immediately instead of waiting
|
||
|
|
# for the email queue to process
|
||
|
|
|
||
|
|
if model_name == 'mail.mail':
|
||
|
|
form_has_email_cc = {'email_cc', 'email_bcc'} & kwargs.keys() or \
|
||
|
|
'email_cc' in kwargs["website_form_signature"]
|
||
|
|
# remove the email_cc information from the signature
|
||
|
|
kwargs["website_form_signature"] = kwargs["website_form_signature"].split(':')[0]
|
||
|
|
if kwargs.get("email_to"):
|
||
|
|
value = kwargs['email_to'] + (':email_cc' if form_has_email_cc else '')
|
||
|
|
hash_value = hmac(model_record.env, 'website_form_signature', value)
|
||
|
|
if not consteq(kwargs["website_form_signature"], hash_value):
|
||
|
|
raise AccessDenied('invalid website_form_signature')
|
||
|
|
request.env[model_name].sudo().browse(id_record).send()
|
||
|
|
|
||
|
|
# Some fields have additional SQL constraints that we can't check generically
|
||
|
|
# Ex: crm.lead.probability which is a float between 0 and 1
|
||
|
|
# TODO: How to get the name of the erroneous field ?
|
||
|
|
except IntegrityError:
|
||
|
|
return json.dumps(False)
|
||
|
|
|
||
|
|
request.session['form_builder_model_model'] = model_record.model
|
||
|
|
request.session['form_builder_model'] = model_record.name
|
||
|
|
request.session['form_builder_id'] = id_record
|
||
|
|
|
||
|
|
return json.dumps({'id': id_record})
|
||
|
|
|
||
|
|
# Constants string to make metadata readable on a text field
|
||
|
|
|
||
|
|
_meta_label = _lt("Metadata") # Title for meta data
|
||
|
|
|
||
|
|
# Dict of dynamically called filters following type of field to be fault tolerent
|
||
|
|
|
||
|
|
def identity(self, field_label, field_input):
|
||
|
|
return field_input
|
||
|
|
|
||
|
|
def integer(self, field_label, field_input):
|
||
|
|
return int(field_input)
|
||
|
|
|
||
|
|
def floating(self, field_label, field_input):
|
||
|
|
return float(field_input)
|
||
|
|
|
||
|
|
def html(self, field_label, field_input):
|
||
|
|
return plaintext2html(field_input)
|
||
|
|
|
||
|
|
def boolean(self, field_label, field_input):
|
||
|
|
return bool(field_input)
|
||
|
|
|
||
|
|
def binary(self, field_label, field_input):
|
||
|
|
return base64.b64encode(field_input.read())
|
||
|
|
|
||
|
|
def one2many(self, field_label, field_input):
|
||
|
|
return [int(i) for i in field_input.split(',')]
|
||
|
|
|
||
|
|
def many2many(self, field_label, field_input, *args):
|
||
|
|
return [(args[0] if args else (6, 0)) + (self.one2many(field_label, field_input),)]
|
||
|
|
|
||
|
|
_input_filters = {
|
||
|
|
'char': identity,
|
||
|
|
'text': identity,
|
||
|
|
'html': html,
|
||
|
|
'date': identity,
|
||
|
|
'datetime': identity,
|
||
|
|
'many2one': integer,
|
||
|
|
'one2many': one2many,
|
||
|
|
'many2many': many2many,
|
||
|
|
'selection': identity,
|
||
|
|
'boolean': boolean,
|
||
|
|
'integer': integer,
|
||
|
|
'float': floating,
|
||
|
|
'binary': binary,
|
||
|
|
'monetary': floating,
|
||
|
|
}
|
||
|
|
|
||
|
|
# Extract all data sent by the form and sort its on several properties
|
||
|
|
def extract_data(self, model, values):
|
||
|
|
dest_model = request.env[model.sudo().model]
|
||
|
|
|
||
|
|
data = {
|
||
|
|
'record': {}, # Values to create record
|
||
|
|
'attachments': [], # Attached files
|
||
|
|
'custom': '', # Custom fields values
|
||
|
|
'meta': '', # Add metadata if enabled
|
||
|
|
}
|
||
|
|
|
||
|
|
authorized_fields = model.with_user(SUPERUSER_ID)._get_form_writable_fields()
|
||
|
|
error_fields = []
|
||
|
|
custom_fields = []
|
||
|
|
|
||
|
|
for field_name, field_value in values.items():
|
||
|
|
# If the value of the field if a file
|
||
|
|
if hasattr(field_value, 'filename'):
|
||
|
|
# Undo file upload field name indexing
|
||
|
|
field_name = field_name.split('[', 1)[0]
|
||
|
|
|
||
|
|
# If it's an actual binary field, convert the input file
|
||
|
|
# If it's not, we'll use attachments instead
|
||
|
|
if field_name in authorized_fields and authorized_fields[field_name]['type'] == 'binary':
|
||
|
|
data['record'][field_name] = base64.b64encode(field_value.read())
|
||
|
|
field_value.stream.seek(0) # do not consume value forever
|
||
|
|
if authorized_fields[field_name]['manual'] and field_name + "_filename" in dest_model:
|
||
|
|
data['record'][field_name + "_filename"] = field_value.filename
|
||
|
|
else:
|
||
|
|
field_value.field_name = field_name
|
||
|
|
data['attachments'].append(field_value)
|
||
|
|
|
||
|
|
# If it's a known field
|
||
|
|
elif field_name in authorized_fields:
|
||
|
|
try:
|
||
|
|
input_filter = self._input_filters[authorized_fields[field_name]['type']]
|
||
|
|
data['record'][field_name] = input_filter(self, field_name, field_value)
|
||
|
|
except ValueError:
|
||
|
|
error_fields.append(field_name)
|
||
|
|
|
||
|
|
if dest_model._name == 'mail.mail' and field_name == 'email_from':
|
||
|
|
# As the "email_from" is used to populate the email_from of the
|
||
|
|
# sent mail.mail, it could be filtered out at sending time if no
|
||
|
|
# outgoing mail server "from_filter" match the sender email.
|
||
|
|
# To make sure the email contains that (important) information
|
||
|
|
# we also add it to the "custom message" that will be included
|
||
|
|
# in the body of the email sent.
|
||
|
|
custom_fields.append((_('email'), field_value))
|
||
|
|
|
||
|
|
# If it's a custom field
|
||
|
|
elif field_name not in ('context', 'website_form_signature'):
|
||
|
|
custom_fields.append((field_name, field_value))
|
||
|
|
|
||
|
|
data['custom'] = "\n".join([u"%s : %s" % v for v in custom_fields])
|
||
|
|
|
||
|
|
# Add metadata if enabled # ICP for retrocompatibility
|
||
|
|
if request.env['ir.config_parameter'].sudo().get_param('website_form_enable_metadata'):
|
||
|
|
environ = request.httprequest.headers.environ
|
||
|
|
data['meta'] += "%s : %s\n%s : %s\n%s : %s\n%s : %s\n" % (
|
||
|
|
"IP", environ.get("REMOTE_ADDR"),
|
||
|
|
"USER_AGENT", environ.get("HTTP_USER_AGENT"),
|
||
|
|
"ACCEPT_LANGUAGE", environ.get("HTTP_ACCEPT_LANGUAGE"),
|
||
|
|
"REFERER", environ.get("HTTP_REFERER")
|
||
|
|
)
|
||
|
|
|
||
|
|
# This function can be defined on any model to provide
|
||
|
|
# a model-specific filtering of the record values
|
||
|
|
# Example:
|
||
|
|
# def website_form_input_filter(self, values):
|
||
|
|
# values['name'] = '%s\'s Application' % values['partner_name']
|
||
|
|
# return values
|
||
|
|
if hasattr(dest_model, "website_form_input_filter"):
|
||
|
|
data['record'] = dest_model.website_form_input_filter(request, data['record'])
|
||
|
|
|
||
|
|
missing_required_fields = [label for label, field in authorized_fields.items() if field['required'] and label not in data['record']]
|
||
|
|
if any(error_fields):
|
||
|
|
raise ValidationError(error_fields + missing_required_fields)
|
||
|
|
|
||
|
|
return data
|
||
|
|
|
||
|
|
def insert_record(self, request, model, values, custom, meta=None):
|
||
|
|
model_name = model.sudo().model
|
||
|
|
if model_name == 'mail.mail':
|
||
|
|
email_from = _('"%s form submission" <%s>') % (request.env.company.name, request.env.company.email)
|
||
|
|
values.update({'reply_to': values.get('email_from'), 'email_from': email_from})
|
||
|
|
record = request.env[model_name].with_user(SUPERUSER_ID).with_context(
|
||
|
|
mail_create_nosubscribe=True,
|
||
|
|
).create(values)
|
||
|
|
if custom or meta:
|
||
|
|
_custom_label = "%s\n___________\n\n" % _("Other Information:") # Title for custom fields
|
||
|
|
if model_name == 'mail.mail':
|
||
|
|
_custom_label = "%s\n___________\n\n" % _("This message has been posted on your website!")
|
||
|
|
default_field = model.website_form_default_field_id
|
||
|
|
default_field_data = values.get(default_field.name, '')
|
||
|
|
custom_content = (default_field_data + "\n\n" if default_field_data else '') \
|
||
|
|
+ (_custom_label + custom + "\n\n" if custom else '') \
|
||
|
|
+ (self._meta_label + "\n________\n\n" + meta if meta else '')
|
||
|
|
|
||
|
|
# If there is a default field configured for this model, use it.
|
||
|
|
# If there isn't, put the custom data in a message instead
|
||
|
|
if default_field.name:
|
||
|
|
if default_field.ttype == 'html' or model_name == 'mail.mail':
|
||
|
|
custom_content = nl2br_enclose(custom_content)
|
||
|
|
record.update({default_field.name: custom_content})
|
||
|
|
elif hasattr(record, '_message_log'):
|
||
|
|
record._message_log(
|
||
|
|
body=nl2br_enclose(custom_content, 'p'),
|
||
|
|
message_type='comment',
|
||
|
|
)
|
||
|
|
|
||
|
|
return record.id
|
||
|
|
|
||
|
|
# Link all files attached on the form
|
||
|
|
def insert_attachment(self, model, id_record, files):
|
||
|
|
orphan_attachment_ids = []
|
||
|
|
model_name = model.sudo().model
|
||
|
|
record = model.env[model_name].browse(id_record)
|
||
|
|
authorized_fields = model.with_user(SUPERUSER_ID)._get_form_writable_fields()
|
||
|
|
for file in files:
|
||
|
|
custom_field = file.field_name not in authorized_fields
|
||
|
|
attachment_value = {
|
||
|
|
'name': file.filename,
|
||
|
|
'datas': base64.encodebytes(file.read()),
|
||
|
|
'res_model': model_name,
|
||
|
|
'res_id': record.id,
|
||
|
|
}
|
||
|
|
attachment_id = request.env['ir.attachment'].sudo().create(attachment_value)
|
||
|
|
if attachment_id and not custom_field:
|
||
|
|
record_sudo = record.sudo()
|
||
|
|
value = [(4, attachment_id.id)]
|
||
|
|
if record_sudo._fields[file.field_name].type == 'many2one':
|
||
|
|
value = attachment_id.id
|
||
|
|
record_sudo[file.field_name] = value
|
||
|
|
else:
|
||
|
|
orphan_attachment_ids.append(attachment_id.id)
|
||
|
|
|
||
|
|
if model_name != 'mail.mail' and hasattr(record, '_message_log') and orphan_attachment_ids:
|
||
|
|
# If some attachments didn't match a field on the model,
|
||
|
|
# we create a mail.message to link them to the record
|
||
|
|
record._message_log(
|
||
|
|
attachment_ids=[(6, 0, orphan_attachment_ids)],
|
||
|
|
body=Markup(_('<p>Attached files: </p>')),
|
||
|
|
message_type='comment',
|
||
|
|
)
|
||
|
|
elif model_name == 'mail.mail' and orphan_attachment_ids:
|
||
|
|
# If the model is mail.mail then we have no other choice but to
|
||
|
|
# attach the custom binary field files on the attachment_ids field.
|
||
|
|
for attachment_id_id in orphan_attachment_ids:
|
||
|
|
record.attachment_ids = [(4, attachment_id_id)]
|