add basic internal token endpoint, verification and tests

Author: artcz

intbot/core/endpoints/webhooks.py

@@ -1,13 +1,20 @@
+import hmac
 import json
+
+from core.models import Webhook
+from django.conf import settings
 from django.http.response import HttpResponseNotAllowed, JsonResponse
 from django.views.decorators.csrf import csrf_exempt
 
-from core.models import Webhook
 
 @csrf_exempt
 def internal_webhook_endpoint(request):
     if request.method == "POST":
-        print(request.body)
+        try:
+            verify_internal_webhook(request)
+        except ValueError as e:
+            return JsonResponse({"status": "bad", "message": str(e)}, status=403)
+
         wh = Webhook.objects.create(
             source="internal",
             content=json.loads(request.body),
@@ -16,3 +23,15 @@ def internal_webhook_endpoint(request):
         return JsonResponse({"status": "created", "guid": wh.uuid})
 
     return HttpResponseNotAllowed("Only POST")
+
+
+def verify_internal_webhook(request):
+    """raise ValueError if incorrect token"""
+
+    if not "Authorization" in request.headers:
+        raise ValueError("Authorization token is missing")
+
+    token = request.headers['Authorization']
+
+    if not hmac.compare_digest(settings.WEBHOOK_INTERNAL_TOKEN, token):
+        raise ValueError("Token doesn't match")

intbot/intbot/settings.py

@@ -138,6 +138,8 @@
         }
     }
 
+    WEBHOOK_INTERNAL_TOKEN = "test-random-token"
+
 
 else:
     raise ValueError(f"Unsupported DJANGO_ENV `{DJANGO_ENV}`")

intbot/tests/test_webhooks.py

@@ -1,9 +1,50 @@
 import pytest
+from django.conf import settings
 from core.models import Webhook
 
+@pytest.mark.django_db
+def test_internal_wh_endpoint_checks_authorization_token(client):
+    webhook_body = {
+        "event": "test1",
+        "content": {
+            "random": "content",
+        },
+    }
+
+    response = client.post(
+        "/webhook/internal/",
+        json=webhook_body,
+        content_type="application/json",
+    )
+
+    assert response.status_code == 403
+    assert response["Content-Type"] == "application/json"
+    assert response.json()["status"] == "bad"
+    assert response.json()["message"] == "Authorization token is missing"
+
+@pytest.mark.django_db
+def test_internal_wh_endpoint_fails_with_bad_token(client):
+    webhook_body = {
+        "event": "test1",
+        "content": {
+            "random": "content",
+        },
+    }
+
+    response = client.post(
+        "/webhook/internal/",
+        json=webhook_body,
+        content_type="application/json",
+        HTTP_AUTHORIZATION="random-incorrect-token",
+    )
+
+    assert response.status_code == 403
+    assert response["Content-Type"] == "application/json"
+    assert response.json()["status"] == "bad"
+    assert response.json()["message"] == "Token doesn't match"
 
 @pytest.mark.django_db
-def test_database_sanity_check(client):
+def test_internal_wh_endpoint_works_with_correct_token(client):
     webhook_body = {
         "event": "test1",
         "content": {
@@ -15,6 +56,7 @@ def test_database_sanity_check(client):
         "/webhook/internal/",
         json=webhook_body,
         content_type="application/json",
+        HTTP_AUTHORIZATION=settings.WEBHOOK_INTERNAL_TOKEN,
     )
 
     wh = Webhook.objects.get()