Merge branch 'develop' into rav/saml2_client

22 files changed, 303 insertions, 205 deletions

diff --git a/synapse/config/emailconfig.py b/synapse/config/emailconfig.py
index cf39936da7..fcd55d3e3d 100644
--- a/synapse/config/emailconfig.py
+++ b/synapse/config/emailconfig.py
@@ -233,11 +233,13 @@ class EmailConfig(Config):
         #   app_name: Matrix
         #
         #   # Enable email notifications by default
+        #   #
         #   notif_for_new_users: True
         #
         #   # Defining a custom URL for Riot is only needed if email notifications
         #   # should contain links to a self-hosted installation of Riot; when set
         #   # the "app_name" setting is ignored
+        #   #
         #   riot_base_url: "http://localhost/riot"
         #
         #   # Enable sending password reset emails via the configured, trusted
@@ -250,16 +252,22 @@ class EmailConfig(Config):
         #   #
         #   # If this option is set to false and SMTP options have not been
         #   # configured, resetting user passwords via email will be disabled
+        #   #
         #   #trust_identity_server_for_password_resets: false
         #
         #   # Configure the time that a validation email or text message code
         #   # will expire after sending
         #   #
         #   # This is currently used for password resets
+        #   #
         #   #validation_token_lifetime: 1h
         #
         #   # Template directory. All template files should be stored within this
-        #   # directory
+        #   # directory. If not set, default templates from within the Synapse
+        #   # package will be used
+        #   #
+        #   # For the list of default templates, please see
+        #   # https://github.com/matrix-org/synapse/tree/master/synapse/res/templates
         #   #
         #   #template_dir: res/templates
         #
diff --git a/synapse/config/password.py b/synapse/config/password.py
index 598f84fc0c..d5b5953f2f 100644
--- a/synapse/config/password.py
+++ b/synapse/config/password.py
@@ -26,6 +26,7 @@ class PasswordConfig(Config):
             password_config = {}
 
         self.password_enabled = password_config.get("enabled", True)
+        self.password_localdb_enabled = password_config.get("localdb_enabled", True)
         self.password_pepper = password_config.get("pepper", "")
 
     def generate_config_section(self, config_dir_path, server_name, **kwargs):
@@ -35,6 +36,12 @@ class PasswordConfig(Config):
            #
            #enabled: false
 
+           # Uncomment to disable authentication against the local password
+           # database. This is ignored if `enabled` is false, and is only useful
+           # if you have other password_providers.
+           #
+           #localdb_enabled: false
+
            # Uncomment and change to a secret random string for extra security.
            # DO NOT CHANGE THIS AFTER INITIAL SETUP!
            #
diff --git a/synapse/config/tls.py b/synapse/config/tls.py
index 8fcf801418..ca508a224f 100644
--- a/synapse/config/tls.py
+++ b/synapse/config/tls.py
@@ -23,7 +23,7 @@ import six
 
 from unpaddedbase64 import encode_base64
 
-from OpenSSL import crypto
+from OpenSSL import SSL, crypto
 from twisted.internet._sslverify import Certificate, trustRootFromCertificates
 
 from synapse.config._base import Config, ConfigError
@@ -81,6 +81,27 @@ class TlsConfig(Config):
             "federation_verify_certificates", True
         )
 
