fix: announce Connection: close from the ASGI worker

gunicorn/asgi/protocol.py

@@ -1291,7 +1291,10 @@ async def send(message):
                     uses_uwsgi and not has_content_length and not omits_body
                 )
 
-                self._send_response_start(response_status, response_headers, request)
+                self._send_response_start(
+                    response_status, response_headers, request,
+                    close=self._will_close(request, response_requires_close),
+                )
 
             elif msg_type == "http.response.body":
                 if not response_started:
@@ -1394,13 +1397,23 @@ async def send(message):
                 self.log.exception("Exception in post_request hook")
 
         # Determine keepalive
+        return not self._will_close(request, response_requires_close)
+
+    def _will_close(self, request, response_requires_close):
+        """Return True when this connection is closed after the response.
+
+        Single source of truth: ``_send_response_start`` consults it to emit
+        ``Connection: close`` and ``_handle_http_request`` returns its inverse
+        as the keepalive decision, so the announced framing cannot drift from
+        the framing actually used.
+        """
         if response_requires_close:
-            return False
+            return True
 
         if request.should_close():
-            return False
+            return True
 
-        return self.worker.alive and self.cfg.keepalive
+        return not (self.worker.alive and self.cfg.keepalive)
 
     def _build_http_scope(self, request, sockname, peername):
         """Build ASGI HTTP scope from parsed request."""
@@ -1531,11 +1544,14 @@ def _send_informational(self, status, headers, request):
         response += "\r\n"
         self._safe_write(response.encode("latin-1"))
 
-    def _send_response_start(self, status, headers, request):
+    def _send_response_start(self, status, headers, request, close=False):
         """Send HTTP response status and headers.
 
         Uses cached status lines and headers for common cases to avoid
         repeated string formatting and encoding.
+
+        ``close`` announces that the connection ends after this response, as
+        RFC 9112 section 9.6 requires of a server that will not reuse it.
         """
         # Get cached status line bytes
         reason = self._get_reason_phrase(status)
@@ -1546,6 +1562,7 @@ def _send_response_start(self, status, headers, request):
 
         has_date = False
         has_server = False
+        has_connection = False
 
         for name, value in headers:
             if isinstance(name, bytes):
@@ -1564,17 +1581,21 @@ def _send_response_start(self, status, headers, request):
 
             parts.append(b"\r\n")
 
-            # Track if Date/Server headers are present
+            # Track if Date/Server/Connection headers are present
             if name_lower == b"date":
                 has_date = True
             elif name_lower == b"server":
                 has_server = True
+            elif name_lower == b"connection":
+                has_connection = True
 
         # Add default headers if not present
         if not has_server:
             parts.append(_CACHED_SERVER_HEADER)
         if not has_date:
             parts.append(_get_cached_date_header())
+        if close and not has_connection:
+            parts.append(b"Connection: close\r\n")
 
         parts.append(b"\r\n")
 

tests/test_asgi_worker.py

@@ -548,6 +548,74 @@ def test_reason_phrases(self):
         assert protocol._get_reason_phrase(500) == "Internal Server Error"
         assert protocol._get_reason_phrase(999) == "Unknown"
 
+    @pytest.mark.parametrize(
+        "requires_close,req_close,alive,keepalive,expected",
+        [
+            (False, False, True, 2, False),   # reusable connection
+            (True, False, True, 2, True),     # response framing forbids reuse
+            (False, True, True, 2, True),     # client sent Connection: close
+            (False, False, False, 2, True),   # worker shutting down
+            (False, False, True, 0, True),    # keepalive disabled
+        ],
+    )
+    def test_will_close_matches_keepalive_decision(
+        self, requires_close, req_close, alive, keepalive, expected
+    ):
+        """_will_close is the inverse of the keepalive decision, case for case."""
+        from gunicorn.asgi.protocol import ASGIProtocol
+
+        worker = mock.Mock()
+        worker.cfg = Config()
+        worker.cfg.set("keepalive", keepalive)
+        worker.alive = alive
+        worker.log = mock.Mock()
+        worker.asgi = mock.Mock()
+
+        protocol = ASGIProtocol(worker)
+        request = mock.Mock()
+        request.should_close.return_value = req_close
+
+        assert protocol._will_close(request, requires_close) is expected
+
+    @pytest.mark.parametrize("close", [True, False])
+    def test_response_start_announces_connection_close(self, close):
+        """RFC 9112 s9.6: a server that will close MUST say so."""
+        from gunicorn.asgi.protocol import ASGIProtocol
+
+        worker = mock.Mock()
+        worker.cfg = Config()
+        worker.log = mock.Mock()
+        worker.asgi = mock.Mock()
+
+        protocol = ASGIProtocol(worker)
+        request = mock.Mock()
+        request.version = (1, 1)
+
+        protocol._send_response_start(
+            200, [(b"content-type", b"text/plain")], request, close=close
+        )
+
+        assert (b"Connection: close\r\n" in protocol._response_buffer) is close
+
+    def test_response_start_keeps_app_connection_header(self):
+        """An app-supplied Connection header is not duplicated."""
+        from gunicorn.asgi.protocol import ASGIProtocol
+
+        worker = mock.Mock()
+        worker.cfg = Config()
+        worker.log = mock.Mock()
+        worker.asgi = mock.Mock()
+
+        protocol = ASGIProtocol(worker)
+        request = mock.Mock()
+        request.version = (1, 1)
+
+        protocol._send_response_start(
+            200, [(b"connection", b"close")], request, close=True
+        )
+
+        assert protocol._response_buffer.lower().count(b"connection:") == 1
+
     def test_scope_building(self):
         """Test HTTP scope building."""
         from gunicorn.asgi.protocol import ASGIProtocol