"""
:mod:`websockets.exceptions` defines the following exception hierarchy:
* :exc:`WebSocketException`
* :exc:`ConnectionClosed`
* :exc:`ConnectionClosedError`
* :exc:`ConnectionClosedOK`
* :exc:`InvalidHandshake`
* :exc:`SecurityError`
* :exc:`InvalidMessage`
* :exc:`InvalidHeader`
* :exc:`InvalidHeaderFormat`
* :exc:`InvalidHeaderValue`
* :exc:`InvalidOrigin`
* :exc:`InvalidUpgrade`
* :exc:`InvalidStatusCode`
* :exc:`NegotiationError`
* :exc:`DuplicateParameter`
* :exc:`InvalidParameterName`
* :exc:`InvalidParameterValue`
* :exc:`AbortHandshake`
* :exc:`RedirectHandshake`
* :exc:`InvalidState`
* :exc:`InvalidURI`
* :exc:`PayloadTooBig`
* :exc:`ProtocolError`
"""
import http
from typing import Optional
from .datastructures import Headers, HeadersLike
__all__ = [
"WebSocketException",
"ConnectionClosed",
"ConnectionClosedError",
"ConnectionClosedOK",
"InvalidHandshake",
"SecurityError",
"InvalidMessage",
"InvalidHeader",
"InvalidHeaderFormat",
"InvalidHeaderValue",
"InvalidOrigin",
"InvalidUpgrade",
"InvalidStatusCode",
"NegotiationError",
"DuplicateParameter",
"InvalidParameterName",
"InvalidParameterValue",
"AbortHandshake",
"RedirectHandshake",
"InvalidState",
"InvalidURI",
"PayloadTooBig",
"ProtocolError",
"WebSocketProtocolError",
]
[docs]class WebSocketException(Exception):
"""
Base class for all exceptions defined by :mod:`websockets`.
"""
# See https://www.iana.org/assignments/websocket/websocket.xhtml
CLOSE_CODES = {
1000: "OK",
1001: "going away",
1002: "protocol error",
1003: "unsupported type",
# 1004 is reserved
1005: "no status code [internal]",
1006: "connection closed abnormally [internal]",
1007: "invalid data",
1008: "policy violation",
1009: "message too big",
1010: "extension required",
1011: "unexpected error",
1012: "service restart",
1013: "try again later",
1014: "bad gateway",
1015: "TLS failure [internal]",
}
def format_close(code: int, reason: str) -> str:
"""
Display a human-readable version of the close code and reason.
"""
if 3000 <= code < 4000:
explanation = "registered"
elif 4000 <= code < 5000:
explanation = "private use"
else:
explanation = CLOSE_CODES.get(code, "unknown")
result = f"code = {code} ({explanation}), "
if reason:
result += f"reason = {reason}"
else:
result += "no reason"
return result
[docs]class ConnectionClosed(WebSocketException):
"""
Raised when trying to interact with a closed connection.
Provides the connection close code and reason in its ``code`` and
``reason`` attributes respectively.
"""
def __init__(self, code: int, reason: str) -> None:
self.code = code
self.reason = reason
super().__init__(format_close(code, reason))
[docs]class ConnectionClosedError(ConnectionClosed):
"""
Like :exc:`ConnectionClosed`, when the connection terminated with an error.
This means the close code is different from 1000 (OK) and 1001 (going away).
"""
def __init__(self, code: int, reason: str) -> None:
assert code != 1000 and code != 1001
super().__init__(code, reason)
[docs]class ConnectionClosedOK(ConnectionClosed):
"""
Like :exc:`ConnectionClosed`, when the connection terminated properly.
This means the close code is 1000 (OK) or 1001 (going away).
"""
def __init__(self, code: int, reason: str) -> None:
assert code == 1000 or code == 1001
super().__init__(code, reason)
[docs]class InvalidHandshake(WebSocketException):
"""
Raised during the handshake when the WebSocket connection fails.
"""
[docs]class SecurityError(InvalidHandshake):
"""
Raised when a handshake request or response breaks a security rule.
Security limits are hard coded.
"""
[docs]class InvalidMessage(InvalidHandshake):
"""
Raised when a handshake request or response is malformed.
"""
[docs]class InvalidOrigin(InvalidHeader):
"""
Raised when the Origin header in a request isn't allowed.
"""
def __init__(self, origin: Optional[str]) -> None:
super().__init__("Origin", origin)
[docs]class InvalidUpgrade(InvalidHeader):
"""
Raised when the Upgrade or Connection header isn't correct.
"""
[docs]class InvalidStatusCode(InvalidHandshake):
"""
Raised when a handshake response status code is invalid.
The integer status code is available in the ``status_code`` attribute.
"""
def __init__(self, status_code: int) -> None:
self.status_code = status_code
message = f"server rejected WebSocket connection: HTTP {status_code}"
super().__init__(message)
[docs]class NegotiationError(InvalidHandshake):
"""
Raised when negotiating an extension fails.
"""
[docs]class DuplicateParameter(NegotiationError):
"""
Raised when a parameter name is repeated in an extension header.
"""
def __init__(self, name: str) -> None:
self.name = name
message = f"duplicate parameter: {name}"
super().__init__(message)
[docs]class InvalidParameterName(NegotiationError):
"""
Raised when a parameter name in an extension header is invalid.
"""
def __init__(self, name: str) -> None:
self.name = name
message = f"invalid parameter name: {name}"
super().__init__(message)
[docs]class InvalidParameterValue(NegotiationError):
"""
Raised when a parameter value in an extension header is invalid.
"""
def __init__(self, name: str, value: Optional[str]) -> None:
self.name = name
self.value = value
if value is None:
message = f"missing value for parameter {name}"
elif value == "":
message = f"empty value for parameter {name}"
else:
message = f"invalid value for parameter {name}: {value}"
super().__init__(message)
[docs]class AbortHandshake(InvalidHandshake):
"""
Raised to abort the handshake on purpose and return a HTTP response.
This exception is an implementation detail.
The public API is :meth:`~legacy.server.WebSocketServerProtocol.process_request`.
"""
def __init__(
self,
status: http.HTTPStatus,
headers: HeadersLike,
body: bytes = b"",
) -> None:
self.status = status
self.headers = Headers(headers)
self.body = body
message = f"HTTP {status}, {len(self.headers)} headers, {len(body)} bytes"
super().__init__(message)
[docs]class RedirectHandshake(InvalidHandshake):
"""
Raised when a handshake gets redirected.
This exception is an implementation detail.
"""
def __init__(self, uri: str) -> None:
self.uri = uri
def __str__(self) -> str:
return f"redirect to {self.uri}"
[docs]class InvalidState(WebSocketException, AssertionError):
"""
Raised when an operation is forbidden in the current state.
This exception is an implementation detail.
It should never be raised in normal circumstances.
"""
[docs]class InvalidURI(WebSocketException):
"""
Raised when connecting to an URI that isn't a valid WebSocket URI.
"""
def __init__(self, uri: str) -> None:
self.uri = uri
message = "{} isn't a valid URI".format(uri)
super().__init__(message)
[docs]class PayloadTooBig(WebSocketException):
"""
Raised when receiving a frame with a payload exceeding the maximum size.
"""
[docs]class ProtocolError(WebSocketException):
"""
Raised when a frame breaks the protocol.
"""
WebSocketProtocolError = ProtocolError # for backwards compatibility