+        # Minimum TLS version to use for outbound federation traffic
+        self.federation_client_minimum_tls_version = str(
+            config.get("federation_client_minimum_tls_version", 1)
+        )
+
+        if self.federation_client_minimum_tls_version not in ["1", "1.1", "1.2", "1.3"]:
+            raise ConfigError(
+                "federation_client_minimum_tls_version must be one of: 1, 1.1, 1.2, 1.3"
+            )
+
+        # Prevent people shooting themselves in the foot here by setting it to
+        # the biggest number blindly
+        if self.federation_client_minimum_tls_version == "1.3":
+            if getattr(SSL, "OP_NO_TLSv1_3", None) is None:
+                raise ConfigError(
+                    (
+                        "federation_client_minimum_tls_version cannot be 1.3, "
+                        "your OpenSSL does not support it"
+                    )
+                )
+
         # Whitelist of domains to not verify certificates for
         fed_whitelist_entries = config.get(
             "federation_certificate_verification_whitelist", []
@@ -261,6 +282,15 @@ class TlsConfig(Config):
         #
         #federation_verify_certificates: false
 
+        # The minimum TLS version that will be used for outbound federation requests.
+        #
+        # Defaults to `1`. Configurable to `1`, `1.1`, `1.2`, or `1.3`. Note
+        # that setting this value higher than `1.2` will prevent federation to most
+        # of the public Matrix network: only configure it to `1.3` if you have an
+        # entirely private federation setup and you can ensure TLS 1.3 support.
+        #
+        #federation_client_minimum_tls_version: 1.2
+
         # Skip federation certificate verification on the following whitelist
         # of domains.
         #
diff --git a/synapse/crypto/context_factory.py b/synapse/crypto/context_factory.py
index 2bc5cc3807..4f48e8e88d 100644
--- a/synapse/crypto/context_factory.py
+++ b/synapse/crypto/context_factory.py
@@ -24,12 +24,25 @@ from OpenSSL import SSL, crypto
 from twisted.internet._sslverify import _defaultCurveName
 from twisted.internet.abstract import isIPAddress, isIPv6Address
 from twisted.internet.interfaces import IOpenSSLClientConnectionCreator
-from twisted.internet.ssl import CertificateOptions, ContextFactory, platformTrust
+from twisted.internet.ssl import (
+    CertificateOptions,
+    ContextFactory,
+    TLSVersion,
+    platformTrust,
+)
 from twisted.python.failure import Failure
 
 logger = logging.getLogger(__name__)
 
 
+_TLS_VERSION_MAP = {
+    "1": TLSVersion.TLSv1_0,
+    "1.1": TLSVersion.TLSv1_1,
+    "1.2": TLSVersion.TLSv1_2,
+    "1.3": TLSVersion.TLSv1_3,
+}
+
+
 class ServerContextFactory(ContextFactory):
     """Factory for PyOpenSSL SSL contexts that are used to handle incoming
     connections."""
@@ -43,16 +56,18 @@ class ServerContextFactory(ContextFactory):
         try:
             _ecCurve = crypto.get_elliptic_curve(_defaultCurveName)
             context.set_tmp_ecdh(_ecCurve)
-
         except Exception:
             logger.exception("Failed to enable elliptic curve for TLS")
-        context.set_options(SSL.OP_NO_SSLv2 | SSL.OP_NO_SSLv3)
+
+        context.set_options(
+            SSL.OP_NO_SSLv2 | SSL.OP_NO_SSLv3 | SSL.OP_NO_TLSv1 | SSL.OP_NO_TLSv1_1
+        )
         context.use_certificate_chain_file(config.tls_certificate_file)
         context.use_privatekey(config.tls_private_key)
 
         # https://hynek.me/articles/hardening-your-web-servers-ssl-ciphers/
         context.set_cipher_list(
-            "ECDH+AESGCM:ECDH+CHACHA20:ECDH+AES256:ECDH+AES128:!aNULL:!SHA1"
+            "ECDH+AESGCM:ECDH+CHACHA20:ECDH+AES256:ECDH+AES128:!aNULL:!SHA1:!AESCCM"
         )
 
     def getContext(self):
@@ -79,10 +94,22 @@ class ClientTLSOptionsFactory(object):
             # Use CA root certs provided by OpenSSL
             trust_root = platformTrust()
 
-        self._verify_ssl_context = CertificateOptions(trustRoot=trust_root).getContext()
+        # "insecurelyLowerMinimumTo" is the argument that will go lower than
+        # Twisted's default, which is why it is marked as "insecure" (since
+        # Twisted's defaults are reasonably secure). But, since Twisted is
+        # moving to TLS 1.2 by default, we want to respect the config option if
+        # it is set to 1.0 (which the alternate option, raiseMinimumTo, will not
+        # let us do).
+        minTLS = _TLS_VERSION_MAP[config.federation_client_minimum_tls_version]
+
+        self._verify_ssl = CertificateOptions(
+            trustRoot=trust_root, insecurelyLowerMinimumTo=minTLS
+        )
+        self._verify_ssl_context = self._verify_ssl.getContext()
         self._verify_ssl_context.set_info_callback(self._context_info_cb)
 
-        self._no_verify_ssl_context = CertificateOptions().getContext()
+        self._no_verify_ssl = CertificateOptions(insecurelyLowerMinimumTo=minTLS)
+        self._no_verify_ssl_context = self._no_verify_ssl.getContext()
         self._no_verify_ssl_context.set_info_callback(self._context_info_cb)
 
     def get_options(self, host):
diff --git a/synapse/handlers/auth.py b/synapse/handlers/auth.py
index 97b21c4093..c8c1ed3246 100644
--- a/synapse/handlers/auth.py
+++ b/synapse/handlers/auth.py
@@ -743,7 +743,7 @@ class AuthHandler(BaseHandler):
                     result = (result, None)
                 defer.returnValue(result)
 
-        if login_type == LoginType.PASSWORD:
+        if login_type == LoginType.PASSWORD and self.hs.config.password_localdb_enabled:
             known_login_type = True
 
             canonical_user_id = yield self._check_local_password(
diff --git a/synapse/handlers/device.py b/synapse/handlers/device.py
index f59d0479b5..99e8413092 100644
--- a/synapse/handlers/device.py
+++ b/synapse/handlers/device.py
@@ -101,9 +101,13 @@ class DeviceWorkerHandler(BaseHandler):
 
         room_ids = yield self.store.get_rooms_for_user(user_id)
 
-        # First we check if any devices have changed
-        changed = yield self.store.get_user_whose_devices_changed(
-            from_token.device_list_key
+        # First we check if any devices have changed for users that we share
+        # rooms with.
+        users_who_share_room = yield self.store.get_users_who_share_room_with_user(
+            user_id
+        )
+        changed = yield self.store.get_users_whose_devices_changed(
+            from_token.device_list_key, users_who_share_room
         )
 
         # Then work out if any users have since joined
@@ -188,10 +192,6 @@ class DeviceWorkerHandler(BaseHandler):
                         break
 
         if possibly_changed or possibly_left:
-            users_who_share_room = yield self.store.get_users_who_share_room_with_user(
-                user_id
-            )
-
             # Take the intersection of the users whose devices may have changed
             # and those that actually still share a room with the user
             possibly_joined = possibly_changed & users_who_share_room
diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py
index 4d6e883802..66b05b4732 100644
--- a/synapse/handlers/room_member.py
+++ b/synapse/handlers/room_member.py
@@ -823,6 +823,7 @@ class RoomMemberHandler(object):
                 "sender": user.to_string(),
                 "state_key": token,
             },
+            ratelimit=False,
             txn_id=txn_id,
         )
 
diff --git a/synapse/handlers/set_password.py b/synapse/handlers/set_password.py
index 5a0995d4fe..d90c9e0108 100644
--- a/synapse/handlers/set_password.py
+++ b/synapse/handlers/set_password.py
@@ -33,6 +33,9 @@ class SetPasswordHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def set_password(self, user_id, newpassword, requester=None):
+        if not self.hs.config.password_localdb_enabled:
+            raise SynapseError(403, "Password change disabled", errcode=Codes.FORBIDDEN)
+
         password_hash = yield self._auth_handler.hash(newpassword)
 
         except_device_id = requester.device_id if requester else None
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index c5188a1f8e..a3f550554f 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -1058,40 +1058,74 @@ class SyncHandler(object):
         newly_left_rooms,
         newly_left_users,
     ):
