diff --git a/django-backend/soroscan/ingest/serializers.py b/django-backend/soroscan/ingest/serializers.py index 6c9fa42b..d2ab078c 100644 --- a/django-backend/soroscan/ingest/serializers.py +++ b/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 diff --git a/django-backend/soroscan/ingest/tests/test_views.py b/django-backend/soroscan/ingest/tests/test_views.py index 4913f910..7df8b468 100644 --- a/django-backend/soroscan/ingest/tests/test_views.py +++ b/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: diff --git a/django-backend/test_webhook_url_validation.py b/django-backend/test_webhook_url_validation.py new file mode 100644 index 00000000..3da2708c --- /dev/null +++ b/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)