Initial routing of events.

4 files changed, 138 insertions, 0 deletions

diff --git a/synapse/api/urls.py b/synapse/api/urls.py
index a918579f50..cf1e4661bb 100644
--- a/synapse/api/urls.py
+++ b/synapse/api/urls.py
@@ -27,6 +27,7 @@ FEDERATION_PREFIX = "/_matrix/federation"
 FEDERATION_V1_PREFIX = FEDERATION_PREFIX + "/v1"
 FEDERATION_V2_PREFIX = FEDERATION_PREFIX + "/v2"
 FEDERATION_UNSTABLE_PREFIX = FEDERATION_PREFIX + "/unstable"
+LINEARIZED_PREFIX = "/_matrix/linearized/unstable"
 STATIC_PREFIX = "/_matrix/static"
 SERVER_KEY_PREFIX = "/_matrix/key"
 MEDIA_R0_PREFIX = "/_matrix/media/r0"
diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py
index 84236ac299..a012cad65d 100644
--- a/synapse/app/homeserver.py
+++ b/synapse/app/homeserver.py
@@ -59,6 +59,7 @@ from synapse.rest import ClientRestResource
 from synapse.rest.admin import AdminRestResource
 from synapse.rest.health import HealthResource
 from synapse.rest.key.v2 import KeyResource
+from synapse.rest.linearized import LinearizedResource
 from synapse.rest.synapse.client import build_synapse_client_resource_tree
 from synapse.rest.well_known import well_known_resource
 from synapse.server import HomeServer
@@ -227,6 +228,13 @@ class SynapseHomeServer(HomeServer):
         if name in ["keys", "federation"]:
             resources[SERVER_KEY_PREFIX] = KeyResource(self)
 
+        if name == "linearized":
+            linearized_resource = LinearizedResource(self)
+            if compress:
+                linearized_resource = gz_wrap(linearized_resource)
+
+            resources["/_matrix/linearized"] = linearized_resource
+
         if name == "metrics" and self.config.metrics.enable_metrics:
             metrics_resource: Resource = MetricsResource(RegistryProxy)
             if compress:
diff --git a/synapse/config/server.py b/synapse/config/server.py
index b46fa51593..04a7790c77 100644
--- a/synapse/config/server.py
+++ b/synapse/config/server.py
@@ -179,6 +179,7 @@ KNOWN_RESOURCES = {
     "federation",
     "health",
     "keys",
+    "linearized",
     "media",
     "metrics",
     "openid",
diff --git a/synapse/rest/linearized.py b/synapse/rest/linearized.py
new file mode 100644
index 0000000000..024191d486
--- /dev/null
+++ b/synapse/rest/linearized.py
@@ -0,0 +1,128 @@
+# Copyright 2023 The Matrix.org Foundation C.I.C.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import logging
+from typing import Dict, List, Tuple
+
+from synapse.api.urls import LINEARIZED_PREFIX
+from synapse.federation.transport.server import Authenticator
+from synapse.federation.transport.server.federation import BaseFederationServerServlet
+from synapse.http.server import JsonResource
+from synapse.http.servlet import parse_string_from_args
+from synapse.types import JsonDict, get_domain_from_id
+
+logger = logging.getLogger(__name__)
+
+
+class LinearizedResource(JsonResource):
+    """Handles incoming federation HTTP requests"""
+
+    def __init__(self, hs: "HomeServer"):
+        self.hs = hs
+        self.clock = hs.get_clock()
+
+        super().__init__(hs, canonical_json=False)
+
+        self.authenticator = Authenticator(hs)
+        self.ratelimiter = hs.get_federation_ratelimiter()
+
+        self.register_servlets()
+
+    def register_servlets(self) -> None:
+        for servletclass in (LinearizedSendServlet, LinearizedInviteServlet):
+            servletclass(
+                self.hs,
+                authenticator=self.authenticator,
+                ratelimiter=self.ratelimiter,
+                server_name=self.hs.hostname,
+            ).register(self)
+
+
+class LinearizedSendServlet(BaseFederationServerServlet):
+    PATH = "/send/(?P<transaction_id>[^/]*)/?"
+    CATEGORY = "Inbound linearized transaction request"
+    PREFIX = LINEARIZED_PREFIX
+
+    # This doesn't seem right. :)
+    REQUIRE_AUTH = False
+
+    # We ratelimit manually in the handler as we queue up the requests and we
+    # don't want to fill up the ratelimiter with blocked requests.
+    RATELIMIT = False
+
+    # This is when someone is trying to send us a bunch of data.
+    async def on_PUT(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        transaction_id: str,
+    ) -> Tuple[int, JsonDict]:
+        """Called on PUT /send/<transaction_id>/
+
+        Args:
+            transaction_id: The transaction_id associated with this request. This
+                is *not* None.
+
+        Returns:
+            Tuple of `(code, response)`, where
+            `response` is a python dict to be converted into JSON that is
+            used as the response body.
+        """
+        # Parse the request
+        try:
+            transaction_data = content
+
+            logger.debug("Decoded %s: %s", transaction_id, str(transaction_data))
+
+            logger.info(
+                "Received txn %s from %s. (PDUs: %d, EDUs: %d)",
+                transaction_id,
+                origin,
+                len(transaction_data.get("pdus", [])),
+                len(transaction_data.get("edus", [])),
+            )
+
+        except Exception as e:
+            logger.exception(e)
+            return 400, {"error": "Invalid transaction"}
+
+        code, response = await self.handler.on_incoming_transaction(
+            origin, transaction_id, self.server_name, transaction_data
+        )
+
+        return code, response
+
+
+class LinearizedInviteServlet(BaseFederationServerServlet):
+    PATH = "/invite"
+    CATEGORY = "Linearized requests"
+    PREFIX = LINEARIZED_PREFIX
+
+    # This doesn't seem right. :)
+    REQUIRE_AUTH = False
+
+    async def on_POST(
+        self, origin: str, content: JsonDict, query: Dict[bytes, List[bytes]]
+    ) -> Tuple[int, JsonDict]:
+        room_version = parse_string_from_args(query, "room_version", required=True)
+        event = content
+
+        # XXX Currently no auth.
+        origin = get_domain_from_id(event["sender"])
+
+        result = await self.handler.on_invite_request(
+            origin, event, room_version_id=room_version
+        )
+
+        return 200, result