diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 7876444..ecf6430 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -79,6 +79,7 @@ jobs:
- "3.11"
- "3.12"
- "3.13"
+ - "3.14"
steps:
- uses: actions/checkout@v4
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8f81dc5..667d648 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -13,6 +13,39 @@ Don't forget to remove deprecated code on each major release!
## [Unreleased]
+- Nothing (yet)
+
+## [4.1.0] - 2026-03-07
+
+!!! tip
+
+ This release includes some changes to the default behavior of `ServeStatic` for security hardening. If you are affected by any of these changes, please read the relevant sections in the documentation on `allow_unsafe_symlinks`.
+
+### Added
+
+- Added support for `zstd` compression on Python 3.14+.
+- Added support for the top-level `servestatic` module to run as a Django app.
+- Added Django system checks to test for common misconfigurations.
+- Added a new `allow_unsafe_symlinks` configuration option for WSGI/ASGI
+- Added a new `SERVESTATIC_ALLOW_UNSAFE_SYMLINKS` configuration option for Django.
+- Added `jxl` image support.
+
+### Changed
+
+- Improved event-loop handling for ASGI file iterator.
+- Installing `servestatic` as a Django app is now the suggested configuration. A warning will appear if it is not detected in `INSTALLED_APPS` when `DEBUG` is `True`.
+- `servestatic.runserver_nostatic` is no longer the recommended Django app installation path. This import path will be retained to ease `WhiteNoise` to `ServeStatic` migration, but now the documentation recommends to use the top-level `servestatic` module instead.
+- For security purposes, `ServeStatic` will no longer follow unsafe symlinks by default. If your symlinks point to files outside of your static root, it is highly recommended to copy them instead. This behavior can be disabled for trusted deployments using `allow_unsafe_symlinks` / `SERVESTATIC_ALLOW_UNSAFE_SYMLINKS`.
+
+### Fixed
+
+- Fixed a range-request edge case where the last byte could be requested but would not be served.
+
+### Security
+
+- Hardened `autorefresh` path matching to prevent potential path traversal or path clobbering.
+- Hardened static file resolution to block symlink breakout by default.
+
## [4.0.0] - 2026-03-05
### Added
@@ -129,7 +162,8 @@ Don't forget to remove deprecated code on each major release!
- Forked from [`whitenoise`](https://github.com/evansd/whitenoise) to add ASGI support.
-[Unreleased]: https://github.com/Archmonger/ServeStatic/compare/4.0.0...HEAD
+[Unreleased]: https://github.com/Archmonger/ServeStatic/compare/4.1.0...HEAD
+[4.1.0]: https://github.com/Archmonger/ServeStatic/compare/4.0.0...4.1.0
[4.0.0]: https://github.com/Archmonger/ServeStatic/compare/3.1.0...4.0.0
[3.1.0]: https://github.com/Archmonger/ServeStatic/compare/3.0.2...3.1.0
[3.0.2]: https://github.com/Archmonger/ServeStatic/compare/3.0.1...3.0.2
diff --git a/README.md b/README.md
index 9ba29b0..b7aef6f 100644
--- a/README.md
+++ b/README.md
@@ -19,39 +19,39 @@
_Production-grade static file server for Python WSGI & ASGI._
-_This project is a fork of [WhiteNoise](https://github.com/evansd/whitenoise) for [ASGI support, bug fixes, new features, and performance upgrades](https://archmonger.github.io/ServeStatic/latest/changelog/)._
+_This project is a fork of [WhiteNoise](https://github.com/evansd/whitenoise) for [ASGI support, bug fixes, security updates, new features, and performance upgrades](https://archmonger.github.io/ServeStatic/latest/changelog/)._
---
-`ServeStatic` simplifies static file serving with minimal lines of configuration. It can operate standalone or transform your existing app into a self-contained unit, without relying on external services like nginx or Amazon S3. This can simplify any production deployment, but is especially useful for platforms like Heroku, OpenShift, and other PaaS providers.
+`ServeStatic` simplifies static file serving with minimal lines of configuration. It enables you to create a self-contained unit (WSGI, ASGI, Django, or 'standalone') without requiring external services like Nginx or Amazon S3. This can simplify any production deployment, but is especially useful for platforms like Heroku, OpenShift, and other PaaS providers.
-It is designed to work seamlessly with CDNs to ensure high performance for traffic-intensive sites, and is compatible with any ASGI/WSGI app. Extra features and auto-configuration are available for [Django](https://www.djangoproject.com/) users.
+It is designed to work seamlessly with CDNs to ensure high performance for traffic-intensive sites, and is compatible with any ASGI/WSGI app. A command-line interface is provided to perform common tasks such as compression, file-name hashing, and manifest generation. Extra features and auto-configuration are available for [Django](https://www.djangoproject.com/) users.
-Best practices are automatically handled by `ServeStatic`, such as:
+When using `ServeStatic`, best practices are automatically handled such as:
- Automatically serving compressed content
- Proper handling of `Accept-Encoding` and `Vary` headers
- Setting far-future cache headers for immutable static files.
-To get started or learn more about `ServeStatic`, visit the [documentation](https://archmonger.github.io/ServeStatic/).
+Visit the [documentation](https://archmonger.github.io/ServeStatic/) to get started or learn more about `ServeStatic` .
## Frequently Asked Questions
### Isn't serving static files from Python horribly inefficient?
-The short answer to this is that if you care about performance and efficiency then you should be using `ServeStatic` behind a CDN like CloudFront. Due to the caching headers `ServeStatic` sends, the vast majority of static requests will be served directly by the CDN without touching your application, so it really doesn't make much difference how efficient `ServeStatic` is.
+The short answer to this is that if you care about performance and efficiency then you should be using `ServeStatic` behind a CDN (such as CloudFront). Due to the caching headers `ServeStatic` sends, the vast majority of static requests will be served directly by the CDN without touching your application, so it really doesn't make much difference how efficient `ServeStatic` is.
That said, `ServeStatic` is pretty efficient. Because it only has to serve a fixed set of files it does all the work of finding files and determining the correct headers upfront on initialization. Requests can then be served with little more than a dictionary lookup to find the appropriate response. Also, when used with gunicorn (and most other WSGI servers) the actual business of pushing the file down the network interface is handled by the kernel's very efficient `sendfile` syscall, not by Python.
### Shouldn't I be pushing my static files to S3 (using Django-Storages)?
-No, you shouldn't. The main problem with this approach is that Amazon S3 cannot currently selectively serve compressed content to your users. Compression using the gzip or the modern brotli algorithm can make dramatic reductions in load time and bandwidth usage. But, in order to do this correctly the server needs to examine the `Accept-Encoding` header of the request to determine which compression formats are supported, and return an appropriate `Vary` header so that intermediate caches know to do the same. This is exactly what `ServeStatic` does, but Amazon S3 currently provides no means of doing this.
+No, you shouldn't. The main problem with this approach is that Amazon S3 cannot currently selectively serve compressed content to your users. Compression using gzip, zstd (Python 3.14+), or brotli can make dramatic reductions in load time and bandwidth usage. But, in order to do this correctly the server needs to examine the `Accept-Encoding` header of the request to determine which compression formats are supported, and return an appropriate `Vary` header so that intermediate caches know to do the same. This is exactly what `ServeStatic` does, but Amazon S3 currently provides no means of doing this.
The second problem with a push-based approach to handling static files is that it adds complexity and fragility to your deployment process: extra libraries specific to your storage backend, extra configuration and authentication keys, and extra tasks that must be run at specific points in the deployment in order for everything to work. With the CDN-as-caching-proxy approach that `ServeStatic` takes there are just two bits of configuration: your application needs the URL of the CDN, and the CDN needs the URL of your application. Everything else is just standard HTTP semantics. This makes your deployments simpler, your life easier, and you happier.
### What's the point in `ServeStatic` when I can use `apache`/`nginx`?
-There are two answers here. One is that ServeStatic is designed to work in situations where `apache`, `nginx`, and the like aren't easily available. But more importantly, it's easy to underestimate what's involved in serving static files correctly. Does your few lines of nginx configuration distinguish between files which might change and files which will never change and set the cache headers appropriately? Did you add the right CORS headers so that your fonts load correctly when served via a CDN? Did you turn on the special nginx setting which allows it to send gzip content in response to an `HTTP/1.0` request, which for some reason CloudFront still uses? Did you install the extension which allows you to serve brotli-encoded content to modern browsers?
+There are two answers here. One is that ServeStatic is designed to work in situations where `apache`, `nginx`, and the like aren't easily available. But more importantly, it's easy to underestimate what's involved in serving static files correctly. Does your few lines of nginx configuration distinguish between files which might change and files which will never change and set the cache headers appropriately? Did you add the right CORS headers so that your fonts load correctly when served via a CDN? Did you turn on the special nginx setting which allows it to send gzip content in response to an `HTTP/1.0` request? Did you install, enable, and configure support for zstd and brotli compression?
None of this is rocket science, but it's fiddly and annoying and `ServeStatic` takes care of all it for you.
diff --git a/docs/src/cli.md b/docs/src/cli.md
index 643d879..1151206 100644
--- a/docs/src/cli.md
+++ b/docs/src/cli.md
@@ -14,7 +14,8 @@ You can either run this during development and commit your generated/compressed
$ servestatic --help
usage: servestatic [-h] [--all] [--hash] [--manifest] [--merge-manifest]
[--compress] [--clear] [-q] [--copy-original] [--no-gzip]
- [--no-brotli]
+ [--no-brotli] [--no-zstd] [--zstd-dict ZSTD_DICT]
+ [--zstd-dict-raw] [--zstd-level ZSTD_LEVEL]
[-e EXCLUDE]
src dest
@@ -34,7 +35,8 @@ options:
--merge-manifest Merge the new manifest with an existing manifest in the
dest directory. Fails if the existing manifest is not
found. (default: False)
- --compress Generate compressed versions (gzip/brotli) of files.
+ --compress Generate compressed versions (gzip/zstd/brotli) of
+ files.
(default: False)
--clear Empty the destination directory before processing.
(default: False)
@@ -46,6 +48,16 @@ options:
--compress). (default: True)
--no-brotli Don't produce brotli '.br' files (only applies with
--compress). (default: True)
+ --no-zstd Don't produce zstd '.zstd' files (only applies with
+ --compress). (default: True)
+ --zstd-dict ZSTD_DICT
+ Path to a zstd dictionary file (only applies with
+ --compress). (default: None)
+ --zstd-dict-raw Treat the zstd dictionary as raw content (only applies
+ with --compress). (default: False)
+ --zstd-level ZSTD_LEVEL
+ Compression level for zstd output (only applies with
+ --compress). (default: None)
-e EXCLUDE, --exclude EXCLUDE
Glob pattern(s) to exclude from processing
(compression/hashing). These files are still copied.
@@ -54,9 +66,13 @@ options:
## Compression Details
-When ServeStatic builds its list of available files, it optionally checks for corresponding files with a `.gz` and a `.br` suffix (e.g., `scripts/app.js`, `scripts/app.js.gz` and `scripts/app.js.br`). If it finds them, it will assume that they are (respectively) gzip and [brotli](https://en.wikipedia.org/wiki/Brotli) compressed versions of the original file and it will serve them in preference to the uncompressed version whenever clients indicate they support that compression format.
+When ServeStatic builds its list of available files, it optionally checks for corresponding files with `.gz`, `.zstd`, and `.br` suffixes (e.g., `scripts/app.js`, `scripts/app.js.gz`, `scripts/app.js.zstd`, and `scripts/app.js.br`). If it finds them, it will serve those compressed versions when clients indicate support via `Accept-Encoding`.
+
+On Python 3.14+, zstd support is built in via the standard library.
+
+On older Python versions, Brotli support is available by installing the [Brotli](https://pypi.org/project/Brotli/) package.
-In order for brotli compression to work, the [Brotli](https://pypi.org/project/Brotli/) python package must be installed.
+You can also pass a custom zstd dictionary with `--zstd-dict` and optionally mark it raw with `--zstd-dict-raw`.
## Hash Details
diff --git a/docs/src/dictionary.txt b/docs/src/dictionary.txt
index bebe042..596cc3a 100644
--- a/docs/src/dictionary.txt
+++ b/docs/src/dictionary.txt
@@ -1,6 +1,7 @@
wsgi
asgi
gzip
+zstd
filesystem
mimetype
mimetypes
@@ -36,3 +37,7 @@ linters
linting
pytest
formatters
+misconfigurations
+encodings
+symlink
+symlinks
diff --git a/docs/src/django-settings.md b/docs/src/django-settings.md
index 212bef1..f44e393 100644
--- a/docs/src/django-settings.md
+++ b/docs/src/django-settings.md
@@ -108,9 +108,19 @@ The W3C [explicitly state](https://www.w3.org/TR/cors/#security) that this behav
---
+## `SERVESTATIC_ALLOW_UNSAFE_SYMLINKS`
+
+**Default:** `False`
+
+Controls whether symlinks that resolve outside configured static roots are allowed.
+
+By default, ServeStatic blocks symlink breakout so requests cannot escape the configured static directory tree. Set this to `True` only if you intentionally depend on symlinks that point outside your static roots and you trust those links.
+
+---
+
## `SERVESTATIC_SKIP_COMPRESS_EXTENSIONS`
-**Default:** `('jpg', 'jpeg', 'png', 'gif', 'webp','zip', 'gz', 'tgz', 'bz2', 'tbz', 'xz', 'br', 'swf', 'flv', 'woff', 'woff2')`
+**Default:** `('jpg', 'jpeg', 'png', 'gif', 'webp','zip', 'gz', 'tgz', 'bz2', 'tbz', 'xz', 'br', 'zstd', 'swf', 'flv', 'woff', 'woff2')`
File extensions to skip when compressing.
@@ -118,6 +128,43 @@ Because the compression process will only create compressed files where this res
---
+## `SERVESTATIC_USE_ZSTD`
+
+**Default:** `True`
+
+Enable or disable zstd output generation when `compression.zstd` is available (Python 3.14+).
+
+---
+
+## `SERVESTATIC_ZSTD_DICTIONARY`
+
+**Default:** `None`
+
+Optional zstd dictionary to improve compression ratio for your asset corpus.
+
+This setting can be either:
+
+- a filesystem path to a trained dictionary file, or
+- raw dictionary bytes / a prebuilt zstd dictionary object supplied by custom storage subclass logic.
+
+---
+
+## `SERVESTATIC_ZSTD_DICTIONARY_IS_RAW`
+
+**Default:** `False`
+
+Set to `True` if `SERVESTATIC_ZSTD_DICTIONARY` points to a raw-content dictionary.
+
+---
+
+## `SERVESTATIC_ZSTD_LEVEL`
+
+**Default:** `None`
+
+Optional zstd compression level.
+
+---
+
## `SERVESTATIC_ADD_HEADERS_FUNCTION`
**Default:** `None`
@@ -196,9 +243,7 @@ If your deployment is more complicated than this (for instance, if you are using
Stores only files with hashed names in `STATIC_ROOT`.
-By default, Django's hashed static files system creates two copies of each file in `STATIC_ROOT`: one using the original name, e.g. `app.js`, and one using the hashed name, e.g. `app.db8f2edc0c8a.js`. If `ServeStatic`'s compression backend is being used this will create another two copies of each of these files (using Gzip and Brotli compression) resulting in six output files for each input file.
-
-In some deployment scenarios it can be important to reduce the size of the build artifact as much as possible. This setting removes the "unhashed" version of the file (which should be not be referenced in any case) which should reduce the space required for static files by half.
+This setting removes the "unhashed" version of the file (which should be not be referenced in any case) which should reduce the space required for static files. In some deployment scenarios it can be important to reduce the size of the build artifact as much as possible.
This setting is only effective if the `ServeStatic` storage backend is being used.
diff --git a/docs/src/django.md b/docs/src/django.md
index 0cbfc96..63ff2b0 100644
--- a/docs/src/django.md
+++ b/docs/src/django.md
@@ -20,6 +20,15 @@ MIDDLEWARE = [
]
```
+To enable ServeStatic's configuration checks, add `servestatic` to `INSTALLED_APPS`:
+
+```python linenums="0"
+INSTALLED_APPS = [
+ "servestatic",
+ # ...
+]
+```
+
That's it! ServeStatic is now configured to serve your static files. For optimal performance, proceed to the next step to enable compression and caching.
??? question "How should I order my middleware?"
@@ -47,15 +56,15 @@ This combines automatic compression with the caching behaviour provided by Djang
If you need to compress files outside of the static files storage system you can use the supplied [command line utility](cli.md).
-??? tip "Enable Brotli compression"
+??? tip "Enable modern compression"
+
+ ServeStatic supports gzip, zstd, and brotli compression.
- As well as the common gzip compression format, ServeStatic supports the newer, more efficient [brotli](https://en.wikipedia.org/wiki/Brotli)
- format. This helps reduce bandwidth and increase loading speed. To enable brotli compression you will need the [Brotli Python
- package](https://pypi.org/project/Brotli/) installed by running `pip install servestatic[brotli]`.
+ On Python 3.14+, zstd support is available from the standard library and is enabled automatically.
- Brotli is supported by [all modern browsers](https://caniuse.com/#feat=brotli). ServeStatic will only serve brotli data to browsers which request it so there are no compatibility issues with enabling brotli support.
+ On older Python versions, install Brotli with `pip install servestatic[brotli]` to get brotli support in addition to gzip.
- Also note that browsers will only request brotli data over an HTTPS connection.
+ ServeStatic only sends compressed responses for encodings requested by the client, so enabling these formats remains backward-compatible.
## Step 3: Make sure Django's `staticfiles` is configured correctly
@@ -93,11 +102,11 @@ See the [reference documentation](./django-settings.md) for a full list of optio
In development Django's `runserver` automatically takes over static file handling. In most cases this is fine, however this means that some of the improvements that ServeStatic makes to static file handling won't be available in development and it opens up the possibility for differences in behaviour between development and production environments. For this reason it's a good idea to use ServeStatic in development as well.
-You can disable Django's static file handling and allow ServeStatic to take over simply by passing the `--nostatic` option to the `runserver` command, but you need to remember to add this option every time you call `runserver`. An easier way is to edit your `settings.py` file and add `servestatic.runserver_nostatic` to the top of your `INSTALLED_APPS` list:
+You can disable Django's static file handling and allow ServeStatic to take over simply by passing the `--nostatic` option to the `runserver` command, but you need to remember to add this option every time you call `runserver`. An easier way is to add `servestatic` to the top of your `INSTALLED_APPS` list:
```python linenums="0"
INSTALLED_APPS = [
- "servestatic.runserver_nostatic",
+ "servestatic",
"django.contrib.staticfiles",
# ...
]
@@ -143,7 +152,7 @@ Below are instruction for setting up ServeStatic with Amazon CloudFront, a popul
By default, CloudFront will discard any `Accept-Encoding` header browsers include in requests, unless the value of the header is gzip. If it is gzip, CloudFront will fetch the uncompressed file from the origin, compress it, and return it to the requesting browser.
- To get CloudFront to not do the compression itself as well as serve files compressed using other algorithms, such as Brotli, you must configure your distribution to [cache based on the Accept-Encoding header](https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/ServingCompressedFiles.html#compressed-content-custom-origin). You can do this in the `Behaviours` tab of your distribution.
+ To get CloudFront to not do the compression itself as well as serve files compressed using other algorithms, such as zstd and Brotli, you must configure your distribution to [cache based on the Accept-Encoding header](https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/ServingCompressedFiles.html#compressed-content-custom-origin). You can do this in the `Behaviours` tab of your distribution.
??? warning "CloudFront SEO issues"
diff --git a/docs/src/quick-start.md b/docs/src/quick-start.md
index 22bdaff..f3d9ae7 100644
--- a/docs/src/quick-start.md
+++ b/docs/src/quick-start.md
@@ -7,6 +7,7 @@ The documentation below is a quick-start guide to using ServeStatic to serve you
Install with:
```bash linenums="0"
+# Note: 'brotli' is an optional extra that adds support for more efficient compression.
pip install servestatic[brotli]
```
diff --git a/docs/src/servestatic.md b/docs/src/servestatic.md
index 1f6d93e..0a40f03 100644
--- a/docs/src/servestatic.md
+++ b/docs/src/servestatic.md
@@ -91,6 +91,16 @@ The W3C [explicitly state](https://www.w3.org/TR/cors/#security) that this behav
---
+### `allow_unsafe_symlinks`
+
+**Default:** `False`
+
+Controls whether symlinks that resolve outside configured static roots are allowed.
+
+By default, ServeStatic blocks symlink breakout so requests cannot escape the configured static directory tree. Set this to `True` only if you intentionally depend on symlinks that point outside your static roots and you trust those links.
+
+---
+
### `add_headers_function`
**Default:** `None`
diff --git a/pyproject.toml b/pyproject.toml
index 4bb0a27..344a0e8 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -21,6 +21,7 @@ classifiers = [
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
+ "Programming Language :: Python :: 3.14",
"Topic :: Communications :: File Sharing",
"Topic :: Internet :: WWW/HTTP",
"Topic :: Internet :: WWW/HTTP :: WSGI :: Middleware",
@@ -51,7 +52,7 @@ installer = "uv"
# >>> Hatch Test Suite <<<
[tool.hatch.envs.hatch-test]
-extra-dependencies = ["pytest-sugar", "requests", "brotli", "pytest-timeout"]
+extra-dependencies = ["pytest-sugar", "httpx", "brotli", "pytest-timeout"]
randomize = true
matrix-name-format = "{variable}-{value}"
timeout = 15
@@ -68,7 +69,7 @@ django = ["5.0"]
# Django 5.1
[[tool.hatch.envs.hatch-test.matrix]]
-python = ["3.10", "3.11", "3.12", "3.13"]
+python = ["3.10", "3.11", "3.12", "3.13", "3.14"]
django = ["5.1"]
[tool.hatch.envs.hatch-test.overrides]
diff --git a/scripts/generate_default_media_types.py b/scripts/generate_default_media_types.py
index 02a4e12..c7ed288 100644
--- a/scripts/generate_default_media_types.py
+++ b/scripts/generate_default_media_types.py
@@ -36,6 +36,7 @@ def main() -> int:
EXTRA_MIMETYPES = {
# Nginx uses application/javascript, but HTML specification recommends text/javascript:
".js": "text/javascript",
+ ".jxl": "image/jxl",
".md": "text/markdown",
".mjs": "text/javascript",
".woff": "application/font-woff",
diff --git a/scripts/security_probe.py b/scripts/security_probe.py
new file mode 100644
index 0000000..052875b
--- /dev/null
+++ b/scripts/security_probe.py
@@ -0,0 +1,274 @@
+"""
+This script performs a series of security probes against both the WSGI and ASGI versions of `ServeStatic` to check
+for vulnerabilities. The results are printed in a report format, indicating whether each probe passed or failed
+for both WSGI and ASGI implementations.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+
+import httpx
+
+from servestatic import ServeStatic, ServeStaticASGI
+
+
+@dataclass
+class ProbeOutcome:
+ expected: str
+ wsgi_ok: bool | None
+ asgi_ok: bool | None
+ wsgi_detail: str
+ asgi_detail: str
+
+
+def fallback_wsgi_app(environ, start_response):
+ status = "404 Not Found"
+ headers = [("Content-Type", "text/plain; charset=utf-8")]
+ start_response(status, headers)
+ return [b"fallback"]
+
+
+async def fallback_asgi_app(scope, receive, send):
+ if scope["type"] != "http":
+ msg = "fallback_asgi_app only supports HTTP"
+ raise RuntimeError(msg)
+
+ await send({
+ "type": "http.response.start",
+ "status": 404,
+ "headers": [(b"content-type", b"text/plain; charset=utf-8")],
+ })
+ await send({"type": "http.response.body", "body": b"fallback"})
+
+
+def prepare_fixture() -> tuple[Path, Path, str, bool]:
+ temp_root = Path(tempfile.mkdtemp(prefix="servestatic-security-"))
+ static_root = temp_root / "static"
+ static_root.mkdir(parents=True, exist_ok=True)
+ (static_root / "with-index").mkdir(parents=True, exist_ok=True)
+
+ (static_root / "app.js").write_text("console.log('ServeStatic security test app');\n", encoding="utf-8")
+ (static_root / "app.js.gz").write_bytes(b"not-a-real-gzip-payload\n")
+ (static_root / "with-index" / "index.html").write_text("index
\n", encoding="utf-8")
+
+ outside_marker = "OUTSIDE_FILE_MARKER"
+ outside_file = temp_root / "outside-secret.txt"
+ outside_file.write_text(outside_marker, encoding="utf-8")
+
+ symlink_created = False
+ symlink_path = static_root / "link-outside.txt"
+ try:
+ os.symlink(outside_file, symlink_path)
+ symlink_created = True
+ except (OSError, NotImplementedError):
+ symlink_created = False
+
+ return temp_root, static_root, outside_marker, symlink_created
+
+
+def check_sync_probes(
+ client: httpx.Client, outside_secret: str, symlink_created: bool
+) -> dict[str, tuple[bool | None, str]]:
+ outcomes: dict[str, tuple[bool | None, str]] = {}
+
+ response = client.get("/static/app.js")
+ ok = response.status_code == 200 and outside_secret.encode() not in response.content
+ outcomes["baseline_static"] = (ok, f"status={response.status_code}")
+
+ response = client.get("/static/%2e%2e/outside-secret.txt")
+ ok = not (response.status_code == 200 and outside_secret.encode() in response.content)
+ outcomes["path_traversal_encoded_dotdot"] = (ok, f"status={response.status_code}")
+
+ response = client.get("/static/..\\outside-secret.txt")
+ ok = not (response.status_code == 200 and outside_secret.encode() in response.content)
+ outcomes["path_traversal_backslash"] = (ok, f"status={response.status_code}")
+
+ response = client.get("/static/app.js.gz")
+ ok = response.status_code != 200 or b"not-a-real-gzip-payload" not in response.content
+ outcomes["direct_compressed_variant_request"] = (ok, f"status={response.status_code}")
+
+ response = client.get("/static/app.js", headers={"Range": "bytes=999999-1000000"})
+ ok = response.status_code == 416 and response.headers.get("Content-Range", "").startswith("bytes */")
+ outcomes["invalid_range_request_handling"] = (
+ ok,
+ f"status={response.status_code}, content-range={response.headers.get('Content-Range')}",
+ )
+
+ response = client.get("/static/with-index/index.html?v=1%0d%0aInjected: yes", follow_redirects=False)
+ location = response.headers.get("Location", "")
+ ok = (
+ response.status_code == 302
+ and "v=1%0d%0aInjected:" in location
+ and "\r" not in location
+ and "\n" not in location
+ )
+ outcomes["redirect_query_header_injection_resistance"] = (
+ ok,
+ f"status={response.status_code}, location={location!r}",
+ )
+
+ response = client.get("/static/app.js%00.txt")
+ ok = response.status_code != 200
+ outcomes["null_byte_encoded_path"] = (ok, f"status={response.status_code}")
+
+ if symlink_created:
+ response = client.get("/static/link-outside.txt")
+ ok = not (response.status_code == 200 and outside_secret.encode() in response.content)
+ outcomes["symlink_escape_from_static_root"] = (ok, f"status={response.status_code}")
+ else:
+ outcomes["symlink_escape_from_static_root"] = (None, "skipped (symlink unavailable)")
+
+ return outcomes
+
+
+async def check_async_probes(
+ client: httpx.AsyncClient, outside_secret: str, symlink_created: bool
+) -> dict[str, tuple[bool | None, str]]:
+ outcomes: dict[str, tuple[bool | None, str]] = {}
+
+ response = await client.get("/static/app.js")
+ ok = response.status_code == 200 and outside_secret.encode() not in response.content
+ outcomes["baseline_static"] = (ok, f"status={response.status_code}")
+
+ response = await client.get("/static/%2e%2e/outside-secret.txt")
+ ok = not (response.status_code == 200 and outside_secret.encode() in response.content)
+ outcomes["path_traversal_encoded_dotdot"] = (ok, f"status={response.status_code}")
+
+ response = await client.get("/static/..\\outside-secret.txt")
+ ok = not (response.status_code == 200 and outside_secret.encode() in response.content)
+ outcomes["path_traversal_backslash"] = (ok, f"status={response.status_code}")
+
+ response = await client.get("/static/app.js.gz")
+ ok = response.status_code != 200 or b"not-a-real-gzip-payload" not in response.content
+ outcomes["direct_compressed_variant_request"] = (ok, f"status={response.status_code}")
+
+ response = await client.get("/static/app.js", headers={"Range": "bytes=999999-1000000"})
+ ok = response.status_code == 416 and response.headers.get("Content-Range", "").startswith("bytes */")
+ outcomes["invalid_range_request_handling"] = (
+ ok,
+ f"status={response.status_code}, content-range={response.headers.get('Content-Range')}",
+ )
+
+ response = await client.get("/static/with-index/index.html?v=1%0d%0aInjected: yes", follow_redirects=False)
+ location = response.headers.get("Location", "")
+ ok = (
+ response.status_code == 302
+ and "v=1%0d%0aInjected:" in location
+ and "\r" not in location
+ and "\n" not in location
+ )
+ outcomes["redirect_query_header_injection_resistance"] = (
+ ok,
+ f"status={response.status_code}, location={location!r}",
+ )
+
+ response = await client.get("/static/app.js%00.txt")
+ ok = response.status_code != 200
+ outcomes["null_byte_encoded_path"] = (ok, f"status={response.status_code}")
+
+ if symlink_created:
+ response = await client.get("/static/link-outside.txt")
+ ok = not (response.status_code == 200 and outside_secret.encode() in response.content)
+ outcomes["symlink_escape_from_static_root"] = (ok, f"status={response.status_code}")
+ else:
+ outcomes["symlink_escape_from_static_root"] = (None, "skipped (symlink unavailable)")
+
+ return outcomes
+
+
+def status_label(value: bool | None) -> str:
+ if value is None:
+ return "SKIP"
+ return "PASS" if value else "FAIL"
+
+
+def print_report(results: dict[str, ProbeOutcome]) -> None:
+ print("# ServeStatic Security Probe Report")
+ print()
+ print("| Probe | Expected | WSGI | ASGI |")
+ print("|---|---|---|---|")
+ for probe_name, result in results.items():
+ print(
+ f"| {probe_name} | {result.expected} | {status_label(result.wsgi_ok)} ({result.wsgi_detail}) | "
+ f"{status_label(result.asgi_ok)} ({result.asgi_detail}) |"
+ )
+
+ findings = []
+ for probe_name, result in results.items():
+ if result.wsgi_ok is False or result.asgi_ok is False:
+ findings.append(probe_name)
+
+ print()
+ if findings:
+ print("Potential Findings:")
+ for finding in findings:
+ print(f"- {finding}")
+ else:
+ print("No exploitable issues detected by this probe set.")
+
+
+def run_probe() -> int:
+ _temp_root, static_root, outside_secret, symlink_created = prepare_fixture()
+
+ wsgi_app = ServeStatic(
+ fallback_wsgi_app,
+ root=static_root,
+ prefix="/static/",
+ index_file=True,
+ autorefresh=True,
+ )
+ asgi_app = ServeStaticASGI(
+ fallback_asgi_app,
+ root=static_root,
+ prefix="/static/",
+ index_file=True,
+ autorefresh=True,
+ )
+
+ with httpx.Client(
+ transport=httpx.WSGITransport(app=wsgi_app, raise_app_exceptions=False),
+ base_url="http://testserver",
+ headers={"Accept-Encoding": "identity"},
+ ) as wsgi_client:
+ wsgi_results = check_sync_probes(wsgi_client, outside_secret, symlink_created)
+
+ async def _run_asgi() -> dict[str, tuple[bool | None, str]]:
+ async with httpx.AsyncClient(
+ transport=httpx.ASGITransport(app=asgi_app, raise_app_exceptions=False),
+ base_url="http://testserver",
+ headers={"Accept-Encoding": "identity"},
+ ) as asgi_client:
+ return await check_async_probes(asgi_client, outside_secret, symlink_created)
+
+ asgi_results = asyncio.run(_run_asgi())
+
+ expected_by_probe = {
+ "baseline_static": "serves known static file",
+ "path_traversal_encoded_dotdot": "must not escape static root",
+ "path_traversal_backslash": "must reject backslash traversal",
+ "direct_compressed_variant_request": "must not serve compressed variants directly",
+ "invalid_range_request_handling": "must return 416 for invalid ranges",
+ "redirect_query_header_injection_resistance": "must preserve query without CRLF header injection",
+ "null_byte_encoded_path": "must not serve null-byte-encoded path",
+ "symlink_escape_from_static_root": "should not allow symlink breakout",
+ }
+
+ combined: dict[str, ProbeOutcome] = {}
+ for probe_name, expected in expected_by_probe.items():
+ wsgi_ok, wsgi_detail = wsgi_results[probe_name]
+ asgi_ok, asgi_detail = asgi_results[probe_name]
+ combined[probe_name] = ProbeOutcome(expected, wsgi_ok, asgi_ok, wsgi_detail, asgi_detail)
+
+ print_report(combined)
+
+ failures = [name for name, result in combined.items() if result.wsgi_ok is False or result.asgi_ok is False]
+ return 1 if failures else 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(run_probe())
diff --git a/src/servestatic/__init__.py b/src/servestatic/__init__.py
index 86490f5..f39279b 100644
--- a/src/servestatic/__init__.py
+++ b/src/servestatic/__init__.py
@@ -3,6 +3,6 @@
from servestatic.asgi import ServeStaticASGI
from servestatic.wsgi import ServeStatic
-__version__ = "4.0.0"
+__version__ = "4.1.0"
__all__ = ["ServeStatic", "ServeStaticASGI"]
diff --git a/src/servestatic/apps.py b/src/servestatic/apps.py
new file mode 100644
index 0000000..8c503dd
--- /dev/null
+++ b/src/servestatic/apps.py
@@ -0,0 +1,16 @@
+"""
+Django app entry-point for ServeStatic. This defines Django-centric start-up behavior, such as
+registering system checks.
+"""
+
+from __future__ import annotations
+
+from django.apps import AppConfig
+
+
+class ServeStaticConfig(AppConfig):
+ name = "servestatic"
+
+ def ready(self):
+ super().ready()
+ from servestatic import checks # noqa: F401
diff --git a/src/servestatic/asgi.py b/src/servestatic/asgi.py
index 598c7fd..b2316b4 100644
--- a/src/servestatic/asgi.py
+++ b/src/servestatic/asgi.py
@@ -26,7 +26,7 @@ async def __call__(self, scope, receive, send) -> None:
static_file = self.files.get(path)
# Serve static file if it exists
- if static_file:
+ if static_file is not None:
return await FileServerASGI(static_file)(scope, receive, send)
# Could not find a static file. Serve the default application instead.
@@ -53,9 +53,10 @@ async def __call__(self, scope, receive, send) -> None:
# Convert ASGI headers into WSGI headers. Allows us to reuse all of our WSGI
# header logic inside of aget_response().
wsgi_headers = {
- "HTTP_" + key.decode().upper().replace("-", "_"): value.decode() for key, value in scope["headers"]
+ "HTTP_" + key.decode("latin-1").upper().replace("-", "_"): value.decode("latin-1")
+ for key, value in scope["headers"]
}
- wsgi_headers["QUERY_STRING"] = scope["query_string"].decode()
+ wsgi_headers["QUERY_STRING"] = scope["query_string"].decode("latin-1")
# Get the ServeStatic file response
response = await self.static_file.aget_response(scope["method"], wsgi_headers)
@@ -66,7 +67,7 @@ async def __call__(self, scope, receive, send) -> None:
"status": response.status,
"headers": [
# Convert headers back to ASGI spec
- (key.lower().replace("_", "-").encode(), value.encode())
+ (key.lower().replace("_", "-").encode("latin-1"), value.encode("latin-1"))
for key, value in response.headers
],
})
diff --git a/src/servestatic/base.py b/src/servestatic/base.py
index 682994b..500c581 100644
--- a/src/servestatic/base.py
+++ b/src/servestatic/base.py
@@ -29,7 +29,7 @@ class ServeStaticBase:
DEFAULT_IMMUTABLE_FILE_TEST = re.compile(r"^.+\.[0-9a-f]{12}\..+$")
__call__: Callable
- """"Subclasses must implement `__call__`"""
+ """Subclasses must implement `__call__`."""
def __init__(
self,
@@ -53,6 +53,7 @@ def __init__(
add_headers_function: Callable[[Headers, str, str], None] | None = None,
index_file: str | bool | None = None,
immutable_file_test: Callable | str | None = None,
+ allow_unsafe_symlinks: bool = False,
):
self.autorefresh = autorefresh
self.max_age = max_age
@@ -61,6 +62,7 @@ def __init__(
self.add_headers_function = add_headers_function
self._immutable_file_test = immutable_file_test
self._immutable_file_test_regex: re.Pattern | None = None
+ self.allow_unsafe_symlinks = allow_unsafe_symlinks
self.media_types = MediaTypes(extra_types=mimetypes)
self.application = application
self.files = {}
@@ -118,9 +120,11 @@ def update_files_dictionary(self, root, prefix):
relative_path = path[len(root) :]
relative_url = relative_path.replace("\\", "/")
url = prefix + relative_url
- self.add_file_to_dictionary(url, path, stat_cache=stat_cache)
+ self.add_file_to_dictionary(url, path, root=root, stat_cache=stat_cache)
- def add_file_to_dictionary(self, url, path, stat_cache=None):
+ def add_file_to_dictionary(self, url, path, root=None, stat_cache=None):
+ if root and not self.path_within_root(root, path):
+ return
if self.is_compressed_variant(path, stat_cache=stat_cache):
return
if self.index_file is not None and url.endswith(f"/{self.index_file}"):
@@ -147,9 +151,28 @@ def candidate_paths_for_url(self, url):
for root, prefix in self.directories:
if url.startswith(prefix):
path = os.path.join(root, url[len(prefix) :])
- if os.path.commonprefix((root, path)) == root:
+ if self.path_within_root(root, path):
yield path
+ def path_within_root(self, root, path):
+ normalized_root = os.path.normpath(os.fspath(root))
+ normalized_path = os.path.normpath(os.fspath(path))
+
+ if not self._path_is_within(normalized_root, normalized_path):
+ return False
+ if getattr(self, "allow_unsafe_symlinks", False):
+ return True
+
+ resolved_root = os.path.realpath(normalized_root)
+ resolved_path = os.path.realpath(normalized_path)
+ return self._path_is_within(resolved_root, resolved_path)
+
+ @staticmethod
+ def _path_is_within(root, path):
+ with contextlib.suppress(ValueError):
+ return os.path.commonpath((root, path)) == root
+ return False
+
def find_file_at_path(self, path, url):
if self.is_compressed_variant(path):
raise MissingFileError(path)
@@ -186,11 +209,12 @@ def url_is_canonical(url):
@staticmethod
def is_compressed_variant(path, stat_cache=None):
- if path[-3:] in {".gz", ".br"}:
- uncompressed_path = path[:-3]
- if stat_cache is None:
- return os.path.isfile(uncompressed_path)
- return uncompressed_path in stat_cache
+ for suffix in (".gz", ".br", ".zstd"):
+ if path.endswith(suffix):
+ uncompressed_path = path[: -len(suffix)]
+ if stat_cache is None:
+ return os.path.isfile(uncompressed_path)
+ return uncompressed_path in stat_cache
return False
def get_static_file(self, path, url, stat_cache=None):
@@ -208,7 +232,7 @@ def get_static_file(self, path, url, stat_cache=None):
path,
headers.items(),
stat_cache=stat_cache,
- encodings={"gzip": f"{path}.gz", "br": f"{path}.br"},
+ encodings={"zstd": f"{path}.zstd", "gzip": f"{path}.gz", "br": f"{path}.br"},
)
def add_mime_headers(self, headers, path, url):
diff --git a/src/servestatic/checks.py b/src/servestatic/checks.py
new file mode 100644
index 0000000..ba62696
--- /dev/null
+++ b/src/servestatic/checks.py
@@ -0,0 +1,214 @@
+"""
+Django system checks for ServeStatic configuration issues.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from collections.abc import Iterable, Mapping
+from importlib import import_module
+
+from django.conf import settings
+from django.core.checks import Error, register
+
+try:
+ zstd = import_module("compression.zstd")
+except ImportError: # pragma: no cover
+ zstd = None
+
+SERVESTATIC_MIDDLEWARE = "servestatic.middleware.ServeStaticMiddleware"
+GZIP_MIDDLEWARE = "django.middleware.gzip.GZipMiddleware"
+
+
+def _is_non_negative_int(value) -> bool:
+ return isinstance(value, int) and not isinstance(value, bool) and value >= 0
+
+
+def _get_setting(name):
+ return getattr(settings, name, None)
+
+
+def _validate_type(name, *, expected, code, message):
+ value = _get_setting(name)
+ if value is None:
+ return []
+ return [] if isinstance(value, expected) else [Error(message, id=code)]
+
+
+def _validate_bool_setting(name, code):
+ return _validate_type(
+ name,
+ expected=bool,
+ code=code,
+ message=f"{name} must be a boolean.",
+ )
+
+
+def _validate_servestatic_root():
+ value = _get_setting("SERVESTATIC_ROOT")
+ if value is None:
+ return []
+ if isinstance(value, (str, os.PathLike)):
+ return []
+ return [Error("SERVESTATIC_ROOT must be a string path, os.PathLike, or None.", id="servestatic.E010")]
+
+
+def _validate_servestatic_max_age():
+ value = _get_setting("SERVESTATIC_MAX_AGE")
+ if value is None or _is_non_negative_int(value):
+ return []
+ return [Error("SERVESTATIC_MAX_AGE must be a non-negative integer or None.", id="servestatic.E014")]
+
+
+def _validate_servestatic_index_file():
+ value = _get_setting("SERVESTATIC_INDEX_FILE")
+ if value is None or isinstance(value, bool):
+ return []
+ if isinstance(value, str) and value:
+ return []
+ return [
+ Error(
+ "SERVESTATIC_INDEX_FILE must be a boolean, a non-empty string, or None.",
+ id="servestatic.E015",
+ )
+ ]
+
+
+def _validate_servestatic_mimetypes():
+ value = _get_setting("SERVESTATIC_MIMETYPES")
+ if value is None:
+ return []
+ if not isinstance(value, Mapping):
+ return [
+ Error("SERVESTATIC_MIMETYPES must be a mapping of string keys to string values.", id="servestatic.E016")
+ ]
+ for key, item in value.items():
+ if not isinstance(key, str) or not key:
+ return [Error("SERVESTATIC_MIMETYPES keys must be non-empty strings.", id="servestatic.E016")]
+ if not isinstance(item, str) or not item:
+ return [Error("SERVESTATIC_MIMETYPES values must be non-empty strings.", id="servestatic.E016")]
+ return []
+
+
+def _validate_servestatic_charset():
+ value = _get_setting("SERVESTATIC_CHARSET")
+ if value is None or (isinstance(value, str) and value):
+ return []
+ return [Error("SERVESTATIC_CHARSET must be a non-empty string.", id="servestatic.E017")]
+
+
+def _validate_servestatic_skip_compress_extensions():
+ value = _get_setting("SERVESTATIC_SKIP_COMPRESS_EXTENSIONS")
+ if value is None:
+ return []
+ if isinstance(value, (str, bytes, bytearray, memoryview)) or not isinstance(value, Iterable):
+ return [Error("SERVESTATIC_SKIP_COMPRESS_EXTENSIONS must be an iterable of strings.", id="servestatic.E019")]
+ for item in value:
+ if not isinstance(item, str) or not item:
+ return [
+ Error("SERVESTATIC_SKIP_COMPRESS_EXTENSIONS must contain non-empty strings.", id="servestatic.E019")
+ ]
+ return []
+
+
+def _validate_servestatic_zstd_dictionary():
+ value = _get_setting("SERVESTATIC_ZSTD_DICTIONARY")
+ if value is None:
+ return []
+ if isinstance(value, (str, os.PathLike, bytes, bytearray, memoryview)):
+ return []
+ if zstd is not None and isinstance(value, zstd.ZstdDict):
+ return []
+ return [
+ Error(
+ "SERVESTATIC_ZSTD_DICTIONARY must be a path, bytes-like value, zstd dictionary object, or None.",
+ id="servestatic.E021",
+ )
+ ]
+
+
+def _validate_servestatic_zstd_level():
+ value = _get_setting("SERVESTATIC_ZSTD_LEVEL")
+ if value is None:
+ return []
+ if isinstance(value, int) and not isinstance(value, bool):
+ return []
+ return [Error("SERVESTATIC_ZSTD_LEVEL must be an integer or None.", id="servestatic.E023")]
+
+
+def _validate_servestatic_add_headers_function():
+ value = _get_setting("SERVESTATIC_ADD_HEADERS_FUNCTION")
+ if value is None or callable(value):
+ return []
+ return [Error("SERVESTATIC_ADD_HEADERS_FUNCTION must be callable or None.", id="servestatic.E024")]
+
+
+def _validate_servestatic_immutable_file_test():
+ value = _get_setting("SERVESTATIC_IMMUTABLE_FILE_TEST")
+ if value is None or callable(value):
+ return []
+ if isinstance(value, str):
+ try:
+ re.compile(value)
+ except re.error:
+ return [Error("SERVESTATIC_IMMUTABLE_FILE_TEST regex is invalid.", id="servestatic.E025")]
+ return []
+ return [Error("SERVESTATIC_IMMUTABLE_FILE_TEST must be callable, regex string, or None.", id="servestatic.E025")]
+
+
+def _validate_servestatic_static_prefix():
+ value = _get_setting("SERVESTATIC_STATIC_PREFIX")
+ if value is None or isinstance(value, str):
+ return []
+ return [Error("SERVESTATIC_STATIC_PREFIX must be a string or None.", id="servestatic.E026")]
+
+
+@register()
+def check_middleware_configuration(app_configs, **kwargs):
+ middleware = list(getattr(settings, "MIDDLEWARE", []))
+
+ if SERVESTATIC_MIDDLEWARE not in middleware or GZIP_MIDDLEWARE not in middleware:
+ return []
+
+ if middleware.index(SERVESTATIC_MIDDLEWARE) < middleware.index(GZIP_MIDDLEWARE):
+ return []
+
+ return [
+ Error(
+ "ServeStatic middleware ordering is invalid.",
+ hint=(
+ "Move 'servestatic.middleware.ServeStaticMiddleware' before "
+ "'django.middleware.gzip.GZipMiddleware' in MIDDLEWARE."
+ ),
+ id="servestatic.E001",
+ )
+ ]
+
+
+@register()
+def check_setting_configuration(app_configs, **kwargs):
+ errors = []
+
+ errors.extend(_validate_servestatic_root())
+ errors.extend(_validate_bool_setting("SERVESTATIC_AUTOREFRESH", "servestatic.E011"))
+ errors.extend(_validate_bool_setting("SERVESTATIC_USE_MANIFEST", "servestatic.E012"))
+ errors.extend(_validate_bool_setting("SERVESTATIC_USE_FINDERS", "servestatic.E013"))
+ errors.extend(_validate_servestatic_max_age())
+ errors.extend(_validate_servestatic_index_file())
+ errors.extend(_validate_servestatic_mimetypes())
+ errors.extend(_validate_servestatic_charset())
+ errors.extend(_validate_bool_setting("SERVESTATIC_ALLOW_ALL_ORIGINS", "servestatic.E018"))
+ errors.extend(_validate_servestatic_skip_compress_extensions())
+ errors.extend(_validate_bool_setting("SERVESTATIC_USE_ZSTD", "servestatic.E020"))
+ errors.extend(_validate_servestatic_zstd_dictionary())
+ errors.extend(_validate_bool_setting("SERVESTATIC_ZSTD_DICTIONARY_IS_RAW", "servestatic.E022"))
+ errors.extend(_validate_servestatic_zstd_level())
+ errors.extend(_validate_servestatic_add_headers_function())
+ errors.extend(_validate_servestatic_immutable_file_test())
+ errors.extend(_validate_servestatic_static_prefix())
+ errors.extend(_validate_bool_setting("SERVESTATIC_KEEP_ONLY_HASHED_FILES", "servestatic.E027"))
+ errors.extend(_validate_bool_setting("SERVESTATIC_MANIFEST_STRICT", "servestatic.E028"))
+ errors.extend(_validate_bool_setting("SERVESTATIC_ALLOW_UNSAFE_SYMLINKS", "servestatic.E029"))
+
+ return errors
diff --git a/src/servestatic/cli.py b/src/servestatic/cli.py
index b5970fc..f4c81a3 100644
--- a/src/servestatic/cli.py
+++ b/src/servestatic/cli.py
@@ -40,7 +40,7 @@ def main(argv=None):
parser.add_argument(
"--compress",
action="store_true",
- help="Generate compressed versions (gzip/brotli) of files.",
+ help="Generate compressed versions (gzip/zstd/brotli) of files.",
)
parser.add_argument(
"--clear",
@@ -66,6 +66,26 @@ def main(argv=None):
dest="use_brotli",
help="Don't produce brotli '.br' files (only applies with --compress).",
)
+ parser.add_argument(
+ "--no-zstd",
+ action="store_false",
+ dest="use_zstd",
+ help="Don't produce zstd '.zstd' files (only applies with --compress).",
+ )
+ parser.add_argument(
+ "--zstd-dict",
+ help="Path to a zstd dictionary file (only applies with --compress).",
+ )
+ parser.add_argument(
+ "--zstd-dict-raw",
+ action="store_true",
+ help="Treat the zstd dictionary as raw content (only applies with --compress).",
+ )
+ parser.add_argument(
+ "--zstd-level",
+ type=int,
+ help="Compression level for zstd output (only applies with --compress).",
+ )
# Exclusion
parser.add_argument(
"-e",
@@ -94,6 +114,8 @@ def main(argv=None):
if not src_path.exists():
parser.error(f"Source directory '{src_path}' does not exist.")
+ if src_path == dest_path:
+ parser.error("Source and destination directories cannot be the same.")
existing_manifest_data = {}
existing_manifest_paths = {}
@@ -114,12 +136,18 @@ def main(argv=None):
log = (lambda _: None) if args.quiet else print
+ def is_excluded(path: Path) -> bool:
+ if not args.exclude:
+ return False
+ rel_path = path.relative_to(dest_path).as_posix()
+ return any(
+ fnmatch.fnmatch(rel_path, pattern) or fnmatch.fnmatch(path.name, pattern) for pattern in args.exclude
+ )
+
if args.clear and dest_path.exists():
- if src_path == dest_path:
- parser.error("Source and destination directories cannot be the same when using --clear.")
log(f"Clearing destination directory {dest_path}...")
for item in dest_path.iterdir():
- if item.is_dir():
+ if item.is_dir() and not item.is_symlink():
shutil.rmtree(item)
else:
item.unlink()
@@ -140,10 +168,7 @@ def main(argv=None):
continue
# Check exclusions
- rel_path = p.relative_to(dest_path).as_posix()
- if args.exclude and any(
- fnmatch.fnmatch(rel_path, pattern) or fnmatch.fnmatch(filename, pattern) for pattern in args.exclude
- ):
+ if is_excluded(p):
excluded_files.append(p)
continue
@@ -206,16 +231,17 @@ def main(argv=None):
continue
# Check exclusions - apply logic against the current file path
- rel_path = p.relative_to(dest_path).as_posix()
- if args.exclude and any(
- fnmatch.fnmatch(rel_path, pattern) or fnmatch.fnmatch(filename, pattern) for pattern in args.exclude
- ):
+ if is_excluded(p):
continue
files_to_compress.append(p)
compressor = Compressor(
use_gzip=args.use_gzip,
use_brotli=args.use_brotli,
+ use_zstd=args.use_zstd,
+ zstd_dict=args.zstd_dict,
+ zstd_dict_is_raw=args.zstd_dict_raw,
+ zstd_level=args.zstd_level,
quiet=args.quiet,
log=log,
# We explicitly rely on the compressor class's default extension exclusion filter
diff --git a/src/servestatic/compress.py b/src/servestatic/compress.py
index fe5f1bf..05e0828 100644
--- a/src/servestatic/compress.py
+++ b/src/servestatic/compress.py
@@ -5,6 +5,7 @@
import os
import re
from concurrent.futures import ThreadPoolExecutor, as_completed
+from importlib import import_module
from io import BytesIO
try:
@@ -12,6 +13,11 @@
except ImportError: # pragma: no cover
brotli = None
+try:
+ zstd = import_module("compression.zstd")
+except ImportError: # pragma: no cover
+ zstd = None
+
class Compressor:
# Extensions that it's not worth trying to compress
@@ -22,6 +28,8 @@ class Compressor:
"png",
"gif",
"webp",
+ "jxl",
+ "avif",
# Compressed files
"zip",
"gz",
@@ -30,6 +38,7 @@ class Compressor:
"tbz",
"xz",
"br",
+ "zstd",
# Flash
"swf",
"flv",
@@ -50,14 +59,46 @@ class Compressor:
"wmv",
)
- def __init__(self, extensions=None, use_gzip=True, use_brotli=True, log=print, quiet=False):
+ def __init__(
+ self,
+ extensions=None,
+ use_gzip=True,
+ use_brotli=True,
+ use_zstd=True,
+ zstd_dict=None,
+ zstd_dict_is_raw=False,
+ zstd_level=None,
+ log=print,
+ quiet=False,
+ ):
if extensions is None:
extensions = self.SKIP_COMPRESS_EXTENSIONS
self.extension_re = self.get_extension_re(extensions)
self.use_gzip = use_gzip
self.use_brotli = use_brotli and (brotli is not None)
+ if zstd_dict is not None and zstd is None:
+ msg = "Zstandard dictionary support requires Python 3.14+ with compression.zstd available"
+ raise RuntimeError(msg)
+ self.use_zstd = use_zstd and (zstd is not None)
+ self.zstd_level = zstd_level
+ self.zstd_dict = self.load_zstd_dictionary(zstd_dict, is_raw=zstd_dict_is_raw)
self.log = (lambda _: None) if quiet else log
+ @staticmethod
+ def load_zstd_dictionary(zstd_dict, *, is_raw=False):
+ if zstd_dict is None:
+ return None
+ if zstd is None:
+ msg = "Zstandard is not available"
+ raise RuntimeError(msg)
+ if isinstance(zstd_dict, (str, os.PathLike)):
+ with open(zstd_dict, "rb") as f:
+ zstd_dict = f.read()
+ if isinstance(zstd_dict, (bytes, bytearray, memoryview)):
+ return zstd.ZstdDict(bytes(zstd_dict), is_raw=is_raw)
+ # Allow callers to provide a pre-built zstd dictionary object.
+ return zstd_dict
+
@staticmethod
def get_extension_re(extensions):
if not extensions:
@@ -73,13 +114,14 @@ def compress(self, path):
stat_result = os.fstat(f.fileno())
data = f.read()
size = len(data)
+ if self.use_zstd:
+ compressed = self.compress_zstd(data, level=self.zstd_level, zstd_dict=self.zstd_dict)
+ if self.is_compressed_effectively("Zstandard", path, size, compressed):
+ filenames.append(self.write_data(path, compressed, ".zstd", stat_result))
if self.use_brotli:
compressed = self.compress_brotli(data)
if self.is_compressed_effectively("Brotli", path, size, compressed):
filenames.append(self.write_data(path, compressed, ".br", stat_result))
- else:
- # If Brotli compression wasn't effective gzip won't be either
- return filenames
if self.use_gzip:
compressed = self.compress_gzip(data)
if self.is_compressed_effectively("Gzip", path, size, compressed):
@@ -102,6 +144,18 @@ def compress_brotli(data):
raise RuntimeError(msg)
return brotli.compress(data)
+ @staticmethod
+ def compress_zstd(data, level=None, zstd_dict=None):
+ if zstd is None:
+ msg = "Zstandard is not available"
+ raise RuntimeError(msg)
+ kwargs = {}
+ if level is not None:
+ kwargs["level"] = level
+ if zstd_dict is not None:
+ kwargs["zstd_dict"] = zstd_dict
+ return zstd.compress(data, **kwargs)
+
def is_compressed_effectively(self, encoding_name, path, orig_size, data):
compressed_size = len(data)
if orig_size == 0:
@@ -128,7 +182,7 @@ def main(argv=None):
parser = argparse.ArgumentParser(
description="Search for all files inside *not* matching "
" and produce compressed versions with "
- "'.gz' and '.br' suffixes (as long as this results in a "
+ "'.gz', '.br', and '.zstd' suffixes (as long as this results in a "
"smaller file)"
)
parser.add_argument("-q", "--quiet", help="Don't produce log output", action="store_true")
@@ -144,12 +198,32 @@ def main(argv=None):
action="store_false",
dest="use_brotli",
)
+ parser.add_argument(
+ "--no-zstd",
+ help="Don't produce zstd '.zstd' files",
+ action="store_false",
+ dest="use_zstd",
+ )
+ parser.add_argument(
+ "--zstd-dict",
+ help="Path to a zstd dictionary file.",
+ )
+ parser.add_argument(
+ "--zstd-dict-raw",
+ help="Treat the zstd dictionary as raw content.",
+ action="store_true",
+ )
+ parser.add_argument(
+ "--zstd-level",
+ help="Compression level for zstd output.",
+ type=int,
+ )
parser.add_argument("root", help="Path root from which to search for files")
default_exclude = ", ".join(Compressor.SKIP_COMPRESS_EXTENSIONS)
parser.add_argument(
"extensions",
nargs="*",
- help=("File extensions to exclude from compression " + f"(default: {default_exclude})"),
+ help=f"File extensions to exclude from compression (default: {default_exclude})",
default=Compressor.SKIP_COMPRESS_EXTENSIONS,
)
args = parser.parse_args(argv)
@@ -158,6 +232,10 @@ def main(argv=None):
extensions=args.extensions,
use_gzip=args.use_gzip,
use_brotli=args.use_brotli,
+ use_zstd=args.use_zstd,
+ zstd_dict=args.zstd_dict,
+ zstd_dict_is_raw=args.zstd_dict_raw,
+ zstd_level=args.zstd_level,
quiet=args.quiet,
)
diff --git a/src/servestatic/management/__init__.py b/src/servestatic/management/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/src/servestatic/management/commands/__init__.py b/src/servestatic/management/commands/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/src/servestatic/management/commands/runserver.py b/src/servestatic/management/commands/runserver.py
new file mode 100644
index 0000000..2df1e30
--- /dev/null
+++ b/src/servestatic/management/commands/runserver.py
@@ -0,0 +1,58 @@
+"""
+Subclass the existing Django 'runserver' command and change the default options
+to disable static file serving, allowing ServeStatic to handle static files.
+
+There is some unpleasant hackery here because we don't know which command class
+to subclass until runtime as it depends on which INSTALLED_APPS we have, so we
+have to determine this dynamically.
+"""
+
+from __future__ import annotations
+
+import contextlib
+from importlib import import_module
+from typing import TYPE_CHECKING, cast
+
+from django.apps import apps
+
+if TYPE_CHECKING:
+ from django.core.management.base import BaseCommand
+
+SERVESTATIC_APP_NAME = "servestatic"
+
+
+def find_fallback_runserver_command():
+ """
+ Return the next highest priority "runserver" command class.
+ """
+ for app_name in iter_lower_priority_apps():
+ module_path = f"{app_name}.management.commands.runserver"
+ with contextlib.suppress(ImportError, AttributeError):
+ return import_module(module_path).Command
+ return None
+
+
+def iter_lower_priority_apps():
+ """
+ Yield all app module names below the current app in INSTALLED_APPS.
+ """
+ reached_servestatic = False
+ for app_config in apps.get_app_configs():
+ if app_config.name == SERVESTATIC_APP_NAME:
+ reached_servestatic = True
+ elif reached_servestatic:
+ yield app_config.name
+ yield "django.core"
+
+
+BaseRunserverCommand = cast("type[BaseCommand]", find_fallback_runserver_command())
+
+
+class Command(BaseRunserverCommand):
+ def add_arguments(self, parser):
+ super().add_arguments(parser)
+ if not parser.description:
+ parser.description = ""
+ if parser.get_default("use_static_handler") is True:
+ parser.set_defaults(use_static_handler=False)
+ parser.description += "\n(Wrapped by 'servestatic' to always enable '--nostatic')"
diff --git a/src/servestatic/media_types.py b/src/servestatic/media_types.py
index d693f3a..8f3ffa7 100644
--- a/src/servestatic/media_types.py
+++ b/src/servestatic/media_types.py
@@ -65,6 +65,7 @@ def default_types() -> dict[str, str]:
".jpg": "image/jpeg",
".js": "text/javascript",
".json": "application/json",
+ ".jxl": "image/jxl",
".kar": "audio/midi",
".kml": "application/vnd.google-earth.kml+xml",
".kmz": "application/vnd.google-earth.kmz",
diff --git a/src/servestatic/middleware.py b/src/servestatic/middleware.py
index fc3bcbd..cfd60c1 100644
--- a/src/servestatic/middleware.py
+++ b/src/servestatic/middleware.py
@@ -3,11 +3,15 @@
import asyncio
import contextlib
import os
+import warnings
+from collections.abc import Awaitable, Callable
+from inspect import iscoroutinefunction
from posixpath import basename, normpath
+from typing import cast
from urllib.parse import urlparse
from urllib.request import url2pathname
-from asgiref.sync import iscoroutinefunction, markcoroutinefunction
+from asgiref.sync import markcoroutinefunction
from django.conf import settings as django_settings
from django.contrib.staticfiles import finders
from django.contrib.staticfiles.storage import (
@@ -30,6 +34,27 @@
__all__ = ["ServeStaticMiddleware"]
+GetResponseCallable = Callable[[HttpRequest], Awaitable[object]]
+
+SERVESTATIC_APP_PATHS = frozenset({
+ "servestatic",
+ "servestatic.apps.ServeStaticConfig",
+ "servestatic.runserver_nostatic",
+ "servestatic.runserver_nostatic.apps.ServeStaticRunserverNoStaticAliasConfig",
+})
+
+
+def has_servestatic_app(installed_apps) -> bool:
+ return bool(set(installed_apps) & SERVESTATIC_APP_PATHS)
+
+
+def is_async_callable(value) -> bool:
+ return iscoroutinefunction(value) or (callable(value) and iscoroutinefunction(value.__call__))
+
+
+def finder_path_is_allowed(directories, url, path, path_within_root) -> bool:
+ return any(root and url.startswith(prefix) and path_within_root(root, path) for root, prefix in directories)
+
class ServeStaticMiddleware(ServeStaticBase):
"""
@@ -39,15 +64,24 @@ class ServeStaticMiddleware(ServeStaticBase):
async_capable = True
sync_capable = False
+ get_response: GetResponseCallable
def __init__(self, get_response=None, settings=django_settings):
- if not iscoroutinefunction(get_response):
+ if not is_async_callable(get_response):
msg = "ServeStaticMiddleware requires an async compatible version of Django."
raise ValueError(msg)
markcoroutinefunction(self)
- self.get_response = get_response
+ installed_apps = getattr(settings, "INSTALLED_APPS", [])
debug = settings.DEBUG
+ if debug and installed_apps and not has_servestatic_app(installed_apps):
+ warnings.warn(
+ "Django checks for ServeStatic are disabled because 'servestatic' is not in "
+ "INSTALLED_APPS. Add 'servestatic' to enable configuration checks.",
+ stacklevel=2,
+ )
+
+ self.get_response = cast("GetResponseCallable", get_response)
autorefresh = getattr(settings, "SERVESTATIC_AUTOREFRESH", debug)
max_age = getattr(settings, "SERVESTATIC_MAX_AGE", 0 if debug else 60)
allow_all_origins = getattr(settings, "SERVESTATIC_ALLOW_ALL_ORIGINS", True)
@@ -56,6 +90,7 @@ def __init__(self, get_response=None, settings=django_settings):
add_headers_function = getattr(settings, "SERVESTATIC_ADD_HEADERS_FUNCTION", None)
self.index_file = getattr(settings, "SERVESTATIC_INDEX_FILE", None)
immutable_file_test = getattr(settings, "SERVESTATIC_IMMUTABLE_FILE_TEST", None)
+ allow_unsafe_symlinks = getattr(settings, "SERVESTATIC_ALLOW_UNSAFE_SYMLINKS", False)
self.use_finders = getattr(settings, "SERVESTATIC_USE_FINDERS", debug)
self.use_manifest = getattr(
settings,
@@ -77,6 +112,7 @@ def __init__(self, get_response=None, settings=django_settings):
add_headers_function=add_headers_function,
index_file=self.index_file,
immutable_file_test=immutable_file_test,
+ allow_unsafe_symlinks=allow_unsafe_symlinks,
)
# Set the static prefix
@@ -116,7 +152,8 @@ async def __call__(self, request):
current_finders = finders.get_finders()
app_dirs = [storage.location for finder in current_finders for storage in finder.storages.values()] # pyright: ignore [reportAttributeAccessIssue]
app_dirs = "\n• ".join(sorted(app_dirs))
- msg = f"ServeStatic did not find the file '{request.path.lstrip(django_settings.STATIC_URL)}' within the following paths:\n• {app_dirs}"
+ missing_path = request.path.removeprefix(django_settings.STATIC_URL)
+ msg = f"ServeStatic did not find the file '{missing_path}' within the following paths:\n• {app_dirs}"
raise MissingFileError(msg)
return await self.get_response(request)
@@ -136,7 +173,7 @@ async def aserve(static_file: StaticFile | Redirect, request: HttpRequest):
return http_response
def add_files_from_finders(self):
- files: dict[str, str] = {}
+ files: dict[str, tuple[str, str | None]] = {}
for finder in finders.get_finders():
for path, storage in finder.list(None):
prefix = (getattr(storage, "prefix", None) or "").strip("/")
@@ -147,12 +184,12 @@ def add_files_from_finders(self):
path.replace("\\", "/"),
))
# Use setdefault as only first matching file should be used
- files.setdefault(url, storage.path(path))
+ files.setdefault(url, (storage.path(path), getattr(storage, "location", None)))
self.insert_directory(storage.location, self.static_prefix)
- stat_cache = stat_files(files.values())
- for url, path in files.items():
- self.add_file_to_dictionary(url, path, stat_cache=stat_cache)
+ stat_cache = stat_files([path for path, _root in files.values()])
+ for url, (path, root) in files.items():
+ self.add_file_to_dictionary(url, path, root=root, stat_cache=stat_cache)
def add_files_from_manifest(self):
if not isinstance(staticfiles_storage, ManifestStaticFilesStorage):
@@ -174,12 +211,14 @@ def add_files_from_manifest(self):
self.add_file_to_dictionary(
f"{self.static_prefix}{original_name}",
staticfiles_storage.path(original_name),
+ root=staticfiles_storage.location,
stat_cache=stat_cache,
)
# Add the hashed file
self.add_file_to_dictionary(
f"{self.static_prefix}{hashed_name}",
staticfiles_storage.path(hashed_name),
+ root=staticfiles_storage.location,
stat_cache=stat_cache,
)
@@ -193,7 +232,7 @@ def candidate_paths_for_url(self, url):
path = url2pathname(relative_url)
normalized_path = normpath(path).lstrip("/")
path = finders.find(normalized_path)
- if path:
+ if path and finder_path_is_allowed(self.directories, url, path, self.path_within_root):
yield path
yield from super().candidate_paths_for_url(url)
diff --git a/src/servestatic/responders.py b/src/servestatic/responders.py
index 6cc4313..33273d1 100644
--- a/src/servestatic/responders.py
+++ b/src/servestatic/responders.py
@@ -162,7 +162,7 @@ def get_range_response(self, range_header, base_headers, file_handle):
msg = "Content-Length header is required for range requests"
raise ValueError(msg)
start, end = self.get_byte_range(range_header, size)
- if start >= end:
+ if start > end:
return self.get_range_not_satisfiable_response(file_handle, size)
if file_handle is not None:
file_handle = SlicedFile(file_handle, start, end)
@@ -185,7 +185,7 @@ async def aget_range_response(self, range_header, base_headers, file_handle):
msg = "Content-Length header is required for range requests"
raise ValueError(msg)
start, end = self.get_byte_range(range_header, size)
- if start >= end:
+ if start > end:
return await self.aget_range_not_satisfiable_response(file_handle, size)
if file_handle is not None:
file_handle = AsyncSlicedFile(file_handle, start, end)
@@ -382,7 +382,7 @@ def stat_regular_file(path, stat_function):
except KeyError as exc:
raise MissingFileError(path) from exc
except OSError as exc:
- if exc.errno in {errno.ENOENT, errno.ENAMETOOLONG}:
+ if exc.errno in {errno.ENOENT, errno.ENAMETOOLONG, errno.ENOTDIR}:
raise MissingFileError(path) from exc
raise
if not stat.S_ISREG(stat_result.st_mode):
diff --git a/src/servestatic/runserver_nostatic/apps.py b/src/servestatic/runserver_nostatic/apps.py
new file mode 100644
index 0000000..0673120
--- /dev/null
+++ b/src/servestatic/runserver_nostatic/apps.py
@@ -0,0 +1,20 @@
+"""Backward-compatible alias app for `servestatic.runserver_nostatic`."""
+
+from __future__ import annotations
+
+from warnings import warn
+
+from servestatic.apps import ServeStaticConfig
+
+
+class ServeStaticRunserverNoStaticAliasConfig(ServeStaticConfig):
+ name = "servestatic.runserver_nostatic"
+ label = "servestatic_runserver_nostatic"
+
+ def ready(self):
+ super().ready()
+ warn(
+ "The 'servestatic.runserver_nostatic' app is deprecated. Use 'servestatic' instead.",
+ DeprecationWarning,
+ stacklevel=2,
+ )
diff --git a/src/servestatic/runserver_nostatic/management/commands/runserver.py b/src/servestatic/runserver_nostatic/management/commands/runserver.py
index ed60ded..75b9484 100644
--- a/src/servestatic/runserver_nostatic/management/commands/runserver.py
+++ b/src/servestatic/runserver_nostatic/management/commands/runserver.py
@@ -1,57 +1,15 @@
-"""
-Subclass the existing 'runserver' command and change the default options
-to disable static file serving, allowing ServeStatic to handle static files.
-
-There is some unpleasant hackery here because we don't know which command class
-to subclass until runtime as it depends on which INSTALLED_APPS we have, so we
-have to determine this dynamically.
-"""
+"""Backward-compatible import path for ``servestatic.runserver_nostatic``."""
from __future__ import annotations
-import contextlib
-from importlib import import_module
-from typing import TYPE_CHECKING, cast
-
-from django.apps import apps
-
-if TYPE_CHECKING:
- from django.core.management.base import BaseCommand
-
-
-def get_next_runserver_command():
- """
- Return the next highest priority "runserver" command class
- """
- for app_name in get_lower_priority_apps():
- module_path = f"{app_name}.management.commands.runserver"
- with contextlib.suppress(ImportError, AttributeError):
- return import_module(module_path).Command
- return None
-
-
-def get_lower_priority_apps():
- """
- Yield all app module names below the current app in the INSTALLED_APPS list
- """
- self_app_name = ".".join(__name__.split(".")[:-3])
- reached_self = False
- for app_config in apps.get_app_configs():
- if app_config.name == self_app_name:
- reached_self = True
- elif reached_self:
- yield app_config.name
- yield "django.core"
-
-
-RunserverCommand = cast("type[BaseCommand]", get_next_runserver_command())
+from servestatic.management.commands import runserver as canonical_runserver
+Command = canonical_runserver.Command
+BaseRunserverCommand = canonical_runserver.BaseRunserverCommand
+find_fallback_runserver_command = canonical_runserver.find_fallback_runserver_command
+iter_lower_priority_apps = canonical_runserver.iter_lower_priority_apps
-class Command(RunserverCommand):
- def add_arguments(self, parser):
- super().add_arguments(parser)
- if not parser.description:
- parser.description = ""
- if parser.get_default("use_static_handler") is True:
- parser.set_defaults(use_static_handler=False)
- parser.description += "\n(Wrapped by 'servestatic.runserver_nostatic' to always enable '--nostatic')"
+# Backward-compatible aliases for previous helper names.
+get_next_runserver_command = find_fallback_runserver_command
+get_lower_priority_apps = iter_lower_priority_apps
+RunserverCommand = BaseRunserverCommand
diff --git a/src/servestatic/storage.py b/src/servestatic/storage.py
index 8154814..780cb54 100644
--- a/src/servestatic/storage.py
+++ b/src/servestatic/storage.py
@@ -23,6 +23,17 @@
_PostProcessT = Iterator[tuple[str, str, bool] | tuple[str, None, RuntimeError]]
+def get_compressor_kwargs(*, quiet: bool) -> dict[str, Any]:
+ return {
+ "extensions": getattr(settings, "SERVESTATIC_SKIP_COMPRESS_EXTENSIONS", None),
+ "use_zstd": getattr(settings, "SERVESTATIC_USE_ZSTD", True),
+ "zstd_dict": getattr(settings, "SERVESTATIC_ZSTD_DICTIONARY", None),
+ "zstd_dict_is_raw": getattr(settings, "SERVESTATIC_ZSTD_DICTIONARY_IS_RAW", False),
+ "zstd_level": getattr(settings, "SERVESTATIC_ZSTD_LEVEL", None),
+ "quiet": quiet,
+ }
+
+
class CompressedStaticFilesStorage(StaticFilesStorage):
"""
StaticFilesStorage subclass that compresses output files.
@@ -34,8 +45,7 @@ def post_process(self, paths: dict[str, Any], dry_run: bool = False, **options:
if dry_run:
return
- extensions = getattr(settings, "SERVESTATIC_SKIP_COMPRESS_EXTENSIONS", None)
- self.compressor = compressor = self.create_compressor(extensions=extensions, quiet=True)
+ self.compressor = compressor = self.create_compressor(**get_compressor_kwargs(quiet=True))
def _compress_path(path: str) -> list[tuple[str, str, bool]]:
compressed: list[tuple[str, str, bool]] = []
@@ -182,8 +192,7 @@ def create_compressor(self, **kwargs): # noqa: PLR6301
return Compressor(**kwargs)
def compress_files(self, paths):
- extensions = getattr(settings, "SERVESTATIC_SKIP_COMPRESS_EXTENSIONS", None)
- self.compressor = compressor = self.create_compressor(extensions=extensions, quiet=True)
+ self.compressor = compressor = self.create_compressor(**get_compressor_kwargs(quiet=True))
def _compress_path(path: str) -> list[tuple[str, str]]:
compressed: list[tuple[str, str]] = []
diff --git a/src/servestatic/utils.py b/src/servestatic/utils.py
index 238affd..1e63914 100644
--- a/src/servestatic/utils.py
+++ b/src/servestatic/utils.py
@@ -5,7 +5,6 @@
import contextlib
import functools
import os
-import threading
from concurrent.futures import ThreadPoolExecutor
from io import IOBase
from typing import TYPE_CHECKING, cast
@@ -74,15 +73,16 @@ def __iter__(self):
thread_executor = concurrent.futures.ThreadPoolExecutor(
max_workers=1, thread_name_prefix="ServeStatic-AsyncFile-Runtime"
)
-
- # Convert from async to sync by stepping through the async iterator and yielding
- # the result of each step.
- generator = self.iterator.__aiter__()
- with contextlib.suppress(GeneratorExit, StopAsyncIteration):
- while True:
- yield thread_executor.submit(loop.run_until_complete, generator.__anext__()).result()
- loop.close()
- thread_executor.shutdown(wait=True)
+ try:
+ # Convert from async to sync by stepping through the async iterator and yielding
+ # the result of each step.
+ generator = self.iterator.__aiter__()
+ with contextlib.suppress(GeneratorExit, StopAsyncIteration):
+ while True:
+ yield thread_executor.submit(loop.run_until_complete, generator.__anext__()).result()
+ finally:
+ loop.close()
+ thread_executor.shutdown(wait=True)
def open_lazy(f):
@@ -131,8 +131,7 @@ def __init__(
)
self.loop: asyncio.AbstractEventLoop | None = None
self.executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="ServeStatic-AsyncFile")
- self.lock = threading.Lock()
- self.file_obj: IOBase = cast("IOBase", None)
+ self.file_obj: IOBase | None = cast("IOBase | None", None)
self.closed = False
self._executor_shutdown = False
@@ -148,22 +147,27 @@ def _shutdown_executor(self):
async def _execute(self, func, *args):
"""Run a function in a dedicated thread (specific to each AsyncFile instance)."""
if self.loop is None:
- self.loop = asyncio.get_event_loop()
- with self.lock:
- return await self.loop.run_in_executor(self.executor, func, *args)
+ self.loop = asyncio.get_running_loop()
+ return await self.loop.run_in_executor(self.executor, func, *args)
async def close(self):
self.closed = True
- if self.file_obj:
+ if self.file_obj is not None:
await self._execute(self.file_obj.close)
self._shutdown_executor()
@open_lazy
async def read(self, size=-1):
+ if self.file_obj is None:
+ msg = "File object was not opened"
+ raise RuntimeError(msg)
return await self._execute(self.file_obj.read, size)
@open_lazy
async def seek(self, offset, whence=0):
+ if self.file_obj is None:
+ msg = "File object was not opened"
+ raise RuntimeError(msg)
return await self._execute(self.file_obj.seek, offset, whence)
@open_lazy
diff --git a/tests/django_settings.py b/tests/django_settings.py
index 49f3281..1cf5c8c 100644
--- a/tests/django_settings.py
+++ b/tests/django_settings.py
@@ -10,7 +10,7 @@
SECRET_KEY = "test_secret"
-INSTALLED_APPS = ["servestatic.runserver_nostatic", "django.contrib.staticfiles"]
+INSTALLED_APPS = ["servestatic", "django.contrib.staticfiles"]
FORCE_SCRIPT_NAME = f"/{AppServer.PREFIX}"
STATIC_URL = f"{FORCE_SCRIPT_NAME}/static/"
diff --git a/tests/test_asgi.py b/tests/test_asgi.py
index b2220f9..eab1a2c 100644
--- a/tests/test_asgi.py
+++ b/tests/test_asgi.py
@@ -3,6 +3,9 @@
import asyncio
import concurrent.futures
import gc
+import os
+import shutil
+import tempfile
from pathlib import Path
import pytest
@@ -160,6 +163,28 @@ def test_async_file_read_raises_after_close():
asyncio.run(async_file.read(1))
+def test_async_file_read_raises_when_lazy_open_returns_none(monkeypatch):
+ async_file = servestatic_utils.AsyncFile(__file__, "rb")
+
+ async def mock_execute(*_args, **_kwargs):
+ await asyncio.sleep(0)
+
+ monkeypatch.setattr(async_file, "_execute", mock_execute)
+ with pytest.raises(RuntimeError, match="not opened"):
+ asyncio.run(async_file.read(1))
+
+
+def test_async_file_seek_raises_when_lazy_open_returns_none(monkeypatch):
+ async_file = servestatic_utils.AsyncFile(__file__, "rb")
+
+ async def mock_execute(*_args, **_kwargs):
+ await asyncio.sleep(0)
+
+ monkeypatch.setattr(async_file, "_execute", mock_execute)
+ with pytest.raises(RuntimeError, match="not opened"):
+ asyncio.run(async_file.seek(0))
+
+
def test_async_file_shutdown_exception_is_ignored(monkeypatch):
async_file = servestatic_utils.AsyncFile(__file__, "rb")
@@ -184,3 +209,51 @@ async def user_app(scope, receive, send):
asyncio.run(app(scope, receive, send))
assert send.status == 200
assert send.body == b"ok"
+
+
+def build_symlink_escape_fixture():
+ tmp_dir = tempfile.mkdtemp()
+ static_dir = Path(tmp_dir) / "static"
+ static_dir.mkdir(parents=True, exist_ok=True)
+
+ outside_content = b"outside-file-marker"
+ outside_path = Path(tmp_dir) / "outside.txt"
+ outside_path.write_bytes(outside_content)
+
+ link_path = static_dir / "link-outside.txt"
+ try:
+ os.symlink(outside_path, link_path)
+ except (OSError, NotImplementedError):
+ shutil.rmtree(tmp_dir)
+ pytest.skip("Symlink creation is unavailable in this environment")
+
+ return Path(tmp_dir), static_dir, outside_content
+
+
+@pytest.mark.parametrize("autorefresh", [True, False])
+def test_asgi_symlink_escape_is_blocked_by_default(autorefresh):
+ tmp_dir, static_dir, _outside_content = build_symlink_escape_fixture()
+ try:
+ app = ServeStaticASGI(None, root=static_dir, autorefresh=autorefresh)
+ scope = AsgiHttpScopeEmulator({"path": "/link-outside.txt"})
+ receive = AsgiReceiveEmulator()
+ send = AsgiSendEmulator()
+ asyncio.run(app(scope, receive, send))
+ assert send.status == 404
+ finally:
+ shutil.rmtree(tmp_dir)
+
+
+@pytest.mark.parametrize("autorefresh", [True, False])
+def test_asgi_symlink_escape_can_be_enabled(autorefresh):
+ tmp_dir, static_dir, outside_content = build_symlink_escape_fixture()
+ try:
+ app = ServeStaticASGI(None, root=static_dir, autorefresh=autorefresh, allow_unsafe_symlinks=True)
+ scope = AsgiHttpScopeEmulator({"path": "/link-outside.txt"})
+ receive = AsgiReceiveEmulator()
+ send = AsgiSendEmulator()
+ asyncio.run(app(scope, receive, send))
+ assert send.status == 200
+ assert send.body == outside_content
+ finally:
+ shutil.rmtree(tmp_dir)
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 821e47c..129e277 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -252,6 +252,18 @@ def test_cli_clear_same_dir(tmp_path, capsys):
assert "cannot be the same" in captured.err
+def test_cli_same_dir_without_clear(tmp_path, capsys):
+ src = tmp_path / "src"
+ src.mkdir()
+ (src / "test.txt").write_text("content")
+
+ with pytest.raises(SystemExit):
+ main(["--manifest", str(src), str(src)])
+
+ captured = capsys.readouterr()
+ assert "cannot be the same" in captured.err
+
+
def test_cli_merge_manifest_success(tmp_path):
src = tmp_path / "src"
src.mkdir()
@@ -393,3 +405,44 @@ def mock_compress(*_args, **_kwargs):
captured = capsys.readouterr()
assert "Error compressing" in captured.out
+
+
+def test_cli_passes_zstd_options_to_compressor(tmp_path, monkeypatch):
+ src = tmp_path / "src"
+ src.mkdir()
+ (src / "test.txt").write_text("content\n" * 1000)
+ dest = tmp_path / "dest"
+ dict_file = tmp_path / "dict.bin"
+ dict_file.write_bytes(b"dict")
+
+ captured_args: dict[str, object] = {}
+
+ class DummyCompressor:
+ def __init__(self, **kwargs):
+ captured_args.update(kwargs)
+
+ @staticmethod
+ def should_compress(_filename):
+ return False
+
+ @staticmethod
+ def compress(_path):
+ return []
+
+ monkeypatch.setattr(servestatic_cli, "Compressor", DummyCompressor)
+
+ main([
+ "--compress",
+ "--zstd-dict",
+ str(dict_file),
+ "--zstd-dict-raw",
+ "--zstd-level",
+ "9",
+ str(src),
+ str(dest),
+ ])
+
+ assert captured_args["use_zstd"] is True
+ assert captured_args["zstd_dict"] == str(dict_file)
+ assert captured_args["zstd_dict_is_raw"] is True
+ assert captured_args["zstd_level"] == 9
diff --git a/tests/test_compress.py b/tests/test_compress.py
index c147f3f..6a1a480 100644
--- a/tests/test_compress.py
+++ b/tests/test_compress.py
@@ -73,7 +73,7 @@ def test_custom_log():
def test_compress():
- compressor = Compressor(use_brotli=False, use_gzip=False)
+ compressor = Compressor(use_brotli=False, use_gzip=False, use_zstd=False)
assert not list(compressor.compress("tests/test_files/static/styles.css"))
@@ -96,3 +96,94 @@ def test_compress_brotli_raises_when_dependency_missing(monkeypatch):
monkeypatch.setattr(compress_module, "brotli", None)
with pytest.raises(RuntimeError, match="Brotli is not installed"):
Compressor.compress_brotli(b"abc")
+
+
+def test_compress_zstd_raises_when_dependency_missing(monkeypatch):
+ monkeypatch.setattr(compress_module, "zstd", None)
+ with pytest.raises(RuntimeError, match="Zstandard is not available"):
+ Compressor.compress_zstd(b"abc")
+
+
+def test_compressor_rejects_dictionary_when_zstd_is_unavailable(monkeypatch):
+ monkeypatch.setattr(compress_module, "zstd", None)
+ with pytest.raises(RuntimeError, match=r"requires Python 3\.14"):
+ Compressor(zstd_dict=b"dict")
+
+
+def test_compress_generates_zstd_with_dictionary(tmp_path, monkeypatch):
+ class FakeZstdDict:
+ def __init__(self, dict_content, is_raw=False):
+ self.dict_content = dict_content
+ self.is_raw = is_raw
+
+ class FakeZstd:
+ def __init__(self):
+ self.last_call = None
+
+ ZstdDict = FakeZstdDict
+
+ def compress(self, data, **kwargs):
+ self.last_call = {"data": data, **kwargs}
+ return b"zstd" + data[:1]
+
+ fake_zstd = FakeZstd()
+ monkeypatch.setattr(compress_module, "zstd", fake_zstd)
+
+ source_path = tmp_path / "styles.css"
+ source_path.write_bytes(b"a" * 1000)
+ dict_path = tmp_path / "dict.bin"
+ dict_path.write_bytes(b"dictionary-bytes")
+
+ compressor = Compressor(
+ use_gzip=False,
+ use_brotli=False,
+ use_zstd=True,
+ zstd_dict=dict_path,
+ zstd_dict_is_raw=True,
+ zstd_level=7,
+ quiet=True,
+ )
+ outputs = compressor.compress(str(source_path))
+
+ assert outputs == [f"{source_path}.zstd"]
+ assert os.path.exists(f"{source_path}.zstd")
+ assert fake_zstd.last_call is not None
+ assert fake_zstd.last_call["level"] == 7
+ assert fake_zstd.last_call["zstd_dict"].dict_content == b"dictionary-bytes"
+ assert fake_zstd.last_call["zstd_dict"].is_raw is True
+
+
+def test_load_zstd_dictionary_raises_when_module_missing(monkeypatch):
+ monkeypatch.setattr(compress_module, "zstd", None)
+ with pytest.raises(RuntimeError, match="Zstandard is not available"):
+ Compressor.load_zstd_dictionary(b"abc")
+
+
+def test_load_zstd_dictionary_from_bytes_uses_zstd_dict(monkeypatch):
+ class FakeZstdDict:
+ def __init__(self, content, is_raw=False):
+ self.content = content
+ self.is_raw = is_raw
+
+ class FakeZstd:
+ ZstdDict = FakeZstdDict
+
+ monkeypatch.setattr(compress_module, "zstd", FakeZstd())
+ loaded = Compressor.load_zstd_dictionary(b"dict-bytes", is_raw=True)
+
+ assert isinstance(loaded, FakeZstdDict)
+ assert loaded.content == b"dict-bytes"
+ assert loaded.is_raw is True
+
+
+def test_load_zstd_dictionary_returns_prebuilt_object(monkeypatch):
+ class FakeZstdDict:
+ pass
+
+ class FakeZstd:
+ ZstdDict = FakeZstdDict
+
+ monkeypatch.setattr(compress_module, "zstd", FakeZstd())
+ sentinel = object()
+
+ assert Compressor.load_zstd_dictionary(sentinel) is sentinel
diff --git a/tests/test_django_servestatic.py b/tests/test_django_servestatic.py
index 5490739..be3ef03 100644
--- a/tests/test_django_servestatic.py
+++ b/tests/test_django_servestatic.py
@@ -2,11 +2,15 @@
import asyncio
import html
+import importlib
+import importlib.util
import os
import shutil
import tempfile
-from contextlib import closing
+import warnings
+from contextlib import closing, suppress
from pathlib import Path
+from typing import Any
from urllib.parse import urljoin, urlparse
import brotli
@@ -17,6 +21,7 @@
from django.contrib.staticfiles import finders, storage
from django.contrib.staticfiles.storage import ManifestStaticFilesStorage
from django.core.asgi import get_asgi_application
+from django.core.checks import run_checks
from django.core.management import call_command
from django.core.wsgi import get_wsgi_application
from django.templatetags.static import static
@@ -37,6 +42,14 @@
Files,
)
+stdlib_zstd: Any | None
+with suppress(ModuleNotFoundError):
+ if importlib.util.find_spec("compression.zstd") is not None:
+ stdlib_zstd = importlib.import_module("compression.zstd")
+
+if "stdlib_zstd" not in globals(): # pragma: no cover
+ stdlib_zstd = None
+
def reset_lazy_object(obj):
obj._wrapped = empty
@@ -188,6 +201,43 @@ def test_get_brotli(server, static_files):
assert response.headers["Vary"] == "Accept-Encoding"
+@pytest.mark.skipif(stdlib_zstd is None, reason="Python 3.14+ zstd module required")
+@pytest.mark.skipif(django.VERSION >= (5, 0), reason="Django <5.0 only")
+@pytest.mark.usefixtures("_collect_static")
+def test_asgi_get_zstd(asgi_application, static_files):
+ assert stdlib_zstd is not None
+ url = storage.staticfiles_storage.url(static_files.js_path)
+ scope = AsgiHttpScopeEmulator({"path": url, "headers": [(b"accept-encoding", b"zstd")]})
+ receive = AsgiReceiveEmulator()
+ send = AsgiSendEmulator()
+ asyncio.run(AsgiAppServer(asgi_application)(scope, receive, send))
+ assert stdlib_zstd.decompress(send.body) == static_files.js_content
+ assert send.headers.get(b"Content-Encoding") == b"zstd"
+ assert send.headers.get(b"Vary") == b"Accept-Encoding"
+
+
+@pytest.mark.skipif(stdlib_zstd is None, reason="Python 3.14+ zstd module required")
+@pytest.mark.skipif(django.VERSION < (5, 0), reason="Django 5.0+ only")
+@pytest.mark.usefixtures("_collect_static")
+def test_asgi_get_zstd_2(asgi_application, static_files):
+ assert stdlib_zstd is not None
+ url = storage.staticfiles_storage.url(static_files.js_path)
+ scope = AsgiHttpScopeEmulator({"path": url, "headers": [(b"accept-encoding", b"zstd")]})
+
+ async def executor():
+ communicator = ApplicationCommunicator(asgi_application, scope)
+ await communicator.send_input(scope)
+ response_start = await communicator.receive_output()
+ response_body = await communicator.receive_output()
+ return response_start | response_body
+
+ response = asyncio.run(executor())
+ headers = dict(response["headers"])
+ assert stdlib_zstd.decompress(response["body"]) == static_files.js_content
+ assert headers.get(b"Content-Encoding") == b"zstd"
+ assert headers.get(b"Vary") == b"Accept-Encoding"
+
+
@pytest.mark.usefixtures("_collect_static")
def test_no_content_type_when_not_modified(server, static_files):
last_mod = "Fri, 11 Apr 2100 11:47:06 GMT"
@@ -312,6 +362,257 @@ def test_middleware_requires_async_get_response():
ServeStaticMiddleware(get_response=lambda request: None)
+@override_settings(MIDDLEWARE=["django.middleware.gzip.GZipMiddleware", "servestatic.middleware.ServeStaticMiddleware"])
+def test_django_check_reports_gzip_middleware_order_error():
+ errors = [error for error in run_checks() if error.id == "servestatic.E001"]
+ assert len(errors) == 1
+
+
+@override_settings(MIDDLEWARE=["servestatic.middleware.ServeStaticMiddleware", "django.middleware.gzip.GZipMiddleware"])
+def test_django_check_accepts_correct_gzip_middleware_order():
+ errors = [error for error in run_checks() if error.id == "servestatic.E001"]
+ assert not errors
+
+
+@pytest.mark.parametrize(
+ ("overrides", "error_id"),
+ [
+ ({"SERVESTATIC_ROOT": 123}, "servestatic.E010"),
+ ({"SERVESTATIC_AUTOREFRESH": "true"}, "servestatic.E011"),
+ ({"SERVESTATIC_USE_MANIFEST": "yes"}, "servestatic.E012"),
+ ({"SERVESTATIC_USE_FINDERS": "yes"}, "servestatic.E013"),
+ ({"SERVESTATIC_MAX_AGE": -1}, "servestatic.E014"),
+ ({"SERVESTATIC_INDEX_FILE": ""}, "servestatic.E015"),
+ ({"SERVESTATIC_MIMETYPES": {".foo": 1}}, "servestatic.E016"),
+ ({"SERVESTATIC_CHARSET": ""}, "servestatic.E017"),
+ ({"SERVESTATIC_ALLOW_ALL_ORIGINS": "yes"}, "servestatic.E018"),
+ ({"SERVESTATIC_SKIP_COMPRESS_EXTENSIONS": "jpg,png"}, "servestatic.E019"),
+ ({"SERVESTATIC_USE_ZSTD": "yes"}, "servestatic.E020"),
+ ({"SERVESTATIC_ZSTD_DICTIONARY": 123}, "servestatic.E021"),
+ ({"SERVESTATIC_ZSTD_DICTIONARY_IS_RAW": "yes"}, "servestatic.E022"),
+ ({"SERVESTATIC_ZSTD_LEVEL": True}, "servestatic.E023"),
+ ({"SERVESTATIC_ADD_HEADERS_FUNCTION": "not-callable"}, "servestatic.E024"),
+ ({"SERVESTATIC_IMMUTABLE_FILE_TEST": "("}, "servestatic.E025"),
+ ({"SERVESTATIC_STATIC_PREFIX": 5}, "servestatic.E026"),
+ ({"SERVESTATIC_KEEP_ONLY_HASHED_FILES": "yes"}, "servestatic.E027"),
+ ({"SERVESTATIC_MANIFEST_STRICT": "yes"}, "servestatic.E028"),
+ ({"SERVESTATIC_ALLOW_UNSAFE_SYMLINKS": "yes"}, "servestatic.E029"),
+ ],
+)
+def test_django_check_reports_invalid_setting_types(overrides, error_id):
+ with override_settings(**overrides):
+ errors = [error for error in run_checks() if error.id == error_id]
+ assert len(errors) == 1
+
+
+@pytest.mark.parametrize(
+ "overrides",
+ [
+ {"SERVESTATIC_ROOT": Path(".")},
+ {"SERVESTATIC_MAX_AGE": None},
+ {"SERVESTATIC_INDEX_FILE": True},
+ {"SERVESTATIC_INDEX_FILE": "index.html"},
+ {"SERVESTATIC_MIMETYPES": {"special-file": "text/plain"}},
+ {"SERVESTATIC_CHARSET": "utf-8"},
+ {"SERVESTATIC_SKIP_COMPRESS_EXTENSIONS": ["jpg", "png"]},
+ {"SERVESTATIC_ZSTD_DICTIONARY": b"dict-bytes"},
+ {"SERVESTATIC_ZSTD_LEVEL": 3},
+ {"SERVESTATIC_ADD_HEADERS_FUNCTION": lambda *_: None},
+ {"SERVESTATIC_IMMUTABLE_FILE_TEST": r"^.+\\.[0-9a-f]{12}\\..+$"},
+ {"SERVESTATIC_IMMUTABLE_FILE_TEST": lambda *_: True},
+ {"SERVESTATIC_STATIC_PREFIX": "/static/"},
+ {"SERVESTATIC_ALLOW_UNSAFE_SYMLINKS": True},
+ ],
+)
+def test_django_check_accepts_valid_setting_values(overrides):
+ with override_settings(**overrides):
+ errors = [
+ error
+ for error in run_checks()
+ if error.id and error.id.startswith("servestatic.E0") and error.id != "servestatic.E001"
+ ]
+ assert not errors
+
+
+def test_django_check_reports_skip_compress_extensions_non_string_items():
+ with override_settings(SERVESTATIC_SKIP_COMPRESS_EXTENSIONS=["jpg", 1]):
+ errors = [error for error in run_checks() if error.id == "servestatic.E019"]
+ assert len(errors) == 1
+
+
+def test_django_check_reports_mimetypes_non_mapping_value():
+ with override_settings(SERVESTATIC_MIMETYPES=[(".foo", "text/plain")]):
+ errors = [error for error in run_checks() if error.id == "servestatic.E016"]
+ assert len(errors) == 1
+
+
+def test_django_check_reports_mimetypes_non_string_key():
+ with override_settings(SERVESTATIC_MIMETYPES={1: "text/plain"}):
+ errors = [error for error in run_checks() if error.id == "servestatic.E016"]
+ assert len(errors) == 1
+
+
+def test_django_check_accepts_boolean_setting_values():
+ with override_settings(SERVESTATIC_AUTOREFRESH=True):
+ errors = [error for error in run_checks() if error.id == "servestatic.E011"]
+ assert not errors
+
+
+@pytest.mark.skipif(stdlib_zstd is None, reason="Python 3.14+ zstd module required")
+def test_django_check_accepts_zstd_dictionary_object():
+ assert stdlib_zstd is not None
+ dictionary = stdlib_zstd.ZstdDict(b"servestatic-checks-dict", is_raw=True)
+ with override_settings(SERVESTATIC_ZSTD_DICTIONARY=dictionary):
+ errors = [error for error in run_checks() if error.id == "servestatic.E021"]
+ assert not errors
+
+
+def test_django_check_reports_immutable_file_test_invalid_type():
+ with override_settings(SERVESTATIC_IMMUTABLE_FILE_TEST=object()):
+ errors = [error for error in run_checks() if error.id == "servestatic.E025"]
+ assert len(errors) == 1
+
+
+def test_middleware_warns_when_servestatic_app_is_missing(async_middleware_response):
+ class Settings:
+ DEBUG = True
+ INSTALLED_APPS = ["some.other.app", "django.contrib.staticfiles"]
+ SERVESTATIC_AUTOREFRESH = False
+ SERVESTATIC_USE_MANIFEST = False
+ SERVESTATIC_USE_FINDERS = False
+ SERVESTATIC_STATIC_PREFIX = "/static/"
+ STATIC_URL = "/static/"
+ STATIC_ROOT = None
+
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ ServeStaticMiddleware(get_response=async_middleware_response, settings=Settings)
+ assert any("checks for ServeStatic are disabled" in str(item.message) for item in caught)
+
+
+def test_middleware_does_not_warn_for_legacy_servestatic_app(async_middleware_response):
+ class Settings:
+ DEBUG = True
+ INSTALLED_APPS = ["servestatic.runserver_nostatic", "django.contrib.staticfiles"]
+ SERVESTATIC_AUTOREFRESH = False
+ SERVESTATIC_USE_MANIFEST = False
+ SERVESTATIC_USE_FINDERS = False
+ SERVESTATIC_STATIC_PREFIX = "/static/"
+ STATIC_URL = "/static/"
+ STATIC_ROOT = None
+
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ ServeStaticMiddleware(get_response=async_middleware_response, settings=Settings)
+ assert all("checks for ServeStatic are disabled" not in str(item.message) for item in caught)
+
+
+def build_symlink_escape_fixture():
+ tmp_dir = tempfile.mkdtemp()
+ static_dir = os.path.join(tmp_dir, "static")
+ os.makedirs(static_dir, exist_ok=True)
+
+ outside_content = b"outside-file-marker"
+ outside_path = os.path.join(tmp_dir, "outside.txt")
+ with open(outside_path, "wb") as outside_file:
+ outside_file.write(outside_content)
+
+ link_path = os.path.join(static_dir, "link-outside.txt")
+ try:
+ os.symlink(outside_path, link_path)
+ except (OSError, NotImplementedError):
+ shutil.rmtree(tmp_dir)
+ pytest.skip("Symlink creation is unavailable in this environment")
+
+ return tmp_dir, static_dir
+
+
+def build_dummy_request(path):
+ request_class = type(
+ "DummyRequest",
+ (),
+ {
+ "path_info": path,
+ "method": "GET",
+ "META": {"HTTP_ACCEPT_ENCODING": "identity"},
+ "path": path,
+ },
+ )
+ return request_class()
+
+
+def test_middleware_blocks_symlink_escape_by_default(async_middleware_response):
+ tmp_dir, static_dir = build_symlink_escape_fixture()
+ try:
+
+ class Settings:
+ DEBUG = False
+ INSTALLED_APPS = ["servestatic", "django.contrib.staticfiles"]
+ SERVESTATIC_ROOT = static_dir
+ SERVESTATIC_AUTOREFRESH = True
+ SERVESTATIC_USE_MANIFEST = False
+ SERVESTATIC_USE_FINDERS = False
+ SERVESTATIC_STATIC_PREFIX = "/"
+ STATIC_URL = "/"
+ STATIC_ROOT = None
+
+ middleware = ServeStaticMiddleware(get_response=async_middleware_response, settings=Settings)
+ response = asyncio.run(middleware(build_dummy_request("/link-outside.txt")))
+ assert response is None
+ finally:
+ shutil.rmtree(tmp_dir)
+
+
+def test_middleware_can_enable_symlink_escape(async_middleware_response):
+ tmp_dir, static_dir = build_symlink_escape_fixture()
+ try:
+
+ class Settings:
+ DEBUG = False
+ INSTALLED_APPS = ["servestatic", "django.contrib.staticfiles"]
+ SERVESTATIC_ROOT = static_dir
+ SERVESTATIC_AUTOREFRESH = True
+ SERVESTATIC_ALLOW_UNSAFE_SYMLINKS = True
+ SERVESTATIC_USE_MANIFEST = False
+ SERVESTATIC_USE_FINDERS = False
+ SERVESTATIC_STATIC_PREFIX = "/"
+ STATIC_URL = "/"
+ STATIC_ROOT = None
+
+ middleware = ServeStaticMiddleware(get_response=async_middleware_response, settings=Settings)
+ response = asyncio.run(middleware(build_dummy_request("/link-outside.txt")))
+ assert response is not None
+ assert response.status_code == 200
+ finally:
+ shutil.rmtree(tmp_dir)
+
+
+def test_middleware_blocks_finder_symlink_escape_in_autorefresh(monkeypatch, async_middleware_response):
+ tmp_dir, static_dir = build_symlink_escape_fixture()
+ link_path = os.path.join(static_dir, "link-outside.txt")
+ try:
+
+ class Settings:
+ DEBUG = False
+ INSTALLED_APPS = ["servestatic", "django.contrib.staticfiles"]
+ SERVESTATIC_ROOT = static_dir
+ SERVESTATIC_AUTOREFRESH = True
+ SERVESTATIC_USE_MANIFEST = False
+ SERVESTATIC_USE_FINDERS = False
+ SERVESTATIC_STATIC_PREFIX = "/"
+ STATIC_URL = "/"
+ STATIC_ROOT = None
+
+ middleware = ServeStaticMiddleware(get_response=async_middleware_response, settings=Settings)
+ middleware.use_finders = True
+ monkeypatch.setattr(middleware_module.finders, "find", lambda _path: link_path)
+
+ response = asyncio.run(middleware(build_dummy_request("/link-outside.txt")))
+ assert response is None
+ finally:
+ shutil.rmtree(tmp_dir)
+
+
def test_add_files_from_manifest_raises_when_storage_has_no_manifest(async_middleware_response):
class Settings:
DEBUG = False
@@ -496,6 +797,16 @@ def test_error_message(server):
assert str(Path(__file__).parent / "test_files" / "static") in response_content
+@override_settings(DEBUG=True)
+def test_error_message_preserves_missing_path_after_prefix(server):
+ response = server.get(f"{settings.STATIC_URL}static.css")
+ response_content = str(response.content.decode())
+ response_content = html.unescape(response_content)
+ response_content = response_content[response_content.index("ServeStatic") :]
+
+ assert "ServeStatic did not find the file 'static.css' within the following paths:" in response_content
+
+
@override_settings(FORCE_SCRIPT_NAME="/subdir", STATIC_URL="static/")
@pytest.mark.usefixtures("_collect_static")
def test_force_script_name(server, static_files):
@@ -518,18 +829,12 @@ def test_force_script_name_with_matching_static_url(server, static_files):
@pytest.mark.usefixtures("_collect_static")
def test_range_response(server, static_files):
- ...
- # FIXME: This test is not working, seemingly due to bugs with AppServer.
-
- # url = storage.staticfiles_storage.url(static_files.js_path)
- # response = server.get(url, headers={"Range": "bytes=0-13"})
- # assert response.content == static_files.js_content[:14]
- # assert response.status_code == 206
- # assert (
- # response.headers["Content-Range"]
- # == f"bytes 0-13/{len(static_files.js_content)}"
- # )
- # assert response.headers["Content-Length"] == "14"
+ url = storage.staticfiles_storage.url(static_files.js_path)
+ response = server.get(url, headers={"Range": "bytes=0-13", "Accept-Encoding": "identity"})
+ assert response.content == static_files.js_content[:14]
+ assert response.status_code == 206
+ assert response.headers["Content-Range"] == f"bytes 0-13/{len(static_files.js_content)}"
+ assert response.headers["Content-Length"] == "14"
@pytest.mark.skipif(django.VERSION >= (5, 0), reason="Django <5.0 only")
diff --git a/tests/test_runserver_nostatic.py b/tests/test_runserver_nostatic.py
index 9076315..c8f11c4 100644
--- a/tests/test_runserver_nostatic.py
+++ b/tests/test_runserver_nostatic.py
@@ -3,6 +3,7 @@
import argparse
from types import SimpleNamespace
+import pytest
from django.core.management import get_commands, load_command_class
@@ -11,7 +12,13 @@ def get_command_instance(name):
return load_command_class(app_name, name)
-def get_runserver_module():
+def get_canonical_runserver_module():
+ import servestatic.management.commands.runserver as runserver_module
+
+ return runserver_module
+
+
+def get_legacy_alias_runserver_module():
import servestatic.runserver_nostatic.management.commands.runserver as runserver_module
return runserver_module
@@ -20,38 +27,67 @@ def get_runserver_module():
def test_command_output():
command = get_command_instance("runserver")
parser = command.create_parser("manage.py", "runserver")
- assert "Wrapped by 'servestatic.runserver_nostatic'" in parser.format_help()
+ assert "Wrapped by 'servestatic'" in parser.format_help()
assert not parser.get_default("use_static_handler")
-def test_get_next_runserver_command_returns_none_when_no_imports_work(monkeypatch):
- runserver_module = get_runserver_module()
- monkeypatch.setattr(runserver_module, "get_lower_priority_apps", lambda: iter(["missing.app"]))
+def test_find_fallback_runserver_command_returns_none_when_no_imports_work(monkeypatch):
+ runserver_module = get_canonical_runserver_module()
+ monkeypatch.setattr(runserver_module, "iter_lower_priority_apps", lambda: iter(["missing.app"]))
def raise_import_error(_module_path):
raise ImportError
monkeypatch.setattr(runserver_module, "import_module", raise_import_error)
- assert runserver_module.get_next_runserver_command() is None
+ assert runserver_module.find_fallback_runserver_command() is None
-def test_get_lower_priority_apps_yields_remaining_apps_and_django_core(monkeypatch):
- runserver_module = get_runserver_module()
- self_app_name = ".".join(runserver_module.__name__.split(".")[:-3])
+def test_iter_lower_priority_apps_yields_remaining_apps_and_django_core(monkeypatch):
+ runserver_module = get_canonical_runserver_module()
app_configs = [
SimpleNamespace(name="before.self"),
- SimpleNamespace(name=self_app_name),
+ SimpleNamespace(name=runserver_module.SERVESTATIC_APP_NAME),
SimpleNamespace(name="after.self"),
]
monkeypatch.setattr(runserver_module.apps, "get_app_configs", lambda: app_configs)
- assert list(runserver_module.get_lower_priority_apps()) == ["after.self", "django.core"]
+ assert list(runserver_module.iter_lower_priority_apps()) == ["after.self", "django.core"]
def test_command_add_arguments_handles_empty_description_without_toggling_false_default(monkeypatch):
- runserver_module = get_runserver_module()
- monkeypatch.setattr(runserver_module.RunserverCommand, "add_arguments", lambda self, parser: None)
+ runserver_module = get_canonical_runserver_module()
+ monkeypatch.setattr(runserver_module.BaseRunserverCommand, "add_arguments", lambda self, parser: None)
parser = argparse.ArgumentParser(description=None)
parser.set_defaults(use_static_handler=False)
runserver_module.Command().add_arguments(parser)
assert parser.description == ""
assert parser.get_default("use_static_handler") is False
+
+
+def test_legacy_module_aliases_canonical_command_exports():
+ canonical = get_canonical_runserver_module()
+ alias = get_legacy_alias_runserver_module()
+
+ assert alias.Command is canonical.Command
+ assert alias.RunserverCommand is canonical.BaseRunserverCommand
+ assert alias.get_next_runserver_command is canonical.find_fallback_runserver_command
+ assert alias.get_lower_priority_apps is canonical.iter_lower_priority_apps
+
+
+def test_legacy_alias_app_config_points_to_runserver_nostatic_path():
+ from servestatic.apps import ServeStaticConfig
+ from servestatic.runserver_nostatic.apps import ServeStaticRunserverNoStaticAliasConfig
+
+ assert issubclass(ServeStaticRunserverNoStaticAliasConfig, ServeStaticConfig)
+ assert ServeStaticRunserverNoStaticAliasConfig.name == "servestatic.runserver_nostatic"
+ assert ServeStaticRunserverNoStaticAliasConfig.label == "servestatic_runserver_nostatic"
+
+
+def test_legacy_alias_app_config_ready_emits_deprecation_warning(monkeypatch):
+ from servestatic.apps import ServeStaticConfig
+ from servestatic.runserver_nostatic.apps import ServeStaticRunserverNoStaticAliasConfig
+
+ monkeypatch.setattr(ServeStaticConfig, "ready", lambda _self: None)
+ app_config = object.__new__(ServeStaticRunserverNoStaticAliasConfig)
+
+ with pytest.warns(DeprecationWarning, match="'servestatic.runserver_nostatic' app is deprecated"):
+ app_config.ready()
diff --git a/tests/test_servestatic.py b/tests/test_servestatic.py
index 0a98084..4e265e5 100644
--- a/tests/test_servestatic.py
+++ b/tests/test_servestatic.py
@@ -341,6 +341,53 @@ def copytree(src, dst):
shutil.copy2(src_path, dst_path)
+def build_symlink_escape_fixture():
+ tmp_dir = tempfile.mkdtemp()
+ static_dir = os.path.join(tmp_dir, "static")
+ os.makedirs(static_dir, exist_ok=True)
+
+ outside_content = b"outside-file-marker"
+ outside_path = os.path.join(tmp_dir, "outside.txt")
+ with open(outside_path, "wb") as outside_file:
+ outside_file.write(outside_content)
+
+ link_path = os.path.join(static_dir, "link-outside.txt")
+ try:
+ os.symlink(outside_path, link_path)
+ except (OSError, NotImplementedError):
+ shutil.rmtree(tmp_dir)
+ pytest.skip("Symlink creation is unavailable in this environment")
+
+ return tmp_dir, static_dir, outside_content
+
+
+@pytest.mark.parametrize("autorefresh", [True, False])
+def test_symlink_escape_is_blocked_by_default(autorefresh):
+ tmp_dir, static_dir, _outside_content = build_symlink_escape_fixture()
+ try:
+ app = ServeStatic(None, root=static_dir, autorefresh=autorefresh)
+ app_server = AppServer(app)
+ with closing(app_server):
+ response = app_server.get(f"/{AppServer.PREFIX}/link-outside.txt")
+ assert response.status_code == 404
+ finally:
+ shutil.rmtree(tmp_dir)
+
+
+@pytest.mark.parametrize("autorefresh", [True, False])
+def test_symlink_escape_can_be_enabled(autorefresh):
+ tmp_dir, static_dir, outside_content = build_symlink_escape_fixture()
+ try:
+ app = ServeStatic(None, root=static_dir, autorefresh=autorefresh, allow_unsafe_symlinks=True)
+ app_server = AppServer(app)
+ with closing(app_server):
+ response = app_server.get(f"/{AppServer.PREFIX}/link-outside.txt")
+ assert response.status_code == 200
+ assert response.content == outside_content
+ finally:
+ shutil.rmtree(tmp_dir)
+
+
def test_immutable_file_test_accepts_regex():
instance = ServeStatic(None, immutable_file_test=r"\.test$")
assert instance.immutable_file_test("", "/myfile.test")
@@ -406,6 +453,27 @@ def test_url_is_canonical_rejects_backslashes():
assert not DummyServeStaticBase.url_is_canonical(r"/static\file.js")
+def test_path_is_within_returns_false_when_commonpath_raises_value_error():
+ assert not DummyServeStaticBase._path_is_within("/tmp/root", "relative/path")
+
+
+def test_path_within_root_normalizes_drive_root_separator(monkeypatch):
+ app = DummyServeStaticBase(None)
+
+ def fake_path_is_within(root, path):
+ return root == "C:\\" and path.startswith("C:\\")
+
+ monkeypatch.setattr(app, "_path_is_within", fake_path_is_within)
+ monkeypatch.setattr(os.path, "realpath", lambda value: value)
+
+ assert app.path_within_root("C:\\", "C:\\static\\file.js")
+
+
+def test_is_compressed_variant_detects_zstd_suffix_with_cache():
+ cache = {"/tmp/app.js": fake_stat_entry(st_mtime=1)}
+ assert DummyServeStaticBase.is_compressed_variant("/tmp/app.js.zstd", stat_cache=cache)
+
+
def test_immutable_file_test_supports_callable():
app = DummyServeStaticBase(None, immutable_file_test=lambda path, url: url.endswith(".ok"))
assert app.immutable_file_test("ignored", "/file.ok")
@@ -431,6 +499,18 @@ def test_get_static_file_omits_allow_all_origins_header_when_disabled():
assert "Access-Control-Allow-Origin" not in headers
+def test_get_static_file_prefers_zstd_when_requested():
+ app = DummyServeStaticBase(None)
+ stat_cache = {
+ __file__: fake_stat_entry(st_size=1000, st_mtime=1),
+ f"{__file__}.zstd": fake_stat_entry(st_size=200, st_mtime=1),
+ }
+ static_file = app.get_static_file(__file__, "/coverage.py", stat_cache=stat_cache)
+ path, headers = static_file.get_path_and_headers({"HTTP_ACCEPT_ENCODING": "zstd, gzip"})
+ assert path == f"{__file__}.zstd"
+ assert Headers(headers)["Content-Encoding"] == "zstd"
+
+
def test_last_modified_not_set_when_mtime_is_zero():
stat_cache = {__file__: fake_stat_entry()}
responder = StaticFile(__file__, [], stat_cache=stat_cache)
@@ -535,6 +615,17 @@ def test_file_size_matches_range_with_range_header():
assert file_size == 14
+def test_single_byte_range_is_supported():
+ stat_cache = {__file__: fake_stat_entry()}
+ responder = StaticFile(__file__, [], stat_cache=stat_cache)
+ response = responder.get_response("GET", {"HTTP_RANGE": "bytes=0-0"})
+ assert int(response.status) == 206
+ assert response.file is not None
+ with open(__file__, "rb") as source:
+ assert response.file.read() == source.read(1)
+ response.file.close()
+
+
def test_chunked_file_size_matches_range_with_range_header():
stat_cache = {__file__: fake_stat_entry()}
responder = StaticFile(__file__, [], stat_cache=stat_cache)
diff --git a/tests/test_storage.py b/tests/test_storage.py
index 56a2f81..1d71e2c 100644
--- a/tests/test_storage.py
+++ b/tests/test_storage.py
@@ -18,6 +18,13 @@
from .utils import Files
+try:
+ from compression import zstd as _zstd # noqa: F401
+
+ HAS_ZSTD = True
+except ImportError: # pragma: no cover
+ HAS_ZSTD = False
+
@pytest.fixture
def setup():
@@ -65,7 +72,11 @@ def _compressed_manifest_storage(setup):
def test_compressed_static_files_storage():
call_command("collectstatic", verbosity=0, interactive=False)
- for name in ["styles.css.gz", "styles.css.br"]:
+ expected_files = ["styles.css.gz", "styles.css.br"]
+ if HAS_ZSTD:
+ expected_files.append("styles.css.zstd")
+
+ for name in expected_files:
path = os.path.join(settings.STATIC_ROOT, name)
assert os.path.exists(path)
@@ -74,7 +85,11 @@ def test_compressed_static_files_storage():
def test_compressed_static_files_storage_dry_run():
call_command("collectstatic", "--dry-run", verbosity=0, interactive=False)
- for name in ["styles.css.gz", "styles.css.br"]:
+ expected_files = ["styles.css.gz", "styles.css.br"]
+ if HAS_ZSTD:
+ expected_files.append("styles.css.zstd")
+
+ for name in expected_files:
path = os.path.join(settings.STATIC_ROOT, name)
assert not os.path.exists(path)
diff --git a/tests/utils.py b/tests/utils.py
index a77d156..ba2f5b9 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -1,12 +1,11 @@
from __future__ import annotations
import os
-import threading
from collections import UserDict
-from wsgiref.simple_server import WSGIRequestHandler, make_server
+from urllib.parse import quote
from wsgiref.util import shift_path_info
-import requests
+import httpx
TEST_FILE_PATH = os.path.join(os.path.dirname(__file__), "test_files")
@@ -18,12 +17,25 @@ class AppServer:
"""
PREFIX = "subdir"
+ DEFAULT_ACCEPT_ENCODING = "gzip, deflate, br"
def __init__(self, application):
self.application = application
- self.server = make_server("127.0.0.1", 0, self.serve_under_prefix, handler_class=WSGIRequestHandler)
+ self.client = httpx.Client(
+ transport=httpx.WSGITransport(app=self.serve_under_prefix, raise_app_exceptions=False),
+ base_url="http://testserver",
+ )
def serve_under_prefix(self, environ, start_response):
+ path_info = environ.get("PATH_INFO", "")
+ try:
+ path_info.encode("iso-8859-1")
+ path_info = ""
+ except UnicodeEncodeError:
+ # WSGI servers expose PATH_INFO as latin-1 decoded bytes. Recreate
+ # that shape for non-ASCII paths so Django can recover UTF-8 bytes.
+ environ["PATH_INFO"] = path_info.encode("utf-8").decode("iso-8859-1")
+
prefix = shift_path_info(environ)
if prefix == self.PREFIX:
return self.application(environ, start_response)
@@ -34,17 +46,52 @@ def get(self, *args, **kwargs):
return self.request("get", *args, **kwargs)
def request(self, method, path, *args, **kwargs):
- domain = self.server.server_address[0]
- port = self.server.server_address[1]
- url = f"http://{domain}:{port}{path}"
- thread = threading.Thread(target=self.server.handle_request)
- thread.start()
- response = requests.request(method, url, *args, **kwargs, timeout=5)
- thread.join()
- return response
+ # Keep compatibility with previous requests-based helpers.
+ allow_redirects = kwargs.pop("allow_redirects", True)
+ headers = dict(kwargs.pop("headers", {}) or {})
+
+ # requests percent-encodes non-ASCII URL paths before they hit WSGI.
+ path = quote(path, safe="/%?=&:+,;@#")
+
+ # Always send Accept-Encoding unless tests explicitly set it to None.
+ has_accept_encoding = any(key.lower() == "accept-encoding" for key in headers)
+ if not has_accept_encoding:
+ headers["Accept-Encoding"] = self.DEFAULT_ACCEPT_ENCODING
+
+ # Preserve existing test semantics where `None` means "no accepted encodings".
+ for key, value in list(headers.items()):
+ if key.lower() == "accept-encoding" and value is None:
+ headers[key] = ""
+
+ # A value of None means "omit this header" in existing tests.
+ headers = {key: value for key, value in headers.items() if value is not None}
+
+ response = self.client.request(
+ method,
+ path,
+ *args,
+ headers=headers,
+ follow_redirects=allow_redirects,
+ **kwargs,
+ )
+ return _ResponseAdapter(response)
def close(self):
- self.server.server_close()
+ self.client.close()
+
+
+class _ResponseAdapter:
+ """Compat adapter that presents a requests-like interface over httpx responses."""
+
+ def __init__(self, response: httpx.Response):
+ self._response = response
+
+ def __getattr__(self, item):
+ return getattr(self._response, item)
+
+ @property
+ def url(self):
+ return str(self._response.url)
class AsgiAppServer: