diff options
Diffstat (limited to 'python/werkzeug/middleware/lint.py')
| -rw-r--r-- | python/werkzeug/middleware/lint.py | 408 |
1 files changed, 0 insertions, 408 deletions
diff --git a/python/werkzeug/middleware/lint.py b/python/werkzeug/middleware/lint.py deleted file mode 100644 index 98f9581..0000000 --- a/python/werkzeug/middleware/lint.py +++ /dev/null @@ -1,408 +0,0 @@ -""" -WSGI Protocol Linter -==================== - -This module provides a middleware that performs sanity checks on the -behavior of the WSGI server and application. It checks that the -:pep:`3333` WSGI spec is properly implemented. It also warns on some -common HTTP errors such as non-empty responses for 304 status codes. - -.. autoclass:: LintMiddleware - -:copyright: 2007 Pallets -:license: BSD-3-Clause -""" -from warnings import warn - -from .._compat import implements_iterator -from .._compat import PY2 -from .._compat import string_types -from ..datastructures import Headers -from ..http import is_entity_header -from ..wsgi import FileWrapper - -try: - from urllib.parse import urlparse -except ImportError: - from urlparse import urlparse - - -class WSGIWarning(Warning): - """Warning class for WSGI warnings.""" - - -class HTTPWarning(Warning): - """Warning class for HTTP warnings.""" - - -def check_string(context, obj, stacklevel=3): - if type(obj) is not str: - warn( - "'%s' requires strings, got '%s'" % (context, type(obj).__name__), - WSGIWarning, - ) - - -class InputStream(object): - def __init__(self, stream): - self._stream = stream - - def read(self, *args): - if len(args) == 0: - warn( - "WSGI does not guarantee an EOF marker on the input stream, thus making" - " calls to 'wsgi.input.read()' unsafe. Conforming servers may never" - " return from this call.", - WSGIWarning, - stacklevel=2, - ) - elif len(args) != 1: - warn( - "Too many parameters passed to 'wsgi.input.read()'.", - WSGIWarning, - stacklevel=2, - ) - return self._stream.read(*args) - - def readline(self, *args): - if len(args) == 0: - warn( - "Calls to 'wsgi.input.readline()' without arguments are unsafe. Use" - " 'wsgi.input.read()' instead.", - WSGIWarning, - stacklevel=2, - ) - elif len(args) == 1: - warn( - "'wsgi.input.readline()' was called with a size hint. WSGI does not" - " support this, although it's available on all major servers.", - WSGIWarning, - stacklevel=2, - ) - else: - raise TypeError("Too many arguments passed to 'wsgi.input.readline()'.") - return self._stream.readline(*args) - - def __iter__(self): - try: - return iter(self._stream) - except TypeError: - warn("'wsgi.input' is not iterable.", WSGIWarning, stacklevel=2) - return iter(()) - - def close(self): - warn("The application closed the input stream!", WSGIWarning, stacklevel=2) - self._stream.close() - - -class ErrorStream(object): - def __init__(self, stream): - self._stream = stream - - def write(self, s): - check_string("wsgi.error.write()", s) - self._stream.write(s) - - def flush(self): - self._stream.flush() - - def writelines(self, seq): - for line in seq: - self.write(line) - - def close(self): - warn("The application closed the error stream!", WSGIWarning, stacklevel=2) - self._stream.close() - - -class GuardedWrite(object): - def __init__(self, write, chunks): - self._write = write - self._chunks = chunks - - def __call__(self, s): - check_string("write()", s) - self._write.write(s) - self._chunks.append(len(s)) - - -@implements_iterator -class GuardedIterator(object): - def __init__(self, iterator, headers_set, chunks): - self._iterator = iterator - if PY2: - self._next = iter(iterator).next - else: - self._next = iter(iterator).__next__ - self.closed = False - self.headers_set = headers_set - self.chunks = chunks - - def __iter__(self): - return self - - def __next__(self): - if self.closed: - warn("Iterated over closed 'app_iter'.", WSGIWarning, stacklevel=2) - - rv = self._next() - - if not self.headers_set: - warn( - "The application returned before it started the response.", - WSGIWarning, - stacklevel=2, - ) - - check_string("application iterator items", rv) - self.chunks.append(len(rv)) - return rv - - def close(self): - self.closed = True - - if hasattr(self._iterator, "close"): - self._iterator.close() - - if self.headers_set: - status_code, headers = self.headers_set - bytes_sent = sum(self.chunks) - content_length = headers.get("content-length", type=int) - - if status_code == 304: - for key, _value in headers: - key = key.lower() - if key not in ("expires", "content-location") and is_entity_header( - key - ): - warn( - "Entity header %r found in 304 response." % key, HTTPWarning - ) - if bytes_sent: - warn("304 responses must not have a body.", HTTPWarning) - elif 100 <= status_code < 200 or status_code == 204: - if content_length != 0: - warn( - "%r responses must have an empty content length." % status_code, - HTTPWarning, - ) - if bytes_sent: - warn( - "%r responses must not have a body." % status_code, HTTPWarning - ) - elif content_length is not None and content_length != bytes_sent: - warn( - "Content-Length and the number of bytes sent to the client do not" - " match.", - WSGIWarning, - ) - - def __del__(self): - if not self.closed: - try: - warn( - "Iterator was garbage collected before it was closed.", WSGIWarning - ) - except Exception: - pass - - -class LintMiddleware(object): - """Warns about common errors in the WSGI and HTTP behavior of the - server and wrapped application. Some of the issues it check are: - - - invalid status codes - - non-bytestrings sent to the WSGI server - - strings returned from the WSGI application - - non-empty conditional responses - - unquoted etags - - relative URLs in the Location header - - unsafe calls to wsgi.input - - unclosed iterators - - Error information is emitted using the :mod:`warnings` module. - - :param app: The WSGI application to wrap. - - .. code-block:: python - - from werkzeug.middleware.lint import LintMiddleware - app = LintMiddleware(app) - """ - - def __init__(self, app): - self.app = app - - def check_environ(self, environ): - if type(environ) is not dict: - warn( - "WSGI environment is not a standard Python dict.", - WSGIWarning, - stacklevel=4, - ) - for key in ( - "REQUEST_METHOD", - "SERVER_NAME", - "SERVER_PORT", - "wsgi.version", - "wsgi.input", - "wsgi.errors", - "wsgi.multithread", - "wsgi.multiprocess", - "wsgi.run_once", - ): - if key not in environ: - warn( - "Required environment key %r not found" % key, - WSGIWarning, - stacklevel=3, - ) - if environ["wsgi.version"] != (1, 0): - warn("Environ is not a WSGI 1.0 environ.", WSGIWarning, stacklevel=3) - - script_name = environ.get("SCRIPT_NAME", "") - path_info = environ.get("PATH_INFO", "") - - if script_name and script_name[0] != "/": - warn( - "'SCRIPT_NAME' does not start with a slash: %r" % script_name, - WSGIWarning, - stacklevel=3, - ) - - if path_info and path_info[0] != "/": - warn( - "'PATH_INFO' does not start with a slash: %r" % path_info, - WSGIWarning, - stacklevel=3, - ) - - def check_start_response(self, status, headers, exc_info): - check_string("status", status) - status_code = status.split(None, 1)[0] - - if len(status_code) != 3 or not status_code.isdigit(): - warn(WSGIWarning("Status code must be three digits"), stacklevel=3) - - if len(status) < 4 or status[3] != " ": - warn( - WSGIWarning( - "Invalid value for status %r. Valid " - "status strings are three digits, a space " - "and a status explanation" - ), - stacklevel=3, - ) - - status_code = int(status_code) - - if status_code < 100: - warn(WSGIWarning("status code < 100 detected"), stacklevel=3) - - if type(headers) is not list: - warn(WSGIWarning("header list is not a list"), stacklevel=3) - - for item in headers: - if type(item) is not tuple or len(item) != 2: - warn(WSGIWarning("Headers must tuple 2-item tuples"), stacklevel=3) - name, value = item - if type(name) is not str or type(value) is not str: - warn(WSGIWarning("header items must be strings"), stacklevel=3) - if name.lower() == "status": - warn( - WSGIWarning( - "The status header is not supported due to " - "conflicts with the CGI spec." - ), - stacklevel=3, - ) - - if exc_info is not None and not isinstance(exc_info, tuple): - warn(WSGIWarning("invalid value for exc_info"), stacklevel=3) - - headers = Headers(headers) - self.check_headers(headers) - - return status_code, headers - - def check_headers(self, headers): - etag = headers.get("etag") - - if etag is not None: - if etag.startswith(("W/", "w/")): - if etag.startswith("w/"): - warn( - HTTPWarning("weak etag indicator should be upcase."), - stacklevel=4, - ) - - etag = etag[2:] - - if not (etag[:1] == etag[-1:] == '"'): - warn(HTTPWarning("unquoted etag emitted."), stacklevel=4) - - location = headers.get("location") - - if location is not None: - if not urlparse(location).netloc: - warn( - HTTPWarning("absolute URLs required for location header"), - stacklevel=4, - ) - - def check_iterator(self, app_iter): - if isinstance(app_iter, string_types): - warn( - "The application returned astring. The response will send one character" - " at a time to the client, which will kill performance. Return a list" - " or iterable instead.", - WSGIWarning, - stacklevel=3, - ) - - def __call__(self, *args, **kwargs): - if len(args) != 2: - warn("A WSGI app takes two arguments.", WSGIWarning, stacklevel=2) - - if kwargs: - warn( - "A WSGI app does not take keyword arguments.", WSGIWarning, stacklevel=2 - ) - - environ, start_response = args - - self.check_environ(environ) - environ["wsgi.input"] = InputStream(environ["wsgi.input"]) - environ["wsgi.errors"] = ErrorStream(environ["wsgi.errors"]) - - # Hook our own file wrapper in so that applications will always - # iterate to the end and we can check the content length. - environ["wsgi.file_wrapper"] = FileWrapper - - headers_set = [] - chunks = [] - - def checking_start_response(*args, **kwargs): - if len(args) not in (2, 3): - warn( - "Invalid number of arguments: %s, expected 2 or 3." % len(args), - WSGIWarning, - stacklevel=2, - ) - - if kwargs: - warn("'start_response' does not take keyword arguments.", WSGIWarning) - - status, headers = args[:2] - - if len(args) == 3: - exc_info = args[2] - else: - exc_info = None - - headers_set[:] = self.check_start_response(status, headers, exc_info) - return GuardedWrite(start_response(status, headers, exc_info), chunks) - - app_iter = self.app(environ, checking_start_response) - self.check_iterator(app_iter) - return GuardedIterator(app_iter, headers_set, chunks) |