Mirrored_Repos/InvenTree
1
0
Fork 0
mirror of https://github.com/inventree/InvenTree synced 2024-08-30 18:33:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

Pulls out the webhook dev for partial merge

Browse Source 
Fixes #2036
This commit is contained in:
Matthias 2021-11-19 22:35:28 +01:00
parent 6a235c9f15
commit 260c51f6fc
No known key found for this signature in database
GPG Key ID: F50EF5741D33E076
8 changed files with 460 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
InvenTree/InvenTree/helpers.py
View File

@ -696,3 +696,18 @@ def clean_decimal(number):
return Decimal(0)
return clean_number.quantize(Decimal(1)) if clean_number == clean_number.to_integral() else clean_number.normalize()
def inheritors(cls):
"""
Return all classes that are subclasses from the supplied cls
"""
subcls = set()
work = [cls]
while work:
parent = work.pop()
for child in parent.__subclasses__():
if child not in subcls:
subcls.add(child)
work.append(child)
return subcls

3
InvenTree/InvenTree/urls.py
View File

@ -67,6 +67,9 @@ apipatterns = [
# Plugin endpoints
url(r'^action/', ActionPluginView.as_view(), name='api-action-plugin'),
# Webhook enpoint
path('', include(common_api_urls)),
# InvenTree information endpoint
url(r'^$', InfoView.as_view(), name='api-inventree-info'),

7
InvenTree/common/admin.py
View File

@ -38,6 +38,11 @@ class UserSettingsAdmin(ImportExportModelAdmin):
return []
class WebhookAdmin(ImportExportModelAdmin):
list_display = ('endpoint_id', 'name', 'active', 'user')
class NotificationEntryAdmin(admin.ModelAdmin):
list_display = ('key', 'uid', 'updated', )
@ -45,4 +50,6 @@ class NotificationEntryAdmin(admin.ModelAdmin):
admin.site.register(common.models.InvenTreeSetting, SettingsAdmin)
admin.site.register(common.models.InvenTreeUserSetting, UserSettingsAdmin)
admin.site.register(common.models.WebhookEndpoint, WebhookAdmin)
admin.site.register(common.models.WebhookMessage, ImportExportModelAdmin)
admin.site.register(common.models.NotificationEntry, NotificationEntryAdmin)

91
InvenTree/common/api.py
View File

@ -5,13 +5,101 @@ Provides a JSON API for common components.
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import json
from django.http.response import HttpResponse
from django.utils.decorators import method_decorator
from django.urls import path
from django.views.decorators.csrf import csrf_exempt
from django.conf.urls import url, include
from rest_framework.views import APIView
from rest_framework.exceptions import NotAcceptable, NotFound
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework import filters, generics, permissions
from django_q.tasks import async_task
import common.models
import common.serializers
from InvenTree.helpers import inheritors
class CsrfExemptMixin(object):
"""
Exempts the view from CSRF requirements.
"""
@method_decorator(csrf_exempt)
def dispatch(self, *args, **kwargs):
return super(CsrfExemptMixin, self).dispatch(*args, **kwargs)
class WebhookView(CsrfExemptMixin, APIView):
"""
Endpoint for receiving webhooks.
"""
authentication_classes = []
permission_classes = []
model_class = common.models.WebhookEndpoint
run_async = False
def post(self, request, endpoint, *args, **kwargs):
# get webhook definition
self._get_webhook(endpoint, request, *args, **kwargs)
# check headers
headers = request.headers
try:
payload = json.loads(request.body)
except json.decoder.JSONDecodeError as error:
raise NotAcceptable(error.msg)
# validate
self.webhook.validate_token(payload, headers, request)
# process data
message = self.webhook.save_data(payload, headers, request)
if self.run_async:
async_task(self._process_payload, message.id)
else:
self._process_result(
self.webhook.process_payload(message, payload, headers),
message,
)
# return results
data = self.webhook.get_return(payload, headers, request)
return HttpResponse(data)
def _process_payload(self, message_id):
message = common.models.WebhookMessage.objects.get(message_id=message_id)
self._process_result(
self.webhook.process_payload(message, message.body, message.header),
message,
)
def _process_result(self, result, message):
if result:
message.worked_on = result
message.save()
else:
message.delete()
def _escalate_object(self, obj):
classes = inheritors(obj.__class__)
for cls in classes:
mdl_name = cls._meta.model_name
if hasattr(obj, mdl_name):
return getattr(obj, mdl_name)
return obj
def _get_webhook(self, endpoint, request, *args, **kwargs):
try:
webhook = self.model_class.objects.get(endpoint_id=endpoint)
self.webhook = self._escalate_object(webhook)
self.webhook.init(request, *args, **kwargs)
return self.webhook.process_webhook()
except self.model_class.DoesNotExist:
raise NotFound()
class SettingsList(generics.ListAPIView):
@ -131,6 +219,7 @@ class UserSettingsDetail(generics.RetrieveUpdateAPIView):
common_api_urls = [
path('webhook/<slug:endpoint>/', WebhookView.as_view(), name='api-webhook'),
# User settings
url(r'^user/', include([
@ -148,6 +237,6 @@ common_api_urls = [
# Global Settings List
url(r'^.*$', GlobalSettingsList.as_view(), name='api-global-setting-list'),
]))
])),
]

