implement atproto_handle() in Web, ActivityPub, ATProto

activitypub.py

@@ -77,8 +77,7 @@ class ActivityPub(User, Protocol):
         assert util.is_web(id), f'{id} is not a URL'
         domain = util.domain_from_link(id)
         assert domain, 'missing domain'
-        assert domain not in common.DOMAIN_BLOCKLIST, \
-            f'{id} is a blocked domain'
+        assert not self.is_blocklisted(domain), f'{id} is a blocked domain'
 
     def web_url(self):
         """Returns this user's web URL aka web_url, eg 'https://foo.com/'."""
@@ -90,7 +89,7 @@ class ActivityPub(User, Protocol):
         return self.ap_actor()
 
     def ap_address(self):
-        """Returns this user's ActivityPub address, eg '@foo.com@foo.com'."""
+        """Returns this user's ActivityPub address, eg '@user@foo.com'."""
         if self.obj and self.obj.as1:
             addr = as2.address(self.as2())
             if addr:
@@ -99,12 +98,17 @@ class ActivityPub(User, Protocol):
         return as2.address(self.key.id())
 
     def ap_actor(self, rest=None):
-        """Returns this user's ActivityPub/AS2 actor id URL.
+        """Returns this user's ActivityPub actor id URL.
 
-        Eg 'https://fed.brid.gy/foo.com'
+        Eg 'https://foo.com/@user'
         """
         return self.key.id()
 
+    def atproto_handle(self):
+        """Returns `[USERNAME].[INSTANCE].AP.brid.gy`."""
+        username, instance = self.ap_address().strip('@').split('@')
+        return f'{username}.{instance}.{self.ABBREV}{common.SUPERDOMAIN}'
+
     @classmethod
     def owns_id(cls, id):
         """Returns None if id is an http(s) URL, False otherwise.

atproto.py

@@ -50,14 +50,7 @@ class ATProto(User, Protocol):
     @ndb.ComputedProperty
     def readable_id(self):
         """Prefers handle, then DID."""
-        did_obj = ATProto.load(self.key.id(), remote=False)
-        if did_obj:
-            handle, _, _ = parse_at_uri(
-                util.get_first(did_obj.raw, 'alsoKnownAs', ''))
-            if handle:
-                return handle
-
-        return self.key.id()
+        return self.atproto_handle() or self.key.id()
 
     def _pre_put_hook(self):
         """Validate id, require did:plc or non-blocklisted did:web.
@@ -80,6 +73,15 @@ class ATProto(User, Protocol):
         assert not self.atproto_did, \
             f"{self.key} shouldn't have atproto_did {self.atproto_did}"
 
+    def atproto_handle(self):
+        """Returns handle if the DID document includes one, otherwise None."""
+        did_obj = ATProto.load(self.key.id(), remote=False)
+        if did_obj:
+            handle, _, _ = parse_at_uri(
+                util.get_first(did_obj.raw, 'alsoKnownAs', ''))
+            if handle:
+                return handle
+
     def web_url(self):
         return bluesky.Bluesky.user_url(self.readable_id)
 
@@ -169,7 +171,7 @@ class ATProto(User, Protocol):
                 return False
         else:
             # create new DID
-            # STATE: (unneeded?) new User.atproto_handle()
+            logger.info(f'Creating new did:plc for {user.key}')
             did_plc = did.create_plc(user.atproto_handle(), privkey=privkey,
                                      pds_hostname=request.host,
                                      post_fn=util.requests_post)
@@ -182,11 +184,12 @@ class ATProto(User, Protocol):
                 user.put()
             update()
 
+        logger.info(f'{user.key} is {user.atproto_did}')
         repo = storage.load_repo(did=user.atproto_did)
         writes = []
         if repo is None:
             # create repo
-            handle = user.readable_id if user.readable_id != user.atproto_did else None
+            handle = user.atproto_handle()
             repo = Repo.create(storage, user.atproto_did, privkey, handle=handle)
             if user.obj and user.obj.as1:
                 # create user profile

tests/test_activitypub.py

@@ -1863,7 +1863,10 @@ class ActivityPubUtilsTest(TestCase):
                                    ignore=['to'])
 
     def test_ap_address(self):
-        user = ActivityPub(obj=Object(id='a', as2={**ACTOR, 'preferredUsername': 'me'}))
+        user = ActivityPub(obj=Object(id='a', as2={
+            **ACTOR,
+            'preferredUsername': 'me',
+        }))
         self.assertEqual('@me@mas.to', user.ap_address())
         self.assertEqual('@me@mas.to', user.readable_id)
 
@@ -1879,6 +1882,13 @@ class ActivityPubUtilsTest(TestCase):
         user = self.make_user('http://foo/actor', cls=ActivityPub)
         self.assertEqual('http://foo/actor', user.ap_actor())
 
+    def test_atproto_handle(self):
+        user = self.make_user('http://a', cls=ActivityPub, obj_as2={
+            'id': 'https://mas.to/users/foo',
+            'preferredUsername': 'me',
+        })
+        self.assertEqual('me.mas.to.ap.brid.gy', user.atproto_handle())
+
     def test_web_url(self):
         user = self.make_user('http://foo/actor', cls=ActivityPub)
         self.assertEqual('http://foo/actor', user.web_url())

tests/test_atproto.py

@@ -164,11 +164,13 @@ class ATProtoTest(TestCase):
         self.assertEqual('https://bsky.app/profile/han.dull', user.web_url())
 
     @patch('requests.get', return_value=requests_response('', status=404))
-    def test_readable_id(self, mock_get):
+    def test_handle_and_readable_id(self, mock_get):
         user = self.make_user('did:plc:foo', cls=ATProto)
+        self.assertIsNone(user.atproto_handle())
         self.assertEqual('did:plc:foo', user.readable_id)
 
         self.store_object(id='did:plc:foo', raw=DID_DOC)
+        self.assertEqual('han.dull', user.atproto_handle())
         self.assertEqual('han.dull', user.readable_id)
 
     def test_ap_address(self):

tests/test_web.py

@@ -1799,6 +1799,9 @@ http://this/404s
         self.assertEqual('http://localhost/user.com', g.user.ap_actor())
         self.assertEqual('http://localhost/user.com/inbox', g.user.ap_actor('inbox'))
 
+    def test_atproto_handle(self, *_):
+        self.assertEqual('user.com.web.brid.gy', g.user.atproto_handle())
+
     def test_check_web_site(self, mock_get, _):
         redir = 'http://localhost/.well-known/webfinger?resource=acct:user.com@user.com'
         mock_get.side_effect = (

web.py

@@ -69,7 +69,7 @@ class Web(User, Protocol):
         id = self.key.id()
         assert re.match(common.DOMAIN_RE, id)
         assert id.lower() == id, f'upper case is not allowed in Web key id: {id}'
-        assert id not in common.DOMAIN_BLOCKLIST, f'{id} is a blocked domain'
+        assert not self.is_blocklisted(id), f'{id} is a blocked domain'
 
     @classmethod
     def get_or_create(cls, id, **kwargs):
@@ -103,6 +103,10 @@ class Web(User, Protocol):
             url += f'/{rest}'
         return url
 
+    def atproto_handle(self):
+        """Returns `[DOMAIN].web.brid.gy`."""
+        return f'{self.key.id()}.{self.ABBREV}{common.SUPERDOMAIN}'
+
     def user_page_path(self, rest=None):
         """Always use domain."""
         path = f'/{self.ABBREV}/{self.key.id()}'