Decode headers as UTF-8 also in ASGI

sanic/asgi.py

@@ -6,7 +6,7 @@
 from urllib.parse import quote
 
 from sanic.compat import Header
-from sanic.exceptions import ServerError
+from sanic.exceptions import BadRequest, ServerError
 from sanic.helpers import Default
 from sanic.http import Stage
 from sanic.log import error_logger, logger
@@ -132,12 +132,20 @@ async def create(
         instance.sanic_app.state.is_started = True
         setattr(instance.transport, "add_task", sanic_app.loop.create_task)
 
-        headers = Header(
-            [
-                (key.decode("latin-1"), value.decode("latin-1"))
-                for key, value in scope.get("headers", [])
-            ]
-        )
+        try:
+            headers = Header(
+                [
+                    (
+                        key.decode("ASCII"),
+                        value.decode(errors="surrogateescape"),
+                    )
+                    for key, value in scope.get("headers", [])
+                ]
+            )
+        except UnicodeDecodeError:
+            raise BadRequest(
+                "Header names can only contain US-ASCII characters"
+            )
         path = (
             scope["path"][1:]
             if scope["path"].startswith("/")

tests/test_asgi.py

@@ -7,6 +7,9 @@
 import pytest
 import uvicorn
 
+from httpx import Headers
+from pytest import MonkeyPatch
+
 from sanic import Sanic
 from sanic.application.state import Mode
 from sanic.asgi import ASGIApp, Lifespan, MockTransport
@@ -626,3 +629,26 @@ async def before_server_stop(_):
             )
         ]
     )
+
+
+@pytest.mark.asyncio
+async def test_asgi_headers_decoding(app: Sanic, monkeypatch: MonkeyPatch):
+    @app.get("/")
+    def handler(request: Request):
+        return text("")
+
+    headers_init = Headers.__init__
+
+    def mocked_headers_init(self, *args, **kwargs):
+        if "encoding" in kwargs:
+            kwargs.pop("encoding")
+        headers_init(self, encoding="utf-8", *args, **kwargs)
+
+    monkeypatch.setattr(Headers, "__init__", mocked_headers_init)
+
+    message = "Header names can only contain US-ASCII characters"
+    with pytest.raises(BadRequest, match=message):
+        _, response = await app.asgi_client.get("/", headers={"😂": "😅"})
+
+    _, response = await app.asgi_client.get("/", headers={"Test-Header": "😅"})
+    assert response.status_code == 200