Closes #22828: Validate Webhook.payload_url as a URL or Jinja2 template (#22832)

docs/models/extras/webhook.md

@@ -32,7 +32,12 @@ The events which will trigger the webhook. At least one event type must be selec
 
 ### URL
 
-The URL to which the webhook HTTP request will be made.
+The URL to which the webhook HTTP request will be made. Must be `http://` or `https://`, though
+part or all of the value may be a Jinja2 template rendered at send time (e.g.
+`http://{{ data.name }}.example.com/hook`, or `{{ data.custom_fields.callback_url }}` if the whole
+URL comes from a template). A literal scheme is always validated as such, even if the rest of the
+URL is templated; otherwise the value is checked only for valid Jinja2 syntax, since its rendered
+value isn't known until the webhook actually fires.
 
 ### HTTP Method
 

netbox/extras/models/models.py

@@ -1,4 +1,5 @@
 import json
+import re
 import urllib.parse
 from pathlib import Path
 
@@ -34,7 +35,7 @@ from netbox.models.features import (
 )
 from netbox.models.mixins import OwnerMixin
 from utilities.html import clean_html
-from utilities.jinja2 import render_jinja2, sanitize_http_header
+from utilities.jinja2 import JINJA2_TEMPLATE_RE, render_jinja2, sanitize_http_header, validate_jinja2_syntax
 from utilities.querydict import dict_to_querydict
 from utilities.querysets import RestrictedQuerySet
 from utilities.tables import get_table_for_model
@@ -51,6 +52,11 @@ __all__ = (
     'Webhook',
 )
 
+# Matches a literal URL scheme (RFC 3986), independent of urlsplit()'s netloc parsing -- which can
+# raise ValueError on a malformed host -- so a payload_url's scheme can always be read even when
+# its host is templated or malformed.
+LITERAL_SCHEME_RE = re.compile(r'^([a-zA-Z][a-zA-Z0-9+.-]*):')
+
 
 class EventRule(CustomFieldsMixin, ExportTemplatesMixin, OwnerMixin, TagsMixin, ChangeLoggedModel):
     """
@@ -187,8 +193,9 @@ class Webhook(CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, OwnerMixin, Ch
         max_length=500,
         verbose_name=_('URL'),
         help_text=_(
-            "This URL will be called using the HTTP method defined when the webhook is called. Jinja2 template "
-            "processing is supported with the same context as the request body."
+            "This URL will be called using the HTTP method defined when the webhook is called. Must be "
+            "http:// or https://. Jinja2 template processing is supported (with the same context as the "
+            "request body) for part or all of the URL."
         )
     )
     http_method = models.CharField(
@@ -273,11 +280,40 @@ class Webhook(CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, OwnerMixin, Ch
     def clean(self):
         super().clean()
 
+        errors = {}
+
         # CA file path requires SSL verification enabled
         if not self.ssl_verification and self.ca_file_path:
-            raise ValidationError({
-                'ca_file_path': _('Do not specify a CA certificate file if SSL verification is disabled.')
-            })
+            errors['ca_file_path'] = _('Do not specify a CA certificate file if SSL verification is disabled.')
+
+        # payload_url may be a literal URL or a Jinja2 template (see its help_text). Skipped when
+        # blank; clean_fields() already flags that.
+        if self.payload_url:
+            if JINJA2_TEMPLATE_RE.search(self.payload_url):
+                # A literal, disallowed scheme (e.g. "file://") can never resolve no matter what
+                # else in the value is templated; anything else is checked for template syntax
+                # only, since its rendered result isn't known here.
+                match = LITERAL_SCHEME_RE.match(self.payload_url)
+                if match and match.group(1).lower() not in ('http', 'https'):
+                    errors['payload_url'] = _("Enter a valid URL, beginning with http:// or https://.")
+                else:
+                    try:
+                        validate_jinja2_syntax(self.payload_url)
+                    except ValidationError as e:
+                        errors['payload_url'] = e
+            else:
+                # Fully literal -- validate directly rather than via URLValidator, which rejects
+                # single-label and underscore hosts that `requests` accepts fine. urlsplit() can
+                # raise ValueError for a malformed netloc (e.g. an unbalanced IPv6 bracket).
+                try:
+                    scheme, netloc = urllib.parse.urlsplit(self.payload_url)[:2]
+                except ValueError:
+                    scheme, netloc = '', ''
+                if scheme not in ('http', 'https') or not netloc:
+                    errors['payload_url'] = _("Enter a valid URL, beginning with http:// or https://.")
+
+        if errors:
+            raise ValidationError(errors)
 
     def render_headers(self, context):
         """

