78 lines
2.1 KiB
Python
78 lines
2.1 KiB
Python
import hashlib
|
|
import hmac
|
|
import logging
|
|
|
|
from django.db import models
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
WEBHOOK_TYPES = [
|
|
("USER_MODIFIED", "User Modified"),
|
|
]
|
|
WEBHOOK_RETRY_COUNT = 5
|
|
_session = None
|
|
|
|
|
|
def create_random_secret() -> str:
|
|
import secrets
|
|
|
|
return secrets.token_urlsafe(32)
|
|
|
|
|
|
class Webhook(models.Model):
|
|
name = models.CharField(max_length=128)
|
|
enabled = models.BooleanField(default=True)
|
|
hook_type = models.CharField(
|
|
max_length=32, choices=WEBHOOK_TYPES, default=WEBHOOK_TYPES[0][0], db_index=True
|
|
)
|
|
url = models.URLField()
|
|
secret = models.CharField(max_length=64, default=create_random_secret)
|
|
description = models.TextField(
|
|
blank=True, help_text="Description of this webhook, for staff-eyes only."
|
|
)
|
|
timeout = models.IntegerField(
|
|
default=3, help_text="Timeout for HTTP calls to this webhook, in seconds."
|
|
)
|
|
app = models.ForeignKey(
|
|
'bid_main.OAuth2Application',
|
|
related_name="webhooks",
|
|
on_delete=models.CASCADE,
|
|
)
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
def send(self, payload: bytes):
|
|
"""Sends a message to the webhook.
|
|
|
|
:param payload: the encoded JSON to send
|
|
"""
|
|
logger.info("Sending %s to %s", payload, self.url)
|
|
|
|
global _session
|
|
if not _session:
|
|
_session = webhook_session()
|
|
|
|
mac = hmac.new(self.secret.encode(), payload, hashlib.sha256)
|
|
_response = _session.post(
|
|
self.url,
|
|
data=payload,
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"X-Webhook-HMAC": mac.hexdigest(),
|
|
},
|
|
timeout=self.timeout,
|
|
)
|
|
_response.raise_for_status()
|
|
|
|
|
|
def webhook_session() -> requests.Session:
|
|
"""Creates a Requests session for sending to webhooks."""
|
|
import requests.adapters
|
|
|
|
sess = requests.Session()
|
|
sess.mount("https://", requests.adapters.HTTPAdapter(max_retries=WEBHOOK_RETRY_COUNT))
|
|
sess.mount("http://", requests.adapters.HTTPAdapter(max_retries=WEBHOOK_RETRY_COUNT))
|
|
return sess
|