feat(api): Check Public Suffix List when creating domain, fixes #88

.env.default

@@ -18,6 +18,7 @@ DESECSTACK_API_EMAIL_HOST_USER=
 DESECSTACK_API_EMAIL_HOST_PASSWORD=
 DESECSTACK_API_EMAIL_PORT=
 DESECSTACK_API_SECRETKEY=
+DESECSTACK_API_PSL_RESOLVER=
 DESECSTACK_DBAPI_PASSWORD_desec=
 DESECSTACK_NORECAPTCHA_SITE_KEY=
 DESECSTACK_NORECAPTCHA_SECRET_KEY=

.travis.yml

@@ -15,6 +15,7 @@ env:
    - DESECSTACK_API_EMAIL_HOST_PASSWORD=password
    - DESECSTACK_API_EMAIL_PORT=25
    - DESECSTACK_API_SECRETKEY=9Fn33T5yGuds
+   - DESECSTACK_API_PSL_RESOLVER=8.8.8.8
    - DESECSTACK_DBAPI_PASSWORD_desec=9Fn33T5yGueeee
    - DESECSTACK_DB_PASSWORD_pdnslord=9Fn33T5yGulkjlskdf
    - DESECSTACK_DB_PASSWORD_pdnsmaster=9Fn33T5yGukjwelt

api/api/settings.py

@@ -141,6 +141,7 @@ AUTH_USER_MODEL = 'desecapi.User'
 DEFAULT_NS = ['ns1.desec.io.', 'ns2.desec.io.']
 
 # Public Suffix settings
+PSL_RESOLVER = os.environ.get('DESECSTACK_API_PSL_RESOLVER')
 LOCAL_PUBLIC_SUFFIXES = {'dedyn.io'}
 
 # PowerDNS API access

api/desecapi/exception_handlers.py

@@ -1,8 +1,10 @@
+import logging
+
 from django.db.utils import OperationalError
 from rest_framework import status
 from rest_framework.response import Response
 from rest_framework.views import exception_handler as drf_exception_handler
-import logging
+from psl_dns.exceptions import UnsupportedRule
 
 
 def exception_handler(exc, context):
@@ -39,4 +41,7 @@ def exception_handler(exc, context):
     if isinstance(exc, OSError):
         return _perform_handling('OSError')
 
+    if isinstance(exc, UnsupportedRule):
+        return _perform_handling('UnsupportedRule')
+
     return drf_exception_handler(exc, context)

api/desecapi/tests/base.py

@@ -1,9 +1,11 @@
 import base64
-from functools import reduce
+from contextlib import nullcontext
+from functools import partial, reduce
 import operator
 import random
 import re
 import string
+from unittest import mock
 
 from django.utils import timezone
 from httpretty import httpretty, core as hr_core
@@ -13,6 +15,7 @@ from rest_framework.utils import json
 
 from api import settings
 from desecapi.models import User, Domain, Token, RRset, RR
+from desecapi.views import DomainList
 
 
 class DesecAPIClient(APIClient):
