MSC4452: Preview URL capability

changelog.d/19715.feature

@@ -0,0 +1 @@
+Add support for "MSC4452 Preview URL capabilities API" which exposes a `io.element.msc4452.preview_url` capability.

synapse/rest/client/capabilities.py

@@ -74,6 +74,9 @@ async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]:
                 "m.get_login_token": {
                     "enabled": self.config.auth.login_via_existing_enabled,
                 },
+                "io.element.msc4452.preview_url": {
+                    "enabled": self.config.media.url_preview_enabled,
+                },
             }
         }
 

synapse/rest/client/media.py

@@ -23,7 +23,7 @@
 import logging
 import re
 
-from synapse.api.errors import Codes, cs_error
+from synapse.api.errors import Codes, SynapseError, cs_error
 from synapse.http.server import (
     HttpServer,
     respond_with_json,
@@ -79,11 +79,13 @@ def __init__(
         self.clock = hs.get_clock()
         self.media_repo = media_repo
         self.media_storage = media_storage
-        assert self.media_repo.url_previewer is not None
         self.url_previewer = self.media_repo.url_previewer
 
     async def on_GET(self, request: SynapseRequest) -> None:
         requester = await self.auth.get_user_by_req(request)
+        if self.url_previewer is None:
+            # If we have no url_previewer then it has been disabled by the server.
+            raise SynapseError(403, "URL Previews are disabled", Codes.FORBIDDEN)
         url = parse_string(request, "url", required=True)
         ts = parse_integer(request, "ts")
         if ts is None:
@@ -299,10 +301,7 @@ async def on_GET(
 
 def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
     media_repo = hs.get_media_repository()
-    if hs.config.media.url_preview_enabled:
-        PreviewURLServlet(hs, media_repo, media_repo.media_storage).register(
-            http_server
-        )
+    PreviewURLServlet(hs, media_repo, media_repo.media_storage).register(http_server)
     MediaConfigResource(hs).register(http_server)
     ThumbnailResource(hs, media_repo, media_repo.media_storage).register(http_server)
     DownloadResource(hs, media_repo).register(http_server)

synapse/rest/media/media_repository_resource.py

@@ -106,8 +106,7 @@ def register_servlets(http_server: HttpServer, hs: "HomeServer") -> None:
         ThumbnailResource(hs, media_repo, media_repo.media_storage).register(
             http_server
         )
-        if hs.config.media.url_preview_enabled:
-            PreviewUrlResource(hs, media_repo, media_repo.media_storage).register(
-                http_server
-            )
+        PreviewUrlResource(hs, media_repo, media_repo.media_storage).register(
+            http_server
+        )
         MediaConfigResource(hs).register(http_server)

synapse/rest/media/preview_url_resource.py

@@ -23,6 +23,7 @@
 import re
 from typing import TYPE_CHECKING
 
+from synapse.api.errors import Codes, SynapseError
 from synapse.http.server import respond_with_json_bytes
 from synapse.http.servlet import RestServlet, parse_integer, parse_string
 from synapse.http.site import SynapseRequest
@@ -65,12 +66,14 @@ def __init__(
         self.clock = hs.get_clock()
         self.media_repo = media_repo
         self.media_storage = media_storage
-        assert self.media_repo.url_previewer is not None
         self.url_previewer = self.media_repo.url_previewer
 
     async def on_GET(self, request: SynapseRequest) -> None:
         # XXX: if get_user_by_req fails, what should we do in an async render?
         requester = await self.auth.get_user_by_req(request)
+        if self.url_previewer is None:
+            # If we have no url_previewer then it has been disabled by the server.
+            raise SynapseError(403, "URL Previews are disabled", Codes.FORBIDDEN)
         url = parse_string(request, "url", required=True)
         ts = parse_integer(request, "ts", default=self.clock.time_msec())
         og = await self.url_previewer.preview(url, requester.user, ts)

tests/rest/client/test_capabilities.py

@@ -276,3 +276,33 @@ def test_get_forget_forced_upon_leave_without_auto_forget(self) -> None:
         self.assertFalse(
             capabilities["org.matrix.msc4267.forget_forced_upon_leave"]["enabled"]
         )
+
+    @override_config(
+        {"url_preview_enabled": True, "url_preview_ip_range_blacklist": ["127.0.0.1"]}
+    )
+    def test_url_previews_enabled(self) -> None:
+        access_token = self.get_success(
+            self.auth_handler.create_access_token_for_user_id(
+                self.user, device_id=None, valid_until_ms=None
+            )
+        )
+        channel = self.make_request("GET", self.url, access_token=access_token)
+        capabilities = channel.json_body["capabilities"]
+        self.assertEqual(channel.code, HTTPStatus.OK)
+        self.assertTrue(capabilities["io.element.msc4452.preview_url"]["enabled"])
+
+    @override_config(
+        {
+            "url_preview_enabled": False,
+        }
+    )
+    def test_url_previews_disabled(self) -> None:
+        access_token = self.get_success(
+            self.auth_handler.create_access_token_for_user_id(
+                self.user, device_id=None, valid_until_ms=None
+            )
+        )
+        channel = self.make_request("GET", self.url, access_token=access_token)
+        capabilities = channel.json_body["capabilities"]
+        self.assertEqual(channel.code, HTTPStatus.OK)
+        self.assertFalse(capabilities["io.element.msc4452.preview_url"]["enabled"])

tests/rest/client/test_media.py

@@ -1573,6 +1573,33 @@ def test_blocked_url(self) -> None:
         self.assertEqual(channel.code, 403, channel.result)
 
 
+class URLPreviewDisabledTests(unittest.HomeserverTestCase):
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        media.register_servlets,
+    ]
+
+    def prepare(
+        self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer
+    ) -> None:
+        self.register_user("user", "password")
+        self.tok = self.login("user", "password")
+
+    def test_disabled_previews(self) -> None:
+        """Tests that disabling URL previews gives back a sane response."""
+        channel = self.make_request(
+            "GET",
+            "/_matrix/client/v1/media/preview_url?url=" + quote("http://example.com"),
+            access_token=self.tok,
+        )
+        self.assertEqual(channel.code, 403, channel.result)
+        self.assertEqual(
+            channel.json_body,
+            {"errcode": "M_FORBIDDEN", "error": "URL Previews are disabled"},
+        )
+
+
 class MediaConfigTest(unittest.HomeserverTestCase):
     servlets = [
         media.register_servlets,