handle protocol bot users in webfinger, ids.translate_handle, Web.own…

…s_handle

for #880

atproto.py

@@ -237,7 +237,7 @@ def pds_for(cls, obj):
 
         return None
 
-    def is_blocklisted(url):
+    def is_blocklisted(url, allow_internal=False):
         # 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)

ids.py

@@ -11,7 +11,13 @@
 from granary.bluesky import BSKY_APP_URL_RE, web_url_to_at_uri
 from oauth_dropins.webutil import util
 
-from common import subdomain_wrap, LOCAL_DOMAINS, PRIMARY_DOMAIN, SUPERDOMAIN
+from common import (
+    LOCAL_DOMAINS,
+    PRIMARY_DOMAIN,
+    PROTOCOL_DOMAINS,
+    subdomain_wrap,
+    SUPERDOMAIN,
+)
 import models
 
 logger = logging.getLogger(__name__)
@@ -153,7 +159,9 @@ def translate_handle(*, handle, from_, to, enhanced):
 
     match from_.LABEL, to.LABEL:
         case _, 'activitypub':
-            domain = handle if enhanced else f'{from_.ABBREV}{SUPERDOMAIN}'
+            domain = f'{from_.ABBREV}{SUPERDOMAIN}'
+            if enhanced or handle == PRIMARY_DOMAIN or handle in PROTOCOL_DOMAINS:
+                domain = handle
             return f'@{handle}@{domain}'
 
         case _, 'atproto' | 'nostr':

tests/test_activitypub.py

@@ -490,13 +490,33 @@ def test_actor_opted_out(self, *_):
         got = self.client.get('/user.com')
         self.assertEqual(404, got.status_code)
 
+    # skip _pre_put_hook since it doesn't allow internal domains
+    @patch.object(Web, '_pre_put_hook', new=lambda self: None)
+    def test_actor_protocol_bot_user(self, *_):
+        """Web users are special cased to drop the /web/ prefix."""
+        actor_as2 = json_loads(util.read('bsky.brid.gy.as2.json'))
+        self.make_user('bsky.brid.gy', cls=Web, obj_as2=actor_as2,
+                       obj_id='https://bsky.brid.gy/')
+
+        got = self.client.get('/bsky.brid.gy')
+        self.assertEqual(200, got.status_code)
+        self.assertEqual(as2.CONTENT_TYPE_LD_PROFILE, got.headers['Content-Type'])
+        self.assert_equals({
+            **actor_as2,
+            'id': 'http://localhost/bsky.brid.gy',
+        }, got.json, ignore=['inbox', 'outbox', 'endpoints', 'followers',
+                             'following', 'publicKey', 'publicKeyPem'])
+
+    # skip _pre_put_hook since it doesn't allow internal domains
+    @patch.object(Web, '_pre_put_hook', new=lambda self: None)
     def test_instance_actor_fetch(self, *_):
         def reset_instance_actor():
             activitypub._INSTANCE_ACTOR = testutil.global_user
         self.addCleanup(reset_instance_actor)
 
         actor_as2 = json_loads(util.read('fed.brid.gy.as2.json'))
-        self.make_user(common.PRIMARY_DOMAIN, cls=Web, obj_as2=actor_as2)
+        self.make_user(common.PRIMARY_DOMAIN, cls=Web, obj_as2=actor_as2,
+                       obj_id='https://fed.brid.gy/')
 
         activitypub._INSTANCE_ACTOR = None
         got = self.client.get(f'/{common.PRIMARY_DOMAIN}')
@@ -2412,5 +2432,3 @@ def test_send_convert_ids(self, mock_post):
             'actor': 'https://fa.brid.gy/ap/fake:user',
             'to': [as2.PUBLIC_AUDIENCE],
         }, json_loads(kwargs['data']))
-
-    # TODO: actor fetch and webfinger for @bsky.brid.gy@bsky.brid.gy both don't work. test and fix those.

tests/test_ids.py

@@ -110,6 +110,10 @@ def test_translate_handle(self):
             (Web, 'user.com', Fake, 'fake:handle:user.com'),
             (Web, 'user.com', Web, 'user.com'),
 
+            # instance actor, protocol bot user
+            (Web, 'fed.brid.gy', ActivityPub, '@fed.brid.gy@fed.brid.gy'),
+            (Web, 'bsky.brid.gy', ActivityPub, '@bsky.brid.gy@bsky.brid.gy'),
+
             (ActivityPub, '@user@instance', ActivityPub, '@user@instance'),
             (ActivityPub, '@user@instance', ATProto, 'user.instance.ap.brid.gy'),
             (ActivityPub, '@user@instance', Fake, 'fake:handle:@user@instance'),
@@ -137,6 +141,10 @@ def test_translate_handle_enhanced(self):
             (ActivityPub, '@user@user', Web, 'https://user'),
             (ActivityPub, '@user@instance', Fake, 'fake:handle:@user@instance'),
             (ATProto, 'user.com', ActivityPub, '@user.com@user.com'),
