patx/micropie
Release 0.32 request and session hardening
Commit a127704 · patx · 2026-07-14T23:09:56-04:00
Comments
No comments yet.
Diff
diff --git a/README.md b/README.md
index f4c908f..96f818e 100644
--- a/README.md
+++ b/README.md
@@ -115,7 +115,9 @@ MicroPie's route handlers map URLs to methods in your `App` subclass. Handler in
- **Automatic Mapping**: URLs map to method names (e.g., `/greet` → `greet`, `/` → `index`).
- **Private Methods**: Methods starting with `_` (e.g., `_private_method`) are private and inaccessible via URLs, returning 404. **Security Note**: Use `_` for sensitive methods to prevent external access.
- **Automatic Argument Binding**: Handler args are populated from path/query/body data by parameter name.
+- **Python Signatures**: Keyword-only parameters bind correctly, `*args` collects remaining path segments, and `**kwargs` collects remaining named request values.
- **Session Isolation**: Session values never bind to handler arguments; read trusted state explicitly from `self.request.session`.
+- **Bounded Bodies**: Requests default to a 16 MiB total limit and multipart text fields to 1 MiB; configure `max_body_size` and `max_form_field_size` on `App`.
- **Request Helpers**:
- `self.request.query(name, default)` for query-string values.
- `self.request.form(name, default)` for URL-encoded or multipart form values.
@@ -149,8 +151,8 @@ class MyApp(App):
Top-level JSON values retain their native types during binding. JSON objects
are not mirrored into `body_params`, so `request.form()` only reads form data.
-Malformed query strings and URL-encoded forms return `400 Bad Request`; valid
-empty values such as `name=` are preserved.
+Query strings and URL-encoded form bodies share MicroPie's permissive parsing
+behavior: blank or bare fields are ignored and malformed encoding is tolerated.
#### **Helper-Based Request Access**
Use helper methods for query, form, and JSON payload access:
@@ -181,6 +183,8 @@ Choose the simplest input style that fits the route:
Do not `await` `self.request.query(...)`, `self.request.form(...)`, or `self.request.json(...)`. They are synchronous accessors over request data MicroPie has already parsed. Uploaded file content is the exception: file bodies are streamed through async queues in `self.request.files`.
+An uploaded file queue yields `None` only after a complete file. If multipart parsing, timeout enforcement, or a size limit aborts an open file, `get()` raises `MultipartFileError`; treat that as a failed upload rather than persisting partial content.
+
By default, MicroPie's route handlers can accept any request method. Check `self.request.method` in a handler when route behavior differs by method. For lower-level request internals such as `query_params` and `body_params`, see `docs/apidocs/reference/request.rst`.
### **Lifecycle Event Handling**
@@ -293,7 +297,7 @@ class MyApp(App):
return f"You have visited {self.request.session['visits']} times."
```
-You also can use the `SessionBackend` class to create your own session backend. You can see an example of this in [examples/sessions](https://github.com/patx/micropie/tree/main/examples/sessions).
+You also can use the `SessionBackend` class to create your own session backend. Session IDs default to validated UUIDv4 values; backends using another ID format can configure `session_id_factory` and `session_id_validator`. You can see an example of a custom backend in [examples/sessions](https://github.com/patx/micropie/tree/main/examples/sessions).
### **Middleware**
MicroPie allows you to create pluggable middleware to hook into the request lifecycle. Take a look at a trivial example using `HttpMiddleware` to send console messages before and after the request is processed. Check out [examples/middleware](https://github.com/patx/micropie/tree/main/examples/middleware) to see more.
@@ -316,7 +320,7 @@ app = Root()
app.middlewares.append(MiddlewareExample())
```
-Middleware provides an easy and **reusable** way to extend the MicroPie framework. We can do things such as rate limiting, checking for max upload size in multipart requests, **explicit routing**, CSRF protection, and more.
+Middleware provides an easy and **reusable** way to extend the MicroPie framework. We can do things such as rate limiting, metadata validation, **explicit routing**, CSRF protection, and more. Request-body byte limits belong on `App` (or an outer ASGI/proxy layer), because `HttpMiddleware` runs after body parsing has begun.
MicroPie apps can be deployed using any ASGI server. For example, using Uvicorn if our application is saved as `app.py` and our `App` subclass is assigned to the `app` variable we can run it with:
```bash
diff --git a/docs/apidocs/conf.py b/docs/apidocs/conf.py
index 2b3f752..8acf402 100644
--- a/docs/apidocs/conf.py
+++ b/docs/apidocs/conf.py
@@ -2,7 +2,7 @@
project = "MicroPie"
author = "Harrison Erd"
-release = "0.31"
+release = "0.32"
extensions = [
"sphinx.ext.autodoc",
diff --git a/docs/apidocs/explanation/architecture.rst b/docs/apidocs/explanation/architecture.rst
index d75982a..923c26c 100644
--- a/docs/apidocs/explanation/architecture.rst
+++ b/docs/apidocs/explanation/architecture.rst
@@ -43,14 +43,17 @@ of each request to avoid leaking state.
Session management
------------------
-Sessions are stored in a pluggable back‑end and identified by a
-random ``session_id`` cookie. The default in‑memory back‑end keeps
-session data in dictionaries keyed by session ID and updates a last
-access timestamp to implement expiration. You can customise the
-back‑end by providing your own implementation of
-:class:`~micropie.SessionBackend`. MicroPie saves sessions after a
-handler returns, only if ``self.request.session`` is non‑empty, to
-avoid creating unnecessary cookies.
+Sessions are stored in a pluggable back-end and identified by a
+random UUIDv4 ``session_id`` cookie by default. Incoming IDs are validated
+before they reach the back-end; custom back-ends can supply a matching ID
+factory and validator. When an empty session becomes populated,
+MicroPie rotates the ID so a client-provided cookie cannot be fixed across
+that state transition. Applications can request another rotation with
+:meth:`~micropie.Request.regenerate_session` when an existing session gains
+privileges. The default in-memory back-end keeps session data in
+dictionaries keyed by session ID and updates a last access timestamp to
+implement expiration. You can customise the back-end by providing your
+own implementation of :class:`~micropie.SessionBackend`.
Middleware pipeline
-------------------
@@ -59,7 +62,10 @@ The middleware hooks allow you to intercept requests before and after
handlers. Middleware can modify the request object, provide early
responses (useful for authentication), and alter the final response.
For WebSockets, separate middleware hooks run before the connection
-handler starts and after it finishes.
+handler starts and after it finishes. HTTP request bodies are parsed
+before ``before_request`` runs, so byte-counting upload limits belong in
+the :class:`~micropie.App` body-limit settings or in an outer ASGI/proxy
+layer rather than in :class:`~micropie.HttpMiddleware`.
Streaming and SSE support
-------------------------
@@ -101,8 +107,8 @@ MicroPie automatically handles common error cases. Requests for
unknown routes result in a ``404 Not Found``. If a required
parameter is missing, MicroPie responds with ``400 Bad Request``.
Unhandled exceptions inside handlers produce a ``500 Internal Server
-Error`` and are printed to standard error. You can override these
-behaviours via middleware.
+Error`` and are reported through the standard ``micropie`` logger. You
+can override these behaviours via middleware.
Extensibility
-------------
diff --git a/docs/apidocs/howto/middleware.rst b/docs/apidocs/howto/middleware.rst
index cb5dba0..7c03e3b 100644
--- a/docs/apidocs/howto/middleware.rst
+++ b/docs/apidocs/howto/middleware.rst
@@ -14,9 +14,9 @@ and implement two asynchronous methods:
* :meth:`~micropie.HttpMiddleware.before_request` – called before the
handler is executed. If this method returns a dictionary with
- ``status_code`` and ``body`` keys, MicroPie immediately sends that
- response and skips calling the handler. You can also provide
- additional headers via a ``headers`` key.
+ response fields, MicroPie immediately sends that response and skips
+ calling the handler. ``status_code``, ``body`` and ``headers`` are all
+ optional; omitted fields use the current defaults.
* :meth:`~micropie.HttpMiddleware.after_request` – called after the
handler has returned a response but before it is sent to the client.
@@ -46,6 +46,17 @@ Register your middleware by appending an instance to
app = MyApp()
app.middlewares.append(LoggingMiddleware())
+Request-body limits
+-------------------
+
+``before_request`` runs after MicroPie has started parsing the request
+body, and :class:`~micropie.HttpMiddleware` does not receive the ASGI
+``receive`` callable. It is therefore too late for this hook to enforce a
+memory-safe upload limit. Configure ``App(max_body_size=...)`` for total
+encoded-body enforcement and ``App(max_form_field_size=...)`` for each
+in-memory multipart text field. Outer ASGI middleware or a reverse proxy
+can add another limit before the request reaches MicroPie.
+
WebSocket middleware
--------------------
diff --git a/docs/apidocs/howto/sessions.rst b/docs/apidocs/howto/sessions.rst
index ed0553b..0b476ea 100644
--- a/docs/apidocs/howto/sessions.rst
+++ b/docs/apidocs/howto/sessions.rst
@@ -98,3 +98,58 @@ production:
.. code-block:: python
app = MyApp(session_cookie_secure=False)
+
+Session fixation protection
+---------------------------
+
+MicroPie validates incoming session IDs as canonical UUIDv4 values before
+passing them to a back-end. UUID validation is input hygiene; the primary
+fixation defense is ID rotation. If a request loads an empty session and
+finishes with populated session data, MicroPie ignores the incoming cookie,
+generates a fresh ID (UUIDv4 under the default policy), and sends it in
+``Set-Cookie``. A client-planted ID therefore cannot survive the transition
+into a new authenticated session.
+
+Applications should explicitly rotate an already populated session when
+its privilege level changes, such as re-authentication or elevation to an
+administrator role:
+
+.. code-block:: python
+
+ class MyApp(App):
+ async def reauthenticate(self):
+ await verify_credentials(self.request)
+ self.request.session["elevated"] = True
+ self.request.regenerate_session()
+ return "Privileges updated"
+
+``regenerate_session()`` preserves the session data, deletes the old
+back-end entry, and persists the data under a fresh ID. It is synchronous
+and is available on HTTP requests only.
+
+Custom session ID formats
+-------------------------
+
+The UUIDv4 policy is the secure default, but a custom back-end may already
+use signed or otherwise structured IDs. Configure both sides of that
+policy together:
+
+.. code-block:: python
+
+ def validate_signed_id(value: str):
+ return value if verify_and_parse_id(value) else None
+
+ app = MyApp(
+ session_backend=backend,
+ session_id_factory=create_signed_id,
+ session_id_validator=validate_signed_id,
+ )
+
+The validator must return a non-empty normalized string or ``None``; the
+returned string is used as both the cookie value and the storage key. It is
+used for HTTP cookies and explicit IDs passed to
+:meth:`~micropie.WebSocket.accept`. Keep custom IDs unguessable, bounded in
+length and safe for both cookies and back-end keys. Passing
+``session_id_validator=None`` disables framework validation and is intended
+only when an outer trusted layer has already validated the value; ID
+rotation still applies.
diff --git a/docs/apidocs/howto/streaming.rst b/docs/apidocs/howto/streaming.rst
index b231ce3..686582e 100644
--- a/docs/apidocs/howto/streaming.rst
+++ b/docs/apidocs/howto/streaming.rst
@@ -9,6 +9,11 @@ the generator is sent as part of the response body. When the
generator finishes, MicroPie automatically sends an empty chunk to
signal completion.
+MicroPie advances synchronous iterators in a worker thread so a blocking
+``next()`` call does not freeze the event loop. Prefer asynchronous
+generators for long-lived streams because they support cooperative
+cancellation and client-disconnect handling directly.
+
Example: number stream
----------------------
@@ -48,8 +53,8 @@ generator the body.
Handling client disconnects
---------------------------
-If the client disconnects while a generator is running, MicroPie
-cancels the generator and closes the underlying response. This
+If the client disconnects while an asynchronous generator is running,
+MicroPie cancels the generator and closes the underlying response. This
behaviour is important for long‑lived streams such as server‑sent
events (SSE). When writing a generator, use ``try``/``finally`` or
``async with`` to perform cleanup when the stream is cancelled.
@@ -63,4 +68,4 @@ Implementing true server‑sent events requires you to set the
properly‑formatted SSE messages (``data: ...\n\n``). Additionally,
MicroPie includes a patch that handles early disconnection of SSE
clients in its HTTP handling code. For a full example, see the
-``examples/sse`` directory in the MicroPie source distribution.
\ No newline at end of file
+``examples/sse`` directory in the MicroPie source distribution.
diff --git a/docs/apidocs/reference/app.rst b/docs/apidocs/reference/app.rst
index 4153556..a67eb98 100644
--- a/docs/apidocs/reference/app.rst
+++ b/docs/apidocs/reference/app.rst
@@ -13,23 +13,43 @@ Uvicorn.
Constructor
-----------
-.. class:: App(session_backend=None, *, body_timeout=5.0, session_timeout=SESSION_TIMEOUT, session_cookie_secure=True)
+.. class:: App(session_backend=None, *, body_timeout=5.0, max_body_size=16777216, max_form_field_size=1048576, session_timeout=SESSION_TIMEOUT, session_cookie_secure=True, session_id_factory=UUIDv4 factory, session_id_validator=UUIDv4 validator)
Create a new application. If *session_backend* is provided it must
be an instance of :class:`~micropie.SessionBackend`. When omitted,
MicroPie uses an in‑memory back‑end.
*body_timeout* is the maximum number of seconds MicroPie waits for
- each non-multipart request-body chunk. The timeout restarts after
- every chunk, so it limits idle time rather than total upload time.
+ each request-body chunk, including multipart uploads. The timeout
+ restarts after every chunk, so it limits idle time rather than total upload time.
Set it to ``None`` to disable the framework deadline.
+ *max_body_size* limits the total encoded request body to 16 MiB by
+ default. MicroPie rejects an oversized ``Content-Length`` before
+ reading the body and also counts actual chunks, covering requests
+ without that header. *max_form_field_size* separately limits each
+ accumulated multipart text field to 1 MiB while allowing file content
+ to remain streamed. Either limit can be disabled with ``None``.
+ Multipart file producers use nonblocking queues so later parts cannot
+ deadlock behind an undrained file; consequently, ``max_body_size`` is
+ also the request-wide payload-byte bound for queued upload content.
+ Disabling it permits unbounded buffering when a handler does not consume
+ files.
+
*session_timeout* is passed to the session back-end whenever session
data is saved. *session_cookie_secure* controls whether generated
session cookies include the ``Secure`` attribute. It defaults to
``True``; applications served over plain HTTP during local development
can set it to ``False`` explicitly.
+ By default, *session_id_factory* creates UUIDv4 values and
+ *session_id_validator* accepts only canonical UUIDv4 strings. A custom
+ storage back-end that owns another identifier format can provide a
+ factory and validator; a validator returns the normalized ID or ``None``
+ to reject it. Setting the validator itself to ``None`` accepts any
+ non-empty cookie value and should only be used with another trusted
+ validation boundary.
+
Attributes
----------
@@ -51,8 +71,19 @@ Attributes
.. attribute:: body_timeout
- The configured per-chunk idle timeout for non-multipart request
- bodies, or ``None`` when timeout enforcement is disabled.
+ The configured per-chunk idle timeout for buffered and multipart
+ request bodies, or ``None`` when timeout enforcement is disabled.
+
+.. attribute:: max_body_size
+
+ Maximum total encoded request-body size in bytes, or ``None``. Bodies
+ exceeding the limit receive ``413 Payload Too Large``.
+
+.. attribute:: max_form_field_size
+
+ Maximum size in bytes of each multipart text field, or ``None``.
+ This is independent of the total body limit so applications permitting
+ large streamed files need not permit equally large in-memory fields.
.. attribute:: session_timeout
@@ -64,6 +95,15 @@ Attributes
Whether HTTP and WebSocket session cookies include the ``Secure``
attribute.
+.. attribute:: session_id_factory
+
+ Callable used whenever MicroPie needs a fresh session ID.
+
+.. attribute:: session_id_validator
+
+ Callable that normalizes an incoming or generated session ID and
+ returns ``None`` to reject it. The default requires canonical UUIDv4.
+
.. attribute:: startup_handlers
A list of asynchronous callables that run during the ASGI
@@ -95,7 +135,10 @@ Methods
Return a tuple ``(302, '', headers)`` representing an HTTP redirect
to *location*. Use this helper in your handlers to redirect the
- client.
+ client. Absolute URLs are allowed, so do not pass an untrusted target
+ without an application-level allow-list. MicroPie percent-encodes
+ non-ASCII URL components, but applications constructing query strings
+ should still use :func:`urllib.parse.urlencode` rather than concatenation.
.. method:: _render_template(name, **kwargs)
@@ -115,3 +158,13 @@ MicroPie writes framework diagnostics and exception tracebacks to the
standard ``micropie`` Python logger. Configure this logger through the
:mod:`logging` package to integrate it with your deployment's handlers
and formatters.
+
+Body-limit placement
+--------------------
+
+MicroPie's :class:`~micropie.HttpMiddleware` hooks run after request-body
+parsing starts and do not receive the ASGI ``receive`` callable.
+They therefore cannot safely implement a byte-counting body limit. Use
+the application settings above for framework enforcement. A reverse
+proxy limit or outer ASGI middleware that wraps ``receive`` remains useful
+as an additional deployment boundary.
diff --git a/docs/apidocs/reference/exceptions.rst b/docs/apidocs/reference/exceptions.rst
index fd1454c..a5dc30e 100644
--- a/docs/apidocs/reference/exceptions.rst
+++ b/docs/apidocs/reference/exceptions.rst
@@ -15,4 +15,14 @@ ConnectionClosed
the WebSocket connection. Catch this exception to detect
disconnection and exit your handler gracefully. The exception does
not carry any attributes; the WebSocket connection has already been
- closed when it is raised.
\ No newline at end of file
+ closed when it is raised.
+
+MultipartFileError
+------------------
+
+.. class:: MultipartFileError
+
+ Raised by an uploaded file's ``content.get()`` operation when the
+ multipart parser stops before that file reaches a clean end-of-file.
+ This includes malformed bodies, receive timeouts and configured size
+ limits. The original parser exception is available as ``cause``.
diff --git a/docs/apidocs/reference/middleware.rst b/docs/apidocs/reference/middleware.rst
index 4754f9e..d785765 100644
--- a/docs/apidocs/reference/middleware.rst
+++ b/docs/apidocs/reference/middleware.rst
@@ -19,8 +19,9 @@ HttpMiddleware
Called before the HTTP handler runs. *request* is the
:class:`~micropie.Request` object. Return a dictionary
``{"status_code": status, "body": body, "headers": headers}``
- to short‑circuit the request and send a response immediately. To
- continue processing, return ``None``.
+ to short‑circuit the request and send a response immediately.
+ Fields are optional and default to the current status/body and an
+ empty header list. To continue processing, return ``None``.
.. method:: after_request(request, status_code, response_body, extra_headers)
@@ -48,4 +49,4 @@ WebSocketMiddleware
.. method:: after_websocket(request)
Called after the WebSocket handler completes. Use this to
- release resources or log activity. Return value is ignored.
\ No newline at end of file
+ release resources or log activity. Return value is ignored.
diff --git a/docs/apidocs/reference/request.rst b/docs/apidocs/reference/request.rst
index 7c051a6..92bb571 100644
--- a/docs/apidocs/reference/request.rst
+++ b/docs/apidocs/reference/request.rst
@@ -44,7 +44,8 @@ Request class
A ``dict`` mapping each query parameter to a list of values.
This is the raw parsed mapping from the query string. For
convenience, use :meth:`~micropie.Request.query` to obtain
- the first value.
+ the first value. Parsing is permissive for compatibility: blank
+ and bare fields are omitted, while malformed encoding is tolerated.
Use this raw attribute when repeated query parameters matter:
@@ -69,9 +70,9 @@ Request class
A ``dict`` mapping URL-encoded and multipart form field names to
lists of string values. JSON objects are not mirrored into this
- mapping; use :meth:`json` for JSON access. URL-encoded forms are
- parsed strictly, preserving empty values and rejecting malformed
- fields or encoding with ``400 Bad Request``.
+ mapping; use :meth:`json` for JSON access. URL-encoded form bodies
+ use the same permissive parsing as query strings: blank and bare
+ fields are omitted, while malformed encoding is tolerated.
Use this raw attribute when you need all submitted values for a
field, or when middleware needs to inspect parsed request fields:
@@ -113,6 +114,15 @@ Request class
Top-level JSON object keys can bind directly to handler arguments
and retain native JSON types; nested values are not stringified.
+ .. method:: regenerate_session()
+
+ Mark the current HTTP session for ID rotation when the response is
+ persisted. Existing session data is preserved, the old back-end
+ entry is deleted, and a fresh session cookie is issued using the
+ application's configured ID factory. Call this
+ after a privilege change in an already populated session. Do not
+ await this method.
+
.. attribute:: session
A ``dict`` for storing per‑client data across requests. See
@@ -125,12 +135,15 @@ Request class
A ``dict`` of uploaded files for multipart/form-data requests.
Each entry is a mapping containing keys ``filename``,
``content_type`` and ``content``, where ``content`` is an
- ``asyncio.Queue`` yielding file chunks as bytes. See
+ ``asyncio.Queue`` subclass yielding file chunks as bytes. See
:doc:`../tutorial/routing` for information on awaiting file fields.
Unlike ``query()``, ``form()`` and ``json()``, file content is
streaming data. Read file chunks asynchronously from the
- ``content`` queue.
+ ``content`` queue. A complete file ends with ``None``. If parsing,
+ timeout or body-limit enforcement aborts an open file, ``get()``
+ raises :class:`~micropie.MultipartFileError` instead; do not treat
+ parser failure as a clean end-of-file.
.. attribute:: headers
diff --git a/docs/apidocs/reference/session.rst b/docs/apidocs/reference/session.rst
index 8d2f5c3..d727a04 100644
--- a/docs/apidocs/reference/session.rst
+++ b/docs/apidocs/reference/session.rst
@@ -6,6 +6,11 @@ interface. Different back‑ends may store session data in memory,
databases, caches or external services. A session is a dictionary of
key/value pairs associated with a ``session_id`` cookie.
+The application owns identifier generation and validation. Canonical
+UUIDv4 is the default; custom storage formats can be enabled with
+``App(session_id_factory=..., session_id_validator=...)``. A storage
+back-end receives only IDs accepted by that policy.
+
SessionBackend
--------------
diff --git a/docs/apidocs/reference/websocket.rst b/docs/apidocs/reference/websocket.rst
index c1e8c4b..04d46fe 100644
--- a/docs/apidocs/reference/websocket.rst
+++ b/docs/apidocs/reference/websocket.rst
@@ -10,7 +10,7 @@ request and passes it as the first argument to your WebSocket handler.
Constructor
-----------
-.. class:: WebSocket(receive, send, *, session_cookie_secure=True)
+.. class:: WebSocket(receive, send, *, session_cookie_secure=True, session_id_validator=UUIDv4 validator)
Create a new WebSocket wrapper around the ASGI ``receive`` and
``send`` callables. You do not instantiate this class yourself;
@@ -25,8 +25,16 @@ Methods
Accept the WebSocket connection. You must call this method before
sending or receiving messages. If you provide a *session_id*,
- MicroPie sets a ``session_id`` cookie during the handshake. The
- optional *subprotocol* argument specifies a negotiated subprotocol.
+ it must pass the owning application's session-ID validator and MicroPie
+ sets it as a cookie during the handshake. The default policy requires
+ canonical UUIDv4; applications with custom back-ends can configure a
+ different factory and validator on :class:`~micropie.App`. The optional
+ *subprotocol* argument specifies a negotiated subprotocol.
+
+ When no session cookie was presented, :meth:`accept` sends the fresh ID
+ allocated by the application so session data written by the handler can
+ be loaded on the next request. An incoming ID whose session loads empty
+ is replaced during the handshake to prevent fixation.
.. method:: receive_text()
@@ -65,7 +73,8 @@ Attributes
.. attribute:: session_id
- The session ID set during the handshake, or ``None`` if not set.
+ The validated session ID allocated or loaded for this connection. When
+ the client did not present one, it is sent as a cookie by :meth:`accept`.
Exceptions
----------
diff --git a/docs/apidocs/tutorial/routing.rst b/docs/apidocs/tutorial/routing.rst
index 9162878..7271017 100644
--- a/docs/apidocs/tutorial/routing.rst
+++ b/docs/apidocs/tutorial/routing.rst
@@ -63,6 +63,10 @@ sources in the following order:
If MicroPie cannot determine a value for a required parameter, it
returns a ``400 Bad Request``.
+Keyword-only parameters are passed by keyword, and ``**kwargs`` collects
+remaining named query, form, JSON and file values after explicit parameters
+have been bound. ``*args`` continues to collect remaining path segments.
+
Session values never participate in automatic argument binding. This
keeps untrusted request inputs and trusted session state in separate
namespaces. Read session values explicitly inside the handler:
@@ -150,11 +154,11 @@ simplest level that fits the handler.
tags = self.request.query_params.get("tag", [])
return {"tags": tags}
-Query strings and URL-encoded forms are parsed strictly. Empty values
-such as ``name=`` are preserved, while fields without ``=``, invalid
-percent escapes and invalid UTF-8 receive ``400 Bad Request``. A
-WebSocket request with a malformed query string is closed with code
-``1008``.
+Query strings and URL-encoded form bodies share MicroPie's permissive parsing
+behavior for compatibility. Blank values such as ``name=`` and bare keys are
+omitted, so they fall through to handler defaults. Malformed percent escapes
+are tolerated, invalid percent-decoded UTF-8 is replaced, and invalid raw
+UTF-8 bytes are ignored. HTTP and WebSocket query strings use the same rules.
Do not await request helpers
----------------------------
@@ -174,8 +178,13 @@ has already parsed for the current request. Do not use ``await`` with
For JSON and URL-encoded form requests, the body has been read before
your handler runs. Multipart file uploads are the main exception:
-uploaded file content is exposed as an ``asyncio.Queue`` in
-``self.request.files`` and should be read asynchronously.
+uploaded file content is exposed as an ``asyncio.Queue`` subclass in
+``self.request.files`` and should be read asynchronously. File producers
+do not block discovery of later multipart parts; the application's total
+body limit bounds queued content. A queue yields ``None`` only for a clean
+end-of-file. If parsing aborts, ``get()`` raises
+:class:`~micropie.MultipartFileError` so a partial file cannot be mistaken
+for a complete upload.
Examples
--------
diff --git a/docs/apidocs/whats_new.rst b/docs/apidocs/whats_new.rst
index 5a60bca..4d0d3dd 100644
--- a/docs/apidocs/whats_new.rst
+++ b/docs/apidocs/whats_new.rst
@@ -11,6 +11,14 @@ releases, consult the `GitHub releases page <https://github.com/patx/micropie/re
Version highlights
------------------
+* **Next release** – Adds request-body and multipart-field limits,
+ per-chunk timeouts for buffered and multipart bodies, deadlock-free
+ multi-file streaming, explicit multipart failure signaling, session
+ fixation protection, configurable session cookie/ID policies, isolated
+ session argument binding, native JSON argument values, and structured
+ framework logging. It also fixes keyword-only/``**kwargs`` binding,
+ WebSocket session-cookie delivery, pending streaming receive tasks,
+ multipart sub-application handoff, and response-header encoding failures.
* **0.31** – Hardens route lookup so non-callable public attributes and
property-like descriptors are not treated as handlers. Descriptor-backed
route methods such as ``functools.partialmethod`` continue to work after
@@ -59,6 +67,18 @@ Version highlights
Upgrade tips
------------
+* **Review request parsing compatibility.** Query strings and URL-encoded form
+ bodies retain their earlier permissive behavior: empty and bare fields are
+ omitted and malformed encoding is tolerated. JSON objects now retain native
+ types and are no longer mirrored into ``body_params``; use ``Request.json()``
+ rather than ``Request.form()`` for JSON payloads.
+* **Review custom session IDs.** Session cookies and explicit WebSocket IDs
+ now require canonical UUIDv4 by default. Back-ends using signed or custom
+ identifiers must configure matching ``session_id_factory`` and
+ ``session_id_validator`` callables.
+* **Handle aborted file streams.** Multipart file queues now raise
+ ``MultipartFileError`` when parsing fails instead of yielding a clean
+ ``None`` sentinel for truncated content.
* **Check optional extras.** When upgrading, confirm that any optional
dependencies you rely on (``jinja2``, ``multipart``, ``orjson``) are
still installed, especially if you pin minimal environments.
diff --git a/docs/release_notes.md b/docs/release_notes.md
index dd8a7bb..4dc9de8 100644
--- a/docs/release_notes.md
+++ b/docs/release_notes.md
@@ -1,6 +1,7 @@
[](https://patx.github.io/micropie)
## Releases Notes
+- **Unreleased** - Harden request parsing, multipart streaming, sessions, parameter binding, logging, and response streaming. This release retains consistent permissive URL-encoded query and form parsing, and introduces native JSON binding without `body_params` mirroring, UUIDv4 session validation by default, configurable custom session-ID policies, request-body limits, and `MultipartFileError` for aborted uploads; review the upgrade notes before deploying.
- **[0.31](https://github.com/patx/micropie/releases/tag/v0.31)** - Security fix: route lookup now ignores non-callable public attributes and property-like descriptors, preventing internal App attributes from being treated as handlers while preserving descriptor-backed route methods.
- **[0.30](https://github.com/patx/micropie/releases/tag/v0.30)** - Remove the public `Request.get_json` attribute. Use `Request.json()` or `Request.json(name, default)`.
- **[0.29](https://github.com/patx/micropie/releases/tag/v0.29)** - Performance upgrades, no more per request signature inspections for routing. 24%-54% increase in req/sec.
diff --git a/examples/file_uploads/app.py b/examples/file_uploads/app.py
index 0106b87..9a0caa4 100644
--- a/examples/file_uploads/app.py
+++ b/examples/file_uploads/app.py
@@ -50,5 +50,5 @@ class Root(App):
return 200, f"Uploaded {file['filename']}"
-# Instantiate the app
-app = Root()
+# Instantiate the app with a 100 MiB encoded-body limit.
+app = Root(max_body_size=100 * 1024 * 1024)
diff --git a/examples/json_api/basic.py b/examples/json_api/basic.py
index 3dfa297..891d36d 100644
--- a/examples/json_api/basic.py
+++ b/examples/json_api/basic.py
@@ -9,7 +9,9 @@ class PasteApp(App):
async def paste(self, pid: str = None):
if self.request.method == "POST":
# Get content from JSON or form, depending on the Content-Type
- content = self.request.form("content")
+ content = self.request.json("content")
+ if content is None:
+ content = self.request.form("content")
pid = str(uuid4())
await db.aset(pid, content)
return {
diff --git a/examples/middleware/sessions.py b/examples/middleware/sessions.py
index e4fbcde..383fe44 100644
--- a/examples/middleware/sessions.py
+++ b/examples/middleware/sessions.py
@@ -1,59 +1,26 @@
+"""Use MicroPie's session-ID policy hooks for signed cookie identifiers."""
+
from html import escape
-import os
import uuid
-from typing import Optional, Dict, List, Tuple, Any
-from itsdangerous import URLSafeTimedSerializer, BadSignature
-from micropie import App, HttpMiddleware, Request, SESSION_TIMEOUT
+from itsdangerous import BadSignature, URLSafeTimedSerializer
-class SignedSessionMiddleware(HttpMiddleware):
- """Middleware to sign and verify session cookies using itsdangerous."""
+from micropie import App
- def __init__(self, app: App, secret_key: str, max_age: int = SESSION_TIMEOUT):
- self.app = app # Store the App instance
- self.serializer = URLSafeTimedSerializer(secret_key)
- self.max_age = max_age
- async def before_request(self, request: Request) -> Optional[Dict]:
- """Verify the session_id cookie before processing the request."""
- cookies = self.app._parse_cookies(request.headers.get("cookie", ""))
- session_id = cookies.get("session_id", "")
- try:
- verified_id = self.serializer.loads(session_id, max_age=self.max_age)
- request.session = await self.app.session_backend.load(verified_id) or {}
- except BadSignature:
- request.session = {} # Invalid or expired session_id
- return None
+serializer = URLSafeTimedSerializer("my-secret-key", salt="session-id")
+
+
+def create_signed_session_id() -> str:
+ return serializer.dumps(str(uuid.uuid4()))
+
- async def after_request(
- self,
- request: Request,
- status_code: int,
- response_body: Any,
- extra_headers: List[Tuple[str, str]],
- ) -> Optional[Dict]:
- """Sign and set the session_id cookie after processing the request."""
- if request.session:
- cookies = self.app._parse_cookies(request.headers.get("cookie", ""))
- session_id = cookies.get("session_id", "") or str(uuid.uuid4())
- try:
- session_id = self.serializer.loads(session_id, max_age=self.max_age)
- except BadSignature:
- session_id = str(uuid.uuid4())
- signed_session_id = self.serializer.dumps(session_id)
- current_session = await self.app.session_backend.load(session_id) or {}
- if current_session != request.session:
- await self.app.session_backend.save(
- session_id, request.session, SESSION_TIMEOUT
- )
- if not cookies.get("session_id"):
- extra_headers.append(
- (
- "Set-Cookie",
- f"session_id={signed_session_id}; Path=/; SameSite=Lax; HttpOnly; Secure;",
- )
- )
+def validate_signed_session_id(value: str):
+ try:
+ serializer.loads(value)
+ except BadSignature:
return None
+ return value
class Root(App):
@@ -66,5 +33,7 @@ class Root(App):
return f"You have visited {escape(str(visits))} times."
-app = Root()
-app.middlewares.append(SignedSessionMiddleware(app=app, secret_key="my-secret-key"))
+app = Root(
+ session_id_factory=create_signed_session_id,
+ session_id_validator=validate_signed_session_id,
+)
diff --git a/examples/middleware/upload.py b/examples/middleware/upload.py
index bcff848..1637c20 100644
--- a/examples/middleware/upload.py
+++ b/examples/middleware/upload.py
@@ -1,40 +1,8 @@
-from micropie import App, HttpMiddleware
-import asyncio
-
-MAX_UPLOAD_SIZE = 100 * 1024 * 1024 # 100MB
+"""Configure upload limits before MicroPie's HTTP middleware layer."""
+from micropie import App
-class MaxUploadSizeMiddleware(HttpMiddleware):
- async def before_request(self, request):
- # Check if we're dealing with a POST, PUT, or PATCH request
- if request.method in ("POST", "PUT", "PATCH"):
- content_length = request.headers.get("content-length")
- # Ensure Content-Length is present and valid
- if content_length is None:
- return {
- "status_code": 400,
- "body": "400 Bad Request: Missing Content-Length header",
- }
- try:
- content_length = int(content_length)
- if content_length > MAX_UPLOAD_SIZE:
- print(
- f"Upload rejected: Content-Length ({content_length}) exceeds {MAX_UPLOAD_SIZE} bytes"
- )
- return {
- "status_code": 413,
- "body": "413 Payload Too Large: Uploaded file exceeds size limit.",
- }
- except ValueError:
- return {
- "status_code": 400,
- "body": "400 Bad Request: Invalid Content-Length header",
- }
- # Continue processing if checks pass
- return None
-
- async def after_request(self, request, status_code, response_body, extra_headers):
- return None
+MAX_UPLOAD_SIZE = 100 * 1024 * 1024 # 100MB
class FileUploadApp(App):
@@ -73,5 +41,4 @@ class FileUploadApp(App):
return {"filename": filename, "content_type": content_type, "size": total_size}
-app = FileUploadApp()
-app.middlewares.append(MaxUploadSizeMiddleware())
+app = FileUploadApp(max_body_size=MAX_UPLOAD_SIZE)
diff --git a/examples/url_shortener/middlewares/csrf.py b/examples/url_shortener/middlewares/csrf.py
index 5298604..863533e 100644
--- a/examples/url_shortener/middlewares/csrf.py
+++ b/examples/url_shortener/middlewares/csrf.py
@@ -1,4 +1,3 @@
-import asyncio
import uuid
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import urlparse
@@ -85,13 +84,6 @@ class CSRFMiddleware(HttpMiddleware):
return token
lst = request.body_params.get(self.body_field)
- if not lst and self._is_multipart(ct):
- while not request.body_parsed:
- lst = request.body_params.get(self.body_field)
- if lst:
- break
- await asyncio.sleep(0)
-
if lst and isinstance(lst, list) and lst:
return lst[0]
diff --git a/micropie.py b/micropie.py
index b7d40d0..2741349 100644
--- a/micropie.py
+++ b/micropie.py
@@ -46,6 +46,8 @@ logger = logging.getLogger(__name__)
_PARAM_EMPTY = inspect.Parameter.empty
_VAR_POSITIONAL = inspect.Parameter.VAR_POSITIONAL
+_VAR_KEYWORD = inspect.Parameter.VAR_KEYWORD
+_KEYWORD_ONLY = inspect.Parameter.KEYWORD_ONLY
_POSITIONAL_PARAM_KINDS = (
inspect.Parameter.POSITIONAL_ONLY,
inspect.Parameter.POSITIONAL_OR_KEYWORD,
@@ -58,7 +60,75 @@ _JSON_HEADER_BYTES = (b"Content-Type", b"application/json")
_DEFAULT_HEADERS_BYTES = [_DEFAULT_HEADER_BYTES]
_JSON_HEADERS_BYTES = [_JSON_HEADER_BYTES]
_ROUTE_ATTR_MISSING = object()
-_INVALID_PERCENT_ESCAPE = re.compile(r"%(?![0-9A-Fa-f]{2})")
+DEFAULT_MAX_BODY_SIZE: int = 16 * 1024 * 1024
+DEFAULT_MAX_FORM_FIELD_SIZE: int = 1024 * 1024
+
+
+class _PayloadTooLarge(Exception):
+ pass
+
+
+class _FormFieldTooLarge(Exception):
+ pass
+
+
+class MultipartFileError(Exception):
+ """Raised while reading a file whose multipart upload did not complete."""
+
+ def __init__(self, cause: BaseException) -> None:
+ self.cause = cause
+ super().__init__("Multipart file upload did not complete")
+
+
+class _MultipartStreamFailure:
+ def __init__(self, cause: BaseException) -> None:
+ self.cause = cause
+
+
+class _MultipartFileQueue(asyncio.Queue):
+ """Queue that distinguishes a complete file from an aborted stream."""
+
+ def __init__(self, maxsize: int = 0) -> None:
+ super().__init__(maxsize=maxsize)
+ self.finished = False
+
+ async def get(self) -> Any:
+ item = await super().get()
+ if isinstance(item, _MultipartStreamFailure):
+ raise MultipartFileError(item.cause) from item.cause
+ return item
+
+ async def finish(self) -> None:
+ if self.finished:
+ return
+ await self.put(None)
+ self.finished = True
+
+ def abort(self, cause: BaseException) -> None:
+ if self.finished:
+ return
+ self.finished = True
+ while True:
+ try:
+ self.put_nowait(_MultipartStreamFailure(cause))
+ return
+ except asyncio.QueueFull:
+ try:
+ asyncio.Queue.get_nowait(self)
+ except asyncio.QueueEmpty:
+ pass
+
+
+def _generate_session_id() -> str:
+ return str(uuid.uuid4())
+
+
+def _next_sync_item(iterator: Any) -> Tuple[bool, Any]:
+ """Advance a synchronous iterator without leaking StopIteration to a Future."""
+ try:
+ return True, next(iterator)
+ except StopIteration:
+ return False, None
def _format_session_cookie(session_id: str, *, secure: bool) -> str:
@@ -73,24 +143,22 @@ def _format_session_cookie(session_id: str, *, secure: bool) -> str:
return "; ".join(attributes) + ";"
-def _parse_urlencoded(data: bytes) -> Dict[str, List[str]]:
- """Parse URL-encoded data without dropping blanks or malformed fields."""
- try:
- encoded = data.decode("ascii")
- except UnicodeDecodeError as exc:
- raise ValueError("URL-encoded input must be ASCII") from exc
- if _INVALID_PERCENT_ESCAPE.search(encoded):
- raise ValueError("Invalid percent escape in URL-encoded input")
+def _validate_session_id(value: Optional[str]) -> Optional[str]:
+ """Return a canonical UUIDv4 session ID, or None for invalid input."""
+ if not value:
+ return None
try:
- return parse_qs(
- encoded,
- keep_blank_values=True,
- strict_parsing=True,
- encoding="utf-8",
- errors="strict",
- )
- except (UnicodeDecodeError, ValueError) as exc:
- raise ValueError("Malformed URL-encoded input") from exc
+ parsed = uuid.UUID(value)
+ except (AttributeError, TypeError, ValueError):
+ return None
+ if parsed.version != 4 or str(parsed) != value.lower():
+ return None
+ return str(parsed)
+
+
+def _parse_urlencoded(data: bytes) -> Dict[str, List[str]]:
+ """Parse URL-encoded data using MicroPie's permissive legacy behavior."""
+ return parse_qs(data.decode("utf-8", "ignore"))
class _HandlerParam(NamedTuple):
@@ -219,6 +287,7 @@ class Request:
for k, v in scope.get("headers", [])
}
self.body_parsed: bool = scope.get("body_parsed", False)
+ self._regenerate_session: bool = False
def query(self, name: str, default: Optional[str] = None) -> Optional[str]:
"""
@@ -260,6 +329,14 @@ class Request:
return self._json.get(name, default)
return default
+ def regenerate_session(self) -> None:
+ """Issue a fresh session ID when this HTTP request is persisted."""
+ if self.scope.get("type") != "http":
+ raise RuntimeError(
+ "Session regeneration is only available for HTTP requests"
+ )
+ self._regenerate_session = True
+
class WebSocketRequest(Request):
"""Represents a WebSocket request in the MicroPie framework."""
@@ -277,6 +354,9 @@ class WebSocket:
send: Callable[[Dict[str, Any]], Awaitable[None]],
*,
session_cookie_secure: bool = True,
+ session_id_validator: Optional[
+ Callable[[str], Optional[str]]
+ ] = _validate_session_id,
) -> None:
"""
Initialize a WebSocket instance.
@@ -285,12 +365,15 @@ class WebSocket:
receive: The ASGI receive callable.
send: The ASGI send callable.
session_cookie_secure: Whether session cookies include the Secure flag.
+ session_id_validator: Normalizes accepted session IDs or returns None.
"""
self.receive = receive
self.send = send
self.accepted = False
self.session_id: Optional[str] = None
self.session_cookie_secure = session_cookie_secure
+ self.session_id_validator = session_id_validator
+ self._session_cookie_pending = False
async def accept(
self, subprotocol: Optional[str] = None, session_id: Optional[str] = None
@@ -309,16 +392,26 @@ class WebSocket:
if message["type"] != "websocket.connect":
raise ValueError(f"Expected websocket.connect, got {message['type']}")
headers = []
- if session_id:
+ if session_id is not None:
+ validated_session_id = (
+ self.session_id_validator(session_id)
+ if self.session_id_validator is not None
+ else session_id
+ )
+ if not isinstance(validated_session_id, str) or not validated_session_id:
+ raise ValueError("session_id was rejected by the session ID validator")
+ self.session_id = validated_session_id
+ self._session_cookie_pending = True
+ if self._session_cookie_pending and self.session_id:
headers.append(
(
"Set-Cookie",
_format_session_cookie(
- session_id, secure=self.session_cookie_secure
+ self.session_id, secure=self.session_cookie_secure
),
)
)
- self.session_id = session_id
+ self._session_cookie_pending = False
await self.send(
{
"type": "websocket.accept",
@@ -508,13 +601,27 @@ class App:
session_backend: Optional[SessionBackend] = None,
*,
body_timeout: Optional[float] = 5.0,
+ max_body_size: Optional[int] = DEFAULT_MAX_BODY_SIZE,
+ max_form_field_size: Optional[int] = DEFAULT_MAX_FORM_FIELD_SIZE,
session_timeout: int = SESSION_TIMEOUT,
session_cookie_secure: bool = True,
+ session_id_factory: Callable[[], str] = _generate_session_id,
+ session_id_validator: Optional[
+ Callable[[str], Optional[str]]
+ ] = _validate_session_id,
) -> None:
if body_timeout is not None and body_timeout <= 0:
raise ValueError("body_timeout must be positive or None")
+ if max_body_size is not None and max_body_size <= 0:
+ raise ValueError("max_body_size must be positive or None")
+ if max_form_field_size is not None and max_form_field_size <= 0:
+ raise ValueError("max_form_field_size must be positive or None")
if session_timeout <= 0:
raise ValueError("session_timeout must be positive")
+ if not callable(session_id_factory):
+ raise ValueError("session_id_factory must be callable")
+ if session_id_validator is not None and not callable(session_id_validator):
+ raise ValueError("session_id_validator must be callable or None")
if JINJA_INSTALLED:
self.env = Environment(
loader=FileSystemLoader("templates"),
@@ -527,8 +634,12 @@ class App:
session_backend or InMemorySessionBackend()
)
self.body_timeout = body_timeout
+ self.max_body_size = max_body_size
+ self.max_form_field_size = max_form_field_size
self.session_timeout = session_timeout
self.session_cookie_secure = session_cookie_secure
+ self.session_id_factory = session_id_factory
+ self.session_id_validator = session_id_validator
self.middlewares: List[HttpMiddleware] = []
self.ws_middlewares: List[WebSocketMiddleware] = []
self.startup_handlers: List[Callable[[], Awaitable[None]]] = []
@@ -536,6 +647,28 @@ class App:
self._handler_cache: Dict[Any, _HandlerInfo] = {}
self._started: bool = False
+ def _normalize_session_id(self, value: Optional[str]) -> Optional[str]:
+ if not value:
+ return None
+ if self.session_id_validator is None:
+ return value
+ try:
+ normalized = self.session_id_validator(value)
+ except Exception:
+ logger.exception("Session ID validator failed")
+ return None
+ return normalized if isinstance(normalized, str) and normalized else None
+
+ def _new_session_id(self) -> str:
+ session_id = self.session_id_factory()
+ normalized = self._normalize_session_id(session_id)
+ if normalized is None:
+ raise RuntimeError(
+ "session_id_factory returned a value rejected by "
+ "session_id_validator"
+ )
+ return normalized
+
@property
def request(self) -> Request:
"""
@@ -693,6 +826,7 @@ class App:
None # background multipart task if started
)
multipart_file_notifications: Optional[asyncio.Queue] = None
+ multipart_parse_error: Optional[BaseException] = None
async def _cancel_parse_task():
if parse_task is None:
@@ -714,16 +848,51 @@ class App:
await self._send_response(send, code, body, headers or [])
return
- async def _await_file_param(name: str) -> Optional[Any]:
+ async def _exit_for_multipart_error(error: BaseException) -> None:
+ if isinstance(error, MultipartFileError):
+ error = error.cause
+ if isinstance(error, _PayloadTooLarge):
+ await _early_exit(413, "413 Payload Too Large")
+ elif isinstance(error, _FormFieldTooLarge):
+ await _early_exit(
+ 413, "413 Payload Too Large: Multipart form field"
+ )
+ elif isinstance(error, TimeoutError):
+ await _early_exit(
+ 408, "408 Request Timeout: Failed to receive body"
+ )
+ else:
+ await _early_exit(400, "400 Bad Request: Malformed multipart body")
+
+ async def _finish_multipart_parse() -> bool:
+ if parse_task is None:
+ return True
+ try:
+ await parse_task
+ request.body_parsed = True
+ return True
+ except asyncio.CancelledError:
+ raise
+ except Exception as exc:
+ await _exit_for_multipart_error(exc)
+ return False
+
+ def _multipart_param(name: str) -> Optional[Any]:
+ values = request.body_params.get(name)
+ if values:
+ return values[0]
+ return request.files.get(name)
+
+ async def _await_multipart_param(name: str) -> Optional[Any]:
"""
- Wait until a multipart file field named `name` is available in request.files,
- or until the background parse_task (if any) completes.
+ Wait until a multipart text or file field is available, or until
+ the background parser completes.
"""
- if name in request.files:
- return request.files[name]
+ if (value := _multipart_param(name)) is not None:
+ return value
if parse_task is None or multipart_file_notifications is None:
return None
- while name not in request.files:
+ while True:
if parse_task.done():
try:
await parse_task
@@ -732,19 +901,34 @@ class App:
pass
break
await multipart_file_notifications.get()
- return request.files.get(name)
+ if (value := _multipart_param(name)) is not None:
+ return value
+ return _multipart_param(name)
try:
# Parse query/cookies/session
- try:
- request.query_params = self._parse_query_string(scope)
- except ValueError:
- await _early_exit(400, "400 Bad Request: Malformed query string")
- return
+ request.query_params = self._parse_query_string(scope)
cookie_header = request.headers.get("cookie", "")
cookies = self._parse_cookies(cookie_header) if cookie_header else {}
+ session_id = self._normalize_session_id(cookies.get("session_id"))
+ if session_id is None:
+ cookies.pop("session_id", None)
+ else:
+ cookies["session_id"] = session_id
request.session = await self._load_session_from_scope(scope, cookies)
+ session_was_empty = not request.session
content_type = request.headers.get("content-type", "")
+ content_length = request.headers.get("content-length")
+ if content_length is not None:
+ if not content_length.isascii() or not content_length.isdigit():
+ await _early_exit(400, "400 Bad Request: Invalid Content-Length")
+ return
+ if (
+ self.max_body_size is not None
+ and int(content_length) > self.max_body_size
+ ):
+ await _early_exit(413, "413 Payload Too Large")
+ return
# Body parsing setup
if (
@@ -765,21 +949,66 @@ class App:
return
multipart_file_notifications = asyncio.Queue()
+ def _terminate_multipart_file_queues(
+ error: BaseException,
+ ) -> None:
+ for file_info in request.files.values():
+ if not isinstance(file_info, dict):
+ continue
+ queue = file_info.get("content")
+ if isinstance(queue, _MultipartFileQueue):
+ queue.abort(error)
+ continue
+ if not isinstance(queue, asyncio.Queue):
+ continue
+ while True:
+ try:
+ queue.put_nowait(_MultipartStreamFailure(error))
+ break
+ except asyncio.QueueFull:
+ try:
+ queue.get_nowait()
+ except asyncio.QueueEmpty:
+ pass
+
async def _run_multipart_parser() -> None:
+ nonlocal multipart_parse_error
+ parser_completed = False
+ parser_error: Optional[BaseException] = None
try:
await self._parse_multipart_into_request(
receive,
boundary_match.group(1).encode("utf-8"),
request,
file_notifications=multipart_file_notifications,
- file_queue_maxsize=2048,
+ file_queue_maxsize=0,
+ max_body_size=self.max_body_size,
+ max_form_field_size=self.max_form_field_size,
+ body_timeout=self.body_timeout,
)
- except asyncio.CancelledError:
+ parser_completed = True
+ except asyncio.CancelledError as exc:
+ parser_error = exc
raise
- except Exception:
+ except (
+ _PayloadTooLarge,
+ _FormFieldTooLarge,
+ TimeoutError,
+ ) as exc:
+ parser_error = exc
+ multipart_parse_error = exc
+ raise
+ except Exception as exc:
+ parser_error = exc
+ multipart_parse_error = exc
logger.exception("Failed to parse multipart request body")
raise
finally:
+ if not parser_completed:
+ _terminate_multipart_file_queues(
+ parser_error
+ or RuntimeError("Multipart parser stopped")
+ )
multipart_file_notifications.put_nowait(None)
# Parse in the background so handlers can drain file queues
@@ -787,6 +1016,7 @@ class App:
parse_task = asyncio.create_task(_run_multipart_parser())
else:
body_chunks: List[bytes] = []
+ body_size = 0
try:
while True:
if self.body_timeout is None:
@@ -795,6 +1025,13 @@ class App:
async with asyncio.timeout(self.body_timeout):
msg = await receive()
if chunk := msg.get("body", b""):
+ body_size += len(chunk)
+ if (
+ self.max_body_size is not None
+ and body_size > self.max_body_size
+ ):
+ await _early_exit(413, "413 Payload Too Large")
+ return
body_chunks.append(chunk)
if not msg.get("more_body"):
break
@@ -814,21 +1051,15 @@ class App:
await _early_exit(400, "400 Bad Request: Bad JSON")
return
else:
- try:
- request.body_params = _parse_urlencoded(body_data)
- except ValueError:
- await _early_exit(
- 400, "400 Bad Request: Malformed form body"
- )
- return
+ request.body_params = _parse_urlencoded(body_data)
request.body_parsed = True
# HTTP middlewares (before)
for mw in self.middlewares:
if result := await mw.before_request(request):
status_code, response_body, extra_headers = (
- result["status_code"],
- result["body"],
+ result.get("status_code", status_code),
+ result.get("body", response_body),
result.get("headers", []),
)
await _early_exit(status_code, response_body, extra_headers)
@@ -836,8 +1067,10 @@ class App:
# Subapp handoff
if hasattr(request, "_subapp"):
- # If we started a multipart parse, cancel it before handing off
- await _cancel_parse_task()
+ # Complete multipart parsing before handoff. Passing a partially
+ # consumed receive() stream would start the child mid-boundary.
+ if not await _finish_multipart_parse():
+ return
new_scope = dict(scope)
new_scope["path"] = request._subapp_path
mount = getattr(request, "_subapp_mount_path", "").strip("/")
@@ -884,8 +1117,8 @@ class App:
return
handler_info = self._get_handler_info(handler)
- # Initialize func_args early to avoid UnboundLocalError
func_args: List[Any] = []
+ func_kwargs: Dict[str, Any] = {}
# Check if index handler accepts parameters (for non-root paths)
if handler == index_handler and path and path != "index":
@@ -904,38 +1137,56 @@ class App:
body_params = request.body_params
json_params = request._json if isinstance(request._json, dict) else {}
files = request.files
+ consumed_names = set()
+ accepts_var_kwargs = False
for param in handler_info.params:
if param.kind == _VAR_POSITIONAL:
func_args.extend(path_params[path_param_index:])
path_param_index = path_param_count
continue
+ if param.kind == _VAR_KEYWORD:
+ accepts_var_kwargs = True
+ continue
param_value = None
+ found = False
if path_param_index < path_param_count:
param_value = path_params[path_param_index]
path_param_index += 1
+ found = True
elif param.name in query_params:
param_value = query_params[param.name][0]
+ found = True
elif param.name in body_params:
param_value = body_params[param.name][0]
+ found = True
elif param.name in json_params:
param_value = json_params[param.name]
+ found = True
elif param.name in files:
param_value = files[param.name]
+ found = True
elif is_multipart:
- param_value = await _await_file_param(param.name)
- if param_value is None and param.default is _PARAM_EMPTY:
+ param_value = await _await_multipart_param(param.name)
+ if multipart_parse_error is not None:
+ await _exit_for_multipart_error(multipart_parse_error)
+ return
+ found = param_value is not None
+ if not found and param.default is _PARAM_EMPTY:
await _early_exit(
400,
- f"400 Bad Request: Missing required parameter '{param.name}'",
+ "400 Bad Request: Missing required parameter "
+ f"'{param.name}'",
)
return
- if param_value is None:
+ if not found:
param_value = param.default
+ found = True
elif param.default is not _PARAM_EMPTY:
param_value = param.default
+ found = True
else:
await _early_exit(
400,
@@ -943,15 +1194,48 @@ class App:
)
return
- func_args.append(param_value)
+ if found:
+ consumed_names.add(param.name)
+ if param.kind == _KEYWORD_ONLY:
+ func_kwargs[param.name] = param_value
+ else:
+ func_args.append(param_value)
+
+ if accepts_var_kwargs:
+ # Multipart dictionaries are populated concurrently. Finish the
+ # parse before snapshotting arbitrary extra names.
+ if is_multipart and not await _finish_multipart_parse():
+ return
+ named_sources = (
+ (query_params, lambda values: values[0], True),
+ (body_params, lambda values: values[0], True),
+ (json_params, lambda value: value, False),
+ (files, lambda value: value, False),
+ )
+ for source, first_value, requires_value in named_sources:
+ for name, value in tuple(source.items()):
+ if name in consumed_names or name in func_kwargs:
+ continue
+ if requires_value and not value:
+ continue
+ func_kwargs[name] = first_value(value)
+
+ # A parser that has already failed must prevent handler side effects,
+ # even when every argument came from query data or an early file part.
+ if multipart_parse_error is not None:
+ await _exit_for_multipart_error(multipart_parse_error)
+ return
# Execute handler
try:
result = (
- await handler(*func_args)
+ await handler(*func_args, **func_kwargs)
if handler_info.is_coroutine
- else handler(*func_args)
+ else handler(*func_args, **func_kwargs)
)
+ except MultipartFileError as exc:
+ await _exit_for_multipart_error(exc)
+ return
except Exception:
logger.exception(
"Unhandled exception in HTTP handler for %s", scope.get("path", "")
@@ -960,13 +1244,8 @@ class App:
return
# Ensure background parser (if any) is finished before finalizing response
- if parse_task is not None:
- try:
- await parse_task
- request.body_parsed = True
- except Exception:
- # The parser runner logs the original exception.
- pass
+ if not await _finish_multipart_parse():
+ return
# Normalize response
if isinstance(result, tuple):
@@ -976,6 +1255,8 @@ class App:
response_body = result
if isinstance(response_body, (dict, list)):
response_body = json.dumps(response_body)
+ if isinstance(response_body, bytes):
+ response_body = response_body.decode("utf-8")
extra_headers.append(("Content-Type", "application/json"))
# HTTP middlewares
@@ -990,12 +1271,13 @@ class App:
)
# Session persistence after middlewares so they can mutate request.session
- session_id = cookies.get("session_id")
-
if request.session:
- # New or updated session
+ rotate_session = request._regenerate_session or session_was_empty
+ if rotate_session and session_id:
+ await self.session_backend.save(session_id, {}, 0)
+ session_id = None
if not session_id:
- session_id = str(uuid.uuid4())
+ session_id = self._new_session_id()
extra_headers.append(
(
"Set-Cookie",
@@ -1049,6 +1331,7 @@ class App:
await gen.aclose()
streaming_task = asyncio.create_task(streamer())
+ msg_task: Optional[asyncio.Task] = None
try:
while True:
msg_task = asyncio.create_task(receive())
@@ -1057,9 +1340,18 @@ class App:
return_when=asyncio.FIRST_COMPLETED,
)
if streaming_task in done:
+ if not msg_task.done():
+ msg_task.cancel()
+ try:
+ await msg_task
+ except asyncio.CancelledError:
+ pass
+ msg_task = None
+ await streaming_task
break
if msg_task in done:
msg = msg_task.result()
+ msg_task = None
if msg["type"] == "http.disconnect":
streaming_task.cancel()
try:
@@ -1068,12 +1360,22 @@ class App:
pass
break
finally:
- if not streaming_task.done():
- streaming_task.cancel()
+ if msg_task is not None:
+ if not msg_task.done():
+ msg_task.cancel()
try:
- await streaming_task
+ await msg_task
except asyncio.CancelledError:
pass
+ except Exception:
+ # Retrieve receive() failures during cancellation.
+ pass
+ if not streaming_task.done():
+ streaming_task.cancel()
+ try:
+ await streaming_task
+ except asyncio.CancelledError:
+ pass
return
else:
await self._send_response(
@@ -1081,6 +1383,7 @@ class App:
)
finally:
+ await _cancel_parse_task()
current_request.reset(token)
async def _asgi_app_websocket(
@@ -1101,16 +1404,16 @@ class App:
token = current_request.set(request)
try:
# Parse request details (query params, cookies, session)
- try:
- request.query_params = self._parse_query_string(scope)
- except ValueError:
- await self._send_websocket_close(
- send, 1008, "Malformed query string"
- )
- return
+ request.query_params = self._parse_query_string(scope)
cookie_header = request.headers.get("cookie", "")
cookies = self._parse_cookies(cookie_header) if cookie_header else {}
+ session_id = self._normalize_session_id(cookies.get("session_id"))
+ if session_id is None:
+ cookies.pop("session_id", None)
+ else:
+ cookies["session_id"] = session_id
request.session = await self._load_session_from_scope(scope, cookies)
+ session_was_empty = not request.session
# Run WebSocket middleware before_websocket
for mw in self.ws_middlewares:
@@ -1147,23 +1450,47 @@ class App:
# Build function arguments
func_args: List[Any] = []
+ func_kwargs: Dict[str, Any] = {}
path_params = request.path_params
path_param_index = 0
path_param_count = len(path_params)
query_params = request.query_params
+ had_session_id = bool(session_id)
+ rotate_session = had_session_id and session_was_empty
+ if rotate_session:
+ await self.session_backend.save(session_id, {}, 0)
+ session_id = None
+ websocket_session_id = session_id or self._new_session_id()
ws = WebSocket(
receive,
send,
session_cookie_secure=self.session_cookie_secure,
+ session_id_validator=self._normalize_session_id,
)
- func_args.append(ws) # First non-self parameter is WebSocket object
- for param in handler_info.params:
- if param.name == "ws": # Skip the WebSocket parameter
- continue
+ ws.session_id = websocket_session_id
+ ws._session_cookie_pending = rotate_session or not had_session_id
+ if not handler_info.params:
+ await self._send_websocket_close(
+ send, 1011, "WebSocket handler must accept a connection argument"
+ )
+ return
+ websocket_param = handler_info.params[0]
+ if websocket_param.kind == _KEYWORD_ONLY:
+ func_kwargs[websocket_param.name] = ws
+ elif websocket_param.kind == _VAR_KEYWORD:
+ func_kwargs["ws"] = ws
+ else:
+ func_args.append(ws)
+ consumed_names = {websocket_param.name}
+ accepts_var_kwargs = websocket_param.kind == _VAR_KEYWORD
+ for param in handler_info.params[1:]:
if param.kind == _VAR_POSITIONAL:
func_args.extend(path_params[path_param_index:])
path_param_index = path_param_count
continue
+ if param.kind == _VAR_KEYWORD:
+ accepts_var_kwargs = True
+ continue
param_value = None
if path_param_index < path_param_count:
param_value = path_params[path_param_index]
@@ -1177,24 +1504,34 @@ class App:
send, 1008, f"Missing required parameter '{param.name}'"
)
return
- func_args.append(param_value)
+ consumed_names.add(param.name)
+ if param.kind == _KEYWORD_ONLY:
+ func_kwargs[param.name] = param_value
+ else:
+ func_args.append(param_value)
- # Set session ID if needed
- session_id = cookies.get("session_id")
- had_session_id = bool(session_id)
- ws.session_id = session_id or str(uuid.uuid4())
+ if accepts_var_kwargs:
+ for name, values in query_params.items():
+ if (
+ name not in consumed_names
+ and name not in func_kwargs
+ and values
+ ):
+ func_kwargs[name] = values[0]
# Execute handler
try:
- await handler(*func_args)
+ await handler(*func_args, **func_kwargs)
except ConnectionClosed:
pass # Normal closure, no need to send another close message
- except Exception as e:
+ except Exception:
logger.exception(
"Unhandled exception in WebSocket handler for %s",
scope.get("path", ""),
)
- await self._send_websocket_close(send, 1011, f"Handler error: {str(e)}")
+ await self._send_websocket_close(
+ send, 1011, "Internal server error"
+ )
return
# Run WebSocket middleware after_websocket
@@ -1239,12 +1576,15 @@ class App:
request: "Request",
*,
file_notifications: Optional[asyncio.Queue] = None,
- file_queue_maxsize: int = 2048,
+ file_queue_maxsize: int = 0,
+ max_body_size: Optional[int] = DEFAULT_MAX_BODY_SIZE,
+ max_form_field_size: Optional[int] = DEFAULT_MAX_FORM_FIELD_SIZE,
+ body_timeout: Optional[float] = 5.0,
) -> None:
"""
Parse multipart directly from ASGI receive() and populate
request.body_params / request.files as parts arrive.
- Uses bounded queues for file parts to apply backpressure.
+ The request-wide body limit bounds producer-nonblocking file queues.
"""
if request.body_params is None:
request.body_params = {}
@@ -1255,13 +1595,25 @@ class App:
current_field_name: Optional[str] = None
current_filename: Optional[str] = None
current_content_type: Optional[str] = None
- current_queue: Optional[asyncio.Queue] = None
+ current_queue: Optional[_MultipartFileQueue] = None
form_value: str = ""
+ form_value_size = 0
+ total_body_size = 0
while True:
- msg = await receive()
+ if body_timeout is None:
+ msg = await receive()
+ else:
+ async with asyncio.timeout(body_timeout):
+ msg = await receive()
body_chunk = msg.get("body", b"")
if body_chunk:
+ total_body_size += len(body_chunk)
+ if (
+ max_body_size is not None
+ and total_body_size > max_body_size
+ ):
+ raise _PayloadTooLarge()
for result in parser.parse(body_chunk):
if isinstance(result, MultipartSegment):
# New part
@@ -1269,10 +1621,11 @@ class App:
current_filename = result.filename
current_content_type = None
form_value = ""
+ form_value_size = 0
# Close previous file stream if open
if current_queue:
- await current_queue.put(None)
+ await current_queue.finish()
current_queue = None
# Pick up content-type for this part if present
@@ -1281,8 +1634,9 @@ class App:
current_content_type = value
if current_filename:
- # File field → bounded queue enforces backpressure
- current_queue = asyncio.Queue(
+ # The request-wide body cap bounds queued bytes while
+ # a nonblocking producer lets later parts be discovered.
+ current_queue = _MultipartFileQueue(
maxsize=file_queue_maxsize
)
request.files[current_field_name] = {
@@ -1299,21 +1653,31 @@ class App:
elif result:
# Part body
if current_queue:
- # May block here if handler isn't draining the queue
await current_queue.put(result)
else:
+ form_value_size += len(result)
+ if (
+ max_form_field_size is not None
+ and form_value_size > max_form_field_size
+ ):
+ raise _FormFieldTooLarge()
form_value += result.decode("utf-8", "ignore")
else:
# End of current part
if current_queue:
- await current_queue.put(None)
+ await current_queue.finish()
current_queue = None
else:
- if form_value and current_field_name:
+ if current_field_name:
request.body_params[current_field_name].append(
form_value
)
+ if file_notifications is not None:
+ file_notifications.put_nowait(
+ current_field_name
+ )
form_value = ""
+ form_value_size = 0
if not msg.get("more_body"):
break
@@ -1322,7 +1686,7 @@ class App:
if current_field_name and form_value and not current_filename:
request.body_params[current_field_name].append(form_value)
if current_queue:
- await current_queue.put(None)
+ await current_queue.finish()
def _prepare_response_headers(
self, extra_headers: Optional[List[Tuple[str, str]]] = None
@@ -1343,9 +1707,16 @@ class App:
"Rejected response header containing a newline: %r: %r", k, v
)
continue
+ try:
+ encoded_header = (k.encode("latin-1"), v.encode("latin-1"))
+ except UnicodeEncodeError:
+ logger.warning(
+ "Rejected response header outside latin-1: %r: %r", k, v
+ )
+ continue
if k.lower() == "content-type":
has_content_type = True
- sanitized_headers.append((k.encode("latin-1"), v.encode("latin-1")))
+ sanitized_headers.append(encoded_header)
if not has_content_type:
sanitized_headers.append(_DEFAULT_HEADER_BYTES)
return sanitized_headers
@@ -1385,12 +1756,26 @@ class App:
await send({"type": "http.response.body", "body": b"", "more_body": False})
return
if hasattr(body, "__iter__") and not isinstance(body, (bytes, str)):
- for chunk in body:
- if isinstance(chunk, str):
- chunk = chunk.encode("utf-8")
- await send(
- {"type": "http.response.body", "body": chunk, "more_body": True}
- )
+ iterator = iter(body)
+ try:
+ while True:
+ has_chunk, chunk = await asyncio.to_thread(
+ _next_sync_item, iterator
+ )
+ if not has_chunk:
+ break
+ if isinstance(chunk, str):
+ chunk = chunk.encode("utf-8")
+ await send(
+ {
+ "type": "http.response.body",
+ "body": chunk,
+ "more_body": True,
+ }
+ )
+ finally:
+ if hasattr(iterator, "close"):
+ await asyncio.to_thread(iterator.close)
await send({"type": "http.response.body", "body": b"", "more_body": False})
return
response_body = body if isinstance(body, bytes) else str(body).encode("utf-8")
@@ -1416,11 +1801,15 @@ class App:
Make a URL safe to put in an HTTP Location header.
Key rule: Location must be ASCII/latin-1 -> percent-encode non-ASCII.
- We percent-encode the PATH but we do NOT touch the query string.
+ Percent-encode non-ASCII URL components without double-encoding ``%``.
"""
p = urlsplit(url)
safe_path = quote(p.path, safe="/%")
- return urlunsplit((p.scheme, p.netloc, safe_path, p.query, p.fragment))
+ safe_query = quote(p.query, safe="=&?/:;+,%")
+ safe_fragment = quote(p.fragment, safe="=&?/:;+,%")
+ return urlunsplit(
+ (p.scheme, p.netloc, safe_path, safe_query, safe_fragment)
+ )
def _redirect(
self,
diff --git a/pyproject.toml b/pyproject.toml
index a696b28..259fde1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "flit_core.buildapi"
[project]
name = "micropie"
-version = "0.31"
+version = "0.32"
description = "An ultra micro ASGI web framework"
keywords = ["micropie", "asgi", "microframework", "http"]
readme = "docs/README.md"
diff --git a/tests.py b/tests.py
index 54de270..d5890ab 100644
--- a/tests.py
+++ b/tests.py
@@ -1,21 +1,30 @@
import asyncio
import functools
import json as std_json
+import threading
import unittest
import uuid
from unittest.mock import AsyncMock, patch
from urllib.parse import parse_qs
from micropie import (
App,
+ DEFAULT_MAX_BODY_SIZE,
+ DEFAULT_MAX_FORM_FIELD_SIZE,
InMemorySessionBackend,
+ MULTIPART_INSTALLED,
Request,
WebSocketRequest,
SESSION_TIMEOUT,
ConnectionClosed,
HttpMiddleware,
+ MultipartFileError,
)
+VALID_SESSION_ID = "00000000-0000-4000-8000-000000000001"
+ATTACKER_SESSION_ID = "00000000-0000-4000-8000-000000000002"
+
+
class CountingSessionBackend(InMemorySessionBackend):
def __init__(self):
super().__init__()
@@ -299,14 +308,194 @@ class TestSession(MicroPieTestCase):
self.assertIn(b"HttpOnly", cookie)
self.assertNotIn(b"Secure", cookie)
- async def test_app_rejects_invalid_timeout_configuration(self):
- """Enabled body and session timeouts must be positive."""
+ async def test_empty_session_transition_rotates_client_supplied_id(self):
+ """A login must never persist under an unknown ID chosen by the client."""
+ backend = CountingSessionBackend()
+ self.app = App(session_backend=backend)
+
+ async def login(self):
+ self.request.session["user_id"] = 42
+ return "logged in"
+
+ setattr(self.app, "login", login.__get__(self.app, App))
+ scope = self.create_mock_scope(
+ path="/login",
+ headers=[
+ (
+ b"cookie",
+ f"session_id={ATTACKER_SESSION_ID}".encode("ascii"),
+ )
+ ],
+ )
+ receive = AsyncMock(
+ return_value={"type": "http.request", "body": b"", "more_body": False}
+ )
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ self.assertEqual(backend.load_calls, [ATTACKER_SESSION_ID])
+ self.assertIn((ATTACKER_SESSION_ID, {}, 0), backend.save_calls)
+ start = next(
+ call[0][0]
+ for call in send.call_args_list
+ if call[0][0]["type"] == "http.response.start"
+ )
+ cookie = next(
+ value for name, value in start["headers"] if name == b"Set-Cookie"
+ ).decode("ascii")
+ new_session_id = cookie.split(";", 1)[0].split("=", 1)[1]
+ self.assertNotEqual(new_session_id, ATTACKER_SESSION_ID)
+ self.assertEqual(uuid.UUID(new_session_id).version, 4)
+ self.assertNotIn(ATTACKER_SESSION_ID, backend.sessions)
+ self.assertEqual(backend.sessions[new_session_id], {"user_id": 42})
+
+ async def test_invalid_session_cookie_never_reaches_backend(self):
+ """Malformed or oversized backend keys should be discarded before load()."""
+ backend = CountingSessionBackend()
+ self.app = App(session_backend=backend)
+
+ async def login(self):
+ self.request.session["user_id"] = 42
+ return "logged in"
+
+ setattr(self.app, "login", login.__get__(self.app, App))
+ scope = self.create_mock_scope(
+ path="/login",
+ headers=[(b"cookie", b"session_id=../../tmp/attacker-session")],
+ )
+ receive = AsyncMock(
+ return_value={"type": "http.request", "body": b"", "more_body": False}
+ )
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ self.assertEqual(backend.load_calls, [])
+ self.assertTrue(backend.save_calls)
+ saved_session_id = backend.save_calls[-1][0]
+ self.assertEqual(uuid.UUID(saved_session_id).version, 4)
+ self.assertNotEqual(saved_session_id, "../../tmp/attacker-session")
+
+ async def test_explicit_session_regeneration_rotates_existing_session(self):
+ """Apps can rotate a populated session during a privilege change."""
+ backend = CountingSessionBackend()
+ await backend.save(VALID_SESSION_ID, {"cart": "item"}, SESSION_TIMEOUT)
+ backend.load_calls.clear()
+ backend.save_calls.clear()
+ self.app = App(session_backend=backend)
+
+ async def reauthenticate(self):
+ self.request.session["user_id"] = 42
+ self.request.regenerate_session()
+ return "reauthenticated"
+
+ setattr(
+ self.app,
+ "reauthenticate",
+ reauthenticate.__get__(self.app, App),
+ )
+ scope = self.create_mock_scope(
+ path="/reauthenticate",
+ headers=[
+ (b"cookie", f"session_id={VALID_SESSION_ID}".encode("ascii"))
+ ],
+ )
+ receive = AsyncMock(
+ return_value={"type": "http.request", "body": b"", "more_body": False}
+ )
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ self.assertIn((VALID_SESSION_ID, {}, 0), backend.save_calls)
+ self.assertNotIn(VALID_SESSION_ID, backend.sessions)
+ new_session_id = backend.save_calls[-1][0]
+ self.assertNotEqual(new_session_id, VALID_SESSION_ID)
+ self.assertEqual(
+ backend.sessions[new_session_id], {"cart": "item", "user_id": 42}
+ )
+ start = next(
+ call[0][0]
+ for call in send.call_args_list
+ if call[0][0]["type"] == "http.response.start"
+ )
+ cookie = next(
+ value for name, value in start["headers"] if name == b"Set-Cookie"
+ )
+ self.assertIn(f"session_id={new_session_id}".encode("ascii"), cookie)
+
+ async def test_custom_session_id_policy_supports_non_uuid_backends(self):
+ """Custom backends can opt into their own validated ID format."""
+ backend = CountingSessionBackend()
+ await backend.save("signed.existing", {"user": "alice"}, SESSION_TIMEOUT)
+ backend.load_calls.clear()
+ backend.save_calls.clear()
+ self.app = App(
+ session_backend=backend,
+ session_id_factory=lambda: "signed.fresh",
+ session_id_validator=lambda value: (
+ value if value.startswith("signed.") else None
+ ),
+ )
+
+ async def profile(self):
+ return self.request.session["user"]
+
+ setattr(self.app, "profile", profile.__get__(self.app, App))
+ scope = self.create_mock_scope(
+ path="/profile",
+ headers=[(b"cookie", b"session_id=signed.existing")],
+ )
+ send = AsyncMock()
+
+ await self.app(
+ scope,
+ AsyncMock(
+ return_value={
+ "type": "http.request",
+ "body": b"",
+ "more_body": False,
+ }
+ ),
+ send,
+ )
+
+ self.assertEqual(backend.load_calls, ["signed.existing"])
+ self.assertEqual(backend.save_calls[-1][0], "signed.existing")
+ send.assert_any_call(
+ {"type": "http.response.body", "body": b"alice", "more_body": False}
+ )
+
+ async def test_app_rejects_invalid_limit_configuration(self):
+ """Enabled body limits and session timeouts must be positive."""
with self.assertRaisesRegex(ValueError, "body_timeout"):
App(body_timeout=0)
+ with self.assertRaisesRegex(ValueError, "max_body_size"):
+ App(max_body_size=0)
+ with self.assertRaisesRegex(ValueError, "max_form_field_size"):
+ App(max_form_field_size=0)
with self.assertRaisesRegex(ValueError, "session_timeout"):
App(session_timeout=0)
-
- self.assertIsNone(App(body_timeout=None).body_timeout)
+ with self.assertRaisesRegex(ValueError, "session_id_factory"):
+ App(session_id_factory=None)
+ with self.assertRaisesRegex(ValueError, "session_id_validator"):
+ App(session_id_validator="not-callable")
+
+ unbounded = App(
+ body_timeout=None,
+ max_body_size=None,
+ max_form_field_size=None,
+ )
+ self.assertIsNone(unbounded.body_timeout)
+ self.assertIsNone(unbounded.max_body_size)
+ self.assertIsNone(unbounded.max_form_field_size)
+
+ defaults = App()
+ self.assertEqual(defaults.max_body_size, DEFAULT_MAX_BODY_SIZE)
+ self.assertEqual(
+ defaults.max_form_field_size, DEFAULT_MAX_FORM_FIELD_SIZE
+ )
class TestFastPaths(MicroPieTestCase):
@@ -336,7 +525,7 @@ class TestFastPaths(MicroPieTestCase):
async def test_request_with_session_cookie_loads_and_persists(self):
"""A session cookie should still load and save the matching session."""
backend = CountingSessionBackend()
- await backend.save("abc", {"user": "alice"}, SESSION_TIMEOUT)
+ await backend.save(VALID_SESSION_ID, {"user": "alice"}, SESSION_TIMEOUT)
backend.load_calls.clear()
backend.save_calls.clear()
self.app = App(session_backend=backend)
@@ -348,7 +537,10 @@ class TestFastPaths(MicroPieTestCase):
setattr(self.app, "index", index.__get__(self.app, App))
scope = self.create_mock_scope(
- path="/index", headers=[(b"cookie", b"session_id=abc")]
+ path="/index",
+ headers=[
+ (b"cookie", f"session_id={VALID_SESSION_ID}".encode("ascii"))
+ ],
)
receive = AsyncMock(
return_value={"type": "http.request", "body": b"", "more_body": False}
@@ -357,9 +549,11 @@ class TestFastPaths(MicroPieTestCase):
await self.app(scope, receive, send)
- self.assertEqual(backend.load_calls, ["abc"])
- self.assertEqual(backend.save_calls[-1][0], "abc")
- self.assertEqual(backend.sessions["abc"], {"user": "alice", "seen": "1"})
+ self.assertEqual(backend.load_calls, [VALID_SESSION_ID])
+ self.assertEqual(backend.save_calls[-1][0], VALID_SESSION_ID)
+ self.assertEqual(
+ backend.sessions[VALID_SESSION_ID], {"user": "alice", "seen": "1"}
+ )
send.assert_any_call(
{"type": "http.response.body", "body": b"alice", "more_body": False}
)
@@ -521,112 +715,107 @@ class TestFastPaths(MicroPieTestCase):
send.assert_any_call({"type": "websocket.send", "text": "lobby:alice"})
+ async def test_http_keyword_only_and_var_keyword_binding(self):
+ """Keyword-only parameters and **kwargs should bind by name."""
-class TestRequestInputParsing(MicroPieTestCase):
- """Tests for strict URL-encoded query and form parsing."""
-
- async def test_blank_query_and_form_values_are_preserved(self):
- """Valid empty values should bind instead of being silently dropped."""
-
- async def query_value(self, value):
- return f"query:<{value}>"
+ async def options(self, *, name="guest", **kwargs):
+ return {"name": name, "extras": kwargs}
- async def form_value(self, value):
- return f"form:<{value}>"
-
- setattr(self.app, "query_value", query_value.__get__(self.app, App))
- setattr(self.app, "form_value", form_value.__get__(self.app, App))
+ setattr(self.app, "options", options.__get__(self.app, App))
+ payload = {"tags": [], "enabled": True}
+ scope = self.create_mock_scope(
+ path="/options",
+ method="POST",
+ headers=[(b"content-type", b"application/json")],
+ query_string=b"name=alice&debug=1",
+ )
+ send = AsyncMock()
- query_send = AsyncMock()
await self.app(
- self.create_mock_scope(path="/query_value", query_string=b"value="),
+ scope,
AsyncMock(
return_value={
"type": "http.request",
- "body": b"",
+ "body": std_json.dumps(payload).encode("utf-8"),
"more_body": False,
}
),
- query_send,
+ send,
+ )
+
+ body = next(
+ call[0][0]["body"]
+ for call in send.call_args_list
+ if call[0][0]["type"] == "http.response.body"
)
- query_send.assert_any_call(
+ self.assertEqual(
+ std_json.loads(body),
{
- "type": "http.response.body",
- "body": b"query:<>",
- "more_body": False,
- }
+ "name": "alice",
+ "extras": {"debug": "1", "tags": [], "enabled": True},
+ },
+ )
+
+ async def test_websocket_keyword_only_and_var_keyword_binding(self):
+ """WebSocket keyword-only parameters and **kwargs use query values."""
+
+ async def ws_options(self, connection, *, room="lobby", **kwargs):
+ await connection.accept()
+ await connection.send_text(f"{room}:{kwargs['debug']}")
+ await connection.close()
+
+ setattr(self.app, "ws_options", ws_options.__get__(self.app, App))
+ scope = self.create_mock_scope(
+ path="/options",
+ query_string=b"room=staff&debug=1",
+ scope_type="websocket",
)
+ send = AsyncMock()
- form_send = AsyncMock()
await self.app(
- self.create_mock_scope(
- path="/form_value",
- method="POST",
- headers=[
- (b"content-type", b"application/x-www-form-urlencoded")
- ],
- ),
- AsyncMock(
- return_value={
- "type": "http.request",
- "body": b"value=",
- "more_body": False,
- }
- ),
- form_send,
+ scope,
+ AsyncMock(return_value={"type": "websocket.connect"}),
+ send,
)
- form_send.assert_any_call(
- {
- "type": "http.response.body",
- "body": b"form:<>",
- "more_body": False,
- }
+
+ send.assert_any_call({"type": "websocket.send", "text": "staff:1"})
+
+
+class TestRequestInputParsing(MicroPieTestCase):
+ """Tests for consistent tolerant URL-encoded parsing."""
+
+ async def test_query_and_form_bodies_share_legacy_tolerant_parsing(self):
+ """The same URL-encoded bytes should produce the same parsed mapping."""
+ cases = (
+ (b"value=", {}),
+ (b"value", {}),
+ (b"value=%ZZ", {"value": ["%ZZ"]}),
+ (b"value=%FF", {"value": ["\ufffd"]}),
+ (b"value=before\xffafter", {"value": ["beforeafter"]}),
)
- async def test_malformed_query_strings_return_400(self):
- """Malformed fields and percent encoding should be rejected."""
- for query_string in (b"missing-equals", b"name=%ZZ", b"name=%FF"):
- with self.subTest(query_string=query_string):
- send = AsyncMock()
- await self.app(
- self.create_mock_scope(
- path="/index", query_string=query_string
- ),
- AsyncMock(
- return_value={
- "type": "http.request",
- "body": b"",
- "more_body": False,
- }
- ),
- send,
- )
+ parsed_bodies = []
- send.assert_any_call(
- {
- "type": "http.response.start",
- "status": 400,
- "headers": [
- (b"Content-Type", b"text/html; charset=utf-8")
- ],
- }
- )
- send.assert_any_call(
- {
- "type": "http.response.body",
- "body": b"400 Bad Request: Malformed query string",
- "more_body": False,
- }
- )
+ async def inspect_form(self):
+ parsed_bodies.append(self.request.body_params)
+ return "OK"
- async def test_malformed_form_bodies_return_400(self):
- """Malformed URL-encoded form fields should be rejected."""
- for body in (b"missing-equals", b"name=%ZZ", b"name=%FF"):
- with self.subTest(body=body):
+ setattr(
+ self.app,
+ "inspect_form",
+ inspect_form.__get__(self.app, App),
+ )
+
+ for encoded, expected in cases:
+ with self.subTest(encoded=encoded):
+ self.assertEqual(
+ self.app._parse_query_string({"query_string": encoded}),
+ expected,
+ )
send = AsyncMock()
await self.app(
self.create_mock_scope(
- path="/index",
+ path="/inspect_form",
method="POST",
headers=[
(
@@ -638,26 +827,17 @@ class TestRequestInputParsing(MicroPieTestCase):
AsyncMock(
return_value={
"type": "http.request",
- "body": body,
+ "body": encoded,
"more_body": False,
}
),
send,
)
-
- send.assert_any_call(
- {
- "type": "http.response.start",
- "status": 400,
- "headers": [
- (b"Content-Type", b"text/html; charset=utf-8")
- ],
- }
- )
+ self.assertEqual(parsed_bodies[-1], expected)
send.assert_any_call(
{
"type": "http.response.body",
- "body": b"400 Bad Request: Malformed form body",
+ "body": b"OK",
"more_body": False,
}
)
@@ -757,79 +937,337 @@ class TestRequestBodyTimeouts(MicroPieTestCase):
)
-class TestMultipartCoordination(MicroPieTestCase):
- """Tests for notification-driven multipart file binding."""
+class TestRequestBodyLimits(MicroPieTestCase):
+ """Tests for declared and streamed request-body limits."""
- def multipart_scope(self):
- return self.create_mock_scope(
- path="/upload",
+ async def test_content_length_rejects_oversized_body_before_receive(self):
+ """An oversized declared body should be rejected without buffering it."""
+ self.app = App(max_body_size=4)
+ scope = self.create_mock_scope(
+ path="/index",
method="POST",
headers=[
- (b"content-type", b"multipart/form-data; boundary=test-boundary")
+ (b"content-type", b"application/json"),
+ (b"content-length", b"5"),
],
)
+ receive = AsyncMock()
+ send = AsyncMock()
- async def test_file_notification_starts_handler_before_parse_completion(self):
- """Handlers can drain a file queue while the parser is still active."""
- parser_may_finish = asyncio.Event()
- handler_received_file = asyncio.Event()
+ await self.app(scope, receive, send)
- async def fake_parse(
- receive,
- boundary,
- request,
- *,
- file_notifications=None,
- file_queue_maxsize=2048,
- ):
- content = asyncio.Queue(maxsize=file_queue_maxsize)
- request.files["upload"] = {
- "filename": "example.txt",
- "content_type": "text/plain",
- "content": content,
+ receive.assert_not_awaited()
+ send.assert_any_call(
+ {
+ "type": "http.response.start",
+ "status": 413,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")],
}
- file_notifications.put_nowait("upload")
- await content.put(b"streamed")
- await content.put(None)
- await parser_may_finish.wait()
-
- self.app._parse_multipart_into_request = fake_parse
-
- async def upload(self, upload):
- chunk = await upload["content"].get()
- handler_received_file.set()
- parser_may_finish.set()
- return chunk
+ )
- setattr(self.app, "upload", upload.__get__(self.app, App))
+ async def test_invalid_content_length_returns_400(self):
+ """Invalid Content-Length values should not bypass early validation."""
+ self.app = App(max_body_size=4)
+ scope = self.create_mock_scope(
+ path="/index",
+ method="POST",
+ headers=[(b"content-length", b"not-a-number")],
+ )
send = AsyncMock()
- with patch("micropie.MULTIPART_INSTALLED", True):
- await asyncio.wait_for(
- self.app(self.multipart_scope(), AsyncMock(), send), timeout=1
- )
+ await self.app(scope, AsyncMock(), send)
- self.assertTrue(handler_received_file.is_set())
send.assert_any_call(
{
"type": "http.response.body",
- "body": b"streamed",
+ "body": b"400 Bad Request: Invalid Content-Length",
"more_body": False,
}
)
- async def test_parser_completion_wakes_missing_file_waiter(self):
- """A missing file should resolve to a 400 without polling or hanging."""
-
- async def fake_parse(
- receive,
- boundary,
- request,
- *,
- file_notifications=None,
- file_queue_maxsize=2048,
- ):
- return None
+ async def test_streamed_body_is_counted_without_content_length(self):
+ """Chunked bodies should receive 413 as soon as actual bytes exceed the cap."""
+ self.app = App(max_body_size=8)
+ chunks = [
+ {"type": "http.request", "body": b"value=", "more_body": True},
+ {"type": "http.request", "body": b"toolong", "more_body": False},
+ ]
+
+ async def receive():
+ return chunks.pop(0)
+
+ scope = self.create_mock_scope(
+ path="/index",
+ method="POST",
+ headers=[(b"content-type", b"application/x-www-form-urlencoded")],
+ )
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ send.assert_any_call(
+ {
+ "type": "http.response.body",
+ "body": b"413 Payload Too Large",
+ "more_body": False,
+ }
+ )
+
+ async def test_body_at_exact_limit_is_allowed(self):
+ """The configured size is inclusive; only larger bodies are rejected."""
+ body = b"value=ok"
+ self.app = App(max_body_size=len(body))
+
+ async def submit(self, value):
+ return value
+
+ setattr(self.app, "submit", submit.__get__(self.app, App))
+ scope = self.create_mock_scope(
+ path="/submit",
+ method="POST",
+ headers=[(b"content-type", b"application/x-www-form-urlencoded")],
+ )
+ send = AsyncMock()
+
+ await self.app(
+ scope,
+ AsyncMock(
+ return_value={
+ "type": "http.request",
+ "body": body,
+ "more_body": False,
+ }
+ ),
+ send,
+ )
+
+ send.assert_any_call(
+ {"type": "http.response.body", "body": b"ok", "more_body": False}
+ )
+
+ @unittest.skipUnless(MULTIPART_INSTALLED, "multipart package is not installed")
+ async def test_multipart_text_field_has_independent_limit(self):
+ """A large text field should fail even when total upload size is allowed."""
+ body = (
+ b"--test-boundary\r\n"
+ b'Content-Disposition: form-data; name="note"\r\n\r\n'
+ b"12345\r\n"
+ b"--test-boundary--\r\n"
+ )
+ self.app = App(
+ max_body_size=len(body) + 1,
+ max_form_field_size=4,
+ )
+
+ async def upload(self, note):
+ return note
+
+ setattr(self.app, "upload", upload.__get__(self.app, App))
+ scope = self.create_mock_scope(
+ path="/upload",
+ method="POST",
+ headers=[
+ (b"content-type", b"multipart/form-data; boundary=test-boundary")
+ ],
+ )
+ send = AsyncMock()
+
+ await asyncio.wait_for(
+ self.app(
+ scope,
+ AsyncMock(
+ return_value={
+ "type": "http.request",
+ "body": body,
+ "more_body": False,
+ }
+ ),
+ send,
+ ),
+ timeout=1,
+ )
+
+ send.assert_any_call(
+ {
+ "type": "http.response.body",
+ "body": b"413 Payload Too Large: Multipart form field",
+ "more_body": False,
+ }
+ )
+
+ @unittest.skipUnless(MULTIPART_INSTALLED, "multipart package is not installed")
+ async def test_multipart_limit_terminates_open_file_queue(self):
+ """A mid-upload 413 should wake a handler already draining the file queue."""
+ prefix = (
+ b"--test-boundary\r\n"
+ b'Content-Disposition: form-data; name="file"; filename="a.txt"\r\n'
+ b"Content-Type: text/plain\r\n\r\n"
+ )
+ first_chunk = prefix + b"hello"
+ final_chunk = b" world\r\n--test-boundary--\r\n"
+ self.app = App(max_body_size=len(first_chunk) + 1)
+ handler_saw_failure = asyncio.Event()
+
+ async def upload(self, file):
+ try:
+ while await file["content"].get() is not None:
+ pass
+ except MultipartFileError:
+ handler_saw_failure.set()
+ raise
+ raise AssertionError("A truncated file was reported as complete")
+
+ setattr(self.app, "upload", upload.__get__(self.app, App))
+ chunks = [
+ {
+ "type": "http.request",
+ "body": first_chunk,
+ "more_body": True,
+ },
+ {
+ "type": "http.request",
+ "body": final_chunk,
+ "more_body": False,
+ },
+ ]
+
+ async def receive():
+ await asyncio.sleep(0)
+ return chunks.pop(0)
+
+ scope = self.create_mock_scope(
+ path="/upload",
+ method="POST",
+ headers=[
+ (b"content-type", b"multipart/form-data; boundary=test-boundary")
+ ],
+ )
+ send = AsyncMock()
+
+ await asyncio.wait_for(self.app(scope, receive, send), timeout=1)
+
+ self.assertTrue(handler_saw_failure.is_set())
+ send.assert_any_call(
+ {
+ "type": "http.response.start",
+ "status": 413,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")],
+ }
+ )
+
+ @unittest.skipUnless(MULTIPART_INSTALLED, "multipart package is not installed")
+ async def test_stalled_multipart_receive_returns_408(self):
+ """Multipart receive calls use the same per-chunk idle timeout."""
+ self.app = App(body_timeout=0.01)
+
+ async def upload(self, file):
+ return "unreachable"
+
+ setattr(self.app, "upload", upload.__get__(self.app, App))
+
+ async def receive():
+ await asyncio.sleep(0.05)
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ scope = self.create_mock_scope(
+ path="/upload",
+ method="POST",
+ headers=[
+ (b"content-type", b"multipart/form-data; boundary=test-boundary")
+ ],
+ )
+ send = AsyncMock()
+
+ await asyncio.wait_for(self.app(scope, receive, send), timeout=1)
+
+ send.assert_any_call(
+ {
+ "type": "http.response.body",
+ "body": b"408 Request Timeout: Failed to receive body",
+ "more_body": False,
+ }
+ )
+
+
+class TestMultipartCoordination(MicroPieTestCase):
+ """Tests for notification-driven multipart file binding."""
+
+ def multipart_scope(self):
+ return self.create_mock_scope(
+ path="/upload",
+ method="POST",
+ headers=[
+ (b"content-type", b"multipart/form-data; boundary=test-boundary")
+ ],
+ )
+
+ async def test_file_notification_starts_handler_before_parse_completion(self):
+ """Handlers can drain a file queue while the parser is still active."""
+ parser_may_finish = asyncio.Event()
+ handler_received_file = asyncio.Event()
+
+ async def fake_parse(
+ receive,
+ boundary,
+ request,
+ *,
+ file_notifications=None,
+ file_queue_maxsize=2048,
+ max_body_size=None,
+ max_form_field_size=None,
+ body_timeout=None,
+ ):
+ content = asyncio.Queue(maxsize=file_queue_maxsize)
+ request.files["upload"] = {
+ "filename": "example.txt",
+ "content_type": "text/plain",
+ "content": content,
+ }
+ file_notifications.put_nowait("upload")
+ await content.put(b"streamed")
+ await content.put(None)
+ await parser_may_finish.wait()
+
+ self.app._parse_multipart_into_request = fake_parse
+
+ async def upload(self, upload):
+ chunk = await upload["content"].get()
+ handler_received_file.set()
+ parser_may_finish.set()
+ return chunk
+
+ setattr(self.app, "upload", upload.__get__(self.app, App))
+ send = AsyncMock()
+
+ with patch("micropie.MULTIPART_INSTALLED", True):
+ await asyncio.wait_for(
+ self.app(self.multipart_scope(), AsyncMock(), send), timeout=1
+ )
+
+ self.assertTrue(handler_received_file.is_set())
+ send.assert_any_call(
+ {
+ "type": "http.response.body",
+ "body": b"streamed",
+ "more_body": False,
+ }
+ )
+
+ async def test_parser_completion_wakes_missing_file_waiter(self):
+ """A missing file should resolve to a 400 without polling or hanging."""
+
+ async def fake_parse(
+ receive,
+ boundary,
+ request,
+ *,
+ file_notifications=None,
+ file_queue_maxsize=2048,
+ max_body_size=None,
+ max_form_field_size=None,
+ body_timeout=None,
+ ):
+ return None
self.app._parse_multipart_into_request = fake_parse
@@ -862,6 +1300,9 @@ class TestMultipartCoordination(MicroPieTestCase):
*,
file_notifications=None,
file_queue_maxsize=2048,
+ max_body_size=None,
+ max_form_field_size=None,
+ body_timeout=None,
):
raise RuntimeError("broken multipart body")
@@ -894,51 +1335,271 @@ class TestMultipartCoordination(MicroPieTestCase):
}
)
- async def test_early_response_cancels_and_retrieves_parser(self):
- """Middleware short-circuiting should cancel the background parser cleanly."""
- parser_started = asyncio.Event()
- parser_cancelled = asyncio.Event()
+ async def test_early_response_cancels_and_retrieves_parser(self):
+ """Middleware short-circuiting should cancel the background parser cleanly."""
+ parser_started = asyncio.Event()
+ parser_cancelled = asyncio.Event()
+
+ async def fake_parse(
+ receive,
+ boundary,
+ request,
+ *,
+ file_notifications=None,
+ file_queue_maxsize=2048,
+ max_body_size=None,
+ max_form_field_size=None,
+ body_timeout=None,
+ ):
+ parser_started.set()
+ try:
+ await asyncio.Future()
+ except asyncio.CancelledError:
+ parser_cancelled.set()
+ raise
+
+ class RejectUpload(HttpMiddleware):
+ async def before_request(self, request):
+ await parser_started.wait()
+ return {"status_code": 413, "body": "too large"}
+
+ async def after_request(
+ self, request, status_code, response_body, extra_headers
+ ):
+ return None
+
+ self.app._parse_multipart_into_request = fake_parse
+ self.app.middlewares.append(RejectUpload())
+ send = AsyncMock()
+
+ with patch("micropie.MULTIPART_INSTALLED", True):
+ await asyncio.wait_for(
+ self.app(self.multipart_scope(), AsyncMock(), send), timeout=1
+ )
+
+ self.assertTrue(parser_cancelled.is_set())
+ send.assert_any_call(
+ {
+ "type": "http.response.start",
+ "status": 413,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")],
+ }
+ )
+
+ async def test_large_first_file_does_not_block_second_file_binding(self):
+ """Multi-file binding must not depend on the first queue being drained."""
+
+ async def fake_parse(
+ receive,
+ boundary,
+ request,
+ *,
+ file_notifications=None,
+ file_queue_maxsize=2048,
+ max_body_size=None,
+ max_form_field_size=None,
+ body_timeout=None,
+ ):
+ self.assertEqual(file_queue_maxsize, 0)
+ first = asyncio.Queue(maxsize=file_queue_maxsize)
+ request.files["file1"] = {
+ "filename": "one.bin",
+ "content_type": "application/octet-stream",
+ "content": first,
+ }
+ file_notifications.put_nowait("file1")
+ for _ in range(4096):
+ await first.put(b"x")
+ await first.put(None)
+
+ second = asyncio.Queue(maxsize=file_queue_maxsize)
+ request.files["file2"] = {
+ "filename": "two.bin",
+ "content_type": "application/octet-stream",
+ "content": second,
+ }
+ file_notifications.put_nowait("file2")
+ await second.put(b"y")
+ await second.put(None)
+
+ self.app._parse_multipart_into_request = fake_parse
+
+ async def upload(self, file1, file2):
+ first_size = 0
+ while (chunk := await file1["content"].get()) is not None:
+ first_size += len(chunk)
+ second_chunk = await file2["content"].get()
+ return f"{first_size}:{second_chunk.decode()}"
+
+ setattr(self.app, "upload", upload.__get__(self.app, App))
+ send = AsyncMock()
+
+ with patch("micropie.MULTIPART_INSTALLED", True):
+ await asyncio.wait_for(
+ self.app(self.multipart_scope(), AsyncMock(), send), timeout=1
+ )
+
+ send.assert_any_call(
+ {
+ "type": "http.response.body",
+ "body": b"4096:y",
+ "more_body": False,
+ }
+ )
+
+ async def test_handler_may_return_without_draining_file_queue(self):
+ """Parser completion cannot deadlock behind an abandoned file queue."""
+ handler_returned = asyncio.Event()
+
+ async def fake_parse(
+ receive,
+ boundary,
+ request,
+ *,
+ file_notifications=None,
+ file_queue_maxsize=2048,
+ max_body_size=None,
+ max_form_field_size=None,
+ body_timeout=None,
+ ):
+ content = asyncio.Queue(maxsize=file_queue_maxsize)
+ request.files["file"] = {
+ "filename": "large.bin",
+ "content_type": "application/octet-stream",
+ "content": content,
+ }
+ file_notifications.put_nowait("file")
+ await handler_returned.wait()
+ for _ in range(4096):
+ await content.put(b"x")
+ await content.put(None)
+
+ self.app._parse_multipart_into_request = fake_parse
+
+ async def upload(self, file):
+ handler_returned.set()
+ return "ignored"
+
+ setattr(self.app, "upload", upload.__get__(self.app, App))
+ send = AsyncMock()
+
+ with patch("micropie.MULTIPART_INSTALLED", True):
+ await asyncio.wait_for(
+ self.app(self.multipart_scope(), AsyncMock(), send), timeout=1
+ )
+
+ send.assert_any_call(
+ {"type": "http.response.body", "body": b"ignored", "more_body": False}
+ )
+
+ async def test_known_parser_failure_prevents_handler_execution(self):
+ """An already-failed parser must block query-only handler side effects."""
+ parser_failed = asyncio.Event()
+ handler_called = False
+
+ async def fake_parse(
+ receive,
+ boundary,
+ request,
+ *,
+ file_notifications=None,
+ file_queue_maxsize=2048,
+ max_body_size=None,
+ max_form_field_size=None,
+ body_timeout=None,
+ ):
+ parser_failed.set()
+ raise RuntimeError("broken multipart body")
+
+ class WaitForFailure(HttpMiddleware):
+ async def before_request(self, request):
+ await parser_failed.wait()
+
+ async def after_request(
+ self, request, status_code, response_body, extra_headers
+ ):
+ return None
+
+ self.app._parse_multipart_into_request = fake_parse
+ self.app.middlewares.append(WaitForFailure())
+
+ async def upload(self, value):
+ nonlocal handler_called
+ handler_called = True
+ return value
+
+ setattr(self.app, "upload", upload.__get__(self.app, App))
+ scope = self.multipart_scope()
+ scope["query_string"] = b"value=unsafe"
+ send = AsyncMock()
+
+ with patch("micropie.MULTIPART_INSTALLED", True), self.assertLogs(
+ "micropie", level="ERROR"
+ ):
+ await asyncio.wait_for(self.app(scope, AsyncMock(), send), timeout=1)
+
+ self.assertFalse(handler_called)
+ send.assert_any_call(
+ {
+ "type": "http.response.start",
+ "status": 400,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")],
+ }
+ )
+
+ @unittest.skipUnless(MULTIPART_INSTALLED, "multipart package is not installed")
+ async def test_multipart_subapp_handoff_uses_completed_parse(self):
+ """A child app must never receive a multipart stream mid-boundary."""
+ child = App()
- async def fake_parse(
- receive,
- boundary,
- request,
- *,
- file_notifications=None,
- file_queue_maxsize=2048,
- ):
- parser_started.set()
- try:
- await asyncio.Future()
- except asyncio.CancelledError:
- parser_cancelled.set()
- raise
+ async def upload(self, file):
+ chunks = []
+ while (chunk := await file["content"].get()) is not None:
+ chunks.append(chunk)
+ return b"".join(chunks)
- class RejectUpload(HttpMiddleware):
+ setattr(child, "upload", upload.__get__(child, App))
+
+ class MountChild(HttpMiddleware):
async def before_request(self, request):
- await parser_started.wait()
- return {"status_code": 413, "body": "too large"}
+ request._subapp = child
+ request._subapp_path = "/upload"
async def after_request(
self, request, status_code, response_body, extra_headers
):
return None
- self.app._parse_multipart_into_request = fake_parse
- self.app.middlewares.append(RejectUpload())
+ self.app.middlewares.append(MountChild())
+ body = (
+ b"--test-boundary\r\n"
+ b'Content-Disposition: form-data; name="file"; filename="a.txt"\r\n'
+ b"Content-Type: text/plain\r\n\r\n"
+ b"complete file\r\n"
+ b"--test-boundary--\r\n"
+ )
send = AsyncMock()
- with patch("micropie.MULTIPART_INSTALLED", True):
- await asyncio.wait_for(
- self.app(self.multipart_scope(), AsyncMock(), send), timeout=1
- )
+ await asyncio.wait_for(
+ self.app(
+ self.multipart_scope(),
+ AsyncMock(
+ return_value={
+ "type": "http.request",
+ "body": body,
+ "more_body": False,
+ }
+ ),
+ send,
+ ),
+ timeout=1,
+ )
- self.assertTrue(parser_cancelled.is_set())
send.assert_any_call(
{
- "type": "http.response.start",
- "status": 413,
- "headers": [(b"Content-Type", b"text/html; charset=utf-8")],
+ "type": "http.response.body",
+ "body": b"complete file",
+ "more_body": False,
}
)
@@ -1168,8 +1829,13 @@ class TestWebSocket(MicroPieTestCase):
await self.app(scope, receive, send)
- send.assert_any_call(
- {"type": "websocket.accept", "subprotocol": None, "headers": []}
+ accept = next(
+ call[0][0]
+ for call in send.call_args_list
+ if call[0][0]["type"] == "websocket.accept"
+ )
+ self.assertTrue(
+ any(name == b"Set-Cookie" for name, _ in accept["headers"])
)
send.assert_any_call({"type": "websocket.send", "text": "Echo: Hello"})
send.assert_any_call(
@@ -1231,11 +1897,18 @@ class TestWebSocket(MicroPieTestCase):
}
)
- async def test_malformed_websocket_query_is_rejected(self):
- """WebSocket requests should reject malformed query encoding."""
+ async def test_malformed_websocket_query_is_tolerated(self):
+ """WebSocket query parsing should use the same permissive behavior."""
+
+ async def ws_echo(self, ws, name):
+ await ws.accept()
+ await ws.send_text(name)
+ await ws.close()
+
+ setattr(self.app, "ws_echo", ws_echo.__get__(self.app, App))
scope = self.create_mock_scope(
path="/echo",
- query_string=b"missing-equals",
+ query_string=b"name=%ZZ",
scope_type="websocket",
)
receive = AsyncMock(return_value={"type": "websocket.connect"})
@@ -1243,20 +1916,14 @@ class TestWebSocket(MicroPieTestCase):
await self.app(scope, receive, send)
- send.assert_any_call(
- {
- "type": "websocket.close",
- "code": 1008,
- "reason": "Malformed query string",
- }
- )
+ send.assert_any_call({"type": "websocket.send", "text": "%ZZ"})
async def test_websocket_cookie_respects_secure_configuration(self):
"""WebSocket handshake cookies should use the App cookie policy."""
self.app = App(session_cookie_secure=False)
async def ws_cookie(self, ws):
- await ws.accept(session_id="abc")
+ await ws.accept(session_id=VALID_SESSION_ID)
await ws.close()
setattr(self.app, "ws_cookie", ws_cookie.__get__(self.app, App))
@@ -1277,6 +1944,38 @@ class TestWebSocket(MicroPieTestCase):
self.assertIn(b"HttpOnly", cookie)
self.assertNotIn(b"Secure", cookie)
+ async def test_websocket_accept_uses_custom_session_id_validator(self):
+ """Explicit non-UUID IDs are allowed by an application's policy."""
+ self.app = App(
+ session_id_factory=lambda: "signed.fresh",
+ session_id_validator=lambda value: (
+ value if value.startswith("signed.") else None
+ ),
+ )
+
+ async def ws_cookie(self, ws):
+ await ws.accept(session_id="signed.explicit")
+ await ws.close()
+
+ setattr(self.app, "ws_cookie", ws_cookie.__get__(self.app, App))
+ send = AsyncMock()
+
+ await self.app(
+ self.create_mock_scope(path="/cookie", scope_type="websocket"),
+ AsyncMock(return_value={"type": "websocket.connect"}),
+ send,
+ )
+
+ accept = next(
+ call[0][0]
+ for call in send.call_args_list
+ if call[0][0]["type"] == "websocket.accept"
+ )
+ cookie = next(
+ value for name, value in accept["headers"] if name == b"Set-Cookie"
+ )
+ self.assertIn(b"session_id=signed.explicit", cookie)
+
async def test_websocket_session_uses_app_timeout(self):
"""WebSocket session saves should receive the configured timeout."""
backend = CountingSessionBackend()
@@ -1295,6 +1994,57 @@ class TestWebSocket(MicroPieTestCase):
await self.app(scope, receive, send)
self.assertEqual(backend.save_calls[-1][2], 321)
+ accept = next(
+ call[0][0]
+ for call in send.call_args_list
+ if call[0][0]["type"] == "websocket.accept"
+ )
+ cookie = next(
+ value for name, value in accept["headers"] if name == b"Set-Cookie"
+ ).decode("ascii")
+ cookie_session_id = cookie.split(";", 1)[0].split("=", 1)[1]
+ self.assertEqual(cookie_session_id, backend.save_calls[-1][0])
+ self.assertEqual(backend.sessions[cookie_session_id], {"user": "alice"})
+
+ async def test_websocket_rotates_unknown_client_session_id(self):
+ """A planted UUID should be replaced during the WebSocket handshake."""
+ backend = CountingSessionBackend()
+ self.app = App(session_backend=backend)
+
+ async def ws_login(self, ws):
+ await ws.accept()
+ self.request.session["user_id"] = 42
+ await ws.close()
+
+ setattr(self.app, "ws_login", ws_login.__get__(self.app, App))
+ scope = self.create_mock_scope(
+ path="/login",
+ scope_type="websocket",
+ headers=[
+ (
+ b"cookie",
+ f"session_id={ATTACKER_SESSION_ID}".encode("ascii"),
+ )
+ ],
+ )
+ receive = AsyncMock(return_value={"type": "websocket.connect"})
+ send = AsyncMock()
+
+ await self.app(scope, receive, send)
+
+ accept = next(
+ call[0][0]
+ for call in send.call_args_list
+ if call[0][0]["type"] == "websocket.accept"
+ )
+ cookie = next(
+ value for name, value in accept["headers"] if name == b"Set-Cookie"
+ ).decode("ascii")
+ new_session_id = cookie.split(";", 1)[0].split("=", 1)[1]
+ self.assertNotEqual(new_session_id, ATTACKER_SESSION_ID)
+ self.assertEqual(uuid.UUID(new_session_id).version, 4)
+ self.assertNotIn(ATTACKER_SESSION_ID, backend.sessions)
+ self.assertEqual(backend.sessions[new_session_id], {"user_id": 42})
async def test_websocket_handler_exception_is_logged(self):
"""WebSocket handler failures should include traceback logging."""
@@ -1315,7 +2065,7 @@ class TestWebSocket(MicroPieTestCase):
{
"type": "websocket.close",
"code": 1011,
- "reason": "Handler error: websocket boom",
+ "reason": "Internal server error",
}
)
@@ -1370,6 +2120,92 @@ class TestMiddleware(MicroPieTestCase):
}
)
+ async def test_partial_before_request_response_uses_defaults(self):
+ """A partial short-circuit mapping should not raise KeyError."""
+
+ class NoContent(HttpMiddleware):
+ async def before_request(self, request):
+ return {"status_code": 204}
+
+ async def after_request(
+ self, request, status_code, response_body, extra_headers
+ ):
+ return None
+
+ self.app.middlewares.append(NoContent())
+ send = AsyncMock()
+
+ await self.app(
+ self.create_mock_scope(),
+ AsyncMock(
+ return_value={
+ "type": "http.request",
+ "body": b"",
+ "more_body": False,
+ }
+ ),
+ send,
+ )
+
+ send.assert_any_call(
+ {
+ "type": "http.response.start",
+ "status": 204,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")],
+ }
+ )
+ send.assert_any_call(
+ {"type": "http.response.body", "body": b"", "more_body": False}
+ )
+
+ async def test_json_subapp_handoff_preserves_json_namespace(self):
+ """Child apps receive JSON through Request.json(), not body_params."""
+ child = App()
+
+ async def submit(self):
+ return {
+ "value": self.request.json("value"),
+ "body_params": self.request.body_params,
+ }
+
+ setattr(child, "submit", submit.__get__(child, App))
+
+ class MountChild(HttpMiddleware):
+ async def before_request(self, request):
+ request._subapp = child
+ request._subapp_path = "/submit"
+
+ async def after_request(
+ self, request, status_code, response_body, extra_headers
+ ):
+ return None
+
+ self.app.middlewares.append(MountChild())
+ send = AsyncMock()
+
+ await self.app(
+ self.create_mock_scope(
+ path="/mounted/submit",
+ method="POST",
+ headers=[(b"content-type", b"application/json")],
+ ),
+ AsyncMock(
+ return_value={
+ "type": "http.request",
+ "body": b'{"value": 7}',
+ "more_body": False,
+ }
+ ),
+ send,
+ )
+
+ body = next(
+ call[0][0]["body"]
+ for call in send.call_args_list
+ if call[0][0]["type"] == "http.response.body"
+ )
+ self.assertEqual(std_json.loads(body), {"value": 7, "body_params": {}})
+
class TestResponseHandling(MicroPieTestCase):
"""Tests for response handling and edge cases."""
@@ -1419,6 +2255,49 @@ class TestResponseHandling(MicroPieTestCase):
}
)
+ async def test_json_response_type_is_stable_for_middleware(self):
+ """Middleware sees text regardless of the selected JSON implementation."""
+ seen_types = []
+
+ class ObserveBody(HttpMiddleware):
+ async def before_request(self, request):
+ return None
+
+ async def after_request(
+ self, request, status_code, response_body, extra_headers
+ ):
+ seen_types.append(type(response_body))
+
+ self.app.middlewares.append(ObserveBody())
+
+ async def index(self):
+ return {"ok": True}
+
+ setattr(self.app, "index", index.__get__(self.app, App))
+ send = AsyncMock()
+
+ with patch("micropie.json.dumps", return_value=b'{"ok":true}'):
+ await self.app(
+ self.create_mock_scope(),
+ AsyncMock(
+ return_value={
+ "type": "http.request",
+ "body": b"",
+ "more_body": False,
+ }
+ ),
+ send,
+ )
+
+ self.assertEqual(seen_types, [str])
+ send.assert_any_call(
+ {
+ "type": "http.response.body",
+ "body": b'{"ok":true}',
+ "more_body": False,
+ }
+ )
+
async def test_json_arguments_keep_native_types_and_form_namespace(self):
"""JSON binding should not stringify or mirror values into form data."""
captured = {}
@@ -1568,6 +2447,91 @@ class TestResponseHandling(MicroPieTestCase):
{"type": "http.response.body", "body": b"Test", "more_body": False}
)
+ async def test_non_latin1_header_is_rejected_before_response_start(self):
+ """An unencodable header should be skipped instead of crashing ASGI."""
+
+ async def index(self):
+ return 200, "OK", [("X-Snowman", "☃")]
+
+ setattr(self.app, "index", index.__get__(self.app, App))
+ send = AsyncMock()
+
+ with self.assertLogs("micropie", level="WARNING"):
+ await self.app(
+ self.create_mock_scope(),
+ AsyncMock(
+ return_value={
+ "type": "http.request",
+ "body": b"",
+ "more_body": False,
+ }
+ ),
+ send,
+ )
+
+ send.assert_any_call(
+ {
+ "type": "http.response.start",
+ "status": 200,
+ "headers": [(b"Content-Type", b"text/html; charset=utf-8")],
+ }
+ )
+
+ async def test_sync_generator_advances_off_event_loop(self):
+ """Blocking synchronous generator steps run in a worker thread."""
+ event_loop_thread = threading.get_ident()
+ generator_threads = []
+
+ def chunks():
+ generator_threads.append(threading.get_ident())
+ yield "chunk"
+
+ async def index(self):
+ return chunks()
+
+ setattr(self.app, "index", index.__get__(self.app, App))
+ send = AsyncMock()
+
+ await self.app(
+ self.create_mock_scope(),
+ AsyncMock(
+ return_value={
+ "type": "http.request",
+ "body": b"",
+ "more_body": False,
+ }
+ ),
+ send,
+ )
+
+ self.assertTrue(generator_threads)
+ self.assertNotEqual(generator_threads[0], event_loop_thread)
+
+ async def test_stream_completion_cancels_pending_receive(self):
+ """Completing a streamed response must not orphan receive()."""
+ receive_cancelled = asyncio.Event()
+
+ async def stream():
+ yield b"done"
+
+ async def index(self):
+ return stream()
+
+ async def receive():
+ try:
+ await asyncio.Future()
+ except asyncio.CancelledError:
+ receive_cancelled.set()
+ raise
+
+ setattr(self.app, "index", index.__get__(self.app, App))
+
+ await asyncio.wait_for(
+ self.app(self.create_mock_scope(), receive, AsyncMock()), timeout=1
+ )
+
+ self.assertTrue(receive_cancelled.is_set())
+
async def test_redirect(self):
"""Test redirect response generation."""
location = "/new-page"
@@ -1579,6 +2543,10 @@ class TestResponseHandling(MicroPieTestCase):
self.assertIn(
("X-Custom", "Value"), headers, "Extra headers should be included"
)
+ _, _, unicode_headers = self.app._redirect("/search?q=snow man ☃")
+ self.assertIn(
+ ("Location", "/search?q=snow%20man%20%E2%98%83"), unicode_headers
+ )
class TestOptionalDependencies(MicroPieTestCase):