diff --git a/Lib/cgi.py b/Lib/cgi.py new file mode 100755 index 000000000..c22c71b38 --- /dev/null +++ b/Lib/cgi.py @@ -0,0 +1,992 @@ +#! /usr/local/bin/python + +# NOTE: the above "/usr/local/bin/python" is NOT a mistake. It is +# intentionally NOT "/usr/bin/env python". On many systems +# (e.g. Solaris), /usr/local/bin is not in $PATH as passed to CGI +# scripts, and /usr/local/bin is the default directory where Python is +# installed, so /usr/bin/env would be unable to find python. Granted, +# binary installations by Linux vendors often install Python in +# /usr/bin. So let those vendors patch cgi.py to match their choice +# of installation. + +"""Support module for CGI (Common Gateway Interface) scripts. + +This module defines a number of utilities for use by CGI scripts +written in Python. +""" + +# History +# ------- +# +# Michael McLay started this module. Steve Majewski changed the +# interface to SvFormContentDict and FormContentDict. The multipart +# parsing was inspired by code submitted by Andreas Paepcke. Guido van +# Rossum rewrote, reformatted and documented the module and is currently +# responsible for its maintenance. +# + +__version__ = "2.6" + + +# Imports +# ======= + +from io import StringIO, BytesIO, TextIOWrapper +from collections.abc import Mapping +import sys +import os +import urllib.parse +from email.parser import FeedParser +from email.message import Message +import html +import locale +import tempfile + +__all__ = ["MiniFieldStorage", "FieldStorage", "parse", "parse_multipart", + "parse_header", "test", "print_exception", "print_environ", + "print_form", "print_directory", "print_arguments", + "print_environ_usage"] + +# Logging support +# =============== + +logfile = "" # Filename to log to, if not empty +logfp = None # File object to log to, if not None + +def initlog(*allargs): + """Write a log message, if there is a log file. + + Even though this function is called initlog(), you should always + use log(); log is a variable that is set either to initlog + (initially), to dolog (once the log file has been opened), or to + nolog (when logging is disabled). + + The first argument is a format string; the remaining arguments (if + any) are arguments to the % operator, so e.g. + log("%s: %s", "a", "b") + will write "a: b" to the log file, followed by a newline. + + If the global logfp is not None, it should be a file object to + which log data is written. + + If the global logfp is None, the global logfile may be a string + giving a filename to open, in append mode. This file should be + world writable!!! If the file can't be opened, logging is + silently disabled (since there is no safe place where we could + send an error message). + + """ + global log, logfile, logfp + if logfile and not logfp: + try: + logfp = open(logfile, "a") + except OSError: + pass + if not logfp: + log = nolog + else: + log = dolog + log(*allargs) + +def dolog(fmt, *args): + """Write a log message to the log file. See initlog() for docs.""" + logfp.write(fmt%args + "\n") + +def nolog(*allargs): + """Dummy function, assigned to log when logging is disabled.""" + pass + +def closelog(): + """Close the log file.""" + global log, logfile, logfp + logfile = '' + if logfp: + logfp.close() + logfp = None + log = initlog + +log = initlog # The current logging function + + +# Parsing functions +# ================= + +# Maximum input we will accept when REQUEST_METHOD is POST +# 0 ==> unlimited input +maxlen = 0 + +def parse(fp=None, environ=os.environ, keep_blank_values=0, strict_parsing=0): + """Parse a query in the environment or from a file (default stdin) + + Arguments, all optional: + + fp : file pointer; default: sys.stdin.buffer + + environ : environment dictionary; default: os.environ + + keep_blank_values: flag indicating whether blank values in + percent-encoded forms should be treated as blank strings. + A true value indicates that blanks should be retained as + blank strings. The default false value indicates that + blank values are to be ignored and treated as if they were + not included. + + strict_parsing: flag indicating what to do with parsing errors. + If false (the default), errors are silently ignored. + If true, errors raise a ValueError exception. + """ + if fp is None: + fp = sys.stdin + + # field keys and values (except for files) are returned as strings + # an encoding is required to decode the bytes read from self.fp + if hasattr(fp,'encoding'): + encoding = fp.encoding + else: + encoding = 'latin-1' + + # fp.read() must return bytes + if isinstance(fp, TextIOWrapper): + fp = fp.buffer + + if not 'REQUEST_METHOD' in environ: + environ['REQUEST_METHOD'] = 'GET' # For testing stand-alone + if environ['REQUEST_METHOD'] == 'POST': + ctype, pdict = parse_header(environ['CONTENT_TYPE']) + if ctype == 'multipart/form-data': + return parse_multipart(fp, pdict) + elif ctype == 'application/x-www-form-urlencoded': + clength = int(environ['CONTENT_LENGTH']) + if maxlen and clength > maxlen: + raise ValueError('Maximum content length exceeded') + qs = fp.read(clength).decode(encoding) + else: + qs = '' # Unknown content-type + if 'QUERY_STRING' in environ: + if qs: qs = qs + '&' + qs = qs + environ['QUERY_STRING'] + elif sys.argv[1:]: + if qs: qs = qs + '&' + qs = qs + sys.argv[1] + environ['QUERY_STRING'] = qs # XXX Shouldn't, really + elif 'QUERY_STRING' in environ: + qs = environ['QUERY_STRING'] + else: + if sys.argv[1:]: + qs = sys.argv[1] + else: + qs = "" + environ['QUERY_STRING'] = qs # XXX Shouldn't, really + return urllib.parse.parse_qs(qs, keep_blank_values, strict_parsing, + encoding=encoding) + + +def parse_multipart(fp, pdict, encoding="utf-8", errors="replace"): + """Parse multipart input. + + Arguments: + fp : input file + pdict: dictionary containing other parameters of content-type header + encoding, errors: request encoding and error handler, passed to + FieldStorage + + Returns a dictionary just like parse_qs(): keys are the field names, each + value is a list of values for that field. For non-file fields, the value + is a list of strings. + """ + # RFC 2026, Section 5.1 : The "multipart" boundary delimiters are always + # represented as 7bit US-ASCII. + boundary = pdict['boundary'].decode('ascii') + ctype = "multipart/form-data; boundary={}".format(boundary) + headers = Message() + headers.set_type(ctype) + headers['Content-Length'] = pdict['CONTENT-LENGTH'] + fs = FieldStorage(fp, headers=headers, encoding=encoding, errors=errors, + environ={'REQUEST_METHOD': 'POST'}) + return {k: fs.getlist(k) for k in fs} + +def _parseparam(s): + while s[:1] == ';': + s = s[1:] + end = s.find(';') + while end > 0 and (s.count('"', 0, end) - s.count('\\"', 0, end)) % 2: + end = s.find(';', end + 1) + if end < 0: + end = len(s) + f = s[:end] + yield f.strip() + s = s[end:] + +def parse_header(line): + """Parse a Content-type like header. + + Return the main content-type and a dictionary of options. + + """ + parts = _parseparam(';' + line) + key = parts.__next__() + pdict = {} + for p in parts: + i = p.find('=') + if i >= 0: + name = p[:i].strip().lower() + value = p[i+1:].strip() + if len(value) >= 2 and value[0] == value[-1] == '"': + value = value[1:-1] + value = value.replace('\\\\', '\\').replace('\\"', '"') + pdict[name] = value + return key, pdict + + +# Classes for field storage +# ========================= + +class MiniFieldStorage: + + """Like FieldStorage, for use when no file uploads are possible.""" + + # Dummy attributes + filename = None + list = None + type = None + file = None + type_options = {} + disposition = None + disposition_options = {} + headers = {} + + def __init__(self, name, value): + """Constructor from field name and value.""" + self.name = name + self.value = value + # self.file = StringIO(value) + + def __repr__(self): + """Return printable representation.""" + return "MiniFieldStorage(%r, %r)" % (self.name, self.value) + + +class FieldStorage: + + """Store a sequence of fields, reading multipart/form-data. + + This class provides naming, typing, files stored on disk, and + more. At the top level, it is accessible like a dictionary, whose + keys are the field names. (Note: None can occur as a field name.) + The items are either a Python list (if there's multiple values) or + another FieldStorage or MiniFieldStorage object. If it's a single + object, it has the following attributes: + + name: the field name, if specified; otherwise None + + filename: the filename, if specified; otherwise None; this is the + client side filename, *not* the file name on which it is + stored (that's a temporary file you don't deal with) + + value: the value as a *string*; for file uploads, this + transparently reads the file every time you request the value + and returns *bytes* + + file: the file(-like) object from which you can read the data *as + bytes* ; None if the data is stored a simple string + + type: the content-type, or None if not specified + + type_options: dictionary of options specified on the content-type + line + + disposition: content-disposition, or None if not specified + + disposition_options: dictionary of corresponding options + + headers: a dictionary(-like) object (sometimes email.message.Message or a + subclass thereof) containing *all* headers + + The class is subclassable, mostly for the purpose of overriding + the make_file() method, which is called internally to come up with + a file open for reading and writing. This makes it possible to + override the default choice of storing all files in a temporary + directory and unlinking them as soon as they have been opened. + + """ + def __init__(self, fp=None, headers=None, outerboundary=b'', + environ=os.environ, keep_blank_values=0, strict_parsing=0, + limit=None, encoding='utf-8', errors='replace', + max_num_fields=None): + """Constructor. Read multipart/* until last part. + + Arguments, all optional: + + fp : file pointer; default: sys.stdin.buffer + (not used when the request method is GET) + Can be : + 1. a TextIOWrapper object + 2. an object whose read() and readline() methods return bytes + + headers : header dictionary-like object; default: + taken from environ as per CGI spec + + outerboundary : terminating multipart boundary + (for internal use only) + + environ : environment dictionary; default: os.environ + + keep_blank_values: flag indicating whether blank values in + percent-encoded forms should be treated as blank strings. + A true value indicates that blanks should be retained as + blank strings. The default false value indicates that + blank values are to be ignored and treated as if they were + not included. + + strict_parsing: flag indicating what to do with parsing errors. + If false (the default), errors are silently ignored. + If true, errors raise a ValueError exception. + + limit : used internally to read parts of multipart/form-data forms, + to exit from the reading loop when reached. It is the difference + between the form content-length and the number of bytes already + read + + encoding, errors : the encoding and error handler used to decode the + binary stream to strings. Must be the same as the charset defined + for the page sending the form (content-type : meta http-equiv or + header) + + max_num_fields: int. If set, then __init__ throws a ValueError + if there are more than n fields read by parse_qsl(). + + """ + method = 'GET' + self.keep_blank_values = keep_blank_values + self.strict_parsing = strict_parsing + self.max_num_fields = max_num_fields + if 'REQUEST_METHOD' in environ: + method = environ['REQUEST_METHOD'].upper() + self.qs_on_post = None + if method == 'GET' or method == 'HEAD': + if 'QUERY_STRING' in environ: + qs = environ['QUERY_STRING'] + elif sys.argv[1:]: + qs = sys.argv[1] + else: + qs = "" + qs = qs.encode(locale.getpreferredencoding(), 'surrogateescape') + fp = BytesIO(qs) + if headers is None: + headers = {'content-type': + "application/x-www-form-urlencoded"} + if headers is None: + headers = {} + if method == 'POST': + # Set default content-type for POST to what's traditional + headers['content-type'] = "application/x-www-form-urlencoded" + if 'CONTENT_TYPE' in environ: + headers['content-type'] = environ['CONTENT_TYPE'] + if 'QUERY_STRING' in environ: + self.qs_on_post = environ['QUERY_STRING'] + if 'CONTENT_LENGTH' in environ: + headers['content-length'] = environ['CONTENT_LENGTH'] + else: + if not (isinstance(headers, (Mapping, Message))): + raise TypeError("headers must be mapping or an instance of " + "email.message.Message") + self.headers = headers + if fp is None: + self.fp = sys.stdin.buffer + # self.fp.read() must return bytes + elif isinstance(fp, TextIOWrapper): + self.fp = fp.buffer + else: + if not (hasattr(fp, 'read') and hasattr(fp, 'readline')): + raise TypeError("fp must be file pointer") + self.fp = fp + + self.encoding = encoding + self.errors = errors + + if not isinstance(outerboundary, bytes): + raise TypeError('outerboundary must be bytes, not %s' + % type(outerboundary).__name__) + self.outerboundary = outerboundary + + self.bytes_read = 0 + self.limit = limit + + # Process content-disposition header + cdisp, pdict = "", {} + if 'content-disposition' in self.headers: + cdisp, pdict = parse_header(self.headers['content-disposition']) + self.disposition = cdisp + self.disposition_options = pdict + self.name = None + if 'name' in pdict: + self.name = pdict['name'] + self.filename = None + if 'filename' in pdict: + self.filename = pdict['filename'] + self._binary_file = self.filename is not None + + # Process content-type header + # + # Honor any existing content-type header. But if there is no + # content-type header, use some sensible defaults. Assume + # outerboundary is "" at the outer level, but something non-false + # inside a multi-part. The default for an inner part is text/plain, + # but for an outer part it should be urlencoded. This should catch + # bogus clients which erroneously forget to include a content-type + # header. + # + # See below for what we do if there does exist a content-type header, + # but it happens to be something we don't understand. + if 'content-type' in self.headers: + ctype, pdict = parse_header(self.headers['content-type']) + elif self.outerboundary or method != 'POST': + ctype, pdict = "text/plain", {} + else: + ctype, pdict = 'application/x-www-form-urlencoded', {} + self.type = ctype + self.type_options = pdict + if 'boundary' in pdict: + self.innerboundary = pdict['boundary'].encode(self.encoding, + self.errors) + else: + self.innerboundary = b"" + + clen = -1 + if 'content-length' in self.headers: + try: + clen = int(self.headers['content-length']) + except ValueError: + pass + if maxlen and clen > maxlen: + raise ValueError('Maximum content length exceeded') + self.length = clen + if self.limit is None and clen >= 0: + self.limit = clen + + self.list = self.file = None + self.done = 0 + if ctype == 'application/x-www-form-urlencoded': + self.read_urlencoded() + elif ctype[:10] == 'multipart/': + self.read_multi(environ, keep_blank_values, strict_parsing) + else: + self.read_single() + + def __del__(self): + try: + self.file.close() + except AttributeError: + pass + + def __enter__(self): + return self + + def __exit__(self, *args): + self.file.close() + + def __repr__(self): + """Return a printable representation.""" + return "FieldStorage(%r, %r, %r)" % ( + self.name, self.filename, self.value) + + def __iter__(self): + return iter(self.keys()) + + def __getattr__(self, name): + if name != 'value': + raise AttributeError(name) + if self.file: + self.file.seek(0) + value = self.file.read() + self.file.seek(0) + elif self.list is not None: + value = self.list + else: + value = None + return value + + def __getitem__(self, key): + """Dictionary style indexing.""" + if self.list is None: + raise TypeError("not indexable") + found = [] + for item in self.list: + if item.name == key: found.append(item) + if not found: + raise KeyError(key) + if len(found) == 1: + return found[0] + else: + return found + + def getvalue(self, key, default=None): + """Dictionary style get() method, including 'value' lookup.""" + if key in self: + value = self[key] + if isinstance(value, list): + return [x.value for x in value] + else: + return value.value + else: + return default + + def getfirst(self, key, default=None): + """ Return the first value received.""" + if key in self: + value = self[key] + if isinstance(value, list): + return value[0].value + else: + return value.value + else: + return default + + def getlist(self, key): + """ Return list of received values.""" + if key in self: + value = self[key] + if isinstance(value, list): + return [x.value for x in value] + else: + return [value.value] + else: + return [] + + def keys(self): + """Dictionary style keys() method.""" + if self.list is None: + raise TypeError("not indexable") + return list(set(item.name for item in self.list)) + + def __contains__(self, key): + """Dictionary style __contains__ method.""" + if self.list is None: + raise TypeError("not indexable") + return any(item.name == key for item in self.list) + + def __len__(self): + """Dictionary style len(x) support.""" + return len(self.keys()) + + def __bool__(self): + if self.list is None: + raise TypeError("Cannot be converted to bool.") + return bool(self.list) + + def read_urlencoded(self): + """Internal: read data in query string format.""" + qs = self.fp.read(self.length) + if not isinstance(qs, bytes): + raise ValueError("%s should return bytes, got %s" \ + % (self.fp, type(qs).__name__)) + qs = qs.decode(self.encoding, self.errors) + if self.qs_on_post: + qs += '&' + self.qs_on_post + query = urllib.parse.parse_qsl( + qs, self.keep_blank_values, self.strict_parsing, + encoding=self.encoding, errors=self.errors, + max_num_fields=self.max_num_fields) + self.list = [MiniFieldStorage(key, value) for key, value in query] + self.skip_lines() + + FieldStorageClass = None + + def read_multi(self, environ, keep_blank_values, strict_parsing): + """Internal: read a part that is itself multipart.""" + ib = self.innerboundary + if not valid_boundary(ib): + raise ValueError('Invalid boundary in multipart form: %r' % (ib,)) + self.list = [] + if self.qs_on_post: + query = urllib.parse.parse_qsl( + self.qs_on_post, self.keep_blank_values, self.strict_parsing, + encoding=self.encoding, errors=self.errors, + max_num_fields=self.max_num_fields) + self.list.extend(MiniFieldStorage(key, value) for key, value in query) + + klass = self.FieldStorageClass or self.__class__ + first_line = self.fp.readline() # bytes + if not isinstance(first_line, bytes): + raise ValueError("%s should return bytes, got %s" \ + % (self.fp, type(first_line).__name__)) + self.bytes_read += len(first_line) + + # Ensure that we consume the file until we've hit our inner boundary + while (first_line.strip() != (b"--" + self.innerboundary) and + first_line): + first_line = self.fp.readline() + self.bytes_read += len(first_line) + + # Propagate max_num_fields into the sub class appropriately + max_num_fields = self.max_num_fields + if max_num_fields is not None: + max_num_fields -= len(self.list) + + while True: + parser = FeedParser() + hdr_text = b"" + while True: + data = self.fp.readline() + hdr_text += data + if not data.strip(): + break + if not hdr_text: + break + # parser takes strings, not bytes + self.bytes_read += len(hdr_text) + parser.feed(hdr_text.decode(self.encoding, self.errors)) + headers = parser.close() + + # Some clients add Content-Length for part headers, ignore them + if 'content-length' in headers: + del headers['content-length'] + + limit = None if self.limit is None \ + else self.limit - self.bytes_read + part = klass(self.fp, headers, ib, environ, keep_blank_values, + strict_parsing, limit, + self.encoding, self.errors, max_num_fields) + + if max_num_fields is not None: + max_num_fields -= 1 + if part.list: + max_num_fields -= len(part.list) + if max_num_fields < 0: + raise ValueError('Max number of fields exceeded') + + self.bytes_read += part.bytes_read + self.list.append(part) + if part.done or self.bytes_read >= self.length > 0: + break + self.skip_lines() + + def read_single(self): + """Internal: read an atomic part.""" + if self.length >= 0: + self.read_binary() + self.skip_lines() + else: + self.read_lines() + self.file.seek(0) + + bufsize = 8*1024 # I/O buffering size for copy to file + + def read_binary(self): + """Internal: read binary data.""" + self.file = self.make_file() + todo = self.length + if todo >= 0: + while todo > 0: + data = self.fp.read(min(todo, self.bufsize)) # bytes + if not isinstance(data, bytes): + raise ValueError("%s should return bytes, got %s" + % (self.fp, type(data).__name__)) + self.bytes_read += len(data) + if not data: + self.done = -1 + break + self.file.write(data) + todo = todo - len(data) + + def read_lines(self): + """Internal: read lines until EOF or outerboundary.""" + if self._binary_file: + self.file = self.__file = BytesIO() # store data as bytes for files + else: + self.file = self.__file = StringIO() # as strings for other fields + if self.outerboundary: + self.read_lines_to_outerboundary() + else: + self.read_lines_to_eof() + + def __write(self, line): + """line is always bytes, not string""" + if self.__file is not None: + if self.__file.tell() + len(line) > 1000: + self.file = self.make_file() + data = self.__file.getvalue() + self.file.write(data) + self.__file = None + if self._binary_file: + # keep bytes + self.file.write(line) + else: + # decode to string + self.file.write(line.decode(self.encoding, self.errors)) + + def read_lines_to_eof(self): + """Internal: read lines until EOF.""" + while 1: + line = self.fp.readline(1<<16) # bytes + self.bytes_read += len(line) + if not line: + self.done = -1 + break + self.__write(line) + + def read_lines_to_outerboundary(self): + """Internal: read lines until outerboundary. + Data is read as bytes: boundaries and line ends must be converted + to bytes for comparisons. + """ + next_boundary = b"--" + self.outerboundary + last_boundary = next_boundary + b"--" + delim = b"" + last_line_lfend = True + _read = 0 + while 1: + if self.limit is not None and _read >= self.limit: + break + line = self.fp.readline(1<<16) # bytes + self.bytes_read += len(line) + _read += len(line) + if not line: + self.done = -1 + break + if delim == b"\r": + line = delim + line + delim = b"" + if line.startswith(b"--") and last_line_lfend: + strippedline = line.rstrip() + if strippedline == next_boundary: + break + if strippedline == last_boundary: + self.done = 1 + break + odelim = delim + if line.endswith(b"\r\n"): + delim = b"\r\n" + line = line[:-2] + last_line_lfend = True + elif line.endswith(b"\n"): + delim = b"\n" + line = line[:-1] + last_line_lfend = True + elif line.endswith(b"\r"): + # We may interrupt \r\n sequences if they span the 2**16 + # byte boundary + delim = b"\r" + line = line[:-1] + last_line_lfend = False + else: + delim = b"" + last_line_lfend = False + self.__write(odelim + line) + + def skip_lines(self): + """Internal: skip lines until outer boundary if defined.""" + if not self.outerboundary or self.done: + return + next_boundary = b"--" + self.outerboundary + last_boundary = next_boundary + b"--" + last_line_lfend = True + while True: + line = self.fp.readline(1<<16) + self.bytes_read += len(line) + if not line: + self.done = -1 + break + if line.endswith(b"--") and last_line_lfend: + strippedline = line.strip() + if strippedline == next_boundary: + break + if strippedline == last_boundary: + self.done = 1 + break + last_line_lfend = line.endswith(b'\n') + + def make_file(self): + """Overridable: return a readable & writable file. + + The file will be used as follows: + - data is written to it + - seek(0) + - data is read from it + + The file is opened in binary mode for files, in text mode + for other fields + + This version opens a temporary file for reading and writing, + and immediately deletes (unlinks) it. The trick (on Unix!) is + that the file can still be used, but it can't be opened by + another process, and it will automatically be deleted when it + is closed or when the current process terminates. + + If you want a more permanent file, you derive a class which + overrides this method. If you want a visible temporary file + that is nevertheless automatically deleted when the script + terminates, try defining a __del__ method in a derived class + which unlinks the temporary files you have created. + + """ + if self._binary_file: + return tempfile.TemporaryFile("wb+") + else: + return tempfile.TemporaryFile("w+", + encoding=self.encoding, newline = '\n') + + +# Test/debug code +# =============== + +def test(environ=os.environ): + """Robust test CGI script, usable as main program. + + Write minimal HTTP headers and dump all information provided to + the script in HTML form. + + """ + print("Content-type: text/html") + print() + sys.stderr = sys.stdout + try: + form = FieldStorage() # Replace with other classes to test those + print_directory() + print_arguments() + print_form(form) + print_environ(environ) + print_environ_usage() + def f(): + exec("testing print_exception() -- italics?") + def g(f=f): + f() + print("

