Enhancement: use charfield for webhook url, custom validation (#9128)

---------

Co-authored-by: Trenton H <797416+stumpylog@users.noreply.github.com>

src/documents/migrations/1063_alter_workflowactionwebhook_url.py

@@ -0,0 +1,22 @@
+# Generated by Django 5.1.6 on 2025-02-16 16:31
+
+from django.db import migrations
+from django.db import models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("documents", "1062_alter_savedviewfilterrule_rule_type"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="workflowactionwebhook",
+            name="url",
+            field=models.CharField(
+                help_text="The destination URL for the notification.",
+                max_length=256,
+                verbose_name="webhook url",
+            ),
+        ),
+    ]

src/documents/models.py

@@ -1203,9 +1203,12 @@ class WorkflowActionEmail(models.Model):
 
 
 class WorkflowActionWebhook(models.Model):
-    url = models.URLField(
+    # We dont use the built-in URLField because it is not flexible enough
+    # validation is handled in the serializer
+    url = models.CharField(
         _("webhook url"),
         null=False,
+        max_length=256,
         help_text=_("The destination URL for the notification."),
     )
 

src/documents/serialisers.py

@@ -58,6 +58,7 @@ from documents.permissions import set_permissions_for_object
 from documents.templating.filepath import validate_filepath_template_and_render
 from documents.templating.utils import convert_format_str_to_template_format
 from documents.validators import uri_validator
+from documents.validators import url_validator
 
 logger = logging.getLogger("paperless.serializers")
 
@@ -1949,6 +1950,10 @@ class WorkflowActionEmailSerializer(serializers.ModelSerializer):
 class WorkflowActionWebhookSerializer(serializers.ModelSerializer):
     id = serializers.IntegerField(allow_null=True, required=False)
 
+    def validate_url(self, url):
+        url_validator(url)
+        return url
+
     class Meta:
         model = WorkflowActionWebhook
         fields = [

src/documents/tests/test_api_workflows.py

@@ -588,3 +588,45 @@ class TestApiWorkflows(DirectoriesMixin, APITestCase):
             content_type="application/json",
         )
         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
+    def test_webhook_action_url_validation(self):
+        """
+        GIVEN:
+            - API request to create a workflow with a notification action
+        WHEN:
+            - API is called
+        THEN:
+            - Correct HTTP response
+        """
+
+        for url, expected_resp_code in [
+            ("https://examplewithouttld:3000/path", status.HTTP_201_CREATED),
+            ("file:///etc/passwd/path", status.HTTP_400_BAD_REQUEST),
+        ]:
+            response = self.client.post(
+                self.ENDPOINT,
+                json.dumps(
+                    {
+                        "name": "Workflow 2",
+                        "order": 1,
+                        "triggers": [
+                            {
+                                "type": WorkflowTrigger.WorkflowTriggerType.CONSUMPTION,
+                                "sources": [DocumentSource.ApiUpload],
+                                "filter_filename": "*",
+                            },
+                        ],
+                        "actions": [
+                            {
+                                "type": WorkflowAction.WorkflowActionType.WEBHOOK,
+                                "webhook": {
+                                    "url": url,
+                                    "include_document": False,
+                                },
+                            },
+                        ],
+                    },
+                ),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, expected_resp_code)

src/documents/validators.py

@@ -4,11 +4,18 @@ from django.core.exceptions import ValidationError
 from django.utils.translation import gettext_lazy as _
 
 
-def uri_validator(value) -> None:
+def uri_validator(value: str, allowed_schemes: set[str] | None = None) -> None:
     """
-    Raises a ValidationError if the given value does not parse as an
-    URI looking thing, which we're defining as a scheme and either network
-    location or path value
+    Validates that the given value parses as a URI with required components
+    and optionally restricts to specific schemes.
+
+    Args:
+        value: The URI string to validate
+        allowed_schemes: Optional set/list of allowed schemes (e.g. {'http', 'https'}).
+                        If None, all schemes are allowed.
+
+    Raises:
+        ValidationError: If the URI is invalid or uses a disallowed scheme
     """
     try:
         parts = urlparse(value)
@@ -22,8 +29,32 @@ def uri_validator(value) -> None:
                 _(f"Unable to parse URI {value}, missing net location or path"),
                 params={"value": value},
             )
+
+        if allowed_schemes and parts.scheme not in allowed_schemes:
+            raise ValidationError(
+                _(
+                    f"URI scheme '{parts.scheme}' is not allowed. Allowed schemes: {', '.join(allowed_schemes)}",
+                ),
+                params={"value": value, "scheme": parts.scheme},
+            )
+
+    except ValidationError:
+        raise
     except Exception as e:
         raise ValidationError(
             _(f"Unable to parse URI {value}"),
             params={"value": value},
         ) from e
+
+
+def url_validator(value) -> None:
+    """
+    Validates that the given value is a valid HTTP or HTTPS URL.
+
+    Args:
+        value: The URL string to validate
+
+    Raises:
+        ValidationError: If the URL is invalid or not using http/https scheme
+    """
+    uri_validator(value, allowed_schemes={"http", "https"})