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
|
import re
import warnings
import six
from django.http import HttpResponse
from django.utils.crypto import constant_time_compare
from django.utils.decorators import method_decorator
from django.views.decorators.csrf import csrf_exempt
from django.views.generic import View
from ..exceptions import AnymailInsecureWebhookWarning, AnymailWebhookValidationFailure
from ..utils import get_anymail_setting, collect_all_methods, get_request_basic_auth
class AnymailBasicAuthMixin(object):
"""Implements webhook basic auth as mixin to AnymailBaseWebhookView."""
# Whether to warn if basic auth is not configured.
# For most ESPs, basic auth is the only webhook security,
# so the default is True. Subclasses can set False if
# they enforce other security (like signed webhooks).
warn_if_no_basic_auth = True
# List of allowable HTTP basic-auth 'user:pass' strings.
basic_auth = None # (Declaring class attr allows override by kwargs in View.as_view.)
def __init__(self, **kwargs):
self.basic_auth = get_anymail_setting('webhook_authorization', default=[],
kwargs=kwargs) # no esp_name -- auth is shared between ESPs
# Allow a single string:
if isinstance(self.basic_auth, six.string_types):
self.basic_auth = [self.basic_auth]
if self.warn_if_no_basic_auth and len(self.basic_auth) < 1:
warnings.warn(
"Your Anymail webhooks are insecure and open to anyone on the web. "
"You should set WEBHOOK_AUTHORIZATION in your ANYMAIL settings. "
"See 'Securing webhooks' in the Anymail docs.",
AnymailInsecureWebhookWarning)
# noinspection PyArgumentList
super(AnymailBasicAuthMixin, self).__init__(**kwargs)
def validate_request(self, request):
"""If configured for webhook basic auth, validate request has correct auth."""
if self.basic_auth:
request_auth = get_request_basic_auth(request)
# Use constant_time_compare to avoid timing attack on basic auth. (It's OK that any()
# can terminate early: we're not trying to protect how many auth strings are allowed,
# just the contents of each individual auth string.)
auth_ok = any(constant_time_compare(request_auth, allowed_auth)
for allowed_auth in self.basic_auth)
if not auth_ok:
# noinspection PyUnresolvedReferences
raise AnymailWebhookValidationFailure(
"Missing or invalid basic auth in Anymail %s webhook" % self.esp_name)
# Mixin note: Django's View.__init__ doesn't cooperate with chaining,
# so all mixins that need __init__ must appear before View in MRO.
class AnymailBaseWebhookView(AnymailBasicAuthMixin, View):
"""Base view for processing ESP event webhooks
ESP-specific implementations should subclass
and implement parse_events. They may also
want to implement validate_request
if additional security is available.
"""
def __init__(self, **kwargs):
super(AnymailBaseWebhookView, self).__init__(**kwargs)
self.validators = collect_all_methods(self.__class__, 'validate_request')
# Subclass implementation:
# Where to send events: either ..signals.inbound or ..signals.tracking
signal = None
def validate_request(self, request):
"""Check validity of webhook post, or raise AnymailWebhookValidationFailure.
AnymailBaseWebhookView includes basic auth validation.
Subclasses can implement (or provide via mixins) if the ESP supports
additional validation (such as signature checking).
*All* definitions of this method in the class chain (including mixins)
will be called. There is no need to chain to the superclass.
(See self.run_validators and collect_all_methods.)
Security note: use django.utils.crypto.constant_time_compare for string
comparisons, to avoid exposing your validation to a timing attack.
"""
# if not constant_time_compare(request.POST['signature'], expected_signature):
# raise AnymailWebhookValidationFailure("...message...")
# (else just do nothing)
pass
def parse_events(self, request):
"""Return a list of normalized AnymailWebhookEvent extracted from ESP post data.
Subclasses must implement.
"""
raise NotImplementedError()
# HTTP handlers (subclasses shouldn't need to override):
http_method_names = ["post", "head", "options"]
@method_decorator(csrf_exempt)
def dispatch(self, request, *args, **kwargs):
return super(AnymailBaseWebhookView, self).dispatch(request, *args, **kwargs)
def head(self, request, *args, **kwargs):
# Some ESPs verify the webhook with a HEAD request at configuration time
return HttpResponse()
def post(self, request, *args, **kwargs):
# Normal Django exception handling will do the right thing:
# - AnymailWebhookValidationFailure will turn into an HTTP 400 response
# (via Django SuspiciousOperation handling)
# - Any other errors (e.g., in signal dispatch) will turn into HTTP 500
# responses (via normal Django error handling). ESPs generally
# treat that as "try again later".
self.run_validators(request)
events = self.parse_events(request)
esp_name = self.esp_name
for event in events:
self.signal.send(sender=self.__class__, event=event, esp_name=esp_name)
return HttpResponse()
# Request validation (subclasses shouldn't need to override):
def run_validators(self, request):
for validator in self.validators:
validator(self, request)
@property
def esp_name(self):
"""
Read-only name of the ESP for this webhook view.
(E.g., MailgunTrackingWebhookView will return "Mailgun")
"""
return re.sub(r'(Tracking|Inbox)WebhookView$', "", self.__class__.__name__)
|
