mirror of
				https://github.com/python/cpython.git
				synced 2025-10-30 05:01:30 +00:00 
			
		
		
		
	
		
			
				
	
	
		
			1037 lines
		
	
	
	
		
			34 KiB
		
	
	
	
		
			Python
		
	
	
		
			Executable file
		
	
	
	
	
			
		
		
	
	
			1037 lines
		
	
	
	
		
			34 KiB
		
	
	
	
		
			Python
		
	
	
		
			Executable file
		
	
	
	
	
| #! /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
 | |
| import sys
 | |
| import os
 | |
| import urllib.parse
 | |
| from email.parser import FeedParser
 | |
| from warnings import warn
 | |
| import html
 | |
| import locale
 | |
| import tempfile
 | |
| 
 | |
| __all__ = ["MiniFieldStorage", "FieldStorage",
 | |
|            "parse", "parse_qs", "parse_qsl", "parse_multipart",
 | |
|            "parse_header", "print_exception", "print_environ",
 | |
|            "print_form", "print_directory", "print_arguments",
 | |
|            "print_environ_usage", "escape"]
 | |
| 
 | |
| # 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 IOError:
 | |
|             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)
 | |
| 
 | |
| 
 | |
| # parse query string function called from urlparse,
 | |
| # this is done in order to maintain backward compatiblity.
 | |
| 
 | |
| def parse_qs(qs, keep_blank_values=0, strict_parsing=0):
 | |
|     """Parse a query given as a string argument."""
 | |
|     warn("cgi.parse_qs is deprecated, use urllib.parse.parse_qs instead",
 | |
|          DeprecationWarning, 2)
 | |
|     return urllib.parse.parse_qs(qs, keep_blank_values, strict_parsing)
 | |
| 
 | |
| def parse_qsl(qs, keep_blank_values=0, strict_parsing=0):
 | |
|     """Parse a query given as a string argument."""
 | |
|     warn("cgi.parse_qsl is deprecated, use urllib.parse.parse_qsl instead",
 | |
|          DeprecationWarning, 2)
 | |
|     return urllib.parse.parse_qsl(qs, keep_blank_values, strict_parsing)
 | |
| 
 | |
| def parse_multipart(fp, pdict):
 | |
|     """Parse multipart input.
 | |
| 
 | |
|     Arguments:
 | |
|     fp   : input file
 | |
|     pdict: dictionary containing other parameters of content-type header
 | |
| 
 | |
|     Returns a dictionary just like parse_qs(): keys are the field names, each
 | |
|     value is a list of values for that field.  This is easy to use but not
 | |
|     much good if you are expecting megabytes to be uploaded -- in that case,
 | |
|     use the FieldStorage class instead which is much more flexible.  Note
 | |
|     that content-type is the raw, unparsed contents of the content-type
 | |
|     header.
 | |
| 
 | |
|     XXX This does not parse nested multipart parts -- use FieldStorage for
 | |
|     that.
 | |
| 
 | |
|     XXX This should really be subsumed by FieldStorage altogether -- no
 | |
|     point in having two implementations of the same parsing algorithm.
 | |
|     Also, FieldStorage protects itself better against certain DoS attacks
 | |
|     by limiting the size of the data read in one chunk.  The API here
 | |
|     does not support that kind of protection.  This also affects parse()
 | |
|     since it can call parse_multipart().
 | |
| 
 | |
|     """
 | |
|     import http.client
 | |
| 
 | |
|     boundary = ""
 | |
|     if 'boundary' in pdict:
 | |
|         boundary = pdict['boundary']
 | |
|     if not valid_boundary(boundary):
 | |
|         raise ValueError('Invalid boundary in multipart form: %r'
 | |
|                             % (boundary,))
 | |
| 
 | |
|     nextpart = "--" + boundary
 | |
|     lastpart = "--" + boundary + "--"
 | |
|     partdict = {}
 | |
|     terminator = ""
 | |
| 
 | |
|     while terminator != lastpart:
 | |
|         bytes = -1
 | |
|         data = None
 | |
|         if terminator:
 | |
|             # At start of next part.  Read headers first.
 | |
|             headers = http.client.parse_headers(fp)
 | |
|             clength = headers.get('content-length')
 | |
|             if clength:
 | |
|                 try:
 | |
|                     bytes = int(clength)
 | |
|                 except ValueError:
 | |
|                     pass
 | |
|             if bytes > 0:
 | |
|                 if maxlen and bytes > maxlen:
 | |