@@ -525,8 +528,9 @@ class DesecTestCase(MockPDNSTestCase):
     """
     client_class = DesecAPIClient
 
-    AUTO_DELEGATION_DOMAINS = list(settings.LOCAL_PUBLIC_SUFFIXES)
-    PUBLIC_SUFFIXES = ['de', 'com', 'io', 'gov.cd', 'edu.ec', 'xxx', 'pinb.gov.pl', 'valer.ostfold.no', 'kota.aichi.jp']
+    AUTO_DELEGATION_DOMAINS = settings.LOCAL_PUBLIC_SUFFIXES
+    PUBLIC_SUFFIXES = {'de', 'com', 'io', 'gov.cd', 'edu.ec', 'xxx', 'pinb.gov.pl', 'valer.ostfold.no',
+                       'kota.aichi.jp', 's3.amazonaws.com', 'wildcard.ck'}
 
     @classmethod
     def reverse(cls, view_name, **kwargs):
@@ -564,13 +568,15 @@ class DesecTestCase(MockPDNSTestCase):
 
     @classmethod
     def random_username(cls, host=None):
-        host = host or cls.random_domain_name(suffix=random.choice(cls.PUBLIC_SUFFIXES))
+        host = host or cls.random_domain_name(cls.PUBLIC_SUFFIXES)
         return cls.random_string() + '+test@' + host.lower()
 
     @classmethod
     def random_domain_name(cls, suffix=None):
         if not suffix:
-            suffix = random.choice(cls.PUBLIC_SUFFIXES)
+            suffix = cls.PUBLIC_SUFFIXES
+        if isinstance(suffix, set):
+            suffix = random.sample(suffix, 1)[0]
         return (random.choice(string.ascii_letters) + cls.random_string() + '--test' + '.' + suffix).lower()
 
     @classmethod
@@ -591,7 +597,7 @@ class DesecTestCase(MockPDNSTestCase):
     @classmethod
     def create_domain(cls, suffix=None, **kwargs):
         kwargs.setdefault('owner', cls.create_user())
-        kwargs.setdefault('name', cls.random_domain_name(suffix=suffix))
+        kwargs.setdefault('name', cls.random_domain_name(suffix))
         domain = Domain(**kwargs)
         domain.save()
         return domain
@@ -679,6 +685,37 @@ class DomainOwnerTestCase(DesecTestCase):
     other_domain = None
     token = None
 
+    def _mock_get_public_suffix(self, domain_name, public_suffixes=None):
+        if public_suffixes is None:
+            public_suffixes = settings.LOCAL_PUBLIC_SUFFIXES | self.PUBLIC_SUFFIXES
+        # Poor man's PSL interpreter. First, find all known suffixes covering the domain.
+        suffixes = [suffix for suffix in public_suffixes
+                    if '.{}'.format(domain_name).endswith('.{}'.format(suffix))]
+        # Also, consider TLD.
+        suffixes += [domain_name.rsplit('.')[-1]]
+        # Select the candidate with the most labels.
+        return max(suffixes, key=lambda suffix: suffix.count('.'))
+
+    @staticmethod
+    def _mock_is_public_suffix(name):
+        return name == DomainList.psl.get_public_suffix(name)
+
+    def get_psl_context_manager(self, side_effect_parameter):
+        if side_effect_parameter is None:
+            return nullcontext()
+
+        if callable(side_effect_parameter):
+            side_effect = side_effect_parameter
+        else:
+            side_effect = partial(self._mock_get_public_suffix, public_suffixes=[side_effect_parameter])
+
+        return mock.patch.object(DomainList.psl, 'get_public_suffix', side_effect=side_effect)
+
+    def setUpMockPatch(self):
+        mock.patch.object(DomainList.psl, 'get_public_suffix', side_effect=self._mock_get_public_suffix).start()
+        mock.patch.object(DomainList.psl, 'is_public_suffix', side_effect=self._mock_is_public_suffix).start()
+        self.addCleanup(mock.patch.stopall)
+
     @classmethod
     def setUpTestDataWithPdns(cls):
         super().setUpTestDataWithPdns()
@@ -686,11 +723,11 @@ class DomainOwnerTestCase(DesecTestCase):
         cls.owner = cls.create_user(dyn=cls.DYN)
 
         cls.my_domains = [
-            cls.create_domain(suffix=random.choice(cls.AUTO_DELEGATION_DOMAINS) if cls.DYN else None, owner=cls.owner)
+            cls.create_domain(suffix=cls.AUTO_DELEGATION_DOMAINS if cls.DYN else None, owner=cls.owner)
             for _ in range(cls.NUM_OWNED_DOMAINS)
         ]
         cls.other_domains = [
-            cls.create_domain(suffix=random.choice(cls.AUTO_DELEGATION_DOMAINS) if cls.DYN else None)
+            cls.create_domain(suffix=cls.AUTO_DELEGATION_DOMAINS if cls.DYN else None)
             for _ in range(cls.NUM_OTHER_DOMAINS)
         ]
 
@@ -705,6 +742,7 @@ class DomainOwnerTestCase(DesecTestCase):
     def setUp(self):
         super().setUp()
         self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token.key)
+        self.setUpMockPatch()
 
 
 class LockedDomainOwnerTestCase(DomainOwnerTestCase):

api/desecapi/tests/test_domains.py

@@ -1,9 +1,9 @@
 import json
-import random
 
 from django.conf import settings
 from django.core import mail
 from django.core.exceptions import ValidationError
+from psl_dns.exceptions import UnsupportedRule
 from rest_framework import status
 
 from desecapi.exceptions import PdnsException
@@ -138,6 +138,40 @@ class DomainOwnerTestCase1(DomainOwnerTestCase):
             response = self.client.post(url, {'name': name})
             self.assertStatus(response, status.HTTP_409_CONFLICT)
 
+    def test_create_public_suffixes(self):
+        for name in self.PUBLIC_SUFFIXES:
+            response = self.client.post(self.reverse('v1:domain-list'), {'name': name})
+            self.assertStatus(response, status.HTTP_409_CONFLICT)
+            self.assertEqual(response.data['code'], 'domain-unavailable')
+
+    def test_create_domain_under_public_suffix_with_private_parent(self):
+        name = 'amazonaws.com'
+        with self.assertPdnsRequests(self.requests_desec_domain_creation(name)[:-1]):
+            Domain(owner=self.create_user(), name=name).save()
+            self.assertTrue(Domain.objects.filter(name=name).exists())
+
+        # If amazonaws.com is owned by another user, we cannot register test.s4.amazonaws.com
+        name = 'test.s4.amazonaws.com'
+        response = self.client.post(self.reverse('v1:domain-list'), {'name': name})
+        self.assertStatus(response, status.HTTP_409_CONFLICT)
+        self.assertEqual(response.data['code'], 'domain-unavailable')
+
+        # s3.amazonaws.com is a public suffix. Therefore, test.s3.amazonaws.com can be
+        # registered even if the parent zone amazonaws.com is owned by another user
+        name = 'test.s3.amazonaws.com'
+        psl_cm = self.get_psl_context_manager('s3.amazonaws.com')
+        with psl_cm, self.assertPdnsRequests(self.requests_desec_domain_creation(name)):
+            response = self.client.post(self.reverse('v1:domain-list'), {'name': name})
+            self.assertStatus(response, status.HTTP_201_CREATED)
+
+    def test_create_domain_under_unsupported_public_suffix_rule(self):
+        # Show lenience if the PSL library produces an UnsupportedRule exception
+        name = 'unsupported.wildcard.test'
+        psl_cm = self.get_psl_context_manager(UnsupportedRule)
+        with psl_cm, self.assertPdnsRequests():
+            response = self.client.post(self.reverse('v1:domain-list'), {'name': name})
+            self.assertStatus(response, status.HTTP_503_SERVICE_UNAVAILABLE)
+
     def test_create_domain_policy(self):
         name = '*.' + self.random_domain_name()
         response = self.client.post(self.reverse('v1:domain-list'), {'name': name})
@@ -283,13 +317,13 @@ class AutoDelegationDomainOwnerTests(DomainOwnerTestCase):
         user_quota = settings.LIMIT_USER_DOMAIN_COUNT_DEFAULT - self.NUM_OWNED_DOMAINS
 
         for i in range(user_quota):
-            name = self.random_domain_name(random.choice(self.AUTO_DELEGATION_DOMAINS))
+            name = self.random_domain_name(self.AUTO_DELEGATION_DOMAINS)
             with self.assertPdnsRequests(self.requests_desec_domain_creation_auto_delegation(name)):
                 response = self.client.post(url, {'name': name})
                 self.assertStatus(response, status.HTTP_201_CREATED)
                 self.assertEqual(len(mail.outbox), i + 1)
 
-        response = self.client.post(url, {'name': self.random_domain_name(random.choice(self.AUTO_DELEGATION_DOMAINS))})
+        response = self.client.post(url, {'name': self.random_domain_name(self.AUTO_DELEGATION_DOMAINS)})
         self.assertStatus(response, status.HTTP_403_FORBIDDEN)
         self.assertEqual(len(mail.outbox), user_quota)
 
@@ -298,7 +332,7 @@ class LockedAutoDelegationDomainOwnerTests(LockedDomainOwnerTestCase):
     DYN = True
 
     def test_unlock_user(self):
-        name = self.random_domain_name(self.AUTO_DELEGATION_DOMAINS[0])
+        name = self.random_domain_name(self.AUTO_DELEGATION_DOMAINS)
 
         # Users should be able to create domains under auto delegated domains even when locked
         response = self.client.post(self.reverse('v1:domain-list'), {'name': name})

api/desecapi/views.py

@@ -7,6 +7,7 @@ from datetime import timedelta
 
 import django.core.exceptions
 import djoser.views
+import psl_dns
 from django.contrib.auth import user_logged_in, user_logged_out
 from django.core.mail import EmailMessage
 from django.db import IntegrityError
@@ -43,7 +44,7 @@ from desecapi.renderers import PlainTextRenderer
 from desecapi.serializers import DomainSerializer, RRsetSerializer, DonationSerializer, TokenSerializer
 
 patternDyn = re.compile(r'^[A-Za-z-][A-Za-z0-9_-]*\.dedyn\.io$')
-patternNonDyn = re.compile(r'^([A-Za-z0-9-][A-Za-z0-9_-]*\.)+[A-Za-z]+$')
+patternNonDyn = re.compile(r'^([A-Za-z0-9-][A-Za-z0-9_-]*\.)*[A-Za-z]+$')
 
 
 class TokenCreateView(djoser.views.TokenCreateView):
@@ -96,13 +97,16 @@ class TokenViewSet(mixins.CreateModelMixin,
 class DomainList(generics.ListCreateAPIView):
     serializer_class = DomainSerializer
     permission_classes = (IsAuthenticated, IsOwner, IsUnlockedOrDyn,)
+    psl = psl_dns.PSL(resolver=settings.PSL_RESOLVER)
 
     def get_queryset(self):
         return Domain.objects.filter(owner=self.request.user.pk)
 
     def perform_create(self, serializer):
+        domain_name = serializer.validated_data['name']
+
         pattern = patternDyn if self.request.user.dyn else patternNonDyn
-        if pattern.match(serializer.validated_data['name']) is None:
+        if pattern.match(domain_name) is None:
             ex = ValidationError(detail={
                 "detail": "This domain name is not well-formed, by policy.",
                 "code": "domain-illformed"}
@@ -110,16 +114,32 @@ class DomainList(generics.ListCreateAPIView):
             ex.status_code = status.HTTP_409_CONFLICT
             raise ex
 
-        # Generate a list containing this and all higher-level domain names
-        domain_name = serializer.validated_data['name']
-        domain_parts = domain_name.split('.')
-        domain_list = {'.'.join(domain_parts[i:]) for i in range(1, len(domain_parts))}
+        # Check if domain is a public suffix
+        try:
+            public_suffix = self.psl.get_public_suffix(domain_name)
+            is_public_suffix = self.psl.is_public_suffix(domain_name)
+        except psl_dns.exceptions.UnsupportedRule as e:
+            # It would probably be fine to just create the domain (with the TLD acting as the
+            # public suffix and setting both public_suffix and is_public_suffix accordingly).
+            # However, in order to allow to investigate the situation, it's better not catch
+            # this exception. Our error handler turns it into a 503 error and makes sure
+            # admins are notified.
+            raise e
+
+        is_restricted_suffix = is_public_suffix and domain_name not in settings.LOCAL_PUBLIC_SUFFIXES
 
-        # Remove public suffixes and then use this list to control registration
-        domain_list -= settings.LOCAL_PUBLIC_SUFFIXES
+        # Generate a list of all domains connecting this one and its public suffix.
+        # If another user owns a zone with one of these names, then the requested
+        # domain is unavailable because it is part of the other user's zone.
+        private_components = domain_name.rsplit(public_suffix, 1)[0].rstrip('.')
+        private_components = private_components.split('.') if private_components else []
+        private_components += [public_suffix]
+        private_domains = ['.'.join(private_components[i:]) for i in range(0, len(private_components) - 1)]
+        assert is_public_suffix or domain_name == private_domains[0]
 
-        queryset = Domain.objects.filter(Q(name=domain_name) | (Q(name__in=domain_list) & ~Q(owner=self.request.user)))
-        if queryset.exists():
+        # Deny registration for non-local public suffixes and for domains covered by other users' zones
+        queryset = Domain.objects.filter(Q(name__in=private_domains) & ~Q(owner=self.request.user))
+        if is_restricted_suffix or queryset.exists():
             ex = ValidationError(detail={"detail": "This domain name is unavailable.", "code": "domain-unavailable"})
             ex.status_code = status.HTTP_409_CONFLICT
             raise ex

api/requirements.txt

@@ -9,3 +9,4 @@ uwsgi~=2.0.0
 django-nocaptcha-recaptcha==0.0.20  # updated manually
 djangorestframework-bulk~=0.2.0
 coverage~=4.5.3
+psl-dns~=1.0rc2

docker-compose.yml

@@ -110,6 +110,7 @@ services:
     - DESECSTACK_API_EMAIL_HOST_PASSWORD
     - DESECSTACK_API_EMAIL_PORT
     - DESECSTACK_API_SECRETKEY
+    - DESECSTACK_API_PSL_RESOLVER
     - DESECSTACK_DBAPI_PASSWORD_desec
     - DESECSTACK_IPV4_REAR_PREFIX16
     - DESECSTACK_IPV6_SUBNET

docs/domains.rst

@@ -103,9 +103,20 @@ and with ``409 Conflict`` otherwise.  This can happen, for example, if there
 already is a domain with the same name or if the domain name is considered
 invalid for policy reasons.
 
-Restrictions on what is a valid domain name apply on a per-user basis.  The
-response body *may* provide further, human-readable information on the policy
-violation that occurred.
+The response body *may* provide further, human-readable information on the
+policy violation that occurred.
+
+Restrictions on what is a valid domain name apply.  In particular, domains
+listed on the `Public Suffix List`_ cannot be registered.  (If you operate a
+public suffix and would like to host it with deSEC, that's certainly possible;
+please contact our support.)
+
+.. _Public Suffix List: https://publicsuffix.org/
+
+Furthermore, restrictions on a per-user basis may apply.  In particular, the
+number of domains a user can create is limited.  If you find yourself affected
+by this limit although you have a legitimate use case, please contact our
+support.
 
 
 Listing Domains