Add further logging for #9533

If in doubt, print it out. Thanks Erik for advising.

.buildkite/.env

@@ -1,13 +0,0 @@
-CI
-BUILDKITE
-BUILDKITE_BUILD_NUMBER
-BUILDKITE_BRANCH
-BUILDKITE_BUILD_NUMBER
-BUILDKITE_JOB_ID
-BUILDKITE_BUILD_URL
-BUILDKITE_PROJECT_SLUG
-BUILDKITE_COMMIT
-BUILDKITE_PULL_REQUEST
-BUILDKITE_TAG
-CODECOV_TOKEN
-TRIAL_FLAGS

.buildkite/merge_base_branch.sh

@@ -1,35 +0,0 @@
-#!/usr/bin/env bash
-
-set -e
-
-if [[ "$BUILDKITE_BRANCH" =~ ^(develop|master|dinsic|shhs|release-.*)$ ]]; then
-    echo "Not merging forward, as this is a release branch"
-    exit 0
-fi
-
-if [[ -z $BUILDKITE_PULL_REQUEST_BASE_BRANCH ]]; then
-    echo "Not a pull request, or hasn't had a PR opened yet..."
-
-    # It probably hasn't had a PR opened yet. Since all PRs land on develop, we
-    # can probably assume it's based on it and will be merged into it.
-    GITBASE="develop"
-else
-    # Get the reference, using the GitHub API
-    GITBASE=$BUILDKITE_PULL_REQUEST_BASE_BRANCH
-fi
-
-echo "--- merge_base_branch $GITBASE"
-
-# Show what we are before
-git --no-pager show -s
-
-# Set up username so it can do a merge
-git config --global user.email bot@matrix.org
-git config --global user.name "A robot"
-
-# Fetch and merge. If it doesn't work, it will raise due to set -e.
-git fetch -u origin $GITBASE
-git merge --no-edit --no-commit origin/$GITBASE
-
-# Show what we are after.
-git --no-pager show -s

.ci/postgres-config.yaml

@@ -3,7 +3,7 @@
 # CI's Docker setup at the point where this file is considered.
 server_name: "localhost:8800"
 
-signing_key_path: ".buildkite/test.signing.key"
+signing_key_path: ".ci/test.signing.key"
 
 report_stats: false
 
@@ -11,7 +11,7 @@ database:
   name: "psycopg2"
   args:
     user: postgres
-    host: postgres
+    host: localhost
     password: postgres
     database: synapse
 

.ci/scripts/postgres_exec.py

@@ -23,7 +23,7 @@ import psycopg2
 # We use "postgres" as a database because it's bound to exist and the "synapse" one
 # doesn't exist yet.
 db_conn = psycopg2.connect(
-    user="postgres", host="postgres", password="postgres", dbname="postgres"
+    user="postgres", host="localhost", password="postgres", dbname="postgres"
 )
 db_conn.autocommit = True
 cur = db_conn.cursor()

.ci/scripts/test_old_deps.sh

@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 
-# this script is run by buildkite in a plain `bionic` container; it installs the
+# this script is run by GitHub Actions in a plain `bionic` container; it installs the
 # minimal requirements for tox and hands over to the py3-old tox environment.
 
 set -ex

.ci/scripts/test_synapse_port_db.sh

@@ -20,22 +20,22 @@ pip install -e .
 echo "--- Generate the signing key"
 
 # Generate the server's signing key.
-python -m synapse.app.homeserver --generate-keys -c .buildkite/sqlite-config.yaml
+python -m synapse.app.homeserver --generate-keys -c .ci/sqlite-config.yaml
 
 echo "--- Prepare test database"
 
 # Make sure the SQLite3 database is using the latest schema and has no pending background update.
-scripts-dev/update_database --database-config .buildkite/sqlite-config.yaml
+scripts-dev/update_database --database-config .ci/sqlite-config.yaml
 
 # Create the PostgreSQL database.
-./.buildkite/scripts/postgres_exec.py "CREATE DATABASE synapse"
+.ci/scripts/postgres_exec.py "CREATE DATABASE synapse"
 
 echo "+++ Run synapse_port_db against test database"
-coverage run scripts/synapse_port_db --sqlite-database .buildkite/test_db.db --postgres-config .buildkite/postgres-config.yaml
+coverage run scripts/synapse_port_db --sqlite-database .ci/test_db.db --postgres-config .ci/postgres-config.yaml
 
 # We should be able to run twice against the same database.
 echo "+++ Run synapse_port_db a second time"
-coverage run scripts/synapse_port_db --sqlite-database .buildkite/test_db.db --postgres-config .buildkite/postgres-config.yaml
+coverage run scripts/synapse_port_db --sqlite-database .ci/test_db.db --postgres-config .ci/postgres-config.yaml
 
 #####
 
@@ -44,14 +44,14 @@ coverage run scripts/synapse_port_db --sqlite-database .buildkite/test_db.db --p
 echo "--- Prepare empty SQLite database"
 
 # we do this by deleting the sqlite db, and then doing the same again.
-rm .buildkite/test_db.db
+rm .ci/test_db.db
 
-scripts-dev/update_database --database-config .buildkite/sqlite-config.yaml
+scripts-dev/update_database --database-config .ci/sqlite-config.yaml
 
 # re-create the PostgreSQL database.
-./.buildkite/scripts/postgres_exec.py \
+.ci/scripts/postgres_exec.py \
   "DROP DATABASE synapse" \
   "CREATE DATABASE synapse"
 
 echo "+++ Run synapse_port_db against empty database"
-coverage run scripts/synapse_port_db --sqlite-database .buildkite/test_db.db --postgres-config .buildkite/postgres-config.yaml
+coverage run scripts/synapse_port_db --sqlite-database .ci/test_db.db --postgres-config .ci/postgres-config.yaml

.ci/sqlite-config.yaml

@@ -3,14 +3,14 @@
 # schema and run background updates on it.
 server_name: "localhost:8800"
 
-signing_key_path: ".buildkite/test.signing.key"
+signing_key_path: ".ci/test.signing.key"
 
 report_stats: false
 
 database:
   name: "sqlite3"
   args:
-    database: ".buildkite/test_db.db"
+    database: ".ci/test_db.db"
 
 # Suppress the key server warning.
 trusted_key_servers: []

.github/workflows/tests.yml

@@ -38,20 +38,15 @@ jobs:
     if: ${{ github.base_ref == 'develop'  || contains(github.base_ref, 'release-') }}
     runs-on: ubuntu-latest
     steps:
-      # Note: This and the script can be simplified once we drop Buildkite. See:
-      #   https://github.com/actions/checkout/issues/266#issuecomment-638346893
-      #   https://github.com/actions/checkout/issues/416
       - uses: actions/checkout@v2
         with:
           ref: ${{ github.event.pull_request.head.sha }}
           fetch-depth: 0
       - uses: actions/setup-python@v2
       - run: pip install tox
-      - name: Patch Buildkite-specific test script
-        run: |
-          sed -i -e 's/\$BUILDKITE_PULL_REQUEST/${{ github.event.number }}/' \
-            scripts-dev/check-newsfragment
       - run: scripts-dev/check-newsfragment
+        env:
+          PULL_REQUEST_NUMBER: ${{ github.event.number }}
 
   lint-sdist:
     runs-on: ubuntu-latest
@@ -144,7 +139,7 @@ jobs:
         uses: docker://ubuntu:bionic # For old python and sqlite
         with:
           workdir: /github/workspace
-          entrypoint: .buildkite/scripts/test_old_deps.sh
+          entrypoint: .ci/scripts/test_old_deps.sh
         env:
           TRIAL_FLAGS: "--jobs=2"
       - name: Dump logs
@@ -197,12 +192,12 @@ jobs:
       volumes:
         - ${{ github.workspace }}:/src
       env:
-        BUILDKITE_BRANCH: ${{ github.head_ref }}
         POSTGRES: ${{ matrix.postgres && 1}}
         MULTI_POSTGRES: ${{ (matrix.postgres == 'multi-postgres') && 1}}
         WORKERS: ${{ matrix.workers && 1 }}
         REDIS: ${{ matrix.redis && 1 }}
         BLACKLIST: ${{ matrix.workers && 'synapse-blacklist-with-workers' }}
+        TOP: ${{ github.workspace }}
 
     strategy:
       fail-fast: false
@@ -232,7 +227,7 @@ jobs:
     steps:
       - uses: actions/checkout@v2
       - name: Prepare test blacklist
-        run: cat sytest-blacklist .buildkite/worker-blacklist > synapse-blacklist-with-workers
+        run: cat sytest-blacklist .ci/worker-blacklist > synapse-blacklist-with-workers
       - name: Run SyTest
         run: /bootstrap.sh synapse
         working-directory: /src
@@ -252,6 +247,8 @@ jobs:
     if: ${{ !failure() && !cancelled() }} # Allow previous steps to be skipped, but not fail
     needs: linting-done
     runs-on: ubuntu-latest
+    env:
+      TOP: ${{ github.workspace }}
     strategy:
       matrix:
         include:
@@ -281,13 +278,7 @@ jobs:
       - uses: actions/setup-python@v2
         with:
           python-version: ${{ matrix.python-version }}
-      - name: Patch Buildkite-specific test scripts
-        run: |
-          sed -i -e 's/host="postgres"/host="localhost"/' .buildkite/scripts/postgres_exec.py
-          sed -i -e 's/host: postgres/host: localhost/' .buildkite/postgres-config.yaml
-          sed -i -e 's|/src/||' .buildkite/{sqlite,postgres}-config.yaml
-          sed -i -e 's/\$TOP/\$GITHUB_WORKSPACE/' .coveragerc
-      - run: .buildkite/scripts/test_synapse_port_db.sh
+      - run: .ci/scripts/test_synapse_port_db.sh
 
   complement:
     if: ${{ !failure() && !cancelled() }}

CONTRIBUTING.md

@@ -13,8 +13,9 @@ This document aims to get you started with contributing to this repo!
 - [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)
+  * [Run the unit tests.](#run-the-unit-tests-twisted-trial)
+  * [Run the integration tests (SyTest).](#run-the-integration-tests-sytest)
+  * [Run the integration tests (Complement).](#run-the-integration-tests-complement)
 - [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)
@@ -197,7 +198,7 @@ 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
+$ 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:buster
 ```
 
 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).

MANIFEST.in

@@ -44,9 +44,9 @@ include book.toml
 include pyproject.toml
 recursive-include changelog.d *
 
-prune .buildkite
 prune .circleci
 prune .github
+prune .ci
 prune contrib
 prune debian
 prune demo/etc

changelog.d/10394.feature

@@ -0,0 +1 @@
+Initial local support for [MSC3266](https://github.com/matrix-org/synapse/pull/10394), Room Summary over the unstable `/rooms/{roomIdOrAlias}/summary` API.

changelog.d/10475.feature

@@ -0,0 +1 @@
+Add support for sending federation requests through a proxy. Contributed by @Bubu and @dklimpel.

changelog.d/10524.feature

@@ -0,0 +1 @@
+Port the PresenceRouter module interface to the new generic interface.

changelog.d/10527.misc

@@ -1 +1 @@
-Prepare for the new spaces summary endpoint (updates to [MSC2946](https://github.com/matrix-org/matrix-doc/pull/2946)).
+Add pagination to the spaces summary based on updates to [MSC2946](https://github.com/matrix-org/matrix-doc/pull/2946).

changelog.d/10530.misc

@@ -1 +1 @@
-Prepare for the new spaces summary endpoint (updates to [MSC2946](https://github.com/matrix-org/matrix-doc/pull/2946)).
+Add pagination to the spaces summary based on updates to [MSC2946](https://github.com/matrix-org/matrix-doc/pull/2946).

changelog.d/10549.feature

@@ -0,0 +1 @@
+Add pagination to the spaces summary based on updates to [MSC2946](https://github.com/matrix-org/matrix-doc/pull/2946).

changelog.d/10552.misc

@@ -0,0 +1 @@
+Update `/batch_send` endpoint to only return `state_events` created by the `state_events_from_before` passed in.

changelog.d/10558.feature

@@ -0,0 +1 @@
+Admin API to delete several media for a specific user. Contributed by @dklimpel.

changelog.d/10561.bugfix

@@ -0,0 +1 @@
+Display an error on User-Interactive Authentication fallback pages when authentication fails. Contributed by Callum Brown.

changelog.d/10569.feature

@@ -0,0 +1 @@
+Add pagination to the spaces summary based on updates to [MSC2946](https://github.com/matrix-org/matrix-doc/pull/2946).

changelog.d/10570.feature

@@ -0,0 +1 @@
+Update the Synapse Grafana dashboard.

changelog.d/10572.misc

@@ -0,0 +1 @@
+Clarify error message when failing to join a restricted room.

changelog.d/10573.misc

@@ -0,0 +1 @@
+Remove references to BuildKite in favour of GitHub Actions.

changelog.d/10574.feature

@@ -0,0 +1 @@
+Add pagination to the spaces summary based on updates to [MSC2946](https://github.com/matrix-org/matrix-doc/pull/2946).

changelog.d/10575.feature

@@ -0,0 +1 @@
+Add pagination to the spaces summary based on updates to [MSC2946](https://github.com/matrix-org/matrix-doc/pull/2946).

changelog.d/10576.misc

@@ -0,0 +1 @@
+Move `/batch_send` endpoint defined by [MSC2716](https://github.com/matrix-org/matrix-doc/pull/2716) to the `/v2_alpha` directory.

changelog.d/10578.feature

@@ -0,0 +1 @@
+Add an admin API (`GET /_synapse/admin/username_available`) to check if a username is available (regardless of registration settings).

changelog.d/10579.feature

@@ -0,0 +1 @@
+Add pagination to the spaces summary based on updates to [MSC2946](https://github.com/matrix-org/matrix-doc/pull/2946).

changelog.d/10580.bugfix

@@ -0,0 +1 @@
+Allow public rooms to be previewed in the spaces summary APIs from [MSC2946](https://github.com/matrix-org/matrix-doc/pull/2946).

changelog.d/10587.misc

@@ -0,0 +1 @@
+Allow multiple custom directories in `read_templates`.

changelog.d/10588.removal

@@ -0,0 +1 @@
+No longer build `.dev` packages for Ubuntu 20.10 LTS Groovy Gorilla, which has now EOLed.

changelog.d/10590.misc

@@ -0,0 +1 @@
+Re-organize the `synapse.federation.transport.server` module to create smaller files.

changelog.d/10591.misc

@@ -0,0 +1 @@
+Clean up some of the federation event authentication code for clarity.

changelog.d/10592.bugfix

@@ -0,0 +1 @@
+Fix a bug introduced in v1.37.1 where an error could occur in the asyncronous processing of PDUs when the queue was empty.

changelog.d/10598.feature

@@ -0,0 +1 @@
+Allow editing a user's `external_ids` via the "Edit User" admin API. Contributed by @dklimpel.

changelog.d/10599.doc

@@ -0,0 +1 @@
+Update CONTRIBUTING.md to fix index links and the instructions for SyTest in docker.

changelog.d/10600.misc

@@ -0,0 +1 @@
+Flatten the `synapse.rest.client` package by moving the contents of `v1` and `v2_alpha` into the parent.

changelog.d/10602.feature

@@ -0,0 +1 @@
+The Synapse manhole no longer needs coroutines to be wrapped in `defer.ensureDeferred`.

changelog.d/10606.bugfix

@@ -0,0 +1 @@
+Fix errors on /sync when read receipt data is a string. Only affects homeservers with the experimental flag for [MSC2285](https://github.com/matrix-org/matrix-doc/pull/2285) enabled. Contributed by @SimonBrandner.

changelog.d/10611.bugfix

@@ -0,0 +1 @@
+Additional validation for the spaces summary API to avoid errors like `ValueError: Stop argument for islice() must be None or an integer`. The missing validation has existed since v1.31.0.

changelog.d/10614.misc

@@ -0,0 +1 @@
+Clean up some of the federation event authentication code for clarity.

changelog.d/10615.misc

@@ -0,0 +1 @@
+Clean up some of the federation event authentication code for clarity.

changelog.d/10620.misc

@@ -0,0 +1 @@
+Fix up a couple of links to the database schema documentation.

changelog.d/10623.bugfix

@@ -0,0 +1 @@
+Revert behaviour introduced in v1.38.0 that strips `org.matrix.msc2732.device_unused_fallback_key_types` from `/sync` when its value is empty. This field should instead always be present according to [MSC2732](https://github.com/matrix-org/matrix-doc/blob/master/proposals/2732-olm-fallback-keys.md).

changelog.d/10629.misc

@@ -0,0 +1 @@
+Convert room member storage tuples to `attrs` classes.

changelog.d/10630.misc

@@ -0,0 +1 @@
+Use auto-attribs for the attrs classes used in sync.

changelog.d/8830.removal

@@ -0,0 +1 @@
+Remove deprecated Shutdown Room and Purge Room Admin API.

contrib/grafana/synapse.json

@@ -54,7 +54,7 @@
   "gnetId": null,
   "graphTooltip": 0,
   "id": null,
-  "iteration": 1621258266004,
+  "iteration": 1628606819564,
   "links": [
     {
       "asDropdown": false,
@@ -307,7 +307,6 @@
       ],
       "thresholds": [
         {
-          "$$hashKey": "object:283",
           "colorMode": "warning",
           "fill": false,
           "line": true,
@@ -316,7 +315,6 @@
           "yaxis": "left"
         },
         {
-          "$$hashKey": "object:284",
           "colorMode": "critical",
           "fill": false,
           "line": true,
@@ -344,7 +342,6 @@
       },
       "yaxes": [
         {
-          "$$hashKey": "object:255",
           "decimals": null,
           "format": "s",
           "label": "",
@@ -354,7 +351,6 @@
           "show": true
         },
         {
-          "$$hashKey": "object:256",
           "format": "hertz",
           "label": "",
           "logBase": 1,
@@ -429,7 +425,6 @@
       ],
       "thresholds": [
         {
-          "$$hashKey": "object:566",
           "colorMode": "critical",
           "fill": true,
           "line": true,
@@ -457,7 +452,6 @@
       },
       "yaxes": [
         {
-          "$$hashKey": "object:538",
           "decimals": null,
           "format": "percentunit",
           "label": null,
@@ -467,7 +461,6 @@
           "show": true
         },
         {
-          "$$hashKey": "object:539",
           "format": "short",
           "label": null,
           "logBase": 1,
@@ -573,7 +566,6 @@
       },
       "yaxes": [
         {
-          "$$hashKey": "object:1560",
           "format": "bytes",
           "logBase": 1,
           "max": null,
@@ -581,7 +573,6 @@
           "show": true
         },
         {
-          "$$hashKey": "object:1561",
           "format": "short",
           "logBase": 1,
           "max": null,
@@ -641,7 +632,6 @@
       "renderer": "flot",
       "seriesOverrides": [
         {
-          "$$hashKey": "object:639",
           "alias": "/max$/",
           "color": "#890F02",
           "fill": 0,
@@ -693,7 +683,6 @@
       },
       "yaxes": [
         {
-          "$$hashKey": "object:650",
           "decimals": null,
           "format": "none",
           "label": "",
@@ -703,7 +692,6 @@
           "show": true
         },
         {
-          "$$hashKey": "object:651",
           "decimals": null,
           "format": "short",
           "label": null,
@@ -783,11 +771,9 @@
           "renderer": "flot",
           "seriesOverrides": [
             {
-              "$$hashKey": "object:1240",
               "alias": "/user/"
             },
             {
-              "$$hashKey": "object:1241",
               "alias": "/system/"
             }
           ],
@@ -817,7 +803,6 @@
           ],
           "thresholds": [
             {
-              "$$hashKey": "object:1278",
               "colorMode": "custom",
               "fillColor": "rgba(255, 255, 255, 1)",
               "line": true,
@@ -827,7 +812,6 @@
               "yaxis": "left"
             },
             {
-              "$$hashKey": "object:1279",
               "colorMode": "custom",
               "fillColor": "rgba(255, 255, 255, 1)",
               "line": true,
@@ -837,7 +821,6 @@
               "yaxis": "left"
             },
             {
-              "$$hashKey": "object:1498",
               "colorMode": "critical",
               "fill": true,
               "line": true,
@@ -865,7 +848,6 @@
           },
           "yaxes": [
             {
-              "$$hashKey": "object:1250",
               "decimals": null,
               "format": "percentunit",
               "label": "",
@@ -875,7 +857,6 @@
               "show": true
             },
             {
-              "$$hashKey": "object:1251",
               "format": "short",
               "logBase": 1,
               "max": null,
@@ -1427,7 +1408,6 @@
           },
           "yaxes": [
             {
-              "$$hashKey": "object:572",
               "format": "percentunit",
               "label": null,
               "logBase": 1,
@@ -1436,7 +1416,6 @@
               "show": true
             },
             {
-              "$$hashKey": "object:573",
               "format": "short",
               "label": null,
               "logBase": 1,
@@ -1720,7 +1699,6 @@
           },
           "yaxes": [
             {
-              "$$hashKey": "object:102",
               "format": "hertz",
               "logBase": 1,
               "max": null,
@@ -1728,7 +1706,6 @@
               "show": true
             },
             {
-              "$$hashKey": "object:103",
               "format": "short",
               "logBase": 1,
               "max": null,
@@ -3425,7 +3402,7 @@
             "h": 9,
             "w": 12,
             "x": 0,
-            "y": 33
+            "y": 6
           },
           "hiddenSeries": false,
           "id": 79,
@@ -3442,9 +3419,12 @@
           "linewidth": 1,
           "links": [],
           "nullPointMode": "null",
+          "options": {
+            "alertThreshold": true
+          },
           "paceLength": 10,
           "percentage": false,
-          "pluginVersion": "7.1.3",
+          "pluginVersion": "7.3.7",
           "pointradius": 5,
           "points": false,
           "renderer": "flot",
@@ -3526,7 +3506,7 @@
             "h": 9,
             "w": 12,
             "x": 12,
-            "y": 33
+            "y": 6
           },
           "hiddenSeries": false,
           "id": 83,
@@ -3543,9 +3523,12 @@
           "linewidth": 1,
           "links": [],
           "nullPointMode": "null",
+          "options": {
+            "alertThreshold": true
+          },
           "paceLength": 10,
           "percentage": false,
-          "pluginVersion": "7.1.3",
+          "pluginVersion": "7.3.7",
           "pointradius": 5,
           "points": false,
           "renderer": "flot",
@@ -3629,7 +3612,7 @@
             "h": 9,
             "w": 12,
             "x": 0,
-            "y": 42
+            "y": 15
           },
           "hiddenSeries": false,
           "id": 109,
@@ -3646,9 +3629,12 @@
           "linewidth": 1,
           "links": [],
           "nullPointMode": "null",
+          "options": {
+            "alertThreshold": true
+          },
           "paceLength": 10,
           "percentage": false,
-          "pluginVersion": "7.1.3",
+          "pluginVersion": "7.3.7",
           "pointradius": 5,
           "points": false,
           "renderer": "flot",
@@ -3733,7 +3719,7 @@
             "h": 9,
             "w": 12,
             "x": 12,
-            "y": 42
+            "y": 15
           },
           "hiddenSeries": false,
           "id": 111,
@@ -3750,9 +3736,12 @@
           "linewidth": 1,
           "links": [],
           "nullPointMode": "null",
+          "options": {
+            "alertThreshold": true
+          },
           "paceLength": 10,
           "percentage": false,
-          "pluginVersion": "7.1.3",
+          "pluginVersion": "7.3.7",
           "pointradius": 5,
           "points": false,
           "renderer": "flot",
@@ -3831,7 +3820,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 51
+            "y": 24
           },
           "hiddenSeries": false,
           "id": 142,
@@ -3847,8 +3836,11 @@
           "lines": true,
           "linewidth": 1,
           "nullPointMode": "null",
+          "options": {
+            "alertThreshold": true
+          },
           "percentage": false,
-          "pluginVersion": "7.1.3",
+          "pluginVersion": "7.3.7",
           "pointradius": 2,
           "points": false,
           "renderer": "flot",
@@ -3931,7 +3923,7 @@
             "h": 9,
             "w": 12,
             "x": 12,
-            "y": 51
+            "y": 24
           },
           "hiddenSeries": false,
           "id": 140,
@@ -3948,9 +3940,12 @@
           "linewidth": 1,
           "links": [],
           "nullPointMode": "null",
+          "options": {
+            "alertThreshold": true
+          },
           "paceLength": 10,
           "percentage": false,
-          "pluginVersion": "7.1.3",
+          "pluginVersion": "7.3.7",
           "pointradius": 5,
           "points": false,
           "renderer": "flot",
@@ -4079,7 +4074,7 @@
             "h": 9,
             "w": 12,
             "x": 0,
-            "y": 59
+            "y": 32
           },
           "heatmap": {},
           "hideZeroBuckets": false,
@@ -4145,7 +4140,7 @@
             "h": 9,
             "w": 12,
             "x": 12,
-            "y": 60
+            "y": 33
           },
           "hiddenSeries": false,
           "id": 162,
@@ -4163,9 +4158,12 @@
           "linewidth": 0,
           "links": [],
           "nullPointMode": "connected",
+          "options": {
+            "alertThreshold": true
+          },
           "paceLength": 10,
           "percentage": false,
-          "pluginVersion": "7.1.3",
+          "pluginVersion": "7.3.7",
           "pointradius": 5,
           "points": false,
           "renderer": "flot",
@@ -4350,7 +4348,7 @@
             "h": 9,
             "w": 12,
             "x": 0,
-            "y": 68
+            "y": 41
           },
           "heatmap": {},
           "hideZeroBuckets": false,
@@ -4396,6 +4394,311 @@
           "yBucketBound": "auto",
           "yBucketNumber": null,
           "yBucketSize": null
+        },
+        {
+          "aliasColors": {},
+          "bars": false,
+          "dashLength": 10,
+          "dashes": false,
+          "datasource": "$datasource",
+          "editable": true,
+          "error": false,
+          "fieldConfig": {
+            "defaults": {
+              "custom": {},
+              "links": []
+            },
+            "overrides": []
+          },
+          "fill": 1,
+          "fillGradient": 0,
+          "grid": {},
+          "gridPos": {
+            "h": 9,
+            "w": 12,
+            "x": 12,
+            "y": 42
+          },
+          "hiddenSeries": false,
+          "id": 203,
+          "legend": {
+            "avg": false,
+            "current": false,
+            "max": false,
+            "min": false,
+            "show": true,
+            "total": false,
+            "values": false
+          },
+          "lines": true,
+          "linewidth": 2,
+          "links": [],
+          "nullPointMode": "null",
+          "options": {
+            "alertThreshold": true
+          },
+          "paceLength": 10,
+          "percentage": false,
+          "pluginVersion": "7.3.7",
+          "pointradius": 5,
+          "points": false,
+          "renderer": "flot",
+          "seriesOverrides": [],
+          "spaceLength": 10,
+          "stack": false,
+          "steppedLine": false,
+          "targets": [
+            {
+              "expr": "synapse_federation_server_oldest_inbound_pdu_in_staging{job=\"$job\",index=~\"$index\",instance=\"$instance\"}",
+              "format": "time_series",
+              "interval": "",
+              "intervalFactor": 1,
+              "legendFormat": "rss {{index}}",
+              "refId": "A",
+              "step": 4
+            }
+          ],
+          "thresholds": [],
+          "timeFrom": null,
+          "timeRegions": [],
+          "timeShift": null,
+          "title": "Age of oldest event in staging area",
+          "tooltip": {
+            "msResolution": false,
+            "shared": true,
+            "sort": 0,
+            "value_type": "cumulative"
+          },
+          "type": "graph",
+          "xaxis": {
+            "buckets": null,
+            "mode": "time",
+            "name": null,
+            "show": true,
+            "values": []
+          },
+          "yaxes": [
+            {
+              "format": "ms",
+              "label": null,
+              "logBase": 1,
+              "max": null,
+              "min": 0,
+              "show": true
+            },
+            {
+              "format": "short",
+              "label": null,
+              "logBase": 1,
+              "max": null,
+              "min": null,
+              "show": true
+            }
+          ],
+          "yaxis": {
+            "align": false,
+            "alignLevel": null
+          }
+        },
+        {
+          "aliasColors": {},
+          "bars": false,
+          "dashLength": 10,
+          "dashes": false,
+          "datasource": "$datasource",
+          "editable": true,
+          "error": false,
+          "fieldConfig": {
+            "defaults": {
+              "custom": {},
+              "links": []
+            },
+            "overrides": []
+          },
+          "fill": 1,
+          "fillGradient": 0,
+          "grid": {},
+          "gridPos": {
+            "h": 9,
+            "w": 12,
+            "x": 0,
+            "y": 50
+          },
+          "hiddenSeries": false,
+          "id": 202,
+          "legend": {
+            "avg": false,
+            "current": false,
+            "max": false,
+            "min": false,
+            "show": true,
+            "total": false,
+            "values": false
+          },
+          "lines": true,
+          "linewidth": 2,
+          "links": [],
+          "nullPointMode": "null",
+          "options": {
+            "alertThreshold": true
+          },
+          "paceLength": 10,
+          "percentage": false,
+          "pluginVersion": "7.3.7",
+          "pointradius": 5,
+          "points": false,
+          "renderer": "flot",
+          "seriesOverrides": [],
+          "spaceLength": 10,
+          "stack": false,
+          "steppedLine": false,
+          "targets": [
+            {
+              "expr": "synapse_federation_server_number_inbound_pdu_in_staging{job=\"$job\",index=~\"$index\",instance=\"$instance\"}",
+              "format": "time_series",
+              "interval": "",
+              "intervalFactor": 1,
+              "legendFormat": "rss {{index}}",
+              "refId": "A",
+              "step": 4
+            }
+          ],
+          "thresholds": [],
+          "timeFrom": null,
+          "timeRegions": [],
+          "timeShift": null,
+          "title": "Number of events in federation staging area",
+          "tooltip": {
+            "msResolution": false,
+            "shared": true,
+            "sort": 0,
+            "value_type": "cumulative"
+          },
+          "type": "graph",
+          "xaxis": {
+            "buckets": null,
+            "mode": "time",
+            "name": null,
+            "show": true,
+            "values": []
+          },
+          "yaxes": [
+            {
+              "format": "none",
+              "label": null,
+              "logBase": 1,
+              "max": null,
+              "min": 0,
+              "show": true
+            },
+            {
+              "format": "short",
+              "label": null,
+              "logBase": 1,
+              "max": null,
+              "min": null,
+              "show": true
+            }
+          ],
+          "yaxis": {
+            "align": false,
+            "alignLevel": null
+          }
+        },
+        {
+          "aliasColors": {},
+          "bars": false,
+          "dashLength": 10,
+          "dashes": false,
+          "datasource": "${DS_PROMETHEUS}",
+          "fieldConfig": {
+            "defaults": {
+              "custom": {}
+            },
+            "overrides": []
+          },
+          "fill": 1,
+          "fillGradient": 0,
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 12,
+            "y": 51
+          },
+          "hiddenSeries": false,
+          "id": 205,
+          "legend": {
+            "avg": false,
+            "current": false,
+            "max": false,
+            "min": false,
+            "show": true,
+            "total": false,
+            "values": false
+          },
+          "lines": true,
+          "linewidth": 1,
+          "nullPointMode": "null",
+          "options": {
+            "alertThreshold": true
+          },
+          "percentage": false,
+          "pluginVersion": "7.3.7",
+          "pointradius": 2,
+          "points": false,
+          "renderer": "flot",
+          "seriesOverrides": [],
+          "spaceLength": 10,
+          "stack": false,
+          "steppedLine": false,
+          "targets": [
+            {
+              "expr": "sum(rate(synapse_federation_soft_failed_events_total{instance=\"$instance\"}[$bucket_size]))",
+              "interval": "",
+              "legendFormat": "soft-failed events",
+              "refId": "A"
+            }
+          ],
+          "thresholds": [],
+          "timeFrom": null,
+          "timeRegions": [],
+          "timeShift": null,
+          "title": "Soft-failed event rate",
+          "tooltip": {
+            "shared": true,
+            "sort": 0,
+            "value_type": "individual"
+          },
+          "type": "graph",
+          "xaxis": {
+            "buckets": null,
+            "mode": "time",
+            "name": null,
+            "show": true,
+            "values": []
+          },
+          "yaxes": [
+            {
+              "format": "hertz",
+              "label": null,
+              "logBase": 1,
+              "max": null,
+              "min": null,
+              "show": true
+            },
+            {
+              "format": "short",
+              "label": null,
+              "logBase": 1,
+              "max": null,
+              "min": null,
+              "show": false
+            }
+          ],
+          "yaxis": {
+            "align": false,
+            "alignLevel": null
+          }
         }
       ],
       "title": "Federation",
@@ -4647,7 +4950,7 @@
             "h": 7,
             "w": 12,
             "x": 0,
-            "y": 8
+            "y": 33
           },
           "hiddenSeries": false,
           "id": 48,
@@ -4749,7 +5052,7 @@
             "h": 7,
             "w": 12,
             "x": 12,
-            "y": 8
+            "y": 33
           },
           "hiddenSeries": false,
           "id": 104,
@@ -4877,7 +5180,7 @@
             "h": 7,
             "w": 12,
             "x": 0,
-            "y": 15
+            "y": 40
           },
           "hiddenSeries": false,
           "id": 10,
@@ -4981,7 +5284,7 @@
             "h": 7,
             "w": 12,
             "x": 12,
-            "y": 15
+            "y": 40
           },
           "hiddenSeries": false,
           "id": 11,
@@ -5086,7 +5389,7 @@
             "h": 7,
             "w": 12,
             "x": 0,
-            "y": 22
+            "y": 47
           },
           "hiddenSeries": false,
           "id": 180,
@@ -5168,6 +5471,126 @@
             "align": false,
             "alignLevel": null
           }
+        },
+        {
+          "aliasColors": {},
+          "bars": false,
+          "dashLength": 10,
+          "dashes": false,
+          "datasource": "$datasource",
+          "fieldConfig": {
+            "defaults": {
+              "custom": {},
+              "links": []
+            },
+            "overrides": []
+          },
+          "fill": 6,
+          "fillGradient": 0,
+          "gridPos": {
+            "h": 9,
+            "w": 12,
+            "x": 12,
+            "y": 47
+          },
+          "hiddenSeries": false,
+          "id": 200,
+          "legend": {
+            "avg": false,
+            "current": false,
+            "max": false,
+            "min": false,
+            "show": true,
+            "total": false,
+            "values": false
+          },
+          "lines": true,
+          "linewidth": 1,
+          "links": [],
+          "nullPointMode": "null",
+          "options": {
+            "alertThreshold": true
+          },
+          "percentage": false,
+          "pluginVersion": "7.3.7",
+          "pointradius": 5,
+          "points": false,
+          "renderer": "flot",
+          "seriesOverrides": [],
+          "spaceLength": 10,
+          "stack": false,
+          "steppedLine": false,
+          "targets": [
+            {
+              "expr": "histogram_quantile(0.99, sum(rate(synapse_storage_schedule_time_bucket{index=~\"$index\",instance=\"$instance\",job=\"$job\"}[$bucket_size])) by (le))",
+              "format": "time_series",
+              "intervalFactor": 1,
+              "legendFormat": "99%",
+              "refId": "D"
+            },
+            {
+              "expr": "histogram_quantile(0.9, sum(rate(synapse_storage_schedule_time_bucket{index=~\"$index\",instance=\"$instance\",job=\"$job\"}[$bucket_size])) by (le))",
+              "format": "time_series",
+              "intervalFactor": 1,
+              "legendFormat": "90%",
+              "refId": "A"
+            },
+            {
+              "expr": "histogram_quantile(0.75, sum(rate(synapse_storage_schedule_time_bucket{index=~\"$index\",instance=\"$instance\",job=\"$job\"}[$bucket_size])) by (le))",
+              "format": "time_series",
+              "intervalFactor": 1,
+              "legendFormat": "75%",
+              "refId": "C"
+            },
+            {
+              "expr": "histogram_quantile(0.5, sum(rate(synapse_storage_schedule_time_bucket{index=~\"$index\",instance=\"$instance\",job=\"$job\"}[$bucket_size])) by (le))",
+              "format": "time_series",
+              "intervalFactor": 1,
+              "legendFormat": "50%",
+              "refId": "B"
+            }
+          ],
+          "thresholds": [],
+          "timeFrom": null,
+          "timeRegions": [],
+          "timeShift": null,
+          "title": "Time waiting for DB connection quantiles",
+          "tooltip": {
+            "shared": true,
+            "sort": 2,
+            "value_type": "individual"
+          },
+          "type": "graph",
+          "xaxis": {
+            "buckets": null,
+            "mode": "time",
+            "name": null,
+            "show": true,
+            "values": []
+          },
+          "yaxes": [
+            {
+              "decimals": null,
+              "format": "s",
+              "label": "",
+              "logBase": 1,
+              "max": null,
+              "min": "0",
+              "show": true
+            },
+            {
+              "format": "short",
+              "label": null,
+              "logBase": 1,
+              "max": null,
+              "min": null,
+              "show": false
+            }
+          ],
+          "yaxis": {
+            "align": false,
+            "alignLevel": null
+          }
         }
       ],
       "repeat": null,
@@ -5916,7 +6339,7 @@
             "h": 10,
             "w": 12,
             "x": 0,
-            "y": 84
+            "y": 35
           },
           "hiddenSeries": false,
           "id": 1,
@@ -6022,7 +6445,7 @@
             "h": 10,
             "w": 12,
             "x": 12,
-            "y": 84
+            "y": 35
           },
           "hiddenSeries": false,
           "id": 8,
@@ -6126,7 +6549,7 @@
             "h": 10,
             "w": 12,
             "x": 0,
-            "y": 94
+            "y": 45
           },
           "hiddenSeries": false,
           "id": 38,
@@ -6226,7 +6649,7 @@
             "h": 10,
             "w": 12,
             "x": 12,
-            "y": 94
+            "y": 45
           },
           "hiddenSeries": false,
           "id": 39,
@@ -6258,8 +6681,9 @@
           "steppedLine": false,
           "targets": [
             {
-              "expr": "topk(10, rate(synapse_util_caches_cache:total{job=\"$job\",index=~\"$index\",instance=\"$instance\"}[$bucket_size]) - rate(synapse_util_caches_cache:hits{job=\"$job\",instance=\"$instance\"}[$bucket_size]))",
+              "expr": "topk(10, rate(synapse_util_caches_cache:total{job=~\"$job\",index=~\"$index\",instance=\"$instance\"}[$bucket_size]) - rate(synapse_util_caches_cache:hits{job=~\"$job\",index=~\"$index\",instance=\"$instance\"}[$bucket_size]))",
               "format": "time_series",
+              "interval": "",
               "intervalFactor": 2,
               "legendFormat": "{{name}} {{job}}-{{index}}",
               "refId": "A",
@@ -6326,7 +6750,7 @@
             "h": 9,
             "w": 12,
             "x": 0,
-            "y": 104
+            "y": 55
           },
           "hiddenSeries": false,
           "id": 65,
@@ -9051,7 +9475,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 119
+            "y": 41
           },
           "hiddenSeries": false,
           "id": 156,
@@ -9089,7 +9513,7 @@
           "steppedLine": false,
           "targets": [
             {
-              "expr": "synapse_admin_mau:current{instance=\"$instance\"}",
+              "expr": "synapse_admin_mau:current{instance=\"$instance\", job=~\"$job\"}",
               "format": "time_series",
               "interval": "",
               "intervalFactor": 1,
@@ -9097,7 +9521,7 @@
               "refId": "A"
             },
             {
-              "expr": "synapse_admin_mau:max{instance=\"$instance\"}",
+              "expr": "synapse_admin_mau:max{instance=\"$instance\", job=~\"$job\"}",
               "format": "time_series",
               "interval": "",
               "intervalFactor": 1,
@@ -9164,7 +9588,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 119
+            "y": 41
           },
           "hiddenSeries": false,
           "id": 160,
@@ -9484,7 +9908,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 73
+            "y": 43
           },
           "hiddenSeries": false,
           "id": 168,
@@ -9516,7 +9940,7 @@
             {
               "expr": "rate(synapse_appservice_api_sent_events{instance=\"$instance\"}[$bucket_size])",
               "interval": "",
-              "legendFormat": "{{exported_service}}",
+              "legendFormat": "{{service}}",
               "refId": "A"
             }
           ],
@@ -9579,7 +10003,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 73
+            "y": 43
           },
           "hiddenSeries": false,
           "id": 171,
@@ -9611,7 +10035,7 @@
             {
               "expr": "rate(synapse_appservice_api_sent_transactions{instance=\"$instance\"}[$bucket_size])",
               "interval": "",
-              "legendFormat": "{{exported_service}}",
+              "legendFormat": "{{service}}",
               "refId": "A"
             }
           ],
@@ -9959,7 +10383,6 @@
           },
           "yaxes": [
             {
-              "$$hashKey": "object:165",
               "format": "hertz",
               "label": null,
               "logBase": 1,
@@ -9968,7 +10391,6 @@
               "show": true
             },
             {
-              "$$hashKey": "object:166",
               "format": "short",
               "label": null,
               "logBase": 1,
@@ -10071,7 +10493,6 @@
           },
           "yaxes": [
             {
-              "$$hashKey": "object:390",
               "format": "hertz",
               "label": null,
               "logBase": 1,
@@ -10080,7 +10501,6 @@
               "show": true
             },
             {
-              "$$hashKey": "object:391",
               "format": "short",
               "label": null,
               "logBase": 1,
@@ -10169,7 +10589,6 @@
           },
           "yaxes": [
             {
-              "$$hashKey": "object:390",
               "format": "hertz",
               "label": null,
               "logBase": 1,
@@ -10178,7 +10597,6 @@
               "show": true
             },
             {
-              "$$hashKey": "object:391",
               "format": "short",
               "label": null,
               "logBase": 1,
@@ -10470,5 +10888,5 @@
   "timezone": "",
   "title": "Synapse",
   "uid": "000000012",
-  "version": 90
+  "version": 99
 }

docs/SUMMARY.md

@@ -51,12 +51,10 @@
       - [Event Reports](admin_api/event_reports.md)
       - [Media](admin_api/media_admin_api.md)
       - [Purge History](admin_api/purge_history_api.md)
-      - [Purge Rooms](admin_api/purge_room.md)
       - [Register Users](admin_api/register_api.md)
       - [Manipulate Room Membership](admin_api/room_membership.md)
       - [Rooms](admin_api/rooms.md)
       - [Server Notices](admin_api/server_notices.md)
-      - [Shutdown Room](admin_api/shutdown_room.md)
       - [Statistics](admin_api/statistics.md)
       - [Users](admin_api/user_admin_api.md)
       - [Server Version](admin_api/version_api.md)

docs/admin_api/media_admin_api.md

@@ -12,6 +12,7 @@
 - [Delete local media](#delete-local-media)
   * [Delete a specific local media](#delete-a-specific-local-media)
   * [Delete local media by date or size](#delete-local-media-by-date-or-size)
+  * [Delete media uploaded by a user](#delete-media-uploaded-by-a-user)
 - [Purge Remote Media API](#purge-remote-media-api)
 
 # Querying media
@@ -47,7 +48,8 @@ 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.md#list-media-of-a-user)
+the use of the
+[List media uploaded by a user](user_admin_api.md#list-media-uploaded-by-a-user)
 Admin API.
 
 # Quarantine media
@@ -281,6 +283,11 @@ The following fields are returned in the JSON response body:
 * `deleted_media`: an array of strings - List of deleted `media_id`
 * `total`: integer - Total number of deleted `media_id`
 
+## Delete media uploaded by a user
+
+You can find details of how to delete multiple media uploaded by a user in
+[User Admin API](user_admin_api.md#delete-media-uploaded-by-a-user).
+
 # Purge Remote Media API
 
 The purge remote media API allows server admins to purge old cached remote media.

docs/admin_api/purge_room.md

@@ -1,21 +0,0 @@
-Deprecated: Purge room API
-==========================
-
-**The old Purge room API is deprecated and will be removed in a future release.
-See the new [Delete Room API](rooms.md#delete-room-api) for more details.**
-
-This API will remove all trace of a room from your database.
-
-All local users must have left the room before it can be removed.
-
-The API is:
-
-```
-POST /_synapse/admin/v1/purge_room
-
-{
-    "room_id": "!room:id"
-}
-```
-
-You must authenticate using the access token of an admin user.

docs/admin_api/shutdown_room.md

@@ -1,102 +0,0 @@
-# Deprecated: Shutdown room API
-
-**The old Shutdown room API is deprecated and will be removed in a future release.
-See the new [Delete Room API](rooms.md#delete-room-api) for more details.**
-
-Shuts down a room, preventing new joins and moves local users and room aliases automatically
-to a new room. The new room will be created with the user specified by the
-`new_room_user_id` parameter as room administrator and will contain a message
-explaining what happened. Users invited to the new room will have power level
--10 by default, and thus be unable to speak. The old room's power levels will be changed to
-disallow any further invites or joins.
-
-The local server will only have the power to move local user and room aliases to
-the new room. Users on other servers will be unaffected.
-
-## API
-
-You will need to authenticate with an access token for an admin user.
-
-### URL
-
-`POST /_synapse/admin/v1/shutdown_room/{room_id}`
-
-### URL Parameters
-
-* `room_id` - The ID of the room (e.g `!someroom:example.com`)
-
-### JSON Body Parameters
-
-* `new_room_user_id` - Required. A string representing the user ID of the user that will admin
-                       the new room that all users in the old room will be moved to.
-* `room_name` - Optional. A string representing the name of the room that new users will be
-                invited to.
-* `message` - Optional. A string containing the first message that will be sent as
-              `new_room_user_id` in the new room. Ideally this will clearly convey why the
-               original room was shut down.
-
-If not specified, the default value of `room_name` is "Content Violation
-Notification". The default value of `message` is "Sharing illegal content on
-othis server is not permitted and rooms in violation will be blocked."
-
-### Response Parameters
-
-* `kicked_users` - An integer number representing the number of users that
-                   were kicked.
-* `failed_to_kick_users` - An integer number representing the number of users
-                           that were not kicked.
-* `local_aliases` - An array of strings representing the local aliases that were migrated from
-                    the old room to the new.
-* `new_room_id` - A string representing the room ID of the new room.
-
-## Example
-
-Request:
-
-```
-POST /_synapse/admin/v1/shutdown_room/!somebadroom%3Aexample.com
-
-{
-    "new_room_user_id": "@someuser:example.com",
-    "room_name": "Content Violation Notification",
-    "message": "Bad Room has been shutdown due to content violations on this server. Please review our Terms of Service."
-}
-```
-
-Response:
-
-```
-{
-    "kicked_users": 5,
-    "failed_to_kick_users": 0,
-    "local_aliases": ["#badroom:example.com", "#evilsaloon:example.com],
-    "new_room_id": "!newroomid:example.com",
-},
-```
-
-## Undoing room shutdowns
-
-*Note*: This guide may be outdated by the time you read it. By nature of room shutdowns being performed at the database level,
-the structure can and does change without notice.
-
-First, it's important to understand that a room shutdown is very destructive. Undoing a shutdown is not as simple as pretending it
-never happened - work has to be done to move forward instead of resetting the past. In fact, in some cases it might not be possible
-to recover at all:
-
-* If the room was invite-only, your users will need to be re-invited.
-* If the room no longer has any members at all, it'll be impossible to rejoin.
-* The first user to rejoin will have to do so via an alias on a different server.
-
-With all that being said, if you still want to try and recover the room:
-
-1. For safety reasons, shut down Synapse.
-2. In the database, run `DELETE FROM blocked_rooms WHERE room_id = '!example:example.org';`
-   * For caution: it's recommended to run this in a transaction: `BEGIN; DELETE ...;`, verify you got 1 result, then `COMMIT;`.
-   * The room ID is the same one supplied to the shutdown room API, not the Content Violation room.
-3. Restart Synapse.
-
-You will have to manually handle, if you so choose, the following:
-
-* Aliases that would have been redirected to the Content Violation room.
-* Users that would have been booted from the room (and will have been force-joined to the Content Violation room).
-* Removal of the Content Violation room if desired.

docs/admin_api/user_admin_api.md

@@ -81,6 +81,16 @@ with a body of:
             "address": "<user_mail_2>"
         }
     ],
+    "external_ids": [
+        {
+            "auth_provider": "<provider1>",
+            "external_id": "<user_id_provider_1>"
+        },
+        {
+            "auth_provider": "<provider2>",
+            "external_id": "<user_id_provider_2>"
+        }
+    ],
     "avatar_url": "<avatar_url>",
     "admin": false,
     "deactivated": false
@@ -90,26 +100,34 @@ with a body of:
 To use it, you will need to authenticate by providing an `access_token` for a
 server admin: [Admin API](../usage/administration/admin_api)
 
+Returns HTTP status code:
+- `201` - When a new user object was created.
+- `200` - When a user was modified.
+
 URL parameters:
 
 - `user_id`: fully-qualified user id: for example, `@user:server.com`.
 
 Body parameters:
 
-- `password`, optional. If provided, the user's password is updated and all
+- `password` - string, optional. If provided, the user's password is updated and all
   devices are logged out.
-
-- `displayname`, optional, defaults to the value of `user_id`.
-
-- `threepids`, optional, allows setting the third-party IDs (email, msisdn)
+- `displayname` - string, optional, defaults to the value of `user_id`.
+- `threepids` - array, optional, allows setting the third-party IDs (email, msisdn)
+  - `medium` - string. Kind of third-party ID, either `email` or `msisdn`.
+  - `address` - string. Value of third-party ID.
   belonging to a user.
-
-- `avatar_url`, optional, must be a
+- `external_ids` - array, optional. Allow setting the identifier of the external identity
+  provider for SSO (Single sign-on). Details in
+  [Sample Configuration File](../usage/configuration/homeserver_sample_config.html)
+  section `sso` and `oidc_providers`.
+  - `auth_provider` - string. ID of the external identity provider. Value of `idp_id`
+    in homeserver configuration.
+  - `external_id` - string, user ID in the external identity provider.
+- `avatar_url` - string, optional, must be a
   [MXC URI](https://matrix.org/docs/spec/client_server/r0.6.0#matrix-content-mxc-uris).
-
-- `admin`, optional, defaults to `false`.
-
-- `deactivated`, optional. If unspecified, deactivation state will be left
+- `admin` - bool, optional, defaults to `false`.
+- `deactivated` - bool, optional. If unspecified, deactivation state will be left
   unchanged on existing accounts and set to `false` for new accounts.
   A user cannot be erased by deactivating with this API. For details on
   deactivating users see [Deactivate Account](#deactivate-account).
@@ -443,8 +461,9 @@ The following fields are returned in the JSON response body:
 - `joined_rooms` - An array of `room_id`.
 - `total` - Number of rooms.
 
+## User media
 
-## List media of a user
+### List media uploaded by a user
 Gets a list of all local media that a specific `user_id` has created.
 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
@@ -543,7 +562,6 @@ The following fields are returned in the JSON response body:
 
 - `media` - An array of objects, each containing information about a media.
   Media objects contain the following fields:
-
   - `created_ts` - integer - Timestamp when the content was uploaded in ms.
   - `last_access_ts` - integer - Timestamp when the content was last accessed in ms.
   - `media_id` - string - The id used to refer to the media.
@@ -551,13 +569,58 @@ The following fields are returned in the JSON response body:
   - `media_type` - string - The MIME-type of the media.
   - `quarantined_by` - string - The user ID that initiated the quarantine request
     for this media.
-
   - `safe_from_quarantine` - bool - Status if this media is safe from quarantining.
   - `upload_name` - string - The name the media was uploaded with.
-
 - `next_token`: integer - Indication for pagination. See above.
 - `total` - integer - Total number of media.
 
+### Delete media uploaded by a user
+
+This API deletes the *local* media from the disk of your own server
+that a specific `user_id` has created. This includes any local thumbnails.
+
+This API will not affect media that has been uploaded to external
+media repositories (e.g https://github.com/turt2live/matrix-media-repo/).
+
+By default, the API deletes media ordered by descending creation date and ascending media ID.
+The newest media is deleted first. You can change the order with parameters
+`order_by` and `dir`. If no `limit` is set the API deletes `100` files per request.
+
+The API is:
+
+```
+DELETE /_synapse/admin/v1/users/<user_id>/media
+```
+
+To use it, you will need to authenticate by providing an `access_token` for a
+server admin: [Admin API](../usage/administration/admin_api)
+
+A response body like the following is returned:
+
+```json
+{
+  "deleted_media": [
+    "abcdefghijklmnopqrstuvwx"
+  ],
+  "total": 1
+}
+```
+
+The following fields are returned in the JSON response body:
+
+* `deleted_media`: an array of strings - List of deleted `media_id`
+* `total`: integer - Total number of deleted `media_id`
+
+**Note**: There is no `next_token`. This is not useful for deleting media, because
+after deleting media the remaining media have a new order.
+
+**Parameters**
+
+This API has the same parameters as
+[List media uploaded by a user](#list-media-uploaded-by-a-user).
+With the parameters you can for example limit the number of files to delete at once or
+delete largest/smallest or newest/oldest files first.
+
 ## Login as a user
 
 Get an access token that can be used to authenticate as that user. Useful for
@@ -1013,3 +1076,22 @@ The following parameters should be set in the URL:
 - `user_id` - The fully qualified MXID: for example, `@user:server.com`. The user must
   be local.
 
+### Check username availability
+
+Checks to see if a username is available, and valid, for the server. See [the client-server 
+API](https://matrix.org/docs/spec/client_server/r0.6.0#get-matrix-client-r0-register-available)
+for more information.
+
+This endpoint will work even if registration is disabled on the server, unlike 
+`/_matrix/client/r0/register/available`.
+
+The API is:
+
+```
+POST /_synapse/admin/v1/username_availabile?username=$localpart
+```
+
+The request and response format is the same as the [/_matrix/client/r0/register/available](https://matrix.org/docs/spec/client_server/r0.6.0#get-matrix-client-r0-register-available) API.
+
+To use it, you will need to authenticate by providing an `access_token` for a
+server admin: [Admin API](../usage/administration/admin_api)

docs/manhole.md

@@ -67,7 +67,7 @@ This gives a Python REPL in which `hs` gives access to the
 `synapse.server.HomeServer` object - which in turn gives access to many other
 parts of the process.
 
-Note that any call which returns a coroutine will need to be wrapped in `ensureDeferred`.
+Note that, prior to Synapse 1.41, any call which returns a coroutine will need to be wrapped in `ensureDeferred`.
 
 As a simple example, retrieving an event from the database:
 

docs/modules.md

@@ -282,6 +282,52 @@ the request is a server admin.
 Modules can modify the `request_content` (by e.g. adding events to its `initial_state`),
 or deny the room's creation by raising a `module_api.errors.SynapseError`.
 
+#### Presence router callbacks
+
+Presence router callbacks allow module developers to specify additional users (local or remote)
+to receive certain presence updates from local users. Presence router callbacks can be 
+registered using the module API's `register_presence_router_callbacks` method.
+
+The available presence router callbacks are:
+
+```python 
+async def get_users_for_states(
+    self,
+    state_updates: Iterable["synapse.api.UserPresenceState"],
+) -> Dict[str, Set["synapse.api.UserPresenceState"]]:
+```
+**Requires** `get_interested_users` to also be registered
+
+Called when processing updates to the presence state of one or more users. This callback can
+be used to instruct the server to forward that presence state to specific users. The module
+must return a dictionary that maps from Matrix user IDs (which can be local or remote) to the
+`UserPresenceState` changes that they should be forwarded.
+
+Synapse will then attempt to send the specified presence updates to each user when possible.
+
+```python
+async def get_interested_users(
+        self, 
+        user_id: str
+) -> Union[Set[str], "synapse.module_api.PRESENCE_ALL_USERS"]
+```
+**Requires** `get_users_for_states` to also be registered
+
+Called when determining which users someone should be able to see the presence state of. This
+callback should return complementary results to `get_users_for_state` or the presence information 
+may not be properly forwarded.
+
+The callback is given the Matrix user ID for a local user that is requesting presence data and
+should return the Matrix user IDs of the users whose presence state they are allowed to
+query. The returned users can be local or remote. 
+
+Alternatively the callback can return `synapse.module_api.PRESENCE_ALL_USERS`
+to indicate that the user should receive updates from all known users.
+
+For example, if the user `@alice:example.org` is passed to this method, and the Set 
+`{"@bob:example.com", "@charlie:somewhere.org"}` is returned, this signifies that Alice 
+should receive presence updates sent by Bob and Charlie, regardless of whether these users 
+share a room.
 
 ### Porting an existing module that uses the old interface
 

docs/presence_router_module.md

@@ -1,3 +1,9 @@
+<h2 style="color:red">
+This page of the Synapse documentation is now deprecated. For up to date
+documentation on setting up or writing a presence router module, please see
+<a href="modules.md">this page</a>.
+</h2>
+
 # Presence Router Module
 
 Synapse supports configuring a module that can specify additional users

docs/sample_config.yaml

@@ -108,20 +108,6 @@ presence:
   #
   #enabled: false
 
-  # Presence routers are third-party modules that can specify additional logic
-  # to where presence updates from users are routed.
-  #
-  presence_router:
-    # The custom module's class. Uncomment to use a custom presence router module.
-    #
-    #module: "my_custom_router.PresenceRouter"
-
-    # Configuration options of the custom module. Refer to your module's
-    # documentation for available options.
-    #
-    #config:
-    #  example_option: 'something'
-
 # 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

docs/setup/forward_proxy.md

@@ -45,18 +45,18 @@ The proxy will be **used** for:
 - recaptcha validation
 - CAS auth validation
 - OpenID Connect
+- Outbound federation
 - Federation (checking public key revocation)
+- Fetching public keys of other servers
+- Downloading remote media
 
 It will **not be used** for:
 
 - Application Services
 - Identity servers
-- Outbound federation
 - In worker configurations
   - connections between workers
   - connections from workers to Redis
-- Fetching public keys of other servers
-- Downloading remote media
 
 ## Troubleshooting
 

docs/upgrade.md

@@ -85,6 +85,54 @@ process, for example:
     dpkg -i matrix-synapse-py3_1.3.0+stretch1_amd64.deb
     ```
 
+# Upgrading to v1.xx.0
+
+## Removal of old Room Admin API
+
+The following admin APIs were deprecated in [Synapse 1.25](https://github.com/matrix-org/synapse/blob/v1.25.0/CHANGES.md#removal-warning)
+(released on 2021-01-13) and have now been removed:
+
+-   `POST /_synapse/admin/v1/purge_room`
+-   `POST /_synapse/admin/v1/shutdown_room/<room_id>`
+
+Any scripts still using the above APIs should be converted to use the 
+[Delete Room API](https://matrix-org.github.io/synapse/latest/admin_api/rooms.html#delete-room-api).
+
+
+# Upgrading to v1.xx.0
+
+## Add support for routing outbound HTTP requests via a proxy for federation
+
+Since Synapse 1.6.0 (2019-11-26) you can set a proxy for outbound HTTP requests via
+http_proxy/https_proxy environment variables. This proxy was set for:
+- push
+- url previews
+- phone-home stats
+- recaptcha validation
+- CAS auth validation
+- OpenID Connect
+- Federation (checking public key revocation)
+
+In this version we have added support for outbound requests for:
+- Outbound federation
+- Downloading remote media
+- Fetching public keys of other servers
+
+These requests use the same proxy configuration. If you have a proxy configuration we
+recommend to verify the configuration. It may be necessary to adjust the `no_proxy`
+environment variable.
+
+See [using a forward proxy with Synapse documentation](setup/forward_proxy.md) for
+details.
+
+## User-interactive authentication fallback templates can now display errors
+
+This may affect you if you make use of custom HTML templates for the
+[reCAPTCHA](../synapse/res/templates/recaptcha.html) or
+[terms](../synapse/res/templates/terms.html) fallback pages.
+
+The template is now provided an `error` variable if the authentication
+process failed. See the default templates linked above for an example.
 
 # Upgrading to v1.39.0
 

mypy.ini

@@ -86,6 +86,7 @@ files =
   tests/test_event_auth.py,
   tests/test_utils,
   tests/handlers/test_password_providers.py,
+  tests/handlers/test_room_summary.py,
   tests/rest/client/v1/test_login.py,
   tests/rest/client/v2_alpha/test_auth.py,
   tests/util/test_itertools.py,

scripts-dev/build_debian_packages

@@ -25,7 +25,6 @@ DISTS = (
     "debian:sid",
     "ubuntu:bionic",  # 18.04 LTS (our EOL forced by Py36 on 2021-12-23)
     "ubuntu:focal",  # 20.04 LTS (our EOL forced by Py38 on 2024-10-14)
-    "ubuntu:groovy",  # 20.10 (EOL 2021-07-07)
     "ubuntu:hirsute",  # 21.04 (EOL 2022-01-05)
 )
 

scripts-dev/check-newsfragment

@@ -11,7 +11,7 @@ set -e
 git remote set-branches --add origin develop
 git fetch -q origin develop
 
-pr="$BUILDKITE_PULL_REQUEST"
+pr="$PULL_REQUEST_NUMBER"
 
 # if there are changes in the debian directory, check that the debian changelog
 # has been updated

scripts-dev/lint.sh

@@ -94,7 +94,7 @@ else
           "scripts-dev/build_debian_packages"
           "scripts-dev/sign_json"
           "scripts-dev/update_database"
-          "contrib" "synctl" "setup.py" "synmark" "stubs" ".buildkite"
+          "contrib" "synctl" "setup.py" "synmark" "stubs" ".ci"
       )
   fi
 fi

synapse/app/_base.py

@@ -37,6 +37,7 @@ from synapse.app import check_bind_error
 from synapse.app.phone_stats_home import start_phone_stats_home
 from synapse.config.homeserver import HomeServerConfig
 from synapse.crypto import context_factory
+from synapse.events.presence_router import load_legacy_presence_router
 from synapse.events.spamcheck import load_legacy_spam_checkers
 from synapse.events.third_party_rules import load_legacy_third_party_event_rules
 from synapse.logging.context import PreserveLoggingContext
@@ -370,6 +371,7 @@ async def start(hs: "HomeServer"):
 
     load_legacy_spam_checkers(hs)
     load_legacy_third_party_event_rules(hs)
+    load_legacy_presence_router(hs)
 
     # If we've configured an expiry time for caches, start the background job now.
     setup_expire_lru_cache_entries(hs)

synapse/app/generic_worker.py

@@ -66,40 +66,40 @@ from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
 from synapse.replication.slave.storage.registration import SlavedRegistrationStore
 from synapse.replication.slave.storage.room import RoomStore
 from synapse.rest.admin import register_servlets_for_media_repo
-from synapse.rest.client.v1 import events, login, presence, room
-from synapse.rest.client.v1.initial_sync import InitialSyncRestServlet
-from synapse.rest.client.v1.profile import (
-    ProfileAvatarURLRestServlet,
-    ProfileDisplaynameRestServlet,
-    ProfileRestServlet,
-)
-from synapse.rest.client.v1.push_rule import PushRuleRestServlet
-from synapse.rest.client.v1.voip import VoipRestServlet
-from synapse.rest.client.v2_alpha import (
+from synapse.rest.client import (
     account_data,
+    events,
     groups,
+    login,
+    presence,
     read_marker,
     receipts,
+    room,
     room_keys,
     sync,
     tags,
     user_directory,
 )
-from synapse.rest.client.v2_alpha._base import client_patterns
-from synapse.rest.client.v2_alpha.account import ThreepidRestServlet
-from synapse.rest.client.v2_alpha.account_data import (
-    AccountDataServlet,
-    RoomAccountDataServlet,
-)
-from synapse.rest.client.v2_alpha.devices import DevicesRestServlet
-from synapse.rest.client.v2_alpha.keys import (
+from synapse.rest.client._base import client_patterns
+from synapse.rest.client.account import ThreepidRestServlet
+from synapse.rest.client.account_data import AccountDataServlet, RoomAccountDataServlet
+from synapse.rest.client.devices import DevicesRestServlet
+from synapse.rest.client.initial_sync import InitialSyncRestServlet
+from synapse.rest.client.keys import (
     KeyChangesServlet,
     KeyQueryServlet,
     OneTimeKeyServlet,
 )
-from synapse.rest.client.v2_alpha.register import RegisterRestServlet
-from synapse.rest.client.v2_alpha.sendtodevice import SendToDeviceRestServlet
+from synapse.rest.client.profile import (
+    ProfileAvatarURLRestServlet,
+    ProfileDisplaynameRestServlet,
+    ProfileRestServlet,
+)
+from synapse.rest.client.push_rule import PushRuleRestServlet
+from synapse.rest.client.register import RegisterRestServlet
+from synapse.rest.client.sendtodevice import SendToDeviceRestServlet
 from synapse.rest.client.versions import VersionsRestServlet
+from synapse.rest.client.voip import VoipRestServlet
 from synapse.rest.health import HealthResource
 from synapse.rest.key.v2 import KeyApiV2Resource
 from synapse.rest.synapse.client import build_synapse_client_resource_tree

synapse/config/_base.py

@@ -237,13 +237,14 @@ class Config:
     def read_templates(
         self,
         filenames: List[str],
-        custom_template_directory: Optional[str] = None,
+        custom_template_directories: Optional[Iterable[str]] = None,
     ) -> List[jinja2.Template]:
         """Load a list of template files from disk using the given variables.
 
         This function will attempt to load the given templates from the default Synapse
-        template directory. If `custom_template_directory` is supplied, that directory
-        is tried first.
+        template directory. If `custom_template_directories` is supplied, any directory
+        in this list is tried (in the order they appear in the list) before trying
+        Synapse's default directory.
 
         Files read are treated as Jinja templates. The templates are not rendered yet
         and have autoescape enabled.
@@ -251,8 +252,8 @@ class Config:
         Args:
             filenames: A list of template filenames to read.
 
-            custom_template_directory: A directory to try to look for the templates
-                before using the default Synapse template directory instead.
+            custom_template_directories: A list of directory to try to look for the
+                templates before using the default Synapse template directory instead.
 
         Raises:
             ConfigError: if the file's path is incorrect or otherwise cannot be read.
@@ -260,20 +261,26 @@ class Config:
         Returns:
             A list of jinja2 templates.
         """
-        search_directories = [self.default_template_dir]
+        search_directories = []
 
-        # The loader will first look in the custom template directory (if specified) for the
-        # given filename. If it doesn't find it, it will use the default template dir instead
-        if custom_template_directory:
-            # Check that the given template directory exists
-            if not self.path_exists(custom_template_directory):
-                raise ConfigError(
-                    "Configured template directory does not exist: %s"
-                    % (custom_template_directory,)
-                )
+        # The loader will first look in the custom template directories (if specified)
+        # for the given filename. If it doesn't find it, it will use the default
+        # template dir instead.
+        if custom_template_directories is not None:
+            for custom_template_directory in custom_template_directories:
+                # Check that the given template directory exists
+                if not self.path_exists(custom_template_directory):
+                    raise ConfigError(
+                        "Configured template directory does not exist: %s"
+                        % (custom_template_directory,)
+                    )
 
-            # Search the custom template directory as well
-            search_directories.insert(0, custom_template_directory)
+                # Search the custom template directory as well
+                search_directories.append(custom_template_directory)
+
+        # Append the default directory at the end of the list so Jinja can fallback on it
+        # if a template is missing from any custom directory.
+        search_directories.append(self.default_template_dir)
 
         # TODO: switch to synapse.util.templates.build_jinja_env
         loader = jinja2.FileSystemLoader(search_directories)

synapse/config/account_validity.py

@@ -88,5 +88,5 @@ class AccountValidityConfig(Config):
                 "account_previously_renewed.html",
                 invalid_token_template_filename,
             ],
-            account_validity_template_dir,
+            (td for td in (account_validity_template_dir,) if td),
         )

synapse/config/emailconfig.py

@@ -257,7 +257,9 @@ class EmailConfig(Config):
                     registration_template_success_html,
                     add_threepid_template_success_html,
                 ],
-                template_dir,
+                (
+                    td for td in (template_dir,) if td
+                ),  # Filter out template_dir if not provided
             )
 
             # Render templates that do not contain any placeholders
@@ -297,7 +299,7 @@ class EmailConfig(Config):
                 self.email_notif_template_text,
             ) = self.read_templates(
                 [notif_template_html, notif_template_text],
-                template_dir,
+                (td for td in (template_dir,) if td),
             )
 
             self.email_notif_for_new_users = email_config.get(
@@ -320,7 +322,7 @@ class EmailConfig(Config):
                 self.account_validity_template_text,
             ) = self.read_templates(
                 [expiry_template_html, expiry_template_text],
-                template_dir,
+                (td for td in (template_dir,) if td),
             )
 
         subjects_config = email_config.get("subjects", {})

synapse/config/experimental.py

@@ -38,3 +38,6 @@ class ExperimentalConfig(Config):
 
         # MSC3244 (room version capabilities)
         self.msc3244_enabled: bool = experimental.get("msc3244_enabled", False)
+
+        # MSC3266 (room summary api)
+        self.msc3266_enabled: bool = experimental.get("msc3266_enabled", False)

synapse/config/server.py

@@ -248,6 +248,7 @@ class ServerConfig(Config):
             self.use_presence = config.get("use_presence", True)
 
         # Custom presence router module
+        # This is the legacy way of configuring it (the config should now be put in the modules section)
         self.presence_router_module_class = None
         self.presence_router_config = None
         presence_router_config = presence_config.get("presence_router")
@@ -858,20 +859,6 @@ class ServerConfig(Config):
           #
           #enabled: false
 
-          # Presence routers are third-party modules that can specify additional logic
-          # to where presence updates from users are routed.
-          #
-          presence_router:
-            # The custom module's class. Uncomment to use a custom presence router module.
-            #
-            #module: "my_custom_router.PresenceRouter"
-
-            # Configuration options of the custom module. Refer to your module's
-            # documentation for available options.
-            #
-            #config:
-            #  example_option: 'something'
-
         # 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

synapse/config/sso.py

@@ -63,7 +63,7 @@ class SSOConfig(Config):
                 "sso_auth_success.html",
                 "sso_auth_bad_user.html",
             ],
-            self.sso_template_dir,
+            (td for td in (self.sso_template_dir,) if td),
         )
 
         # These templates have no placeholders, so render them here

