1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
import json
from datetime import datetime
import hashlib
import hmac
from base64 import b64encode
from django.utils.crypto import constant_time_compare
from django.utils.timezone import utc
from .base import AnymailBaseWebhookView
from ..exceptions import AnymailWebhookValidationFailure, AnymailConfigurationError
from ..signals import tracking, AnymailTrackingEvent, EventType
from ..utils import get_anymail_setting, getfirst, get_request_uri
class MandrillSignatureMixin(object):
"""Validates Mandrill webhook signature"""
# These can be set from kwargs in View.as_view, or pulled from settings in init:
webhook_key = None # required
webhook_url = None # optional; defaults to actual url used
def __init__(self, **kwargs):
# noinspection PyUnresolvedReferences
esp_name = self.esp_name
# webhook_key is required for POST, but not for HEAD when Mandrill validates webhook url.
# Defer "missing setting" error until we actually try to use it in the POST...
webhook_key = get_anymail_setting('webhook_key', esp_name=esp_name, default=None,
kwargs=kwargs, allow_bare=True)
if webhook_key is not None:
self.webhook_key = webhook_key.encode('ascii') # hmac.new requires bytes key in python 3
self.webhook_url = get_anymail_setting('webhook_url', esp_name=esp_name, default=None,
kwargs=kwargs, allow_bare=True)
# noinspection PyArgumentList
super(MandrillSignatureMixin, self).__init__(**kwargs)
def validate_request(self, request):
if self.webhook_key is None:
# issue deferred "missing setting" error (re-call get-setting without a default)
# noinspection PyUnresolvedReferences
get_anymail_setting('webhook_key', esp_name=self.esp_name, allow_bare=True)
try:
signature = request.META["HTTP_X_MANDRILL_SIGNATURE"]
except KeyError:
raise AnymailWebhookValidationFailure("X-Mandrill-Signature header missing from webhook POST")
# Mandrill signs the exact URL (including basic auth, if used) plus the sorted POST params:
url = self.webhook_url or get_request_uri(request)
params = request.POST.dict()
signed_data = url
for key in sorted(params.keys()):
signed_data += key + params[key]
expected_signature = b64encode(hmac.new(key=self.webhook_key, msg=signed_data.encode('utf-8'),
digestmod=hashlib.sha1).digest())
if not constant_time_compare(signature, expected_signature):
raise AnymailWebhookValidationFailure(
"Mandrill webhook called with incorrect signature (for url %r)" % url)
class MandrillBaseWebhookView(MandrillSignatureMixin, AnymailBaseWebhookView):
"""Base view class for Mandrill webhooks"""
warn_if_no_basic_auth = False # because we validate against signature
def parse_events(self, request):
esp_events = json.loads(request.POST['mandrill_events'])
return [self.esp_to_anymail_event(esp_event) for esp_event in esp_events]
def esp_to_anymail_event(self, esp_event):
raise NotImplementedError()
class MandrillTrackingWebhookView(MandrillBaseWebhookView):
signal = tracking
event_types = {
# Message events:
'send': EventType.SENT,
'deferral': EventType.DEFERRED,
'hard_bounce': EventType.BOUNCED,
'soft_bounce': EventType.DEFERRED,
'open': EventType.OPENED,
'click': EventType.CLICKED,
'spam': EventType.COMPLAINED,
'unsub': EventType.UNSUBSCRIBED,
'reject': EventType.REJECTED,
# Sync events (we don't really normalize these well):
'whitelist': EventType.UNKNOWN,
'blacklist': EventType.UNKNOWN,
# Inbound events:
'inbound': EventType.INBOUND,
}
def esp_to_anymail_event(self, esp_event):
esp_type = getfirst(esp_event, ['event', 'type'], None)
event_type = self.event_types.get(esp_type, EventType.UNKNOWN)
if event_type == EventType.INBOUND:
raise AnymailConfigurationError(
"You seem to have set Mandrill's *inbound* webhook URL "
"to Anymail's Mandrill *tracking* webhook URL.")
try:
timestamp = datetime.fromtimestamp(esp_event['ts'], tz=utc)
except (KeyError, ValueError):
timestamp = None
try:
recipient = esp_event['msg']['email']
except KeyError:
try:
recipient = esp_event['reject']['email'] # sync events
except KeyError:
recipient = None
try:
mta_response = esp_event['msg']['diag']
except KeyError:
mta_response = None
try:
description = getfirst(esp_event['reject'], ['detail', 'reason'])
except KeyError:
description = None
try:
metadata = esp_event['msg']['metadata']
except KeyError:
metadata = None
try:
tags = esp_event['msg']['tags']
except KeyError:
tags = None
return AnymailTrackingEvent(
click_url=esp_event.get('url', None),
description=description,
esp_event=esp_event,
event_type=event_type,
message_id=esp_event.get('_id', None),
metadata=metadata,
mta_response=mta_response,
recipient=recipient,
reject_reason=None, # probably map esp_event['msg']['bounce_description'], but insufficient docs
tags=tags,
timestamp=timestamp,
user_agent=esp_event.get('user_agent', None),
)
|
