Merge branch 'release-v0.5.3' of github.com:matrix-org/synapse v0.5.3

6 files changed, 176 insertions, 64 deletions

diff --git a/synapse/__init__.py b/synapse/__init__.py
index d5c2f25484..c99cd96537 100644
--- a/synapse/__init__.py
+++ b/synapse/__init__.py
@@ -16,4 +16,4 @@
 """ This is a reference implementation of a synapse home server.
 """
 
-__version__ = "0.5.2"
+__version__ = "0.5.3"
diff --git a/synapse/api/auth.py b/synapse/api/auth.py
index fb911e51a6..2b0475543d 100644
--- a/synapse/api/auth.py
+++ b/synapse/api/auth.py
@@ -202,7 +202,10 @@ class Auth(object):
 
             # Invites are valid iff caller is in the room and target isn't.
             if not caller_in_room:  # caller isn't joined
-                raise AuthError(403, "You are not in room %s." % event.room_id)
+                raise AuthError(
+                    403,
+                    "%s not in room %s." % (event.user_id, event.room_id,)
+                )
             elif target_in_room:  # the target is already in the room.
                 raise AuthError(403, "%s is already in the room." %
                                      target_user_id)
@@ -225,7 +228,10 @@ class Auth(object):
             # TODO (erikj): Implement kicks.
 
             if not caller_in_room:  # trying to leave a room you aren't joined
-                raise AuthError(403, "You are not in room %s." % event.room_id)
+                raise AuthError(
+                    403,
+                    "%s not in room %s." % (target_user_id, event.room_id,)
+                )
             elif target_user_id != event.user_id:
                 if kick_level:
                     kick_level = int(kick_level)
diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py
index 6bfb30b42d..312d69fcaa 100644
--- a/synapse/federation/replication.py
+++ b/synapse/federation/replication.py
@@ -283,6 +283,22 @@ class ReplicationLayer(object):
 
     @defer.inlineCallbacks
     @log_function
