Polina/robot-control-system
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Загрузить файлы в «venv/Lib/site-packages/websockets»

Browse Source
This commit is contained in:
Polina
2026-07-02 21:02:38 +00:00
parent aee4f85836
commit 4aba318813
5 changed files with 1685 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
venv/Lib/site-packages/websockets/connection.py Normal file
View File

@@ -0,0 +1,12 @@
from __future__ import annotations
import warnings
from .protocol import SEND_EOF, Protocol as Connection, Side, State # noqa: F401
warnings.warn( # deprecated in 11.0 - 2023-04-02
"websockets.connection was renamed to websockets.protocol "
"and Connection was renamed to Protocol",
DeprecationWarning,
)

183
venv/Lib/site-packages/websockets/datastructures.py Normal file
View File

@@ -0,0 +1,183 @@
from __future__ import annotations
from collections.abc import Iterable, Iterator, Mapping, MutableMapping
from typing import Any, Protocol
__all__ = [
"Headers",
"HeadersLike",
"MultipleValuesError",
]
class MultipleValuesError(LookupError):
"""
Exception raised when :class:`Headers` has multiple values for a key.
"""
def __str__(self) -> str:
# Implement the same logic as KeyError_str in Objects/exceptions.c.
if len(self.args) == 1:
return repr(self.args[0])
return super().__str__()
class Headers(MutableMapping[str, str]):
"""
Efficient data structure for manipulating HTTP headers.
A :class:`list` of ``(name, values)`` is inefficient for lookups.
A :class:`dict` doesn't suffice because header names are case-insensitive
and multiple occurrences of headers with the same name are possible.
:class:`Headers` stores HTTP headers in a hybrid data structure to provide
efficient insertions and lookups while preserving the original data.
In order to account for multiple values with minimal hassle,
:class:`Headers` follows this logic:
- When getting a header with ``headers[name]``:
- if there's no value, :exc:`KeyError` is raised;
- if there's exactly one value, it's returned;
- if there's more than one value, :exc:`MultipleValuesError` is raised.
- When setting a header with ``headers[name] = value``, the value is
appended to the list of values for that header.
- When deleting a header with ``del headers[name]``, all values for that
header are removed (this is slow).
Other methods for manipulating headers are consistent with this logic.
As long as no header occurs multiple times, :class:`Headers` behaves like
:class:`dict`, except keys are lower-cased to provide case-insensitivity.
Two methods support manipulating multiple values explicitly:
- :meth:`get_all` returns a list of all values for a header;
- :meth:`raw_items` returns an iterator of ``(name, values)`` pairs.
"""
__slots__ = ["_dict", "_list"]
# Like dict, Headers accepts an optional "mapping or iterable" argument.
def __init__(self, *args: HeadersLike, **kwargs: str) -> None:
self._dict: dict[str, list[str]] = {}
self._list: list[tuple[str, str]] = []
self.update(*args, **kwargs)
def __str__(self) -> str:
return "".join(f"{key}: {value}\r\n" for key, value in self._list) + "\r\n"
def __repr__(self) -> str:
return f"{self.__class__.__name__}({self._list!r})"
def copy(self) -> Headers:
copy = self.__class__()
copy._dict = self._dict.copy()
copy._list = self._list.copy()
return copy
def serialize(self) -> bytes:
# Since headers only contain ASCII characters, we can keep this simple.
return str(self).encode()
# Collection methods
def __contains__(self, key: object) -> bool:
return isinstance(key, str) and key.lower() in self._dict
def __iter__(self) -> Iterator[str]:
return iter(self._dict)
def __len__(self) -> int:
return len(self._dict)
# MutableMapping methods
def __getitem__(self, key: str) -> str:
value = self._dict[key.lower()]
if len(value) == 1:
return value[0]
else:
raise MultipleValuesError(key)
def __setitem__(self, key: str, value: str) -> None:
self._dict.setdefault(key.lower(), []).append(value)
self._list.append((key, value))
def __delitem__(self, key: str) -> None:
key_lower = key.lower()
self._dict.__delitem__(key_lower)
# This is inefficient. Fortunately deleting HTTP headers is uncommon.
self._list = [(k, v) for k, v in self._list if k.lower() != key_lower]
def __eq__(self, other: Any) -> bool:
if not isinstance(other, Headers):
return NotImplemented
return self._dict == other._dict
def clear(self) -> None:
"""
Remove all headers.
"""
self._dict = {}
self._list = []
def update(self, *args: HeadersLike, **kwargs: str) -> None:
"""
Update from a :class:`Headers` instance and/or keyword arguments.
"""
args = tuple(
arg.raw_items() if isinstance(arg, Headers) else arg for arg in args
)
super().update(*args, **kwargs)
# Methods for handling multiple values
def get_all(self, key: str) -> list[str]:
"""
Return the (possibly empty) list of all values for a header.
Args:
key: Header name.
"""
return self._dict.get(key.lower(), [])
def raw_items(self) -> Iterator[tuple[str, str]]:
"""
Return an iterator of all values as ``(name, value)`` pairs.
"""
return iter(self._list)
# copy of _typeshed.SupportsKeysAndGetItem.
class SupportsKeysAndGetItem(Protocol): # pragma: no cover
"""
Dict-like types with ``keys() -> str`` and ``__getitem__(key: str) -> str`` methods.
"""
def keys(self) -> Iterable[str]: ...
def __getitem__(self, key: str) -> str: ...
HeadersLike = (
Headers | Mapping[str, str] | Iterable[tuple[str, str]] | SupportsKeysAndGetItem
)
"""
Types accepted where :class:`Headers` is expected.
In addition to :class:`Headers` itself, this includes dict-like types where both
keys and values are :class:`str`.
"""

