pacnpal/django-anymail
1
0
Fork 0
mirror of https://github.com/pacnpal/django-anymail.git synced 2025-12-20 03:41:05 -05:00
Code Issues Packages Projects Releases Wiki Activity

Drop Python 2 and Django 1.11 support

Browse Source 
Minimum supported versions are now Django 2.0, Python 3.5.

This touches a lot of code, to:
* Remove obsolete portability code and workarounds
  (six, backports of email parsers, test utils, etc.)
* Use Python 3 syntax (class defs, raise ... from, etc.)
* Correct inheritance for mixin classes
* Fix outdated docs content and links
* Suppress Python 3 "unclosed SSLSocket" ResourceWarnings
  that are beyond our control (in integration tests due to boto3, 
  python-sparkpost)
This commit is contained in:
Mike Edmunds
2020-08-01 14:53:10 -07:00
committed by GitHub
parent c803108481
commit 85cec5e9dc
87 changed files with 672 additions and 1278 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
anymail/webhooks/amazon_ses.py
View File

@@ -11,7 +11,7 @@ from ..exceptions import (
_LazyError)
from ..inbound import AnymailInboundMessage
from ..signals import AnymailInboundEvent, AnymailTrackingEvent, EventType, RejectReason, inbound, tracking
from ..utils import combine, get_anymail_setting, getfirst
from ..utils import get_anymail_setting, getfirst
try:
import boto3
@@ -37,7 +37,7 @@ class AmazonSESBaseWebhookView(AnymailBaseWebhookView):
"auto_confirm_sns_subscriptions", esp_name=self.esp_name, kwargs=kwargs, default=True)
# boto3 params for connecting to S3 (inbound downloads) and SNS (auto-confirm subscriptions):
self.session_params, self.client_params = _get_anymail_boto3_params(kwargs=kwargs)
super(AmazonSESBaseWebhookView, self).__init__(**kwargs)
super().__init__(**kwargs)
@staticmethod
def _parse_sns_message(request):
@@ -47,7 +47,7 @@ class AmazonSESBaseWebhookView(AnymailBaseWebhookView):
body = request.body.decode(request.encoding or 'utf-8')
request._sns_message = json.loads(body)
except (TypeError, ValueError, UnicodeDecodeError) as err:
raise AnymailAPIError("Malformed SNS message body %r" % request.body, raised_from=err)
raise AnymailAPIError("Malformed SNS message body %r" % request.body) from err
return request._sns_message
def validate_request(self, request):
@@ -80,7 +80,7 @@ class AmazonSESBaseWebhookView(AnymailBaseWebhookView):
response = HttpResponse(status=401)
response["WWW-Authenticate"] = 'Basic realm="Anymail WEBHOOK_SECRET"'
return response
return super(AmazonSESBaseWebhookView, self).post(request, *args, **kwargs)
return super().post(request, *args, **kwargs)
def parse_events(self, request):
# request *has* been validated by now
@@ -91,11 +91,11 @@ class AmazonSESBaseWebhookView(AnymailBaseWebhookView):
message_string = sns_message.get("Message")
try:
ses_event = json.loads(message_string)
except (TypeError, ValueError):
except (TypeError, ValueError) as err:
if message_string == "Successfully validated SNS topic for Amazon SES event publishing.":
pass # this Notification is generated after SubscriptionConfirmation
else:
raise AnymailAPIError("Unparsable SNS Message %r" % message_string)
raise AnymailAPIError("Unparsable SNS Message %r" % message_string) from err
else:
events = self.esp_to_anymail_events(ses_event, sns_message)
elif sns_type == "SubscriptionConfirmation":
@@ -258,8 +258,7 @@ class AmazonSESTrackingWebhookView(AmazonSESBaseWebhookView):
)
return [
# AnymailTrackingEvent(**common_props, **recipient_props) # Python 3.5+ (PEP-448 syntax)
AnymailTrackingEvent(**combine(common_props, recipient_props))
AnymailTrackingEvent(**common_props, **recipient_props)
for recipient_props in per_recipient_props
]
@@ -306,7 +305,7 @@ class AmazonSESInboundWebhookView(AmazonSESBaseWebhookView):
raise AnymailBotoClientAPIError(
"Anymail AmazonSESInboundWebhookView couldn't download S3 object '{bucket_name}:{object_key}'"
"".format(bucket_name=bucket_name, object_key=object_key),
raised_from=err)
client_error=err) from err
finally:
content.close()
else:
@@ -341,13 +340,9 @@ class AmazonSESInboundWebhookView(AmazonSESBaseWebhookView):
class AnymailBotoClientAPIError(AnymailAPIError, ClientError):
"""An AnymailAPIError that is also a Boto ClientError"""
def __init__(self, *args, **kwargs):
raised_from = kwargs.pop('raised_from')
assert isinstance(raised_from, ClientError)
assert len(kwargs) == 0 # can't support other kwargs
def __init__(self, *args, client_error):
assert isinstance(client_error, ClientError)
# init self as boto ClientError (which doesn't cooperatively subclass):
super(AnymailBotoClientAPIError, self).__init__(
error_response=raised_from.response, operation_name=raised_from.operation_name)
super().__init__(error_response=client_error.response, operation_name=client_error.operation_name)
# emulate AnymailError init:
self.args = args
self.raised_from = raised_from

108
anymail/webhooks/base.py
View File

@@ -1,6 +1,5 @@
import warnings
import six
from django.http import HttpResponse
from django.utils.crypto import constant_time_compare
from django.utils.decorators import method_decorator
@@ -11,62 +10,21 @@ from ..exceptions import AnymailInsecureWebhookWarning, AnymailWebhookValidation
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_secret', 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_SECRET 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
class AnymailCoreWebhookView(View):
"""Common view for processing ESP event webhooks
ESP-specific implementations should subclass
and implement parse_events. They may also
want to implement validate_request
ESP-specific implementations will need to implement parse_events.
ESP-specific implementations should generally subclass
AnymailBaseWebhookView instead, to pick up basic auth.
They may also want to implement validate_request
if additional security is available.
"""
def __init__(self, **kwargs):
super(AnymailBaseWebhookView, self).__init__(**kwargs)
super().__init__(**kwargs)
self.validators = collect_all_methods(self.__class__, 'validate_request')
# Subclass implementation:
@@ -106,7 +64,7 @@ class AnymailBaseWebhookView(AnymailBasicAuthMixin, View):
@method_decorator(csrf_exempt)
def dispatch(self, request, *args, **kwargs):
return super(AnymailBaseWebhookView, self).dispatch(request, *args, **kwargs)
return super().dispatch(request, *args, **kwargs)
def head(self, request, *args, **kwargs):
# Some ESPs verify the webhook with a HEAD request at configuration time
@@ -143,3 +101,51 @@ class AnymailBaseWebhookView(AnymailBasicAuthMixin, View):
"""
raise NotImplementedError("%s.%s must declare esp_name class attr" %
(self.__class__.__module__, self.__class__.__name__))
class AnymailBasicAuthMixin(AnymailCoreWebhookView):
"""Implements webhook basic auth as mixin to AnymailCoreWebhookView."""
# 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_secret', default=[],
kwargs=kwargs) # no esp_name -- auth is shared between ESPs
# Allow a single string:
if isinstance(self.basic_auth, str):
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_SECRET in your ANYMAIL settings. "
"See 'Securing webhooks' in the Anymail docs.",
AnymailInsecureWebhookWarning)
super().__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:
raise AnymailWebhookValidationFailure(
"Missing or invalid basic auth in Anymail %s webhook" % self.esp_name)
class AnymailBaseWebhookView(AnymailBasicAuthMixin, AnymailCoreWebhookView):
"""
Abstract base class for most webhook views, enforcing HTTP basic auth security
"""
pass

16
anymail/webhooks/mailgun.py
View File

@@ -30,11 +30,11 @@ class MailgunBaseWebhookView(AnymailBaseWebhookView):
kwargs=kwargs, allow_bare=True, default=None)
webhook_signing_key = get_anymail_setting('webhook_signing_key', esp_name=self.esp_name,
kwargs=kwargs, default=UNSET if api_key is None else api_key)
self.webhook_signing_key = webhook_signing_key.encode('ascii') # hmac.new requires bytes key in python 3
super(MailgunBaseWebhookView, self).__init__(**kwargs)
self.webhook_signing_key = webhook_signing_key.encode('ascii') # hmac.new requires bytes key
super().__init__(**kwargs)
def validate_request(self, request):
super(MailgunBaseWebhookView, self).validate_request(request) # first check basic auth if enabled
super().validate_request(request) # first check basic auth if enabled
if request.content_type == "application/json":
# New-style webhook: json payload with separate signature block
try:
@@ -45,8 +45,7 @@ class MailgunBaseWebhookView(AnymailBaseWebhookView):
signature = signature_block['signature']
except (KeyError, ValueError, UnicodeDecodeError) as err:
raise AnymailWebhookValidationFailure(
"Mailgun webhook called with invalid payload format",
raised_from=err)
"Mailgun webhook called with invalid payload format") from err
else:
# Legacy webhook: signature fields are interspersed with other POST data
try:
@@ -54,9 +53,10 @@ class MailgunBaseWebhookView(AnymailBaseWebhookView):
# (Fortunately, Django QueryDict is specced to return the last value.)
token = request.POST['token']
timestamp = request.POST['timestamp']
signature = str(request.POST['signature']) # force to same type as hexdigest() (for python2)
except KeyError:
raise AnymailWebhookValidationFailure("Mailgun webhook called without required security fields")
signature = request.POST['signature']
except KeyError as err:
raise AnymailWebhookValidationFailure(
"Mailgun webhook called without required security fields") from err
expected_signature = hmac.new(key=self.webhook_signing_key, msg='{}{}'.format(timestamp, token).encode('ascii'),
digestmod=hashlib.sha256).hexdigest()

13
anymail/webhooks/mandrill.py
View File

@@ -7,14 +7,14 @@ from base64 import b64encode
from django.utils.crypto import constant_time_compare
from django.utils.timezone import utc
from .base import AnymailBaseWebhookView
from .base import AnymailBaseWebhookView, AnymailCoreWebhookView
from ..exceptions import AnymailWebhookValidationFailure
from ..inbound import AnymailInboundMessage
from ..signals import inbound, tracking, AnymailInboundEvent, AnymailTrackingEvent, EventType
from ..utils import get_anymail_setting, getfirst, get_request_uri
class MandrillSignatureMixin(object):
class MandrillSignatureMixin(AnymailCoreWebhookView):
"""Validates Mandrill webhook signature"""
# These can be set from kwargs in View.as_view, or pulled from settings in init:
@@ -22,29 +22,26 @@ class MandrillSignatureMixin(object):
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_key = webhook_key.encode('ascii') # hmac.new requires bytes key
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)
super().__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")
raise AnymailWebhookValidationFailure("X-Mandrill-Signature header missing from webhook POST") from None
# Mandrill signs the exact URL (including basic auth, if used) plus the sorted POST params:
url = self.webhook_url or get_request_uri(request)

7
anymail/webhooks/sendgrid.py
View File

@@ -1,12 +1,13 @@
import json
from datetime import datetime
from email.parser import BytesParser
from email.policy import default as default_policy
from django.utils.timezone import utc
from .base import AnymailBaseWebhookView
from .._email_compat import EmailBytesParser
from ..inbound import AnymailInboundMessage
from ..signals import inbound, tracking, AnymailInboundEvent, AnymailTrackingEvent, EventType, RejectReason
from ..signals import AnymailInboundEvent, AnymailTrackingEvent, EventType, RejectReason, inbound, tracking
class SendGridTrackingWebhookView(AnymailBaseWebhookView):
@@ -204,7 +205,7 @@ class SendGridInboundWebhookView(AnymailBaseWebhookView):
b"\r\n\r\n",
request.body
])
parsed_parts = EmailBytesParser().parsebytes(raw_data).get_payload()
parsed_parts = BytesParser(policy=default_policy).parsebytes(raw_data).get_payload()
for part in parsed_parts:
name = part.get_param('name', header='content-disposition')
if name == 'text':

3
anymail/webhooks/sparkpost.py
View File

@@ -40,7 +40,8 @@ class SparkPostBaseWebhookView(AnymailBaseWebhookView):
# Empty event (SparkPost sometimes sends as a "ping")
event_class = event = None
else:
raise TypeError("Invalid SparkPost webhook event has multiple event classes: %r" % raw_event)
raise TypeError(
"Invalid SparkPost webhook event has multiple event classes: %r" % raw_event) from None
return event_class, event, raw_event
def esp_to_anymail_event(self, event_class, event, raw_event):