test_appservice.py

        Args:
            namespaces: A dictionary containing any user, room or alias namespaces that
                the application service is interested in.

        Returns:
            The registered application service.
        """
        # Create an application service
        appservice = ApplicationService(
            token=random_string(10),
            id=random_string(10),
            sender="@as:example.com",
            rate_limited=False,
            namespaces=namespaces,
            supports_ephemeral=True,
        )

        # Register the application service
        self._services.append(appservice)

        return appservice


class ApplicationServicesHandlerDeviceListsTestCase(unittest.HomeserverTestCase):
    """
    Tests that the ApplicationServicesHandler sends device list updates to application
    services correctly.
    """

    servlets = [
        synapse.rest.admin.register_servlets_for_client_rest_resource,
        login.register_servlets,
        room.register_servlets,
    ]

    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
        # Allow us to modify cached feature flags mid-test
        self.as_handler = hs.get_application_service_handler()

        # Mock ApplicationServiceApi's put_json, so we can verify the raw JSON that
        # will be sent over the wire
        self.put_json = AsyncMock()
        hs.get_application_service_api().put_json = self.put_json  # type: ignore[method-assign]

        # Mock out application services, and allow defining our own in tests
        self._services: List[ApplicationService] = []
        self.hs.get_datastores().main.get_app_services = Mock(  # type: ignore[method-assign]
            return_value=self._services
        )

    # Test across a variety of configuration values
    @parameterized.expand(
        [
            (True, True, True),
            (True, False, False),
            (False, True, False),
            (False, False, False),
        ]
    )
    def test_application_service_receives_device_list_updates(
        self,
        experimental_feature_enabled: bool,
        as_supports_txn_extensions: bool,
        as_should_receive_device_list_updates: bool,
    ) -> None:
        """
        Tests that an application service receives notice of changed device
        lists for a user, when a user changes their device lists.

        Arguments above are populated by parameterized.

        Args:
            as_should_receive_device_list_updates: Whether we expect the AS to receive the
                device list changes.
            experimental_feature_enabled: Whether the "msc3202_transaction_extensions" experimental
                feature is enabled. This feature must be enabled for device lists to ASs to work.
            as_supports_txn_extensions: Whether the application service has explicitly registered
                to receive information defined by MSC3202 - which includes device list changes.
        """
        # Change whether the experimental feature is enabled or disabled before making
        # device list changes
        self.as_handler._msc3202_transaction_extensions_enabled = (
            experimental_feature_enabled
        )

        # Create an appservice that is interested in "local_user"
        appservice = ApplicationService(
            token=random_string(10),
            id=random_string(10),
            sender="@as:example.com",
            rate_limited=False,
            namespaces={
                ApplicationService.NS_USERS: [
                    {
                        "regex": "@local_user:.+",
                        "exclusive": False,
                    }
                ],
            },
            supports_ephemeral=True,
            msc3202_transaction_extensions=as_supports_txn_extensions,
            # Must be set for Synapse to try pushing data to the AS
            hs_token="abcde",
            url="some_url",
        )

        # Register the application service
        self._services.append(appservice)

        # Register a user on the homeserver
        self.local_user = self.register_user("local_user", "password")
        self.local_user_token = self.login("local_user", "password")

        if as_should_receive_device_list_updates:
            # Ensure that the resulting JSON uses the unstable prefix and contains the
            # expected users
            self.put_json.assert_called_once()
            json_body = self.put_json.call_args[1]["json_body"]

            # Our application service should have received a device list update with
            # "local_user" in the "changed" list
            device_list_dict = json_body.get("org.matrix.msc3202.device_lists", {})
            self.assertEqual([], device_list_dict["left"])
            self.assertEqual([self.local_user], device_list_dict["changed"])

        else:
            # No device list changes should have been sent out
            self.put_json.assert_not_called()


class ApplicationServicesHandlerOtkCountsTestCase(unittest.HomeserverTestCase):
    # Argument indices for pulling out arguments from a `send_mock`.
    ARG_OTK_COUNTS = 4
    ARG_FALLBACK_KEYS = 5

    servlets = [
        synapse.rest.admin.register_servlets_for_client_rest_resource,
        login.register_servlets,
        register.register_servlets,
        room.register_servlets,
        sendtodevice.register_servlets,
        receipts.register_servlets,
    ]

    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
        # Mock the ApplicationServiceScheduler's _TransactionController's send method so that
        # we can track what's going out
        self.send_mock = AsyncMock()
        hs.get_application_service_handler().scheduler.txn_ctrl.send = self.send_mock  # type: ignore[method-assign]  # We assign to a method.

        # Define an application service for the tests
        self._service_token = "VERYSECRET"
        self._service = ApplicationService(
            self._service_token,
            "as1",
            "@as.sender:test",
            namespaces={
                "users": [
                    {"regex": "@_as_.*:test", "exclusive": True},
                    {"regex": "@as.sender:test", "exclusive": True},
                ]
            },
            msc3202_transaction_extensions=True,
        )
        self.hs.get_datastores().main.services_cache = [self._service]

        # Register some appservice users
        user_id, device_id = self.register_appservice_user(
            "as.sender", self._service_token
        )
        # With MSC4190 enabled, there will not be a device created
        # during AS registration. However MSC4190 is not enabled
        # in this test. It may become the default behaviour in the
        # future, in which case this test will need to be updated.
        assert device_id is not None
        self._sender_user = user_id
        self._sender_device = device_id

        user_id, device_id = self.register_appservice_user(
            "_as_user1", self._service_token
        )
        assert device_id is not None
        self._namespaced_user = user_id
        self._namespaced_device = device_id

        # Register a real user as well.
        self._real_user = self.register_user("real.user", "meow")
        self._real_user_token = self.login("real.user", "meow")

    async def _add_otks_for_device(
        self, user_id: str, device_id: str, otk_count: int
    ) -> None:
        """
        Add some dummy keys. It doesn't matter if they're not a real algorithm;
        that should be opaque to the server anyway.
        """
        await self.hs.get_datastores().main.add_e2e_one_time_keys(
            user_id,
            device_id,
            self.clock.time_msec(),
            [("algo", f"k{i}", "{}") for i in range(otk_count)],
        )

    async def _add_fallback_key_for_device(
        self, user_id: str, device_id: str, used: bool
    ) -> None:
        """
        Adds a fake fallback key to a device, optionally marking it as used
        right away.
        """
        store = self.hs.get_datastores().main
        await store.set_e2e_fallback_keys(user_id, device_id, {"algo:fk": "fall back!"})
        if used is True:
            # Mark the key as used
            await store.db_pool.simple_update_one(
                table="e2e_fallback_keys_json",
                keyvalues={
                    "user_id": user_id,
                    "device_id": device_id,
                    "algorithm": "algo",
                    "key_id": "fk",
                },
                updatevalues={"used": True},
                desc="_get_fallback_key_set_used",
            )

    def _set_up_devices_and_a_room(self) -> str:
        """
        Helper to set up devices for all the users
        and a room for the users to talk in.
        """

        async def preparation() -> None:
            await self._add_otks_for_device(self._sender_user, self._sender_device, 42)
            await self._add_fallback_key_for_device(
                self._sender_user, self._sender_device, used=True
            )
            await self._add_otks_for_device(
                self._namespaced_user, self._namespaced_device, 36
            )
            await self._add_fallback_key_for_device(
                self._namespaced_user, self._namespaced_device, used=False
            )

            # Register a device for the real user, too, so that we can later ensure
            # that we don't leak information to the AS about the non-AS user.
            await self.hs.get_datastores().main.store_device(
                self._real_user, "REALDEV", "UltraMatrix 3000"
            )
            await self._add_otks_for_device(self._real_user, "REALDEV", 50)

        self.get_success(preparation())

        room_id = self.helper.create_room_as(
            self._real_user, is_public=True, tok=self._real_user_token
        )
        self.helper.join(
            room_id,
            self._namespaced_user,
            tok=self._service_token,
            appservice_user_id=self._namespaced_user,
        )

        # Check it was called for sanity. (This was to send the join event to the AS.)
        self.send_mock.assert_called()
        self.send_mock.reset_mock()

        return room_id

    @override_config(
        {"experimental_features": {"msc3202_transaction_extensions": True}}
    )
    def test_application_services_receive_otk_counts_and_fallback_key_usages_with_pdus(
        self,
    ) -> None:
        """
        Tests that:
        - the AS receives one-time key counts and unused fallback keys for:
            - the specified sender; and
            - any user who is in receipt of the PDUs
        """

        room_id = self._set_up_devices_and_a_room()

        # Send a message into the AS's room
        self.helper.send(room_id, "woof woof", tok=self._real_user_token)

        # Capture what was sent as an AS transaction.
        self.send_mock.assert_called()
        last_args, _last_kwargs = self.send_mock.call_args
        otks: Optional[TransactionOneTimeKeysCount] = last_args[self.ARG_OTK_COUNTS]
        unused_fallbacks: Optional[TransactionUnusedFallbackKeys] = last_args[
            self.ARG_FALLBACK_KEYS
        ]

        self.assertEqual(
            otks,
            {
                "@as.sender:test": {self._sender_device: {"algo": 42}},
                "@_as_user1:test": {self._namespaced_device: {"algo": 36}},
            },
        )
        self.assertEqual(
            unused_fallbacks,
            {
                "@as.sender:test": {self._sender_device: []},
                "@_as_user1:test": {self._namespaced_device: ["algo"]},
            },
        )