Synapse vulnerable to federation denial of service via malformed events

@@ -29,8 +29,13 @@
 # the max size of a (canonical-json-encoded) event
 MAX_PDU_SIZE = 65536
 
-# the "depth" field on events is limited to 2**63 - 1
-MAX_DEPTH = 2**63 - 1
+# Max/min size of ints in canonical JSON
+CANONICALJSON_MAX_INT = (2**53) - 1
+CANONICALJSON_MIN_INT = -CANONICALJSON_MAX_INT
+
+# the "depth" field on events is limited to the same as what
+# canonicaljson accepts
+MAX_DEPTH = CANONICALJSON_MAX_INT
 
 # the maximum length for a room alias is 255 characters
 MAX_ALIAS_LENGTH = 255

@@ -40,6 +40,8 @@
 from canonicaljson import encode_canonical_json
 
 from synapse.api.constants import (
+    CANONICALJSON_MAX_INT,
+    CANONICALJSON_MIN_INT,
     MAX_PDU_SIZE,
     EventContentFields,
     EventTypes,
@@ -61,9 +63,6 @@
 # Find escaped characters, e.g. those with a \ in front of them.
 ESCAPE_SEQUENCE_PATTERN = re.compile(r"\\(.)")
 
-CANONICALJSON_MAX_INT = (2**53) - 1
-CANONICALJSON_MIN_INT = -CANONICALJSON_MAX_INT
-
 
 # Module API callback that allows adding fields to the unsigned section of
 # events that are sent to clients.

@@ -86,9 +86,7 @@ def validate_new(self, event: EventBase, config: HomeServerConfig) -> None:
 
         # Depending on the room version, ensure the data is spec compliant JSON.
         if event.room_version.strict_canonicaljson:
-            # Note that only the client controlled portion of the event is
-            # checked, since we trust the portions of the event we created.
-            validate_canonicaljson(event.content)
+            validate_canonicaljson(event.get_pdu_json())
 
         if event.type == EventTypes.Aliases:
             if "aliases" in event.content:

@@ -20,7 +20,7 @@
 #
 #
 import logging
-from typing import TYPE_CHECKING, Awaitable, Callable, Optional
+from typing import TYPE_CHECKING, Awaitable, Callable, List, Optional, Sequence
 
 from synapse.api.constants import MAX_DEPTH, EventContentFields, EventTypes, Membership
 from synapse.api.errors import Codes, SynapseError
@@ -29,6 +29,7 @@
 from synapse.crypto.keyring import Keyring
 from synapse.events import EventBase, make_event_from_dict
 from synapse.events.utils import prune_event, validate_canonicaljson
+from synapse.federation.units import filter_pdus_for_valid_depth
 from synapse.http.servlet import assert_params_in_dict
 from synapse.logging.opentracing import log_kv, trace
 from synapse.types import JsonDict, get_domain_from_id
@@ -267,6 +268,15 @@ def _is_invite_via_3pid(event: EventBase) -> bool:
     )
 
 
+def parse_events_from_pdu_json(
+    pdus_json: Sequence[JsonDict], room_version: RoomVersion
+) -> List[EventBase]:
+    return [
+        event_from_pdu_json(pdu_json, room_version)
+        for pdu_json in filter_pdus_for_valid_depth(pdus_json)
+    ]
+
+
 def event_from_pdu_json(pdu_json: JsonDict, room_version: RoomVersion) -> EventBase:
     """Construct an EventBase from an event json received over federation
 

@@ -68,6 +68,7 @@
     FederationBase,
     InvalidEventSignatureError,
     event_from_pdu_json,
+    parse_events_from_pdu_json,
 )
 from synapse.federation.transport.client import SendJoinResponse
 from synapse.http.client import is_unknown_endpoint
@@ -349,7 +350,7 @@ async def backfill(
 
         room_version = await self.store.get_room_version(room_id)
 
-        pdus = [event_from_pdu_json(p, room_version) for p in transaction_data_pdus]
+        pdus = parse_events_from_pdu_json(transaction_data_pdus, room_version)
 
         # Check signatures and hash of pdus, removing any from the list that fail checks
         pdus[:] = await self._check_sigs_and_hash_for_pulled_events_and_fetch(
@@ -393,9 +394,7 @@ async def get_pdu_from_destination_raw(
             transaction_data,
         )
 
-        pdu_list: List[EventBase] = [
-            event_from_pdu_json(p, room_version) for p in transaction_data["pdus"]
-        ]
+        pdu_list = parse_events_from_pdu_json(transaction_data["pdus"], room_version)
 
         if pdu_list and pdu_list[0]:
             pdu = pdu_list[0]
@@ -809,7 +808,7 @@ async def get_event_auth(
 
         room_version = await self.store.get_room_version(room_id)
 
-        auth_chain = [event_from_pdu_json(p, room_version) for p in res["auth_chain"]]
+        auth_chain = parse_events_from_pdu_json(res["auth_chain"], room_version)
 
         signed_auth = await self._check_sigs_and_hash_for_pulled_events_and_fetch(
             destination, auth_chain, room_version=room_version
@@ -1529,9 +1528,7 @@ async def get_missing_events(
 
             room_version = await self.store.get_room_version(room_id)
 
-            events = [
-                event_from_pdu_json(e, room_version) for e in content.get("events", [])
-            ]
+            events = parse_events_from_pdu_json(content.get("events", []), room_version)
 
             signed_events = await self._check_sigs_and_hash_for_pulled_events_and_fetch(
                 destination, events, room_version=room_version

@@ -66,7 +66,7 @@
     event_from_pdu_json,
 )
 from synapse.federation.persistence import TransactionActions
-from synapse.federation.units import Edu, Transaction
+from synapse.federation.units import Edu, Transaction, serialize_and_filter_pdus
 from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME
 from synapse.http.servlet import assert_params_in_dict
 from synapse.logging.context import (
@@ -469,7 +469,12 @@ async def _handle_pdus_in_txn(
                 logger.info("Ignoring PDU: %s", e)
                 continue
 
-            event = event_from_pdu_json(p, room_version)
+            try:
+                event = event_from_pdu_json(p, room_version)
+            except SynapseError as e:
+                logger.info("Ignoring PDU for failing to deserialize: %s", e)
+                continue
+
             pdus_by_room.setdefault(room_id, []).append(event)
 
             if event.origin_server_ts > newest_pdu_ts:
@@ -636,8 +641,8 @@ async def _on_context_state_request_compute(
         )
 
         return {
-            "pdus": [pdu.get_pdu_json() for pdu in pdus],
-            "auth_chain": [pdu.get_pdu_json() for pdu in auth_chain],
+            "pdus": serialize_and_filter_pdus(pdus),
+            "auth_chain": serialize_and_filter_pdus(auth_chain),
         }
 
     async def on_pdu_request(
@@ -761,8 +766,8 @@ async def on_send_join_request(
         event_json = event.get_pdu_json(time_now)
         resp = {
             "event": event_json,
-            "state": [p.get_pdu_json(time_now) for p in state_events],
-            "auth_chain": [p.get_pdu_json(time_now) for p in auth_chain_events],
+            "state": serialize_and_filter_pdus(state_events, time_now),
+            "auth_chain": serialize_and_filter_pdus(auth_chain_events, time_now),
             "members_omitted": caller_supports_partial_state,
         }
 
@@ -1005,7 +1010,7 @@ async def on_event_auth(
 
             time_now = self._clock.time_msec()
             auth_pdus = await self.handler.on_event_auth(event_id)
-            res = {"auth_chain": [a.get_pdu_json(time_now) for a in auth_pdus]}
+            res = {"auth_chain": serialize_and_filter_pdus(auth_pdus, time_now)}
         return 200, res
 
     async def on_query_client_keys(
@@ -1090,7 +1095,7 @@ async def on_get_missing_events(
 
             time_now = self._clock.time_msec()
 
-        return {"events": [ev.get_pdu_json(time_now) for ev in missing_events]}
+        return {"events": serialize_and_filter_pdus(missing_events, time_now)}
 
     async def on_openid_userinfo(self, token: str) -> Optional[str]:
         ts_now_ms = self._clock.time_msec()

@@ -24,10 +24,12 @@
 """
 
 import logging
-from typing import List, Optional
+from typing import List, Optional, Sequence
 
 import attr
 
+from synapse.api.constants import CANONICALJSON_MAX_INT, CANONICALJSON_MIN_INT
+from synapse.events import EventBase
 from synapse.types import JsonDict
 
 logger = logging.getLogger(__name__)
@@ -104,8 +106,28 @@ def get_dict(self) -> JsonDict:
         result = {
             "origin": self.origin,
             "origin_server_ts": self.origin_server_ts,
-            "pdus": self.pdus,
+            "pdus": filter_pdus_for_valid_depth(self.pdus),
         }
         if self.edus:
             result["edus"] = self.edus
         return result
+
+
+def filter_pdus_for_valid_depth(pdus: Sequence[JsonDict]) -> List[JsonDict]:
+    filtered_pdus = []
+    for pdu in pdus:
+        # Drop PDUs that have a depth that is outside of the range allowed
+        # by canonical json.
+        if (
+            "depth" in pdu
+            and CANONICALJSON_MIN_INT <= pdu["depth"] <= CANONICALJSON_MAX_INT
+        ):
+            filtered_pdus.append(pdu)
+
+    return filtered_pdus
+
+
+def serialize_and_filter_pdus(
+    pdus: Sequence[EventBase], time_now: Optional[int] = None
+) -> List[JsonDict]:
+    return filter_pdus_for_valid_depth([pdu.get_pdu_json(time_now) for pdu in pdus])