Update cahngelog

Fixes https://github.com/matrix-org/synapse/issues/9572

When a SSO user logs in for the first time, we create a local Matrix user for them. This goes through the register_user flow, which ends up triggering the spam checker. Spam checker modules don't currently have any way to differentiate between a user trying to sign up initially, versus an SSO user (whom has presumably already been approved elsewhere) trying to log in for the first time.

This PR passes `auth_provider_id` as an argument to the `check_registration_for_spam` function. This argument will contain an ID of an SSO provider (`"saml"`, `"cas"`, etc.) if one was used, else `None`.

.circleci/config.yml

@@ -14,7 +14,7 @@ jobs:
           platforms: linux/amd64
       - docker_build:
           tag: -t matrixdotorg/synapse:${CIRCLE_TAG}
-          platforms: linux/amd64,linux/arm/v7,linux/arm64
+          platforms: linux/amd64,linux/arm64
 
   dockerhubuploadlatest:
     docker:
@@ -27,7 +27,7 @@ jobs:
       # until all of the platforms are built.
       - docker_build:
           tag: -t matrixdotorg/synapse:latest
-          platforms: linux/amd64,linux/arm/v7,linux/arm64
+          platforms: linux/amd64,linux/arm64
 
 workflows:
   build:

.git-blame-ignore-revs

@@ -0,0 +1,8 @@
+# Black reformatting (#5482).
+32e7c9e7f20b57dd081023ac42d6931a8da9b3a3
+
+# Target Python 3.5 with black (#8664).
+aff1eb7c671b0a3813407321d2702ec46c71fa56
+
+# Update black to 20.8b1 (#9381).
+0a00b7ff14890987f09112a2ae696c61001e6cf1

.gitignore

@@ -6,13 +6,14 @@
 *.egg
 *.egg-info
 *.lock
-*.pyc
+*.py[cod]
 *.snap
 *.tac
 _trial_temp/
 _trial_temp*/
 /out
 .DS_Store