|                     raise ValueError('Maximum content length exceeded')
 | |
|                 data = fp.read(bytes)
 | |
|             else:
 | |
|                 data = ""
 | |
|         # Read lines until end of part.
 | |
|         lines = []
 | |
|         while 1:
 | |
|             line = fp.readline()
 | |
|             if not line:
 | |
|                 terminator = lastpart # End outer loop
 | |
|                 break
 | |
|             if line.startswith("--"):
 | |
|                 terminator = line.rstrip()
 | |
|                 if terminator in (nextpart, lastpart):
 | |
|                     break
 | |
|             lines.append(line)
 | |
|         # Done with part.
 | |
|         if data is None:
 | |
|             continue
 | |
|         if bytes < 0:
 | |
|             if lines:
 | |
|                 # Strip final line terminator
 | |
|                 line = lines[-1]
 | |
|                 if line[-2:] == "\r\n":
 | |
|                     line = line[:-2]
 | |
|                 elif line[-1:] == "\n":
 | |
|                     line = line[:-1]
 | |
|                 lines[-1] = line
 | |
|                 data = "".join(lines)
 | |
|         line = headers['content-disposition']
 | |
|         if not line:
 | |
|             continue
 | |
|         key, params = parse_header(line)
 | |
|         if key != 'form-data':
 | |
|             continue
 | |
|         if 'name' in params:
 | |
|             name = params['name']
 | |
|         else:
 | |
|             continue
 | |
|         if name in partdict:
 | |
|             partdict[name].append(data)
 | |
|         else:
 | |
|             partdict[name] = [data]
 | |
| 
 | |
|     return partdict
 | |
| 
 | |
| 
 | |
| def _parseparam(s):
 | |
|     while s[:1] == ';':
 | |
|         s = s[1:]
 | |
|         end = s.find(';')
 | |
|         while end > 0 and 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'):
 | |
|         """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)
 | |
| 
 | |
|         """
 | |
|         method = 'GET'
 | |
|         self.keep_blank_values = keep_blank_values
 | |
|         self.strict_parsing = strict_parsing
 | |
|         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']
 | |
|         if fp is None:
 | |
|             self.fp = sys.stdin.buffer
 | |
|         # self.fp.read() must return bytes
 | |
|         elif isinstance(fp, TextIOWrapper):
 | |
|             self.fp = fp.buffer
 | |
|         else:
 | |
|             self.fp = fp
 | |
| 
 | |
|         self.encoding = encoding
 | |
|         self.errors = errors
 | |
| 
 | |
|         self.headers = headers
 | |
|         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)
 | |
|         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:
 | |
|             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 __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 __nonzero__(self):
 | |
|         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
 | |
|         self.list = []
 | |
|         query = urllib.parse.parse_qsl(
 | |
|             qs, self.keep_blank_values, self.strict_parsing,
 | |
|             encoding=self.encoding, errors=self.errors)
 | |
|         for key, value in query:
 | |
|             self.list.append(MiniFieldStorage(key, value))
 | |
|         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)
 | |
|             for key, value in query:
 | |
|                 self.list.append(MiniFieldStorage(key, value))
 | |
|             FieldStorageClass = None
 | |
| 
 | |
|         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)
 | |
|         # first line holds boundary ; ignore it, or check that
 | |
|         # b"--" + ib == first_line.strip() ?
 | |
|         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()
 | |
|             part = klass(self.fp, headers, ib, environ, keep_blank_values,
 | |
|                          strict_parsing,self.limit-self.bytes_read,
 | |
|                          self.encoding, self.errors)
 | |
|             self.bytes_read += part.bytes_read
 | |
|             self.list.append(part)
 | |
|             if self.bytes_read >= self.length:
 | |
|                 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 _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 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
 | |
|             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 os.error as msg:
 | |
|         print("os.error:", 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 escape(s, quote=None):
 | |
|     """Deprecated API."""
 | |
|     warn("cgi.escape is deprecated, use html.escape instead",
 | |
|          PendingDeprecationWarning, stacklevel=2)
 | |
|     s = s.replace("&", "&") # Must be done first!
 | |
|     s = s.replace("<", "<")
 | |
|     s = s.replace(">", ">")
 | |
|     if quote:
 | |
|         s = s.replace('"', """)
 | |
|     return s
 | |
| 
 | |
| 
 | |
| def valid_boundary(s, _vb_pattern=None):
 | |
|     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()
 | 