netbox/extras/tests/test_models.py

@@ -32,6 +32,7 @@ from extras.models import (
     TableConfig,
     Tag,
     TaggedItem,
+    Webhook,
 )
 from extras.models.mixins import RenderTemplateMixin
 from tenancy.models import Tenant, TenantGroup
@@ -1580,6 +1581,95 @@ class ExportTemplateRenderTestCase(TestCase):
         self.assertEqual(response.content.decode(), 'Site A\nSite B\nSite C\n')
 
 
+class WebhookTestCase(TestCase):
+    """Tests for Webhook.clean()'s validation of payload_url (#22828)."""
+
+    def test_payload_url_accepts_literal_url(self):
+        webhook = Webhook(name='Webhook 1', payload_url='http://example.com/hook')
+        webhook.clean()
+
+    def test_payload_url_rejects_non_url(self):
+        webhook = Webhook(name='Webhook 1', payload_url='not-a-url-at-all')
+        with self.assertRaises(ValidationError) as cm:
+            webhook.clean()
+        self.assertIn('payload_url', cm.exception.message_dict)
+
+    def test_payload_url_rejects_disallowed_scheme(self):
+        webhook = Webhook(name='Webhook 1', payload_url='file:///etc/passwd')
+        with self.assertRaises(ValidationError) as cm:
+            webhook.clean()
+        self.assertIn('payload_url', cm.exception.message_dict)
+
+    def test_payload_url_accepts_jinja2_template(self):
+        """A templated payload_url must not be rejected merely for not being a literal URL."""
+        webhook = Webhook(name='Webhook 1', payload_url='http://{{ data.name }}.example.com/hook')
+        webhook.clean()
+
+    def test_payload_url_accepts_template_using_a_registered_filter(self):
+        webhook = Webhook(name='Webhook 1', payload_url="http://example.com/{{ 'HOME' | env }}")
+        webhook.clean()
+
+    def test_payload_url_rejects_malformed_template_syntax(self):
+        webhook = Webhook(name='Webhook 1', payload_url='http://{{ data.name }.example.com/hook')
+        with self.assertRaises(ValidationError) as cm:
+            webhook.clean()
+        self.assertIn('payload_url', cm.exception.message_dict)
+
+    def test_payload_url_rejects_template_with_unregistered_filter(self):
+        webhook = Webhook(
+            name='Webhook 1', payload_url='http://example.com/{{ data.name | totally_unregistered_filter }}'
+        )
+        with self.assertRaises(ValidationError) as cm:
+            webhook.clean()
+        self.assertIn('payload_url', cm.exception.message_dict)
+
+    def test_payload_url_accepts_single_label_host(self):
+        """A Docker/Kubernetes-style internal service name is a legitimate webhook target (#22832)."""
+        webhook = Webhook(name='Webhook 1', payload_url='http://webhook-receiver:8080/hook')
+        webhook.clean()
+
+    def test_payload_url_accepts_underscore_in_hostname(self):
+        """requests accepts an underscore in a hostname even though Django's URLValidator does not (#22832)."""
+        webhook = Webhook(name='Webhook 1', payload_url='http://my_host.example.com/hook')
+        webhook.clean()
+
+    def test_payload_url_rejects_missing_host(self):
+        webhook = Webhook(name='Webhook 1', payload_url='http:///hook')
+        with self.assertRaises(ValidationError) as cm:
+            webhook.clean()
+        self.assertIn('payload_url', cm.exception.message_dict)
+
+    def test_payload_url_rejects_templated_disallowed_scheme(self):
+        """A literal, disallowed scheme must be rejected even when the rest of the URL is templated (#22832)."""
+        webhook = Webhook(name='Webhook 1', payload_url='file:///{{ data.name }}')
+        with self.assertRaises(ValidationError) as cm:
+            webhook.clean()
+        self.assertIn('payload_url', cm.exception.message_dict)
+
+    def test_blank_payload_url_produces_a_single_error(self):
+        """clean() must not add its own error on top of clean_fields()'s for a blank value (#22832)."""
+        webhook = Webhook(name='Webhook 1', payload_url='')
+        with self.assertRaises(ValidationError) as cm:
+            webhook.full_clean()
+        self.assertEqual(cm.exception.message_dict['payload_url'], ['This field cannot be blank.'])
+
+    def test_none_payload_url_does_not_raise_typeerror(self):
+        webhook = Webhook(name='Webhook 1', payload_url=None)
+        webhook.clean()
+
+    def test_payload_url_accepts_fully_templated_value(self):
+        """A value with no literal scheme at all (the scheme itself is templated) must still be usable (#22832)."""
+        webhook = Webhook(name='Webhook 1', payload_url='{{ data.custom_fields.callback_url }}')
+        webhook.clean()
+
+    def test_payload_url_rejects_malformed_bracketed_host_gracefully(self):
+        """A malformed netloc must raise ValidationError, not an uncaught ValueError from urlsplit() (#22832)."""
+        webhook = Webhook(name='Webhook 1', payload_url='http://[2001:db8::1/hook')
+        with self.assertRaises(ValidationError) as cm:
+            webhook.clean()
+        self.assertIn('payload_url', cm.exception.message_dict)
+
+
 class EventRuleTestCase(TestCase):
 
     def test_action_data_clean_accepts_dict(self):