+        """Generate the DeviceLists section of sync
+
+        Args:
+            sync_result_builder (SyncResultBuilder)
+            newly_joined_rooms (set[str]): Set of rooms user has joined since
+                previous sync
+            newly_joined_or_invited_users (set[str]): Set of users that have
+                joined or been invited to a room since previous sync.
+            newly_left_rooms (set[str]): Set of rooms user has left since
+                previous sync
+            newly_left_users (set[str]): Set of users that have left a room
+                we're in since previous sync
+
+        Returns:
+            Deferred[DeviceLists]
+        """
+
         user_id = sync_result_builder.sync_config.user.to_string()
         since_token = sync_result_builder.since_token
 
+        # We're going to mutate these fields, so lets copy them rather than
+        # assume they won't get used later.
+        newly_joined_or_invited_users = set(newly_joined_or_invited_users)
+        newly_left_users = set(newly_left_users)
+
         if since_token and since_token.device_list_key:
-            changed = yield self.store.get_user_whose_devices_changed(
-                since_token.device_list_key
+            # We want to figure out what user IDs the client should refetch
+            # device keys for, and which users we aren't going to track changes
+            # for anymore.
+            #
+            # For the first step we check:
+            #   a. if any users we share a room with have updated their devices,
+            #      and
+            #   b. we also check if we've joined any new rooms, or if a user has
+            #      joined a room we're in.
+            #
+            # For the second step we just find any users we no longer share a
+            # room with by looking at all users that have left a room plus users
+            # that were in a room we've left.
+
+            users_who_share_room = yield self.store.get_users_who_share_room_with_user(
+                user_id
+            )
+
+            # Step 1a, check for changes in devices of users we share a room with
+            users_that_have_changed = yield self.store.get_users_whose_devices_changed(
+                since_token.device_list_key, users_who_share_room
             )
 
-            # TODO: Be more clever than this, i.e. remove users who we already
-            # share a room with?
+            # Step 1b, check for newly joined rooms
             for room_id in newly_joined_rooms:
                 joined_users = yield self.state.get_current_users_in_room(room_id)
                 newly_joined_or_invited_users.update(joined_users)
 
-            for room_id in newly_left_rooms:
-                left_users = yield self.state.get_current_users_in_room(room_id)
-                newly_left_users.update(left_users)
-
             # TODO: Check that these users are actually new, i.e. either they
             # weren't in the previous sync *or* they left and rejoined.
-            changed.update(newly_joined_or_invited_users)
+            users_that_have_changed.update(newly_joined_or_invited_users)
 
-            if not changed and not newly_left_users:
-                defer.returnValue(DeviceLists(changed=[], left=newly_left_users))
+            # Now find users that we no longer track
+            for room_id in newly_left_rooms:
+                left_users = yield self.state.get_current_users_in_room(room_id)
+                newly_left_users.update(left_users)
 
-            users_who_share_room = yield self.store.get_users_who_share_room_with_user(
-                user_id
-            )
+            # Remove any users that we still share a room with.
+            newly_left_users -= users_who_share_room
 
             defer.returnValue(
-                DeviceLists(
-                    changed=users_who_share_room & changed,
-                    left=set(newly_left_users) - users_who_share_room,
-                )
+                DeviceLists(changed=users_that_have_changed, left=newly_left_users)
             )
         else:
             defer.returnValue(DeviceLists(changed=[], left=[]))
diff --git a/synapse/http/server.py b/synapse/http/server.py
index 6fd13e87d1..f067c163c1 100644
--- a/synapse/http/server.py
+++ b/synapse/http/server.py
@@ -16,10 +16,11 @@
 
 import cgi
 import collections
+import http.client
 import logging
-
-from six import PY3
-from six.moves import http_client, urllib
+import types
+import urllib
+from io import BytesIO
 
 from canonicaljson import encode_canonical_json, encode_pretty_printed_json, json
 
@@ -41,11 +42,6 @@ from synapse.api.errors import (
 from synapse.util.caches import intern_dict
 from synapse.util.logcontext import preserve_fn
 
-if PY3:
-    from io import BytesIO
-else:
-    from cStringIO import StringIO as BytesIO
-
 logger = logging.getLogger(__name__)
 
 HTML_ERROR_TEMPLATE = """<!DOCTYPE html>
@@ -75,10 +71,9 @@ def wrap_json_request_handler(h):
     deferred fails with any other type of error we send a 500 reponse.
     """
 
-    @defer.inlineCallbacks
-    def wrapped_request_handler(self, request):
+    async def wrapped_request_handler(self, request):
         try:
-            yield h(self, request)
+            await h(self, request)
         except SynapseError as e:
             code = e.code
             logger.info("%s SynapseError: %s - %s", request, code, e.msg)
@@ -142,10 +137,12 @@ def wrap_html_request_handler(h):
     where "request" must be a SynapseRequest.
     """
 
-    def wrapped_request_handler(self, request):
-        d = defer.maybeDeferred(h, self, request)
-        d.addErrback(_return_html_error, request)
-        return d
+    async def wrapped_request_handler(self, request):
+        try:
+            return await h(self, request)
+        except Exception:
+            f = failure.Failure()
+            return _return_html_error(f, request)
 
     return wrap_async_request_handler(wrapped_request_handler)
 
@@ -171,7 +168,7 @@ def _return_html_error(f, request):
                 exc_info=(f.type, f.value, f.getTracebackObject()),
             )
     else:
-        code = http_client.INTERNAL_SERVER_ERROR
+        code = http.client.INTERNAL_SERVER_ERROR
         msg = "Internal server error"
 
         logger.error(
@@ -201,10 +198,9 @@ def wrap_async_request_handler(h):
     logged until the deferred completes.
     """
 
-    @defer.inlineCallbacks
-    def wrapped_async_request_handler(self, request):
+    async def wrapped_async_request_handler(self, request):
         with request.processing():
-            yield h(self, request)
+            await h(self, request)
 
     # we need to preserve_fn here, because the synchronous render method won't yield for
     # us (obviously)
@@ -270,12 +266,11 @@ class JsonResource(HttpServer, resource.Resource):
     def render(self, request):
         """ This gets called by twisted every time someone sends us a request.
         """
-        self._async_render(request)
+        defer.ensureDeferred(self._async_render(request))
         return NOT_DONE_YET
 
     @wrap_json_request_handler
-    @defer.inlineCallbacks
-    def _async_render(self, request):
+    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.
@@ -292,26 +287,19 @@ class JsonResource(HttpServer, resource.Resource):
         # 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.
-
-        def _unquote(s):
-            if PY3:
-                # On Python 3, unquote is unicode -> unicode
-                return urllib.parse.unquote(s)
-            else:
-                # On Python 2, unquote is bytes -> bytes We need to encode the
-                # URL again (as it was decoded by _get_handler_for request), as
-                # ASCII because it's a URL, and then decode it to get the UTF-8
-                # characters that were quoted.
-                return urllib.parse.unquote(s.encode("ascii")).decode("utf8")
-
         kwargs = intern_dict(
             {
-                name: _unquote(value) if value else value
+                name: urllib.parse.unquote(value) if value else value
                 for name, value in group_dict.items()
             }
         )
 
-        callback_return = yield callback(request, **kwargs)
+        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)
@@ -360,6 +348,23 @@ class JsonResource(HttpServer, resource.Resource):
         )
 
 
