Merge branch 'release-v1.59' into matrix-org-hotfixes

21 files changed, 361 insertions, 127 deletions

diff --git a/synapse/handlers/account_validity.py b/synapse/handlers/account_validity.py
index 05a138410e..33e45e3a11 100644
--- a/synapse/handlers/account_validity.py
+++ b/synapse/handlers/account_validity.py
@@ -23,6 +23,7 @@ from synapse.api.errors import AuthError, StoreError, SynapseError
 from synapse.metrics.background_process_metrics import wrap_as_background_process
 from synapse.types import UserID
 from synapse.util import stringutils
+from synapse.util.async_helpers import delay_cancellation
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -150,7 +151,7 @@ class AccountValidityHandler:
             Whether the user has expired.
         """
         for callback in self._is_user_expired_callbacks:
-            expired = await callback(user_id)
+            expired = await delay_cancellation(callback(user_id))
             if expired is not None:
                 return expired
 
diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py
index 1b57840506..85bd5e4768 100644
--- a/synapse/handlers/appservice.py
+++ b/synapse/handlers/appservice.py
@@ -59,7 +59,7 @@ class ApplicationServicesHandler:
         self.scheduler = hs.get_application_service_scheduler()
         self.started_scheduler = False
         self.clock = hs.get_clock()
-        self.notify_appservices = hs.config.appservice.notify_appservices
+        self.notify_appservices = hs.config.worker.should_notify_appservices
         self.event_sources = hs.get_event_sources()
         self._msc2409_to_device_messages_enabled = (
             hs.config.experimental.msc2409_to_device_messages_enabled
@@ -416,7 +416,7 @@ class ApplicationServicesHandler:
         return typing
 
     async def _handle_receipts(
-        self, service: ApplicationService, new_token: Optional[int]
+        self, service: ApplicationService, new_token: int
     ) -> List[JsonDict]:
         """
         Return the latest read receipts that the given application service should receive.
@@ -447,7 +447,7 @@ class ApplicationServicesHandler:
 
         receipts_source = self.event_sources.sources.receipt
         receipts, _ = await receipts_source.get_new_events_as(
-            service=service, from_key=from_key
+            service=service, from_key=from_key, to_key=new_token
         )
         return receipts
 
diff --git a/synapse/handlers/auth.py b/synapse/handlers/auth.py
index 86991d26ce..1b9050ea96 100644
--- a/synapse/handlers/auth.py
+++ b/synapse/handlers/auth.py
@@ -41,6 +41,7 @@ import pymacaroons
 import unpaddedbase64
 from pymacaroons.exceptions import MacaroonVerificationFailedException
 
+from twisted.internet.defer import CancelledError
 from twisted.web.server import Request
 
 from synapse.api.constants import LoginType
@@ -67,7 +68,7 @@ from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.roommember import ProfileInfo
 from synapse.types import JsonDict, Requester, UserID
 from synapse.util import stringutils as stringutils
-from synapse.util.async_helpers import maybe_awaitable
+from synapse.util.async_helpers import delay_cancellation, maybe_awaitable
 from synapse.util.macaroons import get_value_from_macaroon, satisfy_expiry
 from synapse.util.msisdn import phone_number_to_msisdn
 from synapse.util.stringutils import base62_encode
@@ -481,7 +482,7 @@ class AuthHandler:
             sid = authdict["session"]
 
         # Convert the URI and method to strings.
-        uri = request.uri.decode("utf-8")  # type: ignore
+        uri = request.uri.decode("utf-8")
         method = request.method.decode("utf-8")
 
         # If there's no session ID, create a new session.
@@ -551,7 +552,7 @@ class AuthHandler:
             await self.store.set_ui_auth_clientdict(sid, clientdict)
 
         user_agent = get_request_user_agent(request)
-        clientip = request.getClientIP()
+        clientip = request.getClientAddress().host
 
         await self.store.add_user_agent_ip_to_ui_auth_session(
             session.session_id, user_agent, clientip
@@ -2202,7 +2203,11 @@ class PasswordAuthProvider:
         # other than None (i.e. until a callback returns a success)
         for callback in self.auth_checker_callbacks[login_type]:
             try:
-                result = await callback(username, login_type, login_dict)
+                result = await delay_cancellation(
+                    callback(username, login_type, login_dict)
+                )
+            except CancelledError:
+                raise
             except Exception as e:
                 logger.warning("Failed to run module API callback %s: %s", callback, e)
                 continue
@@ -2263,7 +2268,9 @@ class PasswordAuthProvider:
 
         for callback in self.check_3pid_auth_callbacks:
             try:
-                result = await callback(medium, address, password)
+                result = await delay_cancellation(callback(medium, address, password))
+            except CancelledError:
+                raise
             except Exception as e:
                 logger.warning("Failed to run module API callback %s: %s", callback, e)
                 continue
@@ -2345,7 +2352,7 @@ class PasswordAuthProvider:
         """
         for callback in self.get_username_for_registration_callbacks:
             try:
-                res = await callback(uia_results, params)
+                res = await delay_cancellation(callback(uia_results, params))
 
                 if isinstance(res, str):
                     return res
@@ -2359,6 +2366,8 @@ class PasswordAuthProvider:
                         callback,
                         res,
                     )
+            except CancelledError:
+                raise
             except Exception as e:
                 logger.error(
                     "Module raised an exception in get_username_for_registration: %s",
@@ -2388,7 +2397,7 @@ class PasswordAuthProvider:
         """
         for callback in self.get_displayname_for_registration_callbacks:
             try:
-                res = await callback(uia_results, params)
+                res = await delay_cancellation(callback(uia_results, params))
 
                 if isinstance(res, str):
                     return res
@@ -2402,6 +2411,8 @@ class PasswordAuthProvider:
                         callback,
                         res,
                     )