+__pycache__/
 
 # stuff that is likely to exist when you run a server locally
 /*.db

CHANGES.md

@@ -1,9 +1,285 @@
+Synapse 1.30.1 (2021-03-26)
+===========================
+
+This release is identical to Synapse 1.30.0, with the exception of explicitly
+setting a minimum version of Python's Cryptography library to ensure that users
+of Synapse are protected from the recent [OpenSSL security advisories](https://mta.openssl.org/pipermail/openssl-announce/2021-March/000198.html),
+especially CVE-2021-3449.
+
+Note that Cryptography defaults to bundling its own statically linked copy of
+OpenSSL, which means that you may not be protected by your operating system's
+security updates.
+
+It's also worth noting that Cryptography no longer supports Python 3.5, so
+admins deploying to older environments may not be protected against this or
+future vulnerabilities. Synapse will be dropping support for Python 3.5 at the
+end of March.
+
+
+Updates to the Docker image
+---------------------------
+
+- Ensure that the docker container has up to date versions of openssl. ([\#9697](https://github.com/matrix-org/synapse/issues/9697))
+
+
+Internal Changes
+----------------
+
+- Enforce that `cryptography` dependency is up to date to ensure it has the most recent openssl patches. ([\#9697](https://github.com/matrix-org/synapse/issues/9697))
+
+
+Synapse 1.30.0 (2021-03-22)
+===========================
+
+Note that this release deprecates the ability for appservices to
+call `POST /_matrix/client/r0/register`  without the body parameter `type`. Appservice
+developers should use a `type` value of `m.login.application_service` as
+per [the spec](https://matrix.org/docs/spec/application_service/r0.1.2#server-admin-style-permissions).
+In future releases, calling this endpoint with an access token - but without a `m.login.application_service`
+type - will fail.
+
+
+No significant changes.
+
+
+Synapse 1.30.0rc1 (2021-03-16)
+==============================
+
+Features
+--------
+
+- Add prometheus metrics for number of users successfully registering and logging in. ([\#9510](https://github.com/matrix-org/synapse/issues/9510), [\#9511](https://github.com/matrix-org/synapse/issues/9511), [\#9573](https://github.com/matrix-org/synapse/issues/9573))
+- Add `synapse_federation_last_sent_pdu_time` and `synapse_federation_last_received_pdu_time` prometheus metrics, which monitor federation delays by reporting the timestamps of messages sent and received to a set of remote servers. ([\#9540](https://github.com/matrix-org/synapse/issues/9540))
+- Add support for generating JSON Web Tokens dynamically for use as OIDC client secrets. ([\#9549](https://github.com/matrix-org/synapse/issues/9549))
+- Optimise handling of incomplete room history for incoming federation. ([\#9601](https://github.com/matrix-org/synapse/issues/9601))
+- Finalise support for allowing clients to pick an SSO Identity Provider ([MSC2858](https://github.com/matrix-org/matrix-doc/pull/2858)). ([\#9617](https://github.com/matrix-org/synapse/issues/9617))
+- Tell spam checker modules about the SSO IdP a user registered through if one was used. ([\#9626](https://github.com/matrix-org/synapse/issues/9626))
+
+
+Bugfixes
+--------
+
+- Fix long-standing bug when generating thumbnails for some images with transparency: `TypeError: cannot unpack non-iterable int object`. ([\#9473](https://github.com/matrix-org/synapse/issues/9473))
+- Purge chain cover indexes for events that were purged prior to Synapse v1.29.0. ([\#9542](https://github.com/matrix-org/synapse/issues/9542), [\#9583](https://github.com/matrix-org/synapse/issues/9583))
+- Fix bug where federation requests were not correctly retried on 5xx responses. ([\#9567](https://github.com/matrix-org/synapse/issues/9567))
+- Fix re-activating an account via the admin API when local passwords are disabled. ([\#9587](https://github.com/matrix-org/synapse/issues/9587))
+- Fix a bug introduced in Synapse 1.20 which caused incoming federation transactions to stack up, causing slow recovery from outages. ([\#9597](https://github.com/matrix-org/synapse/issues/9597))
+- Fix a bug introduced in v1.28.0 where the OpenID Connect callback endpoint could error with a `MacaroonInitException`. ([\#9620](https://github.com/matrix-org/synapse/issues/9620))
+- Fix Internal Server Error on `GET /_synapse/client/saml2/authn_response` request. ([\#9623](https://github.com/matrix-org/synapse/issues/9623))
+
+
+Updates to the Docker image
+---------------------------
+
+- Make use of an improved malloc implementation (`jemalloc`) in the docker image. ([\#8553](https://github.com/matrix-org/synapse/issues/8553))
+
+
+Improved Documentation
+----------------------
+
+- Add relayd entry to reverse proxy example configurations. ([\#9508](https://github.com/matrix-org/synapse/issues/9508))
+- Improve the SAML2 upgrade notes for 1.27.0. ([\#9550](https://github.com/matrix-org/synapse/issues/9550))
+- Link to the "List user's media" admin API from the media admin API docs. ([\#9571](https://github.com/matrix-org/synapse/issues/9571))
+- Clarify the spam checker modules documentation example to mention that `parse_config` is a required method. ([\#9580](https://github.com/matrix-org/synapse/issues/9580))
+- Clarify the sample configuration for `stats` settings. ([\#9604](https://github.com/matrix-org/synapse/issues/9604))
+
+
+Deprecations and Removals
+-------------------------
+
+- The `synapse_federation_last_sent_pdu_age` and `synapse_federation_last_received_pdu_age` prometheus metrics have been removed. They are replaced by `synapse_federation_last_sent_pdu_time` and `synapse_federation_last_received_pdu_time`. ([\#9540](https://github.com/matrix-org/synapse/issues/9540))
+- Registering an Application Service user without using the `m.login.application_service` login type will be unsupported in an upcoming Synapse release. ([\#9559](https://github.com/matrix-org/synapse/issues/9559))
+
+
+Internal Changes
+----------------
+
+- Add tests to ResponseCache. ([\#9458](https://github.com/matrix-org/synapse/issues/9458))
+- Add type hints to purge room and server notice admin API. ([\#9520](https://github.com/matrix-org/synapse/issues/9520))
+- Add extra logging to ObservableDeferred when callbacks throw exceptions. ([\#9523](https://github.com/matrix-org/synapse/issues/9523))
+- Fix incorrect type hints. ([\#9528](https://github.com/matrix-org/synapse/issues/9528), [\#9543](https://github.com/matrix-org/synapse/issues/9543), [\#9591](https://github.com/matrix-org/synapse/issues/9591), [\#9608](https://github.com/matrix-org/synapse/issues/9608), [\#9618](https://github.com/matrix-org/synapse/issues/9618))
+- Add an additional test for purging a room. ([\#9541](https://github.com/matrix-org/synapse/issues/9541))
+- Add a `.git-blame-ignore-revs` file with the hashes of auto-formatting. ([\#9560](https://github.com/matrix-org/synapse/issues/9560))
+- Increase the threshold before which outbound federation to a server goes into "catch up" mode, which is expensive for the remote server to handle. ([\#9561](https://github.com/matrix-org/synapse/issues/9561))
+- Fix spurious errors reported by the `config-lint.sh` script. ([\#9562](https://github.com/matrix-org/synapse/issues/9562))
+- Fix type hints and tests for BlacklistingAgentWrapper and BlacklistingReactorWrapper. ([\#9563](https://github.com/matrix-org/synapse/issues/9563))
+- Do not have mypy ignore type hints from unpaddedbase64. ([\#9568](https://github.com/matrix-org/synapse/issues/9568))
+- Improve efficiency of calculating the auth chain in large rooms. ([\#9576](https://github.com/matrix-org/synapse/issues/9576))
+- Convert `synapse.types.Requester` to an `attrs` class. ([\#9586](https://github.com/matrix-org/synapse/issues/9586))
+- Add logging for redis connection setup. ([\#9590](https://github.com/matrix-org/synapse/issues/9590))
+- Improve logging when processing incoming transactions. ([\#9596](https://github.com/matrix-org/synapse/issues/9596))
+- Remove unused `stats.retention` setting, and emit a warning if stats are disabled. ([\#9604](https://github.com/matrix-org/synapse/issues/9604))
+- Prevent attempting to bundle aggregations for state events in /context APIs. ([\#9619](https://github.com/matrix-org/synapse/issues/9619))
+
+
+Synapse 1.29.0 (2021-03-08)
+===========================
+
+Note that synapse now expects an `X-Forwarded-Proto` header when used with a reverse proxy. Please see [UPGRADE.rst](UPGRADE.rst#upgrading-to-v1290) for more details on this change.
+
+
+No significant changes.
+
+
+Synapse 1.29.0rc1 (2021-03-04)
+==============================
+
+Features
+--------
+
+- Add rate limiters to cross-user key sharing requests. ([\#8957](https://github.com/matrix-org/synapse/issues/8957))
+- Add `order_by` to the admin API `GET /_synapse/admin/v1/users/<user_id>/media`. Contributed by @dklimpel. ([\#8978](https://github.com/matrix-org/synapse/issues/8978))
+- Add some configuration settings to make users' profile data more private. ([\#9203](https://github.com/matrix-org/synapse/issues/9203))
+- The `no_proxy` and `NO_PROXY` environment variables are now respected in proxied HTTP clients with the lowercase form taking precedence if both are present. Additionally, the lowercase `https_proxy` environment variable is now respected in proxied HTTP clients on top of existing support for the uppercase `HTTPS_PROXY` form and takes precedence if both are present. Contributed by Timothy Leung. ([\#9372](https://github.com/matrix-org/synapse/issues/9372))
+- Add a configuration option, `user_directory.prefer_local_users`, which when enabled will make it more likely for users on the same server as you to appear above other users. ([\#9383](https://github.com/matrix-org/synapse/issues/9383), [\#9385](https://github.com/matrix-org/synapse/issues/9385))
+- Add support for regenerating thumbnails if they have been deleted but the original image is still stored. ([\#9438](https://github.com/matrix-org/synapse/issues/9438))
+- Add support for `X-Forwarded-Proto` header when using a reverse proxy. ([\#9472](https://github.com/matrix-org/synapse/issues/9472), [\#9501](https://github.com/matrix-org/synapse/issues/9501), [\#9512](https://github.com/matrix-org/synapse/issues/9512), [\#9539](https://github.com/matrix-org/synapse/issues/9539))
+
+
+Bugfixes
+--------
+
+- Fix a bug where users' pushers were not all deleted when they deactivated their account. ([\#9285](https://github.com/matrix-org/synapse/issues/9285), [\#9516](https://github.com/matrix-org/synapse/issues/9516))
+- Fix a bug where a lot of unnecessary presence updates were sent when joining a room. ([\#9402](https://github.com/matrix-org/synapse/issues/9402))
+- Fix a bug that caused multiple calls to the experimental `shared_rooms` endpoint to return stale results. ([\#9416](https://github.com/matrix-org/synapse/issues/9416))
+- Fix a bug in single sign-on which could cause a "No session cookie found" error. ([\#9436](https://github.com/matrix-org/synapse/issues/9436))
+- Fix bug introduced in v1.27.0 where allowing a user to choose their own username when logging in via single sign-on did not work unless an `idp_icon` was defined. ([\#9440](https://github.com/matrix-org/synapse/issues/9440))
+- Fix a bug introduced in v1.26.0 where some sequences were not properly configured when running `synapse_port_db`. ([\#9449](https://github.com/matrix-org/synapse/issues/9449))
+- Fix deleting pushers when using sharded pushers. ([\#9465](https://github.com/matrix-org/synapse/issues/9465), [\#9466](https://github.com/matrix-org/synapse/issues/9466), [\#9479](https://github.com/matrix-org/synapse/issues/9479), [\#9536](https://github.com/matrix-org/synapse/issues/9536))
+- Fix missing startup checks for the consistency of certain PostgreSQL sequences. ([\#9470](https://github.com/matrix-org/synapse/issues/9470))
+- Fix a long-standing bug where the media repository could leak file descriptors while previewing media. ([\#9497](https://github.com/matrix-org/synapse/issues/9497))
+- Properly purge the event chain cover index when purging history. ([\#9498](https://github.com/matrix-org/synapse/issues/9498))
+- Fix missing chain cover index due to a schema delta not being applied correctly. Only affected servers that ran development versions. ([\#9503](https://github.com/matrix-org/synapse/issues/9503))
+- Fix a bug introduced in v1.25.0 where `/_synapse/admin/join/` would fail when given a room alias. ([\#9506](https://github.com/matrix-org/synapse/issues/9506))
+- Prevent presence background jobs from running when presence is disabled. ([\#9530](https://github.com/matrix-org/synapse/issues/9530))
+- Fix rare edge case that caused a background update to fail if the server had rejected an event that had duplicate auth events. ([\#9537](https://github.com/matrix-org/synapse/issues/9537))
+
+
+Improved Documentation
+----------------------
+
+- Update the example systemd config to propagate reloads to individual units. ([\#9463](https://github.com/matrix-org/synapse/issues/9463))
+
+
+Internal Changes
+----------------
+
+- Add documentation and type hints to `parse_duration`. ([\#9432](https://github.com/matrix-org/synapse/issues/9432))
+- Remove vestiges of `uploads_path` configuration setting. ([\#9462](https://github.com/matrix-org/synapse/issues/9462))
+- Add a comment about systemd-python. ([\#9464](https://github.com/matrix-org/synapse/issues/9464))
+- Test that we require validated email for email pushers. ([\#9496](https://github.com/matrix-org/synapse/issues/9496))
+- Allow python to generate bytecode for synapse. ([\#9502](https://github.com/matrix-org/synapse/issues/9502))
+- Fix incorrect type hints. ([\#9515](https://github.com/matrix-org/synapse/issues/9515), [\#9518](https://github.com/matrix-org/synapse/issues/9518))
+- Add type hints to device and event report admin API. ([\#9519](https://github.com/matrix-org/synapse/issues/9519))
+- Add type hints to user admin API. ([\#9521](https://github.com/matrix-org/synapse/issues/9521))
+- Bump the versions of mypy and mypy-zope used for static type checking. ([\#9529](https://github.com/matrix-org/synapse/issues/9529))
+
+
+Synapse 1.28.0 (2021-02-25)
+===========================
+
+Note that this release drops support for ARMv7 in the official Docker images, due to repeated problems building for ARMv7 (and the associated maintenance burden this entails).
+
+This release also fixes the documentation included in v1.27.0 around the callback URI for SAML2 identity providers. If your server is configured to use single sign-on via a SAML2 IdP, you may need to make configuration changes. Please review [UPGRADE.rst](UPGRADE.rst) for more details on these changes.
+
+
+Internal Changes
+----------------
+
+- Revert change in v1.28.0rc1 to remove the deprecated SAML endpoint. ([\#9474](https://github.com/matrix-org/synapse/issues/9474))
+
+
+Synapse 1.28.0rc1 (2021-02-19)
+==============================
+
+Removal warning
+---------------
+
+The v1 list accounts API is deprecated and will be removed in a future release.
+This API was undocumented and misleading. It can be replaced by the
+[v2 list accounts API](https://github.com/matrix-org/synapse/blob/release-v1.28.0/docs/admin_api/user_admin_api.rst#list-accounts),
+which has been available since Synapse 1.7.0 (2019-12-13).
+
+Please check if you're using any scripts which use the admin API and replace
+`GET /_synapse/admin/v1/users/<user_id>` with `GET /_synapse/admin/v2/users`.
+
+
+Features
+--------
+
+- New admin API to get the context of an event: `/_synapse/admin/rooms/{roomId}/context/{eventId}`. ([\#9150](https://github.com/matrix-org/synapse/issues/9150))
+- Further improvements to the user experience of registration via single sign-on. ([\#9300](https://github.com/matrix-org/synapse/issues/9300), [\#9301](https://github.com/matrix-org/synapse/issues/9301))
+- Add hook to spam checker modules that allow checking file uploads and remote downloads. ([\#9311](https://github.com/matrix-org/synapse/issues/9311))
+- Add support for receiving OpenID Connect authentication responses via form `POST`s rather than `GET`s. ([\#9376](https://github.com/matrix-org/synapse/issues/9376))
+- Add the shadow-banning status to the admin API for user info. ([\#9400](https://github.com/matrix-org/synapse/issues/9400))
+
+
+Bugfixes
+--------
+
+- Fix long-standing bug where sending email notifications would fail for rooms that the server had since left. ([\#9257](https://github.com/matrix-org/synapse/issues/9257))
+- Fix bug introduced in Synapse 1.27.0rc1 which meant the "session expired" error page during SSO registration was badly formatted. ([\#9296](https://github.com/matrix-org/synapse/issues/9296))
+- Assert a maximum length for some parameters for spec compliance. ([\#9321](https://github.com/matrix-org/synapse/issues/9321), [\#9393](https://github.com/matrix-org/synapse/issues/9393))
+- Fix additional errors when previewing URLs: "AttributeError 'NoneType' object has no attribute 'xpath'" and "ValueError: Unicode strings with encoding declaration are not supported. Please use bytes input or XML fragments without declaration.". ([\#9333](https://github.com/matrix-org/synapse/issues/9333))
+- Fix a bug causing Synapse to impose the wrong type constraints on fields when processing responses from appservices to `/_matrix/app/v1/thirdparty/user/{protocol}`. ([\#9361](https://github.com/matrix-org/synapse/issues/9361))
+- Fix bug where Synapse would occasionally stop reconnecting to Redis after the connection was lost. ([\#9391](https://github.com/matrix-org/synapse/issues/9391))
+- Fix a long-standing bug when upgrading a room: "TypeError: '>' not supported between instances of 'NoneType' and 'int'". ([\#9395](https://github.com/matrix-org/synapse/issues/9395))
+- Reduce the amount of memory used when generating the URL preview of a file that is larger than the `max_spider_size`. ([\#9421](https://github.com/matrix-org/synapse/issues/9421))
+- Fix a long-standing bug in the deduplication of old presence, resulting in no deduplication. ([\#9425](https://github.com/matrix-org/synapse/issues/9425))
+- The `ui_auth.session_timeout` config option can now be specified in terms of number of seconds/minutes/etc/. Contributed by Rishabh Arya. ([\#9426](https://github.com/matrix-org/synapse/issues/9426))
+- Fix a bug introduced in v1.27.0: "TypeError: int() argument must be a string, a bytes-like object or a number, not 'NoneType." related to the user directory. ([\#9428](https://github.com/matrix-org/synapse/issues/9428))
+
+
+Updates to the Docker image
+---------------------------
+
+- Drop support for ARMv7 in Docker images. ([\#9433](https://github.com/matrix-org/synapse/issues/9433))
+
+
+Improved Documentation
+----------------------
+
+- Reorganize CHANGELOG.md. ([\#9281](https://github.com/matrix-org/synapse/issues/9281))
+- Add note to `auto_join_rooms` config option explaining existing rooms must be publicly joinable. ([\#9291](https://github.com/matrix-org/synapse/issues/9291))
+- Correct name of Synapse's service file in TURN howto. ([\#9308](https://github.com/matrix-org/synapse/issues/9308))
+- Fix the braces in the `oidc_providers` section of the sample config. ([\#9317](https://github.com/matrix-org/synapse/issues/9317))
+- Update installation instructions on Fedora. ([\#9322](https://github.com/matrix-org/synapse/issues/9322))
+- Add HTTP/2 support to the nginx example configuration. Contributed by David Vo. ([\#9390](https://github.com/matrix-org/synapse/issues/9390))
+- Update docs for using Gitea as OpenID provider. ([\#9404](https://github.com/matrix-org/synapse/issues/9404))
+- Document that pusher instances are shardable. ([\#9407](https://github.com/matrix-org/synapse/issues/9407))
+- Fix erroneous documentation from v1.27.0 about updating the SAML2 callback URL. ([\#9434](https://github.com/matrix-org/synapse/issues/9434))
+
+
+Deprecations and Removals
+-------------------------
+
+- Deprecate old admin API `GET /_synapse/admin/v1/users/<user_id>`. ([\#9429](https://github.com/matrix-org/synapse/issues/9429))
+
+
+Internal Changes
+----------------
+
+- Fix 'object name reserved for internal use' errors with recent versions of SQLite. ([\#9003](https://github.com/matrix-org/synapse/issues/9003))
+- Add experimental support for running Synapse with PyPy. ([\#9123](https://github.com/matrix-org/synapse/issues/9123))
+- Deny access to additional IP addresses by default. ([\#9240](https://github.com/matrix-org/synapse/issues/9240))
+- Update the `Cursor` type hints to better match PEP 249. ([\#9299](https://github.com/matrix-org/synapse/issues/9299))
+- Add debug logging for SRV lookups. Contributed by @Bubu. ([\#9305](https://github.com/matrix-org/synapse/issues/9305))
+- Improve logging for OIDC login flow. ([\#9307](https://github.com/matrix-org/synapse/issues/9307))
+- Share the code for handling required attributes between the CAS and SAML handlers. ([\#9326](https://github.com/matrix-org/synapse/issues/9326))
+- Clean up the code to load the metadata for OpenID Connect identity providers. ([\#9362](https://github.com/matrix-org/synapse/issues/9362))
+- Convert tests to use `HomeserverTestCase`. ([\#9377](https://github.com/matrix-org/synapse/issues/9377), [\#9396](https://github.com/matrix-org/synapse/issues/9396))
+- Update the version of black used to 20.8b1. ([\#9381](https://github.com/matrix-org/synapse/issues/9381))
+- Allow OIDC config to override discovered values. ([\#9384](https://github.com/matrix-org/synapse/issues/9384))
+- Remove some dead code from the acceptance of room invites path. ([\#9394](https://github.com/matrix-org/synapse/issues/9394))
+- Clean up an unused method in the presence handler code. ([\#9408](https://github.com/matrix-org/synapse/issues/9408))
+
+
 Synapse 1.27.0 (2021-02-16)
 ===========================
 
 Note that this release includes a change in Synapse to use Redis as a cache ─ as well as a pub/sub mechanism ─ if Redis support is enabled for workers. No action is needed by server administrators, and we do not expect resource usage of the Redis instance to change dramatically.
 
-This release also changes the callback URI for OpenID Connect (OIDC) identity providers. If your server is configured to use single sign-on via an OIDC/OAuth2 IdP, you may need to make configuration changes. Please review [UPGRADE.rst](UPGRADE.rst) for more details on these changes.
+This release also changes the callback URI for OpenID Connect (OIDC) and SAML2 identity providers. If your server is configured to use single sign-on via an OIDC/OAuth2 or SAML2 IdP, you may need to make configuration changes. Please review [UPGRADE.rst](UPGRADE.rst) for more details on these changes.
 
 This release also changes escaping of variables in the HTML templates for SSO or email notifications. If you have customised these templates, please review [UPGRADE.rst](UPGRADE.rst) for more details on these changes.
 

CONTRIBUTING.md

@@ -1,4 +1,31 @@
-# Contributing code to Synapse
+Welcome to Synapse
+
+This document aims to get you started with contributing to this repo! 
+
+- [1. Who can contribute to Synapse?](#1-who-can-contribute-to-synapse)
+- [2. What do I need?](#2-what-do-i-need)
+- [3. Get the source.](#3-get-the-source)
+- [4. Install the dependencies](#4-install-the-dependencies)
+  * [Under Unix (macOS, Linux, BSD, ...)](#under-unix-macos-linux-bsd-)
+  * [Under Windows](#under-windows)
+- [5. Get in touch.](#5-get-in-touch)
+- [6. Pick an issue.](#6-pick-an-issue)
+- [7. Turn coffee and documentation into code and documentation!](#7-turn-coffee-and-documentation-into-code-and-documentation)
+- [8. Test, test, test!](#8-test-test-test)
+  * [Run the linters.](#run-the-linters)
+  * [Run the unit tests.](#run-the-unit-tests)
+  * [Run the integration tests.](#run-the-integration-tests)
+- [9. Submit your patch.](#9-submit-your-patch)
+  * [Changelog](#changelog)
+    + [How do I know what to call the changelog file before I create the PR?](#how-do-i-know-what-to-call-the-changelog-file-before-i-create-the-pr)
+    + [Debian changelog](#debian-changelog)
+  * [Sign off](#sign-off)
+- [10. Turn feedback into better code.](#10-turn-feedback-into-better-code)
+- [11. Find a new issue.](#11-find-a-new-issue)
+- [Notes for maintainers on merging PRs etc](#notes-for-maintainers-on-merging-prs-etc)
+- [Conclusion](#conclusion)
+
+# 1. Who can contribute to Synapse?
 
 Everyone is welcome to contribute code to [matrix.org
 projects](https://github.com/matrix-org), provided that they are willing to
@@ -9,70 +36,179 @@ license the code under the same terms as the project's overall 'outbound'
 license - in our case, this is almost always Apache Software License v2 (see
 [LICENSE](LICENSE)).
 
-## How to contribute
+# 2. What do I need?
+
+The code of Synapse is written in Python 3. To do pretty much anything, you'll need [a recent version of Python 3](https://wiki.python.org/moin/BeginnersGuide/Download).
+
+The source code of Synapse is hosted on GitHub. You will also need [a recent version of git](https://github.com/git-guides/install-git).
+
+For some tests, you will need [a recent version of Docker](https://docs.docker.com/get-docker/).
+
+
+# 3. Get the source.
 
 The preferred and easiest way to contribute changes is to fork the relevant
-project on github, and then [create a pull request](
+project on GitHub, and then [create a pull request](
 https://help.github.com/articles/using-pull-requests/) to ask us to pull your
 changes into our repo.
 
-Some other points to follow:
+Please base your changes on the `develop` branch.
 
- * Please base your changes on the `develop` branch.
+```sh
+git clone git@github.com:YOUR_GITHUB_USER_NAME/synapse.git
+git checkout develop
+```
 
- * Please follow the [code style requirements](#code-style).
+If you need help getting started with git, this is beyond the scope of the document, but you
+can find many good git tutorials on the web.
 
- * Please include a [changelog entry](#changelog) with each PR.
+# 4. Install the dependencies
 
- * Please [sign off](#sign-off) your contribution.
+## Under Unix (macOS, Linux, BSD, ...)
 
- * Please keep an eye on the pull request for feedback from the [continuous
-   integration system](#continuous-integration-and-testing) and try to fix any
-   errors that come up.
+Once you have installed Python 3 and added the source, please open a terminal and
+setup a *virtualenv*, as follows:
 
- * If you need to [update your PR](#updating-your-pull-request), just add new
-   commits to your branch rather than rebasing.
+```sh
+cd path/where/you/have/cloned/the/repository
+python3 -m venv ./env
+source ./env/bin/activate
+pip install -e ".[all,lint,mypy,test]"
+pip install tox
+```
 
-## Code style
+This will install the developer dependencies for the project.
+
+## Under Windows
+
+TBD
+
+
+# 5. Get in touch.
+
+Join our developer community on Matrix: #synapse-dev:matrix.org !
+
+
+# 6. Pick an issue.
+
+Fix your favorite problem or perhaps find a [Good First Issue](https://github.com/matrix-org/synapse/issues?q=is%3Aopen+is%3Aissue+label%3A%22Good+First+Issue%22)
+to work on.
+
+
+# 7. Turn coffee and documentation into code and documentation!
 
 Synapse's code style is documented [here](docs/code_style.md). Please follow
 it, including the conventions for the [sample configuration
 file](docs/code_style.md#configuration-file-format).
 
-Many of the conventions are enforced by scripts which are run as part of the
-[continuous integration system](#continuous-integration-and-testing). To help
-check if you have followed the code style, you can run `scripts-dev/lint.sh`
-locally. You'll need python 3.6 or later, and to install a number of tools:
+There is a growing amount of documentation located in the [docs](docs)
+directory. This documentation is intended primarily for sysadmins running their
+own Synapse instance, as well as developers interacting externally with
+Synapse. [docs/dev](docs/dev) exists primarily to house documentation for
+Synapse developers. [docs/admin_api](docs/admin_api) houses documentation
+regarding Synapse's Admin API, which is used mostly by sysadmins and external
+service developers.
 
-```
-# Install the dependencies
-pip install -e ".[lint,mypy]"
+If you add new files added to either of these folders, please use [GitHub-Flavoured
+Markdown](https://guides.github.com/features/mastering-markdown/).
 
-# Run the linter script
+Some documentation also exists in [Synapse's GitHub
+Wiki](https://github.com/matrix-org/synapse/wiki), although this is primarily
+contributed to by community authors.
+
+
+# 8. Test, test, test!
+<a name="test-test-test"></a>
+
+While you're developing and before submitting a patch, you'll
+want to test your code.
+
+## Run the linters.
+
+The linters look at your code and do two things:
+
+- ensure that your code follows the coding style adopted by the project;
+- catch a number of errors in your code.
+
+They're pretty fast, don't hesitate!
+
+```sh
+source ./env/bin/activate
 ./scripts-dev/lint.sh
 ```
 
-**Note that the script does not just test/check, but also reformats code, so you
-may wish to ensure any new code is committed first**.
+Note that this script *will modify your files* to fix styling errors.
+Make sure that you have saved all your files.
 
-By default, this script checks all files and can take some time; if you alter
-only certain files, you might wish to specify paths as arguments to reduce the
-run-time:
+If you wish to restrict the linters to only the files changed since the last commit
+(much faster!), you can instead run:
 
+```sh
+source ./env/bin/activate
+./scripts-dev/lint.sh -d
 ```
+
+Or if you know exactly which files you wish to lint, you can instead run:
+
+```sh
+source ./env/bin/activate
 ./scripts-dev/lint.sh path/to/file1.py path/to/file2.py path/to/folder
 ```
 
-You can also provide the `-d` option, which will lint the files that have been
-changed since the last git commit. This will often be significantly faster than
-linting the whole codebase.
+## Run the unit tests.
 
-Before pushing new changes, ensure they don't produce linting errors. Commit any
-files that were corrected.
+The unit tests run parts of Synapse, including your changes, to see if anything
+was broken. They are slower than the linters but will typically catch more errors.
+
+```sh
+source ./env/bin/activate
+trial tests
+```
+
+If you wish to only run *some* unit tests, you may specify
+another module instead of `tests` - or a test class or a method:
+
+```sh
+source ./env/bin/activate
+trial tests.rest.admin.test_room tests.handlers.test_admin.ExfiltrateData.test_invite
+```
+
+If your tests fail, you may wish to look at the logs:
+
+```sh
+less _trial_temp/test.log
+```
+
+## Run the integration tests.
+
+The integration tests are a more comprehensive suite of tests. They
+run a full version of Synapse, including your changes, to check if
+anything was broken. They are slower than the unit tests but will
+typically catch more errors.
+
+The following command will let you run the integration test with the most common
+configuration:
+
+```sh
+$ docker run --rm -it -v /path/where/you/have/cloned/the/repository\:/src:ro -v /path/to/where/you/want/logs\:/logs matrixdotorg/sytest-synapse:py37
+```
+
+This configuration should generally cover  your needs. For more details about other configurations, see [documentation in the SyTest repo](https://github.com/matrix-org/sytest/blob/develop/docker/README.md).
+
+
+# 9. Submit your patch.
+
+Once you're happy with your patch, it's time to prepare a Pull Request.
+
+To prepare a Pull Request, please:
+
+1. verify that [all the tests pass](#test-test-test), including the coding style;
+2. [sign off](#sign-off) your contribution;
+3. `git push` your commit to your fork of Synapse;
+4. on GitHub, [create the Pull Request](https://docs.github.com/en/github/collaborating-with-issues-and-pull-requests/creating-a-pull-request);
+5. add a [changelog entry](#changelog) and push it to your Pull Request;
+6. for most contributors, that's all - however, if you are a member of the organization `matrix-org`, on GitHub, please request a review from `matrix.org / Synapse Core`.
 
-Please ensure your changes match the cosmetic style of the existing project,
-and **never** mix cosmetic and functional changes in the same commit, as it
-makes it horribly hard to review otherwise.
 
 ## Changelog
 
@@ -156,24 +292,6 @@ directory, you will need both a regular newsfragment *and* an entry in the
 debian changelog. (Though typically such changes should be submitted as two
 separate pull requests.)
 
-## Documentation
-
-There is a growing amount of documentation located in the [docs](docs)
-directory. This documentation is intended primarily for sysadmins running their
-own Synapse instance, as well as developers interacting externally with
-Synapse. [docs/dev](docs/dev) exists primarily to house documentation for
-Synapse developers. [docs/admin_api](docs/admin_api) houses documentation
-regarding Synapse's Admin API, which is used mostly by sysadmins and external
-service developers.
-
-New files added to both folders should be written in [Github-Flavoured
-Markdown](https://guides.github.com/features/mastering-markdown/), and attempts
-should be made to migrate existing documents to markdown where possible.
-
-Some documentation also exists in [Synapse's Github
-Wiki](https://github.com/matrix-org/synapse/wiki), although this is primarily
-contributed to by community authors.
-
 ## Sign off
 
 In order to have a concrete record that your contribution is intentional
@@ -240,47 +358,36 @@ Git allows you to add this signoff automatically when using the `-s`
 flag to `git commit`, which uses the name and email set in your
 `user.name` and `user.email` git configs.
 
-## Continuous integration and testing
 
-[Buildkite](https://buildkite.com/matrix-dot-org/synapse) will automatically
-run a series of checks and tests against any PR which is opened against the
-project; if your change breaks the build, this will be shown in GitHub, with
-links to the build results. If your build fails, please try to fix the errors
-and update your branch.
+# 10. Turn feedback into better code.
 
-To run unit tests in a local development environment, you can use:
+Once the Pull Request is opened, you will see a few things:
 
-- ``tox -e py35`` (requires tox to be installed by ``pip install tox``)
-  for SQLite-backed Synapse on Python 3.5.
-- ``tox -e py36`` for SQLite-backed Synapse on Python 3.6.
-- ``tox -e py36-postgres`` for PostgreSQL-backed Synapse on Python 3.6
-  (requires a running local PostgreSQL with access to create databases).
-- ``./test_postgresql.sh`` for PostgreSQL-backed Synapse on Python 3.5
-  (requires Docker). Entirely self-contained, recommended if you don't want to
-  set up PostgreSQL yourself.
+1. our automated CI (Continuous Integration) pipeline will run (again) the linters, the unit tests, the integration tests and more;
+2. one or more of the developers will take a look at your Pull Request and offer feedback.
 
-Docker images are available for running the integration tests (SyTest) locally,
-see the [documentation in the SyTest repo](
-https://github.com/matrix-org/sytest/blob/develop/docker/README.md) for more
-information.
+From this point, you should:
 
-## Updating your pull request
+1. Look at the results of the CI pipeline.
+   - If there is any error, fix the error.
+2. If a developer has requested changes, make these changes and let us know if it is ready for a developer to review again.
+3. Create a new commit with the changes.
+   - Please do NOT overwrite the history. New commits make the reviewer's life easier.
+   - Push this commits to your Pull Request.
+4. Back to 1.
 
-If you decide to make changes to your pull request - perhaps to address issues
-raised in a review, or to fix problems highlighted by [continuous
-integration](#continuous-integration-and-testing) - just add new commits to your
-branch, and push to GitHub. The pull request will automatically be updated.
+Once both the CI and the developers are happy, the patch will be merged into Synapse and released shortly!
 
-Please **avoid** rebasing your branch, especially once the PR has been
-reviewed: doing so makes it very difficult for a reviewer to see what has
-changed since a previous review.
+# 11. Find a new issue.
 
-## Notes for maintainers on merging PRs etc
+By now, you know the drill!
+
+# Notes for maintainers on merging PRs etc
 
 There are some notes for those with commit access to the project on how we
 manage git [here](docs/dev/git.md).
 
-## Conclusion
+# Conclusion
 
 That's it! Matrix is a very open and collaborative project as you might expect
 given our obsession with open communication. If we're going to successfully

MANIFEST.in

@@ -20,9 +20,10 @@ recursive-include scripts *
 recursive-include scripts-dev *
 recursive-include synapse *.pyi
 recursive-include tests *.py
-include tests/http/ca.crt
-include tests/http/ca.key
-include tests/http/server.key
+recursive-include tests *.pem
+recursive-include tests *.p8
+recursive-include tests *.crt
+recursive-include tests *.key
 
 recursive-include synapse/res *
 recursive-include synapse/static *.css

README.rst

@@ -183,8 +183,9 @@ Using a reverse proxy with Synapse
 It is recommended to put a reverse proxy such as
 `nginx <https://nginx.org/en/docs/http/ngx_http_proxy_module.html>`_,
 `Apache <https://httpd.apache.org/docs/current/mod/mod_proxy_http.html>`_,
-`Caddy <https://caddyserver.com/docs/quick-starts/reverse-proxy>`_ or
-`HAProxy <https://www.haproxy.org/>`_ in front of Synapse. One advantage of
+`Caddy <https://caddyserver.com/docs/quick-starts/reverse-proxy>`_,
+`HAProxy <https://www.haproxy.org/>`_ or
+`relayd <https://man.openbsd.org/relayd.8>`_ in front of Synapse. One advantage of
 doing so is that it means that you can expose the default https port (443) to
 Matrix clients without needing to run Synapse with root privileges.
 

UPGRADE.rst

@@ -85,23 +85,51 @@ for example:
      wget https://packages.matrix.org/debian/pool/main/m/matrix-synapse-py3/matrix-synapse-py3_1.3.0+stretch1_amd64.deb
      dpkg -i matrix-synapse-py3_1.3.0+stretch1_amd64.deb
 
+Upgrading to v1.29.0
+====================
+
+Requirement for X-Forwarded-Proto header
+----------------------------------------
+
+When using Synapse with a reverse proxy (in particular, when using the
+`x_forwarded` option on an HTTP listener), Synapse now expects to receive an
+`X-Forwarded-Proto` header on incoming HTTP requests. If it is not set, Synapse
+will log a warning on each received request.
+
+To avoid the warning, administrators using a reverse proxy should ensure that
+the reverse proxy sets `X-Forwarded-Proto` header to `https` or `http` to
+indicate the protocol used by the client. See the `reverse proxy documentation
+<docs/reverse_proxy.md>`_, where the example configurations have been updated to
+show how to set this header.
+
+(Users of `Caddy <https://caddyserver.com/>`_ are unaffected, since we believe it
+sets `X-Forwarded-Proto` by default.)
+
 Upgrading to v1.27.0
 ====================
 
-Changes to callback URI for OAuth2 / OpenID Connect
----------------------------------------------------
+Changes to callback URI for OAuth2 / OpenID Connect and SAML2
+-------------------------------------------------------------
 
-This version changes the URI used for callbacks from OAuth2 identity providers. If
-your server is configured for single sign-on via an OpenID Connect or OAuth2 identity
-provider, you will need to add ``[synapse public baseurl]/_synapse/client/oidc/callback``
-to the list of permitted "redirect URIs" at the identity provider.
+This version changes the URI used for callbacks from OAuth2 and SAML2 identity providers:
 
-See `docs/openid.md <docs/openid.md>`_ for more information on setting up OpenID
-Connect.
+* If your server is configured for single sign-on via an OpenID Connect or OAuth2 identity
+  provider, you will need to add ``[synapse public baseurl]/_synapse/client/oidc/callback``
+  to the list of permitted "redirect URIs" at the identity provider.
 
-(Note: a similar change is being made for SAML2; in this case the old URI
-``[synapse public baseurl]/_matrix/saml2`` is being deprecated, but will continue to
-work, so no immediate changes are required for existing installations.)
+  See `docs/openid.md <docs/openid.md>`_ for more information on setting up OpenID
+  Connect.
+
+* If your server is configured for single sign-on via a SAML2 identity provider, you will
+  need to add ``[synapse public baseurl]/_synapse/client/saml2/authn_response`` as a permitted
+  "ACS location" (also known as "allowed callback URLs") at the identity provider.
+
+  The "Issuer" in the "AuthnRequest" to the SAML2 identity provider is also updated to
+  ``[synapse public baseurl]/_synapse/client/saml2/metadata.xml``. If your SAML2 identity
+  provider uses this property to validate or otherwise identify Synapse, its configuration
+  will need to be updated to use the new URL. Alternatively you could create a new, separate
+  "EntityDescriptor" in your SAML2 identity provider with the new URLs and leave the URLs in
+  the existing "EntityDescriptor" as they were.
 
 Changes to HTML templates
 -------------------------

changelog.d/9003.misc

@@ -1 +0,0 @@
-Fix 'object name reserved for internal use' errors with recent versions of SQLite.

changelog.d/9123.misc

@@ -1 +0,0 @@
-Add experimental support for running Synapse with PyPy.

changelog.d/9150.feature

@@ -1 +0,0 @@
-New API /_synapse/admin/rooms/{roomId}/context/{eventId}.

changelog.d/9240.misc

@@ -1 +0,0 @@
-Deny access to additional IP addresses by default.

changelog.d/9257.bugfix

@@ -1 +0,0 @@
-Fix long-standing bug where sending email push would fail for rooms that the server had since left.

changelog.d/9291.doc

@@ -1 +0,0 @@
-Add note to `auto_join_rooms` config option explaining existing rooms must be publicly joinable.

changelog.d/9296.bugfix

@@ -1 +0,0 @@
-Fix bug in Synapse 1.27.0rc1 which meant the "session expired" error page during SSO registration was badly formatted.

changelog.d/9299.misc

@@ -1 +0,0 @@
-Update the `Cursor` type hints to better match PEP 249.

changelog.d/9300.feature

@@ -1 +0,0 @@
-Further improvements to the user experience of registration via single sign-on.

changelog.d/9301.feature

@@ -1 +0,0 @@
-Further improvements to the user experience of registration via single sign-on.

changelog.d/9305.misc

@@ -1 +0,0 @@
-Add debug logging for SRV lookups. Contributed by @Bubu.

changelog.d/9307.misc

@@ -1 +0,0 @@
-Improve logging for OIDC login flow.

changelog.d/9308.doc

@@ -1 +0,0 @@
-Correct name of Synapse's service file in TURN howto.

changelog.d/9311.feature

@@ -1 +0,0 @@
-Add hook to spam checker modules that allow checking file uploads and remote downloads.

changelog.d/9317.doc

@@ -1 +0,0 @@
-Fix the braces in the `oidc_providers` section of the sample config.

changelog.d/9321.bugfix

@@ -1 +0,0 @@
-Assert a maximum length for the `client_secret` parameter for spec compliance.

changelog.d/9322.doc

@@ -1 +0,0 @@
-Update installation instructions on Fedora.

changelog.d/9326.misc

@@ -1 +0,0 @@
-Share the code for handling required attributes between the CAS and SAML handlers.

changelog.d/9333.bugfix

@@ -1 +0,0 @@
-Fix additional errors when previewing URLs: "AttributeError 'NoneType' object has no attribute 'xpath'" and "ValueError: Unicode strings with encoding declaration are not supported. Please use bytes input or XML fragments without declaration.".

changelog.d/9361.bugfix

@@ -1 +0,0 @@
-Fix a bug causing Synapse to impose the wrong type constraints on fields when processing responses from appservices to `/_matrix/app/v1/thirdparty/user/{protocol}`.

changelog.d/9362.misc

@@ -1 +0,0 @@
-Clean up the code to load the metadata for OpenID Connect identity providers.

changelog.d/9376.feature

@@ -1 +0,0 @@
-Add support for receiving OpenID Connect authentication responses via form `POST`s rather than `GET`s.

changelog.d/9377.misc

@@ -1 +0,0 @@
-Convert tests to use `HomeserverTestCase`.

changelog.d/9381.misc

@@ -1 +0,0 @@
-Update the version of black used to 20.8b1.

changelog.d/9384.misc

@@ -1 +0,0 @@
-Allow OIDC config to override discovered values.

changelog.d/9391.bugfix

@@ -1 +0,0 @@
-Fix bug where Synapse would occaisonally stop reconnecting after the connection was lost.

changelog.d/9394.misc

@@ -1 +0,0 @@
-Remove some dead code from the acceptance of room invites path.

changelog.d/9395.bugfix

@@ -1 +0,0 @@
-Fix a long-standing bug when upgrading a room: "TypeError: '>' not supported between instances of 'NoneType' and 'int'".

changelog.d/9396.misc

@@ -1 +0,0 @@
-Convert tests to use `HomeserverTestCase`.

changelog.d/9404.doc

@@ -1 +0,0 @@
-Update docs for using Gitea as OpenID provider.

changelog.d/9407.doc

@@ -1 +0,0 @@
-Document that pusher instances are shardable.

debian/build_virtualenv

@@ -58,10 +58,10 @@ trap "rm -r $tmpdir" EXIT
 cp -r tests "$tmpdir"
 
 PYTHONPATH="$tmpdir" \
-    "${TARGET_PYTHON}" -B -m twisted.trial --reporter=text -j2 tests
+    "${TARGET_PYTHON}" -m twisted.trial --reporter=text -j2 tests
 
 # build the config file
-"${TARGET_PYTHON}" -B "${VIRTUALENV_DIR}/bin/generate_config" \
+"${TARGET_PYTHON}" "${VIRTUALENV_DIR}/bin/generate_config" \
         --config-dir="/etc/matrix-synapse" \
         --data-dir="/var/lib/matrix-synapse" |
     perl -pe '
@@ -87,7 +87,7 @@ PYTHONPATH="$tmpdir" \
 ' > "${PACKAGE_BUILD_DIR}/etc/matrix-synapse/homeserver.yaml"
 
 # build the log config file
-"${TARGET_PYTHON}" -B "${VIRTUALENV_DIR}/bin/generate_log_config" \
+"${TARGET_PYTHON}" "${VIRTUALENV_DIR}/bin/generate_log_config" \
         --output-file="${PACKAGE_BUILD_DIR}/etc/matrix-synapse/log.yaml"
 
 # add a dependency on the right version of python to substvars.

debian/changelog

@@ -1,3 +1,31 @@
+matrix-synapse-py3 (1.30.1) stable; urgency=medium
+
+  * New synapse release 1.30.1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Fri, 26 Mar 2021 12:01:28 +0000
+
+matrix-synapse-py3 (1.30.0) stable; urgency=medium
+
+  * New synapse release 1.30.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Mon, 22 Mar 2021 13:15:34 +0000
+
+matrix-synapse-py3 (1.29.0) stable; urgency=medium
+
+  [ Jonathan de Jong ]
+  * Remove the python -B flag (don't generate bytecode) in scripts and documentation.
+
+  [ Synapse Packaging team ]
+  * New synapse release 1.29.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Mon, 08 Mar 2021 13:51:50 +0000
+
+matrix-synapse-py3 (1.28.0) stable; urgency=medium
+
+  * New synapse release 1.28.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Thu, 25 Feb 2021 10:21:57 +0000
+
 matrix-synapse-py3 (1.27.0) stable; urgency=medium
 
   [ Dan Callahan ]

debian/synctl.1

@@ -44,7 +44,7 @@ Configuration file may be generated as follows:
 .
 .nf
 
-$ python \-B \-m synapse\.app\.homeserver \-c config\.yaml \-\-generate\-config \-\-server\-name=<server name>
+$ python \-m synapse\.app\.homeserver \-c config\.yaml \-\-generate\-config \-\-server\-name=<server name>
 .
 .fi
 .

debian/synctl.ronn

@@ -41,7 +41,7 @@ process.
 
 Configuration file may be generated as follows:
 
-    $ python -B -m synapse.app.homeserver -c config.yaml --generate-config --server-name=<server name>
+    $ python -m synapse.app.homeserver -c config.yaml --generate-config --server-name=<server name>
 
 ## ENVIRONMENT
 

docker/Dockerfile

@@ -20,17 +20,18 @@ 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 \
-    rustc \
-    zlib1g-dev \
- && rm -rf /var/lib/apt/lists/*
+        build-essential \
+        libffi-dev \
+        libjpeg-dev \
+        libpq-dev \
+        libssl-dev \
+        libwebp-dev \
+        libxml++2.6-dev \
+        libxslt1-dev \
+        openssl \
+        rustc \
+        zlib1g-dev \
+        && rm -rf /var/lib/apt/lists/*
 
 # Build dependencies that are not available as wheels, to speed up rebuilds
 RUN pip install --prefix="/install" --no-warn-script-location \
@@ -63,13 +64,16 @@ RUN pip install --prefix="/install" --no-warn-script-location \
 FROM docker.io/python:${PYTHON_VERSION}-slim
 
 RUN apt-get update && apt-get install -y \
-    curl \
-    gosu \
-    libjpeg62-turbo \
-    libpq5 \
-    libwebp6 \
-    xmlsec1 \
- && rm -rf /var/lib/apt/lists/*
+        curl \
+        gosu \
+        libjpeg62-turbo \
+        libpq5 \
+        libwebp6 \
+        xmlsec1 \
+        libjemalloc2 \
+        libssl-dev \
+        openssl \
+        && rm -rf /var/lib/apt/lists/*
 
 COPY --from=builder /install /usr/local
 COPY ./docker/start.py /start.py
@@ -82,4 +86,4 @@ EXPOSE 8008/tcp 8009/tcp 8448/tcp
 ENTRYPOINT ["/start.py"]
 
 HEALTHCHECK --interval=1m --timeout=5s \
-  CMD curl -fSs http://localhost:8008/health || exit 1
+        CMD curl -fSs http://localhost:8008/health || exit 1

docker/README.md

@@ -11,7 +11,6 @@ The image also does *not* provide a TURN server.
 By default, the image expects a single volume, located at ``/data``, that will hold:
 
 * configuration files;
-* temporary files during uploads;
 * uploaded media and thumbnails;
 * the SQLite database if you do not configure postgres;
 * the appservices configuration.
@@ -205,3 +204,8 @@ healthcheck:
   timeout: 10s
   retries: 3
 ```
+
+## Using jemalloc
+
+Jemalloc is embedded in the image and will be used instead of the default allocator.
+You can read about jemalloc by reading the Synapse [README](../README.md)

docker/conf/homeserver.yaml

@@ -89,7 +89,6 @@ federation_rc_concurrent: 3
 ## Files ##
 
 media_store_path: "/data/media"
-uploads_path: "/data/uploads"
 max_upload_size: "{{ SYNAPSE_MAX_UPLOAD_SIZE or "50M" }}"
 max_image_pixels: "32M"
 dynamic_thumbnails: false

docker/start.py

@@ -3,6 +3,7 @@
 import codecs
 import glob
 import os
+import platform
 import subprocess
 import sys
 
@@ -213,6 +214,13 @@ def main(args, environ):
     if "-m" not in args:
         args = ["-m", synapse_worker] + args
 
+    jemallocpath = "/usr/lib/%s-linux-gnu/libjemalloc.so.2" % (platform.machine(),)
+
+    if os.path.isfile(jemallocpath):
+        environ["LD_PRELOAD"] = jemallocpath
+    else:
+        log("Could not find %s, will not use" % (jemallocpath,))
+
     # if there are no config files passed to synapse, try adding the default file
     if not any(p.startswith("--config-path") or p.startswith("-c") for p in args):
         config_dir = environ.get("SYNAPSE_CONFIG_DIR", "/data")
@@ -248,9 +256,9 @@ running with 'migrate_config'. See the README for more details.
     args = ["python"] + args
     if ownership is not None:
         args = ["gosu", ownership] + args
-        os.execv("/usr/sbin/gosu", args)
+        os.execve("/usr/sbin/gosu", args, environ)
     else:
-        os.execv("/usr/local/bin/python", args)
+        os.execve("/usr/local/bin/python", args, environ)
 
 
 if __name__ == "__main__":

docs/admin_api/media_admin_api.md

@@ -1,5 +1,7 @@
 # Contents
-- [List all media in a room](#list-all-media-in-a-room)
+- [Querying media](#querying-media)
+  * [List all media in a room](#list-all-media-in-a-room)
+  * [List all media uploaded by a user](#list-all-media-uploaded-by-a-user)
 - [Quarantine media](#quarantine-media)
   * [Quarantining media by ID](#quarantining-media-by-id)
   * [Quarantining media in a room](#quarantining-media-in-a-room)
@@ -10,7 +12,11 @@
   * [Delete local media by date or size](#delete-local-media-by-date-or-size)
 - [Purge Remote Media API](#purge-remote-media-api)
 
-# List all media in a room
+# Querying media
+
+These APIs allow extracting media information from the homeserver.
+
+## List all media in a room
 
 This API gets a list of known media in a room.
 However, it only shows media from unencrypted events or rooms.
@@ -36,6 +42,12 @@ The API returns a JSON body like the following:
 }
 ```
 
+## List all media uploaded by a user
+
+Listing all media that has been uploaded by a local user can be achieved through
+the use of the [List media of a user](user_admin_api.rst#list-media-of-a-user)
+Admin API.
+
 # Quarantine media
 
 Quarantining media means that it is marked as inaccessible by users. It applies

docs/admin_api/user_admin_api.rst

@@ -29,8 +29,9 @@ It returns a JSON body like the following:
             }
         ],
         "avatar_url": "<avatar_url>",
-        "admin": false,
-        "deactivated": false,
+        "admin": 0,
+        "deactivated": 0,
+        "shadow_banned": 0,
         "password_hash": "$2b$12$p9B4GkqYdRTPGD",
         "creation_ts": 1560432506,
         "appservice_id": null,
@@ -150,6 +151,7 @@ A JSON body is returned with the following shape:
                 "admin": 0,
                 "user_type": null,
                 "deactivated": 0,
+                "shadow_banned": 0,
                 "displayname": "<User One>",
                 "avatar_url": null
             }, {
@@ -158,6 +160,7 @@ A JSON body is returned with the following shape:
                 "admin": 1,
                 "user_type": null,
                 "deactivated": 0,
+                "shadow_banned": 0,
                 "displayname": "<User Two>",
                 "avatar_url": "<avatar_url>"
             }
@@ -262,7 +265,7 @@ The following actions are performed when deactivating an user:
 - Reject all pending invites
 - Remove all account validity information related to the user
 
-The following additional actions are performed during deactivation if``erase``
+The following additional actions are performed during deactivation if ``erase``
 is set to ``true``:
 
 - Remove the user's display name
@@ -376,11 +379,12 @@ The following fields are returned in the JSON response body:
 - ``total`` - Number of rooms.
 
 
-List media of an user
-================================
+List media of a user
+====================
 Gets a list of all local media that a specific ``user_id`` has created.
-The response is ordered by creation date descending and media ID descending.
-The newest media is on top.
+By default, the response is ordered by descending creation date and ascending media ID.
+The newest media is on top. You can change the order with parameters
+``order_by`` and ``dir``.
 
 The API is::
 
@@ -437,6 +441,35 @@ The following parameters should be set in the URL:
   denoting the offset in the returned results. This should be treated as an opaque value and
   not explicitly set to anything other than the return value of ``next_token`` from a previous call.
   Defaults to ``0``.
+- ``order_by`` - The method by which to sort the returned list of media.
+  If the ordered field has duplicates, the second order is always by ascending ``media_id``,
+  which guarantees a stable ordering. Valid values are:
+
+  - ``media_id`` - Media are ordered alphabetically by ``media_id``.
+  - ``upload_name`` - Media are ordered alphabetically by name the media was uploaded with.
+  - ``created_ts`` - Media are ordered by when the content was uploaded in ms.
+    Smallest to largest. This is the default.
+  - ``last_access_ts`` - Media are ordered by when the content was last accessed in ms.
+    Smallest to largest.
+  - ``media_length`` - Media are ordered by length of the media in bytes.
+    Smallest to largest.
+  - ``media_type`` - Media are ordered alphabetically by MIME-type.
+  - ``quarantined_by`` - Media are ordered alphabetically by the user ID that
+    initiated the quarantine request for this media.
+  - ``safe_from_quarantine`` - Media are ordered by the status if this media is safe
+    from quarantining.
+
+- ``dir`` - Direction of media order. Either ``f`` for forwards or ``b`` for backwards.
+  Setting this value to ``b`` will reverse the above sort order. Defaults to ``f``.
+
+If neither ``order_by`` nor ``dir`` is set, the default order is newest media on top
+(corresponds to ``order_by`` = ``created_ts`` and ``dir`` = ``b``).
+
+Caution. The database only has indexes on the columns ``media_id``,
+``user_id`` and ``created_ts``. This means that if a different sort order is used
+(``upload_name``, ``last_access_ts``, ``media_length``, ``media_type``,
+``quarantined_by`` or ``safe_from_quarantine``), this can cause a large load on the
+database, especially for large environments.
 
 **Response**
 

docs/openid.md

@@ -226,7 +226,7 @@ Synapse config:
 oidc_providers:
   - idp_id: github
     idp_name: Github
-    idp_brand: "org.matrix.github"  # optional: styling hint for clients
+    idp_brand: "github"  # optional: styling hint for clients
     discover: false
     issuer: "https://github.com/"
     client_id: "your-client-id" # TO BE FILLED
@@ -252,7 +252,7 @@ oidc_providers:
    oidc_providers:
      - idp_id: google
        idp_name: Google
-       idp_brand: "org.matrix.google"  # optional: styling hint for clients
+       idp_brand: "google"  # optional: styling hint for clients
        issuer: "https://accounts.google.com/"
        client_id: "your-client-id" # TO BE FILLED
        client_secret: "your-client-secret" # TO BE FILLED
@@ -299,7 +299,7 @@ Synapse config:
 oidc_providers:
   - idp_id: gitlab
     idp_name: Gitlab
-    idp_brand: "org.matrix.gitlab"  # optional: styling hint for clients
+    idp_brand: "gitlab"  # optional: styling hint for clients
     issuer: "https://gitlab.com/"
     client_id: "your-client-id" # TO BE FILLED
     client_secret: "your-client-secret" # TO BE FILLED
@@ -334,7 +334,7 @@ Synapse config:
 ```yaml
   - idp_id: facebook
     idp_name: Facebook
-    idp_brand: "org.matrix.facebook"  # optional: styling hint for clients
+    idp_brand: "facebook"  # optional: styling hint for clients
     discover: false
     issuer: "https://facebook.com"
     client_id: "your-client-id" # TO BE FILLED
@@ -386,7 +386,7 @@ oidc_providers:
       config:
         subject_claim: "id"
         localpart_template: "{{ user.login }}"
-        display_name_template: "{{ user.full_name }}" 
+        display_name_template: "{{ user.full_name }}"
 ```
 
 ### XWiki
@@ -401,8 +401,7 @@ oidc_providers:
     idp_name: "XWiki"
     issuer: "https://myxwikihost/xwiki/oidc/"
     client_id: "your-client-id" # TO BE FILLED
-    # Needed until https://github.com/matrix-org/synapse/issues/9212 is fixed
-    client_secret: "dontcare"
+    client_auth_method: none
     scopes: ["openid", "profile"]
     user_profile_method: "userinfo_endpoint"
     user_mapping_provider:
@@ -410,3 +409,40 @@ oidc_providers:
         localpart_template: "{{ user.preferred_username }}"
         display_name_template: "{{ user.name }}"
 ```
+
+## Apple
+
+Configuring "Sign in with Apple" (SiWA) requires an Apple Developer account.
+
+You will need to create a new "Services ID" for SiWA, and create and download a
+private key with "SiWA" enabled.
+
+As well as the private key file, you will need:
+ * Client ID: the "identifier" you gave the "Services ID"
+ * Team ID: a 10-character ID associated with your developer account.
+ * Key ID: the 10-character identifier for the key.
+
+https://help.apple.com/developer-account/?lang=en#/dev77c875b7e has more
+documentation on setting up SiWA.
+
+The synapse config will look like this:
+
+```yaml
+  - idp_id: apple
+    idp_name: Apple
+    issuer: "https://appleid.apple.com"
+    client_id: "your-client-id" # Set to the "identifier" for your "ServicesID"
+    client_auth_method: "client_secret_post"
+    client_secret_jwt_key:
+      key_file: "/path/to/AuthKey_KEYIDCODE.p8"  # point to your key file
+      jwt_header:
+        alg: ES256
+        kid: "KEYIDCODE"   # Set to the 10-char Key ID
+      jwt_payload:
+        iss: TEAMIDCODE    # Set to the 10-char Team ID
+    scopes: ["name", "email", "openid"]
+    authorization_endpoint: https://appleid.apple.com/auth/authorize?response_mode=form_post
+    user_mapping_provider:
+      config:
+        email_template: "{{ user.email }}"
+```

docs/reverse_proxy.md

@@ -3,30 +3,31 @@
 It is recommended to put a reverse proxy such as
 [nginx](https://nginx.org/en/docs/http/ngx_http_proxy_module.html),
 [Apache](https://httpd.apache.org/docs/current/mod/mod_proxy_http.html),
-[Caddy](https://caddyserver.com/docs/quick-starts/reverse-proxy) or
-[HAProxy](https://www.haproxy.org/) in front of Synapse. One advantage
+[Caddy](https://caddyserver.com/docs/quick-starts/reverse-proxy),
+[HAProxy](https://www.haproxy.org/) or
+[relayd](https://man.openbsd.org/relayd.8) in front of Synapse. One advantage
 of doing so is that it means that you can expose the default https port
 (443) to Matrix clients without needing to run Synapse with root
 privileges.
 
+You should configure your reverse proxy to forward requests to `/_matrix` or
+`/_synapse/client` to Synapse, and have it set the `X-Forwarded-For` and
+`X-Forwarded-Proto` request headers.
+
+You should remember that Matrix clients and other Matrix servers do not
+necessarily need to connect to your server via the same server name or
+port. Indeed, clients will use port 443 by default, whereas servers default to
+port 8448. Where these are different, we refer to the 'client port' and the
+'federation port'. See [the Matrix
+specification](https://matrix.org/docs/spec/server_server/latest#resolving-server-names)
+for more details of the algorithm used for federation connections, and
+[delegate.md](<delegate.md>) for instructions on setting up delegation.
+
 **NOTE**: Your reverse proxy must not `canonicalise` or `normalise`
 the requested URI in any way (for example, by decoding `%xx` escapes).
 Beware that Apache *will* canonicalise URIs unless you specify
 `nocanon`.
 
-When setting up a reverse proxy, remember that Matrix clients and other
-Matrix servers do not necessarily need to connect to your server via the
-same server name or port. Indeed, clients will use port 443 by default,
-whereas servers default to port 8448. Where these are different, we
-refer to the 'client port' and the 'federation port'. See [the Matrix
-specification](https://matrix.org/docs/spec/server_server/latest#resolving-server-names)
-for more details of the algorithm used for federation connections, and
-[delegate.md](<delegate.md>) for instructions on setting up delegation.
-
-Endpoints that are part of the standardised Matrix specification are
-located under `/_matrix`, whereas endpoints specific to Synapse are
-located under `/_synapse/client`.
-
 Let's assume that we expect clients to connect to our server at
 `https://matrix.example.com`, and other servers to connect at
 `https://example.com:8448`.  The following sections detail the configuration of
@@ -40,18 +41,21 @@ the reverse proxy and the homeserver.
 
 ```
 server {
-    listen 443 ssl;
-    listen [::]:443 ssl;
+    listen 443 ssl http2;
+    listen [::]:443 ssl http2;
 
     # For the federation port
-    listen 8448 ssl default_server;
-    listen [::]:8448 ssl default_server;
+    listen 8448 ssl http2 default_server;
+    listen [::]:8448 ssl http2 default_server;
 
     server_name matrix.example.com;
 
     location ~* ^(\/_matrix|\/_synapse\/client) {
         proxy_pass http://localhost:8008;
         proxy_set_header X-Forwarded-For $remote_addr;
+        proxy_set_header X-Forwarded-Proto $scheme;
+        proxy_set_header Host $host;
+
         # Nginx by default only allows file uploads up to 1M in size
         # Increase client_max_body_size to match max_upload_size defined in homeserver.yaml
         client_max_body_size 50M;
@@ -102,6 +106,7 @@ example.com:8448 {
     SSLEngine on
     ServerName matrix.example.com;
 
+    RequestHeader set "X-Forwarded-Proto" expr=%{REQUEST_SCHEME}
     AllowEncodedSlashes NoDecode
     ProxyPass /_matrix http://127.0.0.1:8008/_matrix nocanon
     ProxyPassReverse /_matrix http://127.0.0.1:8008/_matrix
@@ -113,6 +118,7 @@ example.com:8448 {
     SSLEngine on
     ServerName example.com;
 
+    RequestHeader set "X-Forwarded-Proto" expr=%{REQUEST_SCHEME}
     AllowEncodedSlashes NoDecode
     ProxyPass /_matrix http://127.0.0.1:8008/_matrix nocanon
     ProxyPassReverse /_matrix http://127.0.0.1:8008/_matrix
@@ -134,6 +140,9 @@ example.com:8448 {
 ```
 frontend https
   bind :::443 v4v6 ssl crt /etc/ssl/haproxy/ strict-sni alpn h2,http/1.1
+  http-request set-header X-Forwarded-Proto https if { ssl_fc }
+  http-request set-header X-Forwarded-Proto http if !{ ssl_fc }
+  http-request set-header X-Forwarded-For %[src]
 
   # Matrix client traffic
   acl matrix-host hdr(host) -i matrix.example.com
@@ -144,12 +153,62 @@ frontend https
 
 frontend matrix-federation
   bind :::8448 v4v6 ssl crt /etc/ssl/haproxy/synapse.pem alpn h2,http/1.1
+  http-request set-header X-Forwarded-Proto https if { ssl_fc }
+  http-request set-header X-Forwarded-Proto http if !{ ssl_fc }
+  http-request set-header X-Forwarded-For %[src]
+
   default_backend matrix
 
 backend matrix
   server matrix 127.0.0.1:8008
 ```
 
+### Relayd
+
+```
+table <webserver>    { 127.0.0.1 }
+table <matrixserver> { 127.0.0.1 }
+
+http protocol "https" {
+    tls { no tlsv1.0, ciphers "HIGH" }
+    tls keypair "example.com"
+    match header set "X-Forwarded-For"   value "$REMOTE_ADDR"
+    match header set "X-Forwarded-Proto" value "https"
+
+    # set CORS header for .well-known/matrix/server, .well-known/matrix/client
+    # httpd does not support setting headers, so do it here
+    match request path "/.well-known/matrix/*" tag "matrix-cors"
+    match response tagged "matrix-cors" header set "Access-Control-Allow-Origin" value "*"
+
+    pass quick path "/_matrix/*"         forward to <matrixserver>
+    pass quick path "/_synapse/client/*" forward to <matrixserver>
+
+    # pass on non-matrix traffic to webserver
+    pass                                 forward to <webserver>
+}
+
+relay "https_traffic" {
+    listen on egress port 443 tls
+    protocol "https"
+    forward to <matrixserver> port 8008 check tcp
+    forward to <webserver>    port 8080 check tcp
+}
+
+http protocol "matrix" {
+    tls { no tlsv1.0, ciphers "HIGH" }
+    tls keypair "example.com"
+    block
+    pass quick path "/_matrix/*"         forward to <matrixserver>
+    pass quick path "/_synapse/client/*" forward to <matrixserver>
+}
+
+relay "matrix_federation" {
+    listen on egress port 8448 tls
+    protocol "matrix"
+    forward to <matrixserver> port 8008 check tcp
+}
+```
+
 ## Homeserver Configuration
 
 You will also want to set `bind_addresses: ['127.0.0.1']` and

docs/sample_config.yaml

@@ -89,8 +89,7 @@ pid_file: DATADIR/homeserver.pid
 # Whether to require authentication to retrieve profile data (avatars,
 # display names) of other users through the client API. Defaults to
 # 'false'. Note that profile data is also available via the federation
-# API, so this setting is of limited value if federation is enabled on
-# the server.
+# API, unless allow_profile_lookup_over_federation is set to false.
 #
 #require_auth_for_profile_requests: true
 
@@ -101,6 +100,14 @@ pid_file: DATADIR/homeserver.pid
 #
 #limit_profile_requests_to_users_who_share_rooms: true
 
+# Uncomment to prevent a user's profile data from being retrieved and
+# displayed in a room until they have joined it. By default, a user's
+# profile data is included in an invite event, regardless of the values
+# of the above two settings, and whether or not the users share a server.
+# Defaults to 'true'.
+#
+#include_profile_data_on_invite: false
+
 # If set to 'true', removes the need for authentication to access the server's
 # public rooms directory through the client API, meaning that anyone can
 # query the room directory. Defaults to 'false'.
@@ -699,6 +706,12 @@ acme:
 #  - matrix.org
 #  - example.com
 
+# Uncomment to disable profile lookup over federation. By default, the
+# Federation API allows other homeservers to obtain profile data of any user
+# on this homeserver. Defaults to 'true'.
+#
+#allow_profile_lookup_over_federation: false
+
 
 ## Caching ##
 
@@ -1766,7 +1779,26 @@ saml2_config:
 #
 #   client_id: Required. oauth2 client id to use.
 #
-#   client_secret: Required. oauth2 client secret to use.
+#   client_secret: oauth2 client secret to use. May be omitted if
+#        client_secret_jwt_key is given, or if client_auth_method is 'none'.
+#
+#   client_secret_jwt_key: Alternative to client_secret: details of a key used
+#      to create a JSON Web Token to be used as an OAuth2 client secret. If
+#      given, must be a dictionary with the following properties:
+#
+#          key: a pem-encoded signing key. Must be a suitable key for the
+#              algorithm specified. Required unless 'key_file' is given.
+#
+#          key_file: the path to file containing a pem-encoded signing key file.
+#              Required unless 'key' is given.
+#
+#          jwt_header: a dictionary giving properties to include in the JWT
+#              header. Must include the key 'alg', giving the algorithm used to
+#              sign the JWT, such as "ES256", using the JWA identifiers in
+#              RFC7518.
+#
+#          jwt_payload: an optional dictionary giving properties to include in
+#              the JWT payload. Normally this should include an 'iss' key.
 #
 #   client_auth_method: auth method to use when exchanging the token. Valid
 #       values are 'client_secret_basic' (default), 'client_secret_post' and
@@ -1887,7 +1919,7 @@ oidc_providers:
   #
   #- idp_id: github
   #  idp_name: Github
-  #  idp_brand: org.matrix.github
+  #  idp_brand: github
   #  discover: false
   #  issuer: "https://github.com/"
   #  client_id: "your-client-id" # TO BE FILLED
@@ -2228,8 +2260,8 @@ password_config:
       #require_uppercase: true
 
 ui_auth:
-    # The number of milliseconds to allow a user-interactive authentication
-    # session to be active.
+    # The amount of time to allow a user-interactive authentication session
+    # to be active.
     #
     # This defaults to 0, meaning the user is queried for their credentials
     # before every action, but this can be overridden to allow a single
@@ -2240,7 +2272,7 @@ ui_auth:
     # Uncomment below to allow for credential validation to last for 15
     # seconds.
     #
-    #session_timeout: 15000
+    #session_timeout: "15s"
 
 
 # Configuration for sending emails from Synapse.
@@ -2530,19 +2562,35 @@ spam_checker:
 
 # User Directory configuration
 #
-# 'enabled' defines whether users can search the user directory. If
-# false then empty responses are returned to all queries. Defaults to
-# true.
-#
-# 'search_all_users' defines whether to search all users visible to your HS
-# when searching the user directory, rather than limiting to users visible
-# in public rooms.  Defaults to false.  If you set it True, you'll have to
-# rebuild the user_directory search indexes, see
-# https://github.com/matrix-org/synapse/blob/master/docs/user_directory.md
-#
-#user_directory:
-#  enabled: true
-#  search_all_users: false
+user_directory:
+    # Defines whether users can search the user directory. If false then
+    # empty responses are returned to all queries. Defaults to true.
+    #
+    # Uncomment to disable the user directory.
+    #
+    #enabled: false
+
+    # Defines whether to search all users visible to your HS when searching
+    # the user directory, rather than limiting to users visible in public
+    # rooms. Defaults to false.
+    #
+    # If you set it true, you'll have to rebuild the user_directory search
+    # indexes, see:
+    # https://github.com/matrix-org/synapse/blob/master/docs/user_directory.md
+    #
+    # Uncomment to return search results containing all known users, even if that
+    # user does not share a room with the requester.
+    #
+    #search_all_users: true
+
+    # Defines whether to prefer local users in search query results.
+    # If True, local users are more likely to appear above remote users
+    # when searching the user directory. Defaults to false.
+    #
+    # Uncomment to prefer local over remote users in user directory search
+    # results.
+    #
+    #prefer_local_users: true
 
 
 # User Consent configuration
@@ -2597,19 +2645,20 @@ spam_checker:
 
 
 
-# Local statistics collection. Used in populating the room directory.
+# Settings for local room and user statistics collection. See
+# docs/room_and_user_statistics.md.
 #
-# 'bucket_size' controls how large each statistics timeslice is. It can
-# be defined in a human readable short form -- e.g. "1d", "1y".
-#
-# 'retention' controls how long historical statistics will be kept for.
-# It can be defined in a human readable short form -- e.g. "1d", "1y".
-#
-#
-#stats:
-#   enabled: true
-#   bucket_size: 1d
-#   retention: 1y
+stats:
+  # Uncomment the following to disable room and user statistics. Note that doing
+  # so may cause certain features (such as the room directory) not to work
+  # correctly.
+  #
+  #enabled: false
+
+  # The size of each timeslice in the room_stats_historical and
+  # user_stats_historical tables, as a time period. Defaults to "1d".
+  #
+  #bucket_size: 1h
 
 
 # Server Notices room configuration

docs/spam_checker.md

@@ -14,6 +14,7 @@ The Python class is instantiated with two objects:
 * An instance of `synapse.module_api.ModuleApi`.
 
 It then implements methods which return a boolean to alter behavior in Synapse.
+All the methods must be defined.
 
 There's a generic method for checking every event (`check_event_for_spam`), as
 well as some specific methods:
@@ -24,13 +25,18 @@ well as some specific methods:
 * `user_may_publish_room`
 * `check_username_for_spam`
 * `check_registration_for_spam`
+* `check_media_file_for_spam`
 
-The details of the each of these methods (as well as their inputs and outputs)
+The details of each of these methods (as well as their inputs and outputs)
 are documented in the `synapse.events.spamcheck.SpamChecker` class.
 
 The `ModuleApi` class provides a way for the custom spam checker class to
 call back into the homeserver internals.
 
+Additionally, a `parse_config` method is mandatory and receives the plugin config
+dictionary. After parsing, It must return an object which will be
+passed to `__init__` later.
+
 ### Example
 
 ```python
@@ -41,6 +47,10 @@ class ExampleSpamChecker:
         self.config = config
         self.api = api
 
+    @staticmethod
+    def parse_config(config):
+        return config
+        
     async def check_event_for_spam(self, foo):
         return False  # allow all events
 
@@ -59,7 +69,13 @@ class ExampleSpamChecker:
     async def check_username_for_spam(self, user_profile):
         return False  # allow all usernames
 
-    async def check_registration_for_spam(self, email_threepid, username, request_info):
+    async def check_registration_for_spam(
+        self,
+        email_threepid,
+        username,
+        request_info,
+        auth_provider_id,
+    ):
         return RegistrationBehaviour.ALLOW  # allow all registrations
 
     async def check_media_file_for_spam(self, file_wrapper, file_info):

docs/systemd-with-workers/system/matrix-synapse-worker@.service

@@ -4,6 +4,7 @@ AssertPathExists=/etc/matrix-synapse/workers/%i.yaml
 
 # This service should be restarted when the synapse target is restarted.
 PartOf=matrix-synapse.target
+ReloadPropagatedFrom=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.

docs/systemd-with-workers/system/matrix-synapse.service

@@ -3,6 +3,7 @@ Description=Synapse master
 
 # This service should be restarted when the synapse target is restarted.
 PartOf=matrix-synapse.target
+ReloadPropagatedFrom=matrix-synapse.target
 
 [Service]
 Type=notify

docs/tcp_replication.md

@@ -220,10 +220,6 @@ Asks the server for the current position of all streams.
 
    Acknowledge receipt of some federation data
 
-#### REMOVE_PUSHER (C)
-
-   Inform the server a pusher should be removed
-
 ### REMOTE_SERVER_UP (S, C)
 
    Inform other processes that a remote server may have come back online.

docs/workers.md

@@ -276,7 +276,8 @@ using):
 
 Ensure that all SSO logins go to a single process.
 For multiple workers not handling the SSO endpoints properly, see
-[#7530](https://github.com/matrix-org/synapse/issues/7530).
+[#7530](https://github.com/matrix-org/synapse/issues/7530) and 
+[#9427](https://github.com/matrix-org/synapse/issues/9427).
 
 Note that a HTTP listener with `client` and `federation` resources must be
 configured in the `worker_listeners` option in the worker config.

mypy.ini

@@ -23,6 +23,7 @@ files =
   synapse/events/validator.py,
   synapse/events/spamcheck.py,
   synapse/federation,
+  synapse/groups,
   synapse/handlers,
   synapse/http/client.py,
   synapse/http/federation/matrix_federation_agent.py,
@@ -68,6 +69,7 @@ files =
   synapse/util/async_helpers.py,
   synapse/util/caches,
   synapse/util/metrics.py,
+  synapse/util/macaroons.py,
   synapse/util/stringutils.py,
   tests/replication,
   tests/test_utils,
@@ -115,9 +117,6 @@ ignore_missing_imports = True
 [mypy-saml2.*]
 ignore_missing_imports = True
 
-[mypy-unpaddedbase64]
-ignore_missing_imports = True
-
 [mypy-canonicaljson]
 ignore_missing_imports = True
 

scripts-dev/config-lint.sh

@@ -2,9 +2,14 @@
 # Find linting errors in Synapse's default config file.
 # Exits with 0 if there are no problems, or another code otherwise.
 
+# cd to the root of the repository
+cd `dirname $0`/..
+
+# Restore backup of sample config upon script exit
+trap "mv docs/sample_config.yaml.bak docs/sample_config.yaml" EXIT
+
 # Fix non-lowercase true/false values
 sed -i.bak -E "s/: +True/: true/g; s/: +False/: false/g;" docs/sample_config.yaml
-rm docs/sample_config.yaml.bak
 
 # Check if anything changed
-git diff --exit-code docs/sample_config.yaml
+diff docs/sample_config.yaml docs/sample_config.yaml.bak

scripts/synapse_port_db

@@ -22,7 +22,7 @@ import logging
 import sys
 import time
 import traceback
-from typing import Dict, Optional, Set
+from typing import Dict, Iterable, Optional, Set
 
 import yaml
 
@@ -47,6 +47,7 @@ from synapse.storage.databases.main.events_bg_updates import (
 from synapse.storage.databases.main.media_repository import (
     MediaRepositoryBackgroundUpdateStore,
 )
+from synapse.storage.databases.main.pusher import PusherWorkerStore
 from synapse.storage.databases.main.registration import (
     RegistrationBackgroundUpdateStore,
     find_max_generated_user_id_localpart,
@@ -177,6 +178,7 @@ class Store(
     UserDirectoryBackgroundUpdateStore,
     EndToEndKeyBackgroundStore,
     StatsStore,
+    PusherWorkerStore,
 ):
     def execute(self, f, *args, **kwargs):
         return self.db_pool.runInteraction(f.__name__, f, *args, **kwargs)
@@ -629,7 +631,13 @@ class Porter(object):
             await self._setup_state_group_id_seq()
             await self._setup_user_id_seq()
             await self._setup_events_stream_seqs()
-            await self._setup_device_inbox_seq()
+            await self._setup_sequence(
+                "device_inbox_sequence", ("device_inbox", "device_federation_outbox")
+            )
+            await self._setup_sequence(
+                "account_data_sequence", ("room_account_data", "room_tags_revisions", "account_data"))
+            await self._setup_sequence("receipts_sequence", ("receipts_linearized", ))
+            await self._setup_auth_chain_sequence()
 
             # Step 3. Get tables.
             self.progress.set_state("Fetching tables")
@@ -854,7 +862,7 @@ class Porter(object):
 
         return done, remaining + done
 
-    async def _setup_state_group_id_seq(self):
+    async def _setup_state_group_id_seq(self) -> None:
         curr_id = await self.sqlite_store.db_pool.simple_select_one_onecol(
             table="state_groups", keyvalues={}, retcol="MAX(id)", allow_none=True
         )
@@ -868,7 +876,7 @@ class Porter(object):
 
         await self.postgres_store.db_pool.runInteraction("setup_state_group_id_seq", r)
 
-    async def _setup_user_id_seq(self):
+    async def _setup_user_id_seq(self) -> None:
         curr_id = await self.sqlite_store.db_pool.runInteraction(
             "setup_user_id_seq", find_max_generated_user_id_localpart
         )
@@ -877,9 +885,9 @@ class Porter(object):
             next_id = curr_id + 1
             txn.execute("ALTER SEQUENCE user_id_seq RESTART WITH %s", (next_id,))
 
-        return self.postgres_store.db_pool.runInteraction("setup_user_id_seq", r)
+        await self.postgres_store.db_pool.runInteraction("setup_user_id_seq", r)
 
-    async def _setup_events_stream_seqs(self):
+    async def _setup_events_stream_seqs(self) -> None:
         """Set the event stream sequences to the correct values.
         """
 
@@ -908,35 +916,46 @@ class Porter(object):
                 (curr_backward_id + 1,),
             )
 
-        return await self.postgres_store.db_pool.runInteraction(
+        await self.postgres_store.db_pool.runInteraction(
             "_setup_events_stream_seqs", _setup_events_stream_seqs_set_pos,
         )
 
-    async def _setup_device_inbox_seq(self):
-        """Set the device inbox sequence to the correct value.
+    async def _setup_sequence(self, sequence_name: str, stream_id_tables: Iterable[str]) -> None:
+        """Set a sequence to the correct value.
         """
-        curr_local_id = await self.sqlite_store.db_pool.simple_select_one_onecol(
-            table="device_inbox",
-            keyvalues={},
-            retcol="COALESCE(MAX(stream_id), 1)",
-            allow_none=True,
-        )
+        current_stream_ids = []
+        for stream_id_table in stream_id_tables:
+            max_stream_id = await self.sqlite_store.db_pool.simple_select_one_onecol(
+                table=stream_id_table,
+                keyvalues={},
+                retcol="COALESCE(MAX(stream_id), 1)",
+                allow_none=True,
+            )
+            current_stream_ids.append(max_stream_id)
 
-        curr_federation_id = await self.sqlite_store.db_pool.simple_select_one_onecol(
-            table="device_federation_outbox",
-            keyvalues={},
-            retcol="COALESCE(MAX(stream_id), 1)",
-            allow_none=True,
-        )
+        next_id = max(current_stream_ids) + 1
 
-        next_id = max(curr_local_id, curr_federation_id) + 1
+        def r(txn):
+            sql = "ALTER SEQUENCE %s RESTART WITH" % (sequence_name, )
+            txn.execute(sql + " %s", (next_id, ))
+
+        await self.postgres_store.db_pool.runInteraction("_setup_%s" % (sequence_name,), r)
+
+    async def _setup_auth_chain_sequence(self) -> None:
+        curr_chain_id = await self.sqlite_store.db_pool.simple_select_one_onecol(
+            table="event_auth_chains", keyvalues={}, retcol="MAX(chain_id)", allow_none=True
+        )
 
         def r(txn):
             txn.execute(
-                "ALTER SEQUENCE device_inbox_sequence RESTART WITH %s", (next_id,)
+                "ALTER SEQUENCE event_auth_chain_id RESTART WITH %s",
+                (curr_chain_id,),
             )
 
-        return self.postgres_store.db_pool.runInteraction("_setup_device_inbox_seq", r)
+        await self.postgres_store.db_pool.runInteraction(
+            "_setup_event_auth_chain_id", r,
+        )
+
 
 
 ##############################################

setup.cfg

@@ -3,6 +3,7 @@ test_suite = tests
 
 [check-manifest]
 ignore =
+    .git-blame-ignore-revs
     contrib
     contrib/*
     docs/*

setup.py

@@ -102,7 +102,7 @@ CONDITIONAL_REQUIREMENTS["lint"] = [
     "flake8",
 ]
 
-CONDITIONAL_REQUIREMENTS["mypy"] = ["mypy==0.790", "mypy-zope==0.2.8"]
+CONDITIONAL_REQUIREMENTS["mypy"] = ["mypy==0.812", "mypy-zope==0.2.11"]
 
 # Dependencies which are exclusively required by unit test code. This is
 # NOT a list of all modules that are necessary to run the unit tests.

stubs/txredisapi.pyi

@@ -17,7 +17,9 @@
 """
 from typing import Any, List, Optional, Type, Union
 
-class RedisProtocol:
+from twisted.internet import protocol
+
+class RedisProtocol(protocol.Protocol):
     def publish(self, channel: str, message: bytes): ...
     async def ping(self) -> None: ...
     async def set(
@@ -52,7 +54,7 @@ def lazyConnection(
 
 class ConnectionHandler: ...
 
-class RedisFactory:
+class RedisFactory(protocol.ReconnectingClientFactory):
     continueTrying: bool
     handler: RedisProtocol
     pool: List[RedisProtocol]

synapse/__init__.py

@@ -48,7 +48,7 @@ try:
 except ImportError:
     pass
 
-__version__ = "1.27.0"
+__version__ = "1.30.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

synapse/api/auth.py

@@ -39,6 +39,7 @@ from synapse.logging import opentracing as opentracing
 from synapse.storage.databases.main.registration import TokenLookupResult
 from synapse.types import StateMap, UserID
 from synapse.util.caches.lrucache import LruCache
+from synapse.util.macaroons import get_value_from_macaroon, satisfy_expiry
 from synapse.util.metrics import Measure
 
 logger = logging.getLogger(__name__)
@@ -163,7 +164,7 @@ class Auth:
 
     async def get_user_by_req(
         self,
-        request: Request,
+        request: SynapseRequest,
         allow_guest: bool = False,
         rights: str = "access",
         allow_expired: bool = False,
@@ -408,7 +409,7 @@ class Auth:
             raise _InvalidMacaroonException()
 
         try:
-            user_id = self.get_user_id_from_macaroon(macaroon)
+            user_id = get_value_from_macaroon(macaroon, "user_id")
 
             guest = False
             for caveat in macaroon.caveats:
@@ -416,7 +417,12 @@ class Auth:
                     guest = True
 
             self.validate_macaroon(macaroon, rights, user_id=user_id)
-        except (pymacaroons.exceptions.MacaroonException, TypeError, ValueError):
+        except (
+            pymacaroons.exceptions.MacaroonException,
+            KeyError,
+            TypeError,
+            ValueError,
+        ):
             raise InvalidClientTokenError("Invalid macaroon passed.")
 
         if rights == "access":
@@ -424,27 +430,6 @@ class Auth:
 
         return user_id, guest
 
-    def get_user_id_from_macaroon(self, macaroon):
-        """Retrieve the user_id given by the caveats on the macaroon.
-
-        Does *not* validate the macaroon.
-
-        Args:
-            macaroon (pymacaroons.Macaroon): The macaroon to validate
-
-        Returns:
-            (str) user id
-
-        Raises:
-            InvalidClientCredentialsError if there is no user_id caveat in the
-                macaroon
-        """
-        user_prefix = "user_id = "
-        for caveat in macaroon.caveats:
-            if caveat.caveat_id.startswith(user_prefix):
-                return caveat.caveat_id[len(user_prefix) :]
-        raise InvalidClientTokenError("No user caveat in macaroon")
-
     def validate_macaroon(self, macaroon, type_string, user_id):
         """
         validate that a Macaroon is understood by and was signed by this server.
@@ -465,21 +450,13 @@ class Auth:
         v.satisfy_exact("type = " + type_string)
         v.satisfy_exact("user_id = %s" % user_id)
         v.satisfy_exact("guest = true")
-        v.satisfy_general(self._verify_expiry)
+        satisfy_expiry(v, self.clock.time_msec)
 
         # access_tokens include a nonce for uniqueness: any value is acceptable
         v.satisfy_general(lambda c: c.startswith("nonce = "))
 
         v.verify(macaroon, self._macaroon_secret_key)
 
-    def _verify_expiry(self, caveat):
-        prefix = "time < "
-        if not caveat.startswith(prefix):
-            return False
-        expiry = int(caveat[len(prefix) :])
-        now = self.hs.get_clock().time_msec()
-        return now < expiry
-
     def get_appservice_by_req(self, request: SynapseRequest) -> ApplicationService:
         token = self.get_access_token_from_request(request)
         service = self.store.get_app_service_by_token(token)

synapse/api/constants.py

@@ -27,6 +27,11 @@ MAX_ALIAS_LENGTH = 255
 # the maximum length for a user id is 255 characters
 MAX_USERID_LENGTH = 255
 
+# The maximum length for a group id is 255 characters
+MAX_GROUPID_LENGTH = 255
+MAX_GROUP_CATEGORYID_LENGTH = 255
+MAX_GROUP_ROLEID_LENGTH = 255
+
 
 class Membership:
 
@@ -93,11 +98,14 @@ class EventTypes:
 
     Retention = "m.room.retention"
 
-    Presence = "m.presence"
-
     Dummy = "org.matrix.dummy_event"
 
 
+class EduTypes:
+    Presence = "m.presence"
+    RoomKeyRequest = "m.room_key_request"
+
+
 class RejectedReason:
     AUTH_ERROR = "auth_error"
 

synapse/api/ratelimiting.py

@@ -14,7 +14,7 @@
 # limitations under the License.
 
 from collections import OrderedDict
-from typing import Any, Optional, Tuple
+from typing import Hashable, Optional, Tuple
 
 from synapse.api.errors import LimitExceededError
 from synapse.types import Requester
@@ -42,7 +42,9 @@ class Ratelimiter:
         #   * How many times an action has occurred since a point in time
         #   * The point in time
         #   * The rate_hz of this particular entry. This can vary per request
-        self.actions = OrderedDict()  # type: OrderedDict[Any, Tuple[float, int, float]]
+        self.actions = (
+            OrderedDict()
+        )  # type: OrderedDict[Hashable, Tuple[float, int, float]]
 
     def can_requester_do_action(
         self,
@@ -82,7 +84,7 @@ class Ratelimiter:
 
     def can_do_action(
         self,
-        key: Any,
+        key: Hashable,
         rate_hz: Optional[float] = None,
         burst_count: Optional[int] = None,
         update: bool = True,
@@ -175,7 +177,7 @@ class Ratelimiter:
 
     def ratelimit(
         self,
-        key: Any,
+        key: Hashable,
         rate_hz: Optional[float] = None,
         burst_count: Optional[int] = None,
         update: bool = True,

synapse/app/__init__.py

@@ -17,8 +17,6 @@ import sys
 
 from synapse import python_dependencies  # noqa: E402
 
-sys.dont_write_bytecode = True
-
 logger = logging.getLogger(__name__)
 
 try:

synapse/app/admin_cmd.py

@@ -210,7 +210,9 @@ def start(config_options):
     config.update_user_directory = False
     config.run_background_tasks = False
     config.start_pushers = False
+    config.pusher_shard_config.instances = []
     config.send_federation = False
+    config.federation_shard_config.instances = []
 
     synapse.events.USE_FROZEN_DICTS = config.use_frozen_dicts
 

synapse/app/generic_worker.py

@@ -23,6 +23,7 @@ from typing_extensions import ContextManager
 
 from twisted.internet import address
 from twisted.web.resource import IResource
+from twisted.web.server import Request
 
 import synapse
 import synapse.events
@@ -190,7 +191,7 @@ class KeyUploadServlet(RestServlet):
         self.http_client = hs.get_simple_http_client()
         self.main_uri = hs.config.worker_main_http_uri
 
-    async def on_POST(self, request, device_id):
+    async def on_POST(self, request: Request, device_id: Optional[str]):
         requester = await self.auth.get_user_by_req(request, allow_guest=True)
         user_id = requester.user.to_string()
         body = parse_json_object_from_request(request)
@@ -223,10 +224,12 @@ class KeyUploadServlet(RestServlet):
                 header: request.requestHeaders.getRawHeaders(header, [])
                 for header in (b"Authorization", b"User-Agent")
             }
-            # Add the previous hop the the X-Forwarded-For header.
+            # Add the previous hop to the X-Forwarded-For header.
             x_forwarded_for = request.requestHeaders.getRawHeaders(
                 b"X-Forwarded-For", []
             )
+            # we use request.client here, since we want the previous hop, not the
+            # original client (as returned by request.getClientAddress()).
             if isinstance(request.client, (address.IPv4Address, address.IPv6Address)):
                 previous_host = request.client.host.encode("ascii")
                 # If the header exists, add to the comma-separated list of the first
@@ -239,6 +242,14 @@ class KeyUploadServlet(RestServlet):
                     x_forwarded_for = [previous_host]
             headers[b"X-Forwarded-For"] = x_forwarded_for
 
+            # Replicate the original X-Forwarded-Proto header. Note that
+            # XForwardedForRequest overrides isSecure() to give us the original protocol
+            # used by the client, as opposed to the protocol used by our upstream proxy
+            # - which is what we want here.
+            headers[b"X-Forwarded-Proto"] = [
+                b"https" if request.isSecure() else b"http"
+            ]
+
             try:
                 result = await self.http_client.post_json_get_json(
                     self.main_uri + request.uri.decode("ascii"), body, headers=headers
@@ -645,9 +656,6 @@ class GenericWorkerServer(HomeServer):
 
         self.get_tcp_replication().start_replication(self)
 
-    async def remove_pusher(self, app_id, push_key, user_id):
-        self.get_tcp_replication().send_remove_pusher(app_id, push_key, user_id)
-
     @cache_in_self
     def get_replication_data_handler(self):
         return GenericWorkerReplicationHandler(self)
@@ -922,22 +930,6 @@ def start(config_options):
         # For other worker types we force this to off.
         config.appservice.notify_appservices = False
 
-    if config.worker_app == "synapse.app.pusher":
-        if config.server.start_pushers:
-            sys.stderr.write(
-                "\nThe pushers must be disabled in the main synapse process"
-                "\nbefore they can be run in a separate worker."
-                "\nPlease add ``start_pushers: false`` to the main config"
-                "\n"
-            )
-            sys.exit(1)
-
-        # Force the pushers to start since they will be disabled in the main config
-        config.server.start_pushers = True
-    else:
-        # For other worker types we force this to off.
-        config.server.start_pushers = False
-
     if config.worker_app == "synapse.app.user_dir":
         if config.server.update_user_directory:
             sys.stderr.write(
@@ -954,22 +946,6 @@ def start(config_options):
         # For other worker types we force this to off.
         config.server.update_user_directory = False
 
-    if config.worker_app == "synapse.app.federation_sender":
-        if config.worker.send_federation:
-            sys.stderr.write(
-                "\nThe send_federation must be disabled in the main synapse process"
-                "\nbefore they can be run in a separate worker."
-                "\nPlease add ``send_federation: false`` to the main config"
-                "\n"
-            )
-            sys.exit(1)
-
-        # Force the pushers to start since they will be disabled in the main config
-        config.worker.send_federation = True
-    else:
-        # For other worker types we force this to off.
-        config.worker.send_federation = False
-
     synapse.events.USE_FROZEN_DICTS = config.use_frozen_dicts
 
     hs = GenericWorkerServer(

synapse/appservice/api.py

@@ -90,7 +90,7 @@ class ApplicationServiceApi(SimpleHttpClient):
         self.clock = hs.get_clock()
 
         self.protocol_meta_cache = ResponseCache(
-            hs, "as_protocol_meta", timeout_ms=HOUR_IN_MS
+            hs.get_clock(), "as_protocol_meta", timeout_ms=HOUR_IN_MS
         )  # type: ResponseCache[Tuple[str, str]]
 
     async def query_user(self, service, user_id):

synapse/config/_base.py

@@ -21,7 +21,7 @@ import os
 from collections import OrderedDict
 from hashlib import sha256
 from textwrap import dedent
-from typing import Any, Iterable, List, MutableMapping, Optional
+from typing import Any, Iterable, List, MutableMapping, Optional, Union
 
 import attr
 import jinja2
@@ -147,7 +147,20 @@ class Config:
         return int(value) * size
 
     @staticmethod
-    def parse_duration(value):
+    def parse_duration(value: Union[str, int]) -> int:
+        """Convert a duration as a string or integer to a number of milliseconds.
+
+        If an integer is provided it is treated as milliseconds and is unchanged.
+
+        String durations can have a suffix of 's', 'm', 'h', 'd', 'w', or 'y'.
+        No suffix is treated as milliseconds.
+
+        Args:
+            value: The duration to parse.
+
+        Returns:
+            The number of milliseconds in the duration.
+        """
         if isinstance(value, int):
             return value
         second = 1000
@@ -199,9 +212,8 @@ class Config:
 
     @classmethod
     def read_file(cls, file_path, config_name):
-        cls.check_file(file_path, config_name)
-        with open(file_path) as file_stream:
-            return file_stream.read()
+        """Deprecated: call read_file directly"""
+        return read_file(file_path, (config_name,))
 
     def read_template(self, filename: str) -> jinja2.Template:
         """Load a template file from disk.
@@ -831,22 +843,23 @@ class ShardedWorkerHandlingConfig:
 
     def should_handle(self, instance_name: str, key: str) -> bool:
         """Whether this instance is responsible for handling the given key."""
-        # If multiple instances are not defined we always return true
-        if not self.instances or len(self.instances) == 1:
-            return True
+        # If no instances are defined we assume some other worker is handling
+        # this.
+        if not self.instances:
+            return False
 
-        return self.get_instance(key) == instance_name
+        return self._get_instance(key) == instance_name
 
-    def get_instance(self, key: str) -> str:
+    def _get_instance(self, key: str) -> str:
         """Get the instance responsible for handling the given key.
 
-        Note: For things like federation sending the config for which instance
-        is sending is known only to the sender instance if there is only one.
-        Therefore `should_handle` should be used where possible.
+        Note: For federation sending and pushers the config for which instance
+        is sending is known only to the sender instance, so we don't expose this
+        method by default.
         """
 
         if not self.instances:
-            return "master"
+            raise Exception("Unknown worker")
 
         if len(self.instances) == 1:
             return self.instances[0]
@@ -863,4 +876,52 @@ class ShardedWorkerHandlingConfig:
         return self.instances[remainder]
 
 
-__all__ = ["Config", "RootConfig", "ShardedWorkerHandlingConfig"]
+@attr.s
+class RoutableShardedWorkerHandlingConfig(ShardedWorkerHandlingConfig):
+    """A version of `ShardedWorkerHandlingConfig` that is used for config
+    options where all instances know which instances are responsible for the
+    sharded work.
+    """
+
+    def __attrs_post_init__(self):
+        # We require that `self.instances` is non-empty.
+        if not self.instances:
+            raise Exception("Got empty list of instances for shard config")
+
+    def get_instance(self, key: str) -> str:
+        """Get the instance responsible for handling the given key."""
+        return self._get_instance(key)
+
+
+def read_file(file_path: Any, config_path: Iterable[str]) -> str:
+    """Check the given file exists, and read it into a string
+
+    If it does not, emit an error indicating the problem
+
+    Args:
+        file_path: the file to be read
+        config_path: where in the configuration file_path came from, so that a useful
+           error can be emitted if it does not exist.
+    Returns:
+        content of the file.
+    Raises:
+        ConfigError if there is a problem reading the file.
+    """
+    if not isinstance(file_path, str):
+        raise ConfigError("%r is not a string", config_path)
+
+    try:
+        os.stat(file_path)
+        with open(file_path) as file_stream:
+            return file_stream.read()
+    except OSError as e:
+        raise ConfigError("Error accessing file %r" % (file_path,), config_path) from e
+
+
+__all__ = [
+    "Config",
+    "RootConfig",
+    "ShardedWorkerHandlingConfig",
+    "RoutableShardedWorkerHandlingConfig",
+    "read_file",
+]

synapse/config/_base.pyi

@@ -149,4 +149,8 @@ class ShardedWorkerHandlingConfig:
     instances: List[str]
     def __init__(self, instances: List[str]) -> None: ...
     def should_handle(self, instance_name: str, key: str) -> bool: ...
+
+class RoutableShardedWorkerHandlingConfig(ShardedWorkerHandlingConfig):
     def get_instance(self, key: str) -> str: ...
+
+def read_file(file_path: Any, config_path: Iterable[str]) -> str: ...

synapse/config/auth.py

@@ -37,7 +37,9 @@ class AuthConfig(Config):
 
         # User-interactive authentication
         ui_auth = config.get("ui_auth") or {}
-        self.ui_auth_session_timeout = ui_auth.get("session_timeout", 0)
+        self.ui_auth_session_timeout = self.parse_duration(
+            ui_auth.get("session_timeout", 0)
+        )
 
     def generate_config_section(self, config_dir_path, server_name, **kwargs):
         return """\
@@ -93,8 +95,8 @@ class AuthConfig(Config):
               #require_uppercase: true
 
         ui_auth:
-            # The number of milliseconds to allow a user-interactive authentication
-            # session to be active.
+            # The amount of time to allow a user-interactive authentication session
+            # to be active.
             #
             # This defaults to 0, meaning the user is queried for their credentials
             # before every action, but this can be overridden to allow a single
@@ -105,5 +107,5 @@ class AuthConfig(Config):
             # Uncomment below to allow for credential validation to last for 15
             # seconds.
             #
-            #session_timeout: 15000
+            #session_timeout: "15s"
         """

synapse/config/federation.py

@@ -41,6 +41,10 @@ class FederationConfig(Config):
         )
         self.federation_metrics_domains = set(federation_metrics_domains)
 
+        self.allow_profile_lookup_over_federation = config.get(
+            "allow_profile_lookup_over_federation", True
+        )
+
     def generate_config_section(self, config_dir_path, server_name, **kwargs):
         return """\
         ## Federation ##
@@ -66,6 +70,12 @@ class FederationConfig(Config):
         #federation_metrics_domains:
         #  - matrix.org
         #  - example.com
+
+        # Uncomment to disable profile lookup over federation. By default, the
+        # Federation API allows other homeservers to obtain profile data of any user
+        # on this homeserver. Defaults to 'true'.
+        #
+        #allow_profile_lookup_over_federation: false
         """
 
 

synapse/config/logger.py

@@ -21,8 +21,10 @@ import threading
 from string import Template
 
 import yaml
+from zope.interface import implementer
 
 from twisted.logger import (
+    ILogObserver,
     LogBeginner,
     STDLibLogObserver,
     eventAsText,
@@ -227,7 +229,8 @@ def _setup_stdlib_logging(config, log_config_path, logBeginner: LogBeginner) ->
 
     threadlocal = threading.local()
 
-    def _log(event):
+    @implementer(ILogObserver)
+    def _log(event: dict) -> None:
         if "log_text" in event:
             if event["log_text"].startswith("DNSDatagramProtocol starting on "):
                 return

synapse/config/oidc_config.py

@@ -15,7 +15,7 @@
 # limitations under the License.
 
 from collections import Counter
-from typing import Iterable, Optional, Tuple, Type
+from typing import Iterable, Mapping, Optional, Tuple, Type
 
 import attr
 
@@ -25,7 +25,7 @@ from synapse.types import Collection, JsonDict
 from synapse.util.module_loader import load_module
 from synapse.util.stringutils import parse_and_validate_mxc_uri
 
-from ._base import Config, ConfigError
+from ._base import Config, ConfigError, read_file
 
 DEFAULT_USER_MAPPING_PROVIDER = "synapse.handlers.oidc_handler.JinjaOidcMappingProvider"
 
@@ -97,7 +97,26 @@ class OIDCConfig(Config):
         #
         #   client_id: Required. oauth2 client id to use.
         #
-        #   client_secret: Required. oauth2 client secret to use.
+        #   client_secret: oauth2 client secret to use. May be omitted if
+        #        client_secret_jwt_key is given, or if client_auth_method is 'none'.
+        #
+        #   client_secret_jwt_key: Alternative to client_secret: details of a key used
+        #      to create a JSON Web Token to be used as an OAuth2 client secret. If
+        #      given, must be a dictionary with the following properties:
+        #
+        #          key: a pem-encoded signing key. Must be a suitable key for the
+        #              algorithm specified. Required unless 'key_file' is given.
+        #
+        #          key_file: the path to file containing a pem-encoded signing key file.
+        #              Required unless 'key' is given.
+        #
+        #          jwt_header: a dictionary giving properties to include in the JWT
+        #              header. Must include the key 'alg', giving the algorithm used to
+        #              sign the JWT, such as "ES256", using the JWA identifiers in
+        #              RFC7518.
+        #
+        #          jwt_payload: an optional dictionary giving properties to include in
+        #              the JWT payload. Normally this should include an 'iss' key.
         #
         #   client_auth_method: auth method to use when exchanging the token. Valid
         #       values are 'client_secret_basic' (default), 'client_secret_post' and
@@ -218,7 +237,7 @@ class OIDCConfig(Config):
           #
           #- idp_id: github
           #  idp_name: Github
-          #  idp_brand: org.matrix.github
+          #  idp_brand: github
           #  discover: false
           #  issuer: "https://github.com/"
           #  client_id: "your-client-id" # TO BE FILLED
@@ -240,7 +259,7 @@ class OIDCConfig(Config):
 # jsonschema definition of the configuration settings for an oidc identity provider
 OIDC_PROVIDER_CONFIG_SCHEMA = {
     "type": "object",
-    "required": ["issuer", "client_id", "client_secret"],
+    "required": ["issuer", "client_id"],
     "properties": {
         "idp_id": {
             "type": "string",
@@ -253,7 +272,12 @@ OIDC_PROVIDER_CONFIG_SCHEMA = {
         "idp_icon": {"type": "string"},
         "idp_brand": {
             "type": "string",
-            # MSC2758-style namespaced identifier
+            "minLength": 1,
+            "maxLength": 255,
+            "pattern": "^[a-z][a-z0-9_.-]*$",
+        },
+        "idp_unstable_brand": {
+            "type": "string",
             "minLength": 1,
             "maxLength": 255,
             "pattern": "^[a-z][a-z0-9_.-]*$",
@@ -262,6 +286,30 @@ OIDC_PROVIDER_CONFIG_SCHEMA = {
         "issuer": {"type": "string"},
         "client_id": {"type": "string"},
         "client_secret": {"type": "string"},
+        "client_secret_jwt_key": {
+            "type": "object",
+            "required": ["jwt_header"],
+            "oneOf": [
+                {"required": ["key"]},
+                {"required": ["key_file"]},
+            ],
+            "properties": {
+                "key": {"type": "string"},
+                "key_file": {"type": "string"},
+                "jwt_header": {
+                    "type": "object",
+                    "required": ["alg"],
+                    "properties": {
+                        "alg": {"type": "string"},
+                    },
+                    "additionalProperties": {"type": "string"},
+                },
+                "jwt_payload": {
+                    "type": "object",
+                    "additionalProperties": {"type": "string"},
+                },
+            },
+        },
         "client_auth_method": {
             "type": "string",
             # the following list is the same as the keys of
@@ -404,15 +452,31 @@ def _parse_oidc_config_dict(
                 "idp_icon must be a valid MXC URI", config_path + ("idp_icon",)
             ) from e
 
+    client_secret_jwt_key_config = oidc_config.get("client_secret_jwt_key")
+    client_secret_jwt_key = None  # type: Optional[OidcProviderClientSecretJwtKey]
+    if client_secret_jwt_key_config is not None:
+        keyfile = client_secret_jwt_key_config.get("key_file")
+        if keyfile:
+            key = read_file(keyfile, config_path + ("client_secret_jwt_key",))
+        else:
+            key = client_secret_jwt_key_config["key"]
+        client_secret_jwt_key = OidcProviderClientSecretJwtKey(
+            key=key,
+            jwt_header=client_secret_jwt_key_config["jwt_header"],
+            jwt_payload=client_secret_jwt_key_config.get("jwt_payload", {}),
+        )
+
     return OidcProviderConfig(
         idp_id=idp_id,
         idp_name=oidc_config.get("idp_name", "OIDC"),
         idp_icon=idp_icon,
         idp_brand=oidc_config.get("idp_brand"),
+        unstable_idp_brand=oidc_config.get("unstable_idp_brand"),
         discover=oidc_config.get("discover", True),
         issuer=oidc_config["issuer"],
         client_id=oidc_config["client_id"],
-        client_secret=oidc_config["client_secret"],
+        client_secret=oidc_config.get("client_secret"),
+        client_secret_jwt_key=client_secret_jwt_key,
         client_auth_method=oidc_config.get("client_auth_method", "client_secret_basic"),
         scopes=oidc_config.get("scopes", ["openid"]),
         authorization_endpoint=oidc_config.get("authorization_endpoint"),
@@ -427,6 +491,18 @@ def _parse_oidc_config_dict(
     )
 
 
+@attr.s(slots=True, frozen=True)
+class OidcProviderClientSecretJwtKey:
+    # a pem-encoded signing key
+    key = attr.ib(type=str)
+
+    # properties to include in the JWT header
+    jwt_header = attr.ib(type=Mapping[str, str])
+
+    # properties to include in the JWT payload.
+    jwt_payload = attr.ib(type=Mapping[str, str])
+
+
 @attr.s(slots=True, frozen=True)
 class OidcProviderConfig:
     # a unique identifier for this identity provider. Used in the 'user_external_ids'
@@ -442,6 +518,9 @@ class OidcProviderConfig:
     # Optional brand identifier for this IdP.
     idp_brand = attr.ib(type=Optional[str])
 
+    # Optional brand identifier for the unstable API (see MSC2858).
+    unstable_idp_brand = attr.ib(type=Optional[str])
+
     # whether the OIDC discovery mechanism is used to discover endpoints
     discover = attr.ib(type=bool)
 
@@ -452,8 +531,13 @@ class OidcProviderConfig:
     # oauth2 client id to use
     client_id = attr.ib(type=str)
 
-    # oauth2 client secret to use
-    client_secret = attr.ib(type=str)
+    # oauth2 client secret to use. if `None`, use client_secret_jwt_key to generate
+    # a secret.
+    client_secret = attr.ib(type=Optional[str])
+
+    # key to use to construct a JWT to use as a client secret. May be `None` if
+    # `client_secret` is set.
+    client_secret_jwt_key = attr.ib(type=Optional[OidcProviderClientSecretJwtKey])
 
     # auth method to use when exchanging the token.
     # Valid values are 'client_secret_basic', 'client_secret_post' and

synapse/config/push.py

@@ -14,7 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from ._base import Config, ShardedWorkerHandlingConfig
+from ._base import Config
 
 
 class PushConfig(Config):
@@ -27,9 +27,6 @@ class PushConfig(Config):
             "group_unread_count_by_room", True
         )
 
-        pusher_instances = config.get("pusher_instances") or []
-        self.pusher_shard_config = ShardedWorkerHandlingConfig(pusher_instances)
-
         # There was a a 'redact_content' setting but mistakenly read from the
         # 'email'section'. Check for the flag in the 'push' section, and log,
         # but do not honour it to avoid nasty surprises when people upgrade.

synapse/config/ratelimiting.py

@@ -102,6 +102,16 @@ class RatelimitConfig(Config):
             defaults={"per_second": 0.01, "burst_count": 3},
         )
 
+        # Ratelimit cross-user key requests:
+        # * For local requests this is keyed by the sending device.
+        # * For requests received over federation this is keyed by the origin.
+        #
+        # Note that this isn't exposed in the configuration as it is obscure.
+        self.rc_key_requests = RateLimitConfig(
+            config.get("rc_key_requests", {}),
+            defaults={"per_second": 20, "burst_count": 100},
+        )
+
         self.rc_3pid_validation = RateLimitConfig(
             config.get("rc_3pid_validation") or {},
             defaults={"per_second": 0.003, "burst_count": 5},

synapse/config/repository.py

@@ -206,7 +206,6 @@ class ContentRepositoryConfig(Config):
 
     def generate_config_section(self, data_dir_path, **kwargs):
         media_store = os.path.join(data_dir_path, "media_store")
-        uploads_path = os.path.join(data_dir_path, "uploads")
 
         formatted_thumbnail_sizes = "".join(
             THUMBNAIL_SIZE_YAML % s for s in DEFAULT_THUMBNAIL_SIZES

synapse/config/server.py

@@ -263,6 +263,12 @@ class ServerConfig(Config):
             False,
         )
 
+        # Whether to retrieve and display profile data for a user when they
+        # are invited to a room
+        self.include_profile_data_on_invite = config.get(
+            "include_profile_data_on_invite", True
+        )
+
         if "restrict_public_rooms_to_local_users" in config and (
             "allow_public_rooms_without_auth" in config
             or "allow_public_rooms_over_federation" in config
@@ -391,7 +397,6 @@ class ServerConfig(Config):
         if self.public_baseurl is not None:
             if self.public_baseurl[-1] != "/":
                 self.public_baseurl += "/"
-        self.start_pushers = config.get("start_pushers", True)
 
         # (undocumented) option for torturing the worker-mode replication a bit,
         # for testing. The value defines the number of milliseconds to pause before
@@ -836,8 +841,7 @@ class ServerConfig(Config):
         # Whether to require authentication to retrieve profile data (avatars,
         # display names) of other users through the client API. Defaults to
         # 'false'. Note that profile data is also available via the federation
-        # API, so this setting is of limited value if federation is enabled on
-        # the server.
+        # API, unless allow_profile_lookup_over_federation is set to false.
         #
         #require_auth_for_profile_requests: true
 
@@ -848,6 +852,14 @@ class ServerConfig(Config):
         #
         #limit_profile_requests_to_users_who_share_rooms: true
 
+        # Uncomment to prevent a user's profile data from being retrieved and
+        # displayed in a room until they have joined it. By default, a user's
+        # profile data is included in an invite event, regardless of the values
+        # of the above two settings, and whether or not the users share a server.
+        # Defaults to 'true'.
+        #
+        #include_profile_data_on_invite: false
+
         # If set to 'true', removes the need for authentication to access the server's
         # public rooms directory through the client API, meaning that anyone can
         # query the room directory. Defaults to 'false'.

synapse/config/stats.py

@@ -13,10 +13,22 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import sys
+import logging
 
 from ._base import Config
 
+ROOM_STATS_DISABLED_WARN = """\
+WARNING: room/user statistics have been disabled via the stats.enabled
+configuration setting. This means that certain features (such as the room
+directory) will not operate correctly. Future versions of Synapse may ignore
+this setting.
+
+To fix this warning, remove the stats.enabled setting from your configuration
+file.
+--------------------------------------------------------------------------------"""
+
+logger = logging.getLogger(__name__)
+
 
 class StatsConfig(Config):
     """Stats Configuration
@@ -28,30 +40,29 @@ class StatsConfig(Config):
     def read_config(self, config, **kwargs):
         self.stats_enabled = True
         self.stats_bucket_size = 86400 * 1000
-        self.stats_retention = sys.maxsize
         stats_config = config.get("stats", None)
         if stats_config:
             self.stats_enabled = stats_config.get("enabled", self.stats_enabled)
             self.stats_bucket_size = self.parse_duration(
                 stats_config.get("bucket_size", "1d")
             )
-            self.stats_retention = self.parse_duration(
-                stats_config.get("retention", "%ds" % (sys.maxsize,))
-            )
+        if not self.stats_enabled:
+            logger.warning(ROOM_STATS_DISABLED_WARN)
 
     def generate_config_section(self, config_dir_path, server_name, **kwargs):
         return """
-        # Local statistics collection. Used in populating the room directory.
+        # Settings for local room and user statistics collection. See
+        # docs/room_and_user_statistics.md.
         #
-        # 'bucket_size' controls how large each statistics timeslice is. It can
-        # be defined in a human readable short form -- e.g. "1d", "1y".
-        #
-        # 'retention' controls how long historical statistics will be kept for.
-        # It can be defined in a human readable short form -- e.g. "1d", "1y".
-        #
-        #
-        #stats:
-        #   enabled: true
-        #   bucket_size: 1d
-        #   retention: 1y
+        stats:
+          # Uncomment the following to disable room and user statistics. Note that doing
+          # so may cause certain features (such as the room directory) not to work
+          # correctly.
+          #
+          #enabled: false
+
+          # The size of each timeslice in the room_stats_historical and
+          # user_stats_historical tables, as a time period. Defaults to "1d".
+          #
+          #bucket_size: 1h
         """

synapse/config/user_directory.py

@@ -24,32 +24,46 @@ class UserDirectoryConfig(Config):
     section = "userdirectory"
 
     def read_config(self, config, **kwargs):
-        self.user_directory_search_enabled = True
-        self.user_directory_search_all_users = False
-        user_directory_config = config.get("user_directory", None)
-        if user_directory_config:
-            self.user_directory_search_enabled = user_directory_config.get(
-                "enabled", True
-            )
-            self.user_directory_search_all_users = user_directory_config.get(
-                "search_all_users", False
-            )
+        user_directory_config = config.get("user_directory") or {}
+        self.user_directory_search_enabled = user_directory_config.get("enabled", True)
+        self.user_directory_search_all_users = user_directory_config.get(
+            "search_all_users", False
+        )
+        self.user_directory_search_prefer_local_users = user_directory_config.get(
+            "prefer_local_users", False
+        )
 
     def generate_config_section(self, config_dir_path, server_name, **kwargs):
         return """
         # User Directory configuration
         #
-        # 'enabled' defines whether users can search the user directory. If
-        # false then empty responses are returned to all queries. Defaults to
-        # true.
-        #
-        # 'search_all_users' defines whether to search all users visible to your HS
-        # when searching the user directory, rather than limiting to users visible
-        # in public rooms.  Defaults to false.  If you set it True, you'll have to
-        # rebuild the user_directory search indexes, see
-        # https://github.com/matrix-org/synapse/blob/master/docs/user_directory.md
-        #
-        #user_directory:
-        #  enabled: true
-        #  search_all_users: false
+        user_directory:
+            # Defines whether users can search the user directory. If false then
+            # empty responses are returned to all queries. Defaults to true.
+            #
+            # Uncomment to disable the user directory.
+            #
+            #enabled: false
+
+            # Defines whether to search all users visible to your HS when searching
+            # the user directory, rather than limiting to users visible in public
+            # rooms. Defaults to false.
+            #
+            # If you set it true, you'll have to rebuild the user_directory search
+            # indexes, see:
+            # https://github.com/matrix-org/synapse/blob/master/docs/user_directory.md
+            #
+            # Uncomment to return search results containing all known users, even if that
+            # user does not share a room with the requester.
+            #
+            #search_all_users: true
+
+            # Defines whether to prefer local users in search query results.
+            # If True, local users are more likely to appear above remote users
+            # when searching the user directory. Defaults to false.
+            #
+            # Uncomment to prefer local over remote users in user directory search
+            # results.
+            #
+            #prefer_local_users: true
         """

synapse/config/workers.py

@@ -17,9 +17,28 @@ from typing import List, Union
 
 import attr
 
-from ._base import Config, ConfigError, ShardedWorkerHandlingConfig
+from ._base import (
+    Config,
+    ConfigError,
+    RoutableShardedWorkerHandlingConfig,
+    ShardedWorkerHandlingConfig,
+)
 from .server import ListenerConfig, parse_listener_def
 
+_FEDERATION_SENDER_WITH_SEND_FEDERATION_ENABLED_ERROR = """
+The send_federation config option must be disabled in the main
+synapse process before they can be run in a separate worker.
+
+Please add ``send_federation: false`` to the main config
+"""
+
+_PUSHER_WITH_START_PUSHERS_ENABLED_ERROR = """
+The start_pushers config option must be disabled in the main
+synapse process before they can be run in a separate worker.
+
+Please add ``start_pushers: false`` to the main config
+"""
+
 
 def _instance_to_list_converter(obj: Union[str, List[str]]) -> List[str]:
     """Helper for allowing parsing a string or list of strings to a config
@@ -103,6 +122,7 @@ class WorkerConfig(Config):
         self.worker_replication_secret = config.get("worker_replication_secret", None)
 
         self.worker_name = config.get("worker_name", self.worker_app)
+        self.instance_name = self.worker_name or "master"
 
         self.worker_main_http_uri = config.get("worker_main_http_uri", None)
 
@@ -118,12 +138,41 @@ class WorkerConfig(Config):
                 )
             )
 
-        # Whether to send federation traffic out in this process. This only
-        # applies to some federation traffic, and so shouldn't be used to
-        # "disable" federation
-        self.send_federation = config.get("send_federation", True)
+        # Handle federation sender configuration.
+        #
+        # There are two ways of configuring which instances handle federation
+        # sending:
+        #   1. The old way where "send_federation" is set to false and running a
+        #      `synapse.app.federation_sender` worker app.
+        #   2. Specifying the workers sending federation in
+        #      `federation_sender_instances`.
+        #
 
-        federation_sender_instances = config.get("federation_sender_instances") or []
+        send_federation = config.get("send_federation", True)
+
+        federation_sender_instances = config.get("federation_sender_instances")
+        if federation_sender_instances is None:
+            # Default to an empty list, which means "another, unknown, worker is
+            # responsible for it".
+            federation_sender_instances = []
+
+            # If no federation sender instances are set we check if
+            # `send_federation` is set, which means use master
+            if send_federation:
+                federation_sender_instances = ["master"]
+
+            if self.worker_app == "synapse.app.federation_sender":
+                if send_federation:
+                    # If we're running federation senders, and not using
+                    # `federation_sender_instances`, then we should have
+                    # explicitly set `send_federation` to false.
+                    raise ConfigError(
+                        _FEDERATION_SENDER_WITH_SEND_FEDERATION_ENABLED_ERROR
+                    )
+
+                federation_sender_instances = [self.worker_name]
+
+        self.send_federation = self.instance_name in federation_sender_instances
         self.federation_shard_config = ShardedWorkerHandlingConfig(
             federation_sender_instances
         )
@@ -164,7 +213,37 @@ class WorkerConfig(Config):
                 "Must only specify one instance to handle `receipts` messages."
             )
 
-        self.events_shard_config = ShardedWorkerHandlingConfig(self.writers.events)
+        if len(self.writers.events) == 0:
+            raise ConfigError("Must specify at least one instance to handle `events`.")
+
+        self.events_shard_config = RoutableShardedWorkerHandlingConfig(
+            self.writers.events
+        )
+
+        # Handle sharded push
+        start_pushers = config.get("start_pushers", True)
+        pusher_instances = config.get("pusher_instances")
+        if pusher_instances is None:
+            # Default to an empty list, which means "another, unknown, worker is
+            # responsible for it".
+            pusher_instances = []
+
+            # If no pushers instances are set we check if `start_pushers` is
+            # set, which means use master
+            if start_pushers:
+                pusher_instances = ["master"]
+
+            if self.worker_app == "synapse.app.pusher":
+                if start_pushers:
+                    # If we're running pushers, and not using
+                    # `pusher_instances`, then we should have explicitly set
+                    # `start_pushers` to false.
+                    raise ConfigError(_PUSHER_WITH_START_PUSHERS_ENABLED_ERROR)
+
+                pusher_instances = [self.instance_name]
+
+        self.start_pushers = self.instance_name in pusher_instances
+        self.pusher_shard_config = ShardedWorkerHandlingConfig(pusher_instances)
 
         # Whether this worker should run background tasks or not.
         #

synapse/events/spamcheck.py

@@ -15,6 +15,7 @@
 # limitations under the License.
 
 import inspect
+import logging
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
 
 from synapse.rest.media.v1._base import FileInfo
@@ -27,6 +28,8 @@ if TYPE_CHECKING:
     import synapse.events
     import synapse.server
 
+logger = logging.getLogger(__name__)
+
 
 class SpamChecker:
     def __init__(self, hs: "synapse.server.HomeServer"):
@@ -190,6 +193,7 @@ class SpamChecker:
         email_threepid: Optional[dict],
         username: Optional[str],
         request_info: Collection[Tuple[str, str]],
+        auth_provider_id: Optional[str] = None,
     ) -> RegistrationBehaviour:
         """Checks if we should allow the given registration request.
 
@@ -198,6 +202,9 @@ class SpamChecker:
             username: The request user name, if any
             request_info: List of tuples of user agent and IP that
                 were used during the registration process.
+            auth_provider_id: The SSO IdP the user used, e.g "oidc", "saml",
+                "cas". If any. Note this does not include users registered
+                via a password provider.
 
         Returns:
             Enum for how the request should be handled
@@ -208,9 +215,25 @@ class SpamChecker:
             # spam checker
             checker = getattr(spam_checker, "check_registration_for_spam", None)
             if checker:
-                behaviour = await maybe_awaitable(
-                    checker(email_threepid, username, request_info)
-                )
+                # Provide auth_provider_id if the function supports it
+                checker_args = inspect.signature(checker)
+                if len(checker_args.parameters) == 4:
+                    d = checker(
+                        email_threepid,
+                        username,
+                        request_info,
+                        auth_provider_id,
+                    )
+                elif len(checker_args.parameters) == 3:
+                    d = checker(email_threepid, username, request_info)
+                else:
+                    logger.error(
+                        "Invalid signature for %s.check_registration_for_spam. Denying registration",
+                        spam_checker.__module__,
+                    )
+                    return RegistrationBehaviour.DENY
+
+                behaviour = await maybe_awaitable(d)
                 assert isinstance(behaviour, RegistrationBehaviour)
                 if behaviour != RegistrationBehaviour.ALLOW:
                     return behaviour

synapse/federation/federation_server.py

@@ -22,6 +22,7 @@ from typing import (
     Awaitable,
     Callable,
     Dict,
+    Iterable,
     List,
     Optional,
     Tuple,
@@ -34,7 +35,7 @@ from twisted.internet import defer
 from twisted.internet.abstract import isIPAddress
 from twisted.python import failure
 
-from synapse.api.constants import EventTypes, Membership
+from synapse.api.constants import EduTypes, EventTypes, Membership
 from synapse.api.errors import (
     AuthError,
     Codes,
@@ -44,6 +45,7 @@ from synapse.api.errors import (
     SynapseError,
     UnsupportedRoomVersionError,
 )
+from synapse.api.ratelimiting import Ratelimiter
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS
 from synapse.events import EventBase
 from synapse.federation.federation_base import FederationBase, event_from_pdu_json
@@ -89,16 +91,15 @@ pdu_process_time = Histogram(
     "Time taken to process an event",
 )
 
-
-last_pdu_age_metric = Gauge(
-    "synapse_federation_last_received_pdu_age",
-    "The age (in seconds) of the last PDU successfully received from the given domain",
+last_pdu_ts_metric = Gauge(
+    "synapse_federation_last_received_pdu_time",
+    "The timestamp of the last PDU which was successfully received from the given domain",
     labelnames=("server_name",),
 )
 
 
 class FederationServer(FederationBase):
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
         super().__init__(hs)
 
         self.auth = hs.get_auth()
@@ -111,14 +112,15 @@ class FederationServer(FederationBase):
         # with FederationHandlerRegistry.
         hs.get_directory_handler()
 
-        self._federation_ratelimiter = hs.get_federation_ratelimiter()
-
         self._server_linearizer = Linearizer("fed_server")
-        self._transaction_linearizer = Linearizer("fed_txn_handler")
+
+        # origins that we are currently processing a transaction from.
+        # a dict from origin to txn id.
+        self._active_transactions = {}  # type: Dict[str, str]
 
         # We cache results for transaction with the same ID
         self._transaction_resp_cache = ResponseCache(
-            hs, "fed_txn_handler", timeout_ms=30000
+            hs.get_clock(), "fed_txn_handler", timeout_ms=30000
         )  # type: ResponseCache[Tuple[str, str]]
 
         self.transaction_actions = TransactionActions(self.store)
@@ -128,10 +130,10 @@ class FederationServer(FederationBase):
         # We cache responses to state queries, as they take a while and often
         # come in waves.
         self._state_resp_cache = ResponseCache(
-            hs, "state_resp", timeout_ms=30000
+            hs.get_clock(), "state_resp", timeout_ms=30000
         )  # type: ResponseCache[Tuple[str, str]]
         self._state_ids_resp_cache = ResponseCache(
-            hs, "state_ids_resp", timeout_ms=30000
+            hs.get_clock(), "state_ids_resp", timeout_ms=30000
         )  # type: ResponseCache[Tuple[str, str]]
 
         self._federation_metrics_domains = (
@@ -168,6 +170,33 @@ class FederationServer(FederationBase):
 
         logger.debug("[%s] Got transaction", transaction_id)
 
+        # Reject malformed transactions early: reject if too many PDUs/EDUs
+        if len(transaction.pdus) > 50 or (  # type: ignore
+            hasattr(transaction, "edus") and len(transaction.edus) > 100  # type: ignore
+        ):
+            logger.info("Transaction PDU or EDU count too large. Returning 400")
+            return 400, {}
+
+        # we only process one transaction from each origin at a time. We need to do
+        # this check here, rather than in _on_incoming_transaction_inner so that we
+        # don't cache the rejection in _transaction_resp_cache (so that if the txn
+        # arrives again later, we can process it).
+        current_transaction = self._active_transactions.get(origin)
+        if current_transaction and current_transaction != transaction_id:
+            logger.warning(
+                "Received another txn %s from %s while still processing %s",
+                transaction_id,
+                origin,
+                current_transaction,
+            )
+            return 429, {
+                "errcode": Codes.UNKNOWN,
+                "error": "Too many concurrent transactions",
+            }
+
+        # CRITICAL SECTION: we must now not await until we populate _active_transactions
+        # in _on_incoming_transaction_inner.
+
         # We wrap in a ResponseCache so that we de-duplicate retried
         # transactions.
         return await self._transaction_resp_cache.wrap(
@@ -181,26 +210,18 @@ class FederationServer(FederationBase):
     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
+        # CRITICAL SECTION: the first thing we must do (before awaiting) is
+        # add an entry to _active_transactions.
+        assert origin not in self._active_transactions
+        self._active_transactions[origin] = transaction.transaction_id  # type: ignore
 
-                result = await self._handle_incoming_transaction(
-                    origin, transaction, request_time
-                )
-
-        return result
+        try:
+            result = await self._handle_incoming_transaction(
+                origin, transaction, request_time
+            )
+            return result
+        finally:
+            del self._active_transactions[origin]
 
     async def _handle_incoming_transaction(
         self, origin: str, transaction: Transaction, request_time: int
@@ -226,19 +247,6 @@ class FederationServer(FederationBase):
 
         logger.debug("[%s] Transaction is new", transaction.transaction_id)  # type: ignore
 
-        # Reject if PDU count > 50 or EDU count > 100
-        if len(transaction.pdus) > 50 or (  # type: ignore
-            hasattr(transaction, "edus") and len(transaction.edus) > 100  # type: ignore
-        ):
-
-            logger.info("Transaction PDU or EDU count too large. Returning 400")
-
-            response = {}
-            await self.transaction_actions.set_response(
-                origin, transaction, 400, response
-            )
-            return 400, response
-
         # We process PDUs and EDUs in parallel. This is important as we don't
         # want to block things like to device messages from reaching clients
         # behind the potentially expensive handling of PDUs.
@@ -334,42 +342,48 @@ class FederationServer(FederationBase):
         # impose a limit to avoid going too crazy with ram/cpu.
 
         async def process_pdus_for_room(room_id: str):
-            logger.debug("Processing PDUs for %s", room_id)
-            try:
-                await self.check_server_matches_acl(origin_host, room_id)
-            except AuthError as e:
-                logger.warning("Ignoring PDUs for room %s from banned server", room_id)
-                for pdu in pdus_by_room[room_id]:
-                    event_id = pdu.event_id
-                    pdu_results[event_id] = e.error_dict()
-                return
+            with nested_logging_context(room_id):
+                logger.debug("Processing PDUs for %s", room_id)
 
-            for pdu in pdus_by_room[room_id]:
-                event_id = pdu.event_id
-                with pdu_process_time.time():
-                    with nested_logging_context(event_id):
-                        try:
-                            await self._handle_received_pdu(origin, pdu)
-                            pdu_results[event_id] = {}
-                        except FederationError as e:
-                            logger.warning("Error handling PDU %s: %s", event_id, e)
-                            pdu_results[event_id] = {"error": str(e)}
-                        except Exception as e:
-                            f = failure.Failure()
-                            pdu_results[event_id] = {"error": str(e)}
-                            logger.error(
-                                "Failed to handle PDU %s",
-                                event_id,
-                                exc_info=(f.type, f.value, f.getTracebackObject()),
-                            )
+                try:
+                    await self.check_server_matches_acl(origin_host, room_id)
+                except AuthError as e:
+                    logger.warning(
+                        "Ignoring PDUs for room %s from banned server", room_id
+                    )
+                    for pdu in pdus_by_room[room_id]:
+                        event_id = pdu.event_id
+                        pdu_results[event_id] = e.error_dict()
+                    return
+
+                for pdu in pdus_by_room[room_id]:
+                    pdu_results[pdu.event_id] = await process_pdu(pdu)
+
+        async def process_pdu(pdu: EventBase) -> JsonDict:
+            event_id = pdu.event_id
+            with pdu_process_time.time():
+                with nested_logging_context(event_id):
+                    try:
+                        await self._handle_received_pdu(origin, pdu)
+                        return {}
+                    except FederationError as e:
+                        logger.warning("Error handling PDU %s: %s", event_id, e)
+                        return {"error": str(e)}
+                    except Exception as e:
+                        f = failure.Failure()
+                        logger.error(
+                            "Failed to handle PDU %s",
+                            event_id,
+                            exc_info=(f.type, f.value, f.getTracebackObject()),  # type: ignore
+                        )
+                        return {"error": str(e)}
 
         await concurrently_execute(
             process_pdus_for_room, pdus_by_room.keys(), TRANSACTION_CONCURRENCY_LIMIT
         )
 
         if newest_pdu_ts and origin in self._federation_metrics_domains:
-            newest_pdu_age = self._clock.time_msec() - newest_pdu_ts
-            last_pdu_age_metric.labels(server_name=origin).set(newest_pdu_age / 1000)
+            last_pdu_ts_metric.labels(server_name=origin).set(newest_pdu_ts / 1000)
 
         return pdu_results
 
@@ -447,18 +461,22 @@ class FederationServer(FederationBase):
 
     async def _on_state_ids_request_compute(self, room_id, event_id):
         state_ids = await self.handler.get_state_ids_for_pdu(room_id, event_id)
-        auth_chain_ids = await self.store.get_auth_chain_ids(state_ids)
+        auth_chain_ids = await self.store.get_auth_chain_ids(room_id, state_ids)
         return {"pdu_ids": state_ids, "auth_chain_ids": auth_chain_ids}
 
     async def _on_context_state_request_compute(
         self, room_id: str, event_id: str
     ) -> Dict[str, list]:
         if event_id:
-            pdus = await self.handler.get_state_for_pdu(room_id, event_id)
+            pdus = await self.handler.get_state_for_pdu(
+                room_id, event_id
+            )  # type: Iterable[EventBase]
         else:
             pdus = (await self.state.get_current_state(room_id)).values()
 
-        auth_chain = await self.store.get_auth_chain([pdu.event_id for pdu in pdus])
+        auth_chain = await self.store.get_auth_chain(
+            room_id, [pdu.event_id for pdu in pdus]
+        )
 
         return {
             "pdus": [pdu.get_pdu_json() for pdu in pdus],
@@ -862,13 +880,22 @@ class FederationHandlerRegistry:
         self.edu_handlers = (
             {}
         )  # type: Dict[str, Callable[[str, dict], Awaitable[None]]]
-        self.query_handlers = {}  # type: Dict[str, Callable[[dict], Awaitable[None]]]
+        self.query_handlers = (
+            {}
+        )  # type: Dict[str, Callable[[dict], Awaitable[JsonDict]]]
 
         # Map from type to instance names that we should route EDU handling to.
         # We randomly choose one instance from the list to route to for each new
         # EDU received.
         self._edu_type_to_instance = {}  # type: Dict[str, List[str]]
 
+        # A rate limiter for incoming room key requests per origin.
+        self._room_key_request_rate_limiter = Ratelimiter(
+            clock=self.clock,
+            rate_hz=self.config.rc_key_requests.per_second,
+            burst_count=self.config.rc_key_requests.burst_count,
+        )
+
     def register_edu_handler(
         self, edu_type: str, handler: Callable[[str, JsonDict], Awaitable[None]]
     ):
@@ -889,7 +916,7 @@ class FederationHandlerRegistry:
         self.edu_handlers[edu_type] = handler
 
     def register_query_handler(
-        self, query_type: str, handler: Callable[[dict], defer.Deferred]
+        self, query_type: str, handler: Callable[[dict], Awaitable[JsonDict]]
     ):
         """Sets the handler callable that will be used to handle an incoming
         federation query of the given type.
@@ -917,7 +944,15 @@ class FederationHandlerRegistry:
         self._edu_type_to_instance[edu_type] = instance_names
 
     async def on_edu(self, edu_type: str, origin: str, content: dict):
-        if not self.config.use_presence and edu_type == "m.presence":
+        if not self.config.use_presence and edu_type == EduTypes.Presence:
+            return
+
+        # If the incoming room key requests from a particular origin are over
+        # the limit, drop them.
+        if (
+            edu_type == EduTypes.RoomKeyRequest
+            and not self._room_key_request_rate_limiter.can_do_action(origin)
+        ):
             return
 
         # Check if we have a handler on this instance
@@ -954,7 +989,7 @@ class FederationHandlerRegistry:
         # Oh well, let's just log and move on.
         logger.warning("No handler registered for EDU type %s", edu_type)
 
-    async def on_query(self, query_type: str, args: dict):
+    async def on_query(self, query_type: str, args: dict) -> JsonDict:
         handler = self.query_handlers.get(query_type)
         if handler:
             return await handler(args)

synapse/federation/sender/__init__.py

@@ -474,7 +474,7 @@ class FederationSender:
             self._processing_pending_presence = False
 
     def send_presence_to_destinations(
-        self, states: List[UserPresenceState], destinations: List[str]
+        self, states: Iterable[UserPresenceState], destinations: Iterable[str]
     ) -> None:
         """Send the given presence states to the given destinations.
         destinations (list[str])

synapse/federation/sender/per_destination_queue.py

@@ -17,6 +17,7 @@ import datetime
 import logging
 from typing import TYPE_CHECKING, Dict, Hashable, Iterable, List, Optional, Tuple, cast
 
+import attr
 from prometheus_client import Counter
 
 from synapse.api.errors import (
@@ -93,6 +94,10 @@ class PerDestinationQueue:
         self._destination = destination
         self.transmission_loop_running = False
 
+        # Flag to signal to any running transmission loop that there is new data
+        # queued up to be sent.
+        self._new_data_to_send = False
+
         # True whilst we are sending events that the remote homeserver missed
         # because it was unreachable. We start in this state so we can perform
         # catch-up at startup.
@@ -108,7 +113,7 @@ class PerDestinationQueue:
         # destination (we are the only updater so this is safe)
         self._last_successful_stream_ordering = None  # type: Optional[int]
 
-        # a list of pending PDUs
+        # a queue of pending PDUs
         self._pending_pdus = []  # type: List[EventBase]
 
         # XXX this is never actually used: see
@@ -208,6 +213,10 @@ class PerDestinationQueue:
         transaction in the background.
         """
 
+        # Mark that we (may) have new things to send, so that any running
+        # transmission loop will recheck whether there is stuff to send.
+        self._new_data_to_send = True
+
         if self.transmission_loop_running:
             # XXX: this can get stuck on by a never-ending
             # request at which point pending_pdus just keeps growing.
@@ -250,125 +259,41 @@ class PerDestinationQueue:
 
             pending_pdus = []
             while True:
-                # We have to keep 2 free slots for presence and rr_edus
-                limit = MAX_EDUS_PER_TRANSACTION - 2
+                self._new_data_to_send = False
 
-                device_update_edus, dev_list_id = await self._get_device_update_edus(
-                    limit
-                )
-
-                limit -= len(device_update_edus)
-
-                (
-                    to_device_edus,
-                    device_stream_id,
-                ) = await self._get_to_device_message_edus(limit)
-
-                pending_edus = device_update_edus + to_device_edus
-
-                # BEGIN CRITICAL SECTION
-                #
-                # In order to avoid a race condition, we need to make sure that
-                # the following code (from popping the queues up to the point
-                # where we decide if we actually have any pending messages) is
-                # atomic - otherwise new PDUs or EDUs might arrive in the
-                # meantime, but not get sent because we hold the
-                # transmission_loop_running flag.
-
-                pending_pdus = self._pending_pdus
-
-                # We can only include at most 50 PDUs per transactions
-                pending_pdus, self._pending_pdus = pending_pdus[:50], pending_pdus[50:]
-
-                pending_edus.extend(self._get_rr_edus(force_flush=False))
-                pending_presence = self._pending_presence
-                self._pending_presence = {}
-                if pending_presence:
-                    pending_edus.append(
-                        Edu(
-                            origin=self._server_name,
-                            destination=self._destination,
-                            edu_type="m.presence",
-                            content={
-                                "push": [
-                                    format_user_presence_state(
-                                        presence, self._clock.time_msec()
-                                    )
-                                    for presence in pending_presence.values()
-                                ]
-                            },
-                        )
-                    )
-
-                pending_edus.extend(
-                    self._pop_pending_edus(MAX_EDUS_PER_TRANSACTION - len(pending_edus))
-                )
-                while (
-                    len(pending_edus) < MAX_EDUS_PER_TRANSACTION
-                    and self._pending_edus_keyed
+                async with _TransactionQueueManager(self) as (
+                    pending_pdus,
+                    pending_edus,
                 ):
-                    _, val = self._pending_edus_keyed.popitem()
-                    pending_edus.append(val)
+                    if not pending_pdus and not pending_edus:
+                        logger.debug("TX [%s] Nothing to send", self._destination)
 
-                if pending_pdus:
-                    logger.debug(
-                        "TX [%s] len(pending_pdus_by_dest[dest]) = %d",
-                        self._destination,
-                        len(pending_pdus),
+                        # If we've gotten told about new things to send during
+                        # checking for things to send, we try looking again.
+                        # Otherwise new PDUs or EDUs might arrive in the meantime,
+                        # but not get sent because we hold the
+                        # `transmission_loop_running` flag.
+                        if self._new_data_to_send:
+                            continue
+                        else:
+                            return
+
+                    if pending_pdus:
+                        logger.debug(
+                            "TX [%s] len(pending_pdus_by_dest[dest]) = %d",
+                            self._destination,
+                            len(pending_pdus),
+                        )
+
+                    await self._transaction_manager.send_new_transaction(
+                        self._destination, pending_pdus, pending_edus
                     )
 
-                if not pending_pdus and not pending_edus:
-                    logger.debug("TX [%s] Nothing to send", self._destination)
-                    self._last_device_stream_id = device_stream_id
-                    return
-
-                # if we've decided to send a transaction anyway, and we have room, we
-                # may as well send any pending RRs
-                if len(pending_edus) < MAX_EDUS_PER_TRANSACTION:
-                    pending_edus.extend(self._get_rr_edus(force_flush=True))
-
-                # END CRITICAL SECTION
-
-                success = await self._transaction_manager.send_new_transaction(
-                    self._destination, pending_pdus, pending_edus
-                )
-                if success:
                     sent_transactions_counter.inc()
                     sent_edus_counter.inc(len(pending_edus))
                     for edu in pending_edus:
                         sent_edus_by_type.labels(edu.edu_type).inc()
-                    # Remove the acknowledged device messages from the database
-                    # Only bother if we actually sent some device messages
-                    if to_device_edus:
-                        await self._store.delete_device_msgs_for_remote(
-                            self._destination, device_stream_id
-                        )
 
-                    # also mark the device updates as sent
-                    if device_update_edus:
-                        logger.info(
-                            "Marking as sent %r %r", self._destination, dev_list_id
-                        )
-                        await self._store.mark_as_sent_devices_by_remote(
-                            self._destination, dev_list_id
-                        )
-
-                    self._last_device_stream_id = device_stream_id
-                    self._last_device_list_stream_id = dev_list_id
-
-                    if pending_pdus:
-                        # we sent some PDUs and it was successful, so update our
-                        # last_successful_stream_ordering in the destinations table.
-                        final_pdu = pending_pdus[-1]
-                        last_successful_stream_ordering = (
-                            final_pdu.internal_metadata.stream_ordering
-                        )
-                        assert last_successful_stream_ordering
-                        await self._store.set_destination_last_successful_stream_ordering(
-                            self._destination, last_successful_stream_ordering
-                        )
-                else:
-                    break
         except NotRetryingDestination as e:
             logger.debug(
                 "TX [%s] not ready for retry yet (next retry at %s) - "
@@ -401,7 +326,7 @@ class PerDestinationQueue:
                 self._pending_presence = {}
                 self._pending_rrs = {}
 
-            self._start_catching_up()
+                self._start_catching_up()
         except FederationDeniedError as e:
             logger.info(e)
         except HttpResponseException as e:
@@ -412,7 +337,6 @@ class PerDestinationQueue:
                 e,
             )
 
-            self._start_catching_up()
         except RequestSendFailed as e:
             logger.warning(
                 "TX [%s] Failed to send transaction: %s", self._destination, e
@@ -422,16 +346,12 @@ class PerDestinationQueue:
                 logger.info(
                     "Failed to send event %s to %s", p.event_id, self._destination
                 )
-
-            self._start_catching_up()
         except Exception:
             logger.exception("TX [%s] Failed to send transaction", self._destination)
             for p in pending_pdus:
                 logger.info(
                     "Failed to send event %s to %s", p.event_id, self._destination
                 )
-
-            self._start_catching_up()
         finally:
             # We want to be *very* sure we clear this after we stop processing
             self.transmission_loop_running = False
@@ -499,13 +419,10 @@ class PerDestinationQueue:
                 rooms = [p.room_id for p in catchup_pdus]
                 logger.info("Catching up rooms to %s: %r", self._destination, rooms)
 
-            success = await self._transaction_manager.send_new_transaction(
+            await self._transaction_manager.send_new_transaction(
                 self._destination, catchup_pdus, []
             )
 
-            if not success:
-                return
-
             sent_transactions_counter.inc()
             final_pdu = catchup_pdus[-1]
             self._last_successful_stream_ordering = cast(
@@ -584,3 +501,135 @@ class PerDestinationQueue:
         """
         self._catching_up = True
         self._pending_pdus = []
+
+
+@attr.s(slots=True)
+class _TransactionQueueManager:
+    """A helper async context manager for pulling stuff off the queues and
+    tracking what was last successfully sent, etc.
+    """
+
+    queue = attr.ib(type=PerDestinationQueue)
+
+    _device_stream_id = attr.ib(type=Optional[int], default=None)
+    _device_list_id = attr.ib(type=Optional[int], default=None)
+    _last_stream_ordering = attr.ib(type=Optional[int], default=None)
+    _pdus = attr.ib(type=List[EventBase], factory=list)
+
+    async def __aenter__(self) -> Tuple[List[EventBase], List[Edu]]:
+        # First we calculate the EDUs we want to send, if any.
+
+        # We start by fetching device related EDUs, i.e device updates and to
+        # device messages. We have to keep 2 free slots for presence and rr_edus.
+        limit = MAX_EDUS_PER_TRANSACTION - 2
+
+        device_update_edus, dev_list_id = await self.queue._get_device_update_edus(
+            limit
+        )
+
+        if device_update_edus:
+            self._device_list_id = dev_list_id
+        else:
+            self.queue._last_device_list_stream_id = dev_list_id
+
+        limit -= len(device_update_edus)
+
+        (
+            to_device_edus,
+            device_stream_id,
+        ) = await self.queue._get_to_device_message_edus(limit)
+
+        if to_device_edus:
+            self._device_stream_id = device_stream_id
+        else:
+            self.queue._last_device_stream_id = device_stream_id
+
+        pending_edus = device_update_edus + to_device_edus
+
+        # Now add the read receipt EDU.
+        pending_edus.extend(self.queue._get_rr_edus(force_flush=False))
+
+        # And presence EDU.
+        if self.queue._pending_presence:
+            pending_edus.append(
+                Edu(
+                    origin=self.queue._server_name,
+                    destination=self.queue._destination,
+                    edu_type="m.presence",
+                    content={
+                        "push": [
+                            format_user_presence_state(
+                                presence, self.queue._clock.time_msec()
+                            )
+                            for presence in self.queue._pending_presence.values()
+                        ]
+                    },
+                )
+            )
+            self.queue._pending_presence = {}
+
+        # Finally add any other types of EDUs if there is room.
+        pending_edus.extend(
+            self.queue._pop_pending_edus(MAX_EDUS_PER_TRANSACTION - len(pending_edus))
+        )
+        while (
+            len(pending_edus) < MAX_EDUS_PER_TRANSACTION
+            and self.queue._pending_edus_keyed
+        ):
+            _, val = self.queue._pending_edus_keyed.popitem()
+            pending_edus.append(val)
+
+        # Now we look for any PDUs to send, by getting up to 50 PDUs from the
+        # queue
+        self._pdus = self.queue._pending_pdus[:50]
+
+        if not self._pdus and not pending_edus:
+            return [], []
+
+        # if we've decided to send a transaction anyway, and we have room, we
+        # may as well send any pending RRs
+        if len(pending_edus) < MAX_EDUS_PER_TRANSACTION:
+            pending_edus.extend(self.queue._get_rr_edus(force_flush=True))
+
+        if self._pdus:
+            self._last_stream_ordering = self._pdus[
+                -1
+            ].internal_metadata.stream_ordering
+            assert self._last_stream_ordering
+
+        return self._pdus, pending_edus
+
+    async def __aexit__(self, exc_type, exc, tb):
+        if exc_type is not None:
+            # Failed to send transaction, so we bail out.
+            return
+
+        # Successfully sent transactions, so we remove pending PDUs from the queue
+        if self._pdus:
+            self.queue._pending_pdus = self.queue._pending_pdus[len(self._pdus) :]
+
+        # Succeeded to send the transaction so we record where we have sent up
+        # to in the various streams
+
+        if self._device_stream_id:
+            await self.queue._store.delete_device_msgs_for_remote(
+                self.queue._destination, self._device_stream_id
+            )
+            self.queue._last_device_stream_id = self._device_stream_id
+
+        # also mark the device updates as sent
+        if self._device_list_id:
+            logger.info(
+                "Marking as sent %r %r", self.queue._destination, self._device_list_id
+            )
+            await self.queue._store.mark_as_sent_devices_by_remote(
+                self.queue._destination, self._device_list_id
+            )
+            self.queue._last_device_list_stream_id = self._device_list_id
+
+        if self._last_stream_ordering:
+            # we sent some PDUs and it was successful, so update our
+            # last_successful_stream_ordering in the destinations table.
+            await self.queue._store.set_destination_last_successful_stream_ordering(
+                self.queue._destination, self._last_stream_ordering
+            )

synapse/federation/sender/transaction_manager.py

@@ -36,9 +36,9 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
-last_pdu_age_metric = Gauge(
-    "synapse_federation_last_sent_pdu_age",
-    "The age (in seconds) of the last PDU successfully sent to the given domain",
+last_pdu_ts_metric = Gauge(
+    "synapse_federation_last_sent_pdu_time",
+    "The timestamp of the last PDU which was successfully sent to the given domain",
     labelnames=("server_name",),
 )
 
@@ -69,15 +69,12 @@ class TransactionManager:
         destination: str,
         pdus: List[EventBase],
         edus: List[Edu],
-    ) -> bool:
+    ) -> None:
         """
         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
@@ -96,8 +93,6 @@ class TransactionManager:
                 edu.strip_context()
 
         with start_active_span_follows_from("send_transaction", span_contexts):
-            success = True
-
             logger.debug("TX [%s] _attempt_new_transaction", destination)
 
             txn_id = str(self._next_txn_id)
@@ -152,45 +147,29 @@ class TransactionManager:
                 response = await self._transport_layer.send_transaction(
                     transaction, json_data_cb
                 )
-                code = 200
             except HttpResponseException as e:
                 code = e.code
                 response = e.response
 
-                if e.code in (401, 404, 429) or 500 <= e.code:
-                    logger.info(
-                        "TX [%s] {%s} got %d response", destination, txn_id, code
-                    )
-                    raise e
+                set_tag(tags.ERROR, True)
 
-            logger.info("TX [%s] {%s} got %d response", destination, txn_id, code)
+                logger.info("TX [%s] {%s} got %d response", destination, txn_id, code)
+                raise
 
-            if code == 200:
-                for e_id, r in response.get("pdus", {}).items():
-                    if "error" in r:
-                        logger.warning(
-                            "TX [%s] {%s} Remote returned error for %s: %s",
-                            destination,
-                            txn_id,
-                            e_id,
-                            r,
-                        )
-            else:
-                for p in pdus:
+            logger.info("TX [%s] {%s} got 200 response", destination, txn_id)
+
+            for e_id, r in response.get("pdus", {}).items():
+                if "error" in r:
                     logger.warning(
-                        "TX [%s] {%s} Failed to send event %s",
+                        "TX [%s] {%s} Remote returned error for %s: %s",
                         destination,
                         txn_id,
-                        p.event_id,
+                        e_id,
+                        r,
                     )
-                success = False
 
-            if success and pdus and destination in self._federation_metrics_domains:
+            if pdus and destination in self._federation_metrics_domains:
                 last_pdu = pdus[-1]
-                last_pdu_age = self.clock.time_msec() - last_pdu.origin_server_ts
-                last_pdu_age_metric.labels(server_name=destination).set(
-                    last_pdu_age / 1000
+                last_pdu_ts_metric.labels(server_name=destination).set(
+                    last_pdu.origin_server_ts / 1000
                 )
-
-            set_tag(tags.ERROR, not success)
-            return success

synapse/federation/transport/server.py

@@ -21,6 +21,7 @@ import re
 from typing import Optional, Tuple, Type
 
 import synapse
+from synapse.api.constants import MAX_GROUP_CATEGORYID_LENGTH, MAX_GROUP_ROLEID_LENGTH
 from synapse.api.errors import Codes, FederationDeniedError, SynapseError
 from synapse.api.room_versions import RoomVersions
 from synapse.api.urls import (
@@ -483,10 +484,9 @@ class FederationQueryServlet(BaseFederationServlet):
 
     # This is when we receive a server-server Query
     async def on_GET(self, origin, content, query, query_type):
-        return await self.handler.on_query_request(
-            query_type,
-            {k.decode("utf8"): v[0].decode("utf-8") for k, v in query.items()},
-        )
+        args = {k.decode("utf8"): v[0].decode("utf-8") for k, v in query.items()}
+        args["origin"] = origin
+        return await self.handler.on_query_request(query_type, args)
 
 
 class FederationMakeJoinServlet(BaseFederationServlet):
@@ -1118,7 +1118,17 @@ class FederationGroupsSummaryRoomsServlet(BaseFederationServlet):
             raise SynapseError(403, "requester_user_id doesn't match origin")
 
         if category_id == "":
-            raise SynapseError(400, "category_id cannot be empty string")
+            raise SynapseError(
+                400, "category_id cannot be empty string", Codes.INVALID_PARAM
+            )
+
+        if len(category_id) > MAX_GROUP_CATEGORYID_LENGTH:
+            raise SynapseError(
+                400,
+                "category_id may not be longer than %s characters"
+                % (MAX_GROUP_CATEGORYID_LENGTH,),
+                Codes.INVALID_PARAM,
+            )
 
         resp = await self.handler.update_group_summary_room(
             group_id,
@@ -1184,6 +1194,14 @@ class FederationGroupsCategoryServlet(BaseFederationServlet):
         if category_id == "":
             raise SynapseError(400, "category_id cannot be empty string")
 
+        if len(category_id) > MAX_GROUP_CATEGORYID_LENGTH:
+            raise SynapseError(
+                400,
+                "category_id may not be longer than %s characters"
+                % (MAX_GROUP_CATEGORYID_LENGTH,),
+                Codes.INVALID_PARAM,
+            )
+
         resp = await self.handler.upsert_group_category(
             group_id, requester_user_id, category_id, content
         )
@@ -1240,7 +1258,17 @@ class FederationGroupsRoleServlet(BaseFederationServlet):
             raise SynapseError(403, "requester_user_id doesn't match origin")
 
         if role_id == "":
-            raise SynapseError(400, "role_id cannot be empty string")
+            raise SynapseError(
+                400, "role_id cannot be empty string", Codes.INVALID_PARAM
+            )
+
+        if len(role_id) > MAX_GROUP_ROLEID_LENGTH:
+            raise SynapseError(
+                400,
+                "role_id may not be longer than %s characters"
+                % (MAX_GROUP_ROLEID_LENGTH,),
+                Codes.INVALID_PARAM,
+            )
 
         resp = await self.handler.update_group_role(
             group_id, requester_user_id, role_id, content
@@ -1285,6 +1313,14 @@ class FederationGroupsSummaryUsersServlet(BaseFederationServlet):
         if role_id == "":
             raise SynapseError(400, "role_id cannot be empty string")
 
+        if len(role_id) > MAX_GROUP_ROLEID_LENGTH:
+            raise SynapseError(
+                400,
+                "role_id may not be longer than %s characters"
+                % (MAX_GROUP_ROLEID_LENGTH,),
+                Codes.INVALID_PARAM,
+            )
+
         resp = await self.handler.update_group_summary_user(
             group_id,
             requester_user_id,

synapse/groups/attestations.py

@@ -37,13 +37,16 @@ An attestation is a signed blob of json that looks like:
 
 import logging
 import random
-from typing import Tuple
+from typing import TYPE_CHECKING, Optional, Tuple
 
 from signedjson.sign import sign_json
 
 from synapse.api.errors import HttpResponseException, RequestSendFailed, SynapseError
 from synapse.metrics.background_process_metrics import run_as_background_process
-from synapse.types import get_domain_from_id
+from synapse.types import JsonDict, get_domain_from_id
+
+if TYPE_CHECKING:
+    from synapse.app.homeserver import HomeServer
 
 logger = logging.getLogger(__name__)
 
@@ -63,15 +66,19 @@ UPDATE_ATTESTATION_TIME_MS = 1 * 24 * 60 * 60 * 1000
 class GroupAttestationSigning:
     """Creates and verifies group attestations."""
 
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
         self.keyring = hs.get_keyring()
         self.clock = hs.get_clock()
         self.server_name = hs.hostname
         self.signing_key = hs.signing_key
 
     async def verify_attestation(
-        self, attestation, group_id, user_id, server_name=None
-    ):
+        self,
+        attestation: JsonDict,
+        group_id: str,
+        user_id: str,
+        server_name: Optional[str] = None,
+    ) -> None:
         """Verifies that the given attestation matches the given parameters.
 
         An optional server_name can be supplied to explicitly set which server's
@@ -100,16 +107,18 @@ class GroupAttestationSigning:
         if valid_until_ms < now:
             raise SynapseError(400, "Attestation expired")
 
+        assert server_name is not None
         await self.keyring.verify_json_for_server(
             server_name, attestation, now, "Group attestation"
         )
 
-    def create_attestation(self, group_id, user_id):
+    def create_attestation(self, group_id: str, user_id: str) -> JsonDict:
         """Create an attestation for the group_id and user_id with default
         validity length.
         """
-        validity_period = DEFAULT_ATTESTATION_LENGTH_MS
-        validity_period *= random.uniform(*DEFAULT_ATTESTATION_JITTER)
+        validity_period = DEFAULT_ATTESTATION_LENGTH_MS * random.uniform(
+            *DEFAULT_ATTESTATION_JITTER
+        )
         valid_until_ms = int(self.clock.time_msec() + validity_period)
 
         return sign_json(
@@ -126,7 +135,7 @@ class GroupAttestationSigning:
 class GroupAttestionRenewer:
     """Responsible for sending and receiving attestation updates."""
 
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
         self.clock = hs.get_clock()
         self.store = hs.get_datastore()
         self.assestations = hs.get_groups_attestation_signing()
@@ -139,7 +148,9 @@ class GroupAttestionRenewer:
                 self._start_renew_attestations, 30 * 60 * 1000
             )
 
-    async def on_renew_attestation(self, group_id, user_id, content):
+    async def on_renew_attestation(
+        self, group_id: str, user_id: str, content: JsonDict
+    ) -> JsonDict:
         """When a remote updates an attestation"""
         attestation = content["attestation"]
 
@@ -154,10 +165,10 @@ class GroupAttestionRenewer:
 
         return {}
 
-    def _start_renew_attestations(self):
+    def _start_renew_attestations(self) -> None:
         return run_as_background_process("renew_attestations", self._renew_attestations)
 
-    async def _renew_attestations(self):
+    async def _renew_attestations(self) -> None:
         """Called periodically to check if we need to update any of our attestations"""
 
         now = self.clock.time_msec()
@@ -166,7 +177,7 @@ class GroupAttestionRenewer:
             now + UPDATE_ATTESTATION_TIME_MS
         )
 
-        async def _renew_attestation(group_user: Tuple[str, str]):
+        async def _renew_attestation(group_user: Tuple[str, str]) -> None:
             group_id, user_id = group_user
             try:
                 if not self.is_mine_id(group_id):

synapse/groups/groups_server.py

@@ -16,12 +16,17 @@
 # limitations under the License.
 
 import logging
+from typing import TYPE_CHECKING, Optional
 
 from synapse.api.errors import Codes, SynapseError
+from synapse.handlers.groups_local import GroupsLocalHandler
 from synapse.handlers.profile import MAX_AVATAR_URL_LEN, MAX_DISPLAYNAME_LEN
-from synapse.types import GroupID, RoomID, UserID, get_domain_from_id
+from synapse.types import GroupID, JsonDict, RoomID, UserID, get_domain_from_id
 from synapse.util.async_helpers import concurrently_execute
 
+if TYPE_CHECKING:
+    from synapse.app.homeserver import HomeServer
+
 logger = logging.getLogger(__name__)
 
 
@@ -39,7 +44,7 @@ MAX_LONG_DESC_LEN = 10000
 
 
 class GroupsServerWorkerHandler:
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
         self.hs = hs
         self.store = hs.get_datastore()
         self.room_list_handler = hs.get_room_list_handler()
@@ -54,16 +59,21 @@ class GroupsServerWorkerHandler:
         self.profile_handler = hs.get_profile_handler()
 
     async def check_group_is_ours(
-        self, group_id, requester_user_id, and_exists=False, and_is_admin=None
-    ):
+        self,
+        group_id: str,
+        requester_user_id: str,
+        and_exists: bool = False,
+        and_is_admin: Optional[str] = None,
+    ) -> Optional[dict]:
         """Check that the group is ours, and optionally if it exists.
 
         If group does exist then return group.
 
         Args:
-            group_id (str)
-            and_exists (bool): whether to also check if group exists
-            and_is_admin (str): whether to also check if given str is a user_id
+            group_id: The group ID to check.
+            requester_user_id: The user ID of the requester.
+            and_exists: whether to also check if group exists
+            and_is_admin: whether to also check if given str is a user_id
                 that is an admin
         """
         if not self.is_mine_id(group_id):
@@ -86,7 +96,9 @@ class GroupsServerWorkerHandler:
 
         return group
 
-    async def get_group_summary(self, group_id, requester_user_id):
+    async def get_group_summary(
+        self, group_id: str, requester_user_id: str
+    ) -> JsonDict:
         """Get the summary for a group as seen by requester_user_id.
 
         The group summary consists of the profile of the room, and a curated
@@ -119,6 +131,8 @@ class GroupsServerWorkerHandler:
             entry = await self.room_list_handler.generate_room_entry(
                 room_id, len(joined_users), with_alias=False, allow_private=True
             )
+            if entry is None:
+                continue
             entry = dict(entry)  # so we don't change what's cached
             entry.pop("room_id", None)
 
@@ -126,22 +140,22 @@ class GroupsServerWorkerHandler:
 
         rooms.sort(key=lambda e: e.get("order", 0))
 
-        for entry in users:
-            user_id = entry["user_id"]
+        for user in users:
+            user_id = user["user_id"]
 
             if not self.is_mine_id(requester_user_id):
                 attestation = await self.store.get_remote_attestation(group_id, user_id)
                 if not attestation:
                     continue
 
-                entry["attestation"] = attestation
+                user["attestation"] = attestation
             else:
-                entry["attestation"] = self.attestations.create_attestation(
+                user["attestation"] = self.attestations.create_attestation(
                     group_id, user_id
                 )
 
             user_profile = await self.profile_handler.get_profile_from_cache(user_id)
-            entry.update(user_profile)
+            user.update(user_profile)
 
         users.sort(key=lambda e: e.get("order", 0))
 
@@ -164,40 +178,43 @@ class GroupsServerWorkerHandler:
             "user": membership_info,
         }
 
-    async def get_group_categories(self, group_id, requester_user_id):
+    async def get_group_categories(
+        self, group_id: str, requester_user_id: str
+    ) -> JsonDict:
         """Get all categories in a group (as seen by user)"""
         await self.check_group_is_ours(group_id, requester_user_id, and_exists=True)
 
         categories = await self.store.get_group_categories(group_id=group_id)
         return {"categories": categories}
 
-    async def get_group_category(self, group_id, requester_user_id, category_id):
+    async def get_group_category(
+        self, group_id: str, requester_user_id: str, category_id: str
+    ) -> JsonDict:
         """Get a specific category in a group (as seen by user)"""
         await self.check_group_is_ours(group_id, requester_user_id, and_exists=True)
 
-        res = await self.store.get_group_category(
+        return await self.store.get_group_category(
             group_id=group_id, category_id=category_id
         )
 
-        logger.info("group %s", res)
-
-        return res
-
-    async def get_group_roles(self, group_id, requester_user_id):
+    async def get_group_roles(self, group_id: str, requester_user_id: str) -> JsonDict:
         """Get all roles in a group (as seen by user)"""
         await self.check_group_is_ours(group_id, requester_user_id, and_exists=True)
 
         roles = await self.store.get_group_roles(group_id=group_id)
         return {"roles": roles}
 
-    async def get_group_role(self, group_id, requester_user_id, role_id):
+    async def get_group_role(
+        self, group_id: str, requester_user_id: str, role_id: str
+    ) -> JsonDict:
         """Get a specific role in a group (as seen by user)"""
         await self.check_group_is_ours(group_id, requester_user_id, and_exists=True)
 
-        res = await self.store.get_group_role(group_id=group_id, role_id=role_id)
-        return res
+        return await self.store.get_group_role(group_id=group_id, role_id=role_id)
 
-    async def get_group_profile(self, group_id, requester_user_id):
+    async def get_group_profile(
+        self, group_id: str, requester_user_id: str
+    ) -> JsonDict:
         """Get the group profile as seen by requester_user_id"""
 
         await self.check_group_is_ours(group_id, requester_user_id)
@@ -219,7 +236,9 @@ class GroupsServerWorkerHandler:
         else:
             raise SynapseError(404, "Unknown group")
 
-    async def get_users_in_group(self, group_id, requester_user_id):
+    async def get_users_in_group(
+        self, group_id: str, requester_user_id: str
+    ) -> JsonDict:
         """Get the users in group as seen by requester_user_id.
 
         The ordering is arbitrary at the moment
@@ -268,7 +287,9 @@ class GroupsServerWorkerHandler:
 
         return {"chunk": chunk, "total_user_count_estimate": len(user_results)}
 
-    async def get_invited_users_in_group(self, group_id, requester_user_id):
+    async def get_invited_users_in_group(
+        self, group_id: str, requester_user_id: str
+    ) -> JsonDict:
         """Get the users that have been invited to a group as seen by requester_user_id.
 
         The ordering is arbitrary at the moment
@@ -298,7 +319,9 @@ class GroupsServerWorkerHandler:
 
         return {"chunk": user_profiles, "total_user_count_estimate": len(invited_users)}
 
-    async def get_rooms_in_group(self, group_id, requester_user_id):
+    async def get_rooms_in_group(
+        self, group_id: str, requester_user_id: str
+    ) -> JsonDict:
         """Get the rooms in group as seen by requester_user_id
 
         This returns rooms in order of decreasing number of joined users
@@ -336,15 +359,20 @@ class GroupsServerWorkerHandler:
 
 
 class GroupsServerHandler(GroupsServerWorkerHandler):
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
         super().__init__(hs)
 
         # Ensure attestations get renewed
         hs.get_groups_attestation_renewer()
 
     async def update_group_summary_room(
-        self, group_id, requester_user_id, room_id, category_id, content
-    ):
+        self,
+        group_id: str,
+        requester_user_id: str,
+        room_id: str,
+        category_id: str,
+        content: JsonDict,
+    ) -> JsonDict:
         """Add/update a room to the group summary"""
         await self.check_group_is_ours(
             group_id, requester_user_id, and_exists=True, and_is_admin=requester_user_id
@@ -367,8 +395,8 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
         return {}
 
     async def delete_group_summary_room(
-        self, group_id, requester_user_id, room_id, category_id
-    ):
+        self, group_id: str, requester_user_id: str, room_id: str, category_id: str
+    ) -> JsonDict:
         """Remove a room from the summary"""
         await self.check_group_is_ours(
             group_id, requester_user_id, and_exists=True, and_is_admin=requester_user_id
@@ -380,7 +408,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         return {}
 
-    async def set_group_join_policy(self, group_id, requester_user_id, content):
+    async def set_group_join_policy(
+        self, group_id: str, requester_user_id: str, content: JsonDict
+    ) -> JsonDict:
         """Sets the group join policy.
 
         Currently supported policies are:
@@ -400,8 +430,8 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
         return {}
 
     async def update_group_category(
-        self, group_id, requester_user_id, category_id, content
-    ):
+        self, group_id: str, requester_user_id: str, category_id: str, content: JsonDict
+    ) -> JsonDict:
         """Add/Update a group category"""
         await self.check_group_is_ours(
             group_id, requester_user_id, and_exists=True, and_is_admin=requester_user_id
@@ -419,7 +449,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         return {}
 
-    async def delete_group_category(self, group_id, requester_user_id, category_id):
+    async def delete_group_category(
+        self, group_id: str, requester_user_id: str, category_id: str
+    ) -> JsonDict:
         """Delete a group category"""
         await self.check_group_is_ours(
             group_id, requester_user_id, and_exists=True, and_is_admin=requester_user_id
@@ -431,7 +463,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         return {}
 
-    async def update_group_role(self, group_id, requester_user_id, role_id, content):
+    async def update_group_role(
+        self, group_id: str, requester_user_id: str, role_id: str, content: JsonDict
+    ) -> JsonDict:
         """Add/update a role in a group"""
         await self.check_group_is_ours(
             group_id, requester_user_id, and_exists=True, and_is_admin=requester_user_id
@@ -447,7 +481,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         return {}
 
-    async def delete_group_role(self, group_id, requester_user_id, role_id):
+    async def delete_group_role(
+        self, group_id: str, requester_user_id: str, role_id: str
+    ) -> JsonDict:
         """Remove role from group"""
         await self.check_group_is_ours(
             group_id, requester_user_id, and_exists=True, and_is_admin=requester_user_id
@@ -458,8 +494,13 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
         return {}
 
     async def update_group_summary_user(
-        self, group_id, requester_user_id, user_id, role_id, content
-    ):
+        self,
+        group_id: str,
+        requester_user_id: str,
+        user_id: str,
+        role_id: str,
+        content: JsonDict,
+    ) -> JsonDict:
         """Add/update a users entry in the group summary"""
         await self.check_group_is_ours(
             group_id, requester_user_id, and_exists=True, and_is_admin=requester_user_id
@@ -480,8 +521,8 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
         return {}
 
     async def delete_group_summary_user(
-        self, group_id, requester_user_id, user_id, role_id
-    ):
+        self, group_id: str, requester_user_id: str, user_id: str, role_id: str
+    ) -> JsonDict:
         """Remove a user from the group summary"""
         await self.check_group_is_ours(
             group_id, requester_user_id, and_exists=True, and_is_admin=requester_user_id
@@ -493,7 +534,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         return {}
 
-    async def update_group_profile(self, group_id, requester_user_id, content):
+    async def update_group_profile(
+        self, group_id: str, requester_user_id: str, content: JsonDict
+    ) -> None:
         """Update the group profile"""
         await self.check_group_is_ours(
             group_id, requester_user_id, and_exists=True, and_is_admin=requester_user_id
@@ -524,7 +567,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         await self.store.update_group_profile(group_id, profile)
 
-    async def add_room_to_group(self, group_id, requester_user_id, room_id, content):
+    async def add_room_to_group(
+        self, group_id: str, requester_user_id: str, room_id: str, content: JsonDict
+    ) -> JsonDict:
         """Add room to group"""
         RoomID.from_string(room_id)  # Ensure valid room id
 
@@ -539,8 +584,13 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
         return {}
 
     async def update_room_in_group(
-        self, group_id, requester_user_id, room_id, config_key, content
-    ):
+        self,
+        group_id: str,
+        requester_user_id: str,
+        room_id: str,
+        config_key: str,
+        content: JsonDict,
+    ) -> JsonDict:
         """Update room in group"""
         RoomID.from_string(room_id)  # Ensure valid room id
 
@@ -559,7 +609,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         return {}
 
-    async def remove_room_from_group(self, group_id, requester_user_id, room_id):
+    async def remove_room_from_group(
+        self, group_id: str, requester_user_id: str, room_id: str
+    ) -> JsonDict:
         """Remove room from group"""
         await self.check_group_is_ours(
             group_id, requester_user_id, and_exists=True, and_is_admin=requester_user_id
@@ -569,12 +621,16 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         return {}
 
-    async def invite_to_group(self, group_id, user_id, requester_user_id, content):
+    async def invite_to_group(
+        self, group_id: str, user_id: str, requester_user_id: str, content: JsonDict
+    ) -> JsonDict:
         """Invite user to group"""
 
         group = await self.check_group_is_ours(
             group_id, requester_user_id, and_exists=True, and_is_admin=requester_user_id
         )
+        if not group:
+            raise SynapseError(400, "Group does not exist", errcode=Codes.BAD_STATE)
 
         # TODO: Check if user knocked
 
@@ -597,6 +653,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         if self.hs.is_mine_id(user_id):
             groups_local = self.hs.get_groups_local_handler()
+            assert isinstance(
+                groups_local, GroupsLocalHandler
+            ), "Workers cannot invites users to groups."
             res = await groups_local.on_invite(group_id, user_id, content)
             local_attestation = None
         else:
@@ -632,6 +691,7 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
                 local_attestation=local_attestation,
                 remote_attestation=remote_attestation,
             )
+            return {"state": "join"}
         elif res["state"] == "invite":
             await self.store.add_group_invite(group_id, user_id)
             return {"state": "invite"}
@@ -640,13 +700,17 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
         else:
             raise SynapseError(502, "Unknown state returned by HS")
 
-    async def _add_user(self, group_id, user_id, content):
+    async def _add_user(
+        self, group_id: str, user_id: str, content: JsonDict
+    ) -> Optional[JsonDict]:
         """Add a user to a group based on a content dict.
 
         See accept_invite, join_group.
         """
         if not self.hs.is_mine_id(user_id):
-            local_attestation = self.attestations.create_attestation(group_id, user_id)
+            local_attestation = self.attestations.create_attestation(
+                group_id, user_id
+            )  # type: Optional[JsonDict]
 
             remote_attestation = content["attestation"]
 
@@ -670,7 +734,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         return local_attestation
 
-    async def accept_invite(self, group_id, requester_user_id, content):
+    async def accept_invite(
+        self, group_id: str, requester_user_id: str, content: JsonDict
+    ) -> JsonDict:
         """User tries to accept an invite to the group.
 
         This is different from them asking to join, and so should error if no
@@ -689,7 +755,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         return {"state": "join", "attestation": local_attestation}
 
-    async def join_group(self, group_id, requester_user_id, content):
+    async def join_group(
+        self, group_id: str, requester_user_id: str, content: JsonDict
+    ) -> JsonDict:
         """User tries to join the group.
 
         This will error if the group requires an invite/knock to join
@@ -698,6 +766,8 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
         group_info = await self.check_group_is_ours(
             group_id, requester_user_id, and_exists=True
         )
+        if not group_info:
+            raise SynapseError(404, "Group does not exist", errcode=Codes.NOT_FOUND)
         if group_info["join_policy"] != "open":
             raise SynapseError(403, "Group is not publicly joinable")
 
@@ -705,25 +775,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         return {"state": "join", "attestation": local_attestation}
 
-    async def knock(self, group_id, requester_user_id, content):
-        """A user requests becoming a member of the group"""
-        await self.check_group_is_ours(group_id, requester_user_id, and_exists=True)
-
-        raise NotImplementedError()
-
-    async def accept_knock(self, group_id, requester_user_id, content):
-        """Accept a users knock to the room.
-
-        Errors if the user hasn't knocked, rather than inviting them.
-        """
-
-        await self.check_group_is_ours(group_id, requester_user_id, and_exists=True)
-
-        raise NotImplementedError()
-
     async def remove_user_from_group(
-        self, group_id, user_id, requester_user_id, content
-    ):
+        self, group_id: str, user_id: str, requester_user_id: str, content: JsonDict
+    ) -> JsonDict:
         """Remove a user from the group; either a user is leaving or an admin
         kicked them.
         """
@@ -745,6 +799,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
         if is_kick:
             if self.hs.is_mine_id(user_id):
                 groups_local = self.hs.get_groups_local_handler()
+                assert isinstance(
+                    groups_local, GroupsLocalHandler
+                ), "Workers cannot remove users from groups."
                 await groups_local.user_removed_from_group(group_id, user_id, {})
             else:
                 await self.transport_client.remove_user_from_group_notification(
@@ -761,14 +818,15 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         return {}
 
-    async def create_group(self, group_id, requester_user_id, content):
-        group = await self.check_group_is_ours(group_id, requester_user_id)
-
+    async def create_group(
+        self, group_id: str, requester_user_id: str, content: JsonDict
+    ) -> JsonDict:
         logger.info("Attempting to create group with ID: %r", group_id)
 
         # parsing the id into a GroupID validates it.
         group_id_obj = GroupID.from_string(group_id)
 
+        group = await self.check_group_is_ours(group_id, requester_user_id)
         if group:
             raise SynapseError(400, "Group already exists")
 
@@ -813,7 +871,7 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
             local_attestation = self.attestations.create_attestation(
                 group_id, requester_user_id
-            )
+            )  # type: Optional[JsonDict]
         else:
             local_attestation = None
             remote_attestation = None
@@ -836,15 +894,14 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
 
         return {"group_id": group_id}
 
-    async def delete_group(self, group_id, requester_user_id):
+    async def delete_group(self, group_id: str, requester_user_id: str) -> None:
         """Deletes a group, kicking out all current members.
 
         Only group admins or server admins can call this request
 
         Args:
-            group_id (str)
-            request_user_id (str)
-
+            group_id: The group ID to delete.
+            requester_user_id: The user requesting to delete the group.
         """
 
         await self.check_group_is_ours(group_id, requester_user_id, and_exists=True)
@@ -867,6 +924,9 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
         async def _kick_user_from_group(user_id):
             if self.hs.is_mine_id(user_id):
                 groups_local = self.hs.get_groups_local_handler()
+                assert isinstance(
+                    groups_local, GroupsLocalHandler
+                ), "Workers cannot kick users from groups."
                 await groups_local.user_removed_from_group(group_id, user_id, {})
             else:
                 await self.transport_client.remove_user_from_group_notification(
@@ -898,7 +958,7 @@ class GroupsServerHandler(GroupsServerWorkerHandler):
         await self.store.delete_group(group_id)
 
 
-def _parse_join_policy_from_contents(content):
+def _parse_join_policy_from_contents(content: JsonDict) -> Optional[str]:
     """Given a content for a request, return the specified join policy or None"""
 
     join_policy_dict = content.get("m.join_policy")
@@ -908,7 +968,7 @@ def _parse_join_policy_from_contents(content):
         return None
 
 
-def _parse_join_policy_dict(join_policy_dict):
+def _parse_join_policy_dict(join_policy_dict: JsonDict) -> str:
     """Given a dict for the "m.join_policy" config return the join policy specified"""
     join_policy_type = join_policy_dict.get("type")
     if not join_policy_type:
@@ -919,7 +979,7 @@ def _parse_join_policy_dict(join_policy_dict):
     return join_policy_type
 
 
-def _parse_visibility_from_contents(content):
+def _parse_visibility_from_contents(content: JsonDict) -> bool:
     """Given a content for a request parse out whether the entity should be
     public or not
     """
@@ -933,7 +993,7 @@ def _parse_visibility_from_contents(content):
     return is_public
 
 
-def _parse_visibility_dict(visibility):
+def _parse_visibility_dict(visibility: JsonDict) -> bool:
     """Given a dict for the "m.visibility" config return if the entity should
     be public or not
     """

synapse/handlers/acme.py

@@ -73,7 +73,9 @@ class AcmeHandler:
                 "Listening for ACME requests on %s:%i", host, self.hs.config.acme_port
             )
             try:
-                self.reactor.listenTCP(self.hs.config.acme_port, srv, interface=host)
+                self.reactor.listenTCP(
+                    self.hs.config.acme_port, srv, backlog=50, interface=host
+                )
             except twisted.internet.error.CannotListenError as e:
                 check_bind_error(e, host, bind_addresses)
 

synapse/handlers/auth.py

@@ -36,7 +36,7 @@ import attr
 import bcrypt
 import pymacaroons
 
-from twisted.web.http import Request
+from twisted.web.server import Request
 
 from synapse.api.constants import LoginType
 from synapse.api.errors import (
@@ -65,6 +65,7 @@ from synapse.storage.roommember import ProfileInfo
 from synapse.types import JsonDict, Requester, UserID
 from synapse.util import stringutils as stringutils
 from synapse.util.async_helpers import maybe_awaitable
+from synapse.util.macaroons import get_value_from_macaroon, satisfy_expiry
 from synapse.util.msisdn import phone_number_to_msisdn
 from synapse.util.threepids import canonicalise_email
 
@@ -170,6 +171,16 @@ class SsoLoginExtraAttributes:
     extra_attributes = attr.ib(type=JsonDict)
 
 
+@attr.s(slots=True, frozen=True)
+class LoginTokenAttributes:
+    """Data we store in a short-term login token"""
+
+    user_id = attr.ib(type=str)
+
+    # the SSO Identity Provider that the user authenticated with, to get this token
+    auth_provider_id = attr.ib(type=str)
+
+
 class AuthHandler(BaseHandler):
     SESSION_EXPIRE_MS = 48 * 60 * 60 * 1000
 
@@ -326,7 +337,8 @@ class AuthHandler(BaseHandler):
                 user is too high to proceed
 
         """
-
+        if not requester.access_token_id:
+            raise ValueError("Cannot validate a user without an access token")
         if self._ui_auth_session_timeout:
             last_validated = await self.store.get_access_token_last_validated(
                 requester.access_token_id
@@ -481,7 +493,7 @@ class AuthHandler(BaseHandler):
             sid = authdict["session"]
 
         # Convert the URI and method to strings.
-        uri = request.uri.decode("utf-8")
+        uri = request.uri.decode("utf-8")  # type: ignore
         method = request.method.decode("utf-8")
 
         # If there's no session ID, create a new session.
@@ -1164,18 +1176,16 @@ class AuthHandler(BaseHandler):
             return None
         return user_id
 
-    async def validate_short_term_login_token_and_get_user_id(self, login_token: str):
-        auth_api = self.hs.get_auth()
-        user_id = None
+    async def validate_short_term_login_token(
+        self, login_token: str
+    ) -> LoginTokenAttributes:
         try:
-            macaroon = pymacaroons.Macaroon.deserialize(login_token)
-            user_id = auth_api.get_user_id_from_macaroon(macaroon)
-            auth_api.validate_macaroon(macaroon, "login", user_id)
+            res = self.macaroon_gen.verify_short_term_login_token(login_token)
         except Exception:
             raise AuthError(403, "Invalid token", errcode=Codes.FORBIDDEN)
 
-        await self.auth.check_auth_blocking(user_id)
-        return user_id
+        await self.auth.check_auth_blocking(res.user_id)
+        return res
 
     async def delete_access_token(self, access_token: str):
         """Invalidate a single access token
@@ -1204,7 +1214,7 @@ class AuthHandler(BaseHandler):
     async def delete_access_tokens_for_user(
         self,
         user_id: str,
-        except_token_id: Optional[str] = None,
+        except_token_id: Optional[int] = None,
         device_id: Optional[str] = None,
     ):
         """Invalidate access tokens belonging to a user
@@ -1397,6 +1407,7 @@ class AuthHandler(BaseHandler):
     async def complete_sso_login(
         self,
         registered_user_id: str,
+        auth_provider_id: str,
         request: Request,
         client_redirect_url: str,
         extra_attributes: Optional[JsonDict] = None,
@@ -1406,6 +1417,9 @@ class AuthHandler(BaseHandler):
 
         Args:
             registered_user_id: The registered user ID to complete SSO login for.
+            auth_provider_id: The id of the SSO Identity provider that was used for
+                login. This will be stored in the login token for future tracking in
+                prometheus metrics.
             request: The request to complete.
             client_redirect_url: The URL to which to redirect the user at the end of the
                 process.
@@ -1427,6 +1441,7 @@ class AuthHandler(BaseHandler):
 
         self._complete_sso_login(
             registered_user_id,
+            auth_provider_id,
             request,
             client_redirect_url,
             extra_attributes,
@@ -1437,6 +1452,7 @@ class AuthHandler(BaseHandler):
     def _complete_sso_login(
         self,
         registered_user_id: str,
+        auth_provider_id: str,
         request: Request,
         client_redirect_url: str,
         extra_attributes: Optional[JsonDict] = None,
@@ -1463,7 +1479,7 @@ class AuthHandler(BaseHandler):
 
         # Create a login token
         login_token = self.macaroon_gen.generate_short_term_login_token(
-            registered_user_id
+            registered_user_id, auth_provider_id=auth_provider_id
         )
 
         # Append the login token to the original redirect URL (i.e. with its query
@@ -1569,15 +1585,48 @@ class MacaroonGenerator:
         return macaroon.serialize()
 
     def generate_short_term_login_token(
-        self, user_id: str, duration_in_ms: int = (2 * 60 * 1000)
+        self,
+        user_id: str,
+        auth_provider_id: str,
+        duration_in_ms: int = (2 * 60 * 1000),
     ) -> str:
         macaroon = self._generate_base_macaroon(user_id)
         macaroon.add_first_party_caveat("type = login")
         now = self.hs.get_clock().time_msec()
         expiry = now + duration_in_ms
         macaroon.add_first_party_caveat("time < %d" % (expiry,))
+        macaroon.add_first_party_caveat("auth_provider_id = %s" % (auth_provider_id,))
         return macaroon.serialize()
 
+    def verify_short_term_login_token(self, token: str) -> LoginTokenAttributes:
+        """Verify a short-term-login macaroon
+
+        Checks that the given token is a valid, unexpired short-term-login token
+        minted by this server.
+
+        Args:
+            token: the login token to verify
+
+        Returns:
+            the user_id that this token is valid for
+
+        Raises:
+            MacaroonVerificationFailedException if the verification failed
+        """
+        macaroon = pymacaroons.Macaroon.deserialize(token)
+        user_id = get_value_from_macaroon(macaroon, "user_id")
+        auth_provider_id = get_value_from_macaroon(macaroon, "auth_provider_id")
+
+        v = pymacaroons.Verifier()
+        v.satisfy_exact("gen = 1")
+        v.satisfy_exact("type = login")
+        v.satisfy_general(lambda c: c.startswith("user_id = "))
+        v.satisfy_general(lambda c: c.startswith("auth_provider_id = "))
+        satisfy_expiry(v, self.hs.get_clock().time_msec)
+        v.verify(macaroon, self.hs.config.key.macaroon_secret_key)
+
+        return LoginTokenAttributes(user_id=user_id, auth_provider_id=auth_provider_id)
+
     def generate_delete_pusher_token(self, user_id: str) -> str:
         macaroon = self._generate_base_macaroon(user_id)
         macaroon.add_first_party_caveat("type = delete_pusher")

synapse/handlers/cas_handler.py

@@ -83,6 +83,7 @@ class CasHandler:
         # the SsoIdentityProvider protocol type.
         self.idp_icon = None
         self.idp_brand = None
+        self.unstable_idp_brand = None
 
         self._sso_handler = hs.get_sso_handler()
 

synapse/handlers/deactivate_account.py

@@ -120,6 +120,11 @@ class DeactivateAccountHandler(BaseHandler):
 
         await self.store.user_set_password_hash(user_id, None)
 
+        # Most of the pushers will have been deleted when we logged out the
+        # associated devices above, but we still need to delete pushers not
+        # associated with devices, e.g. email pushers.
+        await self.store.delete_all_pushers_for_user(user_id)
+
         # Add the user to a table of users pending deactivation (ie.
         # removal from all the rooms they're a member of)
         await self.store.add_user_pending_deactivation(user_id)

synapse/handlers/devicemessage.py

@@ -16,7 +16,9 @@
 import logging
 from typing import TYPE_CHECKING, Any, Dict
 
+from synapse.api.constants import EduTypes
 from synapse.api.errors import SynapseError
+from synapse.api.ratelimiting import Ratelimiter
 from synapse.logging.context import run_in_background
 from synapse.logging.opentracing import (
     get_active_span_text_map,
@@ -25,7 +27,7 @@ from synapse.logging.opentracing import (
     start_active_span,
 )
 from synapse.replication.http.devices import ReplicationUserDevicesResyncRestServlet
-from synapse.types import JsonDict, UserID, get_domain_from_id
+from synapse.types import JsonDict, Requester, UserID, get_domain_from_id
 from synapse.util import json_encoder
 from synapse.util.stringutils import random_string
 
@@ -78,6 +80,12 @@ class DeviceMessageHandler:
                 ReplicationUserDevicesResyncRestServlet.make_client(hs)
             )
 
+        self._ratelimiter = Ratelimiter(
+            clock=hs.get_clock(),
+            rate_hz=hs.config.rc_key_requests.per_second,
+            burst_count=hs.config.rc_key_requests.burst_count,
+        )
+
     async def on_direct_to_device_edu(self, origin: str, content: JsonDict) -> None:
         local_messages = {}
         sender_user_id = content["sender"]
@@ -168,15 +176,27 @@ class DeviceMessageHandler:
 
     async def send_device_message(
         self,
-        sender_user_id: str,
+        requester: Requester,
         message_type: str,
         messages: Dict[str, Dict[str, JsonDict]],
     ) -> None:
+        sender_user_id = requester.user.to_string()
+
         set_tag("number_of_messages", len(messages))
         set_tag("sender", sender_user_id)
         local_messages = {}
         remote_messages = {}  # type: Dict[str, Dict[str, Dict[str, JsonDict]]]
         for user_id, by_device in messages.items():
+            # Ratelimit local cross-user key requests by the sending device.
+            if (
+                message_type == EduTypes.RoomKeyRequest
+                and user_id != sender_user_id
+                and self._ratelimiter.can_do_action(
+                    (sender_user_id, requester.device_id)
+                )
+            ):
+                continue
+
             # we use UserID.from_string to catch invalid user ids
             if self.is_mine(UserID.from_string(user_id)):
                 messages_by_device = {

synapse/handlers/events.py

@@ -17,7 +17,7 @@ import logging
 import random
 from typing import TYPE_CHECKING, Iterable, List, Optional
 
-from synapse.api.constants import EventTypes, Membership
+from synapse.api.constants import EduTypes, EventTypes, Membership
 from synapse.api.errors import AuthError, SynapseError
 from synapse.events import EventBase
 from synapse.handlers.presence import format_user_presence_state
@@ -113,7 +113,7 @@ class EventStreamHandler(BaseHandler):
                     states = await presence_handler.get_states(users)
                     to_add.extend(
                         {
-                            "type": EventTypes.Presence,
+                            "type": EduTypes.Presence,
                             "content": format_user_presence_state(state, time_now),
                         }
                         for state in states

synapse/handlers/federation.py

@@ -201,7 +201,7 @@ class FederationHandler(BaseHandler):
             or pdu.internal_metadata.is_outlier()
         )
         if already_seen:
-            logger.debug("[%s %s]: Already seen pdu", room_id, event_id)
+            logger.debug("Already seen pdu")
             return
 
         # do some initial sanity-checking of the event. In particular, make
@@ -210,18 +210,14 @@ class FederationHandler(BaseHandler):
         try:
             self._sanity_check_event(pdu)
         except SynapseError as err:
-            logger.warning(
-                "[%s %s] Received event failed sanity checks", room_id, event_id
-            )
+            logger.warning("Received event failed sanity checks")
             raise FederationError("ERROR", err.code, err.msg, affected=pdu.event_id)
 
         # If we are currently in the process of joining this room, then we
         # queue up events for later processing.
         if room_id in self.room_queues:
             logger.info(
-                "[%s %s] Queuing PDU from %s for now: join in progress",
-                room_id,
-                event_id,
+                "Queuing PDU from %s for now: join in progress",
                 origin,
             )
             self.room_queues[room_id].append((pdu, origin))
@@ -236,9 +232,7 @@ class FederationHandler(BaseHandler):
         is_in_room = await self.auth.check_host_in_room(room_id, self.server_name)
         if not is_in_room:
             logger.info(
-                "[%s %s] Ignoring PDU from %s as we're not in the room",
-                room_id,
-                event_id,
+                "Ignoring PDU from %s as we're not in the room",
                 origin,
             )
             return None
@@ -250,7 +244,7 @@ class FederationHandler(BaseHandler):
             # We only backfill backwards to the min depth.
             min_depth = await self.get_min_depth_for_context(pdu.room_id)
 
-            logger.debug("[%s %s] min_depth: %d", room_id, event_id, min_depth)
+            logger.debug("min_depth: %d", min_depth)
 
             prevs = set(pdu.prev_event_ids())
             seen = await self.store.have_events_in_timeline(prevs)
@@ -267,17 +261,13 @@ class FederationHandler(BaseHandler):
                     # If we're missing stuff, ensure we only fetch stuff one
                     # at a time.
                     logger.info(
-                        "[%s %s] Acquiring room lock to fetch %d missing prev_events: %s",
-                        room_id,
-                        event_id,
+                        "Acquiring room lock to fetch %d missing prev_events: %s",
                         len(missing_prevs),
                         shortstr(missing_prevs),
                     )
                     with (await self._room_pdu_linearizer.queue(pdu.room_id)):
                         logger.info(
-                            "[%s %s] Acquired room lock to fetch %d missing prev_events",
-                            room_id,
-                            event_id,
+                            "Acquired room lock to fetch %d missing prev_events",
                             len(missing_prevs),
                         )
 
@@ -297,9 +287,7 @@ class FederationHandler(BaseHandler):
 
                         if not prevs - seen:
                             logger.info(
-                                "[%s %s] Found all missing prev_events",
-                                room_id,
-                                event_id,
+                                "Found all missing prev_events",
                             )
 
             if prevs - seen:
@@ -329,9 +317,7 @@ class FederationHandler(BaseHandler):
 
                 if sent_to_us_directly:
                     logger.warning(
-                        "[%s %s] Rejecting: failed to fetch %d prev events: %s",
-                        room_id,
-                        event_id,
+                        "Rejecting: failed to fetch %d prev events: %s",
                         len(prevs - seen),
                         shortstr(prevs - seen),
                     )
@@ -367,17 +353,16 @@ class FederationHandler(BaseHandler):
                     # Ask the remote server for the states we don't
                     # know about
                     for p in prevs - seen:
-                        logger.info(
-                            "Requesting state at missing prev_event %s",
-                            event_id,
-                        )
+                        logger.info("Requesting state after missing prev_event %s", p)
 
                         with nested_logging_context(p):
                             # note that if any of the missing prevs share missing state or
                             # auth events, the requests to fetch those events are deduped
                             # by the get_pdu_cache in federation_client.
-                            (remote_state, _,) = await self._get_state_for_room(
-                                origin, room_id, p, include_event_in_state=True
+                            remote_state = (
+                                await self._get_state_after_missing_prev_event(
+                                    origin, room_id, p
+                                )
                             )
 
                             remote_state_map = {
@@ -414,10 +399,7 @@ class FederationHandler(BaseHandler):
                     state = [event_map[e] for e in state_map.values()]
                 except Exception:
                     logger.warning(
-                        "[%s %s] Error attempting to resolve state at missing "
-                        "prev_events",
-                        room_id,
-                        event_id,
+                        "Error attempting to resolve state at missing " "prev_events",
                         exc_info=True,
                     )
                     raise FederationError(
@@ -454,9 +436,7 @@ class FederationHandler(BaseHandler):
         latest |= seen
 
         logger.info(
-            "[%s %s]: Requesting missing events between %s and %s",
-            room_id,
-            event_id,
+            "Requesting missing events between %s and %s",
             shortstr(latest),
             event_id,
         )
@@ -523,15 +503,11 @@ class FederationHandler(BaseHandler):
             # We failed to get the missing events, but since we need to handle
             # the case of `get_missing_events` not returning the necessary
             # events anyway, it is safe to simply log the error and continue.
-            logger.warning(
-                "[%s %s]: Failed to get prev_events: %s", room_id, event_id, e
-            )
+            logger.warning("Failed to get prev_events: %s", e)
             return
 
         logger.info(
-            "[%s %s]: Got %d prev_events: %s",
-            room_id,
-            event_id,
+            "Got %d prev_events: %s",
             len(missing_events),
             shortstr(missing_events),
         )
@@ -542,9 +518,7 @@ class FederationHandler(BaseHandler):
 
         for ev in missing_events:
             logger.info(
-                "[%s %s] Handling received prev_event %s",
-                room_id,
-                event_id,
+                "Handling received prev_event %s",
                 ev.event_id,
             )
             with nested_logging_context(ev.event_id):
@@ -553,9 +527,7 @@ class FederationHandler(BaseHandler):
                 except FederationError as e:
                     if e.code == 403:
                         logger.warning(
-                            "[%s %s] Received prev_event %s failed history check.",
-                            room_id,
-                            event_id,
+                            "Received prev_event %s failed history check.",
                             ev.event_id,
                         )
                     else:
@@ -566,7 +538,6 @@ class FederationHandler(BaseHandler):
         destination: str,
         room_id: str,
         event_id: str,
-        include_event_in_state: bool = False,
     ) -> Tuple[List[EventBase], List[EventBase]]:
         """Requests all of the room state at a given event from a remote homeserver.
 
@@ -574,11 +545,9 @@ class FederationHandler(BaseHandler):
             destination: The remote homeserver to query for the state.
             room_id: The id of the room we're interested in.
             event_id: The id of the event we want the state at.
-            include_event_in_state: if true, the event itself will be included in the
-                returned state event list.
 
         Returns:
-            A list of events in the state, possibly including the event itself, and
+            A list of events in the state, not including the event itself, and
             a list of events in the auth chain for the given event.
         """
         (
@@ -590,9 +559,6 @@ class FederationHandler(BaseHandler):
 
         desired_events = set(state_event_ids + auth_event_ids)
 
-        if include_event_in_state:
-            desired_events.add(event_id)
-
         event_map = await self._get_events_from_store_or_dest(
             destination, room_id, desired_events
         )
@@ -609,13 +575,6 @@ class FederationHandler(BaseHandler):
             event_map[e_id] for e_id in state_event_ids if e_id in event_map
         ]
 
-        if include_event_in_state:
-            remote_event = event_map.get(event_id)
-            if not remote_event:
-                raise Exception("Unable to get missing prev_event %s" % (event_id,))
-            if remote_event.is_state() and remote_event.rejected_reason is None:
-                remote_state.append(remote_event)
-
         auth_chain = [event_map[e_id] for e_id in auth_event_ids if e_id in event_map]
         auth_chain.sort(key=lambda e: e.depth)
 
@@ -689,6 +648,131 @@ class FederationHandler(BaseHandler):
 
         return fetched_events
 
+    async def _get_state_after_missing_prev_event(
+        self,
+        destination: str,
+        room_id: str,
+        event_id: str,
+    ) -> List[EventBase]:
+        """Requests all of the room state at a given event from a remote homeserver.
+
+        Args:
+            destination: The remote homeserver to query for the state.
+            room_id: The id of the room we're interested in.
+            event_id: The id of the event we want the state at.
+
+        Returns:
+            A list of events in the state, including the event itself
+        """
+        # TODO: This function is basically the same as _get_state_for_room. Can
+        #   we make backfill() use it, rather than having two code paths? I think the
+        #   only difference is that backfill() persists the prev events separately.
+
+        (
+            state_event_ids,
+            auth_event_ids,
+        ) = await self.federation_client.get_room_state_ids(
+            destination, room_id, event_id=event_id
+        )
+
+        logger.debug(
+            "state_ids returned %i state events, %i auth events",
+            len(state_event_ids),
+            len(auth_event_ids),
+        )
+
+        # start by just trying to fetch the events from the store
+        desired_events = set(state_event_ids)
+        desired_events.add(event_id)
+        logger.debug("Fetching %i events from cache/store", len(desired_events))
+        fetched_events = await self.store.get_events(
+            desired_events, allow_rejected=True
+        )
+
+        missing_desired_events = desired_events - fetched_events.keys()
+        logger.debug(
+            "We are missing %i events (got %i)",
+            len(missing_desired_events),
+            len(fetched_events),
+        )
+
+        # We probably won't need most of the auth events, so let's just check which
+        # we have for now, rather than thrashing the event cache with them all
+        # unnecessarily.
+
+        # TODO: we probably won't actually need all of the auth events, since we
+        #   already have a bunch of the state events. It would be nice if the
+        #   federation api gave us a way of finding out which we actually need.
+
+        missing_auth_events = set(auth_event_ids) - fetched_events.keys()
+        missing_auth_events.difference_update(
+            await self.store.have_seen_events(missing_auth_events)
+        )
+        logger.debug("We are also missing %i auth events", len(missing_auth_events))
+
+        missing_events = missing_desired_events | missing_auth_events
+        logger.debug("Fetching %i events from remote", len(missing_events))
+        await self._get_events_and_persist(
+            destination=destination, room_id=room_id, events=missing_events
+        )
+
+        # we need to make sure we re-load from the database to get the rejected
+        # state correct.
+        fetched_events.update(
+            (await self.store.get_events(missing_desired_events, allow_rejected=True))
+        )
+
+        # check for events which were in the wrong room.
+        #
+        # this can happen if a remote server claims that the state or
+        # auth_events at an event in room A are actually events in room B
+
+        bad_events = [
+            (event_id, event.room_id)
+            for event_id, event in fetched_events.items()
+            if event.room_id != room_id
+        ]
+
+        for bad_event_id, bad_room_id in bad_events:
+            # This is a bogus situation, but since we may only discover it a long time
+            # after it happened, we try our best to carry on, by just omitting the
+            # bad events from the returned state set.
+            logger.warning(
+                "Remote server %s claims event %s in room %s is an auth/state "
+                "event in room %s",
+                destination,
+                bad_event_id,
+                bad_room_id,
+                room_id,
+            )
+
+            del fetched_events[bad_event_id]
+
+        # if we couldn't get the prev event in question, that's a problem.
+        remote_event = fetched_events.get(event_id)
+        if not remote_event:
+            raise Exception("Unable to get missing prev_event %s" % (event_id,))
+
+        # missing state at that event is a warning, not a blocker
+        # XXX: this doesn't sound right? it means that we'll end up with incomplete
+        #   state.
+        failed_to_fetch = desired_events - fetched_events.keys()
+        if failed_to_fetch:
+            logger.warning(
+                "Failed to fetch missing state events for %s %s",
+                event_id,
+                failed_to_fetch,
+            )
+
+        remote_state = [
+            fetched_events[e_id] for e_id in state_event_ids if e_id in fetched_events
+        ]
+
+        if remote_event.is_state() and remote_event.rejected_reason is None:
+            remote_state.append(remote_event)
+
+        return remote_state
+
     async def _process_received_pdu(
         self,
         origin: str,
@@ -707,10 +791,7 @@ class FederationHandler(BaseHandler):
                 (ie, we are missing one or more prev_events), the resolved state at the
                 event
         """
-        room_id = event.room_id
-        event_id = event.event_id
-
-        logger.debug("[%s %s] Processing event: %s", room_id, event_id, event)
+        logger.debug("Processing event: %s", event)
 
         try:
             await self._handle_new_event(origin, event, state=state)
@@ -871,7 +952,6 @@ class FederationHandler(BaseHandler):
                 destination=dest,
                 room_id=room_id,
                 event_id=e_id,
-                include_event_in_state=False,
             )
             auth_events.update({a.event_id: a for a in auth})
             auth_events.update({s.event_id: s for s in state})
@@ -1317,7 +1397,7 @@ class FederationHandler(BaseHandler):
     async def on_event_auth(self, event_id: str) -> List[EventBase]:
         event = await self.store.get_event(event_id)
         auth = await self.store.get_auth_chain(
-            list(event.auth_event_ids()), include_given=True
+            event.room_id, list(event.auth_event_ids()), include_given=True
         )
         return list(auth)
 
@@ -1580,7 +1660,7 @@ class FederationHandler(BaseHandler):
         prev_state_ids = await context.get_prev_state_ids()
 
         state_ids = list(prev_state_ids.values())
-        auth_chain = await self.store.get_auth_chain(state_ids)
+        auth_chain = await self.store.get_auth_chain(event.room_id, state_ids)
 
         state = await self.store.get_events(list(prev_state_ids.values()))
 
@@ -2219,7 +2299,7 @@ class FederationHandler(BaseHandler):
 
         # Now get the current auth_chain for the event.
         local_auth_chain = await self.store.get_auth_chain(
-            list(event.auth_event_ids()), include_given=True
+            room_id, list(event.auth_event_ids()), include_given=True
         )
 
         # TODO: Check if we would now reject event_id. If so we need to tell

synapse/handlers/initial_sync.py

@@ -18,7 +18,7 @@ from typing import TYPE_CHECKING, Optional, Tuple
 
 from twisted.internet import defer
 
-from synapse.api.constants import EventTypes, Membership
+from synapse.api.constants import EduTypes, EventTypes, Membership
 from synapse.api.errors import SynapseError
 from synapse.events.validator import EventValidator
 from synapse.handlers.presence import format_user_presence_state
@@ -48,7 +48,7 @@ class InitialSyncHandler(BaseHandler):
         self.clock = hs.get_clock()
         self.validator = EventValidator()
         self.snapshot_cache = ResponseCache(
-            hs, "initial_sync_cache"
+            hs.get_clock(), "initial_sync_cache"
         )  # type: ResponseCache[Tuple[str, Optional[StreamToken], Optional[StreamToken], str, Optional[int], bool, bool]]
         self._event_serializer = hs.get_event_client_serializer()
         self.storage = hs.get_storage()
@@ -412,7 +412,7 @@ class InitialSyncHandler(BaseHandler):
 
             return [
                 {
-                    "type": EventTypes.Presence,
+                    "type": EduTypes.Presence,
                     "content": format_user_presence_state(s, time_now),
                 }
                 for s in states