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

backend/backend/middleware.py

@@ -0,0 +1,75 @@
+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
+        from django.core.cache import cache
+        from django.core.exceptions import DisallowedHost
+        cache_key = f"org_for_host_{host}"
+        org_id = cache.get(cache_key)
+
+        if org_id == "MISSING":
+            raise DisallowedHost(f"Invalid HTTP_HOST header: {host}")
+
+        if org_id:
+            try:
+                org = Organization.objects.get(id=org_id)
+            except Organization.DoesNotExist:
+                cache.delete(cache_key)
+                raise DisallowedHost(f"Invalid HTTP_HOST header: {host}")
+        else:
+            try:
+                org = Organization.objects.get(custom_tracking_domain=host)
+                cache.set(cache_key, org.id, 300)
+            except Organization.DoesNotExist:
+                cache.set(cache_key, "MISSING", 300)
+                # If no matching organization exists, block the request
+                raise DisallowedHost(f"Invalid HTTP_HOST header: {host}")
+
+        # 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:
+            sanitized_path = request.path.replace('\n', '').replace('\r', '')
+            logger.warning(f"Blocked non-tracking access on custom domain {host} for path {sanitized_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/backend/settings.py

@@ -56,7 +56,30 @@ def _normalize_google_redirect_uri(raw_uri: str, backend_base_url: str) -> str:
     'MAILBOX_CREDENTIALS_ENCRYPTION_KEY',
     'fallback-insecure-key-for-local-dev-and-testing' if (DEBUG or TESTING) else '',
 )
-ALLOWED_HOSTS = ['*']
+class DynamicAllowedHosts(list):
+    def __contains__(self, host):
+        from urllib.parse import urlparse
+        import os
+
+        canonical_host = urlparse(os.getenv('BACKEND_BASE_URL', 'https://leadorbit.onrender.com')).hostname or 'leadorbit.onrender.com'
+        if host in ['localhost', '127.0.0.1', '[::1]', canonical_host]:
+            return True
+
+        try:
+            from django.core.cache import cache
+            cache_key = f"allowed_host_{host}"
+            is_allowed = cache.get(cache_key)
+            if is_allowed is not None:
+                return is_allowed
+
+            from tenants.models import Organization
+            is_allowed = Organization.objects.filter(custom_tracking_domain=host).exists()
+            cache.set(cache_key, is_allowed, 300)
+            return is_allowed
+        except Exception:
+            return False
+
+ALLOWED_HOSTS = DynamicAllowedHosts()
 CORS_ALLOWED_ORIGINS = [
     "http://localhost:3000",
     "http://127.0.0.1:3000",
@@ -87,6 +110,7 @@ def _normalize_google_redirect_uri(raw_uri: str, backend_base_url: str) -> str:
     'django.middleware.csrf.CsrfViewMiddleware',
     'django.contrib.auth.middleware.AuthenticationMiddleware',
     'tenants.middleware.TenantMiddleware',  # custom tenant isolation
+    'backend.middleware.CustomDomainMiddleware', # Custom tracking domain routing
     'tenants.security.RateLimitMiddleware',  # API rate limiting
     'tenants.security.SecurityHeadersMiddleware',  # security headers
     'django.contrib.messages.middleware.MessageMiddleware',

backend/campaigns/gmail_service.py

@@ -161,14 +161,21 @@ def _extract_failed_recipients(message, account_email=None):
     return fallback_emails if len(fallback_emails) == 1 else []
 
 
-def build_unsubscribe_url(lead):
+def build_unsubscribe_url(lead, organization=None):
     """
-    Build a signed unsubscribe URL for a lead using the backend base URL.
+    Build a signed unsubscribe URL for a lead using the backend base URL or custom tracking domain.
     """
     from .utils import generate_unsubscribe_token
+    from django.conf import settings
 
     token = generate_unsubscribe_token(lead.id)
-    return f"{settings.BACKEND_BASE_URL}/api/v1/unsubscribe/{lead.id}/{token}/"
+    if organization and organization.custom_tracking_domain:
+        from tenants.utils import is_local_tracking_domain
+        scheme = 'http' if is_local_tracking_domain(organization.custom_tracking_domain) else 'https'
+        base_url = f"{scheme}://{organization.custom_tracking_domain}"
+    else:
+        base_url = settings.BACKEND_BASE_URL.rstrip('/')
+    return f"{base_url}/api/v1/unsubscribe/{lead.id}/{token}/"
 
 
 def send_gmail(account, to_email, subject, body_html, unsubscribe_url=None, thread_id=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,13 @@ 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:
+        from tenants.utils import is_local_tracking_domain
+        scheme = 'http' if is_local_tracking_domain(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 +590,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 +602,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 +611,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,116 @@
+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.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.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.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, ALLOWED_HOSTS=['track.acme.test', 'localhost', '127.0.0.1'])
+class CustomDomainMiddlewareTests(TestCase):
+    def setUp(self):
+        from django.core.cache import cache
+        cache.clear()
+        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,52 @@ 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
+
+            from tenants.utils import is_local_tracking_domain
+
+            if is_local_tracking_domain(self.custom_tracking_domain):
+                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'
+
+                resolver = dns.resolver.Resolver()
+                resolver.timeout = 2.0
+                resolver.lifetime = 5.0
+                answers = resolver.resolve(self.custom_tracking_domain, 'CNAME')
+
+                valid = False
+                for rdata in answers:
+                    target = rdata.target.to_text().rstrip('.').lower()
+                    if target == target_domain.lower():
+                        valid = True
+                        break
+
+                if not valid:
+                    raise ValidationError({'custom_tracking_domain': f'CNAME record must point to {target_domain}'})
+            except dns.resolver.NXDOMAIN as e:
+                raise ValidationError({'custom_tracking_domain': 'Domain does not exist.'}) from e
+            except dns.resolver.NoAnswer as e:
+                raise ValidationError({'custom_tracking_domain': 'No CNAME record found for this domain.'}) from e
+            except dns.resolver.Timeout as e:
+                raise ValidationError({'custom_tracking_domain': 'DNS query timed out.'}) from e
+            except ValidationError:
+                raise
+            except Exception as e:
+                raise ValidationError({'custom_tracking_domain': f'DNS validation failed: {e}'}) from e
 
     def __str__(self):
         return self.name

backend/tenants/utils.py

@@ -0,0 +1,10 @@
+from django.conf import settings
+
+def is_local_tracking_domain(domain):
+    """
+    Returns True if the domain should be treated as a local tracking domain.
+    """
+    if not getattr(settings, 'DEBUG', False):
+        return False
+    domain = domain.lower()
+    return domain.startswith('localhost') or 'localhost' in domain