Method process_extensions

src/websockets/client.py:176–247

Handle the Sec-WebSocket-Extensions HTTP response header. Check that each extension is supported, as well as its parameters. :rfc:`6455` leaves the rules up to the specification of each extension. To provide this level of flexibility, for each extension ac

(self, headers: Headers)

        self.subprotocol = self.process_subprotocol(headers)

    def process_extensions(self, headers: Headers) -> list[Extension]:
        """
        Handle the Sec-WebSocket-Extensions HTTP response header.

        Check that each extension is supported, as well as its parameters.

        :rfc:`6455` leaves the rules up to the specification of each
        extension.

        To provide this level of flexibility, for each extension accepted by
        the server, we check for a match with each extension available in the
        client configuration. If no match is found, an exception is raised.

        If several variants of the same extension are accepted by the server,
        it may be configured several times, which won't make sense in general.
        Extensions must implement their own requirements. For this purpose,
        the list of previously accepted extensions is provided.

        Other requirements, for example related to mandatory extensions or the
        order of extensions, may be implemented by overriding this method.

        Args:
            headers: WebSocket handshake response headers.

        Returns:
            List of accepted extensions.

        Raises:
            InvalidHandshake: To abort the handshake.

        """
        accepted_extensions: list[Extension] = []

        extensions = headers.get_all("Sec-WebSocket-Extensions")

        if extensions:
            if self.available_extensions is None:
                raise NegotiationError("no extensions supported")

            parsed_extensions: list[ExtensionHeader] = sum(
                [parse_extension(header_value) for header_value in extensions], []
            )

            for name, response_params in parsed_extensions:
                for extension_factory in self.available_extensions:
                    # Skip non-matching extensions based on their name.
                    if extension_factory.name != name:
                        continue

                    # Skip non-matching extensions based on their params.
                    try:
                        extension = extension_factory.process_response_params(
                            response_params, accepted_extensions
                        )
                    except NegotiationError:
                        continue

                    # Add matching extension to the final list.