825 lines
		
	
	
		
			29 KiB
		
	
	
	
		
			Python
		
	
	
			
		
		
	
	
			825 lines
		
	
	
		
			29 KiB
		
	
	
	
		
			Python
		
	
	
# Copyright 2017 New Vector Ltd
 | 
						|
# Copyright 2020 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 email.utils
 | 
						|
import logging
 | 
						|
from typing import (
 | 
						|
    TYPE_CHECKING,
 | 
						|
    Any,
 | 
						|
    Callable,
 | 
						|
    Dict,
 | 
						|
    Generator,
 | 
						|
    Iterable,
 | 
						|
    List,
 | 
						|
    Optional,
 | 
						|
    Tuple,
 | 
						|
    Union,
 | 
						|
)
 | 
						|
 | 
						|
import attr
 | 
						|
import jinja2
 | 
						|
 | 
						|
from twisted.internet import defer
 | 
						|
from twisted.web.resource import IResource
 | 
						|
 | 
						|
from synapse.events import EventBase
 | 
						|
from synapse.events.presence_router import PresenceRouter
 | 
						|
from synapse.http.client import SimpleHttpClient
 | 
						|
from synapse.http.server import (
 | 
						|
    DirectServeHtmlResource,
 | 
						|
    DirectServeJsonResource,
 | 
						|
    respond_with_html,
 | 
						|
)
 | 
						|
from synapse.http.servlet import parse_json_object_from_request
 | 
						|
from synapse.http.site import SynapseRequest
 | 
						|
from synapse.logging.context import make_deferred_yieldable, run_in_background
 | 
						|
from synapse.metrics.background_process_metrics import run_as_background_process
 | 
						|
from synapse.storage.database import DatabasePool, LoggingTransaction
 | 
						|
from synapse.storage.databases.main.roommember import ProfileInfo
 | 
						|
from synapse.storage.state import StateFilter
 | 
						|
from synapse.types import (
 | 
						|
    DomainSpecificString,
 | 
						|
    JsonDict,
 | 
						|
    Requester,
 | 
						|
    UserID,
 | 
						|
    UserInfo,
 | 
						|
    create_requester,
 | 
						|
)
 | 
						|
from synapse.util import Clock
 | 
						|
from synapse.util.caches.descriptors import cached
 | 
						|
 | 
						|
if TYPE_CHECKING:
 | 
						|
    from synapse.server import HomeServer
 | 
						|
 | 
						|
"""
 | 
						|
This package defines the 'stable' API which can be used by extension modules which
 | 
						|
are loaded into Synapse.
 | 
						|
"""
 | 
						|
 | 
						|
PRESENCE_ALL_USERS = PresenceRouter.ALL_USERS
 | 
						|
 | 
						|
__all__ = [
 | 
						|
    "errors",
 | 
						|
    "make_deferred_yieldable",
 | 
						|
    "parse_json_object_from_request",
 | 
						|
    "respond_with_html",
 | 
						|
    "run_in_background",
 | 
						|
    "cached",
 | 
						|
    "UserID",
 | 
						|
    "DatabasePool",
 | 
						|
    "LoggingTransaction",
 | 
						|
    "DirectServeHtmlResource",
 | 
						|
    "DirectServeJsonResource",
 | 
						|
    "ModuleApi",
 | 
						|
    "PRESENCE_ALL_USERS",
 | 
						|
]
 | 
						|
 | 
						|
logger = logging.getLogger(__name__)
 | 
						|
 | 
						|
 | 
						|
@attr.s(auto_attribs=True)
 | 
						|
class UserIpAndAgent:
 | 
						|
    """
 | 
						|
    An IP address and user agent used by a user to connect to this homeserver.
 | 
						|
    """
 | 
						|
 | 
						|
    ip: str
 | 
						|
    user_agent: str
 | 
						|
    # The time at which this user agent/ip was last seen.
 | 
						|
    last_seen: int
 | 
						|
 | 
						|
 | 
						|
class ModuleApi:
 | 
						|
    """A proxy object that gets passed to various plugin modules so they
 | 
						|
    can register new users etc if necessary.
 | 
						|
    """
 | 
						|
 | 
						|
    def __init__(self, hs: "HomeServer", auth_handler):
 | 
						|
        self._hs = hs
 | 
						|
 | 
						|
        self._store = hs.get_datastore()
 | 
						|
        self._auth = hs.get_auth()
 | 
						|
        self._auth_handler = auth_handler
 | 
						|
        self._server_name = hs.hostname
 | 
						|
        self._presence_stream = hs.get_event_sources().sources.presence
 | 
						|
        self._state = hs.get_state_handler()
 | 
						|
        self._clock: Clock = hs.get_clock()
 | 
						|
        self._send_email_handler = hs.get_send_email_handler()
 | 
						|
        self.custom_template_dir = hs.config.server.custom_template_directory
 | 
						|
 | 
						|
        try:
 | 
						|
            app_name = self._hs.config.email.email_app_name
 | 
						|
 | 
						|
            self._from_string = self._hs.config.email.email_notif_from % {
 | 
						|
                "app": app_name
 | 
						|
            }
 | 
						|
        except (KeyError, TypeError):
 | 
						|
            # If substitution failed (which can happen if the string contains
 | 
						|
            # placeholders other than just "app", or if the type of the placeholder is
 | 
						|
            # not a string), fall back to the bare strings.
 | 
						|
            self._from_string = self._hs.config.email.email_notif_from
 | 
						|
 | 
						|
        self._raw_from = email.utils.parseaddr(self._from_string)[1]
 | 
						|
 | 
						|
        # We expose these as properties below in order to attach a helpful docstring.
 | 
						|
        self._http_client: SimpleHttpClient = hs.get_simple_http_client()
 | 
						|
        self._public_room_list_manager = PublicRoomListManager(hs)
 | 
						|
 | 
						|
        self._spam_checker = hs.get_spam_checker()
 | 
						|
        self._account_validity_handler = hs.get_account_validity_handler()
 | 
						|
        self._third_party_event_rules = hs.get_third_party_event_rules()
 | 
						|
        self._presence_router = hs.get_presence_router()
 | 
						|
 | 
						|
    #################################################################################
 | 
						|
    # The following methods should only be called during the module's initialisation.
 | 
						|
 | 
						|
    @property
 | 
						|
    def register_spam_checker_callbacks(self):
 | 
						|
        """Registers callbacks for spam checking capabilities."""
 | 
						|
        return self._spam_checker.register_callbacks
 | 
						|
 | 
						|
    @property
 | 
						|
    def register_account_validity_callbacks(self):
 | 
						|
        """Registers callbacks for account validity capabilities."""
 | 
						|
        return self._account_validity_handler.register_account_validity_callbacks
 | 
						|
 | 
						|
    @property
 | 
						|
    def register_third_party_rules_callbacks(self):
 | 
						|
        """Registers callbacks for third party event rules capabilities."""
 | 
						|
        return self._third_party_event_rules.register_third_party_rules_callbacks
 | 
						|
 | 
						|
    @property
 | 
						|
    def register_presence_router_callbacks(self):
 | 
						|
        """Registers callbacks for presence router capabilities."""
 | 
						|
        return self._presence_router.register_presence_router_callbacks
 | 
						|
 | 
						|
    def register_web_resource(self, path: str, resource: IResource):
 | 
						|
        """Registers a web resource to be served at the given path.
 | 
						|
 | 
						|
        This function should be called during initialisation of the module.
 | 
						|
 | 
						|
        If multiple modules register a resource for the same path, the module that
 | 
						|
        appears the highest in the configuration file takes priority.
 | 
						|
 | 
						|
        Args:
 | 
						|
            path: The path to register the resource for.
 | 
						|
            resource: The resource to attach to this path.
 | 
						|
        """
 | 
						|
        self._hs.register_module_web_resource(path, resource)
 | 
						|
 | 
						|
    #########################################################################
 | 
						|
    # The following methods can be called by the module at any point in time.
 | 
						|
 | 
						|
    @property
 | 
						|
    def http_client(self):
 | 
						|
        """Allows making outbound HTTP requests to remote resources.
 | 
						|
 | 
						|
        An instance of synapse.http.client.SimpleHttpClient
 | 
						|
        """
 | 
						|
        return self._http_client
 | 
						|
 | 
						|
    @property
 | 
						|
    def public_room_list_manager(self):
 | 
						|
        """Allows adding to, removing from and checking the status of rooms in the
 | 
						|
        public room list.
 | 
						|
 | 
						|
        An instance of synapse.module_api.PublicRoomListManager
 | 
						|
        """
 | 
						|
        return self._public_room_list_manager
 | 
						|
 | 
						|
    @property
 | 
						|
    def public_baseurl(self) -> str:
 | 
						|
        """The configured public base URL for this homeserver."""
 | 
						|
        return self._hs.config.server.public_baseurl
 | 
						|
 | 
						|
    @property
 | 
						|
    def email_app_name(self) -> str:
 | 
						|
        """The application name configured in the homeserver's configuration."""
 | 
						|
        return self._hs.config.email.email_app_name
 | 
						|
 | 
						|
    async def get_userinfo_by_id(self, user_id: str) -> Optional[UserInfo]:
 | 
						|
        """Get user info by user_id
 | 
						|
 | 
						|
        Args:
 | 
						|
            user_id: Fully qualified user id.
 | 
						|
        Returns:
 | 
						|
            UserInfo object if a user was found, otherwise None
 | 
						|
        """
 | 
						|
        return await self._store.get_userinfo_by_id(user_id)
 | 
						|
 | 
						|
    async def get_user_by_req(
 | 
						|
        self,
 | 
						|
        req: SynapseRequest,
 | 
						|
        allow_guest: bool = False,
 | 
						|
        allow_expired: bool = False,
 | 
						|
    ) -> Requester:
 | 
						|
        """Check the access_token provided for a request
 | 
						|
 | 
						|
        Args:
 | 
						|
            req: Incoming HTTP request
 | 
						|
            allow_guest: True if guest users should be allowed. If this
 | 
						|
                is False, and the access token is for a guest user, an
 | 
						|
                AuthError will be thrown
 | 
						|
            allow_expired: True if expired users should be allowed. If this
 | 
						|
                is False, and the access token is for an expired user, an
 | 
						|
                AuthError will be thrown
 | 
						|
 | 
						|
        Returns:
 | 
						|
            The requester for this request
 | 
						|
 | 
						|
        Raises:
 | 
						|
            InvalidClientCredentialsError: if no user by that token exists,
 | 
						|
                or the token is invalid.
 | 
						|
        """
 | 
						|
        return await self._auth.get_user_by_req(
 | 
						|
            req,
 | 
						|
            allow_guest,
 | 
						|
            allow_expired=allow_expired,
 | 
						|
        )
 | 
						|
 | 
						|
    async def is_user_admin(self, user_id: str) -> bool:
 | 
						|
        """Checks if a user is a server admin.
 | 
						|
 | 
						|
        Args:
 | 
						|
            user_id: The Matrix ID of the user to check.
 | 
						|
 | 
						|
        Returns:
 | 
						|
            True if the user is a server admin, False otherwise.
 | 
						|
        """
 | 
						|
        return await self._store.is_server_admin(UserID.from_string(user_id))
 | 
						|
 | 
						|
    def get_qualified_user_id(self, username):
 | 
						|
        """Qualify a user id, if necessary
 | 
						|
 | 
						|
        Takes a user id provided by the user and adds the @ and :domain to
 | 
						|
        qualify it, if necessary
 | 
						|
 | 
						|
        Args:
 | 
						|
            username (str): provided user id
 | 
						|
 | 
						|
        Returns:
 | 
						|
            str: qualified @user:id
 | 
						|
        """
 | 
						|
        if username.startswith("@"):
 | 
						|
            return username
 | 
						|
        return UserID(username, self._hs.hostname).to_string()
 | 
						|
 | 
						|
    async def get_profile_for_user(self, localpart: str) -> ProfileInfo:
 | 
						|
        """Look up the profile info for the user with the given localpart.
 | 
						|
 | 
						|
        Args:
 | 
						|
            localpart: The localpart to look up profile information for.
 | 
						|
 | 
						|
        Returns:
 | 
						|
            The profile information (i.e. display name and avatar URL).
 | 
						|
        """
 | 
						|
        return await self._store.get_profileinfo(localpart)
 | 
						|
 | 
						|
    async def get_threepids_for_user(self, user_id: str) -> List[Dict[str, str]]:
 | 
						|
        """Look up the threepids (email addresses and phone numbers) associated with the
 | 
						|
        given Matrix user ID.
 | 
						|
 | 
						|
        Args:
 | 
						|
            user_id: The Matrix user ID to look up threepids for.
 | 
						|
 | 
						|
        Returns:
 | 
						|
            A list of threepids, each threepid being represented by a dictionary
 | 
						|
            containing a "medium" key which value is "email" for email addresses and
 | 
						|
            "msisdn" for phone numbers, and an "address" key which value is the
 | 
						|
            threepid's address.
 | 
						|
        """
 | 
						|
        return await self._store.user_get_threepids(user_id)
 | 
						|
 | 
						|
    def check_user_exists(self, user_id):
 | 
						|
        """Check if user exists.
 | 
						|
 | 
						|
        Args:
 | 
						|
            user_id (str): Complete @user:id
 | 
						|
 | 
						|
        Returns:
 | 
						|
            Deferred[str|None]: Canonical (case-corrected) user_id, or None
 | 
						|
               if the user is not registered.
 | 
						|
        """
 | 
						|
        return defer.ensureDeferred(self._auth_handler.check_user_exists(user_id))
 | 
						|
 | 
						|
    @defer.inlineCallbacks
 | 
						|
    def register(self, localpart, displayname=None, emails: Optional[List[str]] = None):
 | 
						|
        """Registers a new user with given localpart and optional displayname, emails.
 | 
						|
 | 
						|
        Also returns an access token for the new user.
 | 
						|
 | 
						|
        Deprecated: avoid this, as it generates a new device with no way to
 | 
						|
        return that device to the user. Prefer separate calls to register_user and
 | 
						|
        register_device.
 | 
						|
 | 
						|
        Args:
 | 
						|
            localpart (str): The localpart of the new user.
 | 
						|
            displayname (str|None): The displayname of the new user.
 | 
						|
            emails (List[str]): Emails to bind to the new user.
 | 
						|
 | 
						|
        Returns:
 | 
						|
            Deferred[tuple[str, str]]: a 2-tuple of (user_id, access_token)
 | 
						|
        """
 | 
						|
        logger.warning(
 | 
						|
            "Using deprecated ModuleApi.register which creates a dummy user device."
 | 
						|
        )
 | 
						|
        user_id = yield self.register_user(localpart, displayname, emails or [])
 | 
						|
        _, access_token, _, _ = yield self.register_device(user_id)
 | 
						|
        return user_id, access_token
 | 
						|
 | 
						|
    def register_user(
 | 
						|
        self, localpart, displayname=None, emails: Optional[List[str]] = None
 | 
						|
    ):
 | 
						|
        """Registers a new user with given localpart and optional displayname, emails.
 | 
						|
 | 
						|
        Args:
 | 
						|
            localpart (str): The localpart of the new user.
 | 
						|
            displayname (str|None): The displayname of the new user.
 | 
						|
            emails (List[str]): Emails to bind to the new user.
 | 
						|
 | 
						|
        Raises:
 | 
						|
            SynapseError if there is an error performing the registration. Check the
 | 
						|
                'errcode' property for more information on the reason for failure
 | 
						|
 | 
						|
        Returns:
 | 
						|
            defer.Deferred[str]: user_id
 | 
						|
        """
 | 
						|
        return defer.ensureDeferred(
 | 
						|
            self._hs.get_registration_handler().register_user(
 | 
						|
                localpart=localpart,
 | 
						|
                default_display_name=displayname,
 | 
						|
                bind_emails=emails or [],
 | 
						|
            )
 | 
						|
        )
 | 
						|
 | 
						|
    def register_device(self, user_id, device_id=None, initial_display_name=None):
 | 
						|
        """Register a device for a user and generate an access token.
 | 
						|
 | 
						|
        Args:
 | 
						|
            user_id (str): full canonical @user:id
 | 
						|
            device_id (str|None): The device ID to check, or None to generate
 | 
						|
                a new one.
 | 
						|
            initial_display_name (str|None): An optional display name for the
 | 
						|
                device.
 | 
						|
 | 
						|
        Returns:
 | 
						|
            defer.Deferred[tuple[str, str]]: Tuple of device ID and access token
 | 
						|
        """
 | 
						|
        return defer.ensureDeferred(
 | 
						|
            self._hs.get_registration_handler().register_device(
 | 
						|
                user_id=user_id,
 | 
						|
                device_id=device_id,
 | 
						|
                initial_display_name=initial_display_name,
 | 
						|
            )
 | 
						|
        )
 | 
						|
 | 
						|
    def record_user_external_id(
 | 
						|
        self, auth_provider_id: str, remote_user_id: str, registered_user_id: str
 | 
						|
    ) -> defer.Deferred:
 | 
						|
        """Record a mapping from an external user id to a mxid
 | 
						|
 | 
						|
        Args:
 | 
						|
            auth_provider: identifier for the remote auth provider
 | 
						|
            external_id: id on that system
 | 
						|
            user_id: complete mxid that it is mapped to
 | 
						|
        """
 | 
						|
        return defer.ensureDeferred(
 | 
						|
            self._store.record_user_external_id(
 | 
						|
                auth_provider_id, remote_user_id, registered_user_id
 | 
						|
            )
 | 
						|
        )
 | 
						|
 | 
						|
    def generate_short_term_login_token(
 | 
						|
        self,
 | 
						|
        user_id: str,
 | 
						|
        duration_in_ms: int = (2 * 60 * 1000),
 | 
						|
        auth_provider_id: str = "",
 | 
						|
    ) -> str:
 | 
						|
        """Generate a login token suitable for m.login.token authentication
 | 
						|
 | 
						|
        Args:
 | 
						|
            user_id: gives the ID of the user that the token is for
 | 
						|
 | 
						|
            duration_in_ms: the time that the token will be valid for
 | 
						|
 | 
						|
            auth_provider_id: the ID of the SSO IdP that the user used to authenticate
 | 
						|
               to get this token, if any. This is encoded in the token so that
 | 
						|
               /login can report stats on number of successful logins by IdP.
 | 
						|
        """
 | 
						|
        return self._hs.get_macaroon_generator().generate_short_term_login_token(
 | 
						|
            user_id,
 | 
						|
            auth_provider_id,
 | 
						|
            duration_in_ms,
 | 
						|
        )
 | 
						|
 | 
						|
    @defer.inlineCallbacks
 | 
						|
    def invalidate_access_token(self, access_token):
 | 
						|
        """Invalidate an access token for a user
 | 
						|
 | 
						|
        Args:
 | 
						|
            access_token(str): access token
 | 
						|
 | 
						|
        Returns:
 | 
						|
            twisted.internet.defer.Deferred - resolves once the access token
 | 
						|
               has been removed.
 | 
						|
 | 
						|
        Raises:
 | 
						|
            synapse.api.errors.AuthError: the access token is invalid
 | 
						|
        """
 | 
						|
        # see if the access token corresponds to a device
 | 
						|
        user_info = yield defer.ensureDeferred(
 | 
						|
            self._auth.get_user_by_access_token(access_token)
 | 
						|
        )
 | 
						|
        device_id = user_info.get("device_id")
 | 
						|
        user_id = user_info["user"].to_string()
 | 
						|
        if device_id:
 | 
						|
            # delete the device, which will also delete its access tokens
 | 
						|
            yield defer.ensureDeferred(
 | 
						|
                self._hs.get_device_handler().delete_device(user_id, device_id)
 | 
						|
            )
 | 
						|
        else:
 | 
						|
            # no associated device. Just delete the access token.
 | 
						|
            yield defer.ensureDeferred(
 | 
						|
                self._auth_handler.delete_access_token(access_token)
 | 
						|
            )
 | 
						|
 | 
						|
    def run_db_interaction(self, desc, func, *args, **kwargs):
 | 
						|
        """Run a function with a database connection
 | 
						|
 | 
						|
        Args:
 | 
						|
            desc (str): description for the transaction, for metrics etc
 | 
						|
            func (func): function to be run. Passed a database cursor object
 | 
						|
                as well as *args and **kwargs
 | 
						|
            *args: positional args to be passed to func
 | 
						|
            **kwargs: named args to be passed to func
 | 
						|
 | 
						|
        Returns:
 | 
						|
            Deferred[object]: result of func
 | 
						|
        """
 | 
						|
        return defer.ensureDeferred(
 | 
						|
            self._store.db_pool.runInteraction(desc, func, *args, **kwargs)
 | 
						|
        )
 | 
						|
 | 
						|
    def complete_sso_login(
 | 
						|
        self, registered_user_id: str, request: SynapseRequest, client_redirect_url: str
 | 
						|
    ):
 | 
						|
        """Complete a SSO login by redirecting the user to a page to confirm whether they
 | 
						|
        want their access token sent to `client_redirect_url`, or redirect them to that
 | 
						|
        URL with a token directly if the URL matches with one of the whitelisted clients.
 | 
						|
 | 
						|
        This is deprecated in favor of complete_sso_login_async.
 | 
						|
 | 
						|
        Args:
 | 
						|
            registered_user_id: The MXID that has been registered as a previous step of
 | 
						|
                of this SSO login.
 | 
						|
            request: The request to respond to.
 | 
						|
            client_redirect_url: The URL to which to offer to redirect the user (or to
 | 
						|
                redirect them directly if whitelisted).
 | 
						|
        """
 | 
						|
        self._auth_handler._complete_sso_login(
 | 
						|
            registered_user_id,
 | 
						|
            "<unknown>",
 | 
						|
            request,
 | 
						|
            client_redirect_url,
 | 
						|
        )
 | 
						|
 | 
						|
    async def complete_sso_login_async(
 | 
						|
        self,
 | 
						|
        registered_user_id: str,
 | 
						|
        request: SynapseRequest,
 | 
						|
        client_redirect_url: str,
 | 
						|
        new_user: bool = False,
 | 
						|
        auth_provider_id: str = "<unknown>",
 | 
						|
    ):
 | 
						|
        """Complete a SSO login by redirecting the user to a page to confirm whether they
 | 
						|
        want their access token sent to `client_redirect_url`, or redirect them to that
 | 
						|
        URL with a token directly if the URL matches with one of the whitelisted clients.
 | 
						|
 | 
						|
        Args:
 | 
						|
            registered_user_id: The MXID that has been registered as a previous step of
 | 
						|
                of this SSO login.
 | 
						|
            request: The request to respond to.
 | 
						|
            client_redirect_url: The URL to which to offer to redirect the user (or to
 | 
						|
                redirect them directly if whitelisted).
 | 
						|
            new_user: set to true to use wording for the consent appropriate to a user
 | 
						|
                who has just registered.
 | 
						|
            auth_provider_id: the ID of the SSO IdP which was used to log in. This
 | 
						|
                is used to track counts of sucessful logins by IdP.
 | 
						|
        """
 | 
						|
        await self._auth_handler.complete_sso_login(
 | 
						|
            registered_user_id,
 | 
						|
            auth_provider_id,
 | 
						|
            request,
 | 
						|
            client_redirect_url,
 | 
						|
            new_user=new_user,
 | 
						|
        )
 | 
						|
 | 
						|
    @defer.inlineCallbacks
 | 
						|
    def get_state_events_in_room(
 | 
						|
        self, room_id: str, types: Iterable[Tuple[str, Optional[str]]]
 | 
						|
    ) -> Generator[defer.Deferred, Any, Iterable[EventBase]]:
 | 
						|
        """Gets current state events for the given room.
 | 
						|
 | 
						|
        (This is exposed for compatibility with the old SpamCheckerApi. We should
 | 
						|
        probably deprecate it and replace it with an async method in a subclass.)
 | 
						|
 | 
						|
        Args:
 | 
						|
            room_id: The room ID to get state events in.
 | 
						|
            types: The event type and state key (using None
 | 
						|
                to represent 'any') of the room state to acquire.
 | 
						|
 | 
						|
        Returns:
 | 
						|
            twisted.internet.defer.Deferred[list(synapse.events.FrozenEvent)]:
 | 
						|
                The filtered state events in the room.
 | 
						|
        """
 | 
						|
        state_ids = yield defer.ensureDeferred(
 | 
						|
            self._store.get_filtered_current_state_ids(
 | 
						|
                room_id=room_id, state_filter=StateFilter.from_types(types)
 | 
						|
            )
 | 
						|
        )
 | 
						|
        state = yield defer.ensureDeferred(self._store.get_events(state_ids.values()))
 | 
						|
        return state.values()
 | 
						|
 | 
						|
    async def create_and_send_event_into_room(self, event_dict: JsonDict) -> EventBase:
 | 
						|
        """Create and send an event into a room. Membership events are currently not supported.
 | 
						|
 | 
						|
        Args:
 | 
						|
            event_dict: A dictionary representing the event to send.
 | 
						|
                Required keys are `type`, `room_id`, `sender` and `content`.
 | 
						|
 | 
						|
        Returns:
 | 
						|
            The event that was sent. If state event deduplication happened, then
 | 
						|
                the previous, duplicate event instead.
 | 
						|
 | 
						|
        Raises:
 | 
						|
            SynapseError if the event was not allowed.
 | 
						|
        """
 | 
						|
        # Create a requester object
 | 
						|
        requester = create_requester(
 | 
						|
            event_dict["sender"], authenticated_entity=self._server_name
 | 
						|
        )
 | 
						|
 | 
						|
        # Create and send the event
 | 
						|
        (
 | 
						|
            event,
 | 
						|
            _,
 | 
						|
        ) = await self._hs.get_event_creation_handler().create_and_send_nonmember_event(
 | 
						|
            requester,
 | 
						|
            event_dict,
 | 
						|
            ratelimit=False,
 | 
						|
            ignore_shadow_ban=True,
 | 
						|
        )
 | 
						|
 | 
						|
        return event
 | 
						|
 | 
						|
    async def send_local_online_presence_to(self, users: Iterable[str]) -> None:
 | 
						|
        """
 | 
						|
        Forces the equivalent of a presence initial_sync for a set of local or remote
 | 
						|
        users. The users will receive presence for all currently online users that they
 | 
						|
        are considered interested in.
 | 
						|
 | 
						|
        Updates to remote users will be sent immediately, whereas local users will receive
 | 
						|
        them on their next sync attempt.
 | 
						|
 | 
						|
        Note that this method can only be run on the process that is configured to write to the
 | 
						|
        presence stream. By default this is the main process.
 | 
						|
        """
 | 
						|
        if self._hs._instance_name not in self._hs.config.worker.writers.presence:
 | 
						|
            raise Exception(
 | 
						|
                "send_local_online_presence_to can only be run "
 | 
						|
                "on the process that is configured to write to the "
 | 
						|
                "presence stream (by default this is the main process)",
 | 
						|
            )
 | 
						|
 | 
						|
        local_users = set()
 | 
						|
        remote_users = set()
 | 
						|
        for user in users:
 | 
						|
            if self._hs.is_mine_id(user):
 | 
						|
                local_users.add(user)
 | 
						|
            else:
 | 
						|
                remote_users.add(user)
 | 
						|
 | 
						|
        # We pull out the presence handler here to break a cyclic
 | 
						|
        # dependency between the presence router and module API.
 | 
						|
        presence_handler = self._hs.get_presence_handler()
 | 
						|
 | 
						|
        if local_users:
 | 
						|
            # Force a presence initial_sync for these users next time they sync.
 | 
						|
            await presence_handler.send_full_presence_to_users(local_users)
 | 
						|
 | 
						|
        for user in remote_users:
 | 
						|
            # Retrieve presence state for currently online users that this user
 | 
						|
            # is considered interested in.
 | 
						|
            presence_events, _ = await self._presence_stream.get_new_events(
 | 
						|
                UserID.from_string(user), from_key=None, include_offline=False
 | 
						|
            )
 | 
						|
 | 
						|
            # Send to remote destinations.
 | 
						|
            destination = UserID.from_string(user).domain
 | 
						|
            presence_handler.get_federation_queue().send_presence_to_destinations(
 | 
						|
                presence_events, destination
 | 
						|
            )
 | 
						|
 | 
						|
    def looping_background_call(
 | 
						|
        self,
 | 
						|
        f: Callable,
 | 
						|
        msec: float,
 | 
						|
        *args,
 | 
						|
        desc: Optional[str] = None,
 | 
						|
        run_on_all_instances: bool = False,
 | 
						|
        **kwargs,
 | 
						|
    ):
 | 
						|
        """Wraps a function as a background process and calls it repeatedly.
 | 
						|
 | 
						|
        NOTE: Will only run on the instance that is configured to run
 | 
						|
        background processes (which is the main process by default), unless
 | 
						|
        `run_on_all_workers` is set.
 | 
						|
 | 
						|
        Waits `msec` initially before calling `f` for the first time.
 | 
						|
 | 
						|
        Args:
 | 
						|
            f: The function to call repeatedly. f can be either synchronous or
 | 
						|
                asynchronous, and must follow Synapse's logcontext rules.
 | 
						|
                More info about logcontexts is available at
 | 
						|
                https://matrix-org.github.io/synapse/latest/log_contexts.html
 | 
						|
            msec: How long to wait between calls in milliseconds.
 | 
						|
            *args: Positional arguments to pass to function.
 | 
						|
            desc: The background task's description. Default to the function's name.
 | 
						|
            run_on_all_instances: Whether to run this on all instances, rather
 | 
						|
                than just the instance configured to run background tasks.
 | 
						|
            **kwargs: Key arguments to pass to function.
 | 
						|
        """
 | 
						|
        if desc is None:
 | 
						|
            desc = f.__name__
 | 
						|
 | 
						|
        if self._hs.config.worker.run_background_tasks or run_on_all_instances:
 | 
						|
            self._clock.looping_call(
 | 
						|
                run_as_background_process,
 | 
						|
                msec,
 | 
						|
                desc,
 | 
						|
                f,
 | 
						|
                *args,
 | 
						|
                **kwargs,
 | 
						|
            )
 | 
						|
        else:
 | 
						|
            logger.warning(
 | 
						|
                "Not running looping call %s as the configuration forbids it",
 | 
						|
                f,
 | 
						|
            )
 | 
						|
 | 
						|
    async def send_mail(
 | 
						|
        self,
 | 
						|
        recipient: str,
 | 
						|
        subject: str,
 | 
						|
        html: str,
 | 
						|
        text: str,
 | 
						|
    ):
 | 
						|
        """Send an email on behalf of the homeserver.
 | 
						|
 | 
						|
        Args:
 | 
						|
            recipient: The email address for the recipient.
 | 
						|
            subject: The email's subject.
 | 
						|
            html: The email's HTML content.
 | 
						|
            text: The email's text content.
 | 
						|
        """
 | 
						|
        await self._send_email_handler.send_email(
 | 
						|
            email_address=recipient,
 | 
						|
            subject=subject,
 | 
						|
            app_name=self.email_app_name,
 | 
						|
            html=html,
 | 
						|
            text=text,
 | 
						|
        )
 | 
						|
 | 
						|
    def read_templates(
 | 
						|
        self,
 | 
						|
        filenames: List[str],
 | 
						|
        custom_template_directory: Optional[str] = None,
 | 
						|
    ) -> List[jinja2.Template]:
 | 
						|
        """Read and load the content of the template files at the given location.
 | 
						|
        By default, Synapse will look for these templates in its configured template
 | 
						|
        directory, but another directory to search in can be provided.
 | 
						|
 | 
						|
        Args:
 | 
						|
            filenames: The name of the template files to look for.
 | 
						|
            custom_template_directory: An additional directory to look for the files in.
 | 
						|
 | 
						|
        Returns:
 | 
						|
            A list containing the loaded templates, with the orders matching the one of
 | 
						|
            the filenames parameter.
 | 
						|
        """
 | 
						|
        return self._hs.config.read_templates(
 | 
						|
            filenames,
 | 
						|
            (td for td in (self.custom_template_dir, custom_template_directory) if td),
 | 
						|
        )
 | 
						|
 | 
						|
    def is_mine(self, id: Union[str, DomainSpecificString]) -> bool:
 | 
						|
        """
 | 
						|
        Checks whether an ID (user id, room, ...) comes from this homeserver.
 | 
						|
 | 
						|
        Args:
 | 
						|
            id: any Matrix id (e.g. user id, room id, ...), either as a raw id,
 | 
						|
                e.g. string "@user:example.com" or as a parsed UserID, RoomID, ...
 | 
						|
        Returns:
 | 
						|
            True if id comes from this homeserver, False otherwise.
 | 
						|
 | 
						|
        Added in Synapse v1.44.0.
 | 
						|
        """
 | 
						|
        if isinstance(id, DomainSpecificString):
 | 
						|
            return self._hs.is_mine(id)
 | 
						|
        else:
 | 
						|
            return self._hs.is_mine_id(id)
 | 
						|
 | 
						|
    async def get_user_ip_and_agents(
 | 
						|
        self, user_id: str, since_ts: int = 0
 | 
						|
    ) -> List[UserIpAndAgent]:
 | 
						|
        """
 | 
						|
        Return the list of user IPs and agents for a user.
 | 
						|
 | 
						|
        Args:
 | 
						|
            user_id: the id of a user, local or remote
 | 
						|
            since_ts: a timestamp in seconds since the epoch,
 | 
						|
                or the epoch itself if not specified.
 | 
						|
        Returns:
 | 
						|
            The list of all UserIpAndAgent that the user has
 | 
						|
            used to connect to this homeserver since `since_ts`.
 | 
						|
            If the user is remote, this list is empty.
 | 
						|
 | 
						|
        Added in Synapse v1.44.0.
 | 
						|
        """
 | 
						|
        # Don't hit the db if this is not a local user.
 | 
						|
        is_mine = False
 | 
						|
        try:
 | 
						|
            # Let's be defensive against ill-formed strings.
 | 
						|
            if self.is_mine(user_id):
 | 
						|
                is_mine = True
 | 
						|
        except Exception:
 | 
						|
            pass
 | 
						|
 | 
						|
        if is_mine:
 | 
						|
            raw_data = await self._store.get_user_ip_and_agents(
 | 
						|
                UserID.from_string(user_id), since_ts
 | 
						|
            )
 | 
						|
            # Sanitize some of the data. We don't want to return tokens.
 | 
						|
            return [
 | 
						|
                UserIpAndAgent(
 | 
						|
                    ip=str(data["ip"]),
 | 
						|
                    user_agent=str(data["user_agent"]),
 | 
						|
                    last_seen=int(data["last_seen"]),
 | 
						|
                )
 | 
						|
                for data in raw_data
 | 
						|
            ]
 | 
						|
        else:
 | 
						|
            return []
 | 
						|
 | 
						|
 | 
						|
