|
#! /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. |
|
""" |
|
|
|
# XXX Perhaps there should be a slimmed version that doesn't contain |
|
# all those backwards compatible and debugging classes and functions? |
|
|
|
# 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 operator import attrgetter |
|
import sys |
|
import os |
|
import UserDict |
|
import urlparse |
|
|
|
from warnings import filterwarnings, catch_warnings, warn |
|
with catch_warnings(): |
|
if sys.py3kwarning: |
|
filterwarnings("ignore", ".*mimetools has been removed", |
|
DeprecationWarning) |
|
filterwarnings("ignore", ".*rfc822 has been removed", |
|
DeprecationWarning) |
|
import mimetools |
|
import rfc822 |
|
|
|
try: |
|
from cStringIO import StringIO |
|
except ImportError: |
|
from StringIO import StringIO |
|
|
|
__all__ = ["MiniFieldStorage", "FieldStorage", "FormContentDict", |
|
"SvFormContentDict", "InterpFormContentDict", "FormContent", |
|
"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 logfp, log |
|
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 |
|
|
|
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 |
|
|
|
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 |
|
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) |
|
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 urlparse.parse_qs(qs, keep_blank_values, strict_parsing) |
|
|
|
|
|
# parse query string function called from urlparse, |
|
# this is done in order to maintain backward compatibility. |
|
|
|
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 urlparse.parse_qs instead", |
|
PendingDeprecationWarning, 2) |
|
return urlparse.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 urlparse.parse_qsl instead", |
|
PendingDeprecationWarning, 2) |
|
return urlparse.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(). |
|
|
|
""" |
|
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 = mimetools.Message(fp) |
|
clength = headers.getheader('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[:2] == "--": |
|
terminator = line.strip() |
|
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) - 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 |
|
|
|
file: the file(-like) object from which you can read the data; |
|
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 rfc822.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="", |
|
environ=os.environ, keep_blank_values=0, strict_parsing=0): |
|
"""Constructor. Read multipart/* until last part. |
|
|
|
Arguments, all optional: |
|
|
|
fp : file pointer; default: sys.stdin |
|
(not used when the request method is GET) |
|
|
|
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. |
|
|
|
""" |
|
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 = "" |
|
fp = StringIO(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'] |
|
self.fp = fp or sys.stdin |
|
self.headers = headers |
|
self.outerboundary = outerboundary |
|
|
|
# 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'] |
|
|
|
# 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 |
|
self.innerboundary = "" |
|
if 'boundary' in pdict: |
|
self.innerboundary = pdict['boundary'] |
|
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 |
|
|
|
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 type(value) is type([]): |
|
return map(attrgetter('value'), 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 type(value) is type([]): |
|
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 type(value) is type([]): |
|
return map(attrgetter('value'), 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 has_key(self, key): |
|
"""Dictionary style has_key() method.""" |
|
if self.list is None: |
|
raise TypeError, "not indexable" |
|
return any(item.name == key 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 self.qs_on_post: |
|
qs += '&' + self.qs_on_post |
|
self.list = list = [] |
|
for key, value in urlparse.parse_qsl(qs, self.keep_blank_values, |
|
self.strict_parsing): |
|
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: |
|
for key, value in urlparse.parse_qsl(self.qs_on_post, |
|
self.keep_blank_values, self.strict_parsing): |
|
self.list.append(MiniFieldStorage(key, value)) |
|
FieldStorageClass = None |
|
|
|
klass = self.FieldStorageClass or self.__class__ |
|
part = klass(self.fp, {}, ib, |
|
environ, keep_blank_values, strict_parsing) |
|
# Throw first part away |
|
while not part.done: |
|
headers = rfc822.Message(self.fp) |
|
part = klass(self.fp, headers, ib, |
|
environ, keep_blank_values, strict_parsing) |
|
self.list.append(part) |
|
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('b') |
|
todo = self.length |
|
if todo >= 0: |
|
while todo > 0: |
|
data = self.fp.read(min(todo, self.bufsize)) |
|
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.""" |
|
self.file = self.__file = StringIO() |
|
if self.outerboundary: |
|
self.read_lines_to_outerboundary() |
|
else: |
|
self.read_lines_to_eof() |
|
|
|
def __write(self, line): |
|
if self.__file is not None: |
|
if self.__file.tell() + len(line) > 1000: |
|
self.file = self.make_file('') |
|
self.file.write(self.__file.getvalue()) |
|
self.__file = None |
|
self.file.write(line) |
|
|
|
def read_lines_to_eof(self): |
|
"""Internal: read lines until EOF.""" |
|
while 1: |
|
line = self.fp.readline(1<<16) |
|
if not line: |
|
self.done = -1 |
|
break |
|
self.__write(line) |
|
|
|
def read_lines_to_outerboundary(self): |
|
"""Internal: read lines until outerboundary.""" |
|
next = "--" + self.outerboundary |
|
last = next + "--" |
|
delim = "" |
|
last_line_lfend = True |
|
while 1: |
|
line = self.fp.readline(1<<16) |
|
if not line: |
|
self.done = -1 |
|
break |
|
if delim == "\r": |
|
line = delim + line |
|
delim = "" |
|
if line[:2] == "--" and last_line_lfend: |
|
strippedline = line.strip() |
|
if strippedline == next: |
|
break |
|
if strippedline == last: |
|
self.done = 1 |
|
break |
|
odelim = delim |
|
if line[-2:] == "\r\n": |
|
delim = "\r\n" |
|
line = line[:-2] |
|
last_line_lfend = True |
|
elif line[-1] == "\n": |
|
delim = "\n" |
|
line = line[:-1] |
|
last_line_lfend = True |
|
elif line[-1] == "\r": |
|
# We may interrupt \r\n sequences if they span the 2**16 |
|
# byte boundary |
|
delim = "\r" |
|
line = line[:-1] |
|
last_line_lfend = False |
|
else: |
|
delim = "" |
|
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 = "--" + self.outerboundary |
|
last = next + "--" |
|
last_line_lfend = True |
|
while 1: |
|
line = self.fp.readline(1<<16) |
|
if not line: |
|
self.done = -1 |
|
break |
|
if line[:2] == "--" and last_line_lfend: |
|
strippedline = line.strip() |
|
if strippedline == next: |
|
break |
|
if strippedline == last: |
|
self.done = 1 |
|
break |
|
last_line_lfend = line.endswith('\n') |
|
|
|
def make_file(self, binary=None): |
|
"""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 'binary' argument is unused -- the file is always opened |
|
in binary mode. |
|
|
|
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. |
|
|
|
""" |
|
import tempfile |
|
return tempfile.TemporaryFile("w+b") |
|
|
|
|
|
|
|
# Backwards Compatibility Classes |
|
# =============================== |
|
|
|
class FormContentDict(UserDict.UserDict): |
|
"""Form content as dictionary with a list of values per field. |
|
|
|
form = FormContentDict() |
|
|
|
form[key] -> [value, value, ...] |
|
key in form -> Boolean |
|
form.keys() -> [key, key, ...] |
|
form.values() -> [[val, val, ...], [val, val, ...], ...] |
|
form.items() -> [(key, [val, val, ...]), (key, [val, val, ...]), ...] |
|
form.dict == {key: [val, val, ...], ...} |
|
|
|
""" |
|
def __init__(self, environ=os.environ, keep_blank_values=0, strict_parsing=0): |
|
self.dict = self.data = parse(environ=environ, |
|
keep_blank_values=keep_blank_values, |
|
strict_parsing=strict_parsing) |
|
self.query_string = environ['QUERY_STRING'] |
|
|
|
|
|
class SvFormContentDict(FormContentDict): |
|
"""Form content as dictionary expecting a single value per field. |
|
|
|
If you only expect a single value for each field, then form[key] |
|
will return that single value. It will raise an IndexError if |
|
that expectation is not true. If you expect a field to have |
|
possible multiple values, than you can use form.getlist(key) to |
|
get all of the values. values() and items() are a compromise: |
|
they return single strings where there is a single value, and |
|
lists of strings otherwise. |
|
|
|
""" |
|
def __getitem__(self, key): |
|
if len(self.dict[key]) > 1: |
|
raise IndexError, 'expecting a single value' |
|
return self.dict[key][0] |
|
def getlist(self, key): |
|
return self.dict[key] |
|
def values(self): |
|
result = [] |
|
for value in self.dict.values(): |
|
if len(value) == 1: |
|
result.append(value[0]) |
|
else: result.append(value) |
|
return result |
|
def items(self): |
|
result = [] |
|
for key, value in self.dict.items(): |
|
if len(value) == 1: |
|
result.append((key, value[0])) |
|
else: result.append((key, value)) |
|
return result |
|
|
|
|
|
class InterpFormContentDict(SvFormContentDict): |
|
"""This class is present for backwards compatibility only.""" |
|
def __getitem__(self, key): |
|
v = SvFormContentDict.__getitem__(self, key) |
|
if v[0] in '0123456789+-.': |
|
try: return int(v) |
|
except ValueError: |
|
try: return float(v) |
|
except ValueError: pass |
|
return v.strip() |
|
def values(self): |
|
result = [] |
|
for key in self.keys(): |
|
try: |
|
result.append(self[key]) |
|
except IndexError: |
|
result.append(self.dict[key]) |
|
return result |
|
def items(self): |
|
result = [] |
|
for key in self.keys(): |
|
try: |
|
result.append((key, self[key])) |
|
except IndexError: |
|
result.append((key, self.dict[key])) |
|
return result |
|
|
|
|
|
class FormContent(FormContentDict): |
|
"""This class is present for backwards compatibility only.""" |
|
def values(self, key): |
|
if key in self.dict :return self.dict[key] |
|
else: return None |
|
def indexed_value(self, key, location): |
|
if key in self.dict: |
|
if len(self.dict[key]) > location: |
|
return self.dict[key][location] |
|
else: return None |
|
else: return None |
|
def value(self, key): |
|
if key in self.dict: return self.dict[key][0] |
|
else: return None |
|
def length(self, key): |
|
return len(self.dict[key]) |
|
def stripped(self, key): |
|
if key in self.dict: return self.dict[key][0].strip() |
|
else: return None |
|
def pars(self): |
|
return self.dict |
|
|
|
|
|
# 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>" % ( |
|
escape("".join(list[:-1])), |
|
escape(list[-1]), |
|
) |
|
del tb |
|
|
|
def print_environ(environ=os.environ): |
|
"""Dump the shell environment as HTML.""" |
|
keys = environ.keys() |
|
keys.sort() |
|
print |
|
print "<H3>Shell Environment:</H3>" |
|
print "<DL>" |
|
for key in keys: |
|
print "<DT>", escape(key), "<DD>", escape(environ[key]) |
|
print "</DL>" |
|
print |
|
|
|
def print_form(form): |
|
"""Dump the contents of a form as HTML.""" |
|
keys = form.keys() |
|
keys.sort() |
|
print |
|
print "<H3>Form Contents:</H3>" |
|
if not keys: |
|
print "<P>No form fields." |
|
print "<DL>" |
|
for key in keys: |
|
print "<DT>" + escape(key) + ":", |
|
value = form[key] |
|
print "<i>" + escape(repr(type(value))) + "</i>" |
|
print "<DD>" + 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, msg: |
|
print "os.error:", escape(str(msg)) |
|
else: |
|
print 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): |
|
'''Replace special characters "&", "<" and ">" to HTML-safe sequences. |
|
If the optional flag quote is true, the quotation mark character (") |
|
is also translated.''' |
|
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="^[ -~]{0,200}[!-~]$"): |
|
import re |
|
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() |