+    def get_event_auth(self, destination, context, event_id):
+        res = yield self.transport_layer.get_event_auth(
+            destination, context, event_id,
+        )
+
+        auth_chain = [
+            self.event_from_pdu_json(p, outlier=True)
+            for p in res["auth_chain"]
+        ]
+
+        auth_chain.sort(key=lambda e: e.depth)
+
+        defer.returnValue(auth_chain)
+
+    @defer.inlineCallbacks
+    @log_function
     def on_backfill_request(self, origin, context, versions, limit):
         pdus = yield self.handler.on_backfill_request(
             origin, context, versions, limit
@@ -549,34 +565,34 @@ class ReplicationLayer(object):
         state = None
 
         # We need to make sure we have all the auth events.
-        for e_id, _ in pdu.auth_events:
-            exists = yield self._get_persisted_pdu(
-                origin,
-                e_id,
-                do_auth=False
-            )
-
-            if not exists:
-                try:
-                    logger.debug(
-                        "_handle_new_pdu fetch missing auth event %s from %s",
-                        e_id,
-                        origin,
-                    )
-
-                    yield self.get_pdu(
-                        origin,
-                        event_id=e_id,
-                        outlier=True,
-                    )
-
-                    logger.debug("Processed pdu %s", e_id)
-                except:
-                    logger.warn(
-                        "Failed to get auth event %s from %s",
-                        e_id,
-                        origin
-                    )
+        # for e_id, _ in pdu.auth_events:
+        #     exists = yield self._get_persisted_pdu(
+        #         origin,
+        #         e_id,
+        #         do_auth=False
+        #     )
+        #
+        #     if not exists:
+        #         try:
+        #             logger.debug(
+        #                 "_handle_new_pdu fetch missing auth event %s from %s",
+        #                 e_id,
+        #                 origin,
+        #             )
+        #
+        #             yield self.get_pdu(
+        #                 origin,
+        #                 event_id=e_id,
+        #                 outlier=True,
+        #             )
+        #
+        #             logger.debug("Processed pdu %s", e_id)
+        #         except:
+        #             logger.warn(
+        #                 "Failed to get auth event %s from %s",
+        #                 e_id,
+        #                 origin
+        #             )
 
         # Get missing pdus if necessary.
         if not pdu.outlier:
@@ -626,6 +642,7 @@ class ReplicationLayer(object):
 
         if not backfilled:
             ret = yield self.handler.on_receive_pdu(
+                origin,
                 pdu,
                 backfilled=backfilled,
                 state=state,
diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py
index d59221a4fb..02202692d4 100644
--- a/synapse/handlers/events.py
+++ b/synapse/handlers/events.py
@@ -53,8 +53,12 @@ class EventStreamHandler(BaseHandler):
             if auth_user not in self._streams_per_user:
                 self._streams_per_user[auth_user] = 0
                 if auth_user in self._stop_timer_per_user:
-                    self.clock.cancel_call_later(
-                        self._stop_timer_per_user.pop(auth_user))
+                    try:
+                        self.clock.cancel_call_later(
+                            self._stop_timer_per_user.pop(auth_user)
+                        )
+                    except:
+                        logger.exception("Failed to cancel event timer")
                 else:
                     yield self.distributor.fire(
                         "started_user_eventstream", auth_user
@@ -95,10 +99,12 @@ class EventStreamHandler(BaseHandler):
                     logger.debug(
                         "_later stopped_user_eventstream %s", auth_user
                     )
+
+                    self._stop_timer_per_user.pop(auth_user, None)
+
                     yield self.distributor.fire(
                         "stopped_user_eventstream", auth_user
                     )
-                    del self._stop_timer_per_user[auth_user]
 
                 logger.debug("Scheduling _later: for %s", auth_user)
                 self._stop_timer_per_user[auth_user] = (
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index fcef602055..925eb5376e 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -101,7 +101,7 @@ class FederationHandler(BaseHandler):
 
     @log_function
     @defer.inlineCallbacks
-    def on_receive_pdu(self, pdu, backfilled, state=None):
+    def on_receive_pdu(self, origin, 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.
         """
@@ -112,7 +112,7 @@ class FederationHandler(BaseHandler):
         # If we are currently in the process of joining this room, then we
         # queue up events for later processing.
         if event.room_id in self.room_queues:
-            self.room_queues[event.room_id].append(pdu)
+            self.room_queues[event.room_id].append((pdu, origin))
             return
 
         logger.debug("Processing event: %s", event.event_id)
@@ -149,14 +149,49 @@ class FederationHandler(BaseHandler):
         # FIXME (erikj): Awful hack to make the case where we are not currently
         # in the room work
         current_state = None
-        if state:
-            is_in_room = yield self.auth.check_host_in_room(
-                event.room_id,
-                self.server_name
+        is_in_room = yield self.auth.check_host_in_room(
+            event.room_id,
+            self.server_name
+        )
+        if not is_in_room and not event.outlier:
+            logger.debug("Got event for room we're not in.")
+
+            replication_layer = self.replication_layer
+            auth_chain = yield replication_layer.get_event_auth(
+                origin,
+                context=event.room_id,
+                event_id=event.event_id,
             )
-            if not is_in_room:
-                logger.debug("Got event for room we're not in.")
-                current_state = state
+
+            for e in auth_chain:
+                e.outlier = True
+                try:
+                    yield self._handle_new_event(e, fetch_missing=False)
+                except:
+                    logger.exception(
+                        "Failed to parse auth event %s",
+                        e.event_id,
+                    )
+
+            if not state:
+                state = yield replication_layer.get_state_for_context(
+                    origin,
+                    context=event.room_id,
+                    event_id=event.event_id,
+                )
+
+            current_state = state
+
+        if state:
+            for e in state:
+                e.outlier = True
+                try:
+                    yield self._handle_new_event(e)
+                except:
+                    logger.exception(
+                        "Failed to parse state event %s",
+                        e.event_id,
+                    )
 
         try:
             yield self._handle_new_event(
@@ -251,6 +286,16 @@ class FederationHandler(BaseHandler):
     @defer.inlineCallbacks
     def on_event_auth(self, event_id):
         auth = yield self.store.get_auth_chain(event_id)
+
+        for event in auth:
+            event.signatures.update(
+                compute_event_signature(
+                    event,
+                    self.hs.hostname,
+                    self.hs.config.signing_key[0]
+                )
+            )
+
         defer.returnValue([e for e in auth])
 
     @log_function
@@ -310,6 +355,7 @@ class FederationHandler(BaseHandler):
 
             state = ret["state"]
             auth_chain = ret["auth_chain"]
+            auth_chain.sort(key=lambda e: e.depth)
 
             logger.debug("do_invite_join auth_chain: %s", auth_chain)
             logger.debug("do_invite_join state: %s", state)
@@ -328,23 +374,32 @@ class FederationHandler(BaseHandler):
 
             for e in auth_chain:
                 e.outlier = True
-                yield self._handle_new_event(e)
-                yield self.notifier.on_new_room_event(
-                    e, extra_users=[joinee]
-                )
+                try:
+                    yield self._handle_new_event(e, fetch_missing=False)
+                except:
+                    logger.exception(
+                        "Failed to parse auth event %s",
+                        e.event_id,
+                    )
 
             for e in state:
                 # FIXME: Auth these.
                 e.outlier = True
-                yield self._handle_new_event(e)
-                yield self.notifier.on_new_room_event(
-                    e, extra_users=[joinee]
-                )
+                try:
+                    yield self._handle_new_event(
+                        e,
+                        fetch_missing=True
+                    )
+                except:
+                    logger.exception(
+                        "Failed to parse state event %s",
+                        e.event_id,
+                    )
 
             yield self._handle_new_event(
                 event,
                 state=state,
-                current_state=state
+                current_state=state,
             )
 
             yield self.notifier.on_new_room_event(
@@ -356,9 +411,9 @@ class FederationHandler(BaseHandler):
             room_queue = self.room_queues[room_id]
             del self.room_queues[room_id]
 
-            for p in room_queue:
+            for p, origin in room_queue:
                 try:
-                    self.on_receive_pdu(p, backfilled=False)
+                    self.on_receive_pdu(origin, p, backfilled=False)
                 except:
                     logger.exception("Couldn't handle pdu")
 
@@ -507,7 +562,17 @@ class FederationHandler(BaseHandler):
                 else:
                     del results[(event.type, event.state_key)]
 
-            defer.returnValue(results.values())
+            res = results.values()
+            for event in res:
+                event.signatures.update(
+                    compute_event_signature(
+                        event,
+                        self.hs.hostname,
+                        self.hs.config.signing_key[0]
+                    )
+                )
+
+            defer.returnValue(res)
         else:
             defer.returnValue([])
 
@@ -540,6 +605,17 @@ class FederationHandler(BaseHandler):
         )
 
         if event:
+            # FIXME: This is a temporary work around where we occasionally
+            # return events slightly differently than when they were
+            # originally signed
+            event.signatures.update(
+                compute_event_signature(
+                    event,
+                    self.hs.hostname,
+                    self.hs.config.signing_key[0]
+                )
+            )
+
             if do_auth:
                 in_room = yield self.auth.check_host_in_room(
                     event.room_id,
@@ -567,11 +643,7 @@ class FederationHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def _handle_new_event(self, event, state=None, backfilled=False,
-                          current_state=None):
-        if state:
-            for s in state:
-                yield self._handle_new_event(s)
-
+                          current_state=None, fetch_missing=True):
         is_new_state = yield self.state_handler.annotate_event_with_state(
             event,
             old_state=state
@@ -611,11 +683,22 @@ class FederationHandler(BaseHandler):
                 )
 
                 if not e:
-                    raise AuthError(
-                        403,
-                        "Can't find auth event %s." % (e_id, )
+                    e = yield self.replication_layer.get_pdu(
+                        event.origin, e_id, outlier=True
                     )
 
+                    if e and fetch_missing:
+                        try:
+                            yield self.on_receive_pdu(event.origin, e, False)
+                        except:
+                            logger.exception(
+                                "Failed to parse auth event %s",
+                                e_id,
+                            )
+
+                if not e:
+                    logger.warn("Can't find auth event %s.", e_id)
+
                 auth_events[(e.type, e.state_key)] = e
 
             if event.type == RoomMemberEvent.TYPE and not event.auth_events:
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index b84735e61c..3405cb365e 100644
--- a/synapse/storage/stream.py
+++ b/synapse/storage/stream.py
@@ -283,7 +283,7 @@ class StreamStore(SQLBaseStore):
 
         sql = (
             "SELECT *, (%(redacted)s) AS redacted FROM events "
-            "WHERE room_id = ? AND stream_ordering <= ? "
+            "WHERE room_id = ? AND stream_ordering <= ? AND outlier = 0 "
             "ORDER BY topological_ordering DESC, stream_ordering DESC LIMIT ? "
         ) % {
             "redacted": del_sql,