+class DirectServeResource(resource.Resource):
+    def render(self, request):
+        """
+        Render the request, using an asynchronous render handler if it exists.
+        """
+        render_callback_name = "_async_render_" + request.method.decode("ascii")
+
+        if hasattr(self, render_callback_name):
+            # Call the handler
+            callback = getattr(self, render_callback_name)
+            defer.ensureDeferred(callback(request))
+
+            return NOT_DONE_YET
+        else:
+            super().render(request)
+
+
 def _options_handler(request):
     """Request handler for OPTIONS requests
 
diff --git a/synapse/metrics/__init__.py b/synapse/metrics/__init__.py
index 1f30179b51..eaf0aaa86e 100644
--- a/synapse/metrics/__init__.py
+++ b/synapse/metrics/__init__.py
@@ -437,7 +437,10 @@ def runUntilCurrentTimer(func):
         counts = gc.get_count()
         for i in (2, 1, 0):
             if threshold[i] < counts[i]:
-                logger.info("Collecting gc %d", i)
+                if i == 0:
+                    logger.debug("Collecting gc %d", i)
+                else:
+                    logger.info("Collecting gc %d", i)
 
                 start = time.time()
                 unreachable = gc.collect(i)
diff --git a/synapse/python_dependencies.py b/synapse/python_dependencies.py
index 13698d9638..6324c00ef1 100644
--- a/synapse/python_dependencies.py
+++ b/synapse/python_dependencies.py
@@ -95,6 +95,7 @@ CONDITIONAL_REQUIREMENTS = {
     "url_preview": ["lxml>=3.5.0"],
     "test": ["mock>=2.0", "parameterized"],
     "sentry": ["sentry-sdk>=0.7.2"],
+    "jwt": ["pyjwt>=1.6.4"],
 }
 
 ALL_OPTIONAL_REQUIREMENTS = set()
diff --git a/synapse/rest/client/v1/login.py b/synapse/rest/client/v1/login.py
index 18aa35a7f9..3051b2171b 100644
--- a/synapse/rest/client/v1/login.py
+++ b/synapse/rest/client/v1/login.py
@@ -340,7 +340,7 @@ class LoginRestServlet(RestServlet):
             }
         else:
             user_id, access_token = (
-                yield self.handlers.registration_handler.register(localpart=user)
+                yield self.registration_handler.register(localpart=user)
             )
 
             device_id = login_submission.get("device_id")
diff --git a/synapse/rest/consent/consent_resource.py b/synapse/rest/consent/consent_resource.py
index 9a32892d8b..624c42441e 100644
--- a/synapse/rest/consent/consent_resource.py
+++ b/synapse/rest/consent/consent_resource.py
@@ -23,13 +23,13 @@ from six.moves import http_client
 import jinja2
 from jinja2 import TemplateNotFound
 
-from twisted.internet import defer
-from twisted.web.resource import Resource
-from twisted.web.server import NOT_DONE_YET
-
 from synapse.api.errors import NotFoundError, StoreError, SynapseError
 from synapse.config import ConfigError
-from synapse.http.server import finish_request, wrap_html_request_handler
+from synapse.http.server import (
+    DirectServeResource,
+    finish_request,
+    wrap_html_request_handler,
+)
 from synapse.http.servlet import parse_string
 from synapse.types import UserID
 
@@ -47,7 +47,7 @@ else:
         return a == b
 
 
-class ConsentResource(Resource):
+class ConsentResource(DirectServeResource):
     """A twisted Resource to display a privacy policy and gather consent to it
 
     When accessed via GET, returns the privacy policy via a template.
