codekingpro/portable-devtools

1from __future__ import annotations2 3import io4import mimetypes5import os6import pkgutil7import re8import sys9import typing as t10import unicodedata11from datetime import datetime12from time import time13from urllib.parse import quote14from zlib import adler3215 16from markupsafe import escape17 18from ._internal import _DictAccessorProperty19from ._internal import _missing20from ._internal import _TAccessorValue21from .datastructures import Headers22from .exceptions import NotFound23from .exceptions import RequestedRangeNotSatisfiable24from .security import _windows_device_files25from .security import safe_join26from .wsgi import wrap_file27 28if t.TYPE_CHECKING:29    from _typeshed.wsgi import WSGIEnvironment30 31    from .wrappers.request import Request32    from .wrappers.response import Response33 34_T = t.TypeVar("_T")35 36_entity_re = re.compile(r"&([^;]+);")37_filename_ascii_strip_re = re.compile(r"[^A-Za-z0-9_.-]")38 39 40class cached_property(property, t.Generic[_T]):41    """A :func:`property` that is only evaluated once. Subsequent access42    returns the cached value. Setting the property sets the cached43    value. Deleting the property clears the cached value, accessing it44    again will evaluate it again.45 46    .. code-block:: python47 48        class Example:49            @cached_property50            def value(self):51                # calculate something important here52                return 4253 54        e = Example()55        e.value  # evaluates56        e.value  # uses cache57        e.value = 16  # sets cache58        del e.value  # clears cache59 60    If the class defines ``__slots__``, it must add ``_cache_{name}`` as61    a slot. Alternatively, it can add ``__dict__``, but that's usually62    not desirable.63 64    .. versionchanged:: 2.165        Works with ``__slots__``.66 67    .. versionchanged:: 2.068        ``del obj.name`` clears the cached value.69    """70 71    def __init__(72        self,73        fget: t.Callable[[t.Any], _T],74        name: str | None = None,75        doc: str | None = None,76    ) -> None:77        super().__init__(fget, doc=doc)78        self.__name__ = name or fget.__name__79        self.slot_name = f"_cache_{self.__name__}"80        self.__module__ = fget.__module__81 82    def __set__(self, obj: object, value: _T) -> None:83        if hasattr(obj, "__dict__"):84            obj.__dict__[self.__name__] = value85        else:86            setattr(obj, self.slot_name, value)87 88    def __get__(self, obj: object, type: type = None) -> _T:  # type: ignore89        if obj is None:90            return self  # type: ignore91 92        obj_dict = getattr(obj, "__dict__", None)93 94        if obj_dict is not None:95            value: _T = obj_dict.get(self.__name__, _missing)96        else:97            value = getattr(obj, self.slot_name, _missing)  # type: ignore[arg-type]98 99        if value is _missing:100            value = self.fget(obj)  # type: ignore101 102            if obj_dict is not None:103                obj.__dict__[self.__name__] = value104            else:105                setattr(obj, self.slot_name, value)106 107        return value108 109    def __delete__(self, obj: object) -> None:110        if hasattr(obj, "__dict__"):111            del obj.__dict__[self.__name__]112        else:113            setattr(obj, self.slot_name, _missing)114 115 116class environ_property(_DictAccessorProperty[_TAccessorValue]):117    """Maps request attributes to environment variables. This works not only118    for the Werkzeug request object, but also any other class with an119    environ attribute:120 121    >>> class Test(object):122    ...     environ = {'key': 'value'}123    ...     test = environ_property('key')124    >>> var = Test()125    >>> var.test126    'value'127 128    If you pass it a second value it's used as default if the key does not129    exist, the third one can be a converter that takes a value and converts130    it.  If it raises :exc:`ValueError` or :exc:`TypeError` the default value131    is used. If no default value is provided `None` is used.132 133    Per default the property is read only.  You have to explicitly enable it134    by passing ``read_only=False`` to the constructor.135    """136 137    read_only = True138 139    def lookup(self, obj: Request) -> WSGIEnvironment:140        return obj.environ141 142 143class header_property(_DictAccessorProperty[_TAccessorValue]):144    """Like `environ_property` but for headers."""145 146    def lookup(self, obj: Request | Response) -> Headers:  # type: ignore[override]147        return obj.headers148 149 150# https://cgit.freedesktop.org/xdg/shared-mime-info/tree/freedesktop.org.xml.in151# https://www.iana.org/assignments/media-types/media-types.xhtml152# Types listed in the XDG mime info that have a charset in the IANA registration.153_charset_mimetypes = {154    "application/ecmascript",155    "application/javascript",156    "application/sql",157    "application/xml",158    "application/xml-dtd",159    "application/xml-external-parsed-entity",160}161 162 163def get_content_type(mimetype: str, charset: str) -> str:164    """Returns the full content type string with charset for a mimetype.165 166    If the mimetype represents text, the charset parameter will be167    appended, otherwise the mimetype is returned unchanged.168 169    :param mimetype: The mimetype to be used as content type.170    :param charset: The charset to be appended for text mimetypes.171    :return: The content type.172 173    .. versionchanged:: 0.15174        Any type that ends with ``+xml`` gets a charset, not just those175        that start with ``application/``. Known text types such as176        ``application/javascript`` are also given charsets.177    """178    if (179        mimetype.startswith("text/")180        or mimetype in _charset_mimetypes181        or mimetype.endswith("+xml")182    ):183        mimetype += f"; charset={charset}"184 185    return mimetype186 187 188def secure_filename(filename: str) -> str:189    r"""Pass it a filename and it will return a secure version of it.  This190    filename can then safely be stored on a regular file system and passed191    to :func:`os.path.join`.  The filename returned is an ASCII only string192    for maximum portability.193 194    On windows systems the function also makes sure that the file is not195    named after one of the special device files.196 197    >>> secure_filename("My cool movie.mov")198    'My_cool_movie.mov'199    >>> secure_filename("../../../etc/passwd")200    'etc_passwd'201    >>> secure_filename('i contain cool \xfcml\xe4uts.txt')202    'i_contain_cool_umlauts.txt'203 204    The function might return an empty filename.  It's your responsibility205    to ensure that the filename is unique and that you abort or206    generate a random filename if the function returned an empty one.207 208    .. versionadded:: 0.5209 210    :param filename: the filename to secure211    """212    filename = unicodedata.normalize("NFKD", filename)213    filename = filename.encode("ascii", "ignore").decode("ascii")214 215    for sep in os.sep, os.path.altsep:216        if sep:217            filename = filename.replace(sep, " ")218    filename = str(_filename_ascii_strip_re.sub("", "_".join(filename.split()))).strip(219        "._"220    )221 222    # on nt a couple of special files are present in each folder.  We223    # have to ensure that the target file is not such a filename.  In224    # this case we prepend an underline225    if (226        os.name == "nt"227        and filename228        and filename.split(".")[0].upper() in _windows_device_files229    ):230        filename = f"_{filename}"231 232    return filename233 234 235def redirect(236    location: str, code: int = 302, Response: type[Response] | None = None237) -> Response:238    """Returns a response object (a WSGI application) that, if called,239    redirects the client to the target location. Supported codes are240    301, 302, 303, 305, 307, and 308. 300 is not supported because241    it's not a real redirect and 304 because it's the answer for a242    request with a request with defined If-Modified-Since headers.243 244    .. versionadded:: 0.6245       The location can now be a unicode string that is encoded using246       the :func:`iri_to_uri` function.247 248    .. versionadded:: 0.10249        The class used for the Response object can now be passed in.250 251    :param location: the location the response should redirect to.252    :param code: the redirect status code. defaults to 302.253    :param class Response: a Response class to use when instantiating a254        response. The default is :class:`werkzeug.wrappers.Response` if255        unspecified.256    """257    if Response is None:258        from .wrappers import Response259 260    html_location = escape(location)261    response = Response(  # type: ignore[misc]262        "<!doctype html>\n"263        "<html lang=en>\n"264        "<title>Redirecting...</title>\n"265        "<h1>Redirecting...</h1>\n"266        "<p>You should be redirected automatically to the target URL: "267        f'<a href="{html_location}">{html_location}</a>. If not, click the link.\n',268        code,269        mimetype="text/html",270    )271    response.headers["Location"] = location272    return response273 274 275def append_slash_redirect(environ: WSGIEnvironment, code: int = 308) -> Response:276    """Redirect to the current URL with a slash appended.277 278    If the current URL is ``/user/42``, the redirect URL will be279    ``42/``. When joined to the current URL during response280    processing or by the browser, this will produce ``/user/42/``.281 282    The behavior is undefined if the path ends with a slash already. If283    called unconditionally on a URL, it may produce a redirect loop.284 285    :param environ: Use the path and query from this WSGI environment286        to produce the redirect URL.287    :param code: the status code for the redirect.288 289    .. versionchanged:: 2.1290        Produce a relative URL that only modifies the last segment.291        Relevant when the current path has multiple segments.292 293    .. versionchanged:: 2.1294        The default status code is 308 instead of 301. This preserves295        the request method and body.296    """297    tail = environ["PATH_INFO"].rpartition("/")[2]298 299    if not tail:300        new_path = "./"301    else:302        new_path = f"{tail}/"303 304    query_string = environ.get("QUERY_STRING")305 306    if query_string:307        new_path = f"{new_path}?{query_string}"308 309    return redirect(new_path, code)310 311 312def send_file(313    path_or_file: os.PathLike[str] | str | t.IO[bytes],314    environ: WSGIEnvironment,315    mimetype: str | None = None,316    as_attachment: bool = False,317    download_name: str | None = None,318    conditional: bool = True,319    etag: bool | str = True,320    last_modified: datetime | int | float | None = None,321    max_age: None | (int | t.Callable[[str | None], int | None]) = None,322    use_x_sendfile: bool = False,323    response_class: type[Response] | None = None,324    _root_path: os.PathLike[str] | str | None = None,325) -> Response:326    """Send the contents of a file to the client.327 328    The first argument can be a file path or a file-like object. Paths329    are preferred in most cases because Werkzeug can manage the file and330    get extra information from the path. Passing a file-like object331    requires that the file is opened in binary mode, and is mostly332    useful when building a file in memory with :class:`io.BytesIO`.333 334    Never pass file paths provided by a user. The path is assumed to be335    trusted, so a user could craft a path to access a file you didn't336    intend. Use :func:`send_from_directory` to safely serve user-provided paths.337 338    If the WSGI server sets a ``file_wrapper`` in ``environ``, it is339    used, otherwise Werkzeug's built-in wrapper is used. Alternatively,340    if the HTTP server supports ``X-Sendfile``, ``use_x_sendfile=True``341    will tell the server to send the given path, which is much more342    efficient than reading it in Python.343 344    :param path_or_file: The path to the file to send, relative to the345        current working directory if a relative path is given.346        Alternatively, a file-like object opened in binary mode. Make347        sure the file pointer is seeked to the start of the data.348    :param environ: The WSGI environ for the current request.349    :param mimetype: The MIME type to send for the file. If not350        provided, it will try to detect it from the file name.351    :param as_attachment: Indicate to a browser that it should offer to352        save the file instead of displaying it.353    :param download_name: The default name browsers will use when saving354        the file. Defaults to the passed file name.355    :param conditional: Enable conditional and range responses based on356        request headers. Requires passing a file path and ``environ``.357    :param etag: Calculate an ETag for the file, which requires passing358        a file path. Can also be a string to use instead.359    :param last_modified: The last modified time to send for the file,360        in seconds. If not provided, it will try to detect it from the361        file path.362    :param max_age: How long the client should cache the file, in363        seconds. If set, ``Cache-Control`` will be ``public``, otherwise364        it will be ``no-cache`` to prefer conditional caching.365    :param use_x_sendfile: Set the ``X-Sendfile`` header to let the366        server to efficiently send the file. Requires support from the367        HTTP server. Requires passing a file path.368    :param response_class: Build the response using this class. Defaults369        to :class:`~werkzeug.wrappers.Response`.370    :param _root_path: Do not use. For internal use only. Use371        :func:`send_from_directory` to safely send files under a path.372 373    .. versionchanged:: 2.0.2374        ``send_file`` only sets a detected ``Content-Encoding`` if375        ``as_attachment`` is disabled.376 377    .. versionadded:: 2.0378        Adapted from Flask's implementation.379 380    .. versionchanged:: 2.0381        ``download_name`` replaces Flask's ``attachment_filename``382         parameter. If ``as_attachment=False``, it is passed with383         ``Content-Disposition: inline`` instead.384 385    .. versionchanged:: 2.0386        ``max_age`` replaces Flask's ``cache_timeout`` parameter.387        ``conditional`` is enabled and ``max_age`` is not set by388        default.389 390    .. versionchanged:: 2.0391        ``etag`` replaces Flask's ``add_etags`` parameter. It can be a392        string to use instead of generating one.393 394    .. versionchanged:: 2.0395        If an encoding is returned when guessing ``mimetype`` from396        ``download_name``, set the ``Content-Encoding`` header.397    """398    if response_class is None:399        from .wrappers import Response400 401        response_class = Response402 403    path: str | None = None404    file: t.IO[bytes] | None = None405    size: int | None = None406    mtime: float | None = None407    headers = Headers()408 409    if isinstance(path_or_file, (os.PathLike, str)) or hasattr(410        path_or_file, "__fspath__"411    ):412        path_or_file = t.cast("os.PathLike[str] | str", path_or_file)413 414        # Flask will pass app.root_path, allowing its send_file wrapper415        # to not have to deal with paths.416        if _root_path is not None:417            path = os.path.join(_root_path, path_or_file)418        else:419            path = os.path.abspath(path_or_file)420 421        stat = os.stat(path)422        size = stat.st_size423        mtime = stat.st_mtime424    else:425        file = path_or_file426 427    if download_name is None and path is not None:428        download_name = os.path.basename(path)429 430    if mimetype is None:431        if download_name is None:432            raise TypeError(433                "Unable to detect the MIME type because a file name is"434                " not available. Either set 'download_name', pass a"435                " path instead of a file, or set 'mimetype'."436            )437 438        mimetype, encoding = mimetypes.guess_type(download_name)439 440        if mimetype is None:441            mimetype = "application/octet-stream"442 443        # Don't send encoding for attachments, it causes browsers to444        # save decompress tar.gz files.445        if encoding is not None and not as_attachment:446            headers.set("Content-Encoding", encoding)447 448    if download_name is not None:449        try:450            download_name.encode("ascii")451        except UnicodeEncodeError:452            simple = unicodedata.normalize("NFKD", download_name)453            simple = simple.encode("ascii", "ignore").decode("ascii")454            # safe = RFC 5987 attr-char455            quoted = quote(download_name, safe="!#$&+-.^_`|~")456            names = {"filename": simple, "filename*": f"UTF-8''{quoted}"}457        else:458            names = {"filename": download_name}459 460        value = "attachment" if as_attachment else "inline"461        headers.set("Content-Disposition", value, **names)462    elif as_attachment:463        raise TypeError(464            "No name provided for attachment. Either set"465            " 'download_name' or pass a path instead of a file."466        )467 468    if use_x_sendfile and path is not None:469        headers["X-Sendfile"] = path470        data = None471    else:472        if file is None:473            file = open(path, "rb")  # type: ignore474        elif isinstance(file, io.BytesIO):475            size = file.getbuffer().nbytes476        elif isinstance(file, io.TextIOBase):477            raise ValueError("Files must be opened in binary mode or use BytesIO.")478 479        data = wrap_file(environ, file)480 481    rv = response_class(482        data, mimetype=mimetype, headers=headers, direct_passthrough=True483    )484 485    if size is not None:486        rv.content_length = size487 488    if last_modified is not None:489        rv.last_modified = last_modified  # type: ignore490    elif mtime is not None:491        rv.last_modified = mtime  # type: ignore492 493    rv.cache_control.no_cache = True494 495    # Flask will pass app.get_send_file_max_age, allowing its send_file496    # wrapper to not have to deal with paths.497    if callable(max_age):498        max_age = max_age(path)499 500    if max_age is not None:501        if max_age > 0:502            rv.cache_control.no_cache = None503            rv.cache_control.public = True504 505        rv.cache_control.max_age = max_age506        rv.expires = int(time() + max_age)  # type: ignore507 508    if isinstance(etag, str):509        rv.set_etag(etag)510    elif etag and path is not None:511        check = adler32(path.encode()) & 0xFFFFFFFF512        rv.set_etag(f"{mtime}-{size}-{check}")513 514    if conditional:515        try:516            rv = rv.make_conditional(environ, accept_ranges=True, complete_length=size)517        except RequestedRangeNotSatisfiable:518            if file is not None:519                file.close()520 521            raise522 523        # Some x-sendfile implementations incorrectly ignore the 304524        # status code and send the file anyway.525        if rv.status_code == 304:526            rv.headers.pop("x-sendfile", None)527 528    return rv529 530 531def send_from_directory(532    directory: os.PathLike[str] | str,533    path: os.PathLike[str] | str,534    environ: WSGIEnvironment,535    **kwargs: t.Any,536) -> Response:537    """Send a file from within a directory using :func:`send_file`.538 539    This is a secure way to serve files from a folder, such as static540    files or uploads. Uses :func:`~werkzeug.security.safe_join` to541    ensure the path coming from the client is not maliciously crafted to542    point outside the specified directory.543 544    If the final path does not point to an existing regular file,545    returns a 404 :exc:`~werkzeug.exceptions.NotFound` error.546 547    :param directory: The directory that ``path`` must be located under. This *must not*548        be a value provided by the client, otherwise it becomes insecure.549    :param path: The path to the file to send, relative to ``directory``. This is the550        part of the path provided by the client, which is checked for security.551    :param environ: The WSGI environ for the current request.552    :param kwargs: Arguments to pass to :func:`send_file`.553 554    .. versionadded:: 2.0555        Adapted from Flask's implementation.556    """557    path_str = safe_join(os.fspath(directory), os.fspath(path))558 559    if path_str is None:560        raise NotFound()561 562    # Flask will pass app.root_path, allowing its send_from_directory563    # wrapper to not have to deal with paths.564    if "_root_path" in kwargs:565        path_str = os.path.join(kwargs["_root_path"], path_str)566 567    if not os.path.isfile(path_str):568        raise NotFound()569 570    return send_file(path_str, environ, **kwargs)571 572 573def import_string(import_name: str, silent: bool = False) -> t.Any:574    """Imports an object based on a string.  This is useful if you want to575    use import paths as endpoints or something similar.  An import path can576    be specified either in dotted notation (``xml.sax.saxutils.escape``)577    or with a colon as object delimiter (``xml.sax.saxutils:escape``).578 579    If `silent` is True the return value will be `None` if the import fails.580 581    :param import_name: the dotted name for the object to import.582    :param silent: if set to `True` import errors are ignored and583                   `None` is returned instead.584    :return: imported object585    """586    import_name = import_name.replace(":", ".")587    try:588        try:589            __import__(import_name)590        except ImportError:591            if "." not in import_name:592                raise593        else:594            return sys.modules[import_name]595 596        module_name, obj_name = import_name.rsplit(".", 1)597        module = __import__(module_name, globals(), locals(), [obj_name])598        try:599            return getattr(module, obj_name)600        except AttributeError as e:601            raise ImportError(e) from None602 603    except ImportError as e:604        if not silent:605            raise ImportStringError(import_name, e).with_traceback(606                sys.exc_info()[2]607            ) from None608 609    return None610 611 612def find_modules(613    import_path: str, include_packages: bool = False, recursive: bool = False614) -> t.Iterator[str]:615    """Finds all the modules below a package.  This can be useful to616    automatically import all views / controllers so that their metaclasses /617    function decorators have a chance to register themselves on the618    application.619 620    Packages are not returned unless `include_packages` is `True`.  This can621    also recursively list modules but in that case it will import all the622    packages to get the correct load path of that module.623 624    :param import_path: the dotted name for the package to find child modules.625    :param include_packages: set to `True` if packages should be returned, too.626    :param recursive: set to `True` if recursion should happen.627    :return: generator628    """629    module = import_string(import_path)630    path = getattr(module, "__path__", None)631    if path is None:632        raise ValueError(f"{import_path!r} is not a package")633    basename = f"{module.__name__}."634    for _importer, modname, ispkg in pkgutil.iter_modules(path):635        modname = basename + modname636        if ispkg:637            if include_packages:638                yield modname639            if recursive:640                yield from find_modules(modname, include_packages, True)641        else:642            yield modname643 644 645class ImportStringError(ImportError):646    """Provides information about a failed :func:`import_string` attempt."""647 648    #: String in dotted notation that failed to be imported.649    import_name: str650    #: Wrapped exception.651    exception: BaseException652 653    def __init__(self, import_name: str, exception: BaseException) -> None:654        self.import_name = import_name655        self.exception = exception656        msg = import_name657        name = ""658        tracked = []659        for part in import_name.replace(":", ".").split("."):660            name = f"{name}.{part}" if name else part661            imported = import_string(name, silent=True)662            if imported:663                tracked.append((name, getattr(imported, "__file__", None)))664            else:665                track = [f"- {n!r} found in {i!r}." for n, i in tracked]666                track.append(f"- {name!r} not found.")667                track_str = "\n".join(track)668                msg = (669                    f"import_string() failed for {import_name!r}. Possible reasons"670                    f" are:\n\n"671                    "- missing __init__.py in a package;\n"672                    "- package or module path not included in sys.path;\n"673                    "- duplicated package or module name taking precedence in"674                    " sys.path;\n"675                    "- missing module, class, function or variable;\n\n"676                    f"Debugged import:\n\n{track_str}\n\n"677                    f"Original exception:\n\n{type(exception).__name__}: {exception}"678                )679                break680 681        super().__init__(msg)682 683    def __repr__(self) -> str:684        return f"<{type(self).__name__}({self.import_name!r}, {self.exception!r})>"685