Custom domain: check CNAME when adding domains

readthedocs/projects/forms.py

@@ -6,6 +6,8 @@
 from re import fullmatch
 from urllib.parse import urlparse
 
+import dns.name
+import dns.resolver
 import pytz
 from allauth.socialaccount.models import SocialAccount
 from django import forms
@@ -1025,8 +1027,71 @@ def clean_domain(self):
             if invalid_domain and domain_string.endswith(invalid_domain):
                 raise forms.ValidationError(f"{invalid_domain} is not a valid domain.")
 
+        self._check_for_suspicious_cname(domain_string)
+
         return domain_string
 
+    def _check_for_suspicious_cname(self, domain):
+        """
+        Check if a domain has a suspicious CNAME record.
+
+        The domain is suspicious if:
+
+        - Has a CNAME pointing to another CNAME.
+          This prevents the subdomain from being hijacked if the last subdomain is already on RTD,
+          but the user didn't register the other subdomain.
+          Example: doc.example.com -> docs.example.com -> readthedocs.io,
+          We don't want to allow doc.example.com to be added.
+        - Has a CNAME pointing to the APEX domain.
+          This prevents a subdomain from being hijacked if the APEX domain is already on RTD.
+          A common case is `www` pointing to the APEX domain, but users didn't register the
+          `www` subdomain, only the APEX domain.
+          Example: www.example.com -> example.com,
+          we don't want to allow www.example.com to be added.
+        """
+        cname = self._get_cname(domain)
+        # Doesn't have a CNAME record, we are good.
+        if not cname:
+            return
+
+        # If the domain has a CNAME pointing to the APEX domain, that's not good.
+        # This check isn't perfect, but it's a good enoug heuristic
+        # to dectect CNAMES like www.example.com -> example.com.
+        if f"{domain}.".endswith(f".{cname}"):
+            raise forms.ValidationError(
+                _(
+                    "This domain has a CNAME record pointing to the APEX domain. "
+                    "Please remove the CNAME before adding the domain.",
+                ),
+            )
+
+        second_cname = self._get_cname(cname)
+        # The domain has a CNAME pointing to another CNAME, That's not good.
+        if second_cname:
+            raise forms.ValidationError(
+                _(
+                    "This domain has a CNAME record pointing to another CNAME. "
+                    "Please remove the CNAME before adding the domain.",
+                ),
+            )
+
+    def _get_cname(self, domain):
+        try:
+            answers = dns.resolver.resolve(domain, "CNAME", lifetime=1)
+            return str(answers[0].target)
+        except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN):
+            return None
+        except dns.resolver.LifetimeTimeout:
+            raise forms.ValidationError(
+                _(
+                    "DNS resolution timed out. Make sure the domain is correct, or try again later."
+                ),
+            )
+        except dns.name.EmptyLabel:
+            raise forms.ValidationError(
+                _("The domain is not valid."),
+            )
+
     def clean_canonical(self):
         canonical = self.cleaned_data["canonical"]
         pk = self.instance.pk

readthedocs/projects/tests/test_domain_views.py

@@ -1,3 +1,6 @@
+from unittest import mock
+
+import dns.resolver
 from django.contrib.auth.models import User
 from django.contrib.messages import get_messages
 from django.test import TestCase, override_settings
@@ -109,6 +112,67 @@ def test_edit_domain_on_subproject(self):
         self.assertEqual(domain.domain, "test.example.com")
         self.assertEqual(domain.canonical, False)
 
