# 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" keys = [] for item in self.list: if item.name not in keys: keys.append(item.name) return keys def has_key(self, key): """Dictionary style has_key() method.""" if self.list is None: raise TypeError, "not indexable" for item in self.list: if item.name == key: return True return False def __contains__(self, key): """Dictionary style __contains__ method.""" if self.list is None: raise TypeError, "not indexable" for item in self.list: if item.name == key: return True return False def __len__(self): """Dictionary style len(x) support.""" return len(self.keys()) def read_urlencoded(self): """Internal: read data in query string format.""" qs = self.fp.read(self.length) self.list = list = [] for key, value in 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 = [] 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 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 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): self.dict = self.data = parse(environ=environ) 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>" printLines 501–1000 of 1071. Previous lines Load more
# 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" keys = [] for item in self.list: if item.name not in keys: keys.append(item.name) return keys def has_key(self, key): """Dictionary style has_key() method.""" if self.list is None: raise TypeError, "not indexable" for item in self.list: if item.name == key: return True return False def __contains__(self, key): """Dictionary style __contains__ method.""" if self.list is None: raise TypeError, "not indexable" for item in self.list: if item.name == key: return True return False def __len__(self): """Dictionary style len(x) support.""" return len(self.keys()) def read_urlencoded(self): """Internal: read data in query string format.""" qs = self.fp.read(self.length) self.list = list = [] for key, value in 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 = [] 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 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 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): self.dict = self.data = parse(environ=environ) 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