class PublicRoomListManager:
 | 
						|
    """Contains methods for adding to, removing from and querying whether a room
 | 
						|
    is in the public room list.
 | 
						|
    """
 | 
						|
 | 
						|
    def __init__(self, hs: "HomeServer"):
 | 
						|
        self._store = hs.get_datastore()
 | 
						|
 | 
						|
    async def room_is_in_public_room_list(self, room_id: str) -> bool:
 | 
						|
        """Checks whether a room is in the public room list.
 | 
						|
 | 
						|
        Args:
 | 
						|
            room_id: The ID of the room.
 | 
						|
 | 
						|
        Returns:
 | 
						|
            Whether the room is in the public room list. Returns False if the room does
 | 
						|
            not exist.
 | 
						|
        """
 | 
						|
        room = await self._store.get_room(room_id)
 | 
						|
        if not room:
 | 
						|
            return False
 | 
						|
 | 
						|
        return room.get("is_public", False)
 | 
						|
 | 
						|
    async def add_room_to_public_room_list(self, room_id: str) -> None:
 | 
						|
        """Publishes a room to the public room list.
 | 
						|
 | 
						|
        Args:
 | 
						|
            room_id: The ID of the room.
 | 
						|
        """
 | 
						|
        await self._store.set_room_is_public(room_id, True)
 | 
						|
 | 
						|
    async def remove_room_from_public_room_list(self, room_id: str) -> None:
 | 
						|
        """Removes a room from the public room list.
 | 
						|
 | 
						|
        Args:
 | 
						|
            room_id: The ID of the room.
 | 
						|
        """
 | 
						|
        await self._store.set_room_is_public(room_id, False)
 |