Add changelog

Remove debug logging and make diff nicer
lint
2019-06-06 16:38:28 +01:00 · 2019-06-06 16:38:18 +01:00 · 2019-06-06 16:14:15 +01:00 · 2019-06-06 16:12:57 +01:00 · 2019-06-06 16:07:21 +01:00 · 2019-06-06 16:03:10 +01:00
172 changed files with 5129 additions and 1457 deletions
@@ -1,3 +1,23 @@
+Synapse 0.99.5.2 (2019-05-30)
+=============================
+
+Bugfixes
+--------
+
+- Fix bug where we leaked extremities when we soft failed events, leading to performance degradation. ([\#5274](https://github.com/matrix-org/synapse/issues/5274), [\#5278](https://github.com/matrix-org/synapse/issues/5278), [\#5291](https://github.com/matrix-org/synapse/issues/5291))
+
+
+Synapse 0.99.5.1 (2019-05-22)
+=============================
+
+0.99.5.1 supersedes 0.99.5 due to malformed debian changelog - no functional changes.
+
+Synapse 0.99.5 (2019-05-22)
+===========================
+
+No significant changes.
+
+
 Synapse 0.99.5rc1 (2019-05-21)
 ==============================

@@ -0,0 +1 @@
+Synapse now more efficiently collates room statistics.
@@ -0,0 +1 @@
+Fix worker registration bug caused by ClientReaderSlavedStore being unable to see get_profileinfo.
@@ -0,0 +1 @@
+Synapse will now serve the experimental "room complexity" API endpoint.
@@ -0,0 +1 @@
+Add experimental support for relations (aka reactions and edits).
@@ -0,0 +1 @@
+Ability to configure default room version.
@@ -0,0 +1 @@
+The base classes for the v1 and v2_alpha REST APIs have been unified.
@@ -0,0 +1 @@
+Simplifications and comments in do_auth.
@@ -0,0 +1 @@
+Remove urllib3 pin as requests 2.22.0 has been released supporting urllib3 1.25.2.
@@ -0,0 +1 @@
+Run black on synapse.crypto.keyring.
@@ -0,0 +1 @@
+Fix appservice timestamp massaging.
@@ -0,0 +1 @@
+Rewrite store_server_verify_key to store several keys at once.
@@ -0,0 +1 @@
+Remove unused VerifyKey.expired and .time_added fields.
@@ -0,0 +1 @@
+Simplify Keyring.process_v2_response.
@@ -0,0 +1 @@
+Store key validity time in the storage layer.
@@ -0,0 +1 @@
+Refactor synapse.crypto.keyring to use a KeyFetcher interface.
@@ -0,0 +1 @@
+Ability to configure default room version.
@@ -0,0 +1 @@
+Simplification to Keyring.wait_for_previous_lookups.
@@ -0,0 +1 @@
+Ensure that server_keys fetched via a notary server are correctly signed.
@@ -0,0 +1 @@
+Show the correct error when logging out and access token is missing.
@@ -0,0 +1 @@
+Fix error code when there is an invalid parameter on /_matrix/client/r0/publicRooms
@@ -0,0 +1 @@
+Fix error when downloading thumbnail with missing width/height parameter.
@@ -0,0 +1 @@
+Synapse now more efficiently collates room statistics.
@@ -0,0 +1 @@
+Fix schema update for account validity.
@@ -0,0 +1 @@
+Fix bug where we leaked extremities when we soft failed events, leading to performance degradation.
@@ -0,0 +1 @@
+Fix "db txn 'update_presence' from sentinel context" log messages.
@@ -0,0 +1 @@
+Allow configuring a range for the account validity startup job.
@@ -0,0 +1 @@
+Fix dropped logcontexts during high outbound traffic.
@@ -0,0 +1 @@
+Fix bug where we leaked extremities when we soft failed events, leading to performance degradation.
@@ -0,0 +1 @@
+Fix docs on resetting the user directory.
@@ -0,0 +1 @@
+Specify the type of reCAPTCHA key to use.
@@ -0,0 +1 @@
+CAS login will now hit the r0 API, not the deprecated v1 one.
@@ -0,0 +1 @@
+Remove spurious debug from MatrixFederationHttpClient.get_json.
@@ -0,0 +1 @@
+Improve logging for logcontext leaks.
@@ -0,0 +1 @@
+Fix bug where we leaked extremities when we soft failed events, leading to performance degradation.
@@ -0,0 +1 @@
+Fix a bug where it is not possible to get events in the federation format with the request `GET /_matrix/client/r0/rooms/{roomId}/messages`.
@@ -0,0 +1 @@
+Fix performance problems with the rooms stats background update.
@@ -0,0 +1 @@
+Refactor keyring.VerifyKeyRequest to use attr.s.
@@ -0,0 +1 @@
+Rewrite get_server_verify_keys, again.
@@ -0,0 +1 @@
+Fix noisy 'no key for server' logs.
@@ -0,0 +1 @@
+Clarify that the admin change password API logs the user out.
@@ -0,0 +1 @@
+Fix bug where a notary server would sometimes forget old keys.
@@ -0,0 +1 @@
+Prevent users from setting huge displaynames and avatar URLs.
@@ -0,0 +1 @@
+Ensure that we have an up-to-date copy of the signing key when validating incoming federation requests.
@@ -0,0 +1 @@
+Synapse now more efficiently collates room statistics.
@@ -0,0 +1 @@
+The base classes for the v1 and v2_alpha REST APIs have been unified.
@@ -0,0 +1 @@
+Improve docstrings on MatrixFederationClient.
@@ -0,0 +1 @@
+Fix various problems which made the signing-key notary server time out for some requests.
@@ -0,0 +1 @@
+Fix bug which would make certain operations (such as room joins) block for 20 minutes while attemoting to fetch verification keys.
@@ -0,0 +1 @@
+Fix a bug where we could rapidly mark a server as unreachable even though it was only down for a few minutes.
@@ -0,0 +1 @@
+Fix a bug where account validity renewal emails could only be sent when email notifs were enabled.
@@ -0,0 +1 @@
+Add ability to perform password reset via email without trusting the identity server.
@@ -1,3 +1,15 @@
+matrix-synapse-py3 (0.99.5.2) stable; urgency=medium
+
+  * New synapse release 0.99.5.2.
+
+ -- Synapse Packaging team <packages@matrix.org>  Thu, 30 May 2019 16:28:07 +0100
+
+matrix-synapse-py3 (0.99.5.1) stable; urgency=medium
+
+  * New synapse release 0.99.5.1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Wed, 22 May 2019 16:22:24 +0000
+
 matrix-synapse-py3 (0.99.4) stable; urgency=medium

  [ Christoph Müller ]
@@ -161,7 +161,7 @@ specify values for `SYNAPSE_CONFIG_PATH`, `SYNAPSE_SERVER_NAME` and
 example:

 ```
-docker run -it --rm
+docker run -it --rm \
    --mount type=volume,src=synapse-data,dst=/data \
    -e SYNAPSE_CONFIG_PATH=/data/homeserver.yaml \
    -e SYNAPSE_SERVER_NAME=my.matrix.host \
@@ -7,6 +7,7 @@ Requires a public/private key pair from:

 https://developers.google.com/recaptcha/

+Must be a reCAPTCHA v2 key using the "I'm not a robot" Checkbox option

 Setting ReCaptcha Keys
 ----------------------
@@ -69,7 +69,7 @@ An empty body may be passed for backwards compatibility.
 Reset password
 ==============

-Changes the password of another user.
+Changes the password of another user. This will automatically log the user out of all their devices.

 The api is::

@@ -83,6 +83,16 @@ pid_file: DATADIR/homeserver.pid
 #
 #restrict_public_rooms_to_local_users: true

+# The default room version for newly created rooms.
+#
+# Known room versions are listed here:
+# https://matrix.org/docs/spec/#complete-list-of-room-versions
+#
+# For example, for room version 1, default_room_version should be set
+# to "1".
+#
+#default_room_version: "1"
+
 # The GC threshold parameters to pass to `gc.set_threshold`, if defined
 #
 #gc_thresholds: [700, 10, 10]
@@ -753,7 +763,9 @@ uploads_path: "DATADIR/uploads"
 # This means that, if a validity period is set, and Synapse is restarted (it will
 # then derive an expiration date from the current validity period), and some time
 # after that the validity period changes and Synapse is restarted, the users'
-# expiration dates won't be updated unless their account is manually renewed.
+# expiration dates won't be updated unless their account is manually renewed. This
+# date will be randomly selected within a range [now + period - d ; now + period],
+# where d is equal to 10% of the validity period.
 #
 #account_validity:
 #  enabled: True
@@ -1006,10 +1018,8 @@ password_config:



-# Enable sending emails for notification events or expiry notices
-# Defining a custom URL for Riot is only needed if email notifications
-# should contain links to a self-hosted installation of Riot; when set
-# the "app_name" setting is ignored.
+# Enable sending emails for password resets, notification events or
+# account expiry notices
 #
 # If your SMTP server requires authentication, the optional smtp_user &
 # smtp_pass variables should be used
@@ -1017,22 +1027,64 @@ password_config:
 #email:
 #   enable_notifs: false
 #   smtp_host: "localhost"
-#   smtp_port: 25
+#   smtp_port: 25 # SSL: 465, STARTTLS: 587
 #   smtp_user: "exampleusername"
 #   smtp_pass: "examplepassword"
 #   require_transport_security: False
 #   notif_from: "Your Friendly %(app)s Home Server <noreply@example.com>"
 #   app_name: Matrix
-#   # if template_dir is unset, uses the example templates that are part of
-#   # the Synapse distribution.
+#
+#   # Enable email notifications by default
+#   notif_for_new_users: True
+#
+#   # Defining a custom URL for Riot is only needed if email notifications
+#   # should contain links to a self-hosted installation of Riot; when set
+#   # the "app_name" setting is ignored
+#   riot_base_url: "http://localhost/riot"
+#
+#   # Enable sending password reset emails via the configured, trusted
+#   # identity servers
+#   #
+#   # IMPORTANT! This will give a malicious or overtaken identity server
+#   # the ability to reset passwords for your users! Make absolutely sure
+#   # that you want to do this! It is strongly recommended that password
+#   # reset emails be sent by the homeserver instead
+#   #
+#   # If this option is set to false and SMTP options have not been
+#   # configured, resetting user passwords via email will be disabled
+#   #trust_identity_server_for_password_resets: false
+#
+#   # Configure the time that a validation email or text message code
+#   # will expire after sending
+#   #
+#   # This is currently used for password resets
+#   #validation_token_lifetime: 1h
+#
+#   # Template directory. All template files should be stored within this
+#   # directory
+#   #
 #   #template_dir: res/templates
+#
+#   # Templates for email notifications
+#   #
 #   notif_template_html: notif_mail.html
 #   notif_template_text: notif_mail.txt
-#   # Templates for account expiry notices.
+#
+#   # Templates for account expiry notices
+#   #
 #   expiry_template_html: notice_expiry.html
 #   expiry_template_text: notice_expiry.txt
-#   notif_for_new_users: True
-#   riot_base_url: "http://localhost/riot"
+#
+#   # Templates for password reset emails sent by the homeserver
+#   #
+#   #password_reset_template_html: password_reset.html
+#   #password_reset_template_text: password_reset.txt
+#
+#   # Templates for password reset success and failure pages that a user
+#   # will see after attempting to reset their password
+#   #
+#   #password_reset_template_success_html: password_reset_success.html
+#   #password_reset_template_failure_html: password_reset_failure.html


 #password_providers:
@@ -1093,9 +1145,9 @@ password_config:
 #
 # '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
-# UPDATE user_directory_stream_pos SET stream_id = NULL;
-# on your database to tell it to rebuild the user_directory search indexes.
+# in public rooms.  Defaults to false.  If you set it True, you'll have to
+# rebuild the user_directory search indexes, see
+# https://github.com/matrix-org/synapse/blob/master/docs/user_directory.md
 #
 #user_directory:
 #  enabled: true
@@ -1153,6 +1205,22 @@ password_config:
 #


+
+# Local statistics collection. Used in populating the room directory.
+#
+# 'bucket_size' controls how large each statistics timeslice is. It can
+# be defined in a human readable short form -- e.g. "1d", "1y".
+#
+# 'retention' controls how long historical statistics will be kept for.
+# It can be defined in a human readable short form -- e.g. "1d", "1y".
+#
+#
+#stats:
+#   enabled: true
+#   bucket_size: 1d
+#   retention: 1y
+
+
 # Server Notices room configuration
 #
 # Uncomment this section to enable a room which can be used to send notices
@@ -7,11 +7,7 @@ who are present in a publicly viewable room present on the server.

 The directory info is stored in various tables, which can (typically after
 DB corruption) get stale or out of sync.  If this happens, for now the
-quickest solution to fix it is:
-
-```
-UPDATE user_directory_stream_pos SET stream_id = NULL;
-```
-
-and restart the synapse, which should then start a background task to
+solution to fix it is to execute the SQL here
+https://github.com/matrix-org/synapse/blob/master/synapse/storage/schema/delta/53/user_dir_populate.sql
+and then restart synapse. This should then start a background task to
 flush the current tables and regenerate the directory.
@@ -20,9 +20,7 @@ class CallVisitor(ast.NodeVisitor):
        else:
            return

-        if name == "client_path_patterns":
-            PATTERNS_V1.append(node.args[0].s)
-        elif name == "client_v2_patterns":
+        if name == "client_patterns":
            PATTERNS_V2.append(node.args[0].s)


@@ -27,4 +27,4 @@ try:
 except ImportError:
    pass

-__version__ = "0.99.5rc1"
+__version__ = "0.99.5.2"
@@ -79,6 +79,7 @@ class EventTypes(object):

    RoomHistoryVisibility = "m.room.history_visibility"
    CanonicalAlias = "m.room.canonical_alias"
+    Encryption = "m.room.encryption"
    RoomAvatar = "m.room.avatar"
    RoomEncryption = "m.room.encryption"
    GuestAccess = "m.room.guest_access"
@@ -339,6 +339,15 @@ class UnsupportedRoomVersionError(SynapseError):
        )


+class ThreepidValidationError(SynapseError):
+    """An error raised when there was a problem authorising an event."""
+
+    def __init__(self, *args, **kwargs):
+        if "errcode" not in kwargs:
+            kwargs["errcode"] = Codes.FORBIDDEN
+        super(ThreepidValidationError, self).__init__(*args, **kwargs)
+
+
 class IncompatibleRoomVersionError(SynapseError):
    """A server is trying to join a room whose version it does not support.

@@ -85,10 +85,6 @@ class RoomVersions(object):
    )


-# the version we will give rooms which are created on this server
-DEFAULT_ROOM_VERSION = RoomVersions.V1
-
-
 KNOWN_ROOM_VERSIONS = {
    v.identifier: v for v in (
        RoomVersions.V1,
@@ -26,6 +26,7 @@ CLIENT_API_PREFIX = "/_matrix/client"
 FEDERATION_PREFIX = "/_matrix/federation"
 FEDERATION_V1_PREFIX = FEDERATION_PREFIX + "/v1"
 FEDERATION_V2_PREFIX = FEDERATION_PREFIX + "/v2"
+FEDERATION_UNSTABLE_PREFIX = FEDERATION_PREFIX + "/unstable"
 STATIC_PREFIX = "/_matrix/static"
 WEB_CLIENT_PREFIX = "/_matrix/client"
 CONTENT_REPO_PREFIX = "/_matrix/content"
@@ -344,15 +344,21 @@ class _LimitedHostnameResolver(object):

    def resolveHostName(self, resolutionReceiver, hostName, portNumber=0,
                        addressTypes=None, transportSemantics='TCP'):
-        # Note this is happening deep within the reactor, so we don't need to
-        # worry about log contexts.
-
        # We need this function to return `resolutionReceiver` so we do all the
        # actual logic involving deferreds in a separate function.
-        self._resolve(
-            resolutionReceiver, hostName, portNumber,
-            addressTypes, transportSemantics,
-        )
+
+        # even though this is happening within the depths of twisted, we need to drop
+        # our logcontext before starting _resolve, otherwise: (a) _resolve will drop
+        # the logcontext if it returns an incomplete deferred; (b) _resolve will
+        # call the resolutionReceiver *with* a logcontext, which it won't be expecting.
+        with PreserveLoggingContext():
+            self._resolve(
+                resolutionReceiver,
+                hostName,
+                portNumber,
+                addressTypes,
+                transportSemantics,
+            )

        return resolutionReceiver

@@ -38,6 +38,7 @@ from synapse.replication.slave.storage.devices import SlavedDeviceStore
 from synapse.replication.slave.storage.directory import DirectoryStore
 from synapse.replication.slave.storage.events import SlavedEventStore
 from synapse.replication.slave.storage.keys import SlavedKeyStore
+from synapse.replication.slave.storage.profile import SlavedProfileStore
 from synapse.replication.slave.storage.push_rule import SlavedPushRuleStore
 from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
 from synapse.replication.slave.storage.registration import SlavedRegistrationStore
@@ -81,6 +82,7 @@ class ClientReaderSlavedStore(
    SlavedApplicationServiceStore,
    SlavedRegistrationStore,
    SlavedTransactionStore,
+    SlavedProfileStore,
    SlavedClientIpStore,
    BaseSlavedStore,
 ):
@@ -37,8 +37,7 @@ from synapse.replication.slave.storage.client_ips import SlavedClientIpStore
 from synapse.replication.slave.storage.devices import SlavedDeviceStore
 from synapse.replication.slave.storage.registration import SlavedRegistrationStore
 from synapse.replication.tcp.client import ReplicationClientHandler
-from synapse.rest.client.v1.base import ClientV1RestServlet, client_path_patterns
-from synapse.rest.client.v2_alpha._base import client_v2_patterns
+from synapse.rest.client.v2_alpha._base import client_patterns
 from synapse.server import HomeServer
 from synapse.storage.engines import create_engine
 from synapse.util.httpresourcetree import create_resource_tree
@@ -49,11 +48,11 @@ from synapse.util.versionstring import get_version_string
 logger = logging.getLogger("synapse.app.frontend_proxy")


-class PresenceStatusStubServlet(ClientV1RestServlet):
-    PATTERNS = client_path_patterns("/presence/(?P<user_id>[^/]*)/status")
+class PresenceStatusStubServlet(RestServlet):
+    PATTERNS = client_patterns("/presence/(?P<user_id>[^/]*)/status")

    def __init__(self, hs):
-        super(PresenceStatusStubServlet, self).__init__(hs)
+        super(PresenceStatusStubServlet, self).__init__()
        self.http_client = hs.get_simple_http_client()
        self.auth = hs.get_auth()
        self.main_uri = hs.config.worker_main_http_uri
@@ -84,7 +83,7 @@ class PresenceStatusStubServlet(ClientV1RestServlet):


 class KeyUploadServlet(RestServlet):
-    PATTERNS = client_v2_patterns("/keys/upload(/(?P<device_id>[^/]+))?$")
+    PATTERNS = client_patterns("/keys/upload(/(?P<device_id>[^/]+))?$")

    def __init__(self, hs):
        """
@@ -176,6 +176,7 @@ class SynapseHomeServer(HomeServer):

            resources.update({
                "/_matrix/client/api/v1": client_resource,
+                "/_synapse/password_reset": client_resource,
                "/_matrix/client/r0": client_resource,
                "/_matrix/client/unstable": client_resource,
                "/_matrix/client/v2_alpha": client_resource,
@@ -1,5 +1,7 @@
 # -*- coding: utf-8 -*-
-# Copyright 2015, 2016 OpenMarket Ltd
+# Copyright 2015-2016 OpenMarket Ltd
+# Copyright 2017-2018 New Vector Ltd
+# Copyright 2019 The Matrix.org Foundation C.I.C.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -29,12 +31,76 @@ logger = logging.getLogger(__name__)

 class EmailConfig(Config):
    def read_config(self, config):
+        # TODO: We should separate better the email configuration from the notification
+        # and account validity config.
+
        self.email_enable_notifs = False

        email_config = config.get("email", {})
-        self.email_enable_notifs = email_config.get("enable_notifs", False)

-        if self.email_enable_notifs:
+        self.email_smtp_host = email_config.get("smtp_host", None)
+        self.email_smtp_port = email_config.get("smtp_port", None)
+        self.email_smtp_user = email_config.get("smtp_user", None)
+        self.email_smtp_pass = email_config.get("smtp_pass", None)
+        self.require_transport_security = email_config.get(
+            "require_transport_security", False
+        )
+        if "app_name" in email_config:
+            self.email_app_name = email_config["app_name"]
+        else:
+            self.email_app_name = "Matrix"
+
+        # TODO: Rename notif_from to something more generic, or have a separate
+        # from for password resets, message notifications, etc?
+        # Currently the email section is a bit bogged down with settings for
+        # multiple functions. Would be good to split it out into separate
+        # sections and only put the common ones under email:
+        self.email_notif_from = email_config.get("notif_from", None)
+        if self.email_notif_from is not None:
+            # make sure it's valid
+            parsed = email.utils.parseaddr(self.email_notif_from)
+            if parsed[1] == '':
+                raise RuntimeError("Invalid notif_from address")
+
+        template_dir = email_config.get("template_dir")
+        # we need an absolute path, because we change directory after starting (and
+        # we don't yet know what auxilliary templates like mail.css we will need).
+        # (Note that loading as package_resources with jinja.PackageLoader doesn't
+        # work for the same reason.)
+        if not template_dir:
+            template_dir = pkg_resources.resource_filename(
+                'synapse', 'res/templates'
+            )
+
+        self.email_template_dir = os.path.abspath(template_dir)
+
+        self.email_enable_notifs = email_config.get("enable_notifs", False)
+        account_validity_renewal_enabled = config.get(
+            "account_validity", {},
+        ).get("renew_at")
+
+        email_trust_identity_server_for_password_resets = email_config.get(
+            "trust_identity_server_for_password_resets", False,
+        )
+        self.email_password_reset_behaviour = (
+            "remote" if email_trust_identity_server_for_password_resets else "local"
+        )
+        if self.email_password_reset_behaviour == "local" and email_config == {}:
+            logger.warn(
+                "User password resets have been disabled due to lack of email config"
+            )
+            self.email_password_reset_behaviour = "off"
+
+        # Get lifetime of a validation token in milliseconds
+        self.email_validation_token_lifetime = self.parse_duration(
+            email_config.get("validation_token_lifetime", "1h")
+        )
+
+        if (
+            self.email_enable_notifs
+            or account_validity_renewal_enabled
+            or self.email_password_reset_behaviour == "local"
+        ):
            # make sure we can import the required deps
            import jinja2
            import bleach
@@ -42,6 +108,68 @@ class EmailConfig(Config):
            jinja2
            bleach

+        if self.email_password_reset_behaviour == "local":
+            required = [
+                "smtp_host",
+                "smtp_port",
+                "notif_from",
+            ]
+
+            missing = []
+            for k in required:
+                if k not in email_config:
+                    missing.append(k)
+
+            if (len(missing) > 0):
+                raise RuntimeError(
+                    "email.password_reset_behaviour is set to 'local' "
+                    "but required keys are missing: %s" %
+                    (", ".join(["email." + k for k in missing]),)
+                )
+
+            # Templates for password reset emails
+            self.email_password_reset_template_html = email_config.get(
+                "password_reset_template_html", "password_reset.html",
+            )
+            self.email_password_reset_template_text = email_config.get(
+                "password_reset_template_text", "password_reset.txt",
+            )
+            self.email_password_reset_failure_template = email_config.get(
+                "password_reset_failure_template", "password_reset_failure.html",
+            )
+            # This template does not support any replaceable variables, so we will
+            # read it from the disk once during setup
+            email_password_reset_success_template = email_config.get(
+                "password_reset_success_template", "password_reset_success.html",
+            )
+
+            # Check templates exist
+            for f in [self.email_password_reset_template_html,
+                      self.email_password_reset_template_text,
+                      self.email_password_reset_failure_template,
+                      email_password_reset_success_template]:
+                p = os.path.join(self.email_template_dir, f)
+                if not os.path.isfile(p):
+                    raise ConfigError("Unable to find template file %s" % (p, ))
+
+            # Retrieve content of web templates
+            filepath = os.path.join(
+                self.email_template_dir,
+                email_password_reset_success_template,
+            )
+            self.email_password_reset_success_html_content = self.read_file(
+                filepath,
+                "email.password_reset_template_success_html",
+            )
+
+            if config.get("public_baseurl") is None:
+                raise RuntimeError(
+                    "email.password_reset_behaviour is set to 'local' but no "
+                    "public_baseurl is set. This is necessary to generate password "
+                    "reset links"
+                )
+
+        if self.email_enable_notifs:
            required = [
                "smtp_host",
                "smtp_port",
@@ -66,34 +194,13 @@ class EmailConfig(Config):
                    "email.enable_notifs is True but no public_baseurl is set"
                )

-            self.email_smtp_host = email_config["smtp_host"]
-            self.email_smtp_port = email_config["smtp_port"]
-            self.email_notif_from = email_config["notif_from"]
            self.email_notif_template_html = email_config["notif_template_html"]
            self.email_notif_template_text = email_config["notif_template_text"]
-            self.email_expiry_template_html = email_config.get(
-                "expiry_template_html", "notice_expiry.html",
-            )
-            self.email_expiry_template_text = email_config.get(
-                "expiry_template_text", "notice_expiry.txt",
-            )
-
-            template_dir = email_config.get("template_dir")
-            # we need an absolute path, because we change directory after starting (and
-            # we don't yet know what auxilliary templates like mail.css we will need).
-            # (Note that loading as package_resources with jinja.PackageLoader doesn't
-            # work for the same reason.)
-            if not template_dir:
-                template_dir = pkg_resources.resource_filename(
-                    'synapse', 'res/templates'
-                )
-            template_dir = os.path.abspath(template_dir)

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

            self.email_notif_for_new_users = email_config.get(
                "notif_for_new_users", True
@@ -101,35 +208,24 @@ class EmailConfig(Config):
            self.email_riot_base_url = email_config.get(
                "riot_base_url", None
            )
-            self.email_smtp_user = email_config.get(
-                "smtp_user", None
-            )
-            self.email_smtp_pass = email_config.get(
-                "smtp_pass", None
-            )
-            self.require_transport_security = email_config.get(
-                "require_transport_security", False
-            )
-            if "app_name" in email_config:
-                self.email_app_name = email_config["app_name"]
-            else:
-                self.email_app_name = "Matrix"

-            # make sure it's valid
-            parsed = email.utils.parseaddr(self.email_notif_from)
-            if parsed[1] == '':
-                raise RuntimeError("Invalid notif_from address")
-        else:
-            self.email_enable_notifs = False
-            # Not much point setting defaults for the rest: it would be an
-            # error for them to be used.
+        if account_validity_renewal_enabled:
+            self.email_expiry_template_html = email_config.get(
+                "expiry_template_html", "notice_expiry.html",
+            )
+            self.email_expiry_template_text = email_config.get(
+                "expiry_template_text", "notice_expiry.txt",
+            )
+
+            for f in self.email_expiry_template_text, self.email_expiry_template_html:
+                p = os.path.join(self.email_template_dir, f)
+                if not os.path.isfile(p):
+                    raise ConfigError("Unable to find email template file %s" % (p, ))

    def default_config(self, config_dir_path, server_name, **kwargs):
        return """
-        # Enable sending emails for notification events or expiry notices
-        # Defining a custom URL for Riot is only needed if email notifications
-        # should contain links to a self-hosted installation of Riot; when set
-        # the "app_name" setting is ignored.
+        # Enable sending emails for password resets, notification events or
+        # account expiry notices
        #
        # If your SMTP server requires authentication, the optional smtp_user &
        # smtp_pass variables should be used
@@ -137,20 +233,62 @@ class EmailConfig(Config):
        #email:
        #   enable_notifs: false
        #   smtp_host: "localhost"
-        #   smtp_port: 25
+        #   smtp_port: 25 # SSL: 465, STARTTLS: 587
        #   smtp_user: "exampleusername"
        #   smtp_pass: "examplepassword"
        #   require_transport_security: False
        #   notif_from: "Your Friendly %(app)s Home Server <noreply@example.com>"
        #   app_name: Matrix
-        #   # if template_dir is unset, uses the example templates that are part of
-        #   # the Synapse distribution.
+        #
+        #   # Enable email notifications by default
+        #   notif_for_new_users: True
+        #
+        #   # Defining a custom URL for Riot is only needed if email notifications
+        #   # should contain links to a self-hosted installation of Riot; when set
+        #   # the "app_name" setting is ignored
+        #   riot_base_url: "http://localhost/riot"
+        #
+        #   # Enable sending password reset emails via the configured, trusted
+        #   # identity servers
+        #   #
+        #   # IMPORTANT! This will give a malicious or overtaken identity server
+        #   # the ability to reset passwords for your users! Make absolutely sure
+        #   # that you want to do this! It is strongly recommended that password
+        #   # reset emails be sent by the homeserver instead
+        #   #
+        #   # If this option is set to false and SMTP options have not been
+        #   # configured, resetting user passwords via email will be disabled
+        #   #trust_identity_server_for_password_resets: false
+        #
+        #   # Configure the time that a validation email or text message code
+        #   # will expire after sending
+        #   #
+        #   # This is currently used for password resets
+        #   #validation_token_lifetime: 1h
+        #
+        #   # Template directory. All template files should be stored within this
+        #   # directory
+        #   #
        #   #template_dir: res/templates
+        #
+        #   # Templates for email notifications
+        #   #
        #   notif_template_html: notif_mail.html
        #   notif_template_text: notif_mail.txt
-        #   # Templates for account expiry notices.
+        #
+        #   # Templates for account expiry notices
+        #   #
        #   expiry_template_html: notice_expiry.html
        #   expiry_template_text: notice_expiry.txt
-        #   notif_for_new_users: True
-        #   riot_base_url: "http://localhost/riot"
+        #
+        #   # Templates for password reset emails sent by the homeserver
+        #   #
+        #   #password_reset_template_html: password_reset.html
+        #   #password_reset_template_text: password_reset.txt
+        #
+        #   # Templates for password reset success and failure pages that a user
+        #   # will see after attempting to reset their password
+        #   #
+        #   #password_reset_template_success_html: password_reset_success.html
+        #   #password_reset_template_failure_html: password_reset_failure.html
        """
@@ -13,6 +13,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+
 from .api import ApiConfig
 from .appservice import AppServiceConfig
 from .captcha import CaptchaConfig
@@ -36,20 +37,41 @@ from .saml2_config import SAML2Config
 from .server import ServerConfig
 from .server_notices_config import ServerNoticesConfig
 from .spam_checker import SpamCheckerConfig
+from .stats import StatsConfig
 from .tls import TlsConfig
 from .user_directory import UserDirectoryConfig
 from .voip import VoipConfig
 from .workers import WorkerConfig


-class HomeServerConfig(ServerConfig, TlsConfig, DatabaseConfig, LoggingConfig,
-                       RatelimitConfig, ContentRepositoryConfig, CaptchaConfig,
-                       VoipConfig, RegistrationConfig, MetricsConfig, ApiConfig,
-                       AppServiceConfig, KeyConfig, SAML2Config, CasConfig,
-                       JWTConfig, PasswordConfig, EmailConfig,
-                       WorkerConfig, PasswordAuthProviderConfig, PushConfig,
-                       SpamCheckerConfig, GroupsConfig, UserDirectoryConfig,
-                       ConsentConfig,
-                       ServerNoticesConfig, RoomDirectoryConfig,
-                       ):
+class HomeServerConfig(
+    ServerConfig,
+    TlsConfig,
+    DatabaseConfig,
+    LoggingConfig,
+    RatelimitConfig,
+    ContentRepositoryConfig,
+    CaptchaConfig,
+    VoipConfig,
+    RegistrationConfig,
+    MetricsConfig,
+    ApiConfig,
+    AppServiceConfig,
+    KeyConfig,
+    SAML2Config,
+    CasConfig,
+    JWTConfig,
+    PasswordConfig,
+    EmailConfig,
+    WorkerConfig,
+    PasswordAuthProviderConfig,
+    PushConfig,
+    SpamCheckerConfig,
+    GroupsConfig,
+    UserDirectoryConfig,
+    ConsentConfig,
+    StatsConfig,
+    ServerNoticesConfig,
+    RoomDirectoryConfig,
+):
    pass
@@ -39,6 +39,8 @@ class AccountValidityConfig(Config):
            else:
                self.renew_email_subject = "Renew your %(app)s account"

+            self.startup_job_max_delta = self.period * 10. / 100.
+
        if self.renew_by_email_enabled and "public_baseurl" not in synapse_config:
            raise ConfigError("Can't send renewal emails without 'public_baseurl'")

@@ -129,7 +131,9 @@ class RegistrationConfig(Config):
        # This means that, if a validity period is set, and Synapse is restarted (it will
        # then derive an expiration date from the current validity period), and some time
        # after that the validity period changes and Synapse is restarted, the users'
-        # expiration dates won't be updated unless their account is manually renewed.
+        # expiration dates won't be updated unless their account is manually renewed. This
+        # date will be randomly selected within a range [now + period - d ; now + period],
+        # where d is equal to 10%% of the validity period.
        #
        #account_validity:
        #  enabled: True
@@ -20,6 +20,7 @@ import os.path

 from netaddr import IPSet

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

@@ -35,6 +36,8 @@ logger = logging.Logger(__name__)
 # in the list.
 DEFAULT_BIND_ADDRESSES = ['::', '0.0.0.0']

+DEFAULT_ROOM_VERSION = "1"
+

 class ServerConfig(Config):

@@ -88,6 +91,22 @@ class ServerConfig(Config):
            "restrict_public_rooms_to_local_users", False,
        )

+        default_room_version = config.get(
+            "default_room_version", DEFAULT_ROOM_VERSION,
+        )
+
+        # Ensure room version is a str
+        default_room_version = str(default_room_version)
+
+        if default_room_version not in KNOWN_ROOM_VERSIONS:
+            raise ConfigError(
+                "Unknown default_room_version: %s, known room versions: %s" %
+                (default_room_version, list(KNOWN_ROOM_VERSIONS.keys()))
+            )
+
+        # Get the actual room version object rather than just the identifier
+        self.default_room_version = KNOWN_ROOM_VERSIONS[default_room_version]
+
        # whether to enable search. If disabled, new entries will not be inserted
        # into the search tables and they will not be indexed. Users will receive
        # errors when attempting to search for messages.
@@ -310,6 +329,10 @@ class ServerConfig(Config):
            unsecure_port = 8008

        pid_file = os.path.join(data_dir_path, "homeserver.pid")
+
+        # Bring DEFAULT_ROOM_VERSION into the local-scope for use in the
+        # default config string
+        default_room_version = DEFAULT_ROOM_VERSION
        return """\
        ## Server ##

@@ -384,6 +407,16 @@ class ServerConfig(Config):
        #
        #restrict_public_rooms_to_local_users: true

+        # The default room version for newly created rooms.
+        #
+        # Known room versions are listed here:
+        # https://matrix.org/docs/spec/#complete-list-of-room-versions
+        #
+        # For example, for room version 1, default_room_version should be set
+        # to "1".
+        #
+        #default_room_version: "%(default_room_version)s"
+
        # The GC threshold parameters to pass to `gc.set_threshold`, if defined
        #
        #gc_thresholds: [700, 10, 10]
@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+# Copyright 2018 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 __future__ import division
+
+import sys
+
+from ._base import Config
+
+
+class StatsConfig(Config):
+    """Stats Configuration
+    Configuration for the behaviour of synapse's stats engine
+    """
+
+    def read_config(self, config):
+        self.stats_enabled = True
+        self.stats_bucket_size = 86400
+        self.stats_retention = sys.maxsize
+        stats_config = config.get("stats", None)
+        if stats_config:
+            self.stats_enabled = stats_config.get("enabled", self.stats_enabled)
+            self.stats_bucket_size = (
+                self.parse_duration(stats_config.get("bucket_size", "1d")) / 1000
+            )
+            self.stats_retention = (
+                self.parse_duration(
+                    stats_config.get("retention", "%ds" % (sys.maxsize,))
+                )
+                / 1000
+            )
+
+    def default_config(self, config_dir_path, server_name, **kwargs):
+        return """
+        # Local statistics collection. Used in populating the room directory.
+        #
+        # 'bucket_size' controls how large each statistics timeslice is. It can
+        # be defined in a human readable short form -- e.g. "1d", "1y".
+        #
+        # 'retention' controls how long historical statistics will be kept for.
+        # It can be defined in a human readable short form -- e.g. "1d", "1y".
+        #
+        #
+        #stats:
+        #   enabled: true
+        #   bucket_size: 1d
+        #   retention: 1y
+        """
@@ -107,7 +107,7 @@ class TlsConfig(Config):
            certs = []
            for ca_file in custom_ca_list:
                logger.debug("Reading custom CA certificate file: %s", ca_file)
-                content = self.read_file(ca_file)
+                content = self.read_file(ca_file, "federation_custom_ca_list")

                # Parse the CA certificates
                try:
@@ -43,9 +43,9 @@ class UserDirectoryConfig(Config):
        #
        # '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
-        # UPDATE user_directory_stream_pos SET stream_id = NULL;
-        # on your database to tell it to rebuild the user_directory search indexes.
+        # in public rooms.  Defaults to false.  If you set it True, you'll have to
+        # rebuild the user_directory search indexes, see
+        # https://github.com/matrix-org/synapse/blob/master/docs/user_directory.md
        #
        #user_directory:
        #  enabled: true
@@ -76,6 +76,7 @@ class EventBuilder(object):
    # someone tries to get them when they don't exist.
    _state_key = attr.ib(default=None)
    _redacts = attr.ib(default=None)
+    _origin_server_ts = attr.ib(default=None)

    internal_metadata = attr.ib(default=attr.Factory(lambda: _EventInternalMetadata({})))

@@ -142,6 +143,9 @@ class EventBuilder(object):
        if self._redacts is not None:
            event_dict["redacts"] = self._redacts

+        if self._origin_server_ts is not None:
+            event_dict["origin_server_ts"] = self._origin_server_ts
+
        defer.returnValue(
            create_local_event_from_event_dict(
                clock=self._clock,
@@ -209,6 +213,7 @@ class EventBuilderFactory(object):
            content=key_values.get("content", {}),
            unsigned=key_values.get("unsigned", {}),
            redacts=key_values.get("redacts", None),
+            origin_server_ts=key_values.get("origin_server_ts", None),
        )


@@ -245,7 +250,7 @@ def create_local_event_from_event_dict(clock, hostname, signing_key,
        event_dict["event_id"] = _create_event_id(clock, hostname)

    event_dict["origin"] = hostname
-    event_dict["origin_server_ts"] = time_now
+    event_dict.setdefault("origin_server_ts", time_now)

    event_dict.setdefault("unsigned", {})
    age = event_dict["unsigned"].pop("age", 0)
@@ -330,12 +330,13 @@ class EventClientSerializer(object):
        )

    @defer.inlineCallbacks
-    def serialize_event(self, event, time_now, **kwargs):
+    def serialize_event(self, event, time_now, bundle_aggregations=True, **kwargs):
        """Serializes a single event.

        Args:
            event (EventBase)
            time_now (int): The current time in milliseconds
+            bundle_aggregations (bool): Whether to bundle in related events
            **kwargs: Arguments to pass to `serialize_event`

        Returns:
@@ -350,7 +351,7 @@ class EventClientSerializer(object):

        # If MSC1849 is enabled then we need to look if thre are any relations
        # we need to bundle in with the event
-        if self.experimental_msc1849_support_enabled:
+        if self.experimental_msc1849_support_enabled and bundle_aggregations:
            annotations = yield self.store.get_aggregation_groups_for_event(
                event_id,
            )
@@ -265,7 +265,7 @@ def _check_sigs_on_pdus(keyring, room_version, pdus):
    ]

    more_deferreds = keyring.verify_json_objects_for_server([
-        (p.sender_domain, p.redacted_pdu_json)
+        (p.sender_domain, p.redacted_pdu_json, 0)
        for p in pdus_to_check_sender
    ])

@@ -298,7 +298,7 @@ def _check_sigs_on_pdus(keyring, room_version, pdus):
        ]

        more_deferreds = keyring.verify_json_objects_for_server([
-            (get_domain_from_id(p.pdu.event_id), p.redacted_pdu_json)
+            (get_domain_from_id(p.pdu.event_id), p.redacted_pdu_json, 0)
            for p in pdus_to_check_event_id
        ])

@@ -23,7 +23,11 @@ from twisted.internet import defer
 import synapse
 from synapse.api.errors import Codes, FederationDeniedError, SynapseError
 from synapse.api.room_versions import RoomVersions
-from synapse.api.urls import FEDERATION_V1_PREFIX, FEDERATION_V2_PREFIX
+from synapse.api.urls import (
+    FEDERATION_UNSTABLE_PREFIX,
+    FEDERATION_V1_PREFIX,
+    FEDERATION_V2_PREFIX,
+)
 from synapse.http.endpoint import parse_and_validate_server_name
 from synapse.http.server import JsonResource
 from synapse.http.servlet import (
@@ -90,6 +94,7 @@ class NoAuthenticationError(AuthenticationError):

 class Authenticator(object):
    def __init__(self, hs):
+        self._clock = hs.get_clock()
        self.keyring = hs.get_keyring()
        self.server_name = hs.hostname
        self.store = hs.get_datastore()
@@ -98,6 +103,7 @@ class Authenticator(object):
    # A method just so we can pass 'self' as the authenticator to the Servlets
    @defer.inlineCallbacks
    def authenticate_request(self, request, content):
+        now = self._clock.time_msec()
        json_request = {
            "method": request.method.decode('ascii'),
            "uri": request.uri.decode('ascii'),
@@ -134,7 +140,7 @@ class Authenticator(object):
                401, "Missing Authorization headers", Codes.UNAUTHORIZED,
            )

-        yield self.keyring.verify_json_for_server(origin, json_request)
+        yield self.keyring.verify_json_for_server(origin, json_request, now)

        logger.info("Request from %s", origin)
        request.authenticated_entity = origin
@@ -1304,6 +1310,30 @@ class FederationGroupsSettingJoinPolicyServlet(BaseFederationServlet):
        defer.returnValue((200, new_content))


+class RoomComplexityServlet(BaseFederationServlet):
+    """
+    Indicates to other servers how complex (and therefore likely
+    resource-intensive) a public room this server knows about is.
+    """
+    PATH = "/rooms/(?P<room_id>[^/]*)/complexity"
+    PREFIX = FEDERATION_UNSTABLE_PREFIX
+
+    @defer.inlineCallbacks
+    def on_GET(self, origin, content, query, room_id):
+
+        store = self.handler.hs.get_datastore()
+
+        is_public = yield store.is_room_world_readable_or_publicly_joinable(
+            room_id
+        )
+
+        if not is_public:
+            raise SynapseError(404, "Room not found", errcode=Codes.INVALID_PARAM)
+
+        complexity = yield store.get_room_complexity(room_id)
+        defer.returnValue((200, complexity))
+
+
 FEDERATION_SERVLET_CLASSES = (
    FederationSendServlet,
    FederationEventServlet,
@@ -1327,6 +1357,7 @@ FEDERATION_SERVLET_CLASSES = (
    FederationThirdPartyInviteExchangeServlet,
    On3pidBindServlet,
    FederationVersionServlet,
+    RoomComplexityServlet,
 )

 OPENID_SERVLET_CLASSES = (
@@ -97,10 +97,11 @@ class GroupAttestationSigning(object):

        # TODO: We also want to check that *new* attestations that people give
        # us to store are valid for at least a little while.
-        if valid_until_ms < self.clock.time_msec():
+        now = self.clock.time_msec()
+        if valid_until_ms < now:
            raise SynapseError(400, "Attestation expired")

-        yield self.keyring.verify_json_for_server(server_name, attestation)
+        yield self.keyring.verify_json_for_server(server_name, attestation, now)

    def create_attestation(self, group_id, user_id):
        """Create an attestation for the group_id and user_id with default
@@ -162,7 +162,7 @@ class AuthHandler(BaseHandler):
        defer.returnValue(params)

    @defer.inlineCallbacks
-    def check_auth(self, flows, clientdict, clientip):
+    def check_auth(self, flows, clientdict, clientip, password_servlet=False):
        """
        Takes a dictionary sent by the client in the login / registration
        protocol and handles the User-Interactive Auth flow.
@@ -186,6 +186,16 @@ class AuthHandler(BaseHandler):

            clientip (str): The IP address of the client.

+            password_servlet (bool): Whether the request originated from
+                PasswordRestServlet.
+                XXX: This is a temporary hack to distinguish between checking
+                for threepid validations locally (in the case of password
+                resets) and using the identity server (in the case of binding
+                a 3PID during registration). Once we start using the
+                homeserver for both tasks, this distinction will no longer be
+                necessary.
+
+
        Returns:
            defer.Deferred[dict, dict, str]: a deferred tuple of
                (creds, params, session_id).
@@ -241,7 +251,9 @@ class AuthHandler(BaseHandler):
        if 'type' in authdict:
            login_type = authdict['type']
            try:
-                result = yield self._check_auth_dict(authdict, clientip)
+                result = yield self._check_auth_dict(
+                    authdict, clientip, password_servlet=password_servlet,
+                )
                if result:
                    creds[login_type] = result
                    self._save_session(session)
@@ -351,7 +363,7 @@ class AuthHandler(BaseHandler):
        return sess.setdefault('serverdict', {}).get(key, default)

    @defer.inlineCallbacks
-    def _check_auth_dict(self, authdict, clientip):
+    def _check_auth_dict(self, authdict, clientip, password_servlet=False):
        """Attempt to validate the auth dict provided by a client

        Args:
@@ -369,7 +381,13 @@ class AuthHandler(BaseHandler):
        login_type = authdict['type']
        checker = self.checkers.get(login_type)
        if checker is not None:
-            res = yield checker(authdict, clientip)
+            # XXX: Temporary workaround for having Synapse handle password resets
+            # See AuthHandler.check_auth for further details
+            res = yield checker(
+                authdict,
+                clientip=clientip,
+                password_servlet=password_servlet,
+            )
            defer.returnValue(res)

        # build a v1-login-style dict out of the authdict and fall back to the
@@ -383,7 +401,7 @@ class AuthHandler(BaseHandler):
        defer.returnValue(canonical_id)

    @defer.inlineCallbacks
-    def _check_recaptcha(self, authdict, clientip):
+    def _check_recaptcha(self, authdict, clientip, **kwargs):
        try:
            user_response = authdict["response"]
        except KeyError:
@@ -429,20 +447,20 @@ class AuthHandler(BaseHandler):
                defer.returnValue(True)
        raise LoginError(401, "", errcode=Codes.UNAUTHORIZED)

-    def _check_email_identity(self, authdict, _):
-        return self._check_threepid('email', authdict)
+    def _check_email_identity(self, authdict, **kwargs):
+        return self._check_threepid('email', authdict, **kwargs)

-    def _check_msisdn(self, authdict, _):
+    def _check_msisdn(self, authdict, **kwargs):
        return self._check_threepid('msisdn', authdict)

-    def _check_dummy_auth(self, authdict, _):
+    def _check_dummy_auth(self, authdict, **kwargs):
        return defer.succeed(True)

-    def _check_terms_auth(self, authdict, _):
+    def _check_terms_auth(self, authdict, **kwargs):
        return defer.succeed(True)

    @defer.inlineCallbacks
-    def _check_threepid(self, medium, authdict):
+    def _check_threepid(self, medium, authdict, password_servlet=False, **kwargs):
        if 'threepid_creds' not in authdict:
            raise LoginError(400, "Missing threepid_creds", Codes.MISSING_PARAM)

@@ -451,7 +469,29 @@ class AuthHandler(BaseHandler):
        identity_handler = self.hs.get_handlers().identity_handler

        logger.info("Getting validated threepid. threepidcreds: %r", (threepid_creds,))
-        threepid = yield identity_handler.threepid_from_creds(threepid_creds)
+        if (
+            not password_servlet
+            or self.hs.config.email_password_reset_behaviour == "remote"
+        ):
+            threepid = yield identity_handler.threepid_from_creds(threepid_creds)
+        elif self.hs.config.email_password_reset_behaviour == "local":
+            row = yield self.store.get_threepid_validation_session(
+                medium,
+                threepid_creds["client_secret"],
+                sid=threepid_creds["sid"],
+            )
+
+            threepid = {
+                "medium": row["medium"],
+                "address": row["address"],
+                "validated_at": row["validated_at"],
+            } if row else None
+
+            if row:
+                # Valid threepid returned, delete from the db
+                yield self.store.delete_threepid_session(threepid_creds["sid"])
+        else:
+            raise SynapseError(400, "Password resets are not enabled on this homeserver")

        if not threepid:
            raise LoginError(401, "", errcode=Codes.UNAUTHORIZED)
@@ -122,6 +122,9 @@ class EventStreamHandler(BaseHandler):

            chunks = yield self._event_serializer.serialize_events(
                events, time_now, as_client_event=as_client_event,
+                # We don't bundle "live" events, as otherwise clients
+                # will end up double counting annotations.
+                bundle_aggregations=False,
            )

            chunk = {
@@ -2013,15 +2013,44 @@ class FederationHandler(BaseHandler):

        Args:
            origin (str):
-            event (synapse.events.FrozenEvent):
+            event (synapse.events.EventBase):
            context (synapse.events.snapshot.EventContext):
-            auth_events (dict[(str, str)->str]):
+            auth_events (dict[(str, str)->synapse.events.EventBase]):
+                Map from (event_type, state_key) to event
+
+                What we expect the event's auth_events to be, based on the event's
+                position in the dag. I think? maybe??
+
+                Also NB that this function adds entries to it.
+        Returns:
+            defer.Deferred[None]
+        """
+        room_version = yield self.store.get_room_version(event.room_id)
+
+        yield self._update_auth_events_and_context_for_auth(
+            origin, event, context, auth_events
+        )
+        try:
+            self.auth.check(room_version, event, auth_events=auth_events)
+        except AuthError as e:
+            logger.warn("Failed auth resolution for %r because %s", event, e)
+            raise e
+
+    @defer.inlineCallbacks
+    def _update_auth_events_and_context_for_auth(
+        self, origin, event, context, auth_events
+    ):
+        """Helper for do_auth. See there for docs.
+
+        Args:
+            origin (str):
+            event (synapse.events.EventBase):
+            context (synapse.events.snapshot.EventContext):
+            auth_events (dict[(str, str)->synapse.events.EventBase]):

        Returns:
            defer.Deferred[None]
        """
-        # Check if we have all the auth events.
-        current_state = set(e.event_id for e in auth_events.values())
        event_auth_events = set(event.auth_event_ids())

        if event.is_state():
@@ -2029,11 +2058,21 @@ class FederationHandler(BaseHandler):
        else:
            event_key = None

-        if event_auth_events - current_state:
+        # if the event's auth_events refers to events which are not in our
+        # calculated auth_events, we need to fetch those events from somewhere.
+        #
+        # we start by fetching them from the store, and then try calling /event_auth/.
+        missing_auth = event_auth_events.difference(
+            e.event_id for e in auth_events.values()
+        )
+
+        if missing_auth:
            # TODO: can we use store.have_seen_events here instead?
            have_events = yield self.store.get_seen_events_with_rejections(
-                event_auth_events - current_state
+                missing_auth
            )
+            logger.debug("Got events %s from store", have_events)
+            missing_auth.difference_update(have_events.keys())
        else:
            have_events = {}

@@ -2042,13 +2081,12 @@ class FederationHandler(BaseHandler):
            for e in auth_events.values()
        })

-        seen_events = set(have_events.keys())
-
-        missing_auth = event_auth_events - seen_events - current_state
-
        if missing_auth:
-            logger.info("Missing auth: %s", missing_auth)
            # If we don't have all the auth events, we need to get them.
+            logger.info(
+                "auth_events contains unknown events: %s",
+                missing_auth,
+            )
            try:
                remote_auth_chain = yield self.federation_client.get_event_auth(
                    origin, event.room_id, event.event_id
@@ -2089,145 +2127,168 @@ class FederationHandler(BaseHandler):
                have_events = yield self.store.get_seen_events_with_rejections(
                    event.auth_event_ids()
                )
-                seen_events = set(have_events.keys())
            except Exception:
                # FIXME:
                logger.exception("Failed to get auth chain")

+        if event.internal_metadata.is_outlier():
+            logger.info("Skipping auth_event fetch for outlier")
+            return
+
        # FIXME: Assumes we have and stored all the state for all the
        # prev_events
-        current_state = set(e.event_id for e in auth_events.values())
-        different_auth = event_auth_events - current_state
+        different_auth = event_auth_events.difference(
+            e.event_id for e in auth_events.values()
+        )
+
+        if not different_auth:
+            return
+
+        logger.info(
+            "auth_events refers to events which are not in our calculated auth "
+            "chain: %s",
+            different_auth,
+        )

        room_version = yield self.store.get_room_version(event.room_id)

-        if different_auth and not event.internal_metadata.is_outlier():
-            # Do auth conflict res.
-            logger.info("Different auth: %s", different_auth)
-
-            different_events = yield logcontext.make_deferred_yieldable(
-                defer.gatherResults([
-                    logcontext.run_in_background(
-                        self.store.get_event,
-                        d,
-                        allow_none=True,
-                        allow_rejected=False,
-                    )
-                    for d in different_auth
-                    if d in have_events and not have_events[d]
-                ], consumeErrors=True)
-            ).addErrback(unwrapFirstError)
-
-            if different_events:
-                local_view = dict(auth_events)
-                remote_view = dict(auth_events)
-                remote_view.update({
-                    (d.type, d.state_key): d for d in different_events if d
-                })
-
-                new_state = yield self.state_handler.resolve_events(
-                    room_version,
-                    [list(local_view.values()), list(remote_view.values())],
-                    event
+        different_events = yield logcontext.make_deferred_yieldable(
+            defer.gatherResults([
+                logcontext.run_in_background(
+                    self.store.get_event,
+                    d,
+                    allow_none=True,
+                    allow_rejected=False,
                )
+                for d in different_auth
+                if d in have_events and not have_events[d]
+            ], consumeErrors=True)
+        ).addErrback(unwrapFirstError)

-                auth_events.update(new_state)
+        if different_events:
+            local_view = dict(auth_events)
+            remote_view = dict(auth_events)
+            remote_view.update({
+                (d.type, d.state_key): d for d in different_events if d
+            })

-                current_state = set(e.event_id for e in auth_events.values())
-                different_auth = event_auth_events - current_state
+            new_state = yield self.state_handler.resolve_events(
+                room_version,
+                [list(local_view.values()), list(remote_view.values())],
+                event
+            )

-                yield self._update_context_for_auth_events(
-                    event, context, auth_events, event_key,
-                )
+            logger.info(
+                "After state res: updating auth_events with new state %s",
+                {
+                    (d.type, d.state_key): d.event_id for d in new_state.values()
+                    if auth_events.get((d.type, d.state_key)) != d
+                },
+            )

-        if different_auth and not event.internal_metadata.is_outlier():
-            logger.info("Different auth after resolution: %s", different_auth)
+            auth_events.update(new_state)

-            # Only do auth resolution if we have something new to say.
-            # We can't rove an auth failure.
-            do_resolution = False
+            different_auth = event_auth_events.difference(
+                e.event_id for e in auth_events.values()
+            )

-            provable = [
-                RejectedReason.NOT_ANCESTOR, RejectedReason.NOT_ANCESTOR,
-            ]
+            yield self._update_context_for_auth_events(
+                event, context, auth_events, event_key,
+            )

-            for e_id in different_auth:
-                if e_id in have_events:
-                    if have_events[e_id] in provable:
-                        do_resolution = True
-                        break
+        if not different_auth:
+            # we're done
+            return

-            if do_resolution:
-                prev_state_ids = yield context.get_prev_state_ids(self.store)
-                # 1. Get what we think is the auth chain.
-                auth_ids = yield self.auth.compute_auth_events(
-                    event, prev_state_ids
-                )
-                local_auth_chain = yield self.store.get_auth_chain(
-                    auth_ids, include_given=True
-                )
+        logger.info(
+            "auth_events still refers to events which are not in the calculated auth "
+            "chain after state resolution: %s",
+            different_auth,
+        )

-                try:
-                    # 2. Get remote difference.
-                    result = yield self.federation_client.query_auth(
-                        origin,
-                        event.room_id,
-                        event.event_id,
-                        local_auth_chain,
-                    )
+        # Only do auth resolution if we have something new to say.
+        # We can't prove an auth failure.
+        do_resolution = False

-                    seen_remotes = yield self.store.have_seen_events(
-                        [e.event_id for e in result["auth_chain"]]
-                    )
+        for e_id in different_auth:
+            if e_id in have_events:
+                if have_events[e_id] == RejectedReason.NOT_ANCESTOR:
+                    do_resolution = True
+                    break

-                    # 3. Process any remote auth chain events we haven't seen.
-                    for ev in result["auth_chain"]:
-                        if ev.event_id in seen_remotes:
-                            continue
+        if not do_resolution:
+            logger.info(
+                "Skipping auth resolution due to lack of provable rejection reasons"
+            )
+            return

-                        if ev.event_id == event.event_id:
-                            continue
+        logger.info("Doing auth resolution")

-                        try:
-                            auth_ids = ev.auth_event_ids()
-                            auth = {
-                                (e.type, e.state_key): e
-                                for e in result["auth_chain"]
-                                if e.event_id in auth_ids
-                                or event.type == EventTypes.Create
-                            }
-                            ev.internal_metadata.outlier = True
+        prev_state_ids = yield context.get_prev_state_ids(self.store)

-                            logger.debug(
-                                "do_auth %s different_auth: %s",
-                                event.event_id, e.event_id
-                            )
-
-                            yield self._handle_new_event(
-                                origin, ev, auth_events=auth
-                            )
-
-                            if ev.event_id in event_auth_events:
-                                auth_events[(ev.type, ev.state_key)] = ev
-                        except AuthError:
-                            pass
-
-                except Exception:
-                    # FIXME:
-                    logger.exception("Failed to query auth chain")
-
-                # 4. Look at rejects and their proofs.
-                # TODO.
-
-                yield self._update_context_for_auth_events(
-                    event, context, auth_events, event_key,
-                )
+        # 1. Get what we think is the auth chain.
+        auth_ids = yield self.auth.compute_auth_events(
+            event, prev_state_ids
+        )
+        local_auth_chain = yield self.store.get_auth_chain(
+            auth_ids, include_given=True
+        )

        try:
-            self.auth.check(room_version, event, auth_events=auth_events)
-        except AuthError as e:
-            logger.warn("Failed auth resolution for %r because %s", event, e)
-            raise e
+            # 2. Get remote difference.
+            result = yield self.federation_client.query_auth(
+                origin,
+                event.room_id,
+                event.event_id,
+                local_auth_chain,
+            )
+
+            seen_remotes = yield self.store.have_seen_events(
+                [e.event_id for e in result["auth_chain"]]
+            )
+
+            # 3. Process any remote auth chain events we haven't seen.
+            for ev in result["auth_chain"]:
+                if ev.event_id in seen_remotes:
+                    continue
+
+                if ev.event_id == event.event_id:
+                    continue
+
+                try:
+                    auth_ids = ev.auth_event_ids()
+                    auth = {
+                        (e.type, e.state_key): e
+                        for e in result["auth_chain"]
+                        if e.event_id in auth_ids
+                        or event.type == EventTypes.Create
+                    }
+                    ev.internal_metadata.outlier = True
+
+                    logger.debug(
+                        "do_auth %s different_auth: %s",
+                        event.event_id, e.event_id
+                    )
+
+                    yield self._handle_new_event(
+                        origin, ev, auth_events=auth
+                    )
+
+                    if ev.event_id in event_auth_events:
+                        auth_events[(ev.type, ev.state_key)] = ev
+                except AuthError:
+                    pass
+
+        except Exception:
+            # FIXME:
+            logger.exception("Failed to query auth chain")
+
+        # 4. Look at rejects and their proofs.
+        # TODO.
+
+        yield self._update_context_for_auth_events(
+            event, context, auth_events, event_key,
+        )

    @defer.inlineCallbacks
    def _update_context_for_auth_events(self, event, context, auth_events,
@@ -247,7 +247,14 @@ class IdentityHandler(BaseHandler):
        defer.returnValue(changed)

    @defer.inlineCallbacks
-    def requestEmailToken(self, id_server, email, client_secret, send_attempt, **kwargs):
+    def requestEmailToken(
+        self,
+        id_server,
+        email,
+        client_secret,
+        send_attempt,
+        next_link=None,
+    ):
        if not self._should_trust_id_server(id_server):
            raise SynapseError(
                400, "Untrusted ID server '%s'" % id_server,
@@ -259,7 +266,9 @@ class IdentityHandler(BaseHandler):
            'client_secret': client_secret,
            'send_attempt': send_attempt,
        }
-        params.update(kwargs)
+
+        if next_link:
+            params.update({'next_link': next_link})

        try:
            data = yield self.http_client.post_json_get_json(
@@ -166,6 +166,9 @@ class MessageHandler(object):
        now = self.clock.time_msec()
        events = yield self._event_serializer.serialize_events(
            room_state.values(), now,
+            # We don't bother bundling aggregations in when asked for state
+            # events, as clients won't use them.
+            bundle_aggregations=False,
        )
        defer.returnValue(events)

@@ -182,17 +182,27 @@ class PresenceHandler(object):
        # Start a LoopingCall in 30s that fires every 5s.
        # The initial delay is to allow disconnected clients a chance to
        # reconnect before we treat them as offline.
+        def run_timeout_handler():
+            return run_as_background_process(
+                "handle_presence_timeouts", self._handle_timeouts
+            )
+
        self.clock.call_later(
            30,
            self.clock.looping_call,
-            self._handle_timeouts,
+            run_timeout_handler,
            5000,
        )

+        def run_persister():
+            return run_as_background_process(
+                "persist_presence_changes", self._persist_unpersisted_changes
+            )
+
        self.clock.call_later(
            60,
            self.clock.looping_call,
-            self._persist_unpersisted_changes,
+            run_persister,
            60 * 1000,
        )

@@ -229,6 +239,7 @@ class PresenceHandler(object):
        )

        if self.unpersisted_users_changes:
+
            yield self.store.update_presence([
                self.user_to_current_state[user_id]
                for user_id in self.unpersisted_users_changes
@@ -240,30 +251,18 @@ class PresenceHandler(object):
        """We periodically persist the unpersisted changes, as otherwise they
        may stack up and slow down shutdown times.
        """
-        logger.info(
-            "Performing _persist_unpersisted_changes. Persisting %d unpersisted changes",
-            len(self.unpersisted_users_changes)
-        )
-
        unpersisted = self.unpersisted_users_changes
        self.unpersisted_users_changes = set()

        if unpersisted:
+            logger.info(
+                "Persisting %d upersisted presence updates", len(unpersisted)
+            )
            yield self.store.update_presence([
                self.user_to_current_state[user_id]
                for user_id in unpersisted
            ])

-        logger.info("Finished _persist_unpersisted_changes")
-
-    @defer.inlineCallbacks
-    def _update_states_and_catch_exception(self, new_states):
-        try:
-            res = yield self._update_states(new_states)
-            defer.returnValue(res)
-        except Exception:
-            logger.exception("Error updating presence")
-
    @defer.inlineCallbacks
    def _update_states(self, new_states):
        """Updates presence of users. Sets the appropriate timeouts. Pokes
@@ -338,45 +337,41 @@ class PresenceHandler(object):
        logger.info("Handling presence timeouts")
        now = self.clock.time_msec()

-        try:
-            with Measure(self.clock, "presence_handle_timeouts"):
-                # Fetch the list of users that *may* have timed out. Things may have
-                # changed since the timeout was set, so we won't necessarily have to
-                # take any action.
-                users_to_check = set(self.wheel_timer.fetch(now))
+        # Fetch the list of users that *may* have timed out. Things may have
+        # changed since the timeout was set, so we won't necessarily have to
+        # take any action.
+        users_to_check = set(self.wheel_timer.fetch(now))

-                # Check whether the lists of syncing processes from an external
-                # process have expired.
-                expired_process_ids = [
-                    process_id for process_id, last_update
-                    in self.external_process_last_updated_ms.items()
-                    if now - last_update > EXTERNAL_PROCESS_EXPIRY
-                ]
-                for process_id in expired_process_ids:
-                    users_to_check.update(
-                        self.external_process_last_updated_ms.pop(process_id, ())
-                    )
-                    self.external_process_last_update.pop(process_id)
+        # Check whether the lists of syncing processes from an external
+        # process have expired.
+        expired_process_ids = [
+            process_id for process_id, last_update
+            in self.external_process_last_updated_ms.items()
+            if now - last_update > EXTERNAL_PROCESS_EXPIRY
+        ]
+        for process_id in expired_process_ids:
+            users_to_check.update(
+                self.external_process_last_updated_ms.pop(process_id, ())
+            )
+            self.external_process_last_update.pop(process_id)

-                states = [
-                    self.user_to_current_state.get(
-                        user_id, UserPresenceState.default(user_id)
-                    )
-                    for user_id in users_to_check
-                ]
+        states = [
+            self.user_to_current_state.get(
+                user_id, UserPresenceState.default(user_id)
+            )
+            for user_id in users_to_check
+        ]

-                timers_fired_counter.inc(len(states))
+        timers_fired_counter.inc(len(states))

-                changes = handle_timeouts(
-                    states,
-                    is_mine_fn=self.is_mine_id,
-                    syncing_user_ids=self.get_currently_syncing_users(),
-                    now=now,
-                )
+        changes = handle_timeouts(
+            states,
+            is_mine_fn=self.is_mine_id,
+            syncing_user_ids=self.get_currently_syncing_users(),
+            now=now,
+        )

-            run_in_background(self._update_states_and_catch_exception, changes)
-        except Exception:
-            logger.exception("Exception in _handle_timeouts loop")
+        return self._update_states(changes)

    @defer.inlineCallbacks
    def bump_presence_active_time(self, user):
@@ -31,6 +31,9 @@ from ._base import BaseHandler

 logger = logging.getLogger(__name__)

+MAX_DISPLAYNAME_LEN = 100
+MAX_AVATAR_URL_LEN = 1000
+

 class BaseProfileHandler(BaseHandler):
    """Handles fetching and updating user profile information.
@@ -162,6 +165,11 @@ class BaseProfileHandler(BaseHandler):
        if not by_admin and target_user != requester.user:
            raise AuthError(400, "Cannot set another user's displayname")

+        if len(new_displayname) > MAX_DISPLAYNAME_LEN:
+            raise SynapseError(
+                400, "Displayname is too long (max %i)" % (MAX_DISPLAYNAME_LEN, ),
+            )
+
        if new_displayname == '':
            new_displayname = None

@@ -217,6 +225,11 @@ class BaseProfileHandler(BaseHandler):
        if not by_admin and target_user != requester.user:
            raise AuthError(400, "Cannot set another user's avatar_url")

+        if len(new_avatar_url) > MAX_AVATAR_URL_LEN:
+            raise SynapseError(
+                400, "Avatar URL is too long (max %i)" % (MAX_AVATAR_URL_LEN, ),
+            )
+
        yield self.store.set_profile_avatar_url(
            target_user.localpart, new_avatar_url
        )
@@ -531,6 +531,8 @@ class RegistrationHandler(BaseHandler):
            A tuple of (user_id, access_token).
        Raises:
            RegistrationError if there was a problem registering.
+
+        NB this is only used in tests. TODO: move it to the test package!
        """
        if localpart is None:
            raise SynapseError(400, "Request must include user id")
@@ -27,7 +27,7 @@ from twisted.internet import defer

 from synapse.api.constants import EventTypes, JoinRules, RoomCreationPreset
 from synapse.api.errors import AuthError, Codes, NotFoundError, StoreError, SynapseError
-from synapse.api.room_versions import DEFAULT_ROOM_VERSION, KNOWN_ROOM_VERSIONS
+from synapse.api.room_versions import KNOWN_ROOM_VERSIONS
 from synapse.storage.state import StateFilter
 from synapse.types import RoomAlias, RoomID, RoomStreamToken, StreamToken, UserID
 from synapse.util import stringutils
@@ -70,6 +70,7 @@ class RoomCreationHandler(BaseHandler):
        self.spam_checker = hs.get_spam_checker()
        self.event_creation_handler = hs.get_event_creation_handler()
        self.room_member_handler = hs.get_room_member_handler()
+        self.config = hs.config

        # linearizer to stop two upgrades happening at once
        self._upgrade_linearizer = Linearizer("room_upgrade_linearizer")
@@ -475,7 +476,11 @@ class RoomCreationHandler(BaseHandler):
        if ratelimit:
            yield self.ratelimit(requester)

-        room_version = config.get("room_version", DEFAULT_ROOM_VERSION.identifier)
+        room_version = config.get(
+            "room_version",
+            self.config.default_room_version.identifier,
+        )
+
        if not isinstance(room_version, string_types):
            raise SynapseError(
                400,
@@ -0,0 +1,325 @@
+# -*- coding: utf-8 -*-
+# Copyright 2018 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.constants import EventTypes, JoinRules, Membership
+from synapse.handlers.state_deltas import StateDeltasHandler
+from synapse.metrics import event_processing_positions
+from synapse.metrics.background_process_metrics import run_as_background_process
+from synapse.types import UserID
+from synapse.util.metrics import Measure
+
+logger = logging.getLogger(__name__)
+
+
+class StatsHandler(StateDeltasHandler):
+    """Handles keeping the *_stats tables updated with a simple time-series of
+    information about the users, rooms and media on the server, such that admins
+    have some idea of who is consuming their resources.
+
+    Heavily derived from UserDirectoryHandler
+    """
+
+    def __init__(self, hs):
+        super(StatsHandler, self).__init__(hs)
+        self.hs = hs
+        self.store = hs.get_datastore()
+        self.state = hs.get_state_handler()
+        self.server_name = hs.hostname
+        self.clock = hs.get_clock()
+        self.notifier = hs.get_notifier()
+        self.is_mine_id = hs.is_mine_id
+        self.stats_bucket_size = hs.config.stats_bucket_size
+
+        # The current position in the current_state_delta stream
+        self.pos = None
+
+        # Guard to ensure we only process deltas one at a time
+        self._is_processing = False
+
+        if hs.config.stats_enabled:
+            self.notifier.add_replication_callback(self.notify_new_event)
+
+            # We kick this off so that we don't have to wait for a change before
+            # we start populating stats
+            self.clock.call_later(0, self.notify_new_event)
+
+    def notify_new_event(self):
+        """Called when there may be more deltas to process
+        """
+        if not self.hs.config.stats_enabled:
+            return
+
+        if self._is_processing:
+            return
+
+        @defer.inlineCallbacks
+        def process():
+            try:
+                yield self._unsafe_process()
+            finally:
+                self._is_processing = False
+
+        self._is_processing = True
+        run_as_background_process("stats.notify_new_event", process)
+
+    @defer.inlineCallbacks
+    def _unsafe_process(self):
+        # If self.pos is None then means we haven't fetched it from DB
+        if self.pos is None:
+            self.pos = yield self.store.get_stats_stream_pos()
+
+        # If still None then the initial background update hasn't happened yet
+        if self.pos is None:
+            defer.returnValue(None)
+
+        # Loop round handling deltas until we're up to date
+        while True:
+            with Measure(self.clock, "stats_delta"):
+                deltas = yield self.store.get_current_state_deltas(self.pos)
+                if not deltas:
+                    return
+
+                logger.info("Handling %d state deltas", len(deltas))
+                yield self._handle_deltas(deltas)
+
+                self.pos = deltas[-1]["stream_id"]
+                yield self.store.update_stats_stream_pos(self.pos)
+
+                event_processing_positions.labels("stats").set(self.pos)
+
+    @defer.inlineCallbacks
+    def _handle_deltas(self, deltas):
+        """
+        Called with the state deltas to process
+        """
+        for delta in deltas:
+            typ = delta["type"]
+            state_key = delta["state_key"]
+            room_id = delta["room_id"]
+            event_id = delta["event_id"]
+            stream_id = delta["stream_id"]
+            prev_event_id = delta["prev_event_id"]
+
+            logger.debug("Handling: %r %r, %s", typ, state_key, event_id)
+
+            token = yield self.store.get_earliest_token_for_room_stats(room_id)
+
+            # If the earliest token to begin from is larger than our current
+            # stream ID, skip processing this delta.
+            if token is not None and token >= stream_id:
+                logger.debug(
+                    "Ignoring: %s as earlier than this room's initial ingestion event",
+                    event_id,
+                )
+                continue
+
+            if event_id is None and prev_event_id is None:
+                # Errr...
+                continue
+
+            event_content = {}
+
+            if event_id is not None:
+                event_content = (yield self.store.get_event(event_id)).content or {}
+
+            # quantise time to the nearest bucket
+            now = yield self.store.get_received_ts(event_id)
+            now = (now // 1000 // self.stats_bucket_size) * self.stats_bucket_size
+
+            if typ == EventTypes.Member:
+                # we could use _get_key_change here but it's a bit inefficient
+                # given we're not testing for a specific result; might as well
+                # just grab the prev_membership and membership strings and
+                # compare them.
+                prev_event_content = {}
+                if prev_event_id is not None:
+                    prev_event_content = (
+                        yield self.store.get_event(prev_event_id)
+                    ).content
+
+                membership = event_content.get("membership", Membership.LEAVE)
+                prev_membership = prev_event_content.get("membership", Membership.LEAVE)
+
+                if prev_membership == membership:
+                    continue
+
+                if prev_membership == Membership.JOIN:
+                    yield self.store.update_stats_delta(
+                        now, "room", room_id, "joined_members", -1
+                    )
+                elif prev_membership == Membership.INVITE:
+                    yield self.store.update_stats_delta(
+                        now, "room", room_id, "invited_members", -1
+                    )
+                elif prev_membership == Membership.LEAVE:
+                    yield self.store.update_stats_delta(
+                        now, "room", room_id, "left_members", -1
+                    )
+                elif prev_membership == Membership.BAN:
+                    yield self.store.update_stats_delta(
+                        now, "room", room_id, "banned_members", -1
+                    )
+                else:
+                    err = "%s is not a valid prev_membership" % (repr(prev_membership),)
+                    logger.error(err)
+                    raise ValueError(err)
+
+                if membership == Membership.JOIN:
+                    yield self.store.update_stats_delta(
+                        now, "room", room_id, "joined_members", +1
+                    )
+                elif membership == Membership.INVITE:
+                    yield self.store.update_stats_delta(
+                        now, "room", room_id, "invited_members", +1
+                    )
+                elif membership == Membership.LEAVE:
+                    yield self.store.update_stats_delta(
+                        now, "room", room_id, "left_members", +1
+                    )
+                elif membership == Membership.BAN:
+                    yield self.store.update_stats_delta(
+                        now, "room", room_id, "banned_members", +1
+                    )
+                else:
+                    err = "%s is not a valid membership" % (repr(membership),)
+                    logger.error(err)
+                    raise ValueError(err)
+
+                user_id = state_key
+                if self.is_mine_id(user_id):
+                    # update user_stats as it's one of our users
+                    public = yield self._is_public_room(room_id)
+
+                    if membership == Membership.LEAVE:
+                        yield self.store.update_stats_delta(
+                            now,
+                            "user",
+                            user_id,
+                            "public_rooms" if public else "private_rooms",
+                            -1,
+                        )
+                    elif membership == Membership.JOIN:
+                        yield self.store.update_stats_delta(
+                            now,
+                            "user",
+                            user_id,
+                            "public_rooms" if public else "private_rooms",
+                            +1,
+                        )
+
+            elif typ == EventTypes.Create:
+                # Newly created room. Add it with all blank portions.
+                yield self.store.update_room_state(
+                    room_id,
+                    {
+                        "join_rules": None,
+                        "history_visibility": None,
+                        "encryption": None,
+                        "name": None,
+                        "topic": None,
+                        "avatar": None,
+                        "canonical_alias": None,
+                    },
+                )
+
+            elif typ == EventTypes.JoinRules:
+                yield self.store.update_room_state(
+                    room_id, {"join_rules": event_content.get("join_rule")}
+                )
+
+                is_public = yield self._get_key_change(
+                    prev_event_id, event_id, "join_rule", JoinRules.PUBLIC
+                )
+                if is_public is not None:
+                    yield self.update_public_room_stats(now, room_id, is_public)
+
+            elif typ == EventTypes.RoomHistoryVisibility:
+                yield self.store.update_room_state(
+                    room_id,
+                    {"history_visibility": event_content.get("history_visibility")},
+                )
+
+                is_public = yield self._get_key_change(
+                    prev_event_id, event_id, "history_visibility", "world_readable"
+                )
+                if is_public is not None:
+                    yield self.update_public_room_stats(now, room_id, is_public)
+
+            elif typ == EventTypes.Encryption:
+                yield self.store.update_room_state(
+                    room_id, {"encryption": event_content.get("algorithm")}
+                )
+            elif typ == EventTypes.Name:
+                yield self.store.update_room_state(
+                    room_id, {"name": event_content.get("name")}
+                )
+            elif typ == EventTypes.Topic:
+                yield self.store.update_room_state(
+                    room_id, {"topic": event_content.get("topic")}
+                )
+            elif typ == EventTypes.RoomAvatar:
+                yield self.store.update_room_state(
+                    room_id, {"avatar": event_content.get("url")}
+                )
+            elif typ == EventTypes.CanonicalAlias:
+                yield self.store.update_room_state(
+                    room_id, {"canonical_alias": event_content.get("alias")}
+                )
+
+    @defer.inlineCallbacks
+    def update_public_room_stats(self, ts, room_id, is_public):
+        """
+        Increment/decrement a user's number of public rooms when a room they are
+        in changes to/from public visibility.
+
+        Args:
+            ts (int): Timestamp in seconds
+            room_id (str)
+            is_public (bool)
+        """
+        # For now, blindly iterate over all local users in the room so that
+        # we can handle the whole problem of copying buckets over as needed
+        user_ids = yield self.store.get_users_in_room(room_id)
+
+        for user_id in user_ids:
+            if self.hs.is_mine(UserID.from_string(user_id)):
+                yield self.store.update_stats_delta(
+                    ts, "user", user_id, "public_rooms", +1 if is_public else -1
+                )
+                yield self.store.update_stats_delta(
+                    ts, "user", user_id, "private_rooms", -1 if is_public else +1
+                )
+
+    @defer.inlineCallbacks
+    def _is_public_room(self, room_id):
+        join_rules = yield self.state.get_current_state(room_id, EventTypes.JoinRules)
+        history_visibility = yield self.state.get_current_state(
+            room_id, EventTypes.RoomHistoryVisibility
+        )
+
+        if (join_rules and join_rules.content.get("join_rule") == JoinRules.PUBLIC) or (
+            (
+                history_visibility
+                and history_visibility.content.get("history_visibility")
+                == "world_readable"
+            )
+        ):
+            defer.returnValue(True)
+        else:
+            defer.returnValue(False)
@@ -285,7 +285,24 @@ class MatrixFederationHttpClient(object):
            request (MatrixFederationRequest): details of request to be sent

            timeout (int|None): number of milliseconds to wait for the response headers
-                (including connecting to the server). 60s by default.
+                (including connecting to the server), *for each attempt*.
+                60s by default.
+
+            long_retries (bool): whether to use the long retry algorithm.
+
+                The regular retry algorithm makes 4 attempts, with intervals
+                [0.5s, 1s, 2s].
+
+                The long retry algorithm makes 11 attempts, with intervals
+                [4s, 16s, 60s, 60s, ...]
+
+                Both algorithms add -20%/+40% jitter to the retry intervals.
+
+                Note that the above intervals are *in addition* to the time spent
+                waiting for the request to complete (up to `timeout` ms).
+
+                NB: the long retry algorithm takes over 20 minutes to complete, with
+                a default timeout of 60s!

            ignore_backoff (bool): true to ignore the historical backoff data
                and try the request anyway.
@@ -566,10 +583,14 @@ class MatrixFederationHttpClient(object):
                the request body. This will be encoded as JSON.
            json_data_callback (callable): A callable returning the dict to
                use as the request body.
-            long_retries (bool): A boolean that indicates whether we should
-                retry for a short or long time.
-            timeout(int): How long to try (in ms) the destination for before
-                giving up. None indicates no timeout.
+
+            long_retries (bool): whether to use the long retry algorithm. See
+                docs on _send_request for details.
+
+            timeout (int|None): number of milliseconds to wait for the response headers
+                (including connecting to the server), *for each attempt*.
+                self._default_timeout (60s) by default.
+
            ignore_backoff (bool): true to ignore the historical backoff data
                and try the request anyway.
            backoff_on_404 (bool): True if we should count a 404 response as
@@ -627,15 +648,22 @@ class MatrixFederationHttpClient(object):
        Args:
            destination (str): The remote server to send the HTTP request
                to.
+
            path (str): The HTTP path.
+
            data (dict): A dict containing the data that will be used as
                the request body. This will be encoded as JSON.
-            long_retries (bool): A boolean that indicates whether we should
-                retry for a short or long time.
-            timeout(int): How long to try (in ms) the destination for before
-                giving up. None indicates no timeout.
+
+            long_retries (bool): whether to use the long retry algorithm. See
+                docs on _send_request for details.
+
+            timeout (int|None): number of milliseconds to wait for the response headers
+                (including connecting to the server), *for each attempt*.
+                self._default_timeout (60s) by default.
+
            ignore_backoff (bool): true to ignore the historical backoff data and
                try the request anyway.
+
            args (dict): query params
        Returns:
            Deferred[dict|list]: Succeeds when we get a 2xx HTTP response. The
@@ -686,14 +714,19 @@ class MatrixFederationHttpClient(object):
        Args:
            destination (str): The remote server to send the HTTP request
                to.
+
            path (str): The HTTP path.
+
            args (dict|None): A dictionary used to create query strings, defaults to
                None.
-            timeout (int): How long to try (in ms) the destination for before
-                giving up. None indicates no timeout and that the request will
-                be retried.
+
+            timeout (int|None): number of milliseconds to wait for the response headers
+                (including connecting to the server), *for each attempt*.
+                self._default_timeout (60s) by default.
+
            ignore_backoff (bool): true to ignore the historical backoff data
                and try the request anyway.
+
            try_trailing_slash_on_400 (bool): True if on a 400 M_UNRECOGNIZED
                response we should try appending a trailing slash to the end of
                the request. Workaround for #3622 in Synapse <= v0.99.3.
@@ -711,10 +744,6 @@ class MatrixFederationHttpClient(object):
            RequestSendFailed: If there were problems connecting to the
                remote, due to e.g. DNS failures, connection timeouts etc.
        """
-        logger.debug("get_json args: %s", args)
-
-        logger.debug("Query bytes: %s Retry DNS: %s", args, retry_on_dns_fail)
-
        request = MatrixFederationRequest(
            method="GET",
            destination=destination,
@@ -746,12 +775,18 @@ class MatrixFederationHttpClient(object):
            destination (str): The remote server to send the HTTP request
                to.
            path (str): The HTTP path.
-            long_retries (bool): A boolean that indicates whether we should
-                retry for a short or long time.
-            timeout(int): How long to try (in ms) the destination for before
-                giving up. None indicates no timeout.
+
+            long_retries (bool): whether to use the long retry algorithm. See
+                docs on _send_request for details.
+
+            timeout (int|None): number of milliseconds to wait for the response headers
+                (including connecting to the server), *for each attempt*.
+                self._default_timeout (60s) by default.
+
            ignore_backoff (bool): true to ignore the historical backoff data and
                try the request anyway.
+
+            args (dict): query params
        Returns:
            Deferred[dict|list]: Succeeds when we get a 2xx HTTP response. The
            result will be the decoded JSON body.
@@ -55,7 +55,7 @@ def parse_integer_from_args(args, name, default=None, required=False):
            return int(args[name][0])
        except Exception:
            message = "Query parameter %r must be an integer" % (name,)
-            raise SynapseError(400, message)
+            raise SynapseError(400, message, errcode=Codes.INVALID_PARAM)
    else:
        if required:
            message = "Missing integer query parameter %r" % (name,)
@@ -80,10 +80,10 @@ ALLOWED_ATTRS = {


 class Mailer(object):
-    def __init__(self, hs, app_name, notif_template_html, notif_template_text):
+    def __init__(self, hs, app_name, template_html, template_text):
        self.hs = hs
-        self.notif_template_html = notif_template_html
-        self.notif_template_text = notif_template_text
+        self.template_html = template_html
+        self.template_text = template_text

        self.sendmail = self.hs.get_sendmail()
        self.store = self.hs.get_datastore()
@@ -93,22 +93,49 @@ class Mailer(object):

        logger.info("Created Mailer for app_name %s" % app_name)

+    @defer.inlineCallbacks
+    def send_password_reset_mail(
+        self,
+        email_address,
+        token,
+        client_secret,
+        sid,
+    ):
+        """Send an email with a password reset link to a user
+
+        Args:
+            email_address (str): Email address we're sending the password
+                reset to
+            token (str): Unique token generated by the server to verify
+                password reset email was received
+            client_secret (str): Unique token generated by the client to
+                group together multiple email sending attempts
+            sid (str): The generated session ID
+        """
+        if email.utils.parseaddr(email_address)[1] == '':
+            raise RuntimeError("Invalid 'to' email address")
+
+        link = (
+            self.hs.config.public_baseurl +
+            "_synapse/password_reset/email/submit_token"
+            "?token=%s&client_secret=%s&sid=%s" %
+            (token, client_secret, sid)
+        )
+
+        template_vars = {
+            "link": link,
+        }
+
+        yield self.send_email(
+            email_address,
+            "[%s] Password Reset Email" % self.hs.config.server_name,
+            template_vars,
+        )
+
    @defer.inlineCallbacks
    def send_notification_mail(self, app_id, user_id, email_address,
                               push_actions, reason):
-        try:
-            from_string = self.hs.config.email_notif_from % {
-                "app": self.app_name
-            }
-        except TypeError:
-            from_string = self.hs.config.email_notif_from
-
-        raw_from = email.utils.parseaddr(from_string)[1]
-        raw_to = email.utils.parseaddr(email_address)[1]
-
-        if raw_to == '':
-            raise RuntimeError("Invalid 'to' address")
-
+        """Send email regarding a user's room notifications"""
        rooms_in_order = deduped_ordered_list(
            [pa['room_id'] for pa in push_actions]
        )
@@ -176,14 +203,36 @@ class Mailer(object):
            "reason": reason,
        }

-        html_text = self.notif_template_html.render(**template_vars)
+        yield self.send_email(
+            email_address,
+            "[%s] %s" % (self.app_name, summary_text),
+            template_vars,
+        )
+
+    @defer.inlineCallbacks
+    def send_email(self, email_address, subject, template_vars):
+        """Send an email with the given information and template text"""
+        try:
+            from_string = self.hs.config.email_notif_from % {
+                "app": self.app_name
+            }
+        except TypeError:
+            from_string = self.hs.config.email_notif_from
+
+        raw_from = email.utils.parseaddr(from_string)[1]
+        raw_to = email.utils.parseaddr(email_address)[1]
+
+        if raw_to == '':
+            raise RuntimeError("Invalid 'to' address")
+
+        html_text = self.template_html.render(**template_vars)
        html_part = MIMEText(html_text, "html", "utf8")

-        plain_text = self.notif_template_text.render(**template_vars)
+        plain_text = self.template_text.render(**template_vars)
        text_part = MIMEText(plain_text, "plain", "utf8")

        multipart_msg = MIMEMultipart('alternative')
-        multipart_msg['Subject'] = "[%s] %s" % (self.app_name, summary_text)
+        multipart_msg['Subject'] = subject
        multipart_msg['From'] = from_string
        multipart_msg['To'] = email_address
        multipart_msg['Date'] = email.utils.formatdate()
@@ -70,8 +70,8 @@ class PusherFactory(object):
            mailer = Mailer(
                hs=self.hs,
                app_name=app_name,
-                notif_template_html=self.notif_template_html,
-                notif_template_text=self.notif_template_text,
+                template_html=self.notif_template_html,
+                template_text=self.notif_template_text,
            )
            self.mailers[app_name] = mailer
        return EmailPusher(self.hs, pusherdict, mailer)
@@ -74,18 +74,10 @@ REQUIREMENTS = [
    "attrs>=17.4.0",

    "netaddr>=0.7.18",
-
-    # requests is a transitive dep of treq, and urlib3 is a transitive dep
-    # of requests, as well as of sentry-sdk.
-    #
-    # As of requests 2.21, requests does not yet support urllib3 1.25.
-    # (If we do not pin it here, pip will give us the latest urllib3
-    # due to the dep via sentry-sdk.)
-    "urllib3<1.25",
 ]

 CONDITIONAL_REQUIREMENTS = {
-    "email.enable_notifs": ["Jinja2>=2.9", "bleach>=1.4.2"],
+    "email": ["Jinja2>=2.9", "bleach>=1.4.2"],
    "matrix-synapse-ldap3": ["matrix-synapse-ldap3>=0.1"],

    # we use execute_batch, which arrived in psycopg 2.7.
@@ -0,0 +1,9 @@
+<html>
+<body>
+    <p>A password reset request has been received for your Matrix account. If this was you, please click the link below to confirm resetting your password:</p>
+
+    <a href="{{ link }}">{{ link }}</a>
+
+    <p>If this was not you, please disregard this email and contact your server administrator. Thank you.</p>
+</body>
+</html>
@@ -0,0 +1,7 @@
+A password reset request has been received for your Matrix account. If this
+was you, please click the link below to confirm resetting your password:
+
+{{ link }}
+
+If this was not you, please disregard this email and contact your server
+administrator. Thank you.
@@ -0,0 +1,6 @@
+<html>
+<head></head>
+<body>
+<p>{{ failure_reason }}. Your password has not been reset.</p>
+</body>
+</html>
@@ -0,0 +1,6 @@
+<html>
+<head></head>
+<body>
+<p>Your password was successfully reset. You may now close this window.</p>
+</body>
+</html>
--- a/Show More
+++ b/Show More
				`@@ -0,0 +1 @@`
				`Synapse now more efficiently collates room statistics.`
				`@@ -0,0 +1 @@`
				`Fix worker registration bug caused by ClientReaderSlavedStore being unable to see get_profileinfo.`
				`@@ -0,0 +1 @@`
				`Synapse will now serve the experimental "room complexity" API endpoint.`
				`@@ -0,0 +1 @@`
				`Add experimental support for relations (aka reactions and edits).`
				`@@ -0,0 +1 @@`
				`The base classes for the v1 and v2_alpha REST APIs have been unified.`
				`@@ -0,0 +1 @@`
				`Remove urllib3 pin as requests 2.22.0 has been released supporting urllib3 1.25.2.`
				`@@ -0,0 +1 @@`
				`Rewrite store_server_verify_key to store several keys at once.`
				`@@ -0,0 +1 @@`
				`Remove unused VerifyKey.expired and .time_added fields.`
				`@@ -0,0 +1 @@`
				`Store key validity time in the storage layer.`
				`@@ -0,0 +1 @@`
				`Refactor synapse.crypto.keyring to use a KeyFetcher interface.`