diff options
Diffstat (limited to 'synapse')
111 files changed, 1460 insertions, 355 deletions
diff --git a/synapse/api/constants.py b/synapse/api/constants.py index 691f8f9adf..8f37d2cf3b 100644 --- a/synapse/api/constants.py +++ b/synapse/api/constants.py @@ -51,6 +51,7 @@ class PresenceState: OFFLINE = "offline" UNAVAILABLE = "unavailable" ONLINE = "online" + BUSY = "org.matrix.msc3026.busy" class JoinRules: @@ -100,6 +101,9 @@ class EventTypes: Dummy = "org.matrix.dummy_event" + MSC1772_SPACE_CHILD = "org.matrix.msc1772.space.child" + MSC1772_SPACE_PARENT = "org.matrix.msc1772.space.parent" + class EduTypes: Presence = "m.presence" @@ -160,6 +164,9 @@ class EventContentFields: # cf https://github.com/matrix-org/matrix-doc/pull/2228 SELF_DESTRUCT_AFTER = "org.matrix.self_destruct_after" + # cf https://github.com/matrix-org/matrix-doc/pull/1772 + MSC1772_ROOM_TYPE = "org.matrix.msc1772.type" + class RoomEncryptionAlgorithms: MEGOLM_V1_AES_SHA2 = "m.megolm.v1.aes-sha2" diff --git a/synapse/app/__init__.py b/synapse/app/__init__.py index 4a9b0129c3..d1a2cd5e19 100644 --- a/synapse/app/__init__.py +++ b/synapse/app/__init__.py @@ -22,7 +22,9 @@ logger = logging.getLogger(__name__) try: python_dependencies.check_requirements() except python_dependencies.DependencyException as e: - sys.stderr.writelines(e.message) + sys.stderr.writelines( + e.message # noqa: B306, DependencyException.message is a property + ) sys.exit(1) diff --git a/synapse/app/generic_worker.py b/synapse/app/generic_worker.py index 274d582d07..caef394e1d 100644 --- a/synapse/app/generic_worker.py +++ b/synapse/app/generic_worker.py @@ -302,6 +302,8 @@ class GenericWorkerPresence(BasePresenceHandler): self.send_stop_syncing, UPDATE_SYNCING_USERS_MS ) + self._busy_presence_enabled = hs.config.experimental.msc3026_enabled + hs.get_reactor().addSystemEventTrigger( "before", "shutdown", @@ -439,8 +441,12 @@ class GenericWorkerPresence(BasePresenceHandler): PresenceState.ONLINE, PresenceState.UNAVAILABLE, PresenceState.OFFLINE, + PresenceState.BUSY, ) - if presence not in valid_presence: + + if presence not in valid_presence or ( + presence == PresenceState.BUSY and not self._busy_presence_enabled + ): raise SynapseError(400, "Invalid presence state") user_id = target_user.to_string() diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index b1c1c51e4d..86f4d9af9d 100644 --- a/synapse/config/experimental.py +++ b/synapse/config/experimental.py @@ -27,3 +27,7 @@ class ExperimentalConfig(Config): # MSC2858 (multiple SSO identity providers) self.msc2858_enabled = experimental.get("msc2858_enabled", False) # type: bool + # Spaces (MSC1772, MSC2946, etc) + self.spaces_enabled = experimental.get("spaces_enabled", False) # type: bool + # MSC3026 (busy presence state) + self.msc3026_enabled = experimental.get("msc3026_enabled", False) # type: bool diff --git a/synapse/config/key.py b/synapse/config/key.py index de964dff13..350ff1d665 100644 --- a/synapse/config/key.py +++ b/synapse/config/key.py @@ -404,7 +404,11 @@ def _parse_key_servers(key_servers, federation_verify_certificates): try: jsonschema.validate(key_servers, TRUSTED_KEY_SERVERS_SCHEMA) except jsonschema.ValidationError as e: - raise ConfigError("Unable to parse 'trusted_key_servers': " + e.message) + raise ConfigError( + "Unable to parse 'trusted_key_servers': {}".format( + e.message # noqa: B306, jsonschema.ValidationError.message is a valid attribute + ) + ) for server in key_servers: server_name = server["server_name"] diff --git a/synapse/config/metrics.py b/synapse/config/metrics.py index dfd27e1523..2b289f4208 100644 --- a/synapse/config/metrics.py +++ b/synapse/config/metrics.py @@ -56,7 +56,9 @@ class MetricsConfig(Config): try: check_requirements("sentry") except DependencyException as e: - raise ConfigError(e.message) + raise ConfigError( + e.message # noqa: B306, DependencyException.message is a property + ) self.sentry_dsn = config["sentry"].get("dsn") if not self.sentry_dsn: diff --git a/synapse/config/oidc_config.py b/synapse/config/oidc_config.py index 2bfb537c15..747ab9a7fe 100644 --- a/synapse/config/oidc_config.py +++ b/synapse/config/oidc_config.py @@ -15,11 +15,12 @@ # limitations under the License. from collections import Counter -from typing import Iterable, Mapping, Optional, Tuple, Type +from typing import Iterable, List, Mapping, Optional, Tuple, Type import attr from synapse.config._util import validate_config +from synapse.config.sso import SsoAttributeRequirement from synapse.python_dependencies import DependencyException, check_requirements from synapse.types import Collection, JsonDict from synapse.util.module_loader import load_module @@ -41,7 +42,9 @@ class OIDCConfig(Config): try: check_requirements("oidc") except DependencyException as e: - raise ConfigError(e.message) from e + raise ConfigError( + e.message # noqa: B306, DependencyException.message is a property + ) from e # check we don't have any duplicate idp_ids now. (The SSO handler will also # check for duplicates when the REST listeners get registered, but that happens @@ -191,6 +194,24 @@ class OIDCConfig(Config): # which is set to the claims returned by the UserInfo Endpoint and/or # in the ID Token. # + # It is possible to configure Synapse to only allow logins if certain attributes + # match particular values in the OIDC userinfo. The requirements can be listed under + # `attribute_requirements` as shown below. All of the listed attributes must + # match for the login to be permitted. Additional attributes can be added to + # userinfo by expanding the `scopes` section of the OIDC config to retrieve + # additional information from the OIDC provider. + # + # If the OIDC claim is a list, then the attribute must match any value in the list. + # Otherwise, it must exactly match the value of the claim. Using the example + # below, the `family_name` claim MUST be "Stephensson", but the `groups` + # claim MUST contain "admin". + # + # attribute_requirements: + # - attribute: family_name + # value: "Stephensson" + # - attribute: groups + # value: "admin" + # # See https://github.com/matrix-org/synapse/blob/master/docs/openid.md # for information on how to configure these options. # @@ -223,6 +244,9 @@ class OIDCConfig(Config): # localpart_template: "{{{{ user.login }}}}" # display_name_template: "{{{{ user.name }}}}" # email_template: "{{{{ user.email }}}}" + # attribute_requirements: + # - attribute: userGroup + # value: "synapseUsers" # For use with Keycloak # @@ -232,6 +256,9 @@ class OIDCConfig(Config): # client_id: "synapse" # client_secret: "copy secret generated in Keycloak UI" # scopes: ["openid", "profile"] + # attribute_requirements: + # - attribute: groups + # value: "admin" # For use with Github # @@ -329,6 +356,10 @@ OIDC_PROVIDER_CONFIG_SCHEMA = { }, "allow_existing_users": {"type": "boolean"}, "user_mapping_provider": {"type": ["object", "null"]}, + "attribute_requirements": { + "type": "array", + "items": SsoAttributeRequirement.JSON_SCHEMA, + }, }, } @@ -465,6 +496,11 @@ def _parse_oidc_config_dict( jwt_header=client_secret_jwt_key_config["jwt_header"], jwt_payload=client_secret_jwt_key_config.get("jwt_payload", {}), ) + # parse attribute_requirements from config (list of dicts) into a list of SsoAttributeRequirement + attribute_requirements = [ + SsoAttributeRequirement(**x) + for x in oidc_config.get("attribute_requirements", []) + ] return OidcProviderConfig( idp_id=idp_id, @@ -488,6 +524,7 @@ def _parse_oidc_config_dict( allow_existing_users=oidc_config.get("allow_existing_users", False), user_mapping_provider_class=user_mapping_provider_class, user_mapping_provider_config=user_mapping_provider_config, + attribute_requirements=attribute_requirements, ) @@ -577,3 +614,6 @@ class OidcProviderConfig: # the config of the user mapping provider user_mapping_provider_config = attr.ib() + + # required attributes to require in userinfo to allow login/registration + attribute_requirements = attr.ib(type=List[SsoAttributeRequirement]) diff --git a/synapse/config/ratelimiting.py b/synapse/config/ratelimiting.py index 847d25122c..3f3997f4e5 100644 --- a/synapse/config/ratelimiting.py +++ b/synapse/config/ratelimiting.py @@ -95,11 +95,11 @@ class RatelimitConfig(Config): self.rc_joins_local = RateLimitConfig( config.get("rc_joins", {}).get("local", {}), - defaults={"per_second": 0.1, "burst_count": 3}, + defaults={"per_second": 0.1, "burst_count": 10}, ) self.rc_joins_remote = RateLimitConfig( config.get("rc_joins", {}).get("remote", {}), - defaults={"per_second": 0.01, "burst_count": 3}, + defaults={"per_second": 0.01, "burst_count": 10}, ) # Ratelimit cross-user key requests: @@ -187,10 +187,10 @@ class RatelimitConfig(Config): #rc_joins: # local: # per_second: 0.1 - # burst_count: 3 + # burst_count: 10 # remote: # per_second: 0.01 - # burst_count: 3 + # burst_count: 10 # #rc_3pid_validation: # per_second: 0.003 diff --git a/synapse/config/repository.py b/synapse/config/repository.py index 69d9de5a43..061c4ec83f 100644 --- a/synapse/config/repository.py +++ b/synapse/config/repository.py @@ -176,7 +176,9 @@ class ContentRepositoryConfig(Config): check_requirements("url_preview") except DependencyException as e: - raise ConfigError(e.message) + raise ConfigError( + e.message # noqa: B306, DependencyException.message is a property + ) if "url_preview_ip_range_blacklist" not in config: raise ConfigError( diff --git a/synapse/config/saml2_config.py b/synapse/config/saml2_config.py index 4b494f217f..6db9cb5ced 100644 --- a/synapse/config/saml2_config.py +++ b/synapse/config/saml2_config.py @@ -76,7 +76,9 @@ class SAML2Config(Config): try: check_requirements("saml2") except DependencyException as e: - raise ConfigError(e.message) + raise ConfigError( + e.message # noqa: B306, DependencyException.message is a property + ) self.saml2_enabled = True diff --git a/synapse/config/tracer.py b/synapse/config/tracer.py index 0c1a854f09..727a1e7008 100644 --- a/synapse/config/tracer.py +++ b/synapse/config/tracer.py @@ -39,7 +39,9 @@ class TracerConfig(Config): try: check_requirements("opentracing") except DependencyException as e: - raise ConfigError(e.message) + raise ConfigError( + e.message # noqa: B306, DependencyException.message is a property + ) # The tracer is enabled so sanitize the config diff --git a/synapse/crypto/context_factory.py b/synapse/crypto/context_factory.py index 14b21796d9..c644b4dfc5 100644 --- a/synapse/crypto/context_factory.py +++ b/synapse/crypto/context_factory.py @@ -191,7 +191,7 @@ def _context_info_cb(ssl_connection, where, ret): # ... we further assume that SSLClientConnectionCreator has set the # '_synapse_tls_verifier' attribute to a ConnectionVerifier object. tls_protocol._synapse_tls_verifier.verify_context_info_cb(ssl_connection, where) - except: # noqa: E722, taken from the twisted implementation + except BaseException: # taken from the twisted implementation logger.exception("Error during info_callback") f = Failure() tls_protocol.failVerification(f) @@ -219,7 +219,7 @@ class SSLClientConnectionCreator: # ... and we also gut-wrench a '_synapse_tls_verifier' attribute into the # tls_protocol so that the SSL context's info callback has something to # call to do the cert verification. - setattr(tls_protocol, "_synapse_tls_verifier", self._verifier) + tls_protocol._synapse_tls_verifier = self._verifier return connection diff --git a/synapse/crypto/keyring.py b/synapse/crypto/keyring.py index 902128a23c..d5fb51513b 100644 --- a/synapse/crypto/keyring.py +++ b/synapse/crypto/keyring.py @@ -57,7 +57,7 @@ from synapse.util.metrics import Measure from synapse.util.retryutils import NotRetryingDestination if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py index 3ec4120f85..8f6b955d17 100644 --- a/synapse/events/__init__.py +++ b/synapse/events/__init__.py @@ -98,7 +98,7 @@ class DefaultDictProperty(DictProperty): class _EventInternalMetadata: - __slots__ = ["_dict", "stream_ordering"] + __slots__ = ["_dict", "stream_ordering", "outlier"] def __init__(self, internal_metadata_dict: JsonDict): # we have to copy the dict, because it turns out that the same dict is @@ -108,7 +108,10 @@ class _EventInternalMetadata: # the stream ordering of this event. None, until it has been persisted. self.stream_ordering = None # type: Optional[int] - outlier = DictProperty("outlier") # type: bool + # whether this event is an outlier (ie, whether we have the state at that point + # in the DAG) + self.outlier = False + out_of_band_membership = DictProperty("out_of_band_membership") # type: bool send_on_behalf_of = DictProperty("send_on_behalf_of") # type: str recheck_redaction = DictProperty("recheck_redaction") # type: bool @@ -129,7 +132,7 @@ class _EventInternalMetadata: return dict(self._dict) def is_outlier(self) -> bool: - return self._dict.get("outlier", False) + return self.outlier def is_out_of_band_membership(self) -> bool: """Whether this is an out of band membership, like an invite or an invite diff --git a/synapse/events/third_party_rules.py b/synapse/events/third_party_rules.py index 02bce8b5c9..9767d23940 100644 --- a/synapse/events/third_party_rules.py +++ b/synapse/events/third_party_rules.py @@ -13,12 +13,15 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import Callable, Union +from typing import TYPE_CHECKING, Union from synapse.events import EventBase from synapse.events.snapshot import EventContext from synapse.types import Requester, StateMap +if TYPE_CHECKING: + from synapse.server import HomeServer + class ThirdPartyEventRules: """Allows server admins to provide a Python module implementing an extra @@ -28,7 +31,7 @@ class ThirdPartyEventRules: behaviours. """ - def __init__(self, hs): + def __init__(self, hs: "HomeServer"): self.third_party_rules = None self.store = hs.get_datastore() @@ -95,10 +98,9 @@ class ThirdPartyEventRules: if self.third_party_rules is None: return True - ret = await self.third_party_rules.on_create_room( + return await self.third_party_rules.on_create_room( requester, config, is_requester_admin ) - return ret async def check_threepid_can_be_invited( self, medium: str, address: str, room_id: str @@ -119,10 +121,9 @@ class ThirdPartyEventRules: state_events = await self._get_state_map_for_room(room_id) - ret = await self.third_party_rules.check_threepid_can_be_invited( + return await self.third_party_rules.check_threepid_can_be_invited( medium, address, state_events ) - return ret async def check_visibility_can_be_modified( self, room_id: str, new_visibility: str @@ -143,7 +144,7 @@ class ThirdPartyEventRules: check_func = getattr( self.third_party_rules, "check_visibility_can_be_modified", None ) - if not check_func or not isinstance(check_func, Callable): + if not check_func or not callable(check_func): return True state_events = await self._get_state_map_for_room(room_id) diff --git a/synapse/events/utils.py b/synapse/events/utils.py index 7ca5c9940a..0f8a3b5ad8 100644 --- a/synapse/events/utils.py +++ b/synapse/events/utils.py @@ -22,6 +22,7 @@ from synapse.api.constants import EventTypes, RelationTypes from synapse.api.errors import Codes, SynapseError from synapse.api.room_versions import RoomVersion from synapse.util.async_helpers import yieldable_gather_results +from synapse.util.frozenutils import unfreeze from . import EventBase @@ -54,6 +55,8 @@ def prune_event(event: EventBase) -> EventBase: event.internal_metadata.stream_ordering ) + pruned_event.internal_metadata.outlier = event.internal_metadata.outlier + # Mark the event as redacted pruned_event.internal_metadata.redacted = True @@ -400,10 +403,19 @@ class EventClientSerializer: # If there is an edit replace the content, preserving existing # relations. + # Ensure we take copies of the edit content, otherwise we risk modifying + # the original event. + edit_content = edit.content.copy() + + # Unfreeze the event content if necessary, so that we may modify it below + edit_content = unfreeze(edit_content) + serialized_event["content"] = edit_content.get("m.new_content", {}) + + # Check for existing relations relations = event.content.get("m.relates_to") - serialized_event["content"] = edit.content.get("m.new_content", {}) if relations: - serialized_event["content"]["m.relates_to"] = relations + # Keep the relations, ensuring we use a dict copy of the original + serialized_event["content"]["m.relates_to"] = relations.copy() else: serialized_event["content"].pop("m.relates_to", None) diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index bee81fc019..afdb5bf2fa 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -27,11 +27,13 @@ from typing import ( List, Mapping, Optional, + Sequence, Tuple, TypeVar, Union, ) +import attr from prometheus_client import Counter from twisted.internet import defer @@ -62,7 +64,7 @@ from synapse.util.caches.expiringcache import ExpiringCache from synapse.util.retryutils import NotRetryingDestination if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) @@ -455,6 +457,7 @@ class FederationClient(FederationBase): description: str, destinations: Iterable[str], callback: Callable[[str], Awaitable[T]], + failover_on_unknown_endpoint: bool = False, ) -> T: """Try an operation on a series of servers, until it succeeds @@ -474,6 +477,10 @@ class FederationClient(FederationBase): next server tried. Normally the stacktrace is logged but this is suppressed if the exception is an InvalidResponseError. + failover_on_unknown_endpoint: if True, we will try other servers if it looks + like a server doesn't support the endpoint. This is typically useful + if the endpoint in question is new or experimental. + Returns: The result of callback, if it succeeds @@ -493,16 +500,31 @@ class FederationClient(FederationBase): except UnsupportedRoomVersionError: raise except HttpResponseException as e: - if not 500 <= e.code < 600: - raise e.to_synapse_error() - else: - logger.warning( - "Failed to %s via %s: %i %s", - description, - destination, - e.code, - e.args[0], - ) + synapse_error = e.to_synapse_error() + failover = False + + if 500 <= e.code < 600: + failover = True + + elif failover_on_unknown_endpoint: + # there is no good way to detect an "unknown" endpoint. Dendrite + # returns a 404 (with no body); synapse returns a 400 + # with M_UNRECOGNISED. + if e.code == 404 or ( + e.code == 400 and synapse_error.errcode == Codes.UNRECOGNIZED + ): + failover = True + + if not failover: + raise synapse_error from e + + logger.warning( + "Failed to %s via %s: %i %s", + description, + destination, + e.code, + e.args[0], + ) except Exception: logger.warning( "Failed to %s via %s", description, destination, exc_info=True @@ -1042,3 +1064,141 @@ class FederationClient(FederationBase): # If we don't manage to find it, return None. It's not an error if a # server doesn't give it to us. return None + + async def get_space_summary( + self, + destinations: Iterable[str], + room_id: str, + suggested_only: bool, + max_rooms_per_space: Optional[int], + exclude_rooms: List[str], + ) -> "FederationSpaceSummaryResult": + """ + Call other servers to get a summary of the given space + + + Args: + destinations: The remote servers. We will try them in turn, omitting any + that have been blacklisted. + + room_id: ID of the space to be queried + + suggested_only: If true, ask the remote server to only return children + with the "suggested" flag set + + max_rooms_per_space: A limit on the number of children to return for each + space + + exclude_rooms: A list of room IDs to tell the remote server to skip + + Returns: + a parsed FederationSpaceSummaryResult + + Raises: + SynapseError if we were unable to get a valid summary from any of the + remote servers + """ + + async def send_request(destination: str) -> FederationSpaceSummaryResult: + res = await self.transport_layer.get_space_summary( + destination=destination, + room_id=room_id, + suggested_only=suggested_only, + max_rooms_per_space=max_rooms_per_space, + exclude_rooms=exclude_rooms, + ) + + try: + return FederationSpaceSummaryResult.from_json_dict(res) + except ValueError as e: + raise InvalidResponseError(str(e)) + + return await self._try_destination_list( + "fetch space summary", + destinations, + send_request, + failover_on_unknown_endpoint=True, + ) + + +@attr.s(frozen=True, slots=True) +class FederationSpaceSummaryEventResult: + """Represents a single event in the result of a successful get_space_summary call. + + It's essentially just a serialised event object, but we do a bit of parsing and + validation in `from_json_dict` and store some of the validated properties in + object attributes. + """ + + event_type = attr.ib(type=str) + state_key = attr.ib(type=str) + via = attr.ib(type=Sequence[str]) + + # the raw data, including the above keys + data = attr.ib(type=JsonDict) + + @classmethod + def from_json_dict(cls, d: JsonDict) -> "FederationSpaceSummaryEventResult": + """Parse an event within the result of a /spaces/ request + + Args: + d: json object to be parsed + + Raises: + ValueError if d is not a valid event + """ + + event_type = d.get("type") + if not isinstance(event_type, str): + raise ValueError("Invalid event: 'event_type' must be a str") + + state_key = d.get("state_key") + if not isinstance(state_key, str): + raise ValueError("Invalid event: 'state_key' must be a str") + + content = d.get("content") + if not isinstance(content, dict): + raise ValueError("Invalid event: 'content' must be a dict") + + via = content.get("via") + if not isinstance(via, Sequence): + raise ValueError("Invalid event: 'via' must be a list") + if any(not isinstance(v, str) for v in via): + raise ValueError("Invalid event: 'via' must be a list of strings") + + return cls(event_type, state_key, via, d) + + +@attr.s(frozen=True, slots=True) +class FederationSpaceSummaryResult: + """Represents the data returned by a successful get_space_summary call.""" + + rooms = attr.ib(type=Sequence[JsonDict]) + events = attr.ib(type=Sequence[FederationSpaceSummaryEventResult]) + + @classmethod + def from_json_dict(cls, d: JsonDict) -> "FederationSpaceSummaryResult": + """Parse the result of a /spaces/ request + + Args: + d: json object to be parsed + + Raises: + ValueError if d is not a valid /spaces/ response + """ + rooms = d.get("rooms") + if not isinstance(rooms, Sequence): + raise ValueError("'rooms' must be a list") + if any(not isinstance(r, dict) for r in rooms): + raise ValueError("Invalid room in 'rooms' list") + + events = d.get("events") + if not isinstance(events, Sequence): + raise ValueError("'events' must be a list") + if any(not isinstance(e, dict) for e in events): + raise ValueError("Invalid event in 'events' list") + parsed_events = [ + FederationSpaceSummaryEventResult.from_json_dict(e) for e in events + ] + + return cls(rooms, parsed_events) diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 9839d3d016..d84e362070 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -35,7 +35,7 @@ from twisted.internet import defer from twisted.internet.abstract import isIPAddress from twisted.python import failure -from synapse.api.constants import EduTypes, EventTypes, Membership +from synapse.api.constants import EduTypes, EventTypes from synapse.api.errors import ( AuthError, Codes, @@ -63,7 +63,7 @@ from synapse.replication.http.federation import ( ReplicationFederationSendEduRestServlet, ReplicationGetQueryRestServlet, ) -from synapse.types import JsonDict, get_domain_from_id +from synapse.types import JsonDict from synapse.util import glob_to_regex, json_decoder, unwrapFirstError from synapse.util.async_helpers import Linearizer, concurrently_execute from synapse.util.caches.response_cache import ResponseCache @@ -727,27 +727,6 @@ class FederationServer(FederationBase): if the event was unacceptable for any other reason (eg, too large, too many prev_events, couldn't find the prev_events) """ - # check that it's actually being sent from a valid destination to - # workaround bug #1753 in 0.18.5 and 0.18.6 - if origin != get_domain_from_id(pdu.sender): - # We continue to accept join events from any server; this is - # necessary for the federation join dance to work correctly. - # (When we join over federation, the "helper" server is - # responsible for sending out the join event, rather than the - # origin. See bug #1893. This is also true for some third party - # invites). - if not ( - pdu.type == "m.room.member" - and pdu.content - and pdu.content.get("membership", None) - in (Membership.JOIN, Membership.INVITE) - ): - logger.info( - "Discarding PDU %s from invalid origin %s", pdu.event_id, origin - ) - return - else: - logger.info("Accepting join PDU %s from %s", pdu.event_id, origin) # We've already checked that we know the room version by this point room_version = await self.store.get_room_version(pdu.room_id) diff --git a/synapse/federation/sender/per_destination_queue.py b/synapse/federation/sender/per_destination_queue.py index cc0d765e5f..89df9a619b 100644 --- a/synapse/federation/sender/per_destination_queue.py +++ b/synapse/federation/sender/per_destination_queue.py @@ -15,7 +15,7 @@ # limitations under the License. import datetime import logging -from typing import TYPE_CHECKING, Dict, Hashable, Iterable, List, Optional, Tuple, cast +from typing import TYPE_CHECKING, Dict, Hashable, Iterable, List, Optional, Tuple import attr from prometheus_client import Counter @@ -77,6 +77,7 @@ class PerDestinationQueue: self._transaction_manager = transaction_manager self._instance_name = hs.get_instance_name() self._federation_shard_config = hs.config.worker.federation_shard_config + self._state = hs.get_state_handler() self._should_send_on_this_instance = True if not self._federation_shard_config.should_handle( @@ -415,22 +416,97 @@ class PerDestinationQueue: "This should not happen." % event_ids ) - if logger.isEnabledFor(logging.INFO): - rooms = [p.room_id for p in catchup_pdus] - logger.info("Catching up rooms to %s: %r", self._destination, rooms) + # We send transactions with events from one room only, as its likely + # that the remote will have to do additional processing, which may + # take some time. It's better to give it small amounts of work + # rather than risk the request timing out and repeatedly being + # retried, and not making any progress. + # + # Note: `catchup_pdus` will have exactly one PDU per room. + for pdu in catchup_pdus: + # The PDU from the DB will be the last PDU in the room from + # *this server* that wasn't sent to the remote. However, other + # servers may have sent lots of events since then, and we want + # to try and tell the remote only about the *latest* events in + # the room. This is so that it doesn't get inundated by events + # from various parts of the DAG, which all need to be processed. + # + # Note: this does mean that in large rooms a server coming back + # online will get sent the same events from all the different + # servers, but the remote will correctly deduplicate them and + # handle it only once. + + # Step 1, fetch the current extremities + extrems = await self._store.get_prev_events_for_room(pdu.room_id) + + if pdu.event_id in extrems: + # If the event is in the extremities, then great! We can just + # use that without having to do further checks. + room_catchup_pdus = [pdu] + else: + # If not, fetch the extremities and figure out which we can + # send. + extrem_events = await self._store.get_events_as_list(extrems) + + new_pdus = [] + for p in extrem_events: + # We pulled this from the DB, so it'll be non-null + assert p.internal_metadata.stream_ordering + + # Filter out events that happened before the remote went + # offline + if ( + p.internal_metadata.stream_ordering + < self._last_successful_stream_ordering + ): + continue - await self._transaction_manager.send_new_transaction( - self._destination, catchup_pdus, [] - ) + # Filter out events where the server is not in the room, + # e.g. it may have left/been kicked. *Ideally* we'd pull + # out the kick and send that, but it's a rare edge case + # so we don't bother for now (the server that sent the + # kick should send it out if its online). + hosts = await self._state.get_hosts_in_room_at_events( + p.room_id, [p.event_id] + ) + if self._destination not in hosts: + continue - sent_transactions_counter.inc() - final_pdu = catchup_pdus[-1] - self._last_successful_stream_ordering = cast( - int, final_pdu.internal_metadata.stream_ordering - ) - await self._store.set_destination_last_successful_stream_ordering( - self._destination, self._last_successful_stream_ordering - ) + new_pdus.append(p) + + # If we've filtered out all the extremities, fall back to + # sending the original event. This should ensure that the + # server gets at least some of missed events (especially if + # the other sending servers are up). + if new_pdus: + room_catchup_pdus = new_pdus + else: + room_catchup_pdus = [pdu] + + logger.info( + "Catching up rooms to %s: %r", self._destination, pdu.room_id + ) + + await self._transaction_manager.send_new_transaction( + self._destination, room_catchup_pdus, [] + ) + + sent_transactions_counter.inc() + + # We pulled this from the DB, so it'll be non-null + assert pdu.internal_metadata.stream_ordering + + # Note that we mark the last successful stream ordering as that + # from the *original* PDU, rather than the PDU(s) we actually + # send. This is because we use it to mark our position in the + # queue of missed PDUs to process. + self._last_successful_stream_ordering = ( + pdu.internal_metadata.stream_ordering + ) + + await self._store.set_destination_last_successful_stream_ordering( + self._destination, self._last_successful_stream_ordering + ) def _get_rr_edus(self, force_flush: bool) -> Iterable[Edu]: if not self._pending_rrs: diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py index 10c4747f97..6aee47c431 100644 --- a/synapse/federation/transport/client.py +++ b/synapse/federation/transport/client.py @@ -16,7 +16,7 @@ import logging import urllib -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional from synapse.api.constants import Membership from synapse.api.errors import Codes, HttpResponseException, SynapseError @@ -26,6 +26,7 @@ from synapse.api.urls import ( FEDERATION_V2_PREFIX, ) from synapse.logging.utils import log_function +from synapse.types import JsonDict logger = logging.getLogger(__name__) @@ -978,6 +979,38 @@ class TransportLayerClient: return self.client.get_json(destination=destination, path=path) + async def get_space_summary( + self, + destination: str, + room_id: str, + suggested_only: bool, + max_rooms_per_space: Optional[int], + exclude_rooms: List[str], + ) -> JsonDict: + """ + Args: + destination: The remote server + room_id: The room ID to ask about. + suggested_only: if True, only suggested rooms will be returned + max_rooms_per_space: an optional limit to the number of children to be + returned per space + exclude_rooms: a list of any rooms we can skip + """ + path = _create_path( + FEDERATION_UNSTABLE_PREFIX, "/org.matrix.msc2946/spaces/%s", room_id + ) + + params = { + "suggested_only": suggested_only, + "exclude_rooms": exclude_rooms, + } + if max_rooms_per_space is not None: + params["max_rooms_per_space"] = max_rooms_per_space + + return await self.client.post_json( + destination=destination, path=path, data=params + ) + def _create_path(federation_prefix, path, *args): """ diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py index 2cf935f38d..84e39c5a46 100644 --- a/synapse/federation/transport/server.py +++ b/synapse/federation/transport/server.py @@ -18,7 +18,7 @@ import functools import logging import re -from typing import Optional, Tuple, Type +from typing import Container, Mapping, Optional, Sequence, Tuple, Type import synapse from synapse.api.constants import MAX_GROUP_CATEGORYID_LENGTH, MAX_GROUP_ROLEID_LENGTH @@ -29,7 +29,7 @@ from synapse.api.urls import ( FEDERATION_V1_PREFIX, FEDERATION_V2_PREFIX, ) -from synapse.http.server import JsonResource +from synapse.http.server import HttpServer, JsonResource from synapse.http.servlet import ( parse_boolean_from_args, parse_integer_from_args, @@ -44,7 +44,8 @@ from synapse.logging.opentracing import ( whitelisted_homeserver, ) from synapse.server import HomeServer -from synapse.types import ThirdPartyInstanceID, get_domain_from_id +from synapse.types import JsonDict, ThirdPartyInstanceID, get_domain_from_id +from synapse.util.ratelimitutils import FederationRateLimiter from synapse.util.stringutils import parse_and_validate_server_name from synapse.util.versionstring import get_version_string @@ -1376,6 +1377,40 @@ class FederationGroupsSettingJoinPolicyServlet(BaseFederationServlet): return 200, new_content +class FederationSpaceSummaryServlet(BaseFederationServlet): + PREFIX = FEDERATION_UNSTABLE_PREFIX + "/org.matrix.msc2946" + PATH = "/spaces/(?P<room_id>[^/]*)" + + async def on_POST( + self, + origin: str, + content: JsonDict, + query: Mapping[bytes, Sequence[bytes]], + room_id: str, + ) -> Tuple[int, JsonDict]: + suggested_only = content.get("suggested_only", False) + if not isinstance(suggested_only, bool): + raise SynapseError( + 400, "'suggested_only' must be a boolean", Codes.BAD_JSON + ) + + exclude_rooms = content.get("exclude_rooms", []) + if not isinstance(exclude_rooms, list) or any( + not isinstance(x, str) for x in exclude_rooms + ): + raise SynapseError(400, "bad value for 'exclude_rooms'", Codes.BAD_JSON) + + max_rooms_per_space = content.get("max_rooms_per_space") + if max_rooms_per_space is not None and not isinstance(max_rooms_per_space, int): + raise SynapseError( + 400, "bad value for 'max_rooms_per_space'", Codes.BAD_JSON + ) + + return 200, await self.handler.federation_space_summary( + room_id, suggested_only, max_rooms_per_space, exclude_rooms + ) + + class RoomComplexityServlet(BaseFederationServlet): """ Indicates to other servers how complex (and therefore likely @@ -1474,18 +1509,24 @@ DEFAULT_SERVLET_GROUPS = ( ) -def register_servlets(hs, resource, authenticator, ratelimiter, servlet_groups=None): +def register_servlets( + hs: HomeServer, + resource: HttpServer, + authenticator: Authenticator, + ratelimiter: FederationRateLimiter, + servlet_groups: Optional[Container[str]] = None, +): """Initialize and register servlet classes. Will by default register all servlets. For custom behaviour, pass in a list of servlet_groups to register. Args: - hs (synapse.server.HomeServer): homeserver - resource (JsonResource): resource class to register to - authenticator (Authenticator): authenticator to use - ratelimiter (util.ratelimitutils.FederationRateLimiter): ratelimiter to use - servlet_groups (list[str], optional): List of servlet groups to register. + hs: homeserver + resource: resource class to register to + authenticator: authenticator to use + ratelimiter: ratelimiter to use + servlet_groups: List of servlet groups to register. Defaults to ``DEFAULT_SERVLET_GROUPS``. """ if not servlet_groups: @@ -1500,6 +1541,14 @@ def register_servlets(hs, resource, authenticator, ratelimiter, servlet_groups=N server_name=hs.hostname, ).register(resource) + if hs.config.experimental.spaces_enabled: + FederationSpaceSummaryServlet( + handler=hs.get_space_summary_handler(), + authenticator=authenticator, + ratelimiter=ratelimiter, + server_name=hs.hostname, + ).register(resource) + if "openid" in servlet_groups: for servletclass in OPENID_SERVLET_CLASSES: servletclass( diff --git a/synapse/groups/attestations.py b/synapse/groups/attestations.py index a3f8d92d08..368c44708d 100644 --- a/synapse/groups/attestations.py +++ b/synapse/groups/attestations.py @@ -46,7 +46,7 @@ from synapse.metrics.background_process_metrics import run_as_background_process from synapse.types import JsonDict, get_domain_from_id if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/groups/groups_server.py b/synapse/groups/groups_server.py index f9a0f40221..4b16a4ac29 100644 --- a/synapse/groups/groups_server.py +++ b/synapse/groups/groups_server.py @@ -25,7 +25,7 @@ from synapse.types import GroupID, JsonDict, RoomID, UserID, get_domain_from_id from synapse.util.async_helpers import concurrently_execute if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index d29b066a56..aade2c4a3a 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -24,7 +24,7 @@ from synapse.api.ratelimiting import Ratelimiter from synapse.types import UserID if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/account_data.py b/synapse/handlers/account_data.py index b1a5df9638..1ce6d697ed 100644 --- a/synapse/handlers/account_data.py +++ b/synapse/handlers/account_data.py @@ -25,7 +25,7 @@ from synapse.replication.http.account_data import ( from synapse.types import JsonDict, UserID if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer class AccountDataHandler: diff --git a/synapse/handlers/account_validity.py b/synapse/handlers/account_validity.py index 664d09da1c..d781bb251d 100644 --- a/synapse/handlers/account_validity.py +++ b/synapse/handlers/account_validity.py @@ -27,7 +27,7 @@ from synapse.types import UserID from synapse.util import stringutils if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/acme.py b/synapse/handlers/acme.py index 132be238dd..2a25af6288 100644 --- a/synapse/handlers/acme.py +++ b/synapse/handlers/acme.py @@ -24,7 +24,7 @@ from twisted.web.resource import Resource from synapse.app import check_bind_error if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/admin.py b/synapse/handlers/admin.py index db68c94c50..c494de49a3 100644 --- a/synapse/handlers/admin.py +++ b/synapse/handlers/admin.py @@ -25,7 +25,7 @@ from synapse.visibility import filter_events_for_client from ._base import BaseHandler if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py index deab8ff2d0..996f9e5deb 100644 --- a/synapse/handlers/appservice.py +++ b/synapse/handlers/appservice.py @@ -38,7 +38,7 @@ from synapse.types import Collection, JsonDict, RoomAlias, RoomStreamToken, User from synapse.util.metrics import Measure if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/auth.py b/synapse/handlers/auth.py index fb5f8118f0..d537ea8137 100644 --- a/synapse/handlers/auth.py +++ b/synapse/handlers/auth.py @@ -70,7 +70,7 @@ from synapse.util.msisdn import phone_number_to_msisdn from synapse.util.threepids import canonicalise_email if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) @@ -886,6 +886,19 @@ class AuthHandler(BaseHandler): ) return result + def can_change_password(self) -> bool: + """Get whether users on this server are allowed to change or set a password. + + Both `config.password_enabled` and `config.password_localdb_enabled` must be true. + + Note that any account (even SSO accounts) are allowed to add passwords if the above + is true. + + Returns: + Whether users on this server are allowed to change or set a password + """ + return self._password_enabled and self._password_localdb_enabled + def get_supported_login_types(self) -> Iterable[str]: """Get a the login types supported for the /login API diff --git a/synapse/handlers/cas_handler.py b/synapse/handlers/cas_handler.py index cb67589f7d..5060936f94 100644 --- a/synapse/handlers/cas_handler.py +++ b/synapse/handlers/cas_handler.py @@ -27,7 +27,7 @@ from synapse.http.site import SynapseRequest from synapse.types import UserID, map_username_to_mxid_localpart if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/deactivate_account.py b/synapse/handlers/deactivate_account.py index 3886d3124d..2bcd8f5435 100644 --- a/synapse/handlers/deactivate_account.py +++ b/synapse/handlers/deactivate_account.py @@ -23,7 +23,7 @@ from synapse.types import Requester, UserID, create_requester from ._base import BaseHandler if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/device.py b/synapse/handlers/device.py index df3cdc8fba..54293d0b9c 100644 --- a/synapse/handlers/device.py +++ b/synapse/handlers/device.py @@ -45,7 +45,7 @@ from synapse.util.retryutils import NotRetryingDestination from ._base import BaseHandler if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) @@ -166,7 +166,7 @@ class DeviceWorkerHandler(BaseHandler): # Fetch the current state at the time. try: - event_ids = await self.store.get_forward_extremeties_for_room( + event_ids = await self.store.get_forward_extremities_for_room_at_stream_ordering( room_id, stream_ordering=stream_ordering ) except errors.StoreError: @@ -907,6 +907,7 @@ class DeviceListUpdater: master_key = result.get("master_key") self_signing_key = result.get("self_signing_key") + ignore_devices = False # If the remote server has more than ~1000 devices for this user # we assume that something is going horribly wrong (e.g. a bot # that logs in and creates a new device every time it tries to @@ -925,6 +926,12 @@ class DeviceListUpdater: len(devices), ) devices = [] + ignore_devices = True + else: + cached_devices = await self.store.get_cached_devices_for_user(user_id) + if cached_devices == {d["device_id"]: d for d in devices}: + devices = [] + ignore_devices = True for device in devices: logger.debug( @@ -934,7 +941,10 @@ class DeviceListUpdater: stream_id, ) - await self.store.update_remote_device_list_cache(user_id, devices, stream_id) + if not ignore_devices: + await self.store.update_remote_device_list_cache( + user_id, devices, stream_id + ) device_ids = [device["device_id"] for device in devices] # Handle cross-signing keys. @@ -945,7 +955,8 @@ class DeviceListUpdater: ) device_ids = device_ids + cross_signing_device_ids - await self.device_handler.notify_device_update(user_id, device_ids) + if device_ids: + await self.device_handler.notify_device_update(user_id, device_ids) # We clobber the seen updates since we've re-synced from a given # point. @@ -973,14 +984,17 @@ class DeviceListUpdater: """ device_ids = [] - if master_key: + current_keys_map = await self.store.get_e2e_cross_signing_keys_bulk([user_id]) + current_keys = current_keys_map.get(user_id) or {} + + if master_key and master_key != current_keys.get("master"): await self.store.set_e2e_cross_signing_key(user_id, "master", master_key) _, verify_key = get_verify_key_from_cross_signing_key(master_key) # verify_key is a VerifyKey from signedjson, which uses # .version to denote the portion of the key ID after the # algorithm and colon, which is the device ID device_ids.append(verify_key.version) - if self_signing_key: + if self_signing_key and self_signing_key != current_keys.get("self_signing"): await self.store.set_e2e_cross_signing_key( user_id, "self_signing", self_signing_key ) diff --git a/synapse/handlers/devicemessage.py b/synapse/handlers/devicemessage.py index 7db4f48965..eb547743be 100644 --- a/synapse/handlers/devicemessage.py +++ b/synapse/handlers/devicemessage.py @@ -32,7 +32,7 @@ from synapse.util import json_encoder from synapse.util.stringutils import random_string if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/e2e_keys.py b/synapse/handlers/e2e_keys.py index 9a946a3cfe..2ad9b6d930 100644 --- a/synapse/handlers/e2e_keys.py +++ b/synapse/handlers/e2e_keys.py @@ -42,7 +42,7 @@ from synapse.util.caches.expiringcache import ExpiringCache from synapse.util.retryutils import NotRetryingDestination if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/e2e_room_keys.py b/synapse/handlers/e2e_room_keys.py index 622cae23be..a910d246d6 100644 --- a/synapse/handlers/e2e_room_keys.py +++ b/synapse/handlers/e2e_room_keys.py @@ -29,7 +29,7 @@ from synapse.types import JsonDict from synapse.util.async_helpers import Linearizer if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/groups_local.py b/synapse/handlers/groups_local.py index bfb95e3eee..a41ca5df9c 100644 --- a/synapse/handlers/groups_local.py +++ b/synapse/handlers/groups_local.py @@ -21,7 +21,7 @@ from synapse.api.errors import HttpResponseException, RequestSendFailed, Synapse from synapse.types import GroupID, JsonDict, get_domain_from_id if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/oidc_handler.py b/synapse/handlers/oidc_handler.py index 6d8551a6d6..bc3630e9e9 100644 --- a/synapse/handlers/oidc_handler.py +++ b/synapse/handlers/oidc_handler.py @@ -280,6 +280,7 @@ class OidcProvider: self._config = provider self._callback_url = hs.config.oidc_callback_url # type: str + self._oidc_attribute_requirements = provider.attribute_requirements self._scopes = provider.scopes self._user_profile_method = provider.user_profile_method @@ -859,6 +860,18 @@ class OidcProvider: ) # otherwise, it's a login + logger.debug("Userinfo for OIDC login: %s", userinfo) + + # Ensure that the attributes of the logged in user meet the required + # attributes by checking the userinfo against attribute_requirements + # In order to deal with the fact that OIDC userinfo can contain many + # types of data, we wrap non-list values in lists. + if not self._sso_handler.check_required_attributes( + request, + {k: v if isinstance(v, list) else [v] for k, v in userinfo.items()}, + self._oidc_attribute_requirements, + ): + return # Call the mapper to register/login the user try: diff --git a/synapse/handlers/password_policy.py b/synapse/handlers/password_policy.py index 6c635cc31b..92cefa11aa 100644 --- a/synapse/handlers/password_policy.py +++ b/synapse/handlers/password_policy.py @@ -21,7 +21,7 @@ from typing import TYPE_CHECKING from synapse.api.errors import Codes, PasswordRefusedError if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 54631b4ee2..da92feacc9 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -104,6 +104,8 @@ class BasePresenceHandler(abc.ABC): self.clock = hs.get_clock() self.store = hs.get_datastore() + self._busy_presence_enabled = hs.config.experimental.msc3026_enabled + active_presence = self.store.take_presence_startup_info() self.user_to_current_state = {state.user_id: state for state in active_presence} @@ -730,8 +732,12 @@ class PresenceHandler(BasePresenceHandler): PresenceState.ONLINE, PresenceState.UNAVAILABLE, PresenceState.OFFLINE, + PresenceState.BUSY, ) - if presence not in valid_presence: + + if presence not in valid_presence or ( + presence == PresenceState.BUSY and not self._busy_presence_enabled + ): raise SynapseError(400, "Invalid presence state") user_id = target_user.to_string() @@ -744,7 +750,9 @@ class PresenceHandler(BasePresenceHandler): msg = status_msg if presence != PresenceState.OFFLINE else None new_fields["status_msg"] = msg - if presence == PresenceState.ONLINE: + if presence == PresenceState.ONLINE or ( + presence == PresenceState.BUSY and self._busy_presence_enabled + ): new_fields["last_active_ts"] = self.clock.time_msec() await self._update_states([prev_state.copy_and_replace(**new_fields)]) diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index dd59392bda..a755363c3f 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -36,7 +36,7 @@ from synapse.types import ( from ._base import BaseHandler if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/read_marker.py b/synapse/handlers/read_marker.py index 6bb2fd936b..a54fe1968e 100644 --- a/synapse/handlers/read_marker.py +++ b/synapse/handlers/read_marker.py @@ -21,7 +21,7 @@ from synapse.util.async_helpers import Linearizer from ._base import BaseHandler if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/receipts.py b/synapse/handlers/receipts.py index 6a6c528849..dbfe9bfaca 100644 --- a/synapse/handlers/receipts.py +++ b/synapse/handlers/receipts.py @@ -20,7 +20,7 @@ from synapse.handlers._base import BaseHandler from synapse.types import JsonDict, ReadReceipt, get_domain_from_id if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py index 1abc8875cb..0fc2bf15d5 100644 --- a/synapse/handlers/register.py +++ b/synapse/handlers/register.py @@ -38,7 +38,7 @@ from synapse.types import RoomAlias, UserID, create_requester from ._base import BaseHandler if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) @@ -437,10 +437,10 @@ class RegistrationHandler(BaseHandler): if RoomAlias.is_valid(r): ( - room_id, + room, remote_room_hosts, ) = await room_member_handler.lookup_room_alias(room_alias) - room_id = room_id.to_string() + room_id = room.to_string() else: raise SynapseError( 400, "%s was not legal room ID or room alias" % (r,) diff --git a/synapse/handlers/room_list.py b/synapse/handlers/room_list.py index 8bfc46c654..924b81db7c 100644 --- a/synapse/handlers/room_list.py +++ b/synapse/handlers/room_list.py @@ -29,7 +29,7 @@ from synapse.util.caches.response_cache import ResponseCache from ._base import BaseHandler if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py index 1660921306..4d20ed8357 100644 --- a/synapse/handlers/room_member.py +++ b/synapse/handlers/room_member.py @@ -155,6 +155,10 @@ class RoomMemberHandler(metaclass=abc.ABCMeta): """ raise NotImplementedError() + @abc.abstractmethod + async def forget(self, user: UserID, room_id: str) -> None: + raise NotImplementedError() + def ratelimit_invite(self, room_id: Optional[str], invitee_user_id: str): """Ratelimit invites by room and by target user. diff --git a/synapse/handlers/room_member_worker.py b/synapse/handlers/room_member_worker.py index 108730a7a1..3a90fc0c16 100644 --- a/synapse/handlers/room_member_worker.py +++ b/synapse/handlers/room_member_worker.py @@ -14,7 +14,7 @@ # limitations under the License. import logging -from typing import List, Optional, Tuple +from typing import TYPE_CHECKING, List, Optional, Tuple from synapse.api.errors import SynapseError from synapse.handlers.room_member import RoomMemberHandler @@ -25,11 +25,14 @@ from synapse.replication.http.membership import ( ) from synapse.types import Requester, UserID +if TYPE_CHECKING: + from synapse.server import HomeServer + logger = logging.getLogger(__name__) class RoomMemberWorkerHandler(RoomMemberHandler): - def __init__(self, hs): + def __init__(self, hs: "HomeServer"): super().__init__(hs) self._remote_join_client = ReplRemoteJoin.make_client(hs) @@ -83,3 +86,6 @@ class RoomMemberWorkerHandler(RoomMemberHandler): await self._notify_change_client( user_id=target.to_string(), room_id=room_id, change="left" ) + + async def forget(self, target: UserID, room_id: str) -> None: + raise RuntimeError("Cannot forget rooms on workers.") diff --git a/synapse/handlers/search.py b/synapse/handlers/search.py index 94062e79cb..d742dfbd53 100644 --- a/synapse/handlers/search.py +++ b/synapse/handlers/search.py @@ -30,7 +30,7 @@ from synapse.visibility import filter_events_for_client from ._base import BaseHandler if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/set_password.py b/synapse/handlers/set_password.py index 84af2dde7e..f98a338ec5 100644 --- a/synapse/handlers/set_password.py +++ b/synapse/handlers/set_password.py @@ -21,7 +21,7 @@ from synapse.types import Requester from ._base import BaseHandler if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) @@ -41,7 +41,7 @@ class SetPasswordHandler(BaseHandler): logout_devices: bool, requester: Optional[Requester] = None, ) -> None: - if not self.hs.config.password_localdb_enabled: + if not self._auth_handler.can_change_password(): raise SynapseError(403, "Password change disabled", errcode=Codes.FORBIDDEN) try: diff --git a/synapse/handlers/space_summary.py b/synapse/handlers/space_summary.py new file mode 100644 index 0000000000..5d9418969d --- /dev/null +++ b/synapse/handlers/space_summary.py @@ -0,0 +1,395 @@ +# -*- coding: utf-8 -*- +# Copyright 2021 The Matrix.org Foundation C.I.C. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import itertools +import logging +from collections import deque +from typing import TYPE_CHECKING, Iterable, List, Optional, Sequence, Set, Tuple, cast + +import attr + +from synapse.api.constants import EventContentFields, EventTypes, HistoryVisibility +from synapse.api.errors import AuthError +from synapse.events import EventBase +from synapse.events.utils import format_event_for_client_v2 +from synapse.types import JsonDict + +if TYPE_CHECKING: + from synapse.server import HomeServer + +logger = logging.getLogger(__name__) + +# number of rooms to return. We'll stop once we hit this limit. +# TODO: allow clients to reduce this with a request param. +MAX_ROOMS = 50 + +# max number of events to return per room. +MAX_ROOMS_PER_SPACE = 50 + +# max number of federation servers to hit per room +MAX_SERVERS_PER_SPACE = 3 + + +class SpaceSummaryHandler: + def __init__(self, hs: "HomeServer"): + self._clock = hs.get_clock() + self._auth = hs.get_auth() + self._room_list_handler = hs.get_room_list_handler() + self._state_handler = hs.get_state_handler() + self._store = hs.get_datastore() + self._event_serializer = hs.get_event_client_serializer() + self._server_name = hs.hostname + self._federation_client = hs.get_federation_client() + + async def get_space_summary( + self, + requester: str, + room_id: str, + suggested_only: bool = False, + max_rooms_per_space: Optional[int] = None, + ) -> JsonDict: + """ + Implementation of the space summary C-S API + + Args: + requester: user id of the user making this request + + room_id: room id to start the summary at + + suggested_only: whether we should only return children with the "suggested" + flag set. + + max_rooms_per_space: an optional limit on the number of child rooms we will + return. This does not apply to the root room (ie, room_id), and + is overridden by MAX_ROOMS_PER_SPACE. + + Returns: + summary dict to return + """ + # first of all, check that the user is in the room in question (or it's + # world-readable) + await self._auth.check_user_in_room_or_world_readable(room_id, requester) + + # the queue of rooms to process + room_queue = deque((_RoomQueueEntry(room_id, ()),)) + + # rooms we have already processed + processed_rooms = set() # type: Set[str] + + # events we have already processed. We don't necessarily have their event ids, + # so instead we key on (room id, state key) + processed_events = set() # type: Set[Tuple[str, str]] + + rooms_result = [] # type: List[JsonDict] + events_result = [] # type: List[JsonDict] + + while room_queue and len(rooms_result) < MAX_ROOMS: + queue_entry = room_queue.popleft() + room_id = queue_entry.room_id + if room_id in processed_rooms: + # already done this room + continue + + logger.debug("Processing room %s", room_id) + + is_in_room = await self._store.is_host_joined(room_id, self._server_name) + + # The client-specified max_rooms_per_space limit doesn't apply to the + # room_id specified in the request, so we ignore it if this is the + # first room we are processing. + max_children = max_rooms_per_space if processed_rooms else None + + if is_in_room: + rooms, events = await self._summarize_local_room( + requester, room_id, suggested_only, max_children + ) + else: + rooms, events = await self._summarize_remote_room( + queue_entry, + suggested_only, + max_children, + exclude_rooms=processed_rooms, + ) + + logger.debug( + "Query of %s returned rooms %s, events %s", + queue_entry.room_id, + [room.get("room_id") for room in rooms], + ["%s->%s" % (ev["room_id"], ev["state_key"]) for ev in events], + ) + + rooms_result.extend(rooms) + + # any rooms returned don't need visiting again + processed_rooms.update(cast(str, room.get("room_id")) for room in rooms) + + # the room we queried may or may not have been returned, but don't process + # it again, anyway. + processed_rooms.add(room_id) + + # XXX: is it ok that we blindly iterate through any events returned by + # a remote server, whether or not they actually link to any rooms in our + # tree? + for ev in events: + # remote servers might return events we have already processed + # (eg, Dendrite returns inward pointers as well as outward ones), so + # we need to filter them out, to avoid returning duplicate links to the + # client. + ev_key = (ev["room_id"], ev["state_key"]) + if ev_key in processed_events: + continue + events_result.append(ev) + + # add the child to the queue. we have already validated + # that the vias are a list of server names. + room_queue.append( + _RoomQueueEntry(ev["state_key"], ev["content"]["via"]) + ) + processed_events.add(ev_key) + + return {"rooms": rooms_result, "events": events_result} + + async def federation_space_summary( + self, + room_id: str, + suggested_only: bool, + max_rooms_per_space: Optional[int], + exclude_rooms: Iterable[str], + ) -> JsonDict: + """ + Implementation of the space summary Federation API + + Args: + room_id: room id to start the summary at + + suggested_only: whether we should only return children with the "suggested" + flag set. + + max_rooms_per_space: an optional limit on the number of child rooms we will + return. Unlike the C-S API, this applies to the root room (room_id). + It is clipped to MAX_ROOMS_PER_SPACE. + + exclude_rooms: a list of rooms to skip over (presumably because the + calling server has already seen them). + + Returns: + summary dict to return + """ + # the queue of rooms to process + room_queue = deque((room_id,)) + + # the set of rooms that we should not walk further. Initialise it with the + # excluded-rooms list; we will add other rooms as we process them so that + # we do not loop. + processed_rooms = set(exclude_rooms) # type: Set[str] + + rooms_result = [] # type: List[JsonDict] + events_result = [] # type: List[JsonDict] + + while room_queue and len(rooms_result) < MAX_ROOMS: + room_id = room_queue.popleft() + if room_id in processed_rooms: + # already done this room + continue + + logger.debug("Processing room %s", room_id) + + rooms, events = await self._summarize_local_room( + None, room_id, suggested_only, max_rooms_per_space + ) + + processed_rooms.add(room_id) + + rooms_result.extend(rooms) + events_result.extend(events) + + # add any children to the queue + room_queue.extend(edge_event["state_key"] for edge_event in events) + + return {"rooms": rooms_result, "events": events_result} + + async def _summarize_local_room( + self, + requester: Optional[str], + room_id: str, + suggested_only: bool, + max_children: Optional[int], + ) -> Tuple[Sequence[JsonDict], Sequence[JsonDict]]: + if not await self._is_room_accessible(room_id, requester): + return (), () + + room_entry = await self._build_room_entry(room_id) + + # look for child rooms/spaces. + child_events = await self._get_child_events(room_id) + + if suggested_only: + # we only care about suggested children + child_events = filter(_is_suggested_child_event, child_events) + + if max_children is None or max_children > MAX_ROOMS_PER_SPACE: + max_children = MAX_ROOMS_PER_SPACE + + now = self._clock.time_msec() + events_result = [] # type: List[JsonDict] + for edge_event in itertools.islice(child_events, max_children): + events_result.append( + await self._event_serializer.serialize_event( + edge_event, + time_now=now, + event_format=format_event_for_client_v2, + ) + ) + return (room_entry,), events_result + + async def _summarize_remote_room( + self, + room: "_RoomQueueEntry", + suggested_only: bool, + max_children: Optional[int], + exclude_rooms: Iterable[str], + ) -> Tuple[Sequence[JsonDict], Sequence[JsonDict]]: + room_id = room.room_id + logger.info("Requesting summary for %s via %s", room_id, room.via) + + # we need to make the exclusion list json-serialisable + exclude_rooms = list(exclude_rooms) + + via = itertools.islice(room.via, MAX_SERVERS_PER_SPACE) + try: + res = await self._federation_client.get_space_summary( + via, + room_id, + suggested_only=suggested_only, + max_rooms_per_space=max_children, + exclude_rooms=exclude_rooms, + ) + except Exception as e: + logger.warning( + "Unable to get summary of %s via federation: %s", + room_id, + e, + exc_info=logger.isEnabledFor(logging.DEBUG), + ) + return (), () + + return res.rooms, tuple( + ev.data + for ev in res.events + if ev.event_type == EventTypes.MSC1772_SPACE_CHILD + ) + + async def _is_room_accessible(self, room_id: str, requester: Optional[str]) -> bool: + # if we have an authenticated requesting user, first check if they are in the + # room + if requester: + try: + await self._auth.check_user_in_room(room_id, requester) + return True + except AuthError: + pass + + # otherwise, check if the room is peekable + hist_vis_ev = await self._state_handler.get_current_state( + room_id, EventTypes.RoomHistoryVisibility, "" + ) + if hist_vis_ev: + hist_vis = hist_vis_ev.content.get("history_visibility") + if hist_vis == HistoryVisibility.WORLD_READABLE: + return True + + logger.info( + "room %s is unpeekable and user %s is not a member, omitting from summary", + room_id, + requester, + ) + return False + + async def _build_room_entry(self, room_id: str) -> JsonDict: + """Generate en entry suitable for the 'rooms' list in the summary response""" + stats = await self._store.get_room_with_stats(room_id) + + # currently this should be impossible because we call + # check_user_in_room_or_world_readable on the room before we get here, so + # there should always be an entry + assert stats is not None, "unable to retrieve stats for %s" % (room_id,) + + current_state_ids = await self._store.get_current_state_ids(room_id) + create_event = await self._store.get_event( + current_state_ids[(EventTypes.Create, "")] + ) + + # TODO: update once MSC1772 lands + room_type = create_event.content.get(EventContentFields.MSC1772_ROOM_TYPE) + + entry = { + "room_id": stats["room_id"], + "name": stats["name"], + "topic": stats["topic"], + "canonical_alias": stats["canonical_alias"], + "num_joined_members": stats["joined_members"], + "avatar_url": stats["avatar"], + "world_readable": ( + stats["history_visibility"] == HistoryVisibility.WORLD_READABLE + ), + "guest_can_join": stats["guest_access"] == "can_join", + "room_type": room_type, + } + + # Filter out Nones – rather omit the field altogether + room_entry = {k: v for k, v in entry.items() if v is not None} + + return room_entry + + async def _get_child_events(self, room_id: str) -> Iterable[EventBase]: + # look for child rooms/spaces. + current_state_ids = await self._store.get_current_state_ids(room_id) + + events = await self._store.get_events_as_list( + [ + event_id + for key, event_id in current_state_ids.items() + # TODO: update once MSC1772 lands + if key[0] == EventTypes.MSC1772_SPACE_CHILD + ] + ) + + # filter out any events without a "via" (which implies it has been redacted) + return (e for e in events if _has_valid_via(e)) + + +@attr.s(frozen=True, slots=True) +class _RoomQueueEntry: + room_id = attr.ib(type=str) + via = attr.ib(type=Sequence[str]) + + +def _has_valid_via(e: EventBase) -> bool: + via = e.content.get("via") + if not via or not isinstance(via, Sequence): + return False + for v in via: + if not isinstance(v, str): + logger.debug("Ignoring edge event %s with invalid via entry", e.event_id) + return False + return True + + +def _is_suggested_child_event(edge_event: EventBase) -> bool: + suggested = edge_event.content.get("suggested") + if isinstance(suggested, bool) and suggested: + return True + logger.debug("Ignorning not-suggested child %s", edge_event.state_key) + return False diff --git a/synapse/handlers/state_deltas.py b/synapse/handlers/state_deltas.py index b3f9875358..ee8f87e59a 100644 --- a/synapse/handlers/state_deltas.py +++ b/synapse/handlers/state_deltas.py @@ -17,7 +17,7 @@ import logging from typing import TYPE_CHECKING, Optional if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/stats.py b/synapse/handlers/stats.py index 924281144c..8730f99d03 100644 --- a/synapse/handlers/stats.py +++ b/synapse/handlers/stats.py @@ -24,7 +24,7 @@ from synapse.metrics.background_process_metrics import run_as_background_process from synapse.types import JsonDict if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index f50257cd57..ee607e6e65 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -80,7 +80,7 @@ class SyncConfig: filter_collection = attr.ib(type=FilterCollection) is_guest = attr.ib(type=bool) request_key = attr.ib(type=Tuple[Any, ...]) - device_id = attr.ib(type=str) + device_id = attr.ib(type=Optional[str]) @attr.s(slots=True, frozen=True) @@ -723,7 +723,9 @@ class SyncHandler: return summary - def get_lazy_loaded_members_cache(self, cache_key: Tuple[str, str]) -> LruCache: + def get_lazy_loaded_members_cache( + self, cache_key: Tuple[str, Optional[str]] + ) -> LruCache: cache = self.lazy_loaded_members_cache.get(cache_key) if cache is None: logger.debug("creating LruCache for %r", cache_key) @@ -1979,8 +1981,10 @@ class SyncHandler: logger.info("User joined room after current token: %s", room_id) - extrems = await self.store.get_forward_extremeties_for_room( - room_id, event_pos.stream + extrems = ( + await self.store.get_forward_extremities_for_room_at_stream_ordering( + room_id, event_pos.stream + ) ) users_in_room = await self.state.get_current_users_in_room(room_id, extrems) if user_id in users_in_room: diff --git a/synapse/handlers/user_directory.py b/synapse/handlers/user_directory.py index 1a8340000a..b121286d95 100644 --- a/synapse/handlers/user_directory.py +++ b/synapse/handlers/user_directory.py @@ -25,7 +25,7 @@ from synapse.types import JsonDict from synapse.util.metrics import Measure if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/http/client.py b/synapse/http/client.py index 1e01e0a9f2..a0caba84e4 100644 --- a/synapse/http/client.py +++ b/synapse/http/client.py @@ -77,7 +77,7 @@ from synapse.util import json_decoder from synapse.util.async_helpers import timeout_deferred if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/http/connectproxyclient.py b/synapse/http/connectproxyclient.py index 856e28454f..b797e3ce80 100644 --- a/synapse/http/connectproxyclient.py +++ b/synapse/http/connectproxyclient.py @@ -19,9 +19,10 @@ from zope.interface import implementer from twisted.internet import defer, protocol from twisted.internet.error import ConnectError -from twisted.internet.interfaces import IStreamClientEndpoint -from twisted.internet.protocol import connectionDone +from twisted.internet.interfaces import IReactorCore, IStreamClientEndpoint +from twisted.internet.protocol import ClientFactory, Protocol, connectionDone from twisted.web import http +from twisted.web.http_headers import Headers logger = logging.getLogger(__name__) @@ -43,23 +44,33 @@ class HTTPConnectProxyEndpoint: Args: reactor: the Twisted reactor to use for the connection - proxy_endpoint (IStreamClientEndpoint): the endpoint to use to connect to the - proxy - host (bytes): hostname that we want to CONNECT to - port (int): port that we want to connect to + proxy_endpoint: the endpoint to use to connect to the proxy + host: hostname that we want to CONNECT to + port: port that we want to connect to + headers: Extra HTTP headers to include in the CONNECT request """ - def __init__(self, reactor, proxy_endpoint, host, port): + def __init__( + self, + reactor: IReactorCore, + proxy_endpoint: IStreamClientEndpoint, + host: bytes, + port: int, + headers: Headers, + ): self._reactor = reactor self._proxy_endpoint = proxy_endpoint self._host = host self._port = port + self._headers = headers def __repr__(self): return "<HTTPConnectProxyEndpoint %s>" % (self._proxy_endpoint,) - def connect(self, protocolFactory): - f = HTTPProxiedClientFactory(self._host, self._port, protocolFactory) + def connect(self, protocolFactory: ClientFactory): + f = HTTPProxiedClientFactory( + self._host, self._port, protocolFactory, self._headers + ) d = self._proxy_endpoint.connect(f) # once the tcp socket connects successfully, we need to wait for the # CONNECT to complete. @@ -74,15 +85,23 @@ class HTTPProxiedClientFactory(protocol.ClientFactory): HTTP Protocol object and run the rest of the connection. Args: - dst_host (bytes): hostname that we want to CONNECT to - dst_port (int): port that we want to connect to - wrapped_factory (protocol.ClientFactory): The original Factory + dst_host: hostname that we want to CONNECT to + dst_port: port that we want to connect to + wrapped_factory: The original Factory + headers: Extra HTTP headers to include in the CONNECT request """ - def __init__(self, dst_host, dst_port, wrapped_factory): + def __init__( + self, + dst_host: bytes, + dst_port: int, + wrapped_factory: ClientFactory, + headers: Headers, + ): self.dst_host = dst_host self.dst_port = dst_port self.wrapped_factory = wrapped_factory + self.headers = headers self.on_connection = defer.Deferred() def startedConnecting(self, connector): @@ -92,7 +111,11 @@ class HTTPProxiedClientFactory(protocol.ClientFactory): wrapped_protocol = self.wrapped_factory.buildProtocol(addr) return HTTPConnectProtocol( - self.dst_host, self.dst_port, wrapped_protocol, self.on_connection + self.dst_host, + self.dst_port, + wrapped_protocol, + self.on_connection, + self.headers, ) def clientConnectionFailed(self, connector, reason): @@ -112,24 +135,37 @@ class HTTPConnectProtocol(protocol.Protocol): """Protocol that wraps an existing Protocol to do a CONNECT handshake at connect Args: - host (bytes): The original HTTP(s) hostname or IPv4 or IPv6 address literal + host: The original HTTP(s) hostname or IPv4 or IPv6 address literal to put in the CONNECT request - port (int): The original HTTP(s) port to put in the CONNECT request + port: The original HTTP(s) port to put in the CONNECT request - wrapped_protocol (interfaces.IProtocol): the original protocol (probably - HTTPChannel or TLSMemoryBIOProtocol, but could be anything really) + wrapped_protocol: the original protocol (probably HTTPChannel or + TLSMemoryBIOProtocol, but could be anything really) - connected_deferred (Deferred): a Deferred which will be callbacked with + connected_deferred: a Deferred which will be callbacked with wrapped_protocol when the CONNECT completes + + headers: Extra HTTP headers to include in the CONNECT request """ - def __init__(self, host, port, wrapped_protocol, connected_deferred): + def __init__( + self, + host: bytes, + port: int, + wrapped_protocol: Protocol, + connected_deferred: defer.Deferred, + headers: Headers, + ): self.host = host self.port = port self.wrapped_protocol = wrapped_protocol self.connected_deferred = connected_deferred - self.http_setup_client = HTTPConnectSetupClient(self.host, self.port) + self.headers = headers + + self.http_setup_client = HTTPConnectSetupClient( + self.host, self.port, self.headers + ) self.http_setup_client.on_connected.addCallback(self.proxyConnected) def connectionMade(self): @@ -154,7 +190,7 @@ class HTTPConnectProtocol(protocol.Protocol): if buf: self.wrapped_protocol.dataReceived(buf) - def dataReceived(self, data): + def dataReceived(self, data: bytes): # if we've set up the HTTP protocol, we can send the data there if self.wrapped_protocol.connected: return self.wrapped_protocol.dataReceived(data) @@ -168,21 +204,29 @@ class HTTPConnectSetupClient(http.HTTPClient): """HTTPClient protocol to send a CONNECT message for proxies and read the response. Args: - host (bytes): The hostname to send in the CONNECT message - port (int): The port to send in the CONNECT message + host: The hostname to send in the CONNECT message + port: The port to send in the CONNECT message + headers: Extra headers to send with the CONNECT message """ - def __init__(self, host, port): + def __init__(self, host: bytes, port: int, headers: Headers): self.host = host self.port = port + self.headers = headers self.on_connected = defer.Deferred() def connectionMade(self): logger.debug("Connected to proxy, sending CONNECT") self.sendCommand(b"CONNECT", b"%s:%d" % (self.host, self.port)) + + # Send any additional specified headers + for name, values in self.headers.getAllRawHeaders(): + for value in values: + self.sendHeader(name, value) + self.endHeaders() - def handleStatus(self, version, status, message): + def handleStatus(self, version: bytes, status: bytes, message: bytes): logger.debug("Got Status: %s %s %s", status, message, version) if status != b"200": raise ProxyConnectError("Unexpected status on CONNECT: %s" % status) diff --git a/synapse/http/proxyagent.py b/synapse/http/proxyagent.py index 3d553ae236..16ec850064 100644 --- a/synapse/http/proxyagent.py +++ b/synapse/http/proxyagent.py @@ -12,10 +12,13 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import base64 import logging import re +from typing import Optional, Tuple from urllib.request import getproxies_environment, proxy_bypass_environment +import attr from zope.interface import implementer from twisted.internet import defer @@ -23,6 +26,7 @@ from twisted.internet.endpoints import HostnameEndpoint, wrapClientTLS from twisted.python.failure import Failure from twisted.web.client import URI, BrowserLikePolicyForHTTPS, _AgentBase from twisted.web.error import SchemeNotSupported +from twisted.web.http_headers import Headers from twisted.web.iweb import IAgent from synapse.http.connectproxyclient import HTTPConnectProxyEndpoint @@ -32,6 +36,22 @@ logger = logging.getLogger(__name__) _VALID_URI = re.compile(br"\A[\x21-\x7e]+\Z") +@attr.s +class ProxyCredentials: + username_password = attr.ib(type=bytes) + + def as_proxy_authorization_value(self) -> bytes: + """ + Return the value for a Proxy-Authorization header (i.e. 'Basic abdef=='). + + Returns: + A transformation of the authentication string the encoded value for + a Proxy-Authorization header. + """ + # Encode as base64 and prepend the authorization type + return b"Basic " + base64.encodebytes(self.username_password) + + @implementer(IAgent) class ProxyAgent(_AgentBase): """An Agent implementation which will use an HTTP proxy if one was requested @@ -96,6 +116,9 @@ class ProxyAgent(_AgentBase): https_proxy = proxies["https"].encode() if "https" in proxies else None no_proxy = proxies["no"] if "no" in proxies else None + # Parse credentials from https proxy connection string if present + self.https_proxy_creds, https_proxy = parse_username_password(https_proxy) + self.http_proxy_endpoint = _http_proxy_endpoint( http_proxy, self.proxy_reactor, **self._endpoint_kwargs ) @@ -175,11 +198,22 @@ class ProxyAgent(_AgentBase): and self.https_proxy_endpoint and not should_skip_proxy ): + connect_headers = Headers() + + # Determine whether we need to set Proxy-Authorization headers + if self.https_proxy_creds: + # Set a Proxy-Authorization header + connect_headers.addRawHeader( + b"Proxy-Authorization", + self.https_proxy_creds.as_proxy_authorization_value(), + ) + endpoint = HTTPConnectProxyEndpoint( self.proxy_reactor, self.https_proxy_endpoint, parsed_uri.host, parsed_uri.port, + headers=connect_headers, ) else: # not using a proxy @@ -208,12 +242,16 @@ class ProxyAgent(_AgentBase): ) -def _http_proxy_endpoint(proxy, reactor, **kwargs): +def _http_proxy_endpoint(proxy: Optional[bytes], reactor, **kwargs): """Parses an http proxy setting and returns an endpoint for the proxy Args: - proxy (bytes|None): the proxy setting + proxy: the proxy setting in the form: [<username>:<password>@]<host>[:<port>] + Note that compared to other apps, this function currently lacks support + for specifying a protocol schema (i.e. protocol://...). + reactor: reactor to be used to connect to the proxy + kwargs: other args to be passed to HostnameEndpoint Returns: @@ -223,16 +261,43 @@ def _http_proxy_endpoint(proxy, reactor, **kwargs): if proxy is None: return None - # currently we only support hostname:port. Some apps also support - # protocol://<host>[:port], which allows a way of requiring a TLS connection to the - # proxy. - + # Parse the connection string host, port = parse_host_port(proxy, default_port=1080) return HostnameEndpoint(reactor, host, port, **kwargs) -def parse_host_port(hostport, default_port=None): - # could have sworn we had one of these somewhere else... +def parse_username_password(proxy: bytes) -> Tuple[Optional[ProxyCredentials], bytes]: + """ + Parses the username and password from a proxy declaration e.g + username:password@hostname:port. + + Args: + proxy: The proxy connection string. + + Returns + An instance of ProxyCredentials and the proxy connection string with any credentials + stripped, i.e u:p@host:port -> host:port. If no credentials were found, the + ProxyCredentials instance is replaced with None. + """ + if proxy and b"@" in proxy: + # We use rsplit here as the password could contain an @ character + credentials, proxy_without_credentials = proxy.rsplit(b"@", 1) + return ProxyCredentials(credentials), proxy_without_credentials + + return None, proxy + + +def parse_host_port(hostport: bytes, default_port: int = None) -> Tuple[bytes, int]: + """ + Parse the hostname and port from a proxy connection byte string. + + Args: + hostport: The proxy connection string. Must be in the form 'host[:port]'. + default_port: The default port to return if one is not found in `hostport`. + + Returns: + A tuple containing the hostname and port. Uses `default_port` if one was not found. + """ if b":" in hostport: host, port = hostport.rsplit(b":", 1) try: diff --git a/synapse/logging/context.py b/synapse/logging/context.py index 1a7ea4fa96..03cf3c2b8e 100644 --- a/synapse/logging/context.py +++ b/synapse/logging/context.py @@ -689,7 +689,7 @@ def run_in_background(f, *args, **kwargs) -> defer.Deferred: current = current_context() try: res = f(*args, **kwargs) - except: # noqa: E722 + except Exception: # the assumption here is that the caller doesn't want to be disturbed # by synchronous exceptions, so let's turn them into Failures. return defer.fail() diff --git a/synapse/push/__init__.py b/synapse/push/__init__.py index f4f7ec96f8..9fc3da49a2 100644 --- a/synapse/push/__init__.py +++ b/synapse/push/__init__.py @@ -21,7 +21,7 @@ import attr from synapse.types import JsonDict, RoomStreamToken if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer @attr.s(slots=True) diff --git a/synapse/push/action_generator.py b/synapse/push/action_generator.py index aaed28650d..38a47a600f 100644 --- a/synapse/push/action_generator.py +++ b/synapse/push/action_generator.py @@ -22,7 +22,7 @@ from synapse.push.bulk_push_rule_evaluator import BulkPushRuleEvaluator from synapse.util.metrics import Measure if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py index c016a83909..1897f59153 100644 --- a/synapse/push/bulk_push_rule_evaluator.py +++ b/synapse/push/bulk_push_rule_evaluator.py @@ -33,7 +33,7 @@ from synapse.util.caches.lrucache import LruCache from .push_rule_evaluator import PushRuleEvaluatorForEvent if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/push/emailpusher.py b/synapse/push/emailpusher.py index 3dc06a79e8..c0968dc7a1 100644 --- a/synapse/push/emailpusher.py +++ b/synapse/push/emailpusher.py @@ -24,7 +24,7 @@ from synapse.push import Pusher, PusherConfig, ThrottleParams from synapse.push.mailer import Mailer if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/push/httppusher.py b/synapse/push/httppusher.py index eb6de8ba72..26af5309c1 100644 --- a/synapse/push/httppusher.py +++ b/synapse/push/httppusher.py @@ -31,7 +31,7 @@ from synapse.push import Pusher, PusherConfig, PusherConfigException from . import push_rule_evaluator, push_tools if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) @@ -290,7 +290,7 @@ class HttpPusher(Pusher): if rejected is False: return False - if isinstance(rejected, list) or isinstance(rejected, tuple): + if isinstance(rejected, (list, tuple)): for pk in rejected: if pk != self.pushkey: # for sanity, we only remove the pushkey if it diff --git a/synapse/push/mailer.py b/synapse/push/mailer.py index d10201b6b3..2e5161de2c 100644 --- a/synapse/push/mailer.py +++ b/synapse/push/mailer.py @@ -40,7 +40,7 @@ from synapse.util.async_helpers import concurrently_execute from synapse.visibility import filter_events_for_client if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/push/pusher.py b/synapse/push/pusher.py index 2aa7918fb4..cb94127850 100644 --- a/synapse/push/pusher.py +++ b/synapse/push/pusher.py @@ -22,7 +22,7 @@ from synapse.push.httppusher import HttpPusher from synapse.push.mailer import Mailer if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/replication/http/federation.py b/synapse/replication/http/federation.py index 8af53b4f28..82ea3b895f 100644 --- a/synapse/replication/http/federation.py +++ b/synapse/replication/http/federation.py @@ -40,6 +40,7 @@ class ReplicationFederationSendEventsRestServlet(ReplicationEndpoint): // containing the event "event_format_version": .., // 1,2,3 etc: the event format version "internal_metadata": { .. serialized internal_metadata .. }, + "outlier": true|false, "rejected_reason": .., // The event.rejected_reason field "context": { .. serialized event context .. }, }], @@ -84,6 +85,7 @@ class ReplicationFederationSendEventsRestServlet(ReplicationEndpoint): "room_version": event.room_version.identifier, "event_format_version": event.format_version, "internal_metadata": event.internal_metadata.get_dict(), + "outlier": event.internal_metadata.is_outlier(), "rejected_reason": event.rejected_reason, "context": serialized_context, } @@ -116,6 +118,7 @@ class ReplicationFederationSendEventsRestServlet(ReplicationEndpoint): event = make_event_from_dict( event_dict, room_ver, internal_metadata, rejected_reason ) + event.internal_metadata.outlier = event_payload["outlier"] context = EventContext.deserialize( self.storage, event_payload["context"] diff --git a/synapse/replication/http/send_event.py b/synapse/replication/http/send_event.py index 8fa104c8d3..a4c5b44292 100644 --- a/synapse/replication/http/send_event.py +++ b/synapse/replication/http/send_event.py @@ -40,6 +40,7 @@ class ReplicationSendEventRestServlet(ReplicationEndpoint): // containing the event "event_format_version": .., // 1,2,3 etc: the event format version "internal_metadata": { .. serialized internal_metadata .. }, + "outlier": true|false, "rejected_reason": .., // The event.rejected_reason field "context": { .. serialized event context .. }, "requester": { .. serialized requester .. }, @@ -79,7 +80,6 @@ class ReplicationSendEventRestServlet(ReplicationEndpoint): ratelimit (bool) extra_users (list(UserID)): Any extra users to notify about event """ - serialized_context = await context.serialize(event, store) payload = { @@ -87,6 +87,7 @@ class ReplicationSendEventRestServlet(ReplicationEndpoint): "room_version": event.room_version.identifier, "event_format_version": event.format_version, "internal_metadata": event.internal_metadata.get_dict(), + "outlier": event.internal_metadata.is_outlier(), "rejected_reason": event.rejected_reason, "context": serialized_context, "requester": requester.serialize(), @@ -108,6 +109,7 @@ class ReplicationSendEventRestServlet(ReplicationEndpoint): event = make_event_from_dict( event_dict, room_ver, internal_metadata, rejected_reason ) + event.internal_metadata.outlier = content["outlier"] requester = Requester.deserialize(self.store, content["requester"]) context = EventContext.deserialize(self.storage, content["context"]) diff --git a/synapse/replication/slave/storage/pushers.py b/synapse/replication/slave/storage/pushers.py index 045bd014da..93161c3dfb 100644 --- a/synapse/replication/slave/storage/pushers.py +++ b/synapse/replication/slave/storage/pushers.py @@ -24,7 +24,7 @@ from ._base import BaseSlavedStore from ._slaved_id_tracker import SlavedIdTracker if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer class SlavedPusherStore(PusherWorkerStore, BaseSlavedStore): diff --git a/synapse/replication/tcp/streams/_base.py b/synapse/replication/tcp/streams/_base.py index f45e7a8c89..3dfee76743 100644 --- a/synapse/replication/tcp/streams/_base.py +++ b/synapse/replication/tcp/streams/_base.py @@ -33,7 +33,7 @@ import attr from synapse.replication.http.streams import ReplicationGetStreamUpdates if TYPE_CHECKING: - import synapse.server + from synapse.server import HomeServer logger = logging.getLogger(__name__) @@ -299,20 +299,23 @@ class TypingStream(Stream): NAME = "typing" ROW_TYPE = TypingStreamRow - def __init__(self, hs): - typing_handler = hs.get_typing_handler() - + def __init__(self, hs: "HomeServer"): writer_instance = hs.config.worker.writers.typing if writer_instance == hs.get_instance_name(): # On the writer, query the typing handler - update_function = typing_handler.get_all_typing_updates + typing_writer_handler = hs.get_typing_writer_handler() + update_function = ( + typing_writer_handler.get_all_typing_updates + ) # type: Callable[[str, int, int, int], Awaitable[Tuple[List[Tuple[int, Any]], int, bool]]] + current_token_function = typing_writer_handler.get_current_token else: # Query the typing writer process update_function = make_http_update_function(hs, self.NAME) + current_token_function = hs.get_typing_handler().get_current_token super().__init__( hs.get_instance_name(), - current_token_without_instance(typing_handler.get_current_token), + current_token_without_instance(current_token_function), update_function, ) @@ -509,7 +512,7 @@ class AccountDataStream(Stream): NAME = "account_data" ROW_TYPE = AccountDataStreamRow - def __init__(self, hs: "synapse.server.HomeServer"): + def __init__(self, hs: "HomeServer"): self.store = hs.get_datastore() super().__init__( hs.get_instance_name(), diff --git a/synapse/rest/admin/media.py b/synapse/rest/admin/media.py index 7fcc48a9d7..40646ef241 100644 --- a/synapse/rest/admin/media.py +++ b/synapse/rest/admin/media.py @@ -28,7 +28,7 @@ from synapse.rest.admin._base import ( from synapse.types import JsonDict if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/rest/admin/users.py b/synapse/rest/admin/users.py index 2c89b62e25..aaa56a7024 100644 --- a/synapse/rest/admin/users.py +++ b/synapse/rest/admin/users.py @@ -271,7 +271,7 @@ class UserRestServletV2(RestServlet): elif not deactivate and user["deactivated"]: if ( "password" not in body - and self.hs.config.password_localdb_enabled + and self.auth_handler.can_change_password() ): raise SynapseError( 400, "Must provide a password to re-activate an account." diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index 5884daea6d..525efdf221 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -18,7 +18,7 @@ import logging import re -from typing import TYPE_CHECKING, List, Optional +from typing import TYPE_CHECKING, List, Optional, Tuple from urllib import parse as urlparse from synapse.api.constants import EventTypes, Membership @@ -35,21 +35,30 @@ from synapse.events.utils import format_event_for_client_v2 from synapse.http.servlet import ( RestServlet, assert_params_in_dict, + parse_boolean, parse_integer, parse_json_object_from_request, parse_string, ) +from synapse.http.site import SynapseRequest from synapse.logging.opentracing import set_tag from synapse.rest.client.transactions import HttpTransactionCache from synapse.rest.client.v2_alpha._base import client_patterns from synapse.storage.state import StateFilter from synapse.streams.config import PaginationConfig -from synapse.types import RoomAlias, RoomID, StreamToken, ThirdPartyInstanceID, UserID +from synapse.types import ( + JsonDict, + RoomAlias, + RoomID, + StreamToken, + ThirdPartyInstanceID, + UserID, +) from synapse.util import json_decoder from synapse.util.stringutils import parse_and_validate_server_name, random_string if TYPE_CHECKING: - import synapse.server + from synapse.server import HomeServer logger = logging.getLogger(__name__) @@ -846,10 +855,10 @@ class RoomTypingRestServlet(RestServlet): "/rooms/(?P<room_id>[^/]*)/typing/(?P<user_id>[^/]*)$", v1=True ) - def __init__(self, hs): + def __init__(self, hs: "HomeServer"): super().__init__() + self.hs = hs self.presence_handler = hs.get_presence_handler() - self.typing_handler = hs.get_typing_handler() self.auth = hs.get_auth() # If we're not on the typing writer instance we should scream if we get @@ -874,16 +883,19 @@ class RoomTypingRestServlet(RestServlet): # Limit timeout to stop people from setting silly typing timeouts. timeout = min(content.get("timeout", 30000), 120000) + # Defer getting the typing handler since it will raise on workers. + typing_handler = self.hs.get_typing_writer_handler() + try: if content["typing"]: - await self.typing_handler.started_typing( + await typing_handler.started_typing( target_user=target_user, requester=requester, room_id=room_id, timeout=timeout, ) else: - await self.typing_handler.stopped_typing( + await typing_handler.stopped_typing( target_user=target_user, requester=requester, room_id=room_id ) except ShadowBanError: @@ -901,7 +913,7 @@ class RoomAliasListServlet(RestServlet): ), ] - def __init__(self, hs: "synapse.server.HomeServer"): + def __init__(self, hs: "HomeServer"): super().__init__() self.auth = hs.get_auth() self.directory_handler = hs.get_directory_handler() @@ -984,7 +996,58 @@ def register_txn_path(servlet, regex_string, http_server, with_get=False): ) -def register_servlets(hs, http_server, is_worker=False): +class RoomSpaceSummaryRestServlet(RestServlet): + PATTERNS = ( + re.compile( + "^/_matrix/client/unstable/org.matrix.msc2946" + "/rooms/(?P<room_id>[^/]*)/spaces$" + ), + ) + + def __init__(self, hs: "HomeServer"): + super().__init__() + self._auth = hs.get_auth() + self._space_summary_handler = hs.get_space_summary_handler() + + async def on_GET( + self, request: SynapseRequest, room_id: str + ) -> Tuple[int, JsonDict]: + requester = await self._auth.get_user_by_req(request, allow_guest=True) + + return 200, await self._space_summary_handler.get_space_summary( + requester.user.to_string(), + room_id, + suggested_only=parse_boolean(request, "suggested_only", default=False), + max_rooms_per_space=parse_integer(request, "max_rooms_per_space"), + ) + + async def on_POST( + self, request: SynapseRequest, room_id: str + ) -> Tuple[int, JsonDict]: + requester = await self._auth.get_user_by_req(request, allow_guest=True) + content = parse_json_object_from_request(request) + + suggested_only = content.get("suggested_only", False) + if not isinstance(suggested_only, bool): + raise SynapseError( + 400, "'suggested_only' must be a boolean", Codes.BAD_JSON + ) + + max_rooms_per_space = content.get("max_rooms_per_space") + if max_rooms_per_space is not None and not isinstance(max_rooms_per_space, int): + raise SynapseError( + 400, "'max_rooms_per_space' must be an integer", Codes.BAD_JSON + ) + + return 200, await self._space_summary_handler.get_space_summary( + requester.user.to_string(), + room_id, + suggested_only=suggested_only, + max_rooms_per_space=max_rooms_per_space, + ) + + +def register_servlets(hs: "HomeServer", http_server, is_worker=False): RoomStateEventRestServlet(hs).register(http_server) RoomMemberListRestServlet(hs).register(http_server) JoinedRoomMemberListRestServlet(hs).register(http_server) @@ -998,6 +1061,9 @@ def register_servlets(hs, http_server, is_worker=False): RoomTypingRestServlet(hs).register(http_server) RoomEventContextServlet(hs).register(http_server) + if hs.config.experimental.spaces_enabled: + RoomSpaceSummaryRestServlet(hs).register(http_server) + # Some servlets only get registered for the main process. if not is_worker: RoomCreateRestServlet(hs).register(http_server) diff --git a/synapse/rest/client/v2_alpha/account.py b/synapse/rest/client/v2_alpha/account.py index adf1d39728..c2ba790bab 100644 --- a/synapse/rest/client/v2_alpha/account.py +++ b/synapse/rest/client/v2_alpha/account.py @@ -45,7 +45,7 @@ from synapse.util.threepids import canonicalise_email, check_3pid_allowed from ._base import client_patterns, interactive_auth_handler if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/rest/client/v2_alpha/capabilities.py b/synapse/rest/client/v2_alpha/capabilities.py index 76879ac559..44ccf10ed4 100644 --- a/synapse/rest/client/v2_alpha/capabilities.py +++ b/synapse/rest/client/v2_alpha/capabilities.py @@ -13,12 +13,18 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging +from typing import TYPE_CHECKING, Tuple from synapse.api.room_versions import KNOWN_ROOM_VERSIONS from synapse.http.servlet import RestServlet +from synapse.http.site import SynapseRequest +from synapse.types import JsonDict from ._base import client_patterns +if TYPE_CHECKING: + from synapse.server import HomeServer + logger = logging.getLogger(__name__) @@ -27,21 +33,16 @@ class CapabilitiesRestServlet(RestServlet): PATTERNS = client_patterns("/capabilities$") - def __init__(self, hs): - """ - Args: - hs (synapse.server.HomeServer): server - """ + def __init__(self, hs: "HomeServer"): super().__init__() self.hs = hs self.config = hs.config self.auth = hs.get_auth() - self.store = hs.get_datastore() + self.auth_handler = hs.get_auth_handler() - async def on_GET(self, request): - requester = await self.auth.get_user_by_req(request, allow_guest=True) - user = await self.store.get_user_by_id(requester.user.to_string()) - change_password = bool(user["password_hash"]) + async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]: + await self.auth.get_user_by_req(request, allow_guest=True) + change_password = self.auth_handler.can_change_password() response = { "capabilities": { @@ -58,5 +59,5 @@ class CapabilitiesRestServlet(RestServlet): return 200, response -def register_servlets(hs, http_server): +def register_servlets(hs: "HomeServer", http_server): CapabilitiesRestServlet(hs).register(http_server) diff --git a/synapse/rest/client/v2_alpha/groups.py b/synapse/rest/client/v2_alpha/groups.py index 5901432fad..08fb6b2b06 100644 --- a/synapse/rest/client/v2_alpha/groups.py +++ b/synapse/rest/client/v2_alpha/groups.py @@ -38,7 +38,7 @@ from synapse.types import GroupID, JsonDict from ._base import client_patterns if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py index 8e52e4cca4..a0db0a054b 100644 --- a/synapse/rest/client/v2_alpha/sync.py +++ b/synapse/rest/client/v2_alpha/sync.py @@ -15,6 +15,7 @@ import itertools import logging +from typing import TYPE_CHECKING, Tuple from synapse.api.constants import PresenceState from synapse.api.errors import Codes, StoreError, SynapseError @@ -26,11 +27,15 @@ from synapse.events.utils import ( from synapse.handlers.presence import format_user_presence_state from synapse.handlers.sync import SyncConfig from synapse.http.servlet import RestServlet, parse_boolean, parse_integer, parse_string -from synapse.types import StreamToken +from synapse.http.site import SynapseRequest +from synapse.types import JsonDict, StreamToken from synapse.util import json_decoder from ._base import client_patterns, set_timeline_upper_limit +if TYPE_CHECKING: + from synapse.server import HomeServer + logger = logging.getLogger(__name__) @@ -73,7 +78,7 @@ class SyncRestServlet(RestServlet): PATTERNS = client_patterns("/sync$") ALLOWED_PRESENCE = {"online", "offline", "unavailable"} - def __init__(self, hs): + def __init__(self, hs: "HomeServer"): super().__init__() self.hs = hs self.auth = hs.get_auth() @@ -85,7 +90,7 @@ class SyncRestServlet(RestServlet): self._server_notices_sender = hs.get_server_notices_sender() self._event_serializer = hs.get_event_client_serializer() - async def on_GET(self, request): + async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]: if b"from" in request.args: # /events used to use 'from', but /sync uses 'since'. # Lets be helpful and whine if we see a 'from'. diff --git a/synapse/rest/client/versions.py b/synapse/rest/client/versions.py index d24a199318..3e3d8839f4 100644 --- a/synapse/rest/client/versions.py +++ b/synapse/rest/client/versions.py @@ -81,6 +81,8 @@ class VersionsRestServlet(RestServlet): "io.element.e2ee_forced.public": self.e2ee_forced_public, "io.element.e2ee_forced.private": self.e2ee_forced_private, "io.element.e2ee_forced.trusted_private": self.e2ee_forced_trusted_private, + # Supports the busy presence state described in MSC3026. + "org.matrix.msc3026.busy_presence": self.config.experimental.msc3026_enabled, }, }, ) diff --git a/synapse/rest/media/v1/config_resource.py b/synapse/rest/media/v1/config_resource.py index 1eff98ef14..c41a7ab412 100644 --- a/synapse/rest/media/v1/config_resource.py +++ b/synapse/rest/media/v1/config_resource.py @@ -23,7 +23,7 @@ from synapse.http.server import DirectServeJsonResource, respond_with_json from synapse.http.site import SynapseRequest if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer class MediaConfigResource(DirectServeJsonResource): diff --git a/synapse/rest/media/v1/download_resource.py b/synapse/rest/media/v1/download_resource.py index 8a43581f1f..5dadaeaf57 100644 --- a/synapse/rest/media/v1/download_resource.py +++ b/synapse/rest/media/v1/download_resource.py @@ -24,8 +24,8 @@ from synapse.http.servlet import parse_boolean from ._base import parse_media_id, respond_404 if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer from synapse.rest.media.v1.media_repository import MediaRepository + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/rest/media/v1/media_repository.py b/synapse/rest/media/v1/media_repository.py index 8b4841ed5d..0c041b542d 100644 --- a/synapse/rest/media/v1/media_repository.py +++ b/synapse/rest/media/v1/media_repository.py @@ -58,7 +58,7 @@ from .thumbnailer import Thumbnailer, ThumbnailError from .upload_resource import UploadResource if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/rest/media/v1/preview_url_resource.py b/synapse/rest/media/v1/preview_url_resource.py index b8895aeaa9..e590a0deab 100644 --- a/synapse/rest/media/v1/preview_url_resource.py +++ b/synapse/rest/media/v1/preview_url_resource.py @@ -54,8 +54,8 @@ from ._base import FileInfo if TYPE_CHECKING: from lxml import etree - from synapse.app.homeserver import HomeServer from synapse.rest.media.v1.media_repository import MediaRepository + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/rest/media/v1/storage_provider.py b/synapse/rest/media/v1/storage_provider.py index e92006faa9..031947557d 100644 --- a/synapse/rest/media/v1/storage_provider.py +++ b/synapse/rest/media/v1/storage_provider.py @@ -29,7 +29,7 @@ from .media_storage import FileResponder logger = logging.getLogger(__name__) if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer class StorageProvider(metaclass=abc.ABCMeta): diff --git a/synapse/rest/media/v1/thumbnail_resource.py b/synapse/rest/media/v1/thumbnail_resource.py index fbcd50f1e2..af802bc0b1 100644 --- a/synapse/rest/media/v1/thumbnail_resource.py +++ b/synapse/rest/media/v1/thumbnail_resource.py @@ -34,8 +34,8 @@ from ._base import ( ) if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer from synapse.rest.media.v1.media_repository import MediaRepository + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/rest/media/v1/upload_resource.py b/synapse/rest/media/v1/upload_resource.py index ae5aef2f7f..0138b2e2d1 100644 --- a/synapse/rest/media/v1/upload_resource.py +++ b/synapse/rest/media/v1/upload_resource.py @@ -26,8 +26,8 @@ from synapse.http.site import SynapseRequest from synapse.rest.media.v1.media_storage import SpamMediaException if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer from synapse.rest.media.v1.media_repository import MediaRepository + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/secrets.py b/synapse/secrets.py index fb6d90a3b7..7939db75e7 100644 --- a/synapse/secrets.py +++ b/synapse/secrets.py @@ -26,10 +26,10 @@ if sys.version_info[0:2] >= (3, 6): import secrets class Secrets: - def token_bytes(self, nbytes=32): + def token_bytes(self, nbytes: int = 32) -> bytes: return secrets.token_bytes(nbytes) - def token_hex(self, nbytes=32): + def token_hex(self, nbytes: int = 32) -> str: return secrets.token_hex(nbytes) @@ -38,8 +38,8 @@ else: import os class Secrets: - def token_bytes(self, nbytes=32): + def token_bytes(self, nbytes: int = 32) -> bytes: return os.urandom(nbytes) - def token_hex(self, nbytes=32): + def token_hex(self, nbytes: int = 32) -> str: return binascii.hexlify(self.token_bytes(nbytes)).decode("ascii") diff --git a/synapse/server.py b/synapse/server.py index 48ac87a124..5e787e2281 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -96,10 +96,11 @@ from synapse.handlers.room import ( RoomShutdownHandler, ) from synapse.handlers.room_list import RoomListHandler -from synapse.handlers.room_member import RoomMemberMasterHandler +from synapse.handlers.room_member import RoomMemberHandler, RoomMemberMasterHandler from synapse.handlers.room_member_worker import RoomMemberWorkerHandler from synapse.handlers.search import SearchHandler from synapse.handlers.set_password import SetPasswordHandler +from synapse.handlers.space_summary import SpaceSummaryHandler from synapse.handlers.sso import SsoHandler from synapse.handlers.stats import StatsHandler from synapse.handlers.sync import SyncHandler @@ -417,10 +418,19 @@ class HomeServer(metaclass=abc.ABCMeta): return PresenceHandler(self) @cache_in_self - def get_typing_handler(self): + def get_typing_writer_handler(self) -> TypingWriterHandler: if self.config.worker.writers.typing == self.get_instance_name(): return TypingWriterHandler(self) else: + raise Exception("Workers cannot write typing") + + @cache_in_self + def get_typing_handler(self) -> FollowerTypingHandler: + if self.config.worker.writers.typing == self.get_instance_name(): + # Use get_typing_writer_handler to ensure that we use the same + # cached version. + return self.get_typing_writer_handler() + else: return FollowerTypingHandler(self) @cache_in_self @@ -630,7 +640,7 @@ class HomeServer(metaclass=abc.ABCMeta): return ThirdPartyEventRules(self) @cache_in_self - def get_room_member_handler(self): + def get_room_member_handler(self) -> RoomMemberHandler: if self.config.worker_app: return RoomMemberWorkerHandler(self) return RoomMemberMasterHandler(self) @@ -640,13 +650,13 @@ class HomeServer(metaclass=abc.ABCMeta): return FederationHandlerRegistry(self) @cache_in_self - def get_server_notices_manager(self): + def get_server_notices_manager(self) -> ServerNoticesManager: if self.config.worker_app: raise Exception("Workers cannot send server notices") return ServerNoticesManager(self) @cache_in_self - def get_server_notices_sender(self): + def get_server_notices_sender(self) -> WorkerServerNoticesSender: if self.config.worker_app: return WorkerServerNoticesSender(self) return ServerNoticesSender(self) @@ -724,6 +734,10 @@ class HomeServer(metaclass=abc.ABCMeta): return AccountDataHandler(self) @cache_in_self + def get_space_summary_handler(self) -> SpaceSummaryHandler: + return SpaceSummaryHandler(self) + + @cache_in_self def get_external_cache(self) -> ExternalCache: return ExternalCache(self) diff --git a/synapse/server_notices/consent_server_notices.py b/synapse/server_notices/consent_server_notices.py index 9137c4edb1..a9349bf9a1 100644 --- a/synapse/server_notices/consent_server_notices.py +++ b/synapse/server_notices/consent_server_notices.py @@ -13,13 +13,16 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging -from typing import Any +from typing import TYPE_CHECKING, Any, Set from synapse.api.errors import SynapseError from synapse.api.urls import ConsentURIBuilder from synapse.config import ConfigError from synapse.types import get_localpart_from_id +if TYPE_CHECKING: + from synapse.server import HomeServer + logger = logging.getLogger(__name__) @@ -28,16 +31,11 @@ class ConsentServerNotices: privacy policy consent, and sends one if we do. """ - def __init__(self, hs): - """ - - Args: - hs (synapse.server.HomeServer): - """ + def __init__(self, hs: "HomeServer"): self._server_notices_manager = hs.get_server_notices_manager() self._store = hs.get_datastore() - self._users_in_progress = set() + self._users_in_progress = set() # type: Set[str] self._current_consent_version = hs.config.user_consent_version self._server_notice_content = hs.config.user_consent_server_notice_content @@ -73,6 +71,10 @@ class ConsentServerNotices: try: u = await self._store.get_user_by_id(user_id) + # The user doesn't exist. + if u is None: + return + if u["is_guest"] and not self._send_to_guests: # don't send to guests return diff --git a/synapse/server_notices/resource_limits_server_notices.py b/synapse/server_notices/resource_limits_server_notices.py index 6652451346..a18a2e76c9 100644 --- a/synapse/server_notices/resource_limits_server_notices.py +++ b/synapse/server_notices/resource_limits_server_notices.py @@ -13,7 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging -from typing import List, Tuple +from typing import TYPE_CHECKING, List, Tuple from synapse.api.constants import ( EventTypes, @@ -24,6 +24,9 @@ from synapse.api.constants import ( from synapse.api.errors import AuthError, ResourceLimitError, SynapseError from synapse.server_notices.server_notices_manager import SERVER_NOTICE_ROOM_TAG +if TYPE_CHECKING: + from synapse.server import HomeServer + logger = logging.getLogger(__name__) @@ -32,11 +35,7 @@ class ResourceLimitsServerNotices: ensures that the client is kept up to date. """ - def __init__(self, hs): - """ - Args: - hs (synapse.server.HomeServer): - """ + def __init__(self, hs: "HomeServer"): self._server_notices_manager = hs.get_server_notices_manager() self._store = hs.get_datastore() self._auth = hs.get_auth() diff --git a/synapse/server_notices/server_notices_manager.py b/synapse/server_notices/server_notices_manager.py index c46b2f047d..144e1da78e 100644 --- a/synapse/server_notices/server_notices_manager.py +++ b/synapse/server_notices/server_notices_manager.py @@ -58,7 +58,7 @@ class ServerNoticesManager: user_id: str, event_content: dict, type: str = EventTypes.Message, - state_key: Optional[bool] = None, + state_key: Optional[str] = None, ) -> EventBase: """Send a notice to the given user diff --git a/synapse/server_notices/server_notices_sender.py b/synapse/server_notices/server_notices_sender.py index 6870b67ca0..965c645889 100644 --- a/synapse/server_notices/server_notices_sender.py +++ b/synapse/server_notices/server_notices_sender.py @@ -12,25 +12,27 @@ # 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 Iterable, Union +from typing import TYPE_CHECKING, Iterable, Union from synapse.server_notices.consent_server_notices import ConsentServerNotices from synapse.server_notices.resource_limits_server_notices import ( ResourceLimitsServerNotices, ) +from synapse.server_notices.worker_server_notices_sender import ( + WorkerServerNoticesSender, +) + +if TYPE_CHECKING: + from synapse.server import HomeServer -class ServerNoticesSender: +class ServerNoticesSender(WorkerServerNoticesSender): """A centralised place which sends server notices automatically when Certain Events take place """ - def __init__(self, hs): - """ - - Args: - hs (synapse.server.HomeServer): - """ + def __init__(self, hs: "HomeServer"): + super().__init__(hs) self._server_notices = ( ConsentServerNotices(hs), ResourceLimitsServerNotices(hs), diff --git a/synapse/server_notices/worker_server_notices_sender.py b/synapse/server_notices/worker_server_notices_sender.py index 9273e61895..c76bd57460 100644 --- a/synapse/server_notices/worker_server_notices_sender.py +++ b/synapse/server_notices/worker_server_notices_sender.py @@ -12,16 +12,17 @@ # 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 + +if TYPE_CHECKING: + from synapse.server import HomeServer class WorkerServerNoticesSender: """Stub impl of ServerNoticesSender which does nothing""" - def __init__(self, hs): - """ - Args: - hs (synapse.server.HomeServer): - """ + def __init__(self, hs: "HomeServer"): + pass async def on_user_syncing(self, user_id: str) -> None: """Called when the user performs a sync operation. diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index a3c52695e9..0b9007e51f 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -36,7 +36,7 @@ from synapse.storage.purge_events import PurgeEventsStorage from synapse.storage.state import StateGroupStorage if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer __all__ = ["Databases", "DataStore"] diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index a25c4093bc..240905329f 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -27,7 +27,7 @@ from synapse.types import Collection, StreamToken, get_domain_from_id from synapse.util import json_decoder if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/storage/background_updates.py b/synapse/storage/background_updates.py index 329660cf0f..ccb06aab39 100644 --- a/synapse/storage/background_updates.py +++ b/synapse/storage/background_updates.py @@ -23,7 +23,7 @@ from synapse.util import json_encoder from . import engines if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer from synapse.storage.database import DatabasePool, LoggingTransaction logger = logging.getLogger(__name__) diff --git a/synapse/storage/database.py b/synapse/storage/database.py index f1ba529a2d..94590e7b45 100644 --- a/synapse/storage/database.py +++ b/synapse/storage/database.py @@ -670,7 +670,7 @@ class DatabasePool: for after_callback, after_args, after_kwargs in after_callbacks: after_callback(*after_args, **after_kwargs) - except: # noqa: E722, as we reraise the exception this is fine. + except Exception: for after_callback, after_args, after_kwargs in exception_callbacks: after_callback(*after_args, **after_kwargs) raise @@ -1906,6 +1906,7 @@ class DatabasePool: retcols: Iterable[str], filters: Optional[Dict[str, Any]] = None, keyvalues: Optional[Dict[str, Any]] = None, + exclude_keyvalues: Optional[Dict[str, Any]] = None, order_direction: str = "ASC", ) -> List[Dict[str, Any]]: """ @@ -1929,7 +1930,10 @@ class DatabasePool: apply a WHERE ? LIKE ? clause. keyvalues: column names and values to select the rows with, or None to not - apply a WHERE clause. + apply a WHERE key = value clause. + exclude_keyvalues: + column names and values to exclude rows with, or None to not + apply a WHERE key != value clause. order_direction: Whether the results should be ordered "ASC" or "DESC". Returns: @@ -1938,7 +1942,7 @@ class DatabasePool: if order_direction not in ["ASC", "DESC"]: raise ValueError("order_direction must be one of 'ASC' or 'DESC'.") - where_clause = "WHERE " if filters or keyvalues else "" + where_clause = "WHERE " if filters or keyvalues or exclude_keyvalues else "" arg_list = [] # type: List[Any] if filters: where_clause += " AND ".join("%s LIKE ?" % (k,) for k in filters) @@ -1947,6 +1951,9 @@ class DatabasePool: if keyvalues: where_clause += " AND ".join("%s = ?" % (k,) for k in keyvalues) arg_list += list(keyvalues.values()) + if exclude_keyvalues: + where_clause += " AND ".join("%s != ?" % (k,) for k in exclude_keyvalues) + arg_list += list(exclude_keyvalues.values()) sql = "SELECT %s FROM %s %s ORDER BY %s %s LIMIT ? OFFSET ?" % ( ", ".join(retcols), diff --git a/synapse/storage/databases/main/appservice.py b/synapse/storage/databases/main/appservice.py index 03a38422a1..85bb853d33 100644 --- a/synapse/storage/databases/main/appservice.py +++ b/synapse/storage/databases/main/appservice.py @@ -32,7 +32,7 @@ from synapse.types import JsonDict from synapse.util import json_encoder if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/storage/databases/main/deviceinbox.py b/synapse/storage/databases/main/deviceinbox.py index 45ca6620a8..691080ce74 100644 --- a/synapse/storage/databases/main/deviceinbox.py +++ b/synapse/storage/databases/main/deviceinbox.py @@ -14,7 +14,7 @@ # limitations under the License. import logging -from typing import List, Tuple +from typing import List, Optional, Tuple from synapse.logging.opentracing import log_kv, set_tag, trace from synapse.replication.tcp.streams import ToDeviceStream @@ -115,7 +115,7 @@ class DeviceInboxWorkerStore(SQLBaseStore): async def get_new_messages_for_device( self, user_id: str, - device_id: str, + device_id: Optional[str], last_stream_id: int, current_stream_id: int, limit: int = 100, @@ -163,7 +163,7 @@ class DeviceInboxWorkerStore(SQLBaseStore): @trace async def delete_messages_for_device( - self, user_id: str, device_id: str, up_to_stream_id: int + self, user_id: str, device_id: Optional[str], up_to_stream_id: int ) -> int: """ Args: diff --git a/synapse/storage/databases/main/event_federation.py b/synapse/storage/databases/main/event_federation.py index 332193ad1c..a956be491a 100644 --- a/synapse/storage/databases/main/event_federation.py +++ b/synapse/storage/databases/main/event_federation.py @@ -793,7 +793,7 @@ class EventFederationWorkerStore(EventsWorkerStore, SignatureWorkerStore, SQLBas return int(min_depth) if min_depth is not None else None - async def get_forward_extremeties_for_room( + async def get_forward_extremities_for_room_at_stream_ordering( self, room_id: str, stream_ordering: int ) -> List[str]: """For a given room_id and stream_ordering, return the forward diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index cd1ceac50e..98dac19a95 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -1270,8 +1270,10 @@ class PersistEventsStore: logger.exception("") raise + # update the stored internal_metadata to update the "outlier" flag. + # TODO: This is unused as of Synapse 1.31. Remove it once we are happy + # to drop backwards-compatibility with 1.30. metadata_json = json_encoder.encode(event.internal_metadata.get_dict()) - sql = "UPDATE event_json SET internal_metadata = ? WHERE event_id = ?" txn.execute(sql, (metadata_json, event.event_id)) @@ -1319,6 +1321,19 @@ class PersistEventsStore: d.pop("redacted_because", None) return d + def get_internal_metadata(event): + im = event.internal_metadata.get_dict() + + # temporary hack for database compatibility with Synapse 1.30 and earlier: + # store the `outlier` flag inside the internal_metadata json as well as in + # the `events` table, so that if anyone rolls back to an older Synapse, + # things keep working. This can be removed once we are happy to drop support + # for that + if event.internal_metadata.is_outlier(): + im["outlier"] = True + + return im + self.db_pool.simple_insert_many_txn( txn, table="event_json", @@ -1327,7 +1342,7 @@ class PersistEventsStore: "event_id": event.event_id, "room_id": event.room_id, "internal_metadata": json_encoder.encode( - event.internal_metadata.get_dict() + get_internal_metadata(event) ), "json": json_encoder.encode(event_dict(event)), "format_version": event.format_version, diff --git a/synapse/storage/databases/main/events_worker.py b/synapse/storage/databases/main/events_worker.py index c04e162ccc..952d4969b2 100644 --- a/synapse/storage/databases/main/events_worker.py +++ b/synapse/storage/databases/main/events_worker.py @@ -799,6 +799,7 @@ class EventsWorkerStore(SQLBaseStore): rejected_reason=rejected_reason, ) original_ev.internal_metadata.stream_ordering = row["stream_ordering"] + original_ev.internal_metadata.outlier = row["outlier"] event_map[event_id] = original_ev @@ -905,7 +906,8 @@ class EventsWorkerStore(SQLBaseStore): ej.json, ej.format_version, r.room_version, - rej.reason + rej.reason, + e.outlier FROM events AS e JOIN event_json AS ej USING (event_id) LEFT JOIN rooms r ON r.room_id = e.room_id @@ -929,6 +931,7 @@ class EventsWorkerStore(SQLBaseStore): "room_version_id": row[5], "rejected_reason": row[6], "redactions": [], + "outlier": row[7], } # check for redactions diff --git a/synapse/storage/databases/main/monthly_active_users.py b/synapse/storage/databases/main/monthly_active_users.py index d788dc0fc6..757da3d55d 100644 --- a/synapse/storage/databases/main/monthly_active_users.py +++ b/synapse/storage/databases/main/monthly_active_users.py @@ -13,7 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging -from typing import Dict, List +from typing import Dict, List, Optional from synapse.metrics.background_process_metrics import wrap_as_background_process from synapse.storage._base import SQLBaseStore @@ -109,7 +109,7 @@ class MonthlyActiveUsersWorkerStore(SQLBaseStore): return users @cached(num_args=1) - async def user_last_seen_monthly_active(self, user_id: str) -> int: + async def user_last_seen_monthly_active(self, user_id: str) -> Optional[int]: """ Checks if a given user is part of the monthly active user group diff --git a/synapse/storage/databases/main/presence.py b/synapse/storage/databases/main/presence.py index 29edab34d4..0ff693a310 100644 --- a/synapse/storage/databases/main/presence.py +++ b/synapse/storage/databases/main/presence.py @@ -13,7 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import List, Tuple +from typing import Dict, List, Tuple from synapse.api.presence import UserPresenceState from synapse.storage._base import SQLBaseStore, make_in_list_sql_clause @@ -157,5 +157,63 @@ class PresenceStore(SQLBaseStore): return {row["user_id"]: UserPresenceState(**row) for row in rows} + async def get_presence_for_all_users( + self, + include_offline: bool = True, + ) -> Dict[str, UserPresenceState]: + """Retrieve the current presence state for all users. + + Note that the presence_stream table is culled frequently, so it should only + contain the latest presence state for each user. + + Args: + include_offline: Whether to include offline presence states + + Returns: + A dict of user IDs to their current UserPresenceState. + """ + users_to_state = {} + + exclude_keyvalues = None + if not include_offline: + # Exclude offline presence state + exclude_keyvalues = {"state": "offline"} + + # This may be a very heavy database query. + # We paginate in order to not block a database connection. + limit = 100 + offset = 0 + while True: + rows = await self.db_pool.runInteraction( + "get_presence_for_all_users", + self.db_pool.simple_select_list_paginate_txn, + "presence_stream", + orderby="stream_id", + start=offset, + limit=limit, + exclude_keyvalues=exclude_keyvalues, + retcols=( + "user_id", + "state", + "last_active_ts", + "last_federation_update_ts", + "last_user_sync_ts", + "status_msg", + "currently_active", + ), + order_direction="ASC", + ) + + for row in rows: + users_to_state[row["user_id"]] = UserPresenceState(**row) + + # We've run out of updates to query + if len(rows) < limit: + break + + offset += limit + + return users_to_state + def get_current_presence_token(self): return self._presence_id_gen.get_current_token() diff --git a/synapse/storage/databases/main/pusher.py b/synapse/storage/databases/main/pusher.py index 85f1ebac98..c65558c280 100644 --- a/synapse/storage/databases/main/pusher.py +++ b/synapse/storage/databases/main/pusher.py @@ -27,7 +27,7 @@ from synapse.util import json_encoder from synapse.util.caches.descriptors import cached, cachedList if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/storage/databases/main/registration.py b/synapse/storage/databases/main/registration.py index eba66ff352..90a8f664ef 100644 --- a/synapse/storage/databases/main/registration.py +++ b/synapse/storage/databases/main/registration.py @@ -1210,6 +1210,7 @@ class RegistrationBackgroundUpdateStore(RegistrationWorkerStore): self._invalidate_cache_and_stream( txn, self.get_user_deactivated_status, (user_id,) ) + self._invalidate_cache_and_stream(txn, self.get_user_by_id, (user_id,)) txn.call_after(self.is_guest.invalidate, (user_id,)) @cached() diff --git a/synapse/storage/databases/main/transactions.py b/synapse/storage/databases/main/transactions.py index 0309661841..b7072f1f5e 100644 --- a/synapse/storage/databases/main/transactions.py +++ b/synapse/storage/databases/main/transactions.py @@ -22,7 +22,6 @@ from canonicaljson import encode_canonical_json from synapse.metrics.background_process_metrics import wrap_as_background_process from synapse.storage._base import SQLBaseStore, db_to_json from synapse.storage.database import DatabasePool, LoggingTransaction -from synapse.storage.engines import PostgresEngine, Sqlite3Engine from synapse.types import JsonDict from synapse.util.caches.expiringcache import ExpiringCache @@ -312,49 +311,23 @@ class TransactionStore(TransactionWorkerStore): stream_ordering: the stream_ordering of the event """ - return await self.db_pool.runInteraction( - "store_destination_rooms_entries", - self._store_destination_rooms_entries_txn, - destinations, - room_id, - stream_ordering, + await self.db_pool.simple_upsert_many( + table="destinations", + key_names=("destination",), + key_values=[(d,) for d in destinations], + value_names=[], + value_values=[], + desc="store_destination_rooms_entries_dests", ) - def _store_destination_rooms_entries_txn( - self, - txn: LoggingTransaction, - destinations: Iterable[str], - room_id: str, - stream_ordering: int, - ) -> None: - - # ensure we have a `destinations` row for this destination, as there is - # a foreign key constraint. - if isinstance(self.database_engine, PostgresEngine): - q = """ - INSERT INTO destinations (destination) - VALUES (?) - ON CONFLICT DO NOTHING; - """ - elif isinstance(self.database_engine, Sqlite3Engine): - q = """ - INSERT OR IGNORE INTO destinations (destination) - VALUES (?); - """ - else: - raise RuntimeError("Unknown database engine") - - txn.execute_batch(q, ((destination,) for destination in destinations)) - rows = [(destination, room_id) for destination in destinations] - - self.db_pool.simple_upsert_many_txn( - txn, + await self.db_pool.simple_upsert_many( table="destination_rooms", key_names=("destination", "room_id"), key_values=rows, value_names=["stream_ordering"], value_values=[(stream_ordering,)] * len(rows), + desc="store_destination_rooms_entries_rooms", ) async def get_destination_last_successful_stream_ordering( diff --git a/synapse/storage/purge_events.py b/synapse/storage/purge_events.py index 4dcd848c59..ad954990a7 100644 --- a/synapse/storage/purge_events.py +++ b/synapse/storage/purge_events.py @@ -20,7 +20,7 @@ from typing import TYPE_CHECKING, Set from synapse.storage.databases import Databases if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer logger = logging.getLogger(__name__) diff --git a/synapse/storage/state.py b/synapse/storage/state.py index d179a41884..2e277a21c4 100644 --- a/synapse/storage/state.py +++ b/synapse/storage/state.py @@ -32,7 +32,7 @@ from synapse.events import EventBase from synapse.types import MutableStateMap, StateMap if TYPE_CHECKING: - from synapse.app.homeserver import HomeServer + from synapse.server import HomeServer from synapse.storage.databases import Databases logger = logging.getLogger(__name__) @@ -449,7 +449,7 @@ class StateGroupStorage: return self.stores.state._get_state_groups_from_groups(groups, state_filter) async def get_state_for_events( - self, event_ids: List[str], state_filter: StateFilter = StateFilter.all() + self, event_ids: Iterable[str], state_filter: StateFilter = StateFilter.all() ) -> Dict[str, StateMap[EventBase]]: """Given a list of event_ids and type tuples, return a list of state dicts for each event. @@ -485,7 +485,7 @@ class StateGroupStorage: return {event: event_to_state[event] for event in event_ids} async def get_state_ids_for_events( - self, event_ids: List[str], state_filter: StateFilter = StateFilter.all() + self, event_ids: Iterable[str], state_filter: StateFilter = StateFilter.all() ) -> Dict[str, StateMap[str]]: """ Get the state dicts corresponding to a list of events, containing the event_ids diff --git a/synapse/util/async_helpers.py b/synapse/util/async_helpers.py index f33c115844..c3b2d981ea 100644 --- a/synapse/util/async_helpers.py +++ b/synapse/util/async_helpers.py @@ -496,7 +496,7 @@ def timeout_deferred( try: deferred.cancel() - except: # noqa: E722, if we throw any exception it'll break time outs + except Exception: # if we throw any exception it'll break time outs logger.exception("Canceller failed during timeout") # the cancel() call should have set off a chain of errbacks which diff --git a/synapse/util/caches/__init__.py b/synapse/util/caches/__init__.py index e676c2cac4..f968706334 100644 --- a/synapse/util/caches/__init__.py +++ b/synapse/util/caches/__init__.py @@ -116,7 +116,7 @@ def register_cache( """ if resizable: if not resize_callback: - resize_callback = getattr(cache, "set_cache_factor") + resize_callback = cache.set_cache_factor # type: ignore add_resizable_cache(cache_name, resize_callback) metric = CacheMetric(cache, cache_type, cache_name, collect_callback) diff --git a/synapse/util/frozenutils.py b/synapse/util/frozenutils.py index 5f7a6dd1d3..5ca2e71e60 100644 --- a/synapse/util/frozenutils.py +++ b/synapse/util/frozenutils.py @@ -36,7 +36,7 @@ def freeze(o): def unfreeze(o): if isinstance(o, (dict, frozendict)): - return dict({k: unfreeze(v) for k, v in o.items()}) + return {k: unfreeze(v) for k, v in o.items()} if isinstance(o, (bytes, str)): return o diff --git a/synapse/visibility.py b/synapse/visibility.py index e39d02602a..ff53a49b3a 100644 --- a/synapse/visibility.py +++ b/synapse/visibility.py @@ -13,7 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging -import operator +from typing import Dict, FrozenSet, List, Optional from synapse.api.constants import ( AccountDataTypes, @@ -21,10 +21,11 @@ from synapse.api.constants import ( HistoryVisibility, Membership, ) +from synapse.events import EventBase from synapse.events.utils import prune_event from synapse.storage import Storage from synapse.storage.state import StateFilter -from synapse.types import get_domain_from_id +from synapse.types import StateMap, get_domain_from_id logger = logging.getLogger(__name__) @@ -48,32 +49,32 @@ MEMBERSHIP_PRIORITY = ( async def filter_events_for_client( storage: Storage, - user_id, - events, - is_peeking=False, - always_include_ids=frozenset(), - filter_send_to_client=True, -): + user_id: str, + events: List[EventBase], + is_peeking: bool = False, + always_include_ids: FrozenSet[str] = frozenset(), + filter_send_to_client: bool = True, +) -> List[EventBase]: """ Check which events a user is allowed to see. If the user can see the event but its sender asked for their data to be erased, prune the content of the event. Args: storage - user_id(str): user id to be checked - events(list[synapse.events.EventBase]): sequence of events to be checked - is_peeking(bool): should be True if: + user_id: user id to be checked + events: sequence of events to be checked + is_peeking: should be True if: * the user is not currently a member of the room, and: * the user has not been a member of the room since the given events - always_include_ids (set(event_id)): set of event ids to specifically + always_include_ids: set of event ids to specifically include (unless sender is ignored) - filter_send_to_client (bool): Whether we're checking an event that's going to be + filter_send_to_client: Whether we're checking an event that's going to be sent to a client. This might not always be the case since this function can also be called to check whether a user can see the state at a given point. Returns: - list[synapse.events.EventBase] + The filtered events. """ # Filter out events that have been soft failed so that we don't relay them # to clients. @@ -90,7 +91,7 @@ async def filter_events_for_client( AccountDataTypes.IGNORED_USER_LIST, user_id ) - ignore_list = frozenset() + ignore_list = frozenset() # type: FrozenSet[str] if ignore_dict_content: ignored_users_dict = ignore_dict_content.get("ignored_users", {}) if isinstance(ignored_users_dict, dict): @@ -107,19 +108,18 @@ async def filter_events_for_client( room_id ] = await storage.main.get_retention_policy_for_room(room_id) - def allowed(event): + def allowed(event: EventBase) -> Optional[EventBase]: """ Args: - event (synapse.events.EventBase): event to check + event: event to check Returns: - None|EventBase: - None if the user cannot see this event at all + None if the user cannot see this event at all - a redacted copy of the event if they can only see a redacted - version + a redacted copy of the event if they can only see a redacted + version - the original event if they can see it as normal. + the original event if they can see it as normal. """ # Only run some checks if these events aren't about to be sent to clients. This is # because, if this is not the case, we're probably only checking if the users can @@ -252,48 +252,46 @@ async def filter_events_for_client( return event - # check each event: gives an iterable[None|EventBase] + # Check each event: gives an iterable of None or (a potentially modified) + # EventBase. filtered_events = map(allowed, events) - # remove the None entries - filtered_events = filter(operator.truth, filtered_events) - - # we turn it into a list before returning it. - return list(filtered_events) + # Turn it into a list and remove None entries before returning. + return [ev for ev in filtered_events if ev] async def filter_events_for_server( storage: Storage, - server_name, - events, - redact=True, - check_history_visibility_only=False, -): + server_name: str, + events: List[EventBase], + redact: bool = True, + check_history_visibility_only: bool = False, +) -> List[EventBase]: """Filter a list of events based on whether given server is allowed to see them. Args: storage - server_name (str) - events (iterable[FrozenEvent]) - redact (bool): Whether to return a redacted version of the event, or + server_name + events + redact: Whether to return a redacted version of the event, or to filter them out entirely. - check_history_visibility_only (bool): Whether to only check the + check_history_visibility_only: Whether to only check the history visibility, rather than things like if the sender has been erased. This is used e.g. during pagination to decide whether to backfill or not. Returns - list[FrozenEvent] + The filtered events. """ - def is_sender_erased(event, erased_senders): + def is_sender_erased(event: EventBase, erased_senders: Dict[str, bool]) -> bool: if erased_senders and erased_senders[event.sender]: logger.info("Sender of %s has been erased, redacting", event.event_id) return True return False - def check_event_is_visible(event, state): + def check_event_is_visible(event: EventBase, state: StateMap[EventBase]) -> bool: history = state.get((EventTypes.RoomHistoryVisibility, ""), None) if history: visibility = history.content.get( |