From c92d64a6c35713aabaed11e8ef1e62d2fb84a875 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 26 Jan 2015 14:33:11 +0000
Subject: [PATCH 01/15] Make it the responsibility of the replication layer to
 check signature and hashes.

---
 synapse/federation/federation_client.py | 108 +++++++++++++++++++++---
 synapse/federation/federation_server.py |  89 ++++++++++++++++---
 synapse/federation/replication.py       |   2 +
 3 files changed, 173 insertions(+), 26 deletions(-)

diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index c80f4c61bc..91b44cd8b3 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -20,6 +20,13 @@ from .units import Edu
 
 from synapse.util.logutils import log_function
 from synapse.events import FrozenEvent
+from synapse.events.utils import prune_event
+
+from syutil.jsonutil import encode_canonical_json
+
+from synapse.crypto.event_signing import check_event_content_hash
+
+from synapse.api.errors import SynapseError
 
 import logging
 
@@ -126,6 +133,11 @@ class FederationClient(object):
             for p in transaction_data["pdus"]
         ]
 
+        for i, pdu in enumerate(pdus):
+            pdus[i] = yield self._check_sigs_and_hash(pdu)
+
+            # FIXME: We should handle signature failures more gracefully.
+
         defer.returnValue(pdus)
 
     @defer.inlineCallbacks
@@ -159,6 +171,22 @@ class FederationClient(object):
                 transaction_data = yield self.transport_layer.get_event(
                     destination, event_id
                 )
+
+                logger.debug("transaction_data %r", transaction_data)
+
+                pdu_list = [
+                    self.event_from_pdu_json(p, outlier=outlier)
+                    for p in transaction_data["pdus"]
+                ]
+
+                if pdu_list:
+                    pdu = pdu_list[0]
+
+                    # Check signatures are correct.
+                    pdu = yield self._check_sigs_and_hash(pdu)
+
+                    break
+
             except Exception as e:
                 logger.info(
                     "Failed to get PDU %s from %s because %s",
@@ -166,18 +194,6 @@ class FederationClient(object):
                 )
                 continue
 
-            logger.debug("transaction_data %r", transaction_data)
-
-            pdu_list = [
-                self.event_from_pdu_json(p, outlier=outlier)
-                for p in transaction_data["pdus"]
-            ]
-
-            if pdu_list:
-                pdu = pdu_list[0]
-                # TODO: We need to check signatures here
-                break
-
         defer.returnValue(pdu)
 
     @defer.inlineCallbacks
@@ -208,6 +224,16 @@ class FederationClient(object):
             for p in result.get("auth_chain", [])
         ]
 
+        for i, pdu in enumerate(pdus):
+            pdus[i] = yield self._check_sigs_and_hash(pdu)
+
+            # FIXME: We should handle signature failures more gracefully.
+
+        for i, pdu in enumerate(auth_chain):
+            auth_chain[i] = yield self._check_sigs_and_hash(pdu)
+
+            # FIXME: We should handle signature failures more gracefully.
+
         defer.returnValue((pdus, auth_chain))
 
     @defer.inlineCallbacks
@@ -222,6 +248,11 @@ class FederationClient(object):
             for p in res["auth_chain"]
         ]
 
+        for i, pdu in enumerate(auth_chain):
+            auth_chain[i] = yield self._check_sigs_and_hash(pdu)
+
+            # FIXME: We should handle signature failures more gracefully.
+
         auth_chain.sort(key=lambda e: e.depth)
 
         defer.returnValue(auth_chain)
@@ -260,6 +291,16 @@ class FederationClient(object):
             for p in content.get("auth_chain", [])
         ]
 