+    @mock.patch("readthedocs.projects.forms.dns.resolver.resolve")
+    def test_create_domain_with_chained_cname_record(self, dns_resolve_mock):
+        dns_resolve_mock.side_effect = [
+            [mock.MagicMock(target="docs.example.com.")],
+            [mock.MagicMock(target="readthedocs.io.")],
+        ]
+        resp = self.client.post(
+            reverse("projects_domains_create", args=[self.project.slug]),
+            data={"domain": "docs2.example.com"},
+        )
+        assert resp.status_code == 200
+        form = resp.context_data["form"]
+        assert not form.is_valid()
+        assert (
+            "This domain has a CNAME record pointing to another CNAME"
+            in form.errors["domain"][0]
+        )
+
+    @mock.patch("readthedocs.projects.forms.dns.resolver.resolve")
+    def test_create_domain_with_cname_record_to_apex_domain(self, dns_resolve_mock):
+        dns_resolve_mock.side_effect = [
+            [mock.MagicMock(target="example.com.")],
+        ]
+        resp = self.client.post(
+            reverse("projects_domains_create", args=[self.project.slug]),
+            data={"domain": "www.example.com"},
+        )
+        assert resp.status_code == 200
+        form = resp.context_data["form"]
+        assert not form.is_valid()
+        assert (
+            "This domain has a CNAME record pointing to the APEX domain"
+            in form.errors["domain"][0]
+        )
+
+    @mock.patch("readthedocs.projects.forms.dns.resolver.resolve")
+    def test_create_domain_cname_timeout(self, dns_resolve_mock):
+        dns_resolve_mock.side_effect = dns.resolver.LifetimeTimeout
+        resp = self.client.post(
+            reverse("projects_domains_create", args=[self.project.slug]),
+            data={"domain": "docs.example.com"},
+        )
+        assert resp.status_code == 200
+        form = resp.context_data["form"]
+        assert not form.is_valid()
+        assert "DNS resolution timed out" in form.errors["domain"][0]
+
+    @mock.patch("readthedocs.projects.forms.dns.resolver.resolve")
+    def test_create_domain_with_single_cname(self, dns_resolve_mock):
+        dns_resolve_mock.side_effect = [
+            [mock.MagicMock(target="readthedocs.io.")],
+            dns.resolver.NoAnswer,
+        ]
+        resp = self.client.post(
+            reverse("projects_domains_create", args=[self.project.slug]),
+            data={"domain": "docs.example.com"},
+        )
+        assert resp.status_code == 302
+        domain = self.project.domains.first()
+        assert domain.domain == "docs.example.com"
+
 
 @override_settings(RTD_ALLOW_ORGANIZATIONS=True)
 class TestDomainViewsWithOrganizations(TestDomainViews):

requirements/deploy.txt

@@ -180,6 +180,8 @@ djangorestframework-api-key==3.0.0
     # via -r requirements/pip.txt
 djangorestframework-jsonp==1.0.2
     # via -r requirements/pip.txt
+dnspython==2.7.0
+    # via -r requirements/pip.txt
 docker==6.1.2
     # via -r requirements/pip.txt
 docutils==0.21.2

requirements/docker.txt

@@ -190,6 +190,8 @@ djangorestframework-api-key==3.0.0
     # via -r requirements/pip.txt
 djangorestframework-jsonp==1.0.2
     # via -r requirements/pip.txt
+dnspython==2.7.0
+    # via -r requirements/pip.txt
 docker==6.1.2
     # via -r requirements/pip.txt
 docutils==0.21.2

requirements/pip.in

@@ -46,6 +46,8 @@ slumber
 pyyaml
 Pygments
 
+dnspython
+
 # Used for Redis cache Django backend (`django.core.cache.backends.redis.RedisCache`)
 redis
 

requirements/pip.txt

@@ -142,6 +142,8 @@ djangorestframework-api-key==3.0.0
     # via -r requirements/pip.in
 djangorestframework-jsonp==1.0.2
     # via -r requirements/pip.in
+dnspython==2.7.0
+    # via -r requirements/pip.in
 docker==6.1.2
     # via -r requirements/pip.in
 docutils==0.21.2

requirements/testing.txt

@@ -187,6 +187,8 @@ djangorestframework-api-key==3.0.0
     # via -r requirements/pip.txt
 djangorestframework-jsonp==1.0.2
     # via -r requirements/pip.txt
+dnspython==2.7.0
+    # via -r requirements/pip.txt
 docker==6.1.2
     # via -r requirements/pip.txt
 docutils==0.21.2