Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/werkzeug/utils.py: 23%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import io
4import mimetypes
5import os
6import pkgutil
7import re
8import sys
9import typing as t
10import unicodedata
11from datetime import datetime
12from time import time
13from urllib.parse import quote
14from zlib import adler32
16from markupsafe import escape
18from ._internal import _DictAccessorProperty
19from ._internal import _missing
20from ._internal import _TAccessorValue
21from .datastructures import Headers
22from .exceptions import NotFound
23from .exceptions import RequestedRangeNotSatisfiable
24from .security import _windows_device_files
25from .security import safe_join
26from .wsgi import wrap_file
28if t.TYPE_CHECKING:
29 from _typeshed.wsgi import WSGIEnvironment
31 from .wrappers.request import Request
32 from .wrappers.response import Response
34_T = t.TypeVar("_T")
36_entity_re = re.compile(r"&([^;]+);")
37_filename_ascii_strip_re = re.compile(r"[^A-Za-z0-9_.-]")
40class cached_property(property, t.Generic[_T]):
41 """A :func:`property` that is only evaluated once. Subsequent access
42 returns the cached value. Setting the property sets the cached
43 value. Deleting the property clears the cached value, accessing it
44 again will evaluate it again.
46 .. code-block:: python
48 class Example:
49 @cached_property
50 def value(self):
51 # calculate something important here
52 return 42
54 e = Example()
55 e.value # evaluates
56 e.value # uses cache
57 e.value = 16 # sets cache
58 del e.value # clears cache
60 If the class defines ``__slots__``, it must add ``_cache_{name}`` as
61 a slot. Alternatively, it can add ``__dict__``, but that's usually
62 not desirable.
64 .. versionchanged:: 2.1
65 Works with ``__slots__``.
67 .. versionchanged:: 2.0
68 ``del obj.name`` clears the cached value.
69 """
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__
82 def __set__(self, obj: object, value: _T) -> None:
83 if hasattr(obj, "__dict__"):
84 obj.__dict__[self.__name__] = value
85 else:
86 setattr(obj, self.slot_name, value)
88 def __get__(self, obj: object, type: type = None) -> _T: # type: ignore
89 if obj is None:
90 return self # type: ignore
92 obj_dict = getattr(obj, "__dict__", None)
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]
99 if value is _missing:
100 value = self.fget(obj) # type: ignore
102 if obj_dict is not None:
103 obj.__dict__[self.__name__] = value
104 else:
105 setattr(obj, self.slot_name, value)
107 return value
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)
116class environ_property(_DictAccessorProperty[_TAccessorValue]):
117 """Maps request attributes to environment variables. This works not only
118 for the Werkzeug request object, but also any other class with an
119 environ attribute:
121 >>> class Test(object):
122 ... environ = {'key': 'value'}
123 ... test = environ_property('key')
124 >>> var = Test()
125 >>> var.test
126 'value'
128 If you pass it a second value it's used as default if the key does not
129 exist, the third one can be a converter that takes a value and converts
130 it. If it raises :exc:`ValueError` or :exc:`TypeError` the default value
131 is used. If no default value is provided `None` is used.
133 Per default the property is read only. You have to explicitly enable it
134 by passing ``read_only=False`` to the constructor.
135 """
137 read_only = True
139 def lookup(self, obj: Request) -> WSGIEnvironment:
140 return obj.environ
143class header_property(_DictAccessorProperty[_TAccessorValue]):
144 """Like `environ_property` but for headers."""
146 def lookup(self, obj: Request | Response) -> Headers: # type: ignore[override]
147 return obj.headers
150# https://cgit.freedesktop.org/xdg/shared-mime-info/tree/freedesktop.org.xml.in
151# https://www.iana.org/assignments/media-types/media-types.xhtml
152# 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}
163def get_content_type(mimetype: str, charset: str) -> str:
164 """Returns the full content type string with charset for a mimetype.
166 If the mimetype represents text, the charset parameter will be
167 appended, otherwise the mimetype is returned unchanged.
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.
173 .. versionchanged:: 0.15
174 Any type that ends with ``+xml`` gets a charset, not just those
175 that start with ``application/``. Known text types such as
176 ``application/javascript`` are also given charsets.
177 """
178 if (
179 mimetype.startswith("text/")
180 or mimetype in _charset_mimetypes
181 or mimetype.endswith("+xml")
182 ):
183 mimetype += f"; charset={charset}"
185 return mimetype
188def secure_filename(filename: str) -> str:
189 r"""Pass it a filename and it will return a secure version of it. This
190 filename can then safely be stored on a regular file system and passed
191 to :func:`os.path.join`. The filename returned is an ASCII only string
192 for maximum portability.
194 On windows systems the function also makes sure that the file is not
195 named after one of the special device files.
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'
204 The function might return an empty filename. It's your responsibility
205 to ensure that the filename is unique and that you abort or
206 generate a random filename if the function returned an empty one.
208 .. versionadded:: 0.5
210 :param filename: the filename to secure
211 """
212 filename = unicodedata.normalize("NFKD", filename)
213 filename = filename.encode("ascii", "ignore").decode("ascii")
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 )
222 # on nt a couple of special files are present in each folder. We
223 # have to ensure that the target file is not such a filename. In
224 # this case we prepend an underline
225 if (
226 os.name == "nt"
227 and filename
228 and filename.split(".")[0].upper() in _windows_device_files
229 ):
230 filename = f"_{filename}"
232 return filename
235def redirect(
236 location: str, code: int = 303, Response: type[Response] | None = None
237) -> Response:
238 """Create a response that redirects the client to the target location.
240 The default ``303 See Other`` status code instructs the client to make a
241 ``GET`` request to the target, regardless of what method the current request
242 is. This produces the correct result for the common use cases: page redirects
243 and form success. The status codes you're likely to use are:
245 - ``303 See Other`` always uses a ``GET`` request.
246 - ``307 Temporary Redirect`` preserves the current method.
247 - ``308 Permanent Redirect`` preserves the current method, and instructs
248 the client to permanently apply the result. This is hard to undo once
249 you've sent it, so be sure the permanence is what you want.
251 Two older codes, ``302 Found`` and ``301 Moved Permanently``, are
252 superseded by ``307`` and ``308`` respectively. They were not consistently
253 implemented by clients, which tend to switch ``POST`` to ``GET`` but
254 preserve other methods. Prefer using ``303``, ``307``, and ``308`` to get
255 the exact behavior you intend. Other ``3xx`` codes are either not defined or
256 have specific use cases.
258 :param location: The URL to redirect to. The client will interpret a
259 relative URL (without the host) as relative to the host it's accessing.
260 :param code: The redirect status code. This affects how the client issues
261 the next request. Defaults to ``303``.
262 :param Response: The response class. Defaults to
263 :class:`werkzeug.wrappers.Response`.
265 .. versionchanged:: 3.2
266 ``code`` defaults to 303 instead of 302.
268 .. versionchanged:: 0.10
269 Added the ``response`` parameter.
271 .. versionchanged:: 0.6
272 ``location`` can contain Unicode characters.
273 """
274 if Response is None:
275 from .wrappers import Response
277 html_location = escape(location)
278 response = Response( # type: ignore[misc]
279 "<!doctype html>\n"
280 "<html lang=en>\n"
281 "<title>Redirecting...</title>\n"
282 "<h1>Redirecting...</h1>\n"
283 "<p>You should be redirected automatically to the target URL: "
284 f'<a href="{html_location}">{html_location}</a>. If not, click the link.\n',
285 code,
286 mimetype="text/html",
287 )
288 response.headers["Location"] = location
289 return response
292def append_slash_redirect(environ: WSGIEnvironment, code: int = 308) -> Response:
293 """Redirect to the current URL with a slash appended.
295 If the current URL is ``/user/42``, the redirect URL will be
296 ``42/``. When joined to the current URL during response
297 processing or by the browser, this will produce ``/user/42/``.
299 The behavior is undefined if the path ends with a slash already. If
300 called unconditionally on a URL, it may produce a redirect loop.
302 :param environ: Use the path and query from this WSGI environment
303 to produce the redirect URL.
304 :param code: the status code for the redirect.
306 .. versionchanged:: 2.1
307 Produce a relative URL that only modifies the last segment.
308 Relevant when the current path has multiple segments.
310 .. versionchanged:: 2.1
311 The default status code is 308 instead of 301. This preserves
312 the request method and body.
313 """
314 tail = environ["PATH_INFO"].rpartition("/")[2]
316 if not tail:
317 new_path = "./"
318 else:
319 new_path = f"{tail}/"
321 query_string = environ.get("QUERY_STRING")
323 if query_string:
324 new_path = f"{new_path}?{query_string}"
326 return redirect(new_path, code)
329def send_file(
330 path_or_file: os.PathLike[str] | str | t.IO[bytes],
331 environ: WSGIEnvironment,
332 mimetype: str | None = None,
333 as_attachment: bool = False,
334 download_name: str | None = None,
335 conditional: bool = True,
336 etag: bool | str = True,
337 last_modified: datetime | int | float | None = None,
338 max_age: None | (int | t.Callable[[str | None], int | None]) = None,
339 use_x_sendfile: bool = False,
340 response_class: type[Response] | None = None,
341 _root_path: os.PathLike[str] | str | None = None,
342) -> Response:
343 """Send the contents of a file to the client.
345 The first argument can be a file path or a file-like object. Paths
346 are preferred in most cases because Werkzeug can manage the file and
347 get extra information from the path. Passing a file-like object
348 requires that the file is opened in binary mode, and is mostly
349 useful when building a file in memory with :class:`io.BytesIO`.
351 Never pass file paths provided by a user. The path is assumed to be
352 trusted, so a user could craft a path to access a file you didn't
353 intend. Use :func:`send_from_directory` to safely serve user-provided paths.
355 If the WSGI server sets a ``file_wrapper`` in ``environ``, it is
356 used, otherwise Werkzeug's built-in wrapper is used. Alternatively,
357 if the HTTP server supports ``X-Sendfile``, ``use_x_sendfile=True``
358 will tell the server to send the given path, which is much more
359 efficient than reading it in Python.
361 :param path_or_file: The path to the file to send, relative to the
362 current working directory if a relative path is given.
363 Alternatively, a file-like object opened in binary mode. Make
364 sure the file pointer is seeked to the start of the data.
365 :param environ: The WSGI environ for the current request.
366 :param mimetype: The MIME type to send for the file. If not
367 provided, it will try to detect it from the file name.
368 :param as_attachment: Indicate to a browser that it should offer to
369 save the file instead of displaying it.
370 :param download_name: The default name browsers will use when saving
371 the file. Defaults to the passed file name.
372 :param conditional: Enable conditional and range responses based on
373 request headers. Requires passing a file path and ``environ``.
374 :param etag: Calculate an ETag for the file, which requires passing
375 a file path. Can also be a string to use instead.
376 :param last_modified: The last modified time to send for the file,
377 in seconds. If not provided, it will try to detect it from the
378 file path.
379 :param max_age: How long the client should cache the file, in
380 seconds. If set, ``Cache-Control`` will be ``public``, otherwise
381 it will be ``no-cache`` to prefer conditional caching.
382 :param use_x_sendfile: Set the ``X-Sendfile`` header to let the
383 server to efficiently send the file. Requires support from the
384 HTTP server. Requires passing a file path.
385 :param response_class: Build the response using this class. Defaults
386 to :class:`~werkzeug.wrappers.Response`.
387 :param _root_path: Do not use. For internal use only. Use
388 :func:`send_from_directory` to safely send files under a path.
390 .. versionchanged:: 2.0.2
391 ``send_file`` only sets a detected ``Content-Encoding`` if
392 ``as_attachment`` is disabled.
394 .. versionadded:: 2.0
395 Adapted from Flask's implementation.
397 .. versionchanged:: 2.0
398 ``download_name`` replaces Flask's ``attachment_filename``
399 parameter. If ``as_attachment=False``, it is passed with
400 ``Content-Disposition: inline`` instead.
402 .. versionchanged:: 2.0
403 ``max_age`` replaces Flask's ``cache_timeout`` parameter.
404 ``conditional`` is enabled and ``max_age`` is not set by
405 default.
407 .. versionchanged:: 2.0
408 ``etag`` replaces Flask's ``add_etags`` parameter. It can be a
409 string to use instead of generating one.
411 .. versionchanged:: 2.0
412 If an encoding is returned when guessing ``mimetype`` from
413 ``download_name``, set the ``Content-Encoding`` header.
414 """
415 if response_class is None:
416 from .wrappers import Response
418 response_class = Response
420 path: str | None = None
421 file: t.IO[bytes] | None = None
422 size: int | None = None
423 mtime: float | None = None
424 headers = Headers()
426 if isinstance(path_or_file, (os.PathLike, str)) or hasattr(
427 path_or_file, "__fspath__"
428 ):
429 path_or_file = t.cast("os.PathLike[str] | str", path_or_file)
431 # Flask will pass app.root_path, allowing its send_file wrapper
432 # to not have to deal with paths.
433 if _root_path is not None:
434 path = os.path.join(_root_path, path_or_file)
435 else:
436 path = os.path.abspath(path_or_file)
438 stat = os.stat(path)
439 size = stat.st_size
440 mtime = stat.st_mtime
441 else:
442 file = path_or_file
444 if download_name is None and path is not None:
445 download_name = os.path.basename(path)
447 if mimetype is None:
448 if download_name is None:
449 raise TypeError(
450 "Unable to detect the MIME type because a file name is"
451 " not available. Either set 'download_name', pass a"
452 " path instead of a file, or set 'mimetype'."
453 )
455 mimetype, encoding = mimetypes.guess_type(download_name)
457 if mimetype is None:
458 mimetype = "application/octet-stream"
460 # Don't send encoding for attachments, it causes browsers to
461 # save decompress tar.gz files.
462 if encoding is not None and not as_attachment:
463 headers.set("Content-Encoding", encoding)
465 if download_name is not None:
466 try:
467 download_name.encode("ascii")
468 except UnicodeEncodeError:
469 simple = unicodedata.normalize("NFKD", download_name)
470 simple = simple.encode("ascii", "ignore").decode("ascii")
471 # safe = RFC 5987 attr-char
472 quoted = quote(download_name, safe="!#$&+-.^_`|~")
473 names = {"filename": simple, "filename*": f"UTF-8''{quoted}"}
474 else:
475 names = {"filename": download_name}
477 value = "attachment" if as_attachment else "inline"
478 headers.set("Content-Disposition", value, **names)
479 elif as_attachment:
480 raise TypeError(
481 "No name provided for attachment. Either set"
482 " 'download_name' or pass a path instead of a file."
483 )
485 if use_x_sendfile and path is not None:
486 headers["X-Sendfile"] = path
487 data = None
488 else:
489 if file is None:
490 file = open(path, "rb") # type: ignore
491 elif isinstance(file, io.BytesIO):
492 size = file.getbuffer().nbytes
493 elif isinstance(file, io.TextIOBase):
494 raise ValueError("Files must be opened in binary mode or use BytesIO.")
496 data = wrap_file(environ, file)
498 rv = response_class(
499 data, mimetype=mimetype, headers=headers, direct_passthrough=True
500 )
502 if size is not None:
503 rv.content_length = size
505 if last_modified is not None:
506 rv.last_modified = last_modified # type: ignore
507 elif mtime is not None:
508 rv.last_modified = mtime # type: ignore
510 rv.cache_control.no_cache = True
512 # Flask will pass app.get_send_file_max_age, allowing its send_file
513 # wrapper to not have to deal with paths.
514 if callable(max_age):
515 max_age = max_age(path)
517 if max_age is not None:
518 if max_age > 0:
519 rv.cache_control.no_cache = None
520 rv.cache_control.public = True
522 rv.cache_control.max_age = max_age
523 rv.expires = int(time() + max_age) # type: ignore
525 if isinstance(etag, str):
526 rv.set_etag(etag)
527 elif etag and path is not None:
528 check = adler32(path.encode()) & 0xFFFFFFFF
529 rv.set_etag(f"{mtime}-{size}-{check}")
531 if conditional:
532 try:
533 rv = rv.make_conditional(environ, accept_ranges=True, complete_length=size)
534 except RequestedRangeNotSatisfiable:
535 if file is not None:
536 file.close()
538 raise
540 # Some x-sendfile implementations incorrectly ignore the 304
541 # status code and send the file anyway.
542 if rv.status_code == 304:
543 rv.headers.pop("X-Sendfile", None)
545 return rv
548def send_from_directory(
549 directory: os.PathLike[str] | str,
550 path: os.PathLike[str] | str,
551 environ: WSGIEnvironment,
552 **kwargs: t.Any,
553) -> Response:
554 """Send a file from within a directory using :func:`send_file`.
556 This is a secure way to serve files from a folder, such as static
557 files or uploads. Uses :func:`~werkzeug.security.safe_join` to
558 ensure the path coming from the client is not maliciously crafted to
559 point outside the specified directory.
561 If the final path does not point to an existing regular file,
562 returns a 404 :exc:`~werkzeug.exceptions.NotFound` error.
564 :param directory: The directory that ``path`` must be located under. This *must not*
565 be a value provided by the client, otherwise it becomes insecure.
566 :param path: The path to the file to send, relative to ``directory``. This is the
567 part of the path provided by the client, which is checked for security.
568 :param environ: The WSGI environ for the current request.
569 :param kwargs: Arguments to pass to :func:`send_file`.
571 .. versionadded:: 2.0
572 Adapted from Flask's implementation.
573 """
574 path_str = safe_join(os.fspath(directory), os.fspath(path))
576 if path_str is None:
577 raise NotFound()
579 # Flask will pass app.root_path, allowing its send_from_directory
580 # wrapper to not have to deal with paths.
581 if "_root_path" in kwargs:
582 path_str = os.path.join(kwargs["_root_path"], path_str)
584 if not os.path.isfile(path_str):
585 raise NotFound()
587 return send_file(path_str, environ, **kwargs)
590def import_string(import_name: str, silent: bool = False) -> t.Any:
591 """Imports an object based on a string. This is useful if you want to
592 use import paths as endpoints or something similar. An import path can
593 be specified either in dotted notation (``xml.sax.saxutils.escape``)
594 or with a colon as object delimiter (``xml.sax.saxutils:escape``).
596 If `silent` is True the return value will be `None` if the import fails.
598 :param import_name: the dotted name for the object to import.
599 :param silent: if set to `True` import errors are ignored and
600 `None` is returned instead.
601 :return: imported object
602 """
603 import_name = import_name.replace(":", ".")
604 try:
605 try:
606 __import__(import_name)
607 except ImportError:
608 if "." not in import_name:
609 raise
610 else:
611 return sys.modules[import_name]
613 module_name, obj_name = import_name.rsplit(".", 1)
614 module = __import__(module_name, globals(), locals(), [obj_name])
615 try:
616 return getattr(module, obj_name)
617 except AttributeError as e:
618 raise ImportError(e) from None
620 except ImportError as e:
621 if not silent:
622 raise ImportStringError(import_name, e).with_traceback(
623 sys.exc_info()[2]
624 ) from None
626 return None
629def find_modules(
630 import_path: str, include_packages: bool = False, recursive: bool = False
631) -> t.Iterator[str]:
632 """Finds all the modules below a package. This can be useful to
633 automatically import all views / controllers so that their metaclasses /
634 function decorators have a chance to register themselves on the
635 application.
637 Packages are not returned unless `include_packages` is `True`. This can
638 also recursively list modules but in that case it will import all the
639 packages to get the correct load path of that module.
641 :param import_path: the dotted name for the package to find child modules.
642 :param include_packages: set to `True` if packages should be returned, too.
643 :param recursive: set to `True` if recursion should happen.
644 :return: generator
645 """
646 module = import_string(import_path)
647 path = getattr(module, "__path__", None)
648 if path is None:
649 raise ValueError(f"{import_path!r} is not a package")
650 basename = f"{module.__name__}."
651 for _importer, modname, ispkg in pkgutil.iter_modules(path):
652 modname = basename + modname
653 if ispkg:
654 if include_packages:
655 yield modname
656 if recursive:
657 yield from find_modules(modname, include_packages, True)
658 else:
659 yield modname
662class ImportStringError(ImportError):
663 """Provides information about a failed :func:`import_string` attempt."""
665 #: String in dotted notation that failed to be imported.
666 import_name: str
667 #: Wrapped exception.
668 exception: BaseException
670 def __init__(self, import_name: str, exception: BaseException) -> None:
671 self.import_name = import_name
672 self.exception = exception
673 msg = import_name
674 name = ""
675 tracked = []
676 for part in import_name.replace(":", ".").split("."):
677 name = f"{name}.{part}" if name else part
678 imported = import_string(name, silent=True)
679 if imported:
680 tracked.append((name, getattr(imported, "__file__", None)))
681 else:
682 track = [f"- {n!r} found in {i!r}." for n, i in tracked]
683 track.append(f"- {name!r} not found.")
684 track_str = "\n".join(track)
685 msg = (
686 f"import_string() failed for {import_name!r}. Possible reasons"
687 f" are:\n\n"
688 "- missing __init__.py in a package;\n"
689 "- package or module path not included in sys.path;\n"
690 "- duplicated package or module name taking precedence in"
691 " sys.path;\n"
692 "- missing module, class, function or variable;\n\n"
693 f"Debugged import:\n\n{track_str}\n\n"
694 f"Original exception:\n\n{type(exception).__name__}: {exception}"
695 )
696 break
698 super().__init__(msg)
700 def __repr__(self) -> str:
701 return f"<{type(self).__name__}({self.import_name!r}, {self.exception!r})>"