mirror of
https://github.com/RustPython/RustPython.git
synced 2026-10-29 11:14:38 +08:00
8 files changed
+2044
-249
No files matched your search
+992
@@ -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() -- <I>italics?</I>")
|
||||
def g(f=f):
|
||||
f()
|
||||
print("<H3>What follows is a test, not an actual exception:</H3>")
|
||||
g()
|
||||
except:
|
||||
print_exception()
|
||||
|
||||
print("<H1>Second try with a small maxlen...</H1>")
|
||||
|
||||
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("<H3>Traceback (most recent call last):</H3>")
|
||||
list = traceback.format_tb(tb, limit) + \
|
||||
traceback.format_exception_only(type, value)
|
||||
print("<PRE>%s<B>%s</B></PRE>" % (
|
||||
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("<H3>Shell Environment:</H3>")
|
||||
print("<DL>")
|
||||
for key in keys:
|
||||
print("<DT>", html.escape(key), "<DD>", html.escape(environ[key]))
|
||||
print("</DL>")
|
||||
print()
|
||||
|
||||
def print_form(form):
|
||||
"""Dump the contents of a form as HTML."""
|
||||
keys = sorted(form.keys())
|
||||
print()
|
||||
print("<H3>Form Contents:</H3>")
|
||||
if not keys:
|
||||
print("<P>No form fields.")
|
||||
print("<DL>")
|
||||
for key in keys:
|
||||
print("<DT>" + html.escape(key) + ":", end=' ')
|
||||
value = form[key]
|
||||
print("<i>" + html.escape(repr(type(value))) + "</i>")
|
||||
print("<DD>" + html.escape(repr(value)))
|
||||
print("</DL>")
|
||||
print()
|
||||
|
||||
def print_directory():
|
||||
"""Dump the current directory as HTML."""
|
||||
print()
|
||||
print("<H3>Current Working Directory:</H3>")
|
||||
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("<H3>Command Line Arguments:</H3>")
|
||||
print()
|
||||
print(sys.argv)
|
||||
print()
|
||||
|
||||
def print_environ_usage():
|
||||
"""Dump a list of environment variables used by CGI as HTML."""
|
||||
print("""
|
||||
<H3>These environment variables could have been set:</H3>
|
||||
<UL>
|
||||
<LI>AUTH_TYPE
|
||||
<LI>CONTENT_LENGTH
|
||||
<LI>CONTENT_TYPE
|
||||
<LI>DATE_GMT
|
||||
<LI>DATE_LOCAL
|
||||
<LI>DOCUMENT_NAME
|
||||
<LI>DOCUMENT_ROOT
|
||||
<LI>DOCUMENT_URI
|
||||
<LI>GATEWAY_INTERFACE
|
||||
<LI>LAST_MODIFIED
|
||||
<LI>PATH
|
||||
<LI>PATH_INFO
|
||||
<LI>PATH_TRANSLATED
|
||||
<LI>QUERY_STRING
|
||||
<LI>REMOTE_ADDR
|
||||
<LI>REMOTE_HOST
|
||||
<LI>REMOTE_IDENT
|
||||
<LI>REMOTE_USER
|
||||
<LI>REQUEST_METHOD
|
||||
<LI>SCRIPT_NAME
|
||||
<LI>SERVER_NAME
|
||||
<LI>SERVER_PORT
|
||||
<LI>SERVER_PROTOCOL
|
||||
<LI>SERVER_ROOT
|
||||
<LI>SERVER_SOFTWARE
|
||||
</UL>
|
||||
In addition, HTTP headers sent by the server may be passed in the
|
||||
environment as well. Here are some common variable names:
|
||||
<UL>
|
||||
<LI>HTTP_ACCEPT
|
||||
<LI>HTTP_CONNECTION
|
||||
<LI>HTTP_HOST
|
||||
<LI>HTTP_PRAGMA
|
||||
<LI>HTTP_REFERER
|
||||
<LI>HTTP_USER_AGENT
|
||||
</UL>
|
||||
""")
|
||||
|
||||
|
||||
# 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()
|
||||
@@ -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()
|
||||
@@ -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('')
|
||||
|
||||
+1
-5
@@ -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):
|
||||
|
||||
+212
-56
@@ -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:
|
||||
<scheme>://<netloc>/<path>?<query>#<fragment>
|
||||
@@ -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('<URL:type://host/path>') --> 'type://host/path'."""
|
||||
"""Transform a string like '<URL:scheme://host/path>' 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)
|
||||
+139
-109
@@ -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('<URL:type://host/path>') --> '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)
|
||||
|
||||
+34
-12
@@ -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"""
|
||||
|
||||
+56
-16
@@ -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<String> {
|
||||
let mut buf = String::new();
|
||||
fn readline(&mut self, size: OptionalOption<i64>, vm: &VirtualMachine) -> PyResult<Vec<u8>> {
|
||||
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<String> {
|
||||
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<i64>, vm: &VirtualMachine) -> PyResult<String> {
|
||||
// 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<Vec<u8>> {
|
||||
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<i64>, vm: &VirtualMachine) -> PyResult<Vec<u8>> {
|
||||
self.buffer(vm)?.readline(size, vm)
|
||||
}
|
||||
|
||||
fn truncate(self, size: OptionalOption<usize>, 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),
|
||||
|
||||
Reference in new issue
Block a user