Method process_extensions

src/websockets/server.py:312–391

Handle the Sec-WebSocket-Extensions HTTP request header. Accept or reject each extension proposed in the client request. Negotiate parameters for accepted extensions. Per :rfc:`6455`, negotiation rules are defined by the specification of each extension.

(
        self,
        headers: Headers,
    )

        return origin

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

        Accept or reject each extension proposed in the client request.
        Negotiate parameters for accepted extensions.

        Per :rfc:`6455`, negotiation rules are defined by the specification of
        each extension.

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

        If several variants of the same extension are proposed by the client,
        it may be accepted 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.

        This process doesn't allow the server to reorder extensions. It can
        only select a subset of the extensions proposed by the client.

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

        Args:
            headers: WebSocket handshake request headers.

        Returns:
            ``Sec-WebSocket-Extensions`` HTTP response header and list of
            accepted extensions.

        Raises:
            InvalidHandshake: If the Sec-WebSocket-Extensions header is invalid.

        """
        response_header_value: str | None = None

        extension_headers: list[ExtensionHeader] = []
        accepted_extensions: list[Extension] = []

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

        if header_values and self.available_extensions:
            parsed_header_values: list[ExtensionHeader] = sum(
                [parse_extension(header_value) for header_value in header_values], []
            )

            for name, request_params in parsed_header_values:
                for ext_factory in self.available_extensions:
                    # Skip non-matching extensions based on their name.
                    if ext_factory.name != name:
                        continue

                    # Skip non-matching extensions based on their params.