diff --git a/docs/models/extras/webhook.md b/docs/models/extras/webhook.md index 1d17bdd49..07db9e33c 100644 --- a/docs/models/extras/webhook.md +++ b/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 diff --git a/netbox/extras/models/models.py b/netbox/extras/models/models.py index 4bd257ccf..fdb371a9f 100644 --- a/netbox/extras/models/models.py +++ b/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): """ diff --git a/netbox/extras/tests/test_models.py b/netbox/extras/tests/test_models.py index 296e7192d..d6862a246 100644 --- a/netbox/extras/tests/test_models.py +++ b/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): diff --git a/netbox/utilities/jinja2.py b/netbox/utilities/jinja2.py index 620697007..5d8f4ffcd 100644 --- a/netbox/utilities/jinja2.py +++ b/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))