patx/wred
Propagate static file errors to applications and ASGI servers
Commit c97bae7 · patx · 2026-10-18T12:53:11-04:00
Propagate static file errors to applications and ASGI servers Preserve root containment with resolved-path validation while allowing file and path exceptions to propagate. Document the behavior and update core and example tests to verify exceptions escape without an HTTP response.
Comments
No comments yet.
Diff
diff --git a/README.md b/README.md
index da9292e..53b4e70 100644
--- a/README.md
+++ b/README.md
@@ -156,8 +156,14 @@ async def asset(request, filename):
`static(filename, root='static')` returns `(bytes, content_type, status)`.
Omitting `root`, passing `''`, or passing `None` uses the `static` directory in the
current working directory. Set `root` to a string or `Path` to use another directory.
-MIME types are guessed; missing, unreadable, or outside-root files return
-JSON 404 (`{"error":"Not Found"}`), including symlink escapes. Files are read in a worker thread, entirely into memory.
+MIME types are guessed. Files are read in a worker thread, entirely into memory.
+All file and path errors propagate to your handler, middleware, or ASGI server.
+Missing files raise `FileNotFoundError`, directories raise `IsADirectoryError`,
+and permission or other filesystem failures retain their original exceptions.
+Resolved paths outside the configured root, including symlink escapes, raise
+`ValueError` before any file is read. Invalid paths can also raise `ValueError`.
+The helper does not turn these errors into HTTP responses; catch exceptions in
+your application or middleware to return a 404 or another status.
Small files. Don't feed it a movie.
## Errors
diff --git a/tests/test_notebook.py b/tests/test_notebook.py
index efcd2ee..ffabb21 100644
--- a/tests/test_notebook.py
+++ b/tests/test_notebook.py
@@ -150,7 +150,8 @@ class NotebookTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual(
(await self.request('/api/health'))[:2], (200, {'status': 'ok'})
)
- self.assertEqual((await self.request('/static/missing.js'))[0], 404)
+ with self.assertRaises(FileNotFoundError):
+ await self.request('/static/missing.js')
async def test_lifespan_starts_and_stops_the_database(self):
worker = self.new_app()
diff --git a/tests/test_wred.py b/tests/test_wred.py
index 413d585..ee8deee 100644
--- a/tests/test_wred.py
+++ b/tests/test_wred.py
@@ -187,44 +187,19 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
return await app.static(data['filename'], **root_options)
cases = [
- ('index.html', 200, b'text/html', b'<p>Hello</p>'),
+ ('index.html', None, b'text/html', b'<p>Hello</p>'),
(
'asset.unknownextension',
- 200,
+ None,
b'application/octet-stream',
b'\x00\xff',
),
- (
- 'missing.txt',
- 404,
- b'application/json',
- b'{"error":"Not Found"}',
- ),
- ('.', 404, b'application/json', b'{"error":"Not Found"}'),
- (
- '../secret.txt',
- 404,
- b'application/json',
- b'{"error":"Not Found"}',
- ),
- (
- str(base / 'secret.txt'),
- 404,
- b'application/json',
- b'{"error":"Not Found"}',
- ),
- (
- 'escape.txt',
- 404,
- b'application/json',
- b'{"error":"Not Found"}',
- ),
- (
- 'invalid\x00.txt',
- 404,
- b'application/json',
- b'{"error":"Not Found"}',
- ),
+ ('missing.txt', FileNotFoundError, None, None),
+ ('.', IsADirectoryError, None, None),
+ ('../secret.txt', ValueError, None, None),
+ (str(base / 'secret.txt'), ValueError, None, None),
+ ('escape.txt', ValueError, None, None),
+ ('invalid\x00.txt', ValueError, None, None),
]
cases = [
(options, *case)
@@ -241,7 +216,7 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
(
{'root': custom_root},
'index.html',
- 200,
+ None,
b'text/html',
b'<p>Custom root</p>',
)
@@ -250,7 +225,7 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
for (
root_options,
filename,
- status,
+ error,
content_type,
body,
) in cases:
@@ -262,21 +237,30 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
'body': orjson.dumps({'filename': filename}),
}
+ scope = {
+ 'type': 'http',
+ 'path': '/file',
+ 'method': 'POST',
+ }
+ if error is not None:
+ messages = []
+
+ async def send(message):
+ messages.append(message)
+
+ with self.assertRaises(error):
+ await app(scope, receive, send)
+ self.assertEqual(messages, [])
+ continue
messages = await call_app(
- app,
- {
- 'type': 'http',
- 'path': '/file',
- 'method': 'POST',
- },
- receive,
+ app, scope, receive
)
self.assertEqual(
messages,
[
{
'type': 'http.response.start',
- 'status': status,
+ 'status': 200,
'headers': [
(b'content-type', content_type)
],
@@ -668,7 +652,7 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
await app({'type': protocol}, receive, send)
self.assertEqual(messages, [])
- async def test_file_read_failure_returns_404(self):
+ async def test_file_read_failure_propagates_without_sending(self):
app = App()
with tempfile.TemporaryDirectory() as directory:
(Path(directory) / 'private.txt').write_text('private')
@@ -682,30 +666,23 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
'File serving should not read the request body'
)
+ messages = []
+
+ async def send(message):
+ messages.append(message)
+
# Simulate the OS refusing a real file read, including when tests
# run as root.
with patch.object(
Path, 'read_bytes', side_effect=PermissionError('denied')
):
- messages = await call_app(
- app,
- {'type': 'http', 'path': '/', 'method': 'GET'},
- receive,
- )
- self.assertEqual(
- messages,
- [
- {
- 'type': 'http.response.start',
- 'status': 404,
- 'headers': [(b'content-type', b'application/json')],
- },
- {
- 'type': 'http.response.body',
- 'body': b'{"error":"Not Found"}',
- },
- ],
- )
+ with self.assertRaisesRegex(PermissionError, 'denied'):
+ await app(
+ {'type': 'http', 'path': '/', 'method': 'GET'},
+ receive,
+ send,
+ )
+ self.assertEqual(messages, [])
async def test_raw_and_custom_response_bodies(self):
app = App()
diff --git a/wred.py b/wred.py
index cfa7110..ac175bf 100644
--- a/wred.py
+++ b/wred.py
@@ -113,16 +113,12 @@ class App:
return params
async def static(self, filename: str | Path, *, root: str | Path | None = 'static') -> tuple[bytes, str, int]:
- """Serve a small file inside root (default: ./static), off the event loop."""
+ """Read a small file inside root off the event loop; errors propagate."""
def read() -> tuple[bytes, str, int]:
- try:
- directory = Path(root or 'static').resolve()
- path = (directory / filename).resolve()
- if not path.is_relative_to(directory) or not path.is_file():
- return b'{"error":"Not Found"}', 'application/json', 404
- return path.read_bytes(), guess_type(path.name)[0] or 'application/octet-stream', 200
- except (OSError, ValueError, RuntimeError):
- return b'{"error":"Not Found"}', 'application/json', 404
+ directory = Path(root or 'static').resolve()
+ path = (directory / filename).resolve()
+ path.relative_to(directory)
+ return path.read_bytes(), guess_type(path.name)[0] or 'application/octet-stream', 200
return await to_thread(read)
def route(self, path: str, methods: list[str] | tuple[str, ...] = ('GET',)) -> Callable: