diff --git a/Lib/cgi.py b/Lib/cgi.py
new file mode 100755
index 000000000..c22c71b38
--- /dev/null
+++ b/Lib/cgi.py
@@ -0,0 +1,992 @@
+#! /usr/local/bin/python
+
+# NOTE: the above "/usr/local/bin/python" is NOT a mistake. It is
+# intentionally NOT "/usr/bin/env python". On many systems
+# (e.g. Solaris), /usr/local/bin is not in $PATH as passed to CGI
+# scripts, and /usr/local/bin is the default directory where Python is
+# installed, so /usr/bin/env would be unable to find python. Granted,
+# binary installations by Linux vendors often install Python in
+# /usr/bin. So let those vendors patch cgi.py to match their choice
+# of installation.
+
+"""Support module for CGI (Common Gateway Interface) scripts.
+
+This module defines a number of utilities for use by CGI scripts
+written in Python.
+"""
+
+# History
+# -------
+#
+# Michael McLay started this module. Steve Majewski changed the
+# interface to SvFormContentDict and FormContentDict. The multipart
+# parsing was inspired by code submitted by Andreas Paepcke. Guido van
+# Rossum rewrote, reformatted and documented the module and is currently
+# responsible for its maintenance.
+#
+
+__version__ = "2.6"
+
+
+# Imports
+# =======
+
+from io import StringIO, BytesIO, TextIOWrapper
+from collections.abc import Mapping
+import sys
+import os
+import urllib.parse
+from email.parser import FeedParser
+from email.message import Message
+import html
+import locale
+import tempfile
+
+__all__ = ["MiniFieldStorage", "FieldStorage", "parse", "parse_multipart",
+ "parse_header", "test", "print_exception", "print_environ",
+ "print_form", "print_directory", "print_arguments",
+ "print_environ_usage"]
+
+# Logging support
+# ===============
+
+logfile = "" # Filename to log to, if not empty
+logfp = None # File object to log to, if not None
+
+def initlog(*allargs):
+ """Write a log message, if there is a log file.
+
+ Even though this function is called initlog(), you should always
+ use log(); log is a variable that is set either to initlog
+ (initially), to dolog (once the log file has been opened), or to
+ nolog (when logging is disabled).
+
+ The first argument is a format string; the remaining arguments (if
+ any) are arguments to the % operator, so e.g.
+ log("%s: %s", "a", "b")
+ will write "a: b" to the log file, followed by a newline.
+
+ If the global logfp is not None, it should be a file object to
+ which log data is written.
+
+ If the global logfp is None, the global logfile may be a string
+ giving a filename to open, in append mode. This file should be
+ world writable!!! If the file can't be opened, logging is
+ silently disabled (since there is no safe place where we could
+ send an error message).
+
+ """
+ global log, logfile, logfp
+ if logfile and not logfp:
+ try:
+ logfp = open(logfile, "a")
+ except OSError:
+ pass
+ if not logfp:
+ log = nolog
+ else:
+ log = dolog
+ log(*allargs)
+
+def dolog(fmt, *args):
+ """Write a log message to the log file. See initlog() for docs."""
+ logfp.write(fmt%args + "\n")
+
+def nolog(*allargs):
+ """Dummy function, assigned to log when logging is disabled."""
+ pass
+
+def closelog():
+ """Close the log file."""
+ global log, logfile, logfp
+ logfile = ''
+ if logfp:
+ logfp.close()
+ logfp = None
+ log = initlog
+
+log = initlog # The current logging function
+
+
+# Parsing functions
+# =================
+
+# Maximum input we will accept when REQUEST_METHOD is POST
+# 0 ==> unlimited input
+maxlen = 0
+
+def parse(fp=None, environ=os.environ, keep_blank_values=0, strict_parsing=0):
+ """Parse a query in the environment or from a file (default stdin)
+
+ Arguments, all optional:
+
+ fp : file pointer; default: sys.stdin.buffer
+
+ environ : environment dictionary; default: os.environ
+
+ keep_blank_values: flag indicating whether blank values in
+ percent-encoded forms should be treated as blank strings.
+ A true value indicates that blanks should be retained as
+ blank strings. The default false value indicates that
+ blank values are to be ignored and treated as if they were
+ not included.
+
+ strict_parsing: flag indicating what to do with parsing errors.
+ If false (the default), errors are silently ignored.
+ If true, errors raise a ValueError exception.
+ """
+ if fp is None:
+ fp = sys.stdin
+
+ # field keys and values (except for files) are returned as strings
+ # an encoding is required to decode the bytes read from self.fp
+ if hasattr(fp,'encoding'):
+ encoding = fp.encoding
+ else:
+ encoding = 'latin-1'
+
+ # fp.read() must return bytes
+ if isinstance(fp, TextIOWrapper):
+ fp = fp.buffer
+
+ if not 'REQUEST_METHOD' in environ:
+ environ['REQUEST_METHOD'] = 'GET' # For testing stand-alone
+ if environ['REQUEST_METHOD'] == 'POST':
+ ctype, pdict = parse_header(environ['CONTENT_TYPE'])
+ if ctype == 'multipart/form-data':
+ return parse_multipart(fp, pdict)
+ elif ctype == 'application/x-www-form-urlencoded':
+ clength = int(environ['CONTENT_LENGTH'])
+ if maxlen and clength > maxlen:
+ raise ValueError('Maximum content length exceeded')
+ qs = fp.read(clength).decode(encoding)
+ else:
+ qs = '' # Unknown content-type
+ if 'QUERY_STRING' in environ:
+ if qs: qs = qs + '&'
+ qs = qs + environ['QUERY_STRING']
+ elif sys.argv[1:]:
+ if qs: qs = qs + '&'
+ qs = qs + sys.argv[1]
+ environ['QUERY_STRING'] = qs # XXX Shouldn't, really
+ elif 'QUERY_STRING' in environ:
+ qs = environ['QUERY_STRING']
+ else:
+ if sys.argv[1:]:
+ qs = sys.argv[1]
+ else:
+ qs = ""
+ environ['QUERY_STRING'] = qs # XXX Shouldn't, really
+ return urllib.parse.parse_qs(qs, keep_blank_values, strict_parsing,
+ encoding=encoding)
+
+
+def parse_multipart(fp, pdict, encoding="utf-8", errors="replace"):
+ """Parse multipart input.
+
+ Arguments:
+ fp : input file
+ pdict: dictionary containing other parameters of content-type header
+ encoding, errors: request encoding and error handler, passed to
+ FieldStorage
+
+ Returns a dictionary just like parse_qs(): keys are the field names, each
+ value is a list of values for that field. For non-file fields, the value
+ is a list of strings.
+ """
+ # RFC 2026, Section 5.1 : The "multipart" boundary delimiters are always
+ # represented as 7bit US-ASCII.
+ boundary = pdict['boundary'].decode('ascii')
+ ctype = "multipart/form-data; boundary={}".format(boundary)
+ headers = Message()
+ headers.set_type(ctype)
+ headers['Content-Length'] = pdict['CONTENT-LENGTH']
+ fs = FieldStorage(fp, headers=headers, encoding=encoding, errors=errors,
+ environ={'REQUEST_METHOD': 'POST'})
+ return {k: fs.getlist(k) for k in fs}
+
+def _parseparam(s):
+ while s[:1] == ';':
+ s = s[1:]
+ end = s.find(';')
+ while end > 0 and (s.count('"', 0, end) - s.count('\\"', 0, end)) % 2:
+ end = s.find(';', end + 1)
+ if end < 0:
+ end = len(s)
+ f = s[:end]
+ yield f.strip()
+ s = s[end:]
+
+def parse_header(line):
+ """Parse a Content-type like header.
+
+ Return the main content-type and a dictionary of options.
+
+ """
+ parts = _parseparam(';' + line)
+ key = parts.__next__()
+ pdict = {}
+ for p in parts:
+ i = p.find('=')
+ if i >= 0:
+ name = p[:i].strip().lower()
+ value = p[i+1:].strip()
+ if len(value) >= 2 and value[0] == value[-1] == '"':
+ value = value[1:-1]
+ value = value.replace('\\\\', '\\').replace('\\"', '"')
+ pdict[name] = value
+ return key, pdict
+
+
+# Classes for field storage
+# =========================
+
+class MiniFieldStorage:
+
+ """Like FieldStorage, for use when no file uploads are possible."""
+
+ # Dummy attributes
+ filename = None
+ list = None
+ type = None
+ file = None
+ type_options = {}
+ disposition = None
+ disposition_options = {}
+ headers = {}
+
+ def __init__(self, name, value):
+ """Constructor from field name and value."""
+ self.name = name
+ self.value = value
+ # self.file = StringIO(value)
+
+ def __repr__(self):
+ """Return printable representation."""
+ return "MiniFieldStorage(%r, %r)" % (self.name, self.value)
+
+
+class FieldStorage:
+
+ """Store a sequence of fields, reading multipart/form-data.
+
+ This class provides naming, typing, files stored on disk, and
+ more. At the top level, it is accessible like a dictionary, whose
+ keys are the field names. (Note: None can occur as a field name.)
+ The items are either a Python list (if there's multiple values) or
+ another FieldStorage or MiniFieldStorage object. If it's a single
+ object, it has the following attributes:
+
+ name: the field name, if specified; otherwise None
+
+ filename: the filename, if specified; otherwise None; this is the
+ client side filename, *not* the file name on which it is
+ stored (that's a temporary file you don't deal with)
+
+ value: the value as a *string*; for file uploads, this
+ transparently reads the file every time you request the value
+ and returns *bytes*
+
+ file: the file(-like) object from which you can read the data *as
+ bytes* ; None if the data is stored a simple string
+
+ type: the content-type, or None if not specified
+
+ type_options: dictionary of options specified on the content-type
+ line
+
+ disposition: content-disposition, or None if not specified
+
+ disposition_options: dictionary of corresponding options
+
+ headers: a dictionary(-like) object (sometimes email.message.Message or a
+ subclass thereof) containing *all* headers
+
+ The class is subclassable, mostly for the purpose of overriding
+ the make_file() method, which is called internally to come up with
+ a file open for reading and writing. This makes it possible to
+ override the default choice of storing all files in a temporary
+ directory and unlinking them as soon as they have been opened.
+
+ """
+ def __init__(self, fp=None, headers=None, outerboundary=b'',
+ environ=os.environ, keep_blank_values=0, strict_parsing=0,
+ limit=None, encoding='utf-8', errors='replace',
+ max_num_fields=None):
+ """Constructor. Read multipart/* until last part.
+
+ Arguments, all optional:
+
+ fp : file pointer; default: sys.stdin.buffer
+ (not used when the request method is GET)
+ Can be :
+ 1. a TextIOWrapper object
+ 2. an object whose read() and readline() methods return bytes
+
+ headers : header dictionary-like object; default:
+ taken from environ as per CGI spec
+
+ outerboundary : terminating multipart boundary
+ (for internal use only)
+
+ environ : environment dictionary; default: os.environ
+
+ keep_blank_values: flag indicating whether blank values in
+ percent-encoded forms should be treated as blank strings.
+ A true value indicates that blanks should be retained as
+ blank strings. The default false value indicates that
+ blank values are to be ignored and treated as if they were
+ not included.
+
+ strict_parsing: flag indicating what to do with parsing errors.
+ If false (the default), errors are silently ignored.
+ If true, errors raise a ValueError exception.
+
+ limit : used internally to read parts of multipart/form-data forms,
+ to exit from the reading loop when reached. It is the difference
+ between the form content-length and the number of bytes already
+ read
+
+ encoding, errors : the encoding and error handler used to decode the
+ binary stream to strings. Must be the same as the charset defined
+ for the page sending the form (content-type : meta http-equiv or
+ header)
+
+ max_num_fields: int. If set, then __init__ throws a ValueError
+ if there are more than n fields read by parse_qsl().
+
+ """
+ method = 'GET'
+ self.keep_blank_values = keep_blank_values
+ self.strict_parsing = strict_parsing
+ self.max_num_fields = max_num_fields
+ if 'REQUEST_METHOD' in environ:
+ method = environ['REQUEST_METHOD'].upper()
+ self.qs_on_post = None
+ if method == 'GET' or method == 'HEAD':
+ if 'QUERY_STRING' in environ:
+ qs = environ['QUERY_STRING']
+ elif sys.argv[1:]:
+ qs = sys.argv[1]
+ else:
+ qs = ""
+ qs = qs.encode(locale.getpreferredencoding(), 'surrogateescape')
+ fp = BytesIO(qs)
+ if headers is None:
+ headers = {'content-type':
+ "application/x-www-form-urlencoded"}
+ if headers is None:
+ headers = {}
+ if method == 'POST':
+ # Set default content-type for POST to what's traditional
+ headers['content-type'] = "application/x-www-form-urlencoded"
+ if 'CONTENT_TYPE' in environ:
+ headers['content-type'] = environ['CONTENT_TYPE']
+ if 'QUERY_STRING' in environ:
+ self.qs_on_post = environ['QUERY_STRING']
+ if 'CONTENT_LENGTH' in environ:
+ headers['content-length'] = environ['CONTENT_LENGTH']
+ else:
+ if not (isinstance(headers, (Mapping, Message))):
+ raise TypeError("headers must be mapping or an instance of "
+ "email.message.Message")
+ self.headers = headers
+ if fp is None:
+ self.fp = sys.stdin.buffer
+ # self.fp.read() must return bytes
+ elif isinstance(fp, TextIOWrapper):
+ self.fp = fp.buffer
+ else:
+ if not (hasattr(fp, 'read') and hasattr(fp, 'readline')):
+ raise TypeError("fp must be file pointer")
+ self.fp = fp
+
+ self.encoding = encoding
+ self.errors = errors
+
+ if not isinstance(outerboundary, bytes):
+ raise TypeError('outerboundary must be bytes, not %s'
+ % type(outerboundary).__name__)
+ self.outerboundary = outerboundary
+
+ self.bytes_read = 0
+ self.limit = limit
+
+ # Process content-disposition header
+ cdisp, pdict = "", {}
+ if 'content-disposition' in self.headers:
+ cdisp, pdict = parse_header(self.headers['content-disposition'])
+ self.disposition = cdisp
+ self.disposition_options = pdict
+ self.name = None
+ if 'name' in pdict:
+ self.name = pdict['name']
+ self.filename = None
+ if 'filename' in pdict:
+ self.filename = pdict['filename']
+ self._binary_file = self.filename is not None
+
+ # Process content-type header
+ #
+ # Honor any existing content-type header. But if there is no
+ # content-type header, use some sensible defaults. Assume
+ # outerboundary is "" at the outer level, but something non-false
+ # inside a multi-part. The default for an inner part is text/plain,
+ # but for an outer part it should be urlencoded. This should catch
+ # bogus clients which erroneously forget to include a content-type
+ # header.
+ #
+ # See below for what we do if there does exist a content-type header,
+ # but it happens to be something we don't understand.
+ if 'content-type' in self.headers:
+ ctype, pdict = parse_header(self.headers['content-type'])
+ elif self.outerboundary or method != 'POST':
+ ctype, pdict = "text/plain", {}
+ else:
+ ctype, pdict = 'application/x-www-form-urlencoded', {}
+ self.type = ctype
+ self.type_options = pdict
+ if 'boundary' in pdict:
+ self.innerboundary = pdict['boundary'].encode(self.encoding,
+ self.errors)
+ else:
+ self.innerboundary = b""
+
+ clen = -1
+ if 'content-length' in self.headers:
+ try:
+ clen = int(self.headers['content-length'])
+ except ValueError:
+ pass
+ if maxlen and clen > maxlen:
+ raise ValueError('Maximum content length exceeded')
+ self.length = clen
+ if self.limit is None and clen >= 0:
+ self.limit = clen
+
+ self.list = self.file = None
+ self.done = 0
+ if ctype == 'application/x-www-form-urlencoded':
+ self.read_urlencoded()
+ elif ctype[:10] == 'multipart/':
+ self.read_multi(environ, keep_blank_values, strict_parsing)
+ else:
+ self.read_single()
+
+ def __del__(self):
+ try:
+ self.file.close()
+ except AttributeError:
+ pass
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *args):
+ self.file.close()
+
+ def __repr__(self):
+ """Return a printable representation."""
+ return "FieldStorage(%r, %r, %r)" % (
+ self.name, self.filename, self.value)
+
+ def __iter__(self):
+ return iter(self.keys())
+
+ def __getattr__(self, name):
+ if name != 'value':
+ raise AttributeError(name)
+ if self.file:
+ self.file.seek(0)
+ value = self.file.read()
+ self.file.seek(0)
+ elif self.list is not None:
+ value = self.list
+ else:
+ value = None
+ return value
+
+ def __getitem__(self, key):
+ """Dictionary style indexing."""
+ if self.list is None:
+ raise TypeError("not indexable")
+ found = []
+ for item in self.list:
+ if item.name == key: found.append(item)
+ if not found:
+ raise KeyError(key)
+ if len(found) == 1:
+ return found[0]
+ else:
+ return found
+
+ def getvalue(self, key, default=None):
+ """Dictionary style get() method, including 'value' lookup."""
+ if key in self:
+ value = self[key]
+ if isinstance(value, list):
+ return [x.value for x in value]
+ else:
+ return value.value
+ else:
+ return default
+
+ def getfirst(self, key, default=None):
+ """ Return the first value received."""
+ if key in self:
+ value = self[key]
+ if isinstance(value, list):
+ return value[0].value
+ else:
+ return value.value
+ else:
+ return default
+
+ def getlist(self, key):
+ """ Return list of received values."""
+ if key in self:
+ value = self[key]
+ if isinstance(value, list):
+ return [x.value for x in value]
+ else:
+ return [value.value]
+ else:
+ return []
+
+ def keys(self):
+ """Dictionary style keys() method."""
+ if self.list is None:
+ raise TypeError("not indexable")
+ return list(set(item.name for item in self.list))
+
+ def __contains__(self, key):
+ """Dictionary style __contains__ method."""
+ if self.list is None:
+ raise TypeError("not indexable")
+ return any(item.name == key for item in self.list)
+
+ def __len__(self):
+ """Dictionary style len(x) support."""
+ return len(self.keys())
+
+ def __bool__(self):
+ if self.list is None:
+ raise TypeError("Cannot be converted to bool.")
+ return bool(self.list)
+
+ def read_urlencoded(self):
+ """Internal: read data in query string format."""
+ qs = self.fp.read(self.length)
+ if not isinstance(qs, bytes):
+ raise ValueError("%s should return bytes, got %s" \
+ % (self.fp, type(qs).__name__))
+ qs = qs.decode(self.encoding, self.errors)
+ if self.qs_on_post:
+ qs += '&' + self.qs_on_post
+ query = urllib.parse.parse_qsl(
+ qs, self.keep_blank_values, self.strict_parsing,
+ encoding=self.encoding, errors=self.errors,
+ max_num_fields=self.max_num_fields)
+ self.list = [MiniFieldStorage(key, value) for key, value in query]
+ self.skip_lines()
+
+ FieldStorageClass = None
+
+ def read_multi(self, environ, keep_blank_values, strict_parsing):
+ """Internal: read a part that is itself multipart."""
+ ib = self.innerboundary
+ if not valid_boundary(ib):
+ raise ValueError('Invalid boundary in multipart form: %r' % (ib,))
+ self.list = []
+ if self.qs_on_post:
+ query = urllib.parse.parse_qsl(
+ self.qs_on_post, self.keep_blank_values, self.strict_parsing,
+ encoding=self.encoding, errors=self.errors,
+ max_num_fields=self.max_num_fields)
+ self.list.extend(MiniFieldStorage(key, value) for key, value in query)
+
+ klass = self.FieldStorageClass or self.__class__
+ first_line = self.fp.readline() # bytes
+ if not isinstance(first_line, bytes):
+ raise ValueError("%s should return bytes, got %s" \
+ % (self.fp, type(first_line).__name__))
+ self.bytes_read += len(first_line)
+
+ # Ensure that we consume the file until we've hit our inner boundary
+ while (first_line.strip() != (b"--" + self.innerboundary) and
+ first_line):
+ first_line = self.fp.readline()
+ self.bytes_read += len(first_line)
+
+ # Propagate max_num_fields into the sub class appropriately
+ max_num_fields = self.max_num_fields
+ if max_num_fields is not None:
+ max_num_fields -= len(self.list)
+
+ while True:
+ parser = FeedParser()
+ hdr_text = b""
+ while True:
+ data = self.fp.readline()
+ hdr_text += data
+ if not data.strip():
+ break
+ if not hdr_text:
+ break
+ # parser takes strings, not bytes
+ self.bytes_read += len(hdr_text)
+ parser.feed(hdr_text.decode(self.encoding, self.errors))
+ headers = parser.close()
+
+ # Some clients add Content-Length for part headers, ignore them
+ if 'content-length' in headers:
+ del headers['content-length']
+
+ limit = None if self.limit is None \
+ else self.limit - self.bytes_read
+ part = klass(self.fp, headers, ib, environ, keep_blank_values,
+ strict_parsing, limit,
+ self.encoding, self.errors, max_num_fields)
+
+ if max_num_fields is not None:
+ max_num_fields -= 1
+ if part.list:
+ max_num_fields -= len(part.list)
+ if max_num_fields < 0:
+ raise ValueError('Max number of fields exceeded')
+
+ self.bytes_read += part.bytes_read
+ self.list.append(part)
+ if part.done or self.bytes_read >= self.length > 0:
+ break
+ self.skip_lines()
+
+ def read_single(self):
+ """Internal: read an atomic part."""
+ if self.length >= 0:
+ self.read_binary()
+ self.skip_lines()
+ else:
+ self.read_lines()
+ self.file.seek(0)
+
+ bufsize = 8*1024 # I/O buffering size for copy to file
+
+ def read_binary(self):
+ """Internal: read binary data."""
+ self.file = self.make_file()
+ todo = self.length
+ if todo >= 0:
+ while todo > 0:
+ data = self.fp.read(min(todo, self.bufsize)) # bytes
+ if not isinstance(data, bytes):
+ raise ValueError("%s should return bytes, got %s"
+ % (self.fp, type(data).__name__))
+ self.bytes_read += len(data)
+ if not data:
+ self.done = -1
+ break
+ self.file.write(data)
+ todo = todo - len(data)
+
+ def read_lines(self):
+ """Internal: read lines until EOF or outerboundary."""
+ if self._binary_file:
+ self.file = self.__file = BytesIO() # store data as bytes for files
+ else:
+ self.file = self.__file = StringIO() # as strings for other fields
+ if self.outerboundary:
+ self.read_lines_to_outerboundary()
+ else:
+ self.read_lines_to_eof()
+
+ def __write(self, line):
+ """line is always bytes, not string"""
+ if self.__file is not None:
+ if self.__file.tell() + len(line) > 1000:
+ self.file = self.make_file()
+ data = self.__file.getvalue()
+ self.file.write(data)
+ self.__file = None
+ if self._binary_file:
+ # keep bytes
+ self.file.write(line)
+ else:
+ # decode to string
+ self.file.write(line.decode(self.encoding, self.errors))
+
+ def read_lines_to_eof(self):
+ """Internal: read lines until EOF."""
+ while 1:
+ line = self.fp.readline(1<<16) # bytes
+ self.bytes_read += len(line)
+ if not line:
+ self.done = -1
+ break
+ self.__write(line)
+
+ def read_lines_to_outerboundary(self):
+ """Internal: read lines until outerboundary.
+ Data is read as bytes: boundaries and line ends must be converted
+ to bytes for comparisons.
+ """
+ next_boundary = b"--" + self.outerboundary
+ last_boundary = next_boundary + b"--"
+ delim = b""
+ last_line_lfend = True
+ _read = 0
+ while 1:
+ if self.limit is not None and _read >= self.limit:
+ break
+ line = self.fp.readline(1<<16) # bytes
+ self.bytes_read += len(line)
+ _read += len(line)
+ if not line:
+ self.done = -1
+ break
+ if delim == b"\r":
+ line = delim + line
+ delim = b""
+ if line.startswith(b"--") and last_line_lfend:
+ strippedline = line.rstrip()
+ if strippedline == next_boundary:
+ break
+ if strippedline == last_boundary:
+ self.done = 1
+ break
+ odelim = delim
+ if line.endswith(b"\r\n"):
+ delim = b"\r\n"
+ line = line[:-2]
+ last_line_lfend = True
+ elif line.endswith(b"\n"):
+ delim = b"\n"
+ line = line[:-1]
+ last_line_lfend = True
+ elif line.endswith(b"\r"):
+ # We may interrupt \r\n sequences if they span the 2**16
+ # byte boundary
+ delim = b"\r"
+ line = line[:-1]
+ last_line_lfend = False
+ else:
+ delim = b""
+ last_line_lfend = False
+ self.__write(odelim + line)
+
+ def skip_lines(self):
+ """Internal: skip lines until outer boundary if defined."""
+ if not self.outerboundary or self.done:
+ return
+ next_boundary = b"--" + self.outerboundary
+ last_boundary = next_boundary + b"--"
+ last_line_lfend = True
+ while True:
+ line = self.fp.readline(1<<16)
+ self.bytes_read += len(line)
+ if not line:
+ self.done = -1
+ break
+ if line.endswith(b"--") and last_line_lfend:
+ strippedline = line.strip()
+ if strippedline == next_boundary:
+ break
+ if strippedline == last_boundary:
+ self.done = 1
+ break
+ last_line_lfend = line.endswith(b'\n')
+
+ def make_file(self):
+ """Overridable: return a readable & writable file.
+
+ The file will be used as follows:
+ - data is written to it
+ - seek(0)
+ - data is read from it
+
+ The file is opened in binary mode for files, in text mode
+ for other fields
+
+ This version opens a temporary file for reading and writing,
+ and immediately deletes (unlinks) it. The trick (on Unix!) is
+ that the file can still be used, but it can't be opened by
+ another process, and it will automatically be deleted when it
+ is closed or when the current process terminates.
+
+ If you want a more permanent file, you derive a class which
+ overrides this method. If you want a visible temporary file
+ that is nevertheless automatically deleted when the script
+ terminates, try defining a __del__ method in a derived class
+ which unlinks the temporary files you have created.
+
+ """
+ if self._binary_file:
+ return tempfile.TemporaryFile("wb+")
+ else:
+ return tempfile.TemporaryFile("w+",
+ encoding=self.encoding, newline = '\n')
+
+
+# Test/debug code
+# ===============
+
+def test(environ=os.environ):
+ """Robust test CGI script, usable as main program.
+
+ Write minimal HTTP headers and dump all information provided to
+ the script in HTML form.
+
+ """
+ print("Content-type: text/html")
+ print()
+ sys.stderr = sys.stdout
+ try:
+ form = FieldStorage() # Replace with other classes to test those
+ print_directory()
+ print_arguments()
+ print_form(form)
+ print_environ(environ)
+ print_environ_usage()
+ def f():
+ exec("testing print_exception() -- italics?")
+ def g(f=f):
+ f()
+ print("What follows is a test, not an actual exception:
")
+ g()
+ except:
+ print_exception()
+
+ print("Second try with a small maxlen...
")
+
+ global maxlen
+ maxlen = 50
+ try:
+ form = FieldStorage() # Replace with other classes to test those
+ print_directory()
+ print_arguments()
+ print_form(form)
+ print_environ(environ)
+ except:
+ print_exception()
+
+def print_exception(type=None, value=None, tb=None, limit=None):
+ if type is None:
+ type, value, tb = sys.exc_info()
+ import traceback
+ print()
+ print("Traceback (most recent call last):
")
+ list = traceback.format_tb(tb, limit) + \
+ traceback.format_exception_only(type, value)
+ print("%s%s
" % (
+ html.escape("".join(list[:-1])),
+ html.escape(list[-1]),
+ ))
+ del tb
+
+def print_environ(environ=os.environ):
+ """Dump the shell environment as HTML."""
+ keys = sorted(environ.keys())
+ print()
+ print("Shell Environment:
")
+ print("")
+ for key in keys:
+ print("- ", html.escape(key), "
- ", html.escape(environ[key]))
+ print("
")
+ print()
+
+def print_form(form):
+ """Dump the contents of a form as HTML."""
+ keys = sorted(form.keys())
+ print()
+ print("Form Contents:
")
+ if not keys:
+ print("No form fields.")
+ print("
")
+ for key in keys:
+ print("- " + html.escape(key) + ":", end=' ')
+ value = form[key]
+ print("" + html.escape(repr(type(value))) + "")
+ print("
- " + html.escape(repr(value)))
+ print("
")
+ print()
+
+def print_directory():
+ """Dump the current directory as HTML."""
+ print()
+ print("Current Working Directory:
")
+ try:
+ pwd = os.getcwd()
+ except OSError as msg:
+ print("OSError:", html.escape(str(msg)))
+ else:
+ print(html.escape(pwd))
+ print()
+
+def print_arguments():
+ print()
+ print("Command Line Arguments:
")
+ print()
+ print(sys.argv)
+ print()
+
+def print_environ_usage():
+ """Dump a list of environment variables used by CGI as HTML."""
+ print("""
+These environment variables could have been set:
+
+- AUTH_TYPE
+
- CONTENT_LENGTH
+
- CONTENT_TYPE
+
- DATE_GMT
+
- DATE_LOCAL
+
- DOCUMENT_NAME
+
- DOCUMENT_ROOT
+
- DOCUMENT_URI
+
- GATEWAY_INTERFACE
+
- LAST_MODIFIED
+
- PATH
+
- PATH_INFO
+
- PATH_TRANSLATED
+
- QUERY_STRING
+
- REMOTE_ADDR
+
- REMOTE_HOST
+
- REMOTE_IDENT
+
- REMOTE_USER
+
- REQUEST_METHOD
+
- SCRIPT_NAME
+
- SERVER_NAME
+
- SERVER_PORT
+
- SERVER_PROTOCOL
+
- SERVER_ROOT
+
- SERVER_SOFTWARE
+
+In addition, HTTP headers sent by the server may be passed in the
+environment as well. Here are some common variable names:
+
+- HTTP_ACCEPT
+
- HTTP_CONNECTION
+
- HTTP_HOST
+
- HTTP_PRAGMA
+
- HTTP_REFERER
+
- HTTP_USER_AGENT
+
+""")
+
+
+# Utilities
+# =========
+
+def valid_boundary(s):
+ import re
+ if isinstance(s, bytes):
+ _vb_pattern = b"^[ -~]{0,200}[!-~]$"
+ else:
+ _vb_pattern = "^[ -~]{0,200}[!-~]$"
+ return re.match(_vb_pattern, s)
+
+# Invoke mainline
+# ===============
+
+# Call test() when this file is run as a script (not imported as a module)
+if __name__ == '__main__':
+ test()
diff --git a/Lib/test/test_cgi.py b/Lib/test/test_cgi.py
new file mode 100644
index 000000000..8ea2484ae
--- /dev/null
+++ b/Lib/test/test_cgi.py
@@ -0,0 +1,609 @@
+import cgi
+import os
+import sys
+import tempfile
+import unittest
+from collections import namedtuple
+from io import StringIO, BytesIO
+from test import support
+
+class HackedSysModule:
+ # The regression test will have real values in sys.argv, which
+ # will completely confuse the test of the cgi module
+ argv = []
+ stdin = sys.stdin
+
+cgi.sys = HackedSysModule()
+
+class ComparableException:
+ def __init__(self, err):
+ self.err = err
+
+ def __str__(self):
+ return str(self.err)
+
+ def __eq__(self, anExc):
+ if not isinstance(anExc, Exception):
+ return NotImplemented
+ return (self.err.__class__ == anExc.__class__ and
+ self.err.args == anExc.args)
+
+ def __getattr__(self, attr):
+ return getattr(self.err, attr)
+
+def do_test(buf, method):
+ env = {}
+ if method == "GET":
+ fp = None
+ env['REQUEST_METHOD'] = 'GET'
+ env['QUERY_STRING'] = buf
+ elif method == "POST":
+ fp = BytesIO(buf.encode('latin-1')) # FieldStorage expects bytes
+ env['REQUEST_METHOD'] = 'POST'
+ env['CONTENT_TYPE'] = 'application/x-www-form-urlencoded'
+ env['CONTENT_LENGTH'] = str(len(buf))
+ else:
+ raise ValueError("unknown method: %s" % method)
+ try:
+ return cgi.parse(fp, env, strict_parsing=1)
+ except Exception as err:
+ return ComparableException(err)
+
+parse_strict_test_cases = [
+ ("", ValueError("bad query field: ''")),
+ ("&", ValueError("bad query field: ''")),
+ ("&&", ValueError("bad query field: ''")),
+ (";", ValueError("bad query field: ''")),
+ (";&;", ValueError("bad query field: ''")),
+ # Should the next few really be valid?
+ ("=", {}),
+ ("=&=", {}),
+ ("=;=", {}),
+ # This rest seem to make sense
+ ("=a", {'': ['a']}),
+ ("&=a", ValueError("bad query field: ''")),
+ ("=a&", ValueError("bad query field: ''")),
+ ("=&a", ValueError("bad query field: 'a'")),
+ ("b=a", {'b': ['a']}),
+ ("b+=a", {'b ': ['a']}),
+ ("a=b=a", {'a': ['b=a']}),
+ ("a=+b=a", {'a': [' b=a']}),
+ ("&b=a", ValueError("bad query field: ''")),
+ ("b&=a", ValueError("bad query field: 'b'")),
+ ("a=a+b&b=b+c", {'a': ['a b'], 'b': ['b c']}),
+ ("a=a+b&a=b+a", {'a': ['a b', 'b a']}),
+ ("x=1&y=2.0&z=2-3.%2b0", {'x': ['1'], 'y': ['2.0'], 'z': ['2-3.+0']}),
+ ("x=1;y=2.0&z=2-3.%2b0", {'x': ['1'], 'y': ['2.0'], 'z': ['2-3.+0']}),
+ ("x=1;y=2.0;z=2-3.%2b0", {'x': ['1'], 'y': ['2.0'], 'z': ['2-3.+0']}),
+ ("Hbc5161168c542333633315dee1182227:key_store_seqid=400006&cuyer=r&view=bustomer&order_id=0bb2e248638833d48cb7fed300000f1b&expire=964546263&lobale=en-US&kid=130003.300038&ss=env",
+ {'Hbc5161168c542333633315dee1182227:key_store_seqid': ['400006'],
+ 'cuyer': ['r'],
+ 'expire': ['964546263'],
+ 'kid': ['130003.300038'],
+ 'lobale': ['en-US'],
+ 'order_id': ['0bb2e248638833d48cb7fed300000f1b'],
+ 'ss': ['env'],
+ 'view': ['bustomer'],
+ }),
+
+ ("group_id=5470&set=custom&_assigned_to=31392&_status=1&_category=100&SUBMIT=Browse",
+ {'SUBMIT': ['Browse'],
+ '_assigned_to': ['31392'],
+ '_category': ['100'],
+ '_status': ['1'],
+ 'group_id': ['5470'],
+ 'set': ['custom'],
+ })
+ ]
+
+def norm(seq):
+ return sorted(seq, key=repr)
+
+def first_elts(list):
+ return [p[0] for p in list]
+
+def first_second_elts(list):
+ return [(p[0], p[1][0]) for p in list]
+
+def gen_result(data, environ):
+ encoding = 'latin-1'
+ fake_stdin = BytesIO(data.encode(encoding))
+ fake_stdin.seek(0)
+ form = cgi.FieldStorage(fp=fake_stdin, environ=environ, encoding=encoding)
+
+ result = {}
+ for k, v in dict(form).items():
+ result[k] = isinstance(v, list) and form.getlist(k) or v.value
+
+ return result
+
+class CgiTests(unittest.TestCase):
+
+ def test_parse_multipart(self):
+ fp = BytesIO(POSTDATA.encode('latin1'))
+ env = {'boundary': BOUNDARY.encode('latin1'),
+ 'CONTENT-LENGTH': '558'}
+ result = cgi.parse_multipart(fp, env)
+ expected = {'submit': [' Add '], 'id': ['1234'],
+ 'file': [b'Testing 123.\n'], 'title': ['']}
+ self.assertEqual(result, expected)
+
+ # TODO RUSTPYTHON - see https://github.com/RustPython/RustPython/issues/935
+ @unittest.expectedFailure
+ def test_parse_multipart_invalid_encoding(self):
+ BOUNDARY = "JfISa01"
+ POSTDATA = """--JfISa01
+Content-Disposition: form-data; name="submit-name"
+Content-Length: 3
+
+\u2603
+--JfISa01"""
+ fp = BytesIO(POSTDATA.encode('utf8'))
+ env = {'boundary': BOUNDARY.encode('latin1'),
+ 'CONTENT-LENGTH': str(len(POSTDATA.encode('utf8')))}
+ result = cgi.parse_multipart(fp, env, encoding="ascii",
+ errors="surrogateescape")
+ expected = {'submit-name': ["\udce2\udc98\udc83"]}
+ self.assertEqual(result, expected)
+ self.assertEqual("\u2603".encode('utf8'),
+ result["submit-name"][0].encode('utf8', 'surrogateescape'))
+
+ def test_fieldstorage_properties(self):
+ fs = cgi.FieldStorage()
+ self.assertFalse(fs)
+ self.assertIn("FieldStorage", repr(fs))
+ self.assertEqual(list(fs), list(fs.keys()))
+ fs.list.append(namedtuple('MockFieldStorage', 'name')('fieldvalue'))
+ self.assertTrue(fs)
+
+ def test_fieldstorage_invalid(self):
+ self.assertRaises(TypeError, cgi.FieldStorage, "not-a-file-obj",
+ environ={"REQUEST_METHOD":"PUT"})
+ self.assertRaises(TypeError, cgi.FieldStorage, "foo", "bar")
+ fs = cgi.FieldStorage(headers={'content-type':'text/plain'})
+ self.assertRaises(TypeError, bool, fs)
+
+ def test_strict(self):
+ for orig, expect in parse_strict_test_cases:
+ # Test basic parsing
+ d = do_test(orig, "GET")
+ self.assertEqual(d, expect, "Error parsing %s method GET" % repr(orig))
+ d = do_test(orig, "POST")
+ self.assertEqual(d, expect, "Error parsing %s method POST" % repr(orig))
+
+ env = {'QUERY_STRING': orig}
+ fs = cgi.FieldStorage(environ=env)
+ if isinstance(expect, dict):
+ # test dict interface
+ self.assertEqual(len(expect), len(fs))
+ self.assertCountEqual(expect.keys(), fs.keys())
+ ##self.assertEqual(norm(expect.values()), norm(fs.values()))
+ ##self.assertEqual(norm(expect.items()), norm(fs.items()))
+ self.assertEqual(fs.getvalue("nonexistent field", "default"), "default")
+ # test individual fields
+ for key in expect.keys():
+ expect_val = expect[key]
+ self.assertIn(key, fs)
+ if len(expect_val) > 1:
+ self.assertEqual(fs.getvalue(key), expect_val)
+ else:
+ self.assertEqual(fs.getvalue(key), expect_val[0])
+
+ def test_log(self):
+ cgi.log("Testing")
+
+ cgi.logfp = StringIO()
+ cgi.initlog("%s", "Testing initlog 1")
+ cgi.log("%s", "Testing log 2")
+ self.assertEqual(cgi.logfp.getvalue(), "Testing initlog 1\nTesting log 2\n")
+ if os.path.exists(os.devnull):
+ cgi.logfp = None
+ cgi.logfile = os.devnull
+ cgi.initlog("%s", "Testing log 3")
+ self.addCleanup(cgi.closelog)
+ cgi.log("Testing log 4")
+
+ # TODO RUSTPYTHON
+ @unittest.expectedFailure
+ def test_fieldstorage_readline(self):
+ # FieldStorage uses readline, which has the capacity to read all
+ # contents of the input file into memory; we use readline's size argument
+ # to prevent that for files that do not contain any newlines in
+ # non-GET/HEAD requests
+ class TestReadlineFile:
+ def __init__(self, file):
+ self.file = file
+ self.numcalls = 0
+
+ def readline(self, size=None):
+ self.numcalls += 1
+ if size:
+ return self.file.readline(size)
+ else:
+ return self.file.readline()
+
+ def __getattr__(self, name):
+ file = self.__dict__['file']
+ a = getattr(file, name)
+ if not isinstance(a, int):
+ setattr(self, name, a)
+ return a
+
+ f = TestReadlineFile(tempfile.TemporaryFile("wb+"))
+ self.addCleanup(f.close)
+ f.write(b'x' * 256 * 1024)
+ f.seek(0)
+ env = {'REQUEST_METHOD':'PUT'}
+ fs = cgi.FieldStorage(fp=f, environ=env)
+ self.addCleanup(fs.file.close)
+ # if we're not chunking properly, readline is only called twice
+ # (by read_binary); if we are chunking properly, it will be called 5 times
+ # as long as the chunksize is 1 << 16.
+ self.assertGreater(f.numcalls, 2)
+ f.close()
+
+ def test_fieldstorage_multipart(self):
+ #Test basic FieldStorage multipart parsing
+ env = {
+ 'REQUEST_METHOD': 'POST',
+ 'CONTENT_TYPE': 'multipart/form-data; boundary={}'.format(BOUNDARY),
+ 'CONTENT_LENGTH': '558'}
+ fp = BytesIO(POSTDATA.encode('latin-1'))
+ fs = cgi.FieldStorage(fp, environ=env, encoding="latin-1")
+ self.assertEqual(len(fs.list), 4)
+ expect = [{'name':'id', 'filename':None, 'value':'1234'},
+ {'name':'title', 'filename':None, 'value':''},
+ {'name':'file', 'filename':'test.txt', 'value':b'Testing 123.\n'},
+ {'name':'submit', 'filename':None, 'value':' Add '}]
+ for x in range(len(fs.list)):
+ for k, exp in expect[x].items():
+ got = getattr(fs.list[x], k)
+ self.assertEqual(got, exp)
+
+ def test_fieldstorage_multipart_leading_whitespace(self):
+ env = {
+ 'REQUEST_METHOD': 'POST',
+ 'CONTENT_TYPE': 'multipart/form-data; boundary={}'.format(BOUNDARY),
+ 'CONTENT_LENGTH': '560'}
+ # Add some leading whitespace to our post data that will cause the
+ # first line to not be the innerboundary.
+ fp = BytesIO(b"\r\n" + POSTDATA.encode('latin-1'))
+ fs = cgi.FieldStorage(fp, environ=env, encoding="latin-1")
+ self.assertEqual(len(fs.list), 4)
+ expect = [{'name':'id', 'filename':None, 'value':'1234'},
+ {'name':'title', 'filename':None, 'value':''},
+ {'name':'file', 'filename':'test.txt', 'value':b'Testing 123.\n'},
+ {'name':'submit', 'filename':None, 'value':' Add '}]
+ for x in range(len(fs.list)):
+ for k, exp in expect[x].items():
+ got = getattr(fs.list[x], k)
+ self.assertEqual(got, exp)
+
+ def test_fieldstorage_multipart_non_ascii(self):
+ #Test basic FieldStorage multipart parsing
+ env = {'REQUEST_METHOD':'POST',
+ 'CONTENT_TYPE': 'multipart/form-data; boundary={}'.format(BOUNDARY),
+ 'CONTENT_LENGTH':'558'}
+ for encoding in ['iso-8859-1','utf-8']:
+ fp = BytesIO(POSTDATA_NON_ASCII.encode(encoding))
+ fs = cgi.FieldStorage(fp, environ=env,encoding=encoding)
+ self.assertEqual(len(fs.list), 1)
+ expect = [{'name':'id', 'filename':None, 'value':'\xe7\xf1\x80'}]
+ for x in range(len(fs.list)):
+ for k, exp in expect[x].items():
+ got = getattr(fs.list[x], k)
+ self.assertEqual(got, exp)
+
+ # TODO RUSTPYTHON
+ @unittest.expectedFailure
+ def test_fieldstorage_multipart_maxline(self):
+ # Issue #18167
+ maxline = 1 << 16
+ self.maxDiff = None
+ def check(content):
+ data = """---123
+Content-Disposition: form-data; name="upload"; filename="fake.txt"
+Content-Type: text/plain
+
+%s
+---123--
+""".replace('\n', '\r\n') % content
+ environ = {
+ 'CONTENT_LENGTH': str(len(data)),
+ 'CONTENT_TYPE': 'multipart/form-data; boundary=-123',
+ 'REQUEST_METHOD': 'POST',
+ }
+ self.assertEqual(gen_result(data, environ),
+ {'upload': content.encode('latin1')})
+ check('x' * (maxline - 1))
+ check('x' * (maxline - 1) + '\r')
+ check('x' * (maxline - 1) + '\r' + 'y' * (maxline - 1))
+
+ def test_fieldstorage_multipart_w3c(self):
+ # Test basic FieldStorage multipart parsing (W3C sample)
+ env = {
+ 'REQUEST_METHOD': 'POST',
+ 'CONTENT_TYPE': 'multipart/form-data; boundary={}'.format(BOUNDARY_W3),
+ 'CONTENT_LENGTH': str(len(POSTDATA_W3))}
+ fp = BytesIO(POSTDATA_W3.encode('latin-1'))
+ fs = cgi.FieldStorage(fp, environ=env, encoding="latin-1")
+ self.assertEqual(len(fs.list), 2)
+ self.assertEqual(fs.list[0].name, 'submit-name')
+ self.assertEqual(fs.list[0].value, 'Larry')
+ self.assertEqual(fs.list[1].name, 'files')
+ files = fs.list[1].value
+ self.assertEqual(len(files), 2)
+ expect = [{'name': None, 'filename': 'file1.txt', 'value': b'... contents of file1.txt ...'},
+ {'name': None, 'filename': 'file2.gif', 'value': b'...contents of file2.gif...'}]
+ for x in range(len(files)):
+ for k, exp in expect[x].items():
+ got = getattr(files[x], k)
+ self.assertEqual(got, exp)
+
+ def test_fieldstorage_part_content_length(self):
+ BOUNDARY = "JfISa01"
+ POSTDATA = """--JfISa01
+Content-Disposition: form-data; name="submit-name"
+Content-Length: 5
+
+Larry
+--JfISa01"""
+ env = {
+ 'REQUEST_METHOD': 'POST',
+ 'CONTENT_TYPE': 'multipart/form-data; boundary={}'.format(BOUNDARY),
+ 'CONTENT_LENGTH': str(len(POSTDATA))}
+ fp = BytesIO(POSTDATA.encode('latin-1'))
+ fs = cgi.FieldStorage(fp, environ=env, encoding="latin-1")
+ self.assertEqual(len(fs.list), 1)
+ self.assertEqual(fs.list[0].name, 'submit-name')
+ self.assertEqual(fs.list[0].value, 'Larry')
+
+ def test_field_storage_multipart_no_content_length(self):
+ fp = BytesIO(b"""--MyBoundary
+Content-Disposition: form-data; name="my-arg"; filename="foo"
+
+Test
+
+--MyBoundary--
+""")
+ env = {
+ "REQUEST_METHOD": "POST",
+ "CONTENT_TYPE": "multipart/form-data; boundary=MyBoundary",
+ "wsgi.input": fp,
+ }
+ fields = cgi.FieldStorage(fp, environ=env)
+
+ self.assertEqual(len(fields["my-arg"].file.read()), 5)
+
+ def test_fieldstorage_as_context_manager(self):
+ fp = BytesIO(b'x' * 10)
+ env = {'REQUEST_METHOD': 'PUT'}
+ with cgi.FieldStorage(fp=fp, environ=env) as fs:
+ content = fs.file.read()
+ self.assertFalse(fs.file.closed)
+ self.assertTrue(fs.file.closed)
+ self.assertEqual(content, 'x' * 10)
+ with self.assertRaisesRegex(ValueError, 'I/O operation on closed file'):
+ fs.file.read()
+
+ _qs_result = {
+ 'key1': 'value1',
+ 'key2': ['value2x', 'value2y'],
+ 'key3': 'value3',
+ 'key4': 'value4'
+ }
+ def testQSAndUrlEncode(self):
+ data = "key2=value2x&key3=value3&key4=value4"
+ environ = {
+ 'CONTENT_LENGTH': str(len(data)),
+ 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
+ 'QUERY_STRING': 'key1=value1&key2=value2y',
+ 'REQUEST_METHOD': 'POST',
+ }
+ v = gen_result(data, environ)
+ self.assertEqual(self._qs_result, v)
+
+ def test_max_num_fields(self):
+ # For application/x-www-form-urlencoded
+ data = '&'.join(['a=a']*11)
+ environ = {
+ 'CONTENT_LENGTH': str(len(data)),
+ 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
+ 'REQUEST_METHOD': 'POST',
+ }
+
+ with self.assertRaises(ValueError):
+ cgi.FieldStorage(
+ fp=BytesIO(data.encode()),
+ environ=environ,
+ max_num_fields=10,
+ )
+
+ # For multipart/form-data
+ data = """---123
+Content-Disposition: form-data; name="a"
+
+3
+---123
+Content-Type: application/x-www-form-urlencoded
+
+a=4
+---123
+Content-Type: application/x-www-form-urlencoded
+
+a=5
+---123--
+"""
+ environ = {
+ 'CONTENT_LENGTH': str(len(data)),
+ 'CONTENT_TYPE': 'multipart/form-data; boundary=-123',
+ 'QUERY_STRING': 'a=1&a=2',
+ 'REQUEST_METHOD': 'POST',
+ }
+
+ # 2 GET entities
+ # 1 top level POST entities
+ # 1 entity within the second POST entity
+ # 1 entity within the third POST entity
+ with self.assertRaises(ValueError):
+ cgi.FieldStorage(
+ fp=BytesIO(data.encode()),
+ environ=environ,
+ max_num_fields=4,
+ )
+ cgi.FieldStorage(
+ fp=BytesIO(data.encode()),
+ environ=environ,
+ max_num_fields=5,
+ )
+
+ def testQSAndFormData(self):
+ data = """---123
+Content-Disposition: form-data; name="key2"
+
+value2y
+---123
+Content-Disposition: form-data; name="key3"
+
+value3
+---123
+Content-Disposition: form-data; name="key4"
+
+value4
+---123--
+"""
+ environ = {
+ 'CONTENT_LENGTH': str(len(data)),
+ 'CONTENT_TYPE': 'multipart/form-data; boundary=-123',
+ 'QUERY_STRING': 'key1=value1&key2=value2x',
+ 'REQUEST_METHOD': 'POST',
+ }
+ v = gen_result(data, environ)
+ self.assertEqual(self._qs_result, v)
+
+ def testQSAndFormDataFile(self):
+ data = """---123
+Content-Disposition: form-data; name="key2"
+
+value2y
+---123
+Content-Disposition: form-data; name="key3"
+
+value3
+---123
+Content-Disposition: form-data; name="key4"
+
+value4
+---123
+Content-Disposition: form-data; name="upload"; filename="fake.txt"
+Content-Type: text/plain
+
+this is the content of the fake file
+
+---123--
+"""
+ environ = {
+ 'CONTENT_LENGTH': str(len(data)),
+ 'CONTENT_TYPE': 'multipart/form-data; boundary=-123',
+ 'QUERY_STRING': 'key1=value1&key2=value2x',
+ 'REQUEST_METHOD': 'POST',
+ }
+ result = self._qs_result.copy()
+ result.update({
+ 'upload': b'this is the content of the fake file\n'
+ })
+ v = gen_result(data, environ)
+ self.assertEqual(result, v)
+
+ def test_parse_header(self):
+ self.assertEqual(
+ cgi.parse_header("text/plain"),
+ ("text/plain", {}))
+ self.assertEqual(
+ cgi.parse_header("text/vnd.just.made.this.up ; "),
+ ("text/vnd.just.made.this.up", {}))
+ self.assertEqual(
+ cgi.parse_header("text/plain;charset=us-ascii"),
+ ("text/plain", {"charset": "us-ascii"}))
+ self.assertEqual(
+ cgi.parse_header('text/plain ; charset="us-ascii"'),
+ ("text/plain", {"charset": "us-ascii"}))
+ self.assertEqual(
+ cgi.parse_header('text/plain ; charset="us-ascii"; another=opt'),
+ ("text/plain", {"charset": "us-ascii", "another": "opt"}))
+ self.assertEqual(
+ cgi.parse_header('attachment; filename="silly.txt"'),
+ ("attachment", {"filename": "silly.txt"}))
+ self.assertEqual(
+ cgi.parse_header('attachment; filename="strange;name"'),
+ ("attachment", {"filename": "strange;name"}))
+ self.assertEqual(
+ cgi.parse_header('attachment; filename="strange;name";size=123;'),
+ ("attachment", {"filename": "strange;name", "size": "123"}))
+ self.assertEqual(
+ cgi.parse_header('form-data; name="files"; filename="fo\\"o;bar"'),
+ ("form-data", {"name": "files", "filename": 'fo"o;bar'}))
+
+ def test_all(self):
+ blacklist = {"logfile", "logfp", "initlog", "dolog", "nolog",
+ "closelog", "log", "maxlen", "valid_boundary"}
+ support.check__all__(self, cgi, blacklist=blacklist)
+
+
+BOUNDARY = "---------------------------721837373350705526688164684"
+
+POSTDATA = """-----------------------------721837373350705526688164684
+Content-Disposition: form-data; name="id"
+
+1234
+-----------------------------721837373350705526688164684
+Content-Disposition: form-data; name="title"
+
+
+-----------------------------721837373350705526688164684
+Content-Disposition: form-data; name="file"; filename="test.txt"
+Content-Type: text/plain
+
+Testing 123.
+
+-----------------------------721837373350705526688164684
+Content-Disposition: form-data; name="submit"
+
+ Add\x20
+-----------------------------721837373350705526688164684--
+"""
+
+POSTDATA_NON_ASCII = """-----------------------------721837373350705526688164684
+Content-Disposition: form-data; name="id"
+
+\xe7\xf1\x80
+-----------------------------721837373350705526688164684
+"""
+
+# http://www.w3.org/TR/html401/interact/forms.html#h-17.13.4
+BOUNDARY_W3 = "AaB03x"
+POSTDATA_W3 = """--AaB03x
+Content-Disposition: form-data; name="submit-name"
+
+Larry
+--AaB03x
+Content-Disposition: form-data; name="files"
+Content-Type: multipart/mixed; boundary=BbC04y
+
+--BbC04y
+Content-Disposition: file; filename="file1.txt"
+Content-Type: text/plain
+
+... contents of file1.txt ...
+--BbC04y
+Content-Disposition: file; filename="file2.gif"
+Content-Type: image/gif
+Content-Transfer-Encoding: binary
+
+...contents of file2.gif...
+--BbC04y--
+--AaB03x--
+"""
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/Lib/test/test_urlparse.py b/Lib/test/test_urlparse.py
index 95d14fe13..61234b764 100644
--- a/Lib/test/test_urlparse.py
+++ b/Lib/test/test_urlparse.py
@@ -522,8 +522,6 @@ class UrlParseTestCase(unittest.TestCase):
self.assertEqual(result.url, defrag)
self.assertEqual(result.fragment, frag)
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_urlsplit_scoped_IPv6(self):
p = urllib.parse.urlsplit('http://[FE80::822a:a8ff:fe49:470c%tESt]:1234')
self.assertEqual(p.hostname, "fe80::822a:a8ff:fe49:470c%tESt")
@@ -883,8 +881,6 @@ class UrlParseTestCase(unittest.TestCase):
errors="ignore")
self.assertEqual(result, [('key', '\u0141-')])
- # TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
def test_parse_qsl_max_num_fields(self):
with self.assertRaises(ValueError):
urllib.parse.parse_qs('&'.join(['a=a']*11), max_num_fields=10)
@@ -949,8 +945,6 @@ class UrlParseTestCase(unittest.TestCase):
self.assertEqual(p2.scheme, 'tel')
self.assertEqual(p2.path, '+31641044153')
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_port_casting_failure_message(self):
message = "Port could not be cast to integer value as 'oracle'"
p1 = urllib.parse.urlparse('http://Server=sde; Service=sde:oracle')
@@ -1003,7 +997,7 @@ class UrlParseTestCase(unittest.TestCase):
self.assertCountEqual(urllib.parse.__all__, expected)
# TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
+ @unittest.expectedFailure
def test_urlsplit_normalization(self):
# Certain characters should never occur in the netloc,
# including under normalization.
@@ -1037,8 +1031,6 @@ class Utility_Tests(unittest.TestCase):
"""Testcase to test the various utility functions in the urllib."""
# In Python 2 this test class was in test_urllib.
- # TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
def test_splittype(self):
splittype = urllib.parse._splittype
self.assertEqual(splittype('type:opaquestring'), ('type', 'opaquestring'))
@@ -1047,8 +1039,6 @@ class Utility_Tests(unittest.TestCase):
self.assertEqual(splittype('type:'), ('type', ''))
self.assertEqual(splittype('type:opaque:string'), ('type', 'opaque:string'))
- # TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
def test_splithost(self):
splithost = urllib.parse._splithost
self.assertEqual(splithost('//www.example.org:80/foo/bar/baz.html'),
@@ -1078,8 +1068,6 @@ class Utility_Tests(unittest.TestCase):
self.assertEqual(splithost("//example.net/file#"),
('example.net', '/file#'))
- # TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
def test_splituser(self):
splituser = urllib.parse._splituser
self.assertEqual(splituser('User:Pass@www.python.org:080'),
@@ -1093,8 +1081,6 @@ class Utility_Tests(unittest.TestCase):
self.assertEqual(splituser('User@example.com:Pass@www.python.org:080'),
('User@example.com:Pass', 'www.python.org:080'))
- # TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
def test_splitpasswd(self):
# Some of the password examples are not sensible, but it is added to
# confirming to RFC2617 and addressing issue4675.
@@ -1113,8 +1099,6 @@ class Utility_Tests(unittest.TestCase):
self.assertEqual(splitpasswd('user'), ('user', None))
self.assertEqual(splitpasswd(':ab'), ('', 'ab'))
- # TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
def test_splitport(self):
splitport = urllib.parse._splitport
self.assertEqual(splitport('parrot:88'), ('parrot', '88'))
@@ -1126,8 +1110,6 @@ class Utility_Tests(unittest.TestCase):
self.assertEqual(splitport('[::1]'), ('[::1]', None))
self.assertEqual(splitport(':88'), ('', '88'))
- # TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
def test_splitnport(self):
splitnport = urllib.parse._splitnport
self.assertEqual(splitnport('parrot:88'), ('parrot', 88))
@@ -1140,8 +1122,6 @@ class Utility_Tests(unittest.TestCase):
self.assertEqual(splitnport('parrot:cheese'), ('parrot', None))
self.assertEqual(splitnport('parrot:cheese', 55), ('parrot', None))
- # TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
def test_splitquery(self):
# Normal cases are exercised by other tests; ensure that we also
# catch cases with no port specified (testcase ensuring coverage)
@@ -1154,8 +1134,6 @@ class Utility_Tests(unittest.TestCase):
('http://python.org/fake', None))
self.assertEqual(splitquery('?foo=bar'), ('', 'foo=bar'))
- # TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
def test_splittag(self):
splittag = urllib.parse._splittag
self.assertEqual(splittag('http://example.com?foo=bar#baz'),
@@ -1168,8 +1146,6 @@ class Utility_Tests(unittest.TestCase):
self.assertEqual(splittag('http://example.com?foo=bar#baz#boo'),
('http://example.com?foo=bar#baz', 'boo'))
- # TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
def test_splitattr(self):
splitattr = urllib.parse._splitattr
self.assertEqual(splitattr('/path;attr1=value1;attr2=value2'),
@@ -1179,8 +1155,6 @@ class Utility_Tests(unittest.TestCase):
('', ['attr1=value1', 'attr2=value2']))
self.assertEqual(splitattr('/path'), ('/path', []))
- # TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
def test_splitvalue(self):
# Normal cases are exercised by other tests; test pathological cases
# with no key/value pairs. (testcase ensuring coverage)
@@ -1191,8 +1165,6 @@ class Utility_Tests(unittest.TestCase):
self.assertEqual(splitvalue('foobar'), ('foobar', None))
self.assertEqual(splitvalue('foo=bar=baz'), ('foo', 'bar=baz'))
- # TODO: RUSTPYTHON
- @unittest.skip("TODO: RUSTPYTHON")
def test_to_bytes(self):
result = urllib.parse._to_bytes('http://www.python.org')
self.assertEqual(result, 'http://www.python.org')
@@ -1208,8 +1180,6 @@ class Utility_Tests(unittest.TestCase):
class DeprecationTest(unittest.TestCase):
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_splittype_deprecation(self):
with self.assertWarns(DeprecationWarning) as cm:
urllib.parse.splittype('')
@@ -1217,8 +1187,6 @@ class DeprecationTest(unittest.TestCase):
'urllib.parse.splittype() is deprecated as of 3.8, '
'use urllib.parse.urlparse() instead')
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_splithost_deprecation(self):
with self.assertWarns(DeprecationWarning) as cm:
urllib.parse.splithost('')
@@ -1226,8 +1194,6 @@ class DeprecationTest(unittest.TestCase):
'urllib.parse.splithost() is deprecated as of 3.8, '
'use urllib.parse.urlparse() instead')
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_splituser_deprecation(self):
with self.assertWarns(DeprecationWarning) as cm:
urllib.parse.splituser('')
@@ -1235,8 +1201,6 @@ class DeprecationTest(unittest.TestCase):
'urllib.parse.splituser() is deprecated as of 3.8, '
'use urllib.parse.urlparse() instead')
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_splitpasswd_deprecation(self):
with self.assertWarns(DeprecationWarning) as cm:
urllib.parse.splitpasswd('')
@@ -1244,8 +1208,6 @@ class DeprecationTest(unittest.TestCase):
'urllib.parse.splitpasswd() is deprecated as of 3.8, '
'use urllib.parse.urlparse() instead')
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_splitport_deprecation(self):
with self.assertWarns(DeprecationWarning) as cm:
urllib.parse.splitport('')
@@ -1253,8 +1215,6 @@ class DeprecationTest(unittest.TestCase):
'urllib.parse.splitport() is deprecated as of 3.8, '
'use urllib.parse.urlparse() instead')
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_splitnport_deprecation(self):
with self.assertWarns(DeprecationWarning) as cm:
urllib.parse.splitnport('')
@@ -1262,8 +1222,6 @@ class DeprecationTest(unittest.TestCase):
'urllib.parse.splitnport() is deprecated as of 3.8, '
'use urllib.parse.urlparse() instead')
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_splitquery_deprecation(self):
with self.assertWarns(DeprecationWarning) as cm:
urllib.parse.splitquery('')
@@ -1271,8 +1229,6 @@ class DeprecationTest(unittest.TestCase):
'urllib.parse.splitquery() is deprecated as of 3.8, '
'use urllib.parse.urlparse() instead')
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_splittag_deprecation(self):
with self.assertWarns(DeprecationWarning) as cm:
urllib.parse.splittag('')
@@ -1280,8 +1236,6 @@ class DeprecationTest(unittest.TestCase):
'urllib.parse.splittag() is deprecated as of 3.8, '
'use urllib.parse.urlparse() instead')
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_splitattr_deprecation(self):
with self.assertWarns(DeprecationWarning) as cm:
urllib.parse.splitattr('')
@@ -1289,8 +1243,6 @@ class DeprecationTest(unittest.TestCase):
'urllib.parse.splitattr() is deprecated as of 3.8, '
'use urllib.parse.urlparse() instead')
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_splitvalue_deprecation(self):
with self.assertWarns(DeprecationWarning) as cm:
urllib.parse.splitvalue('')
@@ -1298,8 +1250,6 @@ class DeprecationTest(unittest.TestCase):
'urllib.parse.splitvalue() is deprecated as of 3.8, '
'use urllib.parse.parse_qsl() instead')
- # TODO: RUSTPYTHON
- @unittest.expectedFailure
def test_to_bytes_deprecation(self):
with self.assertWarns(DeprecationWarning) as cm:
urllib.parse.to_bytes('')
diff --git a/Lib/urllib/error.py b/Lib/urllib/error.py
index c5b675d16..8cd901f13 100644
--- a/Lib/urllib/error.py
+++ b/Lib/urllib/error.py
@@ -16,14 +16,10 @@ import urllib.response
__all__ = ['URLError', 'HTTPError', 'ContentTooShortError']
-# do these error classes make sense?
-# make sure all of the OSError stuff is overridden. we just want to be
-# subtypes.
-
class URLError(OSError):
# URLError is a sub-type of OSError, but it doesn't share any of
# the implementation. need to override __init__ and __str__.
- # It sets self.args for compatibility with other EnvironmentError
+ # It sets self.args for compatibility with other OSError
# subclasses, but args doesn't have the typical format with errno in
# slot 0 and strerror in slot 1. This may be better than nothing.
def __init__(self, reason, filename=None):
diff --git a/Lib/urllib/parse.py b/Lib/urllib/parse.py
index 958767a08..e2b6f133e 100644
--- a/Lib/urllib/parse.py
+++ b/Lib/urllib/parse.py
@@ -30,6 +30,7 @@ test_urlparse.py provides a good indicator of parsing behavior.
import re
import sys
import collections
+import warnings
__all__ = ["urlparse", "urlunparse", "urljoin", "urldefrag",
"urlsplit", "urlunsplit", "urlencode", "parse_qs",
@@ -38,29 +39,37 @@ __all__ = ["urlparse", "urlunparse", "urljoin", "urldefrag",
"DefragResult", "ParseResult", "SplitResult",
"DefragResultBytes", "ParseResultBytes", "SplitResultBytes"]
-# A classification of schemes ('' means apply by default)
-uses_relative = ['ftp', 'http', 'gopher', 'nntp', 'imap',
+# A classification of schemes.
+# The empty string classifies URLs with no scheme specified,
+# being the default value returned by “urlsplit” and “urlparse”.
+
+uses_relative = ['', 'ftp', 'http', 'gopher', 'nntp', 'imap',
'wais', 'file', 'https', 'shttp', 'mms',
- 'prospero', 'rtsp', 'rtspu', '', 'sftp',
+ 'prospero', 'rtsp', 'rtspu', 'sftp',
'svn', 'svn+ssh', 'ws', 'wss']
-uses_netloc = ['ftp', 'http', 'gopher', 'nntp', 'telnet',
+
+uses_netloc = ['', 'ftp', 'http', 'gopher', 'nntp', 'telnet',
'imap', 'wais', 'file', 'mms', 'https', 'shttp',
- 'snews', 'prospero', 'rtsp', 'rtspu', 'rsync', '',
+ 'snews', 'prospero', 'rtsp', 'rtspu', 'rsync',
'svn', 'svn+ssh', 'sftp', 'nfs', 'git', 'git+ssh',
'ws', 'wss']
-uses_params = ['ftp', 'hdl', 'prospero', 'http', 'imap',
+
+uses_params = ['', 'ftp', 'hdl', 'prospero', 'http', 'imap',
'https', 'shttp', 'rtsp', 'rtspu', 'sip', 'sips',
- 'mms', '', 'sftp', 'tel']
+ 'mms', 'sftp', 'tel']
# These are not actually used anymore, but should stay for backwards
# compatibility. (They are undocumented, but have a public-looking name.)
+
non_hierarchical = ['gopher', 'hdl', 'mailto', 'news',
'telnet', 'wais', 'imap', 'snews', 'sip', 'sips']
-uses_query = ['http', 'wais', 'imap', 'https', 'shttp', 'mms',
- 'gopher', 'rtsp', 'rtspu', 'sip', 'sips', '']
-uses_fragment = ['ftp', 'hdl', 'http', 'gopher', 'news',
+
+uses_query = ['', 'http', 'wais', 'imap', 'https', 'shttp', 'mms',
+ 'gopher', 'rtsp', 'rtspu', 'sip', 'sips']
+
+uses_fragment = ['', 'ftp', 'hdl', 'http', 'gopher', 'news',
'nntp', 'wais', 'https', 'shttp', 'snews',
- 'file', 'prospero', '']
+ 'file', 'prospero']
# Characters valid in scheme names
scheme_chars = ('abcdefghijklmnopqrstuvwxyz'
@@ -147,16 +156,22 @@ class _NetlocResultMixinBase(object):
def hostname(self):
hostname = self._hostinfo[0]
if not hostname:
- hostname = None
- elif hostname is not None:
- hostname = hostname.lower()
- return hostname
+ return None
+ # Scoped IPv6 address may have zone info, which must not be lowercased
+ # like http://[fe80::822a:a8ff:fe49:470c%tESt]:1234/keys
+ separator = '%' if isinstance(hostname, str) else b'%'
+ hostname, percent, zone = hostname.partition(separator)
+ return hostname.lower() + percent + zone
@property
def port(self):
port = self._hostinfo[1]
if port is not None:
- port = int(port, 10)
+ try:
+ port = int(port, 10)
+ except ValueError:
+ message = f'Port could not be cast to integer value as {port!r}'
+ raise ValueError(message) from None
if not ( 0 <= port <= 65535):
raise ValueError("Port out of range 0-65535")
return port
@@ -274,7 +289,7 @@ by reference to a primary resource and additional identifying information.
"""
_ParseResultBase.__doc__ = """
-ParseResult(scheme, netloc, path, params, query, fragment)
+ParseResult(scheme, netloc, path, params, query, fragment)
A 6-tuple that contains components of a parsed URL.
"""
@@ -381,6 +396,24 @@ def _splitnetloc(url, start=0):
delim = min(delim, wdelim) # use earliest delim position
return url[start:delim], url[delim:] # return (domain, rest)
+def _checknetloc(netloc):
+ if not netloc or netloc.isascii():
+ return
+ # looking for characters like \u2100 that expand to 'a/c'
+ # IDNA uses NFKC equivalence, so normalize for this check
+ import unicodedata
+ n = netloc.replace('@', '') # ignore characters already included
+ n = n.replace(':', '') # but not the surrounding text
+ n = n.replace('#', '')
+ n = n.replace('?', '')
+ netloc2 = unicodedata.normalize('NFKC', n)
+ if n == netloc2:
+ return
+ for c in '/?#@:':
+ if c in netloc2:
+ raise ValueError("netloc '" + netloc + "' contains invalid " +
+ "characters under NFKC normalization")
+
def urlsplit(url, scheme='', allow_fragments=True):
"""Parse a URL into 5 components:
:///?#
@@ -399,7 +432,6 @@ def urlsplit(url, scheme='', allow_fragments=True):
i = url.find(':')
if i > 0:
if url[:i] == 'http': # optimize the common case
- scheme = url[:i].lower()
url = url[i+1:]
if url[:2] == '//':
netloc, url = _splitnetloc(url, 2)
@@ -410,7 +442,8 @@ def urlsplit(url, scheme='', allow_fragments=True):
url, fragment = url.split('#', 1)
if '?' in url:
url, query = url.split('?', 1)
- v = SplitResult(scheme, netloc, url, query, fragment)
+ _checknetloc(netloc)
+ v = SplitResult('http', netloc, url, query, fragment)
_parse_cache[key] = v
return _coerce_result(v)
for c in url[:i]:
@@ -433,6 +466,7 @@ def urlsplit(url, scheme='', allow_fragments=True):
url, fragment = url.split('#', 1)
if '?' in url:
url, query = url.split('?', 1)
+ _checknetloc(netloc)
v = SplitResult(scheme, netloc, url, query, fragment)
_parse_cache[key] = v
return _coerce_result(v)
@@ -574,7 +608,7 @@ def unquote_to_bytes(string):
# if the function is never called
global _hextobyte
if _hextobyte is None:
- _hextobyte = {(a + b).encode(): bytes([int(a + b, 16)])
+ _hextobyte = {(a + b).encode(): bytes.fromhex(a + b)
for a in _hexdig for b in _hexdig}
for item in bits[1:]:
try:
@@ -612,8 +646,9 @@ def unquote(string, encoding='utf-8', errors='replace'):
append(bits[i + 1])
return ''.join(res)
+
def parse_qs(qs, keep_blank_values=False, strict_parsing=False,
- encoding='utf-8', errors='replace'):
+ encoding='utf-8', errors='replace', max_num_fields=None):
"""Parse a query given as a string argument.
Arguments:
@@ -633,10 +668,16 @@ def parse_qs(qs, keep_blank_values=False, strict_parsing=False,
encoding and errors: specify how to decode percent-encoded sequences
into Unicode characters, as accepted by the bytes.decode() method.
+
+ max_num_fields: int. If set, then throws a ValueError if there
+ are more than n fields read by parse_qsl().
+
+ Returns a dictionary.
"""
parsed_result = {}
pairs = parse_qsl(qs, keep_blank_values, strict_parsing,
- encoding=encoding, errors=errors)
+ encoding=encoding, errors=errors,
+ max_num_fields=max_num_fields)
for name, value in pairs:
if name in parsed_result:
parsed_result[name].append(value)
@@ -644,30 +685,43 @@ def parse_qs(qs, keep_blank_values=False, strict_parsing=False,
parsed_result[name] = [value]
return parsed_result
+
def parse_qsl(qs, keep_blank_values=False, strict_parsing=False,
- encoding='utf-8', errors='replace'):
+ encoding='utf-8', errors='replace', max_num_fields=None):
"""Parse a query given as a string argument.
- Arguments:
+ Arguments:
- qs: percent-encoded query string to be parsed
+ qs: percent-encoded query string to be parsed
- keep_blank_values: flag indicating whether blank values in
- percent-encoded queries should be treated as blank strings. A
- true value indicates that blanks should be retained as blank
- strings. The default false value indicates that blank values
- are to be ignored and treated as if they were not included.
+ keep_blank_values: flag indicating whether blank values in
+ percent-encoded queries should be treated as blank strings.
+ A true value indicates that blanks should be retained as blank
+ strings. The default false value indicates that blank values
+ are to be ignored and treated as if they were not included.
- strict_parsing: flag indicating what to do with parsing errors. If
- false (the default), errors are silently ignored. If true,
- errors raise a ValueError exception.
+ strict_parsing: flag indicating what to do with parsing errors. If
+ false (the default), errors are silently ignored. If true,
+ errors raise a ValueError exception.
- encoding and errors: specify how to decode percent-encoded sequences
- into Unicode characters, as accepted by the bytes.decode() method.
+ encoding and errors: specify how to decode percent-encoded sequences
+ into Unicode characters, as accepted by the bytes.decode() method.
- Returns a list, as G-d intended.
+ max_num_fields: int. If set, then throws a ValueError
+ if there are more than n fields read by parse_qsl().
+
+ Returns a list, as G-d intended.
"""
qs, _coerce_result = _coerce_args(qs)
+
+ # If max_num_fields is defined then check that the number of fields
+ # is less than max_num_fields. This prevents a memory exhaustion DOS
+ # attack via post bodies with many fields.
+ if max_num_fields is not None:
+ num_fields = 1 + qs.count('&') + qs.count(';')
+ if max_num_fields < num_fields:
+ raise ValueError('Max number of fields exceeded')
+
pairs = [s2 for s1 in qs.split('&') for s2 in s1.split(';')]
r = []
for name_value in pairs:
@@ -704,7 +758,7 @@ def unquote_plus(string, encoding='utf-8', errors='replace'):
_ALWAYS_SAFE = frozenset(b'ABCDEFGHIJKLMNOPQRSTUVWXYZ'
b'abcdefghijklmnopqrstuvwxyz'
b'0123456789'
- b'_.-')
+ b'_.-~')
_ALWAYS_SAFE_BYTES = bytes(_ALWAYS_SAFE)
_safe_quoters = {}
@@ -734,22 +788,32 @@ def quote(string, safe='/', encoding=None, errors=None):
"""quote('abc def') -> 'abc%20def'
Each part of a URL, e.g. the path info, the query, etc., has a
- different set of reserved characters that must be quoted.
+ different set of reserved characters that must be quoted. The
+ quote function offers a cautious (not minimal) way to quote a
+ string for most of these parts.
- RFC 2396 Uniform Resource Identifiers (URI): Generic Syntax lists
- the following reserved characters.
+ RFC 3986 Uniform Resource Identifier (URI): Generic Syntax lists
+ the following (un)reserved characters.
- reserved = ";" | "/" | "?" | ":" | "@" | "&" | "=" | "+" |
- "$" | ","
+ unreserved = ALPHA / DIGIT / "-" / "." / "_" / "~"
+ reserved = gen-delims / sub-delims
+ gen-delims = ":" / "/" / "?" / "#" / "[" / "]" / "@"
+ sub-delims = "!" / "$" / "&" / "'" / "(" / ")"
+ / "*" / "+" / "," / ";" / "="
- Each of these characters is reserved in some component of a URL,
+ Each of the reserved characters is reserved in some component of a URL,
but not necessarily in all of them.
- By default, the quote function is intended for quoting the path
- section of a URL. Thus, it will not encode '/'. This character
- is reserved, but in typical usage the quote function is being
- called on a path where the existing slash characters are used as
- reserved characters.
+ The quote function %-escapes all characters that are neither in the
+ unreserved chars ("always safe") nor the additional chars set via the
+ safe arg.
+
+ The default for the safe arg is '/'. The character is reserved, but in
+ typical usage the quote function is being called on a path where the
+ existing slash characters are to be preserved.
+
+ Python 3.7 updates from using RFC 2396 to RFC 3986 to quote URL strings.
+ Now, "~" is included in the set of unreserved characters.
string and safe may be either str or bytes objects. encoding and errors
must not be specified if string is a bytes object.
@@ -893,7 +957,14 @@ def urlencode(query, doseq=False, safe='', encoding=None, errors=None,
l.append(k + '=' + elt)
return '&'.join(l)
+
def to_bytes(url):
+ warnings.warn("urllib.parse.to_bytes() is deprecated as of 3.8",
+ DeprecationWarning, stacklevel=2)
+ return _to_bytes(url)
+
+
+def _to_bytes(url):
"""to_bytes(u"URL") --> 'URL'."""
# Most URL schemes require ASCII. If that changes, the conversion
# can be relaxed.
@@ -906,16 +977,29 @@ def to_bytes(url):
" contains non-ASCII characters")
return url
+
def unwrap(url):
- """unwrap('') --> 'type://host/path'."""
+ """Transform a string like '' into 'scheme://host/path'.
+
+ The string is returned unchanged if it's not a wrapped URL.
+ """
url = str(url).strip()
if url[:1] == '<' and url[-1:] == '>':
url = url[1:-1].strip()
- if url[:4] == 'URL:': url = url[4:].strip()
+ if url[:4] == 'URL:':
+ url = url[4:].strip()
return url
-_typeprog = None
+
def splittype(url):
+ warnings.warn("urllib.parse.splittype() is deprecated as of 3.8, "
+ "use urllib.parse.urlparse() instead",
+ DeprecationWarning, stacklevel=2)
+ return _splittype(url)
+
+
+_typeprog = None
+def _splittype(url):
"""splittype('type:opaquestring') --> 'type', 'opaquestring'."""
global _typeprog
if _typeprog is None:
@@ -927,12 +1011,20 @@ def splittype(url):
return scheme.lower(), data
return None, url
-_hostprog = None
+
def splithost(url):
+ warnings.warn("urllib.parse.splithost() is deprecated as of 3.8, "
+ "use urllib.parse.urlparse() instead",
+ DeprecationWarning, stacklevel=2)
+ return _splithost(url)
+
+
+_hostprog = None
+def _splithost(url):
"""splithost('//host[:port]/path') --> 'host[:port]', '/path'."""
global _hostprog
if _hostprog is None:
- _hostprog = re.compile('//([^/?]*)(.*)', re.DOTALL)
+ _hostprog = re.compile('//([^/#?]*)(.*)', re.DOTALL)
match = _hostprog.match(url)
if match:
@@ -942,32 +1034,64 @@ def splithost(url):
return host_port, path
return None, url
+
def splituser(host):
+ warnings.warn("urllib.parse.splituser() is deprecated as of 3.8, "
+ "use urllib.parse.urlparse() instead",
+ DeprecationWarning, stacklevel=2)
+ return _splituser(host)
+
+
+def _splituser(host):
"""splituser('user[:passwd]@host[:port]') --> 'user[:passwd]', 'host[:port]'."""
user, delim, host = host.rpartition('@')
return (user if delim else None), host
+
def splitpasswd(user):
+ warnings.warn("urllib.parse.splitpasswd() is deprecated as of 3.8, "
+ "use urllib.parse.urlparse() instead",
+ DeprecationWarning, stacklevel=2)
+ return _splitpasswd(user)
+
+
+def _splitpasswd(user):
"""splitpasswd('user:passwd') -> 'user', 'passwd'."""
user, delim, passwd = user.partition(':')
return user, (passwd if delim else None)
+
+def splitport(host):
+ warnings.warn("urllib.parse.splitport() is deprecated as of 3.8, "
+ "use urllib.parse.urlparse() instead",
+ DeprecationWarning, stacklevel=2)
+ return _splitport(host)
+
+
# splittag('/path#tag') --> '/path', 'tag'
_portprog = None
-def splitport(host):
+def _splitport(host):
"""splitport('host:port') --> 'host', 'port'."""
global _portprog
if _portprog is None:
- _portprog = re.compile('(.*):([0-9]*)$', re.DOTALL)
+ _portprog = re.compile('(.*):([0-9]*)', re.DOTALL)
- match = _portprog.match(host)
+ match = _portprog.fullmatch(host)
if match:
host, port = match.groups()
if port:
return host, port
return host, None
+
def splitnport(host, defport=-1):
+ warnings.warn("urllib.parse.splitnport() is deprecated as of 3.8, "
+ "use urllib.parse.urlparse() instead",
+ DeprecationWarning, stacklevel=2)
+ return _splitnport(host, defport)
+
+
+def _splitnport(host, defport=-1):
"""Split host and port, returning numeric port.
Return given default port if no ':' found; defaults to -1.
Return numerical port if a valid number are found after ':'.
@@ -983,27 +1107,59 @@ def splitnport(host, defport=-1):
return host, nport
return host, defport
+
def splitquery(url):
+ warnings.warn("urllib.parse.splitquery() is deprecated as of 3.8, "
+ "use urllib.parse.urlparse() instead",
+ DeprecationWarning, stacklevel=2)
+ return _splitquery(url)
+
+
+def _splitquery(url):
"""splitquery('/path?query') --> '/path', 'query'."""
path, delim, query = url.rpartition('?')
if delim:
return path, query
return url, None
+
def splittag(url):
+ warnings.warn("urllib.parse.splittag() is deprecated as of 3.8, "
+ "use urllib.parse.urlparse() instead",
+ DeprecationWarning, stacklevel=2)
+ return _splittag(url)
+
+
+def _splittag(url):
"""splittag('/path#tag') --> '/path', 'tag'."""
path, delim, tag = url.rpartition('#')
if delim:
return path, tag
return url, None
+
def splitattr(url):
+ warnings.warn("urllib.parse.splitattr() is deprecated as of 3.8, "
+ "use urllib.parse.urlparse() instead",
+ DeprecationWarning, stacklevel=2)
+ return _splitattr(url)
+
+
+def _splitattr(url):
"""splitattr('/path;attr1=value1;attr2=value2;...') ->
'/path', ['attr1=value1', 'attr2=value2', ...]."""
words = url.split(';')
return words[0], words[1:]
+
def splitvalue(attr):
+ warnings.warn("urllib.parse.splitvalue() is deprecated as of 3.8, "
+ "use urllib.parse.parse_qsl() instead",
+ DeprecationWarning, stacklevel=2)
+ return _splitvalue(attr)
+
+
+def _splitvalue(attr):
"""splitvalue('attr=value') --> 'attr', 'value'."""
attr, delim, value = attr.partition('=')
return attr, (value if delim else None)
diff --git a/Lib/urllib/request.py b/Lib/urllib/request.py
index 5f15b74f4..e44073886 100644
--- a/Lib/urllib/request.py
+++ b/Lib/urllib/request.py
@@ -94,7 +94,6 @@ import socket
import string
import sys
import time
-import collections
import tempfile
import contextlib
import warnings
@@ -103,8 +102,8 @@ import warnings
from urllib.error import URLError, HTTPError, ContentTooShortError
from urllib.parse import (
urlparse, urlsplit, urljoin, unwrap, quote, unquote,
- splittype, splithost, splitport, splituser, splitpasswd,
- splitattr, splitquery, splitvalue, splittag, to_bytes,
+ _splittype, _splithost, _splitport, _splituser, _splitpasswd,
+ _splitattr, _splitquery, _splitvalue, _splittag, _to_bytes,
unquote_to_bytes, urlunparse)
from urllib.response import addinfourl, addclosehook
@@ -199,7 +198,7 @@ def urlopen(url, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
global _opener
if cafile or capath or cadefault:
import warnings
- warnings.warn("cafile, cpath and cadefault are deprecated, use a "
+ warnings.warn("cafile, capath and cadefault are deprecated, use a "
"custom context instead.", DeprecationWarning, 2)
if context is not None:
raise ValueError(
@@ -243,7 +242,7 @@ def urlretrieve(url, filename=None, reporthook=None, data=None):
Returns a tuple containing the path to the newly created
data file as well as the resulting HTTPMessage object.
"""
- url_type, path = splittype(url)
+ url_type, path = _splittype(url)
with contextlib.closing(urlopen(url, data)) as fp:
headers = fp.info()
@@ -351,7 +350,7 @@ class Request:
def full_url(self, url):
# unwrap('') --> 'type://host/path'
self._full_url = unwrap(url)
- self._full_url, self.fragment = splittag(self._full_url)
+ self._full_url, self.fragment = _splittag(self._full_url)
self._parse()
@full_url.deleter
@@ -379,10 +378,10 @@ class Request:
self.data = None
def _parse(self):
- self.type, rest = splittype(self._full_url)
+ self.type, rest = _splittype(self._full_url)
if self.type is None:
raise ValueError("unknown url type: %r" % self.full_url)
- self.host, self.selector = splithost(rest)
+ self.host, self.selector = _splithost(rest)
if self.host:
self.host = unquote(self.host)
@@ -427,8 +426,7 @@ class Request:
self.unredirected_hdrs.pop(header_name, None)
def header_items(self):
- hdrs = self.unredirected_hdrs.copy()
- hdrs.update(self.headers)
+ hdrs = {**self.unredirected_hdrs, **self.headers}
return list(hdrs.items())
class OpenerDirector:
@@ -523,6 +521,7 @@ class OpenerDirector:
meth = getattr(processor, meth_name)
req = meth(req)
+ sys.audit('urllib.Request', req.full_url, req.data, req.headers, req.get_method())
response = self._open(req, data)
# post-process response
@@ -684,8 +683,8 @@ class HTTPRedirectHandler(BaseHandler):
newurl = newurl.replace(' ', '%20')
CONTENT_HEADERS = ("content-length", "content-type")
- newheaders = dict((k, v) for k, v in req.headers.items()
- if k.lower() not in CONTENT_HEADERS)
+ newheaders = {k: v for k, v in req.headers.items()
+ if k.lower() not in CONTENT_HEADERS}
return Request(newurl,
headers=newheaders,
origin_req_host=req.origin_req_host,
@@ -769,7 +768,7 @@ def _parse_proxy(proxy):
According to RFC 3986, having an authority component means the URL must
have two slashes after the scheme.
"""
- scheme, r_scheme = splittype(proxy)
+ scheme, r_scheme = _splittype(proxy)
if not r_scheme.startswith("/"):
# authority
scheme = None
@@ -784,9 +783,9 @@ def _parse_proxy(proxy):
if end == -1:
end = None
authority = r_scheme[2:end]
- userinfo, hostport = splituser(authority)
+ userinfo, hostport = _splituser(authority)
if userinfo is not None:
- user, password = splitpasswd(userinfo)
+ user, password = _splitpasswd(userinfo)
else:
user = password = None
return scheme, user, password, hostport
@@ -801,6 +800,7 @@ class ProxyHandler(BaseHandler):
assert hasattr(proxies, 'keys'), "proxies must be a mapping"
self.proxies = proxies
for type, url in proxies.items():
+ type = type.lower()
setattr(self, '%s_open' % type,
lambda r, proxy=url, type=type, meth=self.proxy_open:
meth(r, proxy, type))
@@ -846,7 +846,7 @@ class HTTPPasswordMgr:
self.passwd[realm] = {}
for default_port in True, False:
reduced_uri = tuple(
- [self.reduce_uri(u, default_port) for u in uri])
+ self.reduce_uri(u, default_port) for u in uri)
self.passwd[realm][reduced_uri] = (user, passwd)
def find_user_password(self, realm, authuri):
@@ -873,7 +873,7 @@ class HTTPPasswordMgr:
scheme = None
authority = uri
path = '/'
- host, port = splitport(authority)
+ host, port = _splitport(authority)
if default_port and port is None and scheme is not None:
dport = {"http": 80,
"https": 443,
@@ -945,8 +945,15 @@ class AbstractBasicAuthHandler:
# allow for double- and single-quoted realm values
# (single quotes are a violation of the RFC, but appear in the wild)
- rx = re.compile('(?:.*,)*[ \t]*([^ \t]+)[ \t]+'
- 'realm=(["\']?)([^"\']*)\\2', re.I)
+ rx = re.compile('(?:^|,)' # start of the string or ','
+ '[ \t]*' # optional whitespaces
+ '([^ \t]+)' # scheme like "Basic"
+ '[ \t]+' # mandatory whitespaces
+ # realm=xxx
+ # realm='xxx'
+ # realm="xxx"
+ 'realm=(["\']?)([^"\']*)\\2',
+ re.I)
# XXX could pre-emptively send auth info already accepted (RFC 2617,
# end of section 2, and section 1.2 immediately after "credentials"
@@ -958,27 +965,51 @@ class AbstractBasicAuthHandler:
self.passwd = password_mgr
self.add_password = self.passwd.add_password
+ def _parse_realm(self, header):
+ # parse WWW-Authenticate header: accept multiple challenges per header
+ found_challenge = False
+ for mo in AbstractBasicAuthHandler.rx.finditer(header):
+ scheme, quote, realm = mo.groups()
+ if quote not in ['"', "'"]:
+ warnings.warn("Basic Auth Realm was unquoted",
+ UserWarning, 3)
+
+ yield (scheme, realm)
+
+ found_challenge = True
+
+ if not found_challenge:
+ if header:
+ scheme = header.split()[0]
+ else:
+ scheme = ''
+ yield (scheme, None)
+
def http_error_auth_reqed(self, authreq, host, req, headers):
# host may be an authority (without userinfo) or a URL with an
# authority
- # XXX could be multiple headers
- authreq = headers.get(authreq, None)
+ headers = headers.get_all(authreq)
+ if not headers:
+ # no header found
+ return
- if authreq:
- scheme = authreq.split()[0]
- if scheme.lower() != 'basic':
- raise ValueError("AbstractBasicAuthHandler does not"
- " support the following scheme: '%s'" %
- scheme)
- else:
- mo = AbstractBasicAuthHandler.rx.search(authreq)
- if mo:
- scheme, quote, realm = mo.groups()
- if quote not in ['"',"'"]:
- warnings.warn("Basic Auth Realm was unquoted",
- UserWarning, 2)
- if scheme.lower() == 'basic':
- return self.retry_http_basic_auth(host, req, realm)
+ unsupported = None
+ for header in headers:
+ for scheme, realm in self._parse_realm(header):
+ if scheme.lower() != 'basic':
+ unsupported = scheme
+ continue
+
+ if realm is not None:
+ # Use the first matching Basic challenge.
+ # Ignore following challenges even if they use the Basic
+ # scheme.
+ return self.retry_http_basic_auth(host, req, realm)
+
+ if unsupported is not None:
+ raise ValueError("AbstractBasicAuthHandler does not "
+ "support the following scheme: %r"
+ % (scheme,))
def retry_http_basic_auth(self, host, req, realm):
user, pw = self.passwd.find_user_password(realm, host)
@@ -1144,7 +1175,11 @@ class AbstractDigestAuthHandler:
A2 = "%s:%s" % (req.get_method(),
# XXX selector: what about proxies and full urls
req.selector)
- if qop == 'auth':
+ # NOTE: As per RFC 2617, when server sends "auth,auth-int", the client could use either `auth`
+ # or `auth-int` to the response back. we use `auth` to send the response back.
+ if qop is None:
+ respdig = KD(H(A1), "%s:%s" % (nonce, H(A2)))
+ elif 'auth' in qop.split(','):
if nonce == self.last_nonce:
self.nonce_count += 1
else:
@@ -1152,10 +1187,8 @@ class AbstractDigestAuthHandler:
self.last_nonce = nonce
ncvalue = '%08x' % self.nonce_count
cnonce = self.get_cnonce(nonce)
- noncebit = "%s:%s:%s:%s:%s" % (nonce, ncvalue, cnonce, qop, H(A2))
+ noncebit = "%s:%s:%s:%s:%s" % (nonce, ncvalue, cnonce, 'auth', H(A2))
respdig = KD(H(A1), noncebit)
- elif qop is None:
- respdig = KD(H(A1), "%s:%s" % (nonce, H(A2)))
else:
# XXX handle auth-int.
raise URLError("qop '%s' is not supported." % qop)
@@ -1262,8 +1295,8 @@ class AbstractHTTPHandler(BaseHandler):
sel_host = host
if request.has_proxy():
- scheme, sel = splittype(request.selector)
- sel_host, sel_path = splithost(sel)
+ scheme, sel = _splittype(request.selector)
+ sel_host, sel_path = _splithost(sel)
if not request.has_header('Host'):
request.add_unredirected_header('Host', sel_host)
for name, value in self.parent.addheaders:
@@ -1287,8 +1320,8 @@ class AbstractHTTPHandler(BaseHandler):
h.set_debuglevel(self._debuglevel)
headers = dict(req.unredirected_hdrs)
- headers.update(dict((k, v) for k, v in req.headers.items()
- if k not in headers))
+ headers.update({k: v for k, v in req.headers.items()
+ if k not in headers})
# TODO(jhylton): Should this be redesigned to handle
# persistent connections?
@@ -1300,7 +1333,7 @@ class AbstractHTTPHandler(BaseHandler):
# So make sure the connection gets closed after the (only)
# request.
headers["Connection"] = "close"
- headers = dict((name.title(), val) for name, val in headers.items())
+ headers = {name.title(): val for name, val in headers.items()}
if req._tunnel_host:
tunnel_headers = {}
@@ -1479,7 +1512,7 @@ class FileHandler(BaseHandler):
'Content-type: %s\nContent-length: %d\nLast-modified: %s\n' %
(mtype or 'text/plain', size, modified))
if host:
- host, port = splitport(host)
+ host, port = _splitport(host)
if not host or \
(not port and _safe_gethostbyname(host) in self.get_names()):
if host:
@@ -1488,7 +1521,6 @@ class FileHandler(BaseHandler):
origurl = 'file://' + filename
return addinfourl(open(localfile, 'rb'), headers, origurl)
except OSError as exp:
- # users shouldn't expect OSErrors coming from urlopen()
raise URLError(exp)
raise URLError('file not on local host')
@@ -1505,16 +1537,16 @@ class FTPHandler(BaseHandler):
host = req.host
if not host:
raise URLError('ftp error: no host given')
- host, port = splitport(host)
+ host, port = _splitport(host)
if port is None:
port = ftplib.FTP_PORT
else:
port = int(port)
# username/password handling
- user, host = splituser(host)
+ user, host = _splituser(host)
if user:
- user, passwd = splitpasswd(user)
+ user, passwd = _splitpasswd(user)
else:
passwd = None
host = unquote(host)
@@ -1525,7 +1557,7 @@ class FTPHandler(BaseHandler):
host = socket.gethostbyname(host)
except OSError as msg:
raise URLError(msg)
- path, attrs = splitattr(req.selector)
+ path, attrs = _splitattr(req.selector)
dirs = path.split('/')
dirs = list(map(unquote, dirs))
dirs, file = dirs[:-1], dirs[-1]
@@ -1535,7 +1567,7 @@ class FTPHandler(BaseHandler):
fw = self.connect_ftp(user, passwd, host, port, dirs, req.timeout)
type = file and 'I' or 'D'
for attr in attrs:
- attr, value = splitvalue(attr)
+ attr, value = _splitvalue(attr)
if attr.lower() == 'type' and \
value in ('a', 'A', 'i', 'I', 'd', 'D'):
type = value.upper()
@@ -1658,14 +1690,10 @@ else:
of the 'file' scheme; not recommended for general use."""
return quote(pathname)
-# This really consists of two pieces:
-# (1) a class which handles opening of all sorts of URLs
-# (plus assorted utilities etc.)
-# (2) a set of functions for parsing URLs
-# XXX Should these be separated out into different modules?
-
ftpcache = {}
+
+
class URLopener:
"""Class to open URLs.
This is a class rather than just a subroutine because we may need
@@ -1733,26 +1761,26 @@ class URLopener:
# External interface
def open(self, fullurl, data=None):
"""Use URLopener().open(file) instead of open(file, 'r')."""
- fullurl = unwrap(to_bytes(fullurl))
+ fullurl = unwrap(_to_bytes(fullurl))
fullurl = quote(fullurl, safe="%/:=&?~#+!$,;'@()*[]|")
if self.tempcache and fullurl in self.tempcache:
filename, headers = self.tempcache[fullurl]
fp = open(filename, 'rb')
return addinfourl(fp, headers, fullurl)
- urltype, url = splittype(fullurl)
+ urltype, url = _splittype(fullurl)
if not urltype:
urltype = 'file'
if urltype in self.proxies:
proxy = self.proxies[urltype]
- urltype, proxyhost = splittype(proxy)
- host, selector = splithost(proxyhost)
+ urltype, proxyhost = _splittype(proxy)
+ host, selector = _splithost(proxyhost)
url = (host, fullurl) # Signal special case to open_*()
else:
proxy = None
name = 'open_' + urltype
self.type = urltype
name = name.replace('-', '_')
- if not hasattr(self, name):
+ if not hasattr(self, name) or name == 'open_local_file':
if proxy:
return self.open_unknown_proxy(proxy, fullurl, data)
else:
@@ -1769,28 +1797,28 @@ class URLopener:
def open_unknown(self, fullurl, data=None):
"""Overridable interface to open unknown URL type."""
- type, url = splittype(fullurl)
+ type, url = _splittype(fullurl)
raise OSError('url error', 'unknown url type', type)
def open_unknown_proxy(self, proxy, fullurl, data=None):
"""Overridable interface to open unknown URL type."""
- type, url = splittype(fullurl)
+ type, url = _splittype(fullurl)
raise OSError('url error', 'invalid proxy for %s' % type, proxy)
# External interface
def retrieve(self, url, filename=None, reporthook=None, data=None):
"""retrieve(url) returns (filename, headers) for a local object
or (tempfilename, headers) for a remote object."""
- url = unwrap(to_bytes(url))
+ url = unwrap(_to_bytes(url))
if self.tempcache and url in self.tempcache:
return self.tempcache[url]
- type, url1 = splittype(url)
+ type, url1 = _splittype(url)
if filename is None and (not type or type == 'file'):
try:
fp = self.open_local_file(url1)
hdrs = fp.info()
fp.close()
- return url2pathname(splithost(url1)[1]), hdrs
+ return url2pathname(_splithost(url1)[1]), hdrs
except OSError as msg:
pass
fp = self.open(url, data)
@@ -1799,11 +1827,10 @@ class URLopener:
if filename:
tfp = open(filename, 'wb')
else:
- import tempfile
- garbage, path = splittype(url)
- garbage, path = splithost(path or "")
- path, garbage = splitquery(path or "")
- path, garbage = splitattr(path or "")
+ garbage, path = _splittype(url)
+ garbage, path = _splithost(path or "")
+ path, garbage = _splitquery(path or "")
+ path, garbage = _splitattr(path or "")
suffix = os.path.splitext(path)[1]
(fd, filename) = tempfile.mkstemp(suffix)
self.__tempfiles.append(filename)
@@ -1860,25 +1887,25 @@ class URLopener:
user_passwd = None
proxy_passwd= None
if isinstance(url, str):
- host, selector = splithost(url)
+ host, selector = _splithost(url)
if host:
- user_passwd, host = splituser(host)
+ user_passwd, host = _splituser(host)
host = unquote(host)
realhost = host
else:
host, selector = url
# check whether the proxy contains authorization information
- proxy_passwd, host = splituser(host)
+ proxy_passwd, host = _splituser(host)
# now we proceed with the url we want to obtain
- urltype, rest = splittype(selector)
+ urltype, rest = _splittype(selector)
url = rest
user_passwd = None
if urltype.lower() != 'http':
realhost = None
else:
- realhost, rest = splithost(rest)
+ realhost, rest = _splithost(rest)
if realhost:
- user_passwd, realhost = splituser(realhost)
+ user_passwd, realhost = _splituser(realhost)
if user_passwd:
selector = "%s://%s%s" % (urltype, realhost, rest)
if proxy_bypass(realhost):
@@ -1984,7 +2011,7 @@ class URLopener:
"""Use local file."""
import email.utils
import mimetypes
- host, file = splithost(url)
+ host, file = _splithost(url)
localname = url2pathname(file)
try:
stats = os.stat(localname)
@@ -2001,7 +2028,7 @@ class URLopener:
if file[:1] == '/':
urlfile = 'file://' + file
return addinfourl(open(localname, 'rb'), headers, urlfile)
- host, port = splitport(host)
+ host, port = _splitport(host)
if (not port
and socket.gethostbyname(host) in ((localhost(),) + thishost())):
urlfile = file
@@ -2017,11 +2044,11 @@ class URLopener:
if not isinstance(url, str):
raise URLError('ftp error: proxy support for ftp protocol currently not implemented')
import mimetypes
- host, path = splithost(url)
+ host, path = _splithost(url)
if not host: raise URLError('ftp error: no host given')
- host, port = splitport(host)
- user, host = splituser(host)
- if user: user, passwd = splitpasswd(user)
+ host, port = _splitport(host)
+ user, host = _splituser(host)
+ if user: user, passwd = _splitpasswd(user)
else: passwd = None
host = unquote(host)
user = unquote(user or '')
@@ -2032,7 +2059,7 @@ class URLopener:
port = ftplib.FTP_PORT
else:
port = int(port)
- path, attrs = splitattr(path)
+ path, attrs = _splitattr(path)
path = unquote(path)
dirs = path.split('/')
dirs, file = dirs[:-1], dirs[-1]
@@ -2054,7 +2081,7 @@ class URLopener:
if not file: type = 'D'
else: type = 'I'
for attr in attrs:
- attr, value = splitvalue(attr)
+ attr, value = _splitvalue(attr)
if attr.lower() == 'type' and \
value in ('a', 'A', 'i', 'I', 'd', 'D'):
type = value.upper()
@@ -2237,11 +2264,11 @@ class FancyURLopener(URLopener):
return getattr(self,name)(url, realm, data)
def retry_proxy_http_basic_auth(self, url, realm, data=None):
- host, selector = splithost(url)
+ host, selector = _splithost(url)
newurl = 'http://' + host + selector
proxy = self.proxies['http']
- urltype, proxyhost = splittype(proxy)
- proxyhost, proxyselector = splithost(proxyhost)
+ urltype, proxyhost = _splittype(proxy)
+ proxyhost, proxyselector = _splithost(proxyhost)
i = proxyhost.find('@') + 1
proxyhost = proxyhost[i:]
user, passwd = self.get_user_passwd(proxyhost, realm, i)
@@ -2255,11 +2282,11 @@ class FancyURLopener(URLopener):
return self.open(newurl, data)
def retry_proxy_https_basic_auth(self, url, realm, data=None):
- host, selector = splithost(url)
+ host, selector = _splithost(url)
newurl = 'https://' + host + selector
proxy = self.proxies['https']
- urltype, proxyhost = splittype(proxy)
- proxyhost, proxyselector = splithost(proxyhost)
+ urltype, proxyhost = _splittype(proxy)
+ proxyhost, proxyselector = _splithost(proxyhost)
i = proxyhost.find('@') + 1
proxyhost = proxyhost[i:]
user, passwd = self.get_user_passwd(proxyhost, realm, i)
@@ -2273,7 +2300,7 @@ class FancyURLopener(URLopener):
return self.open(newurl, data)
def retry_http_basic_auth(self, url, realm, data=None):
- host, selector = splithost(url)
+ host, selector = _splithost(url)
i = host.find('@') + 1
host = host[i:]
user, passwd = self.get_user_passwd(host, realm, i)
@@ -2287,7 +2314,7 @@ class FancyURLopener(URLopener):
return self.open(newurl, data)
def retry_https_basic_auth(self, url, realm, data=None):
- host, selector = splithost(url)
+ host, selector = _splithost(url)
i = host.find('@') + 1
host = host[i:]
user, passwd = self.get_user_passwd(host, realm, i)
@@ -2504,23 +2531,26 @@ def proxy_bypass_environment(host, proxies=None):
try:
no_proxy = proxies['no']
except KeyError:
- return 0
+ return False
# '*' is special case for always bypass
if no_proxy == '*':
- return 1
+ return True
+ host = host.lower()
# strip port off host
- hostonly, port = splitport(host)
+ hostonly, port = _splitport(host)
# check if the host ends with any of the DNS suffixes
- no_proxy_list = [proxy.strip() for proxy in no_proxy.split(',')]
- for name in no_proxy_list:
+ for name in no_proxy.split(','):
+ name = name.strip()
if name:
- name = re.escape(name)
- pattern = r'(.+\.)?%s$' % name
- if (re.match(pattern, hostonly, re.I)
- or re.match(pattern, host, re.I)):
- return 1
+ name = name.lstrip('.') # ignore leading dots
+ name = name.lower()
+ if hostonly == name or host == name:
+ return True
+ name = '.' + name
+ if hostonly.endswith(name) or host.endswith(name):
+ return True
# otherwise, don't bypass
- return 0
+ return False
# This code tests an OSX specific data structure but is testable on all
@@ -2539,7 +2569,7 @@ def _proxy_bypass_macosx_sysconf(host, proxy_settings):
"""
from fnmatch import fnmatch
- hostonly, port = splitport(host)
+ hostonly, port = _splitport(host)
def ip2num(ipAddr):
parts = ipAddr.split('.')
@@ -2646,7 +2676,7 @@ elif os.name == 'nt':
for p in proxyServer.split(';'):
protocol, address = p.split('=', 1)
# See if address has a type:// prefix
- if not re.match('^([^/:]+)://', address):
+ if not re.match('(?:[^/:]+)://', address):
address = '%s://%s' % (protocol, address)
proxies[protocol] = address
else:
@@ -2693,7 +2723,7 @@ elif os.name == 'nt':
if not proxyEnable or not proxyOverride:
return 0
# try to make a host list from name and IP address.
- rawHost, port = splitport(host)
+ rawHost, port = _splitport(host)
host = [rawHost]
try:
addr = socket.gethostbyname(rawHost)
diff --git a/Lib/urllib/robotparser.py b/Lib/urllib/robotparser.py
index 9dab4c1c3..c58565e39 100644
--- a/Lib/urllib/robotparser.py
+++ b/Lib/urllib/robotparser.py
@@ -16,6 +16,9 @@ import urllib.request
__all__ = ["RobotFileParser"]
+RequestRate = collections.namedtuple("RequestRate", "requests seconds")
+
+
class RobotFileParser:
""" This class provides a set of methods to read, parse and answer
questions about a single robots.txt file.
@@ -24,6 +27,7 @@ class RobotFileParser:
def __init__(self, url=''):
self.entries = []
+ self.sitemaps = []
self.default_entry = None
self.disallow_all = False
self.allow_all = False
@@ -136,12 +140,14 @@ class RobotFileParser:
# check if all values are sane
if (len(numbers) == 2 and numbers[0].strip().isdigit()
and numbers[1].strip().isdigit()):
- req_rate = collections.namedtuple('req_rate',
- 'requests seconds')
- entry.req_rate = req_rate
- entry.req_rate.requests = int(numbers[0])
- entry.req_rate.seconds = int(numbers[1])
+ entry.req_rate = RequestRate(int(numbers[0]), int(numbers[1]))
state = 2
+ elif line[0] == "sitemap":
+ # According to http://www.sitemaps.org/protocol.html
+ # "This directive is independent of the user-agent line,
+ # so it doesn't matter where you place it in your file."
+ # Therefore we do not change the state of the parser.
+ self.sitemaps.append(line[1])
if state == 2:
self._add_entry(entry)
@@ -180,7 +186,9 @@ class RobotFileParser:
for entry in self.entries:
if entry.applies_to(useragent):
return entry.delay
- return self.default_entry.delay
+ if self.default_entry:
+ return self.default_entry.delay
+ return None
def request_rate(self, useragent):
if not self.mtime():
@@ -188,10 +196,20 @@ class RobotFileParser:
for entry in self.entries:
if entry.applies_to(useragent):
return entry.req_rate
- return self.default_entry.req_rate
+ if self.default_entry:
+ return self.default_entry.req_rate
+ return None
+
+ def site_maps(self):
+ if not self.sitemaps:
+ return None
+ return self.sitemaps
def __str__(self):
- return ''.join([str(entry) + "\n" for entry in self.entries])
+ entries = self.entries
+ if self.default_entry is not None:
+ entries = entries + [self.default_entry]
+ return '\n\n'.join(map(str, entries))
class RuleLine:
@@ -223,10 +241,14 @@ class Entry:
def __str__(self):
ret = []
for agent in self.useragents:
- ret.extend(["User-agent: ", agent, "\n"])
- for line in self.rulelines:
- ret.extend([str(line), "\n"])
- return ''.join(ret)
+ ret.append(f"User-agent: {agent}")
+ if self.delay is not None:
+ ret.append(f"Crawl-delay: {self.delay}")
+ if self.req_rate is not None:
+ rate = self.req_rate
+ ret.append(f"Request-rate: {rate.requests}/{rate.seconds}")
+ ret.extend(map(str, self.rulelines))
+ return '\n'.join(ret)
def applies_to(self, useragent):
"""check if this entry applies to the specified agent"""
diff --git a/vm/src/stdlib/io.rs b/vm/src/stdlib/io.rs
index 3e93f0549..0133faf90 100644
--- a/vm/src/stdlib/io.rs
+++ b/vm/src/stdlib/io.rs
@@ -1,9 +1,11 @@
/*
* I/O core tools.
*/
+use std::convert::TryInto;
use std::fs;
use std::io::{self, prelude::*, Cursor, SeekFrom};
+use bstr::ByteSlice;
use crossbeam_utils::atomic::AtomicCell;
use num_traits::ToPrimitive;
@@ -109,12 +111,39 @@ impl BufferedIO {
self.cursor.position()
}
- fn readline(&mut self) -> Option {
- let mut buf = String::new();
+ fn readline(&mut self, size: OptionalOption, vm: &VirtualMachine) -> PyResult> {
+ let mut buf = Vec::new();
+ let mut left = size
+ .flatten()
+ .filter(|s| *s >= 0)
+ .and_then(|s| s.try_into().ok())
+ .unwrap_or(usize::MAX);
+ if left == 0 {
+ return Ok(buf);
+ }
+ loop {
+ let (done, used) = {
+ let mut available = self.cursor.fill_buf().map_err(|err| os_err(vm, err))?;
+ if left < available.len() {
+ available = &available[..left];
+ }
- match self.cursor.read_line(&mut buf) {
- Ok(_) => Some(buf),
- Err(_) => None,
+ match available.find_byte(b'\n') {
+ Some(i) => {
+ buf.extend_from_slice(&available[..=i]);
+ (true, i + 1)
+ }
+ _ => {
+ buf.extend_from_slice(available);
+ (false, available.len())
+ }
+ }
+ };
+ self.cursor.consume(used);
+ left -= used;
+ if done || used == 0 || left == 0 {
+ return Ok(buf);
+ }
}
}
}
@@ -195,10 +224,12 @@ impl PyStringIORef {
Ok(self.buffer(vm)?.tell())
}
- fn readline(self, vm: &VirtualMachine) -> PyResult {
- match self.buffer(vm)?.readline() {
- Some(line) => Ok(line),
- None => Err(vm.new_value_error("Error Performing Operation".to_owned())),
+ fn readline(self, size: OptionalOption, vm: &VirtualMachine) -> PyResult {
+ // TODO size should correspond to the number of characters, at the moments its the number of
+ // bytes.
+ match String::from_utf8(self.buffer(vm)?.readline(size, vm)?) {
+ Ok(value) => Ok(value),
+ Err(_) => Err(vm.new_value_error("Error Retrieving Value".to_owned())),
}
}
@@ -307,11 +338,8 @@ impl PyBytesIORef {
Ok(self.buffer(vm)?.tell())
}
- fn readline(self, vm: &VirtualMachine) -> PyResult> {
- match self.buffer(vm)?.readline() {
- Some(line) => Ok(line.as_bytes().to_vec()),
- None => Err(vm.new_value_error("Error Performing Operation".to_owned())),
- }
+ fn readline(self, size: OptionalOption, vm: &VirtualMachine) -> PyResult> {
+ self.buffer(vm)?.readline(size, vm)
}
fn truncate(self, size: OptionalOption, vm: &VirtualMachine) -> PyResult<()> {
@@ -812,6 +840,17 @@ fn buffered_writer_seekable(_self: PyObjectRef) -> bool {
true
}
+fn buffered_writer_seek(
+ instance: PyObjectRef,
+ offset: PyObjectRef,
+ how: OptionalArg,
+ vm: &VirtualMachine,
+) -> PyResult {
+ let raw = vm.get_attribute(instance, "raw")?;
+ let args: Vec<_> = std::iter::once(offset).chain(how.into_option()).collect();
+ vm.invoke(&vm.get_attribute(raw, "seek")?, args)
+}
+
#[derive(FromArgs)]
struct TextIOWrapperArgs {
#[pyarg(positional_or_keyword, optional = false)]
@@ -1149,7 +1188,7 @@ pub fn io_open(
// There are 3 possible classes here, each inheriting from the RawBaseIO
// creating || writing || appending => BufferedWriter
let buffered = match mode.chars().next().unwrap() {
- 'w' => {
+ 'w' | 'a' => {
let buffered_writer_class = vm
.get_attribute(io_module.clone(), "BufferedWriter")
.unwrap();
@@ -1162,7 +1201,7 @@ pub fn io_open(
vm.invoke(&buffered_reader_class, vec![file_io_obj.clone()])
}
//TODO: updating => PyBufferedRandom
- _ => unimplemented!("'a' mode is not yet implemented"),
+ _ => unimplemented!("'+' modes is not yet implemented"),
};
match typ.chars().next().unwrap() {
@@ -1237,6 +1276,7 @@ pub fn make_module(vm: &VirtualMachine) -> PyObjectRef {
"__init__" => ctx.new_method(buffered_io_base_init),
"write" => ctx.new_method(buffered_writer_write),
"seekable" => ctx.new_method(buffered_writer_seekable),
+ "seek" => ctx.new_method(buffered_writer_seek),
"fileno" => ctx.new_method(buffered_io_base_fileno),
"tell" => ctx.new_method(buffered_io_base_tell),
"close" => ctx.new_method(buffered_io_base_close),