fix: serve the uWSGI protocol on the ASGI worker

docs/content/2026-news.md

@@ -5,6 +5,13 @@
 
 ### Bug Fixes
 
+- **ASGI worker returned 502 for the uWSGI protocol**: with
+  `--worker-class asgi --protocol uwsgi` behind nginx `uwsgi_pass`, every
+  request failed with `Invalid uWSGI header: incomplete header`. The worker
+  set up the HTTP/1 parser instead of a uWSGI reader, so inbound bytes never
+  reached the uWSGI handler. The connection now starts a reader for the uWSGI
+  protocol, builds the request body for the app, and sets `raw_path`.
+
 - **Chunked framing lines were read without a bound**: a chunk-size line or
   trailer section that never terminates was reread in full on every socket read
   with no size limit, so one request could keep a worker busy without ever

gunicorn/asgi/protocol.py

@@ -455,10 +455,23 @@ def connection_made(self, transport):
         self._is_ssl = ssl_object is not None
         self.writer = transport
 
-        # Setup flow control for HTTP/1.x
+        # Setup flow control
         self._flow_control = FlowControl(transport)
         transport.set_write_buffer_limits(high=HIGH_WATER_LIMIT)
-        self._start_http1()
+        if getattr(self.cfg, 'protocol', 'http') == 'uwsgi':
+            self._start_uwsgi()
+        else:
+            self._start_http1()
+
+    def _start_uwsgi(self):
+        """Commit to the uWSGI protocol on this connection.
+
+        uWSGI framing is pull-parsed from a StreamReader that data_received()
+        feeds, so create the reader here instead of the HTTP/1 callback parser;
+        _handle_connection() dispatches to the uWSGI handler.
+        """
+        self.reader = asyncio.StreamReader()
+        self._task = self.worker.loop.create_task(self._handle_connection())
 
     def _start_http1(self, buffered=b""):
         """Commit to HTTP/1.x and replay anything already read.
@@ -1080,10 +1093,23 @@ async def _handle_connection_uwsgi(self, peername, sockname):
                 await self._handle_websocket(request, sockname, peername)
                 break
 
+            # Build the ASGI body receiver and feed it the uWSGI body. The
+            # HTTP/1 path fills this from parser callbacks; the uWSGI body is
+            # pull-read here (nginx buffers the request upstream).
+            self._body_receiver = BodyReceiver(request, self)
+            if request.content_length:
+                while True:
+                    chunk = await request.read_body()
+                    if not chunk:
+                        break
+                    self._body_receiver.feed(chunk)
+            self._body_receiver.set_complete()
+
             # Handle HTTP request
             keepalive = await self._handle_http_request(
                 request, sockname, peername
             )
+            self._body_receiver = None
 
             # Increment worker request count
             self.worker.nr += 1

gunicorn/asgi/unreader.py

@@ -105,10 +105,14 @@ async def read(self, size=None):
         return data
 
     async def _read_chunk(self):
-        """Read a chunk of data from the underlying stream."""
+        """Read a chunk of data from the underlying stream.
+
+        A lost or reset connection reads as EOF; other errors (including a
+        misconfigured reader) propagate rather than masquerade as EOF.
+        """
         try:
             return await self.reader.read(self.max_chunk)
-        except Exception:
+        except (ConnectionError, OSError):
             return b""
 
     def unread(self, data):

gunicorn/asgi/uwsgi.py

@@ -39,6 +39,7 @@ def __init__(self, cfg, unreader, peer_addr, req_number=1):
         self.method = None
         self.uri = None
         self.path = None
+        self.raw_path = None
         self.query = None
         self.fragment = ""
         self.version = (1, 1)

gunicorn/uwsgi/message.py

@@ -39,6 +39,7 @@ def __init__(self, cfg, unreader, peer_addr, req_number=1):
         self.method = None
         self.uri = None
         self.path = None
+        self.raw_path = None
         self.query = None
         self.fragment = ""
         self.version = (1, 1)  # uWSGI is HTTP/1.1 compatible
@@ -187,6 +188,12 @@ def _extract_request_info(self):
         self.path = self.uwsgi_vars.get('PATH_INFO', '/')
         self.query = self.uwsgi_vars.get('QUERY_STRING', '')
 
+        # raw_path is the undecoded path bytes; REQUEST_URI carries it when the
+        # proxy forwards it, otherwise fall back to the decoded PATH_INFO.
+        request_uri = self.uwsgi_vars.get('REQUEST_URI', '')
+        raw = request_uri.split('?', 1)[0] if request_uri else self.path
+        self.raw_path = raw.encode('latin-1', 'replace')
+
         # Build URI
         if self.query:
             self.uri = "%s?%s" % (self.path, self.query)

tests/support/uwsgi_asgi_app.py

@@ -0,0 +1,26 @@
+"""Minimal ASGI app for the uWSGI-protocol regression test."""
+
+
+async def app(scope, receive, send):
+    if scope["type"] == "lifespan":
+        while True:
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            elif message["type"] == "lifespan.shutdown":
+                await send({"type": "lifespan.shutdown.complete"})
+                return
+        return
+    assert scope["type"] == "http"
+    body = b""
+    while True:
+        message = await receive()
+        body += message.get("body", b"")
+        if not message.get("more_body"):
+            break
+    payload = b"method=%s path=%s query=%s body=%s" % (
+        scope["method"].encode(), scope["path"].encode(),
+        scope["query_string"], body)
+    await send({"type": "http.response.start", "status": 200,
+                "headers": [(b"content-type", b"text/plain")]})
+    await send({"type": "http.response.body", "body": payload})

tests/test_asgi_uwsgi_protocol.py

@@ -0,0 +1,131 @@
+#
+# This file is part of gunicorn released under the MIT license.
+# See the NOTICE for more information.
+
+"""The ASGI worker over the uWSGI protocol, against a live gunicorn.
+
+Spawns gunicorn with ``--worker-class asgi --protocol uwsgi`` on a loopback
+port and speaks the uWSGI binary protocol directly (as nginx ``uwsgi_pass``
+would), asserting the request is served and the body round-trips. Regression
+for the wiring bug where the ASGI worker never fed its uWSGI reader and 502'd
+every request.
+"""
+
+import os
+import socket
+import struct
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+APPS = Path(__file__).parent / "support"
+
+
+def _free_port():
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+def _packet(variables, body=b""):
+    block = b""
+    for key, value in variables:
+        key, value = key.encode(), value.encode()
+        block += struct.pack("<H", len(key)) + key
+        block += struct.pack("<H", len(value)) + value
+    return struct.pack("<BHB", 0, len(block), 0) + block + body
+
+
+class Server:
+    def __init__(self, tmp_path):
+        self.port = _free_port()
+        self.log = tmp_path / f"gunicorn-uwsgi-{self.port}.log"
+        self.proc = subprocess.Popen(
+            [sys.executable, "-m", "gunicorn", "uwsgi_asgi_app:app",
+             "--bind", f"127.0.0.1:{self.port}", "--workers", "1",
+             "--worker-class", "asgi", "--protocol", "uwsgi",
+             "--uwsgi-allow-from", "*", "--keep-alive", "2",
+             "--graceful-timeout", "1", "--log-level", "info"],
+            cwd=str(APPS), stdout=self.log.open("w"), stderr=subprocess.STDOUT)
+        deadline = time.monotonic() + 15
+        while time.monotonic() < deadline:
+            try:
+                with socket.create_connection(("127.0.0.1", self.port), 0.3):
+                    return
+            except OSError:
+                if self.proc.poll() is not None:
+                    break
+                time.sleep(0.05)
+        self.stop()
+        raise RuntimeError(f"gunicorn did not start:\n{self.log.read_text()}")
+
+    def request(self, variables, body=b""):
+        sock = socket.create_connection(("127.0.0.1", self.port), 3)
+        sock.settimeout(3)
+        sock.sendall(_packet(variables, body))
+        data = b""
+        try:
+            while True:
+                chunk = sock.recv(4096)
+                if not chunk:
+                    break
+                data += chunk
+        except socket.timeout:
+            pass
+        sock.close()
+        return data
+
+    def tracebacks(self):
+        return self.log.read_text().count("Traceback")
+
+    def stop(self):
+        if self.proc.poll() is None:
+            self.proc.terminate()
+            try:
+                self.proc.wait(5)
+            except subprocess.TimeoutExpired:
+                self.proc.kill()
+                self.proc.wait()
+
+
+@pytest.fixture
+def server(tmp_path):
+    if not os.environ.get("GUNICORN_TESTING", "1"):
+        pytest.skip("disabled")
+    srv = Server(tmp_path)
+    yield srv
+    srv.stop()
+
+
+def _base_vars(method, path, query="", length=0):
+    return [
+        ("REQUEST_METHOD", method), ("PATH_INFO", path),
+        ("QUERY_STRING", query), ("REQUEST_URI", path + (f"?{query}" if query else "")),
+        ("SERVER_PROTOCOL", "HTTP/1.1"), ("SERVER_NAME", "localhost"),
+        ("SERVER_PORT", "80"), ("CONTENT_LENGTH", str(length)),
+    ]
+
+
+def test_get_is_served(server):
+    resp = server.request(_base_vars("GET", "/hello", "a=1"))
+    assert resp.startswith(b"HTTP/1.1 200"), resp
+    assert b"method=GET path=/hello query=a=1 body=" in resp, resp
+    assert server.tracebacks() == 0
+
+
+def test_post_body_round_trips(server):
+    body = b"payload-1234567890"
+    resp = server.request(_base_vars("POST", "/echo", length=len(body)), body)
+    assert resp.startswith(b"HTTP/1.1 200"), resp
+    assert b"method=POST path=/echo query= body=" + body in resp, resp
+    assert server.tracebacks() == 0
+
+
+def test_keepalive_serves_multiple_on_one_worker(server):
+    for i in range(5):
+        resp = server.request(_base_vars("GET", f"/n{i}"))
+        assert resp.startswith(b"HTTP/1.1 200"), resp
+    assert server.tracebacks() == 0