synapse/events/__init__.py

@@ -396,10 +396,11 @@ class FrozenEvent(EventBase):
         return self.__repr__()
 
     def __repr__(self):
-        return "<FrozenEvent event_id=%r, type=%r, state_key=%r>" % (
+        return "<FrozenEvent event_id=%r, type=%r, state_key=%r, outlier=%s>" % (
             self.get("event_id", None),
             self.get("type", None),
             self.get("state_key", None),
+            self.internal_metadata.is_outlier(),
         )
 
 

synapse/events/presence_router.py

@@ -11,45 +11,115 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
-from typing import TYPE_CHECKING, Dict, Iterable, Set, Union
+import logging
+from typing import (
+    TYPE_CHECKING,
+    Awaitable,
+    Callable,
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    Set,
+    Union,
+)
 
 from synapse.api.presence import UserPresenceState
+from synapse.util.async_helpers import maybe_awaitable
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
 
+GET_USERS_FOR_STATES_CALLBACK = Callable[
+    [Iterable[UserPresenceState]], Awaitable[Dict[str, Set[UserPresenceState]]]
+]
+GET_INTERESTED_USERS_CALLBACK = Callable[
+    [str], Awaitable[Union[Set[str], "PresenceRouter.ALL_USERS"]]
+]
+
+logger = logging.getLogger(__name__)
+
+
+def load_legacy_presence_router(hs: "HomeServer"):
+    """Wrapper that loads a presence router module configured using the old
+    configuration, and registers the hooks they implement.
+    """
+
+    if hs.config.presence_router_module_class is None:
+        return
+
+    module = hs.config.presence_router_module_class
+    config = hs.config.presence_router_config
+    api = hs.get_module_api()
+
+    presence_router = module(config=config, module_api=api)
+
+    # The known hooks. If a module implements a method which name appears in this set,
+    # we'll want to register it.
+    presence_router_methods = {
+        "get_users_for_states",
+        "get_interested_users",
+    }
+
+    # All methods that the module provides should be async, but this wasn't enforced
+    # in the old module system, so we wrap them if needed
+    def async_wrapper(f: Optional[Callable]) -> Optional[Callable[..., Awaitable]]:
+        # f might be None if the callback isn't implemented by the module. In this
+        # case we don't want to register a callback at all so we return None.
+        if f is None:
+            return None
+
+        def run(*args, **kwargs):
+            # mypy doesn't do well across function boundaries so we need to tell it
+            # f is definitely not None.
+            assert f is not None
+
+            return maybe_awaitable(f(*args, **kwargs))
+
+        return run
+
+    # Register the hooks through the module API.
+    hooks = {
+        hook: async_wrapper(getattr(presence_router, hook, None))
+        for hook in presence_router_methods
+    }
+
+    api.register_presence_router_callbacks(**hooks)
+
 
 class PresenceRouter:
     """
     A module that the homeserver will call upon to help route user presence updates to
-    additional destinations. If a custom presence router is configured, calls will be
-    passed to that instead.
+    additional destinations.
     """
 
     ALL_USERS = "ALL"
 
     def __init__(self, hs: "HomeServer"):
-        self.custom_presence_router = None
+        # Initially there are no callbacks
+        self._get_users_for_states_callbacks: List[GET_USERS_FOR_STATES_CALLBACK] = []
+        self._get_interested_users_callbacks: List[GET_INTERESTED_USERS_CALLBACK] = []
 
-        # Check whether a custom presence router module has been configured
-        if hs.config.presence_router_module_class:
-            # Initialise the module
-            self.custom_presence_router = hs.config.presence_router_module_class(
-                config=hs.config.presence_router_config, module_api=hs.get_module_api()
+    def register_presence_router_callbacks(
+        self,
+        get_users_for_states: Optional[GET_USERS_FOR_STATES_CALLBACK] = None,
+        get_interested_users: Optional[GET_INTERESTED_USERS_CALLBACK] = None,
+    ):
+        # PresenceRouter modules are required to implement both of these methods
+        # or neither of them as they are assumed to act in a complementary manner
+        paired_methods = [get_users_for_states, get_interested_users]
+        if paired_methods.count(None) == 1:
+            raise RuntimeError(
+                "PresenceRouter modules must register neither or both of the paired callbacks: "
+                "[get_users_for_states, get_interested_users]"
             )
 
-            # Ensure the module has implemented the required methods
-            required_methods = ["get_users_for_states", "get_interested_users"]
-            for method_name in required_methods:
-                if not hasattr(self.custom_presence_router, method_name):
-                    raise Exception(
-                        "PresenceRouter module '%s' must implement all required methods: %s"
-                        % (
-                            hs.config.presence_router_module_class.__name__,
-                            ", ".join(required_methods),
-                        )
-                    )
+        # Append the methods provided to the lists of callbacks
+        if get_users_for_states is not None:
+            self._get_users_for_states_callbacks.append(get_users_for_states)
+
+        if get_interested_users is not None:
+            self._get_interested_users_callbacks.append(get_interested_users)
 
     async def get_users_for_states(
         self,
@@ -66,14 +136,40 @@ class PresenceRouter:
           A dictionary of user_id -> set of UserPresenceState, indicating which
           presence updates each user should receive.
         """
-        if self.custom_presence_router is not None:
-            # Ask the custom module
-            return await self.custom_presence_router.get_users_for_states(
-                state_updates=state_updates
-            )
 
-        # Don't include any extra destinations for presence updates
-        return {}
+        # Bail out early if we don't have any callbacks to run.
+        if len(self._get_users_for_states_callbacks) == 0:
+            # Don't include any extra destinations for presence updates
+            return {}
+
+        users_for_states = {}
+        # run all the callbacks for get_users_for_states and combine the results
+        for callback in self._get_users_for_states_callbacks:
+            try:
+                result = await callback(state_updates)
+            except Exception as e:
+                logger.warning("Failed to run module API callback %s: %s", callback, e)
+                continue
+
+            if not isinstance(result, Dict):
+                logger.warning(
+                    "Wrong type returned by module API callback %s: %s, expected Dict",
+                    callback,
+                    result,
+                )
+                continue
+
+            for key, new_entries in result.items():
+                if not isinstance(new_entries, Set):
+                    logger.warning(
+                        "Wrong type returned by module API callback %s: %s, expected Set",
+                        callback,
+                        new_entries,
+                    )
+                    break
+                users_for_states.setdefault(key, set()).update(new_entries)
+
+        return users_for_states
 
     async def get_interested_users(self, user_id: str) -> Union[Set[str], ALL_USERS]:
         """
@@ -92,12 +188,36 @@ class PresenceRouter:
             A set of user IDs to return presence updates for, or ALL_USERS to return all
             known updates.
         """
-        if self.custom_presence_router is not None:
-            # Ask the custom module for interested users
-            return await self.custom_presence_router.get_interested_users(
-                user_id=user_id
-            )
 
-        # A custom presence router is not defined.
-        # Don't report any additional interested users
-        return set()
+        # Bail out early if we don't have any callbacks to run.
+        if len(self._get_interested_users_callbacks) == 0:
+            # Don't report any additional interested users
+            return set()
+
+        interested_users = set()
+        # run all the callbacks for get_interested_users and combine the results
+        for callback in self._get_interested_users_callbacks:
+            try:
+                result = await callback(user_id)
+            except Exception as e:
+                logger.warning("Failed to run module API callback %s: %s", callback, e)
+                continue
+
+            # If one of the callbacks returns ALL_USERS then we can stop calling all
+            # of the other callbacks, since the set of interested_users is already as
+            # large as it can possibly be
+            if result == PresenceRouter.ALL_USERS:
+                return PresenceRouter.ALL_USERS
+
+            if not isinstance(result, Set):
+                logger.warning(
+                    "Wrong type returned by module API callback %s: %s, expected set",
+                    callback,
+                    result,
+                )
+                continue
+
+            # Add the new interested users to the set
+            interested_users.update(result)
+
+        return interested_users

synapse/federation/federation_client.py

@@ -1290,6 +1290,88 @@ class FederationClient(FederationBase):
             failover_on_unknown_endpoint=True,
         )
 
+    async def get_room_hierarchy(
+        self,
+        destinations: Iterable[str],
+        room_id: str,
+        suggested_only: bool,
+    ) -> Tuple[JsonDict, Sequence[JsonDict], Sequence[str]]:
+        """
+        Call other servers to get a hierarchy of the given room.
+
+        Performs simple data validates and parsing of the response.
+
+        Args:
+            destinations: The remote servers. We will try them in turn, omitting any
+                that have been blacklisted.
+            room_id: ID of the space to be queried
+            suggested_only:  If true, ask the remote server to only return children
+                with the "suggested" flag set
+
+        Returns:
+            A tuple of:
+                The room as a JSON dictionary.
+                A list of children rooms, as JSON dictionaries.
+                A list of inaccessible children room IDs.
+
+        Raises:
+            SynapseError if we were unable to get a valid summary from any of the
+               remote servers
+        """
+
+        async def send_request(
+            destination: str,
+        ) -> Tuple[JsonDict, Sequence[JsonDict], Sequence[str]]:
+            res = await self.transport_layer.get_room_hierarchy(
+                destination=destination,
+                room_id=room_id,
+                suggested_only=suggested_only,
+            )
+
+            room = res.get("room")
+            if not isinstance(room, dict):
+                raise InvalidResponseError("'room' must be a dict")
+
+            # Validate children_state of the room.
+            children_state = room.get("children_state", [])
+            if not isinstance(children_state, Sequence):
+                raise InvalidResponseError("'room.children_state' must be a list")
+            if any(not isinstance(e, dict) for e in children_state):
+                raise InvalidResponseError("Invalid event in 'children_state' list")
+            try:
+                [
+                    FederationSpaceSummaryEventResult.from_json_dict(e)
+                    for e in children_state
+                ]
+            except ValueError as e:
+                raise InvalidResponseError(str(e))
+
+            # Validate the children rooms.
+            children = res.get("children", [])
+            if not isinstance(children, Sequence):
+                raise InvalidResponseError("'children' must be a list")
+            if any(not isinstance(r, dict) for r in children):
+                raise InvalidResponseError("Invalid room in 'children' list")
+
+            # Validate the inaccessible children.
+            inaccessible_children = res.get("inaccessible_children", [])
+            if not isinstance(inaccessible_children, Sequence):
+                raise InvalidResponseError("'inaccessible_children' must be a list")
+            if any(not isinstance(r, str) for r in inaccessible_children):
+                raise InvalidResponseError(
+                    "Invalid room ID in 'inaccessible_children' list"
+                )
+
+            return room, children, inaccessible_children
+
+        # TODO Fallback to the old federation API and translate the results.
+        return await self._try_destination_list(
+            "fetch room hierarchy",
+            destinations,
+            send_request,
+            failover_on_unknown_endpoint=True,
+        )
+
 
 @attr.s(frozen=True, slots=True, auto_attribs=True)
 class FederationSpaceSummaryEventResult:

synapse/federation/federation_server.py

@@ -972,13 +972,18 @@ class FederationServer(FederationBase):
         # the room, so instead of pulling the event out of the DB and parsing
         # the event we just pull out the next event ID and check if that matches.
         if latest_event is not None and latest_origin is not None:
-            (
-                next_origin,
-                next_event_id,
-            ) = await self.store.get_next_staged_event_id_for_room(room_id)
-            if next_origin != latest_origin or next_event_id != latest_event.event_id:
+            result = await self.store.get_next_staged_event_id_for_room(room_id)
+            if result is None:
                 latest_origin = None
                 latest_event = None
+            else:
+                next_origin, next_event_id = result
+                if (
+                    next_origin != latest_origin
+                    or next_event_id != latest_event.event_id
+                ):
+                    latest_origin = None
+                    latest_event = None
 
         if latest_origin is None or latest_event is None:
             next = await self.store.get_next_staged_event_for_room(
@@ -998,6 +1003,7 @@ class FederationServer(FederationBase):
         # has started processing).
         while True:
             async with lock:
+                logger.info("handling received PDU: %s", event)
                 try:
                     await self.handler.on_receive_pdu(
                         origin, event, sent_to_us_directly=True

synapse/federation/transport/client.py

@@ -1177,6 +1177,28 @@ class TransportLayerClient:
             destination=destination, path=path, data=params
         )
 
+    async def get_room_hierarchy(
+        self,
+        destination: str,
+        room_id: str,
+        suggested_only: bool,
+    ) -> JsonDict:
+        """
+        Args:
+            destination: The remote server
+            room_id: The room ID to ask about.
+            suggested_only: if True, only suggested rooms will be returned
+        """
+        path = _create_path(
+            FEDERATION_UNSTABLE_PREFIX, "/org.matrix.msc2946/hierarchy/%s", room_id
+        )
+
+        return await self.client.get_json(
+            destination=destination,
+            path=path,
+            args={"suggested_only": "true" if suggested_only else "false"},
+        )
+
 
 def _create_path(federation_prefix: str, path: str, *args: str) -> str:
     """

synapse/federation/transport/server/__init__.py

@@ -0,0 +1,332 @@
+# Copyright 2014-2021 The Matrix.org Foundation C.I.C.
+# Copyright 2020 Sorunome
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import logging
+from typing import Dict, Iterable, List, Optional, Tuple, Type
+
+from typing_extensions import Literal
+
+from synapse.api.errors import FederationDeniedError, SynapseError
+from synapse.federation.transport.server._base import (
+    Authenticator,
+    BaseFederationServlet,
+)
+from synapse.federation.transport.server.federation import FEDERATION_SERVLET_CLASSES
+from synapse.federation.transport.server.groups_local import GROUP_LOCAL_SERVLET_CLASSES
+from synapse.federation.transport.server.groups_server import (
+    GROUP_SERVER_SERVLET_CLASSES,
+)
+from synapse.http.server import HttpServer, JsonResource
+from synapse.http.servlet import (
+    parse_boolean_from_args,
+    parse_integer_from_args,
+    parse_string_from_args,
+)
+from synapse.server import HomeServer
+from synapse.types import JsonDict, ThirdPartyInstanceID
+from synapse.util.ratelimitutils import FederationRateLimiter
+
+logger = logging.getLogger(__name__)
+
+
+class TransportLayerServer(JsonResource):
+    """Handles incoming federation HTTP requests"""
+
+    def __init__(self, hs: HomeServer, servlet_groups: Optional[List[str]] = None):
+        """Initialize the TransportLayerServer
+
+        Will by default register all servlets. For custom behaviour, pass in
+        a list of servlet_groups to register.
+
+        Args:
+            hs: homeserver
+            servlet_groups: List of servlet groups to register.
+                Defaults to ``DEFAULT_SERVLET_GROUPS``.
+        """
+        self.hs = hs
+        self.clock = hs.get_clock()
+        self.servlet_groups = servlet_groups
+
+        super().__init__(hs, canonical_json=False)
+
+        self.authenticator = Authenticator(hs)
+        self.ratelimiter = hs.get_federation_ratelimiter()
+
+        self.register_servlets()
+
+    def register_servlets(self) -> None:
+        register_servlets(
+            self.hs,
+            resource=self,
+            ratelimiter=self.ratelimiter,
+            authenticator=self.authenticator,
+            servlet_groups=self.servlet_groups,
+        )
+
+
+class PublicRoomList(BaseFederationServlet):
+    """
+    Fetch the public room list for this server.
+
+    This API returns information in the same format as /publicRooms on the
+    client API, but will only ever include local public rooms and hence is
+    intended for consumption by other homeservers.
+
+    GET /publicRooms HTTP/1.1
+
+    HTTP/1.1 200 OK
+    Content-Type: application/json
+
+    {
+        "chunk": [
+            {
+                "aliases": [
+                    "#test:localhost"
+                ],
+                "guest_can_join": false,
+                "name": "test room",
+                "num_joined_members": 3,
+                "room_id": "!whkydVegtvatLfXmPN:localhost",
+                "world_readable": false
+            }
+        ],
+        "end": "END",
+        "start": "START"
+    }
+    """
+
+    PATH = "/publicRooms"
+
+    def __init__(
+        self,
+        hs: HomeServer,
+        authenticator: Authenticator,
+        ratelimiter: FederationRateLimiter,
+        server_name: str,
+    ):
+        super().__init__(hs, authenticator, ratelimiter, server_name)
+        self.handler = hs.get_room_list_handler()
+        self.allow_access = hs.config.allow_public_rooms_over_federation
+
+    async def on_GET(
+        self, origin: str, content: Literal[None], query: Dict[bytes, List[bytes]]
+    ) -> Tuple[int, JsonDict]:
+        if not self.allow_access:
+            raise FederationDeniedError(origin)
+
+        limit = parse_integer_from_args(query, "limit", 0)
+        since_token = parse_string_from_args(query, "since", None)
+        include_all_networks = parse_boolean_from_args(
+            query, "include_all_networks", default=False
+        )
+        third_party_instance_id = parse_string_from_args(
+            query, "third_party_instance_id", None
+        )
+
+        if include_all_networks:
+            network_tuple = None
+        elif third_party_instance_id:
+            network_tuple = ThirdPartyInstanceID.from_string(third_party_instance_id)
+        else:
+            network_tuple = ThirdPartyInstanceID(None, None)
+
+        if limit == 0:
+            # zero is a special value which corresponds to no limit.
+            limit = None
+
+        data = await self.handler.get_local_public_room_list(
+            limit, since_token, network_tuple=network_tuple, from_federation=True
+        )
+        return 200, data
+
+    async def on_POST(
+        self, origin: str, content: JsonDict, query: Dict[bytes, List[bytes]]
+    ) -> Tuple[int, JsonDict]:
+        # This implements MSC2197 (Search Filtering over Federation)
+        if not self.allow_access:
+            raise FederationDeniedError(origin)
+
+        limit: Optional[int] = int(content.get("limit", 100))
+        since_token = content.get("since", None)
+        search_filter = content.get("filter", None)
+
+        include_all_networks = content.get("include_all_networks", False)
+        third_party_instance_id = content.get("third_party_instance_id", None)
+
+        if include_all_networks:
+            network_tuple = None
+            if third_party_instance_id is not None:
+                raise SynapseError(
+                    400, "Can't use include_all_networks with an explicit network"
+                )
+        elif third_party_instance_id is None:
+            network_tuple = ThirdPartyInstanceID(None, None)
+        else:
+            network_tuple = ThirdPartyInstanceID.from_string(third_party_instance_id)
+
+        if search_filter is None:
+            logger.warning("Nonefilter")
+
+        if limit == 0:
+            # zero is a special value which corresponds to no limit.
+            limit = None
+
+        data = await self.handler.get_local_public_room_list(
+            limit=limit,
+            since_token=since_token,
+            search_filter=search_filter,
+            network_tuple=network_tuple,
+            from_federation=True,
+        )
+
+        return 200, data
+
+
+class FederationGroupsRenewAttestaionServlet(BaseFederationServlet):
+    """A group or user's server renews their attestation"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/renew_attestation/(?P<user_id>[^/]*)"
+
+    def __init__(
+        self,
+        hs: HomeServer,
+        authenticator: Authenticator,
+        ratelimiter: FederationRateLimiter,
+        server_name: str,
+    ):
+        super().__init__(hs, authenticator, ratelimiter, server_name)
+        self.handler = hs.get_groups_attestation_renewer()
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        user_id: str,
+    ) -> Tuple[int, JsonDict]:
+        # We don't need to check auth here as we check the attestation signatures
+
+        new_content = await self.handler.on_renew_attestation(
+            group_id, user_id, content
+        )
+
+        return 200, new_content
+
+
+class OpenIdUserInfo(BaseFederationServlet):
+    """
+    Exchange a bearer token for information about a user.
+
+    The response format should be compatible with:
+        http://openid.net/specs/openid-connect-core-1_0.html#UserInfoResponse
+
+    GET /openid/userinfo?access_token=ABDEFGH HTTP/1.1
+
+    HTTP/1.1 200 OK
+    Content-Type: application/json
+
+    {
+        "sub": "@userpart:example.org",
+    }
+    """
+
+    PATH = "/openid/userinfo"
+
+    REQUIRE_AUTH = False
+
+    def __init__(
+        self,
+        hs: HomeServer,
+        authenticator: Authenticator,
+        ratelimiter: FederationRateLimiter,
+        server_name: str,
+    ):
+        super().__init__(hs, authenticator, ratelimiter, server_name)
+        self.handler = hs.get_federation_server()
+
+    async def on_GET(
+        self,
+        origin: Optional[str],
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+    ) -> Tuple[int, JsonDict]:
+        token = parse_string_from_args(query, "access_token")
+        if token is None:
+            return (
+                401,
+                {"errcode": "M_MISSING_TOKEN", "error": "Access Token required"},
+            )
+
+        user_id = await self.handler.on_openid_userinfo(token)
+
+        if user_id is None:
+            return (
+                401,
+                {
+                    "errcode": "M_UNKNOWN_TOKEN",
+                    "error": "Access Token unknown or expired",
+                },
+            )
+
+        return 200, {"sub": user_id}
+
+
+DEFAULT_SERVLET_GROUPS: Dict[str, Iterable[Type[BaseFederationServlet]]] = {
+    "federation": FEDERATION_SERVLET_CLASSES,
+    "room_list": (PublicRoomList,),
+    "group_server": GROUP_SERVER_SERVLET_CLASSES,
+    "group_local": GROUP_LOCAL_SERVLET_CLASSES,
+    "group_attestation": (FederationGroupsRenewAttestaionServlet,),
+    "openid": (OpenIdUserInfo,),
+}
+
+
+def register_servlets(
+    hs: HomeServer,
+    resource: HttpServer,
+    authenticator: Authenticator,
+    ratelimiter: FederationRateLimiter,
+    servlet_groups: Optional[Iterable[str]] = None,
+):
+    """Initialize and register servlet classes.
+
+    Will by default register all servlets. For custom behaviour, pass in
+    a list of servlet_groups to register.
+
+    Args:
+        hs: homeserver
+        resource: resource class to register to
+        authenticator: authenticator to use
+        ratelimiter: ratelimiter to use
+        servlet_groups: List of servlet groups to register.
+            Defaults to ``DEFAULT_SERVLET_GROUPS``.
+    """
+    if not servlet_groups:
+        servlet_groups = DEFAULT_SERVLET_GROUPS.keys()
+
+    for servlet_group in servlet_groups:
+        # Skip unknown servlet groups.
+        if servlet_group not in DEFAULT_SERVLET_GROUPS:
+            raise RuntimeError(
+                f"Attempting to register unknown federation servlet: '{servlet_group}'"
+            )
+
+        for servletclass in DEFAULT_SERVLET_GROUPS[servlet_group]:
+            servletclass(
+                hs=hs,
+                authenticator=authenticator,
+                ratelimiter=ratelimiter,
+                server_name=hs.hostname,
+            ).register(resource)

synapse/federation/transport/server/_base.py

@@ -0,0 +1,328 @@
+#  Copyright 2021 The Matrix.org Foundation C.I.C.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+import functools
+import logging
+import re
+
+from synapse.api.errors import Codes, FederationDeniedError, SynapseError
+from synapse.api.urls import FEDERATION_V1_PREFIX
+from synapse.http.servlet import parse_json_object_from_request
+from synapse.logging import opentracing
+from synapse.logging.context import run_in_background
+from synapse.logging.opentracing import (
+    SynapseTags,
+    start_active_span,
+    start_active_span_from_request,
+    tags,
+    whitelisted_homeserver,
+)
+from synapse.server import HomeServer
+from synapse.util.ratelimitutils import FederationRateLimiter
+from synapse.util.stringutils import parse_and_validate_server_name
+
+logger = logging.getLogger(__name__)
+
+
+class AuthenticationError(SynapseError):
+    """There was a problem authenticating the request"""
+
+
+class NoAuthenticationError(AuthenticationError):
+    """The request had no authentication information"""
+
+
+class Authenticator:
+    def __init__(self, hs: HomeServer):
+        self._clock = hs.get_clock()
+        self.keyring = hs.get_keyring()
+        self.server_name = hs.hostname
+        self.store = hs.get_datastore()
+        self.federation_domain_whitelist = hs.config.federation_domain_whitelist
+        self.notifier = hs.get_notifier()
+
+        self.replication_client = None
+        if hs.config.worker.worker_app:
+            self.replication_client = hs.get_tcp_replication()
+
+    # A method just so we can pass 'self' as the authenticator to the Servlets
+    async def authenticate_request(self, request, content):
+        now = self._clock.time_msec()
+        json_request = {
+            "method": request.method.decode("ascii"),
+            "uri": request.uri.decode("ascii"),
+            "destination": self.server_name,
+            "signatures": {},
+        }
+
+        if content is not None:
+            json_request["content"] = content
+
+        origin = None
+
+        auth_headers = request.requestHeaders.getRawHeaders(b"Authorization")
+
+        if not auth_headers:
+            raise NoAuthenticationError(
+                401, "Missing Authorization headers", Codes.UNAUTHORIZED
+            )
+
+        for auth in auth_headers:
+            if auth.startswith(b"X-Matrix"):
+                (origin, key, sig) = _parse_auth_header(auth)
+                json_request["origin"] = origin
+                json_request["signatures"].setdefault(origin, {})[key] = sig
+
+        if (
+            self.federation_domain_whitelist is not None
+            and origin not in self.federation_domain_whitelist
+        ):
+            raise FederationDeniedError(origin)
+
+        if origin is None or not json_request["signatures"]:
+            raise NoAuthenticationError(
+                401, "Missing Authorization headers", Codes.UNAUTHORIZED
+            )
+
+        await self.keyring.verify_json_for_server(
+            origin,
+            json_request,
+            now,
+        )
+
+        logger.debug("Request from %s", origin)
+        request.requester = origin
+
+        # If we get a valid signed request from the other side, its probably
+        # alive
+        retry_timings = await self.store.get_destination_retry_timings(origin)
+        if retry_timings and retry_timings.retry_last_ts:
+            run_in_background(self._reset_retry_timings, origin)
+
+        return origin
+
+    async def _reset_retry_timings(self, origin):
+        try:
+            logger.info("Marking origin %r as up", origin)
+            await self.store.set_destination_retry_timings(origin, None, 0, 0)
+
+            # Inform the relevant places that the remote server is back up.
+            self.notifier.notify_remote_server_up(origin)
+            if self.replication_client:
+                # If we're on a worker we try and inform master about this. The
+                # replication client doesn't hook into the notifier to avoid
+                # infinite loops where we send a `REMOTE_SERVER_UP` command to
+                # master, which then echoes it back to us which in turn pokes
+                # the notifier.
+                self.replication_client.send_remote_server_up(origin)
+
+        except Exception:
+            logger.exception("Error resetting retry timings on %s", origin)
+
+
+def _parse_auth_header(header_bytes):
+    """Parse an X-Matrix auth header
+
+    Args:
+        header_bytes (bytes): header value
+
+    Returns:
+        Tuple[str, str, str]: origin, key id, signature.
+
+    Raises:
+        AuthenticationError if the header could not be parsed
+    """
+    try:
+        header_str = header_bytes.decode("utf-8")
+        params = header_str.split(" ")[1].split(",")
+        param_dict = dict(kv.split("=") for kv in params)
+
+        def strip_quotes(value):
+            if value.startswith('"'):
+                return value[1:-1]
+            else:
+                return value
+
+        origin = strip_quotes(param_dict["origin"])
+
+        # ensure that the origin is a valid server name
+        parse_and_validate_server_name(origin)
+
+        key = strip_quotes(param_dict["key"])
+        sig = strip_quotes(param_dict["sig"])
+        return origin, key, sig
+    except Exception as e:
+        logger.warning(
+            "Error parsing auth header '%s': %s",
+            header_bytes.decode("ascii", "replace"),
+            e,
+        )
+        raise AuthenticationError(
+            400, "Malformed Authorization header", Codes.UNAUTHORIZED
+        )
+
+
+class BaseFederationServlet:
+    """Abstract base class for federation servlet classes.
+
+    The servlet object should have a PATH attribute which takes the form of a regexp to
+    match against the request path (excluding the /federation/v1 prefix).
+
+    The servlet should also implement one or more of on_GET, on_POST, on_PUT, to match
+    the appropriate HTTP method. These methods must be *asynchronous* and have the
+    signature:
+
+        on_<METHOD>(self, origin, content, query, **kwargs)
+
+        With arguments:
+
+            origin (unicode|None): The authenticated server_name of the calling server,
+                unless REQUIRE_AUTH is set to False and authentication failed.
+
+            content (unicode|None): decoded json body of the request. None if the
+                request was a GET.
+
+            query (dict[bytes, list[bytes]]): Query params from the request. url-decoded
+                (ie, '+' and '%xx' are decoded) but note that it is *not* utf8-decoded
+                yet.
+
+            **kwargs (dict[unicode, unicode]): the dict mapping keys to path
+                components as specified in the path match regexp.
+
+        Returns:
+            Optional[Tuple[int, object]]: either (response code, response object) to
+                 return a JSON response, or None if the request has already been handled.
+
+        Raises:
+            SynapseError: to return an error code
+
+            Exception: other exceptions will be caught, logged, and a 500 will be
+                returned.
+    """
+
+    PATH = ""  # Overridden in subclasses, the regex to match against the path.
+
+    REQUIRE_AUTH = True
+
+    PREFIX = FEDERATION_V1_PREFIX  # Allows specifying the API version
+
+    RATELIMIT = True  # Whether to rate limit requests or not
+
+    def __init__(
+        self,
+        hs: HomeServer,
+        authenticator: Authenticator,
+        ratelimiter: FederationRateLimiter,
+        server_name: str,
+    ):
+        self.hs = hs
+        self.authenticator = authenticator
+        self.ratelimiter = ratelimiter
+        self.server_name = server_name
+
+    def _wrap(self, func):
+        authenticator = self.authenticator
+        ratelimiter = self.ratelimiter
+
+        @functools.wraps(func)
+        async def new_func(request, *args, **kwargs):
+            """A callback which can be passed to HttpServer.RegisterPaths
+
+            Args:
+                request (twisted.web.http.Request):
+                *args: unused?
+                **kwargs (dict[unicode, unicode]): the dict mapping keys to path
+                    components as specified in the path match regexp.
+
+            Returns:
+                Tuple[int, object]|None: (response code, response object) as returned by
+                    the callback method. None if the request has already been handled.
+            """
+            content = None
+            if request.method in [b"PUT", b"POST"]:
+                # TODO: Handle other method types? other content types?
+                content = parse_json_object_from_request(request)
+
+            try:
+                origin = await authenticator.authenticate_request(request, content)
+            except NoAuthenticationError:
+                origin = None
+                if self.REQUIRE_AUTH:
+                    logger.warning(
+                        "authenticate_request failed: missing authentication"
+                    )
+                    raise
+            except Exception as e:
+                logger.warning("authenticate_request failed: %s", e)
+                raise
+
+            request_tags = {
+                SynapseTags.REQUEST_ID: request.get_request_id(),
+                tags.SPAN_KIND: tags.SPAN_KIND_RPC_SERVER,
+                tags.HTTP_METHOD: request.get_method(),
+                tags.HTTP_URL: request.get_redacted_uri(),
+                tags.PEER_HOST_IPV6: request.getClientIP(),
+                "authenticated_entity": origin,
+                "servlet_name": request.request_metrics.name,
+            }
+
+            # Only accept the span context if the origin is authenticated
+            # and whitelisted
+            if origin and whitelisted_homeserver(origin):
+                scope = start_active_span_from_request(
+                    request, "incoming-federation-request", tags=request_tags
+                )
+            else:
+                scope = start_active_span(
+                    "incoming-federation-request", tags=request_tags
+                )
+
+            with scope:
+                opentracing.inject_response_headers(request.responseHeaders)
+
+                if origin and self.RATELIMIT:
+                    with ratelimiter.ratelimit(origin) as d:
+                        await d
+                        if request._disconnected:
+                            logger.warning(
+                                "client disconnected before we started processing "
+                                "request"
+                            )
+                            return -1, None
+                        response = await func(
+                            origin, content, request.args, *args, **kwargs
+                        )
+                else:
+                    response = await func(
+                        origin, content, request.args, *args, **kwargs
+                    )
+
+            return response
+
+        return new_func
+
+    def register(self, server):
+        pattern = re.compile("^" + self.PREFIX + self.PATH + "$")
+
+        for method in ("GET", "PUT", "POST"):
+            code = getattr(self, "on_%s" % (method), None)
+            if code is None:
+                continue
+
+            server.register_paths(
+                method,
+                (pattern,),
+                self._wrap(code),
+                self.__class__.__name__,
+            )

synapse/federation/transport/server/federation.py

@@ -0,0 +1,706 @@
+#  Copyright 2021 The Matrix.org Foundation C.I.C.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+import logging
+from typing import Dict, List, Mapping, Optional, Sequence, Tuple, Type, Union
+
+from typing_extensions import Literal
+
+import synapse
+from synapse.api.errors import Codes, SynapseError
+from synapse.api.room_versions import RoomVersions
+from synapse.api.urls import FEDERATION_UNSTABLE_PREFIX, FEDERATION_V2_PREFIX
+from synapse.federation.transport.server._base import (
+    Authenticator,
+    BaseFederationServlet,
+)
+from synapse.http.servlet import (
+    parse_boolean_from_args,
+    parse_integer_from_args,
+    parse_string_from_args,
+    parse_strings_from_args,
+)
+from synapse.server import HomeServer
+from synapse.types import JsonDict
+from synapse.util.ratelimitutils import FederationRateLimiter
+from synapse.util.versionstring import get_version_string
+
+logger = logging.getLogger(__name__)
+
+
+class BaseFederationServerServlet(BaseFederationServlet):
+    """Abstract base class for federation servlet classes which provides a federation server handler.
+
+    See BaseFederationServlet for more information.
+    """
+
+    def __init__(
+        self,
+        hs: HomeServer,
+        authenticator: Authenticator,
+        ratelimiter: FederationRateLimiter,
+        server_name: str,
+    ):
+        super().__init__(hs, authenticator, ratelimiter, server_name)
+        self.handler = hs.get_federation_server()
+
+
+class FederationSendServlet(BaseFederationServerServlet):
+    PATH = "/send/(?P<transaction_id>[^/]*)/?"
+
+    # We ratelimit manually in the handler as we queue up the requests and we
+    # don't want to fill up the ratelimiter with blocked requests.
+    RATELIMIT = False
+
+    # This is when someone is trying to send us a bunch of data.
+    async def on_PUT(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        transaction_id: str,
+    ) -> Tuple[int, JsonDict]:
+        """Called on PUT /send/<transaction_id>/
+
+        Args:
+            transaction_id: The transaction_id associated with this request. This
+                is *not* None.
+
+        Returns:
+            Tuple of `(code, response)`, where
+            `response` is a python dict to be converted into JSON that is
+            used as the response body.
+        """
+        # Parse the request
+        try:
+            transaction_data = content
+
+            logger.debug("Decoded %s: %s", transaction_id, str(transaction_data))
+
+            logger.info(
+                "Received txn %s from %s. (PDUs: %d, EDUs: %d)",
+                transaction_id,
+                origin,
+                len(transaction_data.get("pdus", [])),
+                len(transaction_data.get("edus", [])),
+            )
+
+        except Exception as e:
+            logger.exception(e)
+            return 400, {"error": "Invalid transaction"}
+
+        code, response = await self.handler.on_incoming_transaction(
+            origin, transaction_id, self.server_name, transaction_data
+        )
+
+        return code, response
+
+
+class FederationEventServlet(BaseFederationServerServlet):
+    PATH = "/event/(?P<event_id>[^/]*)/?"
+
+    # This is when someone asks for a data item for a given server data_id pair.
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        event_id: str,
+    ) -> Tuple[int, Union[JsonDict, str]]:
+        return await self.handler.on_pdu_request(origin, event_id)
+
+
+class FederationStateV1Servlet(BaseFederationServerServlet):
+    PATH = "/state/(?P<room_id>[^/]*)/?"
+
+    # This is when someone asks for all data for a given room.
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        return await self.handler.on_room_state_request(
+            origin,
+            room_id,
+            parse_string_from_args(query, "event_id", None, required=False),
+        )
+
+
+class FederationStateIdsServlet(BaseFederationServerServlet):
+    PATH = "/state_ids/(?P<room_id>[^/]*)/?"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        return await self.handler.on_state_ids_request(
+            origin,
+            room_id,
+            parse_string_from_args(query, "event_id", None, required=True),
+        )
+
+
+class FederationBackfillServlet(BaseFederationServerServlet):
+    PATH = "/backfill/(?P<room_id>[^/]*)/?"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        versions = [x.decode("ascii") for x in query[b"v"]]
+        limit = parse_integer_from_args(query, "limit", None)
+
+        if not limit:
+            return 400, {"error": "Did not include limit param"}
+
+        return await self.handler.on_backfill_request(origin, room_id, versions, limit)
+
+
+class FederationQueryServlet(BaseFederationServerServlet):
+    PATH = "/query/(?P<query_type>[^/]*)"
+
+    # This is when we receive a server-server Query
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        query_type: str,
+    ) -> Tuple[int, JsonDict]:
+        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(BaseFederationServerServlet):
+    PATH = "/make_join/(?P<room_id>[^/]*)/(?P<user_id>[^/]*)"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+        user_id: str,
+    ) -> Tuple[int, JsonDict]:
+        """
+        Args:
+            origin: The authenticated server_name of the calling server
+
+            content: (GETs don't have bodies)
+
+            query: Query params from the request.
+
+            **kwargs: the dict mapping keys to path components as specified in
+                the path match regexp.
+
+        Returns:
+            Tuple of (response code, response object)
+        """
+        supported_versions = parse_strings_from_args(query, "ver", encoding="utf-8")
+        if supported_versions is None:
+            supported_versions = ["1"]
+
+        result = await self.handler.on_make_join_request(
+            origin, room_id, user_id, supported_versions=supported_versions
+        )
+        return 200, result
+
+
+class FederationMakeLeaveServlet(BaseFederationServerServlet):
+    PATH = "/make_leave/(?P<room_id>[^/]*)/(?P<user_id>[^/]*)"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+        user_id: str,
+    ) -> Tuple[int, JsonDict]:
+        result = await self.handler.on_make_leave_request(origin, room_id, user_id)
+        return 200, result
+
+
+class FederationV1SendLeaveServlet(BaseFederationServerServlet):
+    PATH = "/send_leave/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+
+    async def on_PUT(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+        event_id: str,
+    ) -> Tuple[int, Tuple[int, JsonDict]]:
+        result = await self.handler.on_send_leave_request(origin, content, room_id)
+        return 200, (200, result)
+
+
+class FederationV2SendLeaveServlet(BaseFederationServerServlet):
+    PATH = "/send_leave/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+
+    PREFIX = FEDERATION_V2_PREFIX
+
+    async def on_PUT(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+        event_id: str,
+    ) -> Tuple[int, JsonDict]:
+        result = await self.handler.on_send_leave_request(origin, content, room_id)
+        return 200, result
+
+
+class FederationMakeKnockServlet(BaseFederationServerServlet):
+    PATH = "/make_knock/(?P<room_id>[^/]*)/(?P<user_id>[^/]*)"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+        user_id: str,
+    ) -> Tuple[int, JsonDict]:
+        # Retrieve the room versions the remote homeserver claims to support
+        supported_versions = parse_strings_from_args(
+            query, "ver", required=True, encoding="utf-8"
+        )
+
+        result = await self.handler.on_make_knock_request(
+            origin, room_id, user_id, supported_versions=supported_versions
+        )
+        return 200, result
+
+
+class FederationV1SendKnockServlet(BaseFederationServerServlet):
+    PATH = "/send_knock/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+
+    async def on_PUT(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+        event_id: str,
+    ) -> Tuple[int, JsonDict]:
+        result = await self.handler.on_send_knock_request(origin, content, room_id)
+        return 200, result
+
+
+class FederationEventAuthServlet(BaseFederationServerServlet):
+    PATH = "/event_auth/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+        event_id: str,
+    ) -> Tuple[int, JsonDict]:
+        return await self.handler.on_event_auth(origin, room_id, event_id)
+
+
+class FederationV1SendJoinServlet(BaseFederationServerServlet):
+    PATH = "/send_join/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+
+    async def on_PUT(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+        event_id: str,
+    ) -> Tuple[int, Tuple[int, JsonDict]]:
+        # TODO(paul): assert that event_id parsed from path actually
+        #   match those given in content
+        result = await self.handler.on_send_join_request(origin, content, room_id)
+        return 200, (200, result)
+
+
+class FederationV2SendJoinServlet(BaseFederationServerServlet):
+    PATH = "/send_join/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+
+    PREFIX = FEDERATION_V2_PREFIX
+
+    async def on_PUT(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+        event_id: str,
+    ) -> Tuple[int, JsonDict]:
+        # TODO(paul): assert that event_id parsed from path actually
+        #   match those given in content
+        result = await self.handler.on_send_join_request(origin, content, room_id)
+        return 200, result
+
+
+class FederationV1InviteServlet(BaseFederationServerServlet):
+    PATH = "/invite/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+
+    async def on_PUT(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+        event_id: str,
+    ) -> Tuple[int, Tuple[int, JsonDict]]:
+        # We don't get a room version, so we have to assume its EITHER v1 or
+        # v2. This is "fine" as the only difference between V1 and V2 is the
+        # state resolution algorithm, and we don't use that for processing
+        # invites
+        result = await self.handler.on_invite_request(
+            origin, content, room_version_id=RoomVersions.V1.identifier
+        )
+
+        # V1 federation API is defined to return a content of `[200, {...}]`
+        # due to a historical bug.
+        return 200, (200, result)
+
+
+class FederationV2InviteServlet(BaseFederationServerServlet):
+    PATH = "/invite/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+
+    PREFIX = FEDERATION_V2_PREFIX
+
+    async def on_PUT(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+        event_id: str,
+    ) -> Tuple[int, JsonDict]:
+        # TODO(paul): assert that room_id/event_id parsed from path actually
+        #   match those given in content
+
+        room_version = content["room_version"]
+        event = content["event"]
+        invite_room_state = content["invite_room_state"]
+
+        # Synapse expects invite_room_state to be in unsigned, as it is in v1
+        # API
+
+        event.setdefault("unsigned", {})["invite_room_state"] = invite_room_state
+
+        result = await self.handler.on_invite_request(
+            origin, event, room_version_id=room_version
+        )
+        return 200, result
+
+
+class FederationThirdPartyInviteExchangeServlet(BaseFederationServerServlet):
+    PATH = "/exchange_third_party_invite/(?P<room_id>[^/]*)"
+
+    async def on_PUT(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        await self.handler.on_exchange_third_party_invite_request(content)
+        return 200, {}
+
+
+class FederationClientKeysQueryServlet(BaseFederationServerServlet):
+    PATH = "/user/keys/query"
+
+    async def on_POST(
+        self, origin: str, content: JsonDict, query: Dict[bytes, List[bytes]]
+    ) -> Tuple[int, JsonDict]:
+        return await self.handler.on_query_client_keys(origin, content)
+
+
+class FederationUserDevicesQueryServlet(BaseFederationServerServlet):
+    PATH = "/user/devices/(?P<user_id>[^/]*)"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        user_id: str,
+    ) -> Tuple[int, JsonDict]:
+        return await self.handler.on_query_user_devices(origin, user_id)
+
+
+class FederationClientKeysClaimServlet(BaseFederationServerServlet):
+    PATH = "/user/keys/claim"
+
+    async def on_POST(
+        self, origin: str, content: JsonDict, query: Dict[bytes, List[bytes]]
+    ) -> Tuple[int, JsonDict]:
+        response = await self.handler.on_claim_client_keys(origin, content)
+        return 200, response
+
+
+class FederationGetMissingEventsServlet(BaseFederationServerServlet):
+    # TODO(paul): Why does this path alone end with "/?" optional?
+    PATH = "/get_missing_events/(?P<room_id>[^/]*)/?"
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        limit = int(content.get("limit", 10))
+        earliest_events = content.get("earliest_events", [])
+        latest_events = content.get("latest_events", [])
+
+        result = await self.handler.on_get_missing_events(
+            origin,
+            room_id=room_id,
+            earliest_events=earliest_events,
+            latest_events=latest_events,
+            limit=limit,
+        )
+
+        return 200, result
+
+
+class On3pidBindServlet(BaseFederationServerServlet):
+    PATH = "/3pid/onbind"
+
+    REQUIRE_AUTH = False
+
+    async def on_POST(
+        self, origin: Optional[str], content: JsonDict, query: Dict[bytes, List[bytes]]
+    ) -> Tuple[int, JsonDict]:
+        if "invites" in content:
+            last_exception = None
+            for invite in content["invites"]:
+                try:
+                    if "signed" not in invite or "token" not in invite["signed"]:
+                        message = (
+                            "Rejecting received notification of third-"
+                            "party invite without signed: %s" % (invite,)
+                        )
+                        logger.info(message)
+                        raise SynapseError(400, message)
+                    await self.handler.exchange_third_party_invite(
+                        invite["sender"],
+                        invite["mxid"],
+                        invite["room_id"],
+                        invite["signed"],
+                    )
+                except Exception as e:
+                    last_exception = e
+            if last_exception:
+                raise last_exception
+        return 200, {}
+
+
+class FederationVersionServlet(BaseFederationServlet):
+    PATH = "/version"
+
+    REQUIRE_AUTH = False
+
+    async def on_GET(
+        self,
+        origin: Optional[str],
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+    ) -> Tuple[int, JsonDict]:
+        return (
+            200,
+            {"server": {"name": "Synapse", "version": get_version_string(synapse)}},
+        )
+
+
+class FederationSpaceSummaryServlet(BaseFederationServlet):
+    PREFIX = FEDERATION_UNSTABLE_PREFIX + "/org.matrix.msc2946"
+    PATH = "/spaces/(?P<room_id>[^/]*)"
+
+    def __init__(
+        self,
+        hs: HomeServer,
+        authenticator: Authenticator,
+        ratelimiter: FederationRateLimiter,
+        server_name: str,
+    ):
+        super().__init__(hs, authenticator, ratelimiter, server_name)
+        self.handler = hs.get_room_summary_handler()
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Mapping[bytes, Sequence[bytes]],
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        suggested_only = parse_boolean_from_args(query, "suggested_only", default=False)
+
+        max_rooms_per_space = parse_integer_from_args(query, "max_rooms_per_space")
+        if max_rooms_per_space is not None and max_rooms_per_space < 0:
+            raise SynapseError(
+                400,
+                "Value for 'max_rooms_per_space' must be a non-negative integer",
+                Codes.BAD_JSON,
+            )
+
+        exclude_rooms = parse_strings_from_args(query, "exclude_rooms", default=[])
+
+        return 200, await self.handler.federation_space_summary(
+            origin, room_id, suggested_only, max_rooms_per_space, exclude_rooms
+        )
+
+    # TODO When switching to the stable endpoint, remove the POST handler.
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Mapping[bytes, Sequence[bytes]],
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        suggested_only = content.get("suggested_only", False)
+        if not isinstance(suggested_only, bool):
+            raise SynapseError(
+                400, "'suggested_only' must be a boolean", Codes.BAD_JSON
+            )
+
+        exclude_rooms = content.get("exclude_rooms", [])
+        if not isinstance(exclude_rooms, list) or any(
+            not isinstance(x, str) for x in exclude_rooms
+        ):
+            raise SynapseError(400, "bad value for 'exclude_rooms'", Codes.BAD_JSON)
+
+        max_rooms_per_space = content.get("max_rooms_per_space")
+        if max_rooms_per_space is not None:
+            if not isinstance(max_rooms_per_space, int):
+                raise SynapseError(
+                    400, "bad value for 'max_rooms_per_space'", Codes.BAD_JSON
+                )
+            if max_rooms_per_space < 0:
+                raise SynapseError(
+                    400,
+                    "Value for 'max_rooms_per_space' must be a non-negative integer",
+                    Codes.BAD_JSON,
+                )
+
+        return 200, await self.handler.federation_space_summary(
+            origin, room_id, suggested_only, max_rooms_per_space, exclude_rooms
+        )
+
+
+class FederationRoomHierarchyServlet(BaseFederationServlet):
+    PREFIX = FEDERATION_UNSTABLE_PREFIX + "/org.matrix.msc2946"
+    PATH = "/hierarchy/(?P<room_id>[^/]*)"
+
+    def __init__(
+        self,
+        hs: HomeServer,
+        authenticator: Authenticator,
+        ratelimiter: FederationRateLimiter,
+        server_name: str,
+    ):
+        super().__init__(hs, authenticator, ratelimiter, server_name)
+        self.handler = hs.get_room_summary_handler()
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Mapping[bytes, Sequence[bytes]],
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        suggested_only = parse_boolean_from_args(query, "suggested_only", default=False)
+        return 200, await self.handler.get_federation_hierarchy(
+            origin, room_id, suggested_only
+        )
+
+
+class RoomComplexityServlet(BaseFederationServlet):
+    """
+    Indicates to other servers how complex (and therefore likely
+    resource-intensive) a public room this server knows about is.
+    """
+
+    PATH = "/rooms/(?P<room_id>[^/]*)/complexity"
+    PREFIX = FEDERATION_UNSTABLE_PREFIX
+
+    def __init__(
+        self,
+        hs: HomeServer,
+        authenticator: Authenticator,
+        ratelimiter: FederationRateLimiter,
+        server_name: str,
+    ):
+        super().__init__(hs, authenticator, ratelimiter, server_name)
+        self._store = self.hs.get_datastore()
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        is_public = await self._store.is_room_world_readable_or_publicly_joinable(
+            room_id
+        )
+
+        if not is_public:
+            raise SynapseError(404, "Room not found", errcode=Codes.INVALID_PARAM)
+
+        complexity = await self._store.get_room_complexity(room_id)
+        return 200, complexity
+
+
+FEDERATION_SERVLET_CLASSES: Tuple[Type[BaseFederationServlet], ...] = (
+    FederationSendServlet,
+    FederationEventServlet,
+    FederationStateV1Servlet,
+    FederationStateIdsServlet,
+    FederationBackfillServlet,
+    FederationQueryServlet,
+    FederationMakeJoinServlet,
+    FederationMakeLeaveServlet,
+    FederationEventServlet,
+    FederationV1SendJoinServlet,
+    FederationV2SendJoinServlet,
+    FederationV1SendLeaveServlet,
+    FederationV2SendLeaveServlet,
+    FederationV1InviteServlet,
+    FederationV2InviteServlet,
+    FederationGetMissingEventsServlet,
+    FederationEventAuthServlet,
+    FederationClientKeysQueryServlet,
+    FederationUserDevicesQueryServlet,
+    FederationClientKeysClaimServlet,
+    FederationThirdPartyInviteExchangeServlet,
+    On3pidBindServlet,
+    FederationVersionServlet,
+    RoomComplexityServlet,
+    FederationSpaceSummaryServlet,
+    FederationRoomHierarchyServlet,
+    FederationV1SendKnockServlet,
+    FederationMakeKnockServlet,
+)

synapse/federation/transport/server/groups_local.py

@@ -0,0 +1,113 @@
+#  Copyright 2021 The Matrix.org Foundation C.I.C.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+from typing import Dict, List, Tuple, Type
+
+from synapse.api.errors import SynapseError
+from synapse.federation.transport.server._base import (
+    Authenticator,
+    BaseFederationServlet,
+)
+from synapse.handlers.groups_local import GroupsLocalHandler
+from synapse.server import HomeServer
+from synapse.types import JsonDict, get_domain_from_id
+from synapse.util.ratelimitutils import FederationRateLimiter
+
+
+class BaseGroupsLocalServlet(BaseFederationServlet):
+    """Abstract base class for federation servlet classes which provides a groups local handler.
+
+    See BaseFederationServlet for more information.
+    """
+
+    def __init__(
+        self,
+        hs: HomeServer,
+        authenticator: Authenticator,
+        ratelimiter: FederationRateLimiter,
+        server_name: str,
+    ):
+        super().__init__(hs, authenticator, ratelimiter, server_name)
+        self.handler = hs.get_groups_local_handler()
+
+
+class FederationGroupsLocalInviteServlet(BaseGroupsLocalServlet):
+    """A group server has invited a local user"""
+
+    PATH = "/groups/local/(?P<group_id>[^/]*)/users/(?P<user_id>[^/]*)/invite"
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        user_id: str,
+    ) -> Tuple[int, JsonDict]:
+        if get_domain_from_id(group_id) != origin:
+            raise SynapseError(403, "group_id doesn't match origin")
+
+        assert isinstance(
+            self.handler, GroupsLocalHandler
+        ), "Workers cannot handle group invites."
+
+        new_content = await self.handler.on_invite(group_id, user_id, content)
+
+        return 200, new_content
+
+
+class FederationGroupsRemoveLocalUserServlet(BaseGroupsLocalServlet):
+    """A group server has removed a local user"""
+
+    PATH = "/groups/local/(?P<group_id>[^/]*)/users/(?P<user_id>[^/]*)/remove"
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        user_id: str,
+    ) -> Tuple[int, None]:
+        if get_domain_from_id(group_id) != origin:
+            raise SynapseError(403, "user_id doesn't match origin")
+
+        assert isinstance(
+            self.handler, GroupsLocalHandler
+        ), "Workers cannot handle group removals."
+
+        await self.handler.user_removed_from_group(group_id, user_id, content)
+
+        return 200, None
+
+
+class FederationGroupsBulkPublicisedServlet(BaseGroupsLocalServlet):
+    """Get roles in a group"""
+
+    PATH = "/get_groups_publicised"
+
+    async def on_POST(
+        self, origin: str, content: JsonDict, query: Dict[bytes, List[bytes]]
+    ) -> Tuple[int, JsonDict]:
+        resp = await self.handler.bulk_get_publicised_groups(
+            content["user_ids"], proxy=False
+        )
+
+        return 200, resp
+
+
+GROUP_LOCAL_SERVLET_CLASSES: Tuple[Type[BaseFederationServlet], ...] = (
+    FederationGroupsLocalInviteServlet,
+    FederationGroupsRemoveLocalUserServlet,
+    FederationGroupsBulkPublicisedServlet,
+)

synapse/federation/transport/server/groups_server.py

@@ -0,0 +1,753 @@
+#  Copyright 2021 The Matrix.org Foundation C.I.C.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+from typing import Dict, List, Tuple, Type
+
+from typing_extensions import Literal
+
+from synapse.api.constants import MAX_GROUP_CATEGORYID_LENGTH, MAX_GROUP_ROLEID_LENGTH
+from synapse.api.errors import Codes, SynapseError
+from synapse.federation.transport.server._base import (
+    Authenticator,
+    BaseFederationServlet,
+)
+from synapse.http.servlet import parse_string_from_args
+from synapse.server import HomeServer
+from synapse.types import JsonDict, get_domain_from_id
+from synapse.util.ratelimitutils import FederationRateLimiter
+
+
+class BaseGroupsServerServlet(BaseFederationServlet):
+    """Abstract base class for federation servlet classes which provides a groups server handler.
+
+    See BaseFederationServlet for more information.
+    """
+
+    def __init__(
+        self,
+        hs: HomeServer,
+        authenticator: Authenticator,
+        ratelimiter: FederationRateLimiter,
+        server_name: str,
+    ):
+        super().__init__(hs, authenticator, ratelimiter, server_name)
+        self.handler = hs.get_groups_server_handler()
+
+
+class FederationGroupsProfileServlet(BaseGroupsServerServlet):
+    """Get/set the basic profile of a group on behalf of a user"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/profile"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        new_content = await self.handler.get_group_profile(group_id, requester_user_id)
+
+        return 200, new_content
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        new_content = await self.handler.update_group_profile(
+            group_id, requester_user_id, content
+        )
+
+        return 200, new_content
+
+
+class FederationGroupsSummaryServlet(BaseGroupsServerServlet):
+    PATH = "/groups/(?P<group_id>[^/]*)/summary"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        new_content = await self.handler.get_group_summary(group_id, requester_user_id)
+
+        return 200, new_content
+
+
+class FederationGroupsRoomsServlet(BaseGroupsServerServlet):
+    """Get the rooms in a group on behalf of a user"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/rooms"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        new_content = await self.handler.get_rooms_in_group(group_id, requester_user_id)
+
+        return 200, new_content
+
+
+class FederationGroupsAddRoomsServlet(BaseGroupsServerServlet):
+    """Add/remove room from group"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/room/(?P<room_id>[^/]*)"
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        new_content = await self.handler.add_room_to_group(
+            group_id, requester_user_id, room_id, content
+        )
+
+        return 200, new_content
+
+    async def on_DELETE(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        new_content = await self.handler.remove_room_from_group(
+            group_id, requester_user_id, room_id
+        )
+
+        return 200, new_content
+
+
+class FederationGroupsAddRoomsConfigServlet(BaseGroupsServerServlet):
+    """Update room config in group"""
+
+    PATH = (
+        "/groups/(?P<group_id>[^/]*)/room/(?P<room_id>[^/]*)"
+        "/config/(?P<config_key>[^/]*)"
+    )
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        room_id: str,
+        config_key: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        result = await self.handler.update_room_in_group(
+            group_id, requester_user_id, room_id, config_key, content
+        )
+
+        return 200, result
+
+
+class FederationGroupsUsersServlet(BaseGroupsServerServlet):
+    """Get the users in a group on behalf of a user"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/users"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        new_content = await self.handler.get_users_in_group(group_id, requester_user_id)
+
+        return 200, new_content
+
+
+class FederationGroupsInvitedUsersServlet(BaseGroupsServerServlet):
+    """Get the users that have been invited to a group"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/invited_users"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        new_content = await self.handler.get_invited_users_in_group(
+            group_id, requester_user_id
+        )
+
+        return 200, new_content
+
+
+class FederationGroupsInviteServlet(BaseGroupsServerServlet):
+    """Ask a group server to invite someone to the group"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/users/(?P<user_id>[^/]*)/invite"
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        user_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        new_content = await self.handler.invite_to_group(
+            group_id, user_id, requester_user_id, content
+        )
+
+        return 200, new_content
+
+
+class FederationGroupsAcceptInviteServlet(BaseGroupsServerServlet):
+    """Accept an invitation from the group server"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/users/(?P<user_id>[^/]*)/accept_invite"
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        user_id: str,
+    ) -> Tuple[int, JsonDict]:
+        if get_domain_from_id(user_id) != origin:
+            raise SynapseError(403, "user_id doesn't match origin")
+
+        new_content = await self.handler.accept_invite(group_id, user_id, content)
+
+        return 200, new_content
+
+
+class FederationGroupsJoinServlet(BaseGroupsServerServlet):
+    """Attempt to join a group"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/users/(?P<user_id>[^/]*)/join"
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        user_id: str,
+    ) -> Tuple[int, JsonDict]:
+        if get_domain_from_id(user_id) != origin:
+            raise SynapseError(403, "user_id doesn't match origin")
+
+        new_content = await self.handler.join_group(group_id, user_id, content)
+
+        return 200, new_content
+
+
+class FederationGroupsRemoveUserServlet(BaseGroupsServerServlet):
+    """Leave or kick a user from the group"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/users/(?P<user_id>[^/]*)/remove"
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        user_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        new_content = await self.handler.remove_user_from_group(
+            group_id, user_id, requester_user_id, content
+        )
+
+        return 200, new_content
+
+
+class FederationGroupsSummaryRoomsServlet(BaseGroupsServerServlet):
+    """Add/remove a room from the group summary, with optional category.
+
+    Matches both:
+        - /groups/:group/summary/rooms/:room_id
+        - /groups/:group/summary/categories/:category/rooms/:room_id
+    """
+
+    PATH = (
+        "/groups/(?P<group_id>[^/]*)/summary"
+        "(/categories/(?P<category_id>[^/]+))?"
+        "/rooms/(?P<room_id>[^/]*)"
+    )
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        category_id: str,
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        if category_id == "":
+            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,
+            requester_user_id,
+            room_id=room_id,
+            category_id=category_id,
+            content=content,
+        )
+
+        return 200, resp
+
+    async def on_DELETE(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        category_id: str,
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        if category_id == "":
+            raise SynapseError(400, "category_id cannot be empty string")
+
+        resp = await self.handler.delete_group_summary_room(
+            group_id, requester_user_id, room_id=room_id, category_id=category_id
+        )
+
+        return 200, resp
+
+
+class FederationGroupsCategoriesServlet(BaseGroupsServerServlet):
+    """Get all categories for a group"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/categories/?"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        resp = await self.handler.get_group_categories(group_id, requester_user_id)
+
+        return 200, resp
+
+
+class FederationGroupsCategoryServlet(BaseGroupsServerServlet):
+    """Add/remove/get a category in a group"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/categories/(?P<category_id>[^/]+)"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        category_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        resp = await self.handler.get_group_category(
+            group_id, requester_user_id, category_id
+        )
+
+        return 200, resp
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        category_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        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
+        )
+
+        return 200, resp
+
+    async def on_DELETE(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        category_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        if category_id == "":
+            raise SynapseError(400, "category_id cannot be empty string")
+
+        resp = await self.handler.delete_group_category(
+            group_id, requester_user_id, category_id
+        )
+
+        return 200, resp
+
+
+class FederationGroupsRolesServlet(BaseGroupsServerServlet):
+    """Get roles in a group"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/roles/?"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        resp = await self.handler.get_group_roles(group_id, requester_user_id)
+
+        return 200, resp
+
+
+class FederationGroupsRoleServlet(BaseGroupsServerServlet):
+    """Add/remove/get a role in a group"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/roles/(?P<role_id>[^/]+)"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        role_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        resp = await self.handler.get_group_role(group_id, requester_user_id, role_id)
+
+        return 200, resp
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        role_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        if role_id == "":
+            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
+        )
+
+        return 200, resp
+
+    async def on_DELETE(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        role_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        if role_id == "":
+            raise SynapseError(400, "role_id cannot be empty string")
+
+        resp = await self.handler.delete_group_role(
+            group_id, requester_user_id, role_id
+        )
+
+        return 200, resp
+
+
+class FederationGroupsSummaryUsersServlet(BaseGroupsServerServlet):
+    """Add/remove a user from the group summary, with optional role.
+
+    Matches both:
+        - /groups/:group/summary/users/:user_id
+        - /groups/:group/summary/roles/:role/users/:user_id
+    """
+
+    PATH = (
+        "/groups/(?P<group_id>[^/]*)/summary"
+        "(/roles/(?P<role_id>[^/]+))?"
+        "/users/(?P<user_id>[^/]*)"
+    )
+
+    async def on_POST(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        role_id: str,
+        user_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        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,
+            user_id=user_id,
+            role_id=role_id,
+            content=content,
+        )
+
+        return 200, resp
+
+    async def on_DELETE(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+        role_id: str,
+        user_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        if role_id == "":
+            raise SynapseError(400, "role_id cannot be empty string")
+
+        resp = await self.handler.delete_group_summary_user(
+            group_id, requester_user_id, user_id=user_id, role_id=role_id
+        )
+
+        return 200, resp
+
+
+class FederationGroupsSettingJoinPolicyServlet(BaseGroupsServerServlet):
+    """Sets whether a group is joinable without an invite or knock"""
+
+    PATH = "/groups/(?P<group_id>[^/]*)/settings/m.join_policy"
+
+    async def on_PUT(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: Dict[bytes, List[bytes]],
+        group_id: str,
+    ) -> Tuple[int, JsonDict]:
+        requester_user_id = parse_string_from_args(
+            query, "requester_user_id", required=True
+        )
+        if get_domain_from_id(requester_user_id) != origin:
+            raise SynapseError(403, "requester_user_id doesn't match origin")
+
+        new_content = await self.handler.set_group_join_policy(
+            group_id, requester_user_id, content
+        )
+
+        return 200, new_content
+
+
+GROUP_SERVER_SERVLET_CLASSES: Tuple[Type[BaseFederationServlet], ...] = (
+    FederationGroupsProfileServlet,
+    FederationGroupsSummaryServlet,
+    FederationGroupsRoomsServlet,
+    FederationGroupsUsersServlet,
+    FederationGroupsInvitedUsersServlet,
+    FederationGroupsInviteServlet,
+    FederationGroupsAcceptInviteServlet,
+    FederationGroupsJoinServlet,
+    FederationGroupsRemoveUserServlet,
+    FederationGroupsSummaryRoomsServlet,
+    FederationGroupsCategoriesServlet,
+    FederationGroupsCategoryServlet,
+    FederationGroupsRolesServlet,
+    FederationGroupsRoleServlet,
+    FederationGroupsSummaryUsersServlet,
+    FederationGroupsAddRoomsServlet,
+    FederationGroupsAddRoomsConfigServlet,
+    FederationGroupsSettingJoinPolicyServlet,
+)

synapse/handlers/auth.py

@@ -73,7 +73,7 @@ from synapse.util.stringutils import base62_encode
 from synapse.util.threepids import canonicalise_email
 
 if TYPE_CHECKING:
-    from synapse.rest.client.v1.login import LoginResponse
+    from synapse.rest.client.login import LoginResponse
     from synapse.server import HomeServer
 
 logger = logging.getLogger(__name__)
@@ -461,7 +461,7 @@ class AuthHandler(BaseHandler):
 
         If no auth flows have been completed successfully, raises an
         InteractiveAuthIncompleteError. To handle this, you can use
-        synapse.rest.client.v2_alpha._base.interactive_auth_handler as a
+        synapse.rest.client._base.interactive_auth_handler as a
         decorator.
 
         Args:
@@ -543,7 +543,7 @@ class AuthHandler(BaseHandler):
             # Note that the registration endpoint explicitly removes the
             # "initial_device_display_name" parameter if it is provided
             # without a "password" parameter. See the changes to
-            # synapse.rest.client.v2_alpha.register.RegisterRestServlet.on_POST
+            # synapse.rest.client.register.RegisterRestServlet.on_POST
             # in commit 544722bad23fc31056b9240189c3cbbbf0ffd3f9.
             if not clientdict:
                 clientdict = session.clientdict
@@ -627,23 +627,28 @@ class AuthHandler(BaseHandler):
 
     async def add_oob_auth(
         self, stagetype: str, authdict: Dict[str, Any], clientip: str
-    ) -> bool:
+    ) -> None:
         """
         Adds the result of out-of-band authentication into an existing auth
         session. Currently used for adding the result of fallback auth.
+
+        Raises:
+            LoginError if the stagetype is unknown or the session is missing.
+            LoginError is raised by check_auth if authentication fails.
         """
         if stagetype not in self.checkers:
-            raise LoginError(400, "", Codes.MISSING_PARAM)
-        if "session" not in authdict:
-            raise LoginError(400, "", Codes.MISSING_PARAM)
-
-        result = await self.checkers[stagetype].check_auth(authdict, clientip)
-        if result:
-            await self.store.mark_ui_auth_stage_complete(
-                authdict["session"], stagetype, result
+            raise LoginError(
+                400, f"Unknown UIA stage type: {stagetype}", Codes.INVALID_PARAM
             )
-            return True
-        return False
+        if "session" not in authdict:
+            raise LoginError(400, "Missing session ID", Codes.MISSING_PARAM)
+
+        # If authentication fails a LoginError is raised. Otherwise, store
+        # the successful result.
+        result = await self.checkers[stagetype].check_auth(authdict, clientip)
+        await self.store.mark_ui_auth_stage_complete(
+            authdict["session"], stagetype, result
+        )
 
     def get_session_id(self, clientdict: Dict[str, Any]) -> Optional[str]:
         """

synapse/handlers/event_auth.py

@@ -213,7 +213,7 @@ class EventAuthHandler:
 
             raise AuthError(
                 403,
-                "You do not belong to any of the required rooms to join this room.",
+                "You do not belong to any of the required rooms/spaces to join this room.",
             )
 
     async def has_restricted_join_rules(

synapse/handlers/federation.py

@@ -220,8 +220,6 @@ class FederationHandler(BaseHandler):
         room_id = pdu.room_id
         event_id = pdu.event_id
 
-        logger.info("handling received PDU: %s", pdu)
-
         # We reprocess pdus when we have seen them only as outliers
         existing = await self.store.get_event(
             event_id, allow_none=True, allow_rejected=True
@@ -229,14 +227,19 @@ class FederationHandler(BaseHandler):
 
         # FIXME: Currently we fetch an event again when we already have it
         # if it has been marked as an outlier.
-
-        already_seen = existing and (
-            not existing.internal_metadata.is_outlier()
-            or pdu.internal_metadata.is_outlier()
-        )
-        if already_seen:
-            logger.debug("Already seen pdu")
-            return
+        if existing:
+            if not existing.internal_metadata.is_outlier():
+                logger.info(
+                    "Ignoring received event %s which we have already seen", event_id
+                )
+                return
+            if pdu.internal_metadata.is_outlier():
+                logger.info(
+                    "Ignoring received outlier %s which we already have as an outlier",
+                    event_id,
+                )
+                return
+            logger.info("De-outliering event %s", event_id)
 
         # do some initial sanity-checking of the event. In particular, make
         # sure it doesn't have hundreds of prev_events or auth_events, which
@@ -282,173 +285,172 @@ class FederationHandler(BaseHandler):
         #  - Fetching any missing prev events to fill in gaps in the graph
         #  - Fetching state if we have a hole in the graph
         if not pdu.internal_metadata.is_outlier():
-            # We only backfill backwards to the min depth.
-            min_depth = await self.get_min_depth_for_context(pdu.room_id)
-
-            logger.debug("min_depth: %d", min_depth)
-
             prevs = set(pdu.prev_event_ids())
             seen = await self.store.have_events_in_timeline(prevs)
+            missing_prevs = prevs - seen
 
-            if min_depth is not None and pdu.depth < min_depth:
-                # This is so that we don't notify the user about this
-                # message, to work around the fact that some events will
-                # reference really really old events we really don't want to
-                # send to the clients.
-                pdu.internal_metadata.outlier = True
-            elif min_depth is not None and pdu.depth > min_depth:
-                missing_prevs = prevs - seen
-                if sent_to_us_directly and missing_prevs:
-                    # If we're missing stuff, ensure we only fetch stuff one
-                    # at a time.
-                    logger.info(
-                        "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)):
+            if missing_prevs:
+                if sent_to_us_directly:
+                    # We only backfill backwards to the min depth.
+                    min_depth = await self.get_min_depth_for_context(pdu.room_id)
+                    logger.debug("min_depth: %d", min_depth)
+
+                    if min_depth is not None and pdu.depth > min_depth:
+                        # If we're missing stuff, ensure we only fetch stuff one
+                        # at a time.
                         logger.info(
-                            "Acquired room lock to fetch %d missing prev_events",
+                            "Acquiring room lock to fetch %d missing prev_events: %s",
                             len(missing_prevs),
+                            shortstr(missing_prevs),
                         )
-
-                        try:
-                            await self._get_missing_events_for_pdu(
-                                origin, pdu, prevs, min_depth
+                        with (await self._room_pdu_linearizer.queue(pdu.room_id)):
+                            logger.info(
+                                "Acquired room lock to fetch %d missing prev_events",
+                                len(missing_prevs),
                             )
-                        except Exception as e:
-                            raise Exception(
-                                "Error fetching missing prev_events for %s: %s"
-                                % (event_id, e)
-                            ) from e
+
+                            try:
+                                await self._get_missing_events_for_pdu(
+                                    origin, pdu, prevs, min_depth
+                                )
+                            except Exception as e:
+                                raise Exception(
+                                    "Error fetching missing prev_events for %s: %s"
+                                    % (event_id, e)
+                                ) from e
 
                         # Update the set of things we've seen after trying to
                         # fetch the missing stuff
                         seen = await self.store.have_events_in_timeline(prevs)
+                        missing_prevs = prevs - seen
 
-                        if not prevs - seen:
+                        if not missing_prevs:
+                            logger.info("Found all missing prev_events")
+
+                    if missing_prevs:
+                        # since this event was pushed to us, it is possible for it to
+                        # become the only forward-extremity in the room, and we would then
+                        # trust its state to be the state for the whole room. This is very
+                        # bad. Further, if the event was pushed to us, there is no excuse
+                        # for us not to have all the prev_events. (XXX: apart from
+                        # min_depth?)
+                        #
+                        # We therefore reject any such events.
+                        logger.warning(
+                            "Rejecting: failed to fetch %d prev events: %s",
+                            len(missing_prevs),
+                            shortstr(missing_prevs),
+                        )
+                        raise FederationError(
+                            "ERROR",
+                            403,
+                            (
+                                "Your server isn't divulging details about prev_events "
+                                "referenced in this event."
+                            ),
+                            affected=pdu.event_id,
+                        )
+
+                else:
+                    # We don't have all of the prev_events for this event.
+                    #
+                    # In this case, we need to fall back to asking another server in the
+                    # federation for the state at this event. That's ok provided we then
+                    # resolve the state against other bits of the DAG before using it (which
+                    # will ensure that you can't just take over a room by sending an event,
+                    # withholding its prev_events, and declaring yourself to be an admin in
+                    # the subsequent state request).
+                    #
+                    # Since we're pulling this event as a missing prev_event, then clearly
+                    # this event is not going to become the only forward-extremity and we are
+                    # guaranteed to resolve its state against our existing forward
+                    # extremities, so that should be fine.
+                    #
+                    # XXX this really feels like it could/should be merged with the above,
+                    # but there is an interaction with min_depth that I'm not really
+                    # following.
+                    logger.info(
+                        "Event %s is missing prev_events %s: calculating state for a "
+                        "backwards extremity",
+                        event_id,
+                        shortstr(missing_prevs),
+                    )
+
+                    # Calculate the state after each of the previous events, and
+                    # resolve them to find the correct state at the current event.
+                    event_map = {event_id: pdu}
+                    try:
+                        # Get the state of the events we know about
+                        ours = await self.state_store.get_state_groups_ids(
+                            room_id, seen
+                        )
+
+                        # state_maps is a list of mappings from (type, state_key) to event_id
+                        state_maps: List[StateMap[str]] = list(ours.values())
+
+                        # we don't need this any more, let's delete it.
+                        del ours
+
+                        # Ask the remote server for the states we don't
+                        # know about
+                        for p in missing_prevs:
                             logger.info(
-                                "Found all missing prev_events",
+                                "Requesting state after missing prev_event %s", p
                             )
 
-            if prevs - seen:
-                # We've still not been able to get all of the prev_events for this event.
-                #
-                # In this case, we need to fall back to asking another server in the
-                # federation for the state at this event. That's ok provided we then
-                # resolve the state against other bits of the DAG before using it (which
-                # will ensure that you can't just take over a room by sending an event,
-                # withholding its prev_events, and declaring yourself to be an admin in
-                # the subsequent state request).
-                #
-                # Now, if we're pulling this event as a missing prev_event, then clearly
-                # this event is not going to become the only forward-extremity and we are
-                # guaranteed to resolve its state against our existing forward
-                # extremities, so that should be fine.
-                #
-                # On the other hand, if this event was pushed to us, it is possible for
-                # it to become the only forward-extremity in the room, and we would then
-                # trust its state to be the state for the whole room. This is very bad.
-                # Further, if the event was pushed to us, there is no excuse for us not to
-                # have all the prev_events. We therefore reject any such events.
-                #
-                # XXX this really feels like it could/should be merged with the above,
-                # but there is an interaction with min_depth that I'm not really
-                # following.
-
-                if sent_to_us_directly:
-                    logger.warning(
-                        "Rejecting: failed to fetch %d prev events: %s",
-                        len(prevs - seen),
-                        shortstr(prevs - seen),
-                    )
-                    raise FederationError(
-                        "ERROR",
-                        403,
-                        (
-                            "Your server isn't divulging details about prev_events "
-                            "referenced in this event."
-                        ),
-                        affected=pdu.event_id,
-                    )
-
-                logger.info(
-                    "Event %s is missing prev_events: calculating state for a "
-                    "backwards extremity",
-                    event_id,
-                )
-
-                # Calculate the state after each of the previous events, and
-                # resolve them to find the correct state at the current event.
-                event_map = {event_id: pdu}
-                try:
-                    # Get the state of the events we know about
-                    ours = await self.state_store.get_state_groups_ids(room_id, seen)
-
-                    # state_maps is a list of mappings from (type, state_key) to event_id
-                    state_maps: List[StateMap[str]] = list(ours.values())
-
-                    # we don't need this any more, let's delete it.
-                    del ours
-
-                    # Ask the remote server for the states we don't
-                    # know about
-                    for p in prevs - seen:
-                        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_after_missing_prev_event(
-                                    origin, room_id, 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_after_missing_prev_event(
+                                        origin, room_id, p
+                                    )
                                 )
-                            )
 
-                            remote_state_map = {
-                                (x.type, x.state_key): x.event_id for x in remote_state
-                            }
-                            state_maps.append(remote_state_map)
+                                remote_state_map = {
+                                    (x.type, x.state_key): x.event_id
+                                    for x in remote_state
+                                }
+                                state_maps.append(remote_state_map)
 
-                            for x in remote_state:
-                                event_map[x.event_id] = x
+                                for x in remote_state:
+                                    event_map[x.event_id] = x
 
-                    room_version = await self.store.get_room_version_id(room_id)
-                    state_map = (
-                        await self._state_resolution_handler.resolve_events_with_store(
+                        room_version = await self.store.get_room_version_id(room_id)
+                        state_map = await self._state_resolution_handler.resolve_events_with_store(
                             room_id,
                             room_version,
                             state_maps,
                             event_map,
                             state_res_store=StateResolutionStore(self.store),
                         )
-                    )
 
-                    # We need to give _process_received_pdu the actual state events
-                    # rather than event ids, so generate that now.
+                        # We need to give _process_received_pdu the actual state events
+                        # rather than event ids, so generate that now.
 
-                    # First though we need to fetch all the events that are in
-                    # state_map, so we can build up the state below.
-                    evs = await self.store.get_events(
-                        list(state_map.values()),
-                        get_prev_content=False,
-                        redact_behaviour=EventRedactBehaviour.AS_IS,
-                    )
-                    event_map.update(evs)
+                        # First though we need to fetch all the events that are in
+                        # state_map, so we can build up the state below.
+                        evs = await self.store.get_events(
+                            list(state_map.values()),
+                            get_prev_content=False,
+                            redact_behaviour=EventRedactBehaviour.AS_IS,
+                        )
+                        event_map.update(evs)
 
-                    state = [event_map[e] for e in state_map.values()]
-                except Exception:
-                    logger.warning(
-                        "Error attempting to resolve state at missing " "prev_events",
-                        exc_info=True,
-                    )
-                    raise FederationError(
-                        "ERROR",
-                        403,
-                        "We can't get valid state history.",
-                        affected=event_id,
-                    )
+                        state = [event_map[e] for e in state_map.values()]
+                    except Exception:
+                        logger.warning(
+                            "Error attempting to resolve state at missing "
+                            "prev_events",
+                            exc_info=True,
+                        )
+                        raise FederationError(
+                            "ERROR",
+                            403,
+                            "We can't get valid state history.",
+                            affected=event_id,
+                        )
 
         # A second round of checks for all events. Check that the event passes auth
         # based on `auth_events`, this allows us to assert that the event would
@@ -556,21 +558,14 @@ class FederationHandler(BaseHandler):
             logger.warning("Failed to get prev_events: %s", e)
             return
 
-        logger.info(
-            "Got %d prev_events: %s",
-            len(missing_events),
-            shortstr(missing_events),
-        )
+        logger.info("Got %d prev_events", len(missing_events))
 
         # We want to sort these by depth so we process them and
         # tell clients about them in order.
         missing_events.sort(key=lambda x: x.depth)
 
         for ev in missing_events:
-            logger.info(
-                "Handling received prev_event %s",
-                ev.event_id,
-            )
+            logger.info("Handling received prev_event %s", ev)
             with nested_logging_context(ev.event_id):
                 try:
                     await self.on_receive_pdu(origin, ev, sent_to_us_directly=False)
@@ -1762,10 +1757,8 @@ class FederationHandler(BaseHandler):
         for p, origin in room_queue:
             try:
                 logger.info(
-                    "Processing queued PDU %s which was received "
-                    "while we were joining %s",
-                    p.event_id,
-                    p.room_id,
+                    "Processing queued PDU %s which was received while we were joining",
+                    p,
                 )
                 with nested_logging_context(p.event_id):
                     await self.on_receive_pdu(origin, p, sent_to_us_directly=True)
@@ -2379,6 +2372,7 @@ class FederationHandler(BaseHandler):
                 not event.internal_metadata.is_outlier()
                 and not backfilled
                 and not context.rejected
+                and (await self.store.get_min_depth(event.room_id)) <= event.depth
             ):
                 await self.action_generator.handle_push_actions_for_event(
                     event, context

synapse/handlers/initial_sync.py

@@ -151,7 +151,7 @@ class InitialSyncHandler(BaseHandler):
             limit = 10
 
         async def handle_room(event: RoomsForUser):
-            d = {
+            d: JsonDict = {
                 "room_id": event.room_id,
                 "membership": event.membership,
                 "visibility": (

synapse/handlers/receipts.py

@@ -188,7 +188,14 @@ class ReceiptEventSource:
 
                 new_users = {}
                 for rr_user_id, user_rr in m_read.items():
-                    hidden = user_rr.get("hidden", None)
+                    try:
+                        hidden = user_rr.get("hidden")
+                    except AttributeError:
+                        # Due to https://github.com/matrix-org/synapse/issues/10376
+                        # there are cases where user_rr is a string, in those cases
+                        # we just ignore the read receipt
+                        continue
+
                     if hidden is not True or rr_user_id == user_id:
                         new_users[rr_user_id] = user_rr.copy()
                         # If hidden has a value replace hidden with the correct prefixed key

synapse/handlers/room_summary.py

@@ -16,17 +16,7 @@ import itertools
 import logging
 import re
 from collections import deque
-from typing import (
-    TYPE_CHECKING,
-    Collection,
-    Dict,
-    Iterable,
-    List,
-    Optional,
-    Sequence,
-    Set,
-    Tuple,
-)
+from typing import TYPE_CHECKING, Dict, Iterable, List, Optional, Sequence, Set, Tuple
 
 import attr
 
@@ -38,9 +28,12 @@ from synapse.api.constants import (
     Membership,
     RoomTypes,
 )
+from synapse.api.errors import AuthError, Codes, NotFoundError, SynapseError
 from synapse.events import EventBase
 from synapse.events.utils import format_event_for_client_v2
 from synapse.types import JsonDict
+from synapse.util.caches.response_cache import ResponseCache
+from synapse.util.stringutils import random_string
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -57,16 +50,73 @@ MAX_ROOMS_PER_SPACE = 50
 MAX_SERVERS_PER_SPACE = 3
 
 
-class SpaceSummaryHandler:
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class _PaginationKey:
+    """The key used to find unique pagination session."""
+
+    # The first three entries match the request parameters (and cannot change
+    # during a pagination session).
+    room_id: str
+    suggested_only: bool
+    max_depth: Optional[int]
+    # The randomly generated token.
+    token: str
+
+
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class _PaginationSession:
+    """The information that is stored for pagination."""
+
+    # The time the pagination session was created, in milliseconds.
+    creation_time_ms: int
+    # The queue of rooms which are still to process.
+    room_queue: List["_RoomQueueEntry"]
+    # A set of rooms which have been processed.
+    processed_rooms: Set[str]
+
+
+class RoomSummaryHandler:
+    # The time a pagination session remains valid for.
+    _PAGINATION_SESSION_VALIDITY_PERIOD_MS = 5 * 60 * 1000
+
     def __init__(self, hs: "HomeServer"):
         self._clock = hs.get_clock()
-        self._auth = hs.get_auth()
         self._event_auth_handler = hs.get_event_auth_handler()
         self._store = hs.get_datastore()
         self._event_serializer = hs.get_event_client_serializer()
         self._server_name = hs.hostname
         self._federation_client = hs.get_federation_client()
 
+        # A map of query information to the current pagination state.
+        #
+        # TODO Allow for multiple workers to share this data.
+        # TODO Expire pagination tokens.
+        self._pagination_sessions: Dict[_PaginationKey, _PaginationSession] = {}
+
+        # If a user tries to fetch the same page multiple times in quick succession,
+        # only process the first attempt and return its result to subsequent requests.
+        self._pagination_response_cache: ResponseCache[
+            Tuple[str, bool, Optional[int], Optional[int], Optional[str]]
+        ] = ResponseCache(
+            hs.get_clock(),
+            "get_room_hierarchy",
+        )
+
+    def _expire_pagination_sessions(self):
+        """Expire pagination session which are old."""
+        expire_before = (
+            self._clock.time_msec() - self._PAGINATION_SESSION_VALIDITY_PERIOD_MS
+        )
+        to_expire = []
+
+        for key, value in self._pagination_sessions.items():
+            if value.creation_time_ms < expire_before:
+                to_expire.append(key)
+
+        for key in to_expire:
+            logger.debug("Expiring pagination session id %s", key)
+            del self._pagination_sessions[key]
+
     async def get_space_summary(
         self,
         requester: str,
@@ -92,9 +142,13 @@ class SpaceSummaryHandler:
         Returns:
             summary dict to return
         """
-        # first of all, check that the user is in the room in question (or it's
-        # world-readable)
-        await self._auth.check_user_in_room_or_world_readable(room_id, requester)
+        # First of all, check that the room is accessible.
+        if not await self._is_local_room_accessible(room_id, requester):
+            raise AuthError(
+                403,
+                "User %s not in room %s, and room previews are disabled"
+                % (requester, room_id),
+            )
 
         # the queue of rooms to process
         room_queue = deque((_RoomQueueEntry(room_id, ()),))
@@ -130,10 +184,10 @@ class SpaceSummaryHandler:
                     requester, None, room_id, suggested_only, max_children
                 )
 
-                events: Collection[JsonDict] = []
+                events: Sequence[JsonDict] = []
                 if room_entry:
                     rooms_result.append(room_entry.room)
-                    events = room_entry.children
+                    events = room_entry.children_state_events
 
                 logger.debug(
                     "Query of local room %s returned events %s",
@@ -168,7 +222,7 @@ class SpaceSummaryHandler:
                         room.pop("allowed_spaces", None)
 
                         rooms_result.append(room)
-                        events.extend(room_entry.children)
+                        events.extend(room_entry.children_state_events)
 
                     # All rooms returned don't need visiting again (even if the user
                     # didn't have access to them).
@@ -207,6 +261,212 @@ class SpaceSummaryHandler:
 
         return {"rooms": rooms_result, "events": events_result}
 
+    async def get_room_hierarchy(
+        self,
+        requester: str,
+        requested_room_id: str,
+        suggested_only: bool = False,
+        max_depth: Optional[int] = None,
+        limit: Optional[int] = None,
+        from_token: Optional[str] = None,
+    ) -> JsonDict:
+        """
+        Implementation of the room hierarchy C-S API.
+
+        Args:
+            requester: The user ID of the user making this request.
+            requested_room_id: The room ID to start the hierarchy at (the "root" room).
+            suggested_only: Whether we should only return children with the "suggested"
+                flag set.
+            max_depth: The maximum depth in the tree to explore, must be a
+                non-negative integer.
+
+                0 would correspond to just the root room, 1 would include just
+                the root room's children, etc.
+            limit: An optional limit on the number of rooms to return per
+                page. Must be a positive integer.
+            from_token: An optional pagination token.
+
+        Returns:
+            The JSON hierarchy dictionary.
+        """
+        # If a user tries to fetch the same page multiple times in quick succession,
+        # only process the first attempt and return its result to subsequent requests.
+        #
+        # This is due to the pagination process mutating internal state, attempting
+        # to process multiple requests for the same page will result in errors.
+        return await self._pagination_response_cache.wrap(
+            (requested_room_id, suggested_only, max_depth, limit, from_token),
+            self._get_room_hierarchy,
+            requester,
+            requested_room_id,
+            suggested_only,
+            max_depth,
+            limit,
+            from_token,
+        )
+
+    async def _get_room_hierarchy(
+        self,
+        requester: str,
+        requested_room_id: str,
+        suggested_only: bool = False,
+        max_depth: Optional[int] = None,
+        limit: Optional[int] = None,
+        from_token: Optional[str] = None,
+    ) -> JsonDict:
+        """See docstring for SpaceSummaryHandler.get_room_hierarchy."""
+
+        # First of all, check that the room is accessible.
+        if not await self._is_local_room_accessible(requested_room_id, requester):
+            raise AuthError(
+                403,
+                "User %s not in room %s, and room previews are disabled"
+                % (requester, requested_room_id),
+            )
+
+        # If this is continuing a previous session, pull the persisted data.
+        if from_token:
+            self._expire_pagination_sessions()
+
+            pagination_key = _PaginationKey(
+                requested_room_id, suggested_only, max_depth, from_token
+            )
+            if pagination_key not in self._pagination_sessions:
+                raise SynapseError(400, "Unknown pagination token", Codes.INVALID_PARAM)
+
+            # Load the previous state.
+            pagination_session = self._pagination_sessions[pagination_key]
+            room_queue = pagination_session.room_queue
+            processed_rooms = pagination_session.processed_rooms
+        else:
+            # The queue of rooms to process, the next room is last on the stack.
+            room_queue = [_RoomQueueEntry(requested_room_id, ())]
+
+            # Rooms we have already processed.
+            processed_rooms = set()
+
+        rooms_result: List[JsonDict] = []
+
+        # Cap the limit to a server-side maximum.
+        if limit is None:
+            limit = MAX_ROOMS
+        else:
+            limit = min(limit, MAX_ROOMS)
+
+        # Iterate through the queue until we reach the limit or run out of
+        # rooms to include.
+        while room_queue and len(rooms_result) < limit:
+            queue_entry = room_queue.pop()
+            room_id = queue_entry.room_id
+            current_depth = queue_entry.depth
+            if room_id in processed_rooms:
+                # already done this room
+                continue
+
+            logger.debug("Processing room %s", room_id)
+
+            # A map of summaries for children rooms that might be returned over
+            # federation. The rationale for caching these and *maybe* using them
+            # is to prefer any information local to the homeserver before trusting
+            # data received over federation.
+            children_room_entries: Dict[str, JsonDict] = {}
+            # A set of room IDs which are children that did not have information
+            # returned over federation and are known to be inaccessible to the
+            # current server. We should not reach out over federation to try to
+            # summarise these rooms.
+            inaccessible_children: Set[str] = set()
+
+            # If the room is known locally, summarise it!
+            is_in_room = await self._store.is_host_joined(room_id, self._server_name)
+            if is_in_room:
+                room_entry = await self._summarize_local_room(
+                    requester,
+                    None,
+                    room_id,
+                    suggested_only,
+                    # TODO Handle max children.
+                    max_children=None,
+                )
+
+            # Otherwise, attempt to use information for federation.
+            else:
+                # A previous call might have included information for this room.
+                # It can be used if either:
+                #
+                # 1. The room is not a space.
+                # 2. The maximum depth has been achieved (since no children
+                #    information is needed).
+                if queue_entry.remote_room and (
+                    queue_entry.remote_room.get("room_type") != RoomTypes.SPACE
+                    or (max_depth is not None and current_depth >= max_depth)
+                ):
+                    room_entry = _RoomEntry(
+                        queue_entry.room_id, queue_entry.remote_room
+                    )
+
+                # If the above isn't true, attempt to fetch the room
+                # information over federation.
+                else:
+                    (
+                        room_entry,
+                        children_room_entries,
+                        inaccessible_children,
+                    ) = await self._summarize_remote_room_hierarchy(
+                        queue_entry,
+                        suggested_only,
+                    )
+
+                # Ensure this room is accessible to the requester (and not just
+                # the homeserver).
+                if room_entry and not await self._is_remote_room_accessible(
+                    requester, queue_entry.room_id, room_entry.room
+                ):
+                    room_entry = None
+
+            # This room has been processed and should be ignored if it appears
+            # elsewhere in the hierarchy.
+            processed_rooms.add(room_id)
+
+            # There may or may not be a room entry based on whether it is
+            # inaccessible to the requesting user.
+            if room_entry:
+                # Add the room (including the stripped m.space.child events).
+                rooms_result.append(room_entry.as_json())
+
+                # If this room is not at the max-depth, check if there are any
+                # children to process.
+                if max_depth is None or current_depth < max_depth:
+                    # The children get added in reverse order so that the next
+                    # room to process, according to the ordering, is the last
+                    # item in the list.
+                    room_queue.extend(
+                        _RoomQueueEntry(
+                            ev["state_key"],
+                            ev["content"]["via"],
+                            current_depth + 1,
+                            children_room_entries.get(ev["state_key"]),
+                        )
+                        for ev in reversed(room_entry.children_state_events)
+                        if ev["type"] == EventTypes.SpaceChild
+                        and ev["state_key"] not in inaccessible_children
+                    )
+
+        result: JsonDict = {"rooms": rooms_result}
+
+        # If there's additional data, generate a pagination token (and persist state).
+        if room_queue:
+            next_batch = random_string(24)
+            result["next_batch"] = next_batch
+            pagination_key = _PaginationKey(
+                requested_room_id, suggested_only, max_depth, next_batch
+            )
+            self._pagination_sessions[pagination_key] = _PaginationSession(
+                self._clock.time_msec(), room_queue, processed_rooms
+            )
+
+        return result
+
     async def federation_space_summary(
         self,
         origin: str,
@@ -261,15 +521,78 @@ class SpaceSummaryHandler:
 
             if room_entry:
                 rooms_result.append(room_entry.room)
-                events_result.extend(room_entry.children)
+                events_result.extend(room_entry.children_state_events)
 
                 # add any children to the queue
                 room_queue.extend(
-                    edge_event["state_key"] for edge_event in room_entry.children
+                    edge_event["state_key"]
+                    for edge_event in room_entry.children_state_events
                 )
 
         return {"rooms": rooms_result, "events": events_result}
 
+    async def get_federation_hierarchy(
+        self,
+        origin: str,
+        requested_room_id: str,
+        suggested_only: bool,
+    ):
+        """
+        Implementation of the room hierarchy Federation API.
+
+        This is similar to get_room_hierarchy, but does not recurse into the space.
+        It also considers whether anyone on the server may be able to access the
+        room, as opposed to whether a specific user can.
+
+        Args:
+            origin: The server requesting the spaces summary.
+            requested_room_id: The room ID to start the hierarchy at (the "root" room).
+            suggested_only: whether we should only return children with the "suggested"
+                flag set.
+
+        Returns:
+            The JSON hierarchy dictionary.
+        """
+        root_room_entry = await self._summarize_local_room(
+            None, origin, requested_room_id, suggested_only, max_children=None
+        )
+        if root_room_entry is None:
+            # Room is inaccessible to the requesting server.
+            raise SynapseError(404, "Unknown room: %s" % (requested_room_id,))
+
+        children_rooms_result: List[JsonDict] = []
+        inaccessible_children: List[str] = []
+
+        # Iterate through each child and potentially add it, but not its children,
+        # to the response.
+        for child_room in root_room_entry.children_state_events:
+            room_id = child_room.get("state_key")
+            assert isinstance(room_id, str)
+            # If the room is unknown, skip it.
+            if not await self._store.is_host_joined(room_id, self._server_name):
+                continue
+
+            room_entry = await self._summarize_local_room(
+                None, origin, room_id, suggested_only, max_children=0
+            )
+            # If the room is accessible, include it in the results.
+            #
+            # Note that only the room summary (without information on children)
+            # is included in the summary.
+            if room_entry:
+                children_rooms_result.append(room_entry.room)
+            #  Otherwise, note that the requesting server shouldn't bother
+            #  trying to summarize this room - they do not have access to it.
+            else:
+                inaccessible_children.append(room_id)
+
+        return {
+            # Include the requested room (including the stripped children events).
+            "room": root_room_entry.as_json(),
+            "children": children_rooms_result,
+            "inaccessible_children": inaccessible_children,
+        }
+
     async def _summarize_local_room(
         self,
         requester: Optional[str],
@@ -303,8 +626,9 @@ class SpaceSummaryHandler:
 
         room_entry = await self._build_room_entry(room_id, for_federation=bool(origin))
 
-        # If the room is not a space, return just the room information.
-        if room_entry.get("room_type") != RoomTypes.SPACE:
+        # If the room is not a space or the children don't matter, return just
+        # the room information.
+        if room_entry.get("room_type") != RoomTypes.SPACE or max_children == 0:
             return _RoomEntry(room_id, room_entry)
 
         # Otherwise, look for child rooms/spaces.
@@ -400,29 +724,82 @@ class SpaceSummaryHandler:
 
         return results
 
+    async def _summarize_remote_room_hierarchy(
+        self, room: "_RoomQueueEntry", suggested_only: bool
+    ) -> Tuple[Optional["_RoomEntry"], Dict[str, JsonDict], Set[str]]:
+        """
+        Request room entries and a list of event entries for a given room by querying a remote server.
+
+        Args:
+            room: The room to summarize.
+            suggested_only: True if only suggested children should be returned.
+                Otherwise, all children are returned.
+
+        Returns:
+            A tuple of:
+                The room entry.
+                Partial room data return over federation.
+                A set of inaccessible children room IDs.
+        """
+        room_id = room.room_id
+        logger.info("Requesting summary for %s via %s", room_id, room.via)
+
+        via = itertools.islice(room.via, MAX_SERVERS_PER_SPACE)
+        try:
+            (
+                room_response,
+                children,
+                inaccessible_children,
+            ) = await self._federation_client.get_room_hierarchy(
+                via,
+                room_id,
+                suggested_only=suggested_only,
+            )
+        except Exception as e:
+            logger.warning(
+                "Unable to get hierarchy of %s via federation: %s",
+                room_id,
+                e,
+                exc_info=logger.isEnabledFor(logging.DEBUG),
+            )
+            return None, {}, set()
+
+        # Map the children to their room ID.
+        children_by_room_id = {
+            c["room_id"]: c
+            for c in children
+            if "room_id" in c and isinstance(c["room_id"], str)
+        }
+
+        return (
+            _RoomEntry(room_id, room_response, room_response.pop("children_state", ())),
+            children_by_room_id,
+            set(inaccessible_children),
+        )
+
     async def _is_local_room_accessible(
-        self, room_id: str, requester: Optional[str], origin: Optional[str]
+        self, room_id: str, requester: Optional[str], origin: Optional[str] = None
     ) -> bool:
         """
-        Calculate whether the room should be shown in the spaces summary.
+        Calculate whether the room should be shown to the requester.
 
-        It should be included if:
+        It should return true if:
 
         * The requester is joined or can join the room (per MSC3173).
         * The origin server has any user that is joined or can join the room.
         * The history visibility is set to world readable.
 
         Args:
-            room_id: The room ID to summarize.
+            room_id: The room ID to check accessibility of.
             requester:
-                The user requesting the summary, if it is a local request. None
-                if this is a federation request.
+                The user making the request, if it is a local request.
+                None if this is a federation request.
             origin:
-                The server requesting the summary, if it is a federation request.
+                The server making the request, if it is a federation request.
                 None if this is a local request.
 
         Returns:
-             True if the room should be included in the spaces summary.
+             True if the room is accessible to the requesting user or server.
         """
         state_ids = await self._store.get_current_state_ids(room_id)
 
@@ -516,9 +893,9 @@ class SpaceSummaryHandler:
         self, requester: str, room_id: str, room: JsonDict
     ) -> bool:
         """
-        Calculate whether the room received over federation should be shown in the spaces summary.
+        Calculate whether the room received over federation should be shown to the requester.
 
-        It should be included if:
+        It should return true if:
 
         * The requester is joined or can join the room (per MSC3173).
         * The history visibility is set to world readable.
@@ -530,10 +907,10 @@ class SpaceSummaryHandler:
         Args:
             requester: The user requesting the summary.
             room_id: The room ID returned over federation.
-            room: The summary of the child room returned over federation.
+            room: The summary of the room returned over federation.
 
         Returns:
-            True if the room should be included in the spaces summary.
+            True if the room is accessible to the requesting user.
         """
         # The API doesn't return the room version so assume that a
         # join rule of knock is valid.
@@ -555,11 +932,11 @@ class SpaceSummaryHandler:
         # Finally, check locally if we can access the room. The user might
         # already be in the room (if it was a child room), or there might be a
         # pending invite, etc.
-        return await self._is_local_room_accessible(room_id, requester, None)
+        return await self._is_local_room_accessible(room_id, requester)
 
     async def _build_room_entry(self, room_id: str, for_federation: bool) -> JsonDict:
         """
-        Generate en entry suitable for the 'rooms' list in the summary response.
+        Generate en entry summarising a single room.
 
         Args:
             room_id: The room ID to summarize.
@@ -572,7 +949,7 @@ class SpaceSummaryHandler:
         stats = await self._store.get_room_with_stats(room_id)
 
         # currently this should be impossible because we call
-        # check_user_in_room_or_world_readable on the room before we get here, so
+        # _is_local_room_accessible on the room before we get here, so
         # there should always be an entry
         assert stats is not None, "unable to retrieve stats for %s" % (room_id,)
 
@@ -647,11 +1024,74 @@ class SpaceSummaryHandler:
         # and order to ensure we return stable results.
         return sorted(filter(_has_valid_via, events), key=_child_events_comparison_key)
 
+    async def get_room_summary(
+        self,
+        requester: Optional[str],
+        room_id: str,
+        remote_room_hosts: Optional[List[str]] = None,
+    ) -> JsonDict:
+        """
+        Implementation of the room summary C-S API from MSC3266
+
+        Args:
+            requester:  user id of the user making this request, will be None
+                for unauthenticated requests
+
+            room_id: room id to summarise.
+
+            remote_room_hosts: a list of homeservers to try fetching data through
+                if we don't know it ourselves
+
+        Returns:
+            summary dict to return
+        """
+        is_in_room = await self._store.is_host_joined(room_id, self._server_name)
+
+        if is_in_room:
+            room_entry = await self._summarize_local_room(
+                requester,
+                None,
+                room_id,
+                # Suggested-only doesn't matter since no children are requested.
+                suggested_only=False,
+                max_children=0,
+            )
+
+            if not room_entry:
+                raise NotFoundError("Room not found or is not accessible")
+
+            room_summary = room_entry.room
+
+            # If there was a requester, add their membership.
+            if requester:
+                (
+                    membership,
+                    _,
+                ) = await self._store.get_local_current_membership_for_user_in_room(
+                    requester, room_id
+                )
+
+                room_summary["membership"] = membership or "leave"
+        else:
+            # TODO federation API, descoped from initial unstable implementation
+            #      as MSC needs more maturing on that side.
+            raise SynapseError(400, "Federation is not currently supported.")
+
+        return room_summary
+
 
 @attr.s(frozen=True, slots=True, auto_attribs=True)
 class _RoomQueueEntry:
+    # The room ID of this entry.
     room_id: str
+    # The server to query if the room is not known locally.
     via: Sequence[str]
+    # The minimum number of hops necessary to get to this room (compared to the
+    # originally requested room).
+    depth: int = 0
+    # The room summary for this room returned via federation. This will only be
+    # used if the room is not known locally (and is not a space).
+    remote_room: Optional[JsonDict] = None
 
 
 @attr.s(frozen=True, slots=True, auto_attribs=True)
@@ -662,7 +1102,18 @@ class _RoomEntry:
     # An iterable of the sorted, stripped children events for children of this room.
     #
     # This may not include all children.
-    children: Collection[JsonDict] = ()
+    children_state_events: Sequence[JsonDict] = ()
+
+    def as_json(self) -> JsonDict:
+        """
+        Returns a JSON dictionary suitable for the room hierarchy endpoint.
+
+        It returns the room summary including the stripped m.space.child events
+        as a sub-key.
+        """
+        result = dict(self.room)
+        result["children_state"] = self.children_state_events
+        return result
 
 
 def _has_valid_via(e: EventBase) -> bool:

synapse/handlers/sync.py

@@ -86,20 +86,20 @@ LAZY_LOADED_MEMBERS_CACHE_MAX_SIZE = 100
 SyncRequestKey = Tuple[Any, ...]
 
 
-@attr.s(slots=True, frozen=True)
+@attr.s(slots=True, frozen=True, auto_attribs=True)
 class SyncConfig:
-    user = attr.ib(type=UserID)
-    filter_collection = attr.ib(type=FilterCollection)
-    is_guest = attr.ib(type=bool)
-    request_key = attr.ib(type=SyncRequestKey)
-    device_id = attr.ib(type=Optional[str])
+    user: UserID
+    filter_collection: FilterCollection
+    is_guest: bool
+    request_key: SyncRequestKey
+    device_id: Optional[str]
 
 
-@attr.s(slots=True, frozen=True)
+@attr.s(slots=True, frozen=True, auto_attribs=True)
 class TimelineBatch:
-    prev_batch = attr.ib(type=StreamToken)
-    events = attr.ib(type=List[EventBase])
-    limited = attr.ib(type=bool)
+    prev_batch: StreamToken
+    events: List[EventBase]
+    limited: bool
 
     def __bool__(self) -> bool:
         """Make the result appear empty if there are no updates. This is used
@@ -113,16 +113,16 @@ class TimelineBatch:
 # if there are updates for it, which we check after the instance has been created.
 # This should not be a big deal because we update the notification counts afterwards as
 # well anyway.
-@attr.s(slots=True)
+@attr.s(slots=True, auto_attribs=True)
 class JoinedSyncResult:
-    room_id = attr.ib(type=str)
-    timeline = attr.ib(type=TimelineBatch)
-    state = attr.ib(type=StateMap[EventBase])
-    ephemeral = attr.ib(type=List[JsonDict])
-    account_data = attr.ib(type=List[JsonDict])
-    unread_notifications = attr.ib(type=JsonDict)
-    summary = attr.ib(type=Optional[JsonDict])
-    unread_count = attr.ib(type=int)
+    room_id: str
+    timeline: TimelineBatch
+    state: StateMap[EventBase]
+    ephemeral: List[JsonDict]
+    account_data: List[JsonDict]
+    unread_notifications: JsonDict
+    summary: Optional[JsonDict]
+    unread_count: int
 
     def __bool__(self) -> bool:
         """Make the result appear empty if there are no updates. This is used
@@ -138,12 +138,12 @@ class JoinedSyncResult:
         )
 
 
-@attr.s(slots=True, frozen=True)
+@attr.s(slots=True, frozen=True, auto_attribs=True)
 class ArchivedSyncResult:
-    room_id = attr.ib(type=str)
-    timeline = attr.ib(type=TimelineBatch)
-    state = attr.ib(type=StateMap[EventBase])
-    account_data = attr.ib(type=List[JsonDict])
+    room_id: str
+    timeline: TimelineBatch
+    state: StateMap[EventBase]
+    account_data: List[JsonDict]
 
     def __bool__(self) -> bool:
         """Make the result appear empty if there are no updates. This is used
@@ -152,37 +152,37 @@ class ArchivedSyncResult:
         return bool(self.timeline or self.state or self.account_data)
 
 
-@attr.s(slots=True, frozen=True)
+@attr.s(slots=True, frozen=True, auto_attribs=True)
 class InvitedSyncResult:
-    room_id = attr.ib(type=str)
-    invite = attr.ib(type=EventBase)
+    room_id: str
+    invite: EventBase
 
     def __bool__(self) -> bool:
         """Invited rooms should always be reported to the client"""
         return True
 
 
-@attr.s(slots=True, frozen=True)
+@attr.s(slots=True, frozen=True, auto_attribs=True)
 class KnockedSyncResult:
-    room_id = attr.ib(type=str)
-    knock = attr.ib(type=EventBase)
+    room_id: str
+    knock: EventBase
 
     def __bool__(self) -> bool:
         """Knocked rooms should always be reported to the client"""
         return True
 
 
-@attr.s(slots=True, frozen=True)
+@attr.s(slots=True, frozen=True, auto_attribs=True)
 class GroupsSyncResult:
-    join = attr.ib(type=JsonDict)
-    invite = attr.ib(type=JsonDict)
-    leave = attr.ib(type=JsonDict)
+    join: JsonDict
+    invite: JsonDict
+    leave: JsonDict
 
     def __bool__(self) -> bool:
         return bool(self.join or self.invite or self.leave)
 
 
-@attr.s(slots=True, frozen=True)
+@attr.s(slots=True, frozen=True, auto_attribs=True)
 class DeviceLists:
     """
     Attributes:
@@ -190,27 +190,27 @@ class DeviceLists:
         left: List of user_ids whose devices we no longer track
     """
 
-    changed = attr.ib(type=Collection[str])
-    left = attr.ib(type=Collection[str])
+    changed: Collection[str]
+    left: Collection[str]
 
     def __bool__(self) -> bool:
         return bool(self.changed or self.left)
 
 
-@attr.s(slots=True)
+@attr.s(slots=True, auto_attribs=True)
 class _RoomChanges:
     """The set of room entries to include in the sync, plus the set of joined
     and left room IDs since last sync.
     """
 
-    room_entries = attr.ib(type=List["RoomSyncResultBuilder"])
-    invited = attr.ib(type=List[InvitedSyncResult])
-    knocked = attr.ib(type=List[KnockedSyncResult])
-    newly_joined_rooms = attr.ib(type=List[str])
-    newly_left_rooms = attr.ib(type=List[str])
+    room_entries: List["RoomSyncResultBuilder"]
+    invited: List[InvitedSyncResult]
+    knocked: List[KnockedSyncResult]
+    newly_joined_rooms: List[str]
+    newly_left_rooms: List[str]
 
 
-@attr.s(slots=True, frozen=True)
+@attr.s(slots=True, frozen=True, auto_attribs=True)
 class SyncResult:
     """
     Attributes:
@@ -230,18 +230,18 @@ class SyncResult:
         groups: Group updates, if any
     """
 
-    next_batch = attr.ib(type=StreamToken)
-    presence = attr.ib(type=List[JsonDict])
-    account_data = attr.ib(type=List[JsonDict])
-    joined = attr.ib(type=List[JoinedSyncResult])
-    invited = attr.ib(type=List[InvitedSyncResult])
-    knocked = attr.ib(type=List[KnockedSyncResult])
-    archived = attr.ib(type=List[ArchivedSyncResult])
-    to_device = attr.ib(type=List[JsonDict])
-    device_lists = attr.ib(type=DeviceLists)
-    device_one_time_keys_count = attr.ib(type=JsonDict)
-    device_unused_fallback_key_types = attr.ib(type=List[str])
-    groups = attr.ib(type=Optional[GroupsSyncResult])
+    next_batch: StreamToken
+    presence: List[JsonDict]
+    account_data: List[JsonDict]
+    joined: List[JoinedSyncResult]
+    invited: List[InvitedSyncResult]
+    knocked: List[KnockedSyncResult]
+    archived: List[ArchivedSyncResult]
+    to_device: List[JsonDict]
+    device_lists: DeviceLists
+    device_one_time_keys_count: JsonDict
+    device_unused_fallback_key_types: List[str]
+    groups: Optional[GroupsSyncResult]
 
     def __bool__(self) -> bool:
         """Make the result appear empty if there are no updates. This is used
@@ -701,7 +701,7 @@ class SyncHandler:
         name_id = state_ids.get((EventTypes.Name, ""))
         canonical_alias_id = state_ids.get((EventTypes.CanonicalAlias, ""))
 
-        summary = {}
+        summary: JsonDict = {}
         empty_ms = MemberSummary([], 0)
 
         # TODO: only send these when they change.
@@ -2076,21 +2076,23 @@ class SyncHandler:
         # If the membership's stream ordering is after the given stream
         # ordering, we need to go and work out if the user was in the room
         # before.
-        for room_id, event_pos in joined_rooms:
-            if not event_pos.persisted_after(room_key):
-                joined_room_ids.add(room_id)
+        for joined_room in joined_rooms:
+            if not joined_room.event_pos.persisted_after(room_key):
+                joined_room_ids.add(joined_room.room_id)
                 continue
 
-            logger.info("User joined room after current token: %s", room_id)
+            logger.info("User joined room after current token: %s", joined_room.room_id)
 
             extrems = (
                 await self.store.get_forward_extremities_for_room_at_stream_ordering(
-                    room_id, event_pos.stream
+                    joined_room.room_id, joined_room.event_pos.stream
                 )
             )
-            users_in_room = await self.state.get_current_users_in_room(room_id, extrems)
+            users_in_room = await self.state.get_current_users_in_room(
+                joined_room.room_id, extrems
+            )
             if user_id in users_in_room:
-                joined_room_ids.add(room_id)
+                joined_room_ids.add(joined_room.room_id)
 
         return frozenset(joined_room_ids)
 
@@ -2160,7 +2162,7 @@ def _calculate_state(
     return {event_id_to_key[e]: e for e in state_ids}
 
 
-@attr.s(slots=True)
+@attr.s(slots=True, auto_attribs=True)
 class SyncResultBuilder:
     """Used to help build up a new SyncResult for a user
 
@@ -2182,23 +2184,23 @@ class SyncResultBuilder:
         to_device (list)
     """
 
-    sync_config = attr.ib(type=SyncConfig)
-    full_state = attr.ib(type=bool)
-    since_token = attr.ib(type=Optional[StreamToken])
-    now_token = attr.ib(type=StreamToken)
-    joined_room_ids = attr.ib(type=FrozenSet[str])
+    sync_config: SyncConfig
+    full_state: bool
+    since_token: Optional[StreamToken]
+    now_token: StreamToken
+    joined_room_ids: FrozenSet[str]
 
-    presence = attr.ib(type=List[JsonDict], default=attr.Factory(list))
-    account_data = attr.ib(type=List[JsonDict], default=attr.Factory(list))
-    joined = attr.ib(type=List[JoinedSyncResult], default=attr.Factory(list))
-    invited = attr.ib(type=List[InvitedSyncResult], default=attr.Factory(list))
-    knocked = attr.ib(type=List[KnockedSyncResult], default=attr.Factory(list))
-    archived = attr.ib(type=List[ArchivedSyncResult], default=attr.Factory(list))
-    groups = attr.ib(type=Optional[GroupsSyncResult], default=None)
-    to_device = attr.ib(type=List[JsonDict], default=attr.Factory(list))
+    presence: List[JsonDict] = attr.Factory(list)
+    account_data: List[JsonDict] = attr.Factory(list)
+    joined: List[JoinedSyncResult] = attr.Factory(list)
+    invited: List[InvitedSyncResult] = attr.Factory(list)
+    knocked: List[KnockedSyncResult] = attr.Factory(list)
+    archived: List[ArchivedSyncResult] = attr.Factory(list)
+    groups: Optional[GroupsSyncResult] = None
+    to_device: List[JsonDict] = attr.Factory(list)
 
 
-@attr.s(slots=True)
+@attr.s(slots=True, auto_attribs=True)
 class RoomSyncResultBuilder:
     """Stores information needed to create either a `JoinedSyncResult` or
     `ArchivedSyncResult`.
@@ -2214,10 +2216,10 @@ class RoomSyncResultBuilder:
         upto_token: Latest point to return events from.
     """
 
-    room_id = attr.ib(type=str)
-    rtype = attr.ib(type=str)
-    events = attr.ib(type=Optional[List[EventBase]])
-    newly_joined = attr.ib(type=bool)
-    full_state = attr.ib(type=bool)
-    since_token = attr.ib(type=Optional[StreamToken])
-    upto_token = attr.ib(type=StreamToken)
+    room_id: str
+    rtype: str
+    events: Optional[List[EventBase]]
+    newly_joined: bool
+    full_state: bool
+    since_token: Optional[StreamToken]
+    upto_token: StreamToken

synapse/handlers/ui_auth/checkers.py

@@ -49,7 +49,7 @@ class UserInteractiveAuthChecker:
             clientip: The IP address of the client.
 
         Raises:
-            SynapseError if authentication failed
+            LoginError if authentication failed.
 
         Returns:
             The result of authentication (to pass back to the client?)
@@ -131,7 +131,9 @@ class RecaptchaAuthChecker(UserInteractiveAuthChecker):
             )
             if resp_body["success"]:
                 return True
-        raise LoginError(401, "", errcode=Codes.UNAUTHORIZED)
+        raise LoginError(
+            401, "Captcha authentication failed", errcode=Codes.UNAUTHORIZED
+        )
 
 
 class _BaseThreepidAuthChecker:
@@ -191,7 +193,9 @@ class _BaseThreepidAuthChecker:
             raise AssertionError("Unrecognized threepid medium: %s" % (medium,))
 
         if not threepid:
-            raise LoginError(401, "", errcode=Codes.UNAUTHORIZED)
+            raise LoginError(
+                401, "Unable to get validated threepid", errcode=Codes.UNAUTHORIZED
+            )
 
         if threepid["medium"] != medium:
             raise LoginError(

synapse/http/connectproxyclient.py

@@ -12,8 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import base64
 import logging
+from typing import Optional
 
+import attr
 from zope.interface import implementer
 
 from twisted.internet import defer, protocol
@@ -21,7 +24,6 @@ from twisted.internet.error import ConnectError
 from twisted.internet.interfaces import IReactorCore, IStreamClientEndpoint
 from twisted.internet.protocol import ClientFactory, Protocol, connectionDone
 from twisted.web import http
-from twisted.web.http_headers import Headers
 
 logger = logging.getLogger(__name__)
 
@@ -30,6 +32,22 @@ class ProxyConnectError(ConnectError):
     pass
 
 
+@attr.s
+class ProxyCredentials:
+    username_password = attr.ib(type=bytes)
+
+    def as_proxy_authorization_value(self) -> bytes:
+        """
+        Return the value for a Proxy-Authorization header (i.e. 'Basic abdef==').
+
+        Returns:
+            A transformation of the authentication string the encoded value for
+            a Proxy-Authorization header.
+        """
+        # Encode as base64 and prepend the authorization type
+        return b"Basic " + base64.encodebytes(self.username_password)
+
+
 @implementer(IStreamClientEndpoint)
 class HTTPConnectProxyEndpoint:
     """An Endpoint implementation which will send a CONNECT request to an http proxy
@@ -46,7 +64,7 @@ class HTTPConnectProxyEndpoint:
         proxy_endpoint: the endpoint to use to connect to the proxy
         host: hostname that we want to CONNECT to
         port: port that we want to connect to
-        headers: Extra HTTP headers to include in the CONNECT request
+        proxy_creds: credentials to authenticate at proxy
     """
 
     def __init__(
@@ -55,20 +73,20 @@ class HTTPConnectProxyEndpoint:
         proxy_endpoint: IStreamClientEndpoint,
         host: bytes,
         port: int,
-        headers: Headers,
+        proxy_creds: Optional[ProxyCredentials],
     ):
         self._reactor = reactor
         self._proxy_endpoint = proxy_endpoint
         self._host = host
         self._port = port
-        self._headers = headers
+        self._proxy_creds = proxy_creds
 
     def __repr__(self):
         return "<HTTPConnectProxyEndpoint %s>" % (self._proxy_endpoint,)
 
     def connect(self, protocolFactory: ClientFactory):
         f = HTTPProxiedClientFactory(
-            self._host, self._port, protocolFactory, self._headers
+            self._host, self._port, protocolFactory, self._proxy_creds
         )
         d = self._proxy_endpoint.connect(f)
         # once the tcp socket connects successfully, we need to wait for the
@@ -87,7 +105,7 @@ class HTTPProxiedClientFactory(protocol.ClientFactory):
         dst_host: hostname that we want to CONNECT to
         dst_port: port that we want to connect to
         wrapped_factory: The original Factory
-        headers: Extra HTTP headers to include in the CONNECT request
+        proxy_creds: credentials to authenticate at proxy
     """
 
     def __init__(
@@ -95,12 +113,12 @@ class HTTPProxiedClientFactory(protocol.ClientFactory):
         dst_host: bytes,
         dst_port: int,
         wrapped_factory: ClientFactory,
-        headers: Headers,
+        proxy_creds: Optional[ProxyCredentials],
     ):
         self.dst_host = dst_host
         self.dst_port = dst_port
         self.wrapped_factory = wrapped_factory
-        self.headers = headers
+        self.proxy_creds = proxy_creds
         self.on_connection = defer.Deferred()
 
     def startedConnecting(self, connector):
@@ -114,7 +132,7 @@ class HTTPProxiedClientFactory(protocol.ClientFactory):
             self.dst_port,
             wrapped_protocol,
             self.on_connection,
-            self.headers,
+            self.proxy_creds,
         )
 
     def clientConnectionFailed(self, connector, reason):
@@ -145,7 +163,7 @@ class HTTPConnectProtocol(protocol.Protocol):
         connected_deferred: a Deferred which will be callbacked with
             wrapped_protocol when the CONNECT completes
 
-        headers: Extra HTTP headers to include in the CONNECT request
+        proxy_creds: credentials to authenticate at proxy
     """
 
     def __init__(
@@ -154,16 +172,16 @@ class HTTPConnectProtocol(protocol.Protocol):
         port: int,
         wrapped_protocol: Protocol,
         connected_deferred: defer.Deferred,
-        headers: Headers,
+        proxy_creds: Optional[ProxyCredentials],
     ):
         self.host = host
         self.port = port
         self.wrapped_protocol = wrapped_protocol
         self.connected_deferred = connected_deferred
-        self.headers = headers
+        self.proxy_creds = proxy_creds
 
         self.http_setup_client = HTTPConnectSetupClient(
-            self.host, self.port, self.headers
+            self.host, self.port, self.proxy_creds
         )
         self.http_setup_client.on_connected.addCallback(self.proxyConnected)
 
@@ -205,30 +223,38 @@ class HTTPConnectSetupClient(http.HTTPClient):
     Args:
         host: The hostname to send in the CONNECT message
         port: The port to send in the CONNECT message
-        headers: Extra headers to send with the CONNECT message
+        proxy_creds: credentials to authenticate at proxy
     """
 
-    def __init__(self, host: bytes, port: int, headers: Headers):
+    def __init__(
+        self,
+        host: bytes,
+        port: int,
+        proxy_creds: Optional[ProxyCredentials],
+    ):
         self.host = host
         self.port = port
-        self.headers = headers
+        self.proxy_creds = proxy_creds
         self.on_connected = defer.Deferred()
 
     def connectionMade(self):
         logger.debug("Connected to proxy, sending CONNECT")
         self.sendCommand(b"CONNECT", b"%s:%d" % (self.host, self.port))
 
-        # Send any additional specified headers
-        for name, values in self.headers.getAllRawHeaders():
-            for value in values:
-                self.sendHeader(name, value)
+        # Determine whether we need to set Proxy-Authorization headers
+        if self.proxy_creds:
+            # Set a Proxy-Authorization header
+            self.sendHeader(
+                b"Proxy-Authorization",
+                self.proxy_creds.as_proxy_authorization_value(),
+            )
 
         self.endHeaders()
 
     def handleStatus(self, version: bytes, status: bytes, message: bytes):
         logger.debug("Got Status: %s %s %s", status, message, version)
         if status != b"200":
-            raise ProxyConnectError("Unexpected status on CONNECT: %s" % status)
+            raise ProxyConnectError(f"Unexpected status on CONNECT: {status!s}")
 
     def handleEndHeaders(self):
         logger.debug("End Headers")

synapse/http/federation/matrix_federation_agent.py

@@ -14,6 +14,10 @@
 import logging
 import urllib.parse
 from typing import Any, Generator, List, Optional
+from urllib.request import (  # type: ignore[attr-defined]
+    getproxies_environment,
+    proxy_bypass_environment,
+)
 
 from netaddr import AddrFormatError, IPAddress, IPSet
 from zope.interface import implementer
@@ -30,9 +34,12 @@ from twisted.web.http_headers import Headers
 from twisted.web.iweb import IAgent, IAgentEndpointFactory, IBodyProducer, IResponse
 
 from synapse.crypto.context_factory import FederationPolicyForHTTPS
-from synapse.http.client import BlacklistingAgentWrapper
+from synapse.http import proxyagent
+from synapse.http.client import BlacklistingAgentWrapper, BlacklistingReactorWrapper
+from synapse.http.connectproxyclient import HTTPConnectProxyEndpoint
 from synapse.http.federation.srv_resolver import Server, SrvResolver
 from synapse.http.federation.well_known_resolver import WellKnownResolver
+from synapse.http.proxyagent import ProxyAgent
 from synapse.logging.context import make_deferred_yieldable, run_in_background
 from synapse.types import ISynapseReactor
 from synapse.util import Clock
@@ -57,6 +64,14 @@ class MatrixFederationAgent:
         user_agent:
             The user agent header to use for federation requests.
 
+        ip_whitelist: Allowed IP addresses.
+
+        ip_blacklist: Disallowed IP addresses.
+
+        proxy_reactor: twisted reactor to use for connections to the proxy server
+           reactor might have some blacklisting applied (i.e. for DNS queries),
+           but we need unblocked access to the proxy.
+
         _srv_resolver:
             SrvResolver implementation to use for looking up SRV records. None
             to use a default implementation.
@@ -71,11 +86,18 @@ class MatrixFederationAgent:
         reactor: ISynapseReactor,
         tls_client_options_factory: Optional[FederationPolicyForHTTPS],
         user_agent: bytes,
+        ip_whitelist: IPSet,
         ip_blacklist: IPSet,
         _srv_resolver: Optional[SrvResolver] = None,
         _well_known_resolver: Optional[WellKnownResolver] = None,
     ):
-        self._reactor = reactor
+        # proxy_reactor is not blacklisted
+        proxy_reactor = reactor
+
+        # We need to use a DNS resolver which filters out blacklisted IP
+        # addresses, to prevent DNS rebinding.
+        reactor = BlacklistingReactorWrapper(reactor, ip_whitelist, ip_blacklist)
+
         self._clock = Clock(reactor)
         self._pool = HTTPConnectionPool(reactor)
         self._pool.retryAutomatically = False
@@ -83,24 +105,27 @@ class MatrixFederationAgent:
         self._pool.cachedConnectionTimeout = 2 * 60
 
         self._agent = Agent.usingEndpointFactory(
-            self._reactor,
+            reactor,
             MatrixHostnameEndpointFactory(
-                reactor, tls_client_options_factory, _srv_resolver
+                reactor,
+                proxy_reactor,
+                tls_client_options_factory,
+                _srv_resolver,
             ),
             pool=self._pool,
         )
         self.user_agent = user_agent
 
         if _well_known_resolver is None:
-            # Note that the name resolver has already been wrapped in a
-            # IPBlacklistingResolver by MatrixFederationHttpClient.
             _well_known_resolver = WellKnownResolver(
-                self._reactor,
+                reactor,
                 agent=BlacklistingAgentWrapper(
-                    Agent(
-                        self._reactor,
+                    ProxyAgent(
+                        reactor,
+                        proxy_reactor,
                         pool=self._pool,
                         contextFactory=tls_client_options_factory,
+                        use_proxy=True,
                     ),
                     ip_blacklist=ip_blacklist,
                 ),
@@ -200,10 +225,12 @@ class MatrixHostnameEndpointFactory:
     def __init__(
         self,
         reactor: IReactorCore,
+        proxy_reactor: IReactorCore,
         tls_client_options_factory: Optional[FederationPolicyForHTTPS],
         srv_resolver: Optional[SrvResolver],
     ):
         self._reactor = reactor
+        self._proxy_reactor = proxy_reactor
         self._tls_client_options_factory = tls_client_options_factory
 
         if srv_resolver is None:
@@ -211,9 +238,10 @@ class MatrixHostnameEndpointFactory:
 
         self._srv_resolver = srv_resolver
 
-    def endpointForURI(self, parsed_uri):
+    def endpointForURI(self, parsed_uri: URI):
         return MatrixHostnameEndpoint(
             self._reactor,
+            self._proxy_reactor,
             self._tls_client_options_factory,
             self._srv_resolver,
             parsed_uri,
@@ -227,23 +255,45 @@ class MatrixHostnameEndpoint:
 
     Args:
         reactor: twisted reactor to use for underlying requests
+        proxy_reactor: twisted reactor to use for connections to the proxy server.
+           'reactor' might have some blacklisting applied (i.e. for DNS queries),
+           but we need unblocked access to the proxy.
         tls_client_options_factory:
             factory to use for fetching client tls options, or none to disable TLS.
         srv_resolver: The SRV resolver to use
         parsed_uri: The parsed URI that we're wanting to connect to.
+
+    Raises:
+        ValueError if the environment variables contain an invalid proxy specification.
+        RuntimeError if no tls_options_factory is given for a https connection
     """
 
     def __init__(
         self,
         reactor: IReactorCore,
+        proxy_reactor: IReactorCore,
         tls_client_options_factory: Optional[FederationPolicyForHTTPS],
         srv_resolver: SrvResolver,
         parsed_uri: URI,
     ):
         self._reactor = reactor
-
         self._parsed_uri = parsed_uri
 
+        # http_proxy is not needed because federation is always over TLS
+        proxies = getproxies_environment()
+        https_proxy = proxies["https"].encode() if "https" in proxies else None
+        self.no_proxy = proxies["no"] if "no" in proxies else None
+
+        # endpoint and credentials to use to connect to the outbound https proxy, if any.
+        (
+            self._https_proxy_endpoint,
+            self._https_proxy_creds,
+        ) = proxyagent.http_proxy_endpoint(
+            https_proxy,
+            proxy_reactor,
+            tls_client_options_factory,
+        )
+
         # set up the TLS connection params
         #
         # XXX disabling TLS is really only supported here for the benefit of the
@@ -273,9 +323,33 @@ class MatrixHostnameEndpoint:
             host = server.host
             port = server.port
 
+            should_skip_proxy = False
+            if self.no_proxy is not None:
+                should_skip_proxy = proxy_bypass_environment(
+                    host.decode(),
+                    proxies={"no": self.no_proxy},
+                )
+
+            endpoint: IStreamClientEndpoint
             try:
-                logger.debug("Connecting to %s:%i", host.decode("ascii"), port)
-                endpoint = HostnameEndpoint(self._reactor, host, port)
+                if self._https_proxy_endpoint and not should_skip_proxy:
+                    logger.debug(
+                        "Connecting to %s:%i via %s",
+                        host.decode("ascii"),
+                        port,
+                        self._https_proxy_endpoint,
+                    )
+                    endpoint = HTTPConnectProxyEndpoint(
+                        self._reactor,
+                        self._https_proxy_endpoint,
+                        host,
+                        port,
+                        proxy_creds=self._https_proxy_creds,
+                    )
+                else:
+                    logger.debug("Connecting to %s:%i", host.decode("ascii"), port)
+                    # not using a proxy
+                    endpoint = HostnameEndpoint(self._reactor, host, port)
                 if self._tls_options:
                     endpoint = wrapClientTLS(self._tls_options, endpoint)
                 result = await make_deferred_yieldable(

synapse/http/matrixfederationclient.py

@@ -59,7 +59,6 @@ from synapse.api.errors import (
 from synapse.http import QuieterFileBodyProducer
 from synapse.http.client import (
     BlacklistingAgentWrapper,
-    BlacklistingReactorWrapper,
     BodyExceededMaxSize,
     ByteWriteable,
     encode_query_args,
@@ -69,7 +68,7 @@ from synapse.http.federation.matrix_federation_agent import MatrixFederationAgen
 from synapse.logging import opentracing
 from synapse.logging.context import make_deferred_yieldable
 from synapse.logging.opentracing import set_tag, start_active_span, tags
-from synapse.types import ISynapseReactor, JsonDict
+from synapse.types import JsonDict
 from synapse.util import json_decoder
 from synapse.util.async_helpers import timeout_deferred
 from synapse.util.metrics import Measure
@@ -325,13 +324,7 @@ class MatrixFederationHttpClient:
         self.signing_key = hs.signing_key
         self.server_name = hs.hostname
 
-        # We need to use a DNS resolver which filters out blacklisted IP
-        # addresses, to prevent DNS rebinding.
-        self.reactor: ISynapseReactor = BlacklistingReactorWrapper(
-            hs.get_reactor(),
-            hs.config.federation_ip_range_whitelist,
-            hs.config.federation_ip_range_blacklist,
-        )
+        self.reactor = hs.get_reactor()
 
         user_agent = hs.version_string
         if hs.config.user_agent_suffix:
@@ -342,6 +335,7 @@ class MatrixFederationHttpClient:
             self.reactor,
             tls_client_options_factory,
             user_agent,
+            hs.config.federation_ip_range_whitelist,
             hs.config.federation_ip_range_blacklist,
         )
 

synapse/http/proxyagent.py

@@ -11,7 +11,6 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-import base64
 import logging
 import re
 from typing import Any, Dict, Optional, Tuple
@@ -21,7 +20,6 @@ from urllib.request import (  # type: ignore[attr-defined]
     proxy_bypass_environment,
 )
 
-import attr
 from zope.interface import implementer
 
 from twisted.internet import defer
@@ -38,7 +36,7 @@ from twisted.web.error import SchemeNotSupported
 from twisted.web.http_headers import Headers
 from twisted.web.iweb import IAgent, IBodyProducer, IPolicyForHTTPS
 
-from synapse.http.connectproxyclient import HTTPConnectProxyEndpoint
+from synapse.http.connectproxyclient import HTTPConnectProxyEndpoint, ProxyCredentials
 from synapse.types import ISynapseReactor
 
 logger = logging.getLogger(__name__)
@@ -46,22 +44,6 @@ logger = logging.getLogger(__name__)
 _VALID_URI = re.compile(br"\A[\x21-\x7e]+\Z")
 
 
-@attr.s
-class ProxyCredentials:
-    username_password = attr.ib(type=bytes)
-
-    def as_proxy_authorization_value(self) -> bytes:
-        """
-        Return the value for a Proxy-Authorization header (i.e. 'Basic abdef==').
-
-        Returns:
-            A transformation of the authentication string the encoded value for
-            a Proxy-Authorization header.
-        """
-        # Encode as base64 and prepend the authorization type
-        return b"Basic " + base64.encodebytes(self.username_password)
-
-
 @implementer(IAgent)
 class ProxyAgent(_AgentBase):
     """An Agent implementation which will use an HTTP proxy if one was requested
@@ -95,6 +77,7 @@ class ProxyAgent(_AgentBase):
     Raises:
         ValueError if use_proxy is set and the environment variables
             contain an invalid proxy specification.
+        RuntimeError if no tls_options_factory is given for a https connection
     """
 
     def __init__(
@@ -131,11 +114,11 @@ class ProxyAgent(_AgentBase):
             https_proxy = proxies["https"].encode() if "https" in proxies else None
             no_proxy = proxies["no"] if "no" in proxies else None
 
-        self.http_proxy_endpoint, self.http_proxy_creds = _http_proxy_endpoint(
+        self.http_proxy_endpoint, self.http_proxy_creds = http_proxy_endpoint(
             http_proxy, self.proxy_reactor, contextFactory, **self._endpoint_kwargs
         )
 
-        self.https_proxy_endpoint, self.https_proxy_creds = _http_proxy_endpoint(
+        self.https_proxy_endpoint, self.https_proxy_creds = http_proxy_endpoint(
             https_proxy, self.proxy_reactor, contextFactory, **self._endpoint_kwargs
         )
 
@@ -224,22 +207,12 @@ class ProxyAgent(_AgentBase):
             and self.https_proxy_endpoint
             and not should_skip_proxy
         ):
-            connect_headers = Headers()
-
-            # Determine whether we need to set Proxy-Authorization headers
-            if self.https_proxy_creds:
-                # Set a Proxy-Authorization header
-                connect_headers.addRawHeader(
-                    b"Proxy-Authorization",
-                    self.https_proxy_creds.as_proxy_authorization_value(),
-                )
-
             endpoint = HTTPConnectProxyEndpoint(
                 self.proxy_reactor,
                 self.https_proxy_endpoint,
                 parsed_uri.host,
                 parsed_uri.port,
-                headers=connect_headers,
+                self.https_proxy_creds,
             )
         else:
             # not using a proxy
@@ -268,10 +241,10 @@ class ProxyAgent(_AgentBase):
         )
 
 
-def _http_proxy_endpoint(
+def http_proxy_endpoint(
     proxy: Optional[bytes],
     reactor: IReactorCore,
-    tls_options_factory: IPolicyForHTTPS,
+    tls_options_factory: Optional[IPolicyForHTTPS],
     **kwargs,
 ) -> Tuple[Optional[IStreamClientEndpoint], Optional[ProxyCredentials]]:
     """Parses an http proxy setting and returns an endpoint for the proxy
@@ -294,6 +267,7 @@ def _http_proxy_endpoint(
 
     Raise:
         ValueError if proxy has no hostname or unsupported scheme.
+        RuntimeError if no tls_options_factory is given for a https connection
     """
     if proxy is None:
         return None, None
@@ -305,8 +279,13 @@ def _http_proxy_endpoint(
     proxy_endpoint = HostnameEndpoint(reactor, host, port, **kwargs)
 
     if scheme == b"https":
-        tls_options = tls_options_factory.creatorForNetloc(host, port)
-        proxy_endpoint = wrapClientTLS(tls_options, proxy_endpoint)
+        if tls_options_factory:
+            tls_options = tls_options_factory.creatorForNetloc(host, port)
+            proxy_endpoint = wrapClientTLS(tls_options, proxy_endpoint)
+        else:
+            raise RuntimeError(
+                f"No TLS options for a https connection via proxy {proxy!s}"
+            )
 
     return proxy_endpoint, credentials
 

synapse/http/servlet.py

@@ -14,16 +14,28 @@
 
 """ This module contains base REST classes for constructing REST servlets. """
 import logging
-from typing import Iterable, List, Mapping, Optional, Sequence, overload
+from typing import (
+    TYPE_CHECKING,
+    Iterable,
+    List,
+    Mapping,
+    Optional,
+    Sequence,
+    Tuple,
+    overload,
+)
 
 from typing_extensions import Literal
 
 from twisted.web.server import Request
 
 from synapse.api.errors import Codes, SynapseError
-from synapse.types import JsonDict
+from synapse.types import JsonDict, RoomAlias, RoomID
 from synapse.util import json_decoder
 
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
 logger = logging.getLogger(__name__)
 
 
@@ -663,3 +675,45 @@ class RestServlet:
 
         else:
             raise NotImplementedError("RestServlet must register something.")
+
+
+class ResolveRoomIdMixin:
+    def __init__(self, hs: "HomeServer"):
+        self.room_member_handler = hs.get_room_member_handler()
+
+    async def resolve_room_id(
+        self, room_identifier: str, remote_room_hosts: Optional[List[str]] = None
+    ) -> Tuple[str, Optional[List[str]]]:
+        """
+        Resolve a room identifier to a room ID, if necessary.
+
+        This also performanes checks to ensure the room ID is of the proper form.
+
+        Args:
+            room_identifier: The room ID or alias.
+            remote_room_hosts: The potential remote room hosts to use.
+
+        Returns:
+            The resolved room ID.
+
+        Raises:
+            SynapseError if the room ID is of the wrong form.
+        """
+        if RoomID.is_valid(room_identifier):
+            resolved_room_id = room_identifier
+        elif RoomAlias.is_valid(room_identifier):
+            room_alias = RoomAlias.from_string(room_identifier)
+            (
+                room_id,
+                remote_room_hosts,
+            ) = await self.room_member_handler.lookup_room_alias(room_alias)
+            resolved_room_id = room_id.to_string()
+        else:
+            raise SynapseError(
+                400, "%s was not legal room ID or room alias" % (room_identifier,)
+            )
+        if not resolved_room_id:
+            raise SynapseError(
+                400, "Unknown room ID or room alias %s" % room_identifier
+            )
+        return resolved_room_id, remote_room_hosts

synapse/module_api/__init__.py

@@ -32,6 +32,7 @@ from twisted.internet import defer
 from twisted.web.resource import IResource
 
 from synapse.events import EventBase
+from synapse.events.presence_router import PresenceRouter
 from synapse.http.client import SimpleHttpClient
 from synapse.http.server import (
     DirectServeHtmlResource,
@@ -57,6 +58,8 @@ This package defines the 'stable' API which can be used by extension modules whi
 are loaded into Synapse.
 """
 
+PRESENCE_ALL_USERS = PresenceRouter.ALL_USERS
+
 __all__ = [
     "errors",
     "make_deferred_yieldable",
@@ -70,6 +73,7 @@ __all__ = [
     "DirectServeHtmlResource",
     "DirectServeJsonResource",
     "ModuleApi",
+    "PRESENCE_ALL_USERS",
 ]
 
 logger = logging.getLogger(__name__)
@@ -111,6 +115,7 @@ class ModuleApi:
         self._spam_checker = hs.get_spam_checker()
         self._account_validity_handler = hs.get_account_validity_handler()
         self._third_party_event_rules = hs.get_third_party_event_rules()
+        self._presence_router = hs.get_presence_router()
 
     #################################################################################
     # The following methods should only be called during the module's initialisation.
@@ -130,6 +135,11 @@ class ModuleApi:
         """Registers callbacks for third party event rules capabilities."""
         return self._third_party_event_rules.register_third_party_rules_callbacks
 
+    @property
+    def register_presence_router_callbacks(self):
+        """Registers callbacks for presence router capabilities."""
+        return self._presence_router.register_presence_router_callbacks
+
     def register_web_resource(self, path: str, resource: IResource):
         """Registers a web resource to be served at the given path.
 
@@ -677,7 +687,10 @@ class ModuleApi:
             A list containing the loaded templates, with the orders matching the one of
             the filenames parameter.
         """
-        return self._hs.config.read_templates(filenames, custom_template_directory)
+        return self._hs.config.read_templates(
+            filenames,
+            (td for td in (custom_template_directory,) if td),
+        )
 
 
 class PublicRoomListManager:

synapse/res/templates/recaptcha.html

@@ -16,6 +16,9 @@ function captchaDone() {
 <body>
 <form id="registrationForm" method="post" action="{{ myurl }}">
     <div>
+        {% if error is defined %}
+            <p class="error"><strong>Error: {{ error }}</strong></p>
+        {% endif %}
         <p>
         Hello! We need to prevent computer programs and other automated
         things from creating accounts on this server.

synapse/res/templates/terms.html

@@ -8,6 +8,9 @@
 <body>
 <form id="registrationForm" method="post" action="{{ myurl }}">
     <div>
+        {% if error is defined %}
+            <p class="error"><strong>Error: {{ error }}</strong></p>
+        {% endif %}
         <p>
             Please click the button below if you agree to the
             <a href="{{ terms_url }}">privacy policy of this homeserver.</a>