feat: implement custom tracking domain CNAME integration (#485)

backend/backend/middleware.py

@@ -0,0 +1,57 @@
+import logging
+from django.utils.deprecation import MiddlewareMixin
+from django.http import HttpResponseNotFound
+from django.conf import settings
+from urllib.parse import urlparse
+from tenants.models import Organization
+from tenants.middleware import _thread_locals
+
+logger = logging.getLogger(__name__)
+
+class CustomDomainMiddleware(MiddlewareMixin):
+    """
+    Middleware that intercepts requests with custom domains in the Host header,
+    identifies the associated Organization, and restricts access to ONLY
+    the tracking endpoints, rewriting/routing them to existing handlers.
+    """
+
+    TRACKING_ENDPOINTS = (
+        '/api/v1/clicks/track/',
+        '/api/v1/webhooks/email/',
+        '/api/v1/unsubscribe/',
+    )
+
+    def process_request(self, request):
+        host = request.get_host().split(':')[0].lower()
+        base_url = getattr(settings, 'BACKEND_BASE_URL', 'http://localhost:8000')
+        parsed_base = urlparse(base_url)
+        default_host = (parsed_base.hostname or 'localhost').lower()
+
+        # If the request comes via the default host or localhost, skip custom domain processing
+        if host == default_host or host == '127.0.0.1' or host == 'localhost':
+            return None
+
+        # Check if the host matches a custom tracking domain
+        try:
+            org = Organization.objects.get(custom_tracking_domain=host)
+        except Organization.DoesNotExist:
+            # If no matching organization exists, continue normal request processing.
+            return None
+
+        # Custom domain matched. Identify the correct tenant.
+        _thread_locals.tenant = org
+
+        # Enforce strict tenant isolation: A custom domain must ONLY access tracking resources.
+        is_tracking_endpoint = False
+        for endpoint in self.TRACKING_ENDPOINTS:
+            if request.path.startswith(endpoint):
+                is_tracking_endpoint = True
+                break
+
+        if not is_tracking_endpoint:
+            logger.warning(f"Blocked non-tracking access on custom domain {host} for path {request.path}")
+            return HttpResponseNotFound("Not Found")
+
+        # The request will naturally proceed to the existing tracking handlers
+        # preserving the request method and query parameters.
+        return None

backend/campaigns/tasks.py

@@ -509,7 +509,7 @@ def _execute_call_step(clead, step, now=None):
 
 
 
-def rewrite_email_links(html_body, campaign_lead_id, step_id):
+def rewrite_email_links(html_body, campaign_lead_id, step_id, organization=None):
     """
     Parses the email body, finds all anchor tags, and replaces the href
     with our tracking redirect URL.
@@ -523,7 +523,12 @@ def rewrite_email_links(html_body, campaign_lead_id, step_id):
     token_payload = f"{campaign_lead_id}:{step_id}"
     signed_token = signer.sign(token_payload)
 
-    base_url = getattr(django_settings, 'BACKEND_BASE_URL', 'http://127.0.0.1:8000').rstrip('/')
+    if organization and organization.custom_tracking_domain:
+        scheme = 'http' if getattr(django_settings, 'DEBUG', False) and 'localhost' in organization.custom_tracking_domain else 'https'
+        base_url = f"{scheme}://{organization.custom_tracking_domain}"
+    else:
+        base_url = getattr(django_settings, 'BACKEND_BASE_URL', 'http://127.0.0.1:8000').rstrip('/')
+
     tracking_endpoint = f"{base_url}/api/v1/clicks/track/"
 
     for a_tag in soup.find_all('a', href=True):
@@ -584,7 +589,7 @@ def send_email_step(campaign_lead_id, step_id):
         subject, body = personalize_email(step.template_subject, step.template_body, clead.lead)
 
 
-        body = rewrite_email_links(body, campaign_lead_id, step_id)
+        body = rewrite_email_links(body, campaign_lead_id, step_id, organization=clead.organization)
         # -------------------------------------------
 
         account = clead.campaign.connected_account
@@ -596,7 +601,7 @@ def send_email_step(campaign_lead_id, step_id):
                         clead.lead.email,
                         subject,
                         body,
-                        unsubscribe_url=build_unsubscribe_url(clead.lead),
+                        unsubscribe_url=build_unsubscribe_url(clead.lead, organization=clead.organization),
                     )
                     logger.info(f"Gmail SENT to {clead.lead.email} | msg_id={message_id}")
                 elif account.provider == 'CUSTOM':
@@ -605,7 +610,7 @@ def send_email_step(campaign_lead_id, step_id):
                         clead.lead.email,
                         subject,
                         body,
-                        unsubscribe_url=build_unsubscribe_url(clead.lead),
+                        unsubscribe_url=build_unsubscribe_url(clead.lead, organization=clead.organization),
                     )
                     logger.info(f"SMTP SENT to {clead.lead.email} | msg_id={message_id}")
                 else:

backend/campaigns/test_custom_domains.py

@@ -0,0 +1,114 @@
+from unittest.mock import patch
+from django.test import TestCase, override_settings
+from django.core.exceptions import ValidationError
+from tenants.models import Organization
+from django.conf import settings
+from rest_framework.test import APIClient
+from campaigns.tasks import rewrite_email_links
+from campaigns.gmail_service import build_unsubscribe_url
+
+class MockDnsRdata:
+    def __init__(self, target):
+        self.target_text = target
+
+    @property
+    def target(self):
+        class _Target:
+            def to_text(self_inner):
+                return self.target_text
+        return _Target()
+
+@override_settings(BACKEND_BASE_URL='https://leadorbit.onrender.com')
+class CustomDomainModelTests(TestCase):
+    def setUp(self):
+        self.org = Organization.objects.create(name="Acme Corp")
+
+    @patch('dns.resolver.resolve')
+    def test_valid_cname_configuration(self, mock_resolve):
+        # Mocking dns.resolver to return a valid CNAME pointing to our target
+        mock_resolve.return_value = [MockDnsRdata('leadorbit.onrender.com.')]
+
+        self.org.custom_tracking_domain = 'track.acme.test'
+        # Should not raise exception
+        self.org.clean()
+        self.org.save()
+        self.assertEqual(self.org.custom_tracking_domain, 'track.acme.test')
+
+    @patch('dns.resolver.resolve')
+    def test_invalid_cname_configuration(self, mock_resolve):
+        mock_resolve.return_value = [MockDnsRdata('wrong.target.com.')]
+
+        self.org.custom_tracking_domain = 'track.acme.test'
+        with self.assertRaises(ValidationError) as ctx:
+            self.org.clean()
+        self.assertIn("CNAME record must point to", str(ctx.exception))
+
+    @patch('dns.resolver.resolve')
+    def test_dns_validation_failures(self, mock_resolve):
+        import dns.resolver
+        mock_resolve.side_effect = dns.resolver.NXDOMAIN
+
+        self.org.custom_tracking_domain = 'notexist.acme.test'
+        with self.assertRaises(ValidationError) as ctx:
+            self.org.clean()
+        self.assertIn("Domain does not exist", str(ctx.exception))
+
+
+from django.test import RequestFactory
+
+@override_settings(DEBUG=False)
+class CustomDomainMiddlewareTests(TestCase):
+    def setUp(self):
+        self.factory = RequestFactory()
+        self.org = Organization.objects.create(name="Acme Corp", custom_tracking_domain='track.acme.test')
+        from backend.middleware import CustomDomainMiddleware
+        self.middleware = CustomDomainMiddleware(lambda req: getattr(req, '_fake_response', None))
+
+    def test_middleware_hostname_routing_tracking_endpoint(self):
+        request = self.factory.get('/api/v1/clicks/track/', HTTP_HOST='track.acme.test')
+        request._fake_response = "OK"
+        response = self.middleware(request)
+        self.assertEqual(response, "OK") # Allowed through
+
+    def test_tenant_isolation_non_tracking_endpoint(self):
+        request = self.factory.get('/api/v1/organizations/', HTTP_HOST='track.acme.test')
+        response = self.middleware(request)
+        self.assertEqual(response.status_code, 404)
+
+    def test_default_tracking_fallback(self):
+        request = self.factory.get('/api/v1/organizations/', HTTP_HOST='localhost')
+        request._fake_response = "OK"
+        response = self.middleware(request)
+        self.assertEqual(response, "OK") # Allowed through
+
+
+class CustomDomainURLGenerationTests(TestCase):
+    def setUp(self):
+        self.org_custom = Organization.objects.create(name="Custom Org", custom_tracking_domain='track.custom.test')
+        self.org_default = Organization.objects.create(name="Default Org")
+
+        # Need mock campaign/lead to test rewrite_email_links, or just use string checking
+        from campaigns.models import Campaign, SequenceStep, CampaignLead
+        from leads.models import Lead
+
+        self.campaign = Campaign.objects.create(organization=self.org_custom, name="Test", status="ACTIVE", sent_count=0, bounced_count=0)
+        self.step = SequenceStep.objects.create(organization=self.org_custom, campaign=self.campaign, step_order=1, channel_type="EMAIL")
+        self.lead = Lead.objects.create(organization=self.org_custom, email="test@test.com")
+        self.clead = CampaignLead.objects.create(
+            organization=self.org_custom, campaign=self.campaign, lead=self.lead, current_step=self.step
+        )
+
+    def test_rewrite_email_links_with_custom_domain(self):
+        html = '<a href="https://google.com">Google</a>'
+        rewritten = rewrite_email_links(html, self.clead.id, self.step.id, organization=self.org_custom)
+        self.assertIn('https://track.custom.test/api/v1/clicks/track/?t=', rewritten)
+
+    def test_rewrite_email_links_default_fallback(self):
+        html = '<a href="https://google.com">Google</a>'
+        rewritten = rewrite_email_links(html, self.clead.id, self.step.id, organization=self.org_default)
+        self.assertIn('/api/v1/clicks/track/', rewritten)
+        self.assertNotIn('track.custom.test', rewritten)
+
+    def test_build_unsubscribe_url_with_custom_domain(self):
+        url = build_unsubscribe_url(self.lead, organization=self.org_custom)
+        self.assertIn('https://track.custom.test/api/v1/unsubscribe/', url)

backend/tenants/migrations/0003_organization_custom_tracking_domain.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.0.14 on 2026-07-12 15:38
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('tenants', '0002_organization_enable_ai_personalization_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='organization',
+            name='custom_tracking_domain',
+            field=models.CharField(blank=True, max_length=255, null=True, unique=True),
+        ),
+    ]

backend/tenants/models.py

@@ -9,6 +9,47 @@ class Organization(models.Model):
     created_at = models.DateTimeField(auto_now_add=True)
     gemini_api_key = models.CharField(max_length=255, blank=True, null=True)
     enable_ai_personalization = models.BooleanField(default=True)
+    custom_tracking_domain = models.CharField(max_length=255, blank=True, null=True, unique=True)
+
+    def clean(self):
+        super().clean()
+        if self.custom_tracking_domain:
+            self.custom_tracking_domain = self.custom_tracking_domain.lower().strip()
+
+            from django.conf import settings
+            from django.core.exceptions import ValidationError
+            import dns.resolver
+            from urllib.parse import urlparse
+
+            if settings.DEBUG and self.custom_tracking_domain.startswith('localhost'):
+                return
+
+            try:
+                base_url = getattr(settings, 'BACKEND_BASE_URL', 'https://leadorbit.onrender.com')
+                parsed = urlparse(base_url)
+                target_domain = parsed.hostname or 'leadorbit.onrender.com'
+
+                answers = dns.resolver.resolve(self.custom_tracking_domain, 'CNAME')
+
+                valid = False
+                for rdata in answers:
+                    target = rdata.target.to_text().rstrip('.')
+                    if target == target_domain:
+                        valid = True
+                        break
+
+                if not valid:
+                    raise ValidationError({'custom_tracking_domain': f'CNAME record must point to {target_domain}'})
+            except dns.resolver.NXDOMAIN:
+                raise ValidationError({'custom_tracking_domain': 'Domain does not exist.'})
+            except dns.resolver.NoAnswer:
+                raise ValidationError({'custom_tracking_domain': 'No CNAME record found for this domain.'})
+            except dns.resolver.Timeout:
+                raise ValidationError({'custom_tracking_domain': 'DNS query timed out.'})
+            except ValidationError:
+                raise
+            except Exception as e:
+                raise ValidationError({'custom_tracking_domain': f'DNS validation failed: {str(e)}'})
 
     def __str__(self):
         return self.name