40
InvenTree/common/migrations/0013_webhookendpoint_webhookmessage.py Normal file
View File

@ -0,0 +1,40 @@
# Generated by Django 3.2.5 on 2021-11-19 21:34
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
import uuid
class Migration(migrations.Migration):
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
('common', '0012_notificationentry'),
]
operations = [
migrations.CreateModel(
name='WebhookEndpoint',
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('endpoint_id', models.CharField(default=uuid.uuid4, editable=False, help_text='Endpoint at which this webhook is received', max_length=255, verbose_name='Endpoint')),
('name', models.CharField(blank=True, help_text='Name for this webhook', max_length=255, null=True, verbose_name='Name')),
('active', models.BooleanField(default=True, help_text='Is this webhook active', verbose_name='Active')),
('token', models.CharField(blank=True, default=uuid.uuid4, help_text='Token for access', max_length=255, null=True, verbose_name='Token')),
('secret', models.CharField(blank=True, help_text='Shared secret for HMAC', max_length=255, null=True, verbose_name='Secret')),
('user', models.ForeignKey(blank=True, help_text='User', null=True, on_delete=django.db.models.deletion.SET_NULL, to=settings.AUTH_USER_MODEL, verbose_name='User')),
],
),
migrations.CreateModel(
name='WebhookMessage',
fields=[
('message_id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='Unique identifier for this message', primary_key=True, serialize=False, verbose_name='Message ID')),
('host', models.CharField(editable=False, help_text='Host from which this message was received', max_length=255, verbose_name='Host')),
('header', models.CharField(blank=True, editable=False, help_text='Header of this message', max_length=255, null=True, verbose_name='Header')),
('body', models.JSONField(blank=True, editable=False, help_text='Body of this message', null=True, verbose_name='Body')),
('worked_on', models.BooleanField(default=False, help_text='Was the work on this message finished?', verbose_name='Worked on')),
('endpoint', models.ForeignKey(blank=True, help_text='Endpoint on which this message was received', null=True, on_delete=django.db.models.deletion.SET_NULL, to='common.webhookendpoint', verbose_name='Endpoint')),
],
),
]

186
InvenTree/common/models.py
View File

@ -9,6 +9,12 @@ from __future__ import unicode_literals
import os
import decimal
import math
import uuid
import hmac
import json
import hashlib
import base64
from secrets import compare_digest
from datetime import datetime, timedelta
from django.db import models, transaction
@ -20,6 +26,8 @@ from djmoney.settings import CURRENCY_CHOICES
from djmoney.contrib.exchange.models import convert_money
from djmoney.contrib.exchange.exceptions import MissingRate
from rest_framework.exceptions import PermissionDenied
from django.utils.translation import ugettext_lazy as _
from django.core.validators import MinValueValidator, URLValidator
from django.core.exceptions import ValidationError
@ -1346,6 +1354,184 @@ class ColorTheme(models.Model):
return False
class VerificationMethod:
NONE = 0
TOKEN = 1
HMAC = 2
class WebhookEndpoint(models.Model):
""" Defines a Webhook entdpoint
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):
self.verify = self.VERIFICATION_METHOD
def process_webhook(self):
if self.token:
self.token = self.token
self.verify = VerificationMethod.TOKEN
# TODO make a object-setting
if self.secret:
self.secret = self.secret
self.verify = VerificationMethod.HMAC
# TODO make a object-setting
return True
def validate_token(self, payload, headers, request):
token = headers.get(self.TOKEN_NAME, "")
# no token
if self.verify == VerificationMethod.NONE:
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, headers=None, request=None):
return WebhookMessage.objects.create(
host=request.get_host(),
header=json.dumps({key: val for key, val in headers.items()}),
body=payload,
endpoint=self,
)
def process_payload(self, message, payload=None, headers=None):
return True
def get_return(self, payload, headers=None, request=None):
return self.MESSAGE_OK
class WebhookMessage(models.Model):
""" Defines a webhook message
Attributes:
message_id: Unique identifier for this message,
host: Host from which this message was received,
header: Header of this message,
body: Body of this message,
endpoint: Endpoint on which this message was received,
worked_on: Was the work on this message finished?
"""
message_id = models.UUIDField(
verbose_name=_('Message ID'),
help_text=_('Unique identifier for this message'),
primary_key=True,
default=uuid.uuid4,
editable=False,
)
host = models.CharField(
max_length=255,
verbose_name=_('Host'),
help_text=_('Host from which this message was received'),
editable=False,
)
header = models.CharField(
max_length=255,
blank=True, null=True,
verbose_name=_('Header'),
help_text=_('Header of this message'),
editable=False,
)
body = models.JSONField(
blank=True, null=True,
verbose_name=_('Body'),
help_text=_('Body of this message'),
editable=False,
)
endpoint = models.ForeignKey(
WebhookEndpoint,
on_delete=models.SET_NULL,
blank=True, null=True,
verbose_name=_('Endpoint'),
help_text=_('Endpoint on which this message was received'),
)
worked_on = models.BooleanField(
default=False,
verbose_name=_('Worked on'),
help_text=_('Was the work on this message finished?'),
)
class NotificationEntry(models.Model):
"""
A NotificationEntry records the last time a particular notifaction was sent out.

