feat(api): manage workspace webhooks via the public token API

apps/api/plane/api/serializers/__init__.py

@@ -64,3 +64,4 @@
 from .invite import WorkspaceInviteSerializer
 from .member import ProjectMemberSerializer
 from .sticky import StickySerializer
+from .webhook import WebhookSerializer

apps/api/plane/api/serializers/webhook.py

@@ -0,0 +1,67 @@
+# Copyright (c) 2023-present Plane Software, Inc. and contributors
+# SPDX-License-Identifier: AGPL-3.0-only
+# See the LICENSE file for details.
+
+# Third party imports
+from rest_framework import serializers
+
+# Module imports
+from .base import BaseSerializer
+from plane.db.models import Webhook
+from plane.db.models.webhook import validate_domain, validate_schema
+from plane.utils.webhook import validate_webhook_url
+
+
+class WebhookSerializer(BaseSerializer):
+    """Public token-API serializer for workspace webhooks.
+
+    Mirrors the internal app-API webhook serializer: enforces the schema/domain
+    validators plus the shared SSRF/disallowed-domain guard, and keeps
+    ``secret_key`` server-generated (read-only) so it is returned on create but
+    never accepted as input.
+    """
+
+    url = serializers.URLField(validators=[validate_schema, validate_domain])
+
+    def _validate_webhook_url(self, url):
+        """Validate a webhook URL against SSRF and disallowed-domain rules.
+
+        Thin adapter binding the serializer's request context to the shared
+        ``validate_webhook_url`` guard, mirroring the internal app-API
+        serializer so the SSRF/URL checks cannot drift between the two.
+        """
+        validate_webhook_url(url, self.context.get("request"))
+
+    def create(self, validated_data):
+        url = validated_data.get("url", None)
+        self._validate_webhook_url(url)
+        return Webhook.objects.create(**validated_data)
+
+    def update(self, instance, validated_data):
+        url = validated_data.get("url", None)
+        if url:
+            self._validate_webhook_url(url)
+        return super().update(instance, validated_data)
+
+    class Meta:
+        model = Webhook
+        fields = [
+            "id",
+            "url",
+            "is_active",
+            "secret_key",
+            "project",
+            "issue",
+            "module",
+            "cycle",
+            "issue_comment",
+            "created_at",
+            "updated_at",
+        ]
+        read_only_fields = [
+            "id",
+            "workspace",
+            "secret_key",
+            "created_at",
+            "updated_at",
+        ]

apps/api/plane/api/urls/__init__.py

@@ -14,6 +14,7 @@
 from .work_item import urlpatterns as work_item_patterns
 from .invite import urlpatterns as invite_patterns
 from .sticky import urlpatterns as sticky_patterns
+from .webhook import urlpatterns as webhook_patterns
 
 urlpatterns = [
     *asset_patterns,
@@ -28,4 +29,5 @@
     *work_item_patterns,
     *invite_patterns,
     *sticky_patterns,
+    *webhook_patterns,
 ]

apps/api/plane/api/urls/webhook.py

@@ -0,0 +1,20 @@
+# Copyright (c) 2023-present Plane Software, Inc. and contributors
+# SPDX-License-Identifier: AGPL-3.0-only
+# See the LICENSE file for details.
+
+from django.urls import path
+
+from plane.api.views import WebhookAPIEndpoint, WebhookDetailAPIEndpoint
+
+urlpatterns = [
+    path(
+        "workspaces/<str:slug>/webhooks/",
+        WebhookAPIEndpoint.as_view(http_method_names=["get", "post"]),
+        name="webhooks",
+    ),
+    path(
+        "workspaces/<str:slug>/webhooks/<uuid:pk>/",
+        WebhookDetailAPIEndpoint.as_view(http_method_names=["get", "patch", "delete"]),
+        name="webhook-detail",
+    ),
+]

apps/api/plane/api/views/__init__.py

@@ -63,3 +63,5 @@
 from .invite import WorkspaceInvitationsViewset
 
 from .sticky import StickyViewSet
+
+from .webhook import WebhookAPIEndpoint, WebhookDetailAPIEndpoint

apps/api/plane/api/views/webhook.py

@@ -0,0 +1,217 @@
+# Copyright (c) 2023-present Plane Software, Inc. and contributors
+# SPDX-License-Identifier: AGPL-3.0-only
+# See the LICENSE file for details.
+
+# Django imports
+from django.db import IntegrityError
+
+# Third party imports
+from rest_framework import status
+from rest_framework.response import Response
+from drf_spectacular.utils import (
+    extend_schema,
+    OpenApiResponse,
+    OpenApiRequest,
+    OpenApiParameter,
+    OpenApiTypes,
+)
+
+# Module imports
+from .base import BaseAPIView
+from plane.api.serializers import WebhookSerializer
+from plane.db.models import Webhook, Workspace
+from plane.utils.permissions import WorkspaceOwnerPermission
+from plane.utils.openapi import (
+    WORKSPACE_SLUG_PARAMETER,
+    UNAUTHORIZED_RESPONSE,
+    FORBIDDEN_RESPONSE,
+    WORKSPACE_NOT_FOUND_RESPONSE,
+    CONFLICT_RESPONSE,
+)
+
+# Fields returned to clients everywhere except the create response — the
+# server-generated ``secret_key`` is intentionally withheld here and only ever
+# surfaced once, in the create response, so it is not leaked on every read.
+WEBHOOK_READ_FIELDS = (
+    "id",
+    "url",
+    "is_active",
+    "created_at",
+    "updated_at",
+    "project",
+    "issue",
+    "cycle",
+    "module",
+    "issue_comment",
+)
+
+WEBHOOK_PK_PARAMETER = OpenApiParameter(
+    name="pk",
+    description="Webhook ID",
+    required=True,
+    type=OpenApiTypes.UUID,
+    location=OpenApiParameter.PATH,
+)
+
+
+class WebhookAPIEndpoint(BaseAPIView):
+    """Manage workspace webhooks via the token API.
+
+    Mirrors the internal app-API webhook endpoint: workspace-admin only,
+    enforces the shared URL schema/domain validators and the SSRF /
+    ``WEBHOOK_ALLOWED_IPS`` guard, and generates the signing ``secret_key``
+    server-side (returned only in the create response).
+    """
+
+    permission_classes = [WorkspaceOwnerPermission]
+    serializer_class = WebhookSerializer
+
+    @extend_schema(
+        operation_id="create_webhook",
+        summary="Create webhook",
+        description=(
+            "Register a webhook for the workspace. The signing `secret_key` is "
+            "generated server-side and returned only in this response. The target "
+            "`url` is validated against the SSRF/URL guards (localhost, private "
+            "networks and non-http(s) schemes are rejected)."
+        ),
+        tags=["Webhooks"],
+        parameters=[WORKSPACE_SLUG_PARAMETER],
+        request=OpenApiRequest(request=WebhookSerializer),
+        responses={
+            201: OpenApiResponse(description="Webhook created", response=WebhookSerializer),
+            400: OpenApiResponse(description="Invalid or disallowed webhook URL"),
+            401: UNAUTHORIZED_RESPONSE,
+            403: FORBIDDEN_RESPONSE,
+            404: WORKSPACE_NOT_FOUND_RESPONSE,
+            409: CONFLICT_RESPONSE,
+        },
+    )
+    def post(self, request, slug):
+        workspace = Workspace.objects.get(slug=slug)
+        try:
+            serializer = WebhookSerializer(data=request.data, context={"request": request})
+            if serializer.is_valid():
+                serializer.save(workspace_id=workspace.id)
+                return Response(serializer.data, status=status.HTTP_201_CREATED)
+            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+        except IntegrityError as e:
+            # Only the unique webhook-URL violation maps to 409; any other
+            # integrity error (FK / NOT NULL / ...) is re-raised so it is not
+            # masked as a duplicate-URL conflict.
+            if "already exists" in str(e):
+                return Response(
+                    {"error": "URL already exists for the workspace"},
+                    status=status.HTTP_409_CONFLICT,
+                )
+            raise
+
+    @extend_schema(
+        operation_id="list_webhooks",
+        summary="List webhooks",
+        description=("List all webhooks for the workspace. The `secret_key` is never included in these responses."),
+        tags=["Webhooks"],
+        parameters=[WORKSPACE_SLUG_PARAMETER],
+        responses={
+            200: OpenApiResponse(description="Webhooks", response=WebhookSerializer(many=True)),
+            401: UNAUTHORIZED_RESPONSE,
+            403: FORBIDDEN_RESPONSE,
+            404: WORKSPACE_NOT_FOUND_RESPONSE,
+        },
+    )
+    def get(self, request, slug):
+        webhooks = Webhook.objects.filter(workspace__slug=slug)
+        serializer = WebhookSerializer(webhooks, fields=WEBHOOK_READ_FIELDS, many=True)
+        return Response(serializer.data, status=status.HTTP_200_OK)
+
+
+class WebhookDetailAPIEndpoint(BaseAPIView):
+    """Retrieve, update or delete a single workspace webhook via the token API.
+
+    Split from ``WebhookAPIEndpoint`` so the collection ``GET`` (list) and the
+    detail ``GET`` (retrieve) get distinct, semantically correct OpenAPI
+    ``operationId``s instead of colliding on one shared handler.
+    """
+
+    permission_classes = [WorkspaceOwnerPermission]
+    serializer_class = WebhookSerializer
+
+    @extend_schema(
+        operation_id="retrieve_webhook",
+        summary="Retrieve webhook",
+        description="Retrieve a single webhook by ID. The `secret_key` is never included in this response.",
+        tags=["Webhooks"],
+        parameters=[WORKSPACE_SLUG_PARAMETER, WEBHOOK_PK_PARAMETER],
+        responses={
+            200: OpenApiResponse(description="Webhook", response=WebhookSerializer),
+            401: UNAUTHORIZED_RESPONSE,
+            403: FORBIDDEN_RESPONSE,
+            404: WORKSPACE_NOT_FOUND_RESPONSE,
+        },
+    )
+    def get(self, request, slug, pk):
+        webhook = Webhook.objects.get(workspace__slug=slug, pk=pk)
+        serializer = WebhookSerializer(webhook, fields=WEBHOOK_READ_FIELDS)
+        return Response(serializer.data, status=status.HTTP_200_OK)
+
+    @extend_schema(
+        operation_id="update_webhook",
+        summary="Update webhook",
+        description=(
+            "Update a webhook's target `url`, active state or entity toggles. A "
+            "changed `url` is re-validated against the SSRF/URL guards. The "
+            "`secret_key` cannot be changed here."
+        ),
+        tags=["Webhooks"],
+        parameters=[WORKSPACE_SLUG_PARAMETER, WEBHOOK_PK_PARAMETER],
+        request=OpenApiRequest(request=WebhookSerializer),
+        responses={
+            200: OpenApiResponse(description="Webhook updated", response=WebhookSerializer),
+            400: OpenApiResponse(description="Invalid or disallowed webhook URL"),
+            401: UNAUTHORIZED_RESPONSE,
+            403: FORBIDDEN_RESPONSE,
+            404: WORKSPACE_NOT_FOUND_RESPONSE,
+            409: CONFLICT_RESPONSE,
+        },
+    )
+    def patch(self, request, slug, pk):
+        webhook = Webhook.objects.get(workspace__slug=slug, pk=pk)
+        try:
+            serializer = WebhookSerializer(
+                webhook,
+                data=request.data,
+                context={"request": request},
+                partial=True,
+                fields=WEBHOOK_READ_FIELDS,
+            )
+            if serializer.is_valid():
+                serializer.save()
+                return Response(serializer.data, status=status.HTTP_200_OK)
+            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+        except IntegrityError as e:
+            # Only the unique webhook-URL violation maps to 409; any other
+            # integrity error is re-raised rather than masked as a conflict.
+            if "already exists" in str(e):
+                return Response(
+                    {"error": "URL already exists for the workspace"},
+                    status=status.HTTP_409_CONFLICT,
+                )
+            raise
+
+    @extend_schema(
+        operation_id="delete_webhook",
+        summary="Delete webhook",
+        description="Delete a workspace webhook by ID.",
+        tags=["Webhooks"],
+        parameters=[WORKSPACE_SLUG_PARAMETER, WEBHOOK_PK_PARAMETER],
+        responses={
+            204: OpenApiResponse(description="Webhook deleted"),
+            401: UNAUTHORIZED_RESPONSE,
+            403: FORBIDDEN_RESPONSE,
+            404: WORKSPACE_NOT_FOUND_RESPONSE,
+        },
+    )
+    def delete(self, request, slug, pk):
+        webhook = Webhook.objects.get(workspace__slug=slug, pk=pk)
+        webhook.delete()
+        return Response(status=status.HTTP_204_NO_CONTENT)

