Edit on GitHub

mitmproxy.net.server_spec

Server specs are used to describe an upstream proxy or server.

View Source

 1"""
 2Server specs are used to describe an upstream proxy or server.
 3"""
 4import re
 5from functools import cache
 6from typing import Literal
 7
 8from mitmproxy.net import check
 9
10ServerSpec = tuple[
11    Literal["http", "https", "tls", "dtls", "tcp", "udp", "dns"],
12    tuple[str, int]
13]
14
15server_spec_re = re.compile(
16    r"""
17        ^
18        (?:(?P<scheme>\w+)://)?  # scheme is optional
19        (?P<host>[^:/]+|\[.+\])  # hostname can be DNS name, IPv4, or IPv6 address.
20        (?::(?P<port>\d+))?  #  port is optional
21        /?  #  we allow a trailing backslash, but no path
22        $
23        """,
24    re.VERBOSE,
25)
26
27
28@cache
29def parse(server_spec: str, default_scheme: str) -> ServerSpec:
30    """
31    Parses a server mode specification, e.g.:
32
33     - http://example.com/
34     - example.org
35     - example.com:443
36
37    *Raises:*
38     - ValueError, if the server specification is invalid.
39    """
40    m = server_spec_re.match(server_spec)
41    if not m:
42        raise ValueError(f"Invalid server specification: {server_spec}")
43
44    if m.group("scheme"):
45        scheme = m.group("scheme")
46    else:
47        scheme = default_scheme
48    if scheme not in ("http", "https", "tls", "dtls", "tcp", "udp", "dns"):
49        raise ValueError(f"Invalid server scheme: {scheme}")
50
51    host = m.group("host")
52    # IPv6 brackets
53    if host.startswith("[") and host.endswith("]"):
54        host = host[1:-1]
55    if not check.is_valid_host(host):
56        raise ValueError(f"Invalid hostname: {host}")
57
58    if m.group("port"):
59        port = int(m.group("port"))
60    else:
61        try:
62            port = {
63                "http": 80,
64                "https": 443,
65                "dns": 53,
66            }[scheme]
67        except KeyError:
68            raise ValueError(f"Port specification missing.")
69    if not check.is_valid_port(port):
70        raise ValueError(f"Invalid port: {port}")
71
72    return scheme, (host, port)  # type: ignore

@cache

def parse( server_spec: str, default_scheme: str) -> tuple[typing.Literal['http', 'https', 'tls', 'dtls', 'tcp', 'udp', 'dns'], tuple[str, int]]: View Source

29@cache
30def parse(server_spec: str, default_scheme: str) -> ServerSpec:
31    """
32    Parses a server mode specification, e.g.:
33
34     - http://example.com/
35     - example.org
36     - example.com:443
37
38    *Raises:*
39     - ValueError, if the server specification is invalid.
40    """
41    m = server_spec_re.match(server_spec)
42    if not m:
43        raise ValueError(f"Invalid server specification: {server_spec}")
44
45    if m.group("scheme"):
46        scheme = m.group("scheme")
47    else:
48        scheme = default_scheme
49    if scheme not in ("http", "https", "tls", "dtls", "tcp", "udp", "dns"):
50        raise ValueError(f"Invalid server scheme: {scheme}")
51
52    host = m.group("host")
53    # IPv6 brackets
54    if host.startswith("[") and host.endswith("]"):
55        host = host[1:-1]
56    if not check.is_valid_host(host):
57        raise ValueError(f"Invalid hostname: {host}")
58
59    if m.group("port"):
60        port = int(m.group("port"))
61    else:
62        try:
63            port = {
64                "http": 80,
65                "https": 443,
66                "dns": 53,
67            }[scheme]
68        except KeyError:
69            raise ValueError(f"Port specification missing.")
70    if not check.is_valid_port(port):
71        raise ValueError(f"Invalid port: {port}")
72
73    return scheme, (host, port)  # type: ignore

Parses a server mode specification, e.g.:

http://example.com/
example.org
example.com:443

Raises:

ValueError, if the server specification is invalid.