+            except CancelledError:
+                raise
             except Exception as e:
                 logger.error(
                     "Module raised an exception in get_displayname_for_registration: %s",
@@ -2429,7 +2440,7 @@ class PasswordAuthProvider:
         """
         for callback in self.is_3pid_allowed_callbacks:
             try:
-                res = await callback(medium, address, registration)
+                res = await delay_cancellation(callback(medium, address, registration))
 
                 if res is False:
                     return res
@@ -2443,6 +2454,8 @@ class PasswordAuthProvider:
                         callback,
                         res,
                     )
+            except CancelledError:
+                raise
             except Exception as e:
                 logger.error("Module raised an exception in is_3pid_allowed: %s", e)
                 raise SynapseError(code=500, msg="Internal Server Error")
diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py
index 5b94b00bc3..82a5aac3dd 100644
--- a/synapse/handlers/events.py
+++ b/synapse/handlers/events.py
@@ -164,7 +164,7 @@ class EventHandler:
             event.
         """
         redact_behaviour = (
-            EventRedactBehaviour.AS_IS if show_redacted else EventRedactBehaviour.REDACT
+            EventRedactBehaviour.as_is if show_redacted else EventRedactBehaviour.redact
         )
         event = await self.store.get_event(
             event_id, check_room_id=room_id, redact_behaviour=redact_behaviour
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index d2ba70a814..38dc5b1f6e 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -316,7 +316,7 @@ class FederationHandler:
 
             events_to_check = await self.store.get_events_as_list(
                 event_ids_to_check,
-                redact_behaviour=EventRedactBehaviour.AS_IS,
+                redact_behaviour=EventRedactBehaviour.as_is,
                 get_prev_content=False,
             )
 
@@ -1494,7 +1494,7 @@ class FederationHandler:
 
             events = await self.store.get_events_as_list(
                 batch,
-                redact_behaviour=EventRedactBehaviour.AS_IS,
+                redact_behaviour=EventRedactBehaviour.as_is,
                 allow_rejected=True,
             )
             for event in events:
diff --git a/synapse/handlers/federation_event.py b/synapse/handlers/federation_event.py
index 693b544286..6cf927e4ff 100644
--- a/synapse/handlers/federation_event.py
+++ b/synapse/handlers/federation_event.py
@@ -860,7 +860,7 @@ class FederationEventHandler:
             evs = await self._store.get_events(
                 list(state_map.values()),
                 get_prev_content=False,
-                redact_behaviour=EventRedactBehaviour.AS_IS,
+                redact_behaviour=EventRedactBehaviour.as_is,
             )
             event_map.update(evs)
 
diff --git a/synapse/handlers/identity.py b/synapse/handlers/identity.py
index c183e9c465..9bca2bc4b2 100644
--- a/synapse/handlers/identity.py
+++ b/synapse/handlers/identity.py
@@ -92,7 +92,7 @@ class IdentityHandler:
         """
 
         await self._3pid_validation_ratelimiter_ip.ratelimit(
-            None, (medium, request.getClientIP())
+            None, (medium, request.getClientAddress().host)
         )
         await self._3pid_validation_ratelimiter_address.ratelimit(
             None, (medium, address)
diff --git a/synapse/handlers/initial_sync.py b/synapse/handlers/initial_sync.py
index a7db8feb57..7b94770f97 100644
--- a/synapse/handlers/initial_sync.py
+++ b/synapse/handlers/initial_sync.py
@@ -143,7 +143,7 @@ class InitialSyncHandler:
             to_key=int(now_token.receipt_key),
         )
         if self.hs.config.experimental.msc2285_enabled:
-            receipt = ReceiptEventSource.filter_out_hidden(receipt, user_id)
+            receipt = ReceiptEventSource.filter_out_private(receipt, user_id)
 
         tags_by_room = await self.store.get_tags_for_user(user_id)
 
@@ -449,7 +449,7 @@ class InitialSyncHandler:
             if not receipts:
                 return []
             if self.hs.config.experimental.msc2285_enabled:
-                receipts = ReceiptEventSource.filter_out_hidden(receipts, user_id)
+                receipts = ReceiptEventSource.filter_out_private(receipts, user_id)
             return receipts
 
         presence, receipts, (messages, token) = await make_deferred_yieldable(
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 549b776d6b..e1082954cc 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -1407,7 +1407,7 @@ class EventCreationHandler:
 
                 original_event = await self.store.get_event(
                     event.redacts,
-                    redact_behaviour=EventRedactBehaviour.AS_IS,
+                    redact_behaviour=EventRedactBehaviour.as_is,
                     get_prev_content=False,
                     allow_rejected=False,
                     allow_none=True,
@@ -1427,7 +1427,7 @@ class EventCreationHandler:
             # Validate a newly added alias or newly added alt_aliases.
 
             original_alias = None
-            original_alt_aliases: List[str] = []
+            original_alt_aliases: object = []
 
             original_event_id = event.unsigned.get("replaces_state")
             if original_event_id:
@@ -1455,6 +1455,7 @@ class EventCreationHandler:
             # If the old version of alt_aliases is of an unknown form,
             # completely replace it.
             if not isinstance(original_alt_aliases, (list, tuple)):
+                # TODO: check that the original_alt_aliases' entries are all strings
                 original_alt_aliases = []
 
             # Check that each alias is currently valid.
@@ -1504,7 +1505,7 @@ class EventCreationHandler:
 
             original_event = await self.store.get_event(
                 event.redacts,
-                redact_behaviour=EventRedactBehaviour.AS_IS,
+                redact_behaviour=EventRedactBehaviour.as_is,
                 get_prev_content=False,
                 allow_rejected=False,
                 allow_none=True,
diff --git a/synapse/handlers/oidc.py b/synapse/handlers/oidc.py
index 724b9cfcb4..f6ffb7d18d 100644
--- a/synapse/handlers/oidc.py
+++ b/synapse/handlers/oidc.py
@@ -966,7 +966,7 @@ class OidcProvider:
                         "Mapping provider does not support de-duplicating Matrix IDs"
                     )
 
-                attributes = await self._user_mapping_provider.map_user_attributes(  # type: ignore
+                attributes = await self._user_mapping_provider.map_user_attributes(
                     userinfo, token
                 )
 
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index d078162c29..268481ec19 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -659,27 +659,28 @@ class PresenceHandler(BasePresenceHandler):
         )
 
         now = self.clock.time_msec()
-        for state in self.user_to_current_state.values():
-            self.wheel_timer.insert(
-                now=now, obj=state.user_id, then=state.last_active_ts + IDLE_TIMER
-            )
-            self.wheel_timer.insert(
-                now=now,
-                obj=state.user_id,
-                then=state.last_user_sync_ts + SYNC_ONLINE_TIMEOUT,
-            )
-            if self.is_mine_id(state.user_id):
+        if self._presence_enabled:
+            for state in self.user_to_current_state.values():
                 self.wheel_timer.insert(
-                    now=now,
-                    obj=state.user_id,
-                    then=state.last_federation_update_ts + FEDERATION_PING_INTERVAL,
+                    now=now, obj=state.user_id, then=state.last_active_ts + IDLE_TIMER
                 )
-            else:
                 self.wheel_timer.insert(
                     now=now,
                     obj=state.user_id,
-                    then=state.last_federation_update_ts + FEDERATION_TIMEOUT,
+                    then=state.last_user_sync_ts + SYNC_ONLINE_TIMEOUT,
                 )
+                if self.is_mine_id(state.user_id):
+                    self.wheel_timer.insert(
+                        now=now,
+                        obj=state.user_id,
+                        then=state.last_federation_update_ts + FEDERATION_PING_INTERVAL,
+                    )
+                else:
+                    self.wheel_timer.insert(
+                        now=now,
+                        obj=state.user_id,
+                        then=state.last_federation_update_ts + FEDERATION_TIMEOUT,
+                    )
 
         # Set of users who have presence in the `user_to_current_state` that
         # have not yet been persisted
@@ -804,6 +805,13 @@ class PresenceHandler(BasePresenceHandler):
                 This is currently used to bump the max presence stream ID without changing any
                 user's presence (see PresenceHandler.add_users_to_send_full_presence_to).
         """
+        if not self._presence_enabled:
+            # We shouldn't get here if presence is disabled, but we check anyway
+            # to ensure that we don't a) send out presence federation and b)
+            # don't add things to the wheel timer that will never be handled.
+            logger.warning("Tried to update presence states when presence is disabled")
+            return
+
         now = self.clock.time_msec()
 
         with Measure(self.clock, "presence_update_states"):
@@ -1229,6 +1237,10 @@ class PresenceHandler(BasePresenceHandler):
         ):
             raise SynapseError(400, "Invalid presence state")
 
+        # If presence is disabled, no-op
+        if not self.hs.config.server.use_presence:
+            return
+
         user_id = target_user.to_string()
 
         prev_state = await self.current_state_for_user(user_id)
diff --git a/synapse/handlers/push_rules.py b/synapse/handlers/push_rules.py
new file mode 100644
index 0000000000..2599160bcc
--- /dev/null
+++ b/synapse/handlers/push_rules.py
@@ -0,0 +1,138 @@
+# Copyright 2022 The Matrix.org Foundation C.I.C.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from typing import TYPE_CHECKING, List, Optional, Union
+
+import attr
+
+from synapse.api.errors import SynapseError, UnrecognizedRequestError
+from synapse.push.baserules import BASE_RULE_IDS
+from synapse.storage.push_rule import RuleNotFoundException
+from synapse.types import JsonDict
+
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class RuleSpec:
+    scope: str
+    template: str
+    rule_id: str
+    attr: Optional[str]
+
+
+class PushRulesHandler:
+    """A class to handle changes in push rules for users."""
+
+    def __init__(self, hs: "HomeServer"):
+        self._notifier = hs.get_notifier()
+        self._main_store = hs.get_datastores().main
+
+    async def set_rule_attr(
+        self, user_id: str, spec: RuleSpec, val: Union[bool, JsonDict]
+    ) -> None:
+        """Set an attribute (enabled or actions) on an existing push rule.
+
+        Notifies listeners (e.g. sync handler) of the change.
+
+        Args:
+            user_id: the user for which to modify the push rule.
+            spec: the spec of the push rule to modify.
+            val: the value to change the attribute to.
+
+        Raises:
+            RuleNotFoundException if the rule being modified doesn't exist.
+            SynapseError(400) if the value is malformed.
+            UnrecognizedRequestError if the attribute to change is unknown.
+            InvalidRuleException if we're trying to change the actions on a rule but
+                the provided actions aren't compliant with the spec.
+        """
+        if spec.attr not in ("enabled", "actions"):
+            # for the sake of potential future expansion, shouldn't report
+            # 404 in the case of an unknown request so check it corresponds to
+            # a known attribute first.
+            raise UnrecognizedRequestError()
+
+        namespaced_rule_id = f"global/{spec.template}/{spec.rule_id}"
+        rule_id = spec.rule_id
+        is_default_rule = rule_id.startswith(".")
+        if is_default_rule:
+            if namespaced_rule_id not in BASE_RULE_IDS:
+                raise RuleNotFoundException("Unknown rule %r" % (namespaced_rule_id,))
+        if spec.attr == "enabled":
+            if isinstance(val, dict) and "enabled" in val:
+                val = val["enabled"]
+            if not isinstance(val, bool):
+                # Legacy fallback
+                # This should *actually* take a dict, but many clients pass
+                # bools directly, so let's not break them.
+                raise SynapseError(400, "Value for 'enabled' must be boolean")
+            await self._main_store.set_push_rule_enabled(
+                user_id, namespaced_rule_id, val, is_default_rule
+            )
+        elif spec.attr == "actions":
+            if not isinstance(val, dict):
+                raise SynapseError(400, "Value must be a dict")
+            actions = val.get("actions")
+            if not isinstance(actions, list):
+                raise SynapseError(400, "Value for 'actions' must be dict")
+            check_actions(actions)
+            rule_id = spec.rule_id
+            is_default_rule = rule_id.startswith(".")
+            if is_default_rule:
+                if namespaced_rule_id not in BASE_RULE_IDS:
+                    raise RuleNotFoundException(
+                        "Unknown rule %r" % (namespaced_rule_id,)
+                    )
+            await self._main_store.set_push_rule_actions(
+                user_id, namespaced_rule_id, actions, is_default_rule
+            )
+        else:
+            raise UnrecognizedRequestError()
+
+        self.notify_user(user_id)
+
+    def notify_user(self, user_id: str) -> None:
+        """Notify listeners about a push rule change.
+
+        Args:
+            user_id: the user ID the change is for.
+        """
+        stream_id = self._main_store.get_max_push_rules_stream_id()
+        self._notifier.on_new_event("push_rules_key", stream_id, users=[user_id])
+
+
+def check_actions(actions: List[Union[str, JsonDict]]) -> None:
+    """Check if the given actions are spec compliant.
+
+    Args:
+        actions: the actions to check.
+
+    Raises:
+        InvalidRuleException if the rules aren't compliant with the spec.
+    """
+    if not isinstance(actions, list):
+        raise InvalidRuleException("No actions found")
+
+    for a in actions:
+        if a in ["notify", "dont_notify", "coalesce"]:
+            pass
+        elif isinstance(a, dict) and "set_tweak" in a:
+            pass
+        else:
+            raise InvalidRuleException("Unrecognised action %s" % a)
+
+
+class InvalidRuleException(Exception):
+    pass
diff --git a/synapse/handlers/receipts.py b/synapse/handlers/receipts.py
index 6250bb3bdf..43d615357b 100644
--- a/synapse/handlers/receipts.py
+++ b/synapse/handlers/receipts.py
@@ -14,7 +14,7 @@
 import logging
 from typing import TYPE_CHECKING, Iterable, List, Optional, Tuple
 
-from synapse.api.constants import ReadReceiptEventFields, ReceiptTypes
+from synapse.api.constants import ReceiptTypes
 from synapse.appservice import ApplicationService
 from synapse.streams import EventSource
 from synapse.types import JsonDict, ReadReceipt, UserID, get_domain_from_id
@@ -112,7 +112,7 @@ class ReceiptsHandler:
             )
 
             if not res:
-                # res will be None if this read receipt is 'old'
+                # res will be None if this receipt is 'old'
                 continue
 
             stream_id, max_persisted_id = res
@@ -138,7 +138,7 @@ class ReceiptsHandler:
         return True
 
     async def received_client_receipt(
-        self, room_id: str, receipt_type: str, user_id: str, event_id: str, hidden: bool
+        self, room_id: str, receipt_type: str, user_id: str, event_id: str
     ) -> None:
         """Called when a client tells us a local user has read up to the given
         event_id in the room.
@@ -148,16 +148,14 @@ class ReceiptsHandler:
             receipt_type=receipt_type,
             user_id=user_id,
             event_ids=[event_id],
-            data={"ts": int(self.clock.time_msec()), "hidden": hidden},
+            data={"ts": int(self.clock.time_msec())},
         )
 
         is_new = await self._handle_new_receipts([receipt])
         if not is_new:
             return
 
-        if self.federation_sender and not (
-            self.hs.config.experimental.msc2285_enabled and hidden
-        ):
+        if self.federation_sender and receipt_type != ReceiptTypes.READ_PRIVATE:
             await self.federation_sender.send_read_receipt(receipt)
 
 
@@ -167,46 +165,37 @@ class ReceiptEventSource(EventSource[int, JsonDict]):
         self.config = hs.config
 
     @staticmethod
-    def filter_out_hidden(events: List[JsonDict], user_id: str) -> List[JsonDict]:
+    def filter_out_private(events: List[JsonDict], user_id: str) -> List[JsonDict]:
+        """
+        This method takes in what is returned by
+        get_linearized_receipts_for_rooms() and goes through read receipts
+        filtering out m.read.private receipts if they were not sent by the
+        current user.
+        """
+
         visible_events = []
 
-        # filter out hidden receipts the user shouldn't see
+        # filter out private receipts the user shouldn't see
         for event in events:
             content = event.get("content", {})
             new_event = event.copy()
             new_event["content"] = {}
 
-            for event_id in content.keys():
-                event_content = content.get(event_id, {})
-                m_read = event_content.get(ReceiptTypes.READ, {})
-
-                # If m_read is missing copy over the original event_content as there is nothing to process here
-                if not m_read:
-                    new_event["content"][event_id] = event_content.copy()
-                    continue
-
-                new_users = {}
-                for rr_user_id, user_rr in m_read.items():
-                    try:
-                        hidden = user_rr.get("hidden")
-                    except AttributeError:
-                        # Due to https://github.com/matrix-org/synapse/issues/10376
-                        # there are cases where user_rr is a string, in those cases
-                        # we just ignore the read receipt
-                        continue
-
-                    if hidden is not True or rr_user_id == user_id:
-                        new_users[rr_user_id] = user_rr.copy()
-                        # If hidden has a value replace hidden with the correct prefixed key
-                        if hidden is not None:
-                            new_users[rr_user_id].pop("hidden")
-                            new_users[rr_user_id][
-                                ReadReceiptEventFields.MSC2285_HIDDEN
-                            ] = hidden
-
-                # Set new users unless empty
-                if len(new_users.keys()) > 0:
-                    new_event["content"][event_id] = {ReceiptTypes.READ: new_users}
+            for event_id, event_content in content.items():
+                receipt_event = {}
+                for receipt_type, receipt_content in event_content.items():
+                    if receipt_type == ReceiptTypes.READ_PRIVATE:
+                        user_rr = receipt_content.get(user_id, None)
+                        if user_rr:
+                            receipt_event[ReceiptTypes.READ_PRIVATE] = {
+                                user_id: user_rr.copy()
+                            }
+                    else:
+                        receipt_event[receipt_type] = receipt_content.copy()
+
+                # Only include the receipt event if it is non-empty.
+                if receipt_event:
+                    new_event["content"][event_id] = receipt_event
 
             # Append new_event to visible_events unless empty
             if len(new_event["content"].keys()) > 0:
@@ -234,18 +223,19 @@ class ReceiptEventSource(EventSource[int, JsonDict]):
         )
 
         if self.config.experimental.msc2285_enabled:
-            events = ReceiptEventSource.filter_out_hidden(events, user.to_string())
+            events = ReceiptEventSource.filter_out_private(events, user.to_string())
 
         return events, to_key
 
     async def get_new_events_as(
-        self, from_key: int, service: ApplicationService
+        self, from_key: int, to_key: int, service: ApplicationService
     ) -> Tuple[List[JsonDict], int]:
         """Returns a set of new read receipt events that an appservice
         may be interested in.
 
         Args:
             from_key: the stream position at which events should be fetched from
