sanic/sanic/static.py

from functools import partial, wraps
from mimetypes import guess_type
from os import path
from pathlib import PurePath
from re import sub
from time import gmtime, strftime
from urllib.parse import unquote

from sanic.compat import stat_async
from sanic.exceptions import (
    ContentRangeError,
    FileNotFound,
    HeaderNotFound,
    InvalidUsage,
)
from sanic.handlers import ContentRangeHandler
from sanic.log import error_logger
from sanic.models.futures import FutureStatic
from sanic.response import HTTPResponse, file, file_stream


async def _static_request_handler(
    file_or_directory,
    use_modified_since,
    use_content_range,
    stream_large_files,
    request,
    content_type=None,
    file_uri=None,
):
    # Using this to determine if the URL is trying to break out of the path
    # served.  os.path.realpath seems to be very slow
    if file_uri and "../" in file_uri:
        raise InvalidUsage("Invalid URL")
    # Merge served directory and requested file if provided
    # Strip all / that in the beginning of the URL to help prevent python
    # from herping a derp and treating the uri as an absolute path
    root_path = file_path = file_or_directory
    if file_uri:
        file_path = path.join(file_or_directory, sub("^[/]*", "", file_uri))

    # URL decode the path sent by the browser otherwise we won't be able to
    # match filenames which got encoded (filenames with spaces etc)
    file_path = path.abspath(unquote(file_path))
    if not file_path.startswith(path.abspath(unquote(root_path))):
        error_logger.exception(
            f"File not found: path={file_or_directory}, "
            f"relative_url={file_uri}"
        )
        raise FileNotFound(
            "File not found", path=file_or_directory, relative_url=file_uri
        )
    try:
        headers = {}
        # Check if the client has been sent this file before
        # and it has not been modified since
        stats = None
        if use_modified_since:
            stats = await stat_async(file_path)
            modified_since = strftime(
                "%a, %d %b %Y %H:%M:%S GMT", gmtime(stats.st_mtime)
            )
            if request.headers.get("If-Modified-Since") == modified_since:
                return HTTPResponse(status=304)
            headers["Last-Modified"] = modified_since
        _range = None
        if use_content_range:
            _range = None
            if not stats:
                stats = await stat_async(file_path)
            headers["Accept-Ranges"] = "bytes"
            headers["Content-Length"] = str(stats.st_size)
            if request.method != "HEAD":
                try:
                    _range = ContentRangeHandler(request, stats)
                except HeaderNotFound:
                    pass
                else:
                    del headers["Content-Length"]
                    for key, value in _range.headers.items():
                        headers[key] = value
        headers["Content-Type"] = (
            content_type or guess_type(file_path)[0] or "text/plain"
        )
        if request.method == "HEAD":
            return HTTPResponse(headers=headers)
        else:
            if stream_large_files:
                if type(stream_large_files) == int:
                    threshold = stream_large_files
                else:
                    threshold = 1024 * 1024

                if not stats:
                    stats = await stat_async(file_path)
                if stats.st_size >= threshold:
                    return await file_stream(
                        file_path, headers=headers, _range=_range
                    )
            return await file(file_path, headers=headers, _range=_range)
    except ContentRangeError:
        raise
    except Exception:
        error_logger.exception(
            f"File not found: path={file_or_directory}, "
            f"relative_url={file_uri}"
        )
        raise FileNotFound(
            "File not found", path=file_or_directory, relative_url=file_uri
        )


def register(
    app,
    static: FutureStatic,
):
    # TODO: Though sanic is not a file server, I feel like we should at least
    #       make a good effort here.  Modified-since is nice, but we could
    #       also look into etags, expires, and caching
    """
    Register a static directory handler with Sanic by adding a route to the
    router and registering a handler.

    :param app: Sanic
    :param file_or_directory: File or directory path to serve from
    :type file_or_directory: Union[str,bytes,Path]
    :param uri: URL to serve from
    :type uri: str
    :param pattern: regular expression used to match files in the URL
    :param use_modified_since: If true, send file modified time, and return
                               not modified if the browser's matches the
                               server's
    :param use_content_range: If true, process header for range requests
                              and sends the file part that is requested
    :param stream_large_files: If true, use the file_stream() handler rather
                              than the file() handler to send the file
                              If this is an integer, this represents the
                              threshold size to switch to file_stream()
    :param name: user defined name used for url_for
    :type name: str
    :param content_type: user defined content type for header
    :return: registered static routes
    :rtype: List[sanic.router.Route]
    """

    if isinstance(static.file_or_directory, bytes):
        file_or_directory = static.file_or_directory.decode("utf-8")
    elif isinstance(static.file_or_directory, PurePath):
        file_or_directory = str(static.file_or_directory)
    elif not isinstance(static.file_or_directory, str):
        raise ValueError("Invalid file path string.")
    else:
        file_or_directory = static.file_or_directory

    uri = static.uri
    name = static.name
    # If we're not trying to match a file directly,
    # serve from the folder
    if not path.isfile(file_or_directory):
        uri += "/<file_uri>"

    # special prefix for static files
    # if not static.name.startswith("_static_"):
    #     name = f"_static_{static.name}"

    _handler = wraps(_static_request_handler)(
        partial(
            _static_request_handler,
            file_or_directory,
            static.use_modified_since,
            static.use_content_range,
            static.stream_large_files,
            content_type=static.content_type,
        )
    )

    route, _ = app.route(
        uri=uri,
        methods=["GET", "HEAD"],
        name=name,
        host=static.host,
        strict_slashes=static.strict_slashes,
        static=True,
    )(_handler)

    return route