121
InvenTree/common/tests.py
View File

@ -1,13 +1,14 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from http import HTTPStatus
import json
from datetime import timedelta
from django.test import TestCase
from django.test import TestCase, Client
from django.contrib.auth import get_user_model
from .models import InvenTreeSetting
from .models import NotificationEntry
from .models import InvenTreeSetting, WebhookEndpoint, WebhookMessage, NotificationEntry
from .api import WebhookView
class SettingsTest(TestCase):
@ -90,6 +91,118 @@ class SettingsTest(TestCase):
raise ValueError(f'Non-boolean default value specified for {key}')
class WebhookMessageTests(TestCase):
def setUp(self):
self.endpoint_def = WebhookEndpoint.objects.create()
self.url = f'/api/webhook/{self.endpoint_def.endpoint_id}/'
self.client = Client(enforce_csrf_checks=True)
def test_bad_method(self):
response = self.client.get(self.url)
assert response.status_code == HTTPStatus.METHOD_NOT_ALLOWED
def test_missing_token(self):
response = self.client.post(
self.url,
content_type='application/json',
)
assert response.status_code == HTTPStatus.FORBIDDEN
assert (
json.loads(response.content)['detail'] == WebhookView.model_class.MESSAGE_TOKEN_ERROR
)
def test_bad_token(self):
response = self.client.post(
self.url,
content_type='application/json',
**{'HTTP_TOKEN': '1234567fghj'},
)
assert response.status_code == HTTPStatus.FORBIDDEN
assert (json.loads(response.content)['detail'] == WebhookView.model_class.MESSAGE_TOKEN_ERROR)
def test_bad_url(self):
response = self.client.post(
'/api/webhook/1234/',
content_type='application/json',
)
assert response.status_code == HTTPStatus.NOT_FOUND
def test_bad_json(self):
response = self.client.post(
self.url,
data="{'this': 123}",
content_type='application/json',
**{'HTTP_TOKEN': str(self.endpoint_def.token)},
)
assert response.status_code == HTTPStatus.NOT_ACCEPTABLE
assert (
json.loads(response.content)['detail'] == 'Expecting property name enclosed in double quotes'
)
def test_success_no_token_check(self):
# delete token
self.endpoint_def.token = ''
self.endpoint_def.save()
# check
response = self.client.post(
self.url,
content_type='application/json',
)
assert response.status_code == HTTPStatus.OK
assert str(response.content, 'utf-8') == WebhookView.model_class.MESSAGE_OK
def test_bad_hmac(self):
# delete token
self.endpoint_def.token = ''
self.endpoint_def.secret = '123abc'
self.endpoint_def.save()
# check
response = self.client.post(
self.url,
content_type='application/json',
)
assert response.status_code == HTTPStatus.FORBIDDEN
assert (json.loads(response.content)['detail'] == WebhookView.model_class.MESSAGE_TOKEN_ERROR)
def test_success_hmac(self):
# delete token
self.endpoint_def.token = ''
self.endpoint_def.secret = '123abc'
self.endpoint_def.save()
# check
response = self.client.post(
self.url,
content_type='application/json',
**{'HTTP_TOKEN': str('68MXtc/OiXdA5e2Nq9hATEVrZFpLb3Zb0oau7n8s31I=')},
)
assert response.status_code == HTTPStatus.OK
assert str(response.content, 'utf-8') == WebhookView.model_class.MESSAGE_OK
def test_success(self):
response = self.client.post(
self.url,
data={"this": "is a message"},
content_type='application/json',
**{'HTTP_TOKEN': str(self.endpoint_def.token)},
)
assert response.status_code == HTTPStatus.OK
assert str(response.content, 'utf-8') == WebhookView.model_class.MESSAGE_OK
message = WebhookMessage.objects.get()
assert message.body == {"this": "is a message"}
class NotificationTest(TestCase):
def test_check_notification_entries(self):

2
InvenTree/users/models.py
View File

@ -151,6 +151,8 @@ class RuleSet(models.Model):
'common_colortheme',
'common_inventreesetting',
'common_inventreeusersetting',
'common_webhookendpoint',
'common_webhookmessage',
'common_notificationentry',
'company_contact',
'users_owner',