extend Web/ActivityPub.owns_id() to return False if blocklisted

activitypub.py

@@ -24,6 +24,7 @@ from common import (
     CONTENT_TYPE_HTML,
     error,
     host_url,
+    is_blocklisted,
     NoMicroformats,
     redirect_unwrap,
     redirect_wrap,
@@ -114,7 +115,10 @@ class ActivityPub(User, Protocol):
 
         https://www.w3.org/TR/activitypub/#obj-id
         """
-        return None if util.is_web(id) else False
+        if util.is_web(id) and not is_blocklisted(id):
+            return None
+
+        return False
 
     @classmethod
     def target_for(cls, obj, shared=False):

common.py

@@ -130,16 +130,15 @@ def content_type(resp):
         return type.split(';')[0]
 
 
-def remove_blocklisted(urls):
-    """Returns the subset of input URLs that aren't in our domain blocklist.
+def is_blocklisted(url):
+    """Returns True if the given URL is in our domain blocklist, False otherwise.
 
     Args:
-      urls: sequence of str
+      url: str
 
-    Returns: list of str
+    Returns: boolean
     """
-    return [u for u in urls if not util.domain_or_parent_in(
-              util.domain_from_link(u), DOMAIN_BLOCKLIST)]
+    return util.domain_or_parent_in(util.domain_from_link(url), DOMAIN_BLOCKLIST)
 
 
 def redirect_wrap(url):

protocol.py

@@ -11,7 +11,7 @@ from granary import as1
 import werkzeug.exceptions
 
 import common
-from common import add, error
+from common import add, error, is_blocklisted
 from models import Follower, Object, PROTOCOLS, Target, User
 from oauth_dropins.webutil import util
 from oauth_dropins.webutil.util import json_dumps, json_loads
@@ -134,6 +134,8 @@ class Protocol:
         external HTTP fetches to fetch the id itself or otherwise perform
         discovery.
 
+        Returns False if the id's domain is in :attr:`common.DOMAIN_BLOCKLIST`.
+
         Args:
           id: str
 
@@ -727,7 +729,8 @@ class Protocol:
                 error(f'{verb} missing target URL')
             logger.info(f'original object ids from object: {orig_ids}')
 
-        orig_ids = sorted(common.remove_blocklisted(util.dedupe_urls(orig_ids)))
+        orig_ids = sorted(id for id in util.dedupe_urls(orig_ids)
+                          if not is_blocklisted(id))
         orig_obj = None
         targets = {}
         for id in orig_ids:

tests/test_activitypub.py

@@ -1452,6 +1452,9 @@ class ActivityPubUtilsTest(TestCase):
         self.assertFalse(ActivityPub.owns_id('at://did:plc:foo/bar/123'))
         self.assertFalse(ActivityPub.owns_id('e45fab982'))
 
+        self.assertFalse(ActivityPub.owns_id('https://twitter.com/foo'))
+        self.assertFalse(ActivityPub.owns_id('https://fed.brid.gy/foo'))
+
     def test_postprocess_as2_multiple_in_reply_tos(self):
         self.assert_equals({
             'id': 'http://localhost/r/xyz',

tests/test_web.py

@@ -1803,6 +1803,9 @@ class WebProtocolTest(TestCase):
         g.user.key.delete()
         self.assertIsNone(Web.owns_id('user.com'))
 
+        self.assertFalse(Web.owns_id('https://twitter.com/foo'))
+        self.assertFalse(Web.owns_id('https://fed.brid.gy/foo'))
+
     def test_fetch(self, mock_get, __):
         mock_get.return_value = REPOST