netbox/utilities/jinja2.py

@@ -3,7 +3,10 @@ import os
 import re
 
 from django.apps import apps
+from django.core.exceptions import ValidationError
+from django.utils.translation import gettext_lazy as _
 from jinja2 import BaseLoader, TemplateNotFound
+from jinja2.exceptions import TemplateSyntaxError
 from jinja2.meta import find_referenced_templates
 from jinja2.sandbox import SandboxedEnvironment
 
@@ -11,16 +14,23 @@ from netbox.config import get_config
 
 __all__ = (
     'DEFAULT_JINJA2_FILTERS',
+    'HTTP_HEADER_INVALID_CHARS_RE',
+    'JINJA2_TEMPLATE_RE',
     'DataFileLoader',
     'env_filter',
     'render_jinja2',
     'sanitize_http_header',
+    'validate_jinja2_syntax',
 )
 
 # Control characters (C0 range plus DEL) which are invalid in an HTTP header value. Notably, this includes the
 # carriage return and line feed characters used to smuggle additional headers (CR/LF injection).
 HTTP_HEADER_INVALID_CHARS_RE = re.compile(r'[\x00-\x1f\x7f]')
 
+# Matches the start of a Jinja2 expression, statement, or comment ({{, {%, {#), to detect whether a
+# template-capable field (e.g. Webhook.payload_url) is being used as a literal value or a template.
+JINJA2_TEMPLATE_RE = re.compile(r'\{[{%#]')
+
 
 def env_filter(name):
     """
@@ -86,6 +96,15 @@ class DataFileLoader(BaseLoader):
 # Utility functions
 #
 
+def _jinja2_filters(filters=None):
+    """
+    Build the Jinja2 filter table: defaults, then instance-configured JINJA2_FILTERS, then any
+    filters passed for this call, in increasing precedence. Shared by render_jinja2() and
+    validate_jinja2_syntax() so both see an identical filter table.
+    """
+    return {**DEFAULT_JINJA2_FILTERS, **get_config().JINJA2_FILTERS, **(filters or {})}
+
+
 def render_jinja2(template_code, context, environment_params=None, data_file=None, debug=False, filters=None):
     """
     Render a Jinja2 template with the provided context. Return the rendered content.
@@ -114,15 +133,25 @@ def render_jinja2(template_code, context, environment_params=None, data_file=Non
         environment_params['loader'] = loader
 
     environment = SandboxedEnvironment(**environment_params)
-
-    # Register default filters, then apply any user-defined filters. User-defined entries take precedence so that
-    # existing JINJA2_FILTERS configurations are never overridden. Any filters passed for this render take precedence
-    # over both so that context-specific (e.g. sanitization) filters cannot be shadowed.
-    all_filters = {**DEFAULT_JINJA2_FILTERS, **get_config().JINJA2_FILTERS, **(filters or {})}
-    environment.filters.update(all_filters)
+    environment.filters.update(_jinja2_filters(filters))
 
     if data_file:
         template = environment.get_template(data_file.path)
     else:
         template = environment.from_string(source=template_code)
     return template.render(**context)
+
+
+def validate_jinja2_syntax(template_code, filters=None):
+    """
+    Validate that template_code is syntactically well-formed Jinja2 -- including that any filters
+    it references are registered -- without rendering it, so no context data is required. Pass the
+    same `filters` used at render time (see render_jinja2()) for an identical filter table. Raises
+    django.core.exceptions.ValidationError on failure.
+    """
+    environment = SandboxedEnvironment(loader=BaseLoader())
+    environment.filters.update(_jinja2_filters(filters))
+    try:
+        environment.compile(template_code)
+    except TemplateSyntaxError as e:
+        raise ValidationError(_("Invalid template: {error}").format(error=e))