From 3c57700aa10f6f2312b620785909dff8bd727ce9 Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Wed, 21 Jul 2021 16:08:56 -0400 Subject: [PATCH 1/8] Minor change to generating fake transactions. --- synapse/federation/federation_server.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 145b9161d985..5a10528068ce 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -195,7 +195,7 @@ async def on_backfill_request( origin, room_id, versions, limit ) - res = self._transaction_from_pdus(pdus).get_dict() + res = self._transaction_dict_from_pdus(pdus) return 200, res @@ -538,7 +538,7 @@ async def on_pdu_request( pdu = await self.handler.get_persisted_pdu(origin, event_id) if pdu: - return 200, self._transaction_from_pdus([pdu]).get_dict() + return 200, self._transaction_dict_from_pdus([pdu]) else: return 404, "" @@ -879,7 +879,7 @@ async def on_openid_userinfo(self, token: str) -> Optional[str]: ts_now_ms = self._clock.time_msec() return await self.store.get_user_id_for_open_id_token(token, ts_now_ms) - def _transaction_from_pdus(self, pdu_list: List[EventBase]) -> Transaction: + def _transaction_dict_from_pdus(self, pdu_list: List[EventBase]) -> JsonDict: """Returns a new Transaction containing the given PDUs suitable for transmission. """ @@ -890,7 +890,7 @@ def _transaction_from_pdus(self, pdu_list: List[EventBase]) -> Transaction: pdus=pdus, origin_server_ts=int(time_now), destination=None, - ) + ).get_dict() async def _handle_received_pdu(self, origin: str, pdu: EventBase) -> None: """Process a PDU received in a federation /send/ transaction. From f78cb70ed16c89a47b50000a39b6ca2ca884462c Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Fri, 30 Jul 2021 15:02:38 -0400 Subject: [PATCH 2/8] Pass through the transaction_id/destination explicitly. --- synapse/federation/federation_server.py | 15 +++++++++++---- synapse/federation/transport/server.py | 8 +------- 2 files changed, 12 insertions(+), 11 deletions(-) diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 5a10528068ce..af9332655973 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -200,8 +200,12 @@ async def on_backfill_request( return 200, res async def on_incoming_transaction( - self, origin: str, transaction_data: JsonDict - ) -> Tuple[int, Dict[str, Any]]: + self, + origin: str, + transaction_id: str, + destination: str, + transaction_data: JsonDict, + ) -> Tuple[int, JsonDict]: # If we receive a transaction we should make sure that kick off handling # any old events in the staging area. if not self._started_handling_of_staged_events: @@ -212,8 +216,11 @@ async def on_incoming_transaction( # accurate as possible. request_time = self._clock.time_msec() - transaction = Transaction(**transaction_data) - transaction_id = transaction.transaction_id # type: ignore + transaction = Transaction( + transaction_id=transaction_id, + destination=destination, + **transaction_data, + ) if not transaction_id: raise Exception("Transaction missing transaction_id") diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py index 5e059d6e09d4..e4e77c8d9a68 100644 --- a/synapse/federation/transport/server.py +++ b/synapse/federation/transport/server.py @@ -453,18 +453,12 @@ async def on_PUT( # We should ideally be getting this from the security layer. # origin = body["origin"] - # Add some extra data to the transaction dict that isn't included - # in the request body. - transaction_data.update( - transaction_id=transaction_id, destination=self.server_name - ) - except Exception as e: logger.exception(e) return 400, {"error": "Invalid transaction"} code, response = await self.handler.on_incoming_transaction( - origin, transaction_data + origin, transaction_id, self.server_name, transaction_data ) return code, response From 61a872015f442c46d345e8d1c145f91d0089f602 Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Fri, 30 Jul 2021 15:09:05 -0400 Subject: [PATCH 3/8] Remove obsolete comment. --- synapse/federation/transport/server.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py index e4e77c8d9a68..640f46fff6bf 100644 --- a/synapse/federation/transport/server.py +++ b/synapse/federation/transport/server.py @@ -450,9 +450,6 @@ async def on_PUT( len(transaction_data.get("edus", [])), ) - # We should ideally be getting this from the security layer. - # origin = body["origin"] - except Exception as e: logger.exception(e) return 400, {"error": "Invalid transaction"} From 08220fa6141fe64efdb40e4daa71f59c3d249985 Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Fri, 30 Jul 2021 15:13:49 -0400 Subject: [PATCH 4/8] Remove unnecessary create_new method. --- synapse/federation/sender/transaction_manager.py | 6 +++--- synapse/federation/units.py | 14 -------------- 2 files changed, 3 insertions(+), 17 deletions(-) diff --git a/synapse/federation/sender/transaction_manager.py b/synapse/federation/sender/transaction_manager.py index 72a635830b9a..4d5b5bedb7e0 100644 --- a/synapse/federation/sender/transaction_manager.py +++ b/synapse/federation/sender/transaction_manager.py @@ -104,13 +104,13 @@ async def send_new_transaction( len(edus), ) - transaction = Transaction.create_new( + transaction = Transaction( origin_server_ts=int(self.clock.time_msec()), transaction_id=txn_id, origin=self._server_name, destination=destination, - pdus=pdus, - edus=edus, + pdus=[p.get_pdu_json() for p in pdus], + edus=[edu.get_dict() for edu in edus], ) self._next_txn_id += 1 diff --git a/synapse/federation/units.py b/synapse/federation/units.py index c83a261918c0..e79e8fd4870f 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -108,17 +108,3 @@ def __init__(self, transaction_id=None, pdus: Optional[list] = None, **kwargs): del kwargs["edus"] super().__init__(transaction_id=transaction_id, pdus=pdus or [], **kwargs) - - @staticmethod - def create_new(pdus, **kwargs): - """Used to create a new transaction. Will auto fill out - transaction_id and origin_server_ts keys. - """ - if "origin_server_ts" not in kwargs: - raise KeyError("Require 'origin_server_ts' to construct a Transaction") - if "transaction_id" not in kwargs: - raise KeyError("Require 'transaction_id' to construct a Transaction") - - kwargs["pdus"] = [p.get_pdu_json() for p in pdus] - - return Transaction(**kwargs) From 4150a4074df3f96b5e58b80576294c8093b3e09d Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Fri, 30 Jul 2021 15:15:45 -0400 Subject: [PATCH 5/8] Add some missing type hints. --- synapse/federation/federation_server.py | 4 ++-- synapse/federation/sender/transaction_manager.py | 3 ++- synapse/federation/transport/client.py | 2 +- 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index af9332655973..d6629392466d 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -443,10 +443,10 @@ async def process_pdu(pdu: EventBase) -> JsonDict: return pdu_results - async def _handle_edus_in_txn(self, origin: str, transaction: Transaction): + async def _handle_edus_in_txn(self, origin: str, transaction: Transaction) -> None: """Process the EDUs in a received transaction.""" - async def _process_edu(edu_dict): + async def _process_edu(edu_dict: JsonDict) -> None: received_edus_counter.inc() edu = Edu( diff --git a/synapse/federation/sender/transaction_manager.py b/synapse/federation/sender/transaction_manager.py index 4d5b5bedb7e0..dc555cca0bbf 100644 --- a/synapse/federation/sender/transaction_manager.py +++ b/synapse/federation/sender/transaction_manager.py @@ -27,6 +27,7 @@ tags, whitelisted_homeserver, ) +from synapse.types import JsonDict from synapse.util import json_decoder from synapse.util.metrics import measure_func @@ -131,7 +132,7 @@ async def send_new_transaction( # FIXME (richardv): I also believe it no longer works. We (now?) store # "age_ts" in "unsigned" rather than at the top level. See # https://github.com/matrix-org/synapse/issues/8429. - def json_data_cb(): + def json_data_cb() -> JsonDict: data = transaction.get_dict() now = int(self.clock.time_msec()) if "pdus" in data: diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py index 6a8d3ad4fe6d..90a7c16b62a1 100644 --- a/synapse/federation/transport/client.py +++ b/synapse/federation/transport/client.py @@ -143,7 +143,7 @@ async def send_transaction( """Sends the given Transaction to its destination Args: - transaction (Transaction) + transaction Returns: Succeeds when we get a 2xx HTTP response. The result From 0ffdd0086658f25a188ead16a2a6c0572482054c Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Thu, 5 Aug 2021 07:03:41 -0400 Subject: [PATCH 6/8] Use attrs and remove JsonEncodedObject. --- synapse/federation/federation_server.py | 15 ++-- synapse/federation/units.py | 76 ++++++++---------- synapse/util/jsonobject.py | 102 ------------------------ 3 files changed, 44 insertions(+), 149 deletions(-) delete mode 100644 synapse/util/jsonobject.py diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index d6629392466d..fcd6990c7592 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -219,7 +219,10 @@ async def on_incoming_transaction( transaction = Transaction( transaction_id=transaction_id, destination=destination, - **transaction_data, + origin=origin, + origin_server_ts=transaction_data.get("origin_server_ts"), # type: ignore + pdus=transaction_data.get("pdus"), # type: ignore + edus=transaction_data.get("edus"), ) if not transaction_id: @@ -228,9 +231,7 @@ async def on_incoming_transaction( logger.debug("[%s] Got transaction", transaction_id) # Reject malformed transactions early: reject if too many PDUs/EDUs - if len(transaction.pdus) > 50 or ( # type: ignore - hasattr(transaction, "edus") and len(transaction.edus) > 100 # type: ignore - ): + if len(transaction.pdus) > 50 or len(transaction.edus) > 100: logger.info("Transaction PDU or EDU count too large. Returning 400") return 400, {} @@ -459,7 +460,7 @@ async def _process_edu(edu_dict: JsonDict) -> None: await concurrently_execute( _process_edu, - getattr(transaction, "edus", []), + transaction.edus, TRANSACTION_CONCURRENCY_LIMIT, ) @@ -893,10 +894,12 @@ def _transaction_dict_from_pdus(self, pdu_list: List[EventBase]) -> JsonDict: time_now = self._clock.time_msec() pdus = [p.get_pdu_json(time_now) for p in pdu_list] return Transaction( + # Just need a dummy transaction ID and destination since it won't be used. + transaction_id="", origin=self.server_name, pdus=pdus, origin_server_ts=int(time_now), - destination=None, + destination="", ).get_dict() async def _handle_received_pdu(self, origin: str, pdu: EventBase) -> None: diff --git a/synapse/federation/units.py b/synapse/federation/units.py index e79e8fd4870f..b9b12fbea563 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -17,18 +17,17 @@ """ import logging -from typing import Optional +from typing import List, Optional import attr from synapse.types import JsonDict -from synapse.util.jsonobject import JsonEncodedObject logger = logging.getLogger(__name__) -@attr.s(slots=True) -class Edu(JsonEncodedObject): +@attr.s(slots=True, frozen=True, auto_attribs=True) +class Edu: """An Edu represents a piece of data sent from one homeserver to another. In comparison to Pdus, Edus are not persisted for a long time on disk, are @@ -36,10 +35,10 @@ class Edu(JsonEncodedObject): internal ID or previous references graph. """ - edu_type = attr.ib(type=str) - content = attr.ib(type=dict) - origin = attr.ib(type=str) - destination = attr.ib(type=str) + edu_type: str + content: dict + origin: str + destination: str def get_dict(self) -> JsonDict: return { @@ -55,14 +54,21 @@ def get_internal_dict(self) -> JsonDict: "destination": self.destination, } - def get_context(self): + def get_context(self) -> str: return getattr(self, "content", {}).get("org.matrix.opentracing_context", "{}") - def strip_context(self): + def strip_context(self) -> None: getattr(self, "content", {})["org.matrix.opentracing_context"] = "{}" -class Transaction(JsonEncodedObject): +def _none_to_list(edus: Optional[List[JsonDict]]) -> List[JsonDict]: + if edus is None: + return [] + return edus + + +@attr.s(slots=True, frozen=True, auto_attribs=True) +class Transaction: """A transaction is a list of Pdus and Edus to be sent to a remote home server with some extra metadata. @@ -78,33 +84,21 @@ class Transaction(JsonEncodedObject): """ - valid_keys = [ - "transaction_id", - "origin", - "destination", - "origin_server_ts", - "previous_ids", - "pdus", - "edus", - ] - - internal_keys = ["transaction_id", "destination"] - - required_keys = [ - "transaction_id", - "origin", - "destination", - "origin_server_ts", - "pdus", - ] - - def __init__(self, transaction_id=None, pdus: Optional[list] = None, **kwargs): - """If we include a list of pdus then we decode then as PDU's - automatically. - """ - - # If there's no EDUs then remove the arg - if "edus" in kwargs and not kwargs["edus"]: - del kwargs["edus"] - - super().__init__(transaction_id=transaction_id, pdus=pdus or [], **kwargs) + # Required keys. + transaction_id: str + origin: str + destination: str + origin_server_ts: int + pdus: List[JsonDict] = attr.ib(factory=list, converter=_none_to_list) + edus: List[JsonDict] = attr.ib(factory=list, converter=_none_to_list) + + def get_dict(self) -> JsonDict: + """A JSON-ready dictionary of valid keys which aren't internal.""" + result = { + "origin": self.origin, + "origin_server_ts": self.origin_server_ts, + "pdus": self.pdus, + } + if self.edus: + result["edus"] = self.edus + return result diff --git a/synapse/util/jsonobject.py b/synapse/util/jsonobject.py deleted file mode 100644 index abc12f08374d..000000000000 --- a/synapse/util/jsonobject.py +++ /dev/null @@ -1,102 +0,0 @@ -# Copyright 2014-2016 OpenMarket Ltd -# -# 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. - - -class JsonEncodedObject: - """A common base class for defining protocol units that are represented - as JSON. - - Attributes: - unrecognized_keys (dict): A dict containing all the key/value pairs we - don't recognize. - """ - - valid_keys = [] # keys we will store - """A list of strings that represent keys we know about - and can handle. If we have values for these keys they will be - included in the `dictionary` instance variable. - """ - - internal_keys = [] # keys to ignore while building dict - """A list of strings that should *not* be encoded into JSON. - """ - - required_keys = [] - """A list of strings that we require to exist. If they are not given upon - construction it raises an exception. - """ - - def __init__(self, **kwargs): - """Takes the dict of `kwargs` and loads all keys that are *valid* - (i.e., are included in the `valid_keys` list) into the dictionary` - instance variable. - - Any keys that aren't recognized are added to the `unrecognized_keys` - attribute. - - Args: - **kwargs: Attributes associated with this protocol unit. - """ - for required_key in self.required_keys: - if required_key not in kwargs: - raise RuntimeError("Key %s is required" % required_key) - - self.unrecognized_keys = {} # Keys we were given not listed as valid - for k, v in kwargs.items(): - if k in self.valid_keys or k in self.internal_keys: - self.__dict__[k] = v - else: - self.unrecognized_keys[k] = v - - def get_dict(self): - """Converts this protocol unit into a :py:class:`dict`, ready to be - encoded as JSON. - - The keys it encodes are: `valid_keys` - `internal_keys` - - Returns - dict - """ - d = { - k: _encode(v) - for (k, v) in self.__dict__.items() - if k in self.valid_keys and k not in self.internal_keys - } - d.update(self.unrecognized_keys) - return d - - def get_internal_dict(self): - d = { - k: _encode(v, internal=True) - for (k, v) in self.__dict__.items() - if k in self.valid_keys - } - d.update(self.unrecognized_keys) - return d - - def __str__(self): - return "(%s, %s)" % (self.__class__.__name__, repr(self.__dict__)) - - -def _encode(obj, internal=False): - if type(obj) is list: - return [_encode(o, internal=internal) for o in obj] - - if isinstance(obj, JsonEncodedObject): - if internal: - return obj.get_internal_dict() - else: - return obj.get_dict() - - return obj From 9776a1d8df431f777032074fbca85cf4ad984294 Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Thu, 5 Aug 2021 07:03:55 -0400 Subject: [PATCH 7/8] Remove ignoring type hints in a few places. --- synapse/federation/federation_server.py | 10 +++++----- synapse/federation/persistence.py | 4 ++-- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index fcd6990c7592..0385aadefaf7 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -271,7 +271,7 @@ async def _on_incoming_transaction_inner( # CRITICAL SECTION: the first thing we must do (before awaiting) is # add an entry to _active_transactions. assert origin not in self._active_transactions - self._active_transactions[origin] = transaction.transaction_id # type: ignore + self._active_transactions[origin] = transaction.transaction_id try: result = await self._handle_incoming_transaction( @@ -299,11 +299,11 @@ async def _handle_incoming_transaction( if response: logger.debug( "[%s] We've already responded to this request", - transaction.transaction_id, # type: ignore + transaction.transaction_id, ) return response - logger.debug("[%s] Transaction is new", transaction.transaction_id) # type: ignore + logger.debug("[%s] Transaction is new", transaction.transaction_id) # We process PDUs and EDUs in parallel. This is important as we don't # want to block things like to device messages from reaching clients @@ -342,7 +342,7 @@ async def _handle_pdus_in_txn( report back to the sending server. """ - received_pdus_counter.inc(len(transaction.pdus)) # type: ignore + received_pdus_counter.inc(len(transaction.pdus)) origin_host, _ = parse_server_name(origin) @@ -350,7 +350,7 @@ async def _handle_pdus_in_txn( newest_pdu_ts = 0 - for p in transaction.pdus: # type: ignore + for p in transaction.pdus: # FIXME (richardv): I don't think this works: # https://github.com/matrix-org/synapse/issues/8429 if "unsigned" in p: diff --git a/synapse/federation/persistence.py b/synapse/federation/persistence.py index 2f9c9bc2cdc8..4fead6ca2954 100644 --- a/synapse/federation/persistence.py +++ b/synapse/federation/persistence.py @@ -45,7 +45,7 @@ async def have_responded( `None` if we have not previously responded to this transaction or a 2-tuple of `(int, dict)` representing the response code and response body. """ - transaction_id = transaction.transaction_id # type: ignore + transaction_id = transaction.transaction_id if not transaction_id: raise RuntimeError("Cannot persist a transaction with no transaction_id") @@ -56,7 +56,7 @@ async def set_response( self, origin: str, transaction: Transaction, code: int, response: JsonDict ) -> None: """Persist how we responded to a transaction.""" - transaction_id = transaction.transaction_id # type: ignore + transaction_id = transaction.transaction_id if not transaction_id: raise RuntimeError("Cannot persist a transaction with no transaction_id") From 0a5912458633f52dd50b1b3f5af587560c0f457c Mon Sep 17 00:00:00 2001 From: Patrick Cloke Date: Thu, 5 Aug 2021 07:58:47 -0400 Subject: [PATCH 8/8] Newsfragment --- changelog.d/10542.misc | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/10542.misc diff --git a/changelog.d/10542.misc b/changelog.d/10542.misc new file mode 100644 index 000000000000..44b70b473037 --- /dev/null +++ b/changelog.d/10542.misc @@ -0,0 +1 @@ +Convert `Transaction` and `Edu` objects to attrs.