1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
|
from __future__ import annotations
import dataclasses
import urllib.parse
from typing import Optional, Tuple
from . import exceptions
__all__ = ["parse_uri", "WebSocketURI"]
@dataclasses.dataclass
class WebSocketURI:
"""
WebSocket URI.
Attributes:
secure: :obj:`True` for a ``wss`` URI, :obj:`False` for a ``ws`` URI.
host: Normalized to lower case.
port: Always set even if it's the default.
path: May be empty.
query: May be empty if the URI doesn't include a query component.
username: Available when the URI contains `User Information`_.
password: Available when the URI contains `User Information`_.
.. _User Information: https://www.rfc-editor.org/rfc/rfc3986.html#section-3.2.1
"""
secure: bool
host: str
port: int
path: str
query: str
username: Optional[str] = None
password: Optional[str] = None
@property
def resource_name(self) -> str:
if self.path:
resource_name = self.path
else:
resource_name = "/"
if self.query:
resource_name += "?" + self.query
return resource_name
@property
def user_info(self) -> Optional[Tuple[str, str]]:
if self.username is None:
return None
assert self.password is not None
return (self.username, self.password)
# All characters from the gen-delims and sub-delims sets in RFC 3987.
DELIMS = ":/?#[]@!$&'()*+,;="
def parse_uri(uri: str) -> WebSocketURI:
"""
Parse and validate a WebSocket URI.
Args:
uri: WebSocket URI.
Returns:
WebSocketURI: Parsed WebSocket URI.
Raises:
InvalidURI: if ``uri`` isn't a valid WebSocket URI.
"""
parsed = urllib.parse.urlparse(uri)
if parsed.scheme not in ["ws", "wss"]:
raise exceptions.InvalidURI(uri, "scheme isn't ws or wss")
if parsed.hostname is None:
raise exceptions.InvalidURI(uri, "hostname isn't provided")
if parsed.fragment != "":
raise exceptions.InvalidURI(uri, "fragment identifier is meaningless")
secure = parsed.scheme == "wss"
host = parsed.hostname
port = parsed.port or (443 if secure else 80)
path = parsed.path
query = parsed.query
username = parsed.username
password = parsed.password
# urllib.parse.urlparse accepts URLs with a username but without a
# password. This doesn't make sense for HTTP Basic Auth credentials.
if username is not None and password is None:
raise exceptions.InvalidURI(uri, "username provided without password")
try:
uri.encode("ascii")
except UnicodeEncodeError:
# Input contains non-ASCII characters.
# It must be an IRI. Convert it to a URI.
host = host.encode("idna").decode()
path = urllib.parse.quote(path, safe=DELIMS)
query = urllib.parse.quote(query, safe=DELIMS)
if username is not None:
assert password is not None
username = urllib.parse.quote(username, safe=DELIMS)
password = urllib.parse.quote(password, safe=DELIMS)
return WebSocketURI(secure, host, port, path, query, username, password)
|
