testing/web-platform/tests/tools/wptserve/wptserve/utils.py

   1 import socket
   2 from typing import AnyStr, Dict, List, TypeVar
   3
   4 from .logger import get_logger
   5
   6 KT = TypeVar('KT')
   7 VT = TypeVar('VT')
   8
   9
  10 def isomorphic_decode(s: AnyStr) -> str:
  11     """Decodes a binary string into a text string using iso-8859-1.
  12
  13     Returns `str`. The function is a no-op if the argument already has a text
  14     type. iso-8859-1 is chosen because it is an 8-bit encoding whose code
  15     points range from 0x0 to 0xFF and the values are the same as the binary
  16     representations, so any binary string can be decoded into and encoded from
  17     iso-8859-1 without any errors or data loss. Python 3 also uses iso-8859-1
  18     (or latin-1) extensively in http:
  19     https://github.com/python/cpython/blob/273fc220b25933e443c82af6888eb1871d032fb8/Lib/http/client.py#L213
  20     """
  21     if isinstance(s, str):
  22         return s
  23
  24     if isinstance(s, bytes):
  25         return s.decode("iso-8859-1")
  26
  27     raise TypeError("Unexpected value (expecting string-like): %r" % s)
  28
  29
  30 def isomorphic_encode(s: AnyStr) -> bytes:
  31     """Encodes a text-type string into binary data using iso-8859-1.
  32
  33     Returns `bytes`. The function is a no-op if the argument already has a
  34     binary type. This is the counterpart of isomorphic_decode.
  35     """
  36     if isinstance(s, bytes):
  37         return s
  38
  39     if isinstance(s, str):
  40         return s.encode("iso-8859-1")
  41
  42     raise TypeError("Unexpected value (expecting string-like): %r" % s)
  43
  44
  45 def invert_dict(dict: Dict[KT, List[VT]]) -> Dict[VT, KT]:
  46     rv = {}
  47     for key, values in dict.items():
  48         for value in values:
  49             if value in rv:
  50                 raise ValueError
  51             rv[value] = key
  52     return rv
  53
  54
  55 class HTTPException(Exception):
  56     def __init__(self, code: int, message: str = ""):
  57         self.code = code
  58         self.message = message
  59
  60
  61 def _open_socket(host: str, port: int) -> socket.socket:
  62     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  63     if port != 0:
  64         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  65     sock.bind((host, port))
  66     sock.listen(5)
  67     return sock
  68
  69
  70 def is_bad_port(port: int) -> bool:
  71     """
  72     Bad port as per https://fetch.spec.whatwg.org/#port-blocking
  73     """
  74     return port in [
  75         1,     # tcpmux
  76         7,     # echo
  77         9,     # discard
  78         11,    # systat
  79         13,    # daytime
  80         15,    # netstat
  81         17,    # qotd
  82         19,    # chargen
  83         20,    # ftp-data
  84         21,    # ftp
  85         22,    # ssh
  86         23,    # telnet
  87         25,    # smtp
  88         37,    # time
  89         42,    # name
  90         43,    # nicname
  91         53,    # domain
  92         69,    # tftp
  93         77,    # priv-rjs
  94         79,    # finger
  95         87,    # ttylink
  96         95,    # supdup
  97         101,   # hostriame
  98         102,   # iso-tsap
  99         103,   # gppitnp
 100         104,   # acr-nema
 101         109,   # pop2
 102         110,   # pop3
 103         111,   # sunrpc
 104         113,   # auth
 105         115,   # sftp
 106         117,   # uucp-path
 107         119,   # nntp
 108         123,   # ntp
 109         135,   # loc-srv / epmap
 110         137,   # netbios-ns
 111         139,   # netbios-ssn
 112         143,   # imap2
 113         161,   # snmp
 114         179,   # bgp
 115         389,   # ldap
 116         427,   # afp (alternate)
 117         465,   # smtp (alternate)
 118         512,   # print / exec
 119         513,   # login
 120         514,   # shell
 121         515,   # printer
 122         526,   # tempo
 123         530,   # courier
 124         531,   # chat
 125         532,   # netnews
 126         540,   # uucp
 127         548,   # afp
 128         554,   # rtsp
 129         556,   # remotefs
 130         563,   # nntp+ssl
 131         587,   # smtp (outgoing)
 132         601,   # syslog-conn
 133         636,   # ldap+ssl
 134         989,   # ftps-data
 135         999,   # ftps
 136         993,   # ldap+ssl
 137         995,   # pop3+ssl
 138         1719,  # h323gatestat
 139         1720,  # h323hostcall
 140         1723,  # pptp
 141         2049,  # nfs
 142         3659,  # apple-sasl
 143         4045,  # lockd
 144         5060,  # sip
 145         5061,  # sips
 146         6000,  # x11
 147         6566,  # sane-port
 148         6665,  # irc (alternate)
 149         6666,  # irc (alternate)
 150         6667,  # irc (default)
 151         6668,  # irc (alternate)
 152         6669,  # irc (alternate)
 153         6697,  # irc+tls
 154         10080,  # amanda
 155     ]
 156
 157
 158 def get_port(host: str = '') -> int:
 159     host = host or '127.0.0.1'
 160     port = 0
 161     while True:
 162         free_socket = _open_socket(host, 0)
 163         port = free_socket.getsockname()[1]
 164         free_socket.close()
 165         if not is_bad_port(port):
 166             break
 167     return port
 168
 169 def http2_compatible() -> bool:
 170     # The HTTP/2.0 server requires OpenSSL 1.0.2+.
 171     #
 172     # For systems using other SSL libraries (e.g. LibreSSL), we assume they
 173     # have the necessary support.
 174     import ssl
 175     if not ssl.OPENSSL_VERSION.startswith("OpenSSL"):
 176         logger = get_logger()
 177         logger.warning(
 178             'Skipping HTTP/2.0 compatibility check as system is not using '
 179             'OpenSSL (found: %s)' % ssl.OPENSSL_VERSION)
 180         return True
 181
 182     # Note that OpenSSL's versioning scheme differs between 1.1.1 and
 183     # earlier and 3.0.0. ssl.OPENSSL_VERSION_INFO returns a
 184     #     (major, minor, 0, patch, 0)
 185     # tuple with OpenSSL 3.0.0 and later, and a
 186     #     (major, minor, fix, patch, status)
 187     # tuple for older releases.
 188     # Semantically, "patch" in 3.0.0+ is similar to "fix" in previous versions.
 189     #
 190     # What we do in the check below is allow OpenSSL 3.x.y+, 1.1.x+ and 1.0.2+.
 191     ssl_v = ssl.OPENSSL_VERSION_INFO
 192     return (ssl_v[0] > 1 or
 193             (ssl_v[0] == 1 and
 194              (ssl_v[1] == 1 or
 195               (ssl_v[1] == 0 and ssl_v[2] >= 2))))