diff options
Diffstat (limited to 'Lib/wsgiref/validate.py')
| -rw-r--r-- | Lib/wsgiref/validate.py | 59 |
1 files changed, 35 insertions, 24 deletions
diff --git a/Lib/wsgiref/validate.py b/Lib/wsgiref/validate.py index 04a893d7c61..49eaa514cc5 100644 --- a/Lib/wsgiref/validate.py +++ b/Lib/wsgiref/validate.py @@ -98,7 +98,7 @@ Some of the things this checks: - That it is not a string (it should be a list of a single string; a string will work, but perform horribly). - - That .next() returns a string + - That .__next__() returns a string - That the iterator is not iterated over until start_response has been called (that can signal either a server or application @@ -113,7 +113,6 @@ __all__ = ['validator'] import re import sys -from types import DictType, StringType, TupleType, ListType import warnings header_re = re.compile(r'^[a-zA-Z][a-zA-Z0-9\-_]*$') @@ -128,6 +127,12 @@ def assert_(cond, *args): if not cond: raise AssertionError(*args) +def check_string_type(value, title): + if type (value) is str: + return value + raise AssertionError( + "{0} must be of type str (got {1})".format(title, repr(value))) + def validator(application): """ @@ -189,22 +194,23 @@ class InputWrapper: self.input = wsgi_input def read(self, *args): - assert_(len(args) <= 1) + assert_(len(args) == 1) v = self.input.read(*args) - assert_(type(v) is type("")) + assert_(type(v) is bytes) return v - def readline(self): - v = self.input.readline() - assert_(type(v) is type("")) + def readline(self, *args): + assert_(len(args) <= 1) + v = self.input.readline(*args) + assert_(type(v) is bytes) return v def readlines(self, *args): assert_(len(args) <= 1) lines = self.input.readlines(*args) - assert_(type(lines) is type([])) + assert_(type(lines) is list) for line in lines: - assert_(type(line) is type("")) + assert_(type(line) is bytes) return lines def __iter__(self): @@ -223,7 +229,7 @@ class ErrorWrapper: self.errors = wsgi_errors def write(self, s): - assert_(type(s) is type("")) + assert_(type(s) is str) self.errors.write(s) def flush(self): @@ -242,7 +248,7 @@ class WriteWrapper: self.writer = wsgi_writer def __call__(self, s): - assert_(type(s) is type("")) + assert_(type(s) is bytes) self.writer(s) class PartialIteratorWrapper: @@ -265,10 +271,12 @@ class IteratorWrapper: def __iter__(self): return self - def next(self): + def __next__(self): assert_(not self.closed, "Iterator read after closed") - v = self.iterator.next() + v = next(self.iterator) + if type(v) is not bytes: + assert_(False, "Iterator yielded non-bytestring (%r)" % (v,)) if self.check_start_response is not None: assert_(self.check_start_response, "The application returns and we started iterating over its body, but start_response has not yet been called") @@ -288,7 +296,7 @@ class IteratorWrapper: "Iterator garbage collected without being closed") def check_environ(environ): - assert_(type(environ) is DictType, + assert_(type(environ) is dict, "Environment is not of the right type: %r (environment: %r)" % (type(environ), environ)) @@ -315,11 +323,11 @@ def check_environ(environ): if '.' in key: # Extension, we don't care about its type continue - assert_(type(environ[key]) is StringType, + assert_(type(environ[key]) is str, "Environmental variable %s is not a string: %r (value: %r)" % (key, type(environ[key]), environ[key])) - assert_(type(environ['wsgi.version']) is TupleType, + assert_(type(environ['wsgi.version']) is tuple, "wsgi.version should be a tuple (%r)" % (environ['wsgi.version'],)) assert_(environ['wsgi.url_scheme'] in ('http', 'https'), "wsgi.url_scheme unknown: %r" % environ['wsgi.url_scheme']) @@ -365,8 +373,7 @@ def check_errors(wsgi_errors): % (wsgi_errors, attr)) def check_status(status): - assert_(type(status) is StringType, - "Status must be a string (not %r)" % status) + status = check_string_type(status, "Status") # Implicitly check that we can turn it into an integer: status_code = status.split(None, 1)[0] assert_(len(status_code) == 3, @@ -380,16 +387,18 @@ def check_status(status): % status, WSGIWarning) def check_headers(headers): - assert_(type(headers) is ListType, + assert_(type(headers) is list, "Headers (%r) must be of type list: %r" % (headers, type(headers))) header_names = {} for item in headers: - assert_(type(item) is TupleType, + assert_(type(item) is tuple, "Individual headers (%r) must be of type tuple: %r" % (item, type(item))) assert_(len(item) == 2) name, value = item + name = check_string_type(name, "Header name") + value = check_string_type(value, "Header value") assert_(name.lower() != 'status', "The Status header cannot be used; it conflicts with CGI " "script, and HTTP status is not given through headers " @@ -405,11 +414,13 @@ def check_headers(headers): % (value, bad_header_value_re.search(value).group(0))) def check_content_type(status, headers): + status = check_string_type(status, "Status") code = int(status.split(None, 1)[0]) # @@: need one more person to verify this interpretation of RFC 2616 # http://www.w3.org/Protocols/rfc2616/rfc2616-sec10.html NO_MESSAGE_BODY = (204, 304) for name, value in headers: + name = check_string_type(name, "Header name") if name.lower() == 'content-type': if code not in NO_MESSAGE_BODY: return @@ -419,14 +430,14 @@ def check_content_type(status, headers): assert_(0, "No Content-Type header found in headers (%s)" % headers) def check_exc_info(exc_info): - assert_(exc_info is None or type(exc_info) is type(()), + assert_(exc_info is None or type(exc_info) is tuple, "exc_info (%r) is not a tuple: %r" % (exc_info, type(exc_info))) # More exc_info checks? def check_iterator(iterator): - # Technically a string is legal, which is why it's a really bad + # Technically a bytestring is legal, which is why it's a really bad # idea, because it may cause the response to be returned # character-by-character - assert_(not isinstance(iterator, str), + assert_(not isinstance(iterator, (str, bytes)), "You should not return a string as your application iterator, " - "instead return a single-item list containing that string.") + "instead return a single-item list containing a bytestring.") |