+            to_key: the stream position up to which events should be fetched to
             service: The appservice which may be interested
 
         Returns:
@@ -255,7 +245,6 @@ class ReceiptEventSource(EventSource[int, JsonDict]):
                 * The current read receipt stream token.
         """
         from_key = int(from_key)
-        to_key = self.get_current_key()
 
         if from_key == to_key:
             return [], to_key
diff --git a/synapse/handlers/relations.py b/synapse/handlers/relations.py
index 5efb561273..c2754ec918 100644
--- a/synapse/handlers/relations.py
+++ b/synapse/handlers/relations.py
@@ -11,6 +11,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.
+import collections.abc
 import logging
 from typing import (
     TYPE_CHECKING,
@@ -24,7 +25,6 @@ from typing import (
 )
 
 import attr
-from frozendict import frozendict
 
 from synapse.api.constants import RelationTypes
 from synapse.api.errors import SynapseError
@@ -44,8 +44,6 @@ logger = logging.getLogger(__name__)
 class _ThreadAggregation:
     # The latest event in the thread.
     latest_event: EventBase
-    # The latest edit to the latest event in the thread.
-    latest_edit: Optional[EventBase]
     # The total number of events in the thread.
     count: int
     # True if the current user has sent an event to the thread.
@@ -295,7 +293,7 @@ class RelationsHandler:
 
         for event_id, summary in summaries.items():
             if summary:
-                thread_count, latest_thread_event, edit = summary
+                thread_count, latest_thread_event = summary
 
                 # Subtract off the count of any ignored users.
                 for ignored_user in ignored_users:
@@ -340,7 +338,6 @@ class RelationsHandler:
 
                 results[event_id] = _ThreadAggregation(
                     latest_event=latest_thread_event,
-                    latest_edit=edit,
                     count=thread_count,
                     # If there's a thread summary it must also exist in the
                     # participated dictionary.
@@ -359,15 +356,37 @@ class RelationsHandler:
             user_id: The user requesting the bundled aggregations.
 
         Returns:
-            A map of event ID to the bundled aggregation for the event. Not all
-            events may have bundled aggregations in the results.
+            A map of event ID to the bundled aggregations for the event.
+
+            Not all requested events may exist in the results (if they don't have
+            bundled aggregations).
+
+            The results may include additional events which are related to the
+            requested events.
         """
