summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/websockets/extensions
diff options
context:
space:
mode:
authorcyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
committercyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
commit12cf076118570eebbff08c6b3090e0d4798447a1 (patch)
tree3ba25e17e3c3a5e82316558ba3864b955919ff72 /venv/lib/python3.11/site-packages/websockets/extensions
parentc45662ff3923b34614ddcc8feb9195541166dcc5 (diff)
no venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/websockets/extensions')
-rw-r--r--venv/lib/python3.11/site-packages/websockets/extensions/__init__.py4
-rw-r--r--venv/lib/python3.11/site-packages/websockets/extensions/__pycache__/__init__.cpython-311.pycbin324 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/websockets/extensions/__pycache__/base.cpython-311.pycbin4689 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/websockets/extensions/__pycache__/permessage_deflate.cpython-311.pycbin19873 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/websockets/extensions/base.py133
-rw-r--r--venv/lib/python3.11/site-packages/websockets/extensions/permessage_deflate.py660
6 files changed, 0 insertions, 797 deletions
diff --git a/venv/lib/python3.11/site-packages/websockets/extensions/__init__.py b/venv/lib/python3.11/site-packages/websockets/extensions/__init__.py
deleted file mode 100644
index 02838b9..0000000
--- a/venv/lib/python3.11/site-packages/websockets/extensions/__init__.py
+++ /dev/null
@@ -1,4 +0,0 @@
-from .base import *
-
-
-__all__ = ["Extension", "ClientExtensionFactory", "ServerExtensionFactory"]
diff --git a/venv/lib/python3.11/site-packages/websockets/extensions/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/websockets/extensions/__pycache__/__init__.cpython-311.pyc
deleted file mode 100644
index 746fb9a..0000000
--- a/venv/lib/python3.11/site-packages/websockets/extensions/__pycache__/__init__.cpython-311.pyc
+++ /dev/null
Binary files differ
diff --git a/venv/lib/python3.11/site-packages/websockets/extensions/__pycache__/base.cpython-311.pyc b/venv/lib/python3.11/site-packages/websockets/extensions/__pycache__/base.cpython-311.pyc
deleted file mode 100644
index 1f40641..0000000
--- a/venv/lib/python3.11/site-packages/websockets/extensions/__pycache__/base.cpython-311.pyc
+++ /dev/null
Binary files differ
diff --git a/venv/lib/python3.11/site-packages/websockets/extensions/__pycache__/permessage_deflate.cpython-311.pyc b/venv/lib/python3.11/site-packages/websockets/extensions/__pycache__/permessage_deflate.cpython-311.pyc
deleted file mode 100644
index 925006a..0000000
--- a/venv/lib/python3.11/site-packages/websockets/extensions/__pycache__/permessage_deflate.cpython-311.pyc
+++ /dev/null
Binary files differ
diff --git a/venv/lib/python3.11/site-packages/websockets/extensions/base.py b/venv/lib/python3.11/site-packages/websockets/extensions/base.py
deleted file mode 100644
index 6c481a4..0000000
--- a/venv/lib/python3.11/site-packages/websockets/extensions/base.py
+++ /dev/null
@@ -1,133 +0,0 @@
-from __future__ import annotations
-
-from typing import List, Optional, Sequence, Tuple
-
-from .. import frames
-from ..typing import ExtensionName, ExtensionParameter
-
-
-__all__ = ["Extension", "ClientExtensionFactory", "ServerExtensionFactory"]
-
-
-class Extension:
- """
- Base class for extensions.
-
- """
-
- name: ExtensionName
- """Extension identifier."""
-
- def decode(
- self,
- frame: frames.Frame,
- *,
- max_size: Optional[int] = None,
- ) -> frames.Frame:
- """
- Decode an incoming frame.
-
- Args:
- frame (Frame): incoming frame.
- max_size: maximum payload size in bytes.
-
- Returns:
- Frame: Decoded frame.
-
- Raises:
- PayloadTooBig: if decoding the payload exceeds ``max_size``.
-
- """
- raise NotImplementedError
-
- def encode(self, frame: frames.Frame) -> frames.Frame:
- """
- Encode an outgoing frame.
-
- Args:
- frame (Frame): outgoing frame.
-
- Returns:
- Frame: Encoded frame.
-
- """
- raise NotImplementedError
-
-
-class ClientExtensionFactory:
- """
- Base class for client-side extension factories.
-
- """
-
- name: ExtensionName
- """Extension identifier."""
-
- def get_request_params(self) -> List[ExtensionParameter]:
- """
- Build parameters to send to the server for this extension.
-
- Returns:
- List[ExtensionParameter]: Parameters to send to the server.
-
- """
- raise NotImplementedError
-
- def process_response_params(
- self,
- params: Sequence[ExtensionParameter],
- accepted_extensions: Sequence[Extension],
- ) -> Extension:
- """
- Process parameters received from the server.
-
- Args:
- params (Sequence[ExtensionParameter]): parameters received from
- the server for this extension.
- accepted_extensions (Sequence[Extension]): list of previously
- accepted extensions.
-
- Returns:
- Extension: An extension instance.
-
- Raises:
- NegotiationError: if parameters aren't acceptable.
-
- """
- raise NotImplementedError
-
-
-class ServerExtensionFactory:
- """
- Base class for server-side extension factories.
-
- """
-
- name: ExtensionName
- """Extension identifier."""
-
- def process_request_params(
- self,
- params: Sequence[ExtensionParameter],
- accepted_extensions: Sequence[Extension],
- ) -> Tuple[List[ExtensionParameter], Extension]:
- """
- Process parameters received from the client.
-
- Args:
- params (Sequence[ExtensionParameter]): parameters received from
- the client for this extension.
- accepted_extensions (Sequence[Extension]): list of previously
- accepted extensions.
-
- Returns:
- Tuple[List[ExtensionParameter], Extension]: To accept the offer,
- parameters to send to the client for this extension and an
- extension instance.
-
- Raises:
- NegotiationError: to reject the offer, if parameters received from
- the client aren't acceptable.
-
- """
- raise NotImplementedError
diff --git a/venv/lib/python3.11/site-packages/websockets/extensions/permessage_deflate.py b/venv/lib/python3.11/site-packages/websockets/extensions/permessage_deflate.py
deleted file mode 100644
index b391837..0000000
--- a/venv/lib/python3.11/site-packages/websockets/extensions/permessage_deflate.py
+++ /dev/null
@@ -1,660 +0,0 @@
-from __future__ import annotations
-
-import dataclasses
-import zlib
-from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
-
-from .. import exceptions, frames
-from ..typing import ExtensionName, ExtensionParameter
-from .base import ClientExtensionFactory, Extension, ServerExtensionFactory
-
-
-__all__ = [
- "PerMessageDeflate",
- "ClientPerMessageDeflateFactory",
- "enable_client_permessage_deflate",
- "ServerPerMessageDeflateFactory",
- "enable_server_permessage_deflate",
-]
-
-_EMPTY_UNCOMPRESSED_BLOCK = b"\x00\x00\xff\xff"
-
-_MAX_WINDOW_BITS_VALUES = [str(bits) for bits in range(8, 16)]
-
-
-class PerMessageDeflate(Extension):
- """
- Per-Message Deflate extension.
-
- """
-
- name = ExtensionName("permessage-deflate")
-
- def __init__(
- self,
- remote_no_context_takeover: bool,
- local_no_context_takeover: bool,
- remote_max_window_bits: int,
- local_max_window_bits: int,
- compress_settings: Optional[Dict[Any, Any]] = None,
- ) -> None:
- """
- Configure the Per-Message Deflate extension.
-
- """
- if compress_settings is None:
- compress_settings = {}
-
- assert remote_no_context_takeover in [False, True]
- assert local_no_context_takeover in [False, True]
- assert 8 <= remote_max_window_bits <= 15
- assert 8 <= local_max_window_bits <= 15
- assert "wbits" not in compress_settings
-
- self.remote_no_context_takeover = remote_no_context_takeover
- self.local_no_context_takeover = local_no_context_takeover
- self.remote_max_window_bits = remote_max_window_bits
- self.local_max_window_bits = local_max_window_bits
- self.compress_settings = compress_settings
-
- if not self.remote_no_context_takeover:
- self.decoder = zlib.decompressobj(wbits=-self.remote_max_window_bits)
-
- if not self.local_no_context_takeover:
- self.encoder = zlib.compressobj(
- wbits=-self.local_max_window_bits, **self.compress_settings
- )
-
- # To handle continuation frames properly, we must keep track of
- # whether that initial frame was encoded.
- self.decode_cont_data = False
- # There's no need for self.encode_cont_data because we always encode
- # outgoing frames, so it would always be True.
-
- def __repr__(self) -> str:
- return (
- f"PerMessageDeflate("
- f"remote_no_context_takeover={self.remote_no_context_takeover}, "
- f"local_no_context_takeover={self.local_no_context_takeover}, "
- f"remote_max_window_bits={self.remote_max_window_bits}, "
- f"local_max_window_bits={self.local_max_window_bits})"
- )
-
- def decode(
- self,
- frame: frames.Frame,
- *,
- max_size: Optional[int] = None,
- ) -> frames.Frame:
- """
- Decode an incoming frame.
-
- """
- # Skip control frames.
- if frame.opcode in frames.CTRL_OPCODES:
- return frame
-
- # Handle continuation data frames:
- # - skip if the message isn't encoded
- # - reset "decode continuation data" flag if it's a final frame
- if frame.opcode is frames.OP_CONT:
- if not self.decode_cont_data:
- return frame
- if frame.fin:
- self.decode_cont_data = False
-
- # Handle text and binary data frames:
- # - skip if the message isn't encoded
- # - unset the rsv1 flag on the first frame of a compressed message
- # - set "decode continuation data" flag if it's a non-final frame
- else:
- if not frame.rsv1:
- return frame
- frame = dataclasses.replace(frame, rsv1=False)
- if not frame.fin:
- self.decode_cont_data = True
-
- # Re-initialize per-message decoder.
- if self.remote_no_context_takeover:
- self.decoder = zlib.decompressobj(wbits=-self.remote_max_window_bits)
-
- # Uncompress data. Protect against zip bombs by preventing zlib from
- # decompressing more than max_length bytes (except when the limit is
- # disabled with max_size = None).
- data = frame.data
- if frame.fin:
- data += _EMPTY_UNCOMPRESSED_BLOCK
- max_length = 0 if max_size is None else max_size
- try:
- data = self.decoder.decompress(data, max_length)
- except zlib.error as exc:
- raise exceptions.ProtocolError("decompression failed") from exc
- if self.decoder.unconsumed_tail:
- raise exceptions.PayloadTooBig(f"over size limit (? > {max_size} bytes)")
-
- # Allow garbage collection of the decoder if it won't be reused.
- if frame.fin and self.remote_no_context_takeover:
- del self.decoder
-
- return dataclasses.replace(frame, data=data)
-
- def encode(self, frame: frames.Frame) -> frames.Frame:
- """
- Encode an outgoing frame.
-
- """
- # Skip control frames.
- if frame.opcode in frames.CTRL_OPCODES:
- return frame
-
- # Since we always encode messages, there's no "encode continuation
- # data" flag similar to "decode continuation data" at this time.
-
- if frame.opcode is not frames.OP_CONT:
- # Set the rsv1 flag on the first frame of a compressed message.
- frame = dataclasses.replace(frame, rsv1=True)
- # Re-initialize per-message decoder.
- if self.local_no_context_takeover:
- self.encoder = zlib.compressobj(
- wbits=-self.local_max_window_bits, **self.compress_settings
- )
-
- # Compress data.
- data = self.encoder.compress(frame.data) + self.encoder.flush(zlib.Z_SYNC_FLUSH)
- if frame.fin and data.endswith(_EMPTY_UNCOMPRESSED_BLOCK):
- data = data[:-4]
-
- # Allow garbage collection of the encoder if it won't be reused.
- if frame.fin and self.local_no_context_takeover:
- del self.encoder
-
- return dataclasses.replace(frame, data=data)
-
-
-def _build_parameters(
- server_no_context_takeover: bool,
- client_no_context_takeover: bool,
- server_max_window_bits: Optional[int],
- client_max_window_bits: Optional[Union[int, bool]],
-) -> List[ExtensionParameter]:
- """
- Build a list of ``(name, value)`` pairs for some compression parameters.
-
- """
- params: List[ExtensionParameter] = []
- if server_no_context_takeover:
- params.append(("server_no_context_takeover", None))
- if client_no_context_takeover:
- params.append(("client_no_context_takeover", None))
- if server_max_window_bits:
- params.append(("server_max_window_bits", str(server_max_window_bits)))
- if client_max_window_bits is True: # only in handshake requests
- params.append(("client_max_window_bits", None))
- elif client_max_window_bits:
- params.append(("client_max_window_bits", str(client_max_window_bits)))
- return params
-
-
-def _extract_parameters(
- params: Sequence[ExtensionParameter], *, is_server: bool
-) -> Tuple[bool, bool, Optional[int], Optional[Union[int, bool]]]:
- """
- Extract compression parameters from a list of ``(name, value)`` pairs.
-
- If ``is_server`` is :obj:`True`, ``client_max_window_bits`` may be
- provided without a value. This is only allowed in handshake requests.
-
- """
- server_no_context_takeover: bool = False
- client_no_context_takeover: bool = False
- server_max_window_bits: Optional[int] = None
- client_max_window_bits: Optional[Union[int, bool]] = None
-
- for name, value in params:
- if name == "server_no_context_takeover":
- if server_no_context_takeover:
- raise exceptions.DuplicateParameter(name)
- if value is None:
- server_no_context_takeover = True
- else:
- raise exceptions.InvalidParameterValue(name, value)
-
- elif name == "client_no_context_takeover":
- if client_no_context_takeover:
- raise exceptions.DuplicateParameter(name)
- if value is None:
- client_no_context_takeover = True
- else:
- raise exceptions.InvalidParameterValue(name, value)
-
- elif name == "server_max_window_bits":
- if server_max_window_bits is not None:
- raise exceptions.DuplicateParameter(name)
- if value in _MAX_WINDOW_BITS_VALUES:
- server_max_window_bits = int(value)
- else:
- raise exceptions.InvalidParameterValue(name, value)
-
- elif name == "client_max_window_bits":
- if client_max_window_bits is not None:
- raise exceptions.DuplicateParameter(name)
- if is_server and value is None: # only in handshake requests
- client_max_window_bits = True
- elif value in _MAX_WINDOW_BITS_VALUES:
- client_max_window_bits = int(value)
- else:
- raise exceptions.InvalidParameterValue(name, value)
-
- else:
- raise exceptions.InvalidParameterName(name)
-
- return (
- server_no_context_takeover,
- client_no_context_takeover,
- server_max_window_bits,
- client_max_window_bits,
- )
-
-
-class ClientPerMessageDeflateFactory(ClientExtensionFactory):
- """
- Client-side extension factory for the Per-Message Deflate extension.
-
- Parameters behave as described in `section 7.1 of RFC 7692`_.
-
- .. _section 7.1 of RFC 7692: https://www.rfc-editor.org/rfc/rfc7692.html#section-7.1
-
- Set them to :obj:`True` to include them in the negotiation offer without a
- value or to an integer value to include them with this value.
-
- Args:
- server_no_context_takeover: prevent server from using context takeover.
- client_no_context_takeover: prevent client from using context takeover.
- server_max_window_bits: maximum size of the server's LZ77 sliding window
- in bits, between 8 and 15.
- client_max_window_bits: maximum size of the client's LZ77 sliding window
- in bits, between 8 and 15, or :obj:`True` to indicate support without
- setting a limit.
- compress_settings: additional keyword arguments for :func:`zlib.compressobj`,
- excluding ``wbits``.
-
- """
-
- name = ExtensionName("permessage-deflate")
-
- def __init__(
- self,
- server_no_context_takeover: bool = False,
- client_no_context_takeover: bool = False,
- server_max_window_bits: Optional[int] = None,
- client_max_window_bits: Optional[Union[int, bool]] = True,
- compress_settings: Optional[Dict[str, Any]] = None,
- ) -> None:
- """
- Configure the Per-Message Deflate extension factory.
-
- """
- if not (server_max_window_bits is None or 8 <= server_max_window_bits <= 15):
- raise ValueError("server_max_window_bits must be between 8 and 15")
- if not (
- client_max_window_bits is None
- or client_max_window_bits is True
- or 8 <= client_max_window_bits <= 15
- ):
- raise ValueError("client_max_window_bits must be between 8 and 15")
- if compress_settings is not None and "wbits" in compress_settings:
- raise ValueError(
- "compress_settings must not include wbits, "
- "set client_max_window_bits instead"
- )
-
- self.server_no_context_takeover = server_no_context_takeover
- self.client_no_context_takeover = client_no_context_takeover
- self.server_max_window_bits = server_max_window_bits
- self.client_max_window_bits = client_max_window_bits
- self.compress_settings = compress_settings
-
- def get_request_params(self) -> List[ExtensionParameter]:
- """
- Build request parameters.
-
- """
- return _build_parameters(
- self.server_no_context_takeover,
- self.client_no_context_takeover,
- self.server_max_window_bits,
- self.client_max_window_bits,
- )
-
- def process_response_params(
- self,
- params: Sequence[ExtensionParameter],
- accepted_extensions: Sequence[Extension],
- ) -> PerMessageDeflate:
- """
- Process response parameters.
-
- Return an extension instance.
-
- """
- if any(other.name == self.name for other in accepted_extensions):
- raise exceptions.NegotiationError(f"received duplicate {self.name}")
-
- # Request parameters are available in instance variables.
-
- # Load response parameters in local variables.
- (
- server_no_context_takeover,
- client_no_context_takeover,
- server_max_window_bits,
- client_max_window_bits,
- ) = _extract_parameters(params, is_server=False)
-
- # After comparing the request and the response, the final
- # configuration must be available in the local variables.
-
- # server_no_context_takeover
- #
- # Req. Resp. Result
- # ------ ------ --------------------------------------------------
- # False False False
- # False True True
- # True False Error!
- # True True True
-
- if self.server_no_context_takeover:
- if not server_no_context_takeover:
- raise exceptions.NegotiationError("expected server_no_context_takeover")
-
- # client_no_context_takeover
- #
- # Req. Resp. Result
- # ------ ------ --------------------------------------------------
- # False False False
- # False True True
- # True False True - must change value
- # True True True
-
- if self.client_no_context_takeover:
- if not client_no_context_takeover:
- client_no_context_takeover = True
-
- # server_max_window_bits
-
- # Req. Resp. Result
- # ------ ------ --------------------------------------------------
- # None None None
- # None 8≤M≤15 M
- # 8≤N≤15 None Error!
- # 8≤N≤15 8≤M≤N M
- # 8≤N≤15 N<M≤15 Error!
-
- if self.server_max_window_bits is None:
- pass
-
- else:
- if server_max_window_bits is None:
- raise exceptions.NegotiationError("expected server_max_window_bits")
- elif server_max_window_bits > self.server_max_window_bits:
- raise exceptions.NegotiationError("unsupported server_max_window_bits")
-
- # client_max_window_bits
-
- # Req. Resp. Result
- # ------ ------ --------------------------------------------------
- # None None None
- # None 8≤M≤15 Error!
- # True None None
- # True 8≤M≤15 M
- # 8≤N≤15 None N - must change value
- # 8≤N≤15 8≤M≤N M
- # 8≤N≤15 N<M≤15 Error!
-
- if self.client_max_window_bits is None:
- if client_max_window_bits is not None:
- raise exceptions.NegotiationError("unexpected client_max_window_bits")
-
- elif self.client_max_window_bits is True:
- pass
-
- else:
- if client_max_window_bits is None:
- client_max_window_bits = self.client_max_window_bits
- elif client_max_window_bits > self.client_max_window_bits:
- raise exceptions.NegotiationError("unsupported client_max_window_bits")
-
- return PerMessageDeflate(
- server_no_context_takeover, # remote_no_context_takeover
- client_no_context_takeover, # local_no_context_takeover
- server_max_window_bits or 15, # remote_max_window_bits
- client_max_window_bits or 15, # local_max_window_bits
- self.compress_settings,
- )
-
-
-def enable_client_permessage_deflate(
- extensions: Optional[Sequence[ClientExtensionFactory]],
-) -> Sequence[ClientExtensionFactory]:
- """
- Enable Per-Message Deflate with default settings in client extensions.
-
- If the extension is already present, perhaps with non-default settings,
- the configuration isn't changed.
-
- """
- if extensions is None:
- extensions = []
- if not any(
- extension_factory.name == ClientPerMessageDeflateFactory.name
- for extension_factory in extensions
- ):
- extensions = list(extensions) + [
- ClientPerMessageDeflateFactory(
- compress_settings={"memLevel": 5},
- )
- ]
- return extensions
-
-
-class ServerPerMessageDeflateFactory(ServerExtensionFactory):
- """
- Server-side extension factory for the Per-Message Deflate extension.
-
- Parameters behave as described in `section 7.1 of RFC 7692`_.
-
- .. _section 7.1 of RFC 7692: https://www.rfc-editor.org/rfc/rfc7692.html#section-7.1
-
- Set them to :obj:`True` to include them in the negotiation offer without a
- value or to an integer value to include them with this value.
-
- Args:
- server_no_context_takeover: prevent server from using context takeover.
- client_no_context_takeover: prevent client from using context takeover.
- server_max_window_bits: maximum size of the server's LZ77 sliding window
- in bits, between 8 and 15.
- client_max_window_bits: maximum size of the client's LZ77 sliding window
- in bits, between 8 and 15.
- compress_settings: additional keyword arguments for :func:`zlib.compressobj`,
- excluding ``wbits``.
- require_client_max_window_bits: do not enable compression at all if
- client doesn't advertise support for ``client_max_window_bits``;
- the default behavior is to enable compression without enforcing
- ``client_max_window_bits``.
-
- """
-
- name = ExtensionName("permessage-deflate")
-
- def __init__(
- self,
- server_no_context_takeover: bool = False,
- client_no_context_takeover: bool = False,
- server_max_window_bits: Optional[int] = None,
- client_max_window_bits: Optional[int] = None,
- compress_settings: Optional[Dict[str, Any]] = None,
- require_client_max_window_bits: bool = False,
- ) -> None:
- """
- Configure the Per-Message Deflate extension factory.
-
- """
- if not (server_max_window_bits is None or 8 <= server_max_window_bits <= 15):
- raise ValueError("server_max_window_bits must be between 8 and 15")
- if not (client_max_window_bits is None or 8 <= client_max_window_bits <= 15):
- raise ValueError("client_max_window_bits must be between 8 and 15")
- if compress_settings is not None and "wbits" in compress_settings:
- raise ValueError(
- "compress_settings must not include wbits, "
- "set server_max_window_bits instead"
- )
- if client_max_window_bits is None and require_client_max_window_bits:
- raise ValueError(
- "require_client_max_window_bits is enabled, "
- "but client_max_window_bits isn't configured"
- )
-
- self.server_no_context_takeover = server_no_context_takeover
- self.client_no_context_takeover = client_no_context_takeover
- self.server_max_window_bits = server_max_window_bits
- self.client_max_window_bits = client_max_window_bits
- self.compress_settings = compress_settings
- self.require_client_max_window_bits = require_client_max_window_bits
-
- def process_request_params(
- self,
- params: Sequence[ExtensionParameter],
- accepted_extensions: Sequence[Extension],
- ) -> Tuple[List[ExtensionParameter], PerMessageDeflate]:
- """
- Process request parameters.
-
- Return response params and an extension instance.
-
- """
- if any(other.name == self.name for other in accepted_extensions):
- raise exceptions.NegotiationError(f"skipped duplicate {self.name}")
-
- # Load request parameters in local variables.
- (
- server_no_context_takeover,
- client_no_context_takeover,
- server_max_window_bits,
- client_max_window_bits,
- ) = _extract_parameters(params, is_server=True)
-
- # Configuration parameters are available in instance variables.
-
- # After comparing the request and the configuration, the response must
- # be available in the local variables.
-
- # server_no_context_takeover
- #
- # Config Req. Resp.
- # ------ ------ --------------------------------------------------
- # False False False
- # False True True
- # True False True - must change value to True
- # True True True
-
- if self.server_no_context_takeover:
- if not server_no_context_takeover:
- server_no_context_takeover = True
-
- # client_no_context_takeover
- #
- # Config Req. Resp.
- # ------ ------ --------------------------------------------------
- # False False False
- # False True True (or False)
- # True False True - must change value to True
- # True True True (or False)
-
- if self.client_no_context_takeover:
- if not client_no_context_takeover:
- client_no_context_takeover = True
-
- # server_max_window_bits
-
- # Config Req. Resp.
- # ------ ------ --------------------------------------------------
- # None None None
- # None 8≤M≤15 M
- # 8≤N≤15 None N - must change value
- # 8≤N≤15 8≤M≤N M
- # 8≤N≤15 N<M≤15 N - must change value
-
- if self.server_max_window_bits is None:
- pass
-
- else:
- if server_max_window_bits is None:
- server_max_window_bits = self.server_max_window_bits
- elif server_max_window_bits > self.server_max_window_bits:
- server_max_window_bits = self.server_max_window_bits
-
- # client_max_window_bits
-
- # Config Req. Resp.
- # ------ ------ --------------------------------------------------
- # None None None
- # None True None - must change value
- # None 8≤M≤15 M (or None)
- # 8≤N≤15 None None or Error!
- # 8≤N≤15 True N - must change value
- # 8≤N≤15 8≤M≤N M (or None)
- # 8≤N≤15 N<M≤15 N
-
- if self.client_max_window_bits is None:
- if client_max_window_bits is True:
- client_max_window_bits = self.client_max_window_bits
-
- else:
- if client_max_window_bits is None:
- if self.require_client_max_window_bits:
- raise exceptions.NegotiationError("required client_max_window_bits")
- elif client_max_window_bits is True:
- client_max_window_bits = self.client_max_window_bits
- elif self.client_max_window_bits < client_max_window_bits:
- client_max_window_bits = self.client_max_window_bits
-
- return (
- _build_parameters(
- server_no_context_takeover,
- client_no_context_takeover,
- server_max_window_bits,
- client_max_window_bits,
- ),
- PerMessageDeflate(
- client_no_context_takeover, # remote_no_context_takeover
- server_no_context_takeover, # local_no_context_takeover
- client_max_window_bits or 15, # remote_max_window_bits
- server_max_window_bits or 15, # local_max_window_bits
- self.compress_settings,
- ),
- )
-
-
-def enable_server_permessage_deflate(
- extensions: Optional[Sequence[ServerExtensionFactory]],
-) -> Sequence[ServerExtensionFactory]:
- """
- Enable Per-Message Deflate with default settings in server extensions.
-
- If the extension is already present, perhaps with non-default settings,
- the configuration isn't changed.
-
- """
- if extensions is None:
- extensions = []
- if not any(
- ext_factory.name == ServerPerMessageDeflateFactory.name
- for ext_factory in extensions
- ):
- extensions = list(extensions) + [
- ServerPerMessageDeflateFactory(
- server_max_window_bits=12,
- client_max_window_bits=12,
- compress_settings={"memLevel": 5},
- )
- ]
- return extensions