From 1c445f88f64beabf0bd9bec3950a4a4c0d529e8a Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Wed, 15 Oct 2014 17:09:04 +0100 Subject: persist hashes and origin signatures for PDUs --- synapse/api/events/utils.py | 23 +++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/events/utils.py b/synapse/api/events/utils.py index c3a32be8c1..7fdf45a264 100644 --- a/synapse/api/events/utils.py +++ b/synapse/api/events/utils.py @@ -27,7 +27,14 @@ def prune_event(event): the user has specified, but we do want to keep necessary information like type, state_key etc. """ + return _prune_event_or_pdu(event.type, event) +def prune_pdu(pdu): + """Removes keys that contain unrestricted and non-essential data from a PDU + """ + return _prune_event_or_pdu(pdu.pdu_type, pdu) + +def _prune_event_or_pdu(event_type, event): # Remove all extraneous fields. event.unrecognized_keys = {} @@ -38,25 +45,25 @@ def prune_event(event): if field in event.content: new_content[field] = event.content[field] - if event.type == RoomMemberEvent.TYPE: + if event_type == RoomMemberEvent.TYPE: add_fields("membership") - elif event.type == RoomCreateEvent.TYPE: + elif event_type == RoomCreateEvent.TYPE: add_fields("creator") - elif event.type == RoomJoinRulesEvent.TYPE: + elif event_type == RoomJoinRulesEvent.TYPE: add_fields("join_rule") - elif event.type == RoomPowerLevelsEvent.TYPE: + elif event_type == RoomPowerLevelsEvent.TYPE: # TODO: Actually check these are valid user_ids etc. add_fields("default") for k, v in event.content.items(): if k.startswith("@") and isinstance(v, (int, long)): new_content[k] = v - elif event.type == RoomAddStateLevelEvent.TYPE: + elif event_type == RoomAddStateLevelEvent.TYPE: add_fields("level") - elif event.type == RoomSendEventLevelEvent.TYPE: + elif event_type == RoomSendEventLevelEvent.TYPE: add_fields("level") - elif event.type == RoomOpsPowerLevelsEvent.TYPE: + elif event_type == RoomOpsPowerLevelsEvent.TYPE: add_fields("kick_level", "ban_level", "redact_level") - elif event.type == RoomAliasesEvent.TYPE: + elif event_type == RoomAliasesEvent.TYPE: add_fields("aliases") event.content = new_content -- cgit 1.5.1 From bb04447c44036ebf3ae5dde7a4cc7a7909d50ef6 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Thu, 16 Oct 2014 23:25:12 +0100 Subject: Include hashes of previous pdus when referencing them --- synapse/api/events/__init__.py | 2 +- synapse/federation/pdu_codec.py | 13 ++++--------- synapse/federation/replication.py | 2 +- synapse/federation/units.py | 10 +++++++++- synapse/state.py | 4 ---- synapse/storage/__init__.py | 20 ++++++++++++++------ synapse/storage/pdu.py | 22 ++++++++++++++++------ synapse/storage/schema/signatures.sql | 16 ++++++++++++++++ synapse/storage/signatures.py | 31 +++++++++++++++++++++++++++++++ tests/federation/test_federation.py | 2 +- tests/federation/test_pdu_codec.py | 4 ++-- 11 files changed, 95 insertions(+), 31 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index f66fea2904..a5a55742e0 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -65,13 +65,13 @@ class SynapseEvent(JsonEncodedObject): internal_keys = [ "is_state", - "prev_events", "depth", "destinations", "origin", "outlier", "power_level", "redacted", + "prev_pdus", ] required_keys = [ diff --git a/synapse/federation/pdu_codec.py b/synapse/federation/pdu_codec.py index bcac5f9ae8..11fd7264b3 100644 --- a/synapse/federation/pdu_codec.py +++ b/synapse/federation/pdu_codec.py @@ -45,9 +45,7 @@ class PduCodec(object): kwargs["event_id"] = encode_event_id(pdu.pdu_id, pdu.origin) kwargs["room_id"] = pdu.context kwargs["etype"] = pdu.pdu_type - kwargs["prev_events"] = [ - encode_event_id(p[0], p[1]) for p in pdu.prev_pdus - ] + kwargs["prev_pdus"] = pdu.prev_pdus if hasattr(pdu, "prev_state_id") and hasattr(pdu, "prev_state_origin"): kwargs["prev_state"] = encode_event_id( @@ -78,11 +76,8 @@ class PduCodec(object): d["context"] = event.room_id d["pdu_type"] = event.type - if hasattr(event, "prev_events"): - d["prev_pdus"] = [ - decode_event_id(e, self.server_name) - for e in event.prev_events - ] + if hasattr(event, "prev_pdus"): + d["prev_pdus"] = event.prev_pdus if hasattr(event, "prev_state"): d["prev_state_id"], d["prev_state_origin"] = ( @@ -95,7 +90,7 @@ class PduCodec(object): kwargs = copy.deepcopy(event.unrecognized_keys) kwargs.update({ k: v for k, v in d.items() - if k not in ["event_id", "room_id", "type", "prev_events"] + if k not in ["event_id", "room_id", "type"] }) if "ts" not in kwargs: diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 9363ac7300..788a49b8e8 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -443,7 +443,7 @@ class ReplicationLayer(object): min_depth = yield self.store.get_min_depth_for_context(pdu.context) if min_depth and pdu.depth > min_depth: - for pdu_id, origin in pdu.prev_pdus: + for pdu_id, origin, hashes in pdu.prev_pdus: exists = yield self._get_persisted_pdu(pdu_id, origin) if not exists: diff --git a/synapse/federation/units.py b/synapse/federation/units.py index 3518efb215..6a43007837 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -141,8 +141,16 @@ class Pdu(JsonEncodedObject): for kid, sig in pdu_tuple.signatures.items() } + prev_pdus = [] + for prev_pdu in pdu_tuple.prev_pdu_list: + prev_hashes = pdu_tuple.edge_hashes.get(prev_pdu, {}) + prev_hashes = { + alg: encode_base64(hsh) for alg, hsh in prev_hashes.items() + } + prev_pdus.append((prev_pdu[0], prev_pdu[1], prev_hashes)) + return Pdu( - prev_pdus=pdu_tuple.prev_pdu_list, + prev_pdus=prev_pdus, **args ) else: diff --git a/synapse/state.py b/synapse/state.py index 9db84c9b5c..bc6b928ec7 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -72,10 +72,6 @@ class StateHandler(object): snapshot.fill_out_prev_events(event) - event.prev_events = [ - e for e in event.prev_events if e != event.event_id - ] - current_state = snapshot.prev_state_pdu if current_state: diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index b2a3f0b56c..af05b47932 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -177,6 +177,14 @@ class DataStore(RoomMemberStore, RoomStore, txn, pdu.pdu_id, pdu.origin, key_id, signature_bytes, ) + for prev_pdu_id, prev_origin, prev_hashes in pdu.prev_pdus: + for alg, hash_base64 in prev_hashes.items(): + hash_bytes = decode_base64(hash_base64) + self._store_prev_pdu_hash_txn( + txn, pdu.pdu_id, pdu.origin, prev_pdu_id, prev_origin, alg, + hash_bytes + ) + if pdu.is_state: self._persist_state_txn(txn, pdu.prev_pdus, cols) else: @@ -352,6 +360,7 @@ class DataStore(RoomMemberStore, RoomStore, prev_pdus = self._get_latest_pdus_in_context( txn, room_id ) + if state_type is not None and state_key is not None: prev_state_pdu = self._get_current_state_pdu( txn, room_id, state_type, state_key @@ -401,17 +410,16 @@ class Snapshot(object): self.prev_state_pdu = prev_state_pdu def fill_out_prev_events(self, event): - if hasattr(event, "prev_events"): + if hasattr(event, "prev_pdus"): return - es = [ - "%s@%s" % (p_id, origin) for p_id, origin, _ in self.prev_pdus + event.prev_pdus = [ + (p_id, origin, hashes) + for p_id, origin, hashes, _ in self.prev_pdus ] - event.prev_events = [e for e in es if e != event.event_id] - if self.prev_pdus: - event.depth = max([int(v) for _, _, v in self.prev_pdus]) + 1 + event.depth = max([int(v) for _, _, _, v in self.prev_pdus]) + 1 else: event.depth = 0 diff --git a/synapse/storage/pdu.py b/synapse/storage/pdu.py index 9d624429b7..a423b42dbd 100644 --- a/synapse/storage/pdu.py +++ b/synapse/storage/pdu.py @@ -20,10 +20,13 @@ from ._base import SQLBaseStore, Table, JoinHelper from synapse.federation.units import Pdu from synapse.util.logutils import log_function +from syutil.base64util import encode_base64 + from collections import namedtuple import logging + logger = logging.getLogger(__name__) @@ -64,6 +67,8 @@ class PduStore(SQLBaseStore): for r in PduEdgesTable.decode_results(txn.fetchall()) ] + edge_hashes = self._get_prev_pdu_hashes_txn(txn, pdu_id, origin) + hashes = self._get_pdu_hashes_txn(txn, pdu_id, origin) signatures = self._get_pdu_origin_signatures_txn( txn, pdu_id, origin @@ -86,7 +91,7 @@ class PduStore(SQLBaseStore): row = txn.fetchone() if row: results.append(PduTuple( - PduEntry(*row), edges, hashes, signatures + PduEntry(*row), edges, hashes, signatures, edge_hashes )) return results @@ -310,9 +315,14 @@ class PduStore(SQLBaseStore): (context, ) ) - results = txn.fetchall() + results = [] + for pdu_id, origin, depth in txn.fetchall(): + hashes = self._get_pdu_hashes_txn(txn, pdu_id, origin) + sha256_bytes = hashes["sha256"] + prev_hashes = {"sha256": encode_base64(sha256_bytes)} + results.append((pdu_id, origin, prev_hashes, depth)) - return [(row[0], row[1], row[2]) for row in results] + return results @defer.inlineCallbacks def get_oldest_pdus_in_context(self, context): @@ -431,7 +441,7 @@ class PduStore(SQLBaseStore): "DELETE FROM %s WHERE pdu_id = ? AND origin = ?" % PduForwardExtremitiesTable.table_name ) - txn.executemany(query, prev_pdus) + txn.executemany(query, list(p[:2] for p in prev_pdus)) # We only insert as a forward extremety the new pdu if there are no # other pdus that reference it as a prev pdu @@ -454,7 +464,7 @@ class PduStore(SQLBaseStore): # deleted in a second if they're incorrect anyway. txn.executemany( PduBackwardExtremitiesTable.insert_statement(), - [(i, o, context) for i, o in prev_pdus] + [(i, o, context) for i, o, _ in prev_pdus] ) # Also delete from the backwards extremities table all ones that @@ -915,7 +925,7 @@ This does not include a prev_pdus key. PduTuple = namedtuple( "PduTuple", - ("pdu_entry", "prev_pdu_list", "hashes", "signatures") + ("pdu_entry", "prev_pdu_list", "hashes", "signatures", "edge_hashes") ) """ This is a tuple of a `PduEntry` and a list of `PduIdTuple` that represent the `prev_pdus` key of a PDU. diff --git a/synapse/storage/schema/signatures.sql b/synapse/storage/schema/signatures.sql index 86ee0f2377..a72c4dc35f 100644 --- a/synapse/storage/schema/signatures.sql +++ b/synapse/storage/schema/signatures.sql @@ -34,3 +34,19 @@ CREATE TABLE IF NOT EXISTS pdu_origin_signatures ( CREATE INDEX IF NOT EXISTS pdu_origin_signatures_id ON pdu_origin_signatures ( pdu_id, origin ); + +CREATE TABLE IF NOT EXISTS pdu_edge_hashes( + pdu_id TEXT, + origin TEXT, + prev_pdu_id TEXT, + prev_origin TEXT, + algorithm TEXT, + hash BLOB, + CONSTRAINT uniqueness UNIQUE ( + pdu_id, origin, prev_pdu_id, prev_origin, algorithm + ) +); + +CREATE INDEX IF NOT EXISTS pdu_edge_hashes_id ON pdu_edge_hashes( + pdu_id, origin +); diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index 1f0a680500..1147102489 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.py @@ -88,3 +88,34 @@ class SignatureStore(SQLBaseStore): "signature": buffer(signature_bytes), }) + def _get_prev_pdu_hashes_txn(self, txn, pdu_id, origin): + """Get all the hashes for previous PDUs of a PDU + Args: + txn (cursor): + pdu_id (str): Id of the PDU. + origin (str): Origin of the PDU. + Returns: + dict of (pdu_id, origin) -> dict of algorithm -> hash_bytes. + """ + query = ( + "SELECT prev_pdu_id, prev_origin, algorithm, hash" + " FROM pdu_edge_hashes" + " WHERE pdu_id = ? and origin = ?" + ) + txn.execute(query, (pdu_id, origin)) + results = {} + for prev_pdu_id, prev_origin, algorithm, hash_bytes in txn.fetchall(): + hashes = results.setdefault((prev_pdu_id, prev_origin), {}) + hashes[algorithm] = hash_bytes + return results + + def _store_prev_pdu_hash_txn(self, txn, pdu_id, origin, prev_pdu_id, + prev_origin, algorithm, hash_bytes): + self._simple_insert_txn(txn, "pdu_edge_hashes", { + "pdu_id": pdu_id, + "origin": origin, + "prev_pdu_id": prev_pdu_id, + "prev_origin": prev_origin, + "algorithm": algorithm, + "hash": buffer(hash_bytes), + }) diff --git a/tests/federation/test_federation.py b/tests/federation/test_federation.py index 03b2167cf7..eed50e6335 100644 --- a/tests/federation/test_federation.py +++ b/tests/federation/test_federation.py @@ -41,7 +41,7 @@ def make_pdu(prev_pdus=[], **kwargs): } pdu_fields.update(kwargs) - return PduTuple(PduEntry(**pdu_fields), prev_pdus, {}, {}) + return PduTuple(PduEntry(**pdu_fields), prev_pdus, {}, {}, {}) class FederationTestCase(unittest.TestCase): diff --git a/tests/federation/test_pdu_codec.py b/tests/federation/test_pdu_codec.py index 80851a4258..0ad8cf6641 100644 --- a/tests/federation/test_pdu_codec.py +++ b/tests/federation/test_pdu_codec.py @@ -88,7 +88,7 @@ class PduCodecTestCase(unittest.TestCase): self.assertEquals(pdu.context, event.room_id) self.assertEquals(pdu.is_state, event.is_state) self.assertEquals(pdu.depth, event.depth) - self.assertEquals(["alice@bob.com"], event.prev_events) + self.assertEquals(pdu.prev_pdus, event.prev_pdus) self.assertEquals(pdu.content, event.content) def test_pdu_from_event(self): @@ -144,7 +144,7 @@ class PduCodecTestCase(unittest.TestCase): self.assertEquals(pdu.context, event.room_id) self.assertEquals(pdu.is_state, event.is_state) self.assertEquals(pdu.depth, event.depth) - self.assertEquals(["alice@bob.com"], event.prev_events) + self.assertEquals(pdu.prev_pdus, event.prev_pdus) self.assertEquals(pdu.content, event.content) self.assertEquals(pdu.state_key, event.state_key) -- cgit 1.5.1 From f71627567b4aa58c5aba7e79c6d972b8ac26b449 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 17 Oct 2014 15:04:17 +0100 Subject: Finish implementing the new join dance. --- synapse/api/auth.py | 9 ++ synapse/api/events/factory.py | 14 ++- synapse/federation/replication.py | 66 ++++++++++---- synapse/federation/transport.py | 68 ++++++++++++-- synapse/handlers/federation.py | 181 ++++++++++++++++++-------------------- synapse/state.py | 10 ++- 6 files changed, 222 insertions(+), 126 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index 12ddef1b00..d1eca791ab 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -48,6 +48,15 @@ class Auth(object): """ try: if hasattr(event, "room_id"): + if not event.old_state_events: + # Oh, we don't know what the state of the room was, so we + # are trusting that this is allowed (at least for now) + defer.returnValue(True) + + if hasattr(event, "outlier") and event.outlier: + # TODO (erikj): Auth for outliers is done differently. + defer.returnValue(True) + is_state = hasattr(event, "state_key") if event.type == RoomMemberEvent.TYPE: diff --git a/synapse/api/events/factory.py b/synapse/api/events/factory.py index 0d94850cec..c6d1151cac 100644 --- a/synapse/api/events/factory.py +++ b/synapse/api/events/factory.py @@ -51,12 +51,20 @@ class EventFactory(object): self.clock = hs.get_clock() self.hs = hs + self.event_id_count = 0 + + def create_event_id(self): + i = str(self.event_id_count) + self.event_id_count += 1 + + local_part = str(int(self.clock.time())) + i + random_string(5) + + return "%s@%s" % (local_part, self.hs.hostname) + def create_event(self, etype=None, **kwargs): kwargs["type"] = etype if "event_id" not in kwargs: - kwargs["event_id"] = "%s@%s" % ( - random_string(10), self.hs.hostname - ) + kwargs["event_id"] = self.create_event_id() if "ts" not in kwargs: kwargs["ts"] = int(self.clock.time_msec()) diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 08c29dece5..d482193851 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -244,13 +244,14 @@ class ReplicationLayer(object): pdu = None if pdu_list: pdu = pdu_list[0] - yield self._handle_new_pdu(pdu) + yield self._handle_new_pdu(destination, pdu) defer.returnValue(pdu) @defer.inlineCallbacks @log_function - def get_state_for_context(self, destination, context): + def get_state_for_context(self, destination, context, pdu_id=None, + pdu_origin=None): """Requests all of the `current` state PDUs for a given context from a remote home server. @@ -263,13 +264,14 @@ class ReplicationLayer(object): """ transaction_data = yield self.transport_layer.get_context_state( - destination, context) + destination, context, pdu_id=pdu_id, pdu_origin=pdu_origin, + ) transaction = Transaction(**transaction_data) pdus = [Pdu(outlier=True, **p) for p in transaction.pdus] for pdu in pdus: - yield self._handle_new_pdu(pdu) + yield self._handle_new_pdu(destination, pdu) defer.returnValue(pdus) @@ -315,7 +317,7 @@ class ReplicationLayer(object): dl = [] for pdu in pdu_list: - dl.append(self._handle_new_pdu(pdu)) + dl.append(self._handle_new_pdu(transaction.origin, pdu)) if hasattr(transaction, "edus"): for edu in [Edu(**x) for x in transaction.edus]: @@ -347,14 +349,19 @@ class ReplicationLayer(object): @defer.inlineCallbacks @log_function - def on_context_state_request(self, context): - results = yield self.store.get_current_state_for_context( - context - ) + def on_context_state_request(self, context, pdu_id, pdu_origin): + if pdu_id and pdu_origin: + pdus = yield self.handler.get_state_for_pdu( + pdu_id, pdu_origin + ) + else: + results = yield self.store.get_current_state_for_context( + context + ) + pdus = [Pdu.from_pdu_tuple(p) for p in results] - logger.debug("Context returning %d results", len(results)) + logger.debug("Context returning %d results", len(pdus)) - pdus = [Pdu.from_pdu_tuple(p) for p in results] defer.returnValue((200, self._transaction_from_pdus(pdus).get_dict())) @defer.inlineCallbacks @@ -396,9 +403,10 @@ class ReplicationLayer(object): defer.returnValue( (404, "No handler for Query type '%s'" % (query_type, )) ) - + @defer.inlineCallbacks def on_make_join_request(self, context, user_id): - return self.handler.on_make_join_request(context, user_id) + pdu = yield self.handler.on_make_join_request(context, user_id) + defer.returnValue(pdu.get_dict()) @defer.inlineCallbacks def on_send_join_request(self, origin, content): @@ -406,13 +414,27 @@ class ReplicationLayer(object): state = yield self.handler.on_send_join_request(origin, pdu) defer.returnValue((200, self._transaction_from_pdus(state).get_dict())) + @defer.inlineCallbacks def make_join(self, destination, context, user_id): - return self.transport_layer.make_join( + pdu_dict = yield self.transport_layer.make_join( destination=destination, context=context, user_id=user_id, ) + logger.debug("Got response to make_join: %s", pdu_dict) + + defer.returnValue(Pdu(**pdu_dict)) + + def send_join(self, destination, pdu): + return self.transport_layer.send_join( + destination, + pdu.context, + pdu.pdu_id, + pdu.origin, + pdu.get_dict(), + ) + @defer.inlineCallbacks @log_function def _get_persisted_pdu(self, pdu_id, pdu_origin): @@ -443,7 +465,7 @@ class ReplicationLayer(object): @defer.inlineCallbacks @log_function - def _handle_new_pdu(self, pdu, backfilled=False): + def _handle_new_pdu(self, origin, pdu, backfilled=False): # We reprocess pdus when we have seen them only as outliers existing = yield self._get_persisted_pdu(pdu.pdu_id, pdu.origin) @@ -452,6 +474,8 @@ class ReplicationLayer(object): defer.returnValue({}) return + state = None + # Get missing pdus if necessary. is_new = yield self.pdu_actions.is_new(pdu) if is_new and not pdu.outlier: @@ -475,12 +499,22 @@ class ReplicationLayer(object): except: # TODO(erikj): Do some more intelligent retries. logger.exception("Failed to get PDU") + else: + # We need to get the state at this event, since we have reached + # a backward extremity edge. + state = yield self.get_state_for_context( + origin, pdu.context, pdu.pdu_id, pdu.origin, + ) # Persist the Pdu, but don't mark it as processed yet. yield self.store.persist_event(pdu=pdu) if not backfilled: - ret = yield self.handler.on_receive_pdu(pdu, backfilled=backfilled) + ret = yield self.handler.on_receive_pdu( + pdu, + backfilled=backfilled, + state=state, + ) else: ret = None diff --git a/synapse/federation/transport.py b/synapse/federation/transport.py index 4f552272e6..a0d34fd24d 100644 --- a/synapse/federation/transport.py +++ b/synapse/federation/transport.py @@ -72,7 +72,8 @@ class TransportLayer(object): self.received_handler = None @log_function - def get_context_state(self, destination, context): + def get_context_state(self, destination, context, pdu_id=None, + pdu_origin=None): """ Requests all state for a given context (i.e. room) from the given server. @@ -89,7 +90,14 @@ class TransportLayer(object): subpath = "/state/%s/" % context - return self._do_request_for_transaction(destination, subpath) + args = {} + if pdu_id and pdu_origin: + args["pdu_id"] = pdu_id + args["pdu_origin"] = pdu_origin + + return self._do_request_for_transaction( + destination, subpath, args=args + ) @log_function def get_pdu(self, destination, pdu_origin, pdu_id): @@ -135,8 +143,10 @@ class TransportLayer(object): subpath = "/backfill/%s/" % context - args = {"v": ["%s,%s" % (i, o) for i, o in pdu_tuples]} - args["limit"] = limit + args = { + "v": ["%s,%s" % (i, o) for i, o in pdu_tuples], + "limit": limit, + } return self._do_request_for_transaction( dest, @@ -210,6 +220,23 @@ class TransportLayer(object): defer.returnValue(response) + @defer.inlineCallbacks + @log_function + def send_join(self, destination, context, pdu_id, origin, content): + path = PREFIX + "/send_join/%s/%s/%s" % ( + context, + origin, + pdu_id, + ) + + response = yield self.client.put_json( + destination=destination, + path=path, + data=content, + ) + + defer.returnValue(response) + @defer.inlineCallbacks def _authenticate_request(self, request): json_request = { @@ -330,7 +357,11 @@ class TransportLayer(object): re.compile("^" + PREFIX + "/state/([^/]*)/$"), self._with_authentication( lambda origin, content, query, context: - handler.on_context_state_request(context) + handler.on_context_state_request( + context, + query.get("pdu_id", [None])[0], + query.get("pdu_origin", [None])[0] + ) ) ) @@ -369,7 +400,23 @@ class TransportLayer(object): self.server.register_path( "GET", re.compile("^" + PREFIX + "/make_join/([^/]*)/([^/]*)$"), - self._on_make_join_request + self._with_authentication( + lambda origin, content, query, context, user_id: + self._on_make_join_request( + origin, content, query, context, user_id + ) + ) + ) + + self.server.register_path( + "PUT", + re.compile("^" + PREFIX + "/send_join/([^/]*)/([^/]*)/([^/]*)$"), + self._with_authentication( + lambda origin, content, query, context, pdu_origin, pdu_id: + self._on_send_join_request( + origin, content, query, + ) + ) ) @defer.inlineCallbacks @@ -460,18 +507,23 @@ class TransportLayer(object): context, versions, limit ) + @defer.inlineCallbacks @log_function def _on_make_join_request(self, origin, content, query, context, user_id): - return self.request_handler.on_make_join_request( + content = yield self.request_handler.on_make_join_request( context, user_id, ) + defer.returnValue((200, content)) + @defer.inlineCallbacks @log_function def _on_send_join_request(self, origin, content, query): - return self.request_handler.on_send_join_request( + content = yield self.request_handler.on_send_join_request( origin, content, ) + defer.returnValue((200, content)) + class TransportReceivedHandler(object): """ Callbacks used when we receive a transaction diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index a4f6c739c3..0ae0541bd3 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -20,7 +20,7 @@ from ._base import BaseHandler from synapse.api.events.room import InviteJoinEvent, RoomMemberEvent from synapse.api.constants import Membership from synapse.util.logutils import log_function -from synapse.federation.pdu_codec import PduCodec +from synapse.federation.pdu_codec import PduCodec, encode_event_id from synapse.api.errors import SynapseError from twisted.internet import defer, reactor @@ -87,7 +87,7 @@ class FederationHandler(BaseHandler): @log_function @defer.inlineCallbacks - def on_receive_pdu(self, pdu, backfilled): + def on_receive_pdu(self, pdu, backfilled, state=None): """ Called by the ReplicationLayer when we have a new pdu. We need to do auth checks and put it through the StateHandler. """ @@ -95,7 +95,10 @@ class FederationHandler(BaseHandler): logger.debug("Got event: %s", event.event_id) - yield self.state_handler.annotate_state_groups(event) + if state: + state = [self.pdu_codec.event_from_pdu(p) for p in state] + state = {(e.type, e.state_key): e for e in state} + yield self.state_handler.annotate_state_groups(event, state=state) logger.debug("Event: %s", event) @@ -108,83 +111,55 @@ class FederationHandler(BaseHandler): ) else: is_new_state = False + # TODO: Implement something in federation that allows us to # respond to PDU. - target_is_mine = False - if hasattr(event, "target_host"): - target_is_mine = event.target_host == self.hs.hostname - - if event.type == InviteJoinEvent.TYPE: - if not target_is_mine: - logger.debug("Ignoring invite/join event %s", event) - return - - # If we receive an invite/join event then we need to join the - # sender to the given room. - # TODO: We should probably auth this or some such - content = event.content - content.update({"membership": Membership.JOIN}) - new_event = self.event_factory.create_event( - etype=RoomMemberEvent.TYPE, - state_key=event.user_id, - room_id=event.room_id, - user_id=event.user_id, - membership=Membership.JOIN, - content=content + with (yield self.room_lock.lock(event.room_id)): + yield self.store.persist_event( + event, + backfilled, + is_new_state=is_new_state ) - yield self.hs.get_handlers().room_member_handler.change_membership( - new_event, - do_auth=False, - ) + room = yield self.store.get_room(event.room_id) - else: - with (yield self.room_lock.lock(event.room_id)): - yield self.store.persist_event( - event, - backfilled, - is_new_state=is_new_state + if not room: + # Huh, let's try and get the current state + try: + yield self.replication_layer.get_state_for_context( + event.origin, event.room_id, pdu.pdu_id, pdu.origin, ) - room = yield self.store.get_room(event.room_id) - - if not room: - # Huh, let's try and get the current state - try: - yield self.replication_layer.get_state_for_context( - event.origin, event.room_id - ) - - hosts = yield self.store.get_joined_hosts_for_room( - event.room_id - ) - if self.hs.hostname in hosts: - try: - yield self.store.store_room( - room_id=event.room_id, - room_creator_user_id="", - is_public=False, - ) - except: - pass - except: - logger.exception( - "Failed to get current state for room %s", - event.room_id - ) - - if not backfilled: - extra_users = [] - if event.type == RoomMemberEvent.TYPE: - target_user_id = event.state_key - target_user = self.hs.parse_userid(target_user_id) - extra_users.append(target_user) - - yield self.notifier.on_new_room_event( - event, extra_users=extra_users + hosts = yield self.store.get_joined_hosts_for_room( + event.room_id + ) + if self.hs.hostname in hosts: + try: + yield self.store.store_room( + room_id=event.room_id, + room_creator_user_id="", + is_public=False, + ) + except: + pass + except: + logger.exception( + "Failed to get current state for room %s", + event.room_id ) + if not backfilled: + extra_users = [] + if event.type == RoomMemberEvent.TYPE: + target_user_id = event.state_key + target_user = self.hs.parse_userid(target_user_id) + extra_users.append(target_user) + + yield self.notifier.on_new_room_event( + event, extra_users=extra_users + ) + if event.type == RoomMemberEvent.TYPE: if event.membership == Membership.JOIN: user = self.hs.parse_userid(event.state_key) @@ -214,40 +189,35 @@ class FederationHandler(BaseHandler): @log_function @defer.inlineCallbacks def do_invite_join(self, target_host, room_id, joinee, content, snapshot): - hosts = yield self.store.get_joined_hosts_for_room(room_id) if self.hs.hostname in hosts: # We are already in the room. logger.debug("We're already in the room apparently") defer.returnValue(False) - # First get current state to see if we are already joined. - try: - yield self.replication_layer.get_state_for_context( - target_host, room_id - ) - - hosts = yield self.store.get_joined_hosts_for_room(room_id) - if self.hs.hostname in hosts: - # Oh, we were actually in the room already. - logger.debug("We're already in the room apparently") - defer.returnValue(False) - except Exception: - logger.exception("Failed to get current state") - - new_event = self.event_factory.create_event( - etype=InviteJoinEvent.TYPE, - target_host=target_host, - room_id=room_id, - user_id=joinee, - content=content + pdu = yield self.replication_layer.make_join( + target_host, + room_id, + joinee ) - new_event.destinations = [target_host] + logger.debug("Got response to make_join: %s", pdu) - snapshot.fill_out_prev_events(new_event) - yield self.state_handler.annotate_state_groups(new_event) - yield self.handle_new_event(new_event, snapshot) + event = self.pdu_codec.event_from_pdu(pdu) + + # We should assert some things. + assert(event.type == RoomMemberEvent.TYPE) + assert(event.user_id == joinee) + assert(event.state_key == joinee) + assert(event.room_id == room_id) + + event.event_id = self.event_factory.create_event_id() + event.content = content + + state = yield self.replication_layer.send_join( + target_host, + self.pdu_codec.pdu_from_event(event) + ) # TODO (erikj): Time out here. d = defer.Deferred() @@ -326,14 +296,31 @@ class FederationHandler(BaseHandler): "user_joined_room", user=user, room_id=event.room_id ) - pdu.destinations = yield self.store.get_joined_hosts_for_room( + new_pdu = self.pdu_codec.pdu_from_event(event); + new_pdu.destinations = yield self.store.get_joined_hosts_for_room( event.room_id ) - yield self.replication_layer.send_pdu(pdu) + yield self.replication_layer.send_pdu(new_pdu) defer.returnValue(event.state_events.values()) + @defer.inlineCallbacks + def get_state_for_pdu(self, pdu_id, pdu_origin): + state_groups = yield self.store.get_state_groups( + [encode_event_id(pdu_id, pdu_origin)] + ) + + if state_groups: + defer.returnValue( + [ + self.pdu_codec.pdu_from_event(s) + for s in state_groups[0].state + ] + ) + else: + defer.returnValue([]) + @log_function def _on_user_joined(self, user, room_id): waiters = self.waiting_for_join_list.get((user.to_string(), room_id), []) diff --git a/synapse/state.py b/synapse/state.py index 9be6b716e2..8c4eeb8924 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -130,7 +130,13 @@ class StateHandler(object): defer.returnValue(is_new) @defer.inlineCallbacks - def annotate_state_groups(self, event): + def annotate_state_groups(self, event, state=None): + if state: + event.state_group = None + event.old_state_events = None + event.state_events = state + return + state_groups = yield self.store.get_state_groups( event.prev_events ) @@ -177,7 +183,7 @@ class StateHandler(object): new_powers_deferreds = [] for e in curr_events: new_powers_deferreds.append( - self.store.get_power_level(e.context, e.user_id) + self.store.get_power_level(e.room_id, e.user_id) ) new_powers = yield defer.gatherResults( -- cgit 1.5.1 From 2d1dfb3b34583a4de7e1e53f685c2564a7fc731f Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 28 Oct 2014 16:42:35 +0000 Subject: Begin implementing all the PDU storage stuff in Events land --- synapse/api/events/__init__.py | 4 +- synapse/federation/pdu_codec.py | 11 ++- synapse/storage/__init__.py | 72 ++++++++++---- synapse/storage/_base.py | 53 +++++++---- synapse/storage/event_federation.py | 143 ++++++++++++++++++++++++++++ synapse/storage/schema/event_edges.sql | 51 ++++++++++ synapse/storage/schema/event_signatures.sql | 65 +++++++++++++ synapse/storage/schema/im.sql | 1 + synapse/storage/signatures.py | 127 ++++++++++++++++++++++++ 9 files changed, 485 insertions(+), 42 deletions(-) create mode 100644 synapse/storage/event_federation.py create mode 100644 synapse/storage/schema/event_edges.sql create mode 100644 synapse/storage/schema/event_signatures.sql (limited to 'synapse/api/events') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index a5a55742e0..b855811b98 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -71,7 +71,9 @@ class SynapseEvent(JsonEncodedObject): "outlier", "power_level", "redacted", - "prev_pdus", + "prev_events", + "hashes", + "signatures", ] required_keys = [ diff --git a/synapse/federation/pdu_codec.py b/synapse/federation/pdu_codec.py index 991aae2a56..2cd591410b 100644 --- a/synapse/federation/pdu_codec.py +++ b/synapse/federation/pdu_codec.py @@ -47,7 +47,10 @@ class PduCodec(object): kwargs["event_id"] = encode_event_id(pdu.pdu_id, pdu.origin) kwargs["room_id"] = pdu.context kwargs["etype"] = pdu.pdu_type - kwargs["prev_pdus"] = pdu.prev_pdus + kwargs["prev_events"] = [ + encode_event_id(i, o) + for i, o in pdu.prev_pdus + ] if hasattr(pdu, "prev_state_id") and hasattr(pdu, "prev_state_origin"): kwargs["prev_state"] = encode_event_id( @@ -78,8 +81,8 @@ class PduCodec(object): d["context"] = event.room_id d["pdu_type"] = event.type - if hasattr(event, "prev_pdus"): - d["prev_pdus"] = event.prev_pdus + if hasattr(event, "prev_events"): + d["prev_pdus"] = [decode_event_id(e) for e in event.prev_events] if hasattr(event, "prev_state"): d["prev_state_id"], d["prev_state_origin"] = ( @@ -92,7 +95,7 @@ class PduCodec(object): kwargs = copy.deepcopy(event.unrecognized_keys) kwargs.update({ k: v for k, v in d.items() - if k not in ["event_id", "room_id", "type"] + if k not in ["event_id", "room_id", "type", "prev_events"] }) if "origin_server_ts" not in kwargs: diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index a50e19349a..678de0cf50 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -40,6 +40,7 @@ from .stream import StreamStore from .pdu import StatePduStore, PduStore, PdusTable from .transactions import TransactionStore from .keys import KeyStore +from .event_federation import EventFederationStore from .state import StateStore from .signatures import SignatureStore @@ -69,6 +70,7 @@ SCHEMAS = [ "redactions", "state", "signatures", + "event_edges", ] @@ -83,10 +85,12 @@ class _RollbackButIsFineException(Exception): """ pass + class DataStore(RoomMemberStore, RoomStore, RegistrationStore, StreamStore, ProfileStore, FeedbackStore, PresenceStore, PduStore, StatePduStore, TransactionStore, - DirectoryStore, KeyStore, StateStore, SignatureStore): + DirectoryStore, KeyStore, StateStore, SignatureStore, + EventFederationStore, ): def __init__(self, hs): super(DataStore, self).__init__(hs) @@ -230,6 +234,10 @@ class DataStore(RoomMemberStore, RoomStore, elif event.type == RoomRedactionEvent.TYPE: self._store_redaction(txn, event) + outlier = False + if hasattr(event, "outlier"): + outlier = event.outlier + vals = { "topological_ordering": event.depth, "event_id": event.event_id, @@ -237,20 +245,20 @@ class DataStore(RoomMemberStore, RoomStore, "room_id": event.room_id, "content": json.dumps(event.content), "processed": True, + "outlier": outlier, + "depth": event.depth, } if stream_ordering is not None: vals["stream_ordering"] = stream_ordering - if hasattr(event, "outlier"): - vals["outlier"] = event.outlier - else: - vals["outlier"] = False - unrec = { k: v for k, v in event.get_full_dict().items() - if k not in vals.keys() and k not in ["redacted", "redacted_because"] + if k not in vals.keys() and k not in [ + "redacted", "redacted_because", "signatures", "hashes", + "prev_events", + ] } vals["unrecognized_keys"] = json.dumps(unrec) @@ -264,6 +272,14 @@ class DataStore(RoomMemberStore, RoomStore, ) raise _RollbackButIsFineException("_persist_event") + self._handle_prev_events( + txn, + outlier=outlier, + event_id=event.event_id, + prev_events=event.prev_events, + room_id=event.room_id, + ) + self._store_state_groups_txn(txn, event) is_state = hasattr(event, "state_key") and event.state_key is not None @@ -291,6 +307,28 @@ class DataStore(RoomMemberStore, RoomStore, } ) + signatures = event.signatures.get(event.origin, {}) + + for key_id, signature_base64 in signatures.items(): + signature_bytes = decode_base64(signature_base64) + self._store_event_origin_signature_txn( + txn, event.event_id, key_id, signature_bytes, + ) + + for prev_event_id, prev_hashes in event.prev_events: + for alg, hash_base64 in prev_hashes.items(): + hash_bytes = decode_base64(hash_base64) + self._store_prev_event_hash_txn( + txn, event.event_id, prev_event_id, alg, hash_bytes + ) + + (ref_alg, ref_hash_bytes) = compute_pdu_event_reference_hash(pdu) + self._store_pdu_reference_hash_txn( + txn, pdu.pdu_id, pdu.origin, ref_alg, ref_hash_bytes + ) + + self._update_min_depth_for_room_txn(txn, event.room_id, event.depth) + def _store_redaction(self, txn, event): txn.execute( "INSERT OR IGNORE INTO redactions " @@ -373,7 +411,7 @@ class DataStore(RoomMemberStore, RoomStore, """ def _snapshot(txn): membership_state = self._get_room_member(txn, user_id, room_id) - prev_pdus = self._get_latest_pdus_in_context( + prev_events = self._get_latest_events_in_room( txn, room_id ) @@ -388,7 +426,7 @@ class DataStore(RoomMemberStore, RoomStore, store=self, room_id=room_id, user_id=user_id, - prev_pdus=prev_pdus, + prev_events=prev_events, membership_state=membership_state, state_type=state_type, state_key=state_key, @@ -404,7 +442,7 @@ class Snapshot(object): store (DataStore): The datastore. room_id (RoomId): The room of the snapshot. user_id (UserId): The user this snapshot is for. - prev_pdus (list): The list of PDU ids this snapshot is after. + prev_events (list): The list of event ids this snapshot is after. membership_state (RoomMemberEvent): The current state of the user in the room. state_type (str, optional): State type captured by the snapshot @@ -413,29 +451,29 @@ class Snapshot(object): the previous value of the state type and key in the room. """ - def __init__(self, store, room_id, user_id, prev_pdus, + def __init__(self, store, room_id, user_id, prev_events, membership_state, state_type=None, state_key=None, prev_state_pdu=None): self.store = store self.room_id = room_id self.user_id = user_id - self.prev_pdus = prev_pdus + self.prev_events = prev_events self.membership_state = membership_state self.state_type = state_type self.state_key = state_key self.prev_state_pdu = prev_state_pdu def fill_out_prev_events(self, event): - if hasattr(event, "prev_pdus"): + if hasattr(event, "prev_events"): return - event.prev_pdus = [ + event.prev_events = [ (p_id, origin, hashes) - for p_id, origin, hashes, _ in self.prev_pdus + for p_id, origin, hashes, _ in self.prev_events ] - if self.prev_pdus: - event.depth = max([int(v) for _, _, _, v in self.prev_pdus]) + 1 + if self.prev_events: + event.depth = max([int(v) for _, _, _, v in self.prev_events]) + 1 else: event.depth = 0 diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 1192216971..30732caa83 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -193,7 +193,6 @@ class SQLBaseStore(object): table, keyvalues, retcols=retcols, allow_none=allow_none ) - @defer.inlineCallbacks def _simple_select_one_onecol(self, table, keyvalues, retcol, allow_none=False): """Executes a SELECT query on the named table, which is expected to @@ -204,19 +203,41 @@ class SQLBaseStore(object): keyvalues : dict of column names and values to select the row with retcol : string giving the name of the column to return """ - ret = yield self._simple_select_one( + return self.runInteraction( + "_simple_select_one_onecol_txn", + self._simple_select_one_onecol_txn, + table, keyvalues, retcol, allow_none=allow_none, + ) + + def _simple_select_one_onecol_txn(self, txn, table, keyvalues, retcol, + allow_none=False): + ret = self._simple_select_onecol_txn( + txn, table=table, keyvalues=keyvalues, - retcols=[retcol], - allow_none=allow_none + retcols=retcol, ) if ret: - defer.returnValue(ret[retcol]) + return ret[retcol] else: - defer.returnValue(None) + if allow_none: + return None + else: + raise StoreError(404, "No row found") + + def _simple_select_onecol_txn(self, txn, table, keyvalues, retcol): + sql = "SELECT %(retcol)s FROM %(table)s WHERE %(where)s" % { + "retcol": retcol, + "table": table, + "where": " AND ".join("%s = ?" % k for k in keyvalues.keys()), + } + + txn.execute(sql, keyvalues.values()) + + return [r[0] for r in txn.fetchall()] + - @defer.inlineCallbacks def _simple_select_onecol(self, table, keyvalues, retcol): """Executes a SELECT query on the named table, which returns a list comprising of the values of the named column from the selected rows. @@ -229,19 +250,11 @@ class SQLBaseStore(object): Returns: Deferred: Results in a list """ - sql = "SELECT %(retcol)s FROM %(table)s WHERE %(where)s" % { - "retcol": retcol, - "table": table, - "where": " AND ".join("%s = ?" % k for k in keyvalues.keys()), - } - - def func(txn): - txn.execute(sql, keyvalues.values()) - return txn.fetchall() - - res = yield self.runInteraction("_simple_select_onecol", func) - - defer.returnValue([r[0] for r in res]) + return self.runInteraction( + "_simple_select_onecol", + self._simple_select_onecol_txn, + table, keyvalues, retcol + ) def _simple_select_list(self, table, keyvalues, retcols): """Executes a SELECT query on the named table, which may return zero or diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py new file mode 100644 index 0000000000..27ad9aea4d --- /dev/null +++ b/synapse/storage/event_federation.py @@ -0,0 +1,143 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 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. + +from ._base import SQLBaseStore +from twisted.internet import defer + +import logging + + +logger = logging.getLogger(__name__) + + +class EventFederationStore(SQLBaseStore): + + def _get_latest_events_in_room(self, txn, room_id): + self._simple_select_onecol_txn( + txn, + table="event_forward_extremities", + keyvalues={ + "room_id": room_id, + }, + retcol="event_id", + ) + + results = [] + for pdu_id, origin, depth in txn.fetchall(): + hashes = self._get_pdu_reference_hashes_txn(txn, pdu_id, origin) + sha256_bytes = hashes["sha256"] + prev_hashes = {"sha256": encode_base64(sha256_bytes)} + results.append((pdu_id, origin, prev_hashes, depth)) + + def _get_min_depth_interaction(self, txn, room_id): + min_depth = self._simple_select_one_onecol_txn( + txn, + table="room_depth", + keyvalues={"room_id": room_id,}, + retcol="min_depth", + allow_none=True, + ) + + return int(min_depth) if min_depth is not None else None + + def _update_min_depth_for_room_txn(self, txn, room_id, depth): + min_depth = self._get_min_depth_interaction(txn, room_id) + + do_insert = depth < min_depth if min_depth else True + + if do_insert: + self._simple_insert_txn( + txn, + table="room_depth", + values={ + "room_id": room_id, + "min_depth": depth, + }, + or_replace=True, + ) + + def _handle_prev_events(self, txn, outlier, event_id, prev_events, + room_id): + for e_id in prev_events: + # TODO (erikj): This could be done as a bulk insert + self._simple_insert_txn( + txn, + table="event_edges", + values={ + "event_id": event_id, + "prev_event": e_id, + "room_id": room_id, + } + ) + + # Update the extremities table if this is not an outlier. + if not outlier: + for e_id in prev_events: + # TODO (erikj): This could be done as a bulk insert + self._simple_delete_txn( + txn, + table="event_forward_extremities", + keyvalues={ + "event_id": e_id, + "room_id": room_id, + } + ) + + + + # We only insert as a forward extremity the new pdu if there are no + # other pdus that reference it as a prev pdu + query = ( + "INSERT INTO %(table)s (event_id, room_id) " + "SELECT ?, ? WHERE NOT EXISTS (" + "SELECT 1 FROM %(event_edges)s WHERE " + "prev_event_id = ? " + ")" + ) % { + "table": "event_forward_extremities", + "event_edges": "event_edges", + } + + logger.debug("query: %s", query) + + txn.execute(query, (event_id, room_id, event_id)) + + # Insert all the prev_pdus as a backwards thing, they'll get + # deleted in a second if they're incorrect anyway. + for e_id in prev_events: + # TODO (erikj): This could be done as a bulk insert + self._simple_insert_txn( + txn, + table="event_backward_extremities", + values={ + "event_id": e_id, + "room_id": room_id, + } + ) + + # Also delete from the backwards extremities table all ones that + # reference pdus that we have already seen + query = ( + "DELETE FROM %(event_back)s as b WHERE EXISTS (" + "SELECT 1 FROM %(events)s AS events " + "WHERE " + "b.event_id = events.event_id " + "AND not events.outlier " + ")" + ) % { + "event_back": "event_backward_extremities", + "events": "events", + } + txn.execute(query) \ No newline at end of file diff --git a/synapse/storage/schema/event_edges.sql b/synapse/storage/schema/event_edges.sql new file mode 100644 index 0000000000..6a28314ece --- /dev/null +++ b/synapse/storage/schema/event_edges.sql @@ -0,0 +1,51 @@ + +CREATE TABLE IF NOT EXISTS event_forward_extremities( + event_id TEXT, + room_id TEXT, + CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE +); + +CREATE INDEX IF NOT EXISTS ev_extrem_room ON event_forward_extremities(room_id); +CREATE INDEX IF NOT EXISTS ev_extrem_id ON event_forward_extremities(event_id); +-- + +CREATE TABLE IF NOT EXISTS event_backward_extremities( + event_id TEXT, + room_id TEXT, + CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE +); + +CREATE INDEX IF NOT EXISTS ev_b_extrem_room ON event_backward_extremities(room_id); +CREATE INDEX IF NOT EXISTS ev_b_extrem_id ON event_backward_extremities(event_id); +-- + +CREATE TABLE IF NOT EXISTS event_edges( + event_id TEXT, + prev_event_id TEXT, + room_id TEXT, + CONSTRAINT uniqueness UNIQUE (event_id, prev_event_id, room_id) +); + +CREATE INDEX IF NOT EXISTS ev_edges_id ON event_edges(event_id); +CREATE INDEX IF NOT EXISTS ev_edges_prev_id ON event_edges(prev_event_id); +-- + + +CREATE TABLE IF NOT EXISTS room_depth( + room_id TEXT, + min_depth INTEGER, + CONSTRAINT uniqueness UNIQUE (room_id) +); + +CREATE INDEX IF NOT EXISTS room_depth_room ON room_depth(room_id); +-- + +create TABLE IF NOT EXISTS event_destinations( + event_id TEXT, + destination TEXT, + delivered_ts INTEGER DEFAULT 0, -- or 0 if not delivered + CONSTRAINT uniqueness UNIQUE (event_id, destination) ON CONFLICT REPLACE +); + +CREATE INDEX IF NOT EXISTS event_destinations_id ON event_destinations(event_id); +-- \ No newline at end of file diff --git a/synapse/storage/schema/event_signatures.sql b/synapse/storage/schema/event_signatures.sql new file mode 100644 index 0000000000..5491c7ecec --- /dev/null +++ b/synapse/storage/schema/event_signatures.sql @@ -0,0 +1,65 @@ +/* Copyright 2014 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. + */ + +CREATE TABLE IF NOT EXISTS event_content_hashes ( + event_id TEXT, + algorithm TEXT, + hash BLOB, + CONSTRAINT uniqueness UNIQUE (event_id, algorithm) +); + +CREATE INDEX IF NOT EXISTS event_content_hashes_id ON event_content_hashes( + event_id +); + + +CREATE TABLE IF NOT EXISTS event_reference_hashes ( + event_id TEXT, + algorithm TEXT, + hash BLOB, + CONSTRAINT uniqueness UNIQUE (event_id, algorithm) +); + +CREATE INDEX IF NOT EXISTS event_reference_hashes_id ON event_reference_hashes ( + event_id +); + + +CREATE TABLE IF NOT EXISTS event_origin_signatures ( + event_id TEXT, + origin TEXT, + key_id TEXT, + signature BLOB, + CONSTRAINT uniqueness UNIQUE (event_id, key_id) +); + +CREATE INDEX IF NOT EXISTS event_origin_signatures_id ON event_origin_signatures ( + event_id +); + + +CREATE TABLE IF NOT EXISTS event_edge_hashes( + event_id TEXT, + prev_event_id TEXT, + algorithm TEXT, + hash BLOB, + CONSTRAINT uniqueness UNIQUE ( + event_id, prev_event_id, algorithm + ) +); + +CREATE INDEX IF NOT EXISTS event_edge_hashes_id ON event_edge_hashes( + event_id +); diff --git a/synapse/storage/schema/im.sql b/synapse/storage/schema/im.sql index 3aa83f5c8c..8d6f655993 100644 --- a/synapse/storage/schema/im.sql +++ b/synapse/storage/schema/im.sql @@ -23,6 +23,7 @@ CREATE TABLE IF NOT EXISTS events( unrecognized_keys TEXT, processed BOOL NOT NULL, outlier BOOL NOT NULL, + depth INTEGER DEFAULT 0 NOT NULL, CONSTRAINT ev_uniq UNIQUE (event_id) ); diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index 82be946d3f..b8f8fd44cb 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.py @@ -153,3 +153,130 @@ class SignatureStore(SQLBaseStore): "algorithm": algorithm, "hash": buffer(hash_bytes), }) + + ## Events ## + + def _get_event_content_hashes_txn(self, txn, event_id): + """Get all the hashes for a given Event. + Args: + txn (cursor): + event_id (str): Id for the Event. + Returns: + A dict of algorithm -> hash. + """ + query = ( + "SELECT algorithm, hash" + " FROM event_content_hashes" + " WHERE event_id = ?" + ) + txn.execute(query, (event_id, )) + return dict(txn.fetchall()) + + def _store_event_content_hash_txn(self, txn, event_id, algorithm, + hash_bytes): + """Store a hash for a Event + Args: + txn (cursor): + event_id (str): Id for the Event. + algorithm (str): Hashing algorithm. + hash_bytes (bytes): Hash function output bytes. + """ + self._simple_insert_txn(txn, "event_content_hashes", { + "event_id": event_id, + "algorithm": algorithm, + "hash": buffer(hash_bytes), + }) + + def _get_event_reference_hashes_txn(self, txn, event_id): + """Get all the hashes for a given PDU. + Args: + txn (cursor): + event_id (str): Id for the Event. + Returns: + A dict of algorithm -> hash. + """ + query = ( + "SELECT algorithm, hash" + " FROM event_reference_hashes" + " WHERE event_id = ?" + ) + txn.execute(query, (event_id, )) + return dict(txn.fetchall()) + + def _store_event_reference_hash_txn(self, txn, event_id, algorithm, + hash_bytes): + """Store a hash for a PDU + Args: + txn (cursor): + event_id (str): Id for the Event. + algorithm (str): Hashing algorithm. + hash_bytes (bytes): Hash function output bytes. + """ + self._simple_insert_txn(txn, "event_reference_hashes", { + "event_id": event_id, + "algorithm": algorithm, + "hash": buffer(hash_bytes), + }) + + + def _get_event_origin_signatures_txn(self, txn, event_id): + """Get all the signatures for a given PDU. + Args: + txn (cursor): + event_id (str): Id for the Event. + Returns: + A dict of key_id -> signature_bytes. + """ + query = ( + "SELECT key_id, signature" + " FROM event_origin_signatures" + " WHERE event_id = ? " + ) + txn.execute(query, (event_id, )) + return dict(txn.fetchall()) + + def _store_event_origin_signature_txn(self, txn, event_id, origin, key_id, + signature_bytes): + """Store a signature from the origin server for a PDU. + Args: + txn (cursor): + event_id (str): Id for the Event. + origin (str): origin of the Event. + key_id (str): Id for the signing key. + signature (bytes): The signature. + """ + self._simple_insert_txn(txn, "event_origin_signatures", { + "event_id": event_id, + "origin": origin, + "key_id": key_id, + "signature": buffer(signature_bytes), + }) + + def _get_prev_event_hashes_txn(self, txn, event_id): + """Get all the hashes for previous PDUs of a PDU + Args: + txn (cursor): + event_id (str): Id for the Event. + Returns: + dict of (pdu_id, origin) -> dict of algorithm -> hash_bytes. + """ + query = ( + "SELECT prev_event_id, algorithm, hash" + " FROM event_edge_hashes" + " WHERE event_id = ?" + ) + txn.execute(query, (event_id, )) + results = {} + for prev_event_id, algorithm, hash_bytes in txn.fetchall(): + hashes = results.setdefault(prev_event_id, {}) + hashes[algorithm] = hash_bytes + return results + + def _store_prev_event_hash_txn(self, txn, event_id, prev_event_id, + algorithm, hash_bytes): + self._simple_insert_txn(txn, "event_edge_hashes", { + "event_id": event_id, + "prev_event_id": prev_event_id, + "algorithm": algorithm, + "hash": buffer(hash_bytes), + }) \ No newline at end of file -- cgit 1.5.1 From 12ce441e67a40bb73ac5aca0283c2fe4afac4021 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Oct 2014 17:00:11 +0000 Subject: Convert event ids to be of the form :example.com --- synapse/api/events/factory.py | 6 +++++- synapse/federation/pdu_codec.py | 35 ++++++++++++++++------------------- synapse/handlers/federation.py | 7 +++++-- synapse/server.py | 7 ++++++- synapse/state.py | 17 ++++++++++++----- synapse/types.py | 10 ++++++++++ 6 files changed, 54 insertions(+), 28 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/events/factory.py b/synapse/api/events/factory.py index 06f3bf232b..750096c618 100644 --- a/synapse/api/events/factory.py +++ b/synapse/api/events/factory.py @@ -21,6 +21,8 @@ from synapse.api.events.room import ( RoomRedactionEvent, ) +from synapse.types import EventID + from synapse.util.stringutils import random_string @@ -59,7 +61,9 @@ class EventFactory(object): local_part = str(int(self.clock.time())) + i + random_string(5) - return "%s@%s" % (local_part, self.hs.hostname) + e_id = EventID.create_local(local_part, self.hs) + + return e_id.to_string() def create_event(self, etype=None, **kwargs): kwargs["type"] = etype diff --git a/synapse/federation/pdu_codec.py b/synapse/federation/pdu_codec.py index dccbccb85b..6d31286290 100644 --- a/synapse/federation/pdu_codec.py +++ b/synapse/federation/pdu_codec.py @@ -17,22 +17,11 @@ from .units import Pdu from synapse.crypto.event_signing import ( add_event_pdu_content_hash, sign_event_pdu ) +from synapse.types import EventID import copy -def decode_event_id(event_id, server_name): - parts = event_id.split("@") - if len(parts) < 2: - return (event_id, server_name) - else: - return (parts[0], "".join(parts[1:])) - - -def encode_event_id(pdu_id, origin): - return "%s@%s" % (pdu_id, origin) - - class PduCodec(object): def __init__(self, hs): @@ -40,20 +29,28 @@ class PduCodec(object): self.server_name = hs.hostname self.event_factory = hs.get_event_factory() self.clock = hs.get_clock() + self.hs = hs + + def encode_event_id(self, local, domain): + return EventID.create(local, domain, self.hs).to_string() + + def decode_event_id(self, event_id): + e_id = self.hs.parse_eventid(event_id) + return e_id.localpart, e_id.domain def event_from_pdu(self, pdu): kwargs = {} - kwargs["event_id"] = encode_event_id(pdu.pdu_id, pdu.origin) + kwargs["event_id"] = self.encode_event_id(pdu.pdu_id, pdu.origin) kwargs["room_id"] = pdu.context kwargs["etype"] = pdu.pdu_type kwargs["prev_events"] = [ - (encode_event_id(i, o), s) + (self.encode_event_id(i, o), s) for i, o, s in pdu.prev_pdus ] if hasattr(pdu, "prev_state_id") and hasattr(pdu, "prev_state_origin"): - kwargs["prev_state"] = encode_event_id( + kwargs["prev_state"] = self.encode_event_id( pdu.prev_state_id, pdu.prev_state_origin ) @@ -75,15 +72,15 @@ class PduCodec(object): def pdu_from_event(self, event): d = event.get_full_dict() - d["pdu_id"], d["origin"] = decode_event_id( - event.event_id, self.server_name + d["pdu_id"], d["origin"] = self.decode_event_id( + event.event_id ) d["context"] = event.room_id d["pdu_type"] = event.type if hasattr(event, "prev_events"): def f(e, s): - i, o = decode_event_id(e, self.server_name) + i, o = self.decode_event_id(e) return i, o, s d["prev_pdus"] = [ f(e, s) @@ -92,7 +89,7 @@ class PduCodec(object): if hasattr(event, "prev_state"): d["prev_state_id"], d["prev_state_origin"] = ( - decode_event_id(event.prev_state, self.server_name) + self.decode_event_id(event.prev_state) ) if hasattr(event, "state_key"): diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index da99a4b449..1daeee833b 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -20,9 +20,10 @@ from ._base import BaseHandler from synapse.api.events.room import InviteJoinEvent, RoomMemberEvent from synapse.api.constants import Membership from synapse.util.logutils import log_function -from synapse.federation.pdu_codec import PduCodec, encode_event_id +from synapse.federation.pdu_codec import PduCodec from synapse.api.errors import SynapseError from synapse.util.async import run_on_reactor +from synapse.types import EventID from twisted.internet import defer, reactor @@ -358,7 +359,9 @@ class FederationHandler(BaseHandler): @defer.inlineCallbacks def get_state_for_pdu(self, pdu_id, pdu_origin): - event_id = encode_event_id(pdu_id, pdu_origin) + yield run_on_reactor() + + event_id = EventID.create(pdu_id, pdu_origin, self.hs).to_string() state_groups = yield self.store.get_state_groups( [event_id] diff --git a/synapse/server.py b/synapse/server.py index a4d2d4aba5..d770b20b19 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -28,7 +28,7 @@ from synapse.handlers import Handlers from synapse.rest import RestServletFactory from synapse.state import StateHandler from synapse.storage import DataStore -from synapse.types import UserID, RoomAlias, RoomID +from synapse.types import UserID, RoomAlias, RoomID, EventID from synapse.util import Clock from synapse.util.distributor import Distributor from synapse.util.lockutils import LockManager @@ -143,6 +143,11 @@ class BaseHomeServer(object): object.""" return RoomID.from_string(s, hs=self) + def parse_eventid(self, s): + """Parse the string given by 's' as a Event ID and return a EventID + object.""" + return EventID.from_string(s, hs=self) + def serialize_event(self, e): return serialize_event(self, e) diff --git a/synapse/state.py b/synapse/state.py index a59688e3b4..414701b272 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -16,8 +16,10 @@ from twisted.internet import defer -from synapse.federation.pdu_codec import encode_event_id, decode_event_id from synapse.util.logutils import log_function +from synapse.util.async import run_on_reactor + +from synapse.types import EventID from collections import namedtuple @@ -43,6 +45,7 @@ class StateHandler(object): self.store = hs.get_datastore() self._replication = hs.get_replication_layer() self.server_name = hs.hostname + self.hs = hs @defer.inlineCallbacks @log_function @@ -77,15 +80,17 @@ class StateHandler(object): current_state = snapshot.prev_state_pdu if current_state: - event.prev_state = encode_event_id( - current_state.pdu_id, current_state.origin - ) + event.prev_state = EventID.create( + current_state.pdu_id, current_state.origin, self.hs + ).to_string() # TODO check current_state to see if the min power level is less # than the power level of the user # power_level = self._get_power_level_for_event(event) - pdu_id, origin = decode_event_id(event.event_id, self.server_name) + e_id = self.hs.parse_eventid(event.event_id) + pdu_id = e_id.localpart + origin = e_id.domain yield self.store.update_current_state( pdu_id=pdu_id, @@ -129,6 +134,8 @@ class StateHandler(object): @defer.inlineCallbacks @log_function def annotate_state_groups(self, event, old_state=None): + yield run_on_reactor() + if old_state: event.state_group = None event.old_state_events = old_state diff --git a/synapse/types.py b/synapse/types.py index c51bc8e4f2..649ff2f7d7 100644 --- a/synapse/types.py +++ b/synapse/types.py @@ -78,6 +78,11 @@ class DomainSpecificString( """Create a structure on the local domain""" return cls(localpart=localpart, domain=hs.hostname, is_mine=True) + @classmethod + def create(cls, localpart, domain, hs): + is_mine = domain == hs.hostname + return cls(localpart=localpart, domain=domain, is_mine=is_mine) + class UserID(DomainSpecificString): """Structure representing a user ID.""" @@ -94,6 +99,11 @@ class RoomID(DomainSpecificString): SIGIL = "!" +class EventID(DomainSpecificString): + """Structure representing an event id. """ + SIGIL = "$" + + class StreamToken( namedtuple( "Token", -- cgit 1.5.1 From 0a8b026ccf2d2472bb14303eaf5af8b8cf0efa5d Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 3 Nov 2014 11:32:42 +0000 Subject: Add 'origin' key to events --- synapse/api/events/factory.py | 4 ++++ 1 file changed, 4 insertions(+) (limited to 'synapse/api/events') diff --git a/synapse/api/events/factory.py b/synapse/api/events/factory.py index 750096c618..9134c82eff 100644 --- a/synapse/api/events/factory.py +++ b/synapse/api/events/factory.py @@ -69,6 +69,10 @@ class EventFactory(object): kwargs["type"] = etype if "event_id" not in kwargs: kwargs["event_id"] = self.create_event_id() + kwargs["origin"] = self.hs.hostname + else: + ev_id = self.hs.parse_eventid(kwargs["event_id"]) + kwargs["origin"] = ev_id.domain if "origin_server_ts" not in kwargs: kwargs["origin_server_ts"] = int(self.clock.time_msec()) -- cgit 1.5.1 From ad6eacb3e9424902da9f83c8f106a4f0169c3108 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 3 Nov 2014 13:06:58 +0000 Subject: Rename PDU fields to match that of events. --- synapse/api/events/utils.py | 2 +- synapse/federation/pdu_codec.py | 48 +--------- synapse/federation/replication.py | 72 ++++++--------- synapse/federation/transport.py | 184 +++++++------------------------------- synapse/federation/units.py | 78 +++------------- synapse/handlers/federation.py | 12 ++- 6 files changed, 80 insertions(+), 316 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/events/utils.py b/synapse/api/events/utils.py index 7fdf45a264..31601fd3a9 100644 --- a/synapse/api/events/utils.py +++ b/synapse/api/events/utils.py @@ -32,7 +32,7 @@ def prune_event(event): def prune_pdu(pdu): """Removes keys that contain unrestricted and non-essential data from a PDU """ - return _prune_event_or_pdu(pdu.pdu_type, pdu) + return _prune_event_or_pdu(pdu.type, pdu) def _prune_event_or_pdu(event_type, event): # Remove all extraneous fields. diff --git a/synapse/federation/pdu_codec.py b/synapse/federation/pdu_codec.py index d4c896e163..5ec97a698e 100644 --- a/synapse/federation/pdu_codec.py +++ b/synapse/federation/pdu_codec.py @@ -31,39 +31,16 @@ class PduCodec(object): self.clock = hs.get_clock() self.hs = hs - def encode_event_id(self, local, domain): - return local - - def decode_event_id(self, event_id): - e_id = self.hs.parse_eventid(event_id) - return event_id, e_id.domain - def event_from_pdu(self, pdu): kwargs = {} - kwargs["event_id"] = self.encode_event_id(pdu.pdu_id, pdu.origin) - kwargs["room_id"] = pdu.context - kwargs["etype"] = pdu.pdu_type - kwargs["prev_events"] = [ - (self.encode_event_id(i, o), s) - for i, o, s in pdu.prev_pdus - ] - - if hasattr(pdu, "prev_state_id") and hasattr(pdu, "prev_state_origin"): - kwargs["prev_state"] = self.encode_event_id( - pdu.prev_state_id, pdu.prev_state_origin - ) + kwargs["etype"] = pdu.type kwargs.update({ k: v for k, v in pdu.get_full_dict().items() if k not in [ - "pdu_id", - "context", - "pdu_type", - "prev_pdus", - "prev_state_id", - "prev_state_origin", + "type", ] }) @@ -72,33 +49,12 @@ class PduCodec(object): def pdu_from_event(self, event): d = event.get_full_dict() - d["pdu_id"], d["origin"] = self.decode_event_id( - event.event_id - ) - d["context"] = event.room_id - d["pdu_type"] = event.type - - if hasattr(event, "prev_events"): - def f(e, s): - i, o = self.decode_event_id(e) - return i, o, s - d["prev_pdus"] = [ - f(e, s) - for e, s in event.prev_events - ] - - if hasattr(event, "prev_state"): - d["prev_state_id"], d["prev_state_origin"] = ( - self.decode_event_id(event.prev_state) - ) - if hasattr(event, "state_key"): d["is_state"] = True kwargs = copy.deepcopy(event.unrecognized_keys) kwargs.update({ k: v for k, v in d.items() - if k not in ["event_id", "room_id", "type", "prev_events"] }) if "origin_server_ts" not in kwargs: diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 159af4eed7..838e660a46 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -111,14 +111,6 @@ class ReplicationLayer(object): """Informs the replication layer about a new PDU generated within the home server that should be transmitted to others. - This will fill out various attributes on the PDU object, e.g. the - `prev_pdus` key. - - *Note:* The home server should always call `send_pdu` even if it knows - that it does not need to be replicated to other home servers. This is - in case e.g. someone else joins via a remote home server and then - backfills. - TODO: Figure out when we should actually resolve the deferred. Args: @@ -131,18 +123,12 @@ class ReplicationLayer(object): order = self._order self._order += 1 - logger.debug("[%s] Persisting PDU", pdu.pdu_id) - - # Save *before* trying to send - # yield self.store.persist_event(pdu=pdu) - - logger.debug("[%s] Persisted PDU", pdu.pdu_id) - logger.debug("[%s] transaction_layer.enqueue_pdu... ", pdu.pdu_id) + logger.debug("[%s] transaction_layer.enqueue_pdu... ", pdu.event_id) # TODO, add errback, etc. self._transaction_queue.enqueue_pdu(pdu, order) - logger.debug("[%s] transaction_layer.enqueue_pdu... done", pdu.pdu_id) + logger.debug("[%s] transaction_layer.enqueue_pdu... done", pdu.event_id) @log_function def send_edu(self, destination, edu_type, content): @@ -215,7 +201,7 @@ class ReplicationLayer(object): @defer.inlineCallbacks @log_function - def get_pdu(self, destination, pdu_origin, pdu_id, outlier=False): + def get_pdu(self, destination, event_id, outlier=False): """Requests the PDU with given origin and ID from the remote home server. @@ -224,7 +210,7 @@ class ReplicationLayer(object): Args: destination (str): Which home server to query pdu_origin (str): The home server that originally sent the pdu. - pdu_id (str) + event_id (str) outlier (bool): Indicates whether the PDU is an `outlier`, i.e. if it's from an arbitary point in the context as opposed to part of the current block of PDUs. Defaults to `False` @@ -233,8 +219,9 @@ class ReplicationLayer(object): Deferred: Results in the requested PDU. """ - transaction_data = yield self.transport_layer.get_pdu( - destination, pdu_origin, pdu_id) + transaction_data = yield self.transport_layer.get_event( + destination, event_id + ) transaction = Transaction(**transaction_data) @@ -249,8 +236,7 @@ class ReplicationLayer(object): @defer.inlineCallbacks @log_function - def get_state_for_context(self, destination, context, pdu_id=None, - pdu_origin=None): + def get_state_for_context(self, destination, context, event_id=None): """Requests all of the `current` state PDUs for a given context from a remote home server. @@ -263,7 +249,9 @@ class ReplicationLayer(object): """ transaction_data = yield self.transport_layer.get_context_state( - destination, context, pdu_id=pdu_id, pdu_origin=pdu_origin, + destination, + context, + event_id=event_id, ) transaction = Transaction(**transaction_data) @@ -352,10 +340,10 @@ class ReplicationLayer(object): @defer.inlineCallbacks @log_function - def on_context_state_request(self, context, pdu_id, pdu_origin): - if pdu_id and pdu_origin: + def on_context_state_request(self, context, event_id): + if event_id: pdus = yield self.handler.get_state_for_pdu( - pdu_id, pdu_origin + event_id ) else: raise NotImplementedError("Specify an event") @@ -370,8 +358,8 @@ class ReplicationLayer(object): @defer.inlineCallbacks @log_function - def on_pdu_request(self, pdu_origin, pdu_id): - pdu = yield self._get_persisted_pdu(pdu_id, pdu_origin) + def on_pdu_request(self, event_id): + pdu = yield self._get_persisted_pdu(event_id) if pdu: defer.returnValue( @@ -443,9 +431,8 @@ class ReplicationLayer(object): def send_join(self, destination, pdu): _, content = yield self.transport_layer.send_join( destination, - pdu.context, - pdu.pdu_id, - pdu.origin, + pdu.room_id, + pdu.event_id, pdu.get_dict(), ) @@ -457,13 +444,13 @@ class ReplicationLayer(object): defer.returnValue(pdus) @log_function - def _get_persisted_pdu(self, pdu_id, pdu_origin): + def _get_persisted_pdu(self, event_id): """ Get a PDU from the database with given origin and id. Returns: Deferred: Results in a `Pdu`. """ - return self.handler.get_persisted_pdu(pdu_id, pdu_origin) + return self.handler.get_persisted_pdu(event_id) def _transaction_from_pdus(self, pdu_list): """Returns a new Transaction containing the given PDUs suitable for @@ -487,10 +474,10 @@ class ReplicationLayer(object): @log_function def _handle_new_pdu(self, origin, pdu, backfilled=False): # We reprocess pdus when we have seen them only as outliers - existing = yield self._get_persisted_pdu(pdu.pdu_id, pdu.origin) + existing = yield self._get_persisted_pdu(pdu.event_id) if existing and (not existing.outlier or pdu.outlier): - logger.debug("Already seen pdu %s %s", pdu.pdu_id, pdu.origin) + logger.debug("Already seen pdu %s", pdu.event_id) defer.returnValue({}) return @@ -500,23 +487,22 @@ class ReplicationLayer(object): if not pdu.outlier: # We only backfill backwards to the min depth. min_depth = yield self.handler.get_min_depth_for_context( - pdu.context + pdu.room_id ) if min_depth and pdu.depth > min_depth: - for pdu_id, origin, hashes in pdu.prev_pdus: - exists = yield self._get_persisted_pdu(pdu_id, origin) + for event_id, hashes in pdu.prev_events: + exists = yield self._get_persisted_pdu(event_id) if not exists: - logger.debug("Requesting pdu %s %s", pdu_id, origin) + logger.debug("Requesting pdu %s", event_id) try: yield self.get_pdu( pdu.origin, - pdu_id=pdu_id, - pdu_origin=origin + event_id=event_id, ) - logger.debug("Processed pdu %s %s", pdu_id, origin) + logger.debug("Processed pdu %s", event_id) except: # TODO(erikj): Do some more intelligent retries. logger.exception("Failed to get PDU") @@ -524,7 +510,7 @@ class ReplicationLayer(object): # We need to get the state at this event, since we have reached # a backward extremity edge. state = yield self.get_state_for_context( - origin, pdu.context, pdu.pdu_id, pdu.origin, + origin, pdu.room_id, pdu.event_id, ) # Persist the Pdu, but don't mark it as processed yet. diff --git a/synapse/federation/transport.py b/synapse/federation/transport.py index 7f01b4faaf..04ad7e63ae 100644 --- a/synapse/federation/transport.py +++ b/synapse/federation/transport.py @@ -72,8 +72,7 @@ class TransportLayer(object): self.received_handler = None @log_function - def get_context_state(self, destination, context, pdu_id=None, - pdu_origin=None): + def get_context_state(self, destination, context, event_id=None): """ Requests all state for a given context (i.e. room) from the given server. @@ -91,60 +90,59 @@ class TransportLayer(object): subpath = "/state/%s/" % context args = {} - if pdu_id and pdu_origin: - args["pdu_id"] = pdu_id - args["pdu_origin"] = pdu_origin + if event_id: + args["event_id"] = event_id return self._do_request_for_transaction( destination, subpath, args=args ) @log_function - def get_pdu(self, destination, pdu_origin, pdu_id): + def get_event(self, destination, event_id): """ Requests the pdu with give id and origin from the given server. Args: destination (str): The host name of the remote home server we want to get the state from. - pdu_origin (str): The home server which created the PDU. - pdu_id (str): The id of the PDU being requested. + event_id (str): The id of the event being requested. Returns: Deferred: Results in a dict received from the remote homeserver. """ - logger.debug("get_pdu dest=%s, pdu_origin=%s, pdu_id=%s", - destination, pdu_origin, pdu_id) + logger.debug("get_pdu dest=%s, event_id=%s", + destination, event_id) - subpath = "/pdu/%s/%s/" % (pdu_origin, pdu_id) + subpath = "/event/%s/" % (event_id, ) return self._do_request_for_transaction(destination, subpath) @log_function - def backfill(self, dest, context, pdu_tuples, limit): + def backfill(self, dest, context, event_tuples, limit): """ Requests `limit` previous PDUs in a given context before list of PDUs. Args: dest (str) context (str) - pdu_tuples (list) + event_tuples (list) limt (int) Returns: Deferred: Results in a dict received from the remote homeserver. """ logger.debug( - "backfill dest=%s, context=%s, pdu_tuples=%s, limit=%s", - dest, context, repr(pdu_tuples), str(limit) + "backfill dest=%s, context=%s, event_tuples=%s, limit=%s", + dest, context, repr(event_tuples), str(limit) ) - if not pdu_tuples: + if not event_tuples: + # TODO: raise? return - subpath = "/backfill/%s/" % context + subpath = "/backfill/%s/" % (context,) args = { - "v": ["%s,%s" % (i, o) for i, o in pdu_tuples], + "v": event_tuples, "limit": limit, } @@ -222,11 +220,10 @@ class TransportLayer(object): @defer.inlineCallbacks @log_function - def send_join(self, destination, context, pdu_id, origin, content): - path = PREFIX + "/send_join/%s/%s/%s" % ( + def send_join(self, destination, context, event_id, content): + path = PREFIX + "/send_join/%s/%s" % ( context, - origin, - pdu_id, + event_id, ) code, content = yield self.client.put_json( @@ -242,11 +239,10 @@ class TransportLayer(object): @defer.inlineCallbacks @log_function - def send_invite(self, destination, context, pdu_id, origin, content): - path = PREFIX + "/invite/%s/%s/%s" % ( + def send_invite(self, destination, context, event_id, content): + path = PREFIX + "/invite/%s/%s" % ( context, - origin, - pdu_id, + event_id, ) code, content = yield self.client.put_json( @@ -376,10 +372,10 @@ class TransportLayer(object): # data_id pair. self.server.register_path( "GET", - re.compile("^" + PREFIX + "/pdu/([^/]*)/([^/]*)/$"), + re.compile("^" + PREFIX + "/event/([^/]*)/$"), self._with_authentication( - lambda origin, content, query, pdu_origin, pdu_id: - handler.on_pdu_request(pdu_origin, pdu_id) + lambda origin, content, query, event_id: + handler.on_pdu_request(event_id) ) ) @@ -391,8 +387,7 @@ class TransportLayer(object): lambda origin, content, query, context: handler.on_context_state_request( context, - query.get("pdu_id", [None])[0], - query.get("pdu_origin", [None])[0] + query.get("event_id", [None])[0], ) ) ) @@ -442,9 +437,9 @@ class TransportLayer(object): self.server.register_path( "PUT", - re.compile("^" + PREFIX + "/send_join/([^/]*)/([^/]*)/([^/]*)$"), + re.compile("^" + PREFIX + "/send_join/([^/]*)/([^/]*)$"), self._with_authentication( - lambda origin, content, query, context, pdu_origin, pdu_id: + lambda origin, content, query, context, event_id: self._on_send_join_request( origin, content, query, ) @@ -453,9 +448,9 @@ class TransportLayer(object): self.server.register_path( "PUT", - re.compile("^" + PREFIX + "/invite/([^/]*)/([^/]*)/([^/]*)$"), + re.compile("^" + PREFIX + "/invite/([^/]*)/([^/]*)$"), self._with_authentication( - lambda origin, content, query, context, pdu_origin, pdu_id: + lambda origin, content, query, context, event_id: self._on_invite_request( origin, content, query, ) @@ -548,7 +543,7 @@ class TransportLayer(object): limit = int(limits[-1]) - versions = [v.split(",", 1) for v in v_list] + versions = v_list return self.request_handler.on_backfill_request( context, versions, limit @@ -579,120 +574,3 @@ class TransportLayer(object): ) defer.returnValue((200, content)) - - -class TransportReceivedHandler(object): - """ Callbacks used when we receive a transaction - """ - def on_incoming_transaction(self, transaction): - """ Called on PUT /send/, or on response to a request - that we sent (e.g. a backfill request) - - Args: - transaction (synapse.transaction.Transaction): The transaction that - was sent to us. - - Returns: - twisted.internet.defer.Deferred: A deferred that gets fired when - the transaction has finished being processed. - - The result should be a tuple in the form of - `(response_code, respond_body)`, where `response_body` is a python - dict that will get serialized to JSON. - - On errors, the dict should have an `error` key with a brief message - of what went wrong. - """ - pass - - -class TransportRequestHandler(object): - """ Handlers used when someone want's data from us - """ - def on_pull_request(self, versions): - """ Called on GET /pull/?v=... - - This is hit when a remote home server wants to get all data - after a given transaction. Mainly used when a home server comes back - online and wants to get everything it has missed. - - Args: - versions (list): A list of transaction_ids that should be used to - determine what PDUs the remote side have not yet seen. - - Returns: - Deferred: Resultsin a tuple in the form of - `(response_code, respond_body)`, where `response_body` is a python - dict that will get serialized to JSON. - - On errors, the dict should have an `error` key with a brief message - of what went wrong. - """ - pass - - def on_pdu_request(self, pdu_origin, pdu_id): - """ Called on GET /pdu/// - - Someone wants a particular PDU. This PDU may or may not have originated - from us. - - Args: - pdu_origin (str) - pdu_id (str) - - Returns: - Deferred: Resultsin a tuple in the form of - `(response_code, respond_body)`, where `response_body` is a python - dict that will get serialized to JSON. - - On errors, the dict should have an `error` key with a brief message - of what went wrong. - """ - pass - - def on_context_state_request(self, context): - """ Called on GET /state// - - Gets hit when someone wants all the *current* state for a given - contexts. - - Args: - context (str): The name of the context that we're interested in. - - Returns: - twisted.internet.defer.Deferred: A deferred that gets fired when - the transaction has finished being processed. - - The result should be a tuple in the form of - `(response_code, respond_body)`, where `response_body` is a python - dict that will get serialized to JSON. - - On errors, the dict should have an `error` key with a brief message - of what went wrong. - """ - pass - - def on_backfill_request(self, context, versions, limit): - """ Called on GET /backfill//?v=...&limit=... - - Gets hit when we want to backfill backwards on a given context from - the given point. - - Args: - context (str): The context to backfill - versions (list): A list of 2-tuples representing where to backfill - from, in the form `(pdu_id, origin)` - limit (int): How many pdus to return. - - Returns: - Deferred: Results in a tuple in the form of - `(response_code, respond_body)`, where `response_body` is a python - dict that will get serialized to JSON. - - On errors, the dict should have an `error` key with a brief message - of what went wrong. - """ - pass - - def on_query_request(self): - """ Called on a GET /query/ request. """ diff --git a/synapse/federation/units.py b/synapse/federation/units.py index adc3385644..c94dcf64cf 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -34,13 +34,13 @@ class Pdu(JsonEncodedObject): A Pdu can be classified as "state". For a given context, we can efficiently retrieve all state pdu's that haven't been clobbered. Clobbering is done - via a unique constraint on the tuple (context, pdu_type, state_key). A pdu + via a unique constraint on the tuple (context, type, state_key). A pdu is a state pdu if `is_state` is True. Example pdu:: { - "pdu_id": "78c", + "event_id": "$78c:example.com", "origin_server_ts": 1404835423000, "origin": "bar", "prev_ids": [ @@ -53,14 +53,14 @@ class Pdu(JsonEncodedObject): """ valid_keys = [ - "pdu_id", - "context", + "event_id", + "room_id", "origin", "origin_server_ts", - "pdu_type", + "type", "destinations", "transaction_id", - "prev_pdus", + "prev_events", "depth", "content", "outlier", @@ -68,8 +68,7 @@ class Pdu(JsonEncodedObject): "signatures", "is_state", # Below this are keys valid only for State Pdus. "state_key", - "prev_state_id", - "prev_state_origin", + "prev_state", "required_power_level", "user_id", ] @@ -81,18 +80,18 @@ class Pdu(JsonEncodedObject): ] required_keys = [ - "pdu_id", - "context", + "event_id", + "room_id", "origin", "origin_server_ts", - "pdu_type", + "type", "content", ] # TODO: We need to make this properly load content rather than # just leaving it as a dict. (OR DO WE?!) - def __init__(self, destinations=[], is_state=False, prev_pdus=[], + def __init__(self, destinations=[], is_state=False, prev_events=[], outlier=False, hashes={}, signatures={}, **kwargs): if is_state: for required_key in ["state_key"]: @@ -102,66 +101,13 @@ class Pdu(JsonEncodedObject): super(Pdu, self).__init__( destinations=destinations, is_state=bool(is_state), - prev_pdus=prev_pdus, + prev_events=prev_events, outlier=outlier, hashes=hashes, signatures=signatures, **kwargs ) - @classmethod - def from_pdu_tuple(cls, pdu_tuple): - """ Converts a PduTuple to a Pdu - - Args: - pdu_tuple (synapse.persistence.transactions.PduTuple): The tuple to - convert - - Returns: - Pdu - """ - if pdu_tuple: - d = copy.copy(pdu_tuple.pdu_entry._asdict()) - d["origin_server_ts"] = d.pop("ts") - - for k in d.keys(): - if d[k] is None: - del d[k] - - d["content"] = json.loads(d["content_json"]) - del d["content_json"] - - args = {f: d[f] for f in cls.valid_keys if f in d} - if "unrecognized_keys" in d and d["unrecognized_keys"]: - args.update(json.loads(d["unrecognized_keys"])) - - hashes = { - alg: encode_base64(hsh) - for alg, hsh in pdu_tuple.hashes.items() - } - - signatures = { - kid: encode_base64(sig) - for kid, sig in pdu_tuple.signatures.items() - } - - prev_pdus = [] - for prev_pdu in pdu_tuple.prev_pdu_list: - prev_hashes = pdu_tuple.edge_hashes.get(prev_pdu, {}) - prev_hashes = { - alg: encode_base64(hsh) for alg, hsh in prev_hashes.items() - } - prev_pdus.append((prev_pdu[0], prev_pdu[1], prev_hashes)) - - return Pdu( - prev_pdus=prev_pdus, - hashes=hashes, - signatures=signatures, - **args - ) - else: - return None - def __str__(self): return "(%s, %s)" % (self.__class__.__name__, repr(self.__dict__)) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 18cb1d4e97..bdd28f04bb 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -139,7 +139,7 @@ class FederationHandler(BaseHandler): # Huh, let's try and get the current state try: yield self.replication_layer.get_state_for_context( - event.origin, event.room_id, pdu.pdu_id, pdu.origin, + event.origin, event.room_id, event.event_id, ) hosts = yield self.store.get_joined_hosts_for_room( @@ -368,11 +368,9 @@ class FederationHandler(BaseHandler): ]) @defer.inlineCallbacks - def get_state_for_pdu(self, pdu_id, pdu_origin): + def get_state_for_pdu(self, event_id): yield run_on_reactor() - event_id = EventID.create(pdu_id, pdu_origin, self.hs).to_string() - state_groups = yield self.store.get_state_groups( [event_id] ) @@ -406,7 +404,7 @@ class FederationHandler(BaseHandler): events = yield self.store.get_backfill_events( context, - [self.pdu_codec.encode_event_id(i, o) for i, o in pdu_list], + pdu_list, limit ) @@ -417,14 +415,14 @@ class FederationHandler(BaseHandler): @defer.inlineCallbacks @log_function - def get_persisted_pdu(self, pdu_id, origin): + def get_persisted_pdu(self, event_id): """ Get a PDU from the database with given origin and id. Returns: Deferred: Results in a `Pdu`. """ event = yield self.store.get_event( - self.pdu_codec.encode_event_id(pdu_id, origin), + event_id, allow_none=True, ) -- cgit 1.5.1 From 68698e0ac8c39083f6ab7d377a48b5bead3d3598 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 3 Nov 2014 17:51:42 +0000 Subject: Fix bugs in generating event signatures and hashing --- scripts/check_event_hash.py | 12 +++-- scripts/check_signature.py | 1 - synapse/api/events/__init__.py | 1 + synapse/crypto/event_signing.py | 100 +++++++++++++++------------------------- synapse/federation/pdu_codec.py | 13 +----- synapse/federation/units.py | 11 +---- 6 files changed, 50 insertions(+), 88 deletions(-) (limited to 'synapse/api/events') diff --git a/scripts/check_event_hash.py b/scripts/check_event_hash.py index 9fa4452ee6..7c32f8102a 100644 --- a/scripts/check_event_hash.py +++ b/scripts/check_event_hash.py @@ -6,6 +6,7 @@ import hashlib import sys import json + class dictobj(dict): def __init__(self, *args, **kargs): dict.__init__(self, *args, **kargs) @@ -14,9 +15,12 @@ class dictobj(dict): def get_dict(self): return dict(self) + def get_full_dict(self): + return dict(self) + def main(): - parser = parser = argparse.ArgumentParser() + parser = argparse.ArgumentParser() parser.add_argument("input_json", nargs="?", type=argparse.FileType('r'), default=sys.stdin) args = parser.parse_args() @@ -29,14 +33,14 @@ def main(): } for alg_name in event_json.hashes: - if check_event_pdu_content_hash(event_json, algorithms[alg_name]): + if check_event_content_hash(event_json, algorithms[alg_name]): print "PASS content hash %s" % (alg_name,) else: print "FAIL content hash %s" % (alg_name,) for algorithm in algorithms.values(): - name, h_bytes = compute_pdu_event_reference_hash(event_json, algorithm) - print "Reference hash %s: %s" % (name, encode_base64(bytes)) + name, h_bytes = compute_event_reference_hash(event_json, algorithm) + print "Reference hash %s: %s" % (name, encode_base64(h_bytes)) if __name__=="__main__": main() diff --git a/scripts/check_signature.py b/scripts/check_signature.py index e7964e7e71..e146e18e24 100644 --- a/scripts/check_signature.py +++ b/scripts/check_signature.py @@ -1,5 +1,4 @@ -from synapse.crypto.event_signing import verify_signed_event_pdu from syutil.crypto.jsonsign import verify_signed_json from syutil.crypto.signing_key import ( decode_verify_key_bytes, write_signing_keys diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index b855811b98..168b812311 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -61,6 +61,7 @@ class SynapseEvent(JsonEncodedObject): "prev_content", "prev_state", "redacted_because", + "origin_server_ts", ] internal_keys = [ diff --git a/synapse/crypto/event_signing.py b/synapse/crypto/event_signing.py index 0e8bc7eb6c..de5d2e7465 100644 --- a/synapse/crypto/event_signing.py +++ b/synapse/crypto/event_signing.py @@ -15,11 +15,11 @@ # limitations under the License. -from synapse.federation.units import Pdu -from synapse.api.events.utils import prune_pdu, prune_event +from synapse.api.events.utils import prune_event from syutil.jsonutil import encode_canonical_json from syutil.base64util import encode_base64, decode_base64 -from syutil.crypto.jsonsign import sign_json, verify_signed_json +from syutil.crypto.jsonsign import sign_json +from synapse.api.events.room import GenericEvent import copy import hashlib @@ -28,20 +28,14 @@ import logging logger = logging.getLogger(__name__) -def add_event_pdu_content_hash(pdu, hash_algorithm=hashlib.sha256): - hashed = _compute_content_hash(pdu, hash_algorithm) - pdu.hashes[hashed.name] = encode_base64(hashed.digest()) - return pdu - - -def check_event_pdu_content_hash(pdu, hash_algorithm=hashlib.sha256): +def check_event_content_hash(event, hash_algorithm=hashlib.sha256): """Check whether the hash for this PDU matches the contents""" - computed_hash = _compute_content_hash(pdu, hash_algorithm) - if computed_hash.name not in pdu.hashes: + computed_hash = _compute_content_hash(event, hash_algorithm) + if computed_hash.name not in event.hashes: raise Exception("Algorithm %s not in hashes %s" % ( - computed_hash.name, list(pdu.hashes) + computed_hash.name, list(event.hashes) )) - message_hash_base64 = pdu.hashes[computed_hash.name] + message_hash_base64 = event.hashes[computed_hash.name] try: message_hash_bytes = decode_base64(message_hash_base64) except: @@ -49,70 +43,52 @@ def check_event_pdu_content_hash(pdu, hash_algorithm=hashlib.sha256): return message_hash_bytes == computed_hash.digest() -def _compute_content_hash(pdu, hash_algorithm): - pdu_json = pdu.get_dict() - #TODO: Make "age_ts" key internal - pdu_json.pop("age_ts", None) - pdu_json.pop("unsigned", None) - pdu_json.pop("signatures", None) - pdu_json.pop("hashes", None) - pdu_json_bytes = encode_canonical_json(pdu_json) - return hash_algorithm(pdu_json_bytes) - - -def compute_pdu_event_reference_hash(pdu, hash_algorithm=hashlib.sha256): - tmp_pdu = Pdu(**pdu.get_dict()) - tmp_pdu = prune_pdu(tmp_pdu) - pdu_json = tmp_pdu.get_dict() - pdu_json.pop("signatures", None) - pdu_json_bytes = encode_canonical_json(pdu_json) - hashed = hash_algorithm(pdu_json_bytes) - return (hashed.name, hashed.digest()) +def _compute_content_hash(event, hash_algorithm): + event_json = event.get_full_dict() + #TODO: We need to sign the JSON that is going out via fedaration. + event_json.pop("age_ts", None) + event_json.pop("unsigned", None) + event_json.pop("signatures", None) + event_json.pop("hashes", None) + event_json_bytes = encode_canonical_json(event_json) + return hash_algorithm(event_json_bytes) def compute_event_reference_hash(event, hash_algorithm=hashlib.sha256): - tmp_event = copy.deepcopy(event) + # FIXME(erikj): GenericEvent! + tmp_event = GenericEvent(**event.get_full_dict()) tmp_event = prune_event(tmp_event) event_json = tmp_event.get_dict() event_json.pop("signatures", None) + event_json.pop("age_ts", None) + event_json.pop("unsigned", None) event_json_bytes = encode_canonical_json(event_json) hashed = hash_algorithm(event_json_bytes) return (hashed.name, hashed.digest()) -def sign_event_pdu(pdu, signature_name, signing_key): - tmp_pdu = Pdu(**pdu.get_dict()) - tmp_pdu = prune_pdu(tmp_pdu) - pdu_json = tmp_pdu.get_dict() - pdu_json = sign_json(pdu_json, signature_name, signing_key) - pdu.signatures = pdu_json["signatures"] - return pdu - - -def verify_signed_event_pdu(pdu, signature_name, verify_key): - tmp_pdu = Pdu(**pdu.get_dict()) - tmp_pdu = prune_pdu(tmp_pdu) - pdu_json = tmp_pdu.get_dict() - verify_signed_json(pdu_json, signature_name, verify_key) - - -def add_hashes_and_signatures(event, signature_name, signing_key, - hash_algorithm=hashlib.sha256): +def compute_event_signature(event, signature_name, signing_key): tmp_event = copy.deepcopy(event) tmp_event = prune_event(tmp_event) - redact_json = tmp_event.get_dict() + redact_json = tmp_event.get_full_dict() redact_json.pop("signatures", None) + redact_json.pop("age_ts", None) + redact_json.pop("unsigned", None) + logger.debug("Signing event: %s", redact_json) redact_json = sign_json(redact_json, signature_name, signing_key) - event.signatures = redact_json["signatures"] + return redact_json["signatures"] + + +def add_hashes_and_signatures(event, signature_name, signing_key, + hash_algorithm=hashlib.sha256): + hashed = _compute_content_hash(event, hash_algorithm=hash_algorithm) - event_json = event.get_full_dict() - #TODO: We need to sign the JSON that is going out via fedaration. - event_json.pop("age_ts", None) - event_json.pop("unsigned", None) - event_json.pop("signatures", None) - event_json.pop("hashes", None) - event_json_bytes = encode_canonical_json(event_json) - hashed = hash_algorithm(event_json_bytes) if not hasattr(event, "hashes"): event.hashes = {} event.hashes[hashed.name] = encode_base64(hashed.digest()) + + event.signatures = compute_event_signature( + event, + signature_name=signature_name, + signing_key=signing_key, + ) diff --git a/synapse/federation/pdu_codec.py b/synapse/federation/pdu_codec.py index 5ec97a698e..52c84efb5b 100644 --- a/synapse/federation/pdu_codec.py +++ b/synapse/federation/pdu_codec.py @@ -14,10 +14,6 @@ # limitations under the License. from .units import Pdu -from synapse.crypto.event_signing import ( - add_event_pdu_content_hash, sign_event_pdu -) -from synapse.types import EventID import copy @@ -49,17 +45,10 @@ class PduCodec(object): def pdu_from_event(self, event): d = event.get_full_dict() - if hasattr(event, "state_key"): - d["is_state"] = True - kwargs = copy.deepcopy(event.unrecognized_keys) kwargs.update({ k: v for k, v in d.items() }) - if "origin_server_ts" not in kwargs: - kwargs["origin_server_ts"] = int(self.clock.time_msec()) - pdu = Pdu(**kwargs) - pdu = add_event_pdu_content_hash(pdu) - return sign_event_pdu(pdu, self.server_name, self.signing_key) + return pdu diff --git a/synapse/federation/units.py b/synapse/federation/units.py index c94dcf64cf..c2d8dca8f3 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -65,8 +65,7 @@ class Pdu(JsonEncodedObject): "content", "outlier", "hashes", - "signatures", - "is_state", # Below this are keys valid only for State Pdus. + "signatures", # Below this are keys valid only for State Pdus. "state_key", "prev_state", "required_power_level", @@ -91,16 +90,10 @@ class Pdu(JsonEncodedObject): # TODO: We need to make this properly load content rather than # just leaving it as a dict. (OR DO WE?!) - def __init__(self, destinations=[], is_state=False, prev_events=[], + def __init__(self, destinations=[], prev_events=[], outlier=False, hashes={}, signatures={}, **kwargs): - if is_state: - for required_key in ["state_key"]: - if required_key not in kwargs: - raise RuntimeError("Key %s is required" % required_key) - super(Pdu, self).__init__( destinations=destinations, - is_state=bool(is_state), prev_events=prev_events, outlier=outlier, hashes=hashes, -- cgit 1.5.1 From 4317c8e5835f0c15bf882f737d3e3c2a5b85f73f Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 6 Nov 2014 15:10:55 +0000 Subject: Implement new replace_state and changed prev_state `prev_state` is now a list of previous state ids, similiar to prev_events. `replace_state` now points to what we think was replaced. --- synapse/api/events/__init__.py | 1 + synapse/handlers/directory.py | 5 +- synapse/handlers/federation.py | 4 +- synapse/handlers/message.py | 11 ++-- synapse/handlers/profile.py | 6 +-- synapse/handlers/room.py | 16 ++---- synapse/rest/room.py | 2 +- synapse/state.py | 39 ++------------ synapse/storage/__init__.py | 92 +++++++++++++++++++++++++--------- synapse/storage/_base.py | 66 +++++++++++++++++------- synapse/storage/event_federation.py | 64 ++++++++++++++++++++--- synapse/storage/schema/event_edges.sql | 40 ++++++++++----- synapse/util/jsonobject.py | 2 +- 13 files changed, 220 insertions(+), 128 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 168b812311..fc3f350570 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -60,6 +60,7 @@ class SynapseEvent(JsonEncodedObject): "age_ts", "prev_content", "prev_state", + "replaces_state", "redacted_because", "origin_server_ts", ] diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py index 6e897e915d..164363cdc5 100644 --- a/synapse/handlers/directory.py +++ b/synapse/handlers/directory.py @@ -147,10 +147,7 @@ class DirectoryHandler(BaseHandler): content={"aliases": aliases}, ) - snapshot = yield self.store.snapshot_room( - room_id=room_id, - user_id=user_id, - ) + snapshot = yield self.store.snapshot_room(event) yield self._on_new_room_event( event, snapshot, extra_users=[user_id], suppress_auth=True diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 1464a60937..513ec9a5e3 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -313,9 +313,7 @@ class FederationHandler(BaseHandler): state_key=user_id, ) - snapshot = yield self.store.snapshot_room( - event.room_id, event.user_id, - ) + snapshot = yield self.store.snapshot_room(event) snapshot.fill_out_prev_events(event) yield self.state_handler.annotate_state_groups(event) diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index c6f6ab14d1..8394013df3 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -81,7 +81,7 @@ class MessageHandler(BaseHandler): user = self.hs.parse_userid(event.user_id) assert user.is_mine, "User must be our own: %s" % (user,) - snapshot = yield self.store.snapshot_room(event.room_id, event.user_id) + snapshot = yield self.store.snapshot_room(event) yield self._on_new_room_event( event, snapshot, suppress_auth=suppress_auth @@ -141,12 +141,7 @@ class MessageHandler(BaseHandler): SynapseError if something went wrong. """ - snapshot = yield self.store.snapshot_room( - event.room_id, - event.user_id, - state_type=event.type, - state_key=event.state_key, - ) + snapshot = yield self.store.snapshot_room(event) yield self._on_new_room_event(event, snapshot) @@ -214,7 +209,7 @@ class MessageHandler(BaseHandler): @defer.inlineCallbacks def send_feedback(self, event): - snapshot = yield self.store.snapshot_room(event.room_id, event.user_id) + snapshot = yield self.store.snapshot_room(event) # store message in db yield self._on_new_room_event(event, snapshot) diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index 4cd0a06093..e47814483a 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -17,7 +17,6 @@ from twisted.internet import defer from synapse.api.errors import SynapseError, AuthError, CodeMessageException from synapse.api.constants import Membership -from synapse.api.events.room import RoomMemberEvent from ._base import BaseHandler @@ -196,10 +195,7 @@ class ProfileHandler(BaseHandler): ) for j in joins: - snapshot = yield self.store.snapshot_room( - j.room_id, j.state_key, RoomMemberEvent.TYPE, - j.state_key - ) + snapshot = yield self.store.snapshot_room(j) content = { "membership": j.content["membership"], diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index f176ad39bf..55c893eb58 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -122,10 +122,7 @@ class RoomCreationHandler(BaseHandler): @defer.inlineCallbacks def handle_event(event): - snapshot = yield self.store.snapshot_room( - room_id=room_id, - user_id=user_id, - ) + snapshot = yield self.store.snapshot_room(event) logger.debug("Event: %s", event) @@ -364,10 +361,8 @@ class RoomMemberHandler(BaseHandler): """ target_user_id = event.state_key - snapshot = yield self.store.snapshot_room( - event.room_id, event.user_id, - RoomMemberEvent.TYPE, target_user_id - ) + snapshot = yield self.store.snapshot_room(event) + ## TODO(markjh): get prev state from snapshot. prev_state = yield self.store.get_room_member( target_user_id, event.room_id @@ -442,10 +437,7 @@ class RoomMemberHandler(BaseHandler): content=content, ) - snapshot = yield self.store.snapshot_room( - room_id, joinee.to_string(), RoomMemberEvent.TYPE, - joinee.to_string() - ) + snapshot = yield self.store.snapshot_room(new_event) yield self._do_join(new_event, snapshot, room_host=host, do_auth=True) diff --git a/synapse/rest/room.py b/synapse/rest/room.py index ec0ce78fda..997895dab0 100644 --- a/synapse/rest/room.py +++ b/synapse/rest/room.py @@ -138,7 +138,7 @@ class RoomStateEventRestServlet(RestServlet): raise SynapseError( 404, "Event not found.", errcode=Codes.NOT_FOUND ) - defer.returnValue((200, data[0].get_dict()["content"])) + defer.returnValue((200, data.get_dict()["content"])) @defer.inlineCallbacks def on_PUT(self, request, room_id, event_type, state_key): diff --git a/synapse/state.py b/synapse/state.py index 32744e047c..97a8160a33 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -45,40 +45,6 @@ class StateHandler(object): self.server_name = hs.hostname self.hs = hs - @defer.inlineCallbacks - @log_function - def handle_new_event(self, event, snapshot): - """ Given an event this works out if a) we have sufficient power level - to update the state and b) works out what the prev_state should be. - - Returns: - Deferred: Resolved with a boolean indicating if we successfully - updated the state. - - Raised: - AuthError - """ - # This needs to be done in a transaction. - - if not hasattr(event, "state_key"): - return - - # Now I need to fill out the prev state and work out if it has auth - # (w.r.t. to power levels) - - snapshot.fill_out_prev_events(event) - yield self.annotate_state_groups(event) - - if event.old_state_events: - current_state = event.old_state_events.get( - (event.type, event.state_key) - ) - - if current_state: - event.prev_state = current_state.event_id - - defer.returnValue(True) - @defer.inlineCallbacks @log_function def annotate_state_groups(self, event, old_state=None): @@ -111,7 +77,10 @@ class StateHandler(object): event.old_state_events = copy.deepcopy(new_state) if hasattr(event, "state_key"): - new_state[(event.type, event.state_key)] = event + key = (event.type, event.state_key) + if key in new_state: + event.replaces_state = new_state[key].event_id + new_state[key] = event event.state_group = None event.state_events = new_state diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 6b8fed4502..2d62fc2ed0 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -242,8 +242,8 @@ class DataStore(RoomMemberStore, RoomStore, "state_key": event.state_key, } - if hasattr(event, "prev_state"): - vals["prev_state"] = event.prev_state + if hasattr(event, "replaces_state"): + vals["prev_state"] = event.replaces_state self._simple_insert_txn(txn, "state_events", vals) @@ -258,6 +258,40 @@ class DataStore(RoomMemberStore, RoomStore, } ) + for e_id, h in event.prev_state: + self._simple_insert_txn( + txn, + table="event_edges", + values={ + "event_id": event.event_id, + "prev_event_id": e_id, + "room_id": event.room_id, + "is_state": 1, + }, + or_ignore=True, + ) + + if not backfilled: + self._simple_insert_txn( + txn, + table="state_forward_extremities", + values={ + "event_id": event.event_id, + "room_id": event.room_id, + "type": event.type, + "state_key": event.state_key, + } + ) + + for prev_state_id, _ in event.prev_state: + self._simple_delete_txn( + txn, + table="state_forward_extremities", + keyvalues={ + "event_id": prev_state_id, + } + ) + for hash_alg, hash_base64 in event.hashes.items(): hash_bytes = decode_base64(hash_base64) self._store_event_content_hash_txn( @@ -357,7 +391,7 @@ class DataStore(RoomMemberStore, RoomStore, ], ) - def snapshot_room(self, room_id, user_id, state_type=None, state_key=None): + def snapshot_room(self, event): """Snapshot the room for an update by a user Args: room_id (synapse.types.RoomId): The room to snapshot. @@ -368,16 +402,29 @@ class DataStore(RoomMemberStore, RoomStore, synapse.storage.Snapshot: A snapshot of the state of the room. """ def _snapshot(txn): - membership_state = self._get_room_member(txn, user_id, room_id) - prev_events = self._get_latest_events_in_room(txn, room_id) + prev_events = self._get_latest_events_in_room( + txn, + event.room_id + ) + + prev_state = None + state_key = None + if hasattr(event, "state_key"): + state_key = event.state_key + prev_state = self._get_latest_state_in_room( + txn, + event.room_id, + type=event.type, + state_key=state_key, + ) return Snapshot( store=self, - room_id=room_id, - user_id=user_id, + room_id=event.room_id, + user_id=event.user_id, prev_events=prev_events, - membership_state=membership_state, - state_type=state_type, + prev_state=prev_state, + state_type=event.type, state_key=state_key, ) @@ -400,30 +447,29 @@ class Snapshot(object): """ def __init__(self, store, room_id, user_id, prev_events, - membership_state, state_type=None, state_key=None, - prev_state_pdu=None): + prev_state, state_type=None, state_key=None): self.store = store self.room_id = room_id self.user_id = user_id self.prev_events = prev_events - self.membership_state = membership_state + self.prev_state = prev_state self.state_type = state_type self.state_key = state_key - self.prev_state_pdu = prev_state_pdu def fill_out_prev_events(self, event): - if hasattr(event, "prev_events"): - return + if not hasattr(event, "prev_events"): + event.prev_events = [ + (event_id, hashes) + for event_id, hashes, _ in self.prev_events + ] - event.prev_events = [ - (event_id, hashes) - for event_id, hashes, _ in self.prev_events - ] + if self.prev_events: + event.depth = max([int(v) for _, _, v in self.prev_events]) + 1 + else: + event.depth = 0 - if self.prev_events: - event.depth = max([int(v) for _, _, v in self.prev_events]) + 1 - else: - event.depth = 0 + if not hasattr(event, "prev_state") and self.prev_state is not None: + event.prev_state = self.prev_state def schema_path(schema): diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 7d445b4633..7821fc4726 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -245,7 +245,6 @@ class SQLBaseStore(object): return [r[0] for r in txn.fetchall()] - def _simple_select_onecol(self, table, keyvalues, retcol): """Executes a SELECT query on the named table, which returns a list comprising of the values of the named column from the selected rows. @@ -273,17 +272,30 @@ class SQLBaseStore(object): keyvalues : dict of column names and values to select the rows with retcols : list of strings giving the names of the columns to return """ + return self.runInteraction( + "_simple_select_list", + self._simple_select_list_txn, + table, keyvalues, retcols + ) + + def _simple_select_list_txn(self, txn, table, keyvalues, retcols): + """Executes a SELECT query on the named table, which may return zero or + more rows, returning the result as a list of dicts. + + Args: + txn : Transaction object + table : string giving the table name + keyvalues : dict of column names and values to select the rows with + retcols : list of strings giving the names of the columns to return + """ sql = "SELECT %s FROM %s WHERE %s" % ( ", ".join(retcols), table, - " AND ".join("%s = ?" % (k) for k in keyvalues) + " AND ".join("%s = ?" % (k, ) for k in keyvalues) ) - def func(txn): - txn.execute(sql, keyvalues.values()) - return self.cursor_to_dict(txn) - - return self.runInteraction("_simple_select_list", func) + txn.execute(sql, keyvalues.values()) + return self.cursor_to_dict(txn) def _simple_update_one(self, table, keyvalues, updatevalues, retcols=None): @@ -417,6 +429,10 @@ class SQLBaseStore(object): d.pop("topological_ordering", None) d.pop("processed", None) d["origin_server_ts"] = d.pop("ts", 0) + replaces_state = d.pop("prev_state", None) + + if replaces_state: + d["replaces_state"] = replaces_state d.update(json.loads(row_dict["unrecognized_keys"])) d["content"] = json.loads(d["content"]) @@ -450,16 +466,32 @@ class SQLBaseStore(object): k: encode_base64(v) for k, v in signatures.items() } - ev.prev_events = self._get_prev_events(txn, ev.event_id) - - if hasattr(ev, "prev_state"): - # Load previous state_content. - # TODO: Should we be pulling this out above? - cursor = txn.execute(select_event_sql, (ev.prev_state,)) - prevs = self.cursor_to_dict(cursor) - if prevs: - prev = self._parse_event_from_row(prevs[0]) - ev.prev_content = prev.content + prevs = self._get_prev_events_and_state(txn, ev.event_id) + + ev.prev_events = [ + (e_id, h) + for e_id, h, is_state in prevs + if is_state == 0 + ] + + if hasattr(ev, "state_key"): + ev.prev_state = [ + (e_id, h) + for e_id, h, is_state in prevs + if is_state == 1 + ] + + if hasattr(ev, "replaces_state"): + # Load previous state_content. + # FIXME (erikj): Handle multiple prev_states. + cursor = txn.execute( + select_event_sql, + (ev.replaces_state,) + ) + prevs = self.cursor_to_dict(cursor) + if prevs: + prev = self._parse_event_from_row(prevs[0]) + ev.prev_content = prev.content if not hasattr(ev, "redacted"): logger.debug("Doesn't have redacted key: %s", ev) diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index f427aba879..180a764134 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -69,19 +69,21 @@ class EventFederationStore(SQLBaseStore): return results - def _get_prev_events(self, txn, event_id): - prev_ids = self._simple_select_onecol_txn( + def _get_latest_state_in_room(self, txn, room_id, type, state_key): + event_ids = self._simple_select_onecol_txn( txn, - table="event_edges", + table="state_forward_extremities", keyvalues={ - "event_id": event_id, + "room_id": room_id, + "type": type, + "state_key": state_key, }, - retcol="prev_event_id", + retcol="event_id", ) results = [] - for prev_event_id in prev_ids: - hashes = self._get_event_reference_hashes_txn(txn, prev_event_id) + for event_id in event_ids: + hashes = self._get_event_reference_hashes_txn(txn, event_id) prev_hashes = { k: encode_base64(v) for k, v in hashes.items() if k == "sha256" @@ -90,6 +92,53 @@ class EventFederationStore(SQLBaseStore): return results + def _get_prev_events(self, txn, event_id): + results = self._get_prev_events_and_state( + txn, + event_id, + is_state=0, + ) + + return [(e_id, h, ) for e_id, h, _ in results] + + def _get_prev_state(self, txn, event_id): + results = self._get_prev_events_and_state( + txn, + event_id, + is_state=1, + ) + + return [(e_id, h, ) for e_id, h, _ in results] + + def _get_prev_events_and_state(self, txn, event_id, is_state=None): + keyvalues = { + "event_id": event_id, + } + + if is_state is not None: + keyvalues["is_state"] = is_state + + res = self._simple_select_list_txn( + txn, + table="event_edges", + keyvalues=keyvalues, + retcols=["prev_event_id", "is_state"], + ) + + results = [] + for d in res: + hashes = self._get_event_reference_hashes_txn( + txn, + d["prev_event_id"] + ) + prev_hashes = { + k: encode_base64(v) for k, v in hashes.items() + if k == "sha256" + } + results.append((d["prev_event_id"], prev_hashes, d["is_state"])) + + return results + def get_min_depth(self, room_id): return self.runInteraction( "get_min_depth", @@ -135,6 +184,7 @@ class EventFederationStore(SQLBaseStore): "event_id": event_id, "prev_event_id": e_id, "room_id": room_id, + "is_state": 0, }, or_ignore=True, ) diff --git a/synapse/storage/schema/event_edges.sql b/synapse/storage/schema/event_edges.sql index e5f768c705..51695826a8 100644 --- a/synapse/storage/schema/event_edges.sql +++ b/synapse/storage/schema/event_edges.sql @@ -1,7 +1,7 @@ CREATE TABLE IF NOT EXISTS event_forward_extremities( - event_id TEXT, - room_id TEXT, + event_id TEXT NOT NULL, + room_id TEXT NOT NULL, CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE ); @@ -10,8 +10,8 @@ CREATE INDEX IF NOT EXISTS ev_extrem_id ON event_forward_extremities(event_id); CREATE TABLE IF NOT EXISTS event_backward_extremities( - event_id TEXT, - room_id TEXT, + event_id TEXT NOT NULL, + room_id TEXT NOT NULL, CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE ); @@ -20,10 +20,11 @@ CREATE INDEX IF NOT EXISTS ev_b_extrem_id ON event_backward_extremities(event_id CREATE TABLE IF NOT EXISTS event_edges( - event_id TEXT, - prev_event_id TEXT, - room_id TEXT, - CONSTRAINT uniqueness UNIQUE (event_id, prev_event_id, room_id) + event_id TEXT NOT NULL, + prev_event_id TEXT NOT NULL, + room_id TEXT NOT NULL, + is_state INTEGER NOT NULL, + CONSTRAINT uniqueness UNIQUE (event_id, prev_event_id, room_id, is_state) ); CREATE INDEX IF NOT EXISTS ev_edges_id ON event_edges(event_id); @@ -31,8 +32,8 @@ CREATE INDEX IF NOT EXISTS ev_edges_prev_id ON event_edges(prev_event_id); CREATE TABLE IF NOT EXISTS room_depth( - room_id TEXT, - min_depth INTEGER, + room_id TEXT NOT NULL, + min_depth INTEGER NOT NULL, CONSTRAINT uniqueness UNIQUE (room_id) ); @@ -40,10 +41,25 @@ CREATE INDEX IF NOT EXISTS room_depth_room ON room_depth(room_id); create TABLE IF NOT EXISTS event_destinations( - event_id TEXT, - destination TEXT, + event_id TEXT NOT NULL, + destination TEXT NOT NULL, delivered_ts INTEGER DEFAULT 0, -- or 0 if not delivered CONSTRAINT uniqueness UNIQUE (event_id, destination) ON CONFLICT REPLACE ); CREATE INDEX IF NOT EXISTS event_destinations_id ON event_destinations(event_id); + + +CREATE TABLE IF NOT EXISTS state_forward_extremities( + event_id TEXT NOT NULL, + room_id TEXT NOT NULL, + type TEXT NOT NULL, + state_key TEXT NOT NULL, + CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE +); + +CREATE INDEX IF NOT EXISTS st_extrem_keys ON state_forward_extremities( + room_id, type, state_key +); +CREATE INDEX IF NOT EXISTS st_extrem_id ON state_forward_extremities(event_id); + diff --git a/synapse/util/jsonobject.py b/synapse/util/jsonobject.py index c91eb897a8..e79b68f661 100644 --- a/synapse/util/jsonobject.py +++ b/synapse/util/jsonobject.py @@ -80,7 +80,7 @@ class JsonEncodedObject(object): def get_full_dict(self): d = { - k: v for (k, v) in self.__dict__.items() + k: _encode(v) for (k, v) in self.__dict__.items() if k in self.valid_keys or k in self.internal_keys } d.update(self.unrecognized_keys) -- cgit 1.5.1 From c6766d45b53ba51121ca15ae5806a5ac30f4f5d6 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 6 Nov 2014 15:19:00 +0000 Subject: Don't send prev_state to clients anymore --- synapse/api/events/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'synapse/api/events') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index fc3f350570..98a66144e7 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -59,7 +59,6 @@ class SynapseEvent(JsonEncodedObject): "required_power_level", "age_ts", "prev_content", - "prev_state", "replaces_state", "redacted_because", "origin_server_ts", @@ -76,6 +75,7 @@ class SynapseEvent(JsonEncodedObject): "prev_events", "hashes", "signatures", + "prev_state", ] required_keys = [ -- cgit 1.5.1 From 351c64e99e5677096f9a2ae2cd7e84dbc1887878 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 6 Nov 2014 16:59:13 +0000 Subject: Amalgamate all power levels. Remove concept of reqired power levels, something similiar can be done using the new power level event. --- synapse/api/auth.py | 221 ++++++++++++++--------------------------- synapse/api/events/__init__.py | 3 +- synapse/api/events/factory.py | 7 +- synapse/api/events/room.py | 22 ---- synapse/api/events/utils.py | 23 ++--- synapse/handlers/room.py | 52 +++------- synapse/storage/__init__.py | 12 --- synapse/storage/room.py | 157 ----------------------------- 8 files changed, 102 insertions(+), 395 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index 9eb0491c97..462e97bd90 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -21,8 +21,8 @@ from synapse.api.constants import Membership, JoinRules from synapse.api.errors import AuthError, StoreError, Codes, SynapseError from synapse.api.events.room import ( RoomMemberEvent, RoomPowerLevelsEvent, RoomRedactionEvent, - RoomJoinRulesEvent, RoomOpsPowerLevelsEvent, InviteJoinEvent, - RoomCreateEvent, RoomSendEventLevelEvent, RoomAddStateLevelEvent, + RoomJoinRulesEvent, InviteJoinEvent, + RoomCreateEvent, ) from synapse.util.logutils import log_function @@ -51,6 +51,7 @@ class Auth(object): if event.old_state_events is None: # Oh, we don't know what the state of the room was, so we # are trusting that this is allowed (at least for now) + logger.warn("Trusting event: %s", event.event_id) return True if hasattr(event, "outlier") and event.outlier is True: @@ -64,7 +65,7 @@ class Auth(object): return True if event.type == RoomMemberEvent.TYPE: - self._can_replace_state(event) + self._can_send_event(event) allowed = self.is_membership_change_allowed(event) if allowed: logger.debug("Allowing! %s", event) @@ -72,16 +73,7 @@ class Auth(object): logger.debug("Denying! %s", event) return allowed - if not event.type == InviteJoinEvent.TYPE: - self.check_event_sender_in_room(event) - - if is_state: - # TODO (erikj): This really only should be called for *new* - # state - self._can_add_state(event) - self._can_replace_state(event) - else: - self._can_send_event(event) + self._can_send_event(event) if event.type == RoomPowerLevelsEvent.TYPE: self._check_power_levels(event) @@ -239,21 +231,21 @@ class Auth(object): power_level_event = event.old_state_events.get(key) level = None if power_level_event: - level = power_level_event.content.get(user_id) + level = power_level_event.content.get("users", {}).get(user_id) if not level: - level = power_level_event.content.get("default", 0) + level = power_level_event.content.get("users_default", 0) return level def _get_ops_level_from_event_state(self, event): - key = (RoomOpsPowerLevelsEvent.TYPE, "", ) - ops_event = event.old_state_events.get(key) + key = (RoomPowerLevelsEvent.TYPE, "", ) + power_level_event = event.old_state_events.get(key) - if ops_event: + if power_level_event: return ( - ops_event.content.get("ban_level"), - ops_event.content.get("kick_level"), - ops_event.content.get("redact_level"), + power_level_event.content.get("ban", 50), + power_level_event.content.get("kick", 50), + power_level_event.content.get("redact", 50), ) return None, None, None, @@ -325,13 +317,22 @@ class Auth(object): @log_function def _can_send_event(self, event): - key = (RoomSendEventLevelEvent.TYPE, "", ) + key = (RoomPowerLevelsEvent.TYPE, "", ) send_level_event = event.old_state_events.get(key) send_level = None if send_level_event: - send_level = send_level_event.content.get(event.user_id) + send_level = send_level_event.content.get("events", {}).get( + event.type + ) if not send_level: - send_level = send_level_event.content.get("level", 0) + if hasattr(event, "state_key"): + send_level = send_level_event.content.get( + "state_default", 50 + ) + else: + send_level = send_level_event.content.get( + "events_default", 0 + ) if send_level: send_level = int(send_level) @@ -350,85 +351,21 @@ class Auth(object): if user_level < send_level: raise AuthError( - 403, "You don't have permission to post to the room" - ) - - return True - - def _can_add_state(self, event): - key = (RoomAddStateLevelEvent.TYPE, "", ) - add_level_event = event.old_state_events.get(key) - add_level = None - if add_level_event: - add_level = add_level_event.content.get(event.user_id) - if not add_level: - add_level = add_level_event.content.get("level", 0) - - if add_level: - add_level = int(add_level) - else: - add_level = 0 - - user_level = self._get_power_level_from_event_state( - event, - event.user_id, - ) - - user_level = int(user_level) - - if user_level < add_level: - raise AuthError( - 403, "You don't have permission to add state to the room" + 403, "You don't have permission to post that to the room" ) return True - def _can_replace_state(self, event): - user_level = self._get_power_level_from_event_state( - event, - event.user_id, - ) - - if user_level: - user_level = int(user_level) - else: - user_level = 0 - - logger.debug( - "Checking power level for %s, %s", event.user_id, user_level - ) - - key = (event.type, event.state_key, ) - current_state = event.old_state_events.get(key) - - if current_state and hasattr(current_state, "required_power_level"): - req = current_state.required_power_level - - logger.debug("Checked power level for %s, %s", event.user_id, req) - if user_level < req: - raise AuthError( - 403, - "You don't have permission to change that state" - ) - def _check_redaction(self, event): user_level = self._get_power_level_from_event_state( event, event.user_id, ) - if user_level: - user_level = int(user_level) - else: - user_level = 0 - _, _, redact_level = self._get_ops_level_from_event_state( event ) - if not redact_level: - redact_level = 50 - if user_level < redact_level: raise AuthError( 403, @@ -436,14 +373,9 @@ class Auth(object): ) def _check_power_levels(self, event): - for k, v in event.content.items(): - if k == "default": - continue - - # FIXME (erikj): We don't want hsob_Ts in content. - if k == "hsob_ts": - continue - + user_list = event.content.get("users", {}) + # Validate users + for k, v in user_list.items(): try: self.hs.parse_userid(k) except: @@ -459,72 +391,63 @@ class Auth(object): if not current_state: return - else: - current_state = current_state[0] user_level = self._get_power_level_from_event_state( event, event.user_id, ) - if user_level: - user_level = int(user_level) - else: - user_level = 0 + # Check other levels: + levels_to_check = [ + ("users_default", []), + ("events_default", []), + ("ban", []), + ("redact", []), + ("kick", []), + ] + + old_list = current_state.content.get("users") + for user in set(old_list.keys() + user_list.keys()): + levels_to_check.append( + (user, ["users"]) + ) - old_list = current_state.content + old_list = current_state.content.get("events") + new_list = event.content.get("events") + for ev_id in set(old_list.keys() + new_list.keys()): + levels_to_check.append( + (ev_id, ["events"]) + ) - # FIXME (erikj) - old_people = {k: v for k, v in old_list.items() if k.startswith("@")} - new_people = { - k: v for k, v in event.content.items() - if k.startswith("@") - } + old_state = current_state.content + new_state = event.content - removed = set(old_people.keys()) - set(new_people.keys()) - added = set(new_people.keys()) - set(old_people.keys()) - same = set(old_people.keys()) & set(new_people.keys()) + for level_to_check, dir in levels_to_check: + old_loc = old_state + for d in dir: + old_loc = old_loc.get(d, {}) - for r in removed: - if int(old_list[r]) > user_level: - raise AuthError( - 403, - "You don't have permission to remove user: %s" % (r, ) - ) + new_loc = new_state + for d in dir: + new_loc = new_loc.get(d, {}) - for n in added: - if int(event.content[n]) > user_level: - raise AuthError( - 403, - "You don't have permission to add ops level greater " - "than your own" - ) + if level_to_check in old_loc: + old_level = int(old_loc[level_to_check]) + else: + old_level = None - for s in same: - if int(event.content[s]) != int(old_list[s]): - if int(event.content[s]) > user_level: - raise AuthError( - 403, - "You don't have permission to add ops level greater " - "than your own" - ) + if level_to_check in new_loc: + new_level = int(new_loc[level_to_check]) + else: + new_level = None - if "default" in old_list: - old_default = int(old_list["default"]) + if new_level is not None and old_level is not None: + if new_level == old_level: + continue - if old_default > user_level: + if old_level > user_level or new_level > user_level: raise AuthError( 403, - "You don't have permission to add ops level greater than " - "your own" + "You don't have permission to add ops level greater " + "than your own" ) - - if "default" in event.content: - new_default = int(event.content["default"]) - - if new_default > user_level: - raise AuthError( - 403, - "You don't have permission to add ops level greater " - "than your own" - ) diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 98a66144e7..84d3a98365 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -56,12 +56,12 @@ class SynapseEvent(JsonEncodedObject): "user_id", # sender/initiator "content", # HTTP body, JSON "state_key", - "required_power_level", "age_ts", "prev_content", "replaces_state", "redacted_because", "origin_server_ts", + "auth_chains", ] internal_keys = [ @@ -70,7 +70,6 @@ class SynapseEvent(JsonEncodedObject): "destinations", "origin", "outlier", - "power_level", "redacted", "prev_events", "hashes", diff --git a/synapse/api/events/factory.py b/synapse/api/events/factory.py index 9134c82eff..a1ec708a81 100644 --- a/synapse/api/events/factory.py +++ b/synapse/api/events/factory.py @@ -16,8 +16,8 @@ from synapse.api.events.room import ( RoomTopicEvent, MessageEvent, RoomMemberEvent, FeedbackEvent, InviteJoinEvent, RoomConfigEvent, RoomNameEvent, GenericEvent, - RoomPowerLevelsEvent, RoomJoinRulesEvent, RoomOpsPowerLevelsEvent, - RoomCreateEvent, RoomAddStateLevelEvent, RoomSendEventLevelEvent, + RoomPowerLevelsEvent, RoomJoinRulesEvent, + RoomCreateEvent, RoomRedactionEvent, ) @@ -39,9 +39,6 @@ class EventFactory(object): RoomPowerLevelsEvent, RoomJoinRulesEvent, RoomCreateEvent, - RoomAddStateLevelEvent, - RoomSendEventLevelEvent, - RoomOpsPowerLevelsEvent, RoomRedactionEvent, ] diff --git a/synapse/api/events/room.py b/synapse/api/events/room.py index cd936074fc..25bc883706 100644 --- a/synapse/api/events/room.py +++ b/synapse/api/events/room.py @@ -153,28 +153,6 @@ class RoomPowerLevelsEvent(SynapseStateEvent): def get_content_template(self): return {} - -class RoomAddStateLevelEvent(SynapseStateEvent): - TYPE = "m.room.add_state_level" - - def get_content_template(self): - return {} - - -class RoomSendEventLevelEvent(SynapseStateEvent): - TYPE = "m.room.send_event_level" - - def get_content_template(self): - return {} - - -class RoomOpsPowerLevelsEvent(SynapseStateEvent): - TYPE = "m.room.ops_levels" - - def get_content_template(self): - return {} - - class RoomAliasesEvent(SynapseStateEvent): TYPE = "m.room.aliases" diff --git a/synapse/api/events/utils.py b/synapse/api/events/utils.py index 31601fd3a9..5fc79105b5 100644 --- a/synapse/api/events/utils.py +++ b/synapse/api/events/utils.py @@ -15,7 +15,6 @@ from .room import ( RoomMemberEvent, RoomJoinRulesEvent, RoomPowerLevelsEvent, - RoomAddStateLevelEvent, RoomSendEventLevelEvent, RoomOpsPowerLevelsEvent, RoomAliasesEvent, RoomCreateEvent, ) @@ -52,17 +51,17 @@ def _prune_event_or_pdu(event_type, event): elif event_type == RoomJoinRulesEvent.TYPE: add_fields("join_rule") elif event_type == RoomPowerLevelsEvent.TYPE: - # TODO: Actually check these are valid user_ids etc. - add_fields("default") - for k, v in event.content.items(): - if k.startswith("@") and isinstance(v, (int, long)): - new_content[k] = v - elif event_type == RoomAddStateLevelEvent.TYPE: - add_fields("level") - elif event_type == RoomSendEventLevelEvent.TYPE: - add_fields("level") - elif event_type == RoomOpsPowerLevelsEvent.TYPE: - add_fields("kick_level", "ban_level", "redact_level") + add_fields( + "users", + "users_default", + "events", + "events_default", + "events_default", + "state_default", + "ban", + "kick", + "redact", + ) elif event_type == RoomAliasesEvent.TYPE: add_fields("aliases") diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index 55c893eb58..42a6c9f9bf 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -21,8 +21,7 @@ from synapse.api.constants import Membership, JoinRules from synapse.api.errors import StoreError, SynapseError from synapse.api.events.room import ( RoomMemberEvent, RoomCreateEvent, RoomPowerLevelsEvent, - RoomJoinRulesEvent, RoomAddStateLevelEvent, RoomTopicEvent, - RoomSendEventLevelEvent, RoomOpsPowerLevelsEvent, RoomNameEvent, + RoomTopicEvent, RoomNameEvent, RoomJoinRulesEvent, ) from synapse.util import stringutils from ._base import BaseHandler @@ -139,7 +138,6 @@ class RoomCreationHandler(BaseHandler): etype=RoomNameEvent.TYPE, room_id=room_id, user_id=user_id, - required_power_level=50, content={"name": name}, ) @@ -151,7 +149,6 @@ class RoomCreationHandler(BaseHandler): etype=RoomTopicEvent.TYPE, room_id=room_id, user_id=user_id, - required_power_level=50, content={"topic": topic}, ) @@ -196,7 +193,6 @@ class RoomCreationHandler(BaseHandler): event_keys = { "room_id": room_id, "user_id": creator.to_string(), - "required_power_level": 100, } def create(etype, **content): @@ -213,7 +209,21 @@ class RoomCreationHandler(BaseHandler): power_levels_event = self.event_factory.create_event( etype=RoomPowerLevelsEvent.TYPE, - content={creator.to_string(): 100, "default": 0}, + content={ + "users": { + creator.to_string(): 100, + }, + "users_default": 0, + "events": { + RoomNameEvent.TYPE: 100, + RoomPowerLevelsEvent.TYPE: 100, + }, + "events_default": 0, + "state_default": 50, + "ban": 50, + "kick": 50, + "redact": 50 + }, **event_keys ) @@ -223,30 +233,10 @@ class RoomCreationHandler(BaseHandler): join_rule=join_rule, ) - add_state_event = create( - etype=RoomAddStateLevelEvent.TYPE, - level=100, - ) - - send_event = create( - etype=RoomSendEventLevelEvent.TYPE, - level=0, - ) - - ops = create( - etype=RoomOpsPowerLevelsEvent.TYPE, - ban_level=50, - kick_level=50, - redact_level=50, - ) - return [ creation_event, power_levels_event, join_rules_event, - add_state_event, - send_event, - ops, ] @@ -388,16 +378,6 @@ class RoomMemberHandler(BaseHandler): else: # This is not a JOIN, so we can handle it normally. - # If we're banning someone, set a req power level - if event.membership == Membership.BAN: - if not hasattr(event, "required_power_level") or event.required_power_level is None: - # Add some default required_power_level - user_level = yield self.store.get_power_level( - event.room_id, - event.user_id, - ) - event.required_power_level = user_level - if prev_state and prev_state.membership == event.membership: # double same action, treat this event as a NOOP. defer.returnValue({}) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 2d62fc2ed0..2a1970914f 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -17,13 +17,9 @@ from twisted.internet import defer from synapse.api.events.room import ( RoomMemberEvent, RoomTopicEvent, FeedbackEvent, -# RoomConfigEvent, RoomNameEvent, RoomJoinRulesEvent, RoomPowerLevelsEvent, - RoomAddStateLevelEvent, - RoomSendEventLevelEvent, - RoomOpsPowerLevelsEvent, RoomRedactionEvent, ) @@ -166,14 +162,6 @@ class DataStore(RoomMemberStore, RoomStore, self._store_room_topic_txn(txn, event) elif event.type == RoomJoinRulesEvent.TYPE: self._store_join_rule(txn, event) - elif event.type == RoomPowerLevelsEvent.TYPE: - self._store_power_levels(txn, event) - elif event.type == RoomAddStateLevelEvent.TYPE: - self._store_add_state_level(txn, event) - elif event.type == RoomSendEventLevelEvent.TYPE: - self._store_send_event_level(txn, event) - elif event.type == RoomOpsPowerLevelsEvent.TYPE: - self._store_ops_level(txn, event) elif event.type == RoomRedactionEvent.TYPE: self._store_redaction(txn, event) diff --git a/synapse/storage/room.py b/synapse/storage/room.py index 7e48ce9cc3..0c83c11ad3 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -148,85 +148,6 @@ class RoomStore(SQLBaseStore): else: defer.returnValue(None) - def get_power_level(self, room_id, user_id): - return self.runInteraction( - "get_power_level", - self._get_power_level, - room_id, user_id, - ) - - def _get_power_level(self, txn, room_id, user_id): - sql = ( - "SELECT level FROM room_power_levels as r " - "INNER JOIN current_state_events as c " - "ON r.event_id = c.event_id " - "WHERE c.room_id = ? AND r.user_id = ? " - ) - - rows = txn.execute(sql, (room_id, user_id,)).fetchall() - - if len(rows) == 1: - return rows[0][0] - - sql = ( - "SELECT level FROM room_default_levels as r " - "INNER JOIN current_state_events as c " - "ON r.event_id = c.event_id " - "WHERE c.room_id = ? " - ) - - rows = txn.execute(sql, (room_id,)).fetchall() - - if len(rows) == 1: - return rows[0][0] - else: - return None - - def get_ops_levels(self, room_id): - return self.runInteraction( - "get_ops_levels", - self._get_ops_levels, - room_id, - ) - - def _get_ops_levels(self, txn, room_id): - sql = ( - "SELECT ban_level, kick_level, redact_level " - "FROM room_ops_levels as r " - "INNER JOIN current_state_events as c " - "ON r.event_id = c.event_id " - "WHERE c.room_id = ? " - ) - - rows = txn.execute(sql, (room_id,)).fetchall() - - if len(rows) == 1: - return OpsLevel(rows[0][0], rows[0][1], rows[0][2]) - else: - return OpsLevel(None, None) - - def get_add_state_level(self, room_id): - return self._get_level_from_table("room_add_state_levels", room_id) - - def get_send_event_level(self, room_id): - return self._get_level_from_table("room_send_event_levels", room_id) - - @defer.inlineCallbacks - def _get_level_from_table(self, table, room_id): - sql = ( - "SELECT level FROM %(table)s as r " - "INNER JOIN current_state_events as c " - "ON r.event_id = c.event_id " - "WHERE c.room_id = ? " - ) % {"table": table} - - rows = yield self._execute(None, sql, room_id) - - if len(rows) == 1: - defer.returnValue(rows[0][0]) - else: - defer.returnValue(None) - def _store_room_topic_txn(self, txn, event): self._simple_insert_txn( txn, @@ -260,84 +181,6 @@ class RoomStore(SQLBaseStore): }, ) - def _store_power_levels(self, txn, event): - for user_id, level in event.content.items(): - if user_id == "default": - self._simple_insert_txn( - txn, - "room_default_levels", - { - "event_id": event.event_id, - "room_id": event.room_id, - "level": level, - }, - ) - else: - self._simple_insert_txn( - txn, - "room_power_levels", - { - "event_id": event.event_id, - "room_id": event.room_id, - "user_id": user_id, - "level": level - }, - ) - - def _store_default_level(self, txn, event): - self._simple_insert_txn( - txn, - "room_default_levels", - { - "event_id": event.event_id, - "room_id": event.room_id, - "level": event.content["default_level"], - }, - ) - - def _store_add_state_level(self, txn, event): - self._simple_insert_txn( - txn, - "room_add_state_levels", - { - "event_id": event.event_id, - "room_id": event.room_id, - "level": event.content["level"], - }, - ) - - def _store_send_event_level(self, txn, event): - self._simple_insert_txn( - txn, - "room_send_event_levels", - { - "event_id": event.event_id, - "room_id": event.room_id, - "level": event.content["level"], - }, - ) - - def _store_ops_level(self, txn, event): - content = { - "event_id": event.event_id, - "room_id": event.room_id, - } - - if "kick_level" in event.content: - content["kick_level"] = event.content["kick_level"] - - if "ban_level" in event.content: - content["ban_level"] = event.content["ban_level"] - - if "redact_level" in event.content: - content["redact_level"] = event.content["redact_level"] - - self._simple_insert_txn( - txn, - "room_ops_levels", - content, - ) - class RoomsTable(Table): table_name = "rooms" -- cgit 1.5.1 From bf6b72eb558cca94e209a541188079750bfefea0 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 6 Nov 2014 18:42:18 +0000 Subject: Start implementing auth chains --- synapse/api/auth.py | 3 +- synapse/api/events/__init__.py | 2 +- synapse/handlers/_base.py | 59 ++++++++++++++++++++++++++++++++-- synapse/storage/__init__.py | 12 ++++++- synapse/storage/_base.py | 2 ++ synapse/storage/event_federation.py | 21 ++++++++++++ synapse/storage/schema/event_edges.sql | 10 ++++++ synapse/storage/signatures.py | 12 +++++++ 8 files changed, 115 insertions(+), 6 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index bb25c4ec55..e1302553d7 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -21,8 +21,7 @@ from synapse.api.constants import Membership, JoinRules from synapse.api.errors import AuthError, StoreError, Codes, SynapseError from synapse.api.events.room import ( RoomMemberEvent, RoomPowerLevelsEvent, RoomRedactionEvent, - RoomJoinRulesEvent, InviteJoinEvent, - RoomCreateEvent, + RoomJoinRulesEvent, RoomCreateEvent, ) from synapse.util.logutils import log_function diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 84d3a98365..513a48f568 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -61,7 +61,7 @@ class SynapseEvent(JsonEncodedObject): "replaces_state", "redacted_because", "origin_server_ts", - "auth_chains", + "auth_events", ] internal_keys = [ diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index 509f7b550c..2613fa7fce 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -14,11 +14,15 @@ # limitations under the License. from twisted.internet import defer -from synapse.api.errors import LimitExceededError +from synapse.api.errors import LimitExceededError from synapse.util.async import run_on_reactor - from synapse.crypto.event_signing import add_hashes_and_signatures +from synapse.api.events.room import ( + RoomCreateEvent, RoomMemberEvent, RoomPowerLevelsEvent, RoomJoinRulesEvent, +) +from synapse.api.constants import Membership, JoinRules +from syutil.base64util import encode_base64 import logging @@ -55,6 +59,53 @@ class BaseHandler(object): retry_after_ms=int(1000*(time_allowed - time_now)), ) + @defer.inlineCallbacks + def _add_auth(self, event): + if event.type == RoomCreateEvent.TYPE: + event.auth_events = [] + return + + auth_events = [] + + key = (RoomPowerLevelsEvent.TYPE, "", ) + power_level_event = event.old_state_events.get(key) + + if power_level_event: + auth_events.append(power_level_event.event_id) + + key = (RoomJoinRulesEvent.TYPE, "", ) + join_rule_event = event.old_state_events.get(key) + + key = (RoomMemberEvent.TYPE, event.user_id, ) + member_event = event.old_state_events.get(key) + + if join_rule_event: + join_rule = join_rule_event.content.get("join_rule") + is_public = join_rule == JoinRules.PUBLIC if join_rule else False + + if event.type == RoomMemberEvent.TYPE: + if event.content["membership"] == Membership.JOIN: + if is_public: + auth_events.append(join_rule_event.event_id) + elif member_event: + auth_events.append(member_event.event_id) + + if member_event: + if member_event.content["membership"] == Membership.JOIN: + auth_events.append(member_event.event_id) + + hashes = yield self.store.get_event_reference_hashes( + auth_events + ) + hashes = [ + { + k: encode_base64(v) for k, v in h.items() + if k == "sha256" + } + for h in hashes + ] + event.auth_events = zip(auth_events, hashes) + @defer.inlineCallbacks def _on_new_room_event(self, event, snapshot, extra_destinations=[], extra_users=[], suppress_auth=False): @@ -64,6 +115,8 @@ class BaseHandler(object): yield self.state_handler.annotate_state_groups(event) + yield self._add_auth(event) + logger.debug("Signing event...") add_hashes_and_signatures( @@ -76,6 +129,8 @@ class BaseHandler(object): logger.debug("Authing...") self.auth.check(event, raises=True) logger.debug("Authed") + else: + logger.debug("Suppressed auth.") yield self.store.persist_event(event) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 2a1970914f..48ad4d864f 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -19,7 +19,6 @@ from synapse.api.events.room import ( RoomMemberEvent, RoomTopicEvent, FeedbackEvent, RoomNameEvent, RoomJoinRulesEvent, - RoomPowerLevelsEvent, RoomRedactionEvent, ) @@ -302,6 +301,17 @@ class DataStore(RoomMemberStore, RoomStore, txn, event.event_id, prev_event_id, alg, hash_bytes ) + for auth_id, _ in event.auth_events: + self._simple_insert_txn( + txn, + table="event_auth", + values={ + "event_id": event.event_id, + "room_id": event.room_id, + "auth_id": auth_id, + }, + ) + (ref_alg, ref_hash_bytes) = compute_event_reference_hash(event) self._store_event_reference_hash_txn( txn, event.event_id, ref_alg, ref_hash_bytes diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 7821fc4726..9aa404695d 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -474,6 +474,8 @@ class SQLBaseStore(object): if is_state == 0 ] + ev.auth_events = self._get_auth_events(txn, ev.event_id) + if hasattr(ev, "state_key"): ev.prev_state = [ (e_id, h) diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 180a764134..86c68ebf87 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -139,6 +139,27 @@ class EventFederationStore(SQLBaseStore): return results + def _get_auth_events(self, txn, event_id): + auth_ids = self._simple_select_onecol_txn( + txn, + table="event_auth", + keyvalues={ + "event_id": event_id, + }, + retcol="auth_id", + ) + + results = [] + for auth_id in auth_ids: + hashes = self._get_event_reference_hashes_txn(txn, auth_id) + prev_hashes = { + k: encode_base64(v) for k, v in hashes.items() + if k == "sha256" + } + results.append((auth_id, prev_hashes)) + + return results + def get_min_depth(self, room_id): return self.runInteraction( "get_min_depth", diff --git a/synapse/storage/schema/event_edges.sql b/synapse/storage/schema/event_edges.sql index 51695826a8..be1c72a775 100644 --- a/synapse/storage/schema/event_edges.sql +++ b/synapse/storage/schema/event_edges.sql @@ -63,3 +63,13 @@ CREATE INDEX IF NOT EXISTS st_extrem_keys ON state_forward_extremities( ); CREATE INDEX IF NOT EXISTS st_extrem_id ON state_forward_extremities(event_id); + +CREATE TABLE IF NOT EXISTS event_auth( + event_id TEXT NOT NULL, + auth_id TEXT NOT NULL, + room_id TEXT NOT NULL, + CONSTRAINT uniqueness UNIQUE (event_id, auth_id, room_id) +); + +CREATE INDEX IF NOT EXISTS evauth_edges_id ON event_auth(event_id); +CREATE INDEX IF NOT EXISTS evauth_edges_auth_id ON event_auth(auth_id); \ No newline at end of file diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index b4b3d5d7ea..84a49088a2 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.py @@ -55,6 +55,18 @@ class SignatureStore(SQLBaseStore): or_ignore=True, ) + def get_event_reference_hashes(self, event_ids): + def f(txn): + return [ + self._get_event_reference_hashes_txn(txn, ev) + for ev in event_ids + ] + + return self.runInteraction( + "get_event_reference_hashes", + f + ) + def _get_event_reference_hashes_txn(self, txn, event_id): """Get all the hashes for a given PDU. Args: -- cgit 1.5.1 From 49948d72f3c76f0ca32b844955d3add7922180ea Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 7 Nov 2014 10:42:44 +0000 Subject: Fix joining over federation --- synapse/api/auth.py | 53 +++++++++++++++++++++++++++++++++++++++-- synapse/api/events/__init__.py | 2 +- synapse/handlers/_base.py | 54 +----------------------------------------- synapse/handlers/federation.py | 1 + synapse/storage/__init__.py | 1 + 5 files changed, 55 insertions(+), 56 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index e1302553d7..d4f284bd60 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -24,6 +24,7 @@ from synapse.api.events.room import ( RoomJoinRulesEvent, RoomCreateEvent, ) from synapse.util.logutils import log_function +from syutil.base64util import encode_base64 import logging @@ -61,8 +62,6 @@ class Auth(object): # FIXME return True - self._can_send_event(event) - if event.type == RoomMemberEvent.TYPE: allowed = self.is_membership_change_allowed(event) if allowed: @@ -71,6 +70,8 @@ class Auth(object): logger.debug("Denying! %s", event) return allowed + self._can_send_event(event) + if event.type == RoomPowerLevelsEvent.TYPE: self._check_power_levels(event) @@ -311,6 +312,54 @@ class Auth(object): def is_server_admin(self, user): return self.store.is_server_admin(user) + @defer.inlineCallbacks + def add_auth_events(self, event): + if event.type == RoomCreateEvent.TYPE: + event.auth_events = [] + return + + auth_events = [] + + key = (RoomPowerLevelsEvent.TYPE, "", ) + power_level_event = event.old_state_events.get(key) + + if power_level_event: + auth_events.append(power_level_event.event_id) + + key = (RoomJoinRulesEvent.TYPE, "", ) + join_rule_event = event.old_state_events.get(key) + + key = (RoomMemberEvent.TYPE, event.user_id, ) + member_event = event.old_state_events.get(key) + + if join_rule_event: + join_rule = join_rule_event.content.get("join_rule") + is_public = join_rule == JoinRules.PUBLIC if join_rule else False + + if event.type == RoomMemberEvent.TYPE: + if event.content["membership"] == Membership.JOIN: + if is_public: + auth_events.append(join_rule_event.event_id) + elif member_event: + auth_events.append(member_event.event_id) + + if member_event: + if member_event.content["membership"] == Membership.JOIN: + auth_events.append(member_event.event_id) + + hashes = yield self.store.get_event_reference_hashes( + auth_events + ) + hashes = [ + { + k: encode_base64(v) for k, v in h.items() + if k == "sha256" + } + for h in hashes + ] + event.auth_events = zip(auth_events, hashes) + + @log_function def _can_send_event(self, event): key = (RoomPowerLevelsEvent.TYPE, "", ) diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 513a48f568..e5980c4be3 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -61,7 +61,6 @@ class SynapseEvent(JsonEncodedObject): "replaces_state", "redacted_because", "origin_server_ts", - "auth_events", ] internal_keys = [ @@ -75,6 +74,7 @@ class SynapseEvent(JsonEncodedObject): "hashes", "signatures", "prev_state", + "auth_events", ] required_keys = [ diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index 2613fa7fce..f630280031 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -18,11 +18,6 @@ from twisted.internet import defer from synapse.api.errors import LimitExceededError from synapse.util.async import run_on_reactor from synapse.crypto.event_signing import add_hashes_and_signatures -from synapse.api.events.room import ( - RoomCreateEvent, RoomMemberEvent, RoomPowerLevelsEvent, RoomJoinRulesEvent, -) -from synapse.api.constants import Membership, JoinRules -from syutil.base64util import encode_base64 import logging @@ -59,53 +54,6 @@ class BaseHandler(object): retry_after_ms=int(1000*(time_allowed - time_now)), ) - @defer.inlineCallbacks - def _add_auth(self, event): - if event.type == RoomCreateEvent.TYPE: - event.auth_events = [] - return - - auth_events = [] - - key = (RoomPowerLevelsEvent.TYPE, "", ) - power_level_event = event.old_state_events.get(key) - - if power_level_event: - auth_events.append(power_level_event.event_id) - - key = (RoomJoinRulesEvent.TYPE, "", ) - join_rule_event = event.old_state_events.get(key) - - key = (RoomMemberEvent.TYPE, event.user_id, ) - member_event = event.old_state_events.get(key) - - if join_rule_event: - join_rule = join_rule_event.content.get("join_rule") - is_public = join_rule == JoinRules.PUBLIC if join_rule else False - - if event.type == RoomMemberEvent.TYPE: - if event.content["membership"] == Membership.JOIN: - if is_public: - auth_events.append(join_rule_event.event_id) - elif member_event: - auth_events.append(member_event.event_id) - - if member_event: - if member_event.content["membership"] == Membership.JOIN: - auth_events.append(member_event.event_id) - - hashes = yield self.store.get_event_reference_hashes( - auth_events - ) - hashes = [ - { - k: encode_base64(v) for k, v in h.items() - if k == "sha256" - } - for h in hashes - ] - event.auth_events = zip(auth_events, hashes) - @defer.inlineCallbacks def _on_new_room_event(self, event, snapshot, extra_destinations=[], extra_users=[], suppress_auth=False): @@ -115,7 +63,7 @@ class BaseHandler(object): yield self.state_handler.annotate_state_groups(event) - yield self._add_auth(event) + yield self.auth.add_auth_events(event) logger.debug("Signing event...") diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index f0448a05d8..09593303a4 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -317,6 +317,7 @@ class FederationHandler(BaseHandler): snapshot.fill_out_prev_events(event) yield self.state_handler.annotate_state_groups(event) + yield self.auth.add_auth_events(event) self.auth.check(event, raises=True) pdu = self.pdu_codec.pdu_from_event(event) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 48ad4d864f..96adf20c89 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -310,6 +310,7 @@ class DataStore(RoomMemberStore, RoomStore, "room_id": event.room_id, "auth_id": auth_id, }, + or_ignore=True, ) (ref_alg, ref_hash_bytes) = compute_event_reference_hash(event) -- cgit 1.5.1 From 97a096b507b92d70a2e07d49122b4f5d93b7dac4 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 7 Nov 2014 11:36:40 +0000 Subject: Add hash of current state to events --- synapse/api/events/__init__.py | 1 + synapse/crypto/event_signing.py | 11 ++++++++++- 2 files changed, 11 insertions(+), 1 deletion(-) (limited to 'synapse/api/events') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index e5980c4be3..8d65c29ac1 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -75,6 +75,7 @@ class SynapseEvent(JsonEncodedObject): "signatures", "prev_state", "auth_events", + "state_hash", ] required_keys = [ diff --git a/synapse/crypto/event_signing.py b/synapse/crypto/event_signing.py index de5d2e7465..7d800615fe 100644 --- a/synapse/crypto/event_signing.py +++ b/synapse/crypto/event_signing.py @@ -45,7 +45,7 @@ def check_event_content_hash(event, hash_algorithm=hashlib.sha256): def _compute_content_hash(event, hash_algorithm): event_json = event.get_full_dict() - #TODO: We need to sign the JSON that is going out via fedaration. + # TODO: We need to sign the JSON that is going out via fedaration. event_json.pop("age_ts", None) event_json.pop("unsigned", None) event_json.pop("signatures", None) @@ -81,6 +81,15 @@ def compute_event_signature(event, signature_name, signing_key): def add_hashes_and_signatures(event, signature_name, signing_key, hash_algorithm=hashlib.sha256): + if hasattr(event, "old_state_events"): + state_json_bytes = encode_canonical_json( + [e.event_id for e in event.old_state_events.values()] + ) + hashed = hash_algorithm(state_json_bytes) + event.state_hash = { + hashed.name: encode_base64(hashed.digest()) + } + hashed = _compute_content_hash(event, hash_algorithm=hash_algorithm) if not hasattr(event, "hashes"): -- cgit 1.5.1 From 1c06806f90a6368cdc3b9fa3b9053021b7c40e94 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 10 Nov 2014 10:21:32 +0000 Subject: Finish redaction algorithm. --- synapse/api/events/__init__.py | 4 ++-- synapse/api/events/utils.py | 39 ++++++++++++++++++++++++++------------- synapse/crypto/event_signing.py | 7 ++----- synapse/federation/units.py | 6 ++---- synapse/storage/_base.py | 2 +- 5 files changed, 33 insertions(+), 25 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 8d65c29ac1..f1e53f23ab 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -86,8 +86,8 @@ class SynapseEvent(JsonEncodedObject): def __init__(self, raises=True, **kwargs): super(SynapseEvent, self).__init__(**kwargs) - if "content" in kwargs: - self.check_json(self.content, raises=raises) + # if "content" in kwargs: + # self.check_json(self.content, raises=raises) def get_content_template(self): """ Retrieve the JSON template for this event as a dict. diff --git a/synapse/api/events/utils.py b/synapse/api/events/utils.py index 5fc79105b5..802648f8f7 100644 --- a/synapse/api/events/utils.py +++ b/synapse/api/events/utils.py @@ -18,24 +18,31 @@ from .room import ( RoomAliasesEvent, RoomCreateEvent, ) + def prune_event(event): - """ Prunes the given event of all keys we don't know about or think could - potentially be dodgy. + """ Returns a pruned version of the given event, which removes all keys we + don't know about or think could potentially be dodgy. This is used when we "redact" an event. We want to remove all fields that the user has specified, but we do want to keep necessary information like type, state_key etc. """ - return _prune_event_or_pdu(event.type, event) - -def prune_pdu(pdu): - """Removes keys that contain unrestricted and non-essential data from a PDU - """ - return _prune_event_or_pdu(pdu.type, pdu) + event_type = event.type -def _prune_event_or_pdu(event_type, event): - # Remove all extraneous fields. - event.unrecognized_keys = {} + allowed_keys = [ + "event_id", + "user_id", + "room_id", + "hashes", + "signatures", + "content", + "type", + "state_key", + "depth", + "prev_events", + "prev_state", + "auth_events", + ] new_content = {} @@ -65,6 +72,12 @@ def _prune_event_or_pdu(event_type, event): elif event_type == RoomAliasesEvent.TYPE: add_fields("aliases") - event.content = new_content + allowed_fields = { + k: v + for k, v in event.get_full_dict().items() + if k in allowed_keys + } + + allowed_fields["content"] = new_content - return event + return type(event)(**allowed_fields) diff --git a/synapse/crypto/event_signing.py b/synapse/crypto/event_signing.py index 7d800615fe..056e8f6ca4 100644 --- a/synapse/crypto/event_signing.py +++ b/synapse/crypto/event_signing.py @@ -55,9 +55,7 @@ def _compute_content_hash(event, hash_algorithm): def compute_event_reference_hash(event, hash_algorithm=hashlib.sha256): - # FIXME(erikj): GenericEvent! - tmp_event = GenericEvent(**event.get_full_dict()) - tmp_event = prune_event(tmp_event) + tmp_event = prune_event(event) event_json = tmp_event.get_dict() event_json.pop("signatures", None) event_json.pop("age_ts", None) @@ -68,8 +66,7 @@ def compute_event_reference_hash(event, hash_algorithm=hashlib.sha256): def compute_event_signature(event, signature_name, signing_key): - tmp_event = copy.deepcopy(event) - tmp_event = prune_event(tmp_event) + tmp_event = prune_event(event) redact_json = tmp_event.get_full_dict() redact_json.pop("signatures", None) redact_json.pop("age_ts", None) diff --git a/synapse/federation/units.py b/synapse/federation/units.py index 2070ffe1e2..d98014cac7 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -56,17 +56,15 @@ class Pdu(JsonEncodedObject): "origin_server_ts", "type", "destinations", - "transaction_id", "prev_events", "depth", "content", - "outlier", "hashes", + "user_id", + "auth_events", "signatures", # Below this are keys valid only for State Pdus. "state_key", "prev_state", - "required_power_level", - "user_id", ] internal_keys = [ diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 9aa404695d..3ab81a78d5 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -509,7 +509,7 @@ class SQLBaseStore(object): ) if del_evs: - prune_event(ev) + ev = prune_event(ev) ev.redacted_because = del_evs[0] return events -- cgit 1.5.1 From 5d439b127ba34b951dfd09a7d3c684c2d50df702 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 10 Nov 2014 13:46:44 +0000 Subject: PEP8 --- synapse/api/auth.py | 3 +-- synapse/api/events/room.py | 1 + synapse/federation/replication.py | 1 - synapse/federation/transport.py | 9 ++++++--- synapse/federation/units.py | 7 +++---- synapse/handlers/federation.py | 5 ++++- synapse/storage/__init__.py | 7 ++++--- synapse/storage/event_federation.py | 9 +++------ 8 files changed, 22 insertions(+), 20 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index 48f9d460a3..a5c6964707 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -369,7 +369,6 @@ class Auth(object): ] event.auth_events = zip(auth_events, hashes) - @log_function def _can_send_event(self, event): key = (RoomPowerLevelsEvent.TYPE, "", ) @@ -452,7 +451,7 @@ class Auth(object): event.user_id, ) - # Check other levels: + # Check other levels: levels_to_check = [ ("users_default", []), ("events_default", []), diff --git a/synapse/api/events/room.py b/synapse/api/events/room.py index 25bc883706..8c4ac45d02 100644 --- a/synapse/api/events/room.py +++ b/synapse/api/events/room.py @@ -153,6 +153,7 @@ class RoomPowerLevelsEvent(SynapseStateEvent): def get_content_template(self): return {} + class RoomAliasesEvent(SynapseStateEvent): TYPE = "m.room.aliases" diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index e798304353..bacba36755 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -549,7 +549,6 @@ class ReplicationLayer(object): origin, pdu.room_id, pdu.event_id, ) - if not backfilled: ret = yield self.handler.on_receive_pdu( pdu, diff --git a/synapse/federation/transport.py b/synapse/federation/transport.py index d84a44c211..95c40c6c1b 100644 --- a/synapse/federation/transport.py +++ b/synapse/federation/transport.py @@ -284,7 +284,7 @@ class TransportLayer(object): origin = None if request.method == "PUT": - #TODO: Handle other method types? other content types? + # TODO: Handle other method types? other content types? try: content_bytes = request.content.read() content = json.loads(content_bytes) @@ -296,11 +296,13 @@ class TransportLayer(object): try: params = auth.split(" ")[1].split(",") param_dict = dict(kv.split("=") for kv in params) + def strip_quotes(value): if value.startswith("\""): return value[1:-1] else: return value + origin = strip_quotes(param_dict["origin"]) key = strip_quotes(param_dict["key"]) sig = strip_quotes(param_dict["sig"]) @@ -321,7 +323,7 @@ class TransportLayer(object): if auth.startswith("X-Matrix"): (origin, key, sig) = parse_auth_header(auth) json_request["origin"] = origin - json_request["signatures"].setdefault(origin,{})[key] = sig + json_request["signatures"].setdefault(origin, {})[key] = sig if not json_request["signatures"]: raise SynapseError( @@ -515,7 +517,8 @@ class TransportLayer(object): return try: - code, response = yield self.received_handler.on_incoming_transaction( + handler = self.received_handler + code, response = yield handler.on_incoming_transaction( transaction_data ) except: diff --git a/synapse/federation/units.py b/synapse/federation/units.py index d98014cac7..f4e7b62bd9 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -192,7 +192,9 @@ class Transaction(JsonEncodedObject): transaction_id and origin_server_ts keys. """ if "origin_server_ts" not in kwargs: - raise KeyError("Require 'origin_server_ts' to construct a Transaction") + 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" @@ -204,6 +206,3 @@ class Transaction(JsonEncodedObject): kwargs["pdus"] = [p.get_dict() for p in pdus] return Transaction(**kwargs) - - - diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 587fa308c8..e909af6bd8 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -521,6 +521,9 @@ class FederationHandler(BaseHandler): @log_function def _on_user_joined(self, user, room_id): - waiters = self.waiting_for_join_list.get((user.to_string(), room_id), []) + waiters = self.waiting_for_join_list.get( + (user.to_string(), room_id), + [] + ) while waiters: waiters.pop().callback(None) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 7d810e6a62..4034437f6b 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -494,11 +494,13 @@ def prepare_database(db_conn): user_version = row[0] if user_version > SCHEMA_VERSION: - raise ValueError("Cannot use this database as it is too " + + raise ValueError( + "Cannot use this database as it is too " + "new for the server to understand" ) elif user_version < SCHEMA_VERSION: - logging.info("Upgrading database from version %d", + logging.info( + "Upgrading database from version %d", user_version ) @@ -520,4 +522,3 @@ def prepare_database(db_conn): c.execute("PRAGMA user_version = %d" % SCHEMA_VERSION) c.close() - diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index a707030145..a027db3868 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -215,7 +215,7 @@ class EventFederationStore(SQLBaseStore): min_depth = self._simple_select_one_onecol_txn( txn, table="room_depth", - keyvalues={"room_id": room_id,}, + keyvalues={"room_id": room_id}, retcol="min_depth", allow_none=True, ) @@ -267,10 +267,8 @@ class EventFederationStore(SQLBaseStore): } ) - - - # We only insert as a forward extremity the new pdu if there are no - # other pdus that reference it as a prev pdu + # We only insert as a forward extremity the new pdu if there are + # no other pdus that reference it as a prev pdu query = ( "INSERT OR IGNORE INTO %(table)s (event_id, room_id) " "SELECT ?, ? WHERE NOT EXISTS (" @@ -312,7 +310,6 @@ class EventFederationStore(SQLBaseStore): ) txn.execute(query) - def get_backfill_events(self, room_id, event_list, limit): """Get a list of Events for a given topic that occured before (and including) the pdus in pdu_list. Return a list of max size `limit`. -- cgit 1.5.1 From a8e565eca8cbfcedbdfd812c98a6545c2fc31afd Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 10 Nov 2014 18:24:43 +0000 Subject: Add an EventValidator. Fix bugs in auth ++ storage --- synapse/api/auth.py | 16 +++++++---- synapse/api/events/__init__.py | 61 ----------------------------------------- synapse/handlers/profile.py | 11 +++++--- synapse/rest/base.py | 2 ++ synapse/rest/room.py | 13 +++++++++ synapse/server.py | 5 ++++ synapse/storage/_base.py | 2 +- synapse/storage/registration.py | 6 +++- synapse/storage/room.py | 38 +++++++++++++------------ 9 files changed, 64 insertions(+), 90 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index a5c6964707..6c2d3db26e 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -70,6 +70,7 @@ class Auth(object): logger.debug("Denying! %s", event) return allowed + self.check_event_sender_in_room(event) self._can_send_event(event) if event.type == RoomPowerLevelsEvent.TYPE: @@ -83,8 +84,10 @@ class Auth(object): else: raise AuthError(500, "Unknown event: %s" % event) except AuthError as e: - logger.info("Event auth check failed on event %s with msg: %s", - event, e.msg) + logger.info( + "Event auth check failed on event %s with msg: %s", + event, e.msg + ) logger.info("Denying! %s", event) if raises: raise e @@ -277,7 +280,7 @@ class Auth(object): default=[""] )[0] if user and access_token and ip_addr: - self.store.insert_client_ip( + yield self.store.insert_client_ip( user=user, access_token=access_token, device_id=user_info["device_id"], @@ -349,7 +352,8 @@ class Auth(object): if event.type == RoomMemberEvent.TYPE: e_type = event.content["membership"] if e_type in [Membership.JOIN, Membership.INVITE]: - auth_events.append(join_rule_event.event_id) + if join_rule_event: + auth_events.append(join_rule_event.event_id) if member_event and not is_public: auth_events.append(member_event.event_id) @@ -405,7 +409,9 @@ class Auth(object): if user_level < send_level: raise AuthError( - 403, "You don't have permission to post that to the room" + 403, + "You don't have permission to post that to the room. " + + "user_level (%d) < send_level (%d)" % (user_level, send_level) ) return True diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index f1e53f23ab..1d8bed2906 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -13,7 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -from synapse.api.errors import SynapseError, Codes from synapse.util.jsonobject import JsonEncodedObject @@ -118,66 +117,6 @@ class SynapseEvent(JsonEncodedObject): """ raise NotImplementedError("get_content_template not implemented.") - def check_json(self, content, raises=True): - """Checks the given JSON content abides by the rules of the template. - - Args: - content : A JSON object to check. - raises: True to raise a SynapseError if the check fails. - Returns: - True if the content passes the template. Returns False if the check - fails and raises=False. - Raises: - SynapseError if the check fails and raises=True. - """ - # recursively call to inspect each layer - err_msg = self._check_json(content, self.get_content_template()) - if err_msg: - if raises: - raise SynapseError(400, err_msg, Codes.BAD_JSON) - else: - return False - else: - return True - - def _check_json(self, content, template): - """Check content and template matches. - - If the template is a dict, each key in the dict will be validated with - the content, else it will just compare the types of content and - template. This basic type check is required because this function will - be recursively called and could be called with just strs or ints. - - Args: - content: The content to validate. - template: The validation template. - Returns: - str: An error message if the validation fails, else None. - """ - if type(content) != type(template): - return "Mismatched types: %s" % template - - if type(template) == dict: - for key in template: - if key not in content: - return "Missing %s key" % key - - if type(content[key]) != type(template[key]): - return "Key %s is of the wrong type (got %s, want %s)" % ( - key, type(content[key]), type(template[key])) - - if type(content[key]) == dict: - # we must go deeper - msg = self._check_json(content[key], template[key]) - if msg: - return msg - elif type(content[key]) == list: - # make sure each item type in content matches the template - for entry in content[key]: - msg = self._check_json(entry, template[key][0]) - if msg: - return msg - class SynapseStateEvent(SynapseEvent): diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index e47814483a..834b37f5f3 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -152,10 +152,13 @@ class ProfileHandler(BaseHandler): if not user.is_mine: defer.returnValue(None) - (displayname, avatar_url) = yield defer.gatherResults([ - self.store.get_profile_displayname(user.localpart), - self.store.get_profile_avatar_url(user.localpart), - ]) + (displayname, avatar_url) = yield defer.gatherResults( + [ + self.store.get_profile_displayname(user.localpart), + self.store.get_profile_avatar_url(user.localpart), + ], + consumeErrors=True + ) state["displayname"] = displayname state["avatar_url"] = avatar_url diff --git a/synapse/rest/base.py b/synapse/rest/base.py index dc784c1527..79fc4dfb84 100644 --- a/synapse/rest/base.py +++ b/synapse/rest/base.py @@ -67,6 +67,8 @@ class RestServlet(object): self.auth = hs.get_auth() self.txns = HttpTransactionStore() + self.validator = hs.get_event_validator() + def register(self, http_server): """ Register this servlet with the given HTTP server. """ if hasattr(self, "PATTERN"): diff --git a/synapse/rest/room.py b/synapse/rest/room.py index 5c9c9d3af4..05da0be090 100644 --- a/synapse/rest/room.py +++ b/synapse/rest/room.py @@ -154,6 +154,9 @@ class RoomStateEventRestServlet(RestServlet): user_id=user.to_string(), state_key=urllib.unquote(state_key) ) + + self.validator.validate(event) + if event_type == RoomMemberEvent.TYPE: # membership events are special handler = self.handlers.room_member_handler @@ -188,6 +191,8 @@ class RoomSendEventRestServlet(RestServlet): content=content ) + self.validator.validate(event) + msg_handler = self.handlers.message_handler yield msg_handler.send_message(event) @@ -253,6 +258,9 @@ class JoinRoomAliasServlet(RestServlet): user_id=user.to_string(), state_key=user.to_string() ) + + self.validator.validate(event) + handler = self.handlers.room_member_handler yield handler.change_membership(event) defer.returnValue((200, {})) @@ -424,6 +432,9 @@ class RoomMembershipRestServlet(RestServlet): user_id=user.to_string(), state_key=state_key ) + + self.validator.validate(event) + handler = self.handlers.room_member_handler yield handler.change_membership(event) defer.returnValue((200, {})) @@ -461,6 +472,8 @@ class RoomRedactEventRestServlet(RestServlet): redacts=urllib.unquote(event_id), ) + self.validator.validate(event) + msg_handler = self.handlers.message_handler yield msg_handler.send_message(event) diff --git a/synapse/server.py b/synapse/server.py index d770b20b19..da0a44433a 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -22,6 +22,7 @@ from synapse.federation import initialize_http_replication from synapse.api.events import serialize_event from synapse.api.events.factory import EventFactory +from synapse.api.events.validator import EventValidator from synapse.notifier import Notifier from synapse.api.auth import Auth from synapse.handlers import Handlers @@ -80,6 +81,7 @@ class BaseHomeServer(object): 'event_sources', 'ratelimiter', 'keyring', + 'event_validator', ] def __init__(self, hostname, **kwargs): @@ -223,6 +225,9 @@ class HomeServer(BaseHomeServer): def build_keyring(self): return Keyring(self) + def build_event_validator(self): + return EventValidator(self) + def register_servlets(self): """ Register all servlets associated with this HomeServer. """ diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 2df64bdfeb..a1ee0318f6 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -212,7 +212,7 @@ class SQLBaseStore(object): retcol : string giving the name of the column to return """ return self.runInteraction( - "_simple_select_one_onecol_txn", + "_simple_select_one_onecol", self._simple_select_one_onecol_txn, table, keyvalues, retcol, allow_none=allow_none, ) diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py index a2ca6f9a69..1f89d77344 100644 --- a/synapse/storage/registration.py +++ b/synapse/storage/registration.py @@ -107,13 +107,17 @@ class RegistrationStore(SQLBaseStore): token ) + @defer.inlineCallbacks def is_server_admin(self, user): - return self._simple_select_one_onecol( + res = yield self._simple_select_one_onecol( table="users", keyvalues={"name": user.to_string()}, retcol="admin", + allow_none=True, ) + defer.returnValue(res if res else False) + def _query_for_auth(self, txn, token): sql = ( "SELECT users.name, users.admin, access_tokens.device_id " diff --git a/synapse/storage/room.py b/synapse/storage/room.py index ca70506d28..cc0513b8d2 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -133,26 +133,28 @@ class RoomStore(SQLBaseStore): defer.returnValue(ret) def _store_room_topic_txn(self, txn, event): - self._simple_insert_txn( - txn, - "topics", - { - "event_id": event.event_id, - "room_id": event.room_id, - "topic": event.topic, - } - ) + if hasattr(event, "topic"): + self._simple_insert_txn( + txn, + "topics", + { + "event_id": event.event_id, + "room_id": event.room_id, + "topic": event.topic, + } + ) def _store_room_name_txn(self, txn, event): - self._simple_insert_txn( - txn, - "room_names", - { - "event_id": event.event_id, - "room_id": event.room_id, - "name": event.name, - } - ) + if hasattr(event, "name"): + self._simple_insert_txn( + txn, + "room_names", + { + "event_id": event.event_id, + "room_id": event.room_id, + "name": event.name, + } + ) class RoomsTable(Table): -- cgit 1.5.1 From 0292d991af70b6b29fa7a2aa14c448e18f93a583 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 11 Nov 2014 08:09:28 +0000 Subject: Add EventValidator module --- synapse/api/events/validator.py | 81 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 81 insertions(+) create mode 100644 synapse/api/events/validator.py (limited to 'synapse/api/events') diff --git a/synapse/api/events/validator.py b/synapse/api/events/validator.py new file mode 100644 index 0000000000..59b486d6d7 --- /dev/null +++ b/synapse/api/events/validator.py @@ -0,0 +1,81 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 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. + +from synapse.api.errors import SynapseError, Codes + + +class EventValidator(object): + def __init__(self, hs): + pass + + def validate(self, event): + """Checks the given JSON content abides by the rules of the template. + + Args: + content : A JSON object to check. + raises: True to raise a SynapseError if the check fails. + Returns: + True if the content passes the template. Returns False if the check + fails and raises=False. + Raises: + SynapseError if the check fails and raises=True. + """ + # recursively call to inspect each layer + err_msg = self._check_json_template( + event.content, + event.get_content_template() + ) + if err_msg: + raise SynapseError(400, err_msg, Codes.BAD_JSON) + else: + return True + + def _check_json_template(self, content, template): + """Check content and template matches. + + If the template is a dict, each key in the dict will be validated with + the content, else it will just compare the types of content and + template. This basic type check is required because this function will + be recursively called and could be called with just strs or ints. + + Args: + content: The content to validate. + template: The validation template. + Returns: + str: An error message if the validation fails, else None. + """ + if type(content) != type(template): + return "Mismatched types: %s" % template + + if type(template) == dict: + for key in template: + if key not in content: + return "Missing %s key" % key + + if type(content[key]) != type(template[key]): + return "Key %s is of the wrong type (got %s, want %s)" % ( + key, type(content[key]), type(template[key])) + + if type(content[key]) == dict: + # we must go deeper + msg = self._check_json(content[key], template[key]) + if msg: + return msg + elif type(content[key]) == list: + # make sure each item type in content matches the template + for entry in content[key]: + msg = self._check_json(entry, template[key][0]) + if msg: + return msg \ No newline at end of file -- cgit 1.5.1 From 2cdff00788a4d6075466dcc638fbef90395c1017 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 11 Nov 2014 10:31:47 +0000 Subject: Fix typo in validator --- synapse/api/events/validator.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/events/validator.py b/synapse/api/events/validator.py index 59b486d6d7..2d4f2a3aa7 100644 --- a/synapse/api/events/validator.py +++ b/synapse/api/events/validator.py @@ -70,12 +70,18 @@ class EventValidator(object): if type(content[key]) == dict: # we must go deeper - msg = self._check_json(content[key], template[key]) + msg = self._check_json_template( + content[key], + template[key] + ) if msg: return msg elif type(content[key]) == list: # make sure each item type in content matches the template for entry in content[key]: - msg = self._check_json(entry, template[key][0]) + msg = self._check_json_template( + entry, + template[key][0] + ) if msg: return msg \ No newline at end of file -- cgit 1.5.1 From cb4b6c844a0c9e2d4a96165958ff5680ed82e160 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 14 Nov 2014 21:25:02 +0000 Subject: Merge PDUs and Events into one object --- synapse/api/events/__init__.py | 6 +++ synapse/api/events/utils.py | 2 + synapse/crypto/event_signing.py | 15 ++----- synapse/federation/pdu_codec.py | 54 ------------------------- synapse/federation/replication.py | 61 +++++++++++++++++++--------- synapse/federation/units.py | 79 +------------------------------------ synapse/handlers/federation.py | 65 +++++++++++------------------- synapse/storage/_base.py | 8 ++++ tests/federation/test_federation.py | 8 ++-- tests/handlers/test_federation.py | 5 ++- 10 files changed, 91 insertions(+), 212 deletions(-) delete mode 100644 synapse/federation/pdu_codec.py (limited to 'synapse/api/events') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 1d8bed2906..63c0bd7ae7 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -117,6 +117,12 @@ class SynapseEvent(JsonEncodedObject): """ raise NotImplementedError("get_content_template not implemented.") + def get_pdu_json(self): + pdu_json = self.get_full_dict() + pdu_json.pop("destination", None) + pdu_json.pop("outlier", None) + return pdu_json + class SynapseStateEvent(SynapseEvent): diff --git a/synapse/api/events/utils.py b/synapse/api/events/utils.py index 802648f8f7..d6019d56eb 100644 --- a/synapse/api/events/utils.py +++ b/synapse/api/events/utils.py @@ -42,6 +42,8 @@ def prune_event(event): "prev_events", "prev_state", "auth_events", + "origin", + "origin_server_ts", ] new_content = {} diff --git a/synapse/crypto/event_signing.py b/synapse/crypto/event_signing.py index 79274fd552..4dff2c0ec2 100644 --- a/synapse/crypto/event_signing.py +++ b/synapse/crypto/event_signing.py @@ -16,7 +16,6 @@ from synapse.api.events.utils import prune_event -from synapse.federation.units import Pdu from syutil.jsonutil import encode_canonical_json from syutil.base64util import encode_base64, decode_base64 from syutil.crypto.jsonsign import sign_json @@ -53,8 +52,7 @@ def check_event_content_hash(event, hash_algorithm=hashlib.sha256): def _compute_content_hash(event, hash_algorithm): - event_json = event.get_full_dict() - # TODO: We need to sign the JSON that is going out via fedaration. + event_json = event.get_pdu_json() event_json.pop("age_ts", None) event_json.pop("unsigned", None) event_json.pop("signatures", None) @@ -67,7 +65,7 @@ def _compute_content_hash(event, hash_algorithm): def compute_event_reference_hash(event, hash_algorithm=hashlib.sha256): tmp_event = prune_event(event) - event_json = tmp_event.get_dict() + event_json = tmp_event.get_pdu_json() event_json.pop("signatures", None) event_json.pop("age_ts", None) event_json.pop("unsigned", None) @@ -78,14 +76,7 @@ def compute_event_reference_hash(event, hash_algorithm=hashlib.sha256): def compute_event_signature(event, signature_name, signing_key): tmp_event = prune_event(event) - tmp_event.origin = event.origin - tmp_event.origin_server_ts = event.origin_server_ts - d = tmp_event.get_full_dict() - kwargs = dict(event.unrecognized_keys) - kwargs.update({k: v for k, v in d.items()}) - tmp_pdu = Pdu(**kwargs) - redact_json = tmp_pdu.get_dict() - redact_json.pop("signatures", None) + redact_json = tmp_event.get_pdu_json() redact_json.pop("age_ts", None) redact_json.pop("unsigned", None) logger.debug("Signing event: %s", redact_json) diff --git a/synapse/federation/pdu_codec.py b/synapse/federation/pdu_codec.py deleted file mode 100644 index 52c84efb5b..0000000000 --- a/synapse/federation/pdu_codec.py +++ /dev/null @@ -1,54 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014 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. - -from .units import Pdu - -import copy - - -class PduCodec(object): - - def __init__(self, hs): - self.signing_key = hs.config.signing_key[0] - self.server_name = hs.hostname - self.event_factory = hs.get_event_factory() - self.clock = hs.get_clock() - self.hs = hs - - def event_from_pdu(self, pdu): - kwargs = {} - - kwargs["etype"] = pdu.type - - kwargs.update({ - k: v - for k, v in pdu.get_full_dict().items() - if k not in [ - "type", - ] - }) - - return self.event_factory.create_event(**kwargs) - - def pdu_from_event(self, event): - d = event.get_full_dict() - - kwargs = copy.deepcopy(event.unrecognized_keys) - kwargs.update({ - k: v for k, v in d.items() - }) - - pdu = Pdu(**kwargs) - return pdu diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index a07e307849..8ee74de005 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -19,7 +19,7 @@ a given transport. from twisted.internet import defer -from .units import Transaction, Pdu, Edu +from .units import Transaction, Edu from .persistence import TransactionActions @@ -72,6 +72,8 @@ class ReplicationLayer(object): self._clock = hs.get_clock() + self.event_factory = hs.get_event_factory() + def set_handler(self, handler): """Sets the handler that the replication layer will use to communicate receipt of new PDUs from other home servers. The required methods are @@ -203,7 +205,10 @@ class ReplicationLayer(object): transaction = Transaction(**transaction_data) - pdus = [Pdu(outlier=False, **p) for p in transaction.pdus] + pdus = [ + self.event_from_pdu_json(p, outlier=False) + for p in transaction.pdus + ] for pdu in pdus: yield self._handle_new_pdu(dest, pdu, backfilled=True) @@ -235,7 +240,10 @@ class ReplicationLayer(object): transaction = Transaction(**transaction_data) - pdu_list = [Pdu(outlier=outlier, **p) for p in transaction.pdus] + pdu_list = [ + self.event_from_pdu_json(p, outlier=outlier) + for p in transaction.pdus + ] pdu = None if pdu_list: @@ -265,8 +273,10 @@ class ReplicationLayer(object): ) transaction = Transaction(**transaction_data) - - pdus = [Pdu(outlier=True, **p) for p in transaction.pdus] + pdus = [ + self.event_from_pdu_json(p, outlier=True) + for p in transaction.pdus + ] defer.returnValue(pdus) @@ -293,7 +303,9 @@ class ReplicationLayer(object): p["age_ts"] = int(self._clock.time_msec()) - int(p["age"]) del p["age"] - pdu_list = [Pdu(**p) for p in transaction.pdus] + pdu_list = [ + self.event_from_pdu_json(p) for p in transaction.pdus + ] logger.debug("[%s] Got transaction", transaction.transaction_id) @@ -388,30 +400,30 @@ class ReplicationLayer(object): def on_make_join_request(self, context, user_id): pdu = yield self.handler.on_make_join_request(context, user_id) defer.returnValue({ - "event": pdu.get_dict(), + "event": pdu.get_pdu_json(), }) @defer.inlineCallbacks def on_invite_request(self, origin, content): - pdu = Pdu(**content) + pdu = self.event_from_pdu_json(content) ret_pdu = yield self.handler.on_invite_request(origin, pdu) defer.returnValue( ( 200, { - "event": ret_pdu.get_dict(), + "event": ret_pdu.get_pdu_json(), } ) ) @defer.inlineCallbacks def on_send_join_request(self, origin, content): - pdu = Pdu(**content) + pdu = self.event_from_pdu_json(content) res_pdus = yield self.handler.on_send_join_request(origin, pdu) defer.returnValue((200, { - "state": [p.get_dict() for p in res_pdus["state"]], - "auth_chain": [p.get_dict() for p in res_pdus["auth_chain"]], + "state": [p.get_pdu_json() for p in res_pdus["state"]], + "auth_chain": [p.get_pdu_json() for p in res_pdus["auth_chain"]], })) @defer.inlineCallbacks @@ -421,7 +433,7 @@ class ReplicationLayer(object): ( 200, { - "auth_chain": [a.get_dict() for a in auth_pdus], + "auth_chain": [a.get_pdu_json() for a in auth_pdus], } ) ) @@ -438,7 +450,7 @@ class ReplicationLayer(object): logger.debug("Got response to make_join: %s", pdu_dict) - defer.returnValue(Pdu(**pdu_dict)) + defer.returnValue(self.event_from_pdu_json(pdu_dict)) @defer.inlineCallbacks def send_join(self, destination, pdu): @@ -446,12 +458,15 @@ class ReplicationLayer(object): destination, pdu.room_id, pdu.event_id, - pdu.get_dict(), + pdu.get_pdu_json(), ) logger.debug("Got content: %s", content) - state = [Pdu(outlier=True, **p) for p in content.get("state", [])] + state = [ + self.event_from_pdu_json(p, outlier=True) + for p in content.get("state", []) + ] # FIXME: We probably want to do something with the auth_chain given # to us @@ -468,14 +483,14 @@ class ReplicationLayer(object): destination=destination, context=context, event_id=event_id, - content=pdu.get_dict(), + content=pdu.get_pdu_json(), ) pdu_dict = content["event"] logger.debug("Got response to send_invite: %s", pdu_dict) - defer.returnValue(Pdu(**pdu_dict)) + defer.returnValue(self.event_from_pdu_json(pdu_dict)) @log_function def _get_persisted_pdu(self, origin, event_id): @@ -490,7 +505,7 @@ class ReplicationLayer(object): """Returns a new Transaction containing the given PDUs suitable for transmission. """ - pdus = [p.get_dict() for p in pdu_list] + pdus = [p.get_pdu_json() for p in pdu_list] time_now = self._clock.time_msec() for p in pdus: if "age_ts" in p: @@ -563,6 +578,14 @@ class ReplicationLayer(object): def __str__(self): return "" % self.server_name + def event_from_pdu_json(self, pdu_json, outlier=False): + #TODO: Check we have all the PDU keys here + pdu_json.setdefault("hashes", {}) + pdu_json.setdefault("signatures", {}) + return self.event_factory.create_event( + pdu_json["type"], outlier=outlier, **pdu_json + ) + class _TransactionQueue(object): """This class makes sure we only have one transaction in flight at diff --git a/synapse/federation/units.py b/synapse/federation/units.py index 70412439cd..6e708edb8c 100644 --- a/synapse/federation/units.py +++ b/synapse/federation/units.py @@ -25,83 +25,6 @@ import logging logger = logging.getLogger(__name__) -class Pdu(JsonEncodedObject): - """ A Pdu represents a piece of data sent from a server and is associated - with a context. - - A Pdu can be classified as "state". For a given context, we can efficiently - retrieve all state pdu's that haven't been clobbered. Clobbering is done - via a unique constraint on the tuple (context, type, state_key). A pdu - is a state pdu if `is_state` is True. - - Example pdu:: - - { - "event_id": "$78c:example.com", - "origin_server_ts": 1404835423000, - "origin": "bar", - "prev_ids": [ - ["23b", "foo"], - ["56a", "bar"], - ], - "content": { ... }, - } - - """ - - valid_keys = [ - "event_id", - "room_id", - "origin", - "origin_server_ts", - "type", - "destinations", - "prev_events", - "depth", - "content", - "hashes", - "user_id", - "auth_events", - "signatures", # Below this are keys valid only for State Pdus. - "state_key", - "prev_state", - ] - - internal_keys = [ - "destinations", - "transaction_id", - "outlier", - ] - - required_keys = [ - "event_id", - "room_id", - "origin", - "origin_server_ts", - "type", - "content", - ] - - # TODO: We need to make this properly load content rather than - # just leaving it as a dict. (OR DO WE?!) - - def __init__(self, destinations=[], prev_events=[], - outlier=False, hashes={}, signatures={}, **kwargs): - super(Pdu, self).__init__( - destinations=destinations, - prev_events=prev_events, - outlier=outlier, - hashes=hashes, - signatures=signatures, - **kwargs - ) - - def __str__(self): - return "(%s, %s)" % (self.__class__.__name__, repr(self.__dict__)) - - def __repr__(self): - return "<%s, %s>" % (self.__class__.__name__, repr(self.__dict__)) - class Edu(JsonEncodedObject): """ An Edu represents a piece of data sent from one homeserver to another. @@ -202,6 +125,6 @@ class Transaction(JsonEncodedObject): for p in pdus: p.transaction_id = kwargs["transaction_id"] - kwargs["pdus"] = [p.get_dict() for p in pdus] + kwargs["pdus"] = [p.get_pdu_json() for p in pdus] return Transaction(**kwargs) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index fc00128c56..da38f34e6a 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -22,7 +22,6 @@ from synapse.api.errors import AuthError, FederationError, SynapseError from synapse.api.events.room import RoomMemberEvent from synapse.api.constants import Membership from synapse.util.logutils import log_function -from synapse.federation.pdu_codec import PduCodec from synapse.util.async import run_on_reactor from synapse.crypto.event_signing import ( compute_event_signature, check_event_content_hash @@ -69,8 +68,6 @@ class FederationHandler(BaseHandler): self.replication_layer.set_handler(self) - self.pdu_codec = PduCodec(hs) - # When joining a room we need to queue any events for that room up self.room_queues = {} @@ -92,7 +89,7 @@ class FederationHandler(BaseHandler): yield run_on_reactor() - pdu = self.pdu_codec.pdu_from_event(event) + pdu = event if not hasattr(pdu, "destinations") or not pdu.destinations: pdu.destinations = [] @@ -105,7 +102,7 @@ class FederationHandler(BaseHandler): """ Called by the ReplicationLayer when we have a new pdu. We need to do auth checks and put it through the StateHandler. """ - event = self.pdu_codec.event_from_pdu(pdu) + event = pdu logger.debug("Got event: %s", event.event_id) @@ -118,18 +115,15 @@ class FederationHandler(BaseHandler): logger.debug("Processing event: %s", event.event_id) redacted_event = prune_event(event) - redacted_event.origin = pdu.origin - redacted_event.origin_server_ts = pdu.origin_server_ts - redacted_pdu = self.pdu_codec.pdu_from_event(redacted_event) - redacted_pdu_json = redacted_pdu.get_dict() + redacted_pdu_json = redacted_event.get_pdu_json() try: yield self.keyring.verify_json_for_server( event.origin, redacted_pdu_json ) except SynapseError as e: logger.warn("Signature check failed for %s redacted to %s", - encode_canonical_json(pdu.get_dict()), + encode_canonical_json(pdu.get_pdu_json()), encode_canonical_json(redacted_pdu_json), ) raise FederationError( @@ -147,7 +141,7 @@ class FederationHandler(BaseHandler): event = redacted_event if state: - state = [self.pdu_codec.event_from_pdu(p) for p in state] + state = [p for p in state] is_new_state = yield self.state_handler.annotate_event_with_state( event, @@ -239,7 +233,7 @@ class FederationHandler(BaseHandler): events = [] for pdu in pdus: - event = self.pdu_codec.event_from_pdu(pdu) + event = pdu # FIXME (erikj): Not sure this actually works :/ yield self.state_handler.annotate_event_with_state(event) @@ -260,15 +254,15 @@ class FederationHandler(BaseHandler): destination=target_host, context=event.room_id, event_id=event.event_id, - pdu=self.pdu_codec.pdu_from_event(event) + pdu=event ) - defer.returnValue(self.pdu_codec.event_from_pdu(pdu)) + defer.returnValue(pdu) @defer.inlineCallbacks def on_event_auth(self, event_id): auth = yield self.store.get_auth_chain(event_id) - defer.returnValue([self.pdu_codec.pdu_from_event(e) for e in auth]) + defer.returnValue([e for e in auth]) @log_function @defer.inlineCallbacks @@ -292,7 +286,7 @@ class FederationHandler(BaseHandler): logger.debug("Got response to make_join: %s", pdu) - event = self.pdu_codec.event_from_pdu(pdu) + event = pdu # We should assert some things. assert(event.type == RoomMemberEvent.TYPE) @@ -310,10 +304,10 @@ class FederationHandler(BaseHandler): state = yield self.replication_layer.send_join( target_host, - self.pdu_codec.pdu_from_event(event) + event ) - state = [self.pdu_codec.event_from_pdu(p) for p in state] + state = [p for p in state] logger.debug("do_invite_join state: %s", state) @@ -387,7 +381,7 @@ class FederationHandler(BaseHandler): yield self.auth.add_auth_events(event) self.auth.check(event, raises=True) - pdu = self.pdu_codec.pdu_from_event(event) + pdu = event defer.returnValue(pdu) @@ -397,7 +391,7 @@ class FederationHandler(BaseHandler): """ We have received a join event for a room. Fully process it and respond with the current state and auth chains. """ - event = self.pdu_codec.event_from_pdu(pdu) + event = pdu event.outlier = False @@ -429,7 +423,7 @@ class FederationHandler(BaseHandler): "user_joined_room", user=user, room_id=event.room_id ) - new_pdu = self.pdu_codec.pdu_from_event(event) + new_pdu = event destinations = set() @@ -450,17 +444,10 @@ class FederationHandler(BaseHandler): yield self.replication_layer.send_pdu(new_pdu) auth_chain = yield self.store.get_auth_chain(event.event_id) - pdu_auth_chain = [ - self.pdu_codec.pdu_from_event(e) - for e in auth_chain - ] defer.returnValue({ - "state": [ - self.pdu_codec.pdu_from_event(e) - for e in event.state_events.values() - ], - "auth_chain": pdu_auth_chain, + "state": event.state_events.values(), + "auth_chain": auth_chain, }) @defer.inlineCallbacks @@ -469,7 +456,7 @@ class FederationHandler(BaseHandler): Respond with the now signed event. """ - event = self.pdu_codec.event_from_pdu(pdu) + event = pdu event.outlier = True @@ -493,7 +480,7 @@ class FederationHandler(BaseHandler): event, extra_users=[target_user], ) - defer.returnValue(self.pdu_codec.pdu_from_event(event)) + defer.returnValue(event) @defer.inlineCallbacks def get_state_for_pdu(self, origin, room_id, event_id): @@ -524,12 +511,7 @@ class FederationHandler(BaseHandler): else: del results[(event.type, event.state_key)] - defer.returnValue( - [ - self.pdu_codec.pdu_from_event(s) - for s in results.values() - ] - ) + defer.returnValue(results.values()) else: defer.returnValue([]) @@ -546,10 +528,7 @@ class FederationHandler(BaseHandler): limit ) - defer.returnValue([ - self.pdu_codec.pdu_from_event(e) - for e in events - ]) + defer.returnValue(events) @defer.inlineCallbacks @log_function @@ -572,7 +551,7 @@ class FederationHandler(BaseHandler): if not in_room: raise AuthError(403, "Host not in room.") - defer.returnValue(self.pdu_codec.pdu_from_event(event)) + defer.returnValue(event) else: defer.returnValue(None) diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 30e6eac8db..5d4be09a82 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -492,6 +492,14 @@ class SQLBaseStore(object): for n, s in signatures.items() } + hashes = self._get_event_content_hashes_txn( + txn, ev.event_id, + ) + + ev.hashes = { + k: encode_base64(v) for k, v in hashes.items() + } + prevs = self._get_prev_events_and_state(txn, ev.event_id) ev.prev_events = [ diff --git a/tests/federation/test_federation.py b/tests/federation/test_federation.py index ad09fab392..efac4075dc 100644 --- a/tests/federation/test_federation.py +++ b/tests/federation/test_federation.py @@ -23,7 +23,7 @@ from ..utils import MockHttpResource, MockClock, MockKey from synapse.server import HomeServer from synapse.federation import initialize_http_replication -from synapse.federation.units import Pdu +from synapse.api.events import SynapseEvent def make_pdu(prev_pdus=[], **kwargs): @@ -40,7 +40,7 @@ def make_pdu(prev_pdus=[], **kwargs): } pdu_fields.update(kwargs) - return Pdu(prev_pdus=prev_pdus, **pdu_fields) + return SynapseEvent(prev_pdus=prev_pdus, **pdu_fields) class FederationTestCase(unittest.TestCase): @@ -169,7 +169,7 @@ class FederationTestCase(unittest.TestCase): (200, "OK") ) - pdu = Pdu( + pdu = SynapseEvent( event_id="abc123def456", origin="red", room_id="my-context", @@ -189,7 +189,7 @@ class FederationTestCase(unittest.TestCase): "origin_server_ts": 1000000, "origin": "test", "pdus": [ - pdu.get_dict(), + pdu.get_pdu_json(), ], 'pdu_failures': [], }, diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py index 3f17ca8fb0..e19b073817 100644 --- a/tests/handlers/test_federation.py +++ b/tests/handlers/test_federation.py @@ -19,9 +19,10 @@ from tests import unittest from synapse.api.events.room import ( MessageEvent, ) + +from synapse.api.events import SynapseEvent from synapse.handlers.federation import FederationHandler from synapse.server import HomeServer -from synapse.federation.units import Pdu from mock import NonCallableMock, ANY, Mock @@ -74,7 +75,7 @@ class FederationTestCase(unittest.TestCase): @defer.inlineCallbacks def test_msg(self): - pdu = Pdu( + pdu = SynapseEvent( type=MessageEvent.TYPE, room_id="foo", content={"msgtype": u"fooo"}, -- cgit 1.5.1 From 428581dd056e40adaca41c3593d79f76339f3220 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Tue, 18 Nov 2014 19:18:36 +0000 Subject: SYN-144: Remove bad keys from pdu json objects, convert age_ts to age for all pdus sent. --- synapse/api/events/__init__.py | 13 ++++++++++++- synapse/federation/replication.py | 31 +++++++++++++++++-------------- 2 files changed, 29 insertions(+), 15 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 63c0bd7ae7..0019789dd1 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -117,10 +117,21 @@ class SynapseEvent(JsonEncodedObject): """ raise NotImplementedError("get_content_template not implemented.") - def get_pdu_json(self): + def get_pdu_json(self, time_now=None): pdu_json = self.get_full_dict() pdu_json.pop("destination", None) pdu_json.pop("outlier", None) + pdu_json.pop("replaces_state", None) + pdu_json.pop("redacted", None) + state_hash = pdu_json.pop("state_hash", None) + if state_hash is not None: + pdu_json.setdefault("unsigned", {})["state_hash"] = state_hash + content = pdu_json.get("content", {}) + content.pop("prev", None) + if time_now is not None and "age_ts" in pdu_json: + age = time_now - pdu_json["age_ts"] + pdu_json.setdefault("unsigned", {})["age"] = int(age) + del pdu_json["age_ts"] return pdu_json diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 8ee74de005..2ed303e12d 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -399,19 +399,21 @@ class ReplicationLayer(object): @defer.inlineCallbacks def on_make_join_request(self, context, user_id): pdu = yield self.handler.on_make_join_request(context, user_id) + time_now = self._clock.time_msec() defer.returnValue({ - "event": pdu.get_pdu_json(), + "event": pdu.get_pdu_json(time_now), }) @defer.inlineCallbacks def on_invite_request(self, origin, content): pdu = self.event_from_pdu_json(content) ret_pdu = yield self.handler.on_invite_request(origin, pdu) + time_now = self._clock.time_msec() defer.returnValue( ( 200, { - "event": ret_pdu.get_pdu_json(), + "event": ret_pdu.get_pdu_json(time_now), } ) ) @@ -420,20 +422,21 @@ class ReplicationLayer(object): def on_send_join_request(self, origin, content): pdu = self.event_from_pdu_json(content) res_pdus = yield self.handler.on_send_join_request(origin, pdu) - + time_now = self._clock.time_msec() defer.returnValue((200, { - "state": [p.get_pdu_json() for p in res_pdus["state"]], - "auth_chain": [p.get_pdu_json() for p in res_pdus["auth_chain"]], + "state": [p.get_pdu_json(time_now) for p in res_pdus["state"]], + "auth_chain": [p.get_pdu_json(time_now) for p in res_pdus["auth_chain"]], })) @defer.inlineCallbacks def on_event_auth(self, origin, context, event_id): + time_now = self._clock.time_msec() auth_pdus = yield self.handler.on_event_auth(event_id) defer.returnValue( ( 200, { - "auth_chain": [a.get_pdu_json() for a in auth_pdus], + "auth_chain": [a.get_pdu_json(time_now) for a in auth_pdus], } ) ) @@ -454,11 +457,12 @@ class ReplicationLayer(object): @defer.inlineCallbacks def send_join(self, destination, pdu): + time_now = self._clock.time_msec() _, content = yield self.transport_layer.send_join( destination, pdu.room_id, pdu.event_id, - pdu.get_pdu_json(), + pdu.get_pdu_json(time_now), ) logger.debug("Got content: %s", content) @@ -479,11 +483,12 @@ class ReplicationLayer(object): @defer.inlineCallbacks def send_invite(self, destination, context, event_id, pdu): + time_now = self._clock.time_msec() code, content = yield self.transport_layer.send_invite( destination=destination, context=context, event_id=event_id, - content=pdu.get_pdu_json(), + content=pdu.get_pdu_json(time_now), ) pdu_dict = content["event"] @@ -505,13 +510,8 @@ class ReplicationLayer(object): """Returns a new Transaction containing the given PDUs suitable for transmission. """ - pdus = [p.get_pdu_json() for p in pdu_list] time_now = self._clock.time_msec() - for p in pdus: - if "age_ts" in p: - age = time_now - p["age_ts"] - p.setdefault("unsigned", {})["age"] = int(age) - del p["age_ts"] + pdus = [p.get_pdu_json(time_now) for p in pdu_list] return Transaction( origin=self.server_name, pdus=pdus, @@ -582,6 +582,9 @@ class ReplicationLayer(object): #TODO: Check we have all the PDU keys here pdu_json.setdefault("hashes", {}) pdu_json.setdefault("signatures", {}) + state_hash = pdu_json.get("unsigned", {}).pop("state_hash", None) + if state_hash is not None: + pdu_json["state_hash"] = state_hash return self.event_factory.create_event( pdu_json["type"], outlier=outlier, **pdu_json ) -- cgit 1.5.1 From c5eabe3143bc3398961ed80da041f28e545386bf Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Wed, 19 Nov 2014 16:38:40 +0000 Subject: replace user_id with sender --- synapse/api/events/__init__.py | 4 +++- synapse/federation/replication.py | 3 +++ tests/federation/test_federation.py | 3 +++ tests/handlers/test_federation.py | 3 ++- 4 files changed, 11 insertions(+), 2 deletions(-) (limited to 'synapse/api/events') diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 0019789dd1..a01c4a1351 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -119,7 +119,7 @@ class SynapseEvent(JsonEncodedObject): def get_pdu_json(self, time_now=None): pdu_json = self.get_full_dict() - pdu_json.pop("destination", None) + pdu_json.pop("destinations", None) pdu_json.pop("outlier", None) pdu_json.pop("replaces_state", None) pdu_json.pop("redacted", None) @@ -132,6 +132,8 @@ class SynapseEvent(JsonEncodedObject): age = time_now - pdu_json["age_ts"] pdu_json.setdefault("unsigned", {})["age"] = int(age) del pdu_json["age_ts"] + user_id = pdu_json.pop("user_id") + pdu_json["sender"] = user_id return pdu_json diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 2ed303e12d..65a53ae17c 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -582,6 +582,9 @@ class ReplicationLayer(object): #TODO: Check we have all the PDU keys here pdu_json.setdefault("hashes", {}) pdu_json.setdefault("signatures", {}) + sender = pdu_json.pop("sender", None) + if sender is not None: + pdu_json["user_id"] = sender state_hash = pdu_json.get("unsigned", {}).pop("state_hash", None) if state_hash is not None: pdu_json["state_hash"] = state_hash diff --git a/tests/federation/test_federation.py b/tests/federation/test_federation.py index efac4075dc..73dd289276 100644 --- a/tests/federation/test_federation.py +++ b/tests/federation/test_federation.py @@ -100,6 +100,7 @@ class FederationTestCase(unittest.TestCase): make_pdu( event_id="the-pdu-id", origin="red", + user_id="@a:red", room_id="my-context", type="m.topic", origin_server_ts=123456789000, @@ -145,6 +146,7 @@ class FederationTestCase(unittest.TestCase): make_pdu( event_id="abc123def456", origin="red", + user_id="@a:red", room_id="my-context", type="m.text", origin_server_ts=123456789001, @@ -172,6 +174,7 @@ class FederationTestCase(unittest.TestCase): pdu = SynapseEvent( event_id="abc123def456", origin="red", + user_id="@a:red", room_id="my-context", type="m.text", origin_server_ts=123456789001, diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py index e19b073817..3487a090e9 100644 --- a/tests/handlers/test_federation.py +++ b/tests/handlers/test_federation.py @@ -81,8 +81,9 @@ class FederationTestCase(unittest.TestCase): content={"msgtype": u"fooo"}, origin_server_ts=0, event_id="$a:b", + user_id="@a:b", origin="b", - hashes={"sha256":"PvbCLWrTBxnBsSO7/cJ76072ySTCgI/XGadESRAe02M"}, + hashes={"sha256":"AcLrgtUIqqwaGoHhrEvYG1YLDIsVPYJdSRGhkp3jJp8"}, ) self.datastore.persist_event.return_value = defer.succeed(None) -- cgit 1.5.1