+        for i, pdu in enumerate(state):
+            state[i] = yield self._check_sigs_and_hash(pdu)
+
+            # FIXME: We should handle signature failures more gracefully.
+
+        for i, pdu in enumerate(auth_chain):
+            auth_chain[i] = yield self._check_sigs_and_hash(pdu)
+
+            # FIXME: We should handle signature failures more gracefully.
+
         auth_chain.sort(key=lambda e: e.depth)
 
         defer.returnValue({
@@ -281,7 +322,14 @@ class FederationClient(object):
 
         logger.debug("Got response to send_invite: %s", pdu_dict)
 
-        defer.returnValue(self.event_from_pdu_json(pdu_dict))
+        pdu = self.event_from_pdu_json(pdu_dict)
+
+        # Check signatures are correct.
+        pdu = yield self._check_sigs_and_hash(pdu)
+
+        # FIXME: We should handle signature failures more gracefully.
+
+        defer.returnValue(pdu)
 
     def event_from_pdu_json(self, pdu_json, outlier=False):
         event = FrozenEvent(
@@ -291,3 +339,37 @@ class FederationClient(object):
         event.internal_metadata.outlier = outlier
 
         return event
+
+    @defer.inlineCallbacks
+    def _check_sigs_and_hash(self, pdu):
+        """Throws a SynapseError if the PDU does not have the correct
+        signatures.
+
+        Returns:
+            FrozenEvent: Either the given event or it redacted if it failed the
+            content hash check.
+        """
+        # Check signatures are correct.
+        redacted_event = prune_event(pdu)
+        redacted_pdu_json = redacted_event.get_pdu_json()
+
+        try:
+            yield self.keyring.verify_json_for_server(
+                pdu.origin, redacted_pdu_json
+            )
+        except SynapseError:
+            logger.warn(
+                "Signature check failed for %s redacted to %s",
+                encode_canonical_json(pdu.get_pdu_json()),
+                encode_canonical_json(redacted_pdu_json),
+            )
+            raise
+
+        if not check_event_content_hash(pdu):
+            logger.warn(
+                "Event content has been tampered, redacting %s, %s",
+                pdu.event_id, encode_canonical_json(pdu.get_dict())
+            )
+            defer.returnValue(redacted_event)
+
+        defer.returnValue(pdu)
diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index 0597725ce7..fc5342afaa 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -21,6 +21,13 @@ from .units import Transaction, Edu
 from synapse.util.logutils import log_function
 from synapse.util.logcontext import PreserveLoggingContext
 from synapse.events import FrozenEvent
+from synapse.events.utils import prune_event
+
+from syutil.jsonutil import encode_canonical_json
+
+from synapse.crypto.event_signing import check_event_content_hash
+
+from synapse.api.errors import FederationError, SynapseError
 
 import logging
 
@@ -97,8 +104,10 @@ class FederationServer(object):
         response = yield self.transaction_actions.have_responded(transaction)
 
         if response:
-            logger.debug("[%s] We've already responed to this request",
-                         transaction.transaction_id)
+            logger.debug(
+                "[%s] We've already responed to this request",
+                transaction.transaction_id
+            )
             defer.returnValue(response)
             return
 
@@ -253,6 +262,9 @@ class FederationServer(object):
             origin, pdu.event_id, do_auth=False
         )
 
+        # FIXME: Currently we fetch an event again when we already have it
+        # if it has been marked as an outlier.
+
         already_seen = (
             existing and (
                 not existing.internal_metadata.is_outlier()
@@ -264,14 +276,27 @@ class FederationServer(object):
             defer.returnValue({})
             return
 
+        # Check signature.
+        try:
+            pdu = yield self._check_sigs_and_hash(pdu)
+        except SynapseError as e:
+            raise FederationError(
+                "ERROR",
+                e.code,
+                e.msg,
+                affected=pdu.event_id,
+            )
+
         state = None
 
         auth_chain = []
 
         have_seen = yield self.store.have_events(
-            [e for e, _ in pdu.prev_events]
+            [ev for ev, _ in pdu.prev_events]
         )
 
+        fetch_state = False
+
         # Get missing pdus if necessary.
         if not pdu.internal_metadata.is_outlier():
             # We only backfill backwards to the min depth.
@@ -311,16 +336,20 @@ class FederationServer(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.
-                logger.debug(
-                    "_handle_new_pdu getting state for %s",
-                    pdu.room_id
-                )
-                state, auth_chain = yield self.get_state_for_room(
-                    origin, pdu.room_id, pdu.event_id,
-                )
+                            fetch_state = True
+        else:
+            fetch_state = True
+
+        if fetch_state:
+            # We need to get the state at this event, since we haven't
+            # processed all the prev events.
+            logger.debug(
+                "_handle_new_pdu getting state for %s",
+                pdu.room_id
+            )
+            state, auth_chain = yield self.get_state_for_room(
+                origin, pdu.room_id, pdu.event_id,
+            )
 
         ret = yield self.handler.on_receive_pdu(
             origin,
@@ -343,3 +372,37 @@ class FederationServer(object):
         event.internal_metadata.outlier = outlier
 
         return event
+
+    @defer.inlineCallbacks
+    def _check_sigs_and_hash(self, pdu):
+        """Throws a SynapseError if the PDU does not have the correct
+        signatures.
+
+        Returns:
+            FrozenEvent: Either the given event or it redacted if it failed the
+            content hash check.
+        """
+        # Check signatures are correct.
+        redacted_event = prune_event(pdu)
+        redacted_pdu_json = redacted_event.get_pdu_json()
+
+        try:
+            yield self.keyring.verify_json_for_server(
+                pdu.origin, redacted_pdu_json
+            )
+        except SynapseError:
+            logger.warn(
+                "Signature check failed for %s redacted to %s",
+                encode_canonical_json(pdu.get_pdu_json()),
+                encode_canonical_json(redacted_pdu_json),
+            )
+            raise
+
+        if not check_event_content_hash(pdu):
+            logger.warn(
+                "Event content has been tampered, redacting %s, %s",
+                pdu.event_id, encode_canonical_json(pdu.get_dict())
+            )
+            defer.returnValue(redacted_event)
+
+        defer.returnValue(pdu)
diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py
index 9ef4834927..e442c6c5d5 100644
--- a/synapse/federation/replication.py
+++ b/synapse/federation/replication.py
@@ -51,6 +51,8 @@ class ReplicationLayer(FederationClient, FederationServer):
     def __init__(self, hs, transport_layer):
         self.server_name = hs.hostname
 
+        self.keyring = hs.get_keyring()
+
         self.transport_layer = transport_layer
         self.transport_layer.register_received_handler(self)
         self.transport_layer.register_request_handler(self)

From 0ef5bfd6a9eaaae14e199997658b3d0006abd854 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Wed, 28 Jan 2015 16:16:53 +0000
Subject: [PATCH 02/15] Start implementing auth conflict res

---
 synapse/api/auth.py                     |  38 +++--
 synapse/api/constants.py                |   6 +
 synapse/federation/federation_client.py |  39 +++++
 synapse/handlers/federation.py          | 211 ++++++++++++++++++++----
 synapse/storage/rejections.py           |  10 ++
 synapse/storage/schema/im.sql           |   1 +
 6 files changed, 253 insertions(+), 52 deletions(-)

diff --git a/synapse/api/auth.py b/synapse/api/auth.py
index a342a0e0da..461faa8c78 100644
--- a/synapse/api/auth.py
+++ b/synapse/api/auth.py
@@ -353,9 +353,23 @@ class Auth(object):
     def add_auth_events(self, builder, context):
         yield run_on_reactor()
 
-        if builder.type == EventTypes.Create:
-            builder.auth_events = []
-            return
+        auth_ids = self.compute_auth_events(builder, context)
+
+        auth_events_entries = yield self.store.add_event_hashes(
+            auth_ids
+        )
+
+        builder.auth_events = auth_events_entries
+
+        context.auth_events = {
+            k: v
+            for k, v in context.current_state.items()
+            if v.event_id in auth_ids
+        }
+
+    def compute_auth_events(self, event, context):
+        if event.type == EventTypes.Create:
+            return []
 
         auth_ids = []
 
@@ -368,7 +382,7 @@ class Auth(object):
         key = (EventTypes.JoinRules, "", )
         join_rule_event = context.current_state.get(key)
 
-        key = (EventTypes.Member, builder.user_id, )
+        key = (EventTypes.Member, event.user_id, )
         member_event = context.current_state.get(key)
 
         key = (EventTypes.Create, "", )
@@ -382,8 +396,8 @@ class Auth(object):
         else:
             is_public = False
 
-        if builder.type == EventTypes.Member:
-            e_type = builder.content["membership"]
+        if event.type == EventTypes.Member:
+            e_type = event.content["membership"]
             if e_type in [Membership.JOIN, Membership.INVITE]:
                 if join_rule_event:
                     auth_ids.append(join_rule_event.event_id)
@@ -398,17 +412,7 @@ class Auth(object):
             if member_event.content["membership"] == Membership.JOIN:
                 auth_ids.append(member_event.event_id)
 
-        auth_events_entries = yield self.store.add_event_hashes(
-            auth_ids
-        )
-
-        builder.auth_events = auth_events_entries
-
-        context.auth_events = {
-            k: v
-            for k, v in context.current_state.items()
-            if v.event_id in auth_ids
-        }
+        return auth_ids
 
     @log_function
     def _can_send_event(self, event, auth_events):
diff --git a/synapse/api/constants.py b/synapse/api/constants.py
index 7ee6dcc46e..0d3fc629af 100644
--- a/synapse/api/constants.py
+++ b/synapse/api/constants.py
@@ -74,3 +74,9 @@ class EventTypes(object):
     Message = "m.room.message"
     Topic = "m.room.topic"
     Name = "m.room.name"
+
+
+class RejectedReason(object):
+    AUTH_ERROR = "auth_error"
+    REPLACED = "replaced"
+    NOT_ANCESTOR = "not_ancestor"
diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index 91b44cd8b3..ebcd593506 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -331,6 +331,45 @@ class FederationClient(object):
 
         defer.returnValue(pdu)
 
+    @defer.inlineCallbacks
+    def query_auth(self, destination, room_id, event_id, local_auth):
+        """
+        Params:
+            destination (str)
+            event_it (str)
+            local_auth (list)
+        """
+        time_now = self._clock.time_msec()
+
+        send_content = {
+            "auth_chain": [e.get_pdu_json(time_now) for e in local_auth],
+        }
+
+        code, content = yield self.transport_layer.send_invite(
+            destination=destination,
+            room_id=room_id,
+            event_id=event_id,
+            content=send_content,
+        )
+
+        auth_chain = [
+            (yield self._check_sigs_and_hash(self.event_from_pdu_json(e)))
+            for e in content["auth_chain"]
+        ]
+
+        missing = [
+            (yield self._check_sigs_and_hash(self.event_from_pdu_json(e)))
+            for e in content.get("missing", [])
+        ]
+
+        ret = {
+            "auth_chain": auth_chain,
+            "rejects": content.get("rejects", []),
+            "missing": missing,
+        }
+
+        defer.returnValue(ret)
+
     def event_from_pdu_json(self, pdu_json, outlier=False):
         event = FrozenEvent(
             pdu_json
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index bcdcc90a18..97e3c503b9 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -17,19 +17,16 @@
 
 from ._base import BaseHandler
 
-from synapse.events.utils import prune_event
 from synapse.api.errors import (
-    AuthError, FederationError, SynapseError, StoreError,
+    AuthError, FederationError, StoreError,
 )
-from synapse.api.constants import EventTypes, Membership
+from synapse.api.constants import EventTypes, Membership, RejectedReason
 from synapse.util.logutils import log_function
 from synapse.util.async import run_on_reactor
 from synapse.crypto.event_signing import (
-    compute_event_signature, check_event_content_hash,
-    add_hashes_and_signatures,
+    compute_event_signature, add_hashes_and_signatures,
 )
 from synapse.types import UserID
-from syutil.jsonutil import encode_canonical_json
 
 from twisted.internet import defer
 
@@ -113,33 +110,6 @@ class FederationHandler(BaseHandler):
 
         logger.debug("Processing event: %s", event.event_id)
 
-        redacted_event = prune_event(event)
-
-        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_pdu_json()),
-                encode_canonical_json(redacted_pdu_json),
-            )
-            raise FederationError(
-                "ERROR",
-                e.code,
-                e.msg,
-                affected=event.event_id,
-            )
-
-        if not check_event_content_hash(event):
-            logger.warn(
-                "Event content has been tampered, redacting %s, %s",
-                event.event_id, encode_canonical_json(event.get_dict())
-            )
-            event = redacted_event
-
         logger.debug("Event: %s", event)
 
         # FIXME (erikj): Awful hack to make the case where we are not currently
@@ -180,7 +150,6 @@ class FederationHandler(BaseHandler):
 
         if state:
             for e in state:
-                logging.info("A :) %r", e)
                 e.internal_metadata.outlier = True
                 try:
                     yield self._handle_new_event(e)
@@ -747,7 +716,20 @@ class FederationHandler(BaseHandler):
             event.event_id, event.signatures,
         )
 
-        self.auth.check(event, auth_events=context.auth_events)
+        try:
+            self.auth.check(event, auth_events=context.auth_events)
+        except AuthError:
+            # TODO: Store rejection.
+            context.rejected = RejectedReason.AUTH_ERROR
+
+            yield self.store.persist_event(
+                event,
+                context=context,
+                backfilled=backfilled,
+                is_new_state=False,
+                current_state=current_state,
+            )
+            raise
 
         logger.debug(
             "_handle_new_event: Before persist_event: %s, sigs: %s",
@@ -768,3 +750,162 @@ class FederationHandler(BaseHandler):
         )
 
         defer.returnValue(context)
+
+    @defer.inlineCallbacks
+    def do_auth(self, origin, event, context):
+        for e_id, _ in event.auth_events:
+            pass
+
+        auth_events = set(e_id for e_id, _ in event.auth_events)
+        current_state = set(e.event_id for e in context.auth_events.values())
+
+        missing_auth = auth_events - current_state
+
+        if missing_auth:
+            # Do auth conflict res.
+
+            # 1. Get what we think is the auth chain.
+            auth_ids = self.auth.compute_auth_events(event, context)
+            local_auth_chain = yield self.store.get_auth_chain(auth_ids)
+
+            # 2. Get remote difference.
+            result = yield self.replication_layer.query_auth(
+                origin,
+                event.room_id,
+                event.event_id,
+                local_auth_chain,
+            )
+
+            # 3. Process any remote auth chain events we haven't seen.
+            for e in result.get("missing", []):
+                # TODO.
+                pass
+
+            # 4. Look at rejects and their proofs.
+            # TODO.
+
+        try:
+            self.auth.check(event, auth_events=context.auth_events)
+        except AuthError:
+            raise
+
+    @defer.inlineCallbacks
+    def construct_auth_difference(self, local_auth, remote_auth):
+        """ Given a local and remote auth chain, find the differences. This
+        assumes that we have already processed all events in remote_auth
+
+        Params:
+            local_auth (list)
+            remote_auth (list)
+
+        Returns:
+            dict
+        """
+
+        # TODO: Make sure we are OK with local_auth or remote_auth having more
+        # auth events in them than strictly necessary.
+
+        def sort_fun(ev):
+            return ev.depth, ev.event_id
+
+        # We find the differences by starting at the "bottom" of each list
+        # and iterating up on both lists. The lists are ordered by depth and
+        # then event_id, we iterate up both lists until we find the event ids
+        # don't match. Then we look at depth/event_id to see which side is
+        # missing that event, and iterate only up that list. Repeat.
+
+        remote_list = list(remote_auth)
+        remote_list.sort(key=sort_fun)
+
+        local_list = list(local_auth)
+        local_list.sort(key=sort_fun)
+
+        local_iter = iter(local_list)
+        remote_iter = iter(remote_list)
+
+        current_local = local_iter.next()
+        current_remote = remote_iter.next()
+
+        def get_next(it, opt=None):
+            return it.next() if it.has_next() else opt
+
+        missing_remotes = []
+        missing_locals = []
+        while current_local and current_remote:
+            if current_remote is None:
+                missing_locals.append(current_local)
+                current_local = get_next(local_iter)
+                continue
+
+            if current_local is None:
+                missing_remotes.append(current_remote)
+                current_remote = get_next(remote_iter)
+                continue
+
+            if current_local.event_id == current_remote.event_id:
+                current_local = get_next(local_iter)
+                current_remote = get_next(remote_iter)
+                continue
+
+            if current_local.depth < current_remote.depth:
+                missing_locals.append(current_local)
+                current_local = get_next(local_iter)
+                continue
+
+            if current_local.depth > current_remote.depth:
+                missing_remotes.append(current_remote)
+                current_remote = get_next(remote_iter)
+                continue
+
+            # They have the same depth, so we fall back to the event_id order
+            if current_local.event_id < current_remote.event_id:
+                missing_locals.append(current_local)
+                current_local = get_next(local_iter)
+
+            if current_local.event_id > current_remote.event_id:
+                missing_remotes.append(current_remote)
+                current_remote = get_next(remote_iter)
+                continue
+
+        # missing locals should be sent to the server
+        # We should find why we are missing remotes, as they will have been
+        # rejected.
+
+        # Remove events from missing_remotes if they are referencing a missing
+        # remote. We only care about the "root" rejected ones.
+        missing_remote_ids = [e.event_id for e in missing_remotes]
+        base_remote_rejected = list(missing_remotes)
+        for e in missing_remotes:
+            for e_id, _ in e.auth_events:
+                if e_id in missing_remote_ids:
+                    base_remote_rejected.remove(e)
+
+        reason_map = {}
+
+        for e in base_remote_rejected:
+            reason = yield self.store.get_rejection_reason(e.event_id)
+            if reason is None:
+                # FIXME: ERRR?!
+                raise RuntimeError("")
+
+            reason_map[e.event_id] = reason
+
+            if reason == RejectedReason.AUTH_ERROR:
+                pass
+            elif reason == RejectedReason.REPLACED:
+                # TODO: Get proof
+                pass
+            elif reason == RejectedReason.NOT_ANCESTOR:
+                # TODO: Get proof.
+                pass
+
+        defer.returnValue({
+            "rejects": {
+                e.event_id: {
+                    "reason": reason_map[e.event_id],
+                    "proof": None,
+                }
+                for e in base_remote_rejected
+            },
+            "missing": missing_locals,
+        })
diff --git a/synapse/storage/rejections.py b/synapse/storage/rejections.py
index 7d38b31f44..b7249700d7 100644
--- a/synapse/storage/rejections.py
+++ b/synapse/storage/rejections.py
@@ -31,3 +31,13 @@ class RejectionsStore(SQLBaseStore):
                 "last_failure": self._clock.time_msec(),
             }
         )
+
+    def get_rejection_reason(self, event_id):
+        self._simple_select_one_onecol(
+            table="rejections",
+            retcol="reason",
+            keyvalues={
+                "event_id": event_id,
+            },
+            allow_none=True,
+        )
diff --git a/synapse/storage/schema/im.sql b/synapse/storage/schema/im.sql
index bc7c6b6ed5..5866a387f6 100644
--- a/synapse/storage/schema/im.sql
+++ b/synapse/storage/schema/im.sql
@@ -128,5 +128,6 @@ CREATE TABLE IF NOT EXISTS rejections(
     event_id TEXT NOT NULL,
     reason TEXT NOT NULL,
     last_check TEXT NOT NULL,
+    root_rejected TEXT,
     CONSTRAINT ev_id UNIQUE (event_id) ON CONFLICT REPLACE
 );

From c183cec8f61f5b3488973f01ca5203183a00e6d1 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 29 Jan 2015 13:44:52 +0000
Subject: [PATCH 03/15] Add post_json(...) method to federation client

---
 synapse/http/matrixfederationclient.py | 37 ++++++++++++++++++++++++++
 1 file changed, 37 insertions(+)

diff --git a/synapse/http/matrixfederationclient.py b/synapse/http/matrixfederationclient.py
index 1dda3ba2c7..b1b2916fd3 100644
--- a/synapse/http/matrixfederationclient.py
+++ b/synapse/http/matrixfederationclient.py
@@ -244,6 +244,43 @@ class MatrixFederationHttpClient(object):
 
         defer.returnValue((response.code, body))
 
+    @defer.inlineCallbacks
+    def post_json(self, destination, path, data={}):
+        """ Sends the specifed json data using POST
+
+        Args:
+            destination (str): The remote server to send the HTTP request
+                to.
+            path (str): The HTTP path.
+            data (dict): A dict containing the data that will be used as
+                the request body. This will be encoded as JSON.
+
+        Returns:
+            Deferred: Succeeds when we get a 2xx HTTP response. The result
+            will be the decoded JSON body. On a 4xx or 5xx error response a
+            CodeMessageException is raised.
+        """
+
+        def body_callback(method, url_bytes, headers_dict):
+            self.sign_request(
+                destination, method, url_bytes, headers_dict, data
+            )
+            return None
+
+        response = yield self._create_request(
+            destination.encode("ascii"),
+            "POST",
+            path.encode("ascii"),
+            body_callback=body_callback,
+            headers_dict={"Content-Type": ["application/json"]},
+        )
+
+        logger.debug("Getting resp body")
+        body = yield readBody(response)
+        logger.debug("Got resp body")
+
+        defer.returnValue((response.code, body))
+
     @defer.inlineCallbacks
     def get_json(self, destination, path, args={}, retry_on_dns_fail=True):
         """ GETs some json from the given host homeserver and path

From 5a3a15f5c186af6b818f2b2d3a4dafeee48b4e33 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 29 Jan 2015 13:58:22 +0000
Subject: [PATCH 04/15] Make post_json(...) actually send data.

---
 synapse/http/matrixfederationclient.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/synapse/http/matrixfederationclient.py b/synapse/http/matrixfederationclient.py
index b1b2916fd3..c7bf1b47b8 100644
--- a/synapse/http/matrixfederationclient.py
+++ b/synapse/http/matrixfederationclient.py
@@ -265,7 +265,7 @@ class MatrixFederationHttpClient(object):
             self.sign_request(
                 destination, method, url_bytes, headers_dict, data
             )
-            return None
+            return _JsonProducer(data)
 
         response = yield self._create_request(
             destination.encode("ascii"),

From 78015948a7febb18e000651f72f8f58830a55b93 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 29 Jan 2015 16:50:23 +0000
Subject: [PATCH 05/15] Initial implementation of auth conflict resolution

---
 synapse/events/utils.py                 |   6 +-
 synapse/federation/federation_client.py |   2 +-
 synapse/federation/federation_server.py |  33 ++++
 synapse/federation/transport/client.py  |  16 ++
 synapse/federation/transport/server.py  |  21 ++-
 synapse/handlers/federation.py          | 209 +++++++++++++++---------
 synapse/storage/rejections.py           |   4 +-
 tests/handlers/test_federation.py       |   2 +
 8 files changed, 211 insertions(+), 82 deletions(-)

diff --git a/synapse/events/utils.py b/synapse/events/utils.py
index bcb5457278..10a6b9f264 100644
--- a/synapse/events/utils.py
+++ b/synapse/events/utils.py
@@ -45,12 +45,14 @@ def prune_event(event):
         "membership",
     ]
 
+    event_dict = event.get_dict()
+
     new_content = {}
 
     def add_fields(*fields):
         for field in fields:
             if field in event.content:
-                new_content[field] = event.content[field]
+                new_content[field] = event_dict["content"][field]
 
     if event_type == EventTypes.Member:
         add_fields("membership")
@@ -75,7 +77,7 @@ def prune_event(event):
 
     allowed_fields = {
         k: v
-        for k, v in event.get_dict().items()
+        for k, v in event_dict.items()
         if k in allowed_keys
     }
 
diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index ebcd593506..1173ca817b 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -345,7 +345,7 @@ class FederationClient(object):
             "auth_chain": [e.get_pdu_json(time_now) for e in local_auth],
         }
 
-        code, content = yield self.transport_layer.send_invite(
+        code, content = yield self.transport_layer.send_query_auth(
             destination=destination,
             room_id=room_id,
             event_id=event_id,
diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index fc5342afaa..8cff4e6472 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -230,6 +230,39 @@ class FederationServer(object):
             "auth_chain": [a.get_pdu_json(time_now) for a in auth_pdus],
         }))
 
+    @defer.inlineCallbacks
+    def on_query_auth_request(self, origin, content, event_id):
+        auth_chain = [
+            (yield self._check_sigs_and_hash(self.event_from_pdu_json(e)))
+            for e in content["auth_chain"]
+        ]
+
+        missing = [
+            (yield self._check_sigs_and_hash(self.event_from_pdu_json(e)))
+            for e in content.get("missing", [])
+        ]
+
+        ret = yield self.handler.on_query_auth(
+            origin, event_id, auth_chain, content.get("rejects", []), missing
+        )
+
+        time_now = self._clock.time_msec()
+        send_content = {
+            "auth_chain": [
+                e.get_pdu_json(time_now)
+                for e in ret["auth_chain"]
+            ],
+            "rejects": content.get("rejects", []),
+            "missing": [
+                e.get_pdu_json(time_now)
+                for e in ret.get("missing", [])
+            ],
+        }
+
+        defer.returnValue(
+            (200, send_content)
+        )
+
     @log_function
     def _get_persisted_pdu(self, origin, event_id, do_auth=True):
         """ Get a PDU from the database with given origin and id.
diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py
index e634a3a213..4cb1dea2de 100644
--- a/synapse/federation/transport/client.py
+++ b/synapse/federation/transport/client.py
@@ -213,3 +213,19 @@ class TransportLayerClient(object):
         )
 
         defer.returnValue(response)
+
+    @defer.inlineCallbacks
+    @log_function
+    def send_query_auth(self, destination, room_id, event_id, content):
+        path = PREFIX + "/query_auth/%s/%s" % (room_id, event_id)
+
+        code, content = yield self.client.post_json(
+            destination=destination,
+            path=path,
+            data=content,
+        )
+
+        if not 200 <= code < 300:
+            raise RuntimeError("Got %d from send_invite", code)
+
+        defer.returnValue(json.loads(content))
diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py
index a380a6910b..9c9f8d525b 100644
--- a/synapse/federation/transport/server.py
+++ b/synapse/federation/transport/server.py
@@ -42,7 +42,7 @@ class TransportLayerServer(object):
         content = None
         origin = None
 
-        if request.method == "PUT":
+        if request.method in ["PUT", "POST"]:
             # TODO: Handle other method types? other content types?
             try:
                 content_bytes = request.content.read()
@@ -234,6 +234,16 @@ class TransportLayerServer(object):
                 )
             )
         )
