move common.is_blacklisted to Protocol so that subclasses can override it

activitypub.py

@@ -24,7 +24,6 @@ from common import (
     CONTENT_TYPE_HTML,
     error,
     host_url,
-    is_blocklisted,
     redirect_unwrap,
     redirect_wrap,
     TLD_BLOCKLIST,
@@ -114,7 +113,7 @@ class ActivityPub(User, Protocol):
 
         https://www.w3.org/TR/activitypub/#obj-id
         """
-        if util.is_web(id) and not is_blocklisted(id):
+        if util.is_web(id) and not cls.is_blocklisted(id):
             return None
 
         return False
@@ -164,7 +163,7 @@ class ActivityPub(User, Protocol):
         If `obj.recipient_obj` is set, it's interpreted as the receiving actor
         who we're delivering to and its id is populated into `cc`.
         """
-        if is_blocklisted(url):
+        if cls.is_blocklisted(url):
             logger.info(f'Skipping sending to {url}')
             return False
 

atproto.py

@@ -28,7 +28,6 @@ import common
 from common import (
     add,
     error,
-    is_blocklisted,
     USER_AGENT,
 )
 from flask_app import app, cache
@@ -74,7 +73,7 @@ class ATProto(User, Protocol):
         elif id.startswith('did:web:'):
             domain = id.removeprefix('did:web:')
             assert (re.match(common.DOMAIN_RE, domain)
-                    and not is_blocklisted(domain)), domain
+                    and not Protocol.is_blocklisted(domain)), domain
         else:
             assert False, f'{id} is not valid did:plc or did:web'
 
@@ -126,6 +125,11 @@ class ATProto(User, Protocol):
 
         return common.host_url()
 
+    def is_blocklisted(url):
+        # don't block common.DOMAINS since we want ourselves, ie our own PDS, to
+        # be a valid domain to send to
+        return util.domain_or_parent_in(util.domain_from_link(url), DOMAIN_BLOCKLIST)
+
     @classmethod
     def send(cls, obj, url, log_data=True):
         """Creates a record if we own its repo.

common.py

@@ -50,14 +50,14 @@ LOCAL_DOMAINS = (
 )
 DOMAINS = (PRIMARY_DOMAIN,) + OTHER_DOMAINS + LOCAL_DOMAINS
 # TODO: unify with Bridgy's
-DOMAIN_BLOCKLIST = frozenset((
+DOMAIN_BLOCKLIST = (
     # https://github.com/snarfed/bridgy-fed/issues/348
     'aaronparecki.com',
     'facebook.com',
     'fb.com',
     't.co',
     'twitter.com',
-) + DOMAINS)
+)
 
 CACHE_TIME = timedelta(seconds=60)
 
@@ -129,17 +129,6 @@ def content_type(resp):
         return type.split(';')[0]
 
 
-def is_blocklisted(url):
-    """Returns True if the given URL is in our domain blocklist, False otherwise.
-
-    Args:
-      url: str
-
-    Returns: boolean
-    """
-    return util.domain_or_parent_in(util.domain_from_link(url), DOMAIN_BLOCKLIST)
-
-
 def redirect_wrap(url):
     """Returns a URL on our domain that redirects to this URL.
 

protocol.py

@@ -11,7 +11,7 @@ from granary import as1
 import werkzeug.exceptions
 
 import common
-from common import add, error, is_blocklisted
+from common import add, DOMAIN_BLOCKLIST, DOMAINS, error
 from models import Follower, Object, PROTOCOLS, Target, User
 from oauth_dropins.webutil import util
 from oauth_dropins.webutil.util import json_dumps, json_loads
@@ -344,6 +344,20 @@ class Protocol:
         """
         raise NotImplementedError()
 
+    @classmethod
+    def is_blocklisted(cls, url):
+        """Returns True if we block the given URL and shouldn't deliver to it.
+
+        Default implementation here, subclasses may override.
+
+        Args:
+          url: str
+
+        Returns: boolean
+        """
+        return util.domain_or_parent_in(util.domain_from_link(url),
+                                        DOMAIN_BLOCKLIST + DOMAINS)
+
     @classmethod
     def receive(cls, obj):
         """Handles an incoming activity.
@@ -758,7 +772,7 @@ class Protocol:
         """
         logger.info('Finding recipients and their targets')
         candidates = sorted(id for id in as1.targets(obj.as1)
-                            if not is_blocklisted(id))
+                            if not cls.is_blocklisted(id))
 
         orig_obj = None
         targets = {}
@@ -852,7 +866,7 @@ class Protocol:
             if util.is_web(url)
         ]
         for url in sorted(util.dedupe_urls(candidates.keys())):
-            if is_blocklisted(url):
+            if cls.is_blocklisted(url):
                 logger.info(f'Skipping blocklisted target {url}')
             elif util.is_web(url) and util.domain_from_link(url) in source_domains:
                 logger.info(f'Skipping same-domain target {url}')

web.py

@@ -281,7 +281,7 @@ class Web(User, Protocol):
         verb = obj.as1.get('verb')
         if (verb in ('accept', 'undo')
                 or url not in as1.targets(obj.as1)
-                or common.is_blocklisted(url)):
+                or cls.is_blocklisted(url)):
             logger.info(f'Skipping sending to {url}')
             return False