+
+            # instance actor, protocol bot user
+            (Web, 'fed.brid.gy', ActivityPub, '@fed.brid.gy@fed.brid.gy'),
+            (Web, 'bsky.brid.gy', ActivityPub, '@bsky.brid.gy@bsky.brid.gy'),
         ]:
             with self.subTest(from_=from_.LABEL, to=to.LABEL):
                 self.assertEqual(expected, translate_handle(

tests/test_web.py

@@ -2373,6 +2373,13 @@ def test_handle_as(self, *_):
         self.user.ap_subdomain = 'fed'
         self.assertEqual('@user.com@fed.brid.gy', self.user.handle_as(ActivityPub))
 
+    def test_handle_as_bot_users(self, *_):
+        fed = Web(id='fed.brid.gy', ap_subdomain='fed')
+        self.assertEqual('@fed.brid.gy@fed.brid.gy', fed.handle_as(ActivityPub))
+
+        bsky = Web(id='bsky.brid.gy', ap_subdomain='bsky')
+        self.assertEqual('@bsky.brid.gy@bsky.brid.gy', bsky.handle_as(ActivityPub))
+
     def test_id_as(self, *_):
         self.assertEqual('http://localhost/user.com', self.user.id_as(ActivityPub))
 
@@ -2552,6 +2559,8 @@ def test_owns_handle(self, *_):
         self.assertEqual(False, Web.owns_handle('@foo@bar.com'))
         self.assertEqual(False, Web.owns_handle('foo@bar.com'))
         self.assertEqual(False, Web.owns_handle('localhost'))
+
+        self.assertEqual(True, Web.owns_handle('fed.brid.gy'))
         self.assertEqual(True, Web.owns_handle('bsky.brid.gy'))
 
     def test_handle_to_id(self, *_):

tests/test_webfinger.py

@@ -4,6 +4,7 @@
 import urllib.parse
 
 from granary.as2 import CONTENT_TYPE_LD_PROFILE
+from oauth_dropins.webutil import util
 from oauth_dropins.webutil.testutil import requests_response
 
 # import first so that Fake is defined before URL routes are registered
@@ -341,11 +342,29 @@ def test_create_user(self, mock_get):
         user = Web.get_by_id('user.com')
         assert not user.direct
 
-    def test_fed_brid_gy(self):
+    # skip _pre_put_hook since it doesn't allow internal domains
+    @patch.object(Web, '_pre_put_hook', new=lambda self: None)
+    def test_protocol_bot_user(self):
+        self.make_user('bsky.brid.gy', cls=Web, obj_id='https://bsky.brid.gy/',
+                       ap_subdomain='bsky')
+
+        for id in ('acct:bsky.brid.gy@bsky.brid.gy',
+                   'https://bsky.brid.gy/bsky.brid.gy'):
+            got = self.client.get(f'/.well-known/webfinger?resource={id}')
+            self.assertEqual(200, got.status_code, got.get_data(as_text=True))
+            self.assertEqual('acct:bsky.brid.gy@bsky.brid.gy', got.json['subject'])
+            self.assertEqual(['https://bsky.brid.gy/'], got.json['aliases'])
+            self.assertIn({
+                'href': 'http://localhost/bsky.brid.gy',
+                'rel': 'self',
+                'type': 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"',
+            }, got.json['links'])
+
+    def test_internal_domain_error(self):
         got = self.client.get('/.well-known/webfinger?resource=http://localhost/')
         self.assertEqual(400, got.status_code, got.get_data(as_text=True))
 
-        got = self.client.get('/.well-known/webfinger?resource=acct%3A%40localhost')
+        got = self.client.get('/.well-known/webfinger?resource=acct:@localhost')
         self.assertEqual(400, got.status_code, got.get_data(as_text=True))
 
     @patch('requests.get', return_value=requests_response(

tests/testutil.py

@@ -116,7 +116,7 @@ def handle_to_id(cls, handle):
         return handle.replace(f'{cls.LABEL}:handle:', f'{cls.LABEL}:')
 
     @classmethod
-    def is_blocklisted(cls, url):
+    def is_blocklisted(cls, url, allow_internal=False):
         return url.startswith(f'{cls.LABEL}:blocklisted')
 
     @classmethod

web.py

@@ -359,7 +359,7 @@ def owns_id(cls, id):
 
     @classmethod
     def owns_handle(cls, handle):
-        if handle in PROTOCOL_DOMAINS:
+        if handle == PRIMARY_DOMAIN or handle in PROTOCOL_DOMAINS:
             return True
         elif not is_valid_domain(handle, allow_internal=False):
             return False

webfinger.py

@@ -14,9 +14,10 @@
 
 import activitypub
 import common
-from common import LOCAL_DOMAINS, SUPERDOMAIN
+from common import LOCAL_DOMAINS, PRIMARY_DOMAIN, PROTOCOL_DOMAINS, SUPERDOMAIN
 from flask_app import app, cache
 from protocol import Protocol
+from web import Web
 
 SUBSCRIBE_LINK_REL = 'http://ostatus.org/schema/1.0/subscribe'
 
@@ -58,7 +59,9 @@ def template_vars(self):
         except ValueError:
             id = urlparse(resource).netloc or resource
 
-        if not cls:
+        if id == PRIMARY_DOMAIN or id in PROTOCOL_DOMAINS:
+            cls = Web
+        elif not cls:
             cls = Protocol.for_request(fed='web')
 
         if not cls: