import warnings 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 collect_all_methods, get_anymail_setting, get_request_basic_auth # Mixin note: Django's View.__init__ doesn't cooperate with chaining, # so all mixins that need __init__ must appear before View in MRO. class AnymailCoreWebhookView(View): """Common view for processing ESP event webhooks 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().__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().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. Subclasses must override with class attr. E.g.: esp_name = "Postmark" esp_name = "SendGrid" # (use ESP's preferred capitalization) """ 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. # (Declaring class attr allows override by kwargs in View.as_view.): basic_auth = None def __init__(self, **kwargs): self.basic_auth = get_anymail_setting( "webhook_secret", default=[], # no esp_name -- auth is shared between ESPs kwargs=kwargs, ) # 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