What follows is a test, not an actual exception:

") + g() + except: + print_exception() + + print("

Second try with a small maxlen...

") + + global maxlen + maxlen = 50 + try: + form = FieldStorage() # Replace with other classes to test those + print_directory() + print_arguments() + print_form(form) + print_environ(environ) + except: + print_exception() + +def print_exception(type=None, value=None, tb=None, limit=None): + if type is None: + type, value, tb = sys.exc_info() + import traceback + print() + print("

Traceback (most recent call last):

") + list = traceback.format_tb(tb, limit) + \ + traceback.format_exception_only(type, value) + print("
%s%s
" % ( + html.escape("".join(list[:-1])), + html.escape(list[-1]), + )) + del tb + +def print_environ(environ=os.environ): + """Dump the shell environment as HTML.""" + keys = sorted(environ.keys()) + print() + print("

Shell Environment:

") + print("
") + for key in keys: + print("
", html.escape(key), "
", html.escape(environ[key])) + print("
") + print() + +def print_form(form): + """Dump the contents of a form as HTML.""" + keys = sorted(form.keys()) + print() + print("

Form Contents:

") + if not keys: + print("

No form fields.") + print("

") + for key in keys: + print("
" + html.escape(key) + ":", end=' ') + value = form[key] + print("" + html.escape(repr(type(value))) + "") + print("
" + html.escape(repr(value))) + print("
") + print() + +def print_directory(): + """Dump the current directory as HTML.""" + print() + print("

Current Working Directory:

") + try: + pwd = os.getcwd() + except OSError as msg: + print("OSError:", html.escape(str(msg))) + else: + print(html.escape(pwd)) + print() + +def print_arguments(): + print() + print("

Command Line Arguments:

") + print() + print(sys.argv) + print() + +def print_environ_usage(): + """Dump a list of environment variables used by CGI as HTML.""" + print(""" +

These environment variables could have been set:

+ +In addition, HTTP headers sent by the server may be passed in the +environment as well. Here are some common variable names: + +""") + + +# Utilities +# ========= + +def valid_boundary(s): + import re + if isinstance(s, bytes): + _vb_pattern = b"^[ -~]{0,200}[!-~]$" + else: + _vb_pattern = "^[ -~]{0,200}[!-~]$" + return re.match(_vb_pattern, s) + +# Invoke mainline +# =============== + +# Call test() when this file is run as a script (not imported as a module) +if __name__ == '__main__': + test() diff --git a/Lib/test/test_cgi.py b/Lib/test/test_cgi.py new file mode 100644 index 000000000..ab8677199 --- /dev/null +++ b/Lib/test/test_cgi.py @@ -0,0 +1,603 @@ +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) + + 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") + + 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) + + 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()