Add tests

synapse/appservice/__init__.py

@@ -84,7 +84,7 @@ class ApplicationService:
     NS_USERS = "users"
     NS_ALIASES = "aliases"
     NS_ROOMS = "rooms"
-    NS_PREVIEW_URLS = "uk_half-shot_msc4417_preview_urls"
+    NS_PREVIEW_URLS = "uk.half-shot.msc4417.preview_urls"
     # The ordering here is important as it is used to map database values (which
     # are stored as ints representing the position in this list) to namespace
     # values.
@@ -370,18 +370,21 @@ class ApplicationService:
     def is_room_id_in_namespace(self, room_id: str) -> bool:
         return bool(self._matches_regex(ApplicationService.NS_ROOMS, room_id))
 
+    def is_preview_url_in_namespace(self, url: str) -> bool:
+        return bool(self._matches_regex(ApplicationService.NS_PREVIEW_URLS, url))
+
     def is_exclusive_user(self, user_id: str) -> bool:
         return (
             self._is_exclusive(ApplicationService.NS_USERS, user_id)
             or user_id == self.sender.to_string()
         )
 
+    def is_exclusive_preview_url(self, url: str) -> bool:
+        return self._is_exclusive(ApplicationService.NS_PREVIEW_URLS, url)
+
     def is_interested_in_protocol(self, protocol: str) -> bool:
         return protocol in self.protocols
 
-    def is_interested_in_preview_url(self, url: str) -> Namespace | None:
-        return self._matches_regex(ApplicationService.NS_PREVIEW_URLS, url)
-
     def is_exclusive_alias(self, alias: str) -> bool:
         return self._is_exclusive(ApplicationService.NS_ALIASES, alias)
 

synapse/handlers/appservice.py

@@ -883,15 +883,18 @@ class ApplicationServicesHandler:
     def _get_exclusive_service_for_preview_url(
         self, url: str
     ) -> ApplicationService | None:
-        for service in self.store.get_app_services():
-            ns = service.is_interested_in_preview_url(url)
-            if ns is not None and ns.exclusive:
-                return service
-        return None
+        return next(
+            (
+                service
+                for service in self.store.get_app_services()
+                if service.is_exclusive_preview_url(url)
+            ),
+            None,
+        )
 
     def _get_services_for_preview_url(self, url: str) -> list[ApplicationService]:
         services = self.store.get_app_services()
-        return [s for s in services if s.is_interested_in_preview_url(url)]
+        return [s for s in services if s.is_preview_url_in_namespace(url)]
 
     async def _is_unknown_user(self, user_id: str) -> bool:
         if not self.is_mine_id(user_id):

synapse/media/url_previewer.py

@@ -229,7 +229,7 @@ class UrlPreviewer:
             if as_og_data.result:
                 # This data is never cached in the homeserver, but the appservice
                 # is expected to cache the data.
-                return json_encoder.encode(as_og_data).encode("utf8")
+                return json_encoder.encode(as_og_data.result).encode("utf8")
 
             if as_og_data.exclusive:
                 # This is NOT the same as the Bad Gateway error we usually return for

tests/appservice/test_api.py

@@ -251,3 +251,46 @@ class ApplicationServiceApiTestCase(unittest.HomeserverTestCase):
 
         self.assertEqual(claimed_keys, RESPONSE)
         self.assertEqual(missing, MISSING_KEYS)
+
+    def test_query_preview_url(self) -> None:
+        """
+        Tests that 3pe queries to the appservice are authenticated
+        with the appservice's token.
+        """
+
+        SUCCESS_RESULT = {"og:title": "The matrix.org site!"}
+
+        URL_ENDPOINT = f"{URL}/_matrix/app/unstable/uk.half-shot.msc4417/preview_url"
+
+        QUERY_URL = "https://matrix.org"
+        QUERY_USER = UserID.from_string("@a:user")
+
+        self.request_url = None
+
+        async def get_json(
+            url: str,
+            args: Mapping[Any, Any],
+            headers: Mapping[str | bytes, Sequence[str | bytes]] | None = None,
+        ) -> JsonDict:
+            if not headers or not headers.get(b"Authorization"):
+                raise RuntimeError("Access token should be provided in auth headers.")
+            self.assertEqual(args.get("url"), QUERY_URL)
+            self.assertEqual(args.get("user_id"), QUERY_USER.to_string())
+            self.assertEqual(
+                headers.get(b"Authorization"), [f"Bearer {TOKEN}".encode()]
+            )
+            self.request_url = url
+            if url == URL_ENDPOINT:
+                return SUCCESS_RESULT
+            else:
+                raise RuntimeError(
+                    "URL provided was invalid. This should never be seen."
+                )
+
+        # We assign to a method, which mypy doesn't like.
+        self.api.get_json = Mock(side_effect=get_json)  # type: ignore[method-assign]
+
+        result = self.get_success(
+            self.api.query_preview_url(self.service, QUERY_URL, QUERY_USER)
+        )
+        self.assertEqual(result, SUCCESS_RESULT)