473
venv/Lib/site-packages/websockets/exceptions.py Normal file
View File

@@ -0,0 +1,473 @@
"""
:mod:`websockets.exceptions` defines the following hierarchy of exceptions.
* :exc:`WebSocketException`
* :exc:`ConnectionClosed`
* :exc:`ConnectionClosedOK`
* :exc:`ConnectionClosedError`
* :exc:`InvalidURI`
* :exc:`InvalidProxy`
* :exc:`InvalidHandshake`
* :exc:`SecurityError`
* :exc:`ProxyError`
* :exc:`InvalidProxyMessage`
* :exc:`InvalidProxyStatus`
* :exc:`InvalidMessage`
* :exc:`InvalidStatus`
* :exc:`InvalidStatusCode` (legacy)
* :exc:`InvalidHeader`
* :exc:`InvalidHeaderFormat`
* :exc:`InvalidHeaderValue`
* :exc:`InvalidOrigin`
* :exc:`InvalidUpgrade`
* :exc:`NegotiationError`
* :exc:`DuplicateParameter`
* :exc:`InvalidParameterName`
* :exc:`InvalidParameterValue`
* :exc:`AbortHandshake` (legacy)
* :exc:`RedirectHandshake` (legacy)
* :exc:`ProtocolError` (Sans-I/O)
* :exc:`PayloadTooBig` (Sans-I/O)
* :exc:`InvalidState` (Sans-I/O)
* :exc:`ConcurrencyError`
"""
from __future__ import annotations
import warnings
from .imports import lazy_import
__all__ = [
"WebSocketException",
"ConnectionClosed",
"ConnectionClosedOK",
"ConnectionClosedError",
"InvalidURI",
"InvalidProxy",
"InvalidHandshake",
"SecurityError",
"ProxyError",
"InvalidProxyMessage",
"InvalidProxyStatus",
"InvalidMessage",
"InvalidStatus",
"InvalidHeader",
"InvalidHeaderFormat",
"InvalidHeaderValue",
"InvalidOrigin",
"InvalidUpgrade",
"NegotiationError",
"DuplicateParameter",
"InvalidParameterName",
"InvalidParameterValue",
"ProtocolError",
"PayloadTooBig",
"InvalidState",
"ConcurrencyError",
]
class WebSocketException(Exception):
"""
Base class for all exceptions defined by websockets.
"""
class ConnectionClosed(WebSocketException):
"""
Raised when trying to interact with a closed connection.
Attributes:
rcvd: If a close frame was received, its code and reason are available
in ``rcvd.code`` and ``rcvd.reason``.
sent: If a close frame was sent, its code and reason are available
in ``sent.code`` and ``sent.reason``.
rcvd_then_sent: If close frames were received and sent, this attribute
tells in which order this happened, from the perspective of this
side of the connection.
"""
def __init__(
self,
rcvd: frames.Close | None,
sent: frames.Close | None,
rcvd_then_sent: bool | None = None,
) -> None:
self.rcvd = rcvd
self.sent = sent
self.rcvd_then_sent = rcvd_then_sent
assert (self.rcvd_then_sent is None) == (self.rcvd is None or self.sent is None)
def __str__(self) -> str:
if self.rcvd is None:
if self.sent is None:
return "no close frame received or sent"
else:
return f"sent {self.sent}; no close frame received"
else:
if self.sent is None:
return f"received {self.rcvd}; no close frame sent"
else:
if self.rcvd_then_sent:
return f"received {self.rcvd}; then sent {self.sent}"
else:
return f"sent {self.sent}; then received {self.rcvd}"
# code and reason attributes are provided for backwards-compatibility
@property
def code(self) -> int:
warnings.warn( # deprecated in 13.1 - 2024-09-21
"ConnectionClosed.code is deprecated; "
"use Protocol.close_code or ConnectionClosed.rcvd.code",
DeprecationWarning,
)
if self.rcvd is None:
return frames.CloseCode.ABNORMAL_CLOSURE
return self.rcvd.code
@property
def reason(self) -> str:
warnings.warn( # deprecated in 13.1 - 2024-09-21
"ConnectionClosed.reason is deprecated; "
"use Protocol.close_reason or ConnectionClosed.rcvd.reason",
DeprecationWarning,
)
if self.rcvd is None:
return ""
return self.rcvd.reason
class ConnectionClosedOK(ConnectionClosed):
"""
Like :exc:`ConnectionClosed`, when the connection terminated properly.
A close code with code 1000 (OK) or 1001 (going away) or without a code was
received and sent.
"""
class ConnectionClosedError(ConnectionClosed):
"""
Like :exc:`ConnectionClosed`, when the connection terminated with an error.
A close frame with a code other than 1000 (OK) or 1001 (going away) was
received or sent, or the closing handshake didn't complete properly.
"""
class InvalidURI(WebSocketException):
"""
Raised when connecting to a URI that isn't a valid WebSocket URI.
"""
def __init__(self, uri: str, msg: str) -> None:
self.uri = uri
self.msg = msg
def __str__(self) -> str:
return f"{self.uri} isn't a valid URI: {self.msg}"
class InvalidProxy(WebSocketException):
"""
Raised when connecting via a proxy that isn't valid.
"""
def __init__(self, proxy: str, msg: str) -> None:
self.proxy = proxy
self.msg = msg
def __str__(self) -> str:
return f"{self.proxy} isn't a valid proxy: {self.msg}"
class InvalidHandshake(WebSocketException):
"""
Base class for exceptions raised when the opening handshake fails.
"""
class SecurityError(InvalidHandshake):
"""
Raised when a handshake request or response breaks a security rule.
Security limits can be configured with :doc:`environment variables
<../reference/variables>`.
"""
class ProxyError(InvalidHandshake):
"""
Raised when failing to connect to a proxy.
"""
class InvalidProxyMessage(ProxyError):
"""
Raised when an HTTP proxy response is malformed.
"""
class InvalidProxyStatus(ProxyError):
"""
Raised when an HTTP proxy rejects the connection.
"""
def __init__(self, response: http11.Response) -> None:
self.response = response
def __str__(self) -> str:
return f"proxy rejected connection: HTTP {self.response.status_code:d}"
class InvalidMessage(InvalidHandshake):
"""
Raised when a handshake request or response is malformed.
"""
class InvalidStatus(InvalidHandshake):
"""
Raised when a handshake response rejects the WebSocket upgrade.
"""
def __init__(self, response: http11.Response) -> None:
self.response = response
def __str__(self) -> str:
return (
f"server rejected WebSocket connection: HTTP {self.response.status_code:d}"
)
class InvalidHeader(InvalidHandshake):
"""
Raised when an HTTP header doesn't have a valid format or value.
"""
def __init__(self, name: str, value: str | None = None) -> None:
self.name = name
self.value = value
def __str__(self) -> str:
if self.value is None:
return f"missing {self.name} header"
elif self.value == "":
return f"empty {self.name} header"
else:
return f"invalid {self.name} header: {self.value}"
class InvalidHeaderFormat(InvalidHeader):
"""
Raised when an HTTP header cannot be parsed.
The format of the header doesn't match the grammar for that header.
"""
def __init__(self, name: str, error: str, header: str, pos: int) -> None:
super().__init__(name, f"{error} at {pos} in {header}")
class InvalidHeaderValue(InvalidHeader):
"""
Raised when an HTTP header has a wrong value.
The format of the header is correct but the value isn't acceptable.
"""
class InvalidOrigin(InvalidHeader):
"""
Raised when the Origin header in a request isn't allowed.
"""
def __init__(self, origin: str | None) -> None:
super().__init__("Origin", origin)
class InvalidUpgrade(InvalidHeader):
"""
Raised when the Upgrade or Connection header isn't correct.
"""
class NegotiationError(InvalidHandshake):
"""
Raised when negotiating an extension or a subprotocol fails.
"""
class DuplicateParameter(NegotiationError):
"""
Raised when a parameter name is repeated in an extension header.
"""
def __init__(self, name: str) -> None:
self.name = name
def __str__(self) -> str:
return f"duplicate parameter: {self.name}"
class InvalidParameterName(NegotiationError):
"""
Raised when a parameter name in an extension header is invalid.
"""
def __init__(self, name: str) -> None:
self.name = name
def __str__(self) -> str:
return f"invalid parameter name: {self.name}"
class InvalidParameterValue(NegotiationError):
"""
Raised when a parameter value in an extension header is invalid.
"""
def __init__(self, name: str, value: str | None) -> None:
self.name = name
self.value = value
def __str__(self) -> str:
if self.value is None:
return f"missing value for parameter {self.name}"
elif self.value == "":
return f"empty value for parameter {self.name}"
else:
return f"invalid value for parameter {self.name}: {self.value}"
class ProtocolError(WebSocketException):
"""
Raised when receiving or sending a frame that breaks the protocol.
The Sans-I/O implementation raises this exception when:
* receiving or sending a frame that contains invalid data;
* receiving or sending an invalid sequence of frames.
"""
class PayloadTooBig(WebSocketException):
"""
Raised when parsing a frame with a payload that exceeds the maximum size.
The Sans-I/O layer uses this exception internally. It doesn't bubble up to
the I/O layer.
The :meth:`~websockets.extensions.Extension.decode` method of extensions
must raise :exc:`PayloadTooBig` if decoding a frame would exceed the limit.
"""
def __init__(
self,
size_or_message: int | None | str,
max_size: int | None = None,
current_size: int | None = None,
) -> None:
if isinstance(size_or_message, str):
assert max_size is None
assert current_size is None
warnings.warn( # deprecated in 14.0 - 2024-11-09
"PayloadTooBig(message) is deprecated; "
"change to PayloadTooBig(size, max_size)",
DeprecationWarning,
)
self.message: str | None = size_or_message
else:
self.message = None
self.size: int | None = size_or_message
assert max_size is not None
self.max_size: int = max_size
self.current_size: int | None = None
self.set_current_size(current_size)
def __str__(self) -> str:
if self.message is not None:
return self.message
else:
message = "frame "
if self.size is not None:
message += f"with {self.size} bytes "
if self.current_size is not None:
message += f"after reading {self.current_size} bytes "
message += f"exceeds limit of {self.max_size} bytes"
return message
def set_current_size(self, current_size: int | None) -> None:
assert self.current_size is None
if current_size is not None:
self.max_size += current_size
self.current_size = current_size
class InvalidState(WebSocketException, AssertionError):
"""
Raised when sending a frame is forbidden in the current state.
Specifically, the Sans-I/O layer raises this exception when:
* sending a data frame to a connection in a state other
:attr:`~websockets.protocol.State.OPEN`;
* sending a control frame to a connection in a state other than
:attr:`~websockets.protocol.State.OPEN` or
:attr:`~websockets.protocol.State.CLOSING`.
"""
class ConcurrencyError(WebSocketException, RuntimeError):
"""
Raised when receiving or sending messages concurrently.
WebSocket is a connection-oriented protocol. Reads must be serialized; so
must be writes. However, reading and writing concurrently is possible.
"""
# At the bottom to break import cycles created by type annotations.
from . import frames, http11 # noqa: E402
lazy_import(
globals(),
deprecated_aliases={
# deprecated in 14.0 - 2024-11-09
"AbortHandshake": ".legacy.exceptions",
"InvalidStatusCode": ".legacy.exceptions",
"RedirectHandshake": ".legacy.exceptions",
"WebSocketProtocolError": ".legacy.exceptions",
},
)

