diff options
121 files changed, 3495 insertions, 1638 deletions
diff --git a/.buildkite/pipeline.yml b/.buildkite/pipeline.yml index 369a1ffed1..44b258dca6 100644 --- a/.buildkite/pipeline.yml +++ b/.buildkite/pipeline.yml @@ -90,6 +90,17 @@ steps: image: "python:3.7" propagate-environment: true + - command: + - "python -m pip install tox" + - "tox -e py27-old,codecov" + label: ":python: 2.7 / SQLite / Old Deps" + env: + TRIAL_FLAGS: "-j 2" + plugins: + - docker#v3.0.1: + image: "python:2.7" + propagate-environment: true + - label: ":python: 2.7 / :postgres: 9.4" env: TRIAL_FLAGS: "-j 4" diff --git a/.github/ISSUE_TEMPLATE/BUG_REPORT.md b/.github/ISSUE_TEMPLATE/BUG_REPORT.md index 756759c2d8..5cf844bfb1 100644 --- a/.github/ISSUE_TEMPLATE/BUG_REPORT.md +++ b/.github/ISSUE_TEMPLATE/BUG_REPORT.md @@ -4,9 +4,9 @@ about: Create a report to help us improve --- -<!-- +<!-- -**IF YOU HAVE SUPPORT QUESTIONS ABOUT RUNNING OR CONFIGURING YOUR OWN HOME SERVER**: +**IF YOU HAVE SUPPORT QUESTIONS ABOUT RUNNING OR CONFIGURING YOUR OWN HOME SERVER**: You will likely get better support more quickly if you ask in ** #matrix:matrix.org ** ;) @@ -17,7 +17,7 @@ the necessary data to fix your issue. You can also preview your report before submitting it. You may remove sections that aren't relevant to your particular case. -Text between <!-- and --> marks will be invisible in the report. +Text between <!-- and --​> marks will be invisible in the report. --> @@ -31,7 +31,7 @@ Text between <!-- and --> marks will be invisible in the report. - that reproduce the bug - using hyphens as bullet points -<!-- +<!-- Describe how what happens differs from what you expected. If you can identify any relevant log snippets from _homeserver.log_, please include @@ -48,8 +48,8 @@ those (please be careful to remove any personal or private data). Please surroun If not matrix.org: -<!-- -What version of Synapse is running? +<!-- +What version of Synapse is running? You can find the Synapse version by inspecting the server headers (replace matrix.org with your own homeserver domain): $ curl -v https://matrix.org/_matrix/client/versions 2>&1 | grep "Server:" diff --git a/.gitignore b/.gitignore index a20f3e615d..a84c41b0c9 100644 --- a/.gitignore +++ b/.gitignore @@ -12,11 +12,15 @@ _trial_temp/ _trial_temp*/ # stuff that is likely to exist when you run a server locally +/*.db +/*.log +/*.log.config +/*.pid /*.signing.key -/*.tls.crt -/*.tls.key -/uploads +/env/ +/homeserver*.yaml /media_store/ +/uploads # IDEs /.idea/ diff --git a/INSTALL.md b/INSTALL.md index 2993f3a9e2..de6893530d 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -71,7 +71,8 @@ set this to the hostname of your server. For a more production-ready setup, you will probably want to specify your domain (`example.com`) rather than a matrix-specific hostname here (in the same way that your email address is probably `user@example.com` rather than `user@email.example.com`) - but -doing so may require more advanced setup. - see [Setting up Federation](README.rst#setting-up-federation). Beware that the server name cannot be changed later. +doing so may require more advanced setup: see [Setting up Federation](docs/federate.md). +Beware that the server name cannot be changed later. This command will generate you a config file that you can then customise, but it will also generate a set of keys for you. These keys will allow your Home Server to @@ -374,9 +375,16 @@ To configure Synapse to expose an HTTPS port, you will need to edit * You will also need to uncomment the `tls_certificate_path` and `tls_private_key_path` lines under the `TLS` section. You can either point these settings at an existing certificate and key, or you can - enable Synapse's built-in ACME (Let's Encrypt) support. Instructions - for having Synapse automatically provision and renew federation - certificates through ACME can be found at [ACME.md](docs/ACME.md). + enable Synapse's built-in ACME (Let's Encrypt) support. Instructions + for having Synapse automatically provision and renew federation + certificates through ACME can be found at [ACME.md](docs/ACME.md). If you + are using your own certificate, be sure to use a `.pem` file that includes + the full certificate chain including any intermediate certificates (for + instance, if using certbot, use `fullchain.pem` as your certificate, not + `cert.pem`). + +For those of you upgrading your TLS certificate in readiness for Synapse 1.0, +please take a look at `our guide <docs/MSC1711_certificates_FAQ.md#configuring-certificates-for-compatibility-with-synapse-100>`_. ## Registering a user @@ -402,8 +410,8 @@ This process uses a setting `registration_shared_secret` in `homeserver.yaml`, which is shared between Synapse itself and the `register_new_matrix_user` script. It doesn't matter what it is (a random value is generated by `--generate-config`), but it should be kept secret, as -anyone with knowledge of it can register users on your server even if -`enable_registration` is `false`. +anyone with knowledge of it can register users, including admin accounts, +on your server even if `enable_registration` is `false`. ## Setting up a TURN server diff --git a/README.rst b/README.rst index c04fb4d19a..24afb93d7d 100644 --- a/README.rst +++ b/README.rst @@ -80,7 +80,10 @@ Thanks for using Matrix! Synapse Installation ==================== -For details on how to install synapse, see `<INSTALL.md>`_. +.. _federation: + +* For details on how to install synapse, see `<INSTALL.md>`_. +* For specific details on how to configure Synapse for federation see `docs/federate.md <docs/federate.md>`_ Connecting to Synapse from a client @@ -93,13 +96,13 @@ Unless you are running a test instance of Synapse on your local machine, in general, you will need to enable TLS support before you can successfully connect from a client: see `<INSTALL.md#tls-certificates>`_. -An easy way to get started is to login or register via Riot at -https://riot.im/app/#/login or https://riot.im/app/#/register respectively. +An easy way to get started is to login or register via Riot at +https://riot.im/app/#/login or https://riot.im/app/#/register respectively. You will need to change the server you are logging into from ``matrix.org`` -and instead specify a Homeserver URL of ``https://<server_name>:8448`` -(or just ``https://<server_name>`` if you are using a reverse proxy). -(Leave the identity server as the default - see `Identity servers`_.) -If you prefer to use another client, refer to our +and instead specify a Homeserver URL of ``https://<server_name>:8448`` +(or just ``https://<server_name>`` if you are using a reverse proxy). +(Leave the identity server as the default - see `Identity servers`_.) +If you prefer to use another client, refer to our `client breakdown <https://matrix.org/docs/projects/clients-matrix>`_. If all goes well you should at least be able to log in, create a room, and @@ -151,56 +154,6 @@ server on the same domain. See https://github.com/vector-im/riot-web/issues/1977 and https://developer.github.com/changes/2014-04-25-user-content-security for more details. -Troubleshooting -=============== - -Running out of File Handles ---------------------------- - -If synapse runs out of filehandles, it typically fails badly - live-locking -at 100% CPU, and/or failing to accept new TCP connections (blocking the -connecting client). Matrix currently can legitimately use a lot of file handles, -thanks to busy rooms like #matrix:matrix.org containing hundreds of participating -servers. The first time a server talks in a room it will try to connect -simultaneously to all participating servers, which could exhaust the available -file descriptors between DNS queries & HTTPS sockets, especially if DNS is slow -to respond. (We need to improve the routing algorithm used to be better than -full mesh, but as of June 2017 this hasn't happened yet). - -If you hit this failure mode, we recommend increasing the maximum number of -open file handles to be at least 4096 (assuming a default of 1024 or 256). -This is typically done by editing ``/etc/security/limits.conf`` - -Separately, Synapse may leak file handles if inbound HTTP requests get stuck -during processing - e.g. blocked behind a lock or talking to a remote server etc. -This is best diagnosed by matching up the 'Received request' and 'Processed request' -log lines and looking for any 'Processed request' lines which take more than -a few seconds to execute. Please let us know at #synapse:matrix.org if -you see this failure mode so we can help debug it, however. - -Help!! Synapse eats all my RAM! -------------------------------- - -Synapse's architecture is quite RAM hungry currently - we deliberately -cache a lot of recent room data and metadata in RAM in order to speed up -common requests. We'll improve this in future, but for now the easiest -way to either reduce the RAM usage (at the risk of slowing things down) -is to set the almost-undocumented ``SYNAPSE_CACHE_FACTOR`` environment -variable. The default is 0.5, which can be decreased to reduce RAM usage -in memory constrained enviroments, or increased if performance starts to -degrade. - -Using `libjemalloc <http://jemalloc.net/>`_ can also yield a significant -improvement in overall amount, and especially in terms of giving back RAM -to the OS. To use it, the library must simply be put in the LD_PRELOAD -environment variable when launching Synapse. On Debian, this can be done -by installing the ``libjemalloc1`` package and adding this line to -``/etc/default/matrix-synapse``:: - - LD_PRELOAD=/usr/lib/x86_64-linux-gnu/libjemalloc.so.1 - -This can make a significant difference on Python 2.7 - it's unclear how -much of an improvement it provides on Python 3.x. Upgrading an existing Synapse ============================= @@ -211,100 +164,19 @@ versions of synapse. .. _UPGRADE.rst: UPGRADE.rst -.. _federation: - -Setting up Federation -===================== - -Federation is the process by which users on different servers can participate -in the same room. For this to work, those other servers must be able to contact -yours to send messages. - -The ``server_name`` in your ``homeserver.yaml`` file determines the way that -other servers will reach yours. By default, they will treat it as a hostname -and try to connect to port 8448. This is easy to set up and will work with the -default configuration, provided you set the ``server_name`` to match your -machine's public DNS hostname, and give Synapse a TLS certificate which is -valid for your ``server_name``. - -For a more flexible configuration, you can set up a DNS SRV record. This allows -you to run your server on a machine that might not have the same name as your -domain name. For example, you might want to run your server at -``synapse.example.com``, but have your Matrix user-ids look like -``@user:example.com``. (A SRV record also allows you to change the port from -the default 8448). - -To use a SRV record, first create your SRV record and publish it in DNS. This -should have the format ``_matrix._tcp.<yourdomain.com> <ttl> IN SRV 10 0 <port> -<synapse.server.name>``. The DNS record should then look something like:: - - $ dig -t srv _matrix._tcp.example.com - _matrix._tcp.example.com. 3600 IN SRV 10 0 8448 synapse.example.com. - -Note that the server hostname cannot be an alias (CNAME record): it has to point -directly to the server hosting the synapse instance. - -You can then configure your homeserver to use ``<yourdomain.com>`` as the domain in -its user-ids, by setting ``server_name``:: - - python -m synapse.app.homeserver \ - --server-name <yourdomain.com> \ - --config-path homeserver.yaml \ - --generate-config - python -m synapse.app.homeserver --config-path homeserver.yaml - -If you've already generated the config file, you need to edit the ``server_name`` -in your ``homeserver.yaml`` file. If you've already started Synapse and a -database has been created, you will have to recreate the database. - -If all goes well, you should be able to `connect to your server with a client`__, -and then join a room via federation. (Try ``#matrix-dev:matrix.org`` as a first -step. "Matrix HQ"'s sheer size and activity level tends to make even the -largest boxes pause for thought.) - -.. __: `Connecting to Synapse from a client`_ - -Troubleshooting ---------------- - -You can use the `federation tester <https://matrix.org/federationtester>`_ to -check if your homeserver is all set. - -The typical failure mode with federation is that when you try to join a room, -it is rejected with "401: Unauthorized". Generally this means that other -servers in the room couldn't access yours. (Joining a room over federation is a -complicated dance which requires connections in both directions). - -So, things to check are: - -* If you are not using a SRV record, check that your ``server_name`` (the part - of your user-id after the ``:``) matches your hostname, and that port 8448 on - that hostname is reachable from outside your network. -* If you *are* using a SRV record, check that it matches your ``server_name`` - (it should be ``_matrix._tcp.<server_name>``), and that the port and hostname - it specifies are reachable from outside your network. - -Another common problem is that people on other servers can't join rooms that -you invite them to. This can be caused by an incorrectly-configured reverse -proxy: see `<docs/reverse_proxy.rst>`_ for instructions on how to correctly -configure a reverse proxy. - -Running a Demo Federation of Synapses -------------------------------------- - -If you want to get up and running quickly with a trio of homeservers in a -private federation, there is a script in the ``demo`` directory. This is mainly -useful just for development purposes. See `<demo/README>`_. - Using PostgreSQL ================ -As of Synapse 0.9, `PostgreSQL <https://www.postgresql.org>`_ is supported as an -alternative to the `SQLite <https://sqlite.org/>`_ database that Synapse has -traditionally used for convenience and simplicity. +Synapse offers two database engines: + * `SQLite <https://sqlite.org/>`_ + * `PostgreSQL <https://www.postgresql.org>`_ + +By default Synapse uses SQLite in and doing so trades performance for convenience. +SQLite is only recommended in Synapse for testing purposes or for servers with +light workloads. -The advantages of Postgres include: +Almost all installations should opt to use PostreSQL. Advantages include: * significant performance improvements due to the superior threading and caching model, smarter query optimiser @@ -440,3 +312,54 @@ sphinxcontrib-napoleon:: Building internal API documentation:: python setup.py build_sphinx + +Troubleshooting +=============== + +Running out of File Handles +--------------------------- + +If synapse runs out of file handles, it typically fails badly - live-locking +at 100% CPU, and/or failing to accept new TCP connections (blocking the +connecting client). Matrix currently can legitimately use a lot of file handles, +thanks to busy rooms like #matrix:matrix.org containing hundreds of participating +servers. The first time a server talks in a room it will try to connect +simultaneously to all participating servers, which could exhaust the available +file descriptors between DNS queries & HTTPS sockets, especially if DNS is slow +to respond. (We need to improve the routing algorithm used to be better than +full mesh, but as of March 2019 this hasn't happened yet). + +If you hit this failure mode, we recommend increasing the maximum number of +open file handles to be at least 4096 (assuming a default of 1024 or 256). +This is typically done by editing ``/etc/security/limits.conf`` + +Separately, Synapse may leak file handles if inbound HTTP requests get stuck +during processing - e.g. blocked behind a lock or talking to a remote server etc. +This is best diagnosed by matching up the 'Received request' and 'Processed request' +log lines and looking for any 'Processed request' lines which take more than +a few seconds to execute. Please let us know at #synapse:matrix.org if +you see this failure mode so we can help debug it, however. + +Help!! Synapse eats all my RAM! +------------------------------- + +Synapse's architecture is quite RAM hungry currently - we deliberately +cache a lot of recent room data and metadata in RAM in order to speed up +common requests. We'll improve this in the future, but for now the easiest +way to either reduce the RAM usage (at the risk of slowing things down) +is to set the almost-undocumented ``SYNAPSE_CACHE_FACTOR`` environment +variable. The default is 0.5, which can be decreased to reduce RAM usage +in memory constrained enviroments, or increased if performance starts to +degrade. + +Using `libjemalloc <http://jemalloc.net/>`_ can also yield a significant +improvement in overall amount, and especially in terms of giving back RAM +to the OS. To use it, the library must simply be put in the LD_PRELOAD +environment variable when launching Synapse. On Debian, this can be done +by installing the ``libjemalloc1`` package and adding this line to +``/etc/default/matrix-synapse``:: + + LD_PRELOAD=/usr/lib/x86_64-linux-gnu/libjemalloc.so.1 + +This can make a significant difference on Python 2.7 - it's unclear how +much of an improvement it provides on Python 3.x. diff --git a/changelog.d/4662.misc b/changelog.d/4662.misc new file mode 100644 index 0000000000..f4ec0d6a68 --- /dev/null +++ b/changelog.d/4662.misc @@ -0,0 +1 @@ +Add a systemd setup that supports synapse workers. Contributed by Luca Corbatto. diff --git a/changelog.d/4793.feature b/changelog.d/4793.feature deleted file mode 100644 index 90dba7d122..0000000000 --- a/changelog.d/4793.feature +++ /dev/null @@ -1 +0,0 @@ -Synapse is now permissive about trailing slashes on some of its federation endpoints, allowing zero or more to be present. \ No newline at end of file diff --git a/changelog.d/4821.feature b/changelog.d/4821.feature new file mode 100644 index 0000000000..61d4eb8d60 --- /dev/null +++ b/changelog.d/4821.feature @@ -0,0 +1 @@ +Add configurable rate limiting to the /login endpoint. diff --git a/changelog.d/4832.misc b/changelog.d/4832.misc new file mode 100644 index 0000000000..92022266c6 --- /dev/null +++ b/changelog.d/4832.misc @@ -0,0 +1 @@ +Improve federation documentation, specifically .well-known support. Many thanks to @vaab. diff --git a/changelog.d/4838.bugfix b/changelog.d/4838.bugfix new file mode 100644 index 0000000000..7f4fceabff --- /dev/null +++ b/changelog.d/4838.bugfix @@ -0,0 +1 @@ +Transfer a user's notification settings (push rules) on room upgrade. \ No newline at end of file diff --git a/changelog.d/4843.misc b/changelog.d/4843.misc new file mode 100644 index 0000000000..03d0a3e2e7 --- /dev/null +++ b/changelog.d/4843.misc @@ -0,0 +1 @@ +Add stuff back to the .gitignore. diff --git a/changelog.d/4844.misc b/changelog.d/4844.misc new file mode 100644 index 0000000000..eff6f1c43c --- /dev/null +++ b/changelog.d/4844.misc @@ -0,0 +1 @@ +Clarify what registration_shared_secret allows for. diff --git a/changelog.d/4846.feature b/changelog.d/4846.feature new file mode 100644 index 0000000000..8f792b8890 --- /dev/null +++ b/changelog.d/4846.feature @@ -0,0 +1 @@ +The user directory has been rewritten to make it faster, with less chance of falling behind on a large server. diff --git a/changelog.d/4849.misc b/changelog.d/4849.misc new file mode 100644 index 0000000000..f2cab20b44 --- /dev/null +++ b/changelog.d/4849.misc @@ -0,0 +1 @@ +Update install docs to explicitly state a full-chain (not just the top-level) TLS certificate must be provided to Synapse. This caused some people's Synapse ports to appear correct in a browser but still (rightfully so) upset the federation tester. \ No newline at end of file diff --git a/changelog.d/4852.misc b/changelog.d/4852.misc new file mode 100644 index 0000000000..76ab1e34e7 --- /dev/null +++ b/changelog.d/4852.misc @@ -0,0 +1 @@ + Move client read-receipt processing to federation sender worker. \ No newline at end of file diff --git a/changelog.d/4853.feature b/changelog.d/4853.feature new file mode 100644 index 0000000000..360f92e1de --- /dev/null +++ b/changelog.d/4853.feature @@ -0,0 +1 @@ +Allow passing --daemonize flags to workers in the same way as with master. diff --git a/changelog.d/4855.misc b/changelog.d/4855.misc new file mode 100644 index 0000000000..c4906d2f56 --- /dev/null +++ b/changelog.d/4855.misc @@ -0,0 +1 @@ +Refactor federation TransactionQueue. \ No newline at end of file diff --git a/changelog.d/4863.misc b/changelog.d/4863.misc new file mode 100644 index 0000000000..bfe03cbedc --- /dev/null +++ b/changelog.d/4863.misc @@ -0,0 +1 @@ +Comment out most options in the generated config. diff --git a/changelog.d/4864.feature b/changelog.d/4864.feature new file mode 100644 index 0000000000..57927f2620 --- /dev/null +++ b/changelog.d/4864.feature @@ -0,0 +1 @@ +The user directory has been rewritten to make it faster, with less chance of falling behind on a large server. \ No newline at end of file diff --git a/changelog.d/4865.feature b/changelog.d/4865.feature new file mode 100644 index 0000000000..61d4eb8d60 --- /dev/null +++ b/changelog.d/4865.feature @@ -0,0 +1 @@ +Add configurable rate limiting to the /login endpoint. diff --git a/changelog.d/4870.misc b/changelog.d/4870.misc new file mode 100644 index 0000000000..f287b7d3b0 --- /dev/null +++ b/changelog.d/4870.misc @@ -0,0 +1 @@ +Update Apache setup to remove location syntax. Thanks to @cwmke! diff --git a/changelog.d/4879.misc b/changelog.d/4879.misc new file mode 100644 index 0000000000..574017230c --- /dev/null +++ b/changelog.d/4879.misc @@ -0,0 +1 @@ +Reinstate test case that runs unit tests against oldest supported dependencies. diff --git a/changelog.d/4881.misc b/changelog.d/4881.misc new file mode 100644 index 0000000000..308c21c839 --- /dev/null +++ b/changelog.d/4881.misc @@ -0,0 +1 @@ +Update link to federation docs. diff --git a/changelog.d/4886.bugfix b/changelog.d/4886.bugfix new file mode 100644 index 0000000000..b17aa92485 --- /dev/null +++ b/changelog.d/4886.bugfix @@ -0,0 +1 @@ +fix test_auto_create_auto_join_where_no_consent. diff --git a/changelog.d/4886.misc b/changelog.d/4886.misc new file mode 100644 index 0000000000..b17aa92485 --- /dev/null +++ b/changelog.d/4886.misc @@ -0,0 +1 @@ +fix test_auto_create_auto_join_where_no_consent. diff --git a/changelog.d/4887.feature b/changelog.d/4887.feature new file mode 100644 index 0000000000..e7ff0b9297 --- /dev/null +++ b/changelog.d/4887.feature @@ -0,0 +1 @@ +The user directory has been rewritten to make it faster, with less chance of falling behind on a large server. diff --git a/changelog.d/4888.bugfix b/changelog.d/4888.bugfix new file mode 100644 index 0000000000..0e193709e5 --- /dev/null +++ b/changelog.d/4888.bugfix @@ -0,0 +1,2 @@ +Fix a bug where hs_disabled_message was sometimes not correctly enforced. + diff --git a/changelog.d/4889.misc b/changelog.d/4889.misc new file mode 100644 index 0000000000..f1948db65e --- /dev/null +++ b/changelog.d/4889.misc @@ -0,0 +1 @@ +Use a regular HomeServerConfig object for unit tests rater than a Mock. diff --git a/changelog.d/4890.feature b/changelog.d/4890.feature new file mode 100644 index 0000000000..8d74262250 --- /dev/null +++ b/changelog.d/4890.feature @@ -0,0 +1 @@ +Batch up outgoing read-receipts to reduce federation traffic. diff --git a/changelog.d/4895.feature b/changelog.d/4895.feature new file mode 100644 index 0000000000..5dd7c68194 --- /dev/null +++ b/changelog.d/4895.feature @@ -0,0 +1 @@ +Add option to disable searching the user directory. diff --git a/changelog.d/4895.misc b/changelog.d/4895.misc new file mode 100644 index 0000000000..81a3261538 --- /dev/null +++ b/changelog.d/4895.misc @@ -0,0 +1 @@ +Add some notes about tuning postgres for larger deployments. diff --git a/changelog.d/4896.feature b/changelog.d/4896.feature new file mode 100644 index 0000000000..46ac49a4b4 --- /dev/null +++ b/changelog.d/4896.feature @@ -0,0 +1 @@ +Add option to disable searching of local and remote public room lists. diff --git a/changelog.d/4900.feature b/changelog.d/4900.feature new file mode 100644 index 0000000000..8f792b8890 --- /dev/null +++ b/changelog.d/4900.feature @@ -0,0 +1 @@ +The user directory has been rewritten to make it faster, with less chance of falling behind on a large server. diff --git a/changelog.d/4902.misc b/changelog.d/4902.misc new file mode 100644 index 0000000000..fecc06a6e8 --- /dev/null +++ b/changelog.d/4902.misc @@ -0,0 +1 @@ +Add a config option for torture-testing worker replication. diff --git a/changelog.d/4904.bugfix b/changelog.d/4904.bugfix new file mode 100644 index 0000000000..5c2d7f3cf1 --- /dev/null +++ b/changelog.d/4904.bugfix @@ -0,0 +1 @@ +Fix bug in shutdown room admin API where it would fail if a user in the room hadn't consented to the privacy policy. diff --git a/changelog.d/4905.misc b/changelog.d/4905.misc new file mode 100644 index 0000000000..0f00d5a3d5 --- /dev/null +++ b/changelog.d/4905.misc @@ -0,0 +1 @@ +Log requests which are simulated by the unit tests. diff --git a/changelog.d/4908.bugfix b/changelog.d/4908.bugfix new file mode 100644 index 0000000000..d8c5babf0d --- /dev/null +++ b/changelog.d/4908.bugfix @@ -0,0 +1 @@ +Fix bug where blocked world-readable rooms were still peekable. diff --git a/contrib/systemd-with-workers/README.md b/contrib/systemd-with-workers/README.md new file mode 100644 index 0000000000..74b261e9fb --- /dev/null +++ b/contrib/systemd-with-workers/README.md @@ -0,0 +1,150 @@ +# Setup Synapse with Workers and Systemd + +This is a setup for managing synapse with systemd including support for +managing workers. It provides a `matrix-synapse`, as well as a +`matrix-synapse-worker@` service for any workers you require. Additionally to +group the required services it sets up a `matrix.target`. You can use this to +automatically start any bot- or bridge-services. More on this in +[Bots and Bridges](#bots-and-bridges). + +See the folder [system](system) for any service and target files. + +The folder [workers](workers) contains an example configuration for the +`federation_reader` worker. Pay special attention to the name of the +configuration file. In order to work with the `matrix-synapse-worker@.service` +service, it needs to have the exact same name as the worker app. + +This setup expects neither the homeserver nor any workers to fork. Forking is +handled by systemd. + +## Setup + +1. Adjust your matrix configs. Make sure that the worker config files have the +exact same name as the worker app. Compare `matrix-synapse-worker@.service` for +why. You can find an example worker config in the [workers](workers) folder. See +below for relevant settings in the `homeserver.yaml`. +2. Copy the `*.service` and `*.target` files in [system](system) to +`/etc/systemd/system`. +3. `systemctl enable matrix-synapse.service` this adds the homeserver +app to the `matrix.target` +4. *Optional.* `systemctl enable +matrix-synapse-worker@federation_reader.service` this adds the federation_reader +app to the `matrix-synapse.service` +5. *Optional.* Repeat step 4 for any additional workers you require. +6. *Optional.* Add any bots or bridges by enabling them. +7. Start all matrix related services via `systemctl start matrix.target` +8. *Optional.* Enable autostart of all matrix related services on system boot +via `systemctl enable matrix.target` + +## Usage + +After you have setup you can use the following commands to manage your synapse +installation: + +``` +# Start matrix-synapse, all workers and any enabled bots or bridges. +systemctl start matrix.target + +# Restart matrix-synapse and all workers (not necessarily restarting bots +# or bridges, see "Bots and Bridges") +systemctl restart matrix-synapse.service + +# Stop matrix-synapse and all workers (not necessarily restarting bots +# or bridges, see "Bots and Bridges") +systemctl stop matrix-synapse.service + +# Restart a specific worker (i. e. federation_reader), the homeserver is +# unaffected by this. +systemctl restart matrix-synapse-worker@federation_reader.service + +# Add a new worker (assuming all configs are setup already) +systemctl enable matrix-synapse-worker@federation_writer.service +systemctl restart matrix-synapse.service +``` + +## The Configs + +Make sure the `worker_app` is set in the `homeserver.yaml` and it does not fork. + +``` +worker_app: synapse.app.homeserver +daemonize: false +``` + +None of the workers should fork, as forking is handled by systemd. Hence make +sure this is present in all worker config files. + +``` +worker_daemonize: false +``` + +The config files of all workers are expected to be located in +`/etc/matrix-synapse/workers`. If you want to use a different location you have +to edit the provided `*.service` files accordingly. + +## Bots and Bridges + +Most bots and bridges do not care if the homeserver goes down or is restarted. +Depending on the implementation this may crash them though. So look up the docs +or ask the community of the specific bridge or bot you want to run to make sure +you choose the correct setup. + +Whichever configuration you choose, after the setup the following will enable +automatically starting (and potentially restarting) your bot/bridge with the +`matrix.target`. + +``` +systemctl enable <yourBotOrBridgeName>.service +``` + +**Note** that from an inactive synapse the bots/bridges will only be started with +synapse if you start the `matrix.target`, not if you start the +`matrix-synapse.service`. This is on purpose. Think of `matrix-synapse.service` +as *just* synapse, but `matrix.target` being anything matrix related, including +synapse and any and all enabled bots and bridges. + +### Start with synapse but ignore synapse going down + +If the bridge can handle shutdowns of the homeserver you'll want to install the +service in the `matrix.target` and optionally add a +`After=matrix-synapse.service` dependency to have the bot/bridge start after +synapse on starting everything. + +In this case the service file should look like this. + +``` +[Unit] +# ... +# Optional, this will only ensure that if you start everything, synapse will +# be started before the bot/bridge will be started. +After=matrix-synapse.service + +[Service] +# ... + +[Install] +WantedBy=matrix.target +``` + +### Stop/restart when synapse stops/restarts + +If the bridge can't handle shutdowns of the homeserver you'll still want to +install the service in the `matrix.target` but also have to specify the +`After=matrix-synapse.service` *and* `BindsTo=matrix-synapse.service` +dependencies to have the bot/bridge stop/restart with synapse. + +In this case the service file should look like this. + +``` +[Unit] +# ... +# Mandatory +After=matrix-synapse.service +BindsTo=matrix-synapse.service + +[Service] +# ... + +[Install] +WantedBy=matrix.target +``` diff --git a/contrib/systemd-with-workers/system/matrix-synapse-worker@.service b/contrib/systemd-with-workers/system/matrix-synapse-worker@.service new file mode 100644 index 0000000000..912984b9d2 --- /dev/null +++ b/contrib/systemd-with-workers/system/matrix-synapse-worker@.service @@ -0,0 +1,17 @@ +[Unit] +Description=Synapse Matrix Worker +After=matrix-synapse.service +BindsTo=matrix-synapse.service + +[Service] +Type=simple +User=matrix-synapse +WorkingDirectory=/var/lib/matrix-synapse +EnvironmentFile=/etc/default/matrix-synapse +ExecStart=/opt/venvs/matrix-synapse/bin/python -m synapse.app.%i --config-path=/etc/matrix-synapse/homeserver.yaml --config-path=/etc/matrix-synapse/conf.d/ --config-path=/etc/matrix-synapse/workers/%i.yaml +ExecReload=/bin/kill -HUP $MAINPID +Restart=always +RestartSec=3 + +[Install] +WantedBy=matrix-synapse.service diff --git a/contrib/systemd-with-workers/system/matrix-synapse.service b/contrib/systemd-with-workers/system/matrix-synapse.service new file mode 100644 index 0000000000..8bb4e400dc --- /dev/null +++ b/contrib/systemd-with-workers/system/matrix-synapse.service @@ -0,0 +1,16 @@ +[Unit] +Description=Synapse Matrix Homeserver + +[Service] +Type=simple +User=matrix-synapse +WorkingDirectory=/var/lib/matrix-synapse +EnvironmentFile=/etc/default/matrix-synapse +ExecStartPre=/opt/venvs/matrix-synapse/bin/python -m synapse.app.homeserver --config-path=/etc/matrix-synapse/homeserver.yaml --config-path=/etc/matrix-synapse/conf.d/ --generate-keys +ExecStart=/opt/venvs/matrix-synapse/bin/python -m synapse.app.homeserver --config-path=/etc/matrix-synapse/homeserver.yaml --config-path=/etc/matrix-synapse/conf.d/ +ExecReload=/bin/kill -HUP $MAINPID +Restart=always +RestartSec=3 + +[Install] +WantedBy=matrix.target diff --git a/contrib/systemd-with-workers/system/matrix.target b/contrib/systemd-with-workers/system/matrix.target new file mode 100644 index 0000000000..aff97d03ef --- /dev/null +++ b/contrib/systemd-with-workers/system/matrix.target @@ -0,0 +1,7 @@ +[Unit] +Description=Contains matrix services like synapse, bridges and bots +After=network.target +AllowIsolate=no + +[Install] +WantedBy=multi-user.target diff --git a/contrib/systemd-with-workers/workers/federation_reader.yaml b/contrib/systemd-with-workers/workers/federation_reader.yaml new file mode 100644 index 0000000000..47c54ec0d4 --- /dev/null +++ b/contrib/systemd-with-workers/workers/federation_reader.yaml @@ -0,0 +1,14 @@ +worker_app: synapse.app.federation_reader + +worker_replication_host: 127.0.0.1 +worker_replication_port: 9092 +worker_replication_http_port: 9093 + +worker_listeners: + - type: http + port: 8011 + resources: + - names: [federation] + +worker_daemonize: false +worker_log_config: /etc/matrix-synapse/federation-reader-log.yaml diff --git a/docs/federate.md b/docs/federate.md new file mode 100644 index 0000000000..b7fc09661c --- /dev/null +++ b/docs/federate.md @@ -0,0 +1,123 @@ +Setting up Federation +===================== + +Federation is the process by which users on different servers can participate +in the same room. For this to work, those other servers must be able to contact +yours to send messages. + +The ``server_name`` configured in the Synapse configuration file (often +``homeserver.yaml``) defines how resources (users, rooms, etc.) will be +identified (eg: ``@user:example.com``, ``#room:example.com``). By +default, it is also the domain that other servers will use to +try to reach your server (via port 8448). This is easy to set +up and will work provided you set the ``server_name`` to match your +machine's public DNS hostname, and provide Synapse with a TLS certificate +which is valid for your ``server_name``. + +Once you have completed the steps necessary to federate, you should be able to +join a room via federation. (A good place to start is ``#synapse:matrix.org`` - a +room for Synapse admins.) + + +## Delegation + +For a more flexible configuration, you can have ``server_name`` +resources (eg: ``@user:example.com``) served by a different host and +port (eg: ``synapse.example.com:443``). There are two ways to do this: + +- adding a ``/.well-known/matrix/server`` URL served on ``https://example.com``. +- adding a DNS ``SRV`` record in the DNS zone of domain + ``example.com``. + +Without configuring delegation, the matrix federation will +expect to find your server via ``example.com:8448``. The following methods +allow you retain a `server_name` of `example.com` so that your user IDs, room +aliases, etc continue to look like `*:example.com`, whilst having your +federation traffic routed to a different server. + +### .well-known delegation + +To use this method, you need to be able to alter the +``server_name`` 's https server to serve the ``/.well-known/matrix/server`` +URL. Having an active server (with a valid TLS certificate) serving your +``server_name`` domain is out of the scope of this documentation. + +The URL ``https://<server_name>/.well-known/matrix/server`` should +return a JSON structure containing the key ``m.server`` like so: + + { + "m.server": "<synapse.server.name>[:<yourport>]" + } + +In our example, this would mean that URL ``https://example.com/.well-known/matrix/server`` +should return: + + { + "m.server": "synapse.example.com:443" + } + +Note, specifying a port is optional. If a port is not specified an SRV lookup +is performed, as described below. If the target of the +delegation does not have an SRV record, then the port defaults to 8448. + +Most installations will not need to configure .well-known. However, it can be +useful in cases where the admin is hosting on behalf of someone else and +therefore cannot gain access to the necessary certificate. With .well-known, +federation servers will check for a valid TLS certificate for the delegated +hostname (in our example: ``synapse.example.com``). + +.well-known support first appeared in Synapse v0.99.0. To federate with older +servers you may need to additionally configure SRV delegation. Alternatively, +encourage the server admin in question to upgrade :). + +### DNS SRV delegation + +To use this delegation method, you need to have write access to your +``server_name`` 's domain zone DNS records (in our example it would be +``example.com`` DNS zone). + +This method requires the target server to provide a +valid TLS certificate for the original ``server_name``. + +You need to add a SRV record in your ``server_name`` 's DNS zone with +this format: + + _matrix._tcp.<yourdomain.com> <ttl> IN SRV <priority> <weight> <port> <synapse.server.name> + +In our example, we would need to add this SRV record in the +``example.com`` DNS zone: + + _matrix._tcp.example.com. 3600 IN SRV 10 5 443 synapse.example.com. + +Once done and set up, you can check the DNS record with ``dig -t srv +_matrix._tcp.<server_name>``. In our example, we would expect this: + + $ dig -t srv _matrix._tcp.example.com + _matrix._tcp.example.com. 3600 IN SRV 10 0 443 synapse.example.com. + +Note that the target of a SRV record cannot be an alias (CNAME record): it has to point +directly to the server hosting the synapse instance. + +## Troubleshooting + +You can use the [federation tester]( +<https://matrix.org/federationtester>) to check if your homeserver is +configured correctly. Alternatively try the [JSON API used by the federation tester](https://matrix.org/federationtester/api/report?server_name=DOMAIN). +Note that you'll have to modify this URL to replace ``DOMAIN`` with your +``server_name``. Hitting the API directly provides extra detail. + +The typical failure mode for federation is that when the server tries to join +a room, it is rejected with "401: Unauthorized". Generally this means that other +servers in the room could not access yours. (Joining a room over federation is +a complicated dance which requires connections in both directions). + +Another common problem is that people on other servers can't join rooms that +you invite them to. This can be caused by an incorrectly-configured reverse +proxy: see [reverse_proxy.rst](<reverse_proxy.rst>) for instructions on how to correctly +configure a reverse proxy. + +## Running a Demo Federation of Synapses + +If you want to get up and running quickly with a trio of homeservers in a +private federation, there is a script in the ``demo`` directory. This is mainly +useful just for development purposes. See [demo/README](<../demo/README>). diff --git a/docs/postgres.rst b/docs/postgres.rst index 2377542296..f7ebbed0c3 100644 --- a/docs/postgres.rst +++ b/docs/postgres.rst @@ -49,6 +49,24 @@ As with Debian/Ubuntu, postgres support depends on the postgres python connector export PATH=/usr/pgsql-9.4/bin/:$PATH pip install psycopg2 +Tuning Postgres +=============== + +The default settings should be fine for most deployments. For larger scale +deployments tuning some of the settings is recommended, details of which can be +found at https://wiki.postgresql.org/wiki/Tuning_Your_PostgreSQL_Server. + +In particular, we've found tuning the following values helpful for performance: + +- ``shared_buffers`` +- ``effective_cache_size`` +- ``work_mem`` +- ``maintenance_work_mem`` +- ``autovacuum_work_mem`` + +Note that the appropriate values for those fields depend on the amount of free +memory the database host has available. + Synapse config ============== @@ -129,8 +147,8 @@ Once that has completed, change the synapse config to point at the PostgreSQL database configuration file ``homeserver-postgres.yaml``:: ./synctl stop - mv homeserver.yaml homeserver-old-sqlite.yaml - mv homeserver-postgres.yaml homeserver.yaml + mv homeserver.yaml homeserver-old-sqlite.yaml + mv homeserver-postgres.yaml homeserver.yaml ./synctl start Synapse should now be running against PostgreSQL. diff --git a/docs/reverse_proxy.rst b/docs/reverse_proxy.rst index 6cd129abf4..cc81ceb84b 100644 --- a/docs/reverse_proxy.rst +++ b/docs/reverse_proxy.rst @@ -18,7 +18,7 @@ servers do not necessarily need to connect to your server via the same server name or port. Indeed, clients will use port 443 by default, whereas servers default to port 8448. Where these are different, we refer to the 'client port' and the 'federation port'. See `Setting up federation -<../README.rst#setting-up-federation>`_ for more details of the algorithm used for +<federate.md>`_ for more details of the algorithm used for federation connections. Let's assume that we expect clients to connect to our server at @@ -69,20 +69,16 @@ Let's assume that we expect clients to connect to our server at SSLEngine on ServerName matrix.example.com; - <Location /_matrix> - ProxyPass http://127.0.0.1:8008/_matrix nocanon - ProxyPassReverse http://127.0.0.1:8008/_matrix - </Location> + ProxyPass /_matrix http://127.0.0.1:8008/_matrix nocanon + ProxyPassReverse /_matrix http://127.0.0.1:8008/_matrix </VirtualHost> <VirtualHost *:8448> SSLEngine on ServerName example.com; - - <Location /_matrix> - ProxyPass http://127.0.0.1:8008/_matrix nocanon - ProxyPassReverse http://127.0.0.1:8008/_matrix - </Location> + + ProxyPass /_matrix http://127.0.0.1:8008/_matrix nocanon + ProxyPassReverse /_matrix http://127.0.0.1:8008/_matrix </VirtualHost> * HAProxy:: diff --git a/docs/sample_config.yaml b/docs/sample_config.yaml index b62745dd6e..4ada0fba0e 100644 --- a/docs/sample_config.yaml +++ b/docs/sample_config.yaml @@ -63,11 +63,11 @@ pid_file: DATADIR/homeserver.pid # Zero is used to indicate synapse should set the soft limit to the # hard limit. # -soft_file_limit: 0 +#soft_file_limit: 0 # Set to false to disable presence tracking on this homeserver. # -use_presence: true +#use_presence: false # The GC threshold parameters to pass to `gc.set_threshold`, if defined # @@ -246,6 +246,11 @@ listeners: # See 'ACME support' below to enable auto-provisioning this certificate via # Let's Encrypt. # +# If supplying your own, be sure to use a `.pem` file that includes the +# full certificate chain including any intermediate certificates (for +# instance, if using certbot, use `fullchain.pem` as your certificate, +# not `cert.pem`). +# #tls_certificate_path: "CONFDIR/SERVERNAME.tls.crt" # PEM-encoded private key for TLS @@ -354,7 +359,8 @@ database: database: "DATADIR/homeserver.db" # Number of events to cache in memory. -event_cache_size: "10K" +# +#event_cache_size: 10K ## Logging ## @@ -368,46 +374,77 @@ log_config: "CONFDIR/SERVERNAME.log.config" # Number of messages a client can send per second # -rc_messages_per_second: 0.2 +#rc_messages_per_second: 0.2 # Number of message a client can send before being throttled # -rc_message_burst_count: 10.0 +#rc_message_burst_count: 10.0 + +# Ratelimiting settings for registration and login. +# +# Each ratelimiting configuration is made of two parameters: +# - per_second: number of requests a client can send per second. +# - burst_count: number of requests a client can send before being throttled. +# +# Synapse currently uses the following configurations: +# - one for registration that ratelimits registration requests based on the +# client's IP address. +# - one for login that ratelimits login requests based on the client's IP +# address. +# - one for login that ratelimits login requests based on the account the +# client is attempting to log into. +# - one for login that ratelimits login requests based on the account the +# client is attempting to log into, based on the amount of failed login +# attempts for this account. +# +# The defaults are as shown below. +# +#rc_registration: +# per_second: 0.17 +# burst_count: 3 +# +#rc_login: +# address: +# per_second: 0.17 +# burst_count: 3 +# account: +# per_second: 0.17 +# burst_count: 3 +# failed_attempts: +# per_second: 0.17 +# burst_count: 3 # The federation window size in milliseconds # -federation_rc_window_size: 1000 +#federation_rc_window_size: 1000 # The number of federation requests from a single server in a window # before the server will delay processing the request. # -federation_rc_sleep_limit: 10 +#federation_rc_sleep_limit: 10 # The duration in milliseconds to delay processing events from # remote servers by if they go over the sleep limit. # -federation_rc_sleep_delay: 500 +#federation_rc_sleep_delay: 500 # The maximum number of concurrent federation requests allowed # from a single server # -federation_rc_reject_limit: 50 +#federation_rc_reject_limit: 50 # The number of federation requests to concurrently process from a # single server # -federation_rc_concurrent: 3 +#federation_rc_concurrent: 3 -# Number of registration requests a client can send per second. -# Defaults to 1/minute (0.17). +# Target outgoing federation transaction frequency for sending read-receipts, +# per-room. # -#rc_registration_requests_per_second: 0.17 - -# Number of registration requests a client can send before being -# throttled. -# Defaults to 3. +# If we end up trying to send out more read-receipts, they will get buffered up +# into fewer transactions. # -#rc_registration_request_burst_count: 3.0 +#federation_rr_transactions_per_room_per_second: 50 @@ -436,11 +473,11 @@ uploads_path: "DATADIR/uploads" # The largest allowed upload size in bytes # -max_upload_size: "10M" +#max_upload_size: 10M # Maximum number of pixels that will be thumbnailed # -max_image_pixels: "32M" +#max_image_pixels: 32M # Whether to generate new thumbnails on the fly to precisely match # the resolution requested by the client. If true then whenever @@ -448,32 +485,32 @@ max_image_pixels: "32M" # generate a new thumbnail. If false the server will pick a thumbnail # from a precalculated list. # -dynamic_thumbnails: false +#dynamic_thumbnails: false # List of thumbnails to precalculate when an image is uploaded. # -thumbnail_sizes: -- width: 32 - height: 32 - method: crop -- width: 96 - height: 96 - method: crop -- width: 320 - height: 240 - method: scale -- width: 640 - height: 480 - method: scale -- width: 800 - height: 600 - method: scale +#thumbnail_sizes: +# - width: 32 +# height: 32 +# method: crop +# - width: 96 +# height: 96 +# method: crop +# - width: 320 +# height: 240 +# method: scale +# - width: 640 +# height: 480 +# method: scale +# - width: 800 +# height: 600 +# method: scale # Is the preview URL API enabled? If enabled, you *must* specify # an explicit url_preview_ip_range_blacklist of IPs that the spider is # denied from accessing. # -url_preview_enabled: False +#url_preview_enabled: false # List of IP address CIDR ranges that the URL preview spider is denied # from accessing. There are no defaults: you must explicitly @@ -538,8 +575,8 @@ url_preview_enabled: False # - netloc: '^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$' # The largest allowed URL preview spidering size in bytes -max_spider_size: "10M" - +# +#max_spider_size: 10M ## Captcha ## @@ -547,23 +584,25 @@ max_spider_size: "10M" # This Home Server's ReCAPTCHA public key. # -recaptcha_public_key: "YOUR_PUBLIC_KEY" +#recaptcha_public_key: "YOUR_PUBLIC_KEY" # This Home Server's ReCAPTCHA private key. # -recaptcha_private_key: "YOUR_PRIVATE_KEY" +#recaptcha_private_key: "YOUR_PRIVATE_KEY" # Enables ReCaptcha checks when registering, preventing signup # unless a captcha is answered. Requires a valid ReCaptcha # public/private key. # -enable_registration_captcha: False +#enable_registration_captcha: false # A secret key used to bypass the captcha test entirely. +# #captcha_bypass_secret: "YOUR_SECRET_HERE" # The API endpoint to use for verifying m.login.recaptcha responses. -recaptcha_siteverify_api: "https://www.recaptcha.net/recaptcha/api/siteverify" +# +#recaptcha_siteverify_api: "https://www.recaptcha.net/recaptcha/api/siteverify" ## TURN ## @@ -584,7 +623,7 @@ recaptcha_siteverify_api: "https://www.recaptcha.net/recaptcha/api/siteverify" # How long generated TURN credentials last # -turn_user_lifetime: "1h" +#turn_user_lifetime: 1h # Whether guests should be allowed to use the TURN server. # This defaults to True, otherwise VoIP will be unreliable for guests. @@ -592,15 +631,17 @@ turn_user_lifetime: "1h" # connect to arbitrary endpoints without having first signed up for a # valid account (e.g. by passing a CAPTCHA). # -turn_allow_guests: True +#turn_allow_guests: True ## Registration ## +# # Registration can be rate-limited using the parameters in the "Ratelimiting" # section of this file. # Enable registration for new users. -enable_registration: False +# +#enable_registration: false # The user must provide all of the below types of 3PID when registering. # @@ -611,7 +652,7 @@ enable_registration: False # Explicitly disable asking for MSISDNs from the registration # flow (overrides registrations_require_3pid if MSISDNs are set as required) # -#disable_msisdn_registration: True +#disable_msisdn_registration: true # Mandate that users are only allowed to associate certain formats of # 3PIDs with accounts on this server. @@ -624,8 +665,8 @@ enable_registration: False # - medium: msisdn # pattern: '\+44' -# If set, allows registration by anyone who also has the shared -# secret, even if registration is otherwise disabled. +# If set, allows registration of standard or admin accounts by anyone who +# has the shared secret, even if registration is otherwise disabled. # # registration_shared_secret: <PRIVATE STRING> @@ -635,13 +676,13 @@ enable_registration: False # N.B. that increasing this will exponentially increase the time required # to register or login - e.g. 24 => 2^24 rounds which will take >20 mins. # -bcrypt_rounds: 12 +#bcrypt_rounds: 12 # Allows users to register as guests without a password/email/etc, and # participate in rooms hosted on this server which have been made # accessible to anonymous users. # -allow_guest_access: False +#allow_guest_access: false # The identity server which we suggest that clients should use when users log # in on this server. @@ -657,9 +698,9 @@ allow_guest_access: False # Also defines the ID server which will be called when an account is # deactivated (one will be picked arbitrarily). # -trusted_third_party_id_servers: - - matrix.org - - vector.im +#trusted_third_party_id_servers: +# - matrix.org +# - vector.im # Users who register on this homeserver will automatically be joined # to these rooms @@ -673,14 +714,14 @@ trusted_third_party_id_servers: # Setting to false means that if the rooms are not manually created, # users cannot be auto-joined since they do not exist. # -autocreate_auto_join_rooms: true +#autocreate_auto_join_rooms: true ## Metrics ### # Enable collection and rendering of performance metrics # -enable_metrics: False +#enable_metrics: False # Enable sentry integration # NOTE: While attempts are made to ensure that the logs don't contain @@ -700,22 +741,24 @@ enable_metrics: False # A list of event types that will be included in the room_invite_state # -room_invite_state_types: - - "m.room.join_rules" - - "m.room.canonical_alias" - - "m.room.avatar" - - "m.room.encryption" - - "m.room.name" +#room_invite_state_types: +# - "m.room.join_rules" +# - "m.room.canonical_alias" +# - "m.room.avatar" +# - "m.room.encryption" +# - "m.room.name" -# A list of application service config file to use +# A list of application service config files to use # -app_service_config_files: [] +#app_service_config_files: +# - app_service_1.yaml +# - app_service_2.yaml -# Whether or not to track application service IP addresses. Implicitly +# Uncomment to enable tracking of application service IP addresses. Implicitly # enables MAU tracking for application service users. # -track_appservice_user_ips: False +#track_appservice_user_ips: True # a secret which is used to sign access tokens. If none is specified, @@ -726,7 +769,7 @@ track_appservice_user_ips: False # Used to enable access token expiration. # -expire_access_token: False +#expire_access_token: False # a secret which is used to calculate HMACs for form values, to stop # falsification of values. Must be specified for the User Consent @@ -755,17 +798,16 @@ signing_key_path: "CONFDIR/SERVERNAME.signing.key" # Determines how quickly servers will query to check which keys # are still valid. # -key_refresh_interval: "1d" # 1 Day. +#key_refresh_interval: 1d # The trusted servers to download signing keys from. # -perspectives: - servers: - "matrix.org": - verify_keys: - "ed25519:auto": - key: "Noi6WqcDj0QmPxCNQqgezwTlBKrfqehY1u2FyWP9uYw" - +#perspectives: +# servers: +# "matrix.org": +# verify_keys: +# "ed25519:auto": +# key: "Noi6WqcDj0QmPxCNQqgezwTlBKrfqehY1u2FyWP9uYw" # Enable SAML2 for registration and login. Uses pysaml2. @@ -830,14 +872,15 @@ perspectives: # algorithm: "HS256" - -# Enable password for login. -# password_config: - enabled: true + # Uncomment to disable password login + # + #enabled: false + # Uncomment and change to a secret random string for extra security. # DO NOT CHANGE THIS AFTER INITIAL SETUP! - #pepper: "" + # + #pepper: "EVEN_MORE_SECRET" @@ -906,9 +949,9 @@ password_config: # example_option: 'things' -# Whether to allow non server admins to create groups on this server +# Uncomment to allow non-server-admin users to create groups on this server # -enable_group_creation: false +#enable_group_creation: true # If enabled, non server admins can only create groups with local parts # starting with this prefix @@ -919,6 +962,10 @@ enable_group_creation: false # User Directory configuration # +# 'enabled' defines whether users can search the user directory. If +# false then empty responses are returned to all queries. Defaults to +# true. +# # 'search_all_users' defines whether to search all users visible to your HS # when searching the user directory, rather than limiting to users visible # in public rooms. Defaults to false. If you set it True, you'll have to run @@ -926,6 +973,7 @@ enable_group_creation: false # on your database to tell it to rebuild the user_directory search indexes. # #user_directory: +# enabled: true # search_all_users: false @@ -1001,6 +1049,12 @@ enable_group_creation: false +# Uncomment to disable searching the public room list. When disabled +# blocks searching local and remote room lists for local and remote +# users by always returning an empty list for all queries. +# +#enable_room_list_search: false + # The `alias_creation` option controls who's allowed to create aliases # on this server. # diff --git a/synapse/api/auth.py b/synapse/api/auth.py index 5992d30623..ee646a97e8 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -788,9 +788,11 @@ class Auth(object): # Never fail an auth check for the server notices users or support user # This can be a problem where event creation is prohibited due to blocking - is_support = yield self.store.is_support_user(user_id) - if user_id == self.hs.config.server_notices_mxid or is_support: - return + if user_id is not None: + if user_id == self.hs.config.server_notices_mxid: + return + if (yield self.store.is_support_user(user_id)): + return if self.hs.config.hs_disabled: raise ResourceLimitError( diff --git a/synapse/api/ratelimiting.py b/synapse/api/ratelimiting.py index ad68079eeb..296c4a1c17 100644 --- a/synapse/api/ratelimiting.py +++ b/synapse/api/ratelimiting.py @@ -14,6 +14,8 @@ import collections +from synapse.api.errors import LimitExceededError + class Ratelimiter(object): """ @@ -82,3 +84,13 @@ class Ratelimiter(object): break else: del self.message_counts[key] + + def ratelimit(self, key, time_now_s, rate_hz, burst_count, update=True): + allowed, time_allowed = self.can_do_action( + key, time_now_s, rate_hz, burst_count, update + ) + + if not allowed: + raise LimitExceededError( + retry_after_ms=int(1000 * (time_allowed - time_now_s)), + ) diff --git a/synapse/app/_base.py b/synapse/app/_base.py index 32e8b8a3f5..d4c6c4c8e2 100644 --- a/synapse/app/_base.py +++ b/synapse/app/_base.py @@ -63,12 +63,13 @@ def start_worker_reactor(appname, config): start_reactor( appname, - config.soft_file_limit, - config.gc_thresholds, - config.worker_pid_file, - config.worker_daemonize, - config.worker_cpu_affinity, - logger, + soft_file_limit=config.soft_file_limit, + gc_thresholds=config.gc_thresholds, + pid_file=config.worker_pid_file, + daemonize=config.worker_daemonize, + cpu_affinity=config.worker_cpu_affinity, + print_pidfile=config.print_pidfile, + logger=logger, ) @@ -79,6 +80,7 @@ def start_reactor( pid_file, daemonize, cpu_affinity, + print_pidfile, logger, ): """ Run the reactor in the main process @@ -93,6 +95,7 @@ def start_reactor( pid_file (str): name of pid file to write to if daemonize is True daemonize (bool): true to run the reactor in a background process cpu_affinity (int|None): cpu affinity mask + print_pidfile (bool): whether to print the pid file, if daemonize is True logger (logging.Logger): logger instance to pass to Daemonize """ @@ -124,6 +127,9 @@ def start_reactor( reactor.run() if daemonize: + if print_pidfile: + print(pid_file) + daemon = Daemonize( app=appname, pid=pid_file, diff --git a/synapse/app/federation_sender.py b/synapse/app/federation_sender.py index a461442fdc..9711a7147c 100644 --- a/synapse/app/federation_sender.py +++ b/synapse/app/federation_sender.py @@ -28,6 +28,7 @@ from synapse.config.logger import setup_logging from synapse.federation import send_queue from synapse.http.site import SynapseSite from synapse.metrics import RegistryProxy +from synapse.metrics.background_process_metrics import run_as_background_process from synapse.metrics.resource import METRICS_PREFIX, MetricsResource from synapse.replication.slave.storage.deviceinbox import SlavedDeviceInboxStore from synapse.replication.slave.storage.devices import SlavedDeviceStore @@ -37,8 +38,10 @@ from synapse.replication.slave.storage.receipts import SlavedReceiptsStore from synapse.replication.slave.storage.registration import SlavedRegistrationStore from synapse.replication.slave.storage.transactions import SlavedTransactionStore from synapse.replication.tcp.client import ReplicationClientHandler +from synapse.replication.tcp.streams import ReceiptsStream from synapse.server import HomeServer from synapse.storage.engines import create_engine +from synapse.types import ReadReceipt from synapse.util.async_helpers import Linearizer from synapse.util.httpresourcetree import create_resource_tree from synapse.util.logcontext import LoggingContext, run_in_background @@ -202,6 +205,7 @@ class FederationSenderHandler(object): """ def __init__(self, hs, replication_client): self.store = hs.get_datastore() + self._is_mine_id = hs.is_mine_id self.federation_sender = hs.get_federation_sender() self.replication_client = replication_client @@ -234,6 +238,32 @@ class FederationSenderHandler(object): elif stream_name == "events": self.federation_sender.notify_new_events(token) + # ... and when new receipts happen + elif stream_name == ReceiptsStream.NAME: + run_as_background_process( + "process_receipts_for_federation", self._on_new_receipts, rows, + ) + + @defer.inlineCallbacks + def _on_new_receipts(self, rows): + """ + Args: + rows (iterable[synapse.replication.tcp.streams.ReceiptsStreamRow]): + new receipts to be processed + """ + for receipt in rows: + # we only want to send on receipts for our own users + if not self._is_mine_id(receipt.user_id): + continue + receipt_info = ReadReceipt( + receipt.room_id, + receipt.receipt_type, + receipt.user_id, + [receipt.event_id], + receipt.data, + ) + yield self.federation_sender.send_read_receipt(receipt_info) + @defer.inlineCallbacks def update_token(self, token): try: diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py index e8b6cc3114..869c028d1f 100755 --- a/synapse/app/homeserver.py +++ b/synapse/app/homeserver.py @@ -376,6 +376,7 @@ def setup(config_options): logger.info("Database prepared in %s.", config.database_config['name']) hs.setup() + hs.setup_master() @defer.inlineCallbacks def do_acme(): @@ -636,17 +637,15 @@ def run(hs): # be quite busy the first few minutes clock.call_later(5 * 60, start_phone_stats_home) - if hs.config.daemonize and hs.config.print_pidfile: - print(hs.config.pid_file) - _base.start_reactor( "synapse-homeserver", - hs.config.soft_file_limit, - hs.config.gc_thresholds, - hs.config.pid_file, - hs.config.daemonize, - hs.config.cpu_affinity, - logger, + soft_file_limit=hs.config.soft_file_limit, + gc_thresholds=hs.config.gc_thresholds, + pid_file=hs.config.pid_file, + daemonize=hs.config.daemonize, + cpu_affinity=hs.config.cpu_affinity, + print_pidfile=hs.config.print_pidfile, + logger=logger, ) diff --git a/synapse/config/_base.py b/synapse/config/_base.py index c4d3087fa4..a219a83550 100644 --- a/synapse/config/_base.py +++ b/synapse/config/_base.py @@ -214,14 +214,20 @@ class Config(object): " Defaults to the directory containing the last config file", ) + obj = cls() + + obj.invoke_all("add_arguments", config_parser) + config_args = config_parser.parse_args(argv) config_files = find_config_files(search_paths=config_args.config_path) - obj = cls() obj.read_config_files( config_files, keys_directory=config_args.keys_directory, generate_keys=False ) + + obj.invoke_all("read_arguments", config_args) + return obj @classmethod @@ -399,7 +405,10 @@ class Config(object): self.invoke_all("generate_files", config) return - self.invoke_all("read_config", config) + self.parse_config_dict(config) + + def parse_config_dict(self, config_dict): + self.invoke_all("read_config", config_dict) def find_config_files(search_paths): diff --git a/synapse/config/api.py b/synapse/config/api.py index e8a753f002..5eb4f86fa2 100644 --- a/synapse/config/api.py +++ b/synapse/config/api.py @@ -34,10 +34,10 @@ class ApiConfig(Config): # A list of event types that will be included in the room_invite_state # - room_invite_state_types: - - "{JoinRules}" - - "{CanonicalAlias}" - - "{RoomAvatar}" - - "{RoomEncryption}" - - "{Name}" + #room_invite_state_types: + # - "{JoinRules}" + # - "{CanonicalAlias}" + # - "{RoomAvatar}" + # - "{RoomEncryption}" + # - "{Name}" """.format(**vars(EventTypes)) diff --git a/synapse/config/appservice.py b/synapse/config/appservice.py index c260d59464..9e64c76544 100644 --- a/synapse/config/appservice.py +++ b/synapse/config/appservice.py @@ -37,14 +37,16 @@ class AppServiceConfig(Config): def default_config(cls, **kwargs): return """\ - # A list of application service config file to use + # A list of application service config files to use # - app_service_config_files: [] + #app_service_config_files: + # - app_service_1.yaml + # - app_service_2.yaml - # Whether or not to track application service IP addresses. Implicitly + # Uncomment to enable tracking of application service IP addresses. Implicitly # enables MAU tracking for application service users. # - track_appservice_user_ips: False + #track_appservice_user_ips: True """ diff --git a/synapse/config/captcha.py b/synapse/config/captcha.py index d25196be08..f7eebf26d2 100644 --- a/synapse/config/captcha.py +++ b/synapse/config/captcha.py @@ -18,11 +18,16 @@ from ._base import Config class CaptchaConfig(Config): def read_config(self, config): - self.recaptcha_private_key = config["recaptcha_private_key"] - self.recaptcha_public_key = config["recaptcha_public_key"] - self.enable_registration_captcha = config["enable_registration_captcha"] + self.recaptcha_private_key = config.get("recaptcha_private_key") + self.recaptcha_public_key = config.get("recaptcha_public_key") + self.enable_registration_captcha = config.get( + "enable_registration_captcha", False + ) self.captcha_bypass_secret = config.get("captcha_bypass_secret") - self.recaptcha_siteverify_api = config["recaptcha_siteverify_api"] + self.recaptcha_siteverify_api = config.get( + "recaptcha_siteverify_api", + "https://www.recaptcha.net/recaptcha/api/siteverify", + ) def default_config(self, **kwargs): return """\ @@ -31,21 +36,23 @@ class CaptchaConfig(Config): # This Home Server's ReCAPTCHA public key. # - recaptcha_public_key: "YOUR_PUBLIC_KEY" + #recaptcha_public_key: "YOUR_PUBLIC_KEY" # This Home Server's ReCAPTCHA private key. # - recaptcha_private_key: "YOUR_PRIVATE_KEY" + #recaptcha_private_key: "YOUR_PRIVATE_KEY" # Enables ReCaptcha checks when registering, preventing signup # unless a captcha is answered. Requires a valid ReCaptcha # public/private key. # - enable_registration_captcha: False + #enable_registration_captcha: false # A secret key used to bypass the captcha test entirely. + # #captcha_bypass_secret: "YOUR_SECRET_HERE" # The API endpoint to use for verifying m.login.recaptcha responses. - recaptcha_siteverify_api: "https://www.recaptcha.net/recaptcha/api/siteverify" + # + #recaptcha_siteverify_api: "https://www.recaptcha.net/recaptcha/api/siteverify" """ diff --git a/synapse/config/database.py b/synapse/config/database.py index 63e9cb63f8..3c27ed6b4a 100644 --- a/synapse/config/database.py +++ b/synapse/config/database.py @@ -60,7 +60,8 @@ class DatabaseConfig(Config): database: "%(database_path)s" # Number of events to cache in memory. - event_cache_size: "10K" + # + #event_cache_size: 10K """ % locals() def read_arguments(self, args): diff --git a/synapse/config/groups.py b/synapse/config/groups.py index 46933a904c..e4be172a79 100644 --- a/synapse/config/groups.py +++ b/synapse/config/groups.py @@ -23,9 +23,9 @@ class GroupsConfig(Config): def default_config(self, **kwargs): return """\ - # Whether to allow non server admins to create groups on this server + # Uncomment to allow non-server-admin users to create groups on this server # - enable_group_creation: false + #enable_group_creation: true # If enabled, non server admins can only create groups with local parts # starting with this prefix diff --git a/synapse/config/key.py b/synapse/config/key.py index 35f05fa974..933928885a 100644 --- a/synapse/config/key.py +++ b/synapse/config/key.py @@ -38,15 +38,26 @@ logger = logging.getLogger(__name__) class KeyConfig(Config): def read_config(self, config): - self.signing_key = self.read_signing_key(config["signing_key_path"]) + # the signing key can be specified inline or in a separate file + if "signing_key" in config: + self.signing_key = read_signing_keys([config["signing_key"]]) + else: + self.signing_key = self.read_signing_key(config["signing_key_path"]) + self.old_signing_keys = self.read_old_signing_keys( config.get("old_signing_keys", {}) ) self.key_refresh_interval = self.parse_duration( - config["key_refresh_interval"] + config.get("key_refresh_interval", "1d"), ) self.perspectives = self.read_perspectives( - config["perspectives"] + config.get("perspectives", {}).get("servers", { + "matrix.org": {"verify_keys": { + "ed25519:auto": { + "key": "Noi6WqcDj0QmPxCNQqgezwTlBKrfqehY1u2FyWP9uYw", + } + }} + }) ) self.macaroon_secret_key = config.get( @@ -88,7 +99,7 @@ class KeyConfig(Config): # Used to enable access token expiration. # - expire_access_token: False + #expire_access_token: False # a secret which is used to calculate HMACs for form values, to stop # falsification of values. Must be specified for the User Consent @@ -117,21 +128,21 @@ class KeyConfig(Config): # Determines how quickly servers will query to check which keys # are still valid. # - key_refresh_interval: "1d" # 1 Day. + #key_refresh_interval: 1d # The trusted servers to download signing keys from. # - perspectives: - servers: - "matrix.org": - verify_keys: - "ed25519:auto": - key: "Noi6WqcDj0QmPxCNQqgezwTlBKrfqehY1u2FyWP9uYw" + #perspectives: + # servers: + # "matrix.org": + # verify_keys: + # "ed25519:auto": + # key: "Noi6WqcDj0QmPxCNQqgezwTlBKrfqehY1u2FyWP9uYw" """ % locals() - def read_perspectives(self, perspectives_config): + def read_perspectives(self, perspectives_servers): servers = {} - for server_name, server_config in perspectives_config["servers"].items(): + for server_name, server_config in perspectives_servers.items(): for key_id, key_data in server_config["verify_keys"].items(): if is_signing_algorithm_supported(key_id): key_base64 = key_data["key"] diff --git a/synapse/config/metrics.py b/synapse/config/metrics.py index ed0498c634..2de51979d8 100644 --- a/synapse/config/metrics.py +++ b/synapse/config/metrics.py @@ -24,7 +24,7 @@ MISSING_SENTRY = ( class MetricsConfig(Config): def read_config(self, config): - self.enable_metrics = config["enable_metrics"] + self.enable_metrics = config.get("enable_metrics", False) self.report_stats = config.get("report_stats", None) self.metrics_port = config.get("metrics_port") self.metrics_bind_host = config.get("metrics_bind_host", "127.0.0.1") @@ -48,7 +48,7 @@ class MetricsConfig(Config): # Enable collection and rendering of performance metrics # - enable_metrics: False + #enable_metrics: False # Enable sentry integration # NOTE: While attempts are made to ensure that the logs don't contain diff --git a/synapse/config/password.py b/synapse/config/password.py index 2a52b9db54..eea59e772b 100644 --- a/synapse/config/password.py +++ b/synapse/config/password.py @@ -22,16 +22,21 @@ class PasswordConfig(Config): def read_config(self, config): password_config = config.get("password_config", {}) + if password_config is None: + password_config = {} + self.password_enabled = password_config.get("enabled", True) self.password_pepper = password_config.get("pepper", "") def default_config(self, config_dir_path, server_name, **kwargs): - return """ - # Enable password for login. - # + return """\ password_config: - enabled: true + # Uncomment to disable password login + # + #enabled: false + # Uncomment and change to a secret random string for extra security. # DO NOT CHANGE THIS AFTER INITIAL SETUP! - #pepper: "" + # + #pepper: "EVEN_MORE_SECRET" """ diff --git a/synapse/config/ratelimiting.py b/synapse/config/ratelimiting.py index 093042fdb9..5a68399e63 100644 --- a/synapse/config/ratelimiting.py +++ b/synapse/config/ratelimiting.py @@ -15,23 +15,35 @@ from ._base import Config +class RateLimitConfig(object): + def __init__(self, config): + self.per_second = config.get("per_second", 0.17) + self.burst_count = config.get("burst_count", 3.0) + + class RatelimitConfig(Config): def read_config(self, config): - self.rc_messages_per_second = config["rc_messages_per_second"] - self.rc_message_burst_count = config["rc_message_burst_count"] + self.rc_messages_per_second = config.get("rc_messages_per_second", 0.2) + self.rc_message_burst_count = config.get("rc_message_burst_count", 10.0) - self.federation_rc_window_size = config["federation_rc_window_size"] - self.federation_rc_sleep_limit = config["federation_rc_sleep_limit"] - self.federation_rc_sleep_delay = config["federation_rc_sleep_delay"] - self.federation_rc_reject_limit = config["federation_rc_reject_limit"] - self.federation_rc_concurrent = config["federation_rc_concurrent"] + self.rc_registration = RateLimitConfig(config.get("rc_registration", {})) - self.rc_registration_requests_per_second = config.get( - "rc_registration_requests_per_second", 0.17, + rc_login_config = config.get("rc_login", {}) + self.rc_login_address = RateLimitConfig(rc_login_config.get("address", {})) + self.rc_login_account = RateLimitConfig(rc_login_config.get("account", {})) + self.rc_login_failed_attempts = RateLimitConfig( + rc_login_config.get("failed_attempts", {}), ) - self.rc_registration_request_burst_count = config.get( - "rc_registration_request_burst_count", 3, + + self.federation_rc_window_size = config.get("federation_rc_window_size", 1000) + self.federation_rc_sleep_limit = config.get("federation_rc_sleep_limit", 10) + self.federation_rc_sleep_delay = config.get("federation_rc_sleep_delay", 500) + self.federation_rc_reject_limit = config.get("federation_rc_reject_limit", 50) + self.federation_rc_concurrent = config.get("federation_rc_concurrent", 3) + + self.federation_rr_transactions_per_room_per_second = config.get( + "federation_rr_transactions_per_room_per_second", 50, ) def default_config(self, **kwargs): @@ -40,44 +52,75 @@ class RatelimitConfig(Config): # Number of messages a client can send per second # - rc_messages_per_second: 0.2 + #rc_messages_per_second: 0.2 # Number of message a client can send before being throttled # - rc_message_burst_count: 10.0 + #rc_message_burst_count: 10.0 + + # Ratelimiting settings for registration and login. + # + # Each ratelimiting configuration is made of two parameters: + # - per_second: number of requests a client can send per second. + # - burst_count: number of requests a client can send before being throttled. + # + # Synapse currently uses the following configurations: + # - one for registration that ratelimits registration requests based on the + # client's IP address. + # - one for login that ratelimits login requests based on the client's IP + # address. + # - one for login that ratelimits login requests based on the account the + # client is attempting to log into. + # - one for login that ratelimits login requests based on the account the + # client is attempting to log into, based on the amount of failed login + # attempts for this account. + # + # The defaults are as shown below. + # + #rc_registration: + # per_second: 0.17 + # burst_count: 3 + # + #rc_login: + # address: + # per_second: 0.17 + # burst_count: 3 + # account: + # per_second: 0.17 + # burst_count: 3 + # failed_attempts: + # per_second: 0.17 + # burst_count: 3 # The federation window size in milliseconds # - federation_rc_window_size: 1000 + #federation_rc_window_size: 1000 # The number of federation requests from a single server in a window # before the server will delay processing the request. # - federation_rc_sleep_limit: 10 + #federation_rc_sleep_limit: 10 # The duration in milliseconds to delay processing events from # remote servers by if they go over the sleep limit. # - federation_rc_sleep_delay: 500 + #federation_rc_sleep_delay: 500 # The maximum number of concurrent federation requests allowed # from a single server # - federation_rc_reject_limit: 50 + #federation_rc_reject_limit: 50 # The number of federation requests to concurrently process from a # single server # - federation_rc_concurrent: 3 + #federation_rc_concurrent: 3 - # Number of registration requests a client can send per second. - # Defaults to 1/minute (0.17). + # Target outgoing federation transaction frequency for sending read-receipts, + # per-room. # - #rc_registration_requests_per_second: 0.17 - - # Number of registration requests a client can send before being - # throttled. - # Defaults to 3. + # If we end up trying to send out more read-receipts, they will get buffered up + # into fewer transactions. # - #rc_registration_request_burst_count: 3.0 + #federation_rr_transactions_per_room_per_second: 50 """ diff --git a/synapse/config/registration.py b/synapse/config/registration.py index d34dc9e456..f6b2b9ceee 100644 --- a/synapse/config/registration.py +++ b/synapse/config/registration.py @@ -24,7 +24,7 @@ class RegistrationConfig(Config): def read_config(self, config): self.enable_registration = bool( - strtobool(str(config["enable_registration"])) + strtobool(str(config.get("enable_registration", False))) ) if "disable_registration" in config: self.enable_registration = not bool( @@ -36,7 +36,10 @@ class RegistrationConfig(Config): self.registration_shared_secret = config.get("registration_shared_secret") self.bcrypt_rounds = config.get("bcrypt_rounds", 12) - self.trusted_third_party_id_servers = config["trusted_third_party_id_servers"] + self.trusted_third_party_id_servers = config.get( + "trusted_third_party_id_servers", + ["matrix.org", "vector.im"], + ) self.default_identity_server = config.get("default_identity_server") self.allow_guest_access = config.get("allow_guest_access", False) @@ -64,11 +67,13 @@ class RegistrationConfig(Config): return """\ ## Registration ## + # # Registration can be rate-limited using the parameters in the "Ratelimiting" # section of this file. # Enable registration for new users. - enable_registration: False + # + #enable_registration: false # The user must provide all of the below types of 3PID when registering. # @@ -79,7 +84,7 @@ class RegistrationConfig(Config): # Explicitly disable asking for MSISDNs from the registration # flow (overrides registrations_require_3pid if MSISDNs are set as required) # - #disable_msisdn_registration: True + #disable_msisdn_registration: true # Mandate that users are only allowed to associate certain formats of # 3PIDs with accounts on this server. @@ -92,8 +97,8 @@ class RegistrationConfig(Config): # - medium: msisdn # pattern: '\\+44' - # If set, allows registration by anyone who also has the shared - # secret, even if registration is otherwise disabled. + # If set, allows registration of standard or admin accounts by anyone who + # has the shared secret, even if registration is otherwise disabled. # %(registration_shared_secret)s @@ -103,13 +108,13 @@ class RegistrationConfig(Config): # N.B. that increasing this will exponentially increase the time required # to register or login - e.g. 24 => 2^24 rounds which will take >20 mins. # - bcrypt_rounds: 12 + #bcrypt_rounds: 12 # Allows users to register as guests without a password/email/etc, and # participate in rooms hosted on this server which have been made # accessible to anonymous users. # - allow_guest_access: False + #allow_guest_access: false # The identity server which we suggest that clients should use when users log # in on this server. @@ -125,9 +130,9 @@ class RegistrationConfig(Config): # Also defines the ID server which will be called when an account is # deactivated (one will be picked arbitrarily). # - trusted_third_party_id_servers: - - matrix.org - - vector.im + #trusted_third_party_id_servers: + # - matrix.org + # - vector.im # Users who register on this homeserver will automatically be joined # to these rooms @@ -141,7 +146,7 @@ class RegistrationConfig(Config): # Setting to false means that if the rooms are not manually created, # users cannot be auto-joined since they do not exist. # - autocreate_auto_join_rooms: true + #autocreate_auto_join_rooms: true """ % locals() def add_arguments(self, parser): diff --git a/synapse/config/repository.py b/synapse/config/repository.py index 97db2a5b7a..3f34ad9b2a 100644 --- a/synapse/config/repository.py +++ b/synapse/config/repository.py @@ -19,6 +19,36 @@ from synapse.util.module_loader import load_module from ._base import Config, ConfigError +DEFAULT_THUMBNAIL_SIZES = [ + { + "width": 32, + "height": 32, + "method": "crop", + }, { + "width": 96, + "height": 96, + "method": "crop", + }, { + "width": 320, + "height": 240, + "method": "scale", + }, { + "width": 640, + "height": 480, + "method": "scale", + }, { + "width": 800, + "height": 600, + "method": "scale" + }, +] + +THUMBNAIL_SIZE_YAML = """\ + # - width: %(width)i + # height: %(height)i + # method: %(method)s +""" + MISSING_NETADDR = ( "Missing netaddr library. This is required for URL preview API." ) @@ -77,9 +107,9 @@ def parse_thumbnail_requirements(thumbnail_sizes): class ContentRepositoryConfig(Config): def read_config(self, config): - self.max_upload_size = self.parse_size(config["max_upload_size"]) - self.max_image_pixels = self.parse_size(config["max_image_pixels"]) - self.max_spider_size = self.parse_size(config["max_spider_size"]) + self.max_upload_size = self.parse_size(config.get("max_upload_size", "10M")) + self.max_image_pixels = self.parse_size(config.get("max_image_pixels", "32M")) + self.max_spider_size = self.parse_size(config.get("max_spider_size", "10M")) self.media_store_path = self.ensure_directory(config["media_store_path"]) @@ -139,9 +169,9 @@ class ContentRepositoryConfig(Config): ) self.uploads_path = self.ensure_directory(config["uploads_path"]) - self.dynamic_thumbnails = config["dynamic_thumbnails"] + self.dynamic_thumbnails = config.get("dynamic_thumbnails", False) self.thumbnail_requirements = parse_thumbnail_requirements( - config["thumbnail_sizes"] + config.get("thumbnail_sizes", DEFAULT_THUMBNAIL_SIZES), ) self.url_preview_enabled = config.get("url_preview_enabled", False) if self.url_preview_enabled: @@ -178,6 +208,13 @@ class ContentRepositoryConfig(Config): def default_config(self, data_dir_path, **kwargs): media_store = os.path.join(data_dir_path, "media_store") uploads_path = os.path.join(data_dir_path, "uploads") + + formatted_thumbnail_sizes = "".join( + THUMBNAIL_SIZE_YAML % s for s in DEFAULT_THUMBNAIL_SIZES + ) + # strip final NL + formatted_thumbnail_sizes = formatted_thumbnail_sizes[:-1] + return r""" # Directory where uploaded images and attachments are stored. # @@ -204,11 +241,11 @@ class ContentRepositoryConfig(Config): # The largest allowed upload size in bytes # - max_upload_size: "10M" + #max_upload_size: 10M # Maximum number of pixels that will be thumbnailed # - max_image_pixels: "32M" + #max_image_pixels: 32M # Whether to generate new thumbnails on the fly to precisely match # the resolution requested by the client. If true then whenever @@ -216,32 +253,18 @@ class ContentRepositoryConfig(Config): # generate a new thumbnail. If false the server will pick a thumbnail # from a precalculated list. # - dynamic_thumbnails: false + #dynamic_thumbnails: false # List of thumbnails to precalculate when an image is uploaded. # - thumbnail_sizes: - - width: 32 - height: 32 - method: crop - - width: 96 - height: 96 - method: crop - - width: 320 - height: 240 - method: scale - - width: 640 - height: 480 - method: scale - - width: 800 - height: 600 - method: scale + #thumbnail_sizes: +%(formatted_thumbnail_sizes)s # Is the preview URL API enabled? If enabled, you *must* specify # an explicit url_preview_ip_range_blacklist of IPs that the spider is # denied from accessing. # - url_preview_enabled: False + #url_preview_enabled: false # List of IP address CIDR ranges that the URL preview spider is denied # from accessing. There are no defaults: you must explicitly @@ -306,6 +329,6 @@ class ContentRepositoryConfig(Config): # - netloc: '^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$' # The largest allowed URL preview spidering size in bytes - max_spider_size: "10M" - + # + #max_spider_size: 10M """ % locals() diff --git a/synapse/config/room_directory.py b/synapse/config/room_directory.py index 9b897abe3c..8a9fded4c5 100644 --- a/synapse/config/room_directory.py +++ b/synapse/config/room_directory.py @@ -20,6 +20,10 @@ from ._base import Config, ConfigError class RoomDirectoryConfig(Config): def read_config(self, config): + self.enable_room_list_search = config.get( + "enable_room_list_search", True, + ) + alias_creation_rules = config.get("alias_creation_rules") if alias_creation_rules is not None: @@ -54,6 +58,12 @@ class RoomDirectoryConfig(Config): def default_config(self, config_dir_path, server_name, **kwargs): return """ + # Uncomment to disable searching the public room list. When disabled + # blocks searching local and remote room lists for local and remote + # users by always returning an empty list for all queries. + # + #enable_room_list_search: false + # The `alias_creation` option controls who's allowed to create aliases # on this server. # diff --git a/synapse/config/saml2_config.py b/synapse/config/saml2_config.py index aff0a1f00c..39b9eb29c2 100644 --- a/synapse/config/saml2_config.py +++ b/synapse/config/saml2_config.py @@ -64,7 +64,7 @@ class SAML2Config(Config): } def default_config(self, config_dir_path, server_name, **kwargs): - return """ + return """\ # Enable SAML2 for registration and login. Uses pysaml2. # # `sp_config` is the configuration for the pysaml2 Service Provider. diff --git a/synapse/config/server.py b/synapse/config/server.py index 35a322fee0..08e4e45482 100644 --- a/synapse/config/server.py +++ b/synapse/config/server.py @@ -45,7 +45,7 @@ class ServerConfig(Config): self.pid_file = self.abspath(config.get("pid_file")) self.web_client_location = config.get("web_client_location", None) - self.soft_file_limit = config["soft_file_limit"] + self.soft_file_limit = config.get("soft_file_limit", 0) self.daemonize = config.get("daemonize") self.print_pidfile = config.get("print_pidfile") self.user_agent_suffix = config.get("user_agent_suffix") @@ -126,6 +126,11 @@ class ServerConfig(Config): self.public_baseurl += '/' self.start_pushers = config.get("start_pushers", True) + # (undocumented) option for torturing the worker-mode replication a bit, + # for testing. The value defines the number of milliseconds to pause before + # sending out any replication updates. + self.replication_torture_level = config.get("replication_torture_level") + self.listeners = [] for listener in config.get("listeners", []): if not isinstance(listener.get("port", None), int): @@ -307,11 +312,11 @@ class ServerConfig(Config): # Zero is used to indicate synapse should set the soft limit to the # hard limit. # - soft_file_limit: 0 + #soft_file_limit: 0 # Set to false to disable presence tracking on this homeserver. # - use_presence: true + #use_presence: false # The GC threshold parameters to pass to `gc.set_threshold`, if defined # diff --git a/synapse/config/tls.py b/synapse/config/tls.py index 40045de7ac..f0014902da 100644 --- a/synapse/config/tls.py +++ b/synapse/config/tls.py @@ -181,6 +181,11 @@ class TlsConfig(Config): # See 'ACME support' below to enable auto-provisioning this certificate via # Let's Encrypt. # + # If supplying your own, be sure to use a `.pem` file that includes the + # full certificate chain including any intermediate certificates (for + # instance, if using certbot, use `fullchain.pem` as your certificate, + # not `cert.pem`). + # #tls_certificate_path: "%(tls_certificate_path)s" # PEM-encoded private key for TLS diff --git a/synapse/config/user_directory.py b/synapse/config/user_directory.py index fab3a7d1c8..142754a7dc 100644 --- a/synapse/config/user_directory.py +++ b/synapse/config/user_directory.py @@ -22,9 +22,13 @@ class UserDirectoryConfig(Config): """ def read_config(self, config): + self.user_directory_search_enabled = True self.user_directory_search_all_users = False user_directory_config = config.get("user_directory", None) if user_directory_config: + self.user_directory_search_enabled = ( + user_directory_config.get("enabled", True) + ) self.user_directory_search_all_users = ( user_directory_config.get("search_all_users", False) ) @@ -33,6 +37,10 @@ class UserDirectoryConfig(Config): return """ # User Directory configuration # + # 'enabled' defines whether users can search the user directory. If + # false then empty responses are returned to all queries. Defaults to + # true. + # # 'search_all_users' defines whether to search all users visible to your HS # when searching the user directory, rather than limiting to users visible # in public rooms. Defaults to false. If you set it True, you'll have to run @@ -40,5 +48,6 @@ class UserDirectoryConfig(Config): # on your database to tell it to rebuild the user_directory search indexes. # #user_directory: + # enabled: true # search_all_users: false """ diff --git a/synapse/config/voip.py b/synapse/config/voip.py index 257f7c86e7..2a1f005a37 100644 --- a/synapse/config/voip.py +++ b/synapse/config/voip.py @@ -22,7 +22,9 @@ class VoipConfig(Config): self.turn_shared_secret = config.get("turn_shared_secret") self.turn_username = config.get("turn_username") self.turn_password = config.get("turn_password") - self.turn_user_lifetime = self.parse_duration(config["turn_user_lifetime"]) + self.turn_user_lifetime = self.parse_duration( + config.get("turn_user_lifetime", "1h"), + ) self.turn_allow_guests = config.get("turn_allow_guests", True) def default_config(self, **kwargs): @@ -45,7 +47,7 @@ class VoipConfig(Config): # How long generated TURN credentials last # - turn_user_lifetime: "1h" + #turn_user_lifetime: 1h # Whether guests should be allowed to use the TURN server. # This defaults to True, otherwise VoIP will be unreliable for guests. @@ -53,5 +55,5 @@ class VoipConfig(Config): # connect to arbitrary endpoints without having first signed up for a # valid account (e.g. by passing a CAPTCHA). # - turn_allow_guests: True + #turn_allow_guests: True """ diff --git a/synapse/config/workers.py b/synapse/config/workers.py index 80baf0ce0e..bfbd8b6c91 100644 --- a/synapse/config/workers.py +++ b/synapse/config/workers.py @@ -28,7 +28,7 @@ class WorkerConfig(Config): if self.worker_app == "synapse.app.homeserver": self.worker_app = None - self.worker_listeners = config.get("worker_listeners") + self.worker_listeners = config.get("worker_listeners", []) self.worker_daemonize = config.get("worker_daemonize") self.worker_pid_file = config.get("worker_pid_file") self.worker_log_file = config.get("worker_log_file") @@ -48,6 +48,17 @@ class WorkerConfig(Config): self.worker_main_http_uri = config.get("worker_main_http_uri", None) self.worker_cpu_affinity = config.get("worker_cpu_affinity") + # This option is really only here to support `--manhole` command line + # argument. + manhole = config.get("worker_manhole") + if manhole: + self.worker_listeners.append({ + "port": manhole, + "bind_addresses": ["127.0.0.1"], + "type": "manhole", + "tls": False, + }) + if self.worker_listeners: for listener in self.worker_listeners: bind_address = listener.pop("bind_address", None) @@ -57,3 +68,18 @@ class WorkerConfig(Config): bind_addresses.append(bind_address) elif not bind_addresses: bind_addresses.append('') + + def read_arguments(self, args): + # We support a bunch of command line arguments that override options in + # the config. A lot of these options have a worker_* prefix when running + # on workers so we also have to override them when command line options + # are specified. + + if args.daemonize is not None: + self.worker_daemonize = args.daemonize + if args.log_config is not None: + self.worker_log_config = args.log_config + if args.log_file is not None: + self.worker_log_file = args.log_file + if args.manhole is not None: + self.worker_manhole = args.worker_manhole diff --git a/synapse/federation/send_queue.py b/synapse/federation/send_queue.py index b7d0b25781..04d04a4457 100644 --- a/synapse/federation/send_queue.py +++ b/synapse/federation/send_queue.py @@ -46,7 +46,7 @@ logger = logging.getLogger(__name__) class FederationRemoteSendQueue(object): - """A drop in replacement for TransactionQueue""" + """A drop in replacement for FederationSender""" def __init__(self, hs): self.server_name = hs.hostname @@ -154,13 +154,13 @@ class FederationRemoteSendQueue(object): del self.device_messages[key] def notify_new_events(self, current_id): - """As per TransactionQueue""" + """As per FederationSender""" # We don't need to replicate this as it gets sent down a different # stream. pass def build_and_send_edu(self, destination, edu_type, content, key=None): - """As per TransactionQueue""" + """As per FederationSender""" if destination == self.server_name: logger.info("Not sending EDU to ourselves") return @@ -183,8 +183,17 @@ class FederationRemoteSendQueue(object): self.notifier.on_new_replication_data() + def send_read_receipt(self, receipt): + """As per FederationSender + + Args: + receipt (synapse.types.ReadReceipt): + """ + # nothing to do here: the replication listener will handle it. + pass + def send_presence(self, states): - """As per TransactionQueue + """As per FederationSender Args: states (list(UserPresenceState)) @@ -201,7 +210,7 @@ class FederationRemoteSendQueue(object): self.notifier.on_new_replication_data() def send_device_messages(self, destination): - """As per TransactionQueue""" + """As per FederationSender""" pos = self._next_pos() self.device_messages[pos] = destination self.notifier.on_new_replication_data() @@ -439,7 +448,7 @@ def process_rows_for_federation(transaction_queue, rows): transaction queue ready for sending to the relevant homeservers. Args: - transaction_queue (TransactionQueue) + transaction_queue (FederationSender) rows (list(synapse.replication.tcp.streams.FederationStreamRow)) """ diff --git a/synapse/federation/sender/__init__.py b/synapse/federation/sender/__init__.py new file mode 100644 index 0000000000..1dc041752b --- /dev/null +++ b/synapse/federation/sender/__init__.py @@ -0,0 +1,465 @@ +# -*- coding: utf-8 -*- +# Copyright 2019 New Vector Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import logging + +from six import itervalues + +from prometheus_client import Counter + +from twisted.internet import defer + +import synapse.metrics +from synapse.federation.sender.per_destination_queue import PerDestinationQueue +from synapse.federation.sender.transaction_manager import TransactionManager +from synapse.federation.units import Edu +from synapse.handlers.presence import get_interested_remotes +from synapse.metrics import ( + LaterGauge, + event_processing_loop_counter, + event_processing_loop_room_count, + events_processed_counter, +) +from synapse.metrics.background_process_metrics import run_as_background_process +from synapse.util import logcontext +from synapse.util.metrics import measure_func + +logger = logging.getLogger(__name__) + +sent_pdus_destination_dist_count = Counter( + "synapse_federation_client_sent_pdu_destinations:count", + "Number of PDUs queued for sending to one or more destinations", +) + +sent_pdus_destination_dist_total = Counter( + "synapse_federation_client_sent_pdu_destinations:total", "" + "Total number of PDUs queued for sending across all destinations", +) + + +class FederationSender(object): + def __init__(self, hs): + self.hs = hs + self.server_name = hs.hostname + + self.store = hs.get_datastore() + self.state = hs.get_state_handler() + + self.clock = hs.get_clock() + self.is_mine_id = hs.is_mine_id + + self._transaction_manager = TransactionManager(hs) + + # map from destination to PerDestinationQueue + self._per_destination_queues = {} # type: dict[str, PerDestinationQueue] + + LaterGauge( + "synapse_federation_transaction_queue_pending_destinations", + "", + [], + lambda: sum( + 1 for d in self._per_destination_queues.values() + if d.transmission_loop_running + ), + ) + + # Map of user_id -> UserPresenceState for all the pending presence + # to be sent out by user_id. Entries here get processed and put in + # pending_presence_by_dest + self.pending_presence = {} + + LaterGauge( + "synapse_federation_transaction_queue_pending_pdus", + "", + [], + lambda: sum( + d.pending_pdu_count() for d in self._per_destination_queues.values() + ), + ) + LaterGauge( + "synapse_federation_transaction_queue_pending_edus", + "", + [], + lambda: sum( + d.pending_edu_count() for d in self._per_destination_queues.values() + ), + ) + + self._order = 1 + + self._is_processing = False + self._last_poked_id = -1 + + self._processing_pending_presence = False + + # map from room_id to a set of PerDestinationQueues which we believe are + # awaiting a call to flush_read_receipts_for_room. The presence of an entry + # here for a given room means that we are rate-limiting RR flushes to that room, + # and that there is a pending call to _flush_rrs_for_room in the system. + self._queues_awaiting_rr_flush_by_room = { + } # type: dict[str, set[PerDestinationQueue]] + + self._rr_txn_interval_per_room_ms = ( + 1000.0 / hs.get_config().federation_rr_transactions_per_room_per_second + ) + + def _get_per_destination_queue(self, destination): + """Get or create a PerDestinationQueue for the given destination + + Args: + destination (str): server_name of remote server + + Returns: + PerDestinationQueue + """ + queue = self._per_destination_queues.get(destination) + if not queue: + queue = PerDestinationQueue(self.hs, self._transaction_manager, destination) + self._per_destination_queues[destination] = queue + return queue + + def notify_new_events(self, current_id): + """This gets called when we have some new events we might want to + send out to other servers. + """ + self._last_poked_id = max(current_id, self._last_poked_id) + + if self._is_processing: + return + + # fire off a processing loop in the background + run_as_background_process( + "process_event_queue_for_federation", + self._process_event_queue_loop, + ) + + @defer.inlineCallbacks + def _process_event_queue_loop(self): + try: + self._is_processing = True + while True: + last_token = yield self.store.get_federation_out_pos("events") + next_token, events = yield self.store.get_all_new_events_stream( + last_token, self._last_poked_id, limit=100, + ) + + logger.debug("Handling %s -> %s", last_token, next_token) + + if not events and next_token >= self._last_poked_id: + break + + @defer.inlineCallbacks + def handle_event(event): + # Only send events for this server. + send_on_behalf_of = event.internal_metadata.get_send_on_behalf_of() + is_mine = self.is_mine_id(event.sender) + if not is_mine and send_on_behalf_of is None: + return + + try: + # Get the state from before the event. + # We need to make sure that this is the state from before + # the event and not from after it. + # Otherwise if the last member on a server in a room is + # banned then it won't receive the event because it won't + # be in the room after the ban. + destinations = yield self.state.get_current_hosts_in_room( + event.room_id, latest_event_ids=event.prev_event_ids(), + ) + except Exception: + logger.exception( + "Failed to calculate hosts in room for event: %s", + event.event_id, + ) + return + + destinations = set(destinations) + + if send_on_behalf_of is not None: + # If we are sending the event on behalf of another server + # then it already has the event and there is no reason to + # send the event to it. + destinations.discard(send_on_behalf_of) + + logger.debug("Sending %s to %r", event, destinations) + + self._send_pdu(event, destinations) + + @defer.inlineCallbacks + def handle_room_events(events): + for event in events: + yield handle_event(event) + + events_by_room = {} + for event in events: + events_by_room.setdefault(event.room_id, []).append(event) + + yield logcontext.make_deferred_yieldable(defer.gatherResults( + [ + logcontext.run_in_background(handle_room_events, evs) + for evs in itervalues(events_by_room) + ], + consumeErrors=True + )) + + yield self.store.update_federation_out_pos( + "events", next_token + ) + + if events: + now = self.clock.time_msec() + ts = yield self.store.get_received_ts(events[-1].event_id) + + synapse.metrics.event_processing_lag.labels( + "federation_sender").set(now - ts) + synapse.metrics.event_processing_last_ts.labels( + "federation_sender").set(ts) + + events_processed_counter.inc(len(events)) + + event_processing_loop_room_count.labels( + "federation_sender" + ).inc(len(events_by_room)) + + event_processing_loop_counter.labels("federation_sender").inc() + + synapse.metrics.event_processing_positions.labels( + "federation_sender").set(next_token) + + finally: + self._is_processing = False + + def _send_pdu(self, pdu, destinations): + # We loop through all destinations to see whether we already have + # a transaction in progress. If we do, stick it in the pending_pdus + # table and we'll get back to it later. + + order = self._order + self._order += 1 + + destinations = set(destinations) + destinations.discard(self.server_name) + logger.debug("Sending to: %s", str(destinations)) + + if not destinations: + return + + sent_pdus_destination_dist_total.inc(len(destinations)) + sent_pdus_destination_dist_count.inc() + + for destination in destinations: + self._get_per_destination_queue(destination).send_pdu(pdu, order) + + @defer.inlineCallbacks + def send_read_receipt(self, receipt): + """Send a RR to any other servers in the room + + Args: + receipt (synapse.types.ReadReceipt): receipt to be sent + """ + + # Some background on the rate-limiting going on here. + # + # It turns out that if we attempt to send out RRs as soon as we get them from + # a client, then we end up trying to do several hundred Hz of federation + # transactions. (The number of transactions scales as O(N^2) on the size of a + # room, since in a large room we have both more RRs coming in, and more servers + # to send them to.) + # + # This leads to a lot of CPU load, and we end up getting behind. The solution + # currently adopted is as follows: + # + # The first receipt in a given room is sent out immediately, at time T0. Any + # further receipts are, in theory, batched up for N seconds, where N is calculated + # based on the number of servers in the room to achieve a transaction frequency + # of around 50Hz. So, for example, if there were 100 servers in the room, then + # N would be 100 / 50Hz = 2 seconds. + # + # Then, after T+N, we flush out any receipts that have accumulated, and restart + # the timer to flush out more receipts at T+2N, etc. If no receipts accumulate, + # we stop the cycle and go back to the start. + # + # However, in practice, it is often possible to flush out receipts earlier: in + # particular, if we are sending a transaction to a given server anyway (for + # example, because we have a PDU or a RR in another room to send), then we may + # as well send out all of the pending RRs for that server. So it may be that + # by the time we get to T+N, we don't actually have any RRs left to send out. + # Nevertheless we continue to buffer up RRs for the room in question until we + # reach the point that no RRs arrive between timer ticks. + # + # For even more background, see https://github.com/matrix-org/synapse/issues/4730. + + room_id = receipt.room_id + + # Work out which remote servers should be poked and poke them. + domains = yield self.state.get_current_hosts_in_room(room_id) + domains = [d for d in domains if d != self.server_name] + if not domains: + return + + queues_pending_flush = self._queues_awaiting_rr_flush_by_room.get( + room_id + ) + + # if there is no flush yet scheduled, we will send out these receipts with + # immediate flushes, and schedule the next flush for this room. + if queues_pending_flush is not None: + logger.debug("Queuing receipt for: %r", domains) + else: + logger.debug("Sending receipt to: %r", domains) + self._schedule_rr_flush_for_room(room_id, len(domains)) + + for domain in domains: + queue = self._get_per_destination_queue(domain) + queue.queue_read_receipt(receipt) + + # if there is already a RR flush pending for this room, then make sure this + # destination is registered for the flush + if queues_pending_flush is not None: + queues_pending_flush.add(queue) + else: + queue.flush_read_receipts_for_room(room_id) + + def _schedule_rr_flush_for_room(self, room_id, n_domains): + # that is going to cause approximately len(domains) transactions, so now back + # off for that multiplied by RR_TXN_INTERVAL_PER_ROOM + backoff_ms = self._rr_txn_interval_per_room_ms * n_domains + + logger.debug("Scheduling RR flush in %s in %d ms", room_id, backoff_ms) + self.clock.call_later(backoff_ms, self._flush_rrs_for_room, room_id) + self._queues_awaiting_rr_flush_by_room[room_id] = set() + + def _flush_rrs_for_room(self, room_id): + queues = self._queues_awaiting_rr_flush_by_room.pop(room_id) + logger.debug("Flushing RRs in %s to %s", room_id, queues) + + if not queues: + # no more RRs arrived for this room; we are done. + return + + # schedule the next flush + self._schedule_rr_flush_for_room(room_id, len(queues)) + + for queue in queues: + queue.flush_read_receipts_for_room(room_id) + + @logcontext.preserve_fn # the caller should not yield on this + @defer.inlineCallbacks + def send_presence(self, states): + """Send the new presence states to the appropriate destinations. + + This actually queues up the presence states ready for sending and + triggers a background task to process them and send out the transactions. + + Args: + states (list(UserPresenceState)) + """ + if not self.hs.config.use_presence: + # No-op if presence is disabled. + return + + # First we queue up the new presence by user ID, so multiple presence + # updates in quick successtion are correctly handled + # We only want to send presence for our own users, so lets always just + # filter here just in case. + self.pending_presence.update({ + state.user_id: state for state in states + if self.is_mine_id(state.user_id) + }) + + # We then handle the new pending presence in batches, first figuring + # out the destinations we need to send each state to and then poking it + # to attempt a new transaction. We linearize this so that we don't + # accidentally mess up the ordering and send multiple presence updates + # in the wrong order + if self._processing_pending_presence: + return + + self._processing_pending_presence = True + try: + while True: + states_map = self.pending_presence + self.pending_presence = {} + + if not states_map: + break + + yield self._process_presence_inner(list(states_map.values())) + except Exception: + logger.exception("Error sending presence states to servers") + finally: + self._processing_pending_presence = False + + @measure_func("txnqueue._process_presence") + @defer.inlineCallbacks + def _process_presence_inner(self, states): + """Given a list of states populate self.pending_presence_by_dest and + poke to send a new transaction to each destination + + Args: + states (list(UserPresenceState)) + """ + hosts_and_states = yield get_interested_remotes(self.store, states, self.state) + + for destinations, states in hosts_and_states: + for destination in destinations: + if destination == self.server_name: + continue + self._get_per_destination_queue(destination).send_presence(states) + + def build_and_send_edu(self, destination, edu_type, content, key=None): + """Construct an Edu object, and queue it for sending + + Args: + destination (str): name of server to send to + edu_type (str): type of EDU to send + content (dict): content of EDU + key (Any|None): clobbering key for this edu + """ + if destination == self.server_name: + logger.info("Not sending EDU to ourselves") + return + + edu = Edu( + origin=self.server_name, + destination=destination, + edu_type=edu_type, + content=content, + ) + + self.send_edu(edu, key) + + def send_edu(self, edu, key): + """Queue an EDU for sending + + Args: + edu (Edu): edu to send + key (Any|None): clobbering key for this edu + """ + queue = self._get_per_destination_queue(edu.destination) + if key: + queue.send_keyed_edu(edu, key) + else: + queue.send_edu(edu) + + def send_device_messages(self, destination): + if destination == self.server_name: + logger.info("Not sending device update to ourselves") + return + + self._get_per_destination_queue(destination).attempt_new_transaction() + + def get_current_token(self): + return 0 diff --git a/synapse/federation/sender/per_destination_queue.py b/synapse/federation/sender/per_destination_queue.py new file mode 100644 index 0000000000..be99211003 --- /dev/null +++ b/synapse/federation/sender/per_destination_queue.py @@ -0,0 +1,378 @@ +# -*- coding: utf-8 -*- +# Copyright 2014-2016 OpenMarket Ltd +# Copyright 2019 New Vector Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +import datetime +import logging + +from prometheus_client import Counter + +from twisted.internet import defer + +from synapse.api.errors import ( + FederationDeniedError, + HttpResponseException, + RequestSendFailed, +) +from synapse.events import EventBase +from synapse.federation.units import Edu +from synapse.handlers.presence import format_user_presence_state +from synapse.metrics import sent_transactions_counter +from synapse.metrics.background_process_metrics import run_as_background_process +from synapse.storage import UserPresenceState +from synapse.util.retryutils import NotRetryingDestination, get_retry_limiter + +logger = logging.getLogger(__name__) + + +sent_edus_counter = Counter( + "synapse_federation_client_sent_edus", + "Total number of EDUs successfully sent", +) + +sent_edus_by_type = Counter( + "synapse_federation_client_sent_edus_by_type", + "Number of sent EDUs successfully sent, by event type", + ["type"], +) + + +class PerDestinationQueue(object): + """ + Manages the per-destination transmission queues. + + Args: + hs (synapse.HomeServer): + transaction_sender (TransactionManager): + destination (str): the server_name of the destination that we are managing + transmission for. + """ + def __init__(self, hs, transaction_manager, destination): + self._server_name = hs.hostname + self._clock = hs.get_clock() + self._store = hs.get_datastore() + self._transaction_manager = transaction_manager + + self._destination = destination + self.transmission_loop_running = False + + # a list of tuples of (pending pdu, order) + self._pending_pdus = [] # type: list[tuple[EventBase, int]] + self._pending_edus = [] # type: list[Edu] + + # Pending EDUs by their "key". Keyed EDUs are EDUs that get clobbered + # based on their key (e.g. typing events by room_id) + # Map of (edu_type, key) -> Edu + self._pending_edus_keyed = {} # type: dict[tuple[str, str], Edu] + + # Map of user_id -> UserPresenceState of pending presence to be sent to this + # destination + self._pending_presence = {} # type: dict[str, UserPresenceState] + + # room_id -> receipt_type -> user_id -> receipt_dict + self._pending_rrs = {} + self._rrs_pending_flush = False + + # stream_id of last successfully sent to-device message. + # NB: may be a long or an int. + self._last_device_stream_id = 0 + + # stream_id of last successfully sent device list update. + self._last_device_list_stream_id = 0 + + def __str__(self): + return "PerDestinationQueue[%s]" % self._destination + + def pending_pdu_count(self): + return len(self._pending_pdus) + + def pending_edu_count(self): + return ( + len(self._pending_edus) + + len(self._pending_presence) + + len(self._pending_edus_keyed) + ) + + def send_pdu(self, pdu, order): + """Add a PDU to the queue, and start the transmission loop if neccessary + + Args: + pdu (EventBase): pdu to send + order (int): + """ + self._pending_pdus.append((pdu, order)) + self.attempt_new_transaction() + + def send_presence(self, states): + """Add presence updates to the queue. Start the transmission loop if neccessary. + + Args: + states (iterable[UserPresenceState]): presence to send + """ + self._pending_presence.update({ + state.user_id: state for state in states + }) + self.attempt_new_transaction() + + def queue_read_receipt(self, receipt): + """Add a RR to the list to be sent. Doesn't start the transmission loop yet + (see flush_read_receipts_for_room) + + Args: + receipt (synapse.api.receipt_info.ReceiptInfo): receipt to be queued + """ + self._pending_rrs.setdefault( + receipt.room_id, {}, + ).setdefault( + receipt.receipt_type, {} + )[receipt.user_id] = { + "event_ids": receipt.event_ids, + "data": receipt.data, + } + + def flush_read_receipts_for_room(self, room_id): + # if we don't have any read-receipts for this room, it may be that we've already + # sent them out, so we don't need to flush. + if room_id not in self._pending_rrs: + return + self._rrs_pending_flush = True + self.attempt_new_transaction() + + def send_keyed_edu(self, edu, key): + self._pending_edus_keyed[(edu.edu_type, key)] = edu + self.attempt_new_transaction() + + def send_edu(self, edu): + self._pending_edus.append(edu) + self.attempt_new_transaction() + + def attempt_new_transaction(self): + """Try to start a new transaction to this destination + + If there is already a transaction in progress to this destination, + returns immediately. Otherwise kicks off the process of sending a + transaction in the background. + """ + # list of (pending_pdu, deferred, order) + if self.transmission_loop_running: + # XXX: this can get stuck on by a never-ending + # request at which point pending_pdus just keeps growing. + # we need application-layer timeouts of some flavour of these + # requests + logger.debug( + "TX [%s] Transaction already in progress", + self._destination + ) + return + + logger.debug("TX [%s] Starting transaction loop", self._destination) + + run_as_background_process( + "federation_transaction_transmission_loop", + self._transaction_transmission_loop, + ) + + @defer.inlineCallbacks + def _transaction_transmission_loop(self): + pending_pdus = [] + try: + self.transmission_loop_running = True + + # This will throw if we wouldn't retry. We do this here so we fail + # quickly, but we will later check this again in the http client, + # hence why we throw the result away. + yield get_retry_limiter(self._destination, self._clock, self._store) + + pending_pdus = [] + while True: + device_message_edus, device_stream_id, dev_list_id = ( + yield self._get_new_device_messages() + ) + + # BEGIN CRITICAL SECTION + # + # In order to avoid a race condition, we need to make sure that + # the following code (from popping the queues up to the point + # where we decide if we actually have any pending messages) is + # atomic - otherwise new PDUs or EDUs might arrive in the + # meantime, but not get sent because we hold the + # transmission_loop_running flag. + + pending_pdus = self._pending_pdus + + # We can only include at most 50 PDUs per transactions + pending_pdus, self._pending_pdus = pending_pdus[:50], pending_pdus[50:] + + pending_edus = [] + + pending_edus.extend(self._get_rr_edus(force_flush=False)) + + # We can only include at most 100 EDUs per transactions + pending_edus.extend(self._pop_pending_edus(100 - len(pending_edus))) + + pending_edus.extend( + self._pending_edus_keyed.values() + ) + + self._pending_edus_keyed = {} + + pending_edus.extend(device_message_edus) + + pending_presence = self._pending_presence + self._pending_presence = {} + if pending_presence: + pending_edus.append( + Edu( + origin=self._server_name, + destination=self._destination, + edu_type="m.presence", + content={ + "push": [ + format_user_presence_state( + presence, self._clock.time_msec() + ) + for presence in pending_presence.values() + ] + }, + ) + ) + + if pending_pdus: + logger.debug("TX [%s] len(pending_pdus_by_dest[dest]) = %d", + self._destination, len(pending_pdus)) + + if not pending_pdus and not pending_edus: + logger.debug("TX [%s] Nothing to send", self._destination) + self._last_device_stream_id = device_stream_id + return + + # if we've decided to send a transaction anyway, and we have room, we + # may as well send any pending RRs + if len(pending_edus) < 100: + pending_edus.extend(self._get_rr_edus(force_flush=True)) + + # END CRITICAL SECTION + + success = yield self._transaction_manager.send_new_transaction( + self._destination, pending_pdus, pending_edus + ) + if success: + sent_transactions_counter.inc() + sent_edus_counter.inc(len(pending_edus)) + for edu in pending_edus: + sent_edus_by_type.labels(edu.edu_type).inc() + # Remove the acknowledged device messages from the database + # Only bother if we actually sent some device messages + if device_message_edus: + yield self._store.delete_device_msgs_for_remote( + self._destination, device_stream_id + ) + logger.info( + "Marking as sent %r %r", self._destination, dev_list_id + ) + yield self._store.mark_as_sent_devices_by_remote( + self._destination, dev_list_id + ) + + self._last_device_stream_id = device_stream_id + self._last_device_list_stream_id = dev_list_id + else: + break + except NotRetryingDestination as e: + logger.debug( + "TX [%s] not ready for retry yet (next retry at %s) - " + "dropping transaction for now", + self._destination, + datetime.datetime.fromtimestamp( + (e.retry_last_ts + e.retry_interval) / 1000.0 + ), + ) + except FederationDeniedError as e: + logger.info(e) + except HttpResponseException as e: + logger.warning( + "TX [%s] Received %d response to transaction: %s", + self._destination, e.code, e, + ) + except RequestSendFailed as e: + logger.warning("TX [%s] Failed to send transaction: %s", self._destination, e) + + for p, _ in pending_pdus: + logger.info("Failed to send event %s to %s", p.event_id, + self._destination) + except Exception: + logger.exception( + "TX [%s] Failed to send transaction", + self._destination, + ) + for p, _ in pending_pdus: + logger.info("Failed to send event %s to %s", p.event_id, + self._destination) + finally: + # We want to be *very* sure we clear this after we stop processing + self.transmission_loop_running = False + + def _get_rr_edus(self, force_flush): + if not self._pending_rrs: + return + if not force_flush and not self._rrs_pending_flush: + # not yet time for this lot + return + + edu = Edu( + origin=self._server_name, + destination=self._destination, + edu_type="m.receipt", + content=self._pending_rrs, + ) + self._pending_rrs = {} + self._rrs_pending_flush = False + yield edu + + def _pop_pending_edus(self, limit): + pending_edus = self._pending_edus + pending_edus, self._pending_edus = pending_edus[:limit], pending_edus[limit:] + return pending_edus + + @defer.inlineCallbacks + def _get_new_device_messages(self): + last_device_stream_id = self._last_device_stream_id + to_device_stream_id = self._store.get_to_device_stream_token() + contents, stream_id = yield self._store.get_new_device_msgs_for_remote( + self._destination, last_device_stream_id, to_device_stream_id + ) + edus = [ + Edu( + origin=self._server_name, + destination=self._destination, + edu_type="m.direct_to_device", + content=content, + ) + for content in contents + ] + + last_device_list = self._last_device_list_stream_id + now_stream_id, results = yield self._store.get_devices_by_remote( + self._destination, last_device_list + ) + edus.extend( + Edu( + origin=self._server_name, + destination=self._destination, + edu_type="m.device_list_update", + content=content, + ) + for content in results + ) + defer.returnValue((edus, stream_id, now_stream_id)) diff --git a/synapse/federation/sender/transaction_manager.py b/synapse/federation/sender/transaction_manager.py new file mode 100644 index 0000000000..35e6b8ff5b --- /dev/null +++ b/synapse/federation/sender/transaction_manager.py @@ -0,0 +1,147 @@ +# -*- coding: utf-8 -*- +# Copyright 2019 New Vector Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +import logging + +from twisted.internet import defer + +from synapse.api.errors import HttpResponseException +from synapse.federation.persistence import TransactionActions +from synapse.federation.units import Transaction +from synapse.util.metrics import measure_func + +logger = logging.getLogger(__name__) + + +class TransactionManager(object): + """Helper class which handles building and sending transactions + + shared between PerDestinationQueue objects + """ + def __init__(self, hs): + self._server_name = hs.hostname + self.clock = hs.get_clock() # nb must be called this for @measure_func + self._store = hs.get_datastore() + self._transaction_actions = TransactionActions(self._store) + self._transport_layer = hs.get_federation_transport_client() + + # HACK to get unique tx id + self._next_txn_id = int(self.clock.time_msec()) + + @measure_func("_send_new_transaction") + @defer.inlineCallbacks + def send_new_transaction(self, destination, pending_pdus, pending_edus): + + # Sort based on the order field + pending_pdus.sort(key=lambda t: t[1]) + pdus = [x[0] for x in pending_pdus] + edus = pending_edus + + success = True + + logger.debug("TX [%s] _attempt_new_transaction", destination) + + txn_id = str(self._next_txn_id) + + logger.debug( + "TX [%s] {%s} Attempting new transaction" + " (pdus: %d, edus: %d)", + destination, txn_id, + len(pdus), + len(edus), + ) + + logger.debug("TX [%s] Persisting transaction...", destination) + + transaction = Transaction.create_new( + origin_server_ts=int(self.clock.time_msec()), + transaction_id=txn_id, + origin=self._server_name, + destination=destination, + pdus=pdus, + edus=edus, + ) + + self._next_txn_id += 1 + + yield self._transaction_actions.prepare_to_send(transaction) + + logger.debug("TX [%s] Persisted transaction", destination) + logger.info( + "TX [%s] {%s} Sending transaction [%s]," + " (PDUs: %d, EDUs: %d)", + destination, txn_id, + transaction.transaction_id, + len(pdus), + len(edus), + ) + + # Actually send the transaction + + # FIXME (erikj): This is a bit of a hack to make the Pdu age + # keys work + def json_data_cb(): + data = transaction.get_dict() + now = int(self.clock.time_msec()) + if "pdus" in data: + for p in data["pdus"]: + if "age_ts" in p: + unsigned = p.setdefault("unsigned", {}) + unsigned["age"] = now - int(p["age_ts"]) + del p["age_ts"] + return data + + try: + response = yield self._transport_layer.send_transaction( + transaction, json_data_cb + ) + code = 200 + except HttpResponseException as e: + code = e.code + response = e.response + + if e.code in (401, 404, 429) or 500 <= e.code: + logger.info( + "TX [%s] {%s} got %d response", + destination, txn_id, code + ) + raise e + + logger.info( + "TX [%s] {%s} got %d response", + destination, txn_id, code + ) + + yield self._transaction_actions.delivered( + transaction, code, response + ) + + logger.debug("TX [%s] {%s} Marked as delivered", destination, txn_id) + + if code == 200: + for e_id, r in response.get("pdus", {}).items(): + if "error" in r: + logger.warn( + "TX [%s] {%s} Remote returned error for %s: %s", + destination, txn_id, e_id, r, + ) + else: + for p in pdus: + logger.warn( + "TX [%s] {%s} Failed to send event %s", + destination, txn_id, p.event_id, + ) + success = False + + defer.returnValue(success) diff --git a/synapse/federation/transaction_queue.py b/synapse/federation/transaction_queue.py deleted file mode 100644 index e5e42c647d..0000000000 --- a/synapse/federation/transaction_queue.py +++ /dev/null @@ -1,716 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014-2016 OpenMarket Ltd -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -import datetime -import logging - -from six import itervalues - -from prometheus_client import Counter - -from twisted.internet import defer - -import synapse.metrics -from synapse.api.errors import ( - FederationDeniedError, - HttpResponseException, - RequestSendFailed, -) -from synapse.handlers.presence import format_user_presence_state, get_interested_remotes -from synapse.metrics import ( - LaterGauge, - event_processing_loop_counter, - event_processing_loop_room_count, - events_processed_counter, - sent_transactions_counter, -) -from synapse.metrics.background_process_metrics import run_as_background_process -from synapse.util import logcontext -from synapse.util.metrics import measure_func -from synapse.util.retryutils import NotRetryingDestination, get_retry_limiter - -from .persistence import TransactionActions -from .units import Edu, Transaction - -logger = logging.getLogger(__name__) - -sent_pdus_destination_dist_count = Counter( - "synapse_federation_client_sent_pdu_destinations:count", - "Number of PDUs queued for sending to one or more destinations", -) - -sent_pdus_destination_dist_total = Counter( - "synapse_federation_client_sent_pdu_destinations:total", "" - "Total number of PDUs queued for sending across all destinations", -) - -sent_edus_counter = Counter( - "synapse_federation_client_sent_edus", - "Total number of EDUs successfully sent", -) - -sent_edus_by_type = Counter( - "synapse_federation_client_sent_edus_by_type", - "Number of sent EDUs successfully sent, by event type", - ["type"], -) - - -class TransactionQueue(object): - """This class makes sure we only have one transaction in flight at - a time for a given destination. - - It batches pending PDUs into single transactions. - """ - - def __init__(self, hs): - self.hs = hs - self.server_name = hs.hostname - - self.store = hs.get_datastore() - self.state = hs.get_state_handler() - self.transaction_actions = TransactionActions(self.store) - - self.transport_layer = hs.get_federation_transport_client() - - self.clock = hs.get_clock() - self.is_mine_id = hs.is_mine_id - - # Is a mapping from destinations -> deferreds. Used to keep track - # of which destinations have transactions in flight and when they are - # done - self.pending_transactions = {} - - LaterGauge( - "synapse_federation_transaction_queue_pending_destinations", - "", - [], - lambda: len(self.pending_transactions), - ) - - # Is a mapping from destination -> list of - # tuple(pending pdus, deferred, order) - self.pending_pdus_by_dest = pdus = {} - # destination -> list of tuple(edu, deferred) - self.pending_edus_by_dest = edus = {} - - # Map of user_id -> UserPresenceState for all the pending presence - # to be sent out by user_id. Entries here get processed and put in - # pending_presence_by_dest - self.pending_presence = {} - - # Map of destination -> user_id -> UserPresenceState of pending presence - # to be sent to each destinations - self.pending_presence_by_dest = presence = {} - - # Pending EDUs by their "key". Keyed EDUs are EDUs that get clobbered - # based on their key (e.g. typing events by room_id) - # Map of destination -> (edu_type, key) -> Edu - self.pending_edus_keyed_by_dest = edus_keyed = {} - - LaterGauge( - "synapse_federation_transaction_queue_pending_pdus", - "", - [], - lambda: sum(map(len, pdus.values())), - ) - LaterGauge( - "synapse_federation_transaction_queue_pending_edus", - "", - [], - lambda: ( - sum(map(len, edus.values())) - + sum(map(len, presence.values())) - + sum(map(len, edus_keyed.values())) - ), - ) - - # destination -> stream_id of last successfully sent to-device message. - # NB: may be a long or an int. - self.last_device_stream_id_by_dest = {} - - # destination -> stream_id of last successfully sent device list - # update. - self.last_device_list_stream_id_by_dest = {} - - # HACK to get unique tx id - self._next_txn_id = int(self.clock.time_msec()) - - self._order = 1 - - self._is_processing = False - self._last_poked_id = -1 - - self._processing_pending_presence = False - - def notify_new_events(self, current_id): - """This gets called when we have some new events we might want to - send out to other servers. - """ - self._last_poked_id = max(current_id, self._last_poked_id) - - if self._is_processing: - return - - # fire off a processing loop in the background - run_as_background_process( - "process_event_queue_for_federation", - self._process_event_queue_loop, - ) - - @defer.inlineCallbacks - def _process_event_queue_loop(self): - try: - self._is_processing = True - while True: - last_token = yield self.store.get_federation_out_pos("events") - next_token, events = yield self.store.get_all_new_events_stream( - last_token, self._last_poked_id, limit=100, - ) - - logger.debug("Handling %s -> %s", last_token, next_token) - - if not events and next_token >= self._last_poked_id: - break - - @defer.inlineCallbacks - def handle_event(event): - # Only send events for this server. - send_on_behalf_of = event.internal_metadata.get_send_on_behalf_of() - is_mine = self.is_mine_id(event.sender) - if not is_mine and send_on_behalf_of is None: - return - - try: - # Get the state from before the event. - # We need to make sure that this is the state from before - # the event and not from after it. - # Otherwise if the last member on a server in a room is - # banned then it won't receive the event because it won't - # be in the room after the ban. - destinations = yield self.state.get_current_hosts_in_room( - event.room_id, latest_event_ids=event.prev_event_ids(), - ) - except Exception: - logger.exception( - "Failed to calculate hosts in room for event: %s", - event.event_id, - ) - return - - destinations = set(destinations) - - if send_on_behalf_of is not None: - # If we are sending the event on behalf of another server - # then it already has the event and there is no reason to - # send the event to it. - destinations.discard(send_on_behalf_of) - - logger.debug("Sending %s to %r", event, destinations) - - self._send_pdu(event, destinations) - - @defer.inlineCallbacks - def handle_room_events(events): - for event in events: - yield handle_event(event) - - events_by_room = {} - for event in events: - events_by_room.setdefault(event.room_id, []).append(event) - - yield logcontext.make_deferred_yieldable(defer.gatherResults( - [ - logcontext.run_in_background(handle_room_events, evs) - for evs in itervalues(events_by_room) - ], - consumeErrors=True - )) - - yield self.store.update_federation_out_pos( - "events", next_token - ) - - if events: - now = self.clock.time_msec() - ts = yield self.store.get_received_ts(events[-1].event_id) - - synapse.metrics.event_processing_lag.labels( - "federation_sender").set(now - ts) - synapse.metrics.event_processing_last_ts.labels( - "federation_sender").set(ts) - - events_processed_counter.inc(len(events)) - - event_processing_loop_room_count.labels( - "federation_sender" - ).inc(len(events_by_room)) - - event_processing_loop_counter.labels("federation_sender").inc() - - synapse.metrics.event_processing_positions.labels( - "federation_sender").set(next_token) - - finally: - self._is_processing = False - - def _send_pdu(self, pdu, destinations): - # We loop through all destinations to see whether we already have - # a transaction in progress. If we do, stick it in the pending_pdus - # table and we'll get back to it later. - - order = self._order - self._order += 1 - - destinations = set(destinations) - destinations.discard(self.server_name) - logger.debug("Sending to: %s", str(destinations)) - - if not destinations: - return - - sent_pdus_destination_dist_total.inc(len(destinations)) - sent_pdus_destination_dist_count.inc() - - for destination in destinations: - self.pending_pdus_by_dest.setdefault(destination, []).append( - (pdu, order) - ) - - self._attempt_new_transaction(destination) - - @logcontext.preserve_fn # the caller should not yield on this - @defer.inlineCallbacks - def send_presence(self, states): - """Send the new presence states to the appropriate destinations. - - This actually queues up the presence states ready for sending and - triggers a background task to process them and send out the transactions. - - Args: - states (list(UserPresenceState)) - """ - if not self.hs.config.use_presence: - # No-op if presence is disabled. - return - - # First we queue up the new presence by user ID, so multiple presence - # updates in quick successtion are correctly handled - # We only want to send presence for our own users, so lets always just - # filter here just in case. - self.pending_presence.update({ - state.user_id: state for state in states - if self.is_mine_id(state.user_id) - }) - - # We then handle the new pending presence in batches, first figuring - # out the destinations we need to send each state to and then poking it - # to attempt a new transaction. We linearize this so that we don't - # accidentally mess up the ordering and send multiple presence updates - # in the wrong order - if self._processing_pending_presence: - return - - self._processing_pending_presence = True - try: - while True: - states_map = self.pending_presence - self.pending_presence = {} - - if not states_map: - break - - yield self._process_presence_inner(list(states_map.values())) - except Exception: - logger.exception("Error sending presence states to servers") - finally: - self._processing_pending_presence = False - - @measure_func("txnqueue._process_presence") - @defer.inlineCallbacks - def _process_presence_inner(self, states): - """Given a list of states populate self.pending_presence_by_dest and - poke to send a new transaction to each destination - - Args: - states (list(UserPresenceState)) - """ - hosts_and_states = yield get_interested_remotes(self.store, states, self.state) - - for destinations, states in hosts_and_states: - for destination in destinations: - if destination == self.server_name: - continue - - self.pending_presence_by_dest.setdefault( - destination, {} - ).update({ - state.user_id: state for state in states - }) - - self._attempt_new_transaction(destination) - - def build_and_send_edu(self, destination, edu_type, content, key=None): - """Construct an Edu object, and queue it for sending - - Args: - destination (str): name of server to send to - edu_type (str): type of EDU to send - content (dict): content of EDU - key (Any|None): clobbering key for this edu - """ - if destination == self.server_name: - logger.info("Not sending EDU to ourselves") - return - - edu = Edu( - origin=self.server_name, - destination=destination, - edu_type=edu_type, - content=content, - ) - - self.send_edu(edu, key) - - def send_edu(self, edu, key): - """Queue an EDU for sending - - Args: - edu (Edu): edu to send - key (Any|None): clobbering key for this edu - """ - if key: - self.pending_edus_keyed_by_dest.setdefault( - edu.destination, {} - )[(edu.edu_type, key)] = edu - else: - self.pending_edus_by_dest.setdefault(edu.destination, []).append(edu) - - self._attempt_new_transaction(edu.destination) - - def send_device_messages(self, destination): - if destination == self.server_name: - logger.info("Not sending device update to ourselves") - return - - self._attempt_new_transaction(destination) - - def get_current_token(self): - return 0 - - def _attempt_new_transaction(self, destination): - """Try to start a new transaction to this destination - - If there is already a transaction in progress to this destination, - returns immediately. Otherwise kicks off the process of sending a - transaction in the background. - - Args: - destination (str): - - Returns: - None - """ - # list of (pending_pdu, deferred, order) - if destination in self.pending_transactions: - # XXX: pending_transactions can get stuck on by a never-ending - # request at which point pending_pdus_by_dest just keeps growing. - # we need application-layer timeouts of some flavour of these - # requests - logger.debug( - "TX [%s] Transaction already in progress", - destination - ) - return - - logger.debug("TX [%s] Starting transaction loop", destination) - - run_as_background_process( - "federation_transaction_transmission_loop", - self._transaction_transmission_loop, - destination, - ) - - @defer.inlineCallbacks - def _transaction_transmission_loop(self, destination): - pending_pdus = [] - try: - self.pending_transactions[destination] = 1 - - # This will throw if we wouldn't retry. We do this here so we fail - # quickly, but we will later check this again in the http client, - # hence why we throw the result away. - yield get_retry_limiter(destination, self.clock, self.store) - - pending_pdus = [] - while True: - device_message_edus, device_stream_id, dev_list_id = ( - yield self._get_new_device_messages(destination) - ) - - # BEGIN CRITICAL SECTION - # - # In order to avoid a race condition, we need to make sure that - # the following code (from popping the queues up to the point - # where we decide if we actually have any pending messages) is - # atomic - otherwise new PDUs or EDUs might arrive in the - # meantime, but not get sent because we hold the - # pending_transactions flag. - - pending_pdus = self.pending_pdus_by_dest.pop(destination, []) - - # We can only include at most 50 PDUs per transactions - pending_pdus, leftover_pdus = pending_pdus[:50], pending_pdus[50:] - if leftover_pdus: - self.pending_pdus_by_dest[destination] = leftover_pdus - - pending_edus = self.pending_edus_by_dest.pop(destination, []) - - # We can only include at most 100 EDUs per transactions - pending_edus, leftover_edus = pending_edus[:100], pending_edus[100:] - if leftover_edus: - self.pending_edus_by_dest[destination] = leftover_edus - - pending_presence = self.pending_presence_by_dest.pop(destination, {}) - - pending_edus.extend( - self.pending_edus_keyed_by_dest.pop(destination, {}).values() - ) - - pending_edus.extend(device_message_edus) - if pending_presence: - pending_edus.append( - Edu( - origin=self.server_name, - destination=destination, - edu_type="m.presence", - content={ - "push": [ - format_user_presence_state( - presence, self.clock.time_msec() - ) - for presence in pending_presence.values() - ] - }, - ) - ) - - if pending_pdus: - logger.debug("TX [%s] len(pending_pdus_by_dest[dest]) = %d", - destination, len(pending_pdus)) - - if not pending_pdus and not pending_edus: - logger.debug("TX [%s] Nothing to send", destination) - self.last_device_stream_id_by_dest[destination] = ( - device_stream_id - ) - return - - # END CRITICAL SECTION - - success = yield self._send_new_transaction( - destination, pending_pdus, pending_edus, - ) - if success: - sent_transactions_counter.inc() - sent_edus_counter.inc(len(pending_edus)) - for edu in pending_edus: - sent_edus_by_type.labels(edu.edu_type).inc() - # Remove the acknowledged device messages from the database - # Only bother if we actually sent some device messages - if device_message_edus: - yield self.store.delete_device_msgs_for_remote( - destination, device_stream_id - ) - logger.info("Marking as sent %r %r", destination, dev_list_id) - yield self.store.mark_as_sent_devices_by_remote( - destination, dev_list_id - ) - - self.last_device_stream_id_by_dest[destination] = device_stream_id - self.last_device_list_stream_id_by_dest[destination] = dev_list_id - else: - break - except NotRetryingDestination as e: - logger.debug( - "TX [%s] not ready for retry yet (next retry at %s) - " - "dropping transaction for now", - destination, - datetime.datetime.fromtimestamp( - (e.retry_last_ts + e.retry_interval) / 1000.0 - ), - ) - except FederationDeniedError as e: - logger.info(e) - except HttpResponseException as e: - logger.warning( - "TX [%s] Received %d response to transaction: %s", - destination, e.code, e, - ) - except RequestSendFailed as e: - logger.warning("TX [%s] Failed to send transaction: %s", destination, e) - - for p, _ in pending_pdus: - logger.info("Failed to send event %s to %s", p.event_id, - destination) - except Exception: - logger.exception( - "TX [%s] Failed to send transaction", - destination, - ) - for p, _ in pending_pdus: - logger.info("Failed to send event %s to %s", p.event_id, - destination) - finally: - # We want to be *very* sure we delete this after we stop processing - self.pending_transactions.pop(destination, None) - - @defer.inlineCallbacks - def _get_new_device_messages(self, destination): - last_device_stream_id = self.last_device_stream_id_by_dest.get(destination, 0) - to_device_stream_id = self.store.get_to_device_stream_token() - contents, stream_id = yield self.store.get_new_device_msgs_for_remote( - destination, last_device_stream_id, to_device_stream_id - ) - edus = [ - Edu( - origin=self.server_name, - destination=destination, - edu_type="m.direct_to_device", - content=content, - ) - for content in contents - ] - - last_device_list = self.last_device_list_stream_id_by_dest.get(destination, 0) - now_stream_id, results = yield self.store.get_devices_by_remote( - destination, last_device_list - ) - edus.extend( - Edu( - origin=self.server_name, - destination=destination, - edu_type="m.device_list_update", - content=content, - ) - for content in results - ) - defer.returnValue((edus, stream_id, now_stream_id)) - - @measure_func("_send_new_transaction") - @defer.inlineCallbacks - def _send_new_transaction(self, destination, pending_pdus, pending_edus): - - # Sort based on the order field - pending_pdus.sort(key=lambda t: t[1]) - pdus = [x[0] for x in pending_pdus] - edus = pending_edus - - success = True - - logger.debug("TX [%s] _attempt_new_transaction", destination) - - txn_id = str(self._next_txn_id) - - logger.debug( - "TX [%s] {%s} Attempting new transaction" - " (pdus: %d, edus: %d)", - destination, txn_id, - len(pdus), - len(edus), - ) - - logger.debug("TX [%s] Persisting transaction...", destination) - - transaction = Transaction.create_new( - origin_server_ts=int(self.clock.time_msec()), - transaction_id=txn_id, - origin=self.server_name, - destination=destination, - pdus=pdus, - edus=edus, - ) - - self._next_txn_id += 1 - - yield self.transaction_actions.prepare_to_send(transaction) - - logger.debug("TX [%s] Persisted transaction", destination) - logger.info( - "TX [%s] {%s} Sending transaction [%s]," - " (PDUs: %d, EDUs: %d)", - destination, txn_id, - transaction.transaction_id, - len(pdus), - len(edus), - ) - - # Actually send the transaction - - # FIXME (erikj): This is a bit of a hack to make the Pdu age - # keys work - def json_data_cb(): - data = transaction.get_dict() - now = int(self.clock.time_msec()) - if "pdus" in data: - for p in data["pdus"]: - if "age_ts" in p: - unsigned = p.setdefault("unsigned", {}) - unsigned["age"] = now - int(p["age_ts"]) - del p["age_ts"] - return data - - try: - response = yield self.transport_layer.send_transaction( - transaction, json_data_cb - ) - code = 200 - except HttpResponseException as e: - code = e.code - response = e.response - - if e.code in (401, 404, 429) or 500 <= e.code: - logger.info( - "TX [%s] {%s} got %d response", - destination, txn_id, code - ) - raise e - - logger.info( - "TX [%s] {%s} got %d response", - destination, txn_id, code - ) - - yield self.transaction_actions.delivered( - transaction, code, response - ) - - logger.debug("TX [%s] {%s} Marked as delivered", destination, txn_id) - - if code == 200: - for e_id, r in response.get("pdus", {}).items(): - if "error" in r: - logger.warn( - "TX [%s] {%s} Remote returned error for %s: %s", - destination, txn_id, e_id, r, - ) - else: - for p in pdus: - logger.warn( - "TX [%s] {%s} Failed to send event %s", - destination, txn_id, p.event_id, - ) - success = False - - defer.returnValue(success) diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py index e424c40fdf..0cdb31178f 100644 --- a/synapse/federation/transport/client.py +++ b/synapse/federation/transport/client.py @@ -173,7 +173,7 @@ class TransportLayerClient(object): # generated by the json_data_callback. json_data = transaction.get_dict() - path = _create_v1_path("/send/%s", transaction.transaction_id) + path = _create_v1_path("/send/%s/", transaction.transaction_id) response = yield self.client.put_json( transaction.destination, diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py index efb6bdca48..96d680a5ad 100644 --- a/synapse/federation/transport/server.py +++ b/synapse/federation/transport/server.py @@ -312,7 +312,7 @@ class BaseFederationServlet(object): class FederationSendServlet(BaseFederationServlet): - PATH = "/send/(?P<transaction_id>[^/]*)/?" + PATH = "/send/(?P<transaction_id>[^/]*)/" def __init__(self, handler, server_name, **kwargs): super(FederationSendServlet, self).__init__( @@ -378,7 +378,7 @@ class FederationSendServlet(BaseFederationServlet): class FederationEventServlet(BaseFederationServlet): - PATH = "/event/(?P<event_id>[^/]*)/?" + PATH = "/event/(?P<event_id>[^/]*)/" # This is when someone asks for a data item for a given server data_id pair. def on_GET(self, origin, content, query, event_id): @@ -386,7 +386,7 @@ class FederationEventServlet(BaseFederationServlet): class FederationStateServlet(BaseFederationServlet): - PATH = "/state/(?P<context>[^/]*)/?" + PATH = "/state/(?P<context>[^/]*)/" # This is when someone asks for all data for a given context. def on_GET(self, origin, content, query, context): @@ -398,7 +398,7 @@ class FederationStateServlet(BaseFederationServlet): class FederationStateIdsServlet(BaseFederationServlet): - PATH = "/state_ids/(?P<room_id>[^/]*)/?" + PATH = "/state_ids/(?P<room_id>[^/]*)/" def on_GET(self, origin, content, query, room_id): return self.handler.on_state_ids_request( @@ -409,7 +409,7 @@ class FederationStateIdsServlet(BaseFederationServlet): class FederationBackfillServlet(BaseFederationServlet): - PATH = "/backfill/(?P<context>[^/]*)/?" + PATH = "/backfill/(?P<context>[^/]*)/" def on_GET(self, origin, content, query, context): versions = [x.decode('ascii') for x in query[b"v"]] @@ -1080,7 +1080,7 @@ class FederationGroupsCategoriesServlet(BaseFederationServlet): """Get all categories for a group """ PATH = ( - "/groups/(?P<group_id>[^/]*)/categories/?" + "/groups/(?P<group_id>[^/]*)/categories/" ) @defer.inlineCallbacks @@ -1150,7 +1150,7 @@ class FederationGroupsRolesServlet(BaseFederationServlet): """Get roles in a group """ PATH = ( - "/groups/(?P<group_id>[^/]*)/roles/?" + "/groups/(?P<group_id>[^/]*)/roles/" ) @defer.inlineCallbacks diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index d8d86d6ff3..ac09d03ba9 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -165,6 +165,7 @@ class BaseHandler(object): member_event.room_id, "leave", ratelimit=False, + require_consent=False, ) except Exception as e: logger.exception("Error kicking guest user: %s" % (e,)) diff --git a/synapse/handlers/auth.py b/synapse/handlers/auth.py index 2abd9af94f..caad9ae2dd 100644 --- a/synapse/handlers/auth.py +++ b/synapse/handlers/auth.py @@ -35,6 +35,7 @@ from synapse.api.errors import ( StoreError, SynapseError, ) +from synapse.api.ratelimiting import Ratelimiter from synapse.module_api import ModuleApi from synapse.types import UserID from synapse.util import logcontext @@ -99,6 +100,11 @@ class AuthHandler(BaseHandler): login_types.append(t) self._supported_login_types = login_types + self._account_ratelimiter = Ratelimiter() + self._failed_attempts_ratelimiter = Ratelimiter() + + self._clock = self.hs.get_clock() + @defer.inlineCallbacks def validate_user_via_ui_auth(self, requester, request_body, clientip): """ @@ -568,7 +574,12 @@ class AuthHandler(BaseHandler): Returns: defer.Deferred: (unicode) canonical_user_id, or None if zero or multiple matches + + Raises: + LimitExceededError if the ratelimiter's login requests count for this + user is too high too proceed. """ + self.ratelimit_login_per_account(user_id) res = yield self._find_user_id_and_pwd_hash(user_id) if res is not None: defer.returnValue(res[0]) @@ -634,6 +645,8 @@ class AuthHandler(BaseHandler): StoreError if there was a problem accessing the database SynapseError if there was a problem with the request LoginError if there was an authentication problem. + LimitExceededError if the ratelimiter's login requests count for this + user is too high too proceed. """ if username.startswith('@'): @@ -643,6 +656,8 @@ class AuthHandler(BaseHandler): username, self.hs.hostname ).to_string() + self.ratelimit_login_per_account(qualified_user_id) + login_type = login_submission.get("type") known_login_type = False @@ -715,9 +730,16 @@ class AuthHandler(BaseHandler): if not known_login_type: raise SynapseError(400, "Unknown login type %s" % login_type) - # unknown username or invalid password. We raise a 403 here, but note - # that if we're doing user-interactive login, it turns all LoginErrors - # into a 401 anyway. + # unknown username or invalid password. + self._failed_attempts_ratelimiter.ratelimit( + qualified_user_id.lower(), time_now_s=self._clock.time(), + rate_hz=self.hs.config.rc_login_failed_attempts.per_second, + burst_count=self.hs.config.rc_login_failed_attempts.burst_count, + update=True, + ) + + # We raise a 403 here, but note that if we're doing user-interactive + # login, it turns all LoginErrors into a 401 anyway. raise LoginError( 403, "Invalid password", errcode=Codes.FORBIDDEN @@ -735,6 +757,10 @@ class AuthHandler(BaseHandler): password (unicode): the provided password Returns: (unicode) the canonical_user_id, or None if unknown user / bad password + + Raises: + LimitExceededError if the ratelimiter's login requests count for this + user is too high too proceed. """ lookupres = yield self._find_user_id_and_pwd_hash(user_id) if not lookupres: @@ -763,6 +789,7 @@ class AuthHandler(BaseHandler): auth_api.validate_macaroon(macaroon, "login", True, user_id) except Exception: raise AuthError(403, "Invalid token", errcode=Codes.FORBIDDEN) + self.ratelimit_login_per_account(user_id) yield self.auth.check_auth_blocking(user_id) defer.returnValue(user_id) @@ -934,6 +961,33 @@ class AuthHandler(BaseHandler): else: return defer.succeed(False) + def ratelimit_login_per_account(self, user_id): + """Checks whether the process must be stopped because of ratelimiting. + + Checks against two ratelimiters: the generic one for login attempts per + account and the one specific to failed attempts. + + Args: + user_id (unicode): complete @user:id + + Raises: + LimitExceededError if one of the ratelimiters' login requests count + for this user is too high too proceed. + """ + self._failed_attempts_ratelimiter.ratelimit( + user_id.lower(), time_now_s=self._clock.time(), + rate_hz=self.hs.config.rc_login_failed_attempts.per_second, + burst_count=self.hs.config.rc_login_failed_attempts.burst_count, + update=False, + ) + + self._account_ratelimiter.ratelimit( + user_id.lower(), time_now_s=self._clock.time(), + rate_hz=self.hs.config.rc_login_account.per_second, + burst_count=self.hs.config.rc_login_account.burst_count, + update=True, + ) + @attr.s class MacaroonGenerator(object): diff --git a/synapse/handlers/deactivate_account.py b/synapse/handlers/deactivate_account.py index 75fe50c42c..97d3f31d98 100644 --- a/synapse/handlers/deactivate_account.py +++ b/synapse/handlers/deactivate_account.py @@ -164,6 +164,7 @@ class DeactivateAccountHandler(BaseHandler): room_id, "leave", ratelimit=False, + require_consent=False, ) except Exception: logger.exception( diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py index 8b113307d2..fe128d9c88 100644 --- a/synapse/handlers/directory.py +++ b/synapse/handlers/directory.py @@ -44,6 +44,7 @@ class DirectoryHandler(BaseHandler): self.appservice_handler = hs.get_application_service_handler() self.event_creation_handler = hs.get_event_creation_handler() self.config = hs.config + self.enable_room_list_search = hs.config.enable_room_list_search self.federation = hs.get_federation_client() hs.get_federation_registry().register_query_handler( @@ -411,6 +412,13 @@ class DirectoryHandler(BaseHandler): if visibility not in ["public", "private"]: raise SynapseError(400, "Invalid visibility setting") + if visibility == "public" and not self.enable_room_list_search: + # The room list has been disabled. + raise AuthError( + 403, + "This user is not permitted to publish rooms to the room list" + ) + room = yield self.store.get_room(room_id) if room is None: raise SynapseError(400, "Unknown room") diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py index f772e62c28..d883e98381 100644 --- a/synapse/handlers/events.py +++ b/synapse/handlers/events.py @@ -19,7 +19,7 @@ import random from twisted.internet import defer from synapse.api.constants import EventTypes, Membership -from synapse.api.errors import AuthError +from synapse.api.errors import AuthError, SynapseError from synapse.events import EventBase from synapse.events.utils import serialize_event from synapse.types import UserID @@ -61,6 +61,11 @@ class EventStreamHandler(BaseHandler): If `only_keys` is not None, events from keys will be sent down. """ + if room_id: + blocked = yield self.store.is_room_blocked(room_id) + if blocked: + raise SynapseError(403, "This room has been blocked on this server") + # send any outstanding server notices to the user. yield self._server_notices_sender.on_user_syncing(auth_user_id) diff --git a/synapse/handlers/initial_sync.py b/synapse/handlers/initial_sync.py index 563bb3cea3..7dfae78db0 100644 --- a/synapse/handlers/initial_sync.py +++ b/synapse/handlers/initial_sync.py @@ -18,7 +18,7 @@ import logging from twisted.internet import defer from synapse.api.constants import EventTypes, Membership -from synapse.api.errors import AuthError, Codes +from synapse.api.errors import AuthError, Codes, SynapseError from synapse.events.utils import serialize_event from synapse.events.validator import EventValidator from synapse.handlers.presence import format_user_presence_state @@ -262,6 +262,10 @@ class InitialSyncHandler(BaseHandler): A JSON serialisable dict with the snapshot of the room. """ + blocked = yield self.store.is_room_blocked(room_id) + if blocked: + raise SynapseError(403, "This room has been blocked on this server") + user_id = requester.user.to_string() membership, member_event_id = yield self._check_in_room_or_world_readable( diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index c762b58902..9b41c7b205 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -243,12 +243,19 @@ class EventCreationHandler(object): self.spam_checker = hs.get_spam_checker() - if self.config.block_events_without_consent_error is not None: + self._block_events_without_consent_error = ( + self.config.block_events_without_consent_error + ) + + # we need to construct a ConsentURIBuilder here, as it checks that the necessary + # config options, but *only* if we have a configuration for which we are + # going to need it. + if self._block_events_without_consent_error: self._consent_uri_builder = ConsentURIBuilder(self.config) @defer.inlineCallbacks def create_event(self, requester, event_dict, token_id=None, txn_id=None, - prev_events_and_hashes=None): + prev_events_and_hashes=None, require_consent=True): """ Given a dict from a client, create a new event. @@ -269,6 +276,9 @@ class EventCreationHandler(object): where *hashes* is a map from algorithm to hash. If None, they will be requested from the database. + + require_consent (bool): Whether to check if the requester has + consented to privacy policy. Raises: ResourceLimitError if server is blocked to some resource being exceeded @@ -310,7 +320,7 @@ class EventCreationHandler(object): ) is_exempt = yield self._is_exempt_from_privacy_policy(builder, requester) - if not is_exempt: + if require_consent and not is_exempt: yield self.assert_accepted_privacy_policy(requester) if token_id is not None: @@ -378,7 +388,7 @@ class EventCreationHandler(object): Raises: ConsentNotGivenError: if the user has not given consent yet """ - if self.config.block_events_without_consent_error is None: + if self._block_events_without_consent_error is None: return # exempt AS users from needing consent @@ -405,7 +415,7 @@ class EventCreationHandler(object): consent_uri = self._consent_uri_builder.build_user_consent_uri( requester.user.localpart, ) - msg = self.config.block_events_without_consent_error % { + msg = self._block_events_without_consent_error % { 'consent_uri': consent_uri, } raise ConsentNotGivenError( diff --git a/synapse/handlers/receipts.py b/synapse/handlers/receipts.py index 1728089667..274d2946ad 100644 --- a/synapse/handlers/receipts.py +++ b/synapse/handlers/receipts.py @@ -16,9 +16,8 @@ import logging from twisted.internet import defer -from synapse.types import get_domain_from_id - -from ._base import BaseHandler +from synapse.handlers._base import BaseHandler +from synapse.types import ReadReceipt logger = logging.getLogger(__name__) @@ -42,13 +41,13 @@ class ReceiptsHandler(BaseHandler): """Called when we receive an EDU of type m.receipt from a remote HS. """ receipts = [ - { - "room_id": room_id, - "receipt_type": receipt_type, - "user_id": user_id, - "event_ids": user_values["event_ids"], - "data": user_values.get("data", {}), - } + ReadReceipt( + room_id=room_id, + receipt_type=receipt_type, + user_id=user_id, + event_ids=user_values["event_ids"], + data=user_values.get("data", {}), + ) for room_id, room_values in content.items() for receipt_type, users in room_values.items() for user_id, user_values in users.items() @@ -64,14 +63,12 @@ class ReceiptsHandler(BaseHandler): max_batch_id = None for receipt in receipts: - room_id = receipt["room_id"] - receipt_type = receipt["receipt_type"] - user_id = receipt["user_id"] - event_ids = receipt["event_ids"] - data = receipt["data"] - res = yield self.store.insert_receipt( - room_id, receipt_type, user_id, event_ids, data + receipt.room_id, + receipt.receipt_type, + receipt.user_id, + receipt.event_ids, + receipt.data, ) if not res: @@ -89,7 +86,7 @@ class ReceiptsHandler(BaseHandler): # no new receipts defer.returnValue(False) - affected_room_ids = list(set([r["room_id"] for r in receipts])) + affected_room_ids = list(set([r.room_id for r in receipts])) self.notifier.on_new_event( "receipt_key", max_batch_id, rooms=affected_room_ids @@ -107,49 +104,21 @@ class ReceiptsHandler(BaseHandler): """Called when a client tells us a local user has read up to the given event_id in the room. """ - receipt = { - "room_id": room_id, - "receipt_type": receipt_type, - "user_id": user_id, - "event_ids": [event_id], - "data": { + receipt = ReadReceipt( + room_id=room_id, + receipt_type=receipt_type, + user_id=user_id, + event_ids=[event_id], + data={ "ts": int(self.clock.time_msec()), - } - } + }, + ) is_new = yield self._handle_new_receipts([receipt]) if not is_new: return - # Work out which remote servers should be poked and poke them. - - # TODO: optimise this to move some of the work to the workers. - data = receipt["data"] - - # XXX why does this not use state.get_current_hosts_in_room() ? - users = yield self.state.get_current_user_in_room(room_id) - remotedomains = set(get_domain_from_id(u) for u in users) - remotedomains = remotedomains.copy() - remotedomains.discard(self.server_name) - - logger.debug("Sending receipt to: %r", remotedomains) - - for domain in remotedomains: - self.federation.build_and_send_edu( - destination=domain, - edu_type="m.receipt", - content={ - room_id: { - receipt_type: { - user_id: { - "event_ids": [event_id], - "data": data, - } - } - }, - }, - key=(room_id, receipt_type, user_id), - ) + yield self.federation.send_read_receipt(receipt) @defer.inlineCallbacks def get_receipts_for_room(self, room_id, to_key): diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py index 03130edc54..68f73d3793 100644 --- a/synapse/handlers/register.py +++ b/synapse/handlers/register.py @@ -23,6 +23,7 @@ from synapse.api.constants import LoginType from synapse.api.errors import ( AuthError, Codes, + ConsentNotGivenError, InvalidCaptchaError, LimitExceededError, RegistrationError, @@ -311,6 +312,10 @@ class RegistrationHandler(BaseHandler): ) else: yield self._join_user_to_room(fake_requester, r) + except ConsentNotGivenError as e: + # Technically not necessary to pull out this error though + # moving away from bare excepts is a good thing to do. + logger.error("Failed to join new user to %r: %r", r, e) except Exception as e: logger.error("Failed to join new user to %r: %r", r, e) @@ -629,8 +634,8 @@ class RegistrationHandler(BaseHandler): allowed, time_allowed = self.ratelimiter.can_do_action( address, time_now_s=time_now, - rate_hz=self.hs.config.rc_registration_requests_per_second, - burst_count=self.hs.config.rc_registration_request_burst_count, + rate_hz=self.hs.config.rc_registration.per_second, + burst_count=self.hs.config.rc_registration.burst_count, ) if not allowed: diff --git a/synapse/handlers/room_list.py b/synapse/handlers/room_list.py index afa508d729..d6c9d56007 100644 --- a/synapse/handlers/room_list.py +++ b/synapse/handlers/room_list.py @@ -44,6 +44,7 @@ EMPTY_THIRD_PARTY_ID = ThirdPartyInstanceID(None, None) class RoomListHandler(BaseHandler): def __init__(self, hs): super(RoomListHandler, self).__init__(hs) + self.enable_room_list_search = hs.config.enable_room_list_search self.response_cache = ResponseCache(hs, "room_list") self.remote_response_cache = ResponseCache(hs, "remote_room_list", timeout_ms=30 * 1000) @@ -66,10 +67,17 @@ class RoomListHandler(BaseHandler): appservice and network id to use an appservice specific one. Setting to None returns all public rooms across all lists. """ + if not self.enable_room_list_search: + return defer.succeed({ + "chunk": [], + "total_room_count_estimate": 0, + }) + logger.info( "Getting public room list: limit=%r, since=%r, search=%r, network=%r", limit, since_token, bool(search_filter), network_tuple, ) + if search_filter: # We explicitly don't bother caching searches or requests for # appservice specific lists. @@ -441,6 +449,12 @@ class RoomListHandler(BaseHandler): def get_remote_public_room_list(self, server_name, limit=None, since_token=None, search_filter=None, include_all_networks=False, third_party_instance_id=None,): + if not self.enable_room_list_search: + defer.returnValue({ + "chunk": [], + "total_room_count_estimate": 0, + }) + if search_filter: # We currently don't support searching across federation, so we have # to do it manually without pagination diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py index 190ea2c7b1..71ce5b54e5 100644 --- a/synapse/handlers/room_member.py +++ b/synapse/handlers/room_member.py @@ -160,6 +160,7 @@ class RoomMemberHandler(object): txn_id=None, ratelimit=True, content=None, + require_consent=True, ): user_id = target.to_string() @@ -185,6 +186,7 @@ class RoomMemberHandler(object): token_id=requester.access_token_id, txn_id=txn_id, prev_events_and_hashes=prev_events_and_hashes, + require_consent=require_consent, ) # Check if this event matches the previous membership event for the user. @@ -232,6 +234,10 @@ class RoomMemberHandler(object): self.copy_room_tags_and_direct_to_room( predecessor["room_id"], room_id, user_id, ) + # Move over old push rules + self.store.move_push_rules_from_room_to_room_for_user( + predecessor["room_id"], room_id, user_id, + ) elif event.membership == Membership.LEAVE: if prev_member_event_id: prev_member_event = yield self.store.get_event(prev_member_event_id) @@ -301,6 +307,7 @@ class RoomMemberHandler(object): third_party_signed=None, ratelimit=True, content=None, + require_consent=True, ): key = (room_id,) @@ -315,6 +322,7 @@ class RoomMemberHandler(object): third_party_signed=third_party_signed, ratelimit=ratelimit, content=content, + require_consent=require_consent, ) defer.returnValue(result) @@ -331,6 +339,7 @@ class RoomMemberHandler(object): third_party_signed=None, ratelimit=True, content=None, + require_consent=True, ): content_specified = bool(content) if content is None: @@ -512,6 +521,7 @@ class RoomMemberHandler(object): ratelimit=ratelimit, prev_events_and_hashes=prev_events_and_hashes, content=content, + require_consent=require_consent, ) defer.returnValue(res) diff --git a/synapse/handlers/user_directory.py b/synapse/handlers/user_directory.py index c21da8343a..7dc0e236e7 100644 --- a/synapse/handlers/user_directory.py +++ b/synapse/handlers/user_directory.py @@ -38,18 +38,8 @@ class UserDirectoryHandler(object): world_readable or publically joinable room. We keep a database table up to date by streaming changes of the current state and recalculating whether users should be in the directory or not when necessary. - - For each user in the directory we also store a room_id which is public and that the - user is joined to. This allows us to ignore history_visibility and join_rules changes - for that user in all other public rooms, as we know they'll still be in at least - one public room. """ - INITIAL_ROOM_SLEEP_MS = 50 - INITIAL_ROOM_SLEEP_COUNT = 100 - INITIAL_ROOM_BATCH_SIZE = 100 - INITIAL_USER_SLEEP_MS = 10 - def __init__(self, hs): self.store = hs.get_datastore() self.state = hs.get_state_handler() @@ -59,11 +49,6 @@ class UserDirectoryHandler(object): self.is_mine_id = hs.is_mine_id self.update_user_directory = hs.config.update_user_directory self.search_all_users = hs.config.user_directory_search_all_users - - # When start up for the first time we need to populate the user_directory. - # This is a set of user_id's we've inserted already - self.initially_handled_users = set() - # The current position in the current_state_delta stream self.pos = None @@ -126,7 +111,7 @@ class UserDirectoryHandler(object): # Support users are for diagnostics and should not appear in the user directory. if not is_support: yield self.store.update_profile_in_user_dir( - user_id, profile.display_name, profile.avatar_url, None + user_id, profile.display_name, profile.avatar_url ) @defer.inlineCallbacks @@ -143,10 +128,9 @@ class UserDirectoryHandler(object): if self.pos is None: self.pos = yield self.store.get_user_directory_stream_pos() - # If still None then we need to do the initial fill of directory + # If still None then the initial background update hasn't happened yet if self.pos is None: - yield self._do_initial_spam() - self.pos = yield self.store.get_user_directory_stream_pos() + defer.returnValue(None) # Loop round handling deltas until we're up to date while True: @@ -168,113 +152,6 @@ class UserDirectoryHandler(object): yield self.store.update_user_directory_stream_pos(self.pos) @defer.inlineCallbacks - def _do_initial_spam(self): - """Populates the user_directory from the current state of the DB, used - when synapse first starts with user_directory support - """ - new_pos = yield self.store.get_max_stream_id_in_current_state_deltas() - - # Delete any existing entries just in case there are any - yield self.store.delete_all_from_user_dir() - - # We process by going through each existing room at a time. - room_ids = yield self.store.get_all_rooms() - - logger.info("Doing initial update of user directory. %d rooms", len(room_ids)) - num_processed_rooms = 0 - - for room_id in room_ids: - logger.info("Handling room %d/%d", num_processed_rooms + 1, len(room_ids)) - yield self._handle_initial_room(room_id) - num_processed_rooms += 1 - yield self.clock.sleep(self.INITIAL_ROOM_SLEEP_MS / 1000.0) - - logger.info("Processed all rooms.") - - if self.search_all_users: - num_processed_users = 0 - user_ids = yield self.store.get_all_local_users() - logger.info( - "Doing initial update of user directory. %d users", len(user_ids) - ) - for user_id in user_ids: - # We add profiles for all users even if they don't match the - # include pattern, just in case we want to change it in future - logger.info( - "Handling user %d/%d", num_processed_users + 1, len(user_ids) - ) - yield self._handle_local_user(user_id) - num_processed_users += 1 - yield self.clock.sleep(self.INITIAL_USER_SLEEP_MS / 1000.0) - - logger.info("Processed all users") - - self.initially_handled_users = None - - yield self.store.update_user_directory_stream_pos(new_pos) - - @defer.inlineCallbacks - def _handle_initial_room(self, room_id): - """ - Called when we initially fill out user_directory one room at a time - """ - is_in_room = yield self.store.is_host_joined(room_id, self.server_name) - if not is_in_room: - return - - is_public = yield self.store.is_room_world_readable_or_publicly_joinable( - room_id - ) - - users_with_profile = yield self.state.get_current_user_in_room(room_id) - user_ids = set(users_with_profile) - unhandled_users = user_ids - self.initially_handled_users - - yield self.store.add_profiles_to_user_dir( - {user_id: users_with_profile[user_id] for user_id in unhandled_users}, - ) - - self.initially_handled_users |= unhandled_users - - # We now go and figure out the new users who share rooms with user entries - # We sleep aggressively here as otherwise it can starve resources. - # We also batch up inserts/updates, but try to avoid too many at once. - to_insert = set() - count = 0 - for user_id in user_ids: - if count % self.INITIAL_ROOM_SLEEP_COUNT == 0: - yield self.clock.sleep(self.INITIAL_ROOM_SLEEP_MS / 1000.0) - - if not self.is_mine_id(user_id): - count += 1 - continue - - if self.store.get_if_app_services_interested_in_user(user_id): - count += 1 - continue - - for other_user_id in user_ids: - if user_id == other_user_id: - continue - - if count % self.INITIAL_ROOM_SLEEP_COUNT == 0: - yield self.clock.sleep(self.INITIAL_ROOM_SLEEP_MS / 1000.0) - count += 1 - - user_set = (user_id, other_user_id) - to_insert.add(user_set) - - if len(to_insert) > self.INITIAL_ROOM_BATCH_SIZE: - yield self.store.add_users_who_share_room( - room_id, not is_public, to_insert - ) - to_insert.clear() - - if to_insert: - yield self.store.add_users_who_share_room(room_id, not is_public, to_insert) - to_insert.clear() - - @defer.inlineCallbacks def _handle_deltas(self, deltas): """Called with the state deltas to process """ @@ -423,7 +300,9 @@ class UserDirectoryHandler(object): row = yield self.store.get_user_in_directory(user_id) if not row: - yield self.store.add_profiles_to_user_dir({user_id: profile}) + yield self.store.update_profile_in_user_dir( + user_id, profile.display_name, profile.avatar_url + ) @defer.inlineCallbacks def _handle_new_user(self, room_id, user_id, profile): @@ -435,9 +314,9 @@ class UserDirectoryHandler(object): """ logger.debug("Adding new user to dir, %r", user_id) - row = yield self.store.get_user_in_directory(user_id) - if not row: - yield self.store.add_profiles_to_user_dir({user_id: profile}) + yield self.store.update_profile_in_user_dir( + user_id, profile.display_name, profile.avatar_url + ) is_public = yield self.store.is_room_world_readable_or_publicly_joinable( room_id @@ -445,34 +324,39 @@ class UserDirectoryHandler(object): # Now we update users who share rooms with users. users_with_profile = yield self.state.get_current_user_in_room(room_id) - to_insert = set() + if is_public: + yield self.store.add_users_in_public_rooms(room_id, (user_id,)) + else: + to_insert = set() - # First, if they're our user then we need to update for every user - if self.is_mine_id(user_id): + # First, if they're our user then we need to update for every user + if self.is_mine_id(user_id): - is_appservice = self.store.get_if_app_services_interested_in_user(user_id) + is_appservice = self.store.get_if_app_services_interested_in_user( + user_id + ) - # We don't care about appservice users. - if not is_appservice: - for other_user_id in users_with_profile: - if user_id == other_user_id: - continue + # We don't care about appservice users. + if not is_appservice: + for other_user_id in users_with_profile: + if user_id == other_user_id: + continue - to_insert.add((user_id, other_user_id)) + to_insert.add((user_id, other_user_id)) - # Next we need to update for every local user in the room - for other_user_id in users_with_profile: - if user_id == other_user_id: - continue + # Next we need to update for every local user in the room + for other_user_id in users_with_profile: + if user_id == other_user_id: + continue - is_appservice = self.store.get_if_app_services_interested_in_user( - other_user_id - ) - if self.is_mine_id(other_user_id) and not is_appservice: - to_insert.add((other_user_id, user_id)) + is_appservice = self.store.get_if_app_services_interested_in_user( + other_user_id + ) + if self.is_mine_id(other_user_id) and not is_appservice: + to_insert.add((other_user_id, user_id)) - if to_insert: - yield self.store.add_users_who_share_room(room_id, not is_public, to_insert) + if to_insert: + yield self.store.add_users_who_share_private_room(room_id, to_insert) @defer.inlineCallbacks def _handle_remove_user(self, room_id, user_id): @@ -487,10 +371,10 @@ class UserDirectoryHandler(object): # Remove user from sharing tables yield self.store.remove_user_who_share_room(user_id, room_id) - # Are they still in a room with members? If not, remove them entirely. - users_in_room_with = yield self.store.get_users_who_share_room_from_dir(user_id) + # Are they still in any rooms? If not, remove them entirely. + rooms_user_is_in = yield self.store.get_user_dir_rooms_user_is_in(user_id) - if len(users_in_room_with) == 0: + if len(rooms_user_is_in) == 0: yield self.store.remove_from_user_dir(user_id) @defer.inlineCallbacks @@ -517,9 +401,7 @@ class UserDirectoryHandler(object): new_avatar = event.content.get("avatar_url") if prev_name != new_name or prev_avatar != new_avatar: - yield self.store.update_profile_in_user_dir( - user_id, new_name, new_avatar, room_id - ) + yield self.store.update_profile_in_user_dir(user_id, new_name, new_avatar) @defer.inlineCallbacks def _get_key_change(self, prev_event_id, event_id, key_name, public_value): diff --git a/synapse/replication/tcp/resource.py b/synapse/replication/tcp/resource.py index fd59f1595f..47cdf30bd3 100644 --- a/synapse/replication/tcp/resource.py +++ b/synapse/replication/tcp/resource.py @@ -16,6 +16,7 @@ """ import logging +import random from six import itervalues @@ -74,6 +75,8 @@ class ReplicationStreamer(object): self.notifier = hs.get_notifier() self._server_notices_sender = hs.get_server_notices_sender() + self._replication_torture_level = hs.config.replication_torture_level + # Current connections. self.connections = [] @@ -157,10 +160,23 @@ class ReplicationStreamer(object): for stream in self.streams: stream.advance_current_token() - for stream in self.streams: + all_streams = self.streams + + if self._replication_torture_level is not None: + # there is no guarantee about ordering between the streams, + # so let's shuffle them around a bit when we are in torture mode. + all_streams = list(all_streams) + random.shuffle(all_streams) + + for stream in all_streams: if stream.last_token == stream.upto_token: continue + if self._replication_torture_level: + yield self.clock.sleep( + self._replication_torture_level / 1000.0 + ) + logger.debug( "Getting stream: %s: %s -> %s", stream.NAME, stream.last_token, stream.upto_token diff --git a/synapse/rest/client/v1/admin.py b/synapse/rest/client/v1/admin.py index 2a29f0c2af..e788769639 100644 --- a/synapse/rest/client/v1/admin.py +++ b/synapse/rest/client/v1/admin.py @@ -490,40 +490,54 @@ class ShutdownRoomRestServlet(ClientV1RestServlet): requester_user_id = requester.user.to_string() - logger.info("Shutting down room %r", room_id) + logger.info( + "Shutting down room %r, joining to new room: %r", + room_id, new_room_id, + ) + # This will work even if the room is already blocked, but that is + # desirable in case the first attempt at blocking the room failed below. yield self.store.block_room(room_id, requester_user_id) users = yield self.state.get_current_user_in_room(room_id) kicked_users = [] + failed_to_kick_users = [] for user_id in users: if not self.hs.is_mine_id(user_id): continue logger.info("Kicking %r from %r...", user_id, room_id) - target_requester = create_requester(user_id) - yield self.room_member_handler.update_membership( - requester=target_requester, - target=target_requester.user, - room_id=room_id, - action=Membership.LEAVE, - content={}, - ratelimit=False - ) + try: + target_requester = create_requester(user_id) + yield self.room_member_handler.update_membership( + requester=target_requester, + target=target_requester.user, + room_id=room_id, + action=Membership.LEAVE, + content={}, + ratelimit=False, + require_consent=False, + ) - yield self.room_member_handler.forget(target_requester.user, room_id) + yield self.room_member_handler.forget(target_requester.user, room_id) - yield self.room_member_handler.update_membership( - requester=target_requester, - target=target_requester.user, - room_id=new_room_id, - action=Membership.JOIN, - content={}, - ratelimit=False - ) + yield self.room_member_handler.update_membership( + requester=target_requester, + target=target_requester.user, + room_id=new_room_id, + action=Membership.JOIN, + content={}, + ratelimit=False, + require_consent=False, + ) - kicked_users.append(user_id) + kicked_users.append(user_id) + except Exception: + logger.exception( + "Failed to leave old room and join new room for %r", user_id, + ) + failed_to_kick_users.append(user_id) yield self.event_creation_handler.create_and_send_nonmember_event( room_creator_requester, @@ -544,6 +558,7 @@ class ShutdownRoomRestServlet(ClientV1RestServlet): defer.returnValue((200, { "kicked_users": kicked_users, + "failed_to_kick_users": failed_to_kick_users, "local_aliases": aliases_for_room, "new_room_id": new_room_id, })) diff --git a/synapse/rest/client/v1/login.py b/synapse/rest/client/v1/login.py index 6121c5b6df..8d56effbb8 100644 --- a/synapse/rest/client/v1/login.py +++ b/synapse/rest/client/v1/login.py @@ -22,6 +22,7 @@ from twisted.internet import defer from twisted.web.client import PartialDownloadError from synapse.api.errors import Codes, LoginError, SynapseError +from synapse.api.ratelimiting import Ratelimiter from synapse.http.server import finish_request from synapse.http.servlet import ( RestServlet, @@ -97,6 +98,7 @@ class LoginRestServlet(ClientV1RestServlet): self.registration_handler = hs.get_registration_handler() self.handlers = hs.get_handlers() self._well_known_builder = WellKnownBuilder(hs) + self._address_ratelimiter = Ratelimiter() def on_GET(self, request): flows = [] @@ -129,6 +131,13 @@ class LoginRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_POST(self, request): + self._address_ratelimiter.ratelimit( + request.getClientIP(), time_now_s=self.hs.clock.time(), + rate_hz=self.hs.config.rc_login_address.per_second, + burst_count=self.hs.config.rc_login_address.burst_count, + update=True, + ) + login_submission = parse_json_object_from_request(request) try: if self.jwt_enabled and (login_submission["type"] == @@ -285,6 +294,7 @@ class LoginRestServlet(ClientV1RestServlet): raise LoginError(401, "Invalid JWT", errcode=Codes.UNAUTHORIZED) user_id = UserID(user, self.hs.hostname).to_string() + auth_handler = self.auth_handler registered_user_id = yield auth_handler.check_user_exists(user_id) if registered_user_id: diff --git a/synapse/rest/client/v2_alpha/register.py b/synapse/rest/client/v2_alpha/register.py index 6f34029431..6d235262c8 100644 --- a/synapse/rest/client/v2_alpha/register.py +++ b/synapse/rest/client/v2_alpha/register.py @@ -210,8 +210,8 @@ class RegisterRestServlet(RestServlet): allowed, time_allowed = self.ratelimiter.can_do_action( client_addr, time_now_s=time_now, - rate_hz=self.hs.config.rc_registration_requests_per_second, - burst_count=self.hs.config.rc_registration_request_burst_count, + rate_hz=self.hs.config.rc_registration.per_second, + burst_count=self.hs.config.rc_registration.burst_count, update=False, ) diff --git a/synapse/rest/client/v2_alpha/user_directory.py b/synapse/rest/client/v2_alpha/user_directory.py index cac0624ba7..36b02de37f 100644 --- a/synapse/rest/client/v2_alpha/user_directory.py +++ b/synapse/rest/client/v2_alpha/user_directory.py @@ -59,6 +59,12 @@ class UserDirectorySearchRestServlet(RestServlet): requester = yield self.auth.get_user_by_req(request, allow_guest=False) user_id = requester.user.to_string() + if not self.hs.config.user_directory_search_enabled: + defer.returnValue((200, { + "limited": False, + "results": [], + })) + body = parse_json_object_from_request(request) limit = body.get("limit", 10) diff --git a/synapse/server.py b/synapse/server.py index 72835e8c86..dc8f1ccb8c 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -42,7 +42,7 @@ from synapse.federation.federation_server import ( ReplicationFederationHandlerRegistry, ) from synapse.federation.send_queue import FederationRemoteSendQueue -from synapse.federation.transaction_queue import TransactionQueue +from synapse.federation.sender import FederationSender from synapse.federation.transport.client import TransportLayerClient from synapse.groups.attestations import GroupAttestationSigning, GroupAttestionRenewer from synapse.groups.groups_server import GroupsServerHandler @@ -185,6 +185,10 @@ class HomeServer(object): 'registration_handler', ] + REQUIRED_ON_MASTER_STARTUP = [ + "user_directory_handler", + ] + # This is overridden in derived application classes # (such as synapse.app.homeserver.SynapseHomeServer) and gives the class to be # instantiated during setup() for future return by get_datastore() @@ -221,6 +225,15 @@ class HomeServer(object): conn.commit() logger.info("Finished setting up.") + def setup_master(self): + """ + Some handlers have side effects on instantiation (like registering + background updates). This function causes them to be fetched, and + therefore instantiated, to run those side effects. + """ + for i in self.REQUIRED_ON_MASTER_STARTUP: + getattr(self, "get_" + i)() + def get_reactor(self): """ Fetch the Twisted reactor in use by this HomeServer. @@ -421,7 +434,7 @@ class HomeServer(object): def build_federation_sender(self): if self.should_send_federation(): - return TransactionQueue(self) + return FederationSender(self) elif not self.config.worker_app: return FederationRemoteSendQueue(self) else: diff --git a/synapse/server.pyi b/synapse/server.pyi index fb8df56cd5..3ba3a967c2 100644 --- a/synapse/server.pyi +++ b/synapse/server.pyi @@ -1,5 +1,6 @@ import synapse.api.auth import synapse.config.homeserver +import synapse.federation.sender import synapse.federation.transaction_queue import synapse.federation.transport.client import synapse.handlers @@ -62,7 +63,7 @@ class HomeServer(object): def get_set_password_handler(self) -> synapse.handlers.set_password.SetPasswordHandler: pass - def get_federation_sender(self) -> synapse.federation.transaction_queue.TransactionQueue: + def get_federation_sender(self) -> synapse.federation.sender.FederationSender: pass def get_federation_transport_client(self) -> synapse.federation.transport.client.TransportLayerClient: diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index a0333d5309..7e3903859b 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -767,18 +767,25 @@ class SQLBaseStore(object): """ allvalues = {} allvalues.update(keyvalues) - allvalues.update(values) allvalues.update(insertion_values) + if not values: + latter = "NOTHING" + else: + allvalues.update(values) + latter = ( + "UPDATE SET " + ", ".join(k + "=EXCLUDED." + k for k in values) + ) + sql = ( "INSERT INTO %s (%s) VALUES (%s) " - "ON CONFLICT (%s) DO UPDATE SET %s" + "ON CONFLICT (%s) DO %s" ) % ( table, ", ".join(k for k in allvalues), ", ".join("?" for _ in allvalues), ", ".join(k for k in keyvalues), - ", ".join(k + "=EXCLUDED." + k for k in values), + latter ) txn.execute(sql, list(allvalues.values())) diff --git a/synapse/storage/background_updates.py b/synapse/storage/background_updates.py index 60cdc884e6..a2f8c23a65 100644 --- a/synapse/storage/background_updates.py +++ b/synapse/storage/background_updates.py @@ -52,7 +52,9 @@ class BackgroundUpdatePerformance(object): Returns: A duration in ms as a float """ - if self.total_item_count == 0: + if self.avg_duration_ms == 0: + return 0 + elif self.total_item_count == 0: return None else: # Use the exponential moving average so that we can adapt to @@ -64,7 +66,9 @@ class BackgroundUpdatePerformance(object): Returns: A duration in ms as a float """ - if self.total_item_count == 0: + if self.total_duration_ms == 0: + return 0 + elif self.total_item_count == 0: return None else: return float(self.total_item_count) / float(self.total_duration_ms) diff --git a/synapse/storage/push_rule.py b/synapse/storage/push_rule.py index 6a5028961d..4b8438c3e9 100644 --- a/synapse/storage/push_rule.py +++ b/synapse/storage/push_rule.py @@ -186,6 +186,63 @@ class PushRulesWorkerStore(ApplicationServiceWorkerStore, defer.returnValue(results) @defer.inlineCallbacks + def move_push_rule_from_room_to_room( + self, new_room_id, user_id, rule, + ): + """Move a single push rule from one room to another for a specific user. + + Args: + new_room_id (str): ID of the new room. + user_id (str): ID of user the push rule belongs to. + rule (Dict): A push rule. + """ + # Create new rule id + rule_id_scope = '/'.join(rule["rule_id"].split('/')[:-1]) + new_rule_id = rule_id_scope + "/" + new_room_id + + # Change room id in each condition + for condition in rule.get("conditions", []): + if condition.get("key") == "room_id": + condition["pattern"] = new_room_id + + # Add the rule for the new room + yield self.add_push_rule( + user_id=user_id, + rule_id=new_rule_id, + priority_class=rule["priority_class"], + conditions=rule["conditions"], + actions=rule["actions"], + ) + + # Delete push rule for the old room + yield self.delete_push_rule(user_id, rule["rule_id"]) + + @defer.inlineCallbacks + def move_push_rules_from_room_to_room_for_user( + self, old_room_id, new_room_id, user_id, + ): + """Move all of the push rules from one room to another for a specific + user. + + Args: + old_room_id (str): ID of the old room. + new_room_id (str): ID of the new room. + user_id (str): ID of user to copy push rules for. + """ + # Retrieve push rules for this user + user_push_rules = yield self.get_push_rules_for_user(user_id) + + # Get rules relating to the old room, move them to the new room, then + # delete them from the old room + for rule in user_push_rules: + conditions = rule.get("conditions", []) + if any((c.get("key") == "room_id" and + c.get("pattern") == old_room_id) for c in conditions): + self.move_push_rule_from_room_to_room( + new_room_id, user_id, rule, + ) + + @defer.inlineCallbacks def bulk_get_push_rules_for_room(self, event, context): state_group = context.state_group if not state_group: diff --git a/synapse/storage/room.py b/synapse/storage/room.py index 41c65e112a..a979d4860a 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -500,10 +500,22 @@ class RoomStore(RoomWorkerStore, SearchStore): @defer.inlineCallbacks def block_room(self, room_id, user_id): - yield self._simple_insert( + """Marks the room as blocked. Can be called multiple times. + + Args: + room_id (str): Room to block + user_id (str): Who blocked it + + Returns: + Deferred + """ + yield self._simple_upsert( table="blocked_rooms", - values={ + keyvalues={ "room_id": room_id, + }, + values={}, + insertion_values={ "user_id": user_id, }, desc="block_room", diff --git a/synapse/storage/schema/delta/53/user_dir_populate.sql b/synapse/storage/schema/delta/53/user_dir_populate.sql new file mode 100644 index 0000000000..ffcc896b58 --- /dev/null +++ b/synapse/storage/schema/delta/53/user_dir_populate.sql @@ -0,0 +1,30 @@ +/* Copyright 2019 New Vector Ltd + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +-- Set up staging tables +INSERT INTO background_updates (update_name, progress_json) VALUES + ('populate_user_directory_createtables', '{}'); + +-- Run through each room and update the user directory according to who is in it +INSERT INTO background_updates (update_name, progress_json, depends_on) VALUES + ('populate_user_directory_process_rooms', '{}', 'populate_user_directory_createtables'); + +-- Insert all users, if search_all_users is on +INSERT INTO background_updates (update_name, progress_json, depends_on) VALUES + ('populate_user_directory_process_users', '{}', 'populate_user_directory_process_rooms'); + +-- Clean up staging tables +INSERT INTO background_updates (update_name, progress_json, depends_on) VALUES + ('populate_user_directory_cleanup', '{}', 'populate_user_directory_process_users'); diff --git a/synapse/storage/schema/delta/53/user_share.sql b/synapse/storage/schema/delta/53/user_share.sql index 14424ded0c..5831b1a6f8 100644 --- a/synapse/storage/schema/delta/53/user_share.sql +++ b/synapse/storage/schema/delta/53/user_share.sql @@ -16,9 +16,6 @@ -- Old disused version of the tables below. DROP TABLE IF EXISTS users_who_share_rooms; --- This is no longer used because it's duplicated by the users_who_share_public_rooms -DROP TABLE IF EXISTS users_in_public_rooms; - -- Tables keeping track of what users share rooms. This is a map of local users -- to local or remote users, per room. Remote users cannot be in the user_id -- column, only the other_user_id column. There are two tables, one for public diff --git a/synapse/storage/schema/delta/53/users_in_public_rooms.sql b/synapse/storage/schema/delta/53/users_in_public_rooms.sql new file mode 100644 index 0000000000..f7827ca6d2 --- /dev/null +++ b/synapse/storage/schema/delta/53/users_in_public_rooms.sql @@ -0,0 +1,28 @@ +/* Copyright 2019 New Vector Ltd + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +-- We don't need the old version of this table. +DROP TABLE IF EXISTS users_in_public_rooms; + +-- Old version of users_in_public_rooms +DROP TABLE IF EXISTS users_who_share_public_rooms; + +-- Track what users are in public rooms. +CREATE TABLE IF NOT EXISTS users_in_public_rooms ( + user_id TEXT NOT NULL, + room_id TEXT NOT NULL +); + +CREATE UNIQUE INDEX users_in_public_rooms_u_idx ON users_in_public_rooms(user_id, room_id); diff --git a/synapse/storage/user_directory.py b/synapse/storage/user_directory.py index 2317d22ed6..d360e857d1 100644 --- a/synapse/storage/user_directory.py +++ b/synapse/storage/user_directory.py @@ -16,22 +16,301 @@ import logging import re -from six import iteritems - from twisted.internet import defer from synapse.api.constants import EventTypes, JoinRules +from synapse.storage.background_updates import BackgroundUpdateStore from synapse.storage.engines import PostgresEngine, Sqlite3Engine from synapse.storage.state import StateFilter from synapse.types import get_domain_from_id, get_localpart_from_id -from synapse.util.caches.descriptors import cached, cachedInlineCallbacks - -from ._base import SQLBaseStore +from synapse.util.caches.descriptors import cached logger = logging.getLogger(__name__) -class UserDirectoryStore(SQLBaseStore): +TEMP_TABLE = "_temp_populate_user_directory" + + +class UserDirectoryStore(BackgroundUpdateStore): + + # How many records do we calculate before sending it to + # add_users_who_share_private_rooms? + SHARE_PRIVATE_WORKING_SET = 500 + + def __init__(self, db_conn, hs): + super(UserDirectoryStore, self).__init__(db_conn, hs) + + self.server_name = hs.hostname + + self.register_background_update_handler( + "populate_user_directory_createtables", + self._populate_user_directory_createtables, + ) + self.register_background_update_handler( + "populate_user_directory_process_rooms", + self._populate_user_directory_process_rooms, + ) + self.register_background_update_handler( + "populate_user_directory_process_users", + self._populate_user_directory_process_users, + ) + self.register_background_update_handler( + "populate_user_directory_cleanup", self._populate_user_directory_cleanup + ) + + @defer.inlineCallbacks + def _populate_user_directory_createtables(self, progress, batch_size): + + # Get all the rooms that we want to process. + def _make_staging_area(txn): + sql = ( + "CREATE TABLE IF NOT EXISTS " + + TEMP_TABLE + + "_rooms(room_id TEXT NOT NULL, events BIGINT NOT NULL)" + ) + txn.execute(sql) + + sql = ( + "CREATE TABLE IF NOT EXISTS " + + TEMP_TABLE + + "_position(position TEXT NOT NULL)" + ) + txn.execute(sql) + + # Get rooms we want to process from the database + sql = """ + SELECT room_id, count(*) FROM current_state_events + GROUP BY room_id + """ + txn.execute(sql) + rooms = [{"room_id": x[0], "events": x[1]} for x in txn.fetchall()] + self._simple_insert_many_txn(txn, TEMP_TABLE + "_rooms", rooms) + del rooms + + # If search all users is on, get all the users we want to add. + if self.hs.config.user_directory_search_all_users: + sql = ( + "CREATE TABLE IF NOT EXISTS " + + TEMP_TABLE + + "_users(user_id TEXT NOT NULL)" + ) + txn.execute(sql) + + txn.execute("SELECT name FROM users") + users = [{"user_id": x[0]} for x in txn.fetchall()] + + self._simple_insert_many_txn(txn, TEMP_TABLE + "_users", users) + + new_pos = yield self.get_max_stream_id_in_current_state_deltas() + yield self.runInteraction( + "populate_user_directory_temp_build", _make_staging_area + ) + yield self._simple_insert(TEMP_TABLE + "_position", {"position": new_pos}) + + yield self._end_background_update("populate_user_directory_createtables") + defer.returnValue(1) + + @defer.inlineCallbacks + def _populate_user_directory_cleanup(self, progress, batch_size): + """ + Update the user directory stream position, then clean up the old tables. + """ + position = yield self._simple_select_one_onecol( + TEMP_TABLE + "_position", None, "position" + ) + yield self.update_user_directory_stream_pos(position) + + def _delete_staging_area(txn): + txn.execute("DROP TABLE IF EXISTS " + TEMP_TABLE + "_rooms") + txn.execute("DROP TABLE IF EXISTS " + TEMP_TABLE + "_users") + txn.execute("DROP TABLE IF EXISTS " + TEMP_TABLE + "_position") + + yield self.runInteraction( + "populate_user_directory_cleanup", _delete_staging_area + ) + + yield self._end_background_update("populate_user_directory_cleanup") + defer.returnValue(1) + + @defer.inlineCallbacks + def _populate_user_directory_process_rooms(self, progress, batch_size): + + state = self.hs.get_state_handler() + + # If we don't have progress filed, delete everything. + if not progress: + yield self.delete_all_from_user_dir() + + def _get_next_batch(txn): + sql = """ + SELECT room_id FROM %s + ORDER BY events DESC + LIMIT %s + """ % ( + TEMP_TABLE + "_rooms", + str(batch_size), + ) + txn.execute(sql) + rooms_to_work_on = txn.fetchall() + + if not rooms_to_work_on: + return None + + rooms_to_work_on = [x[0] for x in rooms_to_work_on] + + # Get how many are left to process, so we can give status on how + # far we are in processing + txn.execute("SELECT COUNT(*) FROM " + TEMP_TABLE + "_rooms") + progress["remaining"] = txn.fetchone()[0] + + return rooms_to_work_on + + rooms_to_work_on = yield self.runInteraction( + "populate_user_directory_temp_read", _get_next_batch + ) + + # No more rooms -- complete the transaction. + if not rooms_to_work_on: + yield self._end_background_update("populate_user_directory_process_rooms") + defer.returnValue(1) + + logger.info( + "Processing the next %d rooms of %d remaining" + % (len(rooms_to_work_on), progress["remaining"]) + ) + + for room_id in rooms_to_work_on: + is_in_room = yield self.is_host_joined(room_id, self.server_name) + + if is_in_room: + is_public = yield self.is_room_world_readable_or_publicly_joinable( + room_id + ) + + users_with_profile = yield state.get_current_user_in_room(room_id) + user_ids = set(users_with_profile) + + # Update each user in the user directory. + for user_id, profile in users_with_profile.items(): + yield self.update_profile_in_user_dir( + user_id, profile.display_name, profile.avatar_url + ) + + to_insert = set() + + if is_public: + for user_id in user_ids: + if self.get_if_app_services_interested_in_user(user_id): + continue + + to_insert.add(user_id) + + if to_insert: + yield self.add_users_in_public_rooms(room_id, to_insert) + to_insert.clear() + else: + for user_id in user_ids: + if not self.hs.is_mine_id(user_id): + continue + + if self.get_if_app_services_interested_in_user(user_id): + continue + + for other_user_id in user_ids: + if user_id == other_user_id: + continue + + user_set = (user_id, other_user_id) + to_insert.add(user_set) + + # If it gets too big, stop and write to the database + # to prevent storing too much in RAM. + if len(to_insert) >= self.SHARE_PRIVATE_WORKING_SET: + yield self.add_users_who_share_private_room( + room_id, to_insert + ) + to_insert.clear() + + if to_insert: + yield self.add_users_who_share_private_room(room_id, to_insert) + to_insert.clear() + + # We've finished a room. Delete it from the table. + yield self._simple_delete_one(TEMP_TABLE + "_rooms", {"room_id": room_id}) + # Update the remaining counter. + progress["remaining"] -= 1 + yield self.runInteraction( + "populate_user_directory", + self._background_update_progress_txn, + "populate_user_directory_process_rooms", + progress, + ) + + defer.returnValue(len(rooms_to_work_on)) + + @defer.inlineCallbacks + def _populate_user_directory_process_users(self, progress, batch_size): + """ + If search_all_users is enabled, add all of the users to the user directory. + """ + if not self.hs.config.user_directory_search_all_users: + yield self._end_background_update("populate_user_directory_process_users") + defer.returnValue(1) + + def _get_next_batch(txn): + sql = "SELECT user_id FROM %s LIMIT %s" % ( + TEMP_TABLE + "_users", + str(batch_size), + ) + txn.execute(sql) + users_to_work_on = txn.fetchall() + + if not users_to_work_on: + return None + + users_to_work_on = [x[0] for x in users_to_work_on] + + # Get how many are left to process, so we can give status on how + # far we are in processing + sql = "SELECT COUNT(*) FROM " + TEMP_TABLE + "_users" + txn.execute(sql) + progress["remaining"] = txn.fetchone()[0] + + return users_to_work_on + + users_to_work_on = yield self.runInteraction( + "populate_user_directory_temp_read", _get_next_batch + ) + + # No more users -- complete the transaction. + if not users_to_work_on: + yield self._end_background_update("populate_user_directory_process_users") + defer.returnValue(1) + + logger.info( + "Processing the next %d users of %d remaining" + % (len(users_to_work_on), progress["remaining"]) + ) + + for user_id in users_to_work_on: + profile = yield self.get_profileinfo(get_localpart_from_id(user_id)) + yield self.update_profile_in_user_dir( + user_id, profile.display_name, profile.avatar_url + ) + + # We've finished processing a user. Delete it from the table. + yield self._simple_delete_one(TEMP_TABLE + "_users", {"user_id": user_id}) + # Update the remaining counter. + progress["remaining"] -= 1 + yield self.runInteraction( + "populate_user_directory", + self._background_update_progress_txn, + "populate_user_directory_process_users", + progress, + ) + + defer.returnValue(len(users_to_work_on)) + @defer.inlineCallbacks def is_room_world_readable_or_publicly_joinable(self, room_id): """Check if the room is either world_readable or publically joinable @@ -63,89 +342,16 @@ class UserDirectoryStore(SQLBaseStore): defer.returnValue(False) - def add_profiles_to_user_dir(self, users_with_profile): - """Add profiles to the user directory - - Args: - users_with_profile (dict): Users to add to directory in the form of - mapping of user_id -> ProfileInfo + def update_profile_in_user_dir(self, user_id, display_name, avatar_url): + """ + Update or add a user's profile in the user directory. """ - if isinstance(self.database_engine, PostgresEngine): - # We weight the loclpart most highly, then display name and finally - # server name - sql = """ - INSERT INTO user_directory_search(user_id, vector) - VALUES (?, - setweight(to_tsvector('english', ?), 'A') - || setweight(to_tsvector('english', ?), 'D') - || setweight(to_tsvector('english', COALESCE(?, '')), 'B') - ) - """ - args = ( - ( - user_id, - get_localpart_from_id(user_id), - get_domain_from_id(user_id), - profile.display_name, - ) - for user_id, profile in iteritems(users_with_profile) - ) - elif isinstance(self.database_engine, Sqlite3Engine): - sql = """ - INSERT INTO user_directory_search(user_id, value) - VALUES (?,?) - """ - args = tuple( - ( - user_id, - "%s %s" % (user_id, p.display_name) if p.display_name else user_id, - ) - for user_id, p in iteritems(users_with_profile) - ) - else: - # This should be unreachable. - raise Exception("Unrecognized database engine") - - def _add_profiles_to_user_dir_txn(txn): - txn.executemany(sql, args) - self._simple_insert_many_txn( - txn, - table="user_directory", - values=[ - { - "user_id": user_id, - "room_id": None, - "display_name": profile.display_name, - "avatar_url": profile.avatar_url, - } - for user_id, profile in iteritems(users_with_profile) - ], - ) - for user_id in users_with_profile: - txn.call_after(self.get_user_in_directory.invalidate, (user_id,)) - - return self.runInteraction( - "add_profiles_to_user_dir", _add_profiles_to_user_dir_txn - ) - - @defer.inlineCallbacks - def update_user_in_user_dir(self, user_id, room_id): - yield self._simple_update_one( - table="user_directory", - keyvalues={"user_id": user_id}, - updatevalues={"room_id": room_id}, - desc="update_user_in_user_dir", - ) - self.get_user_in_directory.invalidate((user_id,)) - - def update_profile_in_user_dir(self, user_id, display_name, avatar_url, room_id): def _update_profile_in_user_dir_txn(txn): new_entry = self._simple_upsert_txn( txn, table="user_directory", keyvalues={"user_id": user_id}, - insertion_values={"room_id": room_id}, values={"display_name": display_name, "avatar_url": avatar_url}, lock=False, # We're only inserter ) @@ -242,14 +448,7 @@ class UserDirectoryStore(SQLBaseStore): txn, table="user_directory_search", keyvalues={"user_id": user_id} ) self._simple_delete_txn( - txn, - table="users_who_share_public_rooms", - keyvalues={"user_id": user_id}, - ) - self._simple_delete_txn( - txn, - table="users_who_share_public_rooms", - keyvalues={"other_user_id": user_id}, + txn, table="users_in_public_rooms", keyvalues={"user_id": user_id} ) self._simple_delete_txn( txn, @@ -271,9 +470,9 @@ class UserDirectoryStore(SQLBaseStore): in the given room_id """ user_ids_share_pub = yield self._simple_select_onecol( - table="users_who_share_public_rooms", + table="users_in_public_rooms", keyvalues={"room_id": room_id}, - retcol="other_user_id", + retcol="user_id", desc="get_users_in_dir_due_to_room", ) @@ -290,18 +489,6 @@ class UserDirectoryStore(SQLBaseStore): defer.returnValue(user_ids) @defer.inlineCallbacks - def get_all_rooms(self): - """Get all room_ids we've ever known about, in ascending order of "size" - """ - sql = """ - SELECT room_id FROM current_state_events - GROUP BY room_id - ORDER BY count(*) ASC - """ - rows = yield self._execute("get_all_rooms", None, sql) - defer.returnValue([room_id for room_id, in rows]) - - @defer.inlineCallbacks def get_all_local_users(self): """Get all local users """ @@ -311,26 +498,19 @@ class UserDirectoryStore(SQLBaseStore): rows = yield self._execute("get_all_local_users", None, sql) defer.returnValue([name for name, in rows]) - def add_users_who_share_room(self, room_id, share_private, user_id_tuples): - """Insert entries into the users_who_share_*_rooms table. The first + def add_users_who_share_private_room(self, room_id, user_id_tuples): + """Insert entries into the users_who_share_private_rooms table. The first user should be a local user. Args: room_id (str) - share_private (bool): Is the room private user_id_tuples([(str, str)]): iterable of 2-tuple of user IDs. """ def _add_users_who_share_room_txn(txn): - - if share_private: - tbl = "users_who_share_private_rooms" - else: - tbl = "users_who_share_public_rooms" - self._simple_upsert_many_txn( txn, - table=tbl, + table="users_who_share_private_rooms", key_names=["user_id", "other_user_id", "room_id"], key_values=[ (user_id, other_user_id, room_id) @@ -339,15 +519,35 @@ class UserDirectoryStore(SQLBaseStore): value_names=(), value_values=None, ) - for user_id, other_user_id in user_id_tuples: - txn.call_after( - self.get_users_who_share_room_from_dir.invalidate, (user_id,) - ) return self.runInteraction( "add_users_who_share_room", _add_users_who_share_room_txn ) + def add_users_in_public_rooms(self, room_id, user_ids): + """Insert entries into the users_who_share_private_rooms table. The first + user should be a local user. + + Args: + room_id (str) + user_ids (list[str]) + """ + + def _add_users_in_public_rooms_txn(txn): + + self._simple_upsert_many_txn( + txn, + table="users_in_public_rooms", + key_names=["user_id", "room_id"], + key_values=[(user_id, room_id) for user_id in user_ids], + value_names=(), + value_values=None, + ) + + return self.runInteraction( + "add_users_in_public_rooms", _add_users_in_public_rooms_txn + ) + def remove_user_who_share_room(self, user_id, room_id): """ Deletes entries in the users_who_share_*_rooms table. The first @@ -371,25 +571,18 @@ class UserDirectoryStore(SQLBaseStore): ) self._simple_delete_txn( txn, - table="users_who_share_public_rooms", + table="users_in_public_rooms", keyvalues={"user_id": user_id, "room_id": room_id}, ) - self._simple_delete_txn( - txn, - table="users_who_share_public_rooms", - keyvalues={"other_user_id": user_id, "room_id": room_id}, - ) - txn.call_after( - self.get_users_who_share_room_from_dir.invalidate, (user_id,) - ) return self.runInteraction( "remove_user_who_share_room", _remove_user_who_share_room_txn ) - @cachedInlineCallbacks(max_entries=500000, iterable=True) - def get_users_who_share_room_from_dir(self, user_id): - """Returns the set of users who share a room with `user_id` + @defer.inlineCallbacks + def get_user_dir_rooms_user_is_in(self, user_id): + """ + Returns the rooms that a user is in. Args: user_id(str): Must be a local user @@ -400,23 +593,19 @@ class UserDirectoryStore(SQLBaseStore): rows = yield self._simple_select_onecol( table="users_who_share_private_rooms", keyvalues={"user_id": user_id}, - retcol="other_user_id", - desc="get_users_who_share_room_with_user", + retcol="room_id", + desc="get_rooms_user_is_in", ) pub_rows = yield self._simple_select_onecol( - table="users_who_share_public_rooms", + table="users_in_public_rooms", keyvalues={"user_id": user_id}, - retcol="other_user_id", - desc="get_users_who_share_room_with_user", + retcol="room_id", + desc="get_rooms_user_is_in", ) users = set(pub_rows) users.update(rows) - - # Remove the user themselves from this list. - users.discard(user_id) - defer.returnValue(list(users)) @defer.inlineCallbacks @@ -452,10 +641,9 @@ class UserDirectoryStore(SQLBaseStore): def _delete_all_from_user_dir_txn(txn): txn.execute("DELETE FROM user_directory") txn.execute("DELETE FROM user_directory_search") - txn.execute("DELETE FROM users_who_share_public_rooms") + txn.execute("DELETE FROM users_in_public_rooms") txn.execute("DELETE FROM users_who_share_private_rooms") txn.call_after(self.get_user_in_directory.invalidate_all) - txn.call_after(self.get_users_who_share_room_from_dir.invalidate_all) return self.runInteraction( "delete_all_from_user_dir", _delete_all_from_user_dir_txn @@ -560,23 +748,19 @@ class UserDirectoryStore(SQLBaseStore): """ if self.hs.config.user_directory_search_all_users: - # make s.user_id null to keep the ordering algorithm happy - join_clause = """ - CROSS JOIN (SELECT NULL as user_id) AS s - """ - join_args = () - where_clause = "1=1" + join_args = (user_id,) + where_clause = "user_id != ?" else: - join_clause = """ - LEFT JOIN ( - SELECT other_user_id AS user_id FROM users_who_share_public_rooms - UNION - SELECT other_user_id AS user_id FROM users_who_share_private_rooms - WHERE user_id = ? - ) AS p USING (user_id) - """ join_args = (user_id,) - where_clause = "p.user_id IS NOT NULL" + where_clause = """ + ( + EXISTS (select 1 from users_in_public_rooms WHERE user_id = t.user_id) + OR EXISTS ( + SELECT 1 FROM users_who_share_private_rooms + WHERE user_id = ? AND other_user_id = t.user_id + ) + ) + """ if isinstance(self.database_engine, PostgresEngine): full_query, exact_query, prefix_query = _parse_query_postgres(search_term) @@ -588,9 +772,8 @@ class UserDirectoryStore(SQLBaseStore): # search: (domain, _, display name, localpart) sql = """ SELECT d.user_id AS user_id, display_name, avatar_url - FROM user_directory_search + FROM user_directory_search as t INNER JOIN user_directory AS d USING (user_id) - %s WHERE %s AND vector @@ to_tsquery('english', ?) @@ -617,7 +800,6 @@ class UserDirectoryStore(SQLBaseStore): avatar_url IS NULL LIMIT ? """ % ( - join_clause, where_clause, ) args = join_args + (full_query, exact_query, prefix_query, limit + 1) @@ -626,9 +808,8 @@ class UserDirectoryStore(SQLBaseStore): sql = """ SELECT d.user_id AS user_id, display_name, avatar_url - FROM user_directory_search + FROM user_directory_search as t INNER JOIN user_directory AS d USING (user_id) - %s WHERE %s AND value MATCH ? @@ -638,7 +819,6 @@ class UserDirectoryStore(SQLBaseStore): avatar_url IS NULL LIMIT ? """ % ( - join_clause, where_clause, ) args = join_args + (search_query, limit + 1) diff --git a/synapse/types.py b/synapse/types.py index d8cb64addb..3de94b6335 100644 --- a/synapse/types.py +++ b/synapse/types.py @@ -16,6 +16,8 @@ import re import string from collections import namedtuple +import attr + from synapse.api.errors import SynapseError @@ -455,3 +457,13 @@ class ThirdPartyInstanceID( @classmethod def create(cls, appservice_id, network_id,): return cls(appservice_id=appservice_id, network_id=network_id) + + +@attr.s(slots=True) +class ReadReceipt(object): + """Information about a read-receipt""" + room_id = attr.ib() + receipt_type = attr.ib() + user_id = attr.ib() + event_ids = attr.ib() + data = attr.ib() diff --git a/tests/api/test_auth.py b/tests/api/test_auth.py index d77f20e876..d0d36f96fa 100644 --- a/tests/api/test_auth.py +++ b/tests/api/test_auth.py @@ -345,6 +345,23 @@ class AuthTestCase(unittest.TestCase): self.assertEquals(e.exception.code, 403) @defer.inlineCallbacks + def test_hs_disabled_no_server_notices_user(self): + """Check that 'hs_disabled_message' works correctly when there is no + server_notices user. + """ + # this should be the default, but we had a bug where the test was doing the wrong + # thing, so let's make it explicit + self.hs.config.server_notices_mxid = None + + self.hs.config.hs_disabled = True + self.hs.config.hs_disabled_message = "Reason for being disabled" + with self.assertRaises(ResourceLimitError) as e: + yield self.auth.check_auth_blocking() + self.assertEquals(e.exception.admin_contact, self.hs.config.admin_contact) + self.assertEquals(e.exception.errcode, Codes.RESOURCE_LIMIT_EXCEEDED) + self.assertEquals(e.exception.code, 403) + + @defer.inlineCallbacks def test_server_notices_mxid_special_cased(self): self.hs.config.hs_disabled = True user = "@user:server" diff --git a/tests/federation/test_federation_sender.py b/tests/federation/test_federation_sender.py new file mode 100644 index 0000000000..28e7e27416 --- /dev/null +++ b/tests/federation/test_federation_sender.py @@ -0,0 +1,128 @@ +# -*- coding: utf-8 -*- +# Copyright 2019 New Vector Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from mock import Mock + +from twisted.internet import defer + +from synapse.types import ReadReceipt + +from tests.unittest import HomeserverTestCase + + +class FederationSenderTestCases(HomeserverTestCase): + def make_homeserver(self, reactor, clock): + return super(FederationSenderTestCases, self).setup_test_homeserver( + state_handler=Mock(spec=["get_current_hosts_in_room"]), + federation_transport_client=Mock(spec=["send_transaction"]), + ) + + def test_send_receipts(self): + mock_state_handler = self.hs.get_state_handler() + mock_state_handler.get_current_hosts_in_room.return_value = ["test", "host2"] + + mock_send_transaction = self.hs.get_federation_transport_client().send_transaction + mock_send_transaction.return_value = defer.succeed({}) + + sender = self.hs.get_federation_sender() + receipt = ReadReceipt("room_id", "m.read", "user_id", ["event_id"], {"ts": 1234}) + self.successResultOf(sender.send_read_receipt(receipt)) + + self.pump() + + # expect a call to send_transaction + mock_send_transaction.assert_called_once() + json_cb = mock_send_transaction.call_args[0][1] + data = json_cb() + self.assertEqual(data['edus'], [ + { + 'edu_type': 'm.receipt', + 'content': { + 'room_id': { + 'm.read': { + 'user_id': { + 'event_ids': ['event_id'], + 'data': {'ts': 1234}, + }, + }, + }, + }, + }, + ]) + + def test_send_receipts_with_backoff(self): + """Send two receipts in quick succession; the second should be flushed, but + only after 20ms""" + mock_state_handler = self.hs.get_state_handler() + mock_state_handler.get_current_hosts_in_room.return_value = ["test", "host2"] + + mock_send_transaction = self.hs.get_federation_transport_client().send_transaction + mock_send_transaction.return_value = defer.succeed({}) + + sender = self.hs.get_federation_sender() + receipt = ReadReceipt("room_id", "m.read", "user_id", ["event_id"], {"ts": 1234}) + self.successResultOf(sender.send_read_receipt(receipt)) + + self.pump() + + # expect a call to send_transaction + mock_send_transaction.assert_called_once() + json_cb = mock_send_transaction.call_args[0][1] + data = json_cb() + self.assertEqual(data['edus'], [ + { + 'edu_type': 'm.receipt', + 'content': { + 'room_id': { + 'm.read': { + 'user_id': { + 'event_ids': ['event_id'], + 'data': {'ts': 1234}, + }, + }, + }, + }, + }, + ]) + mock_send_transaction.reset_mock() + + # send the second RR + receipt = ReadReceipt("room_id", "m.read", "user_id", ["other_id"], {"ts": 1234}) + self.successResultOf(sender.send_read_receipt(receipt)) + self.pump() + mock_send_transaction.assert_not_called() + + self.reactor.advance(19) + mock_send_transaction.assert_not_called() + + self.reactor.advance(10) + mock_send_transaction.assert_called_once() + json_cb = mock_send_transaction.call_args[0][1] + data = json_cb() + self.assertEqual(data['edus'], [ + { + 'edu_type': 'm.receipt', + 'content': { + 'room_id': { + 'm.read': { + 'user_id': { + 'event_ids': ['other_id'], + 'data': {'ts': 1234}, + }, + }, + }, + }, + }, + ]) diff --git a/tests/handlers/test_directory.py b/tests/handlers/test_directory.py index 9bf395e923..5b2105bc76 100644 --- a/tests/handlers/test_directory.py +++ b/tests/handlers/test_directory.py @@ -111,7 +111,7 @@ class TestCreateAliasACL(unittest.HomeserverTestCase): servlets = [directory.register_servlets, room.register_servlets] - def prepare(self, hs, reactor, clock): + def prepare(self, reactor, clock, hs): # We cheekily override the config to add custom alias creation rules config = {} config["alias_creation_rules"] = [ @@ -151,3 +151,60 @@ class TestCreateAliasACL(unittest.HomeserverTestCase): ) self.render(request) self.assertEquals(200, channel.code, channel.result) + + +class TestRoomListSearchDisabled(unittest.HomeserverTestCase): + user_id = "@test:test" + + servlets = [directory.register_servlets, room.register_servlets] + + def prepare(self, reactor, clock, hs): + room_id = self.helper.create_room_as(self.user_id) + + request, channel = self.make_request( + "PUT", + b"directory/list/room/%s" % (room_id.encode('ascii'),), + b'{}', + ) + self.render(request) + self.assertEquals(200, channel.code, channel.result) + + self.room_list_handler = hs.get_room_list_handler() + self.directory_handler = hs.get_handlers().directory_handler + + return hs + + def test_disabling_room_list(self): + self.room_list_handler.enable_room_list_search = True + self.directory_handler.enable_room_list_search = True + + # Room list is enabled so we should get some results + request, channel = self.make_request( + "GET", + b"publicRooms", + ) + self.render(request) + self.assertEquals(200, channel.code, channel.result) + self.assertTrue(len(channel.json_body["chunk"]) > 0) + + self.room_list_handler.enable_room_list_search = False + self.directory_handler.enable_room_list_search = False + + # Room list disabled so we should get no results + request, channel = self.make_request( + "GET", + b"publicRooms", + ) + self.render(request) + self.assertEquals(200, channel.code, channel.result) + self.assertTrue(len(channel.json_body["chunk"]) == 0) + + # Room list disabled so we shouldn't be allowed to publish rooms + room_id = self.helper.create_room_as(self.user_id) + request, channel = self.make_request( + "PUT", + b"directory/list/room/%s" % (room_id.encode('ascii'),), + b'{}', + ) + self.render(request) + self.assertEquals(403, channel.code, channel.result) diff --git a/tests/handlers/test_register.py b/tests/handlers/test_register.py index c9c1506273..2217eb2a10 100644 --- a/tests/handlers/test_register.py +++ b/tests/handlers/test_register.py @@ -22,7 +22,7 @@ from synapse.api.errors import ResourceLimitError, SynapseError from synapse.handlers.register import RegistrationHandler from synapse.types import RoomAlias, UserID, create_requester -from tests.utils import setup_test_homeserver +from tests.utils import default_config, setup_test_homeserver from .. import unittest @@ -40,8 +40,16 @@ class RegistrationTestCase(unittest.TestCase): self.mock_distributor = Mock() self.mock_distributor.declare("registered_user") self.mock_captcha_client = Mock() + + hs_config = default_config("test") + + # some of the tests rely on us having a user consent version + hs_config.user_consent_version = "test_consent_version" + hs_config.max_mau_value = 50 + self.hs = yield setup_test_homeserver( self.addCleanup, + config=hs_config, expire_access_token=True, ) self.macaroon_generator = Mock( @@ -50,7 +58,6 @@ class RegistrationTestCase(unittest.TestCase): self.hs.get_macaroon_generator = Mock(return_value=self.macaroon_generator) self.handler = self.hs.get_registration_handler() self.store = self.hs.get_datastore() - self.hs.config.max_mau_value = 50 self.lots_of_users = 100 self.small_number_of_users = 1 @@ -187,12 +194,32 @@ class RegistrationTestCase(unittest.TestCase): @defer.inlineCallbacks def test_auto_create_auto_join_where_no_consent(self): - self.hs.config.user_consent_at_registration = True - self.hs.config.block_events_without_consent_error = "Error" + """Test to ensure that the first user is not auto-joined to a room if + they have not given general consent. + """ + + # Given:- + # * a user must give consent, + # * they have not given that consent + # * The server is configured to auto-join to a room + # (and autocreate if necessary) + + event_creation_handler = self.hs.get_event_creation_handler() + # (Messing with the internals of event_creation_handler is fragile + # but can't see a better way to do this. One option could be to subclass + # the test with custom config.) + event_creation_handler._block_events_without_consent_error = ("Error") + event_creation_handler._consent_uri_builder = Mock() room_alias_str = "#room:test" self.hs.config.auto_join_rooms = [room_alias_str] + + # When:- + # * the user is registered and post consent actions are called res = yield self.handler.register(localpart='jeff') yield self.handler.post_consent_actions(res[0]) + + # Then:- + # * Ensure that they have not been joined to the room rooms = yield self.store.get_rooms_for_user(res[0]) self.assertEqual(len(rooms), 0) diff --git a/tests/handlers/test_typing.py b/tests/handlers/test_typing.py index 6460cbc708..7decb22933 100644 --- a/tests/handlers/test_typing.py +++ b/tests/handlers/test_typing.py @@ -180,7 +180,7 @@ class TypingNotificationsTestCase(unittest.HomeserverTestCase): put_json = self.hs.get_http_client().put_json put_json.assert_called_once_with( "farm", - path="/_matrix/federation/v1/send/1000000", + path="/_matrix/federation/v1/send/1000000/", data=_expect_edu_transaction( "m.typing", content={ @@ -202,7 +202,7 @@ class TypingNotificationsTestCase(unittest.HomeserverTestCase): (request, channel) = self.make_request( "PUT", - "/_matrix/federation/v1/send/1000000", + "/_matrix/federation/v1/send/1000000/", _make_edu_transaction_json( "m.typing", content={ @@ -258,7 +258,7 @@ class TypingNotificationsTestCase(unittest.HomeserverTestCase): put_json = self.hs.get_http_client().put_json put_json.assert_called_once_with( "farm", - path="/_matrix/federation/v1/send/1000000", + path="/_matrix/federation/v1/send/1000000/", data=_expect_edu_transaction( "m.typing", content={ diff --git a/tests/handlers/test_user_directory.py b/tests/handlers/test_user_directory.py index a16a2dc67b..f1d0aa42b6 100644 --- a/tests/handlers/test_user_directory.py +++ b/tests/handlers/test_user_directory.py @@ -16,6 +16,7 @@ from mock import Mock from synapse.api.constants import UserTypes from synapse.rest.client.v1 import admin, login, room +from synapse.rest.client.v2_alpha import user_directory from synapse.storage.roommember import ProfileInfo from tests import unittest @@ -114,13 +115,13 @@ class UserDirectoryTestCase(unittest.HomeserverTestCase): self.helper.join(room, user=u2, tok=u2_token) # Check we have populated the database correctly. - shares_public = self.get_users_who_share_public_rooms() shares_private = self.get_users_who_share_private_rooms() + public_users = self.get_users_in_public_rooms() - self.assertEqual(shares_public, []) self.assertEqual( self._compress_shared(shares_private), set([(u1, u2, room), (u2, u1, room)]) ) + self.assertEqual(public_users, []) # We get one search result when searching for user2 by user1. s = self.get_success(self.handler.search_users(u1, "user2", 10)) @@ -138,11 +139,11 @@ class UserDirectoryTestCase(unittest.HomeserverTestCase): self.helper.leave(room, user=u2, tok=u2_token) # Check we have removed the values. - shares_public = self.get_users_who_share_public_rooms() shares_private = self.get_users_who_share_private_rooms() + public_users = self.get_users_in_public_rooms() - self.assertEqual(shares_public, []) self.assertEqual(self._compress_shared(shares_private), set()) + self.assertEqual(public_users, []) # User1 now gets no search results for any of the other users. s = self.get_success(self.handler.search_users(u1, "user2", 10)) @@ -160,14 +161,16 @@ class UserDirectoryTestCase(unittest.HomeserverTestCase): r.add((i["user_id"], i["other_user_id"], i["room_id"])) return r - def get_users_who_share_public_rooms(self): - return self.get_success( + def get_users_in_public_rooms(self): + r = self.get_success( self.store._simple_select_list( - "users_who_share_public_rooms", - None, - ["user_id", "other_user_id", "room_id"], + "users_in_public_rooms", None, ("user_id", "room_id") ) ) + retval = [] + for i in r: + retval.append((i["user_id"], i["room_id"])) + return retval def get_users_who_share_private_rooms(self): return self.get_success( @@ -178,6 +181,53 @@ class UserDirectoryTestCase(unittest.HomeserverTestCase): ) ) + def _add_background_updates(self): + """ + Add the background updates we need to run. + """ + # Ugh, have to reset this flag + self.store._all_done = False + + self.get_success( + self.store._simple_insert( + "background_updates", + { + "update_name": "populate_user_directory_createtables", + "progress_json": "{}", + }, + ) + ) + self.get_success( + self.store._simple_insert( + "background_updates", + { + "update_name": "populate_user_directory_process_rooms", + "progress_json": "{}", + "depends_on": "populate_user_directory_createtables", + }, + ) + ) + self.get_success( + self.store._simple_insert( + "background_updates", + { + "update_name": "populate_user_directory_process_users", + "progress_json": "{}", + "depends_on": "populate_user_directory_process_rooms", + }, + ) + ) + self.get_success( + self.store._simple_insert( + "background_updates", + { + "update_name": "populate_user_directory_cleanup", + "progress_json": "{}", + "depends_on": "populate_user_directory_process_users", + }, + ) + ) + def test_initial(self): """ The user directory's initial handler correctly updates the search tables. @@ -200,32 +250,24 @@ class UserDirectoryTestCase(unittest.HomeserverTestCase): self.get_success(self.store.update_user_directory_stream_pos(None)) self.get_success(self.store.delete_all_from_user_dir()) - shares_public = self.get_users_who_share_public_rooms() shares_private = self.get_users_who_share_private_rooms() + public_users = self.get_users_in_public_rooms() + # Nothing updated yet self.assertEqual(shares_private, []) - self.assertEqual(shares_public, []) - - # Reset the handled users caches - self.handler.initially_handled_users = set() - - # Do the initial population - d = self.handler._do_initial_spam() + self.assertEqual(public_users, []) - # This takes a while, so pump it a bunch of times to get through the - # sleep delays - for i in range(10): - self.pump(1) + # Do the initial population of the user directory via the background update + self._add_background_updates() - self.get_success(d) + while not self.get_success(self.store.has_completed_background_updates()): + self.get_success(self.store.do_next_background_update(100), by=0.1) - shares_public = self.get_users_who_share_public_rooms() shares_private = self.get_users_who_share_private_rooms() + public_users = self.get_users_in_public_rooms() - # User 1 and User 2 share public rooms - self.assertEqual( - self._compress_shared(shares_public), set([(u1, u2, room), (u2, u1, room)]) - ) + # User 1 and User 2 are in the same public room + self.assertEqual(set(public_users), set([(u1, room), (u2, room)])) # User 1 and User 3 share private rooms self.assertEqual( @@ -233,7 +275,7 @@ class UserDirectoryTestCase(unittest.HomeserverTestCase): set([(u1, u3, private_room), (u3, u1, private_room)]), ) - def test_search_all_users(self): + def test_initial_share_all_users(self): """ Search all users = True means that a user does not have to share a private room with the searching user or be in a public room to be search @@ -243,33 +285,87 @@ class UserDirectoryTestCase(unittest.HomeserverTestCase): self.hs.config.user_directory_search_all_users = True u1 = self.register_user("user1", "pass") - u1_token = self.login(u1, "pass") - u2 = self.register_user("user2", "pass") - u2_token = self.login(u2, "pass") + self.register_user("user2", "pass") u3 = self.register_user("user3", "pass") - # User 1 and User 2 join a room. User 3 never does. - room = self.helper.create_room_as(u1, is_public=True, tok=u1_token) - self.helper.invite(room, src=u1, targ=u2, tok=u1_token) - self.helper.join(room, user=u2, tok=u2_token) - + # Wipe the user dir self.get_success(self.store.update_user_directory_stream_pos(None)) self.get_success(self.store.delete_all_from_user_dir()) - # Reset the handled users caches - self.handler.initially_handled_users = set() + # Do the initial population of the user directory via the background update + self._add_background_updates() - # Do the initial population - d = self.handler._do_initial_spam() + while not self.get_success(self.store.has_completed_background_updates()): + self.get_success(self.store.do_next_background_update(100), by=0.1) - # This takes a while, so pump it a bunch of times to get through the - # sleep delays - for i in range(10): - self.pump(1) + shares_private = self.get_users_who_share_private_rooms() + public_users = self.get_users_in_public_rooms() - self.get_success(d) + # No users share rooms + self.assertEqual(public_users, []) + self.assertEqual(self._compress_shared(shares_private), set([])) # Despite not sharing a room, search_all_users means we get a search # result. s = self.get_success(self.handler.search_users(u1, u3, 10)) self.assertEqual(len(s["results"]), 1) + + # We can find the other two users + s = self.get_success(self.handler.search_users(u1, "user", 10)) + self.assertEqual(len(s["results"]), 2) + + # Registering a user and then searching for them works. + u4 = self.register_user("user4", "pass") + s = self.get_success(self.handler.search_users(u1, u4, 10)) + self.assertEqual(len(s["results"]), 1) + + +class TestUserDirSearchDisabled(unittest.HomeserverTestCase): + user_id = "@test:test" + + servlets = [ + user_directory.register_servlets, + room.register_servlets, + login.register_servlets, + admin.register_servlets, + ] + + def make_homeserver(self, reactor, clock): + config = self.default_config() + config.update_user_directory = True + hs = self.setup_test_homeserver(config=config) + + self.config = hs.config + + return hs + + def test_disabling_room_list(self): + self.config.user_directory_search_enabled = True + + # First we create a room with another user so that user dir is non-empty + # for our user + self.helper.create_room_as(self.user_id) + u2 = self.register_user("user2", "pass") + room = self.helper.create_room_as(self.user_id) + self.helper.join(room, user=u2) + + # Assert user directory is not empty + request, channel = self.make_request( + "POST", + b"user_directory/search", + b'{"search_term":"user2"}', + ) + self.render(request) + self.assertEquals(200, channel.code, channel.result) + self.assertTrue(len(channel.json_body["results"]) > 0) + + # Disable user directory and check search returns nothing + self.config.user_directory_search_enabled = False + request, channel = self.make_request( + "POST", + b"user_directory/search", + b'{"search_term":"user2"}', + ) + self.render(request) + self.assertEquals(200, channel.code, channel.result) + self.assertTrue(len(channel.json_body["results"]) == 0) diff --git a/tests/push/test_email.py b/tests/push/test_email.py index 50ee6910d1..be3fed8de3 100644 --- a/tests/push/test_email.py +++ b/tests/push/test_email.py @@ -63,8 +63,10 @@ class EmailPusherTests(HomeserverTestCase): config.email_smtp_port = 20 config.require_transport_security = False config.email_smtp_user = None + config.email_smtp_pass = None config.email_app_name = "Matrix" config.email_notif_from = "test@example.com" + config.email_riot_base_url = None hs = self.setup_test_homeserver(config=config, sendmail=sendmail) diff --git a/tests/rest/client/v1/test_admin.py b/tests/rest/client/v1/test_admin.py index ea03b7e523..ef38473bd6 100644 --- a/tests/rest/client/v1/test_admin.py +++ b/tests/rest/client/v1/test_admin.py @@ -20,7 +20,7 @@ import json from mock import Mock from synapse.api.constants import UserTypes -from synapse.rest.client.v1 import admin, login +from synapse.rest.client.v1 import admin, events, login, room from tests import unittest @@ -353,3 +353,140 @@ class UserRegisterTestCase(unittest.HomeserverTestCase): self.assertEqual(400, int(channel.result["code"]), msg=channel.result["body"]) self.assertEqual('Invalid user type', channel.json_body["error"]) + + +class ShutdownRoomTestCase(unittest.HomeserverTestCase): + servlets = [ + admin.register_servlets, + login.register_servlets, + events.register_servlets, + room.register_servlets, + room.register_deprecated_servlets, + ] + + def prepare(self, reactor, clock, hs): + self.event_creation_handler = hs.get_event_creation_handler() + hs.config.user_consent_version = "1" + + consent_uri_builder = Mock() + consent_uri_builder.build_user_consent_uri.return_value = ( + "http://example.com" + ) + self.event_creation_handler._consent_uri_builder = consent_uri_builder + + self.store = hs.get_datastore() + + self.admin_user = self.register_user("admin", "pass", admin=True) + self.admin_user_tok = self.login("admin", "pass") + + self.other_user = self.register_user("user", "pass") + self.other_user_token = self.login("user", "pass") + + # Mark the admin user as having consented + self.get_success( + self.store.user_set_consent_version(self.admin_user, "1"), + ) + + def test_shutdown_room_consent(self): + """Test that we can shutdown rooms with local users who have not + yet accepted the privacy policy. This used to fail when we tried to + force part the user from the old room. + """ + self.event_creation_handler._block_events_without_consent_error = None + + room_id = self.helper.create_room_as(self.other_user, tok=self.other_user_token) + + # Assert one user in room + users_in_room = self.get_success( + self.store.get_users_in_room(room_id), + ) + self.assertEqual([self.other_user], users_in_room) + + # Enable require consent to send events + self.event_creation_handler._block_events_without_consent_error = "Error" + + # Assert that the user is getting consent error + self.helper.send( + room_id, + body="foo", tok=self.other_user_token, expect_code=403, + ) + + # Test that the admin can still send shutdown + url = "admin/shutdown_room/" + room_id + request, channel = self.make_request( + "POST", + url.encode('ascii'), + json.dumps({"new_room_user_id": self.admin_user}), + access_token=self.admin_user_tok, + ) + self.render(request) + + self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"]) + + # Assert there is now no longer anyone in the room + users_in_room = self.get_success( + self.store.get_users_in_room(room_id), + ) + self.assertEqual([], users_in_room) + + @unittest.DEBUG + def test_shutdown_room_block_peek(self): + """Test that a world_readable room can no longer be peeked into after + it has been shut down. + """ + + self.event_creation_handler._block_events_without_consent_error = None + + room_id = self.helper.create_room_as(self.other_user, tok=self.other_user_token) + + # Enable world readable + url = "rooms/%s/state/m.room.history_visibility" % (room_id,) + request, channel = self.make_request( + "PUT", + url.encode('ascii'), + json.dumps({"history_visibility": "world_readable"}), + access_token=self.other_user_token, + ) + self.render(request) + self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"]) + + # Test that the admin can still send shutdown + url = "admin/shutdown_room/" + room_id + request, channel = self.make_request( + "POST", + url.encode('ascii'), + json.dumps({"new_room_user_id": self.admin_user}), + access_token=self.admin_user_tok, + ) + self.render(request) + + self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"]) + + # Assert we can no longer peek into the room + self._assert_peek(room_id, expect_code=403) + + def _assert_peek(self, room_id, expect_code): + """Assert that the admin user can (or cannot) peek into the room. + """ + + url = "rooms/%s/initialSync" % (room_id,) + request, channel = self.make_request( + "GET", + url.encode('ascii'), + access_token=self.admin_user_tok, + ) + self.render(request) + self.assertEqual( + expect_code, int(channel.result["code"]), msg=channel.result["body"], + ) + + url = "events?timeout=0&room_id=" + room_id + request, channel = self.make_request( + "GET", + url.encode('ascii'), + access_token=self.admin_user_tok, + ) + self.render(request) + self.assertEqual( + expect_code, int(channel.result["code"]), msg=channel.result["body"], + ) diff --git a/tests/rest/client/v1/test_login.py b/tests/rest/client/v1/test_login.py new file mode 100644 index 0000000000..86312f1096 --- /dev/null +++ b/tests/rest/client/v1/test_login.py @@ -0,0 +1,163 @@ +import json + +from synapse.rest.client.v1 import admin, login + +from tests import unittest + +LOGIN_URL = b"/_matrix/client/r0/login" + + +class LoginRestServletTestCase(unittest.HomeserverTestCase): + + servlets = [ + admin.register_servlets, + login.register_servlets, + ] + + def make_homeserver(self, reactor, clock): + + self.hs = self.setup_test_homeserver() + self.hs.config.enable_registration = True + self.hs.config.registrations_require_3pid = [] + self.hs.config.auto_join_rooms = [] + self.hs.config.enable_registration_captcha = False + + return self.hs + + def test_POST_ratelimiting_per_address(self): + self.hs.config.rc_login_address.burst_count = 5 + self.hs.config.rc_login_address.per_second = 0.17 + + # Create different users so we're sure not to be bothered by the per-user + # ratelimiter. + for i in range(0, 6): + self.register_user("kermit" + str(i), "monkey") + + for i in range(0, 6): + params = { + "type": "m.login.password", + "identifier": { + "type": "m.id.user", + "user": "kermit" + str(i), + }, + "password": "monkey", + } + request_data = json.dumps(params) + request, channel = self.make_request(b"POST", LOGIN_URL, request_data) + self.render(request) + + if i == 5: + self.assertEquals(channel.result["code"], b"429", channel.result) + retry_after_ms = int(channel.json_body["retry_after_ms"]) + else: + self.assertEquals(channel.result["code"], b"200", channel.result) + + # Since we're ratelimiting at 1 request/min, retry_after_ms should be lower + # than 1min. + self.assertTrue(retry_after_ms < 6000) + + self.reactor.advance(retry_after_ms / 1000.) + + params = { + "type": "m.login.password", + "identifier": { + "type": "m.id.user", + "user": "kermit" + str(i), + }, + "password": "monkey", + } + request_data = json.dumps(params) + request, channel = self.make_request(b"POST", LOGIN_URL, params) + self.render(request) + + self.assertEquals(channel.result["code"], b"200", channel.result) + + def test_POST_ratelimiting_per_account(self): + self.hs.config.rc_login_account.burst_count = 5 + self.hs.config.rc_login_account.per_second = 0.17 + + self.register_user("kermit", "monkey") + + for i in range(0, 6): + params = { + "type": "m.login.password", + "identifier": { + "type": "m.id.user", + "user": "kermit", + }, + "password": "monkey", + } + request_data = json.dumps(params) + request, channel = self.make_request(b"POST", LOGIN_URL, request_data) + self.render(request) + + if i == 5: + self.assertEquals(channel.result["code"], b"429", channel.result) + retry_after_ms = int(channel.json_body["retry_after_ms"]) + else: + self.assertEquals(channel.result["code"], b"200", channel.result) + + # Since we're ratelimiting at 1 request/min, retry_after_ms should be lower + # than 1min. + self.assertTrue(retry_after_ms < 6000) + + self.reactor.advance(retry_after_ms / 1000.) + + params = { + "type": "m.login.password", + "identifier": { + "type": "m.id.user", + "user": "kermit", + }, + "password": "monkey", + } + request_data = json.dumps(params) + request, channel = self.make_request(b"POST", LOGIN_URL, params) + self.render(request) + + self.assertEquals(channel.result["code"], b"200", channel.result) + + def test_POST_ratelimiting_per_account_failed_attempts(self): + self.hs.config.rc_login_failed_attempts.burst_count = 5 + self.hs.config.rc_login_failed_attempts.per_second = 0.17 + + self.register_user("kermit", "monkey") + + for i in range(0, 6): + params = { + "type": "m.login.password", + "identifier": { + "type": "m.id.user", + "user": "kermit", + }, + "password": "notamonkey", + } + request_data = json.dumps(params) + request, channel = self.make_request(b"POST", LOGIN_URL, request_data) + self.render(request) + + if i == 5: + self.assertEquals(channel.result["code"], b"429", channel.result) + retry_after_ms = int(channel.json_body["retry_after_ms"]) + else: + self.assertEquals(channel.result["code"], b"403", channel.result) + + # Since we're ratelimiting at 1 request/min, retry_after_ms should be lower + # than 1min. + self.assertTrue(retry_after_ms < 6000) + + self.reactor.advance(retry_after_ms / 1000.) + + params = { + "type": "m.login.password", + "identifier": { + "type": "m.id.user", + "user": "kermit", + }, + "password": "notamonkey", + } + request_data = json.dumps(params) + request, channel = self.make_request(b"POST", LOGIN_URL, params) + self.render(request) + + self.assertEquals(channel.result["code"], b"403", channel.result) diff --git a/tests/rest/client/v2_alpha/test_register.py b/tests/rest/client/v2_alpha/test_register.py index 3600434858..a45e6e5e1f 100644 --- a/tests/rest/client/v2_alpha/test_register.py +++ b/tests/rest/client/v2_alpha/test_register.py @@ -20,6 +20,7 @@ class RegisterRestServletTestCase(unittest.HomeserverTestCase): self.hs.config.registrations_require_3pid = [] self.hs.config.auto_join_rooms = [] self.hs.config.enable_registration_captcha = False + self.hs.config.allow_guest_access = True return self.hs @@ -28,7 +29,7 @@ class RegisterRestServletTestCase(unittest.HomeserverTestCase): as_token = "i_am_an_app_service" appservice = ApplicationService( - as_token, self.hs.config.hostname, + as_token, self.hs.config.server_name, id="1234", namespaces={ "users": [{"regex": r"@as_user.*", "exclusive": True}], @@ -132,7 +133,8 @@ class RegisterRestServletTestCase(unittest.HomeserverTestCase): self.assertEquals(channel.json_body["error"], "Guest access is disabled") def test_POST_ratelimiting_guest(self): - self.hs.config.rc_registration_request_burst_count = 5 + self.hs.config.rc_registration.burst_count = 5 + self.hs.config.rc_registration.per_second = 0.17 for i in range(0, 6): url = self.url + b"?kind=guest" @@ -153,7 +155,8 @@ class RegisterRestServletTestCase(unittest.HomeserverTestCase): self.assertEquals(channel.result["code"], b"200", channel.result) def test_POST_ratelimiting(self): - self.hs.config.rc_registration_request_burst_count = 5 + self.hs.config.rc_registration.burst_count = 5 + self.hs.config.rc_registration.per_second = 0.17 for i in range(0, 6): params = { diff --git a/tests/server.py b/tests/server.py index 37069afdda..ea26dea623 100644 --- a/tests/server.py +++ b/tests/server.py @@ -119,14 +119,7 @@ class FakeSite: server_version_string = b"1" site_tag = "test" - - @property - def access_logger(self): - class FakeLogger: - def info(self, *args, **kwargs): - pass - - return FakeLogger() + access_logger = logging.getLogger("synapse.access.http.fake") def make_request( diff --git a/tests/server_notices/test_resource_limits_server_notices.py b/tests/server_notices/test_resource_limits_server_notices.py index b1551df7ca..3bd9f1e9c1 100644 --- a/tests/server_notices/test_resource_limits_server_notices.py +++ b/tests/server_notices/test_resource_limits_server_notices.py @@ -9,13 +9,16 @@ from synapse.server_notices.resource_limits_server_notices import ( ) from tests import unittest -from tests.utils import setup_test_homeserver +from tests.utils import default_config, setup_test_homeserver class TestResourceLimitsServerNotices(unittest.TestCase): @defer.inlineCallbacks def setUp(self): - self.hs = yield setup_test_homeserver(self.addCleanup) + hs_config = default_config(name="test") + hs_config.server_notices_mxid = "@server:test" + + self.hs = yield setup_test_homeserver(self.addCleanup, config=hs_config) self.server_notices_sender = self.hs.get_server_notices_sender() # relying on [1] is far from ideal, but the only case where diff --git a/tests/storage/test_user_directory.py b/tests/storage/test_user_directory.py index a2a652a235..fd3361404f 100644 --- a/tests/storage/test_user_directory.py +++ b/tests/storage/test_user_directory.py @@ -16,7 +16,6 @@ from twisted.internet import defer from synapse.storage import UserDirectoryStore -from synapse.storage.roommember import ProfileInfo from tests import unittest from tests.utils import setup_test_homeserver @@ -34,15 +33,11 @@ class UserDirectoryStoreTestCase(unittest.TestCase): # alice and bob are both in !room_id. bobby is not but shares # a homeserver with alice. - yield self.store.add_profiles_to_user_dir( - { - ALICE: ProfileInfo(None, "alice"), - BOB: ProfileInfo(None, "bob"), - BOBBY: ProfileInfo(None, "bobby"), - }, - ) - yield self.store.add_users_who_share_room( - "!room:id", False, ((ALICE, BOB), (BOB, ALICE)) + yield self.store.update_profile_in_user_dir(ALICE, "alice", None) + yield self.store.update_profile_in_user_dir(BOB, "bob", None) + yield self.store.update_profile_in_user_dir(BOBBY, "bobby", None) + yield self.store.add_users_in_public_rooms( + "!room:id", (ALICE, BOB) ) @defer.inlineCallbacks diff --git a/tests/unittest.py b/tests/unittest.py index ef31321bc8..7772a47078 100644 --- a/tests/unittest.py +++ b/tests/unittest.py @@ -330,10 +330,10 @@ class HomeserverTestCase(TestCase): """ self.reactor.pump([by] * 100) - def get_success(self, d): + def get_success(self, d, by=0.0): if not isinstance(d, Deferred): return d - self.pump() + self.pump(by=by) return self.successResultOf(d) def register_user(self, username, password, admin=False): diff --git a/tests/utils.py b/tests/utils.py index 9c8dc9dbce..d4ab4209ed 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -28,7 +28,7 @@ from twisted.internet import defer, reactor from synapse.api.constants import EventTypes, RoomVersions from synapse.api.errors import CodeMessageException, cs_error -from synapse.config.server import ServerConfig +from synapse.config.homeserver import HomeServerConfig from synapse.federation.transport import server as federation_server from synapse.http.server import HttpServer from synapse.server import HomeServer @@ -111,14 +111,25 @@ def default_config(name): """ Create a reasonable test config. """ - config = Mock() - config.signing_key = [MockKey()] + config_dict = { + "server_name": name, + "media_store_path": "media", + "uploads_path": "uploads", + + # the test signing key is just an arbitrary ed25519 key to keep the config + # parser happy + "signing_key": "ed25519 a_lPym qvioDNmfExFBRPgdTU+wtFYKq4JfwFRv7sYVgWvmgJg", + } + + config = HomeServerConfig() + config.parse_config_dict(config_dict) + + # TODO: move this stuff into config_dict or get rid of it config.event_cache_size = 1 config.enable_registration = True config.enable_registration_captcha = False config.macaroon_secret_key = "not even a little secret" config.expire_access_token = False - config.server_name = name config.trusted_third_party_id_servers = [] config.room_invite_state_types = [] config.password_providers = [] @@ -151,8 +162,14 @@ def default_config(name): config.admin_contact = None config.rc_messages_per_second = 10000 config.rc_message_burst_count = 10000 - config.rc_registration_request_burst_count = 3.0 - config.rc_registration_requests_per_second = 0.17 + config.rc_registration.per_second = 10000 + config.rc_registration.burst_count = 10000 + config.rc_login_address.per_second = 10000 + config.rc_login_address.burst_count = 10000 + config.rc_login_account.per_second = 10000 + config.rc_login_account.burst_count = 10000 + config.rc_login_failed_attempts.per_second = 10000 + config.rc_login_failed_attempts.burst_count = 10000 config.saml2_enabled = False config.public_baseurl = None config.default_identity_server = None @@ -170,13 +187,6 @@ def default_config(name): # background, which upsets the test runner. config.update_user_directory = False - def is_threepid_reserved(threepid): - return ServerConfig.is_threepid_reserved( - config.mau_limits_reserved_threepids, threepid - ) - - config.is_threepid_reserved.side_effect = is_threepid_reserved - return config @@ -270,7 +280,6 @@ def setup_test_homeserver( db_config=config.database_config, version_string="Synapse/tests", database_engine=db_engine, - room_list_handler=object(), tls_server_context_factory=Mock(), tls_client_options_factory=Mock(), reactor=reactor, @@ -331,6 +340,8 @@ def setup_test_homeserver( cleanup_func(cleanup) hs.setup() + if homeserverToUse.__name__ == "TestHomeServer": + hs.setup_master() else: hs = homeserverToUse( name, @@ -339,7 +350,6 @@ def setup_test_homeserver( config=config, version_string="Synapse/tests", database_engine=db_engine, - room_list_handler=object(), tls_server_context_factory=Mock(), tls_client_options_factory=Mock(), reactor=reactor, diff --git a/tox.ini b/tox.ini index 19080a648f..ef543890f9 100644 --- a/tox.ini +++ b/tox.ini @@ -82,15 +82,23 @@ deps = mock lxml + coverage + commands = /usr/bin/find "{toxinidir}" -name '*.pyc' -delete # Make all greater-thans equals so we test the oldest version of our direct # dependencies, but make the pyopenssl 17.0, which can work against an # OpenSSL 1.1 compiled cryptography (as older ones don't compile on Travis). /bin/sh -c 'python -m synapse.python_dependencies | sed -e "s/>=/==/g" -e "s/psycopg2==2.6//" -e "s/pyopenssl==16.0.0/pyopenssl==17.0.0/" | xargs pip install' + + # Add this so that coverage will run on subprocesses + /bin/sh -c 'echo "import coverage; coverage.process_startup()" > {envsitepackagesdir}/../sitecustomize.py' + # Install Synapse itself. This won't update any libraries. pip install -e . - {envbindir}/trial {env:TRIAL_FLAGS:} {posargs:tests} {env:TOXSUFFIX:} + + {envbindir}/coverage run "{envbindir}/trial" {env:TRIAL_FLAGS:} {posargs:tests} {env:TOXSUFFIX:} + [testenv:packaging] skip_install=True |