@@ -87,7 +87,7 @@ class ConsentResource(Resource):
         Args:
             hs (synapse.server.HomeServer): homeserver
         """
-        Resource.__init__(self)
+        super().__init__()
 
         self.hs = hs
         self.store = hs.get_datastore()
@@ -118,18 +118,12 @@ class ConsentResource(Resource):
 
         self._hmac_secret = hs.config.form_secret.encode("utf-8")
 
-    def render_GET(self, request):
-        self._async_render_GET(request)
-        return NOT_DONE_YET
-
     @wrap_html_request_handler
-    @defer.inlineCallbacks
-    def _async_render_GET(self, request):
+    async def _async_render_GET(self, request):
         """
         Args:
             request (twisted.web.http.Request):
         """
-
         version = parse_string(request, "v", default=self._default_consent_version)
         username = parse_string(request, "u", required=False, default="")
         userhmac = None
@@ -145,7 +139,7 @@ class ConsentResource(Resource):
             else:
                 qualified_user_id = UserID(username, self.hs.hostname).to_string()
 
-            u = yield self.store.get_user_by_id(qualified_user_id)
+            u = await self.store.get_user_by_id(qualified_user_id)
             if u is None:
                 raise NotFoundError("Unknown user")
 
@@ -165,13 +159,8 @@ class ConsentResource(Resource):
         except TemplateNotFound:
             raise NotFoundError("Unknown policy version")
 
-    def render_POST(self, request):
-        self._async_render_POST(request)
-        return NOT_DONE_YET
-
     @wrap_html_request_handler
-    @defer.inlineCallbacks
-    def _async_render_POST(self, request):
+    async def _async_render_POST(self, request):
         """
         Args:
             request (twisted.web.http.Request):
@@ -188,12 +177,12 @@ class ConsentResource(Resource):
             qualified_user_id = UserID(username, self.hs.hostname).to_string()
 
         try:
-            yield self.store.user_set_consent_version(qualified_user_id, version)
+            await self.store.user_set_consent_version(qualified_user_id, version)
         except StoreError as e:
             if e.code != 404:
                 raise
             raise NotFoundError("Unknown user")
-        yield self.registration_handler.post_consent_actions(qualified_user_id)
+        await self.registration_handler.post_consent_actions(qualified_user_id)
 
         try:
             self._render_template(request, "success.html")
diff --git a/synapse/rest/key/v2/remote_key_resource.py b/synapse/rest/key/v2/remote_key_resource.py
index ec8b9d7269..031a316693 100644
--- a/synapse/rest/key/v2/remote_key_resource.py
+++ b/synapse/rest/key/v2/remote_key_resource.py
@@ -16,18 +16,20 @@ import logging
 from io import BytesIO
 
 from twisted.internet import defer
-from twisted.web.resource import Resource
-from twisted.web.server import NOT_DONE_YET
 
 from synapse.api.errors import Codes, SynapseError
 from synapse.crypto.keyring import ServerKeyFetcher
-from synapse.http.server import respond_with_json_bytes, wrap_json_request_handler
+from synapse.http.server import (
+    DirectServeResource,
+    respond_with_json_bytes,
+    wrap_json_request_handler,
+)
 from synapse.http.servlet import parse_integer, parse_json_object_from_request
 
 logger = logging.getLogger(__name__)
 
 