tests/appservice/test_appservice.py

@@ -182,6 +182,32 @@ class ApplicationServiceTestCase(unittest.TestCase):
         )
         self.assertTrue(self.service.is_exclusive_room("!irc_foobar:matrix.org"))
 
+    def test_non_exclusive_url(self) -> None:
+        self.service.namespaces[ApplicationService.NS_PREVIEW_URLS].append(
+            _regex("https:\\/\\/matrix.org.*", exclusive=False)
+        )
+        self.assertFalse(self.service.is_exclusive_preview_url("https://matrix.org"))
+        self.assertFalse(
+            self.service.is_exclusive_preview_url("https://matrix.org/foobar")
+        )
+        self.assertTrue(self.service.is_preview_url_in_namespace("https://matrix.org"))
+        self.assertTrue(
+            self.service.is_preview_url_in_namespace("https://matrix.org/foobar")
+        )
+
+    def test_exclusive_url(self) -> None:
+        self.service.namespaces[ApplicationService.NS_PREVIEW_URLS].append(
+            _regex("https:\\/\\/matrix.org.*", exclusive=True)
+        )
+        self.assertTrue(self.service.is_exclusive_preview_url("https://matrix.org"))
+        self.assertTrue(
+            self.service.is_exclusive_preview_url("https://matrix.org/foobar")
+        )
+        self.assertTrue(self.service.is_preview_url_in_namespace("https://matrix.org"))
+        self.assertTrue(
+            self.service.is_preview_url_in_namespace("https://matrix.org/foobar")
+        )
+
     @defer.inlineCallbacks
     def test_regex_alias_no_match(
         self,

tests/handlers/test_appservice.py

@@ -203,6 +203,81 @@ class AppServiceHandlerTestCase(unittest.TestCase):
         self.assertEqual(result.room_id, room_id)
         self.assertEqual(result.servers, servers)
 
+    async def test_query_url_preview_none(self) -> None:
+        self.mock_store.get_app_services.return_value = []
+        result = await self.handler.query_preview_url(
+            "https://matrix.org", UserID(localpart="a", domain="b")
+        )
+        assert result is not None
+        self.assertIsNone(result.result)
+        self.assertFalse(result.exclusive)
+
+    async def test_query_url_preview_non_exclusive_no_result(self) -> None:
+        self.mock_store.get_app_services.return_value = [self._mkservice_url_preview()]
+        self.mock_as_api.query_preview_url = AsyncMock(return_value=None)
+        result = await self.handler.query_preview_url(
+            "https://matrix.org", UserID(localpart="a", domain="b")
+        )
+        assert result is not None
+        self.assertIsNone(result.result)
+        self.assertFalse(result.exclusive)
+
+    async def test_query_url_preview_non_exclusive_result(self) -> None:
+        return_value = {"og:title": "foobar"}
+        self.mock_store.get_app_services.return_value = [self._mkservice_url_preview()]
+        self.mock_as_api.query_preview_url = AsyncMock(return_value=return_value)
+        result = await self.handler.query_preview_url(
+            "https://matrix.org", UserID(localpart="a", domain="b")
+        )
+        assert result is not None
+        self.assertEqual(result.result, return_value)
+        self.assertFalse(result.exclusive)
+
+    async def test_query_url_preview_exclusive_no_result(self) -> None:
+        self.mock_store.get_app_services.return_value = [
+            self._mkservice_url_preview(True, True)
+        ]
+        self.mock_as_api.query_preview_url = AsyncMock(return_value=None)
+        result = await self.handler.query_preview_url(
+            "https://matrix.org", UserID(localpart="a", domain="b")
+        )
+        assert result is not None
+        self.assertIsNone(result.result)
+        self.assertTrue(result.exclusive)
+
+    async def test_query_url_preview_exclusive_result(self) -> None:
+        return_value = {"og:title": "foobar"}
+        self.mock_store.get_app_services.return_value = [
+            self._mkservice_url_preview(True, True)
+        ]
+        self.mock_as_api.query_preview_url = AsyncMock(return_value=return_value)
+        result = await self.handler.query_preview_url(
+            "https://matrix.org", UserID(localpart="a", domain="b")
+        )
+        assert result is not None
+        self.assertEqual(result.result, return_value)
+        self.assertTrue(result.exclusive)
+
+    async def test_query_url_preview_multiple_services(self) -> None:
+        return_value = {"og:title": "foobar"}
+        url_service = self._mkservice_url_preview()
+        self.mock_store.get_app_services.return_value = [
+            self._mkservice_url_preview(False),
+            url_service,
+        ]
+        query_preview_url = self.mock_as_api.query_preview_url = AsyncMock(
+            return_value=return_value
+        )
+        result = await self.handler.query_preview_url(
+            "https://matrix.org", UserID(localpart="a", domain="b")
+        )
+        assert result is not None
+
+        # Ensure the correct service is used.
+        self.assertEqual(url_service, query_preview_url.call_args_list[0][0][0])
+        self.assertEqual(result.result, return_value)
+        self.assertFalse(result.exclusive)
+
     def test_get_3pe_protocols_no_appservices(self) -> None:
         self.mock_store.get_app_services.return_value = []
         response = self.successResultOf(
@@ -424,6 +499,25 @@ class AppServiceHandlerTestCase(unittest.TestCase):
         service.url = "mock_service_url"
         return service
 
+    def _mkservice_url_preview(self, is_in_namespace=True, is_exclusive=False) -> Mock:
+        """
+        Create a new mock representing an ApplicationService that is or is not interested
+        in any preview urls.
+
+        Args:
+            is_in_namespace: If true, the application service will be interested in the url
+            is_exclusive: If true, the application service will be exclusively interested in the url
+
+        Returns:
+            A mock representing the ApplicationService.
+        """
+        service = Mock()
+        service.is_preview_url_in_namespace.return_value = is_in_namespace
+        service.is_exclusive_preview_url.return_value = is_exclusive
+        service.token = "mock_service_token"
+        service.url = "mock_service_url"
+        return service
+
 
 class ApplicationServicesHandlerSendEventsTestCase(unittest.HomeserverTestCase):
     """

tests/media/test_url_previewer.py

@@ -19,10 +19,14 @@
 #
 #
 import os
+from unittest.mock import AsyncMock
 
 from twisted.internet.testing import MemoryReactor
 
+from synapse.api.errors import SynapseError
+from synapse.handlers.appservice import ApplicationServiceUrlPreviewResult
 from synapse.server import HomeServer
+from synapse.types import UserID
 from synapse.util.clock import Clock
 
 from tests import unittest
@@ -33,6 +37,8 @@ try:
 except ImportError:
     lxml = None  # type: ignore[assignment]
 
+PREVIEW_USER = UserID("a", "b")
+
 
 class URLPreviewTests(unittest.HomeserverTestCase):
     if not lxml:
@@ -118,3 +124,57 @@ class URLPreviewTests(unittest.HomeserverTestCase):
 
         # The TLD is not blocked.
         self.assertFalse(self.url_previewer._is_url_blocked("https://example.com"))
+
+    @override_config({"experimental_features": {"msc4417_enabled": True}})
+    def test_msc4417_forward_to_appservice_no_result_exclusive(self) -> None:
+        """
+        Tests that previews fail if the appservice returns an empty response and is exclusive.
+        """
+        query_preview_url = self.url_previewer.as_services.query_preview_url = (
+            AsyncMock()
+        )
+        query_preview_url.return_value = ApplicationServiceUrlPreviewResult(
+            result=None, exclusive=True
+        )
+        result = self.get_failure(
+            self.url_previewer.preview("https://matrix.org", PREVIEW_USER, 1),
+            SynapseError,
+        )
+        self.assertEquals(result.value.code, 404)
+
+    @override_config({"experimental_features": {"msc4417_enabled": True}})
+    def test_msc4417_forward_to_appservice_result(self) -> None:
+        """
+        Tests that previews succeed with an appservice provided result.
+        """
+        query_preview_url = self.url_previewer.as_services.query_preview_url = (
+            AsyncMock()
+        )
+        query_preview_url.return_value = ApplicationServiceUrlPreviewResult(
+            result={"og:title": "The home of the best protocol in the universe!"},
+            exclusive=False,
+        )
+        result = self.get_success(
+            self.url_previewer.preview("https://matrix.org", PREVIEW_USER, 1)
+        )
+        self.assertEquals(
+            result, b'{"og:title":"The home of the best protocol in the universe!"}'
+        )
+
+    @override_config({"experimental_features": {"msc4417_enabled": True}})
+    def test_msc4417_forward_to_appservice_no_result_non_exclusive(self) -> None:
+        """
+        Tests that previews fall through to the homeserver when the empty result is non-exclusive.
+        """
+        do_preview_mock = self.url_previewer._do_preview = AsyncMock()
+        do_preview_mock.return_value = b"HS provided bytes"
+        query_preview_url = self.url_previewer.as_services.query_preview_url = (
+            AsyncMock()
+        )
+        query_preview_url.return_value = ApplicationServiceUrlPreviewResult(
+            result=None, exclusive=False
+        )
+        result = self.get_success(
+            self.url_previewer.preview("https://matrix.org", PREVIEW_USER, 1),
+        )
+        self.assertEquals(result, b"HS provided bytes")