0
# Webhooks
1
2
Secure webhook integration for processing SendGrid delivery events, bounce notifications, and email engagement tracking. Includes signature verification to ensure webhook authenticity and prevent unauthorized access.
3
4
## Capabilities
5
6
### Webhook Signature Verification
7
8
Cryptographic verification of SendGrid webhook signatures to ensure request authenticity and security.
9
10
```python { .api }
11
@verify_sendgrid_webhook_signature
12
def webhook_view(request):
13
"""
14
Decorator for Django views to verify SendGrid webhook signatures.
15
16
Requires:
17
- SendGrid v6+
18
- SENDGRID_WEBHOOK_VERIFICATION_KEY in Django settings
19
20
Returns HttpResponseNotFound() if signature verification fails.
21
Works with both sync and async view functions.
22
"""
23
24
def check_sendgrid_signature(request) -> bool:
25
"""
26
Verify SendGrid webhook signature against request.
27
28
Parameters:
29
- request (HttpRequest): Django request object containing headers and body
30
31
Returns:
32
- bool: True if signature is valid, False otherwise
33
34
Requires headers:
35
- X-Twilio-Email-Event-Webhook-Signature: Signature to verify
36
- X-Twilio-Email-Event-Webhook-Timestamp: Timestamp for signature
37
"""
38
```
39
40
### Webhook Configuration
41
42
Django settings and setup required for webhook signature verification.
43
44
```python { .api }
45
# Required Django setting
46
SENDGRID_WEBHOOK_VERIFICATION_KEY = str # Verification key from SendGrid console
47
```
48
49
#### Webhook Setup Example
50
51
1. Enable webhook signature verification in SendGrid:
52
53
```python
54
# In SendGrid console:
55
# 1. Go to Settings > Mail Settings > Event Webhook
56
# 2. Enable "Signed Event Webhook"
57
# 3. Copy the verification key
58
```
59
60
2. Configure Django settings:
61
62
```python
63
# settings.py
64
SENDGRID_WEBHOOK_VERIFICATION_KEY = os.environ.get("SENDGRID_WEBHOOK_KEY")
65
66
# Ensure key is present
67
if not SENDGRID_WEBHOOK_VERIFICATION_KEY:
68
raise ImproperlyConfigured("SENDGRID_WEBHOOK_VERIFICATION_KEY is required for webhook security")
69
```
70
71
3. Create webhook view:
72
73
```python
74
# views.py
75
import json
76
from django.http import HttpResponse, HttpRequest
77
from django.views.decorators.csrf import csrf_exempt
78
from django.views.decorators.http import require_POST
79
from sendgrid_backend.decorators import verify_sendgrid_webhook_signature
80
81
@csrf_exempt
82
@require_POST
83
@verify_sendgrid_webhook_signature
84
def sendgrid_webhook_handler(request: HttpRequest) -> HttpResponse:
85
"""Handle SendGrid webhook events with signature verification."""
86
try:
87
events = json.loads(request.body)
88
89
for event in events:
90
process_sendgrid_event(event)
91
92
return HttpResponse("OK")
93
94
except json.JSONDecodeError:
95
return HttpResponse("Invalid JSON", status=400)
96
except Exception as e:
97
# Log error but return success to avoid retries
98
logger.error(f"Webhook processing error: {e}")
99
return HttpResponse("Error processed")
100
101
def process_sendgrid_event(event):
102
"""Process individual SendGrid event."""
103
event_type = event.get('event')
104
email = event.get('email')
105
timestamp = event.get('timestamp')
106
107
# Handle different event types
108
if event_type == 'delivered':
109
handle_delivered_event(event)
110
elif event_type == 'bounce':
111
handle_bounce_event(event)
112
elif event_type == 'open':
113
handle_open_event(event)
114
# ... handle other events
115
```
116
117
### Event Processing Examples
118
119
Comprehensive webhook handler for different SendGrid events:
120
121
```python
122
import json
123
from datetime import datetime
124
from django.db import transaction
125
from django.http import HttpRequest, HttpResponse
126
from django.views.decorators.csrf import csrf_exempt
127
from django.views.decorators.http import require_POST
128
from pytz import utc
129
from sendgrid_backend.decorators import verify_sendgrid_webhook_signature
130
131
# Event type mapping
132
EVENT_TYPES = {
133
'processed': 'processed',
134
'delivered': 'delivered',
135
'bounce': 'bounced',
136
'dropped': 'dropped',
137
'deferred': 'deferred',
138
'open': 'opened',
139
'click': 'clicked',
140
'unsubscribe': 'unsubscribed',
141
'group_unsubscribe': 'group_unsubscribed',
142
'group_resubscribe': 'group_resubscribed',
143
'spamreport': 'spam_reported'
144
}
145
146
@csrf_exempt
147
@require_POST
148
@verify_sendgrid_webhook_signature
149
def comprehensive_webhook_handler(request: HttpRequest) -> HttpResponse:
150
"""
151
Comprehensive webhook handler for all SendGrid events.
152
153
Processes delivery status, engagement tracking, and list management events
154
with proper error handling and logging.
155
"""
156
try:
157
events = json.loads(request.body)
158
159
# Process events in reverse order (newest first)
160
for event_data in reversed(events):
161
with transaction.atomic():
162
process_webhook_event(event_data)
163
164
return HttpResponse("OK")
165
166
except json.JSONDecodeError:
167
logger.error("Invalid JSON in webhook payload")
168
return HttpResponse("Invalid JSON", status=400)
169
170
except Exception as e:
171
logger.error(f"Webhook processing error: {e}")
172
# Return success to prevent SendGrid retries for processing errors
173
return HttpResponse("Processing error logged")
174
175
def process_webhook_event(event_data):
176
"""Process individual webhook event with comprehensive handling."""
177
178
# Extract common event fields
179
event_type = event_data.get('event')
180
email_address = event_data.get('email')
181
timestamp = datetime.fromtimestamp(event_data.get('timestamp', 0), tz=utc)
182
message_id = event_data.get('sg_message_id') or event_data.get('smtp-id')
183
184
# Custom arguments for tracking
185
custom_args = event_data.get('asm', {}) # Unsubscribe group info
186
categories = event_data.get('category', [])
187
188
# Log event for debugging
189
logger.info(f"Processing {event_type} event for {email_address}")
190
191
# Handle specific event types
192
if event_type == 'delivered':
193
handle_delivery_event(event_data, timestamp)
194
195
elif event_type == 'bounce':
196
handle_bounce_event(event_data, timestamp)
197
198
elif event_type == 'open':
199
handle_open_event(event_data, timestamp)
200
201
elif event_type == 'click':
202
handle_click_event(event_data, timestamp)
203
204
elif event_type in ['unsubscribe', 'group_unsubscribe']:
205
handle_unsubscribe_event(event_data, timestamp)
206
207
elif event_type == 'spamreport':
208
handle_spam_report(event_data, timestamp)
209
210
# Update email status in your system
211
update_email_status(message_id, event_type, timestamp, event_data)
212
213
def handle_delivery_event(event_data, timestamp):
214
"""Handle successful email delivery."""
215
message_id = event_data.get('sg_message_id')
216
response_code = event_data.get('response')
217
218
# Update delivery status
219
EmailLog.objects.filter(message_id=message_id).update(
220
status='delivered',
221
delivered_at=timestamp,
222
smtp_response=response_code
223
)
224
225
def handle_bounce_event(event_data, timestamp):
226
"""Handle email bounces with categorization."""
227
bounce_type = event_data.get('type') # 'bounce' or 'blocked'
228
reason = event_data.get('reason', '')
229
email = event_data.get('email')
230
231
# Determine if it's a hard or soft bounce
232
is_hard_bounce = bounce_type == 'bounce' and any(
233
keyword in reason.lower()
234
for keyword in ['invalid', 'not exist', 'unknown', 'rejected']
235
)
236
237
# Update bounce status and potentially suppress future emails
238
if is_hard_bounce:
239
suppress_email_address(email, reason='hard_bounce')
240
241
# Log bounce details
242
BounceLog.objects.create(
243
email=email,
244
bounce_type=bounce_type,
245
reason=reason,
246
timestamp=timestamp,
247
is_hard_bounce=is_hard_bounce
248
)
249
250
def handle_open_event(event_data, timestamp):
251
"""Handle email open tracking."""
252
message_id = event_data.get('sg_message_id')
253
user_agent = event_data.get('useragent', '')
254
ip_address = event_data.get('ip', '')
255
256
# Record email open
257
EmailOpen.objects.create(
258
message_id=message_id,
259
opened_at=timestamp,
260
user_agent=user_agent,
261
ip_address=ip_address
262
)
263
264
def handle_click_event(event_data, timestamp):
265
"""Handle link click tracking."""
266
message_id = event_data.get('sg_message_id')
267
url = event_data.get('url', '')
268
user_agent = event_data.get('useragent', '')
269
270
# Record link click
271
EmailClick.objects.create(
272
message_id=message_id,
273
clicked_url=url,
274
clicked_at=timestamp,
275
user_agent=user_agent
276
)
277
278
def handle_unsubscribe_event(event_data, timestamp):
279
"""Handle unsubscribe events."""
280
email = event_data.get('email')
281
asm_group_id = event_data.get('asm', {}).get('group_id')
282
283
# Add to unsubscribe list
284
UnsubscribeList.objects.update_or_create(
285
email=email,
286
defaults={
287
'unsubscribed_at': timestamp,
288
'asm_group_id': asm_group_id,
289
'is_active': True
290
}
291
)
292
```
293
294
### Async Webhook Support
295
296
Support for async Django views with proper signature verification:
297
298
```python
299
from django.views.decorators.csrf import csrf_exempt
300
from django.views.decorators.http import require_POST
301
from sendgrid_backend.decorators import verify_sendgrid_webhook_signature
302
import asyncio
303
304
@csrf_exempt
305
@require_POST
306
@verify_sendgrid_webhook_signature
307
async def async_webhook_handler(request: HttpRequest) -> HttpResponse:
308
"""Async webhook handler for high-volume event processing."""
309
310
try:
311
events = json.loads(request.body)
312
313
# Process events concurrently
314
tasks = [process_event_async(event) for event in events]
315
await asyncio.gather(*tasks)
316
317
return HttpResponse("OK")
318
319
except Exception as e:
320
logger.error(f"Async webhook error: {e}")
321
return HttpResponse("Error processed")
322
323
async def process_event_async(event_data):
324
"""Async event processing for better performance."""
325
event_type = event_data.get('event')
326
327
# Use async database operations if available
328
if event_type == 'delivered':
329
await update_delivery_status_async(event_data)
330
elif event_type == 'bounce':
331
await handle_bounce_async(event_data)
332
# ... other async handlers
333
```
334
335
### Webhook Security Best Practices
336
337
Additional security measures for webhook endpoints:
338
339
```python
340
from django.conf import settings
341
from django.http import HttpResponseForbidden
342
from functools import wraps
343
344
def additional_webhook_security(view_func):
345
"""Additional security layer for webhooks."""
346
347
@wraps(view_func)
348
def wrapper(request, *args, **kwargs):
349
# Check request origin (optional)
350
allowed_ips = getattr(settings, 'SENDGRID_WEBHOOK_IPS', [])
351
if allowed_ips:
352
client_ip = get_client_ip(request)
353
if client_ip not in allowed_ips:
354
return HttpResponseForbidden("IP not allowed")
355
356
# Check User-Agent
357
user_agent = request.META.get('HTTP_USER_AGENT', '')
358
if not user_agent.startswith('SendGrid'):
359
return HttpResponseForbidden("Invalid User-Agent")
360
361
# Rate limiting (implement as needed)
362
if is_rate_limited(request):
363
return HttpResponseForbidden("Rate limited")
364
365
return view_func(request, *args, **kwargs)
366
367
return wrapper
368
369
@csrf_exempt
370
@require_POST
371
@additional_webhook_security
372
@verify_sendgrid_webhook_signature
373
def secure_webhook_handler(request):
374
"""Webhook handler with additional security measures."""
375
# Process webhook events
376
pass
377
378
def get_client_ip(request):
379
"""Extract client IP from request headers."""
380
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
381
if x_forwarded_for:
382
ip = x_forwarded_for.split(',')[0]
383
else:
384
ip = request.META.get('REMOTE_ADDR')
385
return ip
386
```
387
388
### Webhook Testing
389
390
Utilities for testing webhook functionality:
391
392
```python
393
import json
394
import hmac
395
import hashlib
396
import base64
397
from django.test import TestCase, RequestFactory
398
from django.conf import settings
399
400
class WebhookTestCase(TestCase):
401
"""Test case for SendGrid webhook handling."""
402
403
def setUp(self):
404
self.factory = RequestFactory()
405
self.verification_key = "test-verification-key"
406
407
def create_signed_request(self, payload, timestamp=None):
408
"""Create a properly signed webhook request for testing."""
409
if timestamp is None:
410
timestamp = str(int(time.time()))
411
412
payload_json = json.dumps(payload)
413
414
# Create signature (simplified for testing)
415
signature_payload = timestamp + payload_json
416
signature = base64.b64encode(
417
hmac.new(
418
self.verification_key.encode(),
419
signature_payload.encode(),
420
hashlib.sha256
421
).digest()
422
).decode()
423
424
request = self.factory.post(
425
'/webhook/sendgrid/',
426
data=payload_json,
427
content_type='application/json',
428
HTTP_X_TWILIO_EMAIL_EVENT_WEBHOOK_SIGNATURE=signature,
429
HTTP_X_TWILIO_EMAIL_EVENT_WEBHOOK_TIMESTAMP=timestamp
430
)
431
432
return request
433
434
def test_webhook_signature_verification(self):
435
"""Test webhook signature verification."""
436
payload = [{"event": "delivered", "email": "test@example.com"}]
437
request = self.create_signed_request(payload)
438
439
# Test signature verification
440
is_valid = check_sendgrid_signature(request)
441
self.assertTrue(is_valid)
442
443
def test_webhook_event_processing(self):
444
"""Test webhook event processing."""
445
payload = [{
446
"event": "delivered",
447
"email": "test@example.com",
448
"timestamp": 1234567890,
449
"sg_message_id": "test-message-id"
450
}]
451
452
request = self.create_signed_request(payload)
453
response = sendgrid_webhook_handler(request)
454
455
self.assertEqual(response.status_code, 200)
456
self.assertEqual(response.content.decode(), "OK")
457
```