Closes #22761: Do not follow redirects when sending webhooks

requests.Session.send() defaults allow_redirects=True, and send_webhook()'s call
never overrode it. A webhook destination's response therefore controlled where
the request actually went: a destination that is legitimate at configuration
time but is later compromised (or is malicious from the outset while appearing
legitimate) could respond with a redirect to an arbitrary address -- including
an internal one -- bypassing whatever destination was actually configured.

This is distinct from the documented threat model's carve-out for webhook
authors deliberately targeting arbitrary endpoints (THREAT_MODEL.md): that
carve-out covers the operator's own configured payload_url, not a third
party's response silently redirecting the request elsewhere.

Passes allow_redirects=False to session.send(); a 3xx response now surfaces as
a failed webhook via the existing status-code check, same as any other
non-2xx response.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Brian Tiemann 2026-07-23 15:34:29 -04:00
parent b6ff8654e0
commit e536a4db4a
2 changed files with 77 additions and 2 deletions

View File

@ -1,6 +1,8 @@
import json
import logging
import threading
import uuid
from http.server import BaseHTTPRequestHandler, HTTPServer
from io import BytesIO
from unittest import skipIf
from unittest.mock import Mock, patch
@ -10,8 +12,10 @@ from django.conf import settings
from django.http import HttpResponse
from django.test import RequestFactory, TestCase, tag
from django.urls import reverse
from django.utils import timezone
from PIL import Image
from requests import Session
from requests.exceptions import RequestException
from rest_framework import status
from core.choices import ManagedFileRootPathChoices
@ -869,3 +873,73 @@ class WebhookRenderHeadersTest(TestCase):
self.assertEqual(list(headers.keys()), ['X-Object'])
self.assertNotIn('X-Injected', headers)
self.assertEqual(headers['X-Object'], 'legitX-Injected: evil')
class WebhookRedirectTest(TestCase):
"""
Regression test for #22761: a webhook destination's response must not be able to redirect
the request elsewhere. A destination that is legitimate at configuration time but is later
compromised (or is malicious from the outset while appearing legitimate) could otherwise
respond with a redirect to an arbitrary address -- including an internal one -- that was
never configured as the webhook's target.
"""
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.redirect_target_hits = []
class RedirectTargetHandler(BaseHTTPRequestHandler):
def do_GET(handler_self):
cls.redirect_target_hits.append(handler_self.path)
handler_self.send_response(200)
handler_self.end_headers()
def log_message(handler_self, *args):
pass
class RedirectingHandler(BaseHTTPRequestHandler):
def do_POST(handler_self):
handler_self.send_response(302)
handler_self.send_header('Location', cls.redirect_target_url)
handler_self.end_headers()
def log_message(handler_self, *args):
pass
cls.redirect_target_server = HTTPServer(('127.0.0.1', 0), RedirectTargetHandler)
cls.redirect_target_url = f'http://127.0.0.1:{cls.redirect_target_server.server_port}/unexpected-target'
cls.webhook_server = HTTPServer(('127.0.0.1', 0), RedirectingHandler)
cls.webhook_url = f'http://127.0.0.1:{cls.webhook_server.server_port}/webhook'
threading.Thread(target=cls.redirect_target_server.serve_forever, daemon=True).start()
threading.Thread(target=cls.webhook_server.serve_forever, daemon=True).start()
@classmethod
def tearDownClass(cls):
cls.redirect_target_server.shutdown()
cls.webhook_server.shutdown()
super().tearDownClass()
def test_redirect_is_not_followed(self):
webhook = Webhook.objects.create(name='Redirect Test', payload_url=self.webhook_url)
event_rule = EventRule.objects.create(
name='Redirect Test',
event_types=[OBJECT_CREATED],
action_type=EventRuleActionChoices.WEBHOOK,
action_object_type=ObjectType.objects.get(app_label='extras', model='webhook'),
action_object_id=webhook.id,
)
with self.assertRaises(RequestException):
send_webhook(
event_rule=event_rule,
object_type=ObjectType.objects.get_for_model(Site),
event_type=OBJECT_CREATED,
data={'name': 'Test Site'},
timestamp=timezone.now().isoformat(),
username='testuser',
)
# The webhook's configured destination redirected to this address; it must never
# actually have been reached.
self.assertEqual(self.redirect_target_hits, [])

View File

@ -120,13 +120,14 @@ def send_webhook(event_rule, object_type, event_type, data, timestamp, username,
if webhook.secret != '':
prepared_request.headers['X-Hook-Signature'] = generate_signature(prepared_request.body, webhook.secret)
# Send the request
# Send the request. Redirects are not followed: the destination's response would otherwise
# control where the request actually goes, bypassing whatever destination was configured (#22761).
with requests.Session() as session:
session.verify = webhook.ssl_verification
if webhook.ca_file_path:
session.verify = webhook.ca_file_path
proxies = resolve_proxies(url=url, context={'client': webhook})
response = session.send(prepared_request, proxies=proxies)
response = session.send(prepared_request, proxies=proxies, allow_redirects=False)
if 200 <= response.status_code <= 299:
logger.info(f"Request succeeded; response status {response.status_code}")