Add secure webhook URL validation

django-backend/soroscan/ingest/serializers.py

@@ -292,6 +292,34 @@ class Meta:
             "secret": {"write_only": True},
         }
 
+    def validate_target_url(self, value):
+        """
+        Validate that the target URL uses HTTPS in production mode.
+        HTTP URLs are only allowed when DEBUG=True.
+        """
+        from django.conf import settings
+        from urllib.parse import urlparse
+
+        if not value:
+            raise serializers.ValidationError("Target URL is required.")
+
+        parsed = urlparse(value)
+
+        # In production (DEBUG=False), only allow HTTPS
+        if not settings.DEBUG and parsed.scheme != 'https':
+            raise serializers.ValidationError(
+                "Only HTTPS URLs are allowed in production mode. "
+                "Please use an HTTPS endpoint to ensure secure webhook delivery."
+            )
+
+        # Ensure the scheme is either http or https
+        if parsed.scheme not in ('http', 'https'):
+            raise serializers.ValidationError(
+                f"Invalid URL scheme '{parsed.scheme}'. Only HTTP and HTTPS are supported."
+            )
+
+        return value
+
     def validate_filter_condition(self, value):
         if value in (None, {}):
             return value

django-backend/soroscan/ingest/tests/test_views.py

@@ -357,6 +357,112 @@ def test_webhook_dry_run_rejects_non_object_sample(self, authenticated_client, c
 
         assert response.status_code == status.HTTP_400_BAD_REQUEST
 
+    def test_create_webhook_https_url_accepted_in_production(self, authenticated_client, contract, settings):
+        """Test that HTTPS URLs are accepted when DEBUG=False (production mode)."""
+        settings.DEBUG = False
+        url = reverse("webhook-list")
+        data = {
+            "contract": contract.id,
+            "event_type": "swap",
+            "target_url": "https://secure.example.com/webhook",
+            "is_active": True,
+        }
+        response = authenticated_client.post(url, data)
+
+        assert response.status_code == status.HTTP_201_CREATED
+        assert WebhookSubscription.objects.filter(target_url="https://secure.example.com/webhook").exists()
+
+    def test_create_webhook_http_url_rejected_in_production(self, authenticated_client, contract, settings):
+        """Test that HTTP URLs are rejected when DEBUG=False (production mode)."""
+        settings.DEBUG = False
+        url = reverse("webhook-list")
+        data = {
+            "contract": contract.id,
+            "event_type": "swap",
+            "target_url": "http://insecure.example.com/webhook",
+            "is_active": True,
+        }
+        response = authenticated_client.post(url, data)
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "target_url" in response.data
+        assert "HTTPS" in str(response.data["target_url"])
+        assert "production" in str(response.data["target_url"]).lower()
+
+    def test_create_webhook_http_url_accepted_in_debug(self, authenticated_client, contract, settings):
+        """Test that HTTP URLs are accepted when DEBUG=True (development mode)."""
+        settings.DEBUG = True
+        url = reverse("webhook-list")
+        data = {
+            "contract": contract.id,
+            "event_type": "swap",
+            "target_url": "http://localhost:8000/webhook",
+            "is_active": True,
+        }
+        response = authenticated_client.post(url, data)
+
+        assert response.status_code == status.HTTP_201_CREATED
+        assert WebhookSubscription.objects.filter(target_url="http://localhost:8000/webhook").exists()
+
+    def test_create_webhook_https_url_accepted_in_debug(self, authenticated_client, contract, settings):
+        """Test that HTTPS URLs are also accepted when DEBUG=True."""
+        settings.DEBUG = True
+        url = reverse("webhook-list")
+        data = {
+            "contract": contract.id,
+            "event_type": "swap",
+            "target_url": "https://example.com/webhook",
+            "is_active": True,
+        }
+        response = authenticated_client.post(url, data)
+
+        assert response.status_code == status.HTTP_201_CREATED
+        assert WebhookSubscription.objects.filter(target_url="https://example.com/webhook").exists()
+
+    def test_create_webhook_invalid_scheme_rejected(self, authenticated_client, contract, settings):
+        """Test that URLs with invalid schemes (not http/https) are rejected."""
+        settings.DEBUG = True
+        url = reverse("webhook-list")
+        data = {
+            "contract": contract.id,
+            "event_type": "swap",
+            "target_url": "ftp://example.com/webhook",
+            "is_active": True,
+        }
+        response = authenticated_client.post(url, data)
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "target_url" in response.data
+        assert "ftp" in str(response.data["target_url"]).lower()
+
+    def test_update_webhook_http_url_rejected_in_production(self, authenticated_client, contract, settings):
+        """Test that updating to HTTP URL is rejected in production mode."""
+        settings.DEBUG = False
+        webhook = WebhookSubscriptionFactory(contract=contract, target_url="https://example.com/webhook")
+        url = reverse("webhook-detail", args=[webhook.id])
+        data = {
+            "target_url": "http://insecure.example.com/webhook",
+        }
+        response = authenticated_client.patch(url, data)
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "target_url" in response.data
+        assert "HTTPS" in str(response.data["target_url"])
+
+    def test_update_webhook_https_url_accepted_in_production(self, authenticated_client, contract, settings):
+        """Test that updating to HTTPS URL is accepted in production mode."""
+        settings.DEBUG = False
+        webhook = WebhookSubscriptionFactory(contract=contract, target_url="https://example.com/webhook")
+        url = reverse("webhook-detail", args=[webhook.id])
+        data = {
+            "target_url": "https://new-secure.example.com/webhook",
+        }
+        response = authenticated_client.patch(url, data)
+
+        assert response.status_code == status.HTTP_200_OK
+        webhook.refresh_from_db()
+        assert webhook.target_url == "https://new-secure.example.com/webhook"
+
 
 @pytest.mark.django_db
 class TestRecordEventView:

django-backend/test_webhook_url_validation.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python
+"""
+Quick validation test for webhook URL security.
+This script tests the URL validation logic without requiring the full test suite.
+"""
+import os
+import sys
+import django
+
+# Setup Django
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'soroscan.settings')
+sys.path.insert(0, os.path.dirname(__file__))
+django.setup()
+
+from django.conf import settings
+from rest_framework import serializers
+from urllib.parse import urlparse
+
+
+def validate_target_url(value, debug_mode):
+    """Replicate the validation logic from WebhookSubscriptionSerializer"""
+    if not value:
+        raise serializers.ValidationError("Target URL is required.")
+
+    parsed = urlparse(value)
+
+    # In production (DEBUG=False), only allow HTTPS
+    if not debug_mode and parsed.scheme != 'https':
+        raise serializers.ValidationError(
+            "Only HTTPS URLs are allowed in production mode. "
+            "Please use an HTTPS endpoint to ensure secure webhook delivery."
+        )
+
+    # Ensure the scheme is either http or https
+    if parsed.scheme not in ('http', 'https'):
+        raise serializers.ValidationError(
+            f"Invalid URL scheme '{parsed.scheme}'. Only HTTP and HTTPS are supported."
+        )
+
+    return value
+
+
+def test_validation():
+    """Run validation tests"""
+    print("Testing Webhook URL Validation\n" + "="*50)
+
+    test_cases = [
+        # (url, debug_mode, should_pass, description)
+        ("https://example.com/webhook", False, True, "HTTPS in production"),
+        ("http://example.com/webhook", False, False, "HTTP in production (should fail)"),
+        ("http://localhost:8000/webhook", True, True, "HTTP in debug mode"),
+        ("https://example.com/webhook", True, True, "HTTPS in debug mode"),
+        ("ftp://example.com/webhook", True, False, "Invalid scheme (should fail)"),
+        ("ftp://example.com/webhook", False, False, "Invalid scheme in production (should fail)"),
+    ]
+
+    passed = 0
+    failed = 0
+
+    for url, debug_mode, should_pass, description in test_cases:
+        mode_str = "DEBUG=True" if debug_mode else "DEBUG=False"
+        try:
+            validate_target_url(url, debug_mode)
+            if should_pass:
+                print(f"✓ PASS: {description} ({mode_str})")
+                passed += 1
+            else:
+                print(f"✗ FAIL: {description} ({mode_str}) - Expected validation error but passed")
+                failed += 1
+        except serializers.ValidationError as e:
+            if not should_pass:
+                print(f"✓ PASS: {description} ({mode_str}) - Correctly rejected: {e}")
+                passed += 1
+            else:
+                print(f"✗ FAIL: {description} ({mode_str}) - Unexpected error: {e}")
+                failed += 1
+
+    print("\n" + "="*50)
+    print(f"Results: {passed} passed, {failed} failed")
+
+    return failed == 0
+
+
+if __name__ == "__main__":
+    success = test_validation()
+    sys.exit(0 if success else 1)