How to use the Webhook Reciever

[JanEberlein]

I am currently working on a plugin that is supposed to connect to Lexoffice. For now it should create a sale order when certain events in Lexoffice occur. For that I obviously want to use webhooks.

I have tried quiet a few things with the WebhookEndpoint and have encountered a few roadblocks so far:

1. Header char field for webhook messages too short #9280 This is the same project
2. I expected to receive events via the EventsMixin for incoming calls to webhook endpoints
3. I can not process the payload
4. I can not modify the verification method (This could also be done while processing the payload, given the original message and headers)
5. The documentation regarding the WebhookEndpoint is somewhat lacking.

I also tried using the AppMixin since the shopify plugin uses it to be able inherit the WebhookEndpoint class. That didn't work and to be honest that is too much overhead for the scope of this project.

So my question is:
Am I missing something?
And if not, what else can I do/try?

[matmair]

The way webhooks are used in the Shopify plugin works, I suggest following that.

[JanEberlein]

As I already stated above that didn't work for me.

Are there some setup steps that aren't documented anywhere that I would need to know about?

Edit: I an currently just looking to get this to run locally for development

[matmair]

There is a functional sample app provided above, there is a whole page on plugin development and the whole logic is open source

InvenTree/src/backend/InvenTree/common/models.py

Lines 1279 to 1433 in 7a43c3a

	class WebhookEndpoint(models.Model):
	"""Defines a Webhook endpoint.
	Attributes:
	endpoint_id: Path to the webhook,
	name: Name of the webhook,
	active: Is this webhook active?,
	user: User associated with webhook,
	token: Token for sending a webhook,
	secret: Shared secret for HMAC verification,
	"""
	# Token
	TOKEN_NAME = 'Token'
	VERIFICATION_METHOD = VerificationMethod.NONE
	MESSAGE_OK = 'Message was received.'
	MESSAGE_TOKEN_ERROR = 'Incorrect token in header.'
	endpoint_id = models.CharField(
	max_length=255,
	verbose_name=_('Endpoint'),
	help_text=_('Endpoint at which this webhook is received'),
	default=uuid.uuid4,
	editable=False,
	)
	name = models.CharField(
	max_length=255,
	blank=True,
	null=True,
	verbose_name=_('Name'),
	help_text=_('Name for this webhook'),
	)
	active = models.BooleanField(
	default=True, verbose_name=_('Active'), help_text=_('Is this webhook active')
	)
	user = models.ForeignKey(
	User,
	on_delete=models.SET_NULL,
	blank=True,
	null=True,
	verbose_name=_('User'),
	help_text=_('User'),
	)
	token = models.CharField(
	max_length=255,
	blank=True,
	null=True,
	verbose_name=_('Token'),
	help_text=_('Token for access'),
	default=uuid.uuid4,
	)
	secret = models.CharField(
	max_length=255,
	blank=True,
	null=True,
	verbose_name=_('Secret'),
	help_text=_('Shared secret for HMAC'),
	)
	# To be overridden
	def init(self, request, *args, **kwargs):
	"""Set verification method.
	Args:
	request: Original request object.
	"""
	self.verify = self.VERIFICATION_METHOD
	def process_webhook(self):
	"""Process the webhook incoming.
	This does not deal with the data itself - that happens in process_payload.
	Do not touch or pickle data here - it was not verified to be safe.
	"""
	if self.token:
	self.verify = VerificationMethod.TOKEN
	if self.secret:
	self.verify = VerificationMethod.HMAC
	return True
	def validate_token(self, payload, headers, request):
	"""Make sure that the provided token (if any) confirms to the setting for this endpoint.
	This can be overridden to create your own token validation method.
	"""
	token = headers.get(self.TOKEN_NAME, '')
	# no token
	if self.verify == VerificationMethod.NONE:
	# do nothing as no method was chosen
	pass
	# static token
	elif self.verify == VerificationMethod.TOKEN:
	if not compare_digest(token, self.token):
	raise PermissionDenied(self.MESSAGE_TOKEN_ERROR)
	# hmac token
	elif self.verify == VerificationMethod.HMAC:
	digest = hmac.new(
	self.secret.encode('utf-8'), request.body, hashlib.sha256
	).digest()
	computed_hmac = base64.b64encode(digest)
	if not hmac.compare_digest(computed_hmac, token.encode('utf-8')):
	raise PermissionDenied(self.MESSAGE_TOKEN_ERROR)
	return True
	def save_data(self, payload=None, headers=None, request=None):
	"""Safes payload to database.
	Args:
	payload (optional): Payload that was send along. Defaults to None.
	headers (optional): Headers that were send along. Defaults to None.
	request (optional): Original request object. Defaults to None.
	"""
	return WebhookMessage.objects.create(
	host=request.get_host(),
	header=json.dumps(dict(headers.items())),
	body=payload,
	endpoint=self,
	)
	def process_payload(self, message, payload=None, headers=None) -> bool:
	"""Process a payload.
	Args:
	message: DB entry for this message mm
	payload (optional): Payload that was send along. Defaults to None.
	headers (optional): Headers that were included. Defaults to None.
	Returns:
	bool: Was the message processed
	"""
	return True
	def get_return(self, payload=None, headers=None, request=None) -> str:
	"""Returns the message that should be returned to the endpoint caller.
	Args:
	payload (optional): Payload that was send along. Defaults to None.
	headers (optional): Headers that were send along. Defaults to None.
	request (optional): Original request object. Defaults to None.
	Returns:
	str: Message for caller.
	"""
	return self.MESSAGE_OK

If that is not enough information you might need direct assistance - which we can not give without a concrete source sample and more info than a listicle and "I can not process the payload"