|
|
# -*- coding: utf-8 -*-# Copyright 2016 Tecnativa - Antonio Espinosa# Copyright 2017 Tecnativa - David Vidal# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
import hashlibimport hmacimport jsonimport requestsfrom datetime import datetimefrom odoo import _, api, fields, modelsfrom odoo.exceptions import UserError, ValidationErrorfrom odoo.tools import email_split
import logging_logger = logging.getLogger(__name__)
class MailTrackingEmail(models.Model): _inherit = "mail.tracking.email"
def _country_search(self, country_code): country = False if country_code: country = self.env['res.country'].search([ ('code', '=', country_code.upper()), ]) if country: return country.id return False
@property def _mailgun_mandatory_fields(self): return ('event', 'timestamp', 'token', 'signature', 'tracking_email_id', 'odoo_db')
@property def _mailgun_event_type_mapping(self): return { # Mailgun event type: tracking event type 'delivered': 'delivered', 'opened': 'open', 'clicked': 'click', 'unsubscribed': 'unsub', 'complained': 'spam', 'bounced': 'hard_bounce', 'dropped': 'reject', 'accepted': 'sent', }
def _mailgun_event_type_verify(self, event): event = event or {} mailgun_event_type = event.get('event') if mailgun_event_type not in self._mailgun_event_type_mapping: _logger.error("Mailgun: event type '%s' not supported", mailgun_event_type) return False # OK, event type is valid return True
def _mailgun_signature(self, api_key, timestamp, token): return hmac.new( key=str(api_key), msg='{}{}'.format(str(timestamp), str(token)), digestmod=hashlib.sha256).hexdigest()
def _mailgun_values(self): icp = self.env['ir.config_parameter'] api_key = icp.get_param('mailgun.apikey') if not api_key: raise ValidationError(_('There is no Mailgun API key!')) api_url = icp.get_param( 'mailgun.api_url', 'https://api.mailgun.net/v3') domain = icp.get_param('mail.catchall.domain') if not domain: raise ValidationError(_('A Mailgun domain value is needed!')) validation_key = icp.get_param('mailgun.validation_key') return api_key, api_url, domain, validation_key
def _mailgun_signature_verify(self, event): event = event or {} api_key = self.env['ir.config_parameter'].get_param('mailgun.apikey') if not api_key: _logger.warning("No Mailgun api key configured. " "Please add 'mailgun.apikey' to System parameters " "to enable Mailgun authentication webhoook " "requests. More info at: " "https://documentation.mailgun.com/" "user_manual.html#webhooks") else: timestamp = event.get('timestamp') token = event.get('token') signature = event.get('signature') event_digest = self._mailgun_signature(api_key, timestamp, token) if signature != event_digest: _logger.error("Mailgun: Invalid signature '%s' != '%s'", signature, event_digest) return False # OK, signature is valid return True
def _db_verify(self, event): event = event or {} odoo_db = event.get('odoo_db') current_db = self.env.cr.dbname if odoo_db != current_db: _logger.error("Mailgun: Database '%s' is not the current database", odoo_db) return False # OK, DB is current return True
def _mailgun_metadata(self, mailgun_event_type, event, metadata): # Get Mailgun timestamp when found ts = event.get('timestamp', False) try: ts = float(ts) except: ts = False if ts: dt = datetime.utcfromtimestamp(ts) metadata.update({ 'timestamp': ts, 'time': fields.Datetime.to_string(dt), 'date': fields.Date.to_string(dt), 'mailgun_id': event.get('id', False) }) # Common field mapping mapping = { 'recipient': 'recipient', 'ip': 'ip', 'user_agent': 'user-agent', 'os_family': 'client-os', 'ua_family': 'client-name', 'ua_type': 'client-type', 'url': 'url', } for k, v in mapping.iteritems(): if event.get(v, False): metadata[k] = event[v] # Special field mapping metadata.update({ 'mobile': event.get('device-type') in {'mobile', 'tablet'}, 'user_country_id': self._country_search( event.get('country', False)), }) # Mapping for special events if mailgun_event_type == 'bounced': metadata.update({ 'error_type': event.get('code', False), 'error_description': event.get('error', False), 'error_details': event.get('notification', False), }) elif mailgun_event_type == 'dropped': metadata.update({ 'error_type': event.get('reason', False), 'error_description': event.get('code', False), 'error_details': event.get('description', False), }) elif mailgun_event_type == 'complained': metadata.update({ 'error_type': 'spam', 'error_description': "Recipient '%s' mark this email as spam" % event.get('recipient', False), }) return metadata
def _mailgun_tracking_get(self, event): tracking = False tracking_email_id = event.get('tracking_email_id', False) if tracking_email_id and tracking_email_id.isdigit(): tracking = self.search([('id', '=', tracking_email_id)], limit=1) return tracking
def _event_is_from_mailgun(self, event): event = event or {} return all([k in event for k in self._mailgun_mandatory_fields])
@api.model def event_process(self, request, post, metadata, event_type=None): res = super(MailTrackingEmail, self).event_process( request, post, metadata, event_type=event_type) if res == 'NONE' and self._event_is_from_mailgun(post): if not self._mailgun_signature_verify(post): res = 'ERROR: Signature' elif not self._mailgun_event_type_verify(post): res = 'ERROR: Event type not supported' elif not self._db_verify(post): res = 'ERROR: Invalid DB' else: res = 'OK' if res == 'OK': mailgun_event_type = post.get('event') mapped_event_type = self._mailgun_event_type_mapping.get( mailgun_event_type) or event_type if not mapped_event_type: # pragma: no cover res = 'ERROR: Bad event' tracking = self._mailgun_tracking_get(post) if not tracking: res = 'ERROR: Tracking not found' if res == 'OK': # Complete metadata with mailgun event info metadata = self._mailgun_metadata( mailgun_event_type, post, metadata) # Create event tracking.event_create(mapped_event_type, metadata) if res != 'NONE': if event_type: _logger.info( "Mailgun: event '%s' process '%s'", event_type, res) else: _logger.info("Mailgun: event process '%s'", res) return res
@api.multi def action_manual_check_mailgun(self): """
Manual check against Mailgun API API Documentation: https://documentation.mailgun.com/en/latest/api-events.html """
api_key, api_url, domain, validation_key = self._mailgun_values() for tracking in self: if not tracking.mail_message_id: raise UserError(_('There is no tracked message!')) message_id = tracking.mail_message_id.message_id.replace( "<", "").replace(">", "") res = requests.get( '%s/%s/events' % (api_url, domain), auth=("api", api_key), params={ "begin": tracking.timestamp, "ascending": "yes", "message-id": message_id, } ) if not res or res.status_code != 200: raise ValidationError(_( "Couldn't retrieve Mailgun information")) content = json.loads(res.content, res.apparent_encoding) if "items" not in content: raise ValidationError(_("Event information not longer stored")) for item in content["items"]: if not self.env['mail.tracking.event'].search( # mailgun event hasn't been synced and recipient is the same as # in the evaluated tracking. We use email_split since tracking # recipient could come in format: "example" <to@dest.com> [('mailgun_id', '=', item["id"])]) and ( item.get("recipient", "") == email_split(tracking.recipient)[0]): mapped_event_type = self._mailgun_event_type_mapping.get( item["event"], item["event"]) metadata = self._mailgun_metadata( mapped_event_type, item, {}) tracking.event_create(mapped_event_type, metadata)
@api.multi def check_email_list_validity(self, email_list): """
Checks email list validity with Mailgun's API API documentation: https://documentation.mailgun.com/en/latest/api-email-validation.html """
api_key, api_url, domain, validation_key = self.env[ 'mail.tracking.email']._mailgun_values() if not validation_key: raise UserError(_('You need to configure mailgun.validation_key' ' in order to be able to check mails validity')) result = {} for email in email_list: res = requests.get( "%s/address/validate" % api_url, auth=("api", validation_key), params={ "address": email, "mailbox_verification": True, }) if not res or res.status_code != 200: result[email] = {'result': (_( 'Error %s trying to ' 'check mail' % res.status_code or 'of connection'))} continue content = json.loads(res.content, res.apparent_encoding) if 'mailbox_verification' not in content: result[email] = {'result': ( _("Mailgun Error. Mailbox verification value wasn't" " returned"))} continue # Not a valid address: API sets 'is_valid' as False # and 'mailbox_verification' as None if not content['is_valid']: result[email] = {'result': ( _('%s is not a valid email address. Please check it ' 'in order to avoid sending issues') % (email))} continue # If the mailbox is not valid API returns 'mailbox_verification' # as a string with value 'false' if content['mailbox_verification'] == 'false': result[email] = {'result': ( _('%s failed the mailbox verification. Please check it ' 'in order to avoid sending issues') % (email))} continue # If Mailgun can't complete the validation request the API returns # 'mailbox_verification' as a string set to 'unknown' if content['mailbox_verification'] == 'unknown': result[email] = {'result': ( _("%s couldn't be verified. Either the request couln't be " "completed or the mailbox provider doesn't support " "email verification") % (email))} continue result[email] = {'result': _("The mailbox is correct")} return result
|
