Refs #613: Also block subdomains

activities/models/post.py

@@ -843,7 +843,7 @@ class Post(StatorModel):
                     else:
                         raise TryAgainLater()
                 # If the post is from a blocked domain, stop and drop
-                if author.domain.blocked:
+                if author.domain.recursively_blocked():
                     raise cls.DoesNotExist("Post is from a blocked domain")
                 try:
                     # try again, because fetch_actor() also fetches pinned posts

tests/users/models/test_domain.py

@@ -0,0 +1,23 @@
+import pytest
+
+from users.models import Domain
+
+
+@pytest.mark.django_db
+def test_recursive_block():
+    """
+    Tests that blocking a domain also blocks its subdomains
+    """
+
+    root_domain = Domain.get_remote_domain("evil.com")
+    root_domain.blocked = True
+    root_domain.save()
+
+    # Re-fetching the root should be blocked
+    assert Domain.get_remote_domain("evil.com").recursively_blocked()
+
+    # A sub domain should also be blocked
+    assert Domain.get_remote_domain("terfs.evil.com").recursively_blocked()
+
+    # An unrelated domain should not be blocked
+    assert not Domain.get_remote_domain("example.com").recursively_blocked()

users/models/domain.py

@@ -241,6 +241,24 @@ class Domain(StatorModel):
             return f"{name:.10} - {version:.10}"
         return None
 
+    def recursively_blocked(self) -> bool:
+        """
+        Checks for blocks on all right subsets of this domain, except the very
+        last part of the TLD.
+
+        Yes, I know this weirdly lets you block ".co.uk" or whatever, but
+        people can do that if they want I guess.
+        """
+        # Efficient short-circuit
+        if self.blocked:
+            return True
+        # Build domain list
+        domain_parts = [self.domain]
+        while "." in domain_parts[-1]:
+            domain_parts.append(domain_parts[-1].split(".", 1)[1])
+        # See if any of those are blocked
+        return Domain.objects.filter(domain__in=domain_parts, blocked=True).exists()
+
     ### Config ###
 
     @cached_property

users/views/activitypub.py

@@ -148,7 +148,7 @@ class Inbox(View):
             return HttpResponseBadRequest("Cannot retrieve actor")
 
         # See if it's from a blocked user or domain
-        if identity.blocked or identity.domain.blocked:
+        if identity.blocked or identity.domain.recursively_blocked():
             # I love to lie! Throw it away!
             exceptions.capture_message(
                 f"Inbox: Discarded message from {identity.actor_uri}"