fix: service accounts creating webhooks through API

engine/apps/public_api/serializers/webhooks.py

@@ -3,6 +3,7 @@
 from rest_framework import fields, serializers
 from rest_framework.validators import UniqueTogetherValidator
 
+from apps.user_management.models import ServiceAccountUser
 from apps.webhooks.models import Webhook, WebhookResponse
 from apps.webhooks.models.webhook import PUBLIC_WEBHOOK_HTTP_METHODS, WEBHOOK_FIELD_PLACEHOLDER
 from apps.webhooks.presets.preset_options import WebhookPresetOptions
@@ -161,6 +162,11 @@ def validate_http_method(self, http_method):
     def validate_preset(self, preset):
         raise serializers.ValidationError(PRESET_VALIDATION_MESSAGE)
 
+    def validate_user(self, user):
+        if isinstance(user, ServiceAccountUser):
+            return None
+        return user
+
     def validate(self, data):
         if (
             self.instance

engine/apps/public_api/tests/test_webhooks.py

@@ -1,10 +1,13 @@
 import json
 
+import httpretty
 import pytest
 from django.urls import reverse
 from rest_framework import status
 from rest_framework.test import APIClient
 
+from apps.api import permissions
+from apps.auth_token.tests.helpers import setup_service_account_api_mocks
 from apps.public_api.serializers.webhooks import PRESET_VALIDATION_MESSAGE
 from apps.webhooks.models import Webhook
 from apps.webhooks.tests.test_webhook_presets import ADVANCED_WEBHOOK_PRESET_ID, TEST_WEBHOOK_PRESET_ID
@@ -235,6 +238,47 @@ def test_create_webhook_nested_data(make_organization_and_user_with_token):
     assert response.json() == expected_result
 
 
+@pytest.mark.django_db
+@httpretty.activate(verbose=True, allow_net_connect=False)
+def test_create_webhook_via_service_account(
+    make_organization,
+    make_service_account_for_organization,
+    make_token_for_service_account,
+):
+    organization = make_organization(grafana_url="http://grafana.test")
+    service_account = make_service_account_for_organization(organization)
+    token_string = "glsa_token"
+    make_token_for_service_account(service_account, token_string)
+
+    perms = {
+        permissions.RBACPermission.Permissions.OUTGOING_WEBHOOKS_WRITE.value: ["*"],
+    }
+    setup_service_account_api_mocks(organization.grafana_url, perms)
+
+    client = APIClient()
+    url = reverse("api-public:webhooks-list")
+    data = {
+        "name": "Test outgoing webhook",
+        "url": "https://example.com",
+        "http_method": "POST",
+        "trigger_type": "acknowledge",
+    }
+    response = client.post(
+        url,
+        data=data,
+        format="json",
+        HTTP_AUTHORIZATION=f"{token_string}",
+        HTTP_X_GRAFANA_URL=organization.grafana_url,
+    )
+    if not organization.is_rbac_permissions_enabled:
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+    else:
+        assert response.status_code == status.HTTP_201_CREATED
+        webhook = Webhook.objects.get(public_primary_key=response.data["id"])
+        expected_result = _get_expected_result(webhook)
+        assert response.data == expected_result
+
+
 @pytest.mark.django_db
 def test_update_webhook(
     make_organization_and_user_with_token,

engine/apps/user_management/models/service_account.py

@@ -25,6 +25,10 @@ def id(self):
     def pk(self):
         return self.service_account.id
 
+    @property
+    def current_team(self):
+        return None
+
     @property
     def organization_id(self):
         return self.organization.id