codekingpro/portable-devtools

1"""2Serve Shared Static Files3=========================4 5.. autoclass:: SharedDataMiddleware6    :members: is_allowed7 8:copyright: 2007 Pallets9:license: BSD-3-Clause10"""11 12from __future__ import annotations13 14import collections.abc as cabc15import importlib.util16import mimetypes17import os18import posixpath19import typing as t20from datetime import datetime21from datetime import timezone22from io import BytesIO23from time import time24from zlib import adler3225 26from ..http import http_date27from ..http import is_resource_modified28from ..security import safe_join29from ..utils import get_content_type30from ..wsgi import get_path_info31from ..wsgi import wrap_file32 33_TOpener = t.Callable[[], tuple[t.IO[bytes], datetime, int]]34_TLoader = t.Callable[[t.Optional[str]], tuple[t.Optional[str], t.Optional[_TOpener]]]35 36if t.TYPE_CHECKING:37    from _typeshed.wsgi import StartResponse38    from _typeshed.wsgi import WSGIApplication39    from _typeshed.wsgi import WSGIEnvironment40 41 42class SharedDataMiddleware:43    """A WSGI middleware which provides static content for development44    environments or simple server setups. Its usage is quite simple::45 46        import os47        from werkzeug.middleware.shared_data import SharedDataMiddleware48 49        app = SharedDataMiddleware(app, {50            '/shared': os.path.join(os.path.dirname(__file__), 'shared')51        })52 53    The contents of the folder ``./shared`` will now be available on54    ``http://example.com/shared/``.  This is pretty useful during development55    because a standalone media server is not required. Files can also be56    mounted on the root folder and still continue to use the application because57    the shared data middleware forwards all unhandled requests to the58    application, even if the requests are below one of the shared folders.59 60    If `pkg_resources` is available you can also tell the middleware to serve61    files from package data::62 63        app = SharedDataMiddleware(app, {64            '/static': ('myapplication', 'static')65        })66 67    This will then serve the ``static`` folder in the `myapplication`68    Python package.69 70    The optional `disallow` parameter can be a list of :func:`~fnmatch.fnmatch`71    rules for files that are not accessible from the web.  If `cache` is set to72    `False` no caching headers are sent.73 74    Currently the middleware does not support non-ASCII filenames. If the75    encoding on the file system happens to match the encoding of the URI it may76    work but this could also be by accident. We strongly suggest using ASCII77    only file names for static files.78 79    The middleware will guess the mimetype using the Python `mimetype`80    module.  If it's unable to figure out the charset it will fall back81    to `fallback_mimetype`.82 83    :param app: the application to wrap.  If you don't want to wrap an84                application you can pass it :exc:`NotFound`.85    :param exports: a list or dict of exported files and folders.86    :param disallow: a list of :func:`~fnmatch.fnmatch` rules.87    :param cache: enable or disable caching headers.88    :param cache_timeout: the cache timeout in seconds for the headers.89    :param fallback_mimetype: The fallback mimetype for unknown files.90 91    .. versionchanged:: 1.092        The default ``fallback_mimetype`` is93        ``application/octet-stream``. If a filename looks like a text94        mimetype, the ``utf-8`` charset is added to it.95 96    .. versionadded:: 0.697        Added ``fallback_mimetype``.98 99    .. versionchanged:: 0.5100        Added ``cache_timeout``.101    """102 103    def __init__(104        self,105        app: WSGIApplication,106        exports: (107            cabc.Mapping[str, str | tuple[str, str]]108            | t.Iterable[tuple[str, str | tuple[str, str]]]109        ),110        disallow: None = None,111        cache: bool = True,112        cache_timeout: int = 60 * 60 * 12,113        fallback_mimetype: str = "application/octet-stream",114    ) -> None:115        self.app = app116        self.exports: list[tuple[str, _TLoader]] = []117        self.cache = cache118        self.cache_timeout = cache_timeout119 120        if isinstance(exports, cabc.Mapping):121            exports = exports.items()122 123        for key, value in exports:124            if isinstance(value, tuple):125                loader = self.get_package_loader(*value)126            elif isinstance(value, str):127                if os.path.isfile(value):128                    loader = self.get_file_loader(value)129                else:130                    loader = self.get_directory_loader(value)131            else:132                raise TypeError(f"unknown def {value!r}")133 134            self.exports.append((key, loader))135 136        if disallow is not None:137            from fnmatch import fnmatch138 139            self.is_allowed = lambda x: not fnmatch(x, disallow)140 141        self.fallback_mimetype = fallback_mimetype142 143    def is_allowed(self, filename: str) -> bool:144        """Subclasses can override this method to disallow the access to145        certain files.  However by providing `disallow` in the constructor146        this method is overwritten.147        """148        return True149 150    def _opener(self, filename: str) -> _TOpener:151        return lambda: (152            open(filename, "rb"),153            datetime.fromtimestamp(os.path.getmtime(filename), tz=timezone.utc),154            int(os.path.getsize(filename)),155        )156 157    def get_file_loader(self, filename: str) -> _TLoader:158        return lambda x: (os.path.basename(filename), self._opener(filename))159 160    def get_package_loader(self, package: str, package_path: str) -> _TLoader:161        load_time = datetime.now(timezone.utc)162        spec = importlib.util.find_spec(package)163        reader = spec.loader.get_resource_reader(package)  # type: ignore[union-attr]164 165        def loader(166            path: str | None,167        ) -> tuple[str | None, _TOpener | None]:168            if path is None:169                return None, None170 171            path = safe_join(package_path, path)172 173            if path is None:174                return None, None175 176            basename = posixpath.basename(path)177 178            try:179                resource = reader.open_resource(path)180            except OSError:181                return None, None182 183            if isinstance(resource, BytesIO):184                return (185                    basename,186                    lambda: (resource, load_time, len(resource.getvalue())),187                )188 189            return (190                basename,191                lambda: (192                    resource,193                    datetime.fromtimestamp(194                        os.path.getmtime(resource.name), tz=timezone.utc195                    ),196                    os.path.getsize(resource.name),197                ),198            )199 200        return loader201 202    def get_directory_loader(self, directory: str) -> _TLoader:203        def loader(204            path: str | None,205        ) -> tuple[str | None, _TOpener | None]:206            if path is not None:207                path = safe_join(directory, path)208 209                if path is None:210                    return None, None211            else:212                path = directory213 214            if os.path.isfile(path):215                return os.path.basename(path), self._opener(path)216 217            return None, None218 219        return loader220 221    def generate_etag(self, mtime: datetime, file_size: int, real_filename: str) -> str:222        fn_str = os.fsencode(real_filename)223        timestamp = mtime.timestamp()224        checksum = adler32(fn_str) & 0xFFFFFFFF225        return f"wzsdm-{timestamp}-{file_size}-{checksum}"226 227    def __call__(228        self, environ: WSGIEnvironment, start_response: StartResponse229    ) -> t.Iterable[bytes]:230        path = get_path_info(environ)231        file_loader = None232 233        for search_path, loader in self.exports:234            if search_path == path:235                real_filename, file_loader = loader(None)236 237                if file_loader is not None:238                    break239 240            if not search_path.endswith("/"):241                search_path += "/"242 243            if path.startswith(search_path):244                real_filename, file_loader = loader(path[len(search_path) :])245 246                if file_loader is not None:247                    break248 249        if file_loader is None or not self.is_allowed(real_filename):  # type: ignore250            return self.app(environ, start_response)251 252        guessed_type = mimetypes.guess_type(real_filename)  # type: ignore253        mime_type = get_content_type(guessed_type[0] or self.fallback_mimetype, "utf-8")254        f, mtime, file_size = file_loader()255 256        headers = [("Date", http_date())]257 258        if self.cache:259            timeout = self.cache_timeout260            etag = self.generate_etag(mtime, file_size, real_filename)  # type: ignore261            headers += [262                ("Etag", f'"{etag}"'),263                ("Cache-Control", f"max-age={timeout}, public"),264            ]265 266            if not is_resource_modified(environ, etag, last_modified=mtime):267                f.close()268                start_response("304 Not Modified", headers)269                return []270 271            headers.append(("Expires", http_date(time() + timeout)))272        else:273            headers.append(("Cache-Control", "public"))274 275        headers.extend(276            (277                ("Content-Type", mime_type),278                ("Content-Length", str(file_size)),279                ("Last-Modified", http_date(mtime)),280            )281        )282        start_response("200 OK", headers)283        return wrap_file(environ, f)284