-        # De-duplicate events by ID to handle the same event requested multiple times.
-        #
-        # State events do not get bundled aggregations.
-        events_by_id = {
-            event.event_id: event for event in events if not event.is_state()
-        }
+        # De-duplicated events by ID to handle the same event requested multiple times.
+        events_by_id = {}
+        # A map of event ID to the relation in that event, if there is one.
+        relations_by_id: Dict[str, str] = {}
+        for event in events:
+            # State events do not get bundled aggregations.
+            if event.is_state():
+                continue
+
+            relates_to = event.content.get("m.relates_to")
+            relation_type = None
+            if isinstance(relates_to, collections.abc.Mapping):
+                relation_type = relates_to.get("rel_type")
+                # An event which is a replacement (ie edit) or annotation (ie,
+                # reaction) may not have any other event related to it.
+                if relation_type in (RelationTypes.ANNOTATION, RelationTypes.REPLACE):
+                    continue
+
+            # The event should get bundled aggregations.
+            events_by_id[event.event_id] = event
+            # Track the event's relation information for later.
+            if isinstance(relation_type, str):
+                relations_by_id[event.event_id] = relation_type
 
         # event ID -> bundled aggregation in non-serialized form.
         results: Dict[str, BundledAggregations] = {}
@@ -375,16 +394,34 @@ class RelationsHandler:
         # Fetch any ignored users of the requesting user.
         ignored_users = await self._main_store.ignored_users(user_id)
 
+        # Threads are special as the latest event of a thread might cause additional
+        # events to be fetched. Thus, we check those first!
+
+        # Fetch thread summaries (but only for the directly requested events).
+        threads = await self.get_threads_for_events(
+            # It is not valid to start a thread on an event which itself relates to another event.
+            [eid for eid in events_by_id.keys() if eid not in relations_by_id],
+            user_id,
+            ignored_users,
+        )
+        for event_id, thread in threads.items():
+            results.setdefault(event_id, BundledAggregations()).thread = thread
+
+            # If the latest event in a thread is not already being fetched,
+            # add it. This ensures that the bundled aggregations for the
+            # latest thread event is correct.
+            latest_thread_event = thread.latest_event
+            if latest_thread_event and latest_thread_event.event_id not in events_by_id:
+                events_by_id[latest_thread_event.event_id] = latest_thread_event
+                # Keep relations_by_id in sync with events_by_id:
+                #
+                # We know that the latest event in a thread has a thread relation
+                # (as that is what makes it part of the thread).
+                relations_by_id[latest_thread_event.event_id] = RelationTypes.THREAD
+
         # Fetch other relations per event.
         for event in events_by_id.values():
-            # Do not bundle aggregations for an event which represents an edit or an
-            # annotation. It does not make sense for them to have related events.
-            relates_to = event.content.get("m.relates_to")
-            if isinstance(relates_to, (dict, frozendict)):
-                relation_type = relates_to.get("rel_type")
-                if relation_type in (RelationTypes.ANNOTATION, RelationTypes.REPLACE):
-                    continue
-
+            # Fetch any annotations (ie, reactions) to bundle with this event.
             annotations = await self.get_annotations_for_event(
                 event.event_id, event.room_id, ignored_users=ignored_users
             )
@@ -393,6 +430,7 @@ class RelationsHandler:
                     event.event_id, BundledAggregations()
                 ).annotations = {"chunk": annotations}
 
+            # Fetch any references to bundle with this event.
             references, next_token = await self.get_relations_for_event(
                 event.event_id,
                 event,
@@ -425,10 +463,4 @@ class RelationsHandler:
         for event_id, edit in edits.items():
             results.setdefault(event_id, BundledAggregations()).replace = edit
 
-        threads = await self.get_threads_for_events(
-            events_by_id.keys(), user_id, ignored_users
-        )
-        for event_id, thread in threads.items():
-            results.setdefault(event_id, BundledAggregations()).thread = thread
-
         return results
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index b31f00b517..604eb6ec15 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -57,7 +57,7 @@ from synapse.api.filtering import Filter
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersion
 from synapse.event_auth import validate_event_for_room_version
 from synapse.events import EventBase
-from synapse.events.utils import copy_power_levels_contents
+from synapse.events.utils import copy_and_fixup_power_levels_contents
 from synapse.federation.federation_client import InvalidResponseError
 from synapse.handlers.federation import get_domains_from_state
 from synapse.handlers.relations import BundledAggregations
@@ -337,13 +337,13 @@ class RoomCreationHandler:
         # 50, but if the default PL in a room is 50 or more, then we set the
         # required PL above that.
 
-        pl_content = dict(old_room_pl_state.content)
-        users_default = int(pl_content.get("users_default", 0))
+        pl_content = copy_and_fixup_power_levels_contents(old_room_pl_state.content)
+        users_default: int = pl_content.get("users_default", 0)  # type: ignore[assignment]
         restricted_level = max(users_default + 1, 50)
 
         updated = False
         for v in ("invite", "events_default"):
-            current = int(pl_content.get(v, 0))
+            current: int = pl_content.get(v, 0)  # type: ignore[assignment]
             if current < restricted_level:
                 logger.debug(
                     "Setting level for %s in %s to %i (was %i)",
@@ -380,7 +380,9 @@ class RoomCreationHandler:
                 "state_key": "",
                 "room_id": new_room_id,
                 "sender": requester.user.to_string(),
-                "content": old_room_pl_state.content,
+                "content": copy_and_fixup_power_levels_contents(
+                    old_room_pl_state.content
+                ),
             },
             ratelimit=False,
         )
@@ -471,7 +473,7 @@ class RoomCreationHandler:
         # dict so we can't just copy.deepcopy it.
         initial_state[
             (EventTypes.PowerLevels, "")
-        ] = power_levels = copy_power_levels_contents(
+        ] = power_levels = copy_and_fixup_power_levels_contents(
             initial_state[(EventTypes.PowerLevels, "")]
         )
 
diff --git a/synapse/handlers/room_summary.py b/synapse/handlers/room_summary.py
index 486145f48a..ff24ec8063 100644
--- a/synapse/handlers/room_summary.py
+++ b/synapse/handlers/room_summary.py
@@ -105,6 +105,7 @@ class RoomSummaryHandler:
             hs.get_clock(),
             "get_room_hierarchy",
         )