431
venv/Lib/site-packages/websockets/frames.py Normal file
View File

@@ -0,0 +1,431 @@
from __future__ import annotations
import dataclasses
import enum
import io
import os
import secrets
import struct
from collections.abc import Generator, Sequence
from typing import Callable
from .exceptions import PayloadTooBig, ProtocolError
from .typing import BytesLike
try:
from .speedups import apply_mask
except ImportError:
from .utils import apply_mask
__all__ = [
"Opcode",
"OP_CONT",
"OP_TEXT",
"OP_BINARY",
"OP_CLOSE",
"OP_PING",
"OP_PONG",
"DATA_OPCODES",
"CTRL_OPCODES",
"CloseCode",
"Frame",
"Close",
]
class Opcode(enum.IntEnum):
"""Opcode values for WebSocket frames."""
CONT, TEXT, BINARY = 0x00, 0x01, 0x02
CLOSE, PING, PONG = 0x08, 0x09, 0x0A
OP_CONT = Opcode.CONT
OP_TEXT = Opcode.TEXT
OP_BINARY = Opcode.BINARY
OP_CLOSE = Opcode.CLOSE
OP_PING = Opcode.PING
OP_PONG = Opcode.PONG
DATA_OPCODES = OP_CONT, OP_TEXT, OP_BINARY
CTRL_OPCODES = OP_CLOSE, OP_PING, OP_PONG
class CloseCode(enum.IntEnum):
"""Close code values for WebSocket close frames."""
NORMAL_CLOSURE = 1000
GOING_AWAY = 1001
PROTOCOL_ERROR = 1002
UNSUPPORTED_DATA = 1003
# 1004 is reserved
NO_STATUS_RCVD = 1005
ABNORMAL_CLOSURE = 1006
INVALID_DATA = 1007
POLICY_VIOLATION = 1008
MESSAGE_TOO_BIG = 1009
MANDATORY_EXTENSION = 1010
INTERNAL_ERROR = 1011
SERVICE_RESTART = 1012
TRY_AGAIN_LATER = 1013
BAD_GATEWAY = 1014
TLS_HANDSHAKE = 1015
# See https://www.iana.org/assignments/websocket/websocket.xhtml
CLOSE_CODE_EXPLANATIONS: dict[int, str] = {
CloseCode.NORMAL_CLOSURE: "OK",
CloseCode.GOING_AWAY: "going away",
CloseCode.PROTOCOL_ERROR: "protocol error",
CloseCode.UNSUPPORTED_DATA: "unsupported data",
CloseCode.NO_STATUS_RCVD: "no status received [internal]",
CloseCode.ABNORMAL_CLOSURE: "abnormal closure [internal]",
CloseCode.INVALID_DATA: "invalid frame payload data",
CloseCode.POLICY_VIOLATION: "policy violation",
CloseCode.MESSAGE_TOO_BIG: "message too big",
CloseCode.MANDATORY_EXTENSION: "mandatory extension",
CloseCode.INTERNAL_ERROR: "internal error",
CloseCode.SERVICE_RESTART: "service restart",
CloseCode.TRY_AGAIN_LATER: "try again later",
CloseCode.BAD_GATEWAY: "bad gateway",
CloseCode.TLS_HANDSHAKE: "TLS handshake failure [internal]",
}
# Close code that are allowed in a close frame.
# Using a set optimizes `code in EXTERNAL_CLOSE_CODES`.
EXTERNAL_CLOSE_CODES = {
CloseCode.NORMAL_CLOSURE,
CloseCode.GOING_AWAY,
CloseCode.PROTOCOL_ERROR,
CloseCode.UNSUPPORTED_DATA,
CloseCode.INVALID_DATA,
CloseCode.POLICY_VIOLATION,
CloseCode.MESSAGE_TOO_BIG,
CloseCode.MANDATORY_EXTENSION,
CloseCode.INTERNAL_ERROR,
CloseCode.SERVICE_RESTART,
CloseCode.TRY_AGAIN_LATER,
CloseCode.BAD_GATEWAY,
}
OK_CLOSE_CODES = {
CloseCode.NORMAL_CLOSURE,
CloseCode.GOING_AWAY,
CloseCode.NO_STATUS_RCVD,
}
@dataclasses.dataclass
class Frame:
"""
WebSocket frame.
Attributes:
opcode: Opcode.
data: Payload data.
fin: FIN bit.
rsv1: RSV1 bit.
rsv2: RSV2 bit.
rsv3: RSV3 bit.
Only these fields are needed. The MASK bit, payload length and masking-key
are handled on the fly when parsing and serializing frames.
"""
opcode: Opcode
data: BytesLike
fin: bool = True
rsv1: bool = False
rsv2: bool = False
rsv3: bool = False
# Configure if you want to see more in logs. Should be a multiple of 3.
MAX_LOG_SIZE = int(os.environ.get("WEBSOCKETS_MAX_LOG_SIZE", "75"))
def __str__(self) -> str:
"""
Return a human-readable representation of a frame.
"""
coding = None
length = f"{len(self.data)} byte{'' if len(self.data) == 1 else 's'}"
non_final = "" if self.fin else "continued"
if self.opcode is OP_TEXT:
# Decoding only the beginning and the end is needlessly hard.
# Decode the entire payload then elide later if necessary.
data = repr(bytes(self.data).decode())
elif self.opcode is OP_BINARY:
# We'll show at most the first 16 bytes and the last 8 bytes.
# Encode just what we need, plus two dummy bytes to elide later.
binary = self.data
if len(binary) > self.MAX_LOG_SIZE // 3:
cut = (self.MAX_LOG_SIZE // 3 - 1) // 3 # by default cut = 8
binary = b"".join([binary[: 2 * cut], b"\x00\x00", binary[-cut:]])
data = " ".join(f"{byte:02x}" for byte in binary)
elif self.opcode is OP_CLOSE:
data = str(Close.parse(self.data))
elif self.data:
# We don't know if a Continuation frame contains text or binary.
# Ping and Pong frames could contain UTF-8.
# Attempt to decode as UTF-8 and display it as text; fallback to
# binary. If self.data is a memoryview, it has no decode() method,
# which raises AttributeError.
try:
data = repr(bytes(self.data).decode())
coding = "text"
except (UnicodeDecodeError, AttributeError):
binary = self.data
if len(binary) > self.MAX_LOG_SIZE // 3:
cut = (self.MAX_LOG_SIZE // 3 - 1) // 3 # by default cut = 8
binary = b"".join([binary[: 2 * cut], b"\x00\x00", binary[-cut:]])
data = " ".join(f"{byte:02x}" for byte in binary)
coding = "binary"
else:
data = "''"
if len(data) > self.MAX_LOG_SIZE:
cut = self.MAX_LOG_SIZE // 3 - 1 # by default cut = 24
data = data[: 2 * cut] + "..." + data[-cut:]
metadata = ", ".join(filter(None, [coding, length, non_final]))
return f"{self.opcode.name} {data} [{metadata}]"
@classmethod
def parse(
cls,
read_exact: Callable[[int], Generator[None, None, bytes | bytearray]],
*,
mask: bool,
max_size: int | None = None,
extensions: Sequence[extensions.Extension] | None = None,
) -> Generator[None, None, Frame]:
"""
Parse a WebSocket frame.
This is a generator-based coroutine.
Args:
read_exact: Generator-based coroutine that reads the requested
bytes or raises an exception if there isn't enough data.
mask: Whether the frame should be masked i.e. whether the read
happens on the server side.
max_size: Maximum payload size in bytes.
extensions: List of extensions, applied in reverse order.
Raises:
EOFError: If the connection is closed without a full WebSocket frame.
PayloadTooBig: If the frame's payload size exceeds ``max_size``.
ProtocolError: If the frame contains incorrect values.
"""
# Read the header.
data = yield from read_exact(2)
head1, head2 = struct.unpack("!BB", data)
# While not Pythonic, this is marginally faster than calling bool().
fin = True if head1 & 0b10000000 else False
rsv1 = True if head1 & 0b01000000 else False
rsv2 = True if head1 & 0b00100000 else False
rsv3 = True if head1 & 0b00010000 else False
try:
opcode = Opcode(head1 & 0b00001111)
except ValueError as exc:
raise ProtocolError("invalid opcode") from exc
if (True if head2 & 0b10000000 else False) != mask:
raise ProtocolError("incorrect masking")
length = head2 & 0b01111111
if length == 126:
data = yield from read_exact(2)
(length,) = struct.unpack("!H", data)
elif length == 127:
data = yield from read_exact(8)
(length,) = struct.unpack("!Q", data)
if max_size is not None and length > max_size:
raise PayloadTooBig(length, max_size)
if mask:
mask_bytes = yield from read_exact(4)
# Read the data.
data = yield from read_exact(length)
if mask:
data = apply_mask(data, mask_bytes)
frame = cls(opcode, data, fin, rsv1, rsv2, rsv3)
if extensions is None:
extensions = []
for extension in reversed(extensions):
frame = extension.decode(frame, max_size=max_size)
frame.check()
return frame
def serialize(
self,
*,
mask: bool,
extensions: Sequence[extensions.Extension] | None = None,
) -> bytes:
"""
Serialize a WebSocket frame.
Args:
mask: Whether the frame should be masked i.e. whether the write
happens on the client side.
extensions: List of extensions, applied in order.
Raises:
ProtocolError: If the frame contains incorrect values.
"""
self.check()
if extensions is None:
extensions = []
for extension in extensions:
self = extension.encode(self)
output = io.BytesIO()
# Prepare the header.
head1 = (
(0b10000000 if self.fin else 0)
| (0b01000000 if self.rsv1 else 0)
| (0b00100000 if self.rsv2 else 0)
| (0b00010000 if self.rsv3 else 0)
| self.opcode
)
head2 = 0b10000000 if mask else 0
length = len(self.data)
if length < 126:
output.write(struct.pack("!BB", head1, head2 | length))
elif length < 65536:
output.write(struct.pack("!BBH", head1, head2 | 126, length))
else:
output.write(struct.pack("!BBQ", head1, head2 | 127, length))
if mask:
mask_bytes = secrets.token_bytes(4)
output.write(mask_bytes)
# Prepare the data.
data: BytesLike
if mask:
data = apply_mask(self.data, mask_bytes)
else:
data = self.data
output.write(data)
return output.getvalue()
def check(self) -> None:
"""
Check that reserved bits and opcode have acceptable values.
Raises:
ProtocolError: If a reserved bit or the opcode is invalid.
"""
if self.rsv1 or self.rsv2 or self.rsv3:
raise ProtocolError("reserved bits must be 0")
if self.opcode in CTRL_OPCODES:
if len(self.data) > 125:
raise ProtocolError("control frame too long")
if not self.fin:
raise ProtocolError("fragmented control frame")
@dataclasses.dataclass
class Close:
"""
Code and reason for WebSocket close frames.
Attributes:
code: Close code.
reason: Close reason.
"""
code: CloseCode | int
reason: str
def __str__(self) -> str:
"""
Return a human-readable representation of a close code and reason.
"""
if 3000 <= self.code < 4000:
explanation = "registered"
elif 4000 <= self.code < 5000:
explanation = "private use"
else:
explanation = CLOSE_CODE_EXPLANATIONS.get(self.code, "unknown")
result = f"{self.code} ({explanation})"
if self.reason:
result = f"{result} {self.reason}"
return result
@classmethod
def parse(cls, data: BytesLike) -> Close:
"""
Parse the payload of a close frame.
Args:
data: Payload of the close frame.
Raises:
ProtocolError: If data is ill-formed.
UnicodeDecodeError: If the reason isn't valid UTF-8.
"""
if isinstance(data, memoryview):
raise AssertionError("only compressed outgoing frames use memoryview")
if len(data) >= 2:
(code,) = struct.unpack("!H", data[:2])
reason = data[2:].decode()
close = cls(code, reason)
close.check()
return close
elif len(data) == 0:
return cls(CloseCode.NO_STATUS_RCVD, "")
else:
raise ProtocolError("close frame too short")
def serialize(self) -> bytes:
"""
Serialize the payload of a close frame.
"""
self.check()
return struct.pack("!H", self.code) + self.reason.encode()
def check(self) -> None:
"""
Check that the close code has a valid value for a close frame.
Raises:
ProtocolError: If the close code is invalid.
"""
if not (self.code in EXTERNAL_CLOSE_CODES or 3000 <= self.code < 5000):
raise ProtocolError("invalid status code")
# At the bottom to break import cycles created by type annotations.
from . import extensions # noqa: E402

586
venv/Lib/site-packages/websockets/headers.py Normal file
View File

@@ -0,0 +1,586 @@
from __future__ import annotations
import base64
import binascii
import ipaddress
import re
from collections.abc import Sequence
from typing import Callable, TypeVar, cast
from .exceptions import InvalidHeaderFormat, InvalidHeaderValue
from .typing import (
ConnectionOption,
ExtensionHeader,
ExtensionName,
ExtensionParameter,
Subprotocol,
UpgradeProtocol,
)
__all__ = [
"build_host",
"parse_connection",
"parse_upgrade",
"parse_extension",
"build_extension",
"parse_subprotocol",
"build_subprotocol",
"validate_subprotocols",
"build_www_authenticate_basic",
"parse_authorization_basic",
"build_authorization_basic",
]
T = TypeVar("T")
def build_host(
host: str,
port: int,
secure: bool,
*,
always_include_port: bool = False,
) -> str:
"""
Build a ``Host`` header.
"""
# https://datatracker.ietf.org/doc/html/rfc3986#section-3.2.2
# IPv6 addresses must be enclosed in brackets.
try:
address = ipaddress.ip_address(host)
except ValueError:
# host is a hostname
pass
else:
# host is an IP address
if address.version == 6:
host = f"[{host}]"
if always_include_port or port != (443 if secure else 80):
host = f"{host}:{port}"
return host
# To avoid a dependency on a parsing library, we implement manually the ABNF
# described in https://datatracker.ietf.org/doc/html/rfc6455#section-9.1 and
# https://datatracker.ietf.org/doc/html/rfc7230#appendix-B.
def peek_ahead(header: str, pos: int) -> str | None:
"""
Return the next character from ``header`` at the given position.
Return :obj:`None` at the end of ``header``.
We never need to peek more than one character ahead.
"""
return None if pos == len(header) else header[pos]
_OWS_re = re.compile(r"[\t ]*")
def parse_OWS(header: str, pos: int) -> int:
"""
Parse optional whitespace from ``header`` at the given position.
Return the new position.
The whitespace itself isn't returned because it isn't significant.
"""
# There's always a match, possibly empty, whose content doesn't matter.
match = _OWS_re.match(header, pos)
assert match is not None
return match.end()
_token_re = re.compile(r"[-!#$%&\'*+.^_`|~0-9a-zA-Z]+")
def parse_token(header: str, pos: int, header_name: str) -> tuple[str, int]:
"""
Parse a token from ``header`` at the given position.
Return the token value and the new position.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
match = _token_re.match(header, pos)
if match is None:
raise InvalidHeaderFormat(header_name, "expected token", header, pos)
return match.group(), match.end()
_quoted_string_re = re.compile(
r'"(?:[\x09\x20-\x21\x23-\x5b\x5d-\x7e]|\\[\x09\x20-\x7e\x80-\xff])*"'
)
_unquote_re = re.compile(r"\\([\x09\x20-\x7e\x80-\xff])")
def parse_quoted_string(header: str, pos: int, header_name: str) -> tuple[str, int]:
"""
Parse a quoted string from ``header`` at the given position.
Return the unquoted value and the new position.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
match = _quoted_string_re.match(header, pos)
if match is None:
raise InvalidHeaderFormat(header_name, "expected quoted string", header, pos)
return _unquote_re.sub(r"\1", match.group()[1:-1]), match.end()
_quotable_re = re.compile(r"[\x09\x20-\x7e\x80-\xff]*")
_quote_re = re.compile(r"([\x22\x5c])")
def build_quoted_string(value: str) -> str:
"""
Format ``value`` as a quoted string.
This is the reverse of :func:`parse_quoted_string`.
"""
match = _quotable_re.fullmatch(value)
if match is None:
raise ValueError("invalid characters for quoted-string encoding")
return '"' + _quote_re.sub(r"\\\1", value) + '"'
def parse_list(
parse_item: Callable[[str, int, str], tuple[T, int]],
header: str,
pos: int,
header_name: str,
) -> list[T]:
"""
Parse a comma-separated list from ``header`` at the given position.
This is appropriate for parsing values with the following grammar:
1#item
``parse_item`` parses one item.
``header`` is assumed not to start or end with whitespace.
(This function is designed for parsing an entire header value and
:func:`~websockets.http.read_headers` strips whitespace from values.)
Return a list of items.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
# Per https://datatracker.ietf.org/doc/html/rfc7230#section-7, "a recipient
# MUST parse and ignore a reasonable number of empty list elements";
# hence while loops that remove extra delimiters.
# Remove extra delimiters before the first item.
while peek_ahead(header, pos) == ",":
pos = parse_OWS(header, pos + 1)
items = []
while True:
# Loop invariant: a item starts at pos in header.
item, pos = parse_item(header, pos, header_name)
items.append(item)
pos = parse_OWS(header, pos)
# We may have reached the end of the header.
if pos == len(header):
break
# There must be a delimiter after each element except the last one.
if peek_ahead(header, pos) == ",":
pos = parse_OWS(header, pos + 1)
else:
raise InvalidHeaderFormat(header_name, "expected comma", header, pos)
# Remove extra delimiters before the next item.
while peek_ahead(header, pos) == ",":
pos = parse_OWS(header, pos + 1)
# We may have reached the end of the header.
if pos == len(header):
break
# Since we only advance in the header by one character with peek_ahead()
# or with the end position of a regex match, we can't overshoot the end.
assert pos == len(header)
return items
def parse_connection_option(
header: str, pos: int, header_name: str
) -> tuple[ConnectionOption, int]:
"""
Parse a Connection option from ``header`` at the given position.
Return the protocol value and the new position.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
item, pos = parse_token(header, pos, header_name)
return cast(ConnectionOption, item), pos
def parse_connection(header: str) -> list[ConnectionOption]:
"""
Parse a ``Connection`` header.
Return a list of HTTP connection options.
Args
header: value of the ``Connection`` header.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
return parse_list(parse_connection_option, header, 0, "Connection")
_protocol_re = re.compile(
r"[-!#$%&\'*+.^_`|~0-9a-zA-Z]+(?:/[-!#$%&\'*+.^_`|~0-9a-zA-Z]+)?"
)
def parse_upgrade_protocol(
header: str, pos: int, header_name: str
) -> tuple[UpgradeProtocol, int]:
"""
Parse an Upgrade protocol from ``header`` at the given position.
Return the protocol value and the new position.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
match = _protocol_re.match(header, pos)
if match is None:
raise InvalidHeaderFormat(header_name, "expected protocol", header, pos)
return cast(UpgradeProtocol, match.group()), match.end()
def parse_upgrade(header: str) -> list[UpgradeProtocol]:
"""
Parse an ``Upgrade`` header.
Return a list of HTTP protocols.
Args:
header: Value of the ``Upgrade`` header.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
return parse_list(parse_upgrade_protocol, header, 0, "Upgrade")
def parse_extension_item_param(
header: str, pos: int, header_name: str
) -> tuple[ExtensionParameter, int]:
"""
Parse a single extension parameter from ``header`` at the given position.
Return a ``(name, value)`` pair and the new position.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
# Extract parameter name.
name, pos = parse_token(header, pos, header_name)
pos = parse_OWS(header, pos)
# Extract parameter value, if there is one.
value: str | None = None
if peek_ahead(header, pos) == "=":
pos = parse_OWS(header, pos + 1)
if peek_ahead(header, pos) == '"':
pos_before = pos # for proper error reporting below
value, pos = parse_quoted_string(header, pos, header_name)
# https://datatracker.ietf.org/doc/html/rfc6455#section-9.1 says:
# the value after quoted-string unescaping MUST conform to
# the 'token' ABNF.
if _token_re.fullmatch(value) is None:
raise InvalidHeaderFormat(
header_name, "invalid quoted header content", header, pos_before
)
else:
value, pos = parse_token(header, pos, header_name)
pos = parse_OWS(header, pos)
return (name, value), pos
def parse_extension_item(
header: str, pos: int, header_name: str
) -> tuple[ExtensionHeader, int]:
"""
Parse an extension definition from ``header`` at the given position.
Return an ``(extension name, parameters)`` pair, where ``parameters`` is a
list of ``(name, value)`` pairs, and the new position.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
# Extract extension name.
name, pos = parse_token(header, pos, header_name)
pos = parse_OWS(header, pos)
# Extract all parameters.
parameters = []
while peek_ahead(header, pos) == ";":
pos = parse_OWS(header, pos + 1)
parameter, pos = parse_extension_item_param(header, pos, header_name)
parameters.append(parameter)
return (cast(ExtensionName, name), parameters), pos
def parse_extension(header: str) -> list[ExtensionHeader]:
"""
Parse a ``Sec-WebSocket-Extensions`` header.
Return a list of WebSocket extensions and their parameters in this format::
[
(
'extension name',
[
('parameter name', 'parameter value'),
....
]
),
...
]
Parameter values are :obj:`None` when no value is provided.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
return parse_list(parse_extension_item, header, 0, "Sec-WebSocket-Extensions")
parse_extension_list = parse_extension # alias for backwards compatibility
def build_extension_item(
name: ExtensionName, parameters: Sequence[ExtensionParameter]
) -> str:
"""
Build an extension definition.
This is the reverse of :func:`parse_extension_item`.
"""
return "; ".join(
[cast(str, name)]
+ [
# Quoted strings aren't necessary because values are always tokens.
name if value is None else f"{name}={value}"
for name, value in parameters
]
)
def build_extension(extensions: Sequence[ExtensionHeader]) -> str:
"""
Build a ``Sec-WebSocket-Extensions`` header.
This is the reverse of :func:`parse_extension`.
"""
return ", ".join(
build_extension_item(name, parameters) for name, parameters in extensions
)
build_extension_list = build_extension # alias for backwards compatibility
def parse_subprotocol_item(
header: str, pos: int, header_name: str
) -> tuple[Subprotocol, int]:
"""
Parse a subprotocol from ``header`` at the given position.
Return the subprotocol value and the new position.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
item, pos = parse_token(header, pos, header_name)
return cast(Subprotocol, item), pos
def parse_subprotocol(header: str) -> list[Subprotocol]:
"""
Parse a ``Sec-WebSocket-Protocol`` header.
Return a list of WebSocket subprotocols.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
return parse_list(parse_subprotocol_item, header, 0, "Sec-WebSocket-Protocol")
parse_subprotocol_list = parse_subprotocol # alias for backwards compatibility
def build_subprotocol(subprotocols: Sequence[Subprotocol]) -> str:
"""
Build a ``Sec-WebSocket-Protocol`` header.
This is the reverse of :func:`parse_subprotocol`.
"""
return ", ".join(subprotocols)
build_subprotocol_list = build_subprotocol # alias for backwards compatibility
def validate_subprotocols(subprotocols: Sequence[Subprotocol]) -> None:
"""
Validate that ``subprotocols`` is suitable for :func:`build_subprotocol`.
"""
if not isinstance(subprotocols, Sequence):
raise TypeError("subprotocols must be a list")
if isinstance(subprotocols, str):
raise TypeError("subprotocols must be a list, not a str")
for subprotocol in subprotocols:
if not _token_re.fullmatch(subprotocol):
raise ValueError(f"invalid subprotocol: {subprotocol}")
def build_www_authenticate_basic(realm: str) -> str:
"""
Build a ``WWW-Authenticate`` header for HTTP Basic Auth.
Args:
realm: Identifier of the protection space.
"""
# https://datatracker.ietf.org/doc/html/rfc7617#section-2
realm = build_quoted_string(realm)
charset = build_quoted_string("UTF-8")
return f"Basic realm={realm}, charset={charset}"
_token68_re = re.compile(r"[A-Za-z0-9-._~+/]+=*")
def parse_token68(header: str, pos: int, header_name: str) -> tuple[str, int]:
"""
Parse a token68 from ``header`` at the given position.
Return the token value and the new position.
Raises:
InvalidHeaderFormat: On invalid inputs.
"""
match = _token68_re.match(header, pos)
if match is None:
raise InvalidHeaderFormat(header_name, "expected token68", header, pos)
return match.group(), match.end()
def parse_end(header: str, pos: int, header_name: str) -> None:
"""
Check that parsing reached the end of header.
"""
if pos < len(header):
raise InvalidHeaderFormat(header_name, "trailing data", header, pos)
def parse_authorization_basic(header: str) -> tuple[str, str]:
"""
Parse an ``Authorization`` header for HTTP Basic Auth.
Return a ``(username, password)`` tuple.
Args:
header: Value of the ``Authorization`` header.
Raises:
InvalidHeaderFormat: On invalid inputs.
InvalidHeaderValue: On unsupported inputs.
"""
# https://datatracker.ietf.org/doc/html/rfc7235#section-2.1
# https://datatracker.ietf.org/doc/html/rfc7617#section-2
scheme, pos = parse_token(header, 0, "Authorization")
if scheme.lower() != "basic":
raise InvalidHeaderValue(
"Authorization",
f"unsupported scheme: {scheme}",
)
if peek_ahead(header, pos) != " ":
raise InvalidHeaderFormat(
"Authorization", "expected space after scheme", header, pos
)
pos += 1
basic_credentials, pos = parse_token68(header, pos, "Authorization")
parse_end(header, pos, "Authorization")
try:
user_pass = base64.b64decode(basic_credentials.encode()).decode()
except binascii.Error:
raise InvalidHeaderValue(
"Authorization",
"expected base64-encoded credentials",
) from None
try:
username, password = user_pass.split(":", 1)
except ValueError:
raise InvalidHeaderValue(
"Authorization",
"expected username:password credentials",
) from None
return username, password
def build_authorization_basic(username: str, password: str) -> str:
"""
Build an ``Authorization`` header for HTTP Basic Auth.
This is the reverse of :func:`parse_authorization_basic`.
"""
# https://datatracker.ietf.org/doc/html/rfc7617#section-2
assert ":" not in username
user_pass = f"{username}:{password}"
basic_credentials = base64.b64encode(user_pass.encode()).decode()
return "Basic " + basic_credentials