-class RemoteKey(Resource):
+class RemoteKey(DirectServeResource):
     """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
@@ -94,13 +96,8 @@ class RemoteKey(Resource):
         self.clock = hs.get_clock()
         self.federation_domain_whitelist = hs.config.federation_domain_whitelist
 
-    def render_GET(self, request):
-        self.async_render_GET(request)
-        return NOT_DONE_YET
-
     @wrap_json_request_handler
-    @defer.inlineCallbacks
-    def async_render_GET(self, request):
+    async def _async_render_GET(self, request):
         if len(request.postpath) == 1:
             server, = request.postpath
             query = {server.decode("ascii"): {}}
@@ -114,20 +111,15 @@ class RemoteKey(Resource):
         else:
             raise SynapseError(404, "Not found %r" % request.postpath, Codes.NOT_FOUND)
 
-        yield self.query_keys(request, query, query_remote_on_cache_miss=True)
-
-    def render_POST(self, request):
-        self.async_render_POST(request)
-        return NOT_DONE_YET
+        await self.query_keys(request, query, query_remote_on_cache_miss=True)
 
     @wrap_json_request_handler
-    @defer.inlineCallbacks
-    def async_render_POST(self, request):
+    async def _async_render_POST(self, request):
         content = parse_json_object_from_request(request)
 
         query = content["server_keys"]
 
-        yield self.query_keys(request, query, query_remote_on_cache_miss=True)
+        await self.query_keys(request, query, query_remote_on_cache_miss=True)
 
     @defer.inlineCallbacks
     def query_keys(self, request, query, query_remote_on_cache_miss=False):
diff --git a/synapse/rest/media/v1/config_resource.py b/synapse/rest/media/v1/config_resource.py
index fa3d6680fc..9f747de263 100644
--- a/synapse/rest/media/v1/config_resource.py
+++ b/synapse/rest/media/v1/config_resource.py
@@ -14,31 +14,28 @@
 # limitations under the License.
 #
 
-from twisted.internet import defer
-from twisted.web.resource import Resource
 from twisted.web.server import NOT_DONE_YET
 
-from synapse.http.server import respond_with_json, wrap_json_request_handler
+from synapse.http.server import (
+    DirectServeResource,
+    respond_with_json,
+    wrap_json_request_handler,
+)
 
 
-class MediaConfigResource(Resource):
+class MediaConfigResource(DirectServeResource):
     isLeaf = True
 
     def __init__(self, hs):
-        Resource.__init__(self)
+        super().__init__()
         config = hs.get_config()
         self.clock = hs.get_clock()
         self.auth = hs.get_auth()
         self.limits_dict = {"m.upload.size": config.max_upload_size}
 
-    def render_GET(self, request):
-        self._async_render_GET(request)
-        return NOT_DONE_YET
-
     @wrap_json_request_handler
-    @defer.inlineCallbacks
-    def _async_render_GET(self, request):
-        yield self.auth.get_user_by_req(request)
+    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):
diff --git a/synapse/rest/media/v1/download_resource.py b/synapse/rest/media/v1/download_resource.py
index a21a35f843..66a01559e1 100644
--- a/synapse/rest/media/v1/download_resource.py
+++ b/synapse/rest/media/v1/download_resource.py
@@ -14,37 +14,31 @@
 # limitations under the License.
 import logging
 
-from twisted.internet import defer
-from twisted.web.resource import Resource
-from twisted.web.server import NOT_DONE_YET
-
 import synapse.http.servlet
-from synapse.http.server import set_cors_headers, wrap_json_request_handler
+from synapse.http.server import (
+    DirectServeResource,
+    set_cors_headers,
+    wrap_json_request_handler,
+)
 
 from ._base import parse_media_id, respond_404
 
 logger = logging.getLogger(__name__)
 
 
-class DownloadResource(Resource):
+class DownloadResource(DirectServeResource):
     isLeaf = True
 
     def __init__(self, hs, media_repo):
-        Resource.__init__(self)
-
+        super().__init__()
         self.media_repo = media_repo
         self.server_name = hs.hostname
 
         # this is expected by @wrap_json_request_handler
         self.clock = hs.get_clock()
 
-    def render_GET(self, request):
-        self._async_render_GET(request)
-        return NOT_DONE_YET
-
     @wrap_json_request_handler
-    @defer.inlineCallbacks
-    def _async_render_GET(self, request):
+    async def _async_render_GET(self, request):
         set_cors_headers(request)
         request.setHeader(
             b"Content-Security-Policy",
@@ -58,7 +52,7 @@ class DownloadResource(Resource):
         )
         server_name, media_id, name = parse_media_id(request)
         if server_name == self.server_name:
-            yield self.media_repo.get_local_media(request, media_id, name)
+            await self.media_repo.get_local_media(request, media_id, name)
         else:
             allow_remote = synapse.http.servlet.parse_boolean(
                 request, "allow_remote", default=True
@@ -72,4 +66,4 @@ class DownloadResource(Resource):
                 respond_404(request)
                 return
 
-            yield self.media_repo.get_remote_media(request, server_name, media_id, name)
+            await self.media_repo.get_remote_media(request, server_name, media_id, name)
diff --git a/synapse/rest/media/v1/preview_url_resource.py b/synapse/rest/media/v1/preview_url_resource.py
index de6f292ffb..0337b64dc2 100644
--- a/synapse/rest/media/v1/preview_url_resource.py
+++ b/synapse/rest/media/v1/preview_url_resource.py
@@ -32,12 +32,11 @@ from canonicaljson import json
 
 from twisted.internet import defer
 from twisted.internet.error import DNSLookupError
-from twisted.web.resource import Resource
-from twisted.web.server import NOT_DONE_YET
 
 from synapse.api.errors import Codes, SynapseError
 from synapse.http.client import SimpleHttpClient
 from synapse.http.server import (
+    DirectServeResource,
     respond_with_json,
     respond_with_json_bytes,
     wrap_json_request_handler,
@@ -58,11 +57,11 @@ _charset_match = re.compile(br"<\s*meta[^>]*charset\s*=\s*([a-z0-9-]+)", flags=r
 _content_type_match = re.compile(r'.*; *charset="?(.*?)"?(;|$)', flags=re.I)
 
 
-class PreviewUrlResource(Resource):
+class PreviewUrlResource(DirectServeResource):
     isLeaf = True
 
     def __init__(self, hs, media_repo, media_storage):
-        Resource.__init__(self)
+        super().__init__()
 
         self.auth = hs.get_auth()
         self.clock = hs.get_clock()
@@ -98,16 +97,11 @@ class PreviewUrlResource(Resource):
     def render_OPTIONS(self, request):
         return respond_with_json(request, 200, {}, send_cors=True)
 
-    def render_GET(self, request):
-        self._async_render_GET(request)
-        return NOT_DONE_YET
-
     @wrap_json_request_handler
-    @defer.inlineCallbacks
-    def _async_render_GET(self, request):
+    async def _async_render_GET(self, request):
 
         # XXX: if get_user_by_req fails, what should we do in an async render?
-        requester = yield self.auth.get_user_by_req(request)
+        requester = await self.auth.get_user_by_req(request)
         url = parse_string(request, "url")
         if b"ts" in request.args:
             ts = parse_integer(request, "ts")
@@ -159,7 +153,7 @@ class PreviewUrlResource(Resource):
         else:
             logger.info("Returning cached response")
 
-        og = yield make_deferred_yieldable(observable.observe())
+        og = await make_deferred_yieldable(defer.maybeDeferred(observable.observe))
         respond_with_json_bytes(request, 200, og, send_cors=True)
 
     @defer.inlineCallbacks
diff --git a/synapse/rest/media/v1/thumbnail_resource.py b/synapse/rest/media/v1/thumbnail_resource.py
index ca84c9f139..08329884ac 100644
--- a/synapse/rest/media/v1/thumbnail_resource.py
+++ b/synapse/rest/media/v1/thumbnail_resource.py
@@ -17,10 +17,12 @@
 import logging
 
 from twisted.internet import defer
-from twisted.web.resource import Resource
-from twisted.web.server import NOT_DONE_YET
 
-from synapse.http.server import set_cors_headers, wrap_json_request_handler
+from synapse.http.server import (
+    DirectServeResource,
+    set_cors_headers,
+    wrap_json_request_handler,
+)
 from synapse.http.servlet import parse_integer, parse_string
 
 from ._base import (
@@ -34,11 +36,11 @@ from ._base import (
 logger = logging.getLogger(__name__)
 
 
-class ThumbnailResource(Resource):
+class ThumbnailResource(DirectServeResource):
     isLeaf = True
 
     def __init__(self, hs, media_repo, media_storage):
-        Resource.__init__(self)
+        super().__init__()
 
         self.store = hs.get_datastore()
         self.media_repo = media_repo
@@ -47,13 +49,8 @@ class ThumbnailResource(Resource):
         self.server_name = hs.hostname
         self.clock = hs.get_clock()
 
-    def render_GET(self, request):
-        self._async_render_GET(request)
-        return NOT_DONE_YET
-
     @wrap_json_request_handler
-    @defer.inlineCallbacks
-    def _async_render_GET(self, request):
+    async def _async_render_GET(self, request):
         set_cors_headers(request)
         server_name, media_id, _ = parse_media_id(request)
         width = parse_integer(request, "width", required=True)
@@ -63,21 +60,21 @@ class ThumbnailResource(Resource):
 
         if server_name == self.server_name:
             if self.dynamic_thumbnails:
-                yield self._select_or_generate_local_thumbnail(
+                await self._select_or_generate_local_thumbnail(
                     request, media_id, width, height, method, m_type
                 )
             else:
-                yield self._respond_local_thumbnail(
+                await self._respond_local_thumbnail(
                     request, media_id, width, height, method, m_type
                 )
             self.media_repo.mark_recently_accessed(None, media_id)
         else:
             if self.dynamic_thumbnails:
-                yield self._select_or_generate_remote_thumbnail(
+                await self._select_or_generate_remote_thumbnail(
                     request, server_name, media_id, width, height, method, m_type
                 )
             else:
-                yield self._respond_remote_thumbnail(
+                await self._respond_remote_thumbnail(
                     request, server_name, media_id, width, height, method, m_type
                 )
             self.media_repo.mark_recently_accessed(server_name, media_id)
diff --git a/synapse/rest/media/v1/upload_resource.py b/synapse/rest/media/v1/upload_resource.py
index d1d7e959f0..5d76bbdf68 100644
--- a/synapse/rest/media/v1/upload_resource.py
+++ b/synapse/rest/media/v1/upload_resource.py
@@ -15,22 +15,24 @@
 
 import logging
 
-from twisted.internet import defer
-from twisted.web.resource import Resource
 from twisted.web.server import NOT_DONE_YET
 
 from synapse.api.errors import SynapseError
-from synapse.http.server import respond_with_json, wrap_json_request_handler
+from synapse.http.server import (
+    DirectServeResource,
+    respond_with_json,
+    wrap_json_request_handler,
+)
 from synapse.http.servlet import parse_string
 
 logger = logging.getLogger(__name__)
 
 
-class UploadResource(Resource):
+class UploadResource(DirectServeResource):
     isLeaf = True
 
     def __init__(self, hs, media_repo):
-        Resource.__init__(self)
+        super().__init__()
 
         self.media_repo = media_repo
         self.filepaths = media_repo.filepaths
@@ -41,18 +43,13 @@ class UploadResource(Resource):
         self.max_upload_size = hs.config.max_upload_size
         self.clock = hs.get_clock()
 
-    def render_POST(self, request):
-        self._async_render_POST(request)
-        return NOT_DONE_YET
-
     def render_OPTIONS(self, request):
         respond_with_json(request, 200, {}, send_cors=True)
         return NOT_DONE_YET
 
     @wrap_json_request_handler
-    @defer.inlineCallbacks
-    def _async_render_POST(self, request):
-        requester = yield self.auth.get_user_by_req(request)
+    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
         # already been uploaded to a tmp file at this point
         content_length = request.getHeader(b"Content-Length").decode("ascii")
@@ -81,7 +78,7 @@ class UploadResource(Resource):
         #     disposition = headers.getRawHeaders(b"Content-Disposition")[0]
         # TODO(markjh): parse content-dispostion
 
-        content_uri = yield self.media_repo.create_content(
+        content_uri = await self.media_repo.create_content(
             media_type, upload_name, request.content, content_length, requester.user
         )
 
diff --git a/synapse/rest/saml2/response_resource.py b/synapse/rest/saml2/response_resource.py
index 8ee22473e9..69ecc5e4b4 100644
--- a/synapse/rest/saml2/response_resource.py
+++ b/synapse/rest/saml2/response_resource.py
@@ -14,25 +14,18 @@
 # 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 DirectServeResource, wrap_html_request_handler
 
-from synapse.http.server import wrap_html_request_handler
 
-
-class SAML2ResponseResource(Resource):
+class SAML2ResponseResource(DirectServeResource):
     """A Twisted web resource which handles the SAML response"""
 
     isLeaf = 1
 
     def __init__(self, hs):
-        Resource.__init__(self)
+        super().__init__()
         self._saml_handler = hs.get_saml_handler()
 
-    def render_POST(self, request):
-        self._async_render_POST(request)
-        return NOT_DONE_YET
-
     @wrap_html_request_handler
-    def _async_render_POST(self, request):
-        return self._saml_handler.handle_saml_response(request)
+    async def _async_render_POST(self, request):
+        return await self._saml_handler.handle_saml_response(request)
diff --git a/synapse/storage/devices.py b/synapse/storage/devices.py
index 3413a46675..d2b113a4e7 100644
--- a/synapse/storage/devices.py
+++ b/synapse/storage/devices.py
@@ -24,6 +24,7 @@ from synapse.api.errors import StoreError
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage._base import Cache, SQLBaseStore, db_to_json
 from synapse.storage.background_updates import BackgroundUpdateStore
+from synapse.util import batch_iter
 from synapse.util.caches.descriptors import cached, cachedInlineCallbacks, cachedList
 
 logger = logging.getLogger(__name__)
@@ -391,22 +392,47 @@ class DeviceWorkerStore(SQLBaseStore):
 
         return now_stream_id, []
 
-    @defer.inlineCallbacks
-    def get_user_whose_devices_changed(self, from_key):
-        """Get set of users whose devices have changed since `from_key`.
+    def get_users_whose_devices_changed(self, from_key, user_ids):
+        """Get set of users whose devices have changed since `from_key` that
+        are in the given list of user_ids.
+
+        Args:
+            from_key (str): The device lists stream token
+            user_ids (Iterable[str])
+
+        Returns:
+            Deferred[set[str]]: The set of user_ids whose devices have changed
+            since `from_key`
         """
         from_key = int(from_key)
