move send_file and send_from_directory to Werkzeug

The implementations were moved to Werkzeug, Flask's functions become
wrappers around Werkzeug to pass some Flask-specific values.

cache_timeout is renamed to max_age. SEND_FILE_MAX_AGE_DEFAULT,
app.send_file_max_age_default, and app.get_send_file_max_age defaults
to None. This tells the browser to use conditional requests rather than
a 12 hour cache.

attachment_filename is renamed to download_name, and is always sent if
a name is known.

Deprecate helpers.safe_join in favor of werkzeug.utils.safe_join.

Removed most of the send_file tests, they're tested in Werkzeug.

In the file upload example, renamed the uploaded_file view to
download_file to avoid a common source of confusion.
This commit is contained in:
David Lord 2020-11-05 09:00:57 -08:00
parent 15a49e7297
commit dc11cdb4a4
No known key found for this signature in database
GPG Key ID: 7A1C87E3F5BC42A8
8 changed files with 260 additions and 654 deletions

View File

@ -45,6 +45,20 @@ Unreleased
- Include ``samesite`` and ``secure`` options when removing the
session cookie. :pr:`3726`
- Support passing a ``pathlib.Path`` to ``static_folder``. :pr:`3579`
- ``send_file`` and ``send_from_directory`` are wrappers around the
implementations in ``werkzeug.utils``. :pr:`3828`
- Some ``send_file`` parameters have been renamed, the old names are
deprecated. ``attachment_filename`` is renamed to ``download_name``.
``cache_timeout`` is renamed to ``max_age``. :pr:`3828`
- ``send_file`` passes ``download_name`` even if
``as_attachment=False`` by using ``Content-Disposition: inline``.
:pr:`3828`
- ``send_file`` sets ``conditional=True`` and ``max_age=None`` by
default. ``Cache-Control`` is set to ``no-cache`` if ``max_age`` is
not set, otherwise ``public``. This tells browsers to validate
conditional requests instead of using a timed cache. :pr:`3828`
- ``helpers.safe_join`` is deprecated. Use
``werkzeug.utils.safe_join`` instead. :pr:`3828`
Version 1.1.2

View File

@ -265,11 +265,16 @@ The following configuration values are used internally by Flask:
.. py:data:: SEND_FILE_MAX_AGE_DEFAULT
When serving files, set the cache control max age to this number of
seconds. Can either be a :class:`datetime.timedelta` or an ``int``.
seconds. Can be a :class:`datetime.timedelta` or an ``int``.
Override this value on a per-file basis using
:meth:`~flask.Flask.get_send_file_max_age` on the application or blueprint.
:meth:`~flask.Flask.get_send_file_max_age` on the application or
blueprint.
Default: ``timedelta(hours=12)`` (``43200`` seconds)
If ``None``, ``send_file`` tells the browser to use conditional
requests will be used instead of a timed cache, which is usually
preferable.
Default: ``None``
.. py:data:: SERVER_NAME

View File

@ -63,8 +63,7 @@ the file and redirects the user to the URL for the uploaded file::
if file and allowed_file(file.filename):
filename = secure_filename(file.filename)
file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
return redirect(url_for('uploaded_file',
filename=filename))
return redirect(url_for('download_file', name=filename))
return '''
<!doctype html>
<title>Upload new File</title>
@ -102,31 +101,28 @@ before storing it directly on the filesystem.
>>> secure_filename('../../../../home/username/.bashrc')
'home_username_.bashrc'
Now one last thing is missing: the serving of the uploaded files. In the
:func:`upload_file()` we redirect the user to
``url_for('uploaded_file', filename=filename)``, that is, ``/uploads/filename``.
So we write the :func:`uploaded_file` function to return the file of that name. As
of Flask 0.5 we can use a function that does that for us::
We want to be able to serve the uploaded files so they can be downloaded
by users. We'll define a ``download_file`` view to serve files in the
upload folder by name. ``url_for("download_file", name=name)`` generates
download URLs.
.. code-block:: python
from flask import send_from_directory
@app.route('/uploads/<filename>')
def uploaded_file(filename):
return send_from_directory(app.config['UPLOAD_FOLDER'],
filename)
@app.route('/uploads/<name>')
def download_file(name):
return send_from_directory(app.config["UPLOAD_FOLDER"], name)
Alternatively you can register `uploaded_file` as `build_only` rule and
use the :class:`~werkzeug.wsgi.SharedDataMiddleware`. This also works with
older versions of Flask::
If you're using middleware or the HTTP server to serve files, you can
register the ``download_file`` endpoint as ``build_only`` so ``url_for``
will work without a view function.
from werkzeug.middleware.shared_data import SharedDataMiddleware
app.add_url_rule('/uploads/<filename>', 'uploaded_file',
build_only=True)
app.wsgi_app = SharedDataMiddleware(app.wsgi_app, {
'/uploads': app.config['UPLOAD_FOLDER']
})
.. code-block:: python
If you now run the application everything should work as expected.
app.add_url_rule(
"/uploads/<name>", endpoint="download_file", build_only=True
)
Improving Uploads

View File

@ -55,9 +55,10 @@ from .wrappers import Response
def _make_timedelta(value):
if not isinstance(value, timedelta):
return timedelta(seconds=value)
return value
if value is None or isinstance(value, timedelta):
return value
return timedelta(seconds=value)
class Flask(Scaffold):
@ -234,13 +235,16 @@ class Flask(Scaffold):
"PERMANENT_SESSION_LIFETIME", get_converter=_make_timedelta
)
#: A :class:`~datetime.timedelta` which is used as default cache_timeout
#: for the :func:`send_file` functions. The default is 12 hours.
#: A :class:`~datetime.timedelta` or number of seconds which is used
#: as the default ``max_age`` for :func:`send_file`. The default is
#: ``None``, which tells the browser to use conditional requests
#: instead of a timed cache.
#:
#: This attribute can also be configured from the config with the
#: ``SEND_FILE_MAX_AGE_DEFAULT`` configuration key. This configuration
#: variable can also be set with an integer value used as seconds.
#: Defaults to ``timedelta(hours=12)``
#: Configured with the :data:`SEND_FILE_MAX_AGE_DEFAULT`
#: configuration key.
#:
#: .. versionchanged:: 2.0
#: Defaults to ``None`` instead of 12 hours.
send_file_max_age_default = ConfigAttribute(
"SEND_FILE_MAX_AGE_DEFAULT", get_converter=_make_timedelta
)
@ -297,7 +301,7 @@ class Flask(Scaffold):
"SESSION_COOKIE_SAMESITE": None,
"SESSION_REFRESH_EACH_REQUEST": True,
"MAX_CONTENT_LENGTH": None,
"SEND_FILE_MAX_AGE_DEFAULT": timedelta(hours=12),
"SEND_FILE_MAX_AGE_DEFAULT": None,
"TRAP_BAD_REQUEST_ERRORS": None,
"TRAP_HTTP_EXCEPTIONS": False,
"EXPLAIN_TEMPLATE_LOADING": False,

View File

@ -1,24 +1,16 @@
import io
import mimetypes
import os
import pkgutil
import posixpath
import socket
import sys
import unicodedata
import warnings
from functools import update_wrapper
from threading import RLock
from time import time
from zlib import adler32
import werkzeug.utils
from jinja2 import FileSystemLoader
from werkzeug.datastructures import Headers
from werkzeug.exceptions import BadRequest
from werkzeug.exceptions import NotFound
from werkzeug.exceptions import RequestedRangeNotSatisfiable
from werkzeug.routing import BuildError
from werkzeug.urls import url_quote
from werkzeug.wsgi import wrap_file
from .globals import _app_ctx_stack
from .globals import _request_ctx_stack
@ -444,49 +436,116 @@ def get_flashed_messages(with_categories=False, category_filter=()):
return flashes
def _prepare_send_file_kwargs(
download_name=None,
attachment_filename=None,
max_age=None,
cache_timeout=None,
**kwargs,
):
if attachment_filename is not None:
warnings.warn(
"The 'attachment_filename' parameter has been renamed to 'download_name'."
" The old name will be removed in Flask 2.1.",
DeprecationWarning,
stacklevel=3,
)
download_name = attachment_filename
if cache_timeout is not None:
warnings.warn(
"The 'cache_timeout' parameter has been renamed to 'max_age'. The old name"
" will be removed in Flask 2.1.",
DeprecationWarning,
stacklevel=3,
)
max_age = cache_timeout
if max_age is None:
max_age = current_app.get_send_file_max_age
kwargs.update(
environ=request.environ,
download_name=download_name,
max_age=max_age,
use_x_sendfile=current_app.use_x_sendfile,
response_class=current_app.response_class,
_root_path=current_app.root_path,
)
return kwargs
def send_file(
filename_or_fp,
path_or_file,
mimetype=None,
as_attachment=False,
download_name=None,
attachment_filename=None,
conditional=True,
add_etags=True,
cache_timeout=None,
conditional=False,
last_modified=None,
max_age=None,
cache_timeout=None,
):
"""Sends the contents of a file to the client. This will use the
most efficient method available and configured. By default it will
try to use the WSGI server's file_wrapper support. Alternatively
you can set the application's :attr:`~Flask.use_x_sendfile` attribute
to ``True`` to directly emit an ``X-Sendfile`` header. This however
requires support of the underlying webserver for ``X-Sendfile``.
"""Send the contents of a file to the client.
By default it will try to guess the mimetype for you, but you can
also explicitly provide one. For extra security you probably want
to send certain files as attachment (HTML for instance). The mimetype
guessing requires a `filename` or an `attachment_filename` to be
provided.
The first argument can be a file path or a file-like object. Paths
are preferred in most cases because Werkzeug can manage the file and
get extra information from the path. Passing a file-like object
requires that the file is opened in binary mode, and is mostly
useful when building a file in memory with :class:`io.BytesIO`.
When passing a file-like object instead of a filename, only binary
mode is supported (``open(filename, "rb")``, :class:`~io.BytesIO`,
etc.). Text mode files and :class:`~io.StringIO` will raise a
:exc:`ValueError`.
Never pass file paths provided by a user. The path is assumed to be
trusted, so a user could craft a path to access a file you didn't
intend. Use :func:`send_from_directory` to safely serve
user-requested paths from within a directory.
ETags will also be attached automatically if a `filename` is provided. You
can turn this off by setting `add_etags=False`.
If the WSGI server sets a ``file_wrapper`` in ``environ``, it is
used, otherwise Werkzeug's built-in wrapper is used. Alternatively,
if the HTTP server supports ``X-Sendfile``, configuring Flask with
``USE_X_SENDFILE = True`` will tell the server to send the given
path, which is much more efficient than reading it in Python.
If `conditional=True` and `filename` is provided, this method will try to
upgrade the response stream to support range requests. This will allow
the request to be answered with partial content response.
:param path_or_file: The path to the file to send, relative to the
current working directory if a relative path is given.
Alternatively, a file-like object opened in binary mode. Make
sure the file pointer is seeked to the start of the data.
:param mimetype: The MIME type to send for the file. If not
provided, it will try to detect it from the file name.
:param as_attachment: Indicate to a browser that it should offer to
save the file instead of displaying it.
:param download_name: The default name browsers will use when saving
the file. Defaults to the passed file name.
:param conditional: Enable conditional and range responses based on
request headers. Requires passing a file path and ``environ``.
:param add_etags: Calculate an ETag for the file. Requires passing a
file path.
:param last_modified: The last modified time to send for the file,
in seconds. If not provided, it will try to detect it from the
file path.
:param max_age: How long the client should cache the file, in
seconds. If set, ``Cache-Control`` will be ``public``, otherwise
it will be ``no-cache`` to prefer conditional caching.
Please never pass filenames to this function from user sources;
you should use :func:`send_from_directory` instead.
.. versionchanged:: 2.0
``download_name`` replaces the ``attachment_filename``
parameter. If ``as_attachment=False``, it is passed with
``Content-Disposition: inline`` instead.
.. versionchanged:: 2.0
``max_age`` replaces the ``cache_timeout`` parameter.
``conditional`` is enabled and ``max_age`` is not set by
default.
.. versionchanged:: 2.0
Passing a file-like object that inherits from
:class:`~io.TextIOBase` will raise a :exc:`ValueError` rather
than sending an empty file.
.. versionadded:: 2.0
Moved the implementation to Werkzeug. This is now a wrapper to
pass some Flask-specific arguments.
.. versionchanged:: 1.1
``filename`` may be a :class:`~os.PathLike` object.
@ -498,260 +557,106 @@ def send_file(
compatibility with WSGI servers.
.. versionchanged:: 1.0
UTF-8 filenames, as specified in `RFC 2231`_, are supported.
.. _RFC 2231: https://tools.ietf.org/html/rfc2231#section-4
UTF-8 filenames as specified in :rfc:`2231` are supported.
.. versionchanged:: 0.12
The filename is no longer automatically inferred from file
objects. If you want to use automatic MIME and etag support, pass
a filename via ``filename_or_fp`` or ``attachment_filename``.
The filename is no longer automatically inferred from file
objects. If you want to use automatic MIME and etag support,
pass a filename via ``filename_or_fp`` or
``attachment_filename``.
.. versionchanged:: 0.12
``attachment_filename`` is preferred over ``filename`` for MIME
detection.
``attachment_filename`` is preferred over ``filename`` for MIME
detection.
.. versionchanged:: 0.9
``cache_timeout`` defaults to
:meth:`Flask.get_send_file_max_age`.
``cache_timeout`` defaults to
:meth:`Flask.get_send_file_max_age`.
.. versionchanged:: 0.7
MIME guessing and etag support for file-like objects was
deprecated because it was unreliable. Pass a filename if you are
able to, otherwise attach an etag yourself. This functionality
will be removed in Flask 1.0.
MIME guessing and etag support for file-like objects was
deprecated because it was unreliable. Pass a filename if you are
able to, otherwise attach an etag yourself.
.. versionadded:: 0.5
The ``add_etags``, ``cache_timeout`` and ``conditional``
parameters were added. The default behavior is to add etags.
.. versionchanged:: 0.5
The ``add_etags``, ``cache_timeout`` and ``conditional``
parameters were added. The default behavior is to add etags.
.. versionadded:: 0.2
:param filename_or_fp: The filename of the file to send, relative to
:attr:`~Flask.root_path` if a relative path is specified.
Alternatively, a file-like object opened in binary mode. Make
sure the file pointer is seeked to the start of the data.
``X-Sendfile`` will only be used with filenames.
:param mimetype: the mimetype of the file if provided. If a file path is
given, auto detection happens as fallback, otherwise an
error will be raised.
:param as_attachment: set to ``True`` if you want to send this file with
a ``Content-Disposition: attachment`` header.
:param attachment_filename: the filename for the attachment if it
differs from the file's filename.
:param add_etags: set to ``False`` to disable attaching of etags.
:param conditional: set to ``True`` to enable conditional responses.
:param cache_timeout: the timeout in seconds for the headers. When ``None``
(default), this value is set by
:meth:`~Flask.get_send_file_max_age` of
:data:`~flask.current_app`.
:param last_modified: set the ``Last-Modified`` header to this value,
a :class:`~datetime.datetime` or timestamp.
If a file was passed, this overrides its mtime.
"""
mtime = None
fsize = None
if hasattr(filename_or_fp, "__fspath__"):
filename_or_fp = os.fspath(filename_or_fp)
if isinstance(filename_or_fp, str):
filename = filename_or_fp
if not os.path.isabs(filename):
filename = os.path.join(current_app.root_path, filename)
file = None
if attachment_filename is None:
attachment_filename = os.path.basename(filename)
else:
file = filename_or_fp
filename = None
if mimetype is None:
if attachment_filename is not None:
mimetype = (
mimetypes.guess_type(attachment_filename)[0]
or "application/octet-stream"
)
if mimetype is None:
raise ValueError(
"Unable to infer MIME-type because no filename is available. "
"Please set either `attachment_filename`, pass a filepath to "
"`filename_or_fp` or set your own MIME-type via `mimetype`."
)
headers = Headers()
if as_attachment:
if attachment_filename is None:
raise TypeError("filename unavailable, required for sending as attachment")
if not isinstance(attachment_filename, str):
attachment_filename = attachment_filename.decode("utf-8")
try:
attachment_filename = attachment_filename.encode("ascii")
except UnicodeEncodeError:
quoted = url_quote(attachment_filename, safe="")
filenames = {
"filename": unicodedata.normalize("NFKD", attachment_filename).encode(
"ascii", "ignore"
),
"filename*": f"UTF-8''{quoted}",
}
else:
filenames = {"filename": attachment_filename}
headers.add("Content-Disposition", "attachment", **filenames)
if current_app.use_x_sendfile and filename:
if file is not None:
file.close()
headers["X-Sendfile"] = filename
fsize = os.path.getsize(filename)
data = None
else:
if file is None:
file = open(filename, "rb")
mtime = os.path.getmtime(filename)
fsize = os.path.getsize(filename)
elif isinstance(file, io.BytesIO):
fsize = file.getbuffer().nbytes
elif isinstance(file, io.TextIOBase):
raise ValueError("Files must be opened in binary mode or use BytesIO.")
data = wrap_file(request.environ, file)
if fsize is not None:
headers["Content-Length"] = fsize
rv = current_app.response_class(
data, mimetype=mimetype, headers=headers, direct_passthrough=True
return werkzeug.utils.send_file(
**_prepare_send_file_kwargs(
path_or_file=path_or_file,
environ=request.environ,
mimetype=mimetype,
as_attachment=as_attachment,
download_name=download_name,
attachment_filename=attachment_filename,
conditional=conditional,
add_etags=add_etags,
last_modified=last_modified,
max_age=max_age,
cache_timeout=cache_timeout,
)
)
if last_modified is not None:
rv.last_modified = last_modified
elif mtime is not None:
rv.last_modified = mtime
rv.cache_control.public = True
if cache_timeout is None:
cache_timeout = current_app.get_send_file_max_age(filename)
if cache_timeout is not None:
rv.cache_control.max_age = cache_timeout
rv.expires = int(time() + cache_timeout)
if add_etags and filename is not None:
from warnings import warn
try:
check = (
adler32(
filename.encode("utf-8") if isinstance(filename, str) else filename
)
& 0xFFFFFFFF
)
rv.set_etag(
f"{os.path.getmtime(filename)}-{os.path.getsize(filename)}-{check}"
)
except OSError:
warn(
f"Access {filename} failed, maybe it does not exist, so"
" ignore etags in headers",
stacklevel=2,
)
if conditional:
try:
rv = rv.make_conditional(request, accept_ranges=True, complete_length=fsize)
except RequestedRangeNotSatisfiable:
if file is not None:
file.close()
raise
# make sure we don't send x-sendfile for servers that
# ignore the 304 status code for x-sendfile.
if rv.status_code == 304:
rv.headers.pop("x-sendfile", None)
return rv
def safe_join(directory, *pathnames):
"""Safely join `directory` and zero or more untrusted `pathnames`
components.
"""Safely join zero or more untrusted path components to a base
directory to avoid escaping the base directory.
Example usage::
@app.route('/wiki/<path:filename>')
def wiki_page(filename):
filename = safe_join(app.config['WIKI_FOLDER'], filename)
with open(filename, 'rb') as fd:
content = fd.read() # Read and process the file content...
:param directory: the trusted base directory.
:param pathnames: the untrusted pathnames relative to that directory.
:raises: :class:`~werkzeug.exceptions.NotFound` if one or more passed
paths fall out of its boundaries.
:param directory: The trusted base directory.
:param pathnames: The untrusted path components relative to the
base directory.
:return: A safe path, otherwise ``None``.
"""
warnings.warn(
"'flask.helpers.safe_join' is deprecated and will be removed in"
" 2.1. Use 'werkzeug.utils.safe_join' instead.",
DeprecationWarning,
stacklevel=2,
)
path = werkzeug.utils.safe_join(directory, *pathnames)
parts = [directory]
if path is None:
raise NotFound()
for filename in pathnames:
if filename != "":
filename = posixpath.normpath(filename)
if (
any(sep in filename for sep in _os_alt_seps)
or os.path.isabs(filename)
or filename == ".."
or filename.startswith("../")
):
raise NotFound()
parts.append(filename)
return posixpath.join(*parts)
return path
def send_from_directory(directory, filename, **options):
"""Send a file from a given directory with :func:`send_file`. This
is a secure way to quickly expose static files from an upload folder
or something similar.
def send_from_directory(directory, path, **kwargs):
"""Send a file from within a directory using :func:`send_file`.
Example usage::
.. code-block:: python
@app.route('/uploads/<path:filename>')
def download_file(filename):
return send_from_directory(app.config['UPLOAD_FOLDER'],
filename, as_attachment=True)
@app.route("/uploads/<path:name>")
def download_file(name):
return send_from_directory(
app.config['UPLOAD_FOLDER'], name, as_attachment=True
)
.. admonition:: Sending files and Performance
This is a secure way to serve files from a folder, such as static
files or uploads. Uses :func:`~werkzeug.security.safe_join` to
ensure the path coming from the client is not maliciously crafted to
point outside the specified directory.
It is strongly recommended to activate either ``X-Sendfile`` support in
your webserver or (if no authentication happens) to tell the webserver
to serve files for the given path on its own without calling into the
web application for improved performance.
If the final path does not point to an existing regular file,
raises a 404 :exc:`~werkzeug.exceptions.NotFound` error.
:param directory: The directory that ``path`` must be located under.
:param path: The path to the file to send, relative to
``directory``.
:param kwargs: Arguments to pass to :func:`send_file`.
.. versionadded:: 2.0
Moved the implementation to Werkzeug. This is now a wrapper to
pass some Flask-specific arguments.
.. versionadded:: 0.5
:param directory: the directory where all the files are stored.
:param filename: the filename relative to that directory to
download.
:param options: optional keyword arguments that are directly
forwarded to :func:`send_file`.
"""
filename = os.fspath(filename)
directory = os.fspath(directory)
filename = safe_join(directory, filename)
if not os.path.isabs(filename):
filename = os.path.join(current_app.root_path, filename)
try:
if not os.path.isfile(filename):
raise NotFound()
except (TypeError, ValueError):
raise BadRequest()
options.setdefault("conditional", True)
return send_file(filename, **options)
return werkzeug.utils.send_from_directory(
directory, path, **_prepare_send_file_kwargs(**kwargs)
)
def get_root_path(import_name):
@ -1016,30 +921,25 @@ class _PackageBoundObject:
return FileSystemLoader(os.path.join(self.root_path, self.template_folder))
def get_send_file_max_age(self, filename):
"""Provides default cache_timeout for the :func:`send_file` functions.
"""Used by :func:`send_file` to determine the ``max_age`` cache
value for a given file path if it wasn't passed.
By default, this function returns ``SEND_FILE_MAX_AGE_DEFAULT`` from
the configuration of :data:`~flask.current_app`.
By default, this returns :data:`SEND_FILE_MAX_AGE_DEFAULT` from
the configuration of :data:`~flask.current_app`. This defaults
to ``None``, which tells the browser to use conditional requests
instead of a timed cache, which is usually preferable.
Static file functions such as :func:`send_from_directory` use this
function, and :func:`send_file` calls this function on
:data:`~flask.current_app` when the given cache_timeout is ``None``. If a
cache_timeout is given in :func:`send_file`, that timeout is used;
otherwise, this method is called.
This allows subclasses to change the behavior when sending files based
on the filename. For example, to set the cache timeout for .js files
to 60 seconds::
class MyFlask(flask.Flask):
def get_send_file_max_age(self, name):
if name.lower().endswith('.js'):
return 60
return flask.Flask.get_send_file_max_age(self, name)
.. versionchanged:: 2.0
The default configuration is ``None`` instead of 12 hours.
.. versionadded:: 0.9
"""
return total_seconds(current_app.send_file_max_age_default)
value = current_app.send_file_max_age_default
if value is None:
return None
return total_seconds(value)
def send_static_file(self, filename):
"""Function used internally to send static files from the static
@ -1049,12 +949,11 @@ class _PackageBoundObject:
"""
if not self.has_static_folder:
raise RuntimeError("No static folder for this object")
# Ensure get_send_file_max_age is called in all cases.
# Here, we ensure get_send_file_max_age is called for Blueprints.
cache_timeout = self.get_send_file_max_age(filename)
return send_from_directory(
self.static_folder, filename, cache_timeout=cache_timeout
)
# send_file only knows to call get_send_file_max_age on the app,
# call it here so it works for blueprints too.
max_age = self.get_send_file_max_age(filename)
return send_from_directory(self.static_folder, filename, max_age=max_age)
def open_resource(self, resource, mode="rb"):
"""Opens a resource from the application's resource folder. To see

View File

@ -222,7 +222,7 @@ def test_templates_and_static(test_apps):
assert flask.render_template("nested/nested.txt") == "I'm nested"
def test_default_static_cache_timeout(app):
def test_default_static_max_age(app):
class MyBlueprint(flask.Blueprint):
def get_send_file_max_age(self, filename):
return 100

View File

@ -1,15 +1,7 @@
import datetime
import io
import os
import sys
import pytest
from werkzeug.datastructures import Range
from werkzeug.exceptions import BadRequest
from werkzeug.exceptions import NotFound
from werkzeug.http import http_date
from werkzeug.http import parse_cache_control_header
from werkzeug.http import parse_options_header
import flask
from flask.helpers import get_debug_flag
@ -39,278 +31,45 @@ class PyBytesIO:
class TestSendfile:
def test_send_file_regular(self, app, req_ctx):
def test_send_file(self, app, req_ctx):
rv = flask.send_file("static/index.html")
assert rv.direct_passthrough
assert rv.mimetype == "text/html"
with app.open_resource("static/index.html") as f:
rv.direct_passthrough = False
assert rv.data == f.read()
rv.close()
def test_send_file_xsendfile(self, app, req_ctx):
app.use_x_sendfile = True
rv = flask.send_file("static/index.html")
assert rv.direct_passthrough
assert "x-sendfile" in rv.headers
assert rv.headers["x-sendfile"] == os.path.join(
app.root_path, "static/index.html"
)
assert rv.mimetype == "text/html"
rv.close()
def test_send_file_last_modified(self, app, client):
last_modified = datetime.datetime(1999, 1, 1)
@app.route("/")
def index():
return flask.send_file(
io.BytesIO(b"party like it's"),
last_modified=last_modified,
mimetype="text/plain",
)
rv = client.get("/")
assert rv.last_modified == last_modified
def test_send_file_object_without_mimetype(self, app, req_ctx):
with pytest.raises(ValueError) as excinfo:
flask.send_file(io.BytesIO(b"LOL"))
assert "Unable to infer MIME-type" in str(excinfo.value)
assert "no filename is available" in str(excinfo.value)
flask.send_file(io.BytesIO(b"LOL"), attachment_filename="filename")
@pytest.mark.parametrize(
"opener",
[
lambda app: open(os.path.join(app.static_folder, "index.html"), "rb"),
lambda app: io.BytesIO(b"Test"),
lambda app: PyBytesIO(b"Test"),
],
)
@pytest.mark.usefixtures("req_ctx")
def test_send_file_object(self, app, opener):
file = opener(app)
app.use_x_sendfile = True
rv = flask.send_file(file, mimetype="text/plain")
rv.direct_passthrough = False
assert rv.data
assert rv.mimetype == "text/plain"
assert "x-sendfile" not in rv.headers
rv.close()
@pytest.mark.parametrize(
"opener",
[
lambda app: io.StringIO("Test"),
lambda app: open(os.path.join(app.static_folder, "index.html")),
],
)
@pytest.mark.usefixtures("req_ctx")
def test_send_file_text_fails(self, app, opener):
file = opener(app)
with pytest.raises(ValueError):
flask.send_file(file, mimetype="text/plain")
file.close()
def test_send_file_pathlike(self, app, req_ctx):
rv = flask.send_file(FakePath("static/index.html"))
assert rv.direct_passthrough
assert rv.mimetype == "text/html"
with app.open_resource("static/index.html") as f:
rv.direct_passthrough = False
assert rv.data == f.read()
rv.close()
@pytest.mark.skipif(
not callable(getattr(Range, "to_content_range_header", None)),
reason="not implemented within werkzeug",
)
def test_send_file_range_request(self, app, client):
@app.route("/")
def index():
return flask.send_file("static/index.html", conditional=True)
rv = client.get("/", headers={"Range": "bytes=4-15"})
assert rv.status_code == 206
with app.open_resource("static/index.html") as f:
assert rv.data == f.read()[4:16]
rv.close()
rv = client.get("/", headers={"Range": "bytes=4-"})
assert rv.status_code == 206
with app.open_resource("static/index.html") as f:
assert rv.data == f.read()[4:]
rv.close()
rv = client.get("/", headers={"Range": "bytes=4-1000"})
assert rv.status_code == 206
with app.open_resource("static/index.html") as f:
assert rv.data == f.read()[4:]
rv.close()
rv = client.get("/", headers={"Range": "bytes=-10"})
assert rv.status_code == 206
with app.open_resource("static/index.html") as f:
assert rv.data == f.read()[-10:]
rv.close()
rv = client.get("/", headers={"Range": "bytes=1000-"})
assert rv.status_code == 416
rv.close()
rv = client.get("/", headers={"Range": "bytes=-"})
assert rv.status_code == 416
rv.close()
rv = client.get("/", headers={"Range": "somethingsomething"})
assert rv.status_code == 416
rv.close()
last_modified = datetime.datetime.utcfromtimestamp(
os.path.getmtime(os.path.join(app.root_path, "static/index.html"))
).replace(microsecond=0)
rv = client.get(
"/", headers={"Range": "bytes=4-15", "If-Range": http_date(last_modified)}
)
assert rv.status_code == 206
rv.close()
rv = client.get(
"/",
headers={
"Range": "bytes=4-15",
"If-Range": http_date(datetime.datetime(1999, 1, 1)),
},
)
assert rv.status_code == 200
rv.close()
def test_send_file_range_request_bytesio(self, app, client):
@app.route("/")
def index():
file = io.BytesIO(b"somethingsomething")
return flask.send_file(
file, attachment_filename="filename", conditional=True
)
rv = client.get("/", headers={"Range": "bytes=4-15"})
assert rv.status_code == 206
assert rv.data == b"somethingsomething"[4:16]
rv.close()
def test_send_file_range_request_xsendfile_invalid(self, app, client):
# https://github.com/pallets/flask/issues/2526
app.use_x_sendfile = True
@app.route("/")
def index():
return flask.send_file("static/index.html", conditional=True)
rv = client.get("/", headers={"Range": "bytes=1000-"})
assert rv.status_code == 416
rv.close()
def test_attachment(self, app, req_ctx):
app = flask.Flask(__name__)
with app.test_request_context():
with open(os.path.join(app.root_path, "static/index.html"), "rb") as f:
rv = flask.send_file(
f, as_attachment=True, attachment_filename="index.html"
)
value, options = parse_options_header(rv.headers["Content-Disposition"])
assert value == "attachment"
rv.close()
with open(os.path.join(app.root_path, "static/index.html"), "rb") as f:
rv = flask.send_file(
f, as_attachment=True, attachment_filename="index.html"
)
value, options = parse_options_header(rv.headers["Content-Disposition"])
assert value == "attachment"
assert options["filename"] == "index.html"
assert "filename*" not in rv.headers["Content-Disposition"]
rv.close()
rv = flask.send_file("static/index.html", as_attachment=True)
value, options = parse_options_header(rv.headers["Content-Disposition"])
assert value == "attachment"
assert options["filename"] == "index.html"
rv.close()
rv = flask.send_file(
io.BytesIO(b"Test"),
as_attachment=True,
attachment_filename="index.txt",
add_etags=False,
)
assert rv.mimetype == "text/plain"
value, options = parse_options_header(rv.headers["Content-Disposition"])
assert value == "attachment"
assert options["filename"] == "index.txt"
rv.close()
@pytest.mark.usefixtures("req_ctx")
@pytest.mark.parametrize(
("filename", "ascii", "utf8"),
(
("index.html", "index.html", False),
(
"Ñandúpingüino.txt",
'"Nandu/pinguino.txt"',
"%C3%91and%C3%BA%EF%BC%8Fping%C3%BCino.txt",
),
("Vögel.txt", "Vogel.txt", "V%C3%B6gel.txt"),
# ":/" are not safe in filename* value
("те:/ст", '":/"', "%D1%82%D0%B5%3A%2F%D1%81%D1%82"),
),
)
def test_attachment_filename_encoding(self, filename, ascii, utf8):
rv = flask.send_file(
"static/index.html", as_attachment=True, attachment_filename=filename
)
rv.close()
content_disposition = rv.headers["Content-Disposition"]
assert f"filename={ascii}" in content_disposition
if utf8:
assert f"filename*=UTF-8''{utf8}" in content_disposition
else:
assert "filename*=UTF-8''" not in content_disposition
def test_static_file(self, app, req_ctx):
# default cache timeout is 12 hours
# Default max_age is None.
# Test with static file handler.
rv = app.send_static_file("index.html")
cc = parse_cache_control_header(rv.headers["Cache-Control"])
assert cc.max_age == 12 * 60 * 60
assert rv.cache_control.max_age is None
rv.close()
# Test again with direct use of send_file utility.
# Test with direct use of send_file.
rv = flask.send_file("static/index.html")
cc = parse_cache_control_header(rv.headers["Cache-Control"])
assert cc.max_age == 12 * 60 * 60
assert rv.cache_control.max_age is None
rv.close()
app.config["SEND_FILE_MAX_AGE_DEFAULT"] = 3600
# Test with static file handler.
rv = app.send_static_file("index.html")
cc = parse_cache_control_header(rv.headers["Cache-Control"])
assert cc.max_age == 3600
rv.close()
# Test again with direct use of send_file utility.
rv = flask.send_file("static/index.html")
cc = parse_cache_control_header(rv.headers["Cache-Control"])
assert cc.max_age == 3600
assert rv.cache_control.max_age == 3600
rv.close()
# Test with static file handler.
# Test with direct use of send_file.
rv = flask.send_file("static/index.html")
assert rv.cache_control.max_age == 3600
rv.close()
# Test with pathlib.Path.
rv = app.send_static_file(FakePath("index.html"))
cc = parse_cache_control_header(rv.headers["Cache-Control"])
assert cc.max_age == 3600
assert rv.cache_control.max_age == 3600
rv.close()
class StaticFileApp(flask.Flask):
@ -318,16 +77,16 @@ class TestSendfile:
return 10
app = StaticFileApp(__name__)
with app.test_request_context():
# Test with static file handler.
rv = app.send_static_file("index.html")
cc = parse_cache_control_header(rv.headers["Cache-Control"])
assert cc.max_age == 10
assert rv.cache_control.max_age == 10
rv.close()
# Test again with direct use of send_file utility.
# Test with direct use of send_file.
rv = flask.send_file("static/index.html")
cc = parse_cache_control_header(rv.headers["Cache-Control"])
assert cc.max_age == 10
assert rv.cache_control.max_age == 10
rv.close()
def test_send_from_directory(self, app, req_ctx):
@ -339,28 +98,6 @@ class TestSendfile:
assert rv.data.strip() == b"Hello Subdomain"
rv.close()
def test_send_from_directory_pathlike(self, app, req_ctx):
app.root_path = os.path.join(
os.path.dirname(__file__), "test_apps", "subdomaintestmodule"
)
rv = flask.send_from_directory(FakePath("static"), FakePath("hello.txt"))
rv.direct_passthrough = False
assert rv.data.strip() == b"Hello Subdomain"
rv.close()
def test_send_from_directory_null_character(self, app, req_ctx):
app.root_path = os.path.join(
os.path.dirname(__file__), "test_apps", "subdomaintestmodule"
)
if sys.version_info >= (3, 8):
exception = NotFound
else:
exception = BadRequest
with pytest.raises(exception):
flask.send_from_directory("static", "bad\x00")
class TestUrlFor:
def test_url_for_with_anchor(self, app, req_ctx):
@ -514,47 +251,6 @@ class TestStreaming:
assert rv.data == b"flask"
class TestSafeJoin:
@pytest.mark.parametrize(
"args, expected",
(
(("a/b/c",), "a/b/c"),
(("/", "a/", "b/", "c/"), "/a/b/c"),
(("a", "b", "c"), "a/b/c"),
(("/a", "b/c"), "/a/b/c"),
(("a/b", "X/../c"), "a/b/c"),
(("/a/b", "c/X/.."), "/a/b/c"),
# If last path is '' add a slash
(("/a/b/c", ""), "/a/b/c/"),
# Preserve dot slash
(("/a/b/c", "./"), "/a/b/c/."),
(("a/b/c", "X/.."), "a/b/c/."),
# Base directory is always considered safe
(("../", "a/b/c"), "../a/b/c"),
(("/..",), "/.."),
),
)
def test_safe_join(self, args, expected):
assert flask.safe_join(*args) == expected
@pytest.mark.parametrize(
"args",
(
# path.isabs and ``..'' checks
("/a", "b", "/c"),
("/a", "../b/c"),
("/a", "..", "b/c"),
# Boundaries violations after path normalization
("/a", "b/../b/../../c"),
("/a", "b", "c/../.."),
("/a", "b/../../c"),
),
)
def test_safe_join_exceptions(self, args):
with pytest.raises(NotFound):
print(flask.safe_join(*args))
class TestHelpers:
@pytest.mark.parametrize(
"debug, expected_flag, expected_default_flag",

View File

@ -3,7 +3,6 @@ import platform
import threading
import pytest
from werkzeug.exceptions import NotFound
import flask
@ -56,13 +55,6 @@ def test_memory_consumption():
fire()
def test_safe_join_toplevel_pardir():
from flask.helpers import safe_join
with pytest.raises(NotFound):
safe_join("/foo", "..")
def test_aborting(app):
class Foo(Exception):
whatever = 42