/r/ and Protocol.for_id: handle bad bsky.app URLs better

for https://console.cloud.google.com/errors/detail/CNSB0pOhgIfdbA;locations=global;time=P30D?project=bridgy-federated

Author: snarfed

atproto.py

@@ -241,7 +241,8 @@ def owns_handle(cls, handle, allow_internal=False):
 
     @classmethod
     def handle_to_id(cls, handle):
-        assert cls.owns_handle(handle) is not False
+        if not handle or cls.owns_handle(handle) is False:
+            return None
 
         # TODO: shortcut our own handles? eg snarfed.org.web.brid.gy
 

tests/test_atproto.py

@@ -154,6 +154,11 @@ def test_handle_to_id(self):
         self.make_user('did:plc:user', cls=ATProto)
         self.assertEqual('did:plc:user', ATProto.handle_to_id('ha.nl'))
 
+    def test_handle_to_id_bad(self):
+        for bad in None, '', '.bsky.social':
+            with self.subTest(bad=bad):
+                self.assertIsNone(ATProto.handle_to_id(bad))
+
     def test_handle_to_id_first_opted_out(self):
         self.store_object(id='did:plc:user', raw=DID_DOC)
         user = self.make_user('did:plc:user', cls=ATProto)

tests/test_redirect.py

@@ -50,6 +50,26 @@ def test_redirect_scheme_missing(self):
         got = self.client.get('/r/user.com')
         self.assertEqual(400, got.status_code)
 
+    def test_redirect_not_url(self):
+        got = self.client.get('/r/foo:bar:baz')
+        self.assertEqual(400, got.status_code)
+
+    def test_as2_not_web(self):
+        got = self.client.get('/r/foo:bar:baz',
+                              headers={'Accept': as2.CONTENT_TYPE_LD_PROFILE})
+        self.assertEqual(400, got.status_code)
+
+    def test_redirect_bsky_app_url(self):
+        got = self.client.get('/r/https://bsky.app/profile/.bsky.social')
+        self.assertEqual(301, got.status_code)
+        self.assertEqual('https://bsky.app/profile/.bsky.social',
+                         got.headers['Location'])
+
+    def test_as2_bsky_app_url(self):
+        got = self.client.get('/r/https://bsky.app/profile/.bsky.social',
+                              headers={'Accept': as2.CONTENT_TYPE_LD_PROFILE})
+        self.assertEqual(404, got.status_code)
+
     def test_redirect_url_missing(self):
         got = self.client.get('/r/')
         self.assertEqual(404, got.status_code)