Add uWSGI protocol support to ASGI worker (#3467)

Add uWSGI protocol support to ASGI worker

- Implements AsyncUWSGIRequest class extending sync UWSGIRequest to reuse parsing logic with async I/O
- ASGI protocol handler selects between HTTP and uWSGI based on --protocol config option
- Allows gunicorn's ASGI worker to receive requests from nginx using uwsgi_pass directive
- Includes unit tests and Docker integration tests

docs/content/asgi.md

@@ -33,6 +33,7 @@ The ASGI worker provides:
 - **Lifespan protocol** for startup/shutdown hooks
 - **Optional uvloop** for improved performance
 - **SSL/TLS** support
+- **uWSGI protocol** for nginx `uwsgi_pass` integration
 
 ## Configuration
 
@@ -151,7 +152,7 @@ app = Starlette(routes=[
 
 ## Production Deployment
 
-### With Nginx
+### With Nginx (HTTP Proxy)
 
 ```nginx
 upstream gunicorn {
@@ -181,6 +182,36 @@ server {
 }
 ```
 
+### With Nginx (uWSGI Protocol)
+
+For better performance, you can use nginx's native uWSGI protocol support:
+
+```bash
+gunicorn myapp:app --worker-class asgi --protocol uwsgi --bind 127.0.0.1:8000
+```
+
+```nginx
+upstream gunicorn {
+    server 127.0.0.1:8000;
+}
+
+server {
+    listen 80;
+    server_name example.com;
+
+    location / {
+        uwsgi_pass gunicorn;
+        include uwsgi_params;
+    }
+}
+```
+
+!!! note
+    WebSocket connections are not supported when using the uWSGI protocol.
+    Use HTTP proxy for WebSocket endpoints.
+
+See [uWSGI Protocol](uwsgi.md) for more details on uWSGI protocol configuration.
+
 ### Recommended Settings
 
 For production ASGI deployments:

docs/content/uwsgi.md

@@ -4,6 +4,8 @@ Gunicorn supports the uWSGI binary protocol, allowing it to receive requests fro
 nginx using the `uwsgi_pass` directive. This provides efficient communication
 between nginx and Gunicorn without HTTP overhead.
 
+Both **WSGI** and **ASGI** workers support the uWSGI protocol.
+
 !!! note
     This is the **uWSGI binary protocol**, not the uWSGI server. Gunicorn
     implements the protocol to receive requests from nginx, similar to how
@@ -14,7 +16,11 @@ between nginx and Gunicorn without HTTP overhead.
 Enable uWSGI protocol support:
 
 ```bash
+# WSGI application
 gunicorn myapp:app --protocol uwsgi --bind 127.0.0.1:8000
+
+# ASGI application
+gunicorn myapp:app --worker-class asgi --protocol uwsgi --bind 127.0.0.1:8000
 ```
 
 Configure nginx to forward requests:

gunicorn/asgi/protocol.py

@@ -14,7 +14,9 @@ from datetime import datetime
 
 from gunicorn.asgi.unreader import AsyncUnreader
 from gunicorn.asgi.message import AsyncRequest
+from gunicorn.asgi.uwsgi import AsyncUWSGIRequest
 from gunicorn.http.errors import NoMoreData
+from gunicorn.uwsgi.errors import UWSGIParseException
 
 
 class ASGIResponseInfo:
@@ -92,19 +94,31 @@ class ASGIProtocol(asyncio.Protocol):
                 self.req_count += 1
 
                 try:
-                    # Parse HTTP request
-                    request = await AsyncRequest.parse(
-                        self.cfg,
-                        unreader,
-                        peername,
-                        self.req_count
-                    )
+                    # Parse request based on protocol
+                    protocol = getattr(self.cfg, 'protocol', 'http')
+                    if protocol == 'uwsgi':
+                        request = await AsyncUWSGIRequest.parse(
+                            self.cfg,
+                            unreader,
+                            peername,
+                            self.req_count
+                        )
+                    else:
+                        request = await AsyncRequest.parse(
+                            self.cfg,
+                            unreader,
+                            peername,
+                            self.req_count
+                        )
                 except StopIteration:
                     # No more data, close connection
                     break
                 except NoMoreData:
                     # Client disconnected
                     break
+                except UWSGIParseException as e:
+                    self.log.debug("uWSGI parse error: %s", e)
+                    break
 
                 # Check for WebSocket upgrade
                 if self._is_websocket_upgrade(request):

gunicorn/asgi/uwsgi.py

@@ -0,0 +1,172 @@
+#
+# This file is part of gunicorn released under the MIT license.
+# See the NOTICE for more information.
+
+"""Async uWSGI protocol parser for ASGI workers.
+
+Reuses the parsing logic from gunicorn/uwsgi/message.py, only async I/O differs.
+"""
+
+from gunicorn.uwsgi.message import UWSGIRequest
+from gunicorn.uwsgi.errors import (
+    InvalidUWSGIHeader,
+    UnsupportedModifier,
+)
+
+
+class AsyncUWSGIRequest(UWSGIRequest):
+    """Async version of UWSGIRequest.
+
+    Reuses all parsing logic from the sync version, only async I/O differs.
+    The following methods are reused from the parent class:
+    - _parse_vars() - pure parsing, no I/O
+    - _extract_request_info() - pure transformation
+    - _check_allowed_ip() - no I/O
+    - should_close() - simple logic
+    """
+
+    # pylint: disable=super-init-not-called
+    def __init__(self, cfg, unreader, peer_addr, req_number=1):
+        # Don't call super().__init__ - it does sync parsing
+        # Just initialize attributes
+        self.cfg = cfg
+        self.unreader = unreader
+        self.peer_addr = peer_addr
+        self.remote_addr = peer_addr
+        self.req_number = req_number
+
+        # Initialize all attributes (same as sync version)
+        self.method = None
+        self.uri = None
+        self.path = None
+        self.query = None
+        self.fragment = ""
+        self.version = (1, 1)
+        self.headers = []
+        self.trailers = []
+        self.body = None
+        self.scheme = "https" if cfg.is_ssl else "http"
+        self.must_close = False
+        self.uwsgi_vars = {}
+        self.modifier1 = 0
+        self.modifier2 = 0
+        self.proxy_protocol_info = None
+
+        # Body state
+        self.content_length = 0
+        self.chunked = False
+        self._body_remaining = 0
+
+    # Async factory method - intentionally differs from sync parent:
+    # - async instead of sync (invalid-overridden-method)
+    # - different signature for async I/O (arguments-differ)
+    # pylint: disable=arguments-differ,invalid-overridden-method
+    @classmethod
+    async def parse(cls, cfg, unreader, peer_addr, req_number=1):
+        """Parse a uWSGI request asynchronously.
+
+        Args:
+            cfg: gunicorn config object
+            unreader: AsyncUnreader instance
+            peer_addr: client address tuple
+            req_number: request number on this connection (for keepalive)
+
+        Returns:
+            AsyncUWSGIRequest: Parsed request object
+
+        Raises:
+            InvalidUWSGIHeader: If the uWSGI header is malformed
+            UnsupportedModifier: If modifier1 is not 0
+            ForbiddenUWSGIRequest: If source IP is not allowed
+        """
+        req = cls(cfg, unreader, peer_addr, req_number)
+        req._check_allowed_ip()  # Reuse from parent
+        await req._async_parse()
+        return req
+
+    async def _async_parse(self):
+        """Async version of parse() - reads data then uses sync parsing."""
+        # Read 4-byte header
+        header = await self._async_read_exact(4)
+        if len(header) < 4:
+            raise InvalidUWSGIHeader("incomplete header")
+
+        self.modifier1 = header[0]
+        datasize = int.from_bytes(header[1:3], 'little')
+        self.modifier2 = header[3]
+
+        if self.modifier1 != 0:
+            raise UnsupportedModifier(self.modifier1)
+
+        # Read vars block
+        if datasize > 0:
+            vars_data = await self._async_read_exact(datasize)
+            if len(vars_data) < datasize:
+                raise InvalidUWSGIHeader("incomplete vars block")
+            self._parse_vars(vars_data)  # Reuse sync method
+
+        self._extract_request_info()  # Reuse sync method
+        self._set_body_reader()
+
+    async def _async_read_exact(self, size):
+        """Read exactly size bytes asynchronously."""
+        buf = bytearray()
+        while len(buf) < size:
+            chunk = await self.unreader.read(size - len(buf))
+            if not chunk:
+                break
+            buf.extend(chunk)
+        return bytes(buf)
+
+    def _set_body_reader(self):
+        """Set up body state for async reading."""
+        content_length = 0
+        if 'CONTENT_LENGTH' in self.uwsgi_vars:
+            try:
+                content_length = max(int(self.uwsgi_vars['CONTENT_LENGTH']), 0)
+            except ValueError:
+                content_length = 0
+        self.content_length = content_length
+        self._body_remaining = content_length
+
+    async def read_body(self, size=8192):
+        """Read body chunk asynchronously.
+
+        Args:
+            size: Maximum bytes to read
+
+        Returns:
+            bytes: Body data, empty bytes when body is exhausted
+        """
+        if self._body_remaining <= 0:
+            return b""
+        to_read = min(size, self._body_remaining)
+        data = await self.unreader.read(to_read)
+        if data:
+            self._body_remaining -= len(data)
+        return data
+
+    async def drain_body(self):
+        """Drain unread body data.
+
+        Should be called before reusing connection for keepalive.
+        """
+        while self._body_remaining > 0:
+            data = await self.read_body(8192)
+            if not data:
+                break
+
+    def get_header(self, name):
+        """Get header by name (case-insensitive).
+
+        Args:
+            name: Header name to look up
+
+        Returns:
+            Header value if found, None otherwise
+        """
+        name = name.upper()
+        for h, v in self.headers:
+            if h == name:
+                return v
+        return None

tests/docker/test_asgi_uwsgi/Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.11-slim
+
+WORKDIR /build
+
+# Copy gunicorn source
+COPY . /build/
+
+# Install gunicorn from source
+RUN pip install --no-cache-dir -e .
+
+# Copy test app
+WORKDIR /app
+COPY tests/docker/test_asgi_uwsgi/app.py /app/
+
+# Expose uWSGI port
+EXPOSE 8000
+
+CMD ["gunicorn", "--worker-class", "asgi", "--protocol", "uwsgi", "--bind", "0.0.0.0:8000", "app:app"]

tests/docker/test_asgi_uwsgi/app.py

@@ -0,0 +1,45 @@
+"""Simple ASGI test application for uWSGI protocol testing."""
+
+
+async def app(scope, receive, send):
+    """Simple ASGI application that echoes request info."""
+    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
+
+    if scope["type"] != "http":
+        return
+
+    # Read body
+    body = b""
+    while True:
+        message = await receive()
+        body += message.get("body", b"")
+        if not message.get("more_body", False):
+            break
+
+    # Build response
+    method = scope["method"]
+    path = scope["path"]
+    query = scope.get("query_string", b"").decode("utf-8")
+
+    response_body = f"Method: {method}\nPath: {path}\nQuery: {query}\nBody: {body.decode('utf-8')}\n"
+    response_bytes = response_body.encode("utf-8")
+
+    await send({
+        "type": "http.response.start",
+        "status": 200,
+        "headers": [
+            [b"content-type", b"text/plain"],
+            [b"content-length", str(len(response_bytes)).encode()],
+        ],
+    })
+    await send({
+        "type": "http.response.body",
+        "body": response_bytes,
+    })

tests/docker/test_asgi_uwsgi/docker-compose.yml

@@ -0,0 +1,24 @@
+services:
+  gunicorn:
+    build:
+      context: ../../..
+      dockerfile: tests/docker/test_asgi_uwsgi/Dockerfile
+    command: >
+      gunicorn
+      --worker-class asgi
+      --protocol uwsgi
+      --uwsgi-allow-from '*'
+      --bind 0.0.0.0:8000
+      --workers 1
+      --log-level debug
+      app:app
+    working_dir: /app
+
+  nginx:
+    image: nginx:alpine
+    ports:
+      - "8080:80"
+    volumes:
+      - ./nginx.conf:/etc/nginx/conf.d/default.conf:ro
+    depends_on:
+      - gunicorn

tests/docker/test_asgi_uwsgi/nginx.conf

@@ -0,0 +1,14 @@
+server {
+    listen 80;
+    server_name localhost;
+
+    location / {
+        uwsgi_pass gunicorn:8000;
+        include uwsgi_params;
+    }
+
+    location /health {
+        return 200 "OK";
+        add_header Content-Type text/plain;
+    }
+}

tests/docker/test_asgi_uwsgi/test_uwsgi.sh

@@ -0,0 +1,86 @@
+#!/bin/bash
+# Integration test for ASGI uWSGI protocol support
+#
+# This script tests that gunicorn's ASGI worker correctly handles
+# the uWSGI protocol when nginx forwards requests using uwsgi_pass.
+
+set -e
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+cd "$SCRIPT_DIR"
+
+# Use IPv4 explicitly to avoid Docker IPv6 issues
+BASE_URL="http://127.0.0.1:8080"
+
+cleanup() {
+    echo "Cleaning up..."
+    docker compose down -v 2>/dev/null || true
+}
+
+trap cleanup EXIT
+
+echo "=== Building and starting containers ==="
+docker compose up -d --build
+
+echo "=== Waiting for services to be ready ==="
+sleep 5
+
+echo "=== Running tests ==="
+
+# Test 1: Simple GET request
+echo "Test 1: Simple GET request"
+RESPONSE=$(curl -s "$BASE_URL/")
+if echo "$RESPONSE" | grep -q "Method: GET"; then
+    echo "  PASS: GET request works"
+else
+    echo "  FAIL: GET request failed"
+    echo "  Response: $RESPONSE"
+    exit 1
+fi
+
+# Test 2: GET with query string
+echo "Test 2: GET with query string"
+RESPONSE=$(curl -s "$BASE_URL/search?q=test&page=1")
+if echo "$RESPONSE" | grep -q "Query: q=test&page=1"; then
+    echo "  PASS: Query string works"
+else
+    echo "  FAIL: Query string failed"
+    echo "  Response: $RESPONSE"
+    exit 1
+fi
+
+# Test 3: POST with body
+echo "Test 3: POST with body"
+RESPONSE=$(curl -s -X POST -d "hello=world" "$BASE_URL/submit")
+if echo "$RESPONSE" | grep -q "Method: POST" && echo "$RESPONSE" | grep -q "Body: hello=world"; then
+    echo "  PASS: POST with body works"
+else
+    echo "  FAIL: POST with body failed"
+    echo "  Response: $RESPONSE"
+    exit 1
+fi
+
+# Test 4: Path handling
+echo "Test 4: Path handling"
+RESPONSE=$(curl -s "$BASE_URL/api/v1/users")
+if echo "$RESPONSE" | grep -q "Path: /api/v1/users"; then
+    echo "  PASS: Path handling works"
+else
+    echo "  FAIL: Path handling failed"
+    echo "  Response: $RESPONSE"
+    exit 1
+fi
+
+# Test 5: Multiple requests (keepalive)
+echo "Test 5: Multiple requests (keepalive)"
+for i in 1 2 3; do
+    RESPONSE=$(curl -s "$BASE_URL/request/$i")
+    if ! echo "$RESPONSE" | grep -q "Path: /request/$i"; then
+        echo "  FAIL: Request $i failed"
+        exit 1
+    fi
+done
+echo "  PASS: Multiple requests work"
+
+echo ""
+echo "=== All tests passed! ==="

tests/test_asgi_uwsgi.py

@@ -0,0 +1,472 @@
+#
+# This file is part of gunicorn released under the MIT license.
+# See the NOTICE for more information.
+
+"""
+Tests for ASGI uWSGI protocol parser.
+"""
+
+import pytest
+
+from gunicorn.asgi.unreader import AsyncUnreader
+from gunicorn.asgi.uwsgi import AsyncUWSGIRequest
+from gunicorn.uwsgi.errors import (
+    InvalidUWSGIHeader,
+    UnsupportedModifier,
+    ForbiddenUWSGIRequest,
+)
+
+
+class MockStreamReader:
+    """Mock asyncio.StreamReader for testing."""
+
+    def __init__(self, data):
+        self.data = data
+        self.pos = 0
+
+    async def read(self, size=-1):
+        if self.pos >= len(self.data):
+            return b""
+        if size < 0:
+            result = self.data[self.pos:]
+            self.pos = len(self.data)
+        else:
+            result = self.data[self.pos:self.pos + size]
+            self.pos += size
+        return result
+
+
+class MockConfig:
+    """Mock gunicorn config for testing."""
+
+    def __init__(self):
+        self.is_ssl = False
+        self.uwsgi_allow_ips = ['*']  # Allow all for most tests
+
+
+def build_uwsgi_packet(vars_dict, modifier1=0, modifier2=0):
+    """Build a uWSGI packet from a dictionary of variables.
+
+    Args:
+        vars_dict: Dictionary of uWSGI variables
+        modifier1: uWSGI modifier1 (default 0 for WSGI)
+        modifier2: uWSGI modifier2 (default 0)
+
+    Returns:
+        bytes: Complete uWSGI packet
+    """
+    vars_data = b""
+    for key, value in vars_dict.items():
+        key_bytes = key.encode('latin-1')
+        value_bytes = value.encode('latin-1')
+        vars_data += len(key_bytes).to_bytes(2, 'little')
+        vars_data += key_bytes
+        vars_data += len(value_bytes).to_bytes(2, 'little')
+        vars_data += value_bytes
+
+    # Build header: modifier1 (1 byte) + datasize (2 bytes LE) + modifier2 (1 byte)
+    header = bytes([modifier1])
+    header += len(vars_data).to_bytes(2, 'little')
+    header += bytes([modifier2])
+
+    return header + vars_data
+
+
+# Basic parsing tests
+
+@pytest.mark.asyncio
+async def test_parse_simple_get():
+    """Test parsing a simple GET request."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/test',
+        'QUERY_STRING': '',
+        'HTTP_HOST': 'localhost',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    assert request.method == "GET"
+    assert request.path == "/test"
+    assert request.query == ""
+    assert request.uri == "/test"
+    assert request.version == (1, 1)
+
+
+@pytest.mark.asyncio
+async def test_parse_get_with_query():
+    """Test parsing GET request with query string."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/search',
+        'QUERY_STRING': 'q=test&page=1',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    assert request.method == "GET"
+    assert request.path == "/search"
+    assert request.query == "q=test&page=1"
+    assert request.uri == "/search?q=test&page=1"
+
+
+@pytest.mark.asyncio
+async def test_parse_post_with_content_length():
+    """Test parsing POST request with content length."""
+    body = b"hello=world"
+    vars_dict = {
+        'REQUEST_METHOD': 'POST',
+        'PATH_INFO': '/submit',
+        'CONTENT_LENGTH': str(len(body)),
+        'CONTENT_TYPE': 'application/x-www-form-urlencoded',
+    }
+    packet = build_uwsgi_packet(vars_dict) + body
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    assert request.method == "POST"
+    assert request.path == "/submit"
+    assert request.content_length == len(body)
+
+    # Read body
+    read_body = await request.read_body(100)
+    assert read_body == body
+
+
+@pytest.mark.asyncio
+async def test_parse_headers():
+    """Test that HTTP headers are correctly extracted."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/',
+        'HTTP_HOST': 'example.com',
+        'HTTP_ACCEPT': 'text/html',
+        'HTTP_X_CUSTOM_HEADER': 'custom-value',
+        'CONTENT_TYPE': 'text/plain',
+        'CONTENT_LENGTH': '0',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    # Check headers were extracted correctly
+    assert request.get_header('HOST') == 'example.com'
+    assert request.get_header('ACCEPT') == 'text/html'
+    assert request.get_header('X-CUSTOM-HEADER') == 'custom-value'
+    assert request.get_header('CONTENT-TYPE') == 'text/plain'
+    assert request.get_header('CONTENT-LENGTH') == '0'
+
+
+@pytest.mark.asyncio
+async def test_parse_https_scheme():
+    """Test HTTPS scheme detection."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/',
+        'HTTPS': 'on',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    assert request.scheme == 'https'
+
+
+@pytest.mark.asyncio
+async def test_parse_wsgi_url_scheme():
+    """Test wsgi.url_scheme variable."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/',
+        'wsgi.url_scheme': 'https',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    assert request.scheme == 'https'
+
+
+# Body reading tests
+
+@pytest.mark.asyncio
+async def test_read_body_chunks():
+    """Test reading body in chunks."""
+    body = b"a" * 100
+    vars_dict = {
+        'REQUEST_METHOD': 'POST',
+        'PATH_INFO': '/',
+        'CONTENT_LENGTH': str(len(body)),
+    }
+    packet = build_uwsgi_packet(vars_dict) + body
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    # Read in chunks
+    chunks = []
+    while True:
+        chunk = await request.read_body(30)
+        if not chunk:
+            break
+        chunks.append(chunk)
+
+    assert b"".join(chunks) == body
+
+
+@pytest.mark.asyncio
+async def test_drain_body():
+    """Test draining unread body."""
+    body = b"x" * 50
+    vars_dict = {
+        'REQUEST_METHOD': 'POST',
+        'PATH_INFO': '/',
+        'CONTENT_LENGTH': str(len(body)),
+    }
+    packet = build_uwsgi_packet(vars_dict) + body
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    # Drain without reading
+    await request.drain_body()
+
+    # Further reads should return empty
+    chunk = await request.read_body()
+    assert chunk == b""
+
+
+@pytest.mark.asyncio
+async def test_no_body():
+    """Test request with no body."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    assert request.content_length == 0
+    chunk = await request.read_body()
+    assert chunk == b""
+
+
+# Connection handling tests
+
+@pytest.mark.asyncio
+async def test_should_close_default():
+    """Test default keepalive behavior."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    # Default should be keep-alive (HTTP/1.1 behavior)
+    assert request.should_close() is False
+
+
+@pytest.mark.asyncio
+async def test_should_close_connection_close():
+    """Test connection close header."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/',
+        'HTTP_CONNECTION': 'close',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    assert request.should_close() is True
+
+
+@pytest.mark.asyncio
+async def test_should_close_keepalive():
+    """Test connection keep-alive header."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/',
+        'HTTP_CONNECTION': 'keep-alive',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    assert request.should_close() is False
+
+
+# Error handling tests
+
+@pytest.mark.asyncio
+async def test_incomplete_header():
+    """Test incomplete header raises error."""
+    # Only 2 bytes instead of 4
+    data = b"\x00\x00"
+    reader = MockStreamReader(data)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    with pytest.raises(InvalidUWSGIHeader):
+        await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+
+@pytest.mark.asyncio
+async def test_unsupported_modifier():
+    """Test unsupported modifier1 raises error."""
+    # modifier1 = 1 (not WSGI)
+    header = bytes([1, 0, 0, 0])  # modifier1=1, datasize=0, modifier2=0
+    reader = MockStreamReader(header)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    with pytest.raises(UnsupportedModifier):
+        await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+
+@pytest.mark.asyncio
+async def test_incomplete_vars_block():
+    """Test incomplete vars block raises error."""
+    # Header says 100 bytes of vars, but only 10 provided
+    header = bytes([0])  # modifier1=0
+    header += (100).to_bytes(2, 'little')  # datasize=100
+    header += bytes([0])  # modifier2=0
+    header += b"x" * 10  # Only 10 bytes
+
+    reader = MockStreamReader(header)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    with pytest.raises(InvalidUWSGIHeader):
+        await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+
+@pytest.mark.asyncio
+async def test_forbidden_ip():
+    """Test forbidden IP raises error."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+    cfg.uwsgi_allow_ips = ['10.0.0.1']  # Only allow 10.0.0.1
+
+    with pytest.raises(ForbiddenUWSGIRequest):
+        await AsyncUWSGIRequest.parse(cfg, unreader, ("192.168.1.1", 8000))
+
+
+@pytest.mark.asyncio
+async def test_allowed_ip():
+    """Test allowed IP succeeds."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+    cfg.uwsgi_allow_ips = ['192.168.1.1']
+
+    # Should not raise
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("192.168.1.1", 8000))
+    assert request.method == "GET"
+
+
+@pytest.mark.asyncio
+async def test_unix_socket_allowed():
+    """Test UNIX socket connections are always allowed."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+    cfg.uwsgi_allow_ips = ['10.0.0.1']  # Restrictive IP list
+
+    # UNIX socket peer_addr is not a tuple
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, "/tmp/gunicorn.sock")
+    assert request.method == "GET"
+
+
+# Empty vars block test
+
+@pytest.mark.asyncio
+async def test_empty_vars_block():
+    """Test request with empty vars block uses defaults."""
+    # Header with datasize=0
+    header = bytes([0, 0, 0, 0])
+    reader = MockStreamReader(header)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    # Should use defaults
+    assert request.method == "GET"
+    assert request.path == "/"
+    assert request.query == ""
+
+
+# SSL config test
+
+@pytest.mark.asyncio
+async def test_ssl_config_scheme():
+    """Test SSL config sets https scheme."""
+    vars_dict = {
+        'REQUEST_METHOD': 'GET',
+        'PATH_INFO': '/',
+    }
+    packet = build_uwsgi_packet(vars_dict)
+    reader = MockStreamReader(packet)
+    unreader = AsyncUnreader(reader)
+    cfg = MockConfig()
+    cfg.is_ssl = True
+
+    request = await AsyncUWSGIRequest.parse(cfg, unreader, ("127.0.0.1", 8000))
+
+    assert request.scheme == 'https'