patx/wred
Reject invalid UTF-8 consistently in query parameters
Commit be6ffa1 · patx · 2026-10-18T01:01:47-04:00
Comments
No comments yet.
Diff
diff --git a/README.md b/README.md
index 84d88c6..ccafe98 100644
--- a/README.md
+++ b/README.md
@@ -163,7 +163,7 @@ Small files. Don't feed it a movie.
## Errors
Exceptions propagate to your ASGI server or middleware. Malformed JSON raises
-`orjson.JSONDecodeError`, invalid raw UTF-8 query bytes raise `UnicodeDecodeError`,
+`orjson.JSONDecodeError`, invalid UTF-8 query bytes (raw or percent-encoded) raise `UnicodeDecodeError`,
client disconnects during body reads raise `ConnectionError`, and unsupported
response types raise `TypeError`. Wred does not turn exceptions into responses.
Malformed response tuples can raise `ValueError` during unpacking.
diff --git a/tests/test_wred.py b/tests/test_wred.py
index 434e6d9..9c8b482 100644
--- a/tests/test_wred.py
+++ b/tests/test_wred.py
@@ -233,11 +233,13 @@ class WredTests(unittest.IsolatedAsyncioTestCase):
cases = [
('POST', '/input', b'{', b'', None, orjson.JSONDecodeError),
('GET', '/search', b'', b'\xff', None, UnicodeDecodeError),
+ ('GET', '/search', b'', b'q=%FF', None, UnicodeDecodeError),
+ ('GET', '/search', b'', b'%FF=value', None, UnicodeDecodeError),
('GET', '/invalid', b'', b'', 'unsupported response', TypeError),
('GET', '/invalid', b'', b'', (b'bad', 'text/plain'), ValueError),
]
for method, path, body, query_string, response, error in cases:
- with self.subTest(path=path, response=response):
+ with self.subTest(path=path, query_string=query_string, response=response):
messages = []
async def receive():
diff --git a/wred.py b/wred.py
index c4a01fb..e7d31fe 100644
--- a/wred.py
+++ b/wred.py
@@ -29,7 +29,7 @@ class Request:
def query(self) -> dict[str, str]:
"""Parse and cache query parameters; repeated keys keep their last value."""
if self._query is None:
- self._query = dict(parse_qsl(self.scope.get('query_string', b'').decode(), keep_blank_values=True))
+ self._query = dict(parse_qsl(self.scope.get('query_string', b'').decode(), keep_blank_values=True, errors='strict'))
return self._query
async def body(self) -> bytes: