edq.net.util

  1"""
  2General utilities for working with HTTP items.
  3
  4It is generally assumed that header keys will always be lower case.
  5Or at least, the value will be accessible when looked up with a lower case key,
  6e.g. if the header dict is case insensitive.
  7"""
  8
  9import email.message
 10import errno
 11import io
 12import socket
 13import time
 14import typing
 15import urllib.parse
 16
 17import requests_toolbelt.multipart.decoder
 18
 19import edq.util.dirent
 20
 21DEFAULT_START_PORT: int = 30000
 22DEFAULT_END_PORT: int = 40000
 23DEFAULT_PORT_SEARCH_WAIT_SEC: float = 0.01
 24
 25def find_open_port(
 26        start_port: int = DEFAULT_START_PORT,
 27        end_port: int = DEFAULT_END_PORT,
 28        wait_time: float = DEFAULT_PORT_SEARCH_WAIT_SEC,
 29        ) -> int:
 30    """
 31    Find an open port on this machine within the given range (inclusive).
 32    If no open port is found, an error is raised.
 33    """
 34
 35    for port in range(start_port, end_port + 1):
 36        try:
 37            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 38            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 39            sock.bind(('127.0.0.1', port))
 40
 41            # Explicitly close the port and wait a short amount of time for the port to clear.
 42            # This should not be required because of the socket option above,
 43            # but the cost is small.
 44            sock.close()
 45            time.sleep(DEFAULT_PORT_SEARCH_WAIT_SEC)
 46
 47            return port
 48        except socket.error as ex:
 49            sock.close()
 50
 51            if (ex.errno == errno.EADDRINUSE):
 52                continue
 53
 54            # Unknown error.
 55            raise ex
 56
 57    raise ValueError(f"Could not find open port in [{start_port}, {end_port}].")
 58
 59def parse_request_data(
 60        url: typing.Union[str, None],
 61        headers: typing.Union[email.message.Message, typing.Dict[str, typing.Any]],
 62        body: typing.Union[bytes, str, io.BufferedIOBase],
 63        ) -> typing.Tuple[typing.Dict[str, typing.Any], typing.Dict[str, bytes]]:
 64    """ Parse data and files from an HTTP request URL and body. """
 65
 66    # Parse data from the request body.
 67    request_data, request_files = parse_request_body_data(headers, body)
 68
 69    # Parse parameters from the URL.
 70    if (url is not None):
 71        url_parts = urllib.parse.urlparse(url)
 72        request_data.update(parse_query_string(url_parts.query))
 73
 74    return request_data, request_files
 75
 76def parse_request_body_data(
 77        headers: typing.Union[email.message.Message, typing.Dict[str, typing.Any]],
 78        body: typing.Union[bytes, str, io.BufferedIOBase],
 79        ) -> typing.Tuple[typing.Dict[str, typing.Any], typing.Dict[str, bytes]]:
 80    """ Parse data and files from an HTTP request body. """
 81
 82    data: typing.Dict[str, typing.Any] = {}
 83    files: typing.Dict[str, bytes] = {}
 84
 85    length = int(headers.get('content-length', 0))
 86    if (length == 0):
 87        return data, files
 88
 89    if (isinstance(body, io.BufferedIOBase)):
 90        raw_content = body.read(length)
 91    elif (isinstance(body, str)):
 92        raw_content = body.encode(edq.util.dirent.DEFAULT_ENCODING)
 93    else:
 94        raw_content = body
 95
 96    content_type = headers.get('content-type', '').lower()
 97
 98    if (content_type.startswith('text/plain')):
 99        data[''] = raw_content.decode(edq.util.dirent.DEFAULT_ENCODING).strip()
100        return data, files
101
102    if (content_type in ['', 'application/x-www-form-urlencoded']):
103        data = parse_query_string(raw_content.decode(edq.util.dirent.DEFAULT_ENCODING).strip())
104        return data, files
105
106    if (content_type.startswith('multipart/form-data')):
107        decoder = requests_toolbelt.multipart.decoder.MultipartDecoder(
108            raw_content, content_type, encoding = edq.util.dirent.DEFAULT_ENCODING)
109
110        for multipart_section in decoder.parts:
111            values = parse_content_dispositions(multipart_section.headers)
112
113            name = values.get('name', None)
114            if (name is None):
115                raise ValueError("Could not find name for multipart section.")
116
117            # Look for a "filename" field to indicate a multipart section is a file.
118            # The file's desired name is still in "name", but an alternate name is in "filename".
119            if ('filename' in values):
120                filename = values.get('name', '')
121                if (filename == ''):
122                    raise ValueError("Unable to find filename for multipart section.")
123
124                files[filename] = multipart_section.content
125            else:
126                # Normal Parameter
127                data[name] = multipart_section.text
128
129        return data, files
130
131    raise ValueError(f"Unknown content type: '{content_type}'.")
132
133def parse_content_dispositions(headers: typing.Union[email.message.Message, typing.Dict[str, typing.Any]]) -> typing.Dict[str, typing.Any]:
134    """ Parse a request's content dispositions from headers. """
135
136    values = {}
137    for (key, value) in headers.items():
138        if (isinstance(key, bytes)):
139            key = key.decode(edq.util.dirent.DEFAULT_ENCODING)
140
141        if (isinstance(value, bytes)):
142            value = value.decode(edq.util.dirent.DEFAULT_ENCODING)
143
144        key = key.strip().lower()
145        if (key != 'content-disposition'):
146            continue
147
148        # The Python stdlib recommends using the email library for this parsing,
149        # but I have not had a good experience with it.
150        for part in value.strip().split(';'):
151            part = part.strip()
152
153            parts = part.split('=')
154            if (len(parts) != 2):
155                continue
156
157            cd_key = parts[0].strip()
158            cd_value = parts[1].strip().strip('"')
159
160            values[cd_key] = cd_value
161
162    return values
163
164def parse_query_string(text: str,
165        replace_single_lists: bool = True,
166        keep_blank_values: bool = True,
167        **kwargs: typing.Any) -> typing.Dict[str, typing.Any]:
168    """
169    Parse a query string (like urllib.parse.parse_qs()), and normalize the result.
170    If specified, lists with single values (as returned from urllib.parse.parse_qs()) will be replaced with the single value.
171    """
172
173    results = urllib.parse.parse_qs(text, keep_blank_values = True)
174    for (key, value) in results.items():
175        if (replace_single_lists and (len(value) == 1)):
176            results[key] = value[0]  # type: ignore[assignment]
177
178    return results