codekingpro/portable-devtools

1from __future__ import annotations
2
3import base64
4import binascii
5import ipaddress
6import re
7from collections.abc import Sequence
8from typing import Callable, TypeVar, cast
9
10from .exceptions import InvalidHeaderFormat, InvalidHeaderValue
11from .typing import (
12    ConnectionOption,
13    ExtensionHeader,
14    ExtensionName,
15    ExtensionParameter,
16    Subprotocol,
17    UpgradeProtocol,
18)
19
20
21__all__ = [
22    "build_host",
23    "parse_connection",
24    "parse_upgrade",
25    "parse_extension",
26    "build_extension",
27    "parse_subprotocol",
28    "build_subprotocol",
29    "validate_subprotocols",
30    "build_www_authenticate_basic",
31    "parse_authorization_basic",
32    "build_authorization_basic",
33]
34
35
36T = TypeVar("T")
37
38
39def build_host(
40    host: str,
41    port: int,
42    secure: bool,
43    *,
44    always_include_port: bool = False,
45) -> str:
46    """
47    Build a ``Host`` header.
48
49    """
50    # https://datatracker.ietf.org/doc/html/rfc3986#section-3.2.2
51    # IPv6 addresses must be enclosed in brackets.
52    try:
53        address = ipaddress.ip_address(host)
54    except ValueError:
55        # host is a hostname
56        pass
57    else:
58        # host is an IP address
59        if address.version == 6:
60            host = f"[{host}]"
61
62    if always_include_port or port != (443 if secure else 80):
63        host = f"{host}:{port}"
64
65    return host
66
67
68# To avoid a dependency on a parsing library, we implement manually the ABNF
69# described in https://datatracker.ietf.org/doc/html/rfc6455#section-9.1 and
70# https://datatracker.ietf.org/doc/html/rfc7230#appendix-B.
71
72
73def peek_ahead(header: str, pos: int) -> str | None:
74    """
75    Return the next character from ``header`` at the given position.
76
77    Return :obj:`None` at the end of ``header``.
78
79    We never need to peek more than one character ahead.
80
81    """
82    return None if pos == len(header) else header[pos]
83
84
85_OWS_re = re.compile(r"[\t ]*")
86
87
88def parse_OWS(header: str, pos: int) -> int:
89    """
90    Parse optional whitespace from ``header`` at the given position.
91
92    Return the new position.
93
94    The whitespace itself isn't returned because it isn't significant.
95
96    """
97    # There's always a match, possibly empty, whose content doesn't matter.
98    match = _OWS_re.match(header, pos)
99    assert match is not None
100    return match.end()
101
102
103_token_re = re.compile(r"[-!#$%&\'*+.^_`|~0-9a-zA-Z]+")
104
105
106def parse_token(header: str, pos: int, header_name: str) -> tuple[str, int]:
107    """
108    Parse a token from ``header`` at the given position.
109
110    Return the token value and the new position.
111
112    Raises:
113        InvalidHeaderFormat: On invalid inputs.
114
115    """
116    match = _token_re.match(header, pos)
117    if match is None:
118        raise InvalidHeaderFormat(header_name, "expected token", header, pos)
119    return match.group(), match.end()
120
121
122_quoted_string_re = re.compile(
123    r'"(?:[\x09\x20-\x21\x23-\x5b\x5d-\x7e]|\\[\x09\x20-\x7e\x80-\xff])*"'
124)
125
126
127_unquote_re = re.compile(r"\\([\x09\x20-\x7e\x80-\xff])")
128
129
130def parse_quoted_string(header: str, pos: int, header_name: str) -> tuple[str, int]:
131    """
132    Parse a quoted string from ``header`` at the given position.
133
134    Return the unquoted value and the new position.
135
136    Raises:
137        InvalidHeaderFormat: On invalid inputs.
138
139    """
140    match = _quoted_string_re.match(header, pos)
141    if match is None:
142        raise InvalidHeaderFormat(header_name, "expected quoted string", header, pos)
143    return _unquote_re.sub(r"\1", match.group()[1:-1]), match.end()
144
145
146_quotable_re = re.compile(r"[\x09\x20-\x7e\x80-\xff]*")
147
148
149_quote_re = re.compile(r"([\x22\x5c])")
150
151
152def build_quoted_string(value: str) -> str:
153    """
154    Format ``value`` as a quoted string.
155
156    This is the reverse of :func:`parse_quoted_string`.
157
158    """
159    match = _quotable_re.fullmatch(value)
160    if match is None:
161        raise ValueError("invalid characters for quoted-string encoding")
162    return '"' + _quote_re.sub(r"\\\1", value) + '"'
163
164
165def parse_list(
166    parse_item: Callable[[str, int, str], tuple[T, int]],
167    header: str,
168    pos: int,
169    header_name: str,
170) -> list[T]:
171    """
172    Parse a comma-separated list from ``header`` at the given position.
173
174    This is appropriate for parsing values with the following grammar:
175
176        1#item
177
178    ``parse_item`` parses one item.
179
180    ``header`` is assumed not to start or end with whitespace.
181
182    (This function is designed for parsing an entire header value and
183    :func:`~websockets.http.read_headers` strips whitespace from values.)
184
185    Return a list of items.
186
187    Raises:
188        InvalidHeaderFormat: On invalid inputs.
189
190    """
191    # Per https://datatracker.ietf.org/doc/html/rfc7230#section-7, "a recipient
192    # MUST parse and ignore a reasonable number of empty list elements";
193    # hence while loops that remove extra delimiters.
194
195    # Remove extra delimiters before the first item.
196    while peek_ahead(header, pos) == ",":
197        pos = parse_OWS(header, pos + 1)
198
199    items = []
200    while True:
201        # Loop invariant: a item starts at pos in header.
202        item, pos = parse_item(header, pos, header_name)
203        items.append(item)
204        pos = parse_OWS(header, pos)
205
206        # We may have reached the end of the header.
207        if pos == len(header):
208            break
209
210        # There must be a delimiter after each element except the last one.
211        if peek_ahead(header, pos) == ",":
212            pos = parse_OWS(header, pos + 1)
213        else:
214            raise InvalidHeaderFormat(header_name, "expected comma", header, pos)
215
216        # Remove extra delimiters before the next item.
217        while peek_ahead(header, pos) == ",":
218            pos = parse_OWS(header, pos + 1)
219
220        # We may have reached the end of the header.
221        if pos == len(header):
222            break
223
224    # Since we only advance in the header by one character with peek_ahead()
225    # or with the end position of a regex match, we can't overshoot the end.
226    assert pos == len(header)
227
228    return items
229
230
231def parse_connection_option(
232    header: str, pos: int, header_name: str
233) -> tuple[ConnectionOption, int]:
234    """
235    Parse a Connection option from ``header`` at the given position.
236
237    Return the protocol value and the new position.
238
239    Raises:
240        InvalidHeaderFormat: On invalid inputs.
241
242    """
243    item, pos = parse_token(header, pos, header_name)
244    return cast(ConnectionOption, item), pos
245
246
247def parse_connection(header: str) -> list[ConnectionOption]:
248    """
249    Parse a ``Connection`` header.
250
251    Return a list of HTTP connection options.
252
253    Args
254        header: value of the ``Connection`` header.
255
256    Raises:
257        InvalidHeaderFormat: On invalid inputs.
258
259    """
260    return parse_list(parse_connection_option, header, 0, "Connection")
261
262
263_protocol_re = re.compile(
264    r"[-!#$%&\'*+.^_`|~0-9a-zA-Z]+(?:/[-!#$%&\'*+.^_`|~0-9a-zA-Z]+)?"
265)
266
267
268def parse_upgrade_protocol(
269    header: str, pos: int, header_name: str
270) -> tuple[UpgradeProtocol, int]:
271    """
272    Parse an Upgrade protocol from ``header`` at the given position.
273
274    Return the protocol value and the new position.
275
276    Raises:
277        InvalidHeaderFormat: On invalid inputs.
278
279    """
280    match = _protocol_re.match(header, pos)
281    if match is None:
282        raise InvalidHeaderFormat(header_name, "expected protocol", header, pos)
283    return cast(UpgradeProtocol, match.group()), match.end()
284
285
286def parse_upgrade(header: str) -> list[UpgradeProtocol]:
287    """
288    Parse an ``Upgrade`` header.
289
290    Return a list of HTTP protocols.
291
292    Args:
293        header: Value of the ``Upgrade`` header.
294
295    Raises:
296        InvalidHeaderFormat: On invalid inputs.
297
298    """
299    return parse_list(parse_upgrade_protocol, header, 0, "Upgrade")
300
301
302def parse_extension_item_param(
303    header: str, pos: int, header_name: str
304) -> tuple[ExtensionParameter, int]:
305    """
306    Parse a single extension parameter from ``header`` at the given position.
307
308    Return a ``(name, value)`` pair and the new position.
309
310    Raises:
311        InvalidHeaderFormat: On invalid inputs.
312
313    """
314    # Extract parameter name.
315    name, pos = parse_token(header, pos, header_name)
316    pos = parse_OWS(header, pos)
317    # Extract parameter value, if there is one.
318    value: str | None = None
319    if peek_ahead(header, pos) == "=":
320        pos = parse_OWS(header, pos + 1)
321        if peek_ahead(header, pos) == '"':
322            pos_before = pos  # for proper error reporting below
323            value, pos = parse_quoted_string(header, pos, header_name)
324            # https://datatracker.ietf.org/doc/html/rfc6455#section-9.1 says:
325            # the value after quoted-string unescaping MUST conform to
326            # the 'token' ABNF.
327            if _token_re.fullmatch(value) is None:
328                raise InvalidHeaderFormat(
329                    header_name, "invalid quoted header content", header, pos_before
330                )
331        else:
332            value, pos = parse_token(header, pos, header_name)
333        pos = parse_OWS(header, pos)
334
335    return (name, value), pos
336
337
338def parse_extension_item(
339    header: str, pos: int, header_name: str
340) -> tuple[ExtensionHeader, int]:
341    """
342    Parse an extension definition from ``header`` at the given position.
343
344    Return an ``(extension name, parameters)`` pair, where ``parameters`` is a
345    list of ``(name, value)`` pairs, and the new position.
346
347    Raises:
348        InvalidHeaderFormat: On invalid inputs.
349
350    """
351    # Extract extension name.
352    name, pos = parse_token(header, pos, header_name)
353    pos = parse_OWS(header, pos)
354    # Extract all parameters.
355    parameters = []
356    while peek_ahead(header, pos) == ";":
357        pos = parse_OWS(header, pos + 1)
358        parameter, pos = parse_extension_item_param(header, pos, header_name)
359        parameters.append(parameter)
360    return (cast(ExtensionName, name), parameters), pos
361
362
363def parse_extension(header: str) -> list[ExtensionHeader]:
364    """
365    Parse a ``Sec-WebSocket-Extensions`` header.
366
367    Return a list of WebSocket extensions and their parameters in this format::
368
369        [
370            (
371                'extension name',
372                [
373                    ('parameter name', 'parameter value'),
374                    ....
375                ]
376            ),
377            ...
378        ]
379
380    Parameter values are :obj:`None` when no value is provided.
381
382    Raises:
383        InvalidHeaderFormat: On invalid inputs.
384
385    """
386    return parse_list(parse_extension_item, header, 0, "Sec-WebSocket-Extensions")
387
388
389parse_extension_list = parse_extension  # alias for backwards compatibility
390
391
392def build_extension_item(
393    name: ExtensionName, parameters: Sequence[ExtensionParameter]
394) -> str:
395    """
396    Build an extension definition.
397
398    This is the reverse of :func:`parse_extension_item`.
399
400    """
401    return "; ".join(
402        [cast(str, name)]
403        + [
404            # Quoted strings aren't necessary because values are always tokens.
405            name if value is None else f"{name}={value}"
406            for name, value in parameters
407        ]
408    )
409
410
411def build_extension(extensions: Sequence[ExtensionHeader]) -> str:
412    """
413    Build a ``Sec-WebSocket-Extensions`` header.
414
415    This is the reverse of :func:`parse_extension`.
416
417    """
418    return ", ".join(
419        build_extension_item(name, parameters) for name, parameters in extensions
420    )
421
422
423build_extension_list = build_extension  # alias for backwards compatibility
424
425
426def parse_subprotocol_item(
427    header: str, pos: int, header_name: str
428) -> tuple[Subprotocol, int]:
429    """
430    Parse a subprotocol from ``header`` at the given position.
431
432    Return the subprotocol value and the new position.
433
434    Raises:
435        InvalidHeaderFormat: On invalid inputs.
436
437    """
438    item, pos = parse_token(header, pos, header_name)
439    return cast(Subprotocol, item), pos
440
441
442def parse_subprotocol(header: str) -> list[Subprotocol]:
443    """
444    Parse a ``Sec-WebSocket-Protocol`` header.
445
446    Return a list of WebSocket subprotocols.
447
448    Raises:
449        InvalidHeaderFormat: On invalid inputs.
450
451    """
452    return parse_list(parse_subprotocol_item, header, 0, "Sec-WebSocket-Protocol")
453
454
455parse_subprotocol_list = parse_subprotocol  # alias for backwards compatibility
456
457
458def build_subprotocol(subprotocols: Sequence[Subprotocol]) -> str:
459    """
460    Build a ``Sec-WebSocket-Protocol`` header.
461
462    This is the reverse of :func:`parse_subprotocol`.
463
464    """
465    return ", ".join(subprotocols)
466
467
468build_subprotocol_list = build_subprotocol  # alias for backwards compatibility
469
470
471def validate_subprotocols(subprotocols: Sequence[Subprotocol]) -> None:
472    """
473    Validate that ``subprotocols`` is suitable for :func:`build_subprotocol`.
474
475    """
476    if not isinstance(subprotocols, Sequence):
477        raise TypeError("subprotocols must be a list")
478    if isinstance(subprotocols, str):
479        raise TypeError("subprotocols must be a list, not a str")
480    for subprotocol in subprotocols:
481        if not _token_re.fullmatch(subprotocol):
482            raise ValueError(f"invalid subprotocol: {subprotocol}")
483
484
485def build_www_authenticate_basic(realm: str) -> str:
486    """
487    Build a ``WWW-Authenticate`` header for HTTP Basic Auth.
488
489    Args:
490        realm: Identifier of the protection space.
491
492    """
493    # https://datatracker.ietf.org/doc/html/rfc7617#section-2
494    realm = build_quoted_string(realm)
495    charset = build_quoted_string("UTF-8")
496    return f"Basic realm={realm}, charset={charset}"
497
498
499_token68_re = re.compile(r"[A-Za-z0-9-._~+/]+=*")
500
501
502def parse_token68(header: str, pos: int, header_name: str) -> tuple[str, int]:
503    """
504    Parse a token68 from ``header`` at the given position.
505
506    Return the token value and the new position.
507
508    Raises:
509        InvalidHeaderFormat: On invalid inputs.
510
511    """
512    match = _token68_re.match(header, pos)
513    if match is None:
514        raise InvalidHeaderFormat(header_name, "expected token68", header, pos)
515    return match.group(), match.end()
516
517
518def parse_end(header: str, pos: int, header_name: str) -> None:
519    """
520    Check that parsing reached the end of header.
521
522    """
523    if pos < len(header):
524        raise InvalidHeaderFormat(header_name, "trailing data", header, pos)
525
526
527def parse_authorization_basic(header: str) -> tuple[str, str]:
528    """
529    Parse an ``Authorization`` header for HTTP Basic Auth.
530
531    Return a ``(username, password)`` tuple.
532
533    Args:
534        header: Value of the ``Authorization`` header.
535
536    Raises:
537        InvalidHeaderFormat: On invalid inputs.
538        InvalidHeaderValue: On unsupported inputs.
539
540    """
541    # https://datatracker.ietf.org/doc/html/rfc7235#section-2.1
542    # https://datatracker.ietf.org/doc/html/rfc7617#section-2
543    scheme, pos = parse_token(header, 0, "Authorization")
544    if scheme.lower() != "basic":
545        raise InvalidHeaderValue(
546            "Authorization",
547            f"unsupported scheme: {scheme}",
548        )
549    if peek_ahead(header, pos) != " ":
550        raise InvalidHeaderFormat(
551            "Authorization", "expected space after scheme", header, pos
552        )
553    pos += 1
554    basic_credentials, pos = parse_token68(header, pos, "Authorization")
555    parse_end(header, pos, "Authorization")
556
557    try:
558        user_pass = base64.b64decode(basic_credentials.encode()).decode()
559    except binascii.Error:
560        raise InvalidHeaderValue(
561            "Authorization",
562            "expected base64-encoded credentials",
563        ) from None
564    try:
565        username, password = user_pass.split(":", 1)
566    except ValueError:
567        raise InvalidHeaderValue(
568            "Authorization",
569            "expected username:password credentials",
570        ) from None
571
572    return username, password
573
574
575def build_authorization_basic(username: str, password: str) -> str:
576    """
577    Build an ``Authorization`` header for HTTP Basic Auth.
578
579    This is the reverse of :func:`parse_authorization_basic`.
580
581    """
582    # https://datatracker.ietf.org/doc/html/rfc7617#section-2
583    assert ":" not in username
584    user_pass = f"{username}:{password}"
585    basic_credentials = base64.b64encode(user_pass.encode()).decode()
586    return "Basic " + basic_credentials
587