Make use of new argument to eliminate IS checks outside of registration

Add argument specifying whether 3pid check is happening during registration
Add some logging around checking 3PIDs are allowed
2020-09-29 17:14:44 +01:00 · 2020-09-29 17:13:58 +01:00 · 2020-09-29 15:59:54 +01:00 · 2020-09-29 09:17:11 +01:00 · 2020-09-28 18:41:20 +01:00 · 2020-09-28 15:34:16 +01:00
338 changed files with 8393 additions and 4676 deletions
@@ -1,3 +1,183 @@
+Synapse 1.20.1 (2020-09-24)
+===========================
+
+Bugfixes
+--------
+
+- Fix a bug introduced in v1.20.0 which caused the `synapse_port_db` script to fail. ([\#8386](https://github.com/matrix-org/synapse/issues/8386))
+- Fix a bug introduced in v1.20.0 which caused variables to be incorrectly escaped in Jinja2 templates. ([\#8394](https://github.com/matrix-org/synapse/issues/8394))
+
+
+Synapse 1.20.0 (2020-09-22)
+===========================
+
+No significant changes since v1.20.0rc5.
+
+Removal warning
+---------------
+
+Historically, the [Synapse Admin
+API](https://github.com/matrix-org/synapse/tree/master/docs) has been
+accessible under the `/_matrix/client/api/v1/admin`,
+`/_matrix/client/unstable/admin`, `/_matrix/client/r0/admin` and
+`/_synapse/admin` prefixes. In a future release, we will be dropping support
+for accessing Synapse's Admin API using the `/_matrix/client/*` prefixes. This
+makes it easier for homeserver admins to lock down external access to the Admin
+API endpoints.
+
+Synapse 1.20.0rc5 (2020-09-18)
+==============================
+
+In addition to the below, Synapse 1.20.0rc5 also includes the bug fix that was included in 1.19.3.
+
+Features
+--------
+
+- Add flags to the `/versions` endpoint for whether new rooms default to using E2EE. ([\#8343](https://github.com/matrix-org/synapse/issues/8343))
+
+
+Bugfixes
+--------
+
+- Fix rate limiting of federation `/send` requests. ([\#8342](https://github.com/matrix-org/synapse/issues/8342))
+- Fix a longstanding bug where back pagination over federation could get stuck if it failed to handle a received event. ([\#8349](https://github.com/matrix-org/synapse/issues/8349))
+
+
+Internal Changes
+----------------
+
+- Blacklist [MSC2753](https://github.com/matrix-org/matrix-doc/pull/2753) SyTests until it is implemented. ([\#8285](https://github.com/matrix-org/synapse/issues/8285))
+
+
+Synapse 1.19.3 (2020-09-18)
+===========================
+
+Bugfixes
+--------
+
+- Partially mitigate bug where newly joined servers couldn't get past events in a room when there is a malformed event. ([\#8350](https://github.com/matrix-org/synapse/issues/8350))
+
+
+Synapse 1.20.0rc4 (2020-09-16)
+==============================
+
+Synapse 1.20.0rc4 is identical to 1.20.0rc3, with the addition of the security fix that was included in 1.19.2.
+
+
+Synapse 1.19.2 (2020-09-16)
+===========================
+
+Due to the issue below server admins are encouraged to upgrade as soon as possible.
+
+Bugfixes
+--------
+
+- Fix joining rooms over federation that include malformed events. ([\#8324](https://github.com/matrix-org/synapse/issues/8324))
+
+
+Synapse 1.20.0rc3 (2020-09-11)
+==============================
+
+Bugfixes
+--------
+
+- Fix a bug introduced in v1.20.0rc1 where the wrong exception was raised when invalid JSON data is encountered. ([\#8291](https://github.com/matrix-org/synapse/issues/8291))
+
+
+Synapse 1.20.0rc2 (2020-09-09)
+==============================
+
+Bugfixes
+--------
+
+- Fix a bug introduced in v1.20.0rc1 causing some features related to notifications to misbehave following the implementation of unread counts. ([\#8280](https://github.com/matrix-org/synapse/issues/8280))
+
+
+Synapse 1.20.0rc1 (2020-09-08)
+==============================
+
+Removal warning
+---------------
+
+Some older clients used a [disallowed character](https://matrix.org/docs/spec/client_server/r0.6.1#post-matrix-client-r0-register-email-requesttoken) (`:`) in the `client_secret` parameter of various endpoints. The incorrect behaviour was allowed for backwards compatibility, but is now being removed from Synapse as most users have updated their client. Further context can be found at [\#6766](https://github.com/matrix-org/synapse/issues/6766).
+
+Features
+--------
+
+- Add an endpoint to query your shared rooms with another user as an implementation of [MSC2666](https://github.com/matrix-org/matrix-doc/pull/2666). ([\#7785](https://github.com/matrix-org/synapse/issues/7785))
+- Iteratively encode JSON to avoid blocking the reactor. ([\#8013](https://github.com/matrix-org/synapse/issues/8013), [\#8116](https://github.com/matrix-org/synapse/issues/8116))
+- Add support for shadow-banning users (ignoring any message send requests). ([\#8034](https://github.com/matrix-org/synapse/issues/8034), [\#8092](https://github.com/matrix-org/synapse/issues/8092), [\#8095](https://github.com/matrix-org/synapse/issues/8095), [\#8142](https://github.com/matrix-org/synapse/issues/8142), [\#8152](https://github.com/matrix-org/synapse/issues/8152), [\#8157](https://github.com/matrix-org/synapse/issues/8157), [\#8158](https://github.com/matrix-org/synapse/issues/8158), [\#8176](https://github.com/matrix-org/synapse/issues/8176))
+- Use the default template file when its equivalent is not found in a custom template directory. ([\#8037](https://github.com/matrix-org/synapse/issues/8037), [\#8107](https://github.com/matrix-org/synapse/issues/8107), [\#8252](https://github.com/matrix-org/synapse/issues/8252))
+- Add unread messages count to sync responses, as specified in [MSC2654](https://github.com/matrix-org/matrix-doc/pull/2654). ([\#8059](https://github.com/matrix-org/synapse/issues/8059), [\#8254](https://github.com/matrix-org/synapse/issues/8254), [\#8270](https://github.com/matrix-org/synapse/issues/8270), [\#8274](https://github.com/matrix-org/synapse/issues/8274))
+- Optimise `/federation/v1/user/devices/` API by only returning devices with encryption keys. ([\#8198](https://github.com/matrix-org/synapse/issues/8198))
+
+
+Bugfixes
+--------
+
+- Fix a memory leak by limiting the length of time that messages will be queued for a remote server that has been unreachable. ([\#7864](https://github.com/matrix-org/synapse/issues/7864))
+- Fix `Re-starting finished log context PUT-nnnn` warning when event persistence failed. ([\#8081](https://github.com/matrix-org/synapse/issues/8081))
+- Synapse now correctly enforces the valid characters in the `client_secret` parameter used in various endpoints. ([\#8101](https://github.com/matrix-org/synapse/issues/8101))
+- Fix a bug introduced in v1.7.2 impacting message retention policies that would allow federated homeservers to dictate a retention period that's lower than the configured minimum allowed duration in the configuration file. ([\#8104](https://github.com/matrix-org/synapse/issues/8104))
+- Fix a long-standing bug where invalid JSON would be accepted by Synapse. ([\#8106](https://github.com/matrix-org/synapse/issues/8106))
+- Fix a bug introduced in Synapse v1.12.0 which could cause `/sync` requests to fail with a 404 if you had a very old outstanding room invite. ([\#8110](https://github.com/matrix-org/synapse/issues/8110))
+- Return a proper error code when the rooms of an invalid group are requested. ([\#8129](https://github.com/matrix-org/synapse/issues/8129))
+- Fix a bug which could cause a leaked postgres connection if synapse was set to daemonize. ([\#8131](https://github.com/matrix-org/synapse/issues/8131))
+- Clarify the error code if a user tries to register with a numeric ID. This bug was introduced in v1.15.0. ([\#8135](https://github.com/matrix-org/synapse/issues/8135))
+- Fix a bug where appservices with ratelimiting disabled would still be ratelimited when joining rooms. This bug was introduced in v1.19.0. ([\#8139](https://github.com/matrix-org/synapse/issues/8139))
+- Fix logging in via OpenID Connect with a provider that uses integer user IDs. ([\#8190](https://github.com/matrix-org/synapse/issues/8190))
+- Fix a longstanding bug where user directory updates could break when unexpected profile data was included in events. ([\#8223](https://github.com/matrix-org/synapse/issues/8223))
+- Fix a longstanding bug where stats updates could break when unexpected profile data was included in events. ([\#8226](https://github.com/matrix-org/synapse/issues/8226))
+- Fix slow start times for large servers by removing a table scan of the `users` table from startup code. ([\#8271](https://github.com/matrix-org/synapse/issues/8271))
+
+
+Updates to the Docker image
+---------------------------
+
+- Fix builds of the Docker image on non-x86 platforms. ([\#8144](https://github.com/matrix-org/synapse/issues/8144))
+- Added curl for healthcheck support and readme updates for the change. Contributed by @maquis196. ([\#8147](https://github.com/matrix-org/synapse/issues/8147))
+
+
+Improved Documentation
+----------------------
+
+- Link to matrix-synapse-rest-password-provider in the password provider documentation. ([\#8111](https://github.com/matrix-org/synapse/issues/8111))
+- Updated documentation to note that Synapse does not follow `HTTP 308` redirects due to an upstream library not supporting them. Contributed by Ryan Cole. ([\#8120](https://github.com/matrix-org/synapse/issues/8120))
+- Explain better what GDPR-erased means when deactivating a user. ([\#8189](https://github.com/matrix-org/synapse/issues/8189))
+
+
+Internal Changes
+----------------
+
+- Add filter `name` to the `/users` admin API, which filters by user ID or displayname. Contributed by Awesome Technologies Innovationslabor GmbH. ([\#7377](https://github.com/matrix-org/synapse/issues/7377), [\#8163](https://github.com/matrix-org/synapse/issues/8163))
+- Reduce run times of some unit tests by advancing the reactor a fewer number of times. ([\#7757](https://github.com/matrix-org/synapse/issues/7757))
+- Don't fail `/submit_token` requests on incorrect session ID if `request_token_inhibit_3pid_errors` is turned on. ([\#7991](https://github.com/matrix-org/synapse/issues/7991))
+- Convert various parts of the codebase to async/await. ([\#8071](https://github.com/matrix-org/synapse/issues/8071), [\#8072](https://github.com/matrix-org/synapse/issues/8072), [\#8074](https://github.com/matrix-org/synapse/issues/8074), [\#8075](https://github.com/matrix-org/synapse/issues/8075), [\#8076](https://github.com/matrix-org/synapse/issues/8076), [\#8087](https://github.com/matrix-org/synapse/issues/8087), [\#8100](https://github.com/matrix-org/synapse/issues/8100), [\#8119](https://github.com/matrix-org/synapse/issues/8119), [\#8121](https://github.com/matrix-org/synapse/issues/8121), [\#8133](https://github.com/matrix-org/synapse/issues/8133), [\#8156](https://github.com/matrix-org/synapse/issues/8156), [\#8162](https://github.com/matrix-org/synapse/issues/8162), [\#8166](https://github.com/matrix-org/synapse/issues/8166), [\#8168](https://github.com/matrix-org/synapse/issues/8168), [\#8173](https://github.com/matrix-org/synapse/issues/8173), [\#8191](https://github.com/matrix-org/synapse/issues/8191), [\#8192](https://github.com/matrix-org/synapse/issues/8192), [\#8193](https://github.com/matrix-org/synapse/issues/8193), [\#8194](https://github.com/matrix-org/synapse/issues/8194), [\#8195](https://github.com/matrix-org/synapse/issues/8195), [\#8197](https://github.com/matrix-org/synapse/issues/8197), [\#8199](https://github.com/matrix-org/synapse/issues/8199), [\#8200](https://github.com/matrix-org/synapse/issues/8200), [\#8201](https://github.com/matrix-org/synapse/issues/8201), [\#8202](https://github.com/matrix-org/synapse/issues/8202), [\#8207](https://github.com/matrix-org/synapse/issues/8207), [\#8213](https://github.com/matrix-org/synapse/issues/8213), [\#8214](https://github.com/matrix-org/synapse/issues/8214))
+- Remove some unused database functions. ([\#8085](https://github.com/matrix-org/synapse/issues/8085))
+- Add type hints to various parts of the codebase. ([\#8090](https://github.com/matrix-org/synapse/issues/8090), [\#8127](https://github.com/matrix-org/synapse/issues/8127), [\#8187](https://github.com/matrix-org/synapse/issues/8187), [\#8241](https://github.com/matrix-org/synapse/issues/8241), [\#8140](https://github.com/matrix-org/synapse/issues/8140), [\#8183](https://github.com/matrix-org/synapse/issues/8183), [\#8232](https://github.com/matrix-org/synapse/issues/8232), [\#8235](https://github.com/matrix-org/synapse/issues/8235), [\#8237](https://github.com/matrix-org/synapse/issues/8237), [\#8244](https://github.com/matrix-org/synapse/issues/8244))
+- Return the previous stream token if a non-member event is a duplicate. ([\#8093](https://github.com/matrix-org/synapse/issues/8093), [\#8112](https://github.com/matrix-org/synapse/issues/8112))
+- Separate `get_current_token` into two since there are two different use cases for it. ([\#8113](https://github.com/matrix-org/synapse/issues/8113))
+- Remove `ChainedIdGenerator`. ([\#8123](https://github.com/matrix-org/synapse/issues/8123))
+- Reduce the amount of whitespace in JSON stored and sent in responses. ([\#8124](https://github.com/matrix-org/synapse/issues/8124))
+- Update the test federation client to handle streaming responses. ([\#8130](https://github.com/matrix-org/synapse/issues/8130))
+- Micro-optimisations to `get_auth_chain_ids`. ([\#8132](https://github.com/matrix-org/synapse/issues/8132))
+- Refactor `StreamIdGenerator` and `MultiWriterIdGenerator` to have the same interface. ([\#8161](https://github.com/matrix-org/synapse/issues/8161))
+- Add functions to `MultiWriterIdGen` used by events stream. ([\#8164](https://github.com/matrix-org/synapse/issues/8164), [\#8179](https://github.com/matrix-org/synapse/issues/8179))
+- Fix tests that were broken due to the merge of 1.19.1. ([\#8167](https://github.com/matrix-org/synapse/issues/8167))
+- Make `SlavedIdTracker.advance` have the same interface as `MultiWriterIDGenerator`. ([\#8171](https://github.com/matrix-org/synapse/issues/8171))
+- Remove unused `is_guest` parameter from, and add safeguard to, `MessageHandler.get_room_data`. ([\#8174](https://github.com/matrix-org/synapse/issues/8174), [\#8181](https://github.com/matrix-org/synapse/issues/8181))
+- Standardize the mypy configuration. ([\#8175](https://github.com/matrix-org/synapse/issues/8175))
+- Refactor some of `LoginRestServlet`'s helper methods, and move them to `AuthHandler` for easier reuse. ([\#8182](https://github.com/matrix-org/synapse/issues/8182))
+- Fix `wait_for_stream_position` to allow multiple waiters on same stream ID. ([\#8196](https://github.com/matrix-org/synapse/issues/8196))
+- Make `MultiWriterIDGenerator` work for streams that use negative values. ([\#8203](https://github.com/matrix-org/synapse/issues/8203))
+- Refactor queries for device keys and cross-signatures. ([\#8204](https://github.com/matrix-org/synapse/issues/8204), [\#8205](https://github.com/matrix-org/synapse/issues/8205), [\#8222](https://github.com/matrix-org/synapse/issues/8222), [\#8224](https://github.com/matrix-org/synapse/issues/8224), [\#8225](https://github.com/matrix-org/synapse/issues/8225), [\#8231](https://github.com/matrix-org/synapse/issues/8231), [\#8233](https://github.com/matrix-org/synapse/issues/8233), [\#8234](https://github.com/matrix-org/synapse/issues/8234))
+- Fix type hints for functions decorated with `@cached`. ([\#8240](https://github.com/matrix-org/synapse/issues/8240))
+- Remove obsolete `order` field from federation send queues. ([\#8245](https://github.com/matrix-org/synapse/issues/8245))
+- Stop sub-classing from object. ([\#8249](https://github.com/matrix-org/synapse/issues/8249))
+- Add more logging to debug slow startup. ([\#8264](https://github.com/matrix-org/synapse/issues/8264))
+- Do not attempt to upgrade database schema on worker processes. ([\#8266](https://github.com/matrix-org/synapse/issues/8266), [\#8276](https://github.com/matrix-org/synapse/issues/8276))
+
+
 Synapse 1.19.1 (2020-08-27)
 ===========================

@@ -24,7 +24,7 @@ from twisted.web.client import Agent, readBody
 from twisted.web.http_headers import Headers


-class HttpClient(object):
+class HttpClient:
    """ Interface for talking json over http
    """

@@ -169,7 +169,7 @@ class TwistedHttpClient(HttpClient):
        return d


-class _RawProducer(object):
+class _RawProducer:
    def __init__(self, data):
        self.data = data
        self.body = data
@@ -186,7 +186,7 @@ class _RawProducer(object):
        pass


-class _JsonProducer(object):
+class _JsonProducer:
    """ Used by the twisted http client to create the HTTP body from json
    """

@@ -141,7 +141,7 @@ class CursesStdIO:
        curses.endwin()


-class Callback(object):
+class Callback:
    def __init__(self, stdio):
        self.stdio = stdio

@@ -55,7 +55,7 @@ def excpetion_errback(failure):
    logging.exception(failure)


-class InputOutput(object):
+class InputOutput:
    """ This is responsible for basic I/O so that a user can interact with
    the example app.
    """
@@ -132,7 +132,7 @@ class IOLoggerHandler(logging.Handler):
        self.io.print_log(msg)


-class Room(object):
+class Room:
    """ Used to store (in memory) the current membership state of a room, and
    which home servers we should send PDUs associated with the room to.
    """
@@ -1,3 +1,31 @@
+matrix-synapse-py3 (1.20.1) stable; urgency=medium
+
+  * New synapse release 1.20.1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Thu, 24 Sep 2020 16:25:22 +0100
+
+matrix-synapse-py3 (1.20.0) stable; urgency=medium
+
+  [ Synapse Packaging team ]
+  * New synapse release 1.20.0.
+
+  [ Dexter Chua ]
+  * Use Type=notify in systemd service
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 22 Sep 2020 15:19:32 +0100
+
+matrix-synapse-py3 (1.19.3) stable; urgency=medium
+
+  * New synapse release 1.19.3.
+
+ -- Synapse Packaging team <packages@matrix.org>  Fri, 18 Sep 2020 14:59:30 +0100
+
+matrix-synapse-py3 (1.19.2) stable; urgency=medium
+
+  * New synapse release 1.19.2.
+
+ -- Synapse Packaging team <packages@matrix.org>  Wed, 16 Sep 2020 12:50:30 +0100
+
 matrix-synapse-py3 (1.19.1) stable; urgency=medium

  * New synapse release 1.19.1.
@@ -2,7 +2,7 @@
 Description=Synapse Matrix homeserver

 [Service]
-Type=simple
+Type=notify
 User=matrix-synapse
 WorkingDirectory=/var/lib/matrix-synapse
 EnvironmentFile=/etc/default/matrix-synapse
@@ -19,11 +19,16 @@ ARG PYTHON_VERSION=3.7
 FROM docker.io/python:${PYTHON_VERSION}-slim as builder

 # install the OS build deps
-
-
 RUN apt-get update && apt-get install -y \
    build-essential \
+    libffi-dev \
+    libjpeg-dev \
    libpq-dev \
+    libssl-dev \
+    libwebp-dev \
+    libxml++2.6-dev \
+    libxslt1-dev \
+    zlib1g-dev \
 && rm -rf /var/lib/apt/lists/*

 # Build dependencies that are not available as wheels, to speed up rebuilds
@@ -55,9 +60,12 @@ RUN pip install --prefix="/install" --no-warn-script-location \
 FROM docker.io/python:${PYTHON_VERSION}-slim

 RUN apt-get update && apt-get install -y \
-    libpq5 \
-    xmlsec1 \
+    curl \
    gosu \
+    libjpeg62-turbo \
+    libpq5 \
+    libwebp6 \
+    xmlsec1 \
 && rm -rf /var/lib/apt/lists/*

 COPY --from=builder /install /usr/local
@@ -69,3 +77,6 @@ VOLUME ["/data"]
 EXPOSE 8008/tcp 8009/tcp 8448/tcp

 ENTRYPOINT ["/start.py"]
+
+HEALTHCHECK --interval=1m --timeout=5s \
+  CMD curl -fSs http://localhost:8008/health || exit 1
@@ -162,3 +162,32 @@ docker build -t matrixdotorg/synapse -f docker/Dockerfile .

 You can choose to build a different docker image by changing the value of the `-f` flag to
 point to another Dockerfile.
+
+## Disabling the healthcheck
+
+If you are using a non-standard port or tls inside docker you can disable the healthcheck
+whilst running the above `docker run` commands. 
+
+```
+   --no-healthcheck
+```
+## Setting custom healthcheck on docker run
+
+If you wish to point the healthcheck at a different port with docker command, add the following
+
+```
+  --health-cmd 'curl -fSs http://localhost:1234/health'
+```
+
+## Setting the healthcheck in docker-compose file
+
+You can add the following to set a custom healthcheck in a docker compose file.
+You will need version >2.1 for this to work. 
+
+```
+healthcheck:
+  test: ["CMD", "curl", "-fSs", "http://localhost:8008/health"]
+  interval: 1m
+  timeout: 10s
+  retries: 3
+```
@@ -108,7 +108,7 @@ The api is::

    GET /_synapse/admin/v2/users?from=0&limit=10&guests=false

-To use it, you will need to authenticate by providing an `access_token` for a
+To use it, you will need to authenticate by providing an ``access_token`` for a
 server admin: see `README.rst <README.rst>`_.

 The parameter ``from`` is optional but used for pagination, denoting the
@@ -119,8 +119,11 @@ from a previous call.
 The parameter ``limit`` is optional but is used for pagination, denoting the
 maximum number of items to return in this call. Defaults to ``100``.

-The parameter ``user_id`` is optional and filters to only users with user IDs
-that contain this value.
+The parameter ``user_id`` is optional and filters to only return users with user IDs
+that contain this value. This parameter is ignored when using the ``name`` parameter.
+
+The parameter ``name`` is optional and filters to only return users with user ID localparts
+**or** displaynames that contain this value.

 The parameter ``guests`` is optional and if ``false`` will **exclude** guest users.
 Defaults to ``true`` to include guest users.
@@ -211,9 +214,11 @@ Deactivate Account

 This API deactivates an account. It removes active access tokens, resets the
 password, and deletes third-party IDs (to prevent the user requesting a
-password reset). It can also mark the user as GDPR-erased (stopping their data
-from distributed further, and deleting it entirely if there are no other
-references to it).
+password reset).
+
+It can also mark the user as GDPR-erased. This means messages sent by the
+user will still be visible by anyone that was in the room when these messages
+were sent, but hidden from users joining the room afterwards.

 The api is::

@@ -47,6 +47,18 @@ you invite them to. This can be caused by an incorrectly-configured reverse
 proxy: see [reverse_proxy.md](<reverse_proxy.md>) for instructions on how to correctly
 configure a reverse proxy.

+### Known issues
+
+**HTTP `308 Permanent Redirect` redirects are not followed**: Due to missing features
+in the HTTP library used by Synapse, 308 redirects are currently not followed by
+federating servers, which can cause `M_UNKNOWN` or `401 Unauthorized` errors. This
+may affect users who are redirecting apex-to-www (e.g. `example.com` -> `www.example.com`),
+and especially users of the Kubernetes *Nginx Ingress* module, which uses 308 redirect
+codes by default. For those Kubernetes users, [this Stackoverflow post](https://stackoverflow.com/a/52617528/5096871) 
+might be helpful. For other users, switching to a `301 Moved Permanently` code may be
+an option. 308 redirect codes will be supported properly in a future
+release of Synapse.
+
 ## Running a demo federation of Synapses

 If you want to get up and running quickly with a trio of homeservers in a
@@ -14,6 +14,7 @@ password auth provider module implementations:

 * [matrix-synapse-ldap3](https://github.com/matrix-org/matrix-synapse-ldap3/)
 * [matrix-synapse-shared-secret-auth](https://github.com/devture/matrix-synapse-shared-secret-auth)
+* [matrix-synapse-rest-password-provider](https://github.com/ma1uta/matrix-synapse-rest-password-provider)

 ## Required methods

@@ -378,11 +378,10 @@ retention:
  #  min_lifetime: 1d
  #  max_lifetime: 1y

-  # Retention policy limits. If set, a user won't be able to send a
-  # 'm.room.retention' event which features a 'min_lifetime' or a 'max_lifetime'
-  # that's not within this range. This is especially useful in closed federations,
-  # in which server admins can make sure every federating server applies the same
-  # rules.
+  # Retention policy limits. If set, and the state of a room contains a
+  # 'm.room.retention' event in its state which contains a 'min_lifetime' or a
+  # 'max_lifetime' that's out of these bounds, Synapse will cap the room's policy
+  # to these limits when running purge jobs.
  #
  #allowed_lifetime_min: 1d
  #allowed_lifetime_max: 1y
@@ -408,12 +407,19 @@ retention:
  # (e.g. every 12h), but not want that purge to be performed by a job that's
  # iterating over every room it knows, which could be heavy on the server.
  #
+  # If any purge job is configured, it is strongly recommended to have at least
+  # a single job with neither 'shortest_max_lifetime' nor 'longest_max_lifetime'
+  # set, or one job without 'shortest_max_lifetime' and one job without
+  # 'longest_max_lifetime' set. Otherwise some rooms might be ignored, even if
+  # 'allowed_lifetime_min' and 'allowed_lifetime_max' are set, because capping a
+  # room's policy to these values is done after the policies are retrieved from
+  # Synapse's database (which is done using the range specified in a purge job's
+  # configuration).
+  #
  #purge_jobs:
-  #  - shortest_max_lifetime: 1d
-  #    longest_max_lifetime: 3d
+  #  - longest_max_lifetime: 3d
  #    interval: 12h
  #  - shortest_max_lifetime: 3d
-  #    longest_max_lifetime: 1y
  #    interval: 1d

 # Inhibits the /requestToken endpoints from returning an error that might leak
@@ -1133,6 +1139,11 @@ account_validity:
 # Mandate that users are only allowed to associate certain formats of
 # 3PIDs with accounts on this server.
 #
+# Use an Identity Server to establish which 3PIDs are allowed to register?
+# Overrides allowed_local_3pids below.
+#
+#check_is_for_allowed_local_3pids: matrix.org
+#
 #allowed_local_3pids:
 #  - medium: email
 #    pattern: '.*@matrix\.org'
@@ -2002,9 +2013,7 @@ email:
  # Directory in which Synapse will try to find the template files below.
  # If not set, default templates from within the Synapse package will be used.
  #
-  # DO NOT UNCOMMENT THIS SETTING unless you want to customise the templates.
-  # If you *do* uncomment it, you will need to make sure that all the templates
-  # below are in the directory.
+  # Do not uncomment this setting unless you want to customise the templates.
  #
  # Synapse will look for the following templates in this directory:
  #
@@ -1,9 +1,14 @@
 [Unit]
 Description=Synapse %i
 AssertPathExists=/etc/matrix-synapse/workers/%i.yaml
+
 # This service should be restarted when the synapse target is restarted.
 PartOf=matrix-synapse.target

+# if this is started at the same time as the main, let the main process start
+# first, to initialise the database schema.
+After=matrix-synapse.service
+
 [Service]
 Type=notify
 NotifyAccess=main
@@ -1,11 +1,66 @@
 [mypy]
 namespace_packages = True
-plugins = mypy_zope:plugin
+plugins = mypy_zope:plugin, scripts-dev/mypy_synapse_plugin.py
 follow_imports = silent
 check_untyped_defs = True
 show_error_codes = True
 show_traceback = True
 mypy_path = stubs
+files =
+  synapse/api,
+  synapse/appservice,
+  synapse/config,
+  synapse/event_auth.py,
+  synapse/events/builder.py,
+  synapse/events/spamcheck.py,
+  synapse/federation,
+  synapse/handlers/auth.py,
+  synapse/handlers/cas_handler.py,
+  synapse/handlers/directory.py,
+  synapse/handlers/events.py,
+  synapse/handlers/federation.py,
+  synapse/handlers/identity.py,
+  synapse/handlers/initial_sync.py,
+  synapse/handlers/message.py,
+  synapse/handlers/oidc_handler.py,
+  synapse/handlers/pagination.py,
+  synapse/handlers/presence.py,
+  synapse/handlers/room.py,
+  synapse/handlers/room_member.py,
+  synapse/handlers/room_member_worker.py,
+  synapse/handlers/saml_handler.py,
+  synapse/handlers/sync.py,
+  synapse/handlers/ui_auth,
+  synapse/http/federation/well_known_resolver.py,
+  synapse/http/server.py,
+  synapse/http/site.py,
+  synapse/logging/,
+  synapse/metrics,
+  synapse/module_api,
+  synapse/notifier.py,
+  synapse/push/pusherpool.py,
+  synapse/push/push_rule_evaluator.py,
+  synapse/replication,
+  synapse/rest,
+  synapse/server.py,
+  synapse/server_notices,
+  synapse/spam_checker_api,
+  synapse/state,
+  synapse/storage/databases/main/stream.py,
+  synapse/storage/databases/main/ui_auth.py,
+  synapse/storage/database.py,
+  synapse/storage/engines,
+  synapse/storage/state.py,
+  synapse/storage/util,
+  synapse/streams,
+  synapse/types.py,
+  synapse/util/caches/descriptors.py,
+  synapse/util/caches/stream_change_cache.py,
+  synapse/util/metrics.py,
+  tests/replication,
+  tests/test_utils,
+  tests/rest/client/v2_alpha/test_auth.py,
+  tests/util/test_stream_change_cache.py

 [mypy-pymacaroons.*]
 ignore_missing_imports = True
@@ -21,10 +21,12 @@ import argparse
 import base64
 import json
 import sys
+from typing import Any, Optional
 from urllib import parse as urlparse

 import nacl.signing
 import requests
+import signedjson.types
 import srvlookup
 import yaml
 from requests.adapters import HTTPAdapter
@@ -69,7 +71,9 @@ def encode_canonical_json(value):
    ).encode("UTF-8")


-def sign_json(json_object, signing_key, signing_name):
+def sign_json(
+    json_object: Any, signing_key: signedjson.types.SigningKey, signing_name: str
+) -> Any:
    signatures = json_object.pop("signatures", {})
    unsigned = json_object.pop("unsigned", None)

@@ -122,7 +126,14 @@ def read_signing_keys(stream):
    return keys


-def request_json(method, origin_name, origin_key, destination, path, content):
+def request(
+    method: Optional[str],
+    origin_name: str,
+    origin_key: signedjson.types.SigningKey,
+    destination: str,
+    path: str,
+    content: Optional[str],
+) -> requests.Response:
    if method is None:
        if content is None:
            method = "GET"
@@ -159,11 +170,14 @@ def request_json(method, origin_name, origin_key, destination, path, content):
    if method == "POST":
        headers["Content-Type"] = "application/json"

-    result = s.request(
-        method=method, url=dest, headers=headers, verify=False, data=content
+    return s.request(
+        method=method,
+        url=dest,
+        headers=headers,
+        verify=False,
+        data=content,
+        stream=True,
    )
-    sys.stderr.write("Status Code: %d\n" % (result.status_code,))
-    return result.json()


 def main():
@@ -222,7 +236,7 @@ def main():
    with open(args.signing_key_path) as f:
        key = read_signing_keys(f)[0]

-    result = request_json(
+    result = request(
        args.method,
        args.server_name,
        key,
@@ -231,7 +245,12 @@ def main():
        content=args.body,
    )

-    json.dump(result, sys.stdout)
+    sys.stderr.write("Status Code: %d\n" % (result.status_code,))
+
+    for chunk in result.iter_content():
+        # we write raw utf8 to stdout.
+        sys.stdout.buffer.write(chunk)
+
    print("")


@@ -15,7 +15,7 @@ from synapse.storage.pdu import PduStore
 from synapse.storage.signatures import SignatureStore


-class Store(object):
+class Store:
    _get_pdu_tuples = PduStore.__dict__["_get_pdu_tuples"]
    _get_pdu_content_hashes_txn = SignatureStore.__dict__["_get_pdu_content_hashes_txn"]
    _get_prev_pdu_hashes_txn = SignatureStore.__dict__["_get_prev_pdu_hashes_txn"]
@@ -0,0 +1,85 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 The Matrix.org Foundation C.I.C.
+#
+# 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.
+
+"""This is a mypy plugin for Synpase to deal with some of the funky typing that
+can crop up, e.g the cache descriptors.
+"""
+
+from typing import Callable, Optional
+
+from mypy.plugin import MethodSigContext, Plugin
+from mypy.typeops import bind_self
+from mypy.types import CallableType
+
+
+class SynapsePlugin(Plugin):
+    def get_method_signature_hook(
+        self, fullname: str
+    ) -> Optional[Callable[[MethodSigContext], CallableType]]:
+        if fullname.startswith(
+            "synapse.util.caches.descriptors._CachedFunction.__call__"
+        ):
+            return cached_function_method_signature
+        return None
+
+
+def cached_function_method_signature(ctx: MethodSigContext) -> CallableType:
+    """Fixes the `_CachedFunction.__call__` signature to be correct.
+
+    It already has *almost* the correct signature, except:
+
+        1. the `self` argument needs to be marked as "bound"; and
+        2. any `cache_context` argument should be removed.
+    """
+
+    # First we mark this as a bound function signature.
+    signature = bind_self(ctx.default_signature)
+
+    # Secondly, we remove any "cache_context" args.
+    #
+    # Note: We should be only doing this if `cache_context=True` is set, but if
+    # it isn't then the code will raise an exception when its called anyway, so
+    # its not the end of the world.
+    context_arg_index = None
+    for idx, name in enumerate(signature.arg_names):
+        if name == "cache_context":
+            context_arg_index = idx
+            break
+
+    if context_arg_index:
+        arg_types = list(signature.arg_types)
+        arg_types.pop(context_arg_index)
+
+        arg_names = list(signature.arg_names)
+        arg_names.pop(context_arg_index)
+
+        arg_kinds = list(signature.arg_kinds)
+        arg_kinds.pop(context_arg_index)
+
+        signature = signature.copy_modified(
+            arg_types=arg_types, arg_names=arg_names, arg_kinds=arg_kinds,
+        )
+
+    return signature
+
+
+def plugin(version: str):
+    # This is the entry point of the plugin, and let's us deal with the fact
+    # that the mypy plugin interface is *not* stable by looking at the version
+    # string.
+    #
+    # However, since we pin the version of mypy Synapse uses in CI, we don't
+    # really care.
+    return SynapsePlugin
@@ -89,6 +89,7 @@ BOOLEAN_COLUMNS = {
    "redactions": ["have_censored"],
    "room_stats_state": ["is_federatable"],
    "local_media_repository": ["safe_from_quarantine"],
+    "users": ["shadow_banned"],
 }


@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 The Matrix.org Foundation C.I.C.
+#
+# 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.
+
+# Stub for frozendict.
+
+from typing import (
+    Any,
+    Hashable,
+    Iterable,
+    Iterator,
+    Mapping,
+    overload,
+    Tuple,
+    TypeVar,
+)
+
+_KT = TypeVar("_KT", bound=Hashable)  # Key type.
+_VT = TypeVar("_VT")  # Value type.
+
+class frozendict(Mapping[_KT, _VT]):
+    @overload
+    def __init__(self, **kwargs: _VT) -> None: ...
+    @overload
+    def __init__(self, __map: Mapping[_KT, _VT], **kwargs: _VT) -> None: ...
+    @overload
+    def __init__(
+        self, __iterable: Iterable[Tuple[_KT, _VT]], **kwargs: _VT
+    ) -> None: ...
+    def __getitem__(self, key: _KT) -> _VT: ...
+    def __contains__(self, key: Any) -> bool: ...
+    def copy(self, **add_or_replace: Any) -> frozendict: ...
+    def __iter__(self) -> Iterator[_KT]: ...
+    def __len__(self) -> int: ...
+    def __repr__(self) -> str: ...
+    def __hash__(self) -> int: ...
@@ -48,7 +48,7 @@ try:
 except ImportError:
    pass

-__version__ = "1.19.1"
+__version__ = "1.20.1"

 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
@@ -58,7 +58,7 @@ class _InvalidMacaroonException(Exception):
    pass


-class Auth(object):
+class Auth:
    """
    FIXME: This class contains a mix of functions for authenticating users
    of our client-server API and authenticating events added to room graphs.
@@ -213,6 +213,7 @@ class Auth(object):
            user = user_info["user"]
            token_id = user_info["token_id"]
            is_guest = user_info["is_guest"]
+            shadow_banned = user_info["shadow_banned"]

            # Deny the request if the user account has expired.
            if self._account_validity.enabled and not allow_expired:
@@ -252,7 +253,12 @@ class Auth(object):
                opentracing.set_tag("device_id", device_id)

            return synapse.types.create_requester(
-                user, token_id, is_guest, device_id, app_service=app_service
+                user,
+                token_id,
+                is_guest,
+                shadow_banned,
+                device_id,
+                app_service=app_service,
            )
        except KeyError:
            raise MissingClientTokenError()
@@ -297,6 +303,7 @@ class Auth(object):
            dict that includes:
               `user` (UserID)
               `is_guest` (bool)
+               `shadow_banned` (bool)
               `token_id` (int|None): access token id. May be None if guest
               `device_id` (str|None): device corresponding to access token
        Raises:
@@ -356,6 +363,7 @@ class Auth(object):
                ret = {
                    "user": user,
                    "is_guest": True,
+                    "shadow_banned": False,
                    "token_id": None,
                    # all guests get the same device id
                    "device_id": GUEST_DEVICE_ID,
@@ -365,6 +373,7 @@ class Auth(object):
                ret = {
                    "user": user,
                    "is_guest": False,
+                    "shadow_banned": False,
                    "token_id": None,
                    "device_id": None,
                }
@@ -488,6 +497,7 @@ class Auth(object):
            "user": UserID.from_string(ret.get("name")),
            "token_id": ret.get("token_id", None),
            "is_guest": False,
+            "shadow_banned": ret.get("shadow_banned"),
            "device_id": ret.get("device_id"),
            "valid_until_ms": ret.get("valid_until_ms"),
        }
@@ -22,7 +22,7 @@ from synapse.config.server import is_threepid_reserved
 logger = logging.getLogger(__name__)


-class AuthBlocking(object):
+class AuthBlocking:
    def __init__(self, hs):
        self.store = hs.get_datastore()

@@ -28,7 +28,7 @@ MAX_ALIAS_LENGTH = 255
 MAX_USERID_LENGTH = 255


-class Membership(object):
+class Membership:

    """Represents the membership states of a user in a room."""

@@ -40,7 +40,7 @@ class Membership(object):
    LIST = (INVITE, JOIN, KNOCK, LEAVE, BAN)


-class PresenceState(object):
+class PresenceState:
    """Represents the presence state of a user."""

    OFFLINE = "offline"
@@ -48,14 +48,14 @@ class PresenceState(object):
    ONLINE = "online"


-class JoinRules(object):
+class JoinRules:
    PUBLIC = "public"
    KNOCK = "knock"
    INVITE = "invite"
    PRIVATE = "private"


-class LoginType(object):
+class LoginType:
    PASSWORD = "m.login.password"
    EMAIL_IDENTITY = "m.login.email.identity"
    MSISDN = "m.login.msisdn"
@@ -65,7 +65,7 @@ class LoginType(object):
    DUMMY = "m.login.dummy"


-class EventTypes(object):
+class EventTypes:
    Member = "m.room.member"
    Create = "m.room.create"
    Tombstone = "m.room.tombstone"
@@ -96,17 +96,17 @@ class EventTypes(object):
    Presence = "m.presence"


-class RejectedReason(object):
+class RejectedReason:
    AUTH_ERROR = "auth_error"


-class RoomCreationPreset(object):
+class RoomCreationPreset:
    PRIVATE_CHAT = "private_chat"
    PUBLIC_CHAT = "public_chat"
    TRUSTED_PRIVATE_CHAT = "trusted_private_chat"


-class ThirdPartyEntityKind(object):
+class ThirdPartyEntityKind:
    USER = "user"
    LOCATION = "location"

@@ -115,7 +115,7 @@ ServerNoticeMsgType = "m.server_notice"
 ServerNoticeLimitReached = "m.server_notice.usage_limit_reached"


-class UserTypes(object):
+class UserTypes:
    """Allows for user type specific behaviour. With the benefit of hindsight
    'admin' and 'guest' users should also be UserTypes. Normal users are type None
    """
@@ -125,7 +125,7 @@ class UserTypes(object):
    ALL_USER_TYPES = (SUPPORT, BOT)


-class RelationTypes(object):
+class RelationTypes:
    """The types of relations known to this server.
    """

@@ -134,14 +134,14 @@ class RelationTypes(object):
    REFERENCE = "m.reference"


-class LimitBlockingTypes(object):
+class LimitBlockingTypes:
    """Reasons that a server may be blocked"""

    MONTHLY_ACTIVE_USER = "monthly_active_user"
    HS_DISABLED = "hs_disabled"


-class EventContentFields(object):
+class EventContentFields:
    """Fields found in events' content, regardless of type."""

    # Labels for the event, cf https://github.com/matrix-org/matrix-doc/pull/2326
@@ -152,6 +152,6 @@ class EventContentFields(object):
    SELF_DESTRUCT_AFTER = "org.matrix.self_destruct_after"


-class RoomEncryptionAlgorithms(object):
+class RoomEncryptionAlgorithms:
    MEGOLM_V1_AES_SHA2 = "m.megolm.v1.aes-sha2"
    DEFAULT = MEGOLM_V1_AES_SHA2
@@ -21,17 +21,17 @@ import typing
 from http import HTTPStatus
 from typing import Dict, List, Optional, Union

-from canonicaljson import json
-
 from twisted.web import http

+from synapse.util import json_decoder
+
 if typing.TYPE_CHECKING:
    from synapse.types import JsonDict

 logger = logging.getLogger(__name__)


-class Codes(object):
+class Codes:
    UNRECOGNIZED = "M_UNRECOGNIZED"
    UNAUTHORIZED = "M_UNAUTHORIZED"
    FORBIDDEN = "M_FORBIDDEN"
@@ -593,7 +593,7 @@ class HttpResponseException(CodeMessageException):
        # try to parse the body as json, to get better errcode/msg, but
        # default to M_UNKNOWN with the HTTP status as the error text
        try:
-            j = json.loads(self.response.decode("utf-8"))
+            j = json_decoder.decode(self.response.decode("utf-8"))
        except ValueError:
            j = {}

@@ -604,3 +604,11 @@ class HttpResponseException(CodeMessageException):
        errmsg = j.pop("error", self.msg)

        return ProxiedRequestError(self.code, errmsg, errcode, j)
+
+
+class ShadowBanError(Exception):
+    """
+    Raised when a shadow-banned user attempts to perform an action.
+
+    This should be caught and a proper "fake" success response sent to the user.
+    """
@@ -23,7 +23,7 @@ from jsonschema import FormatChecker

 from synapse.api.constants import EventContentFields
 from synapse.api.errors import SynapseError
-from synapse.storage.presence import UserPresenceState
+from synapse.api.presence import UserPresenceState
 from synapse.types import RoomID, UserID

 FILTER_SCHEMA = {
@@ -130,7 +130,7 @@ def matrix_user_id_validator(user_id_str):
    return UserID.from_string(user_id_str)


-class Filtering(object):
+class Filtering:
    def __init__(self, hs):
        super(Filtering, self).__init__()
        self.store = hs.get_datastore()
@@ -168,7 +168,7 @@ class Filtering(object):
            raise SynapseError(400, str(e))


-class FilterCollection(object):
+class FilterCollection:
    def __init__(self, filter_json):
        self._filter_json = filter_json

@@ -249,7 +249,7 @@ class FilterCollection(object):
        )


-class Filter(object):
+class Filter:
    def __init__(self, filter_json):
        self.filter_json = filter_json

@@ -21,7 +21,7 @@ from synapse.types import Requester
 from synapse.util import Clock


-class Ratelimiter(object):
+class Ratelimiter:
    """
    Ratelimit actions marked by arbitrary keys.

@@ -18,7 +18,7 @@ from typing import Dict
 import attr


-class EventFormatVersions(object):
+class EventFormatVersions:
    """This is an internal enum for tracking the version of the event format,
    independently from the room version.
    """
@@ -35,20 +35,20 @@ KNOWN_EVENT_FORMAT_VERSIONS = {
 }


-class StateResolutionVersions(object):
+class StateResolutionVersions:
    """Enum to identify the state resolution algorithms"""

    V1 = 1  # room v1 state res
    V2 = 2  # MSC1442 state res: room v2 and later


-class RoomDisposition(object):
+class RoomDisposition:
    STABLE = "stable"
    UNSTABLE = "unstable"


@attr.s(slots=True, frozen=True)
-class RoomVersion(object):
+class RoomVersion:
    """An object which describes the unique attributes of a room version."""

    identifier = attr.ib()  # str; the identifier for this version
@@ -69,7 +69,7 @@ class RoomVersion(object):
    limit_notifications_power_levels = attr.ib(type=bool)


-class RoomVersions(object):
+class RoomVersions:
    V1 = RoomVersion(
        "1",
        RoomDisposition.STABLE,
@@ -33,7 +33,7 @@ MEDIA_PREFIX = "/_matrix/media/r0"
 LEGACY_MEDIA_PREFIX = "/_matrix/media/v1"


-class ConsentURIBuilder(object):
+class ConsentURIBuilder:
    def __init__(self, hs_config):
        """
        Args:
@@ -334,6 +334,13 @@ def install_dns_limiter(reactor, max_dns_requests_in_flight=100):
    This is to workaround https://twistedmatrix.com/trac/ticket/9620, where we
    can run out of file descriptors and infinite loop if we attempt to do too
    many DNS queries at once
+
+    XXX: I'm confused by this. reactor.nameResolver does not use twisted.names unless
+    you explicitly install twisted.names as the resolver; rather it uses a GAIResolver
+    backed by the reactor's default threadpool (which is limited to 10 threads). So
+    (a) I don't understand why twisted ticket 9620 is relevant, and (b) I don't
+    understand why we would run out of FDs if we did too many lookups at once.
+    -- richvdh 2020/08/29
    """
    new_resolver = _LimitedHostnameResolver(
        reactor.nameResolver, max_dns_requests_in_flight
@@ -342,7 +349,7 @@ def install_dns_limiter(reactor, max_dns_requests_in_flight=100):
    reactor.installNameResolver(new_resolver)


-class _LimitedHostnameResolver(object):
+class _LimitedHostnameResolver:
    """Wraps a IHostnameResolver, limiting the number of in-flight DNS lookups.
    """

@@ -402,7 +409,7 @@ class _LimitedHostnameResolver(object):
            yield deferred


-class _DeferredResolutionReceiver(object):
+class _DeferredResolutionReceiver:
    """Wraps a IResolutionReceiver and simply resolves the given deferred when
    resolution is complete
    """
@@ -79,8 +79,7 @@ class AdminCmdServer(HomeServer):
        pass


-@defer.inlineCallbacks
-def export_data_command(hs, args):
+async def export_data_command(hs, args):
    """Export data for a user.

    Args:
@@ -91,10 +90,8 @@ def export_data_command(hs, args):
    user_id = args.user_id
    directory = args.output_directory

-    res = yield defer.ensureDeferred(
-        hs.get_handlers().admin_handler.export_user_data(
-            user_id, FileExfiltrationWriter(user_id, directory=directory)
-        )
+    res = await hs.get_handlers().admin_handler.export_user_data(
+        user_id, FileExfiltrationWriter(user_id, directory=directory)
    )
    print(res)

@@ -232,14 +229,15 @@ def start(config_options):
    # We also make sure that `_base.start` gets run before we actually run the
    # command.

-    @defer.inlineCallbacks
-    def run(_reactor):
+    async def run():
        with LoggingContext("command"):
-            yield _base.start(ss, [])
-            yield args.func(ss, args)
+            _base.start(ss, [])
+            await args.func(ss, args)

    _base.start_worker_reactor(
-        "synapse-admin-cmd", config, run_command=lambda: task.react(run)
+        "synapse-admin-cmd",
+        config,
+        run_command=lambda: task.react(lambda _reactor: defer.ensureDeferred(run())),
    )


@@ -745,7 +745,7 @@ class GenericWorkerReplicationHandler(ReplicationDataHandler):
            self.send_handler.wake_destination(server)


-class FederationSenderHandler(object):
+class FederationSenderHandler:
    """Processes the fedration replication stream

    This class is only instantiate on the worker responsible for sending outbound
@@ -411,26 +411,24 @@ def setup(config_options):

        return provision

-    @defer.inlineCallbacks
-    def reprovision_acme():
+    async def reprovision_acme():
        """
        Provision a certificate from ACME, if required, and reload the TLS
        certificate if it's renewed.
        """
-        reprovisioned = yield defer.ensureDeferred(do_acme())
+        reprovisioned = await do_acme()
        if reprovisioned:
            _base.refresh_certificate(hs)

-    @defer.inlineCallbacks
-    def start():
+    async def start():
        try:
            # Run the ACME provisioning code, if it's enabled.
            if hs.config.acme_enabled:
                acme = hs.get_acme_handler()
                # Start up the webservices which we will respond to ACME
                # challenges with, and then provision.
-                yield defer.ensureDeferred(acme.start_listening())
-                yield defer.ensureDeferred(do_acme())
+                await acme.start_listening()
+                await do_acme()

                # Check if it needs to be reprovisioned every day.
                hs.get_clock().looping_call(reprovision_acme, 24 * 60 * 60 * 1000)
@@ -439,8 +437,8 @@ def setup(config_options):
            if hs.config.oidc_enabled:
                oidc = hs.get_oidc_handler()
                # Loading the provider metadata also ensures the provider config is valid.
-                yield defer.ensureDeferred(oidc.load_metadata())
-                yield defer.ensureDeferred(oidc.load_jwks())
+                await oidc.load_metadata()
+                await oidc.load_jwks()

            _base.start(hs, config.listeners)

@@ -456,7 +454,7 @@ def setup(config_options):
                reactor.stop()
            sys.exit(1)

-    reactor.callWhenRunning(start)
+    reactor.callWhenRunning(lambda: defer.ensureDeferred(start()))

    return hs

@@ -14,20 +14,25 @@
 # limitations under the License.
 import logging
 import re
+from typing import TYPE_CHECKING

 from synapse.api.constants import EventTypes
+from synapse.appservice.api import ApplicationServiceApi
 from synapse.types import GroupID, get_domain_from_id
 from synapse.util.caches.descriptors import cached

+if TYPE_CHECKING:
+    from synapse.storage.databases.main import DataStore
+
 logger = logging.getLogger(__name__)


-class ApplicationServiceState(object):
+class ApplicationServiceState:
    DOWN = "down"
    UP = "up"


-class AppServiceTransaction(object):
+class AppServiceTransaction:
    """Represents an application service transaction."""

    def __init__(self, service, id, events):
@@ -35,19 +40,19 @@ class AppServiceTransaction(object):
        self.id = id
        self.events = events

-    def send(self, as_api):
+    async def send(self, as_api: ApplicationServiceApi) -> bool:
        """Sends this transaction using the provided AS API interface.

        Args:
-            as_api(ApplicationServiceApi): The API to use to send.
+            as_api: The API to use to send.
        Returns:
-            An Awaitable which resolves to True if the transaction was sent.
+            True if the transaction was sent.
        """
-        return as_api.push_bulk(
+        return await as_api.push_bulk(
            service=self.service, events=self.events, txn_id=self.id
        )

-    def complete(self, store):
+    async def complete(self, store: "DataStore") -> None:
        """Completes this transaction as successful.

        Marks this transaction ID on the application service and removes the
@@ -55,13 +60,11 @@ class AppServiceTransaction(object):

        Args:
            store: The database store to operate on.
-        Returns:
-            A Deferred which resolves to True if the transaction was completed.
        """
-        return store.complete_appservice_txn(service=self.service, txn_id=self.id)
+        await store.complete_appservice_txn(service=self.service, txn_id=self.id)


-class ApplicationService(object):
+class ApplicationService:
    """Defines an application service. This definition is mostly what is
    provided to the /register AS API.

@@ -14,18 +14,20 @@
 # limitations under the License.
 import logging
 import urllib
+from typing import TYPE_CHECKING, Optional

 from prometheus_client import Counter

-from twisted.internet import defer
-
 from synapse.api.constants import EventTypes, ThirdPartyEntityKind
 from synapse.api.errors import CodeMessageException
 from synapse.events.utils import serialize_event
 from synapse.http.client import SimpleHttpClient
-from synapse.types import ThirdPartyInstanceID
+from synapse.types import JsonDict, ThirdPartyInstanceID
 from synapse.util.caches.response_cache import ResponseCache

+if TYPE_CHECKING:
+    from synapse.appservice import ApplicationService
+
 logger = logging.getLogger(__name__)

 sent_transactions_counter = Counter(
@@ -163,19 +165,20 @@ class ApplicationServiceApi(SimpleHttpClient):
            logger.warning("query_3pe to %s threw exception %s", uri, ex)
            return []

-    def get_3pe_protocol(self, service, protocol):
+    async def get_3pe_protocol(
+        self, service: "ApplicationService", protocol: str
+    ) -> Optional[JsonDict]:
        if service.url is None:
            return {}

-        @defer.inlineCallbacks
-        def _get():
+        async def _get() -> Optional[JsonDict]:
            uri = "%s%s/thirdparty/protocol/%s" % (
                service.url,
                APP_SERVICE_PREFIX,
                urllib.parse.quote(protocol),
            )
            try:
-                info = yield defer.ensureDeferred(self.get_json(uri, {}))
+                info = await self.get_json(uri, {})

                if not _is_valid_3pe_metadata(info):
                    logger.warning(
@@ -196,7 +199,7 @@ class ApplicationServiceApi(SimpleHttpClient):
                return None

        key = (service.id, protocol)
-        return self.protocol_meta_cache.wrap(key, _get)
+        return await self.protocol_meta_cache.wrap(key, _get)

    async def push_bulk(self, service, events, txn_id=None):
        if service.url is None:
@@ -57,7 +57,7 @@ from synapse.metrics.background_process_metrics import run_as_background_process
 logger = logging.getLogger(__name__)


-class ApplicationServiceScheduler(object):
+class ApplicationServiceScheduler:
    """ Public facing API for this module. Does the required DI to tie the
    components together. This also serves as the "event_pool", which in this
    case is a simple array.
@@ -86,7 +86,7 @@ class ApplicationServiceScheduler(object):
        self.queuer.enqueue(service, event)


-class _ServiceQueuer(object):
+class _ServiceQueuer:
    """Queue of events waiting to be sent to appservices.

    Groups events into transactions per-appservice, and sends them on to the
@@ -133,7 +133,7 @@ class _ServiceQueuer(object):
            self.requests_in_flight.discard(service.id)


-class _TransactionController(object):
+class _TransactionController:
    """Transaction manager.

    Builds AppServiceTransactions and runs their lifecycle. Also starts a Recoverer
@@ -209,7 +209,7 @@ class _TransactionController(object):
        return state == ApplicationServiceState.UP or state is None


-class _Recoverer(object):
+class _Recoverer:
    """Manages retries and backoff for a DOWN appservice.

    We have one of these for each appservice which is currently considered DOWN.
@@ -18,12 +18,16 @@
 import argparse
 import errno
 import os
+import time
+import urllib.parse
 from collections import OrderedDict
 from hashlib import sha256
 from textwrap import dedent
-from typing import Any, List, MutableMapping, Optional
+from typing import Any, Callable, List, MutableMapping, Optional

 import attr
+import jinja2
+import pkg_resources
 import yaml


@@ -84,7 +88,7 @@ def path_exists(file_path):
        return False


-class Config(object):
+class Config:
    """
    A configuration section, containing configuration keys and values.

@@ -100,6 +104,11 @@ class Config(object):
    def __init__(self, root_config=None):
        self.root = root_config

+        # Get the path to the default Synapse template directory
+        self.default_template_dir = pkg_resources.resource_filename(
+            "synapse", "res/templates"
+        )
+
    def __getattr__(self, item: str) -> Any:
        """
        Try and fetch a configuration option that does not exist on this class.
@@ -184,8 +193,103 @@ class Config(object):
        with open(file_path) as file_stream:
            return file_stream.read()

+    def read_templates(
+        self,
+        filenames: List[str],
+        custom_template_directory: Optional[str] = None,
+        autoescape: bool = False,
+    ) -> List[jinja2.Template]:
+        """Load a list of template files from disk using the given variables.

-class RootConfig(object):
+        This function will attempt to load the given templates from the default Synapse
+        template directory. If `custom_template_directory` is supplied, that directory
+        is tried first.
+
+        Files read are treated as Jinja templates. These templates are not rendered yet.
+
+        Args:
+            filenames: A list of template filenames to read.
+
+            custom_template_directory: A directory to try to look for the templates
+                before using the default Synapse template directory instead.
+
+            autoescape: Whether to autoescape variables before inserting them into the
+                template.
+
+        Raises:
+            ConfigError: if the file's path is incorrect or otherwise cannot be read.
+
+        Returns:
+            A list of jinja2 templates.
+        """
+        templates = []
+        search_directories = [self.default_template_dir]
+
+        # The loader will first look in the custom template directory (if specified) for the
+        # given filename. If it doesn't find it, it will use the default template dir instead
+        if custom_template_directory:
+            # Check that the given template directory exists
+            if not self.path_exists(custom_template_directory):
+                raise ConfigError(
+                    "Configured template directory does not exist: %s"
+                    % (custom_template_directory,)
+                )
+
+            # Search the custom template directory as well
+            search_directories.insert(0, custom_template_directory)
+
+        loader = jinja2.FileSystemLoader(search_directories)
+        env = jinja2.Environment(loader=loader, autoescape=autoescape)
+
+        # Update the environment with our custom filters
+        env.filters.update(
+            {
+                "format_ts": _format_ts_filter,
+                "mxc_to_http": _create_mxc_to_http_filter(self.public_baseurl),
+            }
+        )
+
+        for filename in filenames:
+            # Load the template
+            template = env.get_template(filename)
+            templates.append(template)
+
+        return templates
+
+
+def _format_ts_filter(value: int, format: str):
+    return time.strftime(format, time.localtime(value / 1000))
+
+
+def _create_mxc_to_http_filter(public_baseurl: str) -> Callable:
+    """Create and return a jinja2 filter that converts MXC urls to HTTP
+
+    Args:
+        public_baseurl: The public, accessible base URL of the homeserver
+    """
+
+    def mxc_to_http_filter(value, width, height, resize_method="crop"):
+        if value[0:6] != "mxc://":
+            return ""
+
+        server_and_media_id = value[6:]
+        fragment = None
+        if "#" in server_and_media_id:
+            server_and_media_id, fragment = server_and_media_id.split("#", 1)
+            fragment = "#" + fragment
+
+        params = {"width": width, "height": height, "method": resize_method}
+        return "%s_matrix/media/v1/thumbnail/%s?%s%s" % (
+            public_baseurl,
+            server_and_media_id,
+            urllib.parse.urlencode(params),
+            fragment or "",
+        )
+
+    return mxc_to_http_filter
+
+
+class RootConfig:
    """
    Holder of an application's configuration.

@@ -33,7 +33,7 @@ _DEFAULT_FACTOR_SIZE = 0.5
 _DEFAULT_EVENT_CACHE_SIZE = "10K"


-class CacheProperties(object):
+class CacheProperties:
    def __init__(self):
        # The default factor size for all caches
        self.default_factor_size = float(
@@ -23,7 +23,6 @@ from enum import Enum
 from typing import Optional

 import attr
-import pkg_resources

 from ._base import Config, ConfigError

@@ -98,21 +97,18 @@ class EmailConfig(Config):
            if parsed[1] == "":
                raise RuntimeError("Invalid notif_from address")

+        # A user-configurable template directory
        template_dir = email_config.get("template_dir")
-        # we need an absolute path, because we change directory after starting (and
-        # we don't yet know what auxiliary templates like mail.css we will need).
-        # (Note that loading as package_resources with jinja.PackageLoader doesn't
-        # work for the same reason.)
-        if not template_dir:
-            template_dir = pkg_resources.resource_filename("synapse", "res/templates")
-
-        self.email_template_dir = os.path.abspath(template_dir)
+        if isinstance(template_dir, str):
+            # We need an absolute path, because we change directory after starting (and
+            # we don't yet know what auxiliary templates like mail.css we will need).
+            template_dir = os.path.abspath(template_dir)
+        elif template_dir is not None:
+            # If template_dir is something other than a str or None, warn the user
+            raise ConfigError("Config option email.template_dir must be type str")

        self.email_enable_notifs = email_config.get("enable_notifs", False)

-        account_validity_config = config.get("account_validity") or {}
-        account_validity_renewal_enabled = account_validity_config.get("renew_at")
-
        self.threepid_behaviour_email = (
            # Have Synapse handle the email sending if account_threepid_delegates.email
            # is not defined
@@ -166,19 +162,6 @@ class EmailConfig(Config):
            email_config.get("validation_token_lifetime", "1h")
        )

-        if (
-            self.email_enable_notifs
-            or account_validity_renewal_enabled
-            or self.threepid_behaviour_email == ThreepidBehaviour.LOCAL
-        ):
-            # make sure we can import the required deps
-            import bleach
-            import jinja2
-
-            # prevent unused warnings
-            jinja2
-            bleach
-
        if self.threepid_behaviour_email == ThreepidBehaviour.LOCAL:
            missing = []
            if not self.email_notif_from:
@@ -196,49 +179,49 @@ class EmailConfig(Config):

            # These email templates have placeholders in them, and thus must be
            # parsed using a templating engine during a request
-            self.email_password_reset_template_html = email_config.get(
+            password_reset_template_html = email_config.get(
                "password_reset_template_html", "password_reset.html"
            )
-            self.email_password_reset_template_text = email_config.get(
+            password_reset_template_text = email_config.get(
                "password_reset_template_text", "password_reset.txt"
            )
-            self.email_registration_template_html = email_config.get(
+            registration_template_html = email_config.get(
                "registration_template_html", "registration.html"
            )
-            self.email_registration_template_text = email_config.get(
+            registration_template_text = email_config.get(
                "registration_template_text", "registration.txt"
            )
-            self.email_add_threepid_template_html = email_config.get(
+            add_threepid_template_html = email_config.get(
                "add_threepid_template_html", "add_threepid.html"
            )
-            self.email_add_threepid_template_text = email_config.get(
+            add_threepid_template_text = email_config.get(
                "add_threepid_template_text", "add_threepid.txt"
            )

-            self.email_password_reset_template_failure_html = email_config.get(
+            password_reset_template_failure_html = email_config.get(
                "password_reset_template_failure_html", "password_reset_failure.html"
            )
-            self.email_registration_template_failure_html = email_config.get(
+            registration_template_failure_html = email_config.get(
                "registration_template_failure_html", "registration_failure.html"
            )
-            self.email_add_threepid_template_failure_html = email_config.get(
+            add_threepid_template_failure_html = email_config.get(
                "add_threepid_template_failure_html", "add_threepid_failure.html"
            )

            # These templates do not support any placeholder variables, so we
            # will read them from disk once during setup
-            email_password_reset_template_success_html = email_config.get(
+            password_reset_template_success_html = email_config.get(
                "password_reset_template_success_html", "password_reset_success.html"
            )
-            email_registration_template_success_html = email_config.get(
+            registration_template_success_html = email_config.get(
                "registration_template_success_html", "registration_success.html"
            )
-            email_add_threepid_template_success_html = email_config.get(
+            add_threepid_template_success_html = email_config.get(
                "add_threepid_template_success_html", "add_threepid_success.html"
            )

-            # Check templates exist
-            for f in [
+            # Read all templates from disk
+            (
                self.email_password_reset_template_html,
                self.email_password_reset_template_text,
                self.email_registration_template_html,
@@ -248,32 +231,36 @@ class EmailConfig(Config):
                self.email_password_reset_template_failure_html,
                self.email_registration_template_failure_html,
                self.email_add_threepid_template_failure_html,
-                email_password_reset_template_success_html,
-                email_registration_template_success_html,
-                email_add_threepid_template_success_html,
-            ]:
-                p = os.path.join(self.email_template_dir, f)
-                if not os.path.isfile(p):
-                    raise ConfigError("Unable to find template file %s" % (p,))
+                password_reset_template_success_html_template,
+                registration_template_success_html_template,
+                add_threepid_template_success_html_template,
+            ) = self.read_templates(
+                [
+                    password_reset_template_html,
+                    password_reset_template_text,
+                    registration_template_html,
+                    registration_template_text,
+                    add_threepid_template_html,
+                    add_threepid_template_text,
+                    password_reset_template_failure_html,
+                    registration_template_failure_html,
+                    add_threepid_template_failure_html,
+                    password_reset_template_success_html,
+                    registration_template_success_html,
+                    add_threepid_template_success_html,
+                ],
+                template_dir,
+            )

-            # Retrieve content of web templates
-            filepath = os.path.join(
-                self.email_template_dir, email_password_reset_template_success_html
+            # Render templates that do not contain any placeholders
+            self.email_password_reset_template_success_html_content = (
+                password_reset_template_success_html_template.render()
            )
-            self.email_password_reset_template_success_html = self.read_file(
-                filepath, "email.password_reset_template_success_html"
+            self.email_registration_template_success_html_content = (
+                registration_template_success_html_template.render()
            )
-            filepath = os.path.join(
-                self.email_template_dir, email_registration_template_success_html
-            )
-            self.email_registration_template_success_html_content = self.read_file(
-                filepath, "email.registration_template_success_html"
-            )
-            filepath = os.path.join(
-                self.email_template_dir, email_add_threepid_template_success_html
-            )
-            self.email_add_threepid_template_success_html_content = self.read_file(
-                filepath, "email.add_threepid_template_success_html"
+            self.email_add_threepid_template_success_html_content = (
+                add_threepid_template_success_html_template.render()
            )

        if self.email_enable_notifs:
@@ -290,17 +277,19 @@ class EmailConfig(Config):
                    % (", ".join(missing),)
                )

-            self.email_notif_template_html = email_config.get(
+            notif_template_html = email_config.get(
                "notif_template_html", "notif_mail.html"
            )
-            self.email_notif_template_text = email_config.get(
+            notif_template_text = email_config.get(
                "notif_template_text", "notif_mail.txt"
            )

-            for f in self.email_notif_template_text, self.email_notif_template_html:
-                p = os.path.join(self.email_template_dir, f)
-                if not os.path.isfile(p):
-                    raise ConfigError("Unable to find email template file %s" % (p,))
+            (
+                self.email_notif_template_html,
+                self.email_notif_template_text,
+            ) = self.read_templates(
+                [notif_template_html, notif_template_text], template_dir,
+            )

            self.email_notif_for_new_users = email_config.get(
                "notif_for_new_users", True
@@ -309,18 +298,20 @@ class EmailConfig(Config):
                "client_base_url", email_config.get("riot_base_url", None)
            )

-        if account_validity_renewal_enabled:
-            self.email_expiry_template_html = email_config.get(
+        if self.account_validity.renew_by_email_enabled:
+            expiry_template_html = email_config.get(
                "expiry_template_html", "notice_expiry.html"
            )
-            self.email_expiry_template_text = email_config.get(
+            expiry_template_text = email_config.get(
                "expiry_template_text", "notice_expiry.txt"
            )

-            for f in self.email_expiry_template_text, self.email_expiry_template_html:
-                p = os.path.join(self.email_template_dir, f)
-                if not os.path.isfile(p):
-                    raise ConfigError("Unable to find email template file %s" % (p,))
+            (
+                self.account_validity_template_html,
+                self.account_validity_template_text,
+            ) = self.read_templates(
+                [expiry_template_html, expiry_template_text], template_dir,
+            )

        subjects_config = email_config.get("subjects", {})
        subjects = {}
@@ -400,9 +391,7 @@ class EmailConfig(Config):
          # Directory in which Synapse will try to find the template files below.
          # If not set, default templates from within the Synapse package will be used.
          #
-          # DO NOT UNCOMMENT THIS SETTING unless you want to customise the templates.
-          # If you *do* uncomment it, you will need to make sure that all the templates
-          # below are in the directory.
+          # Do not uncomment this setting unless you want to customise the templates.
          #
          # Synapse will look for the following templates in this directory:
          #
@@ -82,7 +82,7 @@ logger = logging.getLogger(__name__)


@attr.s
-class TrustedKeyServer(object):
+class TrustedKeyServer:
    # string: name of the server.
    server_name = attr.ib()

@@ -22,7 +22,7 @@ from ._base import Config, ConfigError


@attr.s
-class MetricsFlags(object):
+class MetricsFlags:
    known_servers = attr.ib(default=False, validator=attr.validators.instance_of(bool))

    @classmethod
@@ -17,7 +17,7 @@ from typing import Dict
 from ._base import Config


-class RateLimitConfig(object):
+class RateLimitConfig:
    def __init__(
        self,
        config: Dict[str, float],
@@ -27,7 +27,7 @@ class RateLimitConfig(object):
        self.burst_count = config.get("burst_count", defaults["burst_count"])


-class FederationRateLimitConfig(object):
+class FederationRateLimitConfig:
    _items_and_default = {
        "window_size": 1000,
        "sleep_limit": 10,
@@ -100,6 +100,9 @@ class RegistrationConfig(Config):

        self.registrations_require_3pid = config.get("registrations_require_3pid", [])
        self.allowed_local_3pids = config.get("allowed_local_3pids", [])
+        self.check_is_for_allowed_local_3pids = config.get(
+            "check_is_for_allowed_local_3pids", None
+        )
        self.enable_3pid_lookup = config.get("enable_3pid_lookup", True)
        self.registration_shared_secret = config.get("registration_shared_secret")

@@ -294,6 +297,11 @@ class RegistrationConfig(Config):
        # Mandate that users are only allowed to associate certain formats of
        # 3PIDs with accounts on this server.
        #
+        # Use an Identity Server to establish which 3PIDs are allowed to register?
+        # Overrides allowed_local_3pids below.
+        #
+        #check_is_for_allowed_local_3pids: matrix.org
+        #
        #allowed_local_3pids:
        #  - medium: email
        #    pattern: '.*@matrix\\.org'
@@ -22,7 +22,7 @@ from ._base import Config, ConfigError
 logger = logging.Logger(__name__)


-class RoomDefaultEncryptionTypes(object):
+class RoomDefaultEncryptionTypes:
    """Possible values for the encryption_enabled_by_default_for_room_type config option"""

    ALL = "all"
@@ -149,7 +149,7 @@ class RoomDirectoryConfig(Config):
        return False


-class _RoomDirectoryRule(object):
+class _RoomDirectoryRule:
    """Helper class to test whether a room directory action is allowed, like
    creating an alias or publishing a room.
    """
@@ -18,8 +18,6 @@ import logging
 from typing import Any, List

 import attr
-import jinja2
-import pkg_resources

 from synapse.python_dependencies import DependencyException, check_requirements
 from synapse.util.module_loader import load_module, load_python_module
@@ -171,15 +169,11 @@ class SAML2Config(Config):
            saml2_config.get("saml_session_lifetime", "15m")
        )

-        template_dir = saml2_config.get("template_dir")
-        if not template_dir:
-            template_dir = pkg_resources.resource_filename("synapse", "res/templates",)
-
-        loader = jinja2.FileSystemLoader(template_dir)
-        # enable auto-escape here, to having to remember to escape manually in the
-        # template
-        env = jinja2.Environment(loader=loader, autoescape=True)
-        self.saml2_error_html_template = env.get_template("saml_error.html")
+        # We enable autoescape here as the message may potentially come from a
+        # remote resource
+        self.saml2_error_html_template = self.read_templates(
+            ["saml_error.html"], saml2_config.get("template_dir"), autoescape=True
+        )[0]

    def _default_saml_config_dict(
        self, required_attributes: set, optional_attributes: set
@@ -26,7 +26,6 @@ import yaml

 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS
 from synapse.http.endpoint import parse_and_validate_server_name
-from synapse.python_dependencies import DependencyException, check_requirements

 from ._base import Config, ConfigError

@@ -425,7 +424,7 @@ class ServerConfig(Config):
        self.gc_thresholds = read_gc_thresholds(config.get("gc_thresholds", None))

        @attr.s
-        class LimitRemoteRoomsConfig(object):
+        class LimitRemoteRoomsConfig:
            enabled = attr.ib(
                validator=attr.validators.instance_of(bool), default=False
            )
@@ -508,8 +507,6 @@ class ServerConfig(Config):
                )
            )

-        _check_resource_config(self.listeners)
-
        self.cleanup_extremities_with_dummy_events = config.get(
            "cleanup_extremities_with_dummy_events", True
        )
@@ -964,11 +961,10 @@ class ServerConfig(Config):
          #  min_lifetime: 1d
          #  max_lifetime: 1y

-          # Retention policy limits. If set, a user won't be able to send a
-          # 'm.room.retention' event which features a 'min_lifetime' or a 'max_lifetime'
-          # that's not within this range. This is especially useful in closed federations,
-          # in which server admins can make sure every federating server applies the same
-          # rules.
+          # Retention policy limits. If set, and the state of a room contains a
+          # 'm.room.retention' event in its state which contains a 'min_lifetime' or a
+          # 'max_lifetime' that's out of these bounds, Synapse will cap the room's policy
+          # to these limits when running purge jobs.
          #
          #allowed_lifetime_min: 1d
          #allowed_lifetime_max: 1y
@@ -994,12 +990,19 @@ class ServerConfig(Config):
          # (e.g. every 12h), but not want that purge to be performed by a job that's
          # iterating over every room it knows, which could be heavy on the server.
          #
+          # If any purge job is configured, it is strongly recommended to have at least
+          # a single job with neither 'shortest_max_lifetime' nor 'longest_max_lifetime'
+          # set, or one job without 'shortest_max_lifetime' and one job without
+          # 'longest_max_lifetime' set. Otherwise some rooms might be ignored, even if
+          # 'allowed_lifetime_min' and 'allowed_lifetime_max' are set, because capping a
+          # room's policy to these values is done after the policies are retrieved from
+          # Synapse's database (which is done using the range specified in a purge job's
+          # configuration).
+          #
          #purge_jobs:
-          #  - shortest_max_lifetime: 1d
-          #    longest_max_lifetime: 3d
+          #  - longest_max_lifetime: 3d
          #    interval: 12h
          #  - shortest_max_lifetime: 3d
-          #    longest_max_lifetime: 1y
          #    interval: 1d

        # Inhibits the /requestToken endpoints from returning an error that might leak
@@ -1133,20 +1136,3 @@ def _warn_if_webclient_configured(listeners: Iterable[ListenerConfig]) -> None:
                if name == "webclient":
                    logger.warning(NO_MORE_WEB_CLIENT_WARNING)
                    return
-
-
-def _check_resource_config(listeners: Iterable[ListenerConfig]) -> None:
-    resource_names = {
-        res_name
-        for listener in listeners
-        if listener.http_options
-        for res in listener.http_options.resources
-        for res_name in res.names
-    }
-
-    for resource in resource_names:
-        if resource == "consent":
-            try:
-                check_requirements("resources.consent")
-            except DependencyException as e:
-                raise ConfigError(e.message)
@@ -12,11 +12,8 @@
 # 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.
-import os
 from typing import Any, Dict

-import pkg_resources
-
 from ._base import Config


@@ -29,22 +26,32 @@ class SSOConfig(Config):
    def read_config(self, config, **kwargs):
        sso_config = config.get("sso") or {}  # type: Dict[str, Any]

-        # Pick a template directory in order of:
-        # * The sso-specific template_dir
-        # * /path/to/synapse/install/res/templates
+        # The sso-specific template_dir
        template_dir = sso_config.get("template_dir")
-        if not template_dir:
-            template_dir = pkg_resources.resource_filename("synapse", "res/templates",)

-        self.sso_template_dir = template_dir
-        self.sso_account_deactivated_template = self.read_file(
-            os.path.join(self.sso_template_dir, "sso_account_deactivated.html"),
-            "sso_account_deactivated_template",
+        # Read templates from disk
+        (
+            self.sso_redirect_confirm_template,
+            self.sso_auth_confirm_template,
+            self.sso_error_template,
+            sso_account_deactivated_template,
+            sso_auth_success_template,
+        ) = self.read_templates(
+            [
+                "sso_redirect_confirm.html",
+                "sso_auth_confirm.html",
+                "sso_error.html",
+                "sso_account_deactivated.html",
+                "sso_auth_success.html",
+            ],
+            template_dir,
        )
-        self.sso_auth_success_template = self.read_file(
-            os.path.join(self.sso_template_dir, "sso_auth_success.html"),
-            "sso_auth_success_template",
+
+        # These templates have no placeholders, so render them here
+        self.sso_account_deactivated_template = (
+            sso_account_deactivated_template.render()
        )
+        self.sso_auth_success_template = sso_auth_success_template.render()

        self.sso_client_whitelist = sso_config.get("client_whitelist") or []

@@ -83,7 +83,7 @@ class ServerContextFactory(ContextFactory):


@implementer(IPolicyForHTTPS)
-class FederationPolicyForHTTPS(object):
+class FederationPolicyForHTTPS:
    """Factory for Twisted SSLClientConnectionCreators that are used to make connections
    to remote servers for federation.

@@ -152,7 +152,7 @@ class FederationPolicyForHTTPS(object):


@implementer(IPolicyForHTTPS)
-class RegularPolicyForHTTPS(object):
+class RegularPolicyForHTTPS:
    """Factory for Twisted SSLClientConnectionCreators that are used to make connections
    to remote servers, for other than federation.

@@ -189,7 +189,7 @@ def _context_info_cb(ssl_connection, where, ret):


@implementer(IOpenSSLClientConnectionCreator)
-class SSLClientConnectionCreator(object):
+class SSLClientConnectionCreator:
    """Creates openssl connection objects for client connections.

    Replaces twisted.internet.ssl.ClientTLSOptions
@@ -214,7 +214,7 @@ class SSLClientConnectionCreator(object):
        return connection


-class ConnectionVerifier(object):
+class ConnectionVerifier:
    """Set the SNI, and do cert verification

    This is a thing which is attached to the TLSMemoryBIOProtocol, and is called by
@@ -57,7 +57,7 @@ logger = logging.getLogger(__name__)


@attr.s(slots=True, cmp=False)
-class VerifyJsonRequest(object):
+class VerifyJsonRequest:
    """
    A request to verify a JSON object.

@@ -96,7 +96,7 @@ class KeyLookupError(ValueError):
    pass


-class Keyring(object):
+class Keyring:
    def __init__(self, hs, key_fetchers=None):
        self.clock = hs.get_clock()

@@ -420,7 +420,7 @@ class Keyring(object):
        remaining_requests.difference_update(completed)


-class KeyFetcher(object):
+class KeyFetcher:
    async def get_keys(self, keys_to_fetch):
        """
        Args:
@@ -456,7 +456,7 @@ class StoreKeyFetcher(KeyFetcher):
        return keys


-class BaseV2KeyFetcher(object):
+class BaseV2KeyFetcher:
    def __init__(self, hs):
        self.store = hs.get_datastore()
        self.config = hs.get_config()
@@ -757,9 +757,8 @@ class ServerKeyFetcher(BaseV2KeyFetcher):
            except Exception:
                logger.exception("Error getting keys %s from %s", key_ids, server_name)

-        return await yieldable_gather_results(
-            get_key, keys_to_fetch.items()
-        ).addCallback(lambda _: results)
+        await yieldable_gather_results(get_key, keys_to_fetch.items())
+        return results

    async def get_server_verify_key_v2_direct(self, server_name, key_ids):
        """
@@ -769,7 +768,7 @@ class ServerKeyFetcher(BaseV2KeyFetcher):
            key_ids (iterable[str]):

        Returns:
-            Deferred[dict[str, FetchKeyResult]]: map from key ID to lookup result
+            dict[str, FetchKeyResult]: map from key ID to lookup result

        Raises:
            KeyLookupError if there was a problem making the lookup
@@ -47,7 +47,7 @@ def check(
    Args:
        room_version_obj: the version of the room
        event: the event being checked.
-        auth_events (dict: event-key -> event): the existing room state.
+        auth_events: the existing room state.

    Raises:
        AuthError if the checks fail
@@ -18,7 +18,7 @@
 import abc
 import os
 from distutils.util import strtobool
-from typing import Dict, Optional, Type
+from typing import Dict, Optional, Tuple, Type

 from unpaddedbase64 import encode_base64

@@ -96,7 +96,7 @@ class DefaultDictProperty(DictProperty):
        return instance._dict.get(self.key, self.default)


-class _EventInternalMetadata(object):
+class _EventInternalMetadata:
    __slots__ = ["_dict"]

    def __init__(self, internal_metadata_dict: JsonDict):
@@ -120,7 +120,7 @@ class _EventInternalMetadata(object):
    # be here
    before = DictProperty("before")  # type: str
    after = DictProperty("after")  # type: str
-    order = DictProperty("order")  # type: int
+    order = DictProperty("order")  # type: Tuple[int, int]

    def get_dict(self) -> JsonDict:
        return dict(self._dict)
@@ -133,6 +133,8 @@ class _EventInternalMetadata(object):
        rejection. This is needed as those events are marked as outliers, but
        they still need to be processed as if they're new events (e.g. updating
        invite state in the database, relaying to clients, etc).
+
+        (Added in synapse 0.99.0, so may be unreliable for events received before that)
        """
        return self._dict.get("out_of_band_membership", False)

@@ -12,7 +12,7 @@
 # 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 typing import Optional
+from typing import Any, Dict, List, Optional, Tuple, Union

 import attr
 from nacl.signing import SigningKey
@@ -36,7 +36,7 @@ from synapse.util.stringutils import random_string


@attr.s(slots=True, cmp=False, frozen=True)
-class EventBuilder(object):
+class EventBuilder:
    """A format independent event builder used to build up the event content
    before signing the event.

@@ -97,14 +97,14 @@ class EventBuilder(object):
    def is_state(self):
        return self._state_key is not None

-    async def build(self, prev_event_ids):
+    async def build(self, prev_event_ids: List[str]) -> EventBase:
        """Transform into a fully signed and hashed event

        Args:
-            prev_event_ids (list[str]): The event IDs to use as the prev events
+            prev_event_ids: The event IDs to use as the prev events

        Returns:
-            FrozenEvent
+            The signed and hashed event.
        """

        state_ids = await self._state.get_current_state_ids(
@@ -114,8 +114,13 @@ class EventBuilder(object):

        format_version = self.room_version.event_format
        if format_version == EventFormatVersions.V1:
-            auth_events = await self._store.add_event_hashes(auth_ids)
-            prev_events = await self._store.add_event_hashes(prev_event_ids)
+            # The types of auth/prev events changes between event versions.
+            auth_events = await self._store.add_event_hashes(
+                auth_ids
+            )  # type: Union[List[str], List[Tuple[str, Dict[str, str]]]]
+            prev_events = await self._store.add_event_hashes(
+                prev_event_ids
+            )  # type: Union[List[str], List[Tuple[str, Dict[str, str]]]]
        else:
            auth_events = auth_ids
            prev_events = prev_event_ids
@@ -138,7 +143,7 @@ class EventBuilder(object):
            "unsigned": self.unsigned,
            "depth": depth,
            "prev_state": [],
-        }
+        }  # type: Dict[str, Any]

        if self.is_state():
            event_dict["state_key"] = self._state_key
@@ -159,7 +164,7 @@ class EventBuilder(object):
        )


-class EventBuilderFactory(object):
+class EventBuilderFactory:
    def __init__(self, hs):
        self.clock = hs.get_clock()
        self.hostname = hs.hostname
@@ -15,16 +15,17 @@
 # limitations under the License.

 import inspect
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional, Tuple

-from synapse.spam_checker_api import SpamCheckerApi
+from synapse.spam_checker_api import RegistrationBehaviour, SpamCheckerApi
+from synapse.types import Collection

 MYPY = False
 if MYPY:
    import synapse.server


-class SpamChecker(object):
+class SpamChecker:
    def __init__(self, hs: "synapse.server.HomeServer"):
        self.spam_checkers = []  # type: List[Any]

@@ -160,3 +161,33 @@ class SpamChecker(object):
                    return True

        return False
+
+    def check_registration_for_spam(
+        self,
+        email_threepid: Optional[dict],
+        username: Optional[str],
+        request_info: Collection[Tuple[str, str]],
+    ) -> RegistrationBehaviour:
+        """Checks if we should allow the given registration request.
+
+        Args:
+            email_threepid: The email threepid used for registering, if any
+            username: The request user name, if any
+            request_info: List of tuples of user agent and IP that
+                were used during the registration process.
+
+        Returns:
+            Enum for how the request should be handled
+        """
+
+        for spam_checker in self.spam_checkers:
+            # For backwards compatibility, only run if the method exists on the
+            # spam checker
+            checker = getattr(spam_checker, "check_registration_for_spam", None)
+            if checker:
+                behaviour = checker(email_threepid, username, request_info)
+                assert isinstance(behaviour, RegistrationBehaviour)
+                if behaviour != RegistrationBehaviour.ALLOW:
+                    return behaviour
+
+        return RegistrationBehaviour.ALLOW
@@ -18,7 +18,7 @@ from synapse.events.snapshot import EventContext
 from synapse.types import Requester


-class ThirdPartyEventRules(object):
+class ThirdPartyEventRules:
    """Allows server admins to provide a Python module implementing an extra
    set of rules to apply when processing events.

@@ -322,7 +322,7 @@ def serialize_event(
    return d


-class EventClientSerializer(object):
+class EventClientSerializer:
    """Serializes events that are to be sent to clients.

    This is used for bundling extra information with any events to be sent to
@@ -20,7 +20,7 @@ from synapse.events.utils import validate_canonicaljson
 from synapse.types import EventID, RoomID, UserID


-class EventValidator(object):
+class EventValidator:
    def validate_new(self, event, config):
        """Validates the event has roughly the right format

@@ -74,15 +74,14 @@ class EventValidator(object):
                        )

        if event.type == EventTypes.Retention:
-            self._validate_retention(event, config)
+            self._validate_retention(event)

-    def _validate_retention(self, event, config):
+    def _validate_retention(self, event):
        """Checks that an event that defines the retention policy for a room respects the
-        boundaries imposed by the server's administrator.
+        format enforced by the spec.

        Args:
            event (FrozenEvent): The event to validate.
-            config (Config): The homeserver's configuration.
        """
        min_lifetime = event.content.get("min_lifetime")
        max_lifetime = event.content.get("max_lifetime")
@@ -95,32 +94,6 @@ class EventValidator(object):
                    errcode=Codes.BAD_JSON,
                )

-            if (
-                config.retention_allowed_lifetime_min is not None
-                and min_lifetime < config.retention_allowed_lifetime_min
-            ):
-                raise SynapseError(
-                    code=400,
-                    msg=(
-                        "'min_lifetime' can't be lower than the minimum allowed"
-                        " value enforced by the server's administrator"
-                    ),
-                    errcode=Codes.BAD_JSON,
-                )
-
-            if (
-                config.retention_allowed_lifetime_max is not None
-                and min_lifetime > config.retention_allowed_lifetime_max
-            ):
-                raise SynapseError(
-                    code=400,
-                    msg=(
-                        "'min_lifetime' can't be greater than the maximum allowed"
-                        " value enforced by the server's administrator"
-                    ),
-                    errcode=Codes.BAD_JSON,
-                )
-
        if max_lifetime is not None:
            if not isinstance(max_lifetime, int):
                raise SynapseError(
@@ -129,32 +102,6 @@ class EventValidator(object):
                    errcode=Codes.BAD_JSON,
                )

-            if (
-                config.retention_allowed_lifetime_min is not None
-                and max_lifetime < config.retention_allowed_lifetime_min
-            ):
-                raise SynapseError(
-                    code=400,
-                    msg=(
-                        "'max_lifetime' can't be lower than the minimum allowed value"
-                        " enforced by the server's administrator"
-                    ),
-                    errcode=Codes.BAD_JSON,
-                )
-
-            if (
-                config.retention_allowed_lifetime_max is not None
-                and max_lifetime > config.retention_allowed_lifetime_max
-            ):
-                raise SynapseError(
-                    code=400,
-                    msg=(
-                        "'max_lifetime' can't be greater than the maximum allowed"
-                        " value enforced by the server's administrator"
-                    ),
-                    errcode=Codes.BAD_JSON,
-                )
-
        if (
            min_lifetime is not None
            and max_lifetime is not None
@@ -39,7 +39,7 @@ from synapse.types import JsonDict, get_domain_from_id
 logger = logging.getLogger(__name__)


-class FederationBase(object):
+class FederationBase:
    def __init__(self, hs):
        self.hs = hs

@@ -54,7 +54,7 @@ from synapse.events import EventBase, builder
 from synapse.federation.federation_base import FederationBase, event_from_pdu_json
 from synapse.logging.context import make_deferred_yieldable, preserve_fn
 from synapse.logging.utils import log_function
-from synapse.types import JsonDict
+from synapse.types import JsonDict, get_domain_from_id
 from synapse.util import unwrapFirstError
 from synapse.util.caches.expiringcache import ExpiringCache
 from synapse.util.retryutils import NotRetryingDestination
@@ -217,11 +217,9 @@ class FederationClient(FederationBase):
            for p in transaction_data["pdus"]
        ]

-        # FIXME: We should handle signature failures more gracefully.
-        pdus[:] = await make_deferred_yieldable(
-            defer.gatherResults(
-                self._check_sigs_and_hashes(room_version, pdus), consumeErrors=True,
-            ).addErrback(unwrapFirstError)
+        # Check signatures and hash of pdus, removing any from the list that fail checks
+        pdus[:] = await self._check_sigs_and_hash_and_fetch(
+            dest, pdus, outlier=True, room_version=room_version
        )

        return pdus
@@ -386,10 +384,11 @@ class FederationClient(FederationBase):
                    pdu.event_id, allow_rejected=True, allow_none=True
                )

-            if not res and pdu.origin != origin:
+            pdu_origin = get_domain_from_id(pdu.sender)
+            if not res and pdu_origin != origin:
                try:
                    res = await self.get_pdu(
-                        destinations=[pdu.origin],
+                        destinations=[pdu_origin],
                        event_id=pdu.event_id,
                        room_version=room_version,
                        outlier=outlier,
@@ -28,7 +28,6 @@ from typing import (
    Union,
 )

-from canonicaljson import json
 from prometheus_client import Counter, Histogram

 from twisted.internet import defer
@@ -63,7 +62,7 @@ from synapse.replication.http.federation import (
    ReplicationGetQueryRestServlet,
 )
 from synapse.types import JsonDict, get_domain_from_id
-from synapse.util import glob_to_regex, unwrapFirstError
+from synapse.util import glob_to_regex, json_decoder, unwrapFirstError
 from synapse.util.async_helpers import Linearizer, concurrently_execute
 from synapse.util.caches.response_cache import ResponseCache

@@ -98,10 +97,16 @@ class FederationServer(FederationBase):
        self.state = hs.get_state_handler()

        self.device_handler = hs.get_device_handler()
+        self._federation_ratelimiter = hs.get_federation_ratelimiter()

        self._server_linearizer = Linearizer("fed_server")
        self._transaction_linearizer = Linearizer("fed_txn_handler")

+        # We cache results for transaction with the same ID
+        self._transaction_resp_cache = ResponseCache(
+            hs, "fed_txn_handler", timeout_ms=30000
+        )
+
        self.transaction_actions = TransactionActions(self.store)

        self.registry = hs.get_federation_registry()
@@ -136,22 +141,44 @@ class FederationServer(FederationBase):
        request_time = self._clock.time_msec()

        transaction = Transaction(**transaction_data)
+        transaction_id = transaction.transaction_id  # type: ignore

-        if not transaction.transaction_id:  # type: ignore
+        if not transaction_id:
            raise Exception("Transaction missing transaction_id")

-        logger.debug("[%s] Got transaction", transaction.transaction_id)  # type: ignore
+        logger.debug("[%s] Got transaction", transaction_id)

-        # use a linearizer to ensure that we don't process the same transaction
-        # multiple times in parallel.
-        with (
-            await self._transaction_linearizer.queue(
-                (origin, transaction.transaction_id)  # type: ignore
-            )
-        ):
-            result = await self._handle_incoming_transaction(
-                origin, transaction, request_time
-            )
+        # We wrap in a ResponseCache so that we de-duplicate retried
+        # transactions.
+        return await self._transaction_resp_cache.wrap(
+            (origin, transaction_id),
+            self._on_incoming_transaction_inner,
+            origin,
+            transaction,
+            request_time,
+        )
+
+    async def _on_incoming_transaction_inner(
+        self, origin: str, transaction: Transaction, request_time: int
+    ) -> Tuple[int, Dict[str, Any]]:
+        # Use a linearizer to ensure that transactions from a remote are
+        # processed in order.
+        with await self._transaction_linearizer.queue(origin):
+            # We rate limit here *after* we've queued up the incoming requests,
+            # so that we don't fill up the ratelimiter with blocked requests.
+            #
+            # This is important as the ratelimiter allows N concurrent requests
+            # at a time, and only starts ratelimiting if there are more requests
+            # than that being processed at a time. If we queued up requests in
+            # the linearizer/response cache *after* the ratelimiting then those
+            # queued up requests would count as part of the allowed limit of N
+            # concurrent requests.
+            with self._federation_ratelimiter.ratelimit(origin) as d:
+                await d
+
+                result = await self._handle_incoming_transaction(
+                    origin, transaction, request_time
+                )

        return result

@@ -551,7 +578,7 @@ class FederationServer(FederationBase):
            for device_id, keys in device_keys.items():
                for key_id, json_str in keys.items():
                    json_result.setdefault(user_id, {})[device_id] = {
-                        key_id: json.loads(json_str)
+                        key_id: json_decoder.decode(json_str)
                    }

        logger.info(
@@ -786,7 +813,7 @@ def _acl_entry_matches(server_name: str, acl_entry: str) -> Match:
    return regex.match(server_name)


-class FederationHandlerRegistry(object):
+class FederationHandlerRegistry:
    """Allows classes to register themselves as handlers for a given EDU or
    query type for incoming federation traffic.
    """
@@ -20,13 +20,16 @@ These actions are mostly only used by the :py:mod:`.replication` module.
 """

 import logging
+from typing import Optional, Tuple

+from synapse.federation.units import Transaction
 from synapse.logging.utils import log_function
+from synapse.types import JsonDict

 logger = logging.getLogger(__name__)


-class TransactionActions(object):
+class TransactionActions:
    """ Defines persistence actions that relate to handling Transactions.
    """

@@ -34,30 +37,32 @@ class TransactionActions(object):
        self.store = datastore

    @log_function
-    def have_responded(self, origin, transaction):
-        """ Have we already responded to a transaction with the same id and
+    async def have_responded(
+        self, origin: str, transaction: Transaction
+    ) -> Optional[Tuple[int, JsonDict]]:
+        """Have we already responded to a transaction with the same id and
        origin?

        Returns:
-            Deferred: Results in `None` if we have not previously responded to
-            this transaction or a 2-tuple of `(int, dict)` representing the
-            response code and response body.
+            `None` if we have not previously responded to this transaction or a
+            2-tuple of `(int, dict)` representing the response code and response body.
        """
-        if not transaction.transaction_id:
+        transaction_id = transaction.transaction_id  # type: ignore
+        if not transaction_id:
            raise RuntimeError("Cannot persist a transaction with no transaction_id")

-        return self.store.get_received_txn_response(transaction.transaction_id, origin)
+        return await self.store.get_received_txn_response(transaction_id, origin)

    @log_function
-    def set_response(self, origin, transaction, code, response):
-        """ Persist how we responded to a transaction.
-
-        Returns:
-            Deferred
+    async def set_response(
+        self, origin: str, transaction: Transaction, code: int, response: JsonDict
+    ) -> None:
+        """Persist how we responded to a transaction.
        """
-        if not transaction.transaction_id:
+        transaction_id = transaction.transaction_id  # type: ignore
+        if not transaction_id:
            raise RuntimeError("Cannot persist a transaction with no transaction_id")

-        return self.store.set_received_txn_response(
-            transaction.transaction_id, origin, code, response
+        await self.store.set_received_txn_response(
+            transaction_id, origin, code, response
        )
@@ -37,8 +37,8 @@ from sortedcontainers import SortedDict

 from twisted.internet import defer

+from synapse.api.presence import UserPresenceState
 from synapse.metrics import LaterGauge
-from synapse.storage.presence import UserPresenceState
 from synapse.util.metrics import Measure

 from .units import Edu
@@ -46,7 +46,7 @@ from .units import Edu
 logger = logging.getLogger(__name__)


-class FederationRemoteSendQueue(object):
+class FederationRemoteSendQueue:
    """A drop in replacement for FederationSender"""

    def __init__(self, hs):
@@ -365,7 +365,7 @@ class FederationRemoteSendQueue(object):
        )


-class BaseFederationRow(object):
+class BaseFederationRow:
    """Base class for rows to be sent in the federation stream.

    Specifies how to identify, serialize and deserialize the different types.
@@ -22,6 +22,7 @@ from twisted.internet import defer

 import synapse
 import synapse.metrics
+from synapse.api.presence import UserPresenceState
 from synapse.events import EventBase
 from synapse.federation.sender.per_destination_queue import PerDestinationQueue
 from synapse.federation.sender.transaction_manager import TransactionManager
@@ -39,7 +40,6 @@ from synapse.metrics import (
    events_processed_counter,
 )
 from synapse.metrics.background_process_metrics import run_as_background_process
-from synapse.storage.presence import UserPresenceState
 from synapse.types import ReadReceipt
 from synapse.util.metrics import Measure, measure_func

@@ -56,7 +56,7 @@ sent_pdus_destination_dist_total = Counter(
 )


-class FederationSender(object):
+class FederationSender:
    def __init__(self, hs: "synapse.server.HomeServer"):
        self.hs = hs
        self.server_name = hs.hostname
@@ -108,8 +108,6 @@ class FederationSender(object):
            ),
        )

-        self._order = 1
-
        self._is_processing = False
        self._last_poked_id = -1

@@ -272,9 +270,6 @@ class FederationSender(object):
        # a transaction in progress. If we do, stick it in the pending_pdus
        # table and we'll get back to it later.

-        order = self._order
-        self._order += 1
-
        destinations = set(destinations)
        destinations.discard(self.server_name)
        logger.debug("Sending to: %s", str(destinations))
@@ -286,7 +281,7 @@ class FederationSender(object):
        sent_pdus_destination_dist_count.inc()

        for destination in destinations:
-            self._get_per_destination_queue(destination).send_pdu(pdu, order)
+            self._get_per_destination_queue(destination).send_pdu(pdu)

    async def send_read_receipt(self, receipt: ReadReceipt) -> None:
        """Send a RR to any other servers in the room
@@ -329,10 +324,10 @@ class FederationSender(object):
        room_id = receipt.room_id

        # Work out which remote servers should be poked and poke them.
-        domains = await self.state.get_current_hosts_in_room(room_id)
+        domains_set = await self.state.get_current_hosts_in_room(room_id)
        domains = [
            d
-            for d in domains
+            for d in domains_set
            if d != self.server_name
            and self._federation_shard_config.should_handle(self._instance_name, d)
        ]
@@ -24,12 +24,12 @@ from synapse.api.errors import (
    HttpResponseException,
    RequestSendFailed,
 )
+from synapse.api.presence import UserPresenceState
 from synapse.events import EventBase
 from synapse.federation.units import Edu
 from synapse.handlers.presence import format_user_presence_state
 from synapse.metrics import sent_transactions_counter
 from synapse.metrics.background_process_metrics import run_as_background_process
-from synapse.storage.presence import UserPresenceState
 from synapse.types import ReadReceipt
 from synapse.util.retryutils import NotRetryingDestination, get_retry_limiter

@@ -53,7 +53,7 @@ sent_edus_by_type = Counter(
 )


-class PerDestinationQueue(object):
+class PerDestinationQueue:
    """
    Manages the per-destination transmission queues.

@@ -92,8 +92,8 @@ class PerDestinationQueue(object):
        self._destination = destination
        self.transmission_loop_running = False

-        # a list of tuples of (pending pdu, order)
-        self._pending_pdus = []  # type: List[Tuple[EventBase, int]]
+        # a list of pending PDUs
+        self._pending_pdus = []  # type: List[EventBase]

        # XXX this is never actually used: see
        # https://github.com/matrix-org/synapse/issues/7549
@@ -132,14 +132,13 @@ class PerDestinationQueue(object):
            + len(self._pending_edus_keyed)
        )

-    def send_pdu(self, pdu: EventBase, order: int) -> None:
+    def send_pdu(self, pdu: EventBase) -> None:
        """Add a PDU to the queue, and start the transmission loop if necessary

        Args:
            pdu: pdu to send
-            order
        """
-        self._pending_pdus.append((pdu, order))
+        self._pending_pdus.append(pdu)
        self.attempt_new_transaction()

    def send_presence(self, states: Iterable[UserPresenceState]) -> None:
@@ -185,7 +184,7 @@ class PerDestinationQueue(object):
        returns immediately. Otherwise kicks off the process of sending a
        transaction in the background.
        """
-        # list of (pending_pdu, deferred, order)
+
        if self.transmission_loop_running:
            # XXX: this can get stuck on by a never-ending
            # request at which point pending_pdus just keeps growing.
@@ -210,7 +209,7 @@ class PerDestinationQueue(object):
        )

    async def _transaction_transmission_loop(self) -> None:
-        pending_pdus = []  # type: List[Tuple[EventBase, int]]
+        pending_pdus = []  # type: List[EventBase]
        try:
            self.transmission_loop_running = True

@@ -337,6 +336,28 @@ class PerDestinationQueue(object):
                    (e.retry_last_ts + e.retry_interval) / 1000.0
                ),
            )
+
+            if e.retry_interval > 60 * 60 * 1000:
+                # we won't retry for another hour!
+                # (this suggests a significant outage)
+                # We drop pending PDUs and EDUs because otherwise they will
+                # rack up indefinitely.
+                # Note that:
+                # - the EDUs that are being dropped here are those that we can
+                #   afford to drop (specifically, only typing notifications,
+                #   read receipts and presence updates are being dropped here)
+                # - Other EDUs such as to_device messages are queued with a
+                #   different mechanism
+                # - this is all volatile state that would be lost if the
+                #   federation sender restarted anyway
+
+                # dropping read receipts is a bit sad but should be solved
+                # through another mechanism, because this is all volatile!
+                self._pending_pdus = []
+                self._pending_edus = []
+                self._pending_edus_keyed = {}
+                self._pending_presence = {}
+                self._pending_rrs = {}
        except FederationDeniedError as e:
            logger.info(e)
        except HttpResponseException as e:
@@ -351,13 +372,13 @@ class PerDestinationQueue(object):
                "TX [%s] Failed to send transaction: %s", self._destination, e
            )

-            for p, _ in pending_pdus:
+            for p in pending_pdus:
                logger.info(
                    "Failed to send event %s to %s", p.event_id, self._destination
                )
        except Exception:
            logger.exception("TX [%s] Failed to send transaction", self._destination)
-            for p, _ in pending_pdus:
+            for p in pending_pdus:
                logger.info(
                    "Failed to send event %s to %s", p.event_id, self._destination
                )
@@ -13,9 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
-from typing import TYPE_CHECKING, List, Tuple
-
-from canonicaljson import json
+from typing import TYPE_CHECKING, List

 from synapse.api.errors import HttpResponseException
 from synapse.events import EventBase
@@ -28,6 +26,7 @@ from synapse.logging.opentracing import (
    tags,
    whitelisted_homeserver,
 )
+from synapse.util import json_decoder
 from synapse.util.metrics import measure_func

 if TYPE_CHECKING:
@@ -36,7 +35,7 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)


-class TransactionManager(object):
+class TransactionManager:
    """Helper class which handles building and sending transactions

    shared between PerDestinationQueue objects
@@ -54,11 +53,17 @@ class TransactionManager(object):

    @measure_func("_send_new_transaction")
    async def send_new_transaction(
-        self,
-        destination: str,
-        pending_pdus: List[Tuple[EventBase, int]],
-        pending_edus: List[Edu],
-    ):
+        self, destination: str, pdus: List[EventBase], edus: List[Edu],
+    ) -> bool:
+        """
+        Args:
+            destination: The destination to send to (e.g. 'example.org')
+            pdus: In-order list of PDUs to send
+            edus: List of EDUs to send
+
+        Returns:
+            True iff the transaction was successful
+        """

        # Make a transaction-sending opentracing span. This span follows on from
        # all the edus in that transaction. This needs to be done since there is
@@ -68,20 +73,14 @@ class TransactionManager(object):
        span_contexts = []
        keep_destination = whitelisted_homeserver(destination)

-        for edu in pending_edus:
+        for edu in edus:
            context = edu.get_context()
            if context:
-                span_contexts.append(extract_text_map(json.loads(context)))
+                span_contexts.append(extract_text_map(json_decoder.decode(context)))
            if keep_destination:
                edu.strip_context()

        with start_active_span_follows_from("send_transaction", span_contexts):
-
-            # Sort based on the order field
-            pending_pdus.sort(key=lambda t: t[1])
-            pdus = [x[0] for x in pending_pdus]
-            edus = pending_edus
-
            success = True

            logger.debug("TX [%s] _attempt_new_transaction", destination)
@@ -30,7 +30,7 @@ from synapse.logging.utils import log_function
 logger = logging.getLogger(__name__)


-class TransportLayerClient(object):
+class TransportLayerClient:
    """Sends federation HTTP requests to other servers"""

    def __init__(self, hs):
@@ -45,7 +45,6 @@ from synapse.logging.opentracing import (
 )
 from synapse.server import HomeServer
 from synapse.types import ThirdPartyInstanceID, get_domain_from_id
-from synapse.util.ratelimitutils import FederationRateLimiter
 from synapse.util.versionstring import get_version_string

 logger = logging.getLogger(__name__)
@@ -72,9 +71,7 @@ class TransportLayerServer(JsonResource):
        super(TransportLayerServer, self).__init__(hs, canonical_json=False)

        self.authenticator = Authenticator(hs)
-        self.ratelimiter = FederationRateLimiter(
-            self.clock, config=hs.config.rc_federation
-        )
+        self.ratelimiter = hs.get_federation_ratelimiter()

        self.register_servlets()

@@ -100,7 +97,7 @@ class NoAuthenticationError(AuthenticationError):
    pass


-class Authenticator(object):
+class Authenticator:
    def __init__(self, hs: HomeServer):
        self._clock = hs.get_clock()
        self.keyring = hs.get_keyring()
@@ -228,7 +225,7 @@ def _parse_auth_header(header_bytes):
        )


-class BaseFederationServlet(object):
+class BaseFederationServlet:
    """Abstract base class for federation servlet classes.

    The servlet object should have a PATH attribute which takes the form of a regexp to
@@ -272,6 +269,8 @@ class BaseFederationServlet(object):

    PREFIX = FEDERATION_V1_PREFIX  # Allows specifying the API version

+    RATELIMIT = True  # Whether to rate limit requests or not
+
    def __init__(self, handler, authenticator, ratelimiter, server_name):
        self.handler = handler
        self.authenticator = authenticator
@@ -335,7 +334,7 @@ class BaseFederationServlet(object):
                )

            with scope:
-                if origin:
+                if origin and self.RATELIMIT:
                    with ratelimiter.ratelimit(origin) as d:
                        await d
                        if request._disconnected:
@@ -372,6 +371,10 @@ class BaseFederationServlet(object):
 class FederationSendServlet(BaseFederationServlet):
    PATH = "/send/(?P<transaction_id>[^/]*)/?"

+    # We ratelimit manually in the handler as we queue up the requests and we
+    # don't want to fill up the ratelimiter with blocked requests.
+    RATELIMIT = False
+
    def __init__(self, handler, server_name, **kwargs):
        super(FederationSendServlet, self).__init__(
            handler, server_name=server_name, **kwargs
@@ -107,9 +107,7 @@ class Transaction(JsonEncodedObject):
        if "edus" in kwargs and not kwargs["edus"]:
            del kwargs["edus"]

-        super(Transaction, self).__init__(
-            transaction_id=transaction_id, pdus=pdus, **kwargs
-        )
+        super().__init__(transaction_id=transaction_id, pdus=pdus, **kwargs)

    @staticmethod
    def create_new(pdus, **kwargs):
@@ -60,7 +60,7 @@ DEFAULT_ATTESTATION_JITTER = (0.9, 1.3)
 UPDATE_ATTESTATION_TIME_MS = 1 * 24 * 60 * 60 * 1000


-class GroupAttestationSigning(object):
+class GroupAttestationSigning:
    """Creates and verifies group attestations.
    """

@@ -124,7 +124,7 @@ class GroupAttestationSigning(object):
        )


-class GroupAttestionRenewer(object):
+class GroupAttestionRenewer:
    """Responsible for sending and receiving attestation updates.
    """

@@ -32,7 +32,7 @@ logger = logging.getLogger(__name__)
 # TODO: Flairs


-class GroupsServerWorkerHandler(object):
+class GroupsServerWorkerHandler:
    def __init__(self, hs):
        self.hs = hs
        self.store = hs.get_datastore()
@@ -719,6 +719,27 @@ class GroupsServerHandler(GroupsServerWorkerHandler):

        raise NotImplementedError()

+    async def change_user_admin_in_group(
+        self, group_id, user_id, want_admin, requester_user_id, content
+    ):
+        """Promotes or demotes a user in a group.
+        """
+
+        await self.check_group_is_ours(group_id, requester_user_id, and_exists=True)
+
+        if requester_user_id == user_id:
+            raise SynapseError(400, "User cannot target themselves")
+
+        is_admin = await self.store.is_user_admin_in_group(
+            group_id, requester_user_id
+        )
+        if not is_admin:
+            raise SynapseError(403, "User is not admin in group")
+
+        await self.store.change_user_admin_in_group(group_id, user_id, want_admin)
+
+        return {}
+
    async def remove_user_from_group(
        self, group_id, user_id, requester_user_id, content
    ):
@@ -20,7 +20,7 @@ from .identity import IdentityHandler
 from .search import SearchHandler


-class Handlers(object):
+class Handlers:

    """ Deprecated. A collection of handlers.

@@ -25,7 +25,7 @@ from synapse.types import UserID
 logger = logging.getLogger(__name__)


-class BaseHandler(object):
+class BaseHandler:
    """
    Common base class for the event handlers.
    """
@@ -14,7 +14,7 @@
 # limitations under the License.


-class AccountDataEventSource(object):
+class AccountDataEventSource:
    def __init__(self, hs):
        self.store = hs.get_datastore()

@@ -26,15 +26,10 @@ from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.types import UserID
 from synapse.util import stringutils

-try:
-    from synapse.push.mailer import load_jinja2_templates
-except ImportError:
-    load_jinja2_templates = None
-
 logger = logging.getLogger(__name__)


-class AccountValidityHandler(object):
+class AccountValidityHandler:
    def __init__(self, hs):
        self.hs = hs
        self.config = hs.config
@@ -47,9 +42,11 @@ class AccountValidityHandler(object):
        if (
            self._account_validity.enabled
            and self._account_validity.renew_by_email_enabled
-            and load_jinja2_templates
        ):
            # Don't do email-specific configuration if renewal by email is disabled.
+            self._template_html = self.config.account_validity_template_html
+            self._template_text = self.config.account_validity_template_text
+
            try:
                app_name = self.hs.config.email_app_name

@@ -65,17 +62,6 @@ class AccountValidityHandler(object):

            self._raw_from = email.utils.parseaddr(self._from_string)[1]

-            self._template_html, self._template_text = load_jinja2_templates(
-                self.config.email_template_dir,
-                [
-                    self.config.email_expiry_template_html,
-                    self.config.email_expiry_template_text,
-                ],
-                apply_format_ts_filter=True,
-                apply_mxc_to_http_filter=True,
-                public_baseurl=self.config.public_baseurl,
-            )
-
            # Check the renewal emails to send and send them every 30min.
            def send_emails():
                # run as a background process to make sure that the database transactions
@@ -34,7 +34,7 @@ solutions, please read https://github.com/matrix-org/synapse/blob/master/docs/AC
 --------------------------------------------------------------------------------"""


-class AcmeHandler(object):
+class AcmeHandler:
    def __init__(self, hs):
        self.hs = hs
        self.reactor = hs.get_reactor()
@@ -78,7 +78,7 @@ def create_issuing_service(reactor, acme_url, account_key_file, well_known_resou

@attr.s
@implementer(ICertificateStore)
-class ErsatzStore(object):
+class ErsatzStore:
    """
    A store that only stores in memory.
    """
@@ -197,7 +197,7 @@ class AdminHandler(BaseHandler):
        return writer.finished()


-class ExfiltrationWriter(object):
+class ExfiltrationWriter:
    """Interface used to specify how to write exported data.
    """

@@ -34,7 +34,7 @@ logger = logging.getLogger(__name__)
 events_processed_counter = Counter("synapse_handlers_appservice_events_processed", "")


-class ApplicationServicesHandler(object):
+class ApplicationServicesHandler:
    def __init__(self, hs):
        self.store = hs.get_datastore()
        self.is_mine_id = hs.is_mine_id
@@ -42,9 +42,9 @@ from synapse.http.site import SynapseRequest
 from synapse.logging.context import defer_to_thread
 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.types import JsonDict, Requester, UserID
 from synapse.util import stringutils as stringutils
+from synapse.util.msisdn import phone_number_to_msisdn
 from synapse.util.threepids import canonicalise_email

 from ._base import BaseHandler
@@ -52,6 +52,91 @@ from ._base import BaseHandler
 logger = logging.getLogger(__name__)


+def convert_client_dict_legacy_fields_to_identifier(
+    submission: JsonDict,
+) -> Dict[str, str]:
+    """
+    Convert a legacy-formatted login submission to an identifier dict.
+
+    Legacy login submissions (used in both login and user-interactive authentication)
+    provide user-identifying information at the top-level instead.
+
+    These are now deprecated and replaced with identifiers:
+    https://matrix.org/docs/spec/client_server/r0.6.1#identifier-types
+
+    Args:
+        submission: The client dict to convert
+
+    Returns:
+        The matching identifier dict
+
+    Raises:
+        SynapseError: If the format of the client dict is invalid
+    """
+    identifier = submission.get("identifier", {})
+
+    # Generate an m.id.user identifier if "user" parameter is present
+    user = submission.get("user")
+    if user:
+        identifier = {"type": "m.id.user", "user": user}
+
+    # Generate an m.id.thirdparty identifier if "medium" and "address" parameters are present
+    medium = submission.get("medium")
+    address = submission.get("address")
+    if medium and address:
+        identifier = {
+            "type": "m.id.thirdparty",
+            "medium": medium,
+            "address": address,
+        }
+
+    # We've converted valid, legacy login submissions to an identifier. If the
+    # submission still doesn't have an identifier, it's invalid
+    if not identifier:
+        raise SynapseError(400, "Invalid login submission", Codes.INVALID_PARAM)
+
+    # Ensure the identifier has a type
+    if "type" not in identifier:
+        raise SynapseError(
+            400, "'identifier' dict has no key 'type'", errcode=Codes.MISSING_PARAM,
+        )
+
+    return identifier
+
+
+def login_id_phone_to_thirdparty(identifier: JsonDict) -> Dict[str, str]:
+    """
+    Convert a phone login identifier type to a generic threepid identifier.
+
+    Args:
+        identifier: Login identifier dict of type 'm.id.phone'
+
+    Returns:
+        An equivalent m.id.thirdparty identifier dict
+    """
+    if "country" not in identifier or (
+        # The specification requires a "phone" field, while Synapse used to require a "number"
+        # field. Accept both for backwards compatibility.
+        "phone" not in identifier
+        and "number" not in identifier
+    ):
+        raise SynapseError(
+            400, "Invalid phone-type identifier", errcode=Codes.INVALID_PARAM
+        )
+
+    # Accept both "phone" and "number" as valid keys in m.id.phone
+    phone_number = identifier.get("phone", identifier["number"])
+
+    # Convert user-provided phone number to a consistent representation
+    msisdn = phone_number_to_msisdn(identifier["country"], phone_number)
+
+    return {
+        "type": "m.id.thirdparty",
+        "medium": "msisdn",
+        "address": msisdn,
+    }
+
+
 class AuthHandler(BaseHandler):
    SESSION_EXPIRE_MS = 48 * 60 * 60 * 1000

@@ -132,18 +217,17 @@ class AuthHandler(BaseHandler):
        # after the SSO completes and before redirecting them back to their client.
        # It notifies the user they are about to give access to their matrix account
        # to the client.
-        self._sso_redirect_confirm_template = load_jinja2_templates(
-            hs.config.sso_template_dir, ["sso_redirect_confirm.html"],
-        )[0]
+        self._sso_redirect_confirm_template = hs.config.sso_redirect_confirm_template
+
        # The following template is shown during user interactive authentication
        # in the fallback auth scenario. It notifies the user that they are
        # authenticating for an operation to occur on their account.
-        self._sso_auth_confirm_template = load_jinja2_templates(
-            hs.config.sso_template_dir, ["sso_auth_confirm.html"],
-        )[0]
+        self._sso_auth_confirm_template = hs.config.sso_auth_confirm_template
+
        # The following template is shown after a successful user interactive
        # authentication session. It tells the user they can close the window.
        self._sso_auth_success_template = hs.config.sso_auth_success_template
+
        # The following template is shown during the SSO authentication process if
        # the account is deactivated.
        self._sso_account_deactivated_template = (
@@ -366,6 +450,14 @@ class AuthHandler(BaseHandler):
            # authentication flow.
            await self.store.set_ui_auth_clientdict(sid, clientdict)

+        user_agent = request.requestHeaders.getRawHeaders(b"User-Agent", default=[b""])[
+            0
+        ].decode("ascii", "surrogateescape")
+
+        await self.store.add_user_agent_ip_to_ui_auth_session(
+            session.session_id, user_agent, clientip
+        )
+
        if not authdict:
            raise InteractiveAuthIncompleteError(
                session.session_id, self._auth_dict_for_flows(flows, session.session_id)
@@ -1144,7 +1236,7 @@ class AuthHandler(BaseHandler):


@attr.s
-class MacaroonGenerator(object):
+class MacaroonGenerator:

    hs = attr.ib()

@@ -35,6 +35,7 @@ class CasHandler:
    """

    def __init__(self, hs):
+        self.hs = hs
        self._hostname = hs.hostname
        self._auth_handler = hs.get_auth_handler()
        self._registration_handler = hs.get_registration_handler()
@@ -210,8 +211,16 @@ class CasHandler:

        else:
            if not registered_user_id:
+                # Pull out the user-agent and IP from the request.
+                user_agent = request.requestHeaders.getRawHeaders(
+                    b"User-Agent", default=[b""]
+                )[0].decode("ascii", "surrogateescape")
+                ip_address = self.hs.get_ip_from_request(request)
+
                registered_user_id = await self._registration_handler.register_user(
-                    localpart=localpart, default_display_name=user_display_name
+                    localpart=localpart,
+                    default_display_name=user_display_name,
+                    user_agent_ips=(user_agent, ip_address),
                )

            await self._auth_handler.complete_sso_login(
@@ -234,7 +234,9 @@ class DeviceWorkerHandler(BaseHandler):
        return result

    async def on_federation_query_user_devices(self, user_id):
-        stream_id, devices = await self.store.get_devices_with_keys_by_user(user_id)
+        stream_id, devices = await self.store.get_e2e_device_keys_for_federation_query(
+            user_id
+        )
        master_key = await self.store.get_e2e_cross_signing_key(user_id, "master")
        self_signing_key = await self.store.get_e2e_cross_signing_key(
            user_id, "self_signing"
@@ -495,7 +497,7 @@ def _update_device_from_client_ips(device, client_ips):
    device.update({"last_seen_ts": ip.get("last_seen"), "last_seen_ip": ip.get("ip")})


-class DeviceListUpdater(object):
+class DeviceListUpdater:
    "Handles incoming device list updates from federation and updates the DB"

    def __init__(self, hs, device_handler):
@@ -16,8 +16,6 @@
 import logging
 from typing import Any, Dict

-from canonicaljson import json
-
 from synapse.api.errors import SynapseError
 from synapse.logging.context import run_in_background
 from synapse.logging.opentracing import (
@@ -27,12 +25,13 @@ from synapse.logging.opentracing import (
    start_active_span,
 )
 from synapse.types import UserID, get_domain_from_id
+from synapse.util import json_encoder
 from synapse.util.stringutils import random_string

 logger = logging.getLogger(__name__)


-class DeviceMessageHandler(object):
+class DeviceMessageHandler:
    def __init__(self, hs):
        """
        Args:
@@ -174,7 +173,7 @@ class DeviceMessageHandler(object):
                    "sender": sender_user_id,
                    "type": message_type,
                    "message_id": message_id,
-                    "org.matrix.opentracing_context": json.dumps(context),
+                    "org.matrix.opentracing_context": json_encoder.encode(context),
                }

        log_kv({"local_messages": local_messages})
@@ -23,6 +23,7 @@ from synapse.api.errors import (
    CodeMessageException,
    Codes,
    NotFoundError,
+    ShadowBanError,
    StoreError,
    SynapseError,
 )
@@ -199,6 +200,8 @@ class DirectoryHandler(BaseHandler):

        try:
            await self._update_canonical_alias(requester, user_id, room_id, room_alias)
+        except ShadowBanError as e:
+            logger.info("Failed to update alias events due to shadow-ban: %s", e)
        except AuthError as e:
            logger.info("Failed to update alias events: %s", e)

@@ -292,6 +295,9 @@ class DirectoryHandler(BaseHandler):
        """
        Send an updated canonical alias event if the removed alias was set as
        the canonical alias or listed in the alt_aliases field.
+
+        Raises:
+            ShadowBanError if the requester has been shadow-banned.
        """
        alias_event = await self.state.get_current_state(
            room_id, EventTypes.CanonicalAlias, ""
@@ -19,7 +19,7 @@ import logging
 from typing import Dict, List, Optional, Tuple

 import attr
-from canonicaljson import encode_canonical_json, json
+from canonicaljson import encode_canonical_json
 from signedjson.key import VerifyKey, decode_verify_key_bytes
 from signedjson.sign import SignatureVerifyException, verify_signed_json
 from unpaddedbase64 import decode_base64
@@ -35,7 +35,7 @@ from synapse.types import (
    get_domain_from_id,
    get_verify_key_from_cross_signing_key,
 )
-from synapse.util import unwrapFirstError
+from synapse.util import json_decoder, unwrapFirstError
 from synapse.util.async_helpers import Linearizer
 from synapse.util.caches.expiringcache import ExpiringCache
 from synapse.util.retryutils import NotRetryingDestination
@@ -43,7 +43,7 @@ from synapse.util.retryutils import NotRetryingDestination
 logger = logging.getLogger(__name__)


-class E2eKeysHandler(object):
+class E2eKeysHandler:
    def __init__(self, hs):
        self.store = hs.get_datastore()
        self.federation = hs.get_federation_client()
@@ -353,7 +353,7 @@ class E2eKeysHandler(object):
            # make sure that each queried user appears in the result dict
            result_dict[user_id] = {}

-        results = await self.store.get_e2e_device_keys(local_query)
+        results = await self.store.get_e2e_device_keys_for_cs_api(local_query)

        # Build the result structure
        for user_id, device_keys in results.items():
@@ -404,7 +404,7 @@ class E2eKeysHandler(object):
            for device_id, keys in device_keys.items():
                for key_id, json_bytes in keys.items():
                    json_result.setdefault(user_id, {})[device_id] = {
-                        key_id: json.loads(json_bytes)
+                        key_id: json_decoder.decode(json_bytes)
                    }

        @trace
@@ -734,7 +734,7 @@ class E2eKeysHandler(object):
            # fetch our stored devices.  This is used to 1. verify
            # signatures on the master key, and 2. to compare with what
            # was sent if the device was signed
-            devices = await self.store.get_e2e_device_keys([(user_id, None)])
+            devices = await self.store.get_e2e_device_keys_for_cs_api([(user_id, None)])

            if user_id not in devices:
                raise NotFoundError("No device keys found")
@@ -1186,7 +1186,7 @@ def _exception_to_failure(e):


 def _one_time_keys_match(old_key_json, new_key):
-    old_key = json.loads(old_key_json)
+    old_key = json_decoder.decode(old_key_json)

    # if either is a string rather than an object, they must match exactly
    if not isinstance(old_key, dict) or not isinstance(new_key, dict):
@@ -1212,7 +1212,7 @@ class SignatureListItem:
    signature = attr.ib()


-class SigningKeyEduUpdater(object):
+class SigningKeyEduUpdater:
    """Handles incoming signing key updates from federation and updates the DB"""

    def __init__(self, hs, e2e_keys_handler):
@@ -29,7 +29,7 @@ from synapse.util.async_helpers import Linearizer
 logger = logging.getLogger(__name__)


-class E2eRoomKeysHandler(object):
+class E2eRoomKeysHandler:
    """
    Implements an optional realtime backup mechanism for encrypted E2E megolm room keys.
    This gives a way for users to store and recover their megolm keys if they lose all
@@ -15,29 +15,30 @@

 import logging
 import random
+from typing import TYPE_CHECKING, Iterable, List, Optional

 from synapse.api.constants import EventTypes, Membership
 from synapse.api.errors import AuthError, SynapseError
 from synapse.events import EventBase
 from synapse.handlers.presence import format_user_presence_state
 from synapse.logging.utils import log_function
-from synapse.types import UserID
+from synapse.streams.config import PaginationConfig
+from synapse.types import JsonDict, UserID
 from synapse.visibility import filter_events_for_client

 from ._base import BaseHandler

+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+
 logger = logging.getLogger(__name__)


 class EventStreamHandler(BaseHandler):
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        super(EventStreamHandler, self).__init__(hs)

-        # Count of active streams per user
-        self._streams_per_user = {}
-        # Grace timers per user to delay the "stopped" signal
-        self._stop_timer_per_user = {}
-
        self.distributor = hs.get_distributor()
        self.distributor.declare("started_user_eventstream")
        self.distributor.declare("stopped_user_eventstream")
@@ -52,14 +53,14 @@ class EventStreamHandler(BaseHandler):
    @log_function
    async def get_stream(
        self,
-        auth_user_id,
-        pagin_config,
-        timeout=0,
-        as_client_event=True,
-        affect_presence=True,
-        room_id=None,
-        is_guest=False,
-    ):
+        auth_user_id: str,
+        pagin_config: PaginationConfig,
+        timeout: int = 0,
+        as_client_event: bool = True,
+        affect_presence: bool = True,
+        room_id: Optional[str] = None,
+        is_guest: bool = False,
+    ) -> JsonDict:
        """Fetches the events stream for a given user.
        """

@@ -98,7 +99,7 @@ class EventStreamHandler(BaseHandler):

            # When the user joins a new room, or another user joins a currently
            # joined room, we need to send down presence for those users.
-            to_add = []
+            to_add = []  # type: List[JsonDict]
            for event in events:
                if not isinstance(event, EventBase):
                    continue
@@ -110,7 +111,7 @@ class EventStreamHandler(BaseHandler):
                        # Send down presence for everyone in the room.
                        users = await self.state.get_current_users_in_room(
                            event.room_id
-                        )
+                        )  # type: Iterable[str]
                    else:
                        users = [event.state_key]

@@ -144,20 +145,22 @@ class EventStreamHandler(BaseHandler):


 class EventHandler(BaseHandler):
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        super(EventHandler, self).__init__(hs)
        self.storage = hs.get_storage()

-    async def get_event(self, user, room_id, event_id):
+    async def get_event(
+        self, user: UserID, room_id: Optional[str], event_id: str
+    ) -> Optional[EventBase]:
        """Retrieve a single specified event.

        Args:
-            user (synapse.types.UserID): The user requesting the event
-            room_id (str|None): The expected room id. We'll return None if the
+            user: The user requesting the event
+            room_id: The expected room id. We'll return None if the
                event's room does not match.
-            event_id (str): The event ID to obtain.
+            event_id: The event ID to obtain.
        Returns:
-            dict: An event, or None if there is no event matching this ID.
+            An event, or None if there is no event matching this ID.
        Raises:
            SynapseError if there was a problem retrieving this event, or
            AuthError if the user does not have the rights to inspect this
@@ -72,7 +72,13 @@ from synapse.replication.http.federation import (
 from synapse.replication.http.membership import ReplicationUserJoinedLeftRoomRestServlet
 from synapse.state import StateResolutionStore, resolve_events_with_store
 from synapse.storage.databases.main.events_worker import EventRedactBehaviour
-from synapse.types import JsonDict, StateMap, UserID, get_domain_from_id
+from synapse.types import (
+    JsonDict,
+    MutableStateMap,
+    StateMap,
+    UserID,
+    get_domain_from_id,
+)
 from synapse.util.async_helpers import Linearizer, concurrently_execute
 from synapse.util.distributor import user_joined_room
 from synapse.util.retryutils import NotRetryingDestination
@@ -96,7 +102,7 @@ class _NewEventInfo:

    event = attr.ib(type=EventBase)
    state = attr.ib(type=Optional[Sequence[EventBase]], default=None)
-    auth_events = attr.ib(type=Optional[StateMap[EventBase]], default=None)
+    auth_events = attr.ib(type=Optional[MutableStateMap[EventBase]], default=None)


 class FederationHandler(BaseHandler):
@@ -434,11 +440,11 @@ class FederationHandler(BaseHandler):
        if not prevs - seen:
            return

-        latest = await self.store.get_latest_event_ids_in_room(room_id)
+        latest_list = await self.store.get_latest_event_ids_in_room(room_id)

        # We add the prev events that we have seen to the latest
        # list to ensure the remote server doesn't give them to us
-        latest = set(latest)
+        latest = set(latest_list)
        latest |= seen

        logger.info(
@@ -775,7 +781,7 @@ class FederationHandler(BaseHandler):
                    # keys across all devices.
                    current_keys = [
                        key
-                        for device in cached_devices
+                        for device in cached_devices.values()
                        for key in device.get("keys", {}).get("keys", {}).values()
                    ]

@@ -937,15 +943,26 @@ class FederationHandler(BaseHandler):

        return events

-    async def maybe_backfill(self, room_id, current_depth):
+    async def maybe_backfill(
+        self, room_id: str, current_depth: int, limit: int
+    ) -> bool:
        """Checks the database to see if we should backfill before paginating,
        and if so do.
+
+        Args:
+            room_id
+            current_depth: The depth from which we're paginating from. This is
+                used to decide if we should backfill and what extremities to
+                use.
+            limit: The number of events that the pagination request will
+                return. This is used as part of the heuristic to decide if we
+                should back paginate.
        """
        extremities = await self.store.get_oldest_events_with_depth_in_room(room_id)

        if not extremities:
            logger.debug("Not backfilling as no extremeties found.")
-            return
+            return False

        # We only want to paginate if we can actually see the events we'll get,
        # as otherwise we'll just spend a lot of resources to get redacted
@@ -998,16 +1015,54 @@ class FederationHandler(BaseHandler):
        sorted_extremeties_tuple = sorted(extremities.items(), key=lambda e: -int(e[1]))
        max_depth = sorted_extremeties_tuple[0][1]

+        # If we're approaching an extremity we trigger a backfill, otherwise we
+        # no-op.
+        #
+        # We chose twice the limit here as then clients paginating backwards
+        # will send pagination requests that trigger backfill at least twice
+        # using the most recent extremity before it gets removed (see below). We
+        # chose more than one times the limit in case of failure, but choosing a
+        # much larger factor will result in triggering a backfill request much
+        # earlier than necessary.
+        if current_depth - 2 * limit > max_depth:
+            logger.debug(
+                "Not backfilling as we don't need to. %d < %d - 2 * %d",
+                max_depth,
+                current_depth,
+                limit,
+            )
+            return False
+
+        logger.debug(
+            "room_id: %s, backfill: current_depth: %s, max_depth: %s, extrems: %s",
+            room_id,
+            current_depth,
+            max_depth,
+            sorted_extremeties_tuple,
+        )
+
+        # We ignore extremities that have a greater depth than our current depth
+        # as:
+        #    1. we don't really care about getting events that have happened
+        #       before our current position; and
+        #    2. we have likely previously tried and failed to backfill from that
+        #       extremity, so to avoid getting "stuck" requesting the same
+        #       backfill repeatedly we drop those extremities.
+        filtered_sorted_extremeties_tuple = [
+            t for t in sorted_extremeties_tuple if int(t[1]) <= current_depth
+        ]
+
+        # However, we need to check that the filtered extremities are non-empty.
+        # If they are empty then either we can a) bail or b) still attempt to
+        # backill. We opt to try backfilling anyway just in case we do get
+        # relevant events.
+        if filtered_sorted_extremeties_tuple:
+            sorted_extremeties_tuple = filtered_sorted_extremeties_tuple
+
        # We don't want to specify too many extremities as it causes the backfill
        # request URI to be too long.
        extremities = dict(sorted_extremeties_tuple[:5])

-        if current_depth > max_depth:
-            logger.debug(
-                "Not backfilling as we don't need to. %d < %d", max_depth, current_depth
-            )
-            return
-
        # Now we need to decide which hosts to hit first.

        # First we try hosts that are already in the room
@@ -1777,9 +1832,7 @@ class FederationHandler(BaseHandler):
        """Returns the state at the event. i.e. not including said event.
        """

-        event = await self.store.get_event(
-            event_id, allow_none=False, check_room_id=room_id
-        )
+        event = await self.store.get_event(event_id, check_room_id=room_id)

        state_groups = await self.state_store.get_state_groups(room_id, [event_id])

@@ -1805,9 +1858,7 @@ class FederationHandler(BaseHandler):
    async def get_state_ids_for_pdu(self, room_id: str, event_id: str) -> List[str]:
        """Returns the state at the event. i.e. not including said event.
        """
-        event = await self.store.get_event(
-            event_id, allow_none=False, check_room_id=room_id
-        )
+        event = await self.store.get_event(event_id, check_room_id=room_id)

        state_groups = await self.state_store.get_state_groups_ids(room_id, [event_id])

@@ -1877,8 +1928,8 @@ class FederationHandler(BaseHandler):
        else:
            return None

-    def get_min_depth_for_context(self, context):
-        return self.store.get_min_depth(context)
+    async def get_min_depth_for_context(self, context):
+        return await self.store.get_min_depth(context)

    async def _handle_new_event(
        self, origin, event, state=None, auth_events=None, backfilled=False
@@ -2057,7 +2108,7 @@ class FederationHandler(BaseHandler):
        origin: str,
        event: EventBase,
        state: Optional[Iterable[EventBase]],
-        auth_events: Optional[StateMap[EventBase]],
+        auth_events: Optional[MutableStateMap[EventBase]],
        backfilled: bool,
    ) -> EventContext:
        context = await self.state_handler.compute_event_context(event, old_state=state)
@@ -2107,8 +2158,8 @@ class FederationHandler(BaseHandler):
        if backfilled or event.internal_metadata.is_outlier():
            return

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

        if extrem_ids == prev_event_ids:
@@ -2138,10 +2189,12 @@ class FederationHandler(BaseHandler):
            )
            state_sets = list(state_sets.values())
            state_sets.append(state)
-            current_state_ids = await self.state_handler.resolve_events(
+            current_states = 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()}
+            current_state_ids = {
+                k: e.event_id for k, e in current_states.items()
+            }  # type: StateMap[str]
        else:
            current_state_ids = await self.state_handler.get_current_state_ids(
                event.room_id, latest_event_ids=extrem_ids
@@ -2153,11 +2206,13 @@ class FederationHandler(BaseHandler):

        # 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]
+        current_state_ids_list = [
+            e for k, e in current_state_ids.items() if k in auth_types
+        ]

-        current_auth_events = await self.store.get_events(current_state_ids)
+        auth_events_map = await self.store.get_events(current_state_ids_list)
        current_auth_events = {
-            (e.type, e.state_key): e for e in current_auth_events.values()
+            (e.type, e.state_key): e for e in auth_events_map.values()
        }

        try:
@@ -2173,9 +2228,7 @@ class FederationHandler(BaseHandler):
        if not in_room:
            raise AuthError(403, "Host not in room.")

-        event = await self.store.get_event(
-            event_id, allow_none=False, check_room_id=room_id
-        )
+        event = await self.store.get_event(event_id, check_room_id=room_id)

        # Just go through and process each event in `remote_auth_chain`. We
        # don't want to fall into the trap of `missing` being wrong.
@@ -2227,7 +2280,7 @@ class FederationHandler(BaseHandler):
        origin: str,
        event: EventBase,
        context: EventContext,
-        auth_events: StateMap[EventBase],
+        auth_events: MutableStateMap[EventBase],
    ) -> EventContext:
        """

@@ -2278,7 +2331,7 @@ class FederationHandler(BaseHandler):
        origin: str,
        event: EventBase,
        context: EventContext,
-        auth_events: StateMap[EventBase],
+        auth_events: MutableStateMap[EventBase],
    ) -> EventContext:
        """Helper for do_auth. See there for docs.

@@ -52,7 +52,7 @@ def _create_rerouter(func_name):
    return f


-class GroupsLocalWorkerHandler(object):
+class GroupsLocalWorkerHandler:
    def __init__(self, hs):
        self.hs = hs
        self.store = hs.get_datastore()
@@ -461,6 +461,25 @@ class GroupsLocalHandler(GroupsLocalWorkerHandler):

        return {"state": "invite", "user_profile": user_profile}

+    async def change_user_admin_in_group(
+        self, group_id, user_id, want_admin, requester_user_id, content
+    ):
+        """Promotes or demotes a user in a group.
+        """
+
+        if not self.is_mine_id(user_id):
+            raise SynapseError(400, "User not on this server")
+
+        # TODO: We should probably support federation, but this is fine for now
+        if not self.is_mine_id(group_id):
+            raise SynapseError(400, "Group not on this server")
+
+        res = await self.groups_server_handler.change_user_admin_in_group(
+            group_id, user_id, want_admin, requester_user_id, content
+        )
+
+        return res
+
    async def remove_user_from_group(
        self, group_id, user_id, requester_user_id, content
    ):
@@ -21,8 +21,6 @@ import logging
 import urllib.parse
 from typing import Awaitable, Callable, Dict, List, Optional, Tuple

-from canonicaljson import json
-
 from twisted.internet.error import TimeoutError

 from synapse.api.errors import (
@@ -34,6 +32,7 @@ from synapse.api.errors import (
 from synapse.config.emailconfig import ThreepidBehaviour
 from synapse.http.client import SimpleHttpClient
 from synapse.types import JsonDict, Requester
+from synapse.util import json_decoder
 from synapse.util.hash import sha256_and_url_safe_base64
 from synapse.util.stringutils import assert_valid_client_secret, random_string

@@ -177,7 +176,7 @@ class IdentityHandler(BaseHandler):
        except TimeoutError:
            raise SynapseError(500, "Timed out contacting identity server")
        except CodeMessageException as e:
-            data = json.loads(e.msg)  # XXX WAT?
+            data = json_decoder.decode(e.msg)  # XXX WAT?
            return data

        logger.info("Got 404 when POSTing JSON %s, falling back to v1 URL", bind_url)
@@ -14,6 +14,7 @@
 # limitations under the License.

 import logging
+from typing import TYPE_CHECKING

 from twisted.internet import defer

@@ -22,8 +23,9 @@ from synapse.api.errors import SynapseError
 from synapse.events.validator import EventValidator
 from synapse.handlers.presence import format_user_presence_state
 from synapse.logging.context import make_deferred_yieldable, run_in_background
+from synapse.storage.roommember import RoomsForUser
 from synapse.streams.config import PaginationConfig
-from synapse.types import StreamToken, UserID
+from synapse.types import JsonDict, Requester, StreamToken, UserID
 from synapse.util import unwrapFirstError
 from synapse.util.async_helpers import concurrently_execute
 from synapse.util.caches.response_cache import ResponseCache
@@ -31,11 +33,15 @@ from synapse.visibility import filter_events_for_client

 from ._base import BaseHandler

+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+
 logger = logging.getLogger(__name__)


 class InitialSyncHandler(BaseHandler):
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        super(InitialSyncHandler, self).__init__(hs)
        self.hs = hs
        self.state = hs.get_state_handler()
@@ -48,27 +54,25 @@ class InitialSyncHandler(BaseHandler):

    def snapshot_all_rooms(
        self,
-        user_id=None,
-        pagin_config=None,
-        as_client_event=True,
-        include_archived=False,
-    ):
+        user_id: str,
+        pagin_config: PaginationConfig,
+        as_client_event: bool = True,
+        include_archived: bool = False,
+    ) -> JsonDict:
        """Retrieve a snapshot of all rooms the user is invited or has joined.

        This snapshot may include messages for all rooms where the user is
        joined, depending on the pagination config.

        Args:
-            user_id (str): The ID of the user making the request.
-            pagin_config (synapse.api.streams.PaginationConfig): The pagination
-            config used to determine how many messages *PER ROOM* to return.
-            as_client_event (bool): True to get events in client-server format.
-            include_archived (bool): True to get rooms that the user has left
+            user_id: The ID of the user making the request.
+            pagin_config: The pagination config used to determine how many
+                messages *PER ROOM* to return.
+            as_client_event: True to get events in client-server format.
+            include_archived: True to get rooms that the user has left
        Returns:
-            A list of dicts with "room_id" and "membership" keys for all rooms
-            the user is currently invited or joined in on. Rooms where the user
-            is joined on, may return a "messages" key with messages, depending
-            on the specified PaginationConfig.
+            A JsonDict with the same format as the response to `/intialSync`
+            API
        """
        key = (
            user_id,
@@ -91,11 +95,11 @@ class InitialSyncHandler(BaseHandler):

    async def _snapshot_all_rooms(
        self,
-        user_id=None,
-        pagin_config=None,
-        as_client_event=True,
-        include_archived=False,
-    ):
+        user_id: str,
+        pagin_config: PaginationConfig,
+        as_client_event: bool = True,
+        include_archived: bool = False,
+    ) -> JsonDict:

        memberships = [Membership.INVITE, Membership.JOIN]
        if include_archived:
@@ -134,7 +138,7 @@ class InitialSyncHandler(BaseHandler):
        if limit is None:
            limit = 10

-        async def handle_room(event):
+        async def handle_room(event: RoomsForUser):
            d = {
                "room_id": event.room_id,
                "membership": event.membership,
@@ -251,17 +255,18 @@ class InitialSyncHandler(BaseHandler):

        return ret

-    async def room_initial_sync(self, requester, room_id, pagin_config=None):
+    async def room_initial_sync(
+        self, requester: Requester, room_id: str, pagin_config: PaginationConfig
+    ) -> JsonDict:
        """Capture the a snapshot of a room. If user is currently a member of
        the room this will be what is currently in the room. If the user left
        the room this will be what was in the room when they left.

        Args:
-            requester(Requester): The user to get a snapshot for.
-            room_id(str): The room to get a snapshot of.
-            pagin_config(synapse.streams.config.PaginationConfig):
-                The pagination config used to determine how many messages to
-                return.
+            requester: The user to get a snapshot for.
+            room_id: The room to get a snapshot of.
+            pagin_config: The pagination config used to determine how many
+                messages to return.
        Raises:
            AuthError if the user wasn't in the room.
        Returns:
@@ -305,8 +310,14 @@ class InitialSyncHandler(BaseHandler):
        return result

    async def _room_initial_sync_parted(
-        self, user_id, room_id, pagin_config, membership, member_event_id, is_peeking
-    ):
+        self,
+        user_id: str,
+        room_id: str,
+        pagin_config: PaginationConfig,
+        membership: Membership,
+        member_event_id: str,
+        is_peeking: bool,
+    ) -> JsonDict:
        room_state = await self.state_store.get_state_for_events([member_event_id])

        room_state = room_state[member_event_id]
@@ -350,8 +361,13 @@ class InitialSyncHandler(BaseHandler):
        }

    async def _room_initial_sync_joined(
-        self, user_id, room_id, pagin_config, membership, is_peeking
-    ):
+        self,
+        user_id: str,
+        room_id: str,
+        pagin_config: PaginationConfig,
+        membership: Membership,
+        is_peeking: bool,
+    ) -> JsonDict:
        current_state = await self.state.get_current_state(room_id=room_id)

        # TODO: These concurrently
@@ -15,9 +15,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
+import random
 from typing import TYPE_CHECKING, Dict, List, Optional, Tuple

-from canonicaljson import encode_canonical_json, json
+from canonicaljson import encode_canonical_json

 from twisted.internet.interfaces import IDelayedCall

@@ -34,6 +35,7 @@ from synapse.api.errors import (
    Codes,
    ConsentNotGivenError,
    NotFoundError,
+    ShadowBanError,
    SynapseError,
 )
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersions
@@ -47,14 +49,8 @@ from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.replication.http.send_event import ReplicationSendEventRestServlet
 from synapse.storage.databases.main.events_worker import EventRedactBehaviour
 from synapse.storage.state import StateFilter
-from synapse.types import (
-    Collection,
-    Requester,
-    RoomAlias,
-    StreamToken,
-    UserID,
-    create_requester,
-)
+from synapse.types import Requester, RoomAlias, StreamToken, UserID, create_requester
+from synapse.util import json_decoder
 from synapse.util.async_helpers import Linearizer
 from synapse.util.frozenutils import frozendict_json_encoder
 from synapse.util.metrics import measure_func
@@ -68,7 +64,7 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)


-class MessageHandler(object):
+class MessageHandler:
    """Contains some read only APIs to get state about a room
    """

@@ -92,12 +88,7 @@ class MessageHandler(object):
            )

    async def get_room_data(
-        self,
-        user_id: str,
-        room_id: str,
-        event_type: str,
-        state_key: str,
-        is_guest: bool,
+        self, user_id: str, room_id: str, event_type: str, state_key: str,
    ) -> dict:
        """ Get data from a room.

@@ -106,11 +97,10 @@ class MessageHandler(object):
            room_id
            event_type
            state_key
-            is_guest
        Returns:
            The path data content.
        Raises:
-            SynapseError if something went wrong.
+            SynapseError or AuthError if the user is not in the room
        """
        (
            membership,
@@ -127,6 +117,16 @@ class MessageHandler(object):
                [membership_event_id], StateFilter.from_types([key])
            )
            data = room_state[membership_event_id].get(key)
+        else:
+            # check_user_in_room_or_world_readable, if it doesn't raise an AuthError, should
+            # only ever return a Membership.JOIN/LEAVE object
+            #
+            # Safeguard in case it returned something else
+            logger.error(
+                "Attempted to retrieve data from a room for a user that has never been in it. "
+                "This should not have happened."
+            )
+            raise SynapseError(403, "User not in room", errcode=Codes.FORBIDDEN)

        return data

@@ -361,7 +361,7 @@ class MessageHandler(object):
 _DUMMY_EVENT_ROOM_EXCLUSION_EXPIRY = 7 * 24 * 60 * 60 * 1000


-class EventCreationHandler(object):
+class EventCreationHandler:
    def __init__(self, hs: "HomeServer"):
        self.hs = hs
        self.auth = hs.get_auth()
@@ -439,7 +439,7 @@ class EventCreationHandler(object):
        event_dict: dict,
        token_id: Optional[str] = None,
        txn_id: Optional[str] = None,
-        prev_event_ids: Optional[Collection[str]] = None,
+        prev_event_ids: Optional[List[str]] = None,
        require_consent: bool = True,
    ) -> Tuple[EventBase, EventContext]:
        """
@@ -644,37 +644,48 @@ class EventCreationHandler(object):
        event: EventBase,
        context: EventContext,
        ratelimit: bool = True,
+        ignore_shadow_ban: bool = False,
    ) -> int:
        """
        Persists and notifies local clients and federation of an event.

        Args:
-            requester
-            event the event to send.
-            context: the context of the event.
+            requester: The requester sending the event.
+            event: The event to send.
+            context: The context of the event.
            ratelimit: Whether to rate limit this send.
+            ignore_shadow_ban: True if shadow-banned users should be allowed to
+                send this event.

        Return:
            The stream_id of the persisted event.
+
+        Raises:
+            ShadowBanError if the requester has been shadow-banned.
        """
        if event.type == EventTypes.Member:
            raise SynapseError(
                500, "Tried to send member event through non-member codepath"
            )

+        if not ignore_shadow_ban and requester.shadow_banned:
+            # We randomly sleep a bit just to annoy the requester.
+            await self.clock.sleep(random.randint(1, 10))
+            raise ShadowBanError()
+
        user = UserID.from_string(event.sender)

        assert self.hs.is_mine(user), "User must be our own: %s" % (user,)

        if event.is_state():
-            prev_state = await self.deduplicate_state_event(event, context)
-            if prev_state is not None:
+            prev_event = await self.deduplicate_state_event(event, context)
+            if prev_event is not None:
                logger.info(
                    "Not bothering to persist state event %s duplicated by %s",
                    event.event_id,
-                    prev_state.event_id,
+                    prev_event.event_id,
                )
-                return prev_state
+                return await self.store.get_stream_id_for_event(prev_event.event_id)

        return await self.handle_new_client_event(
            requester=requester, event=event, context=context, ratelimit=ratelimit
@@ -682,27 +693,32 @@ class EventCreationHandler(object):

    async def deduplicate_state_event(
        self, event: EventBase, context: EventContext
-    ) -> None:
+    ) -> Optional[EventBase]:
        """
        Checks whether event is in the latest resolved state in context.

-        If so, returns the version of the event in context.
-        Otherwise, returns None.
+        Args:
+            event: The event to check for duplication.
+            context: The event context.
+
+        Returns:
+            The previous verion of the event is returned, if it is found in the
+            event context. Otherwise, None is returned.
        """
        prev_state_ids = await context.get_prev_state_ids()
        prev_event_id = prev_state_ids.get((event.type, event.state_key))
        if not prev_event_id:
-            return
+            return None
        prev_event = await self.store.get_event(prev_event_id, allow_none=True)
        if not prev_event:
-            return
+            return None

        if prev_event and event.user_id == prev_event.user_id:
            prev_content = encode_canonical_json(prev_event.content)
            next_content = encode_canonical_json(event.content)
            if prev_content == next_content:
                return prev_event
-        return
+        return None

    async def create_and_send_nonmember_event(
        self,
@@ -710,12 +726,28 @@ class EventCreationHandler(object):
        event_dict: dict,
        ratelimit: bool = True,
        txn_id: Optional[str] = None,
+        ignore_shadow_ban: bool = False,
    ) -> Tuple[EventBase, int]:
        """
        Creates an event, then sends it.

        See self.create_event and self.send_nonmember_event.
+
+        Args:
+            requester: The requester sending the event.
+            event_dict: An entire event.
+            ratelimit: Whether to rate limit this send.
+            txn_id: The transaction ID.
+            ignore_shadow_ban: True if shadow-banned users should be allowed to
+                send this event.
+
+        Raises:
+            ShadowBanError if the requester has been shadow-banned.
        """
+        if not ignore_shadow_ban and requester.shadow_banned:
+            # We randomly sleep a bit just to annoy the requester.
+            await self.clock.sleep(random.randint(1, 10))
+            raise ShadowBanError()

        # We limit the number of concurrent event sends in a room so that we
        # don't fork the DAG too much. If we don't limit then we can end up in
@@ -734,7 +766,11 @@ class EventCreationHandler(object):
                raise SynapseError(403, spam_error, Codes.FORBIDDEN)

            stream_id = await self.send_nonmember_event(
-                requester, event, context, ratelimit=ratelimit
+                requester,
+                event,
+                context,
+                ratelimit=ratelimit,
+                ignore_shadow_ban=ignore_shadow_ban,
            )
        return event, stream_id

@@ -743,7 +779,7 @@ class EventCreationHandler(object):
        self,
        builder: EventBuilder,
        requester: Optional[Requester] = None,
-        prev_event_ids: Optional[Collection[str]] = None,
+        prev_event_ids: Optional[List[str]] = None,
    ) -> Tuple[EventBase, EventContext]:
        """Create a new event for a local client

@@ -859,7 +895,7 @@ class EventCreationHandler(object):
        # Ensure that we can round trip before trying to persist in db
        try:
            dump = frozendict_json_encoder.encode(event.content)
-            json.loads(dump)
+            json_decoder.decode(dump)
        except Exception:
            logger.exception("Failed to encode content: %r", event.content)
            raise
@@ -891,9 +927,7 @@ class EventCreationHandler(object):
        except Exception:
            # Ensure that we actually remove the entries in the push actions
            # staging area, if we calculated them.
-            run_in_background(
-                self.store.remove_push_actions_from_staging, event.event_id
-            )
+            await self.store.remove_push_actions_from_staging(event.event_id)
            raise

    async def _validate_canonical_alias(
@@ -957,7 +991,7 @@ class EventCreationHandler(object):
                    allow_none=True,
                )

-                is_admin_redaction = (
+                is_admin_redaction = bool(
                    original_event and event.sender != original_event.sender
                )

@@ -1077,8 +1111,8 @@ class EventCreationHandler(object):
            auth_events_ids = 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_map = await self.store.get_events(auth_events_ids)
+            auth_events = {(e.type, e.state_key): e for e in auth_events_map.values()}

            room_version = await self.store.get_room_version_id(event.room_id)
            room_version_obj = KNOWN_ROOM_VERSIONS[room_version]
@@ -1176,8 +1210,14 @@ class EventCreationHandler(object):

                    event.internal_metadata.proactively_send = False

+                    # Since this is a dummy-event it is OK if it is sent by a
+                    # shadow-banned user.
                    await self.send_nonmember_event(
-                        requester, event, context, ratelimit=False
+                        requester,
+                        event,
+                        context,
+                        ratelimit=False,
+                        ignore_shadow_ban=True,
                    )
                    dummy_event_sent = True
                    break
@@ -12,7 +12,6 @@
 # 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.
-import json
 import logging
 from typing import TYPE_CHECKING, Dict, Generic, List, Optional, Tuple, TypeVar
 from urllib.parse import urlencode
@@ -38,8 +37,8 @@ from synapse.config import ConfigError
 from synapse.http.server import respond_with_html
 from synapse.http.site import SynapseRequest
 from synapse.logging.context import make_deferred_yieldable
-from synapse.push.mailer import load_jinja2_templates
 from synapse.types import UserID, map_username_to_mxid_localpart
+from synapse.util import json_decoder

 if TYPE_CHECKING:
    from synapse.server import HomeServer
@@ -94,6 +93,7 @@ class OidcHandler:
    """

    def __init__(self, hs: "HomeServer"):
+        self.hs = hs
        self._callback_url = hs.config.oidc_callback_url  # type: str
        self._scopes = hs.config.oidc_scopes  # type: List[str]
        self._client_auth = ClientAuth(
@@ -123,9 +123,7 @@ class OidcHandler:
        self._hostname = hs.hostname  # type: str
        self._server_name = hs.config.server_name  # type: str
        self._macaroon_secret_key = hs.config.macaroon_secret_key
-        self._error_template = load_jinja2_templates(
-            hs.config.sso_template_dir, ["sso_error.html"]
-        )[0]
+        self._error_template = hs.config.sso_error_template

        # identifier for the external_ids table
        self._auth_provider_id = "oidc"
@@ -370,7 +368,7 @@ class OidcHandler:
            # and check for an error field. If not, we respond with a generic
            # error message.
            try:
-                resp = json.loads(resp_body.decode("utf-8"))
+                resp = json_decoder.decode(resp_body.decode("utf-8"))
                error = resp["error"]
                description = resp.get("error_description", error)
            except (ValueError, KeyError):
@@ -387,7 +385,7 @@ class OidcHandler:

        # Since it is a not a 5xx code, body should be a valid JSON. It will
        # raise if not.
-        resp = json.loads(resp_body.decode("utf-8"))
+        resp = json_decoder.decode(resp_body.decode("utf-8"))

        if "error" in resp:
            error = resp["error"]
@@ -692,9 +690,17 @@ class OidcHandler:
                self._render_error(request, "invalid_token", str(e))
                return

+        # Pull out the user-agent and IP from the request.
+        user_agent = request.requestHeaders.getRawHeaders(b"User-Agent", default=[b""])[
+            0
+        ].decode("ascii", "surrogateescape")
+        ip_address = self.hs.get_ip_from_request(request)
+
        # Call the mapper to register/login the user
        try:
-            user_id = await self._map_userinfo_to_user(userinfo, token)
+            user_id = await self._map_userinfo_to_user(
+                userinfo, token, user_agent, ip_address
+            )
        except MappingException as e:
            logger.exception("Could not map user")
            self._render_error(request, "mapping_error", str(e))
@@ -831,7 +837,9 @@ class OidcHandler:
        now = self._clock.time_msec()
        return now < expiry

-    async def _map_userinfo_to_user(self, userinfo: UserInfo, token: Token) -> str:
+    async def _map_userinfo_to_user(
+        self, userinfo: UserInfo, token: Token, user_agent: str, ip_address: str
+    ) -> str:
        """Maps a UserInfo object to a mxid.

        UserInfo should have a claim that uniquely identifies users. This claim
@@ -846,6 +854,8 @@ class OidcHandler:
        Args:
            userinfo: an object representing the user
            token: a dict with the tokens obtained from the provider
+            user_agent: The user agent of the client making the request.
+            ip_address: The IP address of the client making the request.

        Raises:
            MappingException: if there was an error while mapping some properties
@@ -859,6 +869,9 @@ class OidcHandler:
            raise MappingException(
                "Failed to extract subject from OIDC response: %s" % (e,)
            )
+        # Some OIDC providers use integer IDs, but Synapse expects external IDs
+        # to be strings.
+        remote_user_id = str(remote_user_id)

        logger.info(
            "Looking for existing mapping for user %s:%s",
@@ -902,7 +915,9 @@ class OidcHandler:
        # It's the first time this user is logging in and the mapped mxid was
        # not taken, register the user
        registered_user_id = await self._registration_handler.register_user(
-            localpart=localpart, default_display_name=attributes["display_name"],
+            localpart=localpart,
+            default_display_name=attributes["display_name"],
+            user_agent_ips=(user_agent, ip_address),
        )

        await self._datastore.record_user_external_id(
@@ -14,23 +14,30 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
+from typing import TYPE_CHECKING, Any, Dict, Optional, Set

 from twisted.python.failure import Failure

 from synapse.api.constants import EventTypes, Membership
 from synapse.api.errors import SynapseError
+from synapse.api.filtering import Filter
 from synapse.logging.context import run_in_background
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.state import StateFilter
-from synapse.types import RoomStreamToken
+from synapse.streams.config import PaginationConfig
+from synapse.types import Requester, RoomStreamToken
 from synapse.util.async_helpers import ReadWriteLock
 from synapse.util.stringutils import random_string
 from synapse.visibility import filter_events_for_client

+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+
 logger = logging.getLogger(__name__)


-class PurgeStatus(object):
+class PurgeStatus:
    """Object tracking the status of a purge request

    This class contains information on the progress of a purge request, for
@@ -58,14 +65,14 @@ class PurgeStatus(object):
        return {"status": PurgeStatus.STATUS_TEXT[self.status]}


-class PaginationHandler(object):
+class PaginationHandler:
    """Handles pagination and purge history requests.

    These are in the same handler due to the fact we need to block clients
    paginating during a purge.
    """

-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        self.hs = hs
        self.auth = hs.get_auth()
        self.store = hs.get_datastore()
@@ -75,13 +82,16 @@ class PaginationHandler(object):
        self._server_name = hs.hostname

        self.pagination_lock = ReadWriteLock()
-        self._purges_in_progress_by_room = set()
+        self._purges_in_progress_by_room = set()  # type: Set[str]
        # map from purge id to PurgeStatus
-        self._purges_by_id = {}
+        self._purges_by_id = {}  # type: Dict[str, PurgeStatus]
        self._event_serializer = hs.get_event_client_serializer()

        self._retention_default_max_lifetime = hs.config.retention_default_max_lifetime

+        self._retention_allowed_lifetime_min = hs.config.retention_allowed_lifetime_min
+        self._retention_allowed_lifetime_max = hs.config.retention_allowed_lifetime_max
+
        if hs.config.retention_enabled:
            # Run the purge jobs described in the configuration file.
            for job in hs.config.retention_purge_jobs:
@@ -96,7 +106,9 @@ class PaginationHandler(object):
                    job["longest_max_lifetime"],
                )

-    async def purge_history_for_rooms_in_range(self, min_ms, max_ms):
+    async def purge_history_for_rooms_in_range(
+        self, min_ms: Optional[int], max_ms: Optional[int]
+    ):
        """Purge outdated events from rooms within the given retention range.

        If a default retention policy is defined in the server's configuration and its
@@ -104,14 +116,14 @@ class PaginationHandler(object):
        retention policy.

        Args:
-            min_ms (int|None): Duration in milliseconds that define the lower limit of
+            min_ms: Duration in milliseconds that define the lower limit of
                the range to handle (exclusive). If None, it means that the range has no
                lower limit.
-            max_ms (int|None): Duration in milliseconds that define the upper limit of
+            max_ms: Duration in milliseconds that define the upper limit of
                the range to handle (inclusive). If None, it means that the range has no
                upper limit.
        """
-        # We want the storage layer to to include rooms with no retention policy in its
+        # We want the storage layer to include rooms with no retention policy in its
        # return value only if a default retention policy is defined in the server's
        # configuration and that policy's 'max_lifetime' is either lower (or equal) than
        # max_ms or higher than min_ms (or both).
@@ -152,13 +164,32 @@ class PaginationHandler(object):
                )
                continue

-            max_lifetime = retention_policy["max_lifetime"]
+            # If max_lifetime is None, it means that the room has no retention policy.
+            # Given we only retrieve such rooms when there's a default retention policy
+            # defined in the server's configuration, we can safely assume that's the
+            # case and use it for this room.
+            max_lifetime = (
+                retention_policy["max_lifetime"] or self._retention_default_max_lifetime
+            )

-            if max_lifetime is None:
-                # If max_lifetime is None, it means that include_null equals True,
-                # therefore we can safely assume that there is a default policy defined
-                # in the server's configuration.
-                max_lifetime = self._retention_default_max_lifetime
+            # Cap the effective max_lifetime to be within the range allowed in the
+            # config.
+            # We do this in two steps:
+            #   1. Make sure it's higher or equal to the minimum allowed value, and if
+            #      it's not replace it with that value. This is because the server
+            #      operator can be required to not delete information before a given
+            #      time, e.g. to comply with freedom of information laws.
+            #   2. Make sure the resulting value is lower or equal to the maximum allowed
+            #      value, and if it's not replace it with that value. This is because the
+            #      server operator can be required to delete any data after a specific
+            #      amount of time.
+            if self._retention_allowed_lifetime_min is not None:
+                max_lifetime = max(self._retention_allowed_lifetime_min, max_lifetime)
+
+            if self._retention_allowed_lifetime_max is not None:
+                max_lifetime = min(max_lifetime, self._retention_allowed_lifetime_max)
+
+            logger.debug("[purge] max_lifetime for room %s: %s", room_id, max_lifetime)

            # Figure out what token we should start purging at.
            ts = self.clock.time_msec() - max_lifetime
@@ -195,18 +226,19 @@ class PaginationHandler(object):
                "_purge_history", self._purge_history, purge_id, room_id, token, True,
            )

-    def start_purge_history(self, room_id, token, delete_local_events=False):
+    def start_purge_history(
+        self, room_id: str, token: str, delete_local_events: bool = False
+    ) -> str:
        """Start off a history purge on a room.

        Args:
-            room_id (str): The room to purge from
-
-            token (str): topological token to delete events before
-            delete_local_events (bool): True to delete local events as well as
+            room_id: The room to purge from
+            token: topological token to delete events before
+            delete_local_events: True to delete local events as well as
                remote ones

        Returns:
-            str: unique ID for this purge transaction.
+            unique ID for this purge transaction.
        """
        if room_id in self._purges_in_progress_by_room:
            raise SynapseError(
@@ -225,15 +257,16 @@ class PaginationHandler(object):
        )
        return purge_id

-    async def _purge_history(self, purge_id, room_id, token, delete_local_events):
+    async def _purge_history(
+        self, purge_id: str, room_id: str, token: str, delete_local_events: bool
+    ) -> None:
        """Carry out a history purge on a room.

        Args:
-            purge_id (str): The id for this purge
-            room_id (str): The room to purge from
-            token (str): topological token to delete events before
-            delete_local_events (bool): True to delete local events as well as
-                remote ones
+            purge_id: The id for this purge
+            room_id: The room to purge from
+            token: topological token to delete events before
+            delete_local_events: True to delete local events as well as remote ones
        """
        self._purges_in_progress_by_room.add(room_id)
        try:
@@ -258,20 +291,17 @@ class PaginationHandler(object):

            self.hs.get_reactor().callLater(24 * 3600, clear_purge)

-    def get_purge_status(self, purge_id):
+    def get_purge_status(self, purge_id: str) -> Optional[PurgeStatus]:
        """Get the current status of an active purge

        Args:
-            purge_id (str): purge_id returned by start_purge_history
-
-        Returns:
-            PurgeStatus|None
+            purge_id: purge_id returned by start_purge_history
        """
        return self._purges_by_id.get(purge_id)

-    async def purge_room(self, room_id):
+    async def purge_room(self, room_id: str) -> None:
        """Purge the given room from the database"""
-        with (await self.pagination_lock.write(room_id)):
+        with await self.pagination_lock.write(room_id):
            # check we know about the room
            await self.store.get_room_version_id(room_id)

@@ -285,23 +315,22 @@ class PaginationHandler(object):

    async def get_messages(
        self,
-        requester,
-        room_id=None,
-        pagin_config=None,
-        as_client_event=True,
-        event_filter=None,
-    ):
+        requester: Requester,
+        room_id: str,
+        pagin_config: PaginationConfig,
+        as_client_event: bool = True,
+        event_filter: Optional[Filter] = None,
+    ) -> Dict[str, Any]:
        """Get messages in a room.

        Args:
-            requester (Requester): The user requesting messages.
-            room_id (str): The room they want messages from.
-            pagin_config (synapse.api.streams.PaginationConfig): The pagination
-                config rules to apply, if any.
-            as_client_event (bool): True to get events in client-server format.
-            event_filter (Filter): Filter to apply to results or None
+            requester: The user requesting messages.
+            room_id: The room they want messages from.
+            pagin_config: The pagination config rules to apply, if any.
+            as_client_event: True to get events in client-server format.
+            event_filter: Filter to apply to results or None
        Returns:
-            dict: Pagination API results
+            Pagination API results
        """
        user_id = requester.user.to_string()

@@ -321,7 +350,7 @@ class PaginationHandler(object):

        source_config = pagin_config.get_source_config("room")

-        with (await self.pagination_lock.read(room_id)):
+        with await self.pagination_lock.read(room_id):
            (
                membership,
                member_event_id,
@@ -333,9 +362,9 @@ class PaginationHandler(object):
                # if we're going backwards, we might need to backfill. This
                # requires that we have a topo token.
                if room_token.topological:
-                    max_topo = room_token.topological
+                    curr_topo = room_token.topological
                else:
-                    max_topo = await self.store.get_max_topological_token(
+                    curr_topo = await self.store.get_current_topological_token(
                        room_id, room_token.stream
                    )

@@ -343,15 +372,19 @@ class PaginationHandler(object):
                    # If they have left the room then clamp the token to be before
                    # they left the room, to save the effort of loading from the
                    # database.
+
+                    # This is only None if the room is world_readable, in which
+                    # case "JOIN" would have been returned.
+                    assert member_event_id
+
                    leave_token = await self.store.get_topological_token_for_event(
                        member_event_id
                    )
-                    leave_token = RoomStreamToken.parse(leave_token)
-                    if leave_token.topological < max_topo:
+                    if RoomStreamToken.parse(leave_token).topological < curr_topo:
                        source_config.from_key = str(leave_token)

                await self.hs.get_handlers().federation_handler.maybe_backfill(
-                    room_id, max_topo
+                    room_id, curr_topo, limit=source_config.limit,
                )

            events, next_key = await self.store.paginate_room_events(
@@ -394,8 +427,8 @@ class PaginationHandler(object):
            )

            if state_ids:
-                state = await self.store.get_events(list(state_ids.values()))
-                state = state.values()
+                state_dict = await self.store.get_events(list(state_ids.values()))
+                state = state_dict.values()

        time_now = self.clock.time_msec()

@@ -22,7 +22,7 @@ from synapse.api.errors import Codes, PasswordRefusedError
 logger = logging.getLogger(__name__)


-class PasswordPolicyHandler(object):
+class PasswordPolicyHandler:
    def __init__(self, hs):
        self.policy = hs.config.password_policy
        self.enabled = hs.config.password_policy_enabled
@@ -33,14 +33,14 @@ from typing_extensions import ContextManager
 import synapse.metrics
 from synapse.api.constants import EventTypes, Membership, PresenceState
 from synapse.api.errors import SynapseError
+from synapse.api.presence import UserPresenceState
 from synapse.logging.context import run_in_background
 from synapse.logging.utils import log_function
 from synapse.metrics import LaterGauge
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.state import StateHandler
 from synapse.storage.databases.main import DataStore
-from synapse.storage.presence import UserPresenceState
-from synapse.types import JsonDict, UserID, get_domain_from_id
+from synapse.types import Collection, JsonDict, UserID, get_domain_from_id
 from synapse.util.async_helpers import Linearizer
 from synapse.util.caches.descriptors import cached
 from synapse.util.metrics import Measure
@@ -1010,7 +1010,7 @@ def format_user_presence_state(state, now, include_user_id=True):
    return content


-class PresenceEventSource(object):
+class PresenceEventSource:
    def __init__(self, hs):
        # We can't call get_presence_handler here because there's a cycle:
        #
@@ -1318,7 +1318,7 @@ async def get_interested_parties(

 async def get_interested_remotes(
    store: DataStore, states: List[UserPresenceState], state_handler: StateHandler
-) -> List[Tuple[List[str], List[UserPresenceState]]]:
+) -> List[Tuple[Collection[str], List[UserPresenceState]]]:
    """Given a list of presence states figure out which remote servers
    should be sent which.

@@ -1334,7 +1334,7 @@ async def get_interested_remotes(
        each tuple the list of UserPresenceState should be sent to each
        destination
    """
-    hosts_and_states = []
+    hosts_and_states = []  # type: List[Tuple[Collection[str], List[UserPresenceState]]]

    # First we look up the rooms each user is in (as well as any explicit
    # subscriptions), then for each distinct room we look up the remote
@@ -14,6 +14,7 @@
 # limitations under the License.

 import logging
+import random

 from synapse.api.errors import (
    AuthError,
@@ -160,6 +161,9 @@ class BaseProfileHandler(BaseHandler):
                    Codes.FORBIDDEN,
                )

+        if not isinstance(new_displayname, str):
+            raise SynapseError(400, "Invalid displayname")
+
        if len(new_displayname) > MAX_DISPLAYNAME_LEN:
            raise SynapseError(
                400, "Displayname is too long (max %i)" % (MAX_DISPLAYNAME_LEN,)
@@ -213,8 +217,14 @@ class BaseProfileHandler(BaseHandler):
    async def set_avatar_url(
        self, target_user, requester, new_avatar_url, by_admin=False
    ):
-        """target_user is the user whose avatar_url is to be changed;
-        auth_user is the user attempting to make this change."""
+        """Set a new avatar URL for a user.
+
+        Args:
+            target_user (UserID): the user whose avatar URL is to be changed.
+            requester (Requester): The user attempting to make this change.
+            new_avatar_url (str): The avatar URL to give this user.
+            by_admin (bool): Whether this change was made by an administrator.
+        """
        if not self.hs.is_mine(target_user):
            raise SynapseError(400, "User is not hosted on this homeserver")

@@ -228,6 +238,9 @@ class BaseProfileHandler(BaseHandler):
                    400, "Changing avatar is disabled on this server", Codes.FORBIDDEN
                )

+        if not isinstance(new_avatar_url, str):
+            raise SynapseError(400, "Invalid displayname")
+
        if len(new_avatar_url) > MAX_AVATAR_URL_LEN:
            raise SynapseError(
                400, "Avatar URL is too long (max %i)" % (MAX_AVATAR_URL_LEN,)
@@ -278,6 +291,12 @@ class BaseProfileHandler(BaseHandler):

        await self.ratelimit(requester)

+        # Do not actually update the room state for shadow-banned users.
+        if requester.shadow_banned:
+            # We randomly sleep a bit just to annoy the requester.
+            await self.clock.sleep(random.randint(1, 10))
+            return
+
        room_ids = await self.store.get_rooms_for_user(target_user.to_string())

        for room_id in room_ids:
@@ -123,7 +123,7 @@ class ReceiptsHandler(BaseHandler):
        await self.federation.send_read_receipt(receipt)


-class ReceiptEventSource(object):
+class ReceiptEventSource:
    def __init__(self, hs):
        self.store = hs.get_datastore()

@@ -26,6 +26,7 @@ from synapse.replication.http.register import (
    ReplicationPostRegisterActionsServlet,
    ReplicationRegisterServlet,
 )
+from synapse.spam_checker_api import RegistrationBehaviour
 from synapse.storage.state import StateFilter
 from synapse.types import RoomAlias, UserID, create_requester

@@ -52,6 +53,8 @@ class RegistrationHandler(BaseHandler):
        self.macaroon_gen = hs.get_macaroon_generator()
        self._server_notices_mxid = hs.config.server_notices_mxid

+        self.spam_checker = hs.get_spam_checker()
+
        if hs.config.worker_app:
            self._register_client = ReplicationRegisterServlet.make_client(hs)
            self._register_device_client = RegisterDeviceReplicationServlet.make_client(
@@ -124,7 +127,9 @@ class RegistrationHandler(BaseHandler):
            try:
                int(localpart)
                raise SynapseError(
-                    400, "Numeric user IDs are reserved for guest users."
+                    400,
+                    "Numeric user IDs are reserved for guest users.",
+                    errcode=Codes.INVALID_USERNAME,
                )
            except ValueError:
                pass
@@ -142,6 +147,7 @@ class RegistrationHandler(BaseHandler):
        address=None,
        bind_emails=[],
        by_admin=False,
+        user_agent_ips=None,
    ):
        """Registers a new client on the server.

@@ -159,6 +165,8 @@ class RegistrationHandler(BaseHandler):
            bind_emails (List[str]): list of emails to bind to this account.
            by_admin (bool): True if this registration is being made via the
              admin api, otherwise False.
+            user_agent_ips (List[(str, str)]): Tuples of IP addresses and user-agents used
+                during the registration process.
        Returns:
            str: user_id
        Raises:
@@ -166,6 +174,24 @@ class RegistrationHandler(BaseHandler):
        """
        self.check_registration_ratelimit(address)

+        result = self.spam_checker.check_registration_for_spam(
+            threepid, localpart, user_agent_ips or [],
+        )
+
+        if result == RegistrationBehaviour.DENY:
+            logger.info(
+                "Blocked registration of %r", localpart,
+            )
+            # We return a 429 to make it not obvious that they've been
+            # denied.
+            raise SynapseError(429, "Rate limited")
+
+        shadow_banned = result == RegistrationBehaviour.SHADOW_BAN
+        if shadow_banned:
+            logger.info(
+                "Shadow banning registration of %r", localpart,
+            )
+
        # do not check_auth_blocking if the call is coming through the Admin API
        if not by_admin:
            await self.auth.check_auth_blocking(threepid=threepid)
@@ -194,6 +220,7 @@ class RegistrationHandler(BaseHandler):
                admin=admin,
                user_type=user_type,
                address=address,
+                shadow_banned=shadow_banned,
            )

            if self.hs.config.user_directory_search_all_users:
@@ -224,6 +251,7 @@ class RegistrationHandler(BaseHandler):
                        make_guest=make_guest,
                        create_profile_with_displayname=default_display_name,
                        address=address,
+                        shadow_banned=shadow_banned,
                    )

                    # Successfully registered
@@ -529,6 +557,7 @@ class RegistrationHandler(BaseHandler):
        admin=False,
        user_type=None,
        address=None,
+        shadow_banned=False,
    ):
        """Register user in the datastore.

@@ -546,6 +575,7 @@ class RegistrationHandler(BaseHandler):
            user_type (str|None): type of user. One of the values from
                api.constants.UserTypes, or None for a normal user.
            address (str|None): the IP address used to perform the registration.
+            shadow_banned (bool): Whether to shadow-ban the user

        Returns:
            Awaitable
@@ -561,6 +591,7 @@ class RegistrationHandler(BaseHandler):
                admin=admin,
                user_type=user_type,
                address=address,
+                shadow_banned=shadow_banned,
            )
        else:
            return self.store.register_user(
@@ -572,6 +603,7 @@ class RegistrationHandler(BaseHandler):
                create_profile_with_displayname=create_profile_with_displayname,
                admin=admin,
                user_type=user_type,
+                shadow_banned=shadow_banned,
            )

    async def register_device(
@@ -20,9 +20,10 @@
 import itertools
 import logging
 import math
+import random
 import string
 from collections import OrderedDict
-from typing import Awaitable, Optional, Tuple
+from typing import TYPE_CHECKING, Any, Awaitable, Dict, List, Optional, Tuple

 from synapse.api.constants import (
    EventTypes,
@@ -32,11 +33,15 @@ from synapse.api.constants import (
    RoomEncryptionAlgorithms,
 )
 from synapse.api.errors import AuthError, Codes, NotFoundError, StoreError, SynapseError
+from synapse.api.filtering import Filter
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersion
+from synapse.events import EventBase
 from synapse.events.utils import copy_power_levels_contents
 from synapse.http.endpoint import parse_and_validate_server_name
 from synapse.storage.state import StateFilter
 from synapse.types import (
+    JsonDict,
+    MutableStateMap,
    Requester,
    RoomAlias,
    RoomID,
@@ -47,12 +52,15 @@ from synapse.types import (
    create_requester,
 )
 from synapse.util import stringutils
-from synapse.util.async_helpers import Linearizer, maybe_awaitable
+from synapse.util.async_helpers import Linearizer
 from synapse.util.caches.response_cache import ResponseCache
 from synapse.visibility import filter_events_for_client

 from ._base import BaseHandler

+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
 logger = logging.getLogger(__name__)

 id_server_scheme = "https://"
@@ -61,7 +69,7 @@ FIVE_MINUTES_IN_MS = 5 * 60 * 1000


 class RoomCreationHandler(BaseHandler):
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        super(RoomCreationHandler, self).__init__(hs)

        self.spam_checker = hs.get_spam_checker()
@@ -92,7 +100,7 @@ class RoomCreationHandler(BaseHandler):
                "guest_can_join": False,
                "power_level_content_override": {},
            },
-        }
+        }  # type: Dict[str, Dict[str, Any]]

        # Modify presets to selectively enable encryption by default per homeserver config
        for preset_name, preset_config in self._presets_dict.items():
@@ -129,6 +137,9 @@ class RoomCreationHandler(BaseHandler):

        Returns:
            the new room id
+
+        Raises:
+            ShadowBanError if the requester is shadow-banned.
        """
        await self.ratelimit(requester)

@@ -164,6 +175,15 @@ class RoomCreationHandler(BaseHandler):
    async def _upgrade_room(
        self, requester: Requester, old_room_id: str, new_version: RoomVersion
    ):
+        """
+        Args:
+            requester: the user requesting the upgrade
+            old_room_id: the id of the room to be replaced
+            new_versions: the version to upgrade the room to
+
+        Raises:
+            ShadowBanError if the requester is shadow-banned.
+        """
        user_id = requester.user.to_string()

        # start by allocating a new room id
@@ -215,6 +235,9 @@ class RoomCreationHandler(BaseHandler):

        old_room_state = await tombstone_context.get_current_state_ids()

+        # We know the tombstone event isn't an outlier so it has current state.
+        assert old_room_state is not None
+
        # update any aliases
        await self._move_aliases_to_new_room(
            requester, old_room_id, new_room_id, old_room_state
@@ -247,6 +270,9 @@ class RoomCreationHandler(BaseHandler):
            old_room_id: the id of the room to be replaced
            new_room_id: the id of the replacement room
            old_room_state: the state map for the old room
+
+        Raises:
+            ShadowBanError if the requester is shadow-banned.
        """
        old_room_pl_event_id = old_room_state.get((EventTypes.PowerLevels, ""))

@@ -425,7 +451,7 @@ class RoomCreationHandler(BaseHandler):
        old_room_member_state_events = await self.store.get_events(
            old_room_member_state_ids.values()
        )
-        for k, old_event in old_room_member_state_events.items():
+        for old_event in old_room_member_state_events.values():
            # Only transfer ban events
            if (
                "membership" in old_event.content
@@ -528,17 +554,21 @@ class RoomCreationHandler(BaseHandler):
            logger.error("Unable to send updated alias events in new room: %s", e)

    async def create_room(
-        self, requester, config, ratelimit=True, creator_join_profile=None
+        self,
+        requester: Requester,
+        config: JsonDict,
+        ratelimit: bool = True,
+        creator_join_profile: Optional[JsonDict] = None,
    ) -> Tuple[dict, int]:
        """ Creates a new room.

        Args:
-            requester (synapse.types.Requester):
+            requester:
                The user who requested the room creation.
-            config (dict) : A dict of configuration options.
-            ratelimit (bool): set to False to disable the rate limiter
+            config : A dict of configuration options.
+            ratelimit: set to False to disable the rate limiter

-            creator_join_profile (dict|None):
+            creator_join_profile:
                Set to override the displayname and avatar for the creating
                user in this room. If unset, displayname and avatar will be
                derived from the user's profile. If set, should contain the
@@ -601,6 +631,7 @@ class RoomCreationHandler(BaseHandler):
                Codes.UNSUPPORTED_ROOM_VERSION,
            )

+        room_alias = None
        if "room_alias_name" in config:
            for wchar in string.whitespace:
                if wchar in config["room_alias_name"]:
@@ -611,9 +642,8 @@ class RoomCreationHandler(BaseHandler):

            if mapping:
                raise SynapseError(400, "Room alias already taken", Codes.ROOM_IN_USE)
-        else:
-            room_alias = None

+        invite_3pid_list = config.get("invite_3pid", [])
        invite_list = config.get("invite", [])
        for i in invite_list:
            try:
@@ -622,6 +652,14 @@ class RoomCreationHandler(BaseHandler):
            except Exception:
                raise SynapseError(400, "Invalid user_id: %s" % (i,))

+        if (invite_list or invite_3pid_list) and requester.shadow_banned:
+            # We randomly sleep a bit just to annoy the requester.
+            await self.clock.sleep(random.randint(1, 10))
+
+            # Allow the request to go through, but remove any associated invites.
+            invite_3pid_list = []
+            invite_list = []
+
        await self.event_creation_handler.assert_accepted_privacy_policy(requester)

        power_level_content_override = config.get("power_level_content_override")
@@ -636,8 +674,6 @@ class RoomCreationHandler(BaseHandler):
                % (user_id,),
            )

-        invite_3pid_list = config.get("invite_3pid", [])
-
        visibility = config.get("visibility", None)
        is_public = visibility == "public"

@@ -732,6 +768,8 @@ class RoomCreationHandler(BaseHandler):
            if is_direct:
                content["is_direct"] = is_direct

+            # Note that update_membership with an action of "invite" can raise a
+            # ShadowBanError, but this was handled above by emptying invite_list.
            _, last_stream_id = await self.room_member_handler.update_membership(
                requester,
                UserID.from_string(invitee),
@@ -746,6 +784,8 @@ class RoomCreationHandler(BaseHandler):
            id_access_token = invite_3pid.get("id_access_token")  # optional
            address = invite_3pid["address"]
            medium = invite_3pid["medium"]
+            # Note that do_3pid_invite can raise a  ShadowBanError, but this was
+            # handled above by emptying invite_3pid_list.
            last_stream_id = await self.hs.get_room_member_handler().do_3pid_invite(
                room_id,
                requester.user,
@@ -771,23 +811,30 @@ class RoomCreationHandler(BaseHandler):

    async def _send_events_for_new_room(
        self,
-        creator,  # A Requester object.
-        room_id,
-        preset_config,
-        invite_list,
-        initial_state,
-        creation_content,
-        room_alias=None,
-        power_level_content_override=None,  # Doesn't apply when initial state has power level state event content
-        creator_join_profile=None,
+        creator: Requester,
+        room_id: str,
+        preset_config: str,
+        invite_list: List[str],
+        initial_state: MutableStateMap,
+        creation_content: JsonDict,
+        room_alias: Optional[RoomAlias] = None,
+        power_level_content_override: Optional[JsonDict] = None,
+        creator_join_profile: Optional[JsonDict] = None,
    ) -> int:
        """Sends the initial events into a new room.

+        `power_level_content_override` doesn't apply when initial state has
+        power level state event content.
+
        Returns:
            The stream_id of the last event persisted.
        """

-        def create(etype, content, **kwargs):
+        creator_id = creator.user.to_string()
+
+        event_keys = {"room_id": room_id, "sender": creator_id, "state_key": ""}
+
+        def create(etype: str, content: JsonDict, **kwargs) -> JsonDict:
            e = {"type": etype, "content": content}

            e.update(event_keys)
@@ -795,23 +842,21 @@ class RoomCreationHandler(BaseHandler):

            return e

-        async def send(etype, content, **kwargs) -> int:
+        async def send(etype: str, content: JsonDict, **kwargs) -> int:
            event = create(etype, content, **kwargs)
            logger.debug("Sending %s in new room", etype)
+            # Allow these events to be sent even if the user is shadow-banned to
+            # allow the room creation to complete.
            (
                _,
                last_stream_id,
            ) = await self.event_creation_handler.create_and_send_nonmember_event(
-                creator, event, ratelimit=False
+                creator, event, ratelimit=False, ignore_shadow_ban=True,
            )
            return last_stream_id

        config = self._presets_dict[preset_config]

-        creator_id = creator.user.to_string()
-
-        event_keys = {"room_id": room_id, "sender": creator_id, "state_key": ""}
-
        creation_content.update({"creator": creator_id})
        await send(etype=EventTypes.Create, content=creation_content)

@@ -852,7 +897,7 @@ class RoomCreationHandler(BaseHandler):
                "kick": 50,
                "redact": 50,
                "invite": 50,
-            }
+            }  # type: JsonDict

            if config["original_invitees_have_ops"]:
                for invitee in invite_list:
@@ -906,7 +951,7 @@ class RoomCreationHandler(BaseHandler):
        return last_sent_stream_id

    async def _generate_room_id(
-        self, creator_id: str, is_public: str, room_version: RoomVersion,
+        self, creator_id: str, is_public: bool, room_version: RoomVersion,
    ):
        # autogen room IDs and try to create it. We may clash, so just
        # try a few times till one goes through, giving up eventually.
@@ -929,24 +974,31 @@ class RoomCreationHandler(BaseHandler):
        raise StoreError(500, "Couldn't generate a room ID.")


-class RoomContextHandler(object):
-    def __init__(self, hs):
+class RoomContextHandler:
+    def __init__(self, hs: "HomeServer"):
        self.hs = hs
        self.store = hs.get_datastore()
        self.storage = hs.get_storage()
        self.state_store = self.storage.state

-    async def get_event_context(self, user, room_id, event_id, limit, event_filter):
+    async def get_event_context(
+        self,
+        user: UserID,
+        room_id: str,
+        event_id: str,
+        limit: int,
+        event_filter: Optional[Filter],
+    ) -> Optional[JsonDict]:
        """Retrieves events, pagination tokens and state around a given event
        in a room.

        Args:
-            user (UserID)
-            room_id (str)
-            event_id (str)
-            limit (int): The maximum number of events to return in total
+            user
+            room_id
+            event_id
+            limit: The maximum number of events to return in total
                (excluding state).
-            event_filter (Filter|None): the filter to apply to the events returned
+            event_filter: the filter to apply to the events returned
                (excluding the target event_id)

        Returns:
@@ -1032,13 +1084,19 @@ class RoomContextHandler(object):
        return results


-class RoomEventSource(object):
-    def __init__(self, hs):
+class RoomEventSource:
+    def __init__(self, hs: "HomeServer"):
        self.store = hs.get_datastore()

    async def get_new_events(
-        self, user, from_key, limit, room_ids, is_guest, explicit_room_id=None
-    ):
+        self,
+        user: UserID,
+        from_key: str,
+        limit: int,
+        room_ids: List[str],
+        is_guest: bool,
+        explicit_room_id: Optional[str] = None,
+    ) -> Tuple[List[EventBase], str]:
        # We just ignore the key for now.

        to_key = self.get_current_key()
@@ -1088,7 +1146,7 @@ class RoomEventSource(object):
        return self.store.get_room_events_max_id(room_id)


-class RoomShutdownHandler(object):
+class RoomShutdownHandler:

    DEFAULT_MESSAGE = (
        "Sharing illegal content on this server is not permitted and rooms in"
@@ -1096,7 +1154,7 @@ class RoomShutdownHandler(object):
    )
    DEFAULT_ROOM_NAME = "Content Violation Notification"

-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        self.hs = hs
        self.room_member_handler = hs.get_room_member_handler()
        self._room_creation_handler = hs.get_room_creation_handler()
@@ -1272,9 +1330,7 @@ class RoomShutdownHandler(object):
                ratelimit=False,
            )

-            aliases_for_room = await maybe_awaitable(
-                self.store.get_aliases_for_room(room_id)
-            )
+            aliases_for_room = await self.store.get_aliases_for_room(room_id)

            await self.store.update_aliases_for_room(
                room_id, new_room_id, requester_user_id
--- a/Show More
+++ b/Show More