codekingpro/portable-devtools

1"""2WSGI Protocol Linter3====================4 5This module provides a middleware that performs sanity checks on the6behavior of the WSGI server and application. It checks that the7:pep:`3333` WSGI spec is properly implemented. It also warns on some8common HTTP errors such as non-empty responses for 304 status codes.9 10.. autoclass:: LintMiddleware11 12:copyright: 2007 Pallets13:license: BSD-3-Clause14"""15 16from __future__ import annotations17 18import typing as t19from types import TracebackType20from urllib.parse import urlparse21from warnings import warn22 23from ..datastructures import Headers24from ..http import is_entity_header25from ..wsgi import FileWrapper26 27if t.TYPE_CHECKING:28    from _typeshed.wsgi import StartResponse29    from _typeshed.wsgi import WSGIApplication30    from _typeshed.wsgi import WSGIEnvironment31 32 33class WSGIWarning(Warning):34    """Warning class for WSGI warnings."""35 36 37class HTTPWarning(Warning):38    """Warning class for HTTP warnings."""39 40 41def check_type(context: str, obj: object, need: type = str) -> None:42    if type(obj) is not need:43        warn(44            f"{context!r} requires {need.__name__!r}, got {type(obj).__name__!r}.",45            WSGIWarning,46            stacklevel=3,47        )48 49 50class InputStream:51    def __init__(self, stream: t.IO[bytes]) -> None:52        self._stream = stream53 54    def read(self, *args: t.Any) -> bytes:55        if len(args) == 0:56            warn(57                "WSGI does not guarantee an EOF marker on the input stream, thus making"58                " calls to 'wsgi.input.read()' unsafe. Conforming servers may never"59                " return from this call.",60                WSGIWarning,61                stacklevel=2,62            )63        elif len(args) != 1:64            warn(65                "Too many parameters passed to 'wsgi.input.read()'.",66                WSGIWarning,67                stacklevel=2,68            )69        return self._stream.read(*args)70 71    def readline(self, *args: t.Any) -> bytes:72        if len(args) == 0:73            warn(74                "Calls to 'wsgi.input.readline()' without arguments are unsafe. Use"75                " 'wsgi.input.read()' instead.",76                WSGIWarning,77                stacklevel=2,78            )79        elif len(args) == 1:80            warn(81                "'wsgi.input.readline()' was called with a size hint. WSGI does not"82                " support this, although it's available on all major servers.",83                WSGIWarning,84                stacklevel=2,85            )86        else:87            raise TypeError("Too many arguments passed to 'wsgi.input.readline()'.")88        return self._stream.readline(*args)89 90    def __iter__(self) -> t.Iterator[bytes]:91        try:92            return iter(self._stream)93        except TypeError:94            warn("'wsgi.input' is not iterable.", WSGIWarning, stacklevel=2)95            return iter(())96 97    def close(self) -> None:98        warn("The application closed the input stream!", WSGIWarning, stacklevel=2)99        self._stream.close()100 101 102class ErrorStream:103    def __init__(self, stream: t.IO[str]) -> None:104        self._stream = stream105 106    def write(self, s: str) -> None:107        check_type("wsgi.error.write()", s, str)108        self._stream.write(s)109 110    def flush(self) -> None:111        self._stream.flush()112 113    def writelines(self, seq: t.Iterable[str]) -> None:114        for line in seq:115            self.write(line)116 117    def close(self) -> None:118        warn("The application closed the error stream!", WSGIWarning, stacklevel=2)119        self._stream.close()120 121 122class GuardedWrite:123    def __init__(self, write: t.Callable[[bytes], object], chunks: list[int]) -> None:124        self._write = write125        self._chunks = chunks126 127    def __call__(self, s: bytes) -> None:128        check_type("write()", s, bytes)129        self._write(s)130        self._chunks.append(len(s))131 132 133class GuardedIterator:134    def __init__(135        self,136        iterator: t.Iterable[bytes],137        headers_set: tuple[int, Headers],138        chunks: list[int],139    ) -> None:140        self._iterator = iterator141        self._next = iter(iterator).__next__142        self.closed = False143        self.headers_set = headers_set144        self.chunks = chunks145 146    def __iter__(self) -> GuardedIterator:147        return self148 149    def __next__(self) -> bytes:150        if self.closed:151            warn("Iterated over closed 'app_iter'.", WSGIWarning, stacklevel=2)152 153        rv = self._next()154 155        if not self.headers_set:156            warn(157                "The application returned before it started the response.",158                WSGIWarning,159                stacklevel=2,160            )161 162        check_type("application iterator items", rv, bytes)163        self.chunks.append(len(rv))164        return rv165 166    def close(self) -> None:167        self.closed = True168 169        if hasattr(self._iterator, "close"):170            self._iterator.close()171 172        if self.headers_set:173            status_code, headers = self.headers_set174            bytes_sent = sum(self.chunks)175            content_length = headers.get("content-length", type=int)176 177            if status_code == 304:178                for key, _value in headers:179                    key = key.lower()180                    if key not in ("expires", "content-location") and is_entity_header(181                        key182                    ):183                        warn(184                            f"Entity header {key!r} found in 304 response.",185                            HTTPWarning,186                            stacklevel=2,187                        )188                if bytes_sent:189                    warn(190                        "304 responses must not have a body.",191                        HTTPWarning,192                        stacklevel=2,193                    )194            elif 100 <= status_code < 200 or status_code == 204:195                if content_length != 0:196                    warn(197                        f"{status_code} responses must have an empty content length.",198                        HTTPWarning,199                        stacklevel=2,200                    )201                if bytes_sent:202                    warn(203                        f"{status_code} responses must not have a body.",204                        HTTPWarning,205                        stacklevel=2,206                    )207            elif content_length is not None and content_length != bytes_sent:208                warn(209                    "Content-Length and the number of bytes sent to the"210                    " client do not match.",211                    WSGIWarning,212                    stacklevel=2,213                )214 215    def __del__(self) -> None:216        if not self.closed:217            try:218                warn(219                    "Iterator was garbage collected before it was closed.",220                    WSGIWarning,221                    stacklevel=2,222                )223            except Exception:224                pass225 226 227class LintMiddleware:228    """Warns about common errors in the WSGI and HTTP behavior of the229    server and wrapped application. Some of the issues it checks are:230 231    -   invalid status codes232    -   non-bytes sent to the WSGI server233    -   strings returned from the WSGI application234    -   non-empty conditional responses235    -   unquoted etags236    -   relative URLs in the Location header237    -   unsafe calls to wsgi.input238    -   unclosed iterators239 240    Error information is emitted using the :mod:`warnings` module.241 242    :param app: The WSGI application to wrap.243 244    .. code-block:: python245 246        from werkzeug.middleware.lint import LintMiddleware247        app = LintMiddleware(app)248    """249 250    def __init__(self, app: WSGIApplication) -> None:251        self.app = app252 253    def check_environ(self, environ: WSGIEnvironment) -> None:254        if type(environ) is not dict:  # noqa: E721255            warn(256                "WSGI environment is not a standard Python dict.",257                WSGIWarning,258                stacklevel=4,259            )260        for key in (261            "REQUEST_METHOD",262            "SERVER_NAME",263            "SERVER_PORT",264            "wsgi.version",265            "wsgi.input",266            "wsgi.errors",267            "wsgi.multithread",268            "wsgi.multiprocess",269            "wsgi.run_once",270        ):271            if key not in environ:272                warn(273                    f"Required environment key {key!r} not found",274                    WSGIWarning,275                    stacklevel=3,276                )277        if environ["wsgi.version"] != (1, 0):278            warn("Environ is not a WSGI 1.0 environ.", WSGIWarning, stacklevel=3)279 280        script_name = environ.get("SCRIPT_NAME", "")281        path_info = environ.get("PATH_INFO", "")282 283        if script_name and script_name[0] != "/":284            warn(285                f"'SCRIPT_NAME' does not start with a slash: {script_name!r}",286                WSGIWarning,287                stacklevel=3,288            )289 290        if path_info and path_info[0] != "/":291            warn(292                f"'PATH_INFO' does not start with a slash: {path_info!r}",293                WSGIWarning,294                stacklevel=3,295            )296 297    def check_start_response(298        self,299        status: str,300        headers: list[tuple[str, str]],301        exc_info: None | (tuple[type[BaseException], BaseException, TracebackType]),302    ) -> tuple[int, Headers]:303        check_type("status", status, str)304        status_code_str = status.split(None, 1)[0]305 306        if len(status_code_str) != 3 or not status_code_str.isdecimal():307            warn("Status code must be three digits.", WSGIWarning, stacklevel=3)308 309        if len(status) < 4 or status[3] != " ":310            warn(311                f"Invalid value for status {status!r}. Valid status strings are three"312                " digits, a space and a status explanation.",313                WSGIWarning,314                stacklevel=3,315            )316 317        status_code = int(status_code_str)318 319        if status_code < 100:320            warn("Status code < 100 detected.", WSGIWarning, stacklevel=3)321 322        if type(headers) is not list:  # noqa: E721323            warn("Header list is not a list.", WSGIWarning, stacklevel=3)324 325        for item in headers:326            if type(item) is not tuple or len(item) != 2:327                warn("Header items must be 2-item tuples.", WSGIWarning, stacklevel=3)328            name, value = item329            if type(name) is not str or type(value) is not str:  # noqa: E721330                warn(331                    "Header keys and values must be strings.", WSGIWarning, stacklevel=3332                )333            if name.lower() == "status":334                warn(335                    "The status header is not supported due to"336                    " conflicts with the CGI spec.",337                    WSGIWarning,338                    stacklevel=3,339                )340 341        if exc_info is not None and not isinstance(exc_info, tuple):342            warn("Invalid value for exc_info.", WSGIWarning, stacklevel=3)343 344        headers_obj = Headers(headers)345        self.check_headers(headers_obj)346 347        return status_code, headers_obj348 349    def check_headers(self, headers: Headers) -> None:350        etag = headers.get("etag")351 352        if etag is not None:353            if etag.startswith(("W/", "w/")):354                if etag.startswith("w/"):355                    warn(356                        "Weak etag indicator should be upper case.",357                        HTTPWarning,358                        stacklevel=4,359                    )360 361                etag = etag[2:]362 363            if not (etag[:1] == etag[-1:] == '"'):364                warn("Unquoted etag emitted.", HTTPWarning, stacklevel=4)365 366        location = headers.get("location")367 368        if location is not None:369            if not urlparse(location).netloc:370                warn(371                    "Absolute URLs required for location header.",372                    HTTPWarning,373                    stacklevel=4,374                )375 376    def check_iterator(self, app_iter: t.Iterable[bytes]) -> None:377        if isinstance(app_iter, str):378            warn(379                "The application returned a string. The response will send one"380                " character at a time to the client, which will kill performance."381                " Return a list or iterable instead.",382                WSGIWarning,383                stacklevel=3,384            )385 386    def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Iterable[bytes]:387        if len(args) != 2:388            warn("A WSGI app takes two arguments.", WSGIWarning, stacklevel=2)389 390        if kwargs:391            warn(392                "A WSGI app does not take keyword arguments.", WSGIWarning, stacklevel=2393            )394 395        environ: WSGIEnvironment = args[0]396        start_response: StartResponse = args[1]397 398        self.check_environ(environ)399        environ["wsgi.input"] = InputStream(environ["wsgi.input"])400        environ["wsgi.errors"] = ErrorStream(environ["wsgi.errors"])401 402        # Hook our own file wrapper in so that applications will always403        # iterate to the end and we can check the content length.404        environ["wsgi.file_wrapper"] = FileWrapper405 406        headers_set: list[t.Any] = []407        chunks: list[int] = []408 409        def checking_start_response(410            *args: t.Any, **kwargs: t.Any411        ) -> t.Callable[[bytes], None]:412            if len(args) not in {2, 3}:413                warn(414                    f"Invalid number of arguments: {len(args)}, expected 2 or 3.",415                    WSGIWarning,416                    stacklevel=2,417                )418 419            if kwargs:420                warn(421                    "'start_response' does not take keyword arguments.",422                    WSGIWarning,423                    stacklevel=2,424                )425 426            status: str = args[0]427            headers: list[tuple[str, str]] = args[1]428            exc_info: (429                None | (tuple[type[BaseException], BaseException, TracebackType])430            ) = args[2] if len(args) == 3 else None431 432            headers_set[:] = self.check_start_response(status, headers, exc_info)433            return GuardedWrite(start_response(status, headers, exc_info), chunks)434 435        app_iter = self.app(environ, t.cast("StartResponse", checking_start_response))436        self.check_iterator(app_iter)437        return GuardedIterator(438            app_iter, t.cast(tuple[int, Headers], headers_set), chunks439        )440