apps/api/plane/app/serializers/webhook.py

@@ -2,57 +2,27 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 # See the LICENSE file for details.
 
-# Python imports
-import logging
-from urllib.parse import urlparse
-
 # Third party imports
 from rest_framework import serializers
 
-# Django imports
-from django.conf import settings
-
 # Module imports
 from .base import DynamicBaseSerializer
 from plane.db.models import Webhook, WebhookLog
 from plane.db.models.webhook import validate_domain, validate_schema
-from plane.utils.ip_address import validate_url
-
-logger = logging.getLogger(__name__)
+from plane.utils.webhook import validate_webhook_url
 
 
 class WebhookSerializer(DynamicBaseSerializer):
     url = serializers.URLField(validators=[validate_schema, validate_domain])
 
     def _validate_webhook_url(self, url):
-        """Validate a webhook URL against SSRF and disallowed domain rules."""
-        try:
-            validate_url(
-                url,
-                allowed_ips=settings.WEBHOOK_ALLOWED_IPS,
-                allowed_hosts=settings.WEBHOOK_ALLOWED_HOSTS,
-            )
-        except ValueError as e:
-            logger.warning("Webhook URL validation failed for %s: %s", url, e)
-            raise serializers.ValidationError({"url": "Invalid or disallowed webhook URL."})
-
-        hostname = (urlparse(url).hostname or "").rstrip(".").lower()
-
-        # Hosts explicitly trusted via WEBHOOK_ALLOWED_HOSTS bypass the
-        # disallowed-domain check — they're already trusted for SSRF, so
-        # the loop-back guard would only get in the way of legitimate
-        # sibling services that share a parent domain with Plane.
-        if hostname in settings.WEBHOOK_ALLOWED_HOSTS:
-            return
-
-        request = self.context.get("request")
-        disallowed_domains = list(settings.WEBHOOK_DISALLOWED_DOMAINS)
-        if request:
-            request_host = request.get_host().split(":")[0].rstrip(".").lower()
-            disallowed_domains.append(request_host)
+        """Validate a webhook URL against SSRF and disallowed-domain rules.
 
-        if any(hostname == domain or hostname.endswith("." + domain) for domain in disallowed_domains):
-            raise serializers.ValidationError({"url": "URL domain or its subdomain is not allowed."})
+        Thin adapter binding the serializer's request context to the shared
+        ``validate_webhook_url`` guard so the SSRF/URL checks live in a single
+        place shared with the public token API.
+        """
+        validate_webhook_url(url, self.context.get("request"))
 
     def create(self, validated_data):
         url = validated_data.get("url", None)