Merge commit '5cdca53aa' into dinsic

* commit '5cdca53aa': Merge different Resource implementation classes (#7732) Fix inconsistent handling of upper and lower cases of email addresses. (#7021) Allow YAML config file to contain None (#7779) Fix a typo. Move 1.15.2 after 1.16.0rc2. 1.16.0rc2 Remove an extraneous space. Add links to the fixes. Fix tense in the release notes. Hack to add push priority to push notifications (#7765) Add early returns to `_check_for_soft_fail` (#7769) Use symbolic names for replication stream names (#7768) Type checking for `FederationHandler` (#7770) Fix new metric where we used ms instead of seconds (#7771) Fix incorrect error message when database CTYPE was set incorrectly. (#7760) Pin link in CHANGES.md Fixes to CHANGES.md
2020-08-03 17:08:31 -07:00
parent 678f9fb88e 5cdca53aa0
commit 60a96e4232
53 changed files with 1144 additions and 521 deletions
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,13 +1,55 @@
+Synapse 1.16.0rc2 (2020-07-02)
+==============================
+
+Synapse 1.16.0rc2 includes the security fixes released with Synapse 1.15.2.
+Please see [below](#synapse-1152-2020-07-02) for more details.
+
+Improved Documentation
+----------------------
+
+- Update postgres image in example `docker-compose.yaml` to tag `12-alpine`. ([\#7696](https://github.com/matrix-org/synapse/issues/7696))
+
+
+Internal Changes
+----------------
+
+- Add some metrics for inbound and outbound federation latencies: `synapse_federation_server_pdu_process_time` and `synapse_event_processing_lag_by_event`. ([\#7771](https://github.com/matrix-org/synapse/issues/7771))
+
+
+Synapse 1.15.2 (2020-07-02)
+===========================
+
+Due to the two security issues highlighted below, server administrators are
+encouraged to update Synapse. We are not aware of these vulnerabilities being
+exploited in the wild.
+
+Security advisory
+-----------------
+
+* A malicious homeserver could force Synapse to reset the state in a room to a
+  small subset of the correct state. This affects all Synapse deployments which
+  federate with untrusted servers. ([96e9afe6](https://github.com/matrix-org/synapse/commit/96e9afe62500310977dc3cbc99a8d16d3d2fa15c))
+* HTML pages served via Synapse were vulnerable to clickjacking attacks. This
+  predominantly affects homeservers with single-sign-on enabled, but all server
+  administrators are encouraged to upgrade. ([ea26e9a9](https://github.com/matrix-org/synapse/commit/ea26e9a98b0541fc886a1cb826a38352b7599dbe))
+
+  This was reported by [Quentin Gliech](https://sandhose.fr/).
+
+
 Synapse 1.16.0rc1 (2020-07-01)
 ==============================

+Note that this release deprecates the `m.login.jwt` login method, renaming it
+to `org.matrix.login.jwt`, as `m.login.jwt` is not part of the Matrix spec.
+Otherwise the behaviour is identical. Synapse will accept both names for now,
+but this may change in a future release.
+
 Features
 --------

 - Add an option to enable encryption by default for new rooms. ([\#7639](https://github.com/matrix-org/synapse/issues/7639))
- Add support for running multiple media repository workers. See [docs/workers.md](docs/workers.md) for instructions. ([\#7706](https://github.com/matrix-org/synapse/issues/7706))
+- Add support for running multiple media repository workers. See [docs/workers.md](https://github.com/matrix-org/synapse/blob/release-v1.16.0/docs/workers.md) for instructions. ([\#7706](https://github.com/matrix-org/synapse/issues/7706))
 - Media can now be marked as safe from quarantined. ([\#7718](https://github.com/matrix-org/synapse/issues/7718))
- Add unread messages count to sync responses. ([\#7761](https://github.com/matrix-org/synapse/issues/7761))
 - Expand the configuration options for auto-join rooms. ([\#7763](https://github.com/matrix-org/synapse/issues/7763))


@@ -15,17 +57,17 @@ Bugfixes
 --------

 - Remove `user_id` from the response to `GET /_matrix/client/r0/presence/{userId}/status` to match the specification. ([\#7606](https://github.com/matrix-org/synapse/issues/7606))
- In working mode, ensure that replicated data has not already been received. ([\#7648](https://github.com/matrix-org/synapse/issues/7648))
+- In worker mode, ensure that replicated data has not already been received. ([\#7648](https://github.com/matrix-org/synapse/issues/7648))
 - Fix intermittent exception during startup, introduced in Synapse 1.14.0. ([\#7663](https://github.com/matrix-org/synapse/issues/7663))
 - Include a user-agent for federation and well-known requests. ([\#7677](https://github.com/matrix-org/synapse/issues/7677))
- Accept the proper field (`phone`) for the `m.id.phone` identifier type. The legacy field of `number` is still accepted as a fallback. Bug introduced in v0.20.0-rc1. ([\#7687](https://github.com/matrix-org/synapse/issues/7687))
- Fix "Starting db txn 'get_completed_ui_auth_stages' from sentinel context" warning. The bug was introduced in 1.13.0rc1. ([\#7688](https://github.com/matrix-org/synapse/issues/7688))
- Compare the URI and method during user interactive authentication (instead of the URI twice). Bug introduced in 1.13.0rc1. ([\#7689](https://github.com/matrix-org/synapse/issues/7689))
+- Accept the proper field (`phone`) for the `m.id.phone` identifier type. The legacy field of `number` is still accepted as a fallback. Bug introduced in v0.20.0. ([\#7687](https://github.com/matrix-org/synapse/issues/7687))
+- Fix "Starting db txn 'get_completed_ui_auth_stages' from sentinel context" warning. The bug was introduced in 1.13.0. ([\#7688](https://github.com/matrix-org/synapse/issues/7688))
+- Compare the URI and method during user interactive authentication (instead of the URI twice). Bug introduced in 1.13.0. ([\#7689](https://github.com/matrix-org/synapse/issues/7689))
 - Fix a long standing bug where the response to the `GET room_keys/version` endpoint had the incorrect type for the `etag` field. ([\#7691](https://github.com/matrix-org/synapse/issues/7691))
 - Fix logged error during device resync in opentracing. Broke in v1.14.0. ([\#7698](https://github.com/matrix-org/synapse/issues/7698))
 - Do not break push rule evaluation when receiving an event with a non-string body. This is a long-standing bug. ([\#7701](https://github.com/matrix-org/synapse/issues/7701))
 - Fixs a long standing bug which resulted in an exception: "TypeError: argument of type 'ObservableDeferred' is not iterable". ([\#7708](https://github.com/matrix-org/synapse/issues/7708))
- The `synapse_port_db` script no longer fails when the `ui_auth_sessions` table is non-empty. This bug has existed since v1.13.0rc1. ([\#7711](https://github.com/matrix-org/synapse/issues/7711))
+- The `synapse_port_db` script no longer fails when the `ui_auth_sessions` table is non-empty. This bug has existed since v1.13.0. ([\#7711](https://github.com/matrix-org/synapse/issues/7711))
 - Synapse will now fetch media from the proper specified URL (using the r0 prefix instead of the unspecified v1). ([\#7714](https://github.com/matrix-org/synapse/issues/7714))
 - Fix the tables ignored by `synapse_port_db` to be in sync the current database schema. ([\#7717](https://github.com/matrix-org/synapse/issues/7717))
 - Fix missing `Content-Length` on HTTP responses from the metrics handler. ([\#7730](https://github.com/matrix-org/synapse/issues/7730))
@@ -51,7 +93,7 @@ Internal Changes

 - Refactor getting replication updates from database. ([\#7636](https://github.com/matrix-org/synapse/issues/7636))
 - Clean-up the login fallback code. ([\#7657](https://github.com/matrix-org/synapse/issues/7657))
- Increase the default SAML session expirary time to 15 minutes. ([\#7664](https://github.com/matrix-org/synapse/issues/7664))
+- Increase the default SAML session expiry time to 15 minutes. ([\#7664](https://github.com/matrix-org/synapse/issues/7664))
 - Convert the device message and pagination handlers to async/await. ([\#7678](https://github.com/matrix-org/synapse/issues/7678))
 - Convert typing handler to async/await. ([\#7679](https://github.com/matrix-org/synapse/issues/7679))
 - Require `parameterized` package version to be at least 0.7.0. ([\#7680](https://github.com/matrix-org/synapse/issues/7680))
@@ -68,26 +110,6 @@ Internal Changes
 - Add some metrics for inbound and outbound federation latencies: `synapse_federation_server_pdu_process_time` and `synapse_event_processing_lag_by_event`. ([\#7755](https://github.com/matrix-org/synapse/issues/7755))


-Synapse 1.15.2 (2020-07-02)
-===========================
-
-Due to the two security issues highlight below, server administrators are
-encouraged to update Synapse. We are not aware of these vulnerabilities being
-exploited in the wild.
-
-Security advisory
-----------------
-
-* A malicious homeserver could force Synapse to reset the state in a room to a
-  small subset of the correct state. This affects all Synapse deployments which
-  federate with untrusted servers.
-* HTML pages served via Synapse were vulnerable to clickjacking attacks. This
-  predominantly affects homeservers with single-sign-on enabled, but all server
-  administrators are encouraged to upgrade.
-
-  This was reported by [Quentin Gliech](https://sandhose.fr/).
-
-
 Synapse 1.15.1 (2020-06-16)
 ===========================

--- a/changelog.d/7021.bugfix
+++ b/changelog.d/7021.bugfix
@@ -0,0 +1 @@
+Fix inconsistent handling of upper and lower case in email addresses when used as identifiers for login, etc. Contributed by @dklimpel.
--- a/changelog.d/7696.doc
+++ b/changelog.d/7696.doc
@@ -1 +0,0 @@
-Update postgres image in example `docker-compose.yaml` to tag `12-alpine`.
--- a/changelog.d/7732.bugfix
+++ b/changelog.d/7732.bugfix
@@ -0,0 +1 @@
+Fix "Tried to close a non-active scope!" error messages when opentracing is enabled.
--- a/changelog.d/7760.bugfix
+++ b/changelog.d/7760.bugfix
@@ -0,0 +1 @@
+Fix incorrect error message when database CTYPE was set incorrectly.
--- a/changelog.d/7765.misc
+++ b/changelog.d/7765.misc
@@ -0,0 +1 @@
+Send push notifications with a high or low priority depending upon whether they may generate user-observable effects.
--- a/changelog.d/7768.misc
+++ b/changelog.d/7768.misc
@@ -0,0 +1 @@
+Use symbolic names for replication stream names.
--- a/changelog.d/7769.misc
+++ b/changelog.d/7769.misc
@@ -0,0 +1 @@
+Add early returns to `_check_for_soft_fail`.
--- a/changelog.d/7770.misc
+++ b/changelog.d/7770.misc
@@ -0,0 +1 @@
+Fix up `synapse.handlers.federation` to pass mypy.
--- a/changelog.d/7779.bugfix
+++ b/changelog.d/7779.bugfix
@@ -0,0 +1 @@
+Fix synctl to handle empty config files correctly. Contributed by @kotovalexarian.
--- a/synapse/init.py
+++ b/synapse/init.py
@@ -36,7 +36,7 @@ try:
 except ImportError:
    pass

-__version__ = "1.15.2"
+__version__ = "1.16.0rc2"

 if bool(os.environ.get("SYNAPSE_TEST_PATCH_LOG_CONTEXTS", False)):
    # We import here so that we don't have to install a bunch of deps when
--- a/synapse/federation/sender/init.py
+++ b/synapse/federation/sender/init.py
@@ -209,7 +209,7 @@ class FederationSender(object):

                        synapse.metrics.event_processing_lag_by_event.labels(
                            "federation_sender"
-                        ).observe(now - ts)
+                        ).observe((now - ts) / 1000)

                async def handle_room_events(events: Iterable[EventBase]) -> None:
                    with Measure(self.clock, "handle_room_events"):
--- a/synapse/federation/transport/server.py
+++ b/synapse/federation/transport/server.py
@@ -362,11 +362,7 @@ class BaseFederationServlet(object):
                continue

            server.register_paths(
-                method,
-                (pattern,),
-                self._wrap(code),
-                self.__class__.__name__,
-                trace=False,
+                method, (pattern,), self._wrap(code), self.__class__.__name__,
            )


--- a/synapse/handlers/appservice.py
+++ b/synapse/handlers/appservice.py
@@ -118,7 +118,7 @@ class ApplicationServicesHandler(object):
                        ts = yield self.store.get_received_ts(event.event_id)
                        synapse.metrics.event_processing_lag_by_event.labels(
                            "appservice_sender"
-                        ).observe(now - ts)
+                        ).observe((now - ts) / 1000)

                    @defer.inlineCallbacks
                    def handle_room_events(events):
--- a/synapse/handlers/auth.py
+++ b/synapse/handlers/auth.py
@@ -45,6 +45,7 @@ from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.module_api import ModuleApi
 from synapse.push.mailer import load_jinja2_templates
 from synapse.types import Requester, UserID
+from synapse.util.threepids import canonicalise_email

 from ._base import BaseHandler

@@ -928,7 +929,7 @@ class AuthHandler(BaseHandler):
        # for the presence of an email address during password reset was
        # case sensitive).
        if medium == "email":
-            address = address.lower()
+            address = canonicalise_email(address)

        await self.store.user_add_threepid(
            user_id, medium, address, validated_at, self.hs.get_clock().time_msec()
@@ -956,7 +957,7 @@ class AuthHandler(BaseHandler):

        # 'Canonicalise' email addresses as per above
        if medium == "email":
-            address = address.lower()
+            address = canonicalise_email(address)

        identity_handler = self.hs.get_handlers().identity_handler
        result = await identity_handler.try_unbind_threepid(
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -19,8 +19,9 @@

 import itertools
 import logging
+from collections import Container
 from http import HTTPStatus
-from typing import Dict, Iterable, List, Optional, Sequence, Tuple
+from typing import Dict, Iterable, List, Optional, Sequence, Tuple, Union

 import attr
 from signedjson.key import decode_verify_key_bytes
@@ -745,6 +746,9 @@ class FederationHandler(BaseHandler):
                # device and recognize the algorithm then we can work out the
                # exact key to expect. Otherwise check it matches any key we
                # have for that device.
+
+                current_keys = []  # type: Container[str]
+
                if device:
                    keys = device.get("keys", {}).get("keys", {})

@@ -761,15 +765,15 @@ class FederationHandler(BaseHandler):
                        current_keys = keys.values()
                elif device_id:
                    # We don't have any keys for the device ID.
-                    current_keys = []
+                    pass
                else:
                    # The event didn't include a device ID, so we just look for
                    # keys across all devices.
-                    current_keys = (
+                    current_keys = [
                        key
                        for device in cached_devices
                        for key in device.get("keys", {}).get("keys", {}).values()
-                    )
+                    ]

                # We now check that the sender key matches (one of) the expected
                # keys.
@@ -1014,7 +1018,7 @@ class FederationHandler(BaseHandler):
                if e_type == EventTypes.Member and event.membership == Membership.JOIN
            ]

-            joined_domains = {}
+            joined_domains = {}  # type: Dict[str, int]
            for u, d in joined_users:
                try:
                    dom = get_domain_from_id(u)
@@ -1280,14 +1284,15 @@ class FederationHandler(BaseHandler):
        try:
            # Try the host we successfully got a response to /make_join/
            # request first.
+            host_list = list(target_hosts)
            try:
-                target_hosts.remove(origin)
-                target_hosts.insert(0, origin)
+                host_list.remove(origin)
+                host_list.insert(0, origin)
            except ValueError:
                pass

            ret = await self.federation_client.send_join(
-                target_hosts, event, room_version_obj
+                host_list, event, room_version_obj
            )

            origin = ret["origin"]
@@ -1594,13 +1599,14 @@ class FederationHandler(BaseHandler):

        # Try the host that we succesfully called /make_leave/ on first for
        # the /send_leave/ request.
+        host_list = list(target_hosts)
        try:
-            target_hosts.remove(origin)
-            target_hosts.insert(0, origin)
+            host_list.remove(origin)
+            host_list.insert(0, origin)
        except ValueError:
            pass

-        await self.federation_client.send_leave(target_hosts, event)
+        await self.federation_client.send_leave(host_list, event)

        context = await self.state_handler.compute_event_context(event)
        stream_id = await self.persist_events_and_notify([(event, context)])
@@ -1614,7 +1620,7 @@ class FederationHandler(BaseHandler):
        user_id: str,
        membership: str,
        content: JsonDict = {},
-        params: Optional[Dict[str, str]] = None,
+        params: Optional[Dict[str, Union[str, Iterable[str]]]] = None,
    ) -> Tuple[str, EventBase, RoomVersion]:
        (
            origin,
@@ -2028,8 +2034,8 @@ class FederationHandler(BaseHandler):
            auth_events_ids = await self.auth.compute_auth_events(
                event, prev_state_ids, for_verification=True
            )
-            auth_events = await self.store.get_events(auth_events_ids)
-            auth_events = {(e.type, e.state_key): e for e in auth_events.values()}
+            auth_events_x = await self.store.get_events(auth_events_ids)
+            auth_events = {(e.type, e.state_key): e for e in auth_events_x.values()}

        # This is a hack to fix some old rooms where the initial join event
        # didn't reference the create event in its auth events.
@@ -2065,76 +2071,67 @@ class FederationHandler(BaseHandler):
        # For new (non-backfilled and non-outlier) events we check if the event
        # passes auth based on the current state. If it doesn't then we
        # "soft-fail" the event.
-        do_soft_fail_check = not backfilled and not event.internal_metadata.is_outlier()
-        if do_soft_fail_check:
-            extrem_ids = await self.store.get_latest_event_ids_in_room(event.room_id)
+        if backfilled or event.internal_metadata.is_outlier():
+            return

-            extrem_ids = set(extrem_ids)
-            prev_event_ids = set(event.prev_event_ids())
+        extrem_ids = await self.store.get_latest_event_ids_in_room(event.room_id)
+        extrem_ids = set(extrem_ids)
+        prev_event_ids = set(event.prev_event_ids())

-            if extrem_ids == prev_event_ids:
-                # If they're the same then the current state is the same as the
-                # state at the event, so no point rechecking auth for soft fail.
-                do_soft_fail_check = False
+        if extrem_ids == prev_event_ids:
+            # If they're the same then the current state is the same as the
+            # state at the event, so no point rechecking auth for soft fail.
+            return

-        if do_soft_fail_check:
-            room_version = await self.store.get_room_version_id(event.room_id)
-            room_version_obj = KNOWN_ROOM_VERSIONS[room_version]
+        room_version = await self.store.get_room_version_id(event.room_id)
+        room_version_obj = KNOWN_ROOM_VERSIONS[room_version]

-            # Calculate the "current state".
-            if state is not None:
-                # If we're explicitly given the state then we won't have all the
-                # prev events, and so we have a gap in the graph. In this case
-                # we want to be a little careful as we might have been down for
-                # a while and have an incorrect view of the current state,
-                # however we still want to do checks as gaps are easy to
-                # maliciously manufacture.
-                #
-                # So we use a "current state" that is actually a state
-                # resolution across the current forward extremities and the
-                # given state at the event. This should correctly handle cases
-                # like bans, especially with state res v2.
+        # Calculate the "current state".
+        if state is not None:
+            # If we're explicitly given the state then we won't have all the
+            # prev events, and so we have a gap in the graph. In this case
+            # we want to be a little careful as we might have been down for
+            # a while and have an incorrect view of the current state,
+            # however we still want to do checks as gaps are easy to
+            # maliciously manufacture.
+            #
+            # So we use a "current state" that is actually a state
+            # resolution across the current forward extremities and the
+            # given state at the event. This should correctly handle cases
+            # like bans, especially with state res v2.

-                state_sets = await self.state_store.get_state_groups(
-                    event.room_id, extrem_ids
-                )
-                state_sets = list(state_sets.values())
-                state_sets.append(state)
-                current_state_ids = await self.state_handler.resolve_events(
-                    room_version, state_sets, event
-                )
-                current_state_ids = {
-                    k: e.event_id for k, e in current_state_ids.items()
-                }
-            else:
-                current_state_ids = await self.state_handler.get_current_state_ids(
-                    event.room_id, latest_event_ids=extrem_ids
-                )
-
-            logger.debug(
-                "Doing soft-fail check for %s: state %s",
-                event.event_id,
-                current_state_ids,
+            state_sets = await self.state_store.get_state_groups(
+                event.room_id, extrem_ids
+            )
+            state_sets = list(state_sets.values())
+            state_sets.append(state)
+            current_state_ids = await self.state_handler.resolve_events(
+                room_version, state_sets, event
+            )
+            current_state_ids = {k: e.event_id for k, e in current_state_ids.items()}
+        else:
+            current_state_ids = await self.state_handler.get_current_state_ids(
+                event.room_id, latest_event_ids=extrem_ids
            )

-            # Now check if event pass auth against said current state
-            auth_types = auth_types_for_event(event)
-            current_state_ids = [
-                e for k, e in current_state_ids.items() if k in auth_types
-            ]
+        logger.debug(
+            "Doing soft-fail check for %s: state %s", event.event_id, current_state_ids,
+        )

-            current_auth_events = await self.store.get_events(current_state_ids)
-            current_auth_events = {
-                (e.type, e.state_key): e for e in current_auth_events.values()
-            }
+        # Now check if event pass auth against said current state
+        auth_types = auth_types_for_event(event)
+        current_state_ids = [e for k, e in current_state_ids.items() if k in auth_types]

-            try:
-                event_auth.check(
-                    room_version_obj, event, auth_events=current_auth_events
-                )
-            except AuthError as e:
-                logger.warning("Soft-failing %r because %s", event, e)
-                event.internal_metadata.soft_failed = True
+        current_auth_events = await self.store.get_events(current_state_ids)
+        current_auth_events = {
+            (e.type, e.state_key): e for e in current_auth_events.values()
+        }
+
+        try:
+            event_auth.check(room_version_obj, event, auth_events=current_auth_events)
+        except AuthError as e:
+            logger.warning("Soft-failing %r because %s", event, e)
+            event.internal_metadata.soft_failed = True

    async def on_query_auth(
        self, origin, event_id, room_id, remote_auth_chain, rejects, missing
@@ -2303,10 +2300,10 @@ class FederationHandler(BaseHandler):
                    remote_auth_chain = await self.federation_client.get_event_auth(
                        origin, event.room_id, event.event_id
                    )
-                except RequestSendFailed as e:
+                except RequestSendFailed as e1:
                    # The other side isn't around or doesn't implement the
                    # endpoint, so lets just bail out.
-                    logger.info("Failed to get event auth from remote: %s", e)
+                    logger.info("Failed to get event auth from remote: %s", e1)
                    return context

                seen_remotes = await self.store.have_seen_events(
@@ -2784,7 +2781,8 @@ class FederationHandler(BaseHandler):

        logger.debug("Checking auth on event %r", event.content)

-        last_exception = None
+        last_exception = None  # type: Optional[Exception]
+
        # for each public key in the 3pid invite event
        for public_key_object in self.hs.get_auth().get_public_keys(invite_event):
            try:
@@ -2838,6 +2836,12 @@ class FederationHandler(BaseHandler):
                        return
            except Exception as e:
                last_exception = e
+
+        if last_exception is None:
+            # we can only get here if get_public_keys() returned an empty list
+            # TODO: make this better
+            raise RuntimeError("no public key in invite event")
+
        raise last_exception

    async def _check_key_revocation(self, public_key, url):
--- a/synapse/http/additional_resource.py
+++ b/synapse/http/additional_resource.py
@@ -13,13 +13,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

-from twisted.web.resource import Resource
-from twisted.web.server import NOT_DONE_YET
-
-from synapse.http.server import wrap_json_request_handler
+from synapse.http.server import DirectServeJsonResource


-class AdditionalResource(Resource):
+class AdditionalResource(DirectServeJsonResource):
    """Resource wrapper for additional_resources

    If the user has configured additional_resources, we need to wrap the
@@ -41,16 +38,10 @@ class AdditionalResource(Resource):
            handler ((twisted.web.server.Request) -> twisted.internet.defer.Deferred):
                function to be called to handle the request.
        """
-        Resource.__init__(self)
+        super().__init__()
        self._handler = handler

-        # required by the request_handler wrapper
-        self.clock = hs.get_clock()
-
-    def render(self, request):
-        self._async_render(request)
-        return NOT_DONE_YET
-
-    @wrap_json_request_handler
    def _async_render(self, request):
+        # Cheekily pass the result straight through, so we don't need to worry
+        # if its an awaitable or not.
        return self._handler(request)
--- a/synapse/http/server.py
+++ b/synapse/http/server.py
@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

+import abc
 import collections
 import html
 import logging
@@ -21,7 +22,7 @@ import types
 import urllib
 from http import HTTPStatus
 from io import BytesIO
-from typing import Awaitable, Callable, TypeVar, Union
+from typing import Any, Callable, Dict, Tuple, Union

 import jinja2
 from canonicaljson import encode_canonical_json, encode_pretty_printed_json, json
@@ -62,99 +63,43 @@ HTML_ERROR_TEMPLATE = """<!DOCTYPE html>
 """


-def wrap_json_request_handler(h):
-    """Wraps a request handler method with exception handling.
-
-    Also does the wrapping with request.processing as per wrap_async_request_handler.
-
-    The handler method must have a signature of "handle_foo(self, request)",
-    where "request" must be a SynapseRequest.
-
-    The handler must return a deferred or a coroutine. If the deferred succeeds
-    we assume that a response has been sent. If the deferred fails with a SynapseError we use
-    it to send a JSON response with the appropriate HTTP reponse code. If the
-    deferred fails with any other type of error we send a 500 reponse.
+def return_json_error(f: failure.Failure, request: SynapseRequest) -> None:
+    """Sends a JSON error response to clients.
    """

-    async def wrapped_request_handler(self, request):
-        try:
-            await h(self, request)
-        except SynapseError as e:
-            code = e.code
-            logger.info("%s SynapseError: %s - %s", request, code, e.msg)
+    if f.check(SynapseError):
+        error_code = f.value.code
+        error_dict = f.value.error_dict()

-            # Only respond with an error response if we haven't already started
-            # writing, otherwise lets just kill the connection
-            if request.startedWriting:
-                if request.transport:
-                    try:
-                        request.transport.abortConnection()
-                    except Exception:
-                        # abortConnection throws if the connection is already closed
-                        pass
-            else:
-                respond_with_json(
-                    request,
-                    code,
-                    e.error_dict(),
-                    send_cors=True,
-                    pretty_print=_request_user_agent_is_curl(request),
-                )
+        logger.info("%s SynapseError: %s - %s", request, error_code, f.value.msg)
+    else:
+        error_code = 500
+        error_dict = {"error": "Internal server error", "errcode": Codes.UNKNOWN}

-        except Exception:
-            # failure.Failure() fishes the original Failure out
-            # of our stack, and thus gives us a sensible stack
-            # trace.
-            f = failure.Failure()
-            logger.error(
-                "Failed handle request via %r: %r",
-                request.request_metrics.name,
-                request,
-                exc_info=(f.type, f.value, f.getTracebackObject()),
-            )
-            # Only respond with an error response if we haven't already started
-            # writing, otherwise lets just kill the connection
-            if request.startedWriting:
-                if request.transport:
-                    try:
-                        request.transport.abortConnection()
-                    except Exception:
-                        # abortConnection throws if the connection is already closed
-                        pass
-            else:
-                respond_with_json(
-                    request,
-                    500,
-                    {"error": "Internal server error", "errcode": Codes.UNKNOWN},
-                    send_cors=True,
-                    pretty_print=_request_user_agent_is_curl(request),
-                )
+        logger.error(
+            "Failed handle request via %r: %r",
+            request.request_metrics.name,
+            request,
+            exc_info=(f.type, f.value, f.getTracebackObject()),
+        )

-    return wrap_async_request_handler(wrapped_request_handler)
-
-
-TV = TypeVar("TV")
-
-
-def wrap_html_request_handler(
-    h: Callable[[TV, SynapseRequest], Awaitable]
-) -> Callable[[TV, SynapseRequest], Awaitable[None]]:
-    """Wraps a request handler method with exception handling.
-
-    Also does the wrapping with request.processing as per wrap_async_request_handler.
-
-    The handler method must have a signature of "handle_foo(self, request)",
-    where "request" must be a SynapseRequest.
-    """
-
-    async def wrapped_request_handler(self, request):
-        try:
-            await h(self, request)
-        except Exception:
-            f = failure.Failure()
-            return_html_error(f, request, HTML_ERROR_TEMPLATE)
-
-    return wrap_async_request_handler(wrapped_request_handler)
+    # Only respond with an error response if we haven't already started writing,
+    # otherwise lets just kill the connection
+    if request.startedWriting:
+        if request.transport:
+            try:
+                request.transport.abortConnection()
+            except Exception:
+                # abortConnection throws if the connection is already closed
+                pass
+    else:
+        respond_with_json(
+            request,
+            error_code,
+            error_dict,
+            send_cors=True,
+            pretty_print=_request_user_agent_is_curl(request),
+        )


 def return_html_error(
@@ -249,7 +194,113 @@ class HttpServer(object):
        pass


-class JsonResource(HttpServer, resource.Resource):
+class _AsyncResource(resource.Resource, metaclass=abc.ABCMeta):
+    """Base class for resources that have async handlers.
+
+    Sub classes can either implement `_async_render_<METHOD>` to handle
+    requests by method, or override `_async_render` to handle all requests.
+
+    Args:
+        extract_context: Whether to attempt to extract the opentracing
+            context from the request the servlet is handling.
+    """
+
+    def __init__(self, extract_context=False):
+        super().__init__()
+
+        self._extract_context = extract_context
+
+    def render(self, request):
+        """ This gets called by twisted every time someone sends us a request.
+        """
+        defer.ensureDeferred(self._async_render_wrapper(request))
+        return NOT_DONE_YET
+
+    @wrap_async_request_handler
+    async def _async_render_wrapper(self, request):
+        """This is a wrapper that delegates to `_async_render` and handles
+        exceptions, return values, metrics, etc.
+        """
+        try:
+            request.request_metrics.name = self.__class__.__name__
+
+            with trace_servlet(request, self._extract_context):
+                callback_return = await self._async_render(request)
+
+                if callback_return is not None:
+                    code, response = callback_return
+                    self._send_response(request, code, response)
+        except Exception:
+            # failure.Failure() fishes the original Failure out
+            # of our stack, and thus gives us a sensible stack
+            # trace.
+            f = failure.Failure()
+            self._send_error_response(f, request)
+
+    async def _async_render(self, request):
+        """Delegates to `_async_render_<METHOD>` methods, or returns a 400 if
+        no appropriate method exists. Can be overriden in sub classes for
+        different routing.
+        """
+
+        method_handler = getattr(
+            self, "_async_render_%s" % (request.method.decode("ascii"),), None
+        )
+        if method_handler:
+            raw_callback_return = method_handler(request)
+
+            # Is it synchronous? We'll allow this for now.
+            if isinstance(raw_callback_return, (defer.Deferred, types.CoroutineType)):
+                callback_return = await raw_callback_return
+            else:
+                callback_return = raw_callback_return
+
+            return callback_return
+
+        _unrecognised_request_handler(request)
+
+    @abc.abstractmethod
+    def _send_response(
+        self, request: SynapseRequest, code: int, response_object: Any,
+    ) -> None:
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def _send_error_response(
+        self, f: failure.Failure, request: SynapseRequest,
+    ) -> None:
+        raise NotImplementedError()
+
+
+class DirectServeJsonResource(_AsyncResource):
+    """A resource that will call `self._async_on_<METHOD>` on new requests,
+    formatting responses and errors as JSON.
+    """
+
+    def _send_response(
+        self, request, code, response_object,
+    ):
+        """Implements _AsyncResource._send_response
+        """
+        # TODO: Only enable CORS for the requests that need it.
+        respond_with_json(
+            request,
+            code,
+            response_object,
+            send_cors=True,
+            pretty_print=_request_user_agent_is_curl(request),
+            canonical_json=self.canonical_json,
+        )
+
+    def _send_error_response(
+        self, f: failure.Failure, request: SynapseRequest,
+    ) -> None:
+        """Implements _AsyncResource._send_error_response
+        """
+        return_json_error(f, request)
+
+
+class JsonResource(DirectServeJsonResource):
    """ This implements the HttpServer interface and provides JSON support for
    Resources.

@@ -269,17 +320,15 @@ class JsonResource(HttpServer, resource.Resource):
        "_PathEntry", ["pattern", "callback", "servlet_classname"]
    )

-    def __init__(self, hs, canonical_json=True):
-        resource.Resource.__init__(self)
+    def __init__(self, hs, canonical_json=True, extract_context=False):
+        super().__init__(extract_context)

        self.canonical_json = canonical_json
        self.clock = hs.get_clock()
        self.path_regexs = {}
        self.hs = hs

-    def register_paths(
-        self, method, path_patterns, callback, servlet_classname, trace=True
-    ):
+    def register_paths(self, method, path_patterns, callback, servlet_classname):
        """
        Registers a request handler against a regular expression. Later request URLs are
        checked against these regular expressions in order to identify an appropriate
@@ -295,74 +344,23 @@ class JsonResource(HttpServer, resource.Resource):

            servlet_classname (str): The name of the handler to be used in prometheus
                and opentracing logs.
-
-            trace (bool): Whether we should start a span to trace the servlet.
        """
        method = method.encode("utf-8")  # method is bytes on py3

-        if trace:
-            # We don't extract the context from the servlet because we can't
-            # trust the sender
-            callback = trace_servlet(servlet_classname)(callback)
-
        for path_pattern in path_patterns:
            logger.debug("Registering for %s %s", method, path_pattern.pattern)
            self.path_regexs.setdefault(method, []).append(
                self._PathEntry(path_pattern, callback, servlet_classname)
            )

-    def render(self, request):
-        """ This gets called by twisted every time someone sends us a request.
-        """
-        defer.ensureDeferred(self._async_render(request))
-        return NOT_DONE_YET
-
-    @wrap_json_request_handler
-    async def _async_render(self, request):
-        """ This gets called from render() every time someone sends us a request.
-            This checks if anyone has registered a callback for that method and
-            path.
-        """
-        callback, servlet_classname, group_dict = self._get_handler_for_request(request)
-
-        # Make sure we have a name for this handler in prometheus.
-        request.request_metrics.name = servlet_classname
-
-        # Now trigger the callback. If it returns a response, we send it
-        # here. If it throws an exception, that is handled by the wrapper
-        # installed by @request_handler.
-        kwargs = intern_dict(
-            {
-                name: urllib.parse.unquote(value) if value else value
-                for name, value in group_dict.items()
-            }
-        )
-
-        callback_return = callback(request, **kwargs)
-
-        # Is it synchronous? We'll allow this for now.
-        if isinstance(callback_return, (defer.Deferred, types.CoroutineType)):
-            callback_return = await callback_return
-
-        if callback_return is not None:
-            code, response = callback_return
-            self._send_response(request, code, response)
-
-    def _get_handler_for_request(self, request):
-        """Finds a callback method to handle the given request
-
-        Args:
-            request (twisted.web.http.Request):
+    def _get_handler_for_request(
+        self, request: SynapseRequest
+    ) -> Tuple[Callable, str, Dict[str, str]]:
+        """Finds a callback method to handle the given request.

        Returns:
-            Tuple[Callable, str, dict[unicode, unicode]]: callback method, the
-                label to use for that method in prometheus metrics, and the
-                dict mapping keys to path components as specified in the
-                handler's path match regexp.
-
-                The callback will normally be a method registered via
-                register_paths, so will return (possibly via Deferred) either
-                None, or a tuple of (http code, response body).
+            A tuple of the callback to use, the name of the servlet, and the
+            key word arguments to pass to the callback
        """
        request_path = request.path.decode("ascii")

@@ -377,42 +375,59 @@ class JsonResource(HttpServer, resource.Resource):
        # Huh. No one wanted to handle that? Fiiiiiine. Send 400.
        return _unrecognised_request_handler, "unrecognised_request_handler", {}

-    def _send_response(
-        self, request, code, response_json_object, response_code_message=None
-    ):
-        # TODO: Only enable CORS for the requests that need it.
-        respond_with_json(
-            request,
-            code,
-            response_json_object,
-            send_cors=True,
-            response_code_message=response_code_message,
-            pretty_print=_request_user_agent_is_curl(request),
-            canonical_json=self.canonical_json,
+    async def _async_render(self, request):
+        callback, servlet_classname, group_dict = self._get_handler_for_request(request)
+
+        # Make sure we have an appopriate name for this handler in prometheus
+        # (rather than the default of JsonResource).
+        request.request_metrics.name = servlet_classname
+
+        # Now trigger the callback. If it returns a response, we send it
+        # here. If it throws an exception, that is handled by the wrapper
+        # installed by @request_handler.
+        kwargs = intern_dict(
+            {
+                name: urllib.parse.unquote(value) if value else value
+                for name, value in group_dict.items()
+            }
        )

+        raw_callback_return = callback(request, **kwargs)

-class DirectServeResource(resource.Resource):
-    def render(self, request):
+        # Is it synchronous? We'll allow this for now.
+        if isinstance(raw_callback_return, (defer.Deferred, types.CoroutineType)):
+            callback_return = await raw_callback_return
+        else:
+            callback_return = raw_callback_return
+
+        return callback_return
+
+
+class DirectServeHtmlResource(_AsyncResource):
+    """A resource that will call `self._async_on_<METHOD>` on new requests,
+    formatting responses and errors as HTML.
+    """
+
+    # The error template to use for this resource
+    ERROR_TEMPLATE = HTML_ERROR_TEMPLATE
+
+    def _send_response(
+        self, request: SynapseRequest, code: int, response_object: Any,
+    ):
+        """Implements _AsyncResource._send_response
        """
-        Render the request, using an asynchronous render handler if it exists.
+        # We expect to get bytes for us to write
+        assert isinstance(response_object, bytes)
+        html_bytes = response_object
+
+        respond_with_html_bytes(request, 200, html_bytes)
+
+    def _send_error_response(
+        self, f: failure.Failure, request: SynapseRequest,
+    ) -> None:
+        """Implements _AsyncResource._send_error_response
        """
-        async_render_callback_name = "_async_render_" + request.method.decode("ascii")
-
-        # Try and get the async renderer
-        callback = getattr(self, async_render_callback_name, None)
-
-        # No async renderer for this request method.
-        if not callback:
-            return super().render(request)
-
-        resp = trace_servlet(self.__class__.__name__)(callback)(request)
-
-        # If it's a coroutine, turn it into a Deferred
-        if isinstance(resp, types.CoroutineType):
-            defer.ensureDeferred(resp)
-
-        return NOT_DONE_YET
+        return_html_error(f, request, self.ERROR_TEMPLATE)


 class StaticResource(File):
--- a/synapse/logging/opentracing.py
+++ b/synapse/logging/opentracing.py
@@ -169,7 +169,6 @@ import contextlib
 import inspect
 import logging
 import re
-import types
 from functools import wraps
 from typing import TYPE_CHECKING, Dict, Optional, Type

@@ -182,6 +181,7 @@ from synapse.config import ConfigError

 if TYPE_CHECKING:
    from synapse.server import HomeServer
+    from synapse.http.site import SynapseRequest

 # Helper class

@@ -793,48 +793,42 @@ def tag_args(func):
    return _tag_args_inner


-def trace_servlet(servlet_name, extract_context=False):
-    """Decorator which traces a serlet. It starts a span with some servlet specific
-    tags such as the servlet_name and request information
+@contextlib.contextmanager
+def trace_servlet(request: "SynapseRequest", extract_context: bool = False):
+    """Returns a context manager which traces a request. It starts a span
+    with some servlet specific tags such as the request metrics name and
+    request information.

    Args:
-        servlet_name (str): The name to be used for the span's operation_name
-        extract_context (bool): Whether to attempt to extract the opentracing
+        request
+        extract_context: Whether to attempt to extract the opentracing
            context from the request the servlet is handling.
-
    """

-    def _trace_servlet_inner_1(func):
-        if not opentracing:
-            return func
+    if opentracing is None:
+        yield
+        return

-        @wraps(func)
-        async def _trace_servlet_inner(request, *args, **kwargs):
-            request_tags = {
-                "request_id": request.get_request_id(),
-                tags.SPAN_KIND: tags.SPAN_KIND_RPC_SERVER,
-                tags.HTTP_METHOD: request.get_method(),
-                tags.HTTP_URL: request.get_redacted_uri(),
-                tags.PEER_HOST_IPV6: request.getClientIP(),
-            }
+    request_tags = {
+        "request_id": request.get_request_id(),
+        tags.SPAN_KIND: tags.SPAN_KIND_RPC_SERVER,
+        tags.HTTP_METHOD: request.get_method(),
+        tags.HTTP_URL: request.get_redacted_uri(),
+        tags.PEER_HOST_IPV6: request.getClientIP(),
+    }

-            if extract_context:
-                scope = start_active_span_from_request(
-                    request, servlet_name, tags=request_tags
-                )
-            else:
-                scope = start_active_span(servlet_name, tags=request_tags)
+    request_name = request.request_metrics.name
+    if extract_context:
+        scope = start_active_span_from_request(request, request_name, tags=request_tags)
+    else:
+        scope = start_active_span(request_name, tags=request_tags)

-            with scope:
-                result = func(request, *args, **kwargs)
+    with scope:
+        try:
+            yield
+        finally:
+            # We set the operation name again in case its changed (which happens
+            # with JsonResource).
+            scope.span.set_operation_name(request.request_metrics.name)

-                if not isinstance(result, (types.CoroutineType, defer.Deferred)):
-                    # Some servlets aren't async and just return results
-                    # directly, so we handle that here.
-                    return result
-
-                return await result
-
-        return _trace_servlet_inner
-
-    return _trace_servlet_inner_1
+            scope.span.set_tag("request_tag", request.request_metrics.start_context.tag)
--- a/synapse/push/httppusher.py
+++ b/synapse/push/httppusher.py
@@ -20,6 +20,7 @@ from prometheus_client import Counter
 from twisted.internet import defer
 from twisted.internet.error import AlreadyCalled, AlreadyCancelled

+from synapse.api.constants import EventTypes
 from synapse.logging import opentracing
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.push import PusherConfigException
@@ -305,12 +306,23 @@ class HttpPusher(object):

    @defer.inlineCallbacks
    def _build_notification_dict(self, event, tweaks, badge):
+        priority = "low"
+        if (
+            event.type == EventTypes.Encrypted
+            or tweaks.get("highlight")
+            or tweaks.get("sound")
+        ):
+            # HACK send our push as high priority only if it generates a sound, highlight
+            #  or may do so (i.e. is encrypted so has unknown effects).
+            priority = "high"
+
        if self.data.get("format") == "event_id_only":
            d = {
                "notification": {
                    "event_id": event.event_id,
                    "room_id": event.room_id,
                    "counts": {"unread": badge},
+                    "prio": priority,
                    "devices": [
                        {
                            "app_id": self.app_id,
@@ -334,9 +346,8 @@ class HttpPusher(object):
                "room_id": event.room_id,
                "type": event.type,
                "sender": event.user_id,
-                "counts": {  # -- we don't mark messages as read yet so
-                    # we have no way of knowing
-                    # Just set the badge to 1 until we have read receipts
+                "prio": priority,
+                "counts": {
                    "unread": badge,
                    # 'missed_calls': 2
                },
--- a/synapse/replication/http/init.py
+++ b/synapse/replication/http/init.py
@@ -30,7 +30,8 @@ REPLICATION_PREFIX = "/_synapse/replication"

 class ReplicationRestResource(JsonResource):
    def __init__(self, hs):
-        JsonResource.__init__(self, hs, canonical_json=False)
+        # We enable extracting jaeger contexts here as these are internal APIs.
+        super().__init__(hs, canonical_json=False, extract_context=True)
        self.register_servlets(hs)

    def register_servlets(self, hs):
--- a/synapse/replication/http/_base.py
+++ b/synapse/replication/http/_base.py
@@ -28,11 +28,7 @@ from synapse.api.errors import (
    RequestSendFailed,
    SynapseError,
 )
-from synapse.logging.opentracing import (
-    inject_active_span_byte_dict,
-    trace,
-    trace_servlet,
-)
+from synapse.logging.opentracing import inject_active_span_byte_dict, trace
 from synapse.util.caches.response_cache import ResponseCache
 from synapse.util.stringutils import random_string

@@ -240,11 +236,8 @@ class ReplicationEndpoint(object):
        args = "/".join("(?P<%s>[^/]+)" % (arg,) for arg in url_args)
        pattern = re.compile("^/_synapse/replication/%s/%s$" % (self.NAME, args))

-        handler = trace_servlet(self.__class__.__name__, extract_context=True)(handler)
-        # We don't let register paths trace this servlet using the default tracing
-        # options because we wish to extract the context explicitly.
        http_server.register_paths(
-            method, [pattern], handler, self.__class__.__name__, trace=False
+            method, [pattern], handler, self.__class__.__name__,
        )

    def _cached_handler(self, request, txn_id, **kwargs):
--- a/synapse/replication/slave/storage/account_data.py
+++ b/synapse/replication/slave/storage/account_data.py
@@ -16,6 +16,7 @@

 from synapse.replication.slave.storage._base import BaseSlavedStore
 from synapse.replication.slave.storage._slaved_id_tracker import SlavedIdTracker
+from synapse.replication.tcp.streams import AccountDataStream, TagAccountDataStream
 from synapse.storage.data_stores.main.account_data import AccountDataWorkerStore
 from synapse.storage.data_stores.main.tags import TagsWorkerStore
 from synapse.storage.database import Database
@@ -39,12 +40,12 @@ class SlavedAccountDataStore(TagsWorkerStore, AccountDataWorkerStore, BaseSlaved
        return self._account_data_id_gen.get_current_token()

    def process_replication_rows(self, stream_name, instance_name, token, rows):
-        if stream_name == "tag_account_data":
+        if stream_name == TagAccountDataStream.NAME:
            self._account_data_id_gen.advance(token)
            for row in rows:
                self.get_tags_for_user.invalidate((row.user_id,))
                self._account_data_stream_cache.entity_has_changed(row.user_id, token)
-        elif stream_name == "account_data":
+        elif stream_name == AccountDataStream.NAME:
            self._account_data_id_gen.advance(token)
            for row in rows:
                if not row.room_id:
--- a/synapse/replication/slave/storage/deviceinbox.py
+++ b/synapse/replication/slave/storage/deviceinbox.py
@@ -15,6 +15,7 @@

 from synapse.replication.slave.storage._base import BaseSlavedStore
 from synapse.replication.slave.storage._slaved_id_tracker import SlavedIdTracker
+from synapse.replication.tcp.streams import ToDeviceStream
 from synapse.storage.data_stores.main.deviceinbox import DeviceInboxWorkerStore
 from synapse.storage.database import Database
 from synapse.util.caches.expiringcache import ExpiringCache
@@ -44,7 +45,7 @@ class SlavedDeviceInboxStore(DeviceInboxWorkerStore, BaseSlavedStore):
        )

    def process_replication_rows(self, stream_name, instance_name, token, rows):
-        if stream_name == "to_device":
+        if stream_name == ToDeviceStream.NAME:
            self._device_inbox_id_gen.advance(token)
            for row in rows:
                if row.entity.startswith("@"):
--- a/synapse/replication/slave/storage/groups.py
+++ b/synapse/replication/slave/storage/groups.py
@@ -15,6 +15,7 @@

 from synapse.replication.slave.storage._base import BaseSlavedStore
 from synapse.replication.slave.storage._slaved_id_tracker import SlavedIdTracker
+from synapse.replication.tcp.streams import GroupServerStream
 from synapse.storage.data_stores.main.group_server import GroupServerWorkerStore
 from synapse.storage.database import Database
 from synapse.util.caches.stream_change_cache import StreamChangeCache
@@ -38,7 +39,7 @@ class SlavedGroupServerStore(GroupServerWorkerStore, BaseSlavedStore):
        return self._group_updates_id_gen.get_current_token()

    def process_replication_rows(self, stream_name, instance_name, token, rows):
-        if stream_name == "groups":
+        if stream_name == GroupServerStream.NAME:
            self._group_updates_id_gen.advance(token)
            for row in rows:
                self._group_updates_stream_cache.entity_has_changed(row.user_id, token)
--- a/synapse/replication/slave/storage/presence.py
+++ b/synapse/replication/slave/storage/presence.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

+from synapse.replication.tcp.streams import PresenceStream
 from synapse.storage import DataStore
 from synapse.storage.data_stores.main.presence import PresenceStore
 from synapse.storage.database import Database
@@ -42,7 +43,7 @@ class SlavedPresenceStore(BaseSlavedStore):
        return self._presence_id_gen.get_current_token()

    def process_replication_rows(self, stream_name, instance_name, token, rows):
-        if stream_name == "presence":
+        if stream_name == PresenceStream.NAME:
            self._presence_id_gen.advance(token)
            for row in rows:
                self.presence_stream_cache.entity_has_changed(row.user_id, token)
--- a/synapse/replication/slave/storage/push_rule.py
+++ b/synapse/replication/slave/storage/push_rule.py
@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

+from synapse.replication.tcp.streams import PushRulesStream
 from synapse.storage.data_stores.main.push_rule import PushRulesWorkerStore

 from .events import SlavedEventStore
@@ -30,7 +31,7 @@ class SlavedPushRuleStore(SlavedEventStore, PushRulesWorkerStore):
        return self._push_rules_stream_id_gen.get_current_token()

    def process_replication_rows(self, stream_name, instance_name, token, rows):
-        if stream_name == "push_rules":
+        if stream_name == PushRulesStream.NAME:
            self._push_rules_stream_id_gen.advance(token)
            for row in rows:
                self.get_push_rules_for_user.invalidate((row.user_id,))
--- a/synapse/replication/slave/storage/pushers.py
+++ b/synapse/replication/slave/storage/pushers.py
@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

+from synapse.replication.tcp.streams import PushersStream
 from synapse.storage.data_stores.main.pusher import PusherWorkerStore
 from synapse.storage.database import Database

@@ -32,6 +33,6 @@ class SlavedPusherStore(PusherWorkerStore, BaseSlavedStore):
        return self._pushers_id_gen.get_current_token()

    def process_replication_rows(self, stream_name, instance_name, token, rows):
-        if stream_name == "pushers":
+        if stream_name == PushersStream.NAME:
            self._pushers_id_gen.advance(token)
        return super().process_replication_rows(stream_name, instance_name, token, rows)
--- a/synapse/replication/slave/storage/receipts.py
+++ b/synapse/replication/slave/storage/receipts.py
@@ -14,20 +14,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

+from synapse.replication.tcp.streams import ReceiptsStream
 from synapse.storage.data_stores.main.receipts import ReceiptsWorkerStore
 from synapse.storage.database import Database

 from ._base import BaseSlavedStore
 from ._slaved_id_tracker import SlavedIdTracker

-# So, um, we want to borrow a load of functions intended for reading from
-# a DataStore, but we don't want to take functions that either write to the
-# DataStore or are cached and don't have cache invalidation logic.
-#
-# Rather than write duplicate versions of those functions, or lift them to
-# a common base class, we going to grab the underlying __func__ object from
-# the method descriptor on the DataStore and chuck them into our class.
-

 class SlavedReceiptsStore(ReceiptsWorkerStore, BaseSlavedStore):
    def __init__(self, database: Database, db_conn, hs):
@@ -52,7 +45,7 @@ class SlavedReceiptsStore(ReceiptsWorkerStore, BaseSlavedStore):
        self.get_receipts_for_room.invalidate((room_id, receipt_type))

    def process_replication_rows(self, stream_name, instance_name, token, rows):
-        if stream_name == "receipts":
+        if stream_name == ReceiptsStream.NAME:
            self._receipts_id_gen.advance(token)
            for row in rows:
                self.invalidate_caches_for_receipt(
--- a/synapse/replication/slave/storage/room.py
+++ b/synapse/replication/slave/storage/room.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

+from synapse.replication.tcp.streams import PublicRoomsStream
 from synapse.storage.data_stores.main.room import RoomWorkerStore
 from synapse.storage.database import Database

@@ -31,7 +32,7 @@ class RoomStore(RoomWorkerStore, BaseSlavedStore):
        return self._public_room_id_gen.get_current_token()

    def process_replication_rows(self, stream_name, instance_name, token, rows):
-        if stream_name == "public_rooms":
+        if stream_name == PublicRoomsStream.NAME:
            self._public_room_id_gen.advance(token)

        return super().process_replication_rows(stream_name, instance_name, token, rows)
--- a/synapse/rest/client/v1/login.py
+++ b/synapse/rest/client/v1/login.py
@@ -28,6 +28,7 @@ from synapse.rest.client.v2_alpha._base import client_patterns
 from synapse.rest.well_known import WellKnownBuilder
 from synapse.types import UserID
 from synapse.util.msisdn import phone_number_to_msisdn
+from synapse.util.threepids import canonicalise_email

 logger = logging.getLogger(__name__)

@@ -206,11 +207,14 @@ class LoginRestServlet(RestServlet):
            if medium is None or address is None:
                raise SynapseError(400, "Invalid thirdparty identifier")

+            # For emails, canonicalise the address.
+            # We store all email addresses canonicalised in the DB.
+            # (See add_threepid in synapse/handlers/auth.py)
            if medium == "email":
-                # For emails, transform the address to lowercase.
-                # We store all email addreses as lowercase in the DB.
-                # (See add_threepid in synapse/handlers/auth.py)
-                address = address.lower()
+                try:
+                    address = canonicalise_email(address)
+                except ValueError as e:
+                    raise SynapseError(400, str(e))

            # We also apply account rate limiting using the 3PID as a key, as
            # otherwise using 3PID bypasses the ratelimiting based on user ID.
--- a/synapse/rest/client/v2_alpha/account.py
+++ b/synapse/rest/client/v2_alpha/account.py
@@ -35,7 +35,7 @@ from synapse.push.mailer import Mailer, load_jinja2_templates
 from synapse.types import UserID
 from synapse.util.msisdn import phone_number_to_msisdn
 from synapse.util.stringutils import assert_valid_client_secret, random_string
-from synapse.util.threepids import check_3pid_allowed
+from synapse.util.threepids import canonicalise_email, check_3pid_allowed

 from ._base import client_patterns, interactive_auth_handler

@@ -88,7 +88,15 @@ class EmailPasswordRequestTokenRestServlet(RestServlet):
        client_secret = body["client_secret"]
        assert_valid_client_secret(client_secret)

-        email = body["email"]
+        # Canonicalise the email address. The addresses are all stored canonicalised
+        # in the database. This allows the user to reset his password without having to
+        # know the exact spelling (eg. upper and lower case) of address in the database.
+        # Stored in the database "foo@bar.com"
+        # User requests with "FOO@bar.com" would raise a Not Found error
+        try:
+            email = canonicalise_email(body["email"])
+        except ValueError as e:
+            raise SynapseError(400, str(e))
        send_attempt = body["send_attempt"]
        next_link = body.get("next_link")  # Optional param

@@ -99,6 +107,10 @@ class EmailPasswordRequestTokenRestServlet(RestServlet):
                Codes.THREEPID_DENIED,
            )

+        # The email will be sent to the stored address.
+        # This avoids a potential account hijack by requesting a password reset to
+        # an email address which is controlled by the attacker but which, after
+        # canonicalisation, matches the one in our database.
        existing_user_id = await self.hs.get_datastore().get_user_id_by_threepid(
            "email", email
        )
@@ -284,10 +296,13 @@ class PasswordRestServlet(RestServlet):
                if "medium" not in threepid or "address" not in threepid:
                    raise SynapseError(500, "Malformed threepid")
                if threepid["medium"] == "email":
-                    # For emails, transform the address to lowercase.
-                    # We store all email addreses as lowercase in the DB.
+                    # For emails, canonicalise the address.
+                    # We store all email addresses canonicalised in the DB.
                    # (See add_threepid in synapse/handlers/auth.py)
-                    threepid["address"] = threepid["address"].lower()
+                    try:
+                        threepid["address"] = canonicalise_email(threepid["address"])
+                    except ValueError as e:
+                        raise SynapseError(400, str(e))
                # if using email, we must know about the email they're authing with!
                threepid_user_id = await self.datastore.get_user_id_by_threepid(
                    threepid["medium"], threepid["address"]
@@ -420,7 +435,16 @@ class EmailThreepidRequestTokenRestServlet(RestServlet):
        client_secret = body["client_secret"]
        assert_valid_client_secret(client_secret)

-        email = body["email"]
+        # Canonicalise the email address. The addresses are all stored canonicalised
+        # in the database.
+        # This ensures that the validation email is sent to the canonicalised address
+        # as it will later be entered into the database.
+        # Otherwise the email will be sent to "FOO@bar.com" and stored as
+        # "foo@bar.com" in database.
+        try:
+            email = canonicalise_email(body["email"])
+        except ValueError as e:
+            raise SynapseError(400, str(e))
        send_attempt = body["send_attempt"]
        next_link = body.get("next_link")  # Optional param

@@ -431,11 +455,7 @@ class EmailThreepidRequestTokenRestServlet(RestServlet):
                Codes.THREEPID_DENIED,
            )

-        assert_valid_client_secret(body["client_secret"])
-
-        existing_user_id = await self.store.get_user_id_by_threepid(
-            "email", body["email"]
-        )
+        existing_user_id = await self.store.get_user_id_by_threepid("email", email)

        if existing_user_id is not None:
            if self.config.request_token_inhibit_3pid_errors:
--- a/synapse/rest/client/v2_alpha/register.py
+++ b/synapse/rest/client/v2_alpha/register.py
@@ -49,7 +49,7 @@ from synapse.push.mailer import load_jinja2_templates
 from synapse.util.msisdn import phone_number_to_msisdn
 from synapse.util.ratelimitutils import FederationRateLimiter
 from synapse.util.stringutils import assert_valid_client_secret, random_string
-from synapse.util.threepids import check_3pid_allowed
+from synapse.util.threepids import canonicalise_email, check_3pid_allowed

 from ._base import client_patterns, interactive_auth_handler

@@ -118,7 +118,14 @@ class EmailRegisterRequestTokenRestServlet(RestServlet):
        client_secret = body["client_secret"]
        assert_valid_client_secret(client_secret)

-        email = body["email"]
+        # For emails, canonicalise the address.
+        # We store all email addresses canonicalised in the DB.
+        # (See on_POST in EmailThreepidRequestTokenRestServlet
+        # in synapse/rest/client/v2_alpha/account.py)
+        try:
+            email = canonicalise_email(body["email"])
+        except ValueError as e:
+            raise SynapseError(400, str(e))
        send_attempt = body["send_attempt"]
        next_link = body.get("next_link")  # Optional param

@@ -130,7 +137,7 @@ class EmailRegisterRequestTokenRestServlet(RestServlet):
            )

        existing_user_id = await self.hs.get_datastore().get_user_id_by_threepid(
-            "email", body["email"]
+            "email", email
        )

        if existing_user_id is not None:
@@ -625,6 +632,15 @@ class RegisterRestServlet(RestServlet):
                    if login_type in auth_result:
                        medium = auth_result[login_type]["medium"]
                        address = auth_result[login_type]["address"]
+                        # For emails, canonicalise the address.
+                        # We store all email addresses canonicalised in the DB.
+                        # (See on_POST in EmailThreepidRequestTokenRestServlet
+                        # in synapse/rest/client/v2_alpha/account.py)
+                        if medium == "email":
+                            try:
+                                address = canonicalise_email(address)
+                            except ValueError as e:
+                                raise SynapseError(400, str(e))

                        existing_user_id = await self.store.get_user_id_by_threepid(
                            medium, address
--- a/synapse/rest/consent/consent_resource.py
+++ b/synapse/rest/consent/consent_resource.py
@@ -26,11 +26,7 @@ from twisted.internet import defer

 from synapse.api.errors import NotFoundError, StoreError, SynapseError
 from synapse.config import ConfigError
-from synapse.http.server import (
-    DirectServeResource,
-    respond_with_html,
-    wrap_html_request_handler,
-)
+from synapse.http.server import DirectServeHtmlResource, respond_with_html
 from synapse.http.servlet import parse_string
 from synapse.types import UserID

@@ -48,7 +44,7 @@ else:
        return a == b


-class ConsentResource(DirectServeResource):
+class ConsentResource(DirectServeHtmlResource):
    """A twisted Resource to display a privacy policy and gather consent to it

    When accessed via GET, returns the privacy policy via a template.
@@ -119,7 +115,6 @@ class ConsentResource(DirectServeResource):

        self._hmac_secret = hs.config.form_secret.encode("utf-8")

-    @wrap_html_request_handler
    async def _async_render_GET(self, request):
        """
        Args:
@@ -160,7 +155,6 @@ class ConsentResource(DirectServeResource):
        except TemplateNotFound:
            raise NotFoundError("Unknown policy version")

-    @wrap_html_request_handler
    async def _async_render_POST(self, request):
        """
        Args:
--- a/synapse/rest/key/v2/remote_key_resource.py
+++ b/synapse/rest/key/v2/remote_key_resource.py
@@ -20,17 +20,13 @@ from signedjson.sign import sign_json

 from synapse.api.errors import Codes, SynapseError
 from synapse.crypto.keyring import ServerKeyFetcher
-from synapse.http.server import (
-    DirectServeResource,
-    respond_with_json_bytes,
-    wrap_json_request_handler,
-)
+from synapse.http.server import DirectServeJsonResource, respond_with_json_bytes
 from synapse.http.servlet import parse_integer, parse_json_object_from_request

 logger = logging.getLogger(__name__)


-class RemoteKey(DirectServeResource):
+class RemoteKey(DirectServeJsonResource):
    """HTTP resource for retreiving the TLS certificate and NACL signature
    verification keys for a collection of servers. Checks that the reported
    X.509 TLS certificate matches the one used in the HTTPS connection. Checks
@@ -92,13 +88,14 @@ class RemoteKey(DirectServeResource):
    isLeaf = True

    def __init__(self, hs):
+        super().__init__()
+
        self.fetcher = ServerKeyFetcher(hs)
        self.store = hs.get_datastore()
        self.clock = hs.get_clock()
        self.federation_domain_whitelist = hs.config.federation_domain_whitelist
        self.config = hs.config

-    @wrap_json_request_handler
    async def _async_render_GET(self, request):
        if len(request.postpath) == 1:
            (server,) = request.postpath
@@ -115,7 +112,6 @@ class RemoteKey(DirectServeResource):

        await self.query_keys(request, query, query_remote_on_cache_miss=True)

-    @wrap_json_request_handler
    async def _async_render_POST(self, request):
        content = parse_json_object_from_request(request)

--- a/synapse/rest/media/v1/config_resource.py
+++ b/synapse/rest/media/v1/config_resource.py
@@ -14,16 +14,10 @@
 # limitations under the License.
 #

-from twisted.web.server import NOT_DONE_YET
-
-from synapse.http.server import (
-    DirectServeResource,
-    respond_with_json,
-    wrap_json_request_handler,
-)
+from synapse.http.server import DirectServeJsonResource, respond_with_json


-class MediaConfigResource(DirectServeResource):
+class MediaConfigResource(DirectServeJsonResource):
    isLeaf = True

    def __init__(self, hs):
@@ -33,11 +27,9 @@ class MediaConfigResource(DirectServeResource):
        self.auth = hs.get_auth()
        self.limits_dict = {"m.upload.size": config.max_upload_size}

-    @wrap_json_request_handler
    async def _async_render_GET(self, request):
        await self.auth.get_user_by_req(request)
        respond_with_json(request, 200, self.limits_dict, send_cors=True)

-    def render_OPTIONS(self, request):
+    async def _async_render_OPTIONS(self, request):
        respond_with_json(request, 200, {}, send_cors=True)
-        return NOT_DONE_YET
--- a/synapse/rest/media/v1/download_resource.py
+++ b/synapse/rest/media/v1/download_resource.py
@@ -15,18 +15,14 @@
 import logging

 import synapse.http.servlet
-from synapse.http.server import (
-    DirectServeResource,
-    set_cors_headers,
-    wrap_json_request_handler,
-)
+from synapse.http.server import DirectServeJsonResource, set_cors_headers

 from ._base import parse_media_id, respond_404

 logger = logging.getLogger(__name__)


-class DownloadResource(DirectServeResource):
+class DownloadResource(DirectServeJsonResource):
    isLeaf = True

    def __init__(self, hs, media_repo):
@@ -34,10 +30,6 @@ class DownloadResource(DirectServeResource):
        self.media_repo = media_repo
        self.server_name = hs.hostname

-        # this is expected by @wrap_json_request_handler
-        self.clock = hs.get_clock()
-
-    @wrap_json_request_handler
    async def _async_render_GET(self, request):
        set_cors_headers(request)
        request.setHeader(
--- a/synapse/rest/media/v1/preview_url_resource.py
+++ b/synapse/rest/media/v1/preview_url_resource.py
@@ -34,10 +34,9 @@ from twisted.internet.error import DNSLookupError
 from synapse.api.errors import Codes, SynapseError
 from synapse.http.client import SimpleHttpClient
 from synapse.http.server import (
-    DirectServeResource,
+    DirectServeJsonResource,
    respond_with_json,
    respond_with_json_bytes,
-    wrap_json_request_handler,
 )
 from synapse.http.servlet import parse_integer, parse_string
 from synapse.logging.context import make_deferred_yieldable, run_in_background
@@ -58,7 +57,7 @@ OG_TAG_NAME_MAXLEN = 50
 OG_TAG_VALUE_MAXLEN = 1000


-class PreviewUrlResource(DirectServeResource):
+class PreviewUrlResource(DirectServeJsonResource):
    isLeaf = True

    def __init__(self, hs, media_repo, media_storage):
@@ -108,11 +107,10 @@ class PreviewUrlResource(DirectServeResource):
                self._start_expire_url_cache_data, 10 * 1000
            )

-    def render_OPTIONS(self, request):
+    async def _async_render_OPTIONS(self, request):
        request.setHeader(b"Allow", b"OPTIONS, GET")
-        return respond_with_json(request, 200, {}, send_cors=True)
+        respond_with_json(request, 200, {}, send_cors=True)

-    @wrap_json_request_handler
    async def _async_render_GET(self, request):

        # XXX: if get_user_by_req fails, what should we do in an async render?
--- a/synapse/rest/media/v1/thumbnail_resource.py
+++ b/synapse/rest/media/v1/thumbnail_resource.py
@@ -16,11 +16,7 @@

 import logging

-from synapse.http.server import (
-    DirectServeResource,
-    set_cors_headers,
-    wrap_json_request_handler,
-)
+from synapse.http.server import DirectServeJsonResource, set_cors_headers
 from synapse.http.servlet import parse_integer, parse_string

 from ._base import (
@@ -34,7 +30,7 @@ from ._base import (
 logger = logging.getLogger(__name__)


-class ThumbnailResource(DirectServeResource):
+class ThumbnailResource(DirectServeJsonResource):
    isLeaf = True

    def __init__(self, hs, media_repo, media_storage):
@@ -45,9 +41,7 @@ class ThumbnailResource(DirectServeResource):
        self.media_storage = media_storage
        self.dynamic_thumbnails = hs.config.dynamic_thumbnails
        self.server_name = hs.hostname
-        self.clock = hs.get_clock()

-    @wrap_json_request_handler
    async def _async_render_GET(self, request):
        set_cors_headers(request)
        server_name, media_id, _ = parse_media_id(request)
--- a/synapse/rest/media/v1/upload_resource.py
+++ b/synapse/rest/media/v1/upload_resource.py
@@ -15,20 +15,14 @@

 import logging

-from twisted.web.server import NOT_DONE_YET
-
 from synapse.api.errors import Codes, SynapseError
-from synapse.http.server import (
-    DirectServeResource,
-    respond_with_json,
-    wrap_json_request_handler,
-)
+from synapse.http.server import DirectServeJsonResource, respond_with_json
 from synapse.http.servlet import parse_string

 logger = logging.getLogger(__name__)


-class UploadResource(DirectServeResource):
+class UploadResource(DirectServeJsonResource):
    isLeaf = True

    def __init__(self, hs, media_repo):
@@ -43,11 +37,9 @@ class UploadResource(DirectServeResource):
        self.max_upload_size = hs.config.max_upload_size
        self.clock = hs.get_clock()

-    def render_OPTIONS(self, request):
+    async def _async_render_OPTIONS(self, request):
        respond_with_json(request, 200, {}, send_cors=True)
-        return NOT_DONE_YET

-    @wrap_json_request_handler
    async def _async_render_POST(self, request):
        requester = await self.auth.get_user_by_req(request)
        # TODO: The checks here are a bit late. The content will have
--- a/synapse/rest/oidc/callback_resource.py
+++ b/synapse/rest/oidc/callback_resource.py
@@ -14,18 +14,17 @@
 # limitations under the License.
 import logging

-from synapse.http.server import DirectServeResource, wrap_html_request_handler
+from synapse.http.server import DirectServeHtmlResource

 logger = logging.getLogger(__name__)


-class OIDCCallbackResource(DirectServeResource):
+class OIDCCallbackResource(DirectServeHtmlResource):
    isLeaf = 1

    def __init__(self, hs):
        super().__init__()
        self._oidc_handler = hs.get_oidc_handler()

-    @wrap_html_request_handler
    async def _async_render_GET(self, request):
-        return await self._oidc_handler.handle_oidc_callback(request)
+        await self._oidc_handler.handle_oidc_callback(request)
--- a/synapse/rest/saml2/response_resource.py
+++ b/synapse/rest/saml2/response_resource.py
@@ -16,10 +16,10 @@
 from twisted.python import failure

 from synapse.api.errors import SynapseError
-from synapse.http.server import DirectServeResource, return_html_error
+from synapse.http.server import DirectServeHtmlResource, return_html_error


-class SAML2ResponseResource(DirectServeResource):
+class SAML2ResponseResource(DirectServeHtmlResource):
    """A Twisted web resource which handles the SAML response"""

    isLeaf = 1
--- a/synapse/storage/data_stores/main/cache.py
+++ b/synapse/storage/data_stores/main/cache.py
@@ -19,7 +19,9 @@ import logging
 from typing import Any, Iterable, Optional, Tuple

 from synapse.api.constants import EventTypes
+from synapse.replication.tcp.streams import BackfillStream, CachesStream
 from synapse.replication.tcp.streams.events import (
+    EventsStream,
    EventsStreamCurrentStateRow,
    EventsStreamEventRow,
 )
@@ -71,10 +73,10 @@ class CacheInvalidationWorkerStore(SQLBaseStore):
        )

    def process_replication_rows(self, stream_name, instance_name, token, rows):
-        if stream_name == "events":
+        if stream_name == EventsStream.NAME:
            for row in rows:
                self._process_event_stream_row(token, row)
-        elif stream_name == "backfill":
+        elif stream_name == BackfillStream.NAME:
            for row in rows:
                self._invalidate_caches_for_event(
                    -token,
@@ -86,7 +88,7 @@ class CacheInvalidationWorkerStore(SQLBaseStore):
                    row.relates_to,
                    backfilled=True,
                )
-        elif stream_name == "caches":
+        elif stream_name == CachesStream.NAME:
            if self._cache_id_gen:
                self._cache_id_gen.advance(instance_name, token)

--- a/synapse/storage/data_stores/main/events_worker.py
+++ b/synapse/storage/data_stores/main/events_worker.py
@@ -38,6 +38,8 @@ from synapse.events.utils import prune_event
 from synapse.logging.context import PreserveLoggingContext, current_context
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.replication.slave.storage._slaved_id_tracker import SlavedIdTracker
+from synapse.replication.tcp.streams import BackfillStream
+from synapse.replication.tcp.streams.events import EventsStream
 from synapse.storage._base import SQLBaseStore, make_in_list_sql_clause
 from synapse.storage.database import Database
 from synapse.storage.util.id_generators import StreamIdGenerator
@@ -113,9 +115,9 @@ class EventsWorkerStore(SQLBaseStore):
        self._event_fetch_ongoing = 0

    def process_replication_rows(self, stream_name, instance_name, token, rows):
-        if stream_name == "events":
+        if stream_name == EventsStream.NAME:
            self._stream_id_gen.advance(token)
-        elif stream_name == "backfill":
+        elif stream_name == BackfillStream.NAME:
            self._backfill_id_gen.advance(-token)

        super().process_replication_rows(stream_name, instance_name, token, rows)
--- a/synapse/storage/engines/postgres.py
+++ b/synapse/storage/engines/postgres.py
@@ -92,7 +92,7 @@ class PostgresEngine(BaseDatabaseEngine):
            errors.append("    - 'COLLATE' is set to %r. Should be 'C'" % (collation,))

        if ctype != "C":
-            errors.append("    - 'CTYPE' is set to %r. Should be 'C'" % (collation,))
+            errors.append("    - 'CTYPE' is set to %r. Should be 'C'" % (ctype,))

        if errors:
            raise IncorrectDatabaseSetup(
--- a/synapse/util/threepids.py
+++ b/synapse/util/threepids.py
@@ -78,3 +78,26 @@ def check_3pid_allowed(hs, medium, address):
        defer.returnValue(True)

    defer.returnValue(False)
+
+
+def canonicalise_email(address: str) -> str:
+    """'Canonicalise' email address
+    Case folding of local part of email address and lowercase domain part
+    See MSC2265, https://github.com/matrix-org/matrix-doc/pull/2265
+
+    Args:
+        address: email address to be canonicalised
+    Returns:
+        The canonical form of the email address
+    Raises:
+        ValueError if the address could not be parsed.
+    """
+
+    address = address.strip()
+
+    parts = address.split("@")
+    if len(parts) != 2:
+        logger.debug("Couldn't parse email address %s", address)
+        raise ValueError("Unable to parse email address")
+
+    return parts[0].casefold() + "@" + parts[1].lower()
--- a/3
+++ b/3
@@ -239,7 +239,8 @@ def main():
    for config_file in config_files:
        with open(config_file) as file_stream:
            yaml_config = yaml.safe_load(file_stream)
-        config.update(yaml_config)
+        if yaml_config is not None:
+            config.update(yaml_config)

    pidfile = config["pid_file"]
    cache_factor = config.get("synctl_cache_factor")
--- a/tests/http/test_additional_resource.py
+++ b/tests/http/test_additional_resource.py
@@ -0,0 +1,62 @@
+# -*- coding: utf-8 -*-
+# Copyright 2018 New Vector 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.http.additional_resource import AdditionalResource
+from synapse.http.server import respond_with_json
+
+from tests.unittest import HomeserverTestCase
+
+
+class _AsyncTestCustomEndpoint:
+    def __init__(self, config, module_api):
+        pass
+
+    async def handle_request(self, request):
+        respond_with_json(request, 200, {"some_key": "some_value_async"})
+
+
+class _SyncTestCustomEndpoint:
+    def __init__(self, config, module_api):
+        pass
+
+    async def handle_request(self, request):
+        respond_with_json(request, 200, {"some_key": "some_value_sync"})
+
+
+class AdditionalResourceTests(HomeserverTestCase):
+    """Very basic tests that `AdditionalResource` works correctly with sync
+    and async handlers.
+    """
+
+    def test_async(self):
+        handler = _AsyncTestCustomEndpoint({}, None).handle_request
+        self.resource = AdditionalResource(self.hs, handler)
+
+        request, channel = self.make_request("GET", "/")
+        self.render(request)
+
+        self.assertEqual(request.code, 200)
+        self.assertEqual(channel.json_body, {"some_key": "some_value_async"})
+
+    def test_sync(self):
+        handler = _SyncTestCustomEndpoint({}, None).handle_request
+        self.resource = AdditionalResource(self.hs, handler)
+
+        request, channel = self.make_request("GET", "/")
+        self.render(request)
+
+        self.assertEqual(request.code, 200)
+        self.assertEqual(channel.json_body, {"some_key": "some_value_sync"})
--- a/tests/push/test_http.py
+++ b/tests/push/test_http.py
@@ -25,7 +25,6 @@ from tests.unittest import HomeserverTestCase


 class HTTPPusherTests(HomeserverTestCase):
-
    servlets = [
        synapse.rest.admin.register_servlets_for_client_rest_resource,
        room.register_servlets,
@@ -35,7 +34,6 @@ class HTTPPusherTests(HomeserverTestCase):
    hijack_auth = False

    def make_homeserver(self, reactor, clock):
-
        self.push_attempts = []

        m = Mock()
@@ -90,9 +88,6 @@ class HTTPPusherTests(HomeserverTestCase):
        # Create a room
        room = self.helper.create_room_as(user_id, tok=access_token)

-        # Invite the other person
-        self.helper.invite(room=room, src=user_id, tok=access_token, targ=other_user_id)
-
        # The other user joins
        self.helper.join(room=room, user=other_user_id, tok=other_access_token)

@@ -157,3 +152,350 @@ class HTTPPusherTests(HomeserverTestCase):
        pushers = list(pushers)
        self.assertEqual(len(pushers), 1)
        self.assertTrue(pushers[0]["last_stream_ordering"] > last_stream_ordering)
+
+    def test_sends_high_priority_for_encrypted(self):
+        """
+        The HTTP pusher will send pushes at high priority if they correspond
+        to an encrypted message.
+        This will happen both in 1:1 rooms and larger rooms.
+        """
+        # Register the user who gets notified
+        user_id = self.register_user("user", "pass")
+        access_token = self.login("user", "pass")
+
+        # Register the user who sends the message
+        other_user_id = self.register_user("otheruser", "pass")
+        other_access_token = self.login("otheruser", "pass")
+
+        # Register a third user
+        yet_another_user_id = self.register_user("yetanotheruser", "pass")
+        yet_another_access_token = self.login("yetanotheruser", "pass")
+
+        # Create a room
+        room = self.helper.create_room_as(user_id, tok=access_token)
+
+        # The other user joins
+        self.helper.join(room=room, user=other_user_id, tok=other_access_token)
+
+        # Register the pusher
+        user_tuple = self.get_success(
+            self.hs.get_datastore().get_user_by_access_token(access_token)
+        )
+        token_id = user_tuple["token_id"]
+
+        self.get_success(
+            self.hs.get_pusherpool().add_pusher(
+                user_id=user_id,
+                access_token=token_id,
+                kind="http",
+                app_id="m.http",
+                app_display_name="HTTP Push Notifications",
+                device_display_name="pushy push",
+                pushkey="a@example.com",
+                lang=None,
+                data={"url": "example.com"},
+            )
+        )
+
+        # Send an encrypted event
+        # I know there'd normally be set-up of an encrypted room first
+        # but this will do for our purposes
+        self.helper.send_event(
+            room,
+            "m.room.encrypted",
+            content={
+                "algorithm": "m.megolm.v1.aes-sha2",
+                "sender_key": "6lImKbzK51MzWLwHh8tUM3UBBSBrLlgup/OOCGTvumM",
+                "ciphertext": "AwgAErABoRxwpMipdgiwXgu46rHiWQ0DmRj0qUlPrMraBUDk"
+                "leTnJRljpuc7IOhsYbLY3uo2WI0ab/ob41sV+3JEIhODJPqH"
+                "TK7cEZaIL+/up9e+dT9VGF5kRTWinzjkeqO8FU5kfdRjm+3w"
+                "0sy3o1OCpXXCfO+faPhbV/0HuK4ndx1G+myNfK1Nk/CxfMcT"
+                "BT+zDS/Df/QePAHVbrr9uuGB7fW8ogW/ulnydgZPRluusFGv"
+                "J3+cg9LoPpZPAmv5Me3ec7NtdlfN0oDZ0gk3TiNkkhsxDG9Y"
+                "YcNzl78USI0q8+kOV26Bu5dOBpU4WOuojXZHJlP5lMgdzLLl"
+                "EQ0",
+                "session_id": "IigqfNWLL+ez/Is+Duwp2s4HuCZhFG9b9CZKTYHtQ4A",
+                "device_id": "AHQDUSTAAA",
+            },
+            tok=other_access_token,
+        )
+
+        # Advance time a bit, so the pusher will register something has happened
+        self.pump()
+
+        # Make the push succeed
+        self.push_attempts[0][0].callback({})
+        self.pump()
+
+        # Check our push made it with high priority
+        self.assertEqual(len(self.push_attempts), 1)
+        self.assertEqual(self.push_attempts[0][1], "example.com")
+        self.assertEqual(self.push_attempts[0][2]["notification"]["prio"], "high")
+
+        # Add yet another person — we want to make this room not a 1:1
+        # (as encrypted messages in a 1:1 currently have tweaks applied
+        #  so it doesn't properly exercise the condition of all encrypted
+        #  messages need to be high).
+        self.helper.join(
+            room=room, user=yet_another_user_id, tok=yet_another_access_token
+        )
+
+        # Check no push notifications are sent regarding the membership changes
+        # (that would confuse the test)
+        self.pump()
+        self.assertEqual(len(self.push_attempts), 1)
+
+        # Send another encrypted event
+        self.helper.send_event(
+            room,
+            "m.room.encrypted",
+            content={
+                "ciphertext": "AwgAEoABtEuic/2DF6oIpNH+q/PonzlhXOVho8dTv0tzFr5m"
+                "9vTo50yabx3nxsRlP2WxSqa8I07YftP+EKWCWJvTkg6o7zXq"
+                "6CK+GVvLQOVgK50SfvjHqJXN+z1VEqj+5mkZVN/cAgJzoxcH"
+                "zFHkwDPJC8kQs47IHd8EO9KBUK4v6+NQ1uE/BIak4qAf9aS/"
+                "kI+f0gjn9IY9K6LXlah82A/iRyrIrxkCkE/n0VfvLhaWFecC"
+                "sAWTcMLoF6fh1Jpke95mljbmFSpsSd/eEQw",
+                "device_id": "SRCFTWTHXO",
+                "session_id": "eMA+bhGczuTz1C5cJR1YbmrnnC6Goni4lbvS5vJ1nG4",
+                "algorithm": "m.megolm.v1.aes-sha2",
+                "sender_key": "rC/XSIAiYrVGSuaHMop8/pTZbku4sQKBZwRwukgnN1c",
+            },
+            tok=other_access_token,
+        )
+
+        # Advance time a bit, so the pusher will register something has happened
+        self.pump()
+        self.assertEqual(len(self.push_attempts), 2)
+        self.assertEqual(self.push_attempts[1][1], "example.com")
+        self.assertEqual(self.push_attempts[1][2]["notification"]["prio"], "high")
+
+    def test_sends_high_priority_for_one_to_one_only(self):
+        """
+        The HTTP pusher will send pushes at high priority if they correspond
+        to a message in a one-to-one room.
+        """
+        # Register the user who gets notified
+        user_id = self.register_user("user", "pass")
+        access_token = self.login("user", "pass")
+
+        # Register the user who sends the message
+        other_user_id = self.register_user("otheruser", "pass")
+        other_access_token = self.login("otheruser", "pass")
+
+        # Register a third user
+        yet_another_user_id = self.register_user("yetanotheruser", "pass")
+        yet_another_access_token = self.login("yetanotheruser", "pass")
+
+        # Create a room
+        room = self.helper.create_room_as(user_id, tok=access_token)
+
+        # The other user joins
+        self.helper.join(room=room, user=other_user_id, tok=other_access_token)
+
+        # Register the pusher
+        user_tuple = self.get_success(
+            self.hs.get_datastore().get_user_by_access_token(access_token)
+        )
+        token_id = user_tuple["token_id"]
+
+        self.get_success(
+            self.hs.get_pusherpool().add_pusher(
+                user_id=user_id,
+                access_token=token_id,
+                kind="http",
+                app_id="m.http",
+                app_display_name="HTTP Push Notifications",
+                device_display_name="pushy push",
+                pushkey="a@example.com",
+                lang=None,
+                data={"url": "example.com"},
+            )
+        )
+
+        # Send a message
+        self.helper.send(room, body="Hi!", tok=other_access_token)
+
+        # Advance time a bit, so the pusher will register something has happened
+        self.pump()
+
+        # Make the push succeed
+        self.push_attempts[0][0].callback({})
+        self.pump()
+
+        # Check our push made it with high priority — this is a one-to-one room
+        self.assertEqual(len(self.push_attempts), 1)
+        self.assertEqual(self.push_attempts[0][1], "example.com")
+        self.assertEqual(self.push_attempts[0][2]["notification"]["prio"], "high")
+
+        # Yet another user joins
+        self.helper.join(
+            room=room, user=yet_another_user_id, tok=yet_another_access_token
+        )
+
+        # Check no push notifications are sent regarding the membership changes
+        # (that would confuse the test)
+        self.pump()
+        self.assertEqual(len(self.push_attempts), 1)
+
+        # Send another event
+        self.helper.send(room, body="Welcome!", tok=other_access_token)
+
+        # Advance time a bit, so the pusher will register something has happened
+        self.pump()
+        self.assertEqual(len(self.push_attempts), 2)
+        self.assertEqual(self.push_attempts[1][1], "example.com")
+
+        # check that this is low-priority
+        self.assertEqual(self.push_attempts[1][2]["notification"]["prio"], "low")
+
+    def test_sends_high_priority_for_mention(self):
+        """
+        The HTTP pusher will send pushes at high priority if they correspond
+        to a message containing the user's display name.
+        """
+        # Register the user who gets notified
+        user_id = self.register_user("user", "pass")
+        access_token = self.login("user", "pass")
+
+        # Register the user who sends the message
+        other_user_id = self.register_user("otheruser", "pass")
+        other_access_token = self.login("otheruser", "pass")
+
+        # Register a third user
+        yet_another_user_id = self.register_user("yetanotheruser", "pass")
+        yet_another_access_token = self.login("yetanotheruser", "pass")
+
+        # Create a room
+        room = self.helper.create_room_as(user_id, tok=access_token)
+
+        # The other users join
+        self.helper.join(room=room, user=other_user_id, tok=other_access_token)
+        self.helper.join(
+            room=room, user=yet_another_user_id, tok=yet_another_access_token
+        )
+
+        # Register the pusher
+        user_tuple = self.get_success(
+            self.hs.get_datastore().get_user_by_access_token(access_token)
+        )
+        token_id = user_tuple["token_id"]
+
+        self.get_success(
+            self.hs.get_pusherpool().add_pusher(
+                user_id=user_id,
+                access_token=token_id,
+                kind="http",
+                app_id="m.http",
+                app_display_name="HTTP Push Notifications",
+                device_display_name="pushy push",
+                pushkey="a@example.com",
+                lang=None,
+                data={"url": "example.com"},
+            )
+        )
+
+        # Send a message
+        self.helper.send(room, body="Oh, user, hello!", tok=other_access_token)
+
+        # Advance time a bit, so the pusher will register something has happened
+        self.pump()
+
+        # Make the push succeed
+        self.push_attempts[0][0].callback({})
+        self.pump()
+
+        # Check our push made it with high priority
+        self.assertEqual(len(self.push_attempts), 1)
+        self.assertEqual(self.push_attempts[0][1], "example.com")
+        self.assertEqual(self.push_attempts[0][2]["notification"]["prio"], "high")
+
+        # Send another event, this time with no mention
+        self.helper.send(room, body="Are you there?", tok=other_access_token)
+
+        # Advance time a bit, so the pusher will register something has happened
+        self.pump()
+        self.assertEqual(len(self.push_attempts), 2)
+        self.assertEqual(self.push_attempts[1][1], "example.com")
+
+        # check that this is low-priority
+        self.assertEqual(self.push_attempts[1][2]["notification"]["prio"], "low")
+
+    def test_sends_high_priority_for_atroom(self):
+        """
+        The HTTP pusher will send pushes at high priority if they correspond
+        to a message that contains @room.
+        """
+        # Register the user who gets notified
+        user_id = self.register_user("user", "pass")
+        access_token = self.login("user", "pass")
+
+        # Register the user who sends the message
+        other_user_id = self.register_user("otheruser", "pass")
+        other_access_token = self.login("otheruser", "pass")
+
+        # Register a third user
+        yet_another_user_id = self.register_user("yetanotheruser", "pass")
+        yet_another_access_token = self.login("yetanotheruser", "pass")
+
+        # Create a room (as other_user so the power levels are compatible with
+        # other_user sending @room).
+        room = self.helper.create_room_as(other_user_id, tok=other_access_token)
+
+        # The other users join
+        self.helper.join(room=room, user=user_id, tok=access_token)
+        self.helper.join(
+            room=room, user=yet_another_user_id, tok=yet_another_access_token
+        )
+
+        # Register the pusher
+        user_tuple = self.get_success(
+            self.hs.get_datastore().get_user_by_access_token(access_token)
+        )
+        token_id = user_tuple["token_id"]
+
+        self.get_success(
+            self.hs.get_pusherpool().add_pusher(
+                user_id=user_id,
+                access_token=token_id,
+                kind="http",
+                app_id="m.http",
+                app_display_name="HTTP Push Notifications",
+                device_display_name="pushy push",
+                pushkey="a@example.com",
+                lang=None,
+                data={"url": "example.com"},
+            )
+        )
+
+        # Send a message
+        self.helper.send(
+            room,
+            body="@room eeek! There's a spider on the table!",
+            tok=other_access_token,
+        )
+
+        # Advance time a bit, so the pusher will register something has happened
+        self.pump()
+
+        # Make the push succeed
+        self.push_attempts[0][0].callback({})
+        self.pump()
+
+        # Check our push made it with high priority
+        self.assertEqual(len(self.push_attempts), 1)
+        self.assertEqual(self.push_attempts[0][1], "example.com")
+        self.assertEqual(self.push_attempts[0][2]["notification"]["prio"], "high")
+
+        # Send another event, this time as someone without the power of @room
+        self.helper.send(
+            room, body="@room the spider is gone", tok=yet_another_access_token
+        )
+
+        # Advance time a bit, so the pusher will register something has happened
+        self.pump()
+        self.assertEqual(len(self.push_attempts), 2)
+        self.assertEqual(self.push_attempts[1][1], "example.com")
+
+        # check that this is low-priority
+        self.assertEqual(self.push_attempts[1][2]["notification"]["prio"], "low")
--- a/tests/rest/client/v2_alpha/test_account.py
+++ b/tests/rest/client/v2_alpha/test_account.py
@@ -108,6 +108,46 @@ class PasswordResetTestCase(unittest.HomeserverTestCase):
        # Assert we can't log in with the old password
        self.attempt_wrong_password_login("kermit", old_password)

+    def test_basic_password_reset_canonicalise_email(self):
+        """Test basic password reset flow
+        Request password reset with different spelling
+        """
+        old_password = "monkey"
+        new_password = "kangeroo"
+
+        user_id = self.register_user("kermit", old_password)
+        self.login("kermit", old_password)
+
+        email_profile = "test@example.com"
+        email_passwort_reset = "TEST@EXAMPLE.COM"
+
+        # Add a threepid
+        self.get_success(
+            self.store.user_add_threepid(
+                user_id=user_id,
+                medium="email",
+                address=email_profile,
+                validated_at=0,
+                added_at=0,
+            )
+        )
+
+        client_secret = "foobar"
+        session_id = self._request_token(email_passwort_reset, client_secret)
+
+        self.assertEquals(len(self.email_attempts), 1)
+        link = self._get_link_from_email()
+
+        self._validate_token(link)
+
+        self._reset_password(new_password, session_id, client_secret)
+
+        # Assert we can log in with the new password
+        self.login("kermit", new_password)
+
+        # Assert we can't log in with the old password
+        self.attempt_wrong_password_login("kermit", old_password)
+
    def test_cant_reset_password_without_clicking_link(self):
        """Test that we do actually need to click the link in the email
        """
@@ -386,44 +426,67 @@ class ThreepidEmailRestTestCase(unittest.HomeserverTestCase):
        self.email = "test@example.com"
        self.url_3pid = b"account/3pid"

-    def test_add_email(self):
-        """Test adding an email to profile
-        """
-        client_secret = "foobar"
-        session_id = self._request_token(self.email, client_secret)
+    def test_add_valid_email(self):
+        self.get_success(self._add_email(self.email, self.email))

-        self.assertEquals(len(self.email_attempts), 1)
-        link = self._get_link_from_email()
-
-        self._validate_token(link)
-
-        request, channel = self.make_request(
-            "POST",
-            b"/_matrix/client/unstable/account/3pid/add",
-            {
-                "client_secret": client_secret,
-                "sid": session_id,
-                "auth": {
-                    "type": "m.login.password",
-                    "user": self.user_id,
-                    "password": "test",
-                },
-            },
-            access_token=self.user_id_tok,
+    def test_add_valid_email_second_time(self):
+        self.get_success(self._add_email(self.email, self.email))
+        self.get_success(
+            self._request_token_invalid_email(
+                self.email,
+                expected_errcode=Codes.THREEPID_IN_USE,
+                expected_error="Email is already in use",
+            )
        )

-        self.render(request)
-        self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
-
-        # Get user
-        request, channel = self.make_request(
-            "GET", self.url_3pid, access_token=self.user_id_tok,
+    def test_add_valid_email_second_time_canonicalise(self):
+        self.get_success(self._add_email(self.email, self.email))
+        self.get_success(
+            self._request_token_invalid_email(
+                "TEST@EXAMPLE.COM",
+                expected_errcode=Codes.THREEPID_IN_USE,
+                expected_error="Email is already in use",
+            )
        )
-        self.render(request)

-        self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
-        self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
-        self.assertEqual(self.email, channel.json_body["threepids"][0]["address"])
+    def test_add_email_no_at(self):
+        self.get_success(
+            self._request_token_invalid_email(
+                "address-without-at.bar",
+                expected_errcode=Codes.UNKNOWN,
+                expected_error="Unable to parse email address",
+            )
+        )
+
+    def test_add_email_two_at(self):
+        self.get_success(
+            self._request_token_invalid_email(
+                "foo@foo@test.bar",
+                expected_errcode=Codes.UNKNOWN,
+                expected_error="Unable to parse email address",
+            )
+        )
+
+    def test_add_email_bad_format(self):
+        self.get_success(
+            self._request_token_invalid_email(
+                "user@bad.example.net@good.example.com",
+                expected_errcode=Codes.UNKNOWN,
+                expected_error="Unable to parse email address",
+            )
+        )
+
+    def test_add_email_domain_to_lower(self):
+        self.get_success(self._add_email("foo@TEST.BAR", "foo@test.bar"))
+
+    def test_add_email_domain_with_umlaut(self):
+        self.get_success(self._add_email("foo@Öumlaut.com", "foo@öumlaut.com"))
+
+    def test_add_email_address_casefold(self):
+        self.get_success(self._add_email("Strauß@Example.com", "strauss@example.com"))
+
+    def test_address_trim(self):
+        self.get_success(self._add_email(" foo@test.bar ", "foo@test.bar"))

    def test_add_email_if_disabled(self):
        """Test adding email to profile when doing so is disallowed
@@ -616,6 +679,19 @@ class ThreepidEmailRestTestCase(unittest.HomeserverTestCase):

        return channel.json_body["sid"]

+    def _request_token_invalid_email(
+        self, email, expected_errcode, expected_error, client_secret="foobar",
+    ):
+        request, channel = self.make_request(
+            "POST",
+            b"account/3pid/email/requestToken",
+            {"client_secret": client_secret, "email": email, "send_attempt": 1},
+        )
+        self.render(request)
+        self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"])
+        self.assertEqual(expected_errcode, channel.json_body["errcode"])
+        self.assertEqual(expected_error, channel.json_body["error"])
+
    def _validate_token(self, link):
        # Remove the host
        path = link.replace("https://example.com", "")
@@ -643,3 +719,42 @@ class ThreepidEmailRestTestCase(unittest.HomeserverTestCase):
        assert match, "Could not find link in email"

        return match.group(0)
+
+    def _add_email(self, request_email, expected_email):
+        """Test adding an email to profile
+        """
+        client_secret = "foobar"
+        session_id = self._request_token(request_email, client_secret)
+
+        self.assertEquals(len(self.email_attempts), 1)
+        link = self._get_link_from_email()
+
+        self._validate_token(link)
+
+        request, channel = self.make_request(
+            "POST",
+            b"/_matrix/client/unstable/account/3pid/add",
+            {
+                "client_secret": client_secret,
+                "sid": session_id,
+                "auth": {
+                    "type": "m.login.password",
+                    "user": self.user_id,
+                    "password": "test",
+                },
+            },
+            access_token=self.user_id_tok,
+        )
+
+        self.render(request)
+        self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
+
+        # Get user
+        request, channel = self.make_request(
+            "GET", self.url_3pid, access_token=self.user_id_tok,
+        )
+        self.render(request)
+
+        self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
+        self.assertEqual("email", channel.json_body["threepids"][0]["medium"])
+        self.assertEqual(expected_email, channel.json_body["threepids"][0]["address"])
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -24,12 +24,7 @@ from twisted.web.server import NOT_DONE_YET

 from synapse.api.errors import Codes, RedirectException, SynapseError
 from synapse.config.server import parse_listener_def
-from synapse.http.server import (
-    DirectServeResource,
-    JsonResource,
-    OptionsResource,
-    wrap_html_request_handler,
-)
+from synapse.http.server import DirectServeHtmlResource, JsonResource, OptionsResource
 from synapse.http.site import SynapseSite, logger
 from synapse.logging.context import make_deferred_yieldable
 from synapse.util import Clock
@@ -256,12 +251,11 @@ class OptionsResourceTests(unittest.TestCase):


 class WrapHtmlRequestHandlerTests(unittest.TestCase):
-    class TestResource(DirectServeResource):
+    class TestResource(DirectServeHtmlResource):
        callback = None

-        @wrap_html_request_handler
        async def _async_render_GET(self, request):
-            return await self.callback(request)
+            await self.callback(request)

    def setUp(self):
        self.reactor = ThreadedMemoryReactorClock()
--- a/tests/util/test_threepids.py
+++ b/tests/util/test_threepids.py
@@ -0,0 +1,49 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 Dirk Klimpel
+#
+# 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.util.threepids import canonicalise_email
+
+from tests.unittest import HomeserverTestCase
+
+
+class CanonicaliseEmailTests(HomeserverTestCase):
+    def test_no_at(self):
+        with self.assertRaises(ValueError):
+            canonicalise_email("address-without-at.bar")
+
+    def test_two_at(self):
+        with self.assertRaises(ValueError):
+            canonicalise_email("foo@foo@test.bar")
+
+    def test_bad_format(self):
+        with self.assertRaises(ValueError):
+            canonicalise_email("user@bad.example.net@good.example.com")
+
+    def test_valid_format(self):
+        self.assertEqual(canonicalise_email("foo@test.bar"), "foo@test.bar")
+
+    def test_domain_to_lower(self):
+        self.assertEqual(canonicalise_email("foo@TEST.BAR"), "foo@test.bar")
+
+    def test_domain_with_umlaut(self):
+        self.assertEqual(canonicalise_email("foo@Öumlaut.com"), "foo@öumlaut.com")
+
+    def test_address_casefold(self):
+        self.assertEqual(
+            canonicalise_email("Strauß@Example.com"), "strauss@example.com"
+        )
+
+    def test_address_trim(self):
+        self.assertEqual(canonicalise_email(" foo@test.bar "), "foo@test.bar")
--- a/tox.ini
+++ b/tox.ini
@@ -186,6 +186,7 @@ commands = mypy \
            synapse/handlers/auth.py \
            synapse/handlers/cas_handler.py \
            synapse/handlers/directory.py \
+            synapse/handlers/federation.py \
            synapse/handlers/oidc_handler.py \
            synapse/handlers/presence.py \
            synapse/handlers/room_member.py \
				`@@ -0,0 +1 @@`
				`Fix inconsistent handling of upper and lower case in email addresses when used as identifiers for login, etc. Contributed by @dklimpel.`
				`@@ -1 +0,0 @@`
				Update postgres image in example `docker-compose.yaml` to tag `12-alpine`.
				`@@ -0,0 +1 @@`
				`Fix "Tried to close a non-active scope!" error messages when opentracing is enabled.`
				`@@ -0,0 +1 @@`
				`Fix incorrect error message when database CTYPE was set incorrectly.`
				`@@ -0,0 +1 @@`
				`Send push notifications with a high or low priority depending upon whether they may generate user-observable effects.`
				`@@ -0,0 +1 @@`
				`Use symbolic names for replication stream names.`
				`@@ -0,0 +1 @@`
				Fix up `synapse.handlers.federation` to pass mypy.
				`@@ -0,0 +1 @@`
				`Fix synctl to handle empty config files correctly. Contributed by @kotovalexarian.`