-        changed = self._device_list_stream_cache.get_all_entities_changed(from_key)
-        if changed is not None:
-            defer.returnValue(set(changed))
 
-        sql = """
-            SELECT DISTINCT user_id FROM device_lists_stream WHERE stream_id > ?
-        """
-        rows = yield self._execute(
-            "get_user_whose_devices_changed", None, sql, from_key
+        # Get set of users who *may* have changed. Users not in the returned
+        # list have definitely not changed.
+        to_check = list(
+            self._device_list_stream_cache.get_entities_changed(user_ids, from_key)
+        )
+
+        if not to_check:
+            return defer.succeed(set())
+
+        def _get_users_whose_devices_changed_txn(txn):
+            changes = set()
+
+            sql = """
+                SELECT DISTINCT user_id FROM device_lists_stream
+                WHERE stream_id > ?
+                AND user_id IN (%s)
+            """
+
+            for chunk in batch_iter(to_check, 100):
+                txn.execute(sql % (",".join("?" for _ in chunk),), (from_key,) + chunk)
+                changes.update(user_id for user_id, in txn)
+
+            return changes
+
+        return self.runInteraction(
+            "get_users_whose_devices_changed", _get_users_whose_devices_changed_txn
         )
-        defer.returnValue(set(row[0] for row in rows))
 
     def get_all_device_list_changes_for_remotes(self, from_key, to_key):
         """Return a list of `(stream_id, user_id, destination)` which is the