+        self._msc3266_enabled = hs.config.experimental.msc3266_enabled
 
     async def get_room_hierarchy(
         self,
@@ -630,7 +631,7 @@ class RoomSummaryHandler:
         return False
 
     async def _is_remote_room_accessible(
-        self, requester: str, room_id: str, room: JsonDict
+        self, requester: Optional[str], room_id: str, room: JsonDict
     ) -> bool:
         """
         Calculate whether the room received over federation should be shown to the requester.
@@ -645,7 +646,8 @@ class RoomSummaryHandler:
         due to an invite, etc.
 
         Args:
-            requester: The user requesting the summary.
+            requester: The user requesting the summary. If not passed only world
+                readability is checked.
             room_id: The room ID returned over federation.
             room: The summary of the room returned over federation.
 
@@ -659,6 +661,8 @@ class RoomSummaryHandler:
             or room.get("world_readable") is True
         ):
             return True
+        elif not requester:
+            return False
 
         # Check if the user is a member of any of the allowed rooms from the response.
         allowed_rooms = room.get("allowed_room_ids")
@@ -715,6 +719,10 @@ class RoomSummaryHandler:
             "room_type": create_event.content.get(EventContentFields.ROOM_TYPE),
         }
 
+        if self._msc3266_enabled:
+            entry["im.nheko.summary.version"] = stats["version"]
+            entry["im.nheko.summary.encryption"] = stats["encryption"]
+
         # Federation requests need to provide additional information so the
         # requested server is able to filter the response appropriately.
         if for_federation:
@@ -812,9 +820,45 @@ class RoomSummaryHandler:
 
                 room_summary["membership"] = membership or "leave"
         else:
-            # TODO federation API, descoped from initial unstable implementation
-            #      as MSC needs more maturing on that side.
-            raise SynapseError(400, "Federation is not currently supported.")
+            # Reuse the hierarchy query over federation
+            if remote_room_hosts is None:
+                raise SynapseError(400, "Missing via to query remote room")
+
+            (
+                room_entry,
+                children_room_entries,
+                inaccessible_children,
+            ) = await self._summarize_remote_room_hierarchy(
+                _RoomQueueEntry(room_id, remote_room_hosts),
+                suggested_only=True,
+            )
+
+            # The results over federation might include rooms that we, as the
+            # requesting server, are allowed to see, but the requesting user is
+            # not permitted to see.
+            #
+            # Filter the returned results to only what is accessible to the user.
+            if not room_entry or not await self._is_remote_room_accessible(
+                requester, room_entry.room_id, room_entry.room
+            ):
+                raise NotFoundError("Room not found or is not accessible")
+
+            room = dict(room_entry.room)
+            room.pop("allowed_room_ids", None)
+
+            # If there was a requester, add their membership.
+            # We keep the membership in the local membership table unless the
+            # room is purged even for remote rooms.
+            if requester:
+                (
+                    membership,
+                    _,
+                ) = await self._store.get_local_current_membership_for_user_in_room(
+                    requester, room_id
+                )
+                room["membership"] = membership or "leave"
+
+            return room
 
         return room_summary
 
diff --git a/synapse/handlers/search.py b/synapse/handlers/search.py
index 102dd4b57d..5619f8f50e 100644
--- a/synapse/handlers/search.py
+++ b/synapse/handlers/search.py
@@ -357,7 +357,7 @@ class SearchHandler:
             itertools.chain(
                 # The events_before and events_after for each context.
                 itertools.chain.from_iterable(
-                    itertools.chain(context["events_before"], context["events_after"])  # type: ignore[arg-type]
+                    itertools.chain(context["events_before"], context["events_after"])
                     for context in contexts.values()
                 ),
                 # The returned events.
@@ -373,10 +373,10 @@ class SearchHandler:
 
         for context in contexts.values():
             context["events_before"] = self._event_serializer.serialize_events(
-                context["events_before"], time_now, bundle_aggregations=aggregations  # type: ignore[arg-type]
+                context["events_before"], time_now, bundle_aggregations=aggregations
             )
             context["events_after"] = self._event_serializer.serialize_events(
-                context["events_after"], time_now, bundle_aggregations=aggregations  # type: ignore[arg-type]
+                context["events_after"], time_now, bundle_aggregations=aggregations
             )
 
         results = [
diff --git a/synapse/handlers/sso.py b/synapse/handlers/sso.py
index e4fe94e557..1e171f3f71 100644
--- a/synapse/handlers/sso.py
+++ b/synapse/handlers/sso.py
@@ -468,7 +468,7 @@ class SsoHandler:
                     auth_provider_id,
                     remote_user_id,
                     get_request_user_agent(request),
-                    request.getClientIP(),
+                    request.getClientAddress().host,
                 )
                 new_user = True
             elif self._sso_update_profile_information:
@@ -928,7 +928,7 @@ class SsoHandler:
             session.auth_provider_id,
             session.remote_user_id,
             get_request_user_agent(request),
-            request.getClientIP(),
+            request.getClientAddress().host,
         )
 
         logger.info(
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 5125126a80..2c555a66d0 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -1045,7 +1045,7 @@ class SyncHandler:
             last_unread_event_id = await self.store.get_last_receipt_event_id_for_user(
                 user_id=sync_config.user.to_string(),
                 room_id=room_id,
-                receipt_type=ReceiptTypes.READ,
+                receipt_types=(ReceiptTypes.READ, ReceiptTypes.READ_PRIVATE),
             )
 
             return await self.store.get_unread_event_push_actions_by_room_for_user(
diff --git a/synapse/handlers/ui_auth/checkers.py b/synapse/handlers/ui_auth/checkers.py
index 472b029af3..05cebb5d4d 100644
--- a/synapse/handlers/ui_auth/checkers.py
+++ b/synapse/handlers/ui_auth/checkers.py
@@ -256,7 +256,9 @@ class RegistrationTokenAuthChecker(UserInteractiveAuthChecker):
     def __init__(self, hs: "HomeServer"):
         super().__init__(hs)
         self.hs = hs
-        self._enabled = bool(hs.config.registration.registration_requires_token)
+        self._enabled = bool(
+            hs.config.registration.registration_requires_token
+        ) or bool(hs.config.registration.enable_registration_token_3pid_bypass)
         self.store = hs.get_datastores().main
 
     def is_enabled(self) -> bool:
diff --git a/synapse/handlers/user_directory.py b/synapse/handlers/user_directory.py
index 048fd4bb82..74f7fdfe6c 100644
--- a/synapse/handlers/user_directory.py
+++ b/synapse/handlers/user_directory.py
@@ -60,7 +60,7 @@ class UserDirectoryHandler(StateDeltasHandler):
         self.clock = hs.get_clock()
         self.notifier = hs.get_notifier()
         self.is_mine_id = hs.is_mine_id
-        self.update_user_directory = hs.config.server.update_user_directory
+        self.update_user_directory = hs.config.worker.should_update_user_directory
         self.search_all_users = hs.config.userdirectory.user_directory_search_all_users
         self.spam_checker = hs.get_spam_checker()
         # The current position in the current_state_delta stream