+        self.server.register_path(
+            "POST",
+            re.compile("^" + PREFIX + "/query_auth/([^/]*)/([^/]*)$"),
+            self._with_authentication(
+                lambda origin, content, query, context, event_id:
+                self._on_query_auth_request(
+                    origin, content, event_id,
+                )
+            )
+        )
 
     @defer.inlineCallbacks
     @log_function
@@ -325,3 +335,12 @@ class TransportLayerServer(object):
         )
 
         defer.returnValue((200, content))
+
+    @defer.inlineCallbacks
+    @log_function
+    def _on_query_auth_request(self, origin, content, event_id):
+        new_content = yield self.request_handler.on_query_auth_request(
+            origin, content, event_id
+        )
+
+        defer.returnValue((200, new_content))
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 97e3c503b9..14c26d8cea 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -126,7 +126,7 @@ class FederationHandler(BaseHandler):
 
             if not state:
                 state, auth_chain = yield replication.get_state_for_room(
-                    origin, context=event.room_id, event_id=event.event_id,
+                    origin, room_id=event.room_id, event_id=event.event_id,
                 )
 
             if not auth_chain:
@@ -139,7 +139,7 @@ class FederationHandler(BaseHandler):
             for e in auth_chain:
                 e.internal_metadata.outlier = True
                 try:
-                    yield self._handle_new_event(e, fetch_auth_from=origin)
+                    yield self._handle_new_event(origin, e)
                 except:
                     logger.exception(
                         "Failed to handle auth event %s",
@@ -152,7 +152,7 @@ class FederationHandler(BaseHandler):
             for e in state:
                 e.internal_metadata.outlier = True
                 try:
-                    yield self._handle_new_event(e)
+                    yield self._handle_new_event(origin, e)
                 except:
                     logger.exception(
                         "Failed to handle state event %s",
@@ -161,6 +161,7 @@ class FederationHandler(BaseHandler):
 
         try:
             yield self._handle_new_event(
+                origin,
                 event,
                 state=state,
                 backfilled=backfilled,
@@ -363,7 +364,14 @@ class FederationHandler(BaseHandler):
             for e in auth_chain:
                 e.internal_metadata.outlier = True
                 try:
-                    yield self._handle_new_event(e)
+                    auth_ids = [e_id for e_id, _ in e.auth_events]
+                    auth = {
+                        (e.type, e.state_key): e for e in auth_chain
+                        if e.event_id in auth_ids
+                    }
+                    yield self._handle_new_event(
+                        target_host, e, auth_events=auth
+                    )
                 except:
                     logger.exception(
                         "Failed to handle auth event %s",
@@ -374,8 +382,13 @@ class FederationHandler(BaseHandler):
                 # FIXME: Auth these.
                 e.internal_metadata.outlier = True
                 try:
+                    auth_ids = [e_id for e_id, _ in e.auth_events]
+                    auth = {
+                        (e.type, e.state_key): e for e in auth_chain
+                        if e.event_id in auth_ids
+                    }
                     yield self._handle_new_event(
-                        e, fetch_auth_from=target_host
+                        target_host, e, auth_events=auth
                     )
                 except:
                     logger.exception(
@@ -384,6 +397,7 @@ class FederationHandler(BaseHandler):
                     )
 
             yield self._handle_new_event(
+                target_host,
                 new_event,
                 state=state,
                 current_state=state,
@@ -450,7 +464,7 @@ class FederationHandler(BaseHandler):
 
         event.internal_metadata.outlier = False
 
-        context = yield self._handle_new_event(event)
+        context = yield self._handle_new_event(origin, event)
 
         logger.debug(
             "on_send_join_request: After _handle_new_event: %s, sigs: %s",
@@ -651,11 +665,12 @@ class FederationHandler(BaseHandler):
             waiters.pop().callback(None)
 
     @defer.inlineCallbacks
-    def _handle_new_event(self, event, state=None, backfilled=False,
-                          current_state=None, fetch_auth_from=None):
+    @log_function
+    def _handle_new_event(self, origin, event, state=None, backfilled=False,
+                          current_state=None, auth_events=None):
 
         logger.debug(
-            "_handle_new_event: Before annotate: %s, sigs: %s",
+            "_handle_new_event: %s, sigs: %s",
             event.event_id, event.signatures,
         )
 
@@ -663,62 +678,34 @@ class FederationHandler(BaseHandler):
             event, old_state=state
         )
 
+        if not auth_events:
+            auth_events = context.auth_events
+
         logger.debug(
-            "_handle_new_event: Before auth fetch: %s, sigs: %s",
-            event.event_id, event.signatures,
+            "_handle_new_event: %s, auth_events: %s",
+            event.event_id, auth_events,
         )
 
         is_new_state = not event.internal_metadata.is_outlier()
 
-        known_ids = set(
-            [s.event_id for s in context.auth_events.values()]
-        )
-
-        for e_id, _ in event.auth_events:
-            if e_id not in known_ids:
-                e = yield self.store.get_event(e_id, allow_none=True)
-
-                if not e and fetch_auth_from is not None:
-                    # Grab the auth_chain over federation if we are missing
-                    # auth events.
-                    auth_chain = yield self.replication_layer.get_event_auth(
-                        fetch_auth_from, event.event_id, event.room_id
-                    )
-                    for auth_event in auth_chain:
-                        yield self._handle_new_event(auth_event)
-                    e = yield self.store.get_event(e_id, allow_none=True)
-
-                if not e:
-                    # TODO: Do some conflict res to make sure that we're
-                    # not the ones who are wrong.
-                    logger.info(
-                        "Rejecting %s as %s not in db or %s",
-                        event.event_id, e_id, known_ids,
-                    )
-                    # FIXME: How does raising AuthError work with federation?
-                    raise AuthError(403, "Cannot find auth event")
-
-                context.auth_events[(e.type, e.state_key)] = e
-
-        logger.debug(
-            "_handle_new_event: Before hack: %s, sigs: %s",
-            event.event_id, event.signatures,
-        )
-
+        # This is a hack to fix some old rooms where the initial join event
+        # didn't reference the create event in its auth events.
         if event.type == EventTypes.Member and not event.auth_events:
             if len(event.prev_events) == 1:
                 c = yield self.store.get_event(event.prev_events[0][0])
                 if c.type == EventTypes.Create:
-                    context.auth_events[(c.type, c.state_key)] = c
-
-        logger.debug(
-            "_handle_new_event: Before auth check: %s, sigs: %s",
-            event.event_id, event.signatures,
-        )
+                    auth_events[(c.type, c.state_key)] = c
 
         try:
-            self.auth.check(event, auth_events=context.auth_events)
-        except AuthError:
+            yield self.do_auth(
+                origin, event, context, auth_events=auth_events
+            )
+        except AuthError as e:
+            logger.warn(
+                "Rejecting %s because %s",
+                event.event_id, e.msg
+            )
+
             # TODO: Store rejection.
             context.rejected = RejectedReason.AUTH_ERROR
 
@@ -731,11 +718,6 @@ class FederationHandler(BaseHandler):
             )
             raise
 
-        logger.debug(
-            "_handle_new_event: Before persist_event: %s, sigs: %s",
-            event.event_id, event.signatures,
-        )
-
         yield self.store.persist_event(
             event,
             context=context,
@@ -744,25 +726,73 @@ class FederationHandler(BaseHandler):
             current_state=current_state,
         )
 
-        logger.debug(
-            "_handle_new_event: After persist_event: %s, sigs: %s",
-            event.event_id, event.signatures,
-        )
-
         defer.returnValue(context)
 
     @defer.inlineCallbacks
-    def do_auth(self, origin, event, context):
-        for e_id, _ in event.auth_events:
-            pass
+    def on_query_auth(self, origin, event_id, remote_auth_chain, rejects,
+                      missing):
+        # Just go through and process each event in `remote_auth_chain`. We
+        # don't want to fall into the trap of `missing` being wrong.
+        for e in remote_auth_chain:
+            try:
+                yield self._handle_new_event(origin, e)
+            except AuthError:
+                pass
 
-        auth_events = set(e_id for e_id, _ in event.auth_events)
-        current_state = set(e.event_id for e in context.auth_events.values())
+        # Now get the current auth_chain for the event.
+        local_auth_chain = yield self.store.get_auth_chain([event_id])
 
-        missing_auth = auth_events - current_state
+        # TODO: Check if we would now reject event_id. If so we need to tell
+        # everyone.
+
+        ret = yield self.construct_auth_difference(
+            local_auth_chain, remote_auth_chain
+        )
+
+        logger.debug("on_query_auth reutrning: %s", ret)
+
+        defer.returnValue(ret)
+
+    @defer.inlineCallbacks
+    @log_function
+    def do_auth(self, origin, event, context, auth_events):
+        # Check if we have all the auth events.
+        res = yield self.store.have_events(
+            [e_id for e_id, _ in event.auth_events]
+        )
+
+        event_auth_events = set(e_id for e_id, _ in event.auth_events)
+        seen_events = set(res.keys())
+
+        missing_auth = event_auth_events - seen_events
 
         if missing_auth:
+            logger.debug("Missing auth: %s", missing_auth)
+            # If we don't have all the auth events, we need to get them.
+            remote_auth_chain = yield self.replication_layer.get_event_auth(
+                origin, event.room_id, event.event_id
+            )
+
+            for e in remote_auth_chain:
+                try:
+                    auth_ids = [e_id for e_id, _ in e.auth_events]
+                    auth = {
+                        (e.type, e.state_key): e for e in remote_auth_chain
+                        if e.event_id in auth_ids
+                    }
+                    yield self._handle_new_event(
+                        origin, e, auth_events=auth
+                    )
+                    auth_events[(e.type, e.state_key)] = e
+                except AuthError:
+                    pass
+
+        current_state = set(e.event_id for e in auth_events.values())
+        different_auth = event_auth_events - current_state
+
+        if different_auth and not event.internal_metadata.is_outlier():
             # Do auth conflict res.
+            logger.debug("Different auth: %s", different_auth)
 
             # 1. Get what we think is the auth chain.
             auth_ids = self.auth.compute_auth_events(event, context)
@@ -778,14 +808,24 @@ class FederationHandler(BaseHandler):
 
             # 3. Process any remote auth chain events we haven't seen.
             for e in result.get("missing", []):
-                # TODO.
-                pass
+                try:
+                    auth_ids = [e_id for e_id, _ in e.auth_events]
+                    auth = {
+                        (e.type, e.state_key): e for e in result["auth_chain"]
+                        if e.event_id in auth_ids
+                    }
+                    yield self._handle_new_event(
+                        origin, e, auth_events=auth
+                    )
+                    auth_events[(e.type, e.state_key)] = e
+                except AuthError:
+                    pass
 
             # 4. Look at rejects and their proofs.
             # TODO.
 
         try:
-            self.auth.check(event, auth_events=context.auth_events)
+            self.auth.check(event, auth_events=auth_events)
         except AuthError:
             raise
 
@@ -802,12 +842,16 @@ class FederationHandler(BaseHandler):
             dict
         """
 
+        logger.debug("construct_auth_difference Start!")
+
         # TODO: Make sure we are OK with local_auth or remote_auth having more
         # auth events in them than strictly necessary.
 
         def sort_fun(ev):
             return ev.depth, ev.event_id
 
+        logger.debug("construct_auth_difference after sort_fun!")
+
         # We find the differences by starting at the "bottom" of each list
         # and iterating up on both lists. The lists are ordered by depth and
         # then event_id, we iterate up both lists until we find the event ids
@@ -823,11 +867,18 @@ class FederationHandler(BaseHandler):
         local_iter = iter(local_list)
         remote_iter = iter(remote_list)
 
-        current_local = local_iter.next()
-        current_remote = remote_iter.next()
+        logger.debug("construct_auth_difference before get_next!")
 
         def get_next(it, opt=None):
-            return it.next() if it.has_next() else opt
+            try:
+                return it.next()
+            except:
+                return opt
+
+        current_local = get_next(local_iter)
+        current_remote = get_next(remote_iter)
+
+        logger.debug("construct_auth_difference before while")
 
         missing_remotes = []
         missing_locals = []
@@ -867,6 +918,8 @@ class FederationHandler(BaseHandler):
                 current_remote = get_next(remote_iter)
                 continue
 
+        logger.debug("construct_auth_difference after while")
+
         # missing locals should be sent to the server
         # We should find why we are missing remotes, as they will have been
         # rejected.
@@ -886,6 +939,7 @@ class FederationHandler(BaseHandler):
             reason = yield self.store.get_rejection_reason(e.event_id)
             if reason is None:
                 # FIXME: ERRR?!
+                logger.warn("Could not find reason for %s", e.event_id)
                 raise RuntimeError("")
 
             reason_map[e.event_id] = reason
@@ -899,7 +953,10 @@ class FederationHandler(BaseHandler):
                 # TODO: Get proof.
                 pass
 
+        logger.debug("construct_auth_difference returning")
+
         defer.returnValue({
+            "auth_chain": local_auth,
             "rejects": {
                 e.event_id: {
                     "reason": reason_map[e.event_id],
diff --git a/synapse/storage/rejections.py b/synapse/storage/rejections.py
index b7249700d7..4e1a9a2783 100644
--- a/synapse/storage/rejections.py
+++ b/synapse/storage/rejections.py
@@ -28,12 +28,12 @@ class RejectionsStore(SQLBaseStore):
             values={
                 "event_id": event_id,
                 "reason": reason,
-                "last_failure": self._clock.time_msec(),
+                "last_check": self._clock.time_msec(),
             }
         )
 
     def get_rejection_reason(self, event_id):
-        self._simple_select_one_onecol(
+        return self._simple_select_one_onecol(
             table="rejections",
             retcol="reason",
             keyvalues={
diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py
index ed21defd13..44dbce6bea 100644
--- a/tests/handlers/test_federation.py
+++ b/tests/handlers/test_federation.py
@@ -52,6 +52,7 @@ class FederationTestCase(unittest.TestCase):
                 "get_room",
                 "get_destination_retry_timings",
                 "set_destination_retry_timings",
+                "have_events",
             ]),
             resource_for_federation=NonCallableMock(),
             http_client=NonCallableMock(spec_set=[]),
@@ -90,6 +91,7 @@ class FederationTestCase(unittest.TestCase):
         self.datastore.persist_event.return_value = defer.succeed(None)
         self.datastore.get_room.return_value = defer.succeed(True)
         self.auth.check_host_in_room.return_value = defer.succeed(True)
+        self.datastore.have_events.return_value = defer.succeed({})
 
         def annotate(ev, old_state=None):
             context = Mock()

From c1c7b398270cc19fcf9410b9159f741092e0510b Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 30 Jan 2015 10:30:54 +0000
Subject: [PATCH 06/15] Fix bug where we changes in outlier in metadata dict
 propogated to other events

---
 synapse/events/__init__.py | 2 +-
 synapse/events/builder.py  | 5 +++--
 synapse/events/utils.py    | 5 ++++-
 3 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py
index 4252e5ab5c..bf07951027 100644
--- a/synapse/events/__init__.py
+++ b/synapse/events/__init__.py
@@ -18,7 +18,7 @@ from synapse.util.frozenutils import freeze, unfreeze
 
 class _EventInternalMetadata(object):
     def __init__(self, internal_metadata_dict):
-        self.__dict__ = internal_metadata_dict
+        self.__dict__ = dict(internal_metadata_dict)
 
     def get_dict(self):
         return dict(self.__dict__)
diff --git a/synapse/events/builder.py b/synapse/events/builder.py
index a9b1b99a10..9d45bdb892 100644
--- a/synapse/events/builder.py
+++ b/synapse/events/builder.py
@@ -23,14 +23,15 @@ import copy
 
 
 class EventBuilder(EventBase):
-    def __init__(self, key_values={}):
+    def __init__(self, key_values={}, internal_metadata_dict={}):
         signatures = copy.deepcopy(key_values.pop("signatures", {}))
         unsigned = copy.deepcopy(key_values.pop("unsigned", {}))
 
         super(EventBuilder, self).__init__(
             key_values,
             signatures=signatures,
-            unsigned=unsigned
+            unsigned=unsigned,
+            internal_metadata_dict=internal_metadata_dict,
         )
 
     def build(self):
diff --git a/synapse/events/utils.py b/synapse/events/utils.py
index 10a6b9f264..08d6d6fa47 100644
--- a/synapse/events/utils.py
+++ b/synapse/events/utils.py
@@ -88,7 +88,10 @@ def prune_event(event):
     if "age_ts" in event.unsigned:
         allowed_fields["unsigned"]["age_ts"] = event.unsigned["age_ts"]
 
-    return type(event)(allowed_fields)
+    return type(event)(
+        allowed_fields,
+        internal_metadata_dict=event.internal_metadata.get_dict()
+    )
 
 
 def serialize_event(hs, e, client_event=True):

From c1d860870b418b9583078022d0babe557b73760f Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 30 Jan 2015 10:48:47 +0000
Subject: [PATCH 07/15] Fix regression where we no longer correctly handled the
 case of gaps in our event graph

---
 synapse/federation/federation_server.py | 3 +++
 synapse/handlers/federation.py          | 8 ++++++--
 2 files changed, 9 insertions(+), 2 deletions(-)

diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index 8cff4e6472..845a07a3a3 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -366,10 +366,13 @@ class FederationServer(object):
                                 logger.debug("Processed pdu %s", event_id)
                             else:
                                 logger.warn("Failed to get PDU %s", event_id)
+                                fetch_state = True
                         except:
                             # TODO(erikj): Do some more intelligent retries.
                             logger.exception("Failed to get PDU")
                             fetch_state = True
+            else:
+                fetch_state = True
         else:
             fetch_state = True
 
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 14c26d8cea..de47a97e6b 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -119,7 +119,7 @@ class FederationHandler(BaseHandler):
             event.room_id,
             self.server_name
         )
-        if not is_in_room and not event.internal_metadata.outlier:
+        if not is_in_room and not event.internal_metadata.is_outlier():
             logger.debug("Got event for room we're not in.")
 
             replication = self.replication_layer
@@ -780,6 +780,7 @@ class FederationHandler(BaseHandler):
                         (e.type, e.state_key): e for e in remote_auth_chain
                         if e.event_id in auth_ids
                     }
+                    e.internal_metadata.outlier = True
                     yield self._handle_new_event(
                         origin, e, auth_events=auth
                     )
@@ -787,6 +788,8 @@ class FederationHandler(BaseHandler):
                 except AuthError:
                     pass
 
+        # FIXME: Assumes we have and stored all the state for all the
+        # prev_events
         current_state = set(e.event_id for e in auth_events.values())
         different_auth = event_auth_events - current_state
 
@@ -814,6 +817,7 @@ class FederationHandler(BaseHandler):
                         (e.type, e.state_key): e for e in result["auth_chain"]
                         if e.event_id in auth_ids
                     }
+                    e.internal_metadata.outlier = True
                     yield self._handle_new_event(
                         origin, e, auth_events=auth
                     )
@@ -882,7 +886,7 @@ class FederationHandler(BaseHandler):
 
         missing_remotes = []
         missing_locals = []
-        while current_local and current_remote:
+        while current_local or current_remote:
             if current_remote is None:
                 missing_locals.append(current_local)
                 current_local = get_next(local_iter)

From 823999716eb506740b34d87fcdaaf3ebbe4f760c Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 30 Jan 2015 11:07:57 +0000
Subject: [PATCH 08/15] Fix bug in timeout handling in keyclient

---
 synapse/crypto/keyclient.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/synapse/crypto/keyclient.py b/synapse/crypto/keyclient.py
index 9c910fa3fc..cdb6279764 100644
--- a/synapse/crypto/keyclient.py
+++ b/synapse/crypto/keyclient.py
@@ -61,9 +61,11 @@ class SynapseKeyClientProtocol(HTTPClient):
 
     def __init__(self):
         self.remote_key = defer.Deferred()
+        self.host = None
 
     def connectionMade(self):
-        logger.debug("Connected to %s", self.transport.getHost())
+        self.host = self.transport.getHost()
+        logger.debug("Connected to %s", self.host)
         self.sendCommand(b"GET", b"/_matrix/key/v1/")
         self.endHeaders()
         self.timer = reactor.callLater(
@@ -92,8 +94,7 @@ class SynapseKeyClientProtocol(HTTPClient):
         self.timer.cancel()
 
     def on_timeout(self):
-        logger.debug("Timeout waiting for response from %s",
-                     self.transport.getHost())
+        logger.debug("Timeout waiting for response from %s", self.host)
         self.remote_key.errback(IOError("Timeout waiting for response"))
         self.transport.abortConnection()
 

From 0c2d245fdf5f24f99138f1c60c4e6bed52cf5d5a Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 30 Jan 2015 11:08:52 +0000
Subject: [PATCH 09/15] Update the current state of an event if we update auth
 events.

---
 synapse/handlers/federation.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index de47a97e6b..cc22f21cd1 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -706,7 +706,6 @@ class FederationHandler(BaseHandler):
                 event.event_id, e.msg
             )
 
-            # TODO: Store rejection.
             context.rejected = RejectedReason.AUTH_ERROR
 
             yield self.store.persist_event(
@@ -828,6 +827,9 @@ class FederationHandler(BaseHandler):
             # 4. Look at rejects and their proofs.
             # TODO.
 
+            context.current_state.update(auth_events)
+            context.state_group = None
+
         try:
             self.auth.check(event, auth_events=auth_events)
         except AuthError:

From 2c9e136d5721d1797e153268f4562c27887a6201 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 30 Jan 2015 11:14:33 +0000
Subject: [PATCH 10/15] Fix bad merge fo python_dependencies.py

---
 synapse/python_dependencies.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/synapse/python_dependencies.py b/synapse/python_dependencies.py
index ba9308803c..4182ad990f 100644
--- a/synapse/python_dependencies.py
+++ b/synapse/python_dependencies.py
@@ -6,7 +6,6 @@ logger = logging.getLogger(__name__)
 REQUIREMENTS = {
     "syutil==0.0.2": ["syutil"],
     "matrix_angular_sdk==0.6.0": ["syweb>=0.6.0"],
-    "matrix_angular_sdk>=0.6.0": ["syweb>=0.6.0"],
     "Twisted>=14.0.0": ["twisted>=14.0.0"],
     "service_identity>=1.0.0": ["service_identity>=1.0.0"],
     "pyopenssl>=0.14": ["OpenSSL>=0.14"],

From 0adf3e54454d499045d2bfd8f6bcfece4e8fe6ed Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 30 Jan 2015 11:16:41 +0000
Subject: [PATCH 11/15] Revert accidental bumping of angluar_sdk dep

---
 setup.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/setup.py b/setup.py
index 733cfa8318..043cd044a7 100755
--- a/setup.py
+++ b/setup.py
@@ -32,7 +32,7 @@ setup(
     description="Reference Synapse Home Server",
     install_requires=[
         "syutil==0.0.2",
-        "matrix_angular_sdk==0.6.1",
+        "matrix_angular_sdk==0.6.0",
         "Twisted>=14.0.0",
         "service_identity>=1.0.0",
         "pyopenssl>=0.14",
@@ -47,7 +47,7 @@ setup(
     dependency_links=[
         "https://github.com/matrix-org/syutil/tarball/v0.0.2#egg=syutil-0.0.2",
         "https://github.com/pyca/pynacl/tarball/d4d3175589b892f6ea7c22f466e0e223853516fa#egg=pynacl-0.3.0",
-        "https://github.com/matrix-org/matrix-angular-sdk/tarball/v0.6.0/#egg=matrix_angular_sdk-0.6.1",
+        "https://github.com/matrix-org/matrix-angular-sdk/tarball/v0.6.0/#egg=matrix_angular_sdk-0.6.0",
     ],
     setup_requires=[
         "setuptools_trial",

From a70a801184814d116ed5b10a952e17c45df7bfc8 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 30 Jan 2015 13:34:01 +0000
Subject: [PATCH 12/15] Fix bug where we superfluously asked for current state.
 Change API of /query_auth/ so that we don't duplicate events in the response.

---
 synapse/api/auth.py                     |  2 +
 synapse/federation/federation_client.py |  7 +---
 synapse/federation/federation_server.py | 12 +++---
 synapse/handlers/federation.py          | 51 +++++++++----------------
 synapse/state.py                        | 20 ++++++++--
 5 files changed, 43 insertions(+), 49 deletions(-)

diff --git a/synapse/api/auth.py b/synapse/api/auth.py
index 3471afd7e7..37e31d2b6f 100644
--- a/synapse/api/auth.py
+++ b/synapse/api/auth.py
@@ -102,6 +102,8 @@ class Auth(object):
     def check_host_in_room(self, room_id, host):
         curr_state = yield self.state.get_current_state(room_id)
 
+        logger.debug("Got curr_state %s", curr_state)
+
         for event in curr_state:
             if event.type == EventTypes.Member:
                 try:
diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index 1173ca817b..e1539bd0e0 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -357,15 +357,10 @@ class FederationClient(object):
             for e in content["auth_chain"]
         ]
 
-        missing = [
-            (yield self._check_sigs_and_hash(self.event_from_pdu_json(e)))
-            for e in content.get("missing", [])
-        ]
-
         ret = {
             "auth_chain": auth_chain,
             "rejects": content.get("rejects", []),
-            "missing": missing,
+            "missing": content.get("missing", []),
         }
 
         defer.returnValue(ret)
diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index 845a07a3a3..84ed0a0ba0 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -252,11 +252,8 @@ class FederationServer(object):
                 e.get_pdu_json(time_now)
                 for e in ret["auth_chain"]
             ],
-            "rejects": content.get("rejects", []),
-            "missing": [
-                e.get_pdu_json(time_now)
-                for e in ret.get("missing", [])
-            ],
+            "rejects": ret.get("rejects", []),
+            "missing": ret.get("missing", []),
         }
 
         defer.returnValue(
@@ -372,7 +369,10 @@ class FederationServer(object):
                             logger.exception("Failed to get PDU")
                             fetch_state = True
             else:
-                fetch_state = True
+                prevs = {e_id for e_id, _ in pdu.prev_events}
+                seen = set(have_seen.keys())
+                if prevs - seen:
+                    fetch_state = True
         else:
             fetch_state = True
 
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index cc22f21cd1..35cad4182a 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -121,38 +121,18 @@ class FederationHandler(BaseHandler):
         )
         if not is_in_room and not event.internal_metadata.is_outlier():
             logger.debug("Got event for room we're not in.")
-
-            replication = self.replication_layer
-
-            if not state:
-                state, auth_chain = yield replication.get_state_for_room(
-                    origin, room_id=event.room_id, event_id=event.event_id,
-                )
-
-            if not auth_chain:
-                auth_chain = yield replication.get_event_auth(
-                    origin,
-                    context=event.room_id,
-                    event_id=event.event_id,
-                )
-
-            for e in auth_chain:
-                e.internal_metadata.outlier = True
-                try:
-                    yield self._handle_new_event(origin, e)
-                except:
-                    logger.exception(
-                        "Failed to handle auth event %s",
-                        e.event_id,
-                    )
-
             current_state = state
 
-        if state:
+        if state and auth_chain is not None:
             for e in state:
                 e.internal_metadata.outlier = True
                 try:
-                    yield self._handle_new_event(origin, e)
+                    auth_ids = [e_id for e_id, _ in e.auth_events]
+                    auth = {
+                        (e.type, e.state_key): e for e in auth_chain
+                        if e.event_id in auth_ids
+                    }
+                    yield self._handle_new_event(origin, e, auth_events=auth)
                 except:
                     logger.exception(
                         "Failed to handle state event %s",
@@ -809,18 +789,23 @@ class FederationHandler(BaseHandler):
             )
 
             # 3. Process any remote auth chain events we haven't seen.
-            for e in result.get("missing", []):
+            for missing_id in result.get("missing", []):
                 try:
-                    auth_ids = [e_id for e_id, _ in e.auth_events]
+                    for e in result["auth_chain"]:
+                        if e.event_id == missing_id:
+                            ev = e
+                            break
+
+                    auth_ids = [e_id for e_id, _ in ev.auth_events]
                     auth = {
                         (e.type, e.state_key): e for e in result["auth_chain"]
                         if e.event_id in auth_ids
                     }
-                    e.internal_metadata.outlier = True
+                    ev.internal_metadata.outlier = True
                     yield self._handle_new_event(
-                        origin, e, auth_events=auth
+                        origin, ev, auth_events=auth
                     )
-                    auth_events[(e.type, e.state_key)] = e
+                    auth_events[(ev.type, ev.state_key)] = ev
                 except AuthError:
                     pass
 
@@ -970,5 +955,5 @@ class FederationHandler(BaseHandler):
                 }
                 for e in base_remote_rejected
             },
-            "missing": missing_locals,
+            "missing": [e.event_id for e in missing_locals],
         })
diff --git a/synapse/state.py b/synapse/state.py
index d9fdfb34be..e6632978b5 100644
--- a/synapse/state.py
+++ b/synapse/state.py
@@ -166,10 +166,17 @@ class StateHandler(object):
         first is the name of a state group if one and only one is involved,
         otherwise `None`.
         """
+        logger.debug("resolve_state_groups event_ids %s", event_ids)
+
         state_groups = yield self.store.get_state_groups(
             event_ids
         )
 
+        logger.debug(
+            "resolve_state_groups state_groups %s",
+            state_groups.keys()
+        )
+
         group_names = set(state_groups.keys())
         if len(group_names) == 1:
             name, state_list = state_groups.items().pop()
@@ -205,6 +212,15 @@ class StateHandler(object):
             if len(v.values()) > 1
         }
 
+        logger.debug(
+            "resolve_state_groups Unconflicted state: %s",
+            unconflicted_state.values(),
+        )
+        logger.debug(
+            "resolve_state_groups Conflicted state: %s",
+            conflicted_state.values(),
+        )
+
         if event_type:
             prev_states_events = conflicted_state.get(
                 (event_type, state_key), []
@@ -240,10 +256,6 @@ class StateHandler(object):
             1. power levels
             2. memberships
             3. other events.
-
-        :param conflicted_state:
-        :param auth_events:
-        :return:
         """
         resolved_state = {}
         power_key = (EventTypes.PowerLevels, "")

From 7a1e881665ab72e29ff931f88ee2b65e8e690f74 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 30 Jan 2015 15:56:32 +0000
Subject: [PATCH 13/15] Remove debug logging

---
 synapse/state.py | 9 ---------
 1 file changed, 9 deletions(-)

diff --git a/synapse/state.py b/synapse/state.py
index 038e5eba11..8a056ee955 100644
--- a/synapse/state.py
+++ b/synapse/state.py
@@ -214,15 +214,6 @@ class StateHandler(object):
             if len(v.values()) > 1
         }
 
-        logger.debug(
-            "resolve_state_groups Unconflicted state: %s",
-            unconflicted_state.values(),
-        )
-        logger.debug(
-            "resolve_state_groups Conflicted state: %s",
-            conflicted_state.values(),
-        )
-
         if event_type:
             prev_states_events = conflicted_state.get(
                 (event_type, state_key), []

From b724a809c46f782231e18b70a632ce1ea9c540da Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 30 Jan 2015 15:57:53 +0000
Subject: [PATCH 14/15] Only auth_events with event if event in
 event.auth_events

---
 synapse/handlers/federation.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 35cad4182a..40836e0c51 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -805,7 +805,9 @@ class FederationHandler(BaseHandler):
                     yield self._handle_new_event(
                         origin, ev, auth_events=auth
                     )
-                    auth_events[(ev.type, ev.state_key)] = ev
+
+                    if ev.event_id in event_auth_events:
+                        auth_events[(ev.type, ev.state_key)] = ev
                 except AuthError:
                     pass
 

From 776ac820f9bb7f0e9c2fae9facbee05b0132079e Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 30 Jan 2015 15:58:28 +0000
Subject: [PATCH 15/15] Briefly doc structure of query_auth API.

---
 synapse/federation/federation_server.py | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index 84ed0a0ba0..5fbd8b19de 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -232,6 +232,24 @@ class FederationServer(object):
 
     @defer.inlineCallbacks
     def on_query_auth_request(self, origin, content, event_id):
+        """
+        Content is a dict with keys::
+            auth_chain (list): A list of events that give the auth chain.
+            missing (list): A list of event_ids indicating what the other
+              side (`origin`) think we're missing.
+            rejects (dict): A mapping from event_id to a 2-tuple of reason
+              string and a proof (or None) of why the event was rejected.
+              The keys of this dict give the list of events the `origin` has
+              rejected.
+
+        Args:
+            origin (str)
+            content (dict)
+            event_id (str)
+
+        Returns:
+            Deferred: Results in `dict` with the same format as `content`
+        """
         auth_chain = [
             (yield self._check_sigs_and_hash(self.event_from_pdu_json(e)))
             for e in content["auth_chain"]