Implement asgiref tls extension

scripts/generate

@@ -0,0 +1,11 @@
+#!/bin/sh -e
+
+if [ -d 'venv' ] ; then
+    PREFIX="venv/bin/"
+else
+    PREFIX=""
+fi
+
+set -x
+
+${PREFIX}python -m tools.generate_tls_const

tests/conftest.py

@@ -56,6 +56,13 @@ def tls_certificate(tls_certificate_authority: trustme.CA) -> trustme.LeafCert:
     )
 
 
+@pytest.fixture
+def tls_client_certificate(request, tls_certificate_authority: trustme.CA) -> trustme.LeafCert:
+    return tls_certificate_authority.issue_cert(
+        "client@example.com", common_name=getattr(request, "param", "uvicorn client")
+    )
+
+
 @pytest.fixture
 def tls_ca_certificate_pem_path(tls_certificate_authority: trustme.CA):
     with tls_certificate_authority.cert_pem.tempfile() as ca_cert_pem:
@@ -109,6 +116,13 @@ def tls_ca_ssl_context(tls_certificate_authority: trustme.CA) -> ssl.SSLContext:
     return ssl_ctx
 
 
+@pytest.fixture
+def tls_client_certificate_pem_path(tls_client_certificate: trustme.LeafCert):
+    private_key_and_cert_chain = tls_client_certificate.private_key_and_cert_chain_pem
+    with private_key_and_cert_chain.tempfile() as client_cert_pem:
+        yield client_cert_pem
+
+
 @pytest.fixture(scope="package")
 def reload_directory_structure(tmp_path_factory: pytest.TempPathFactory):
     """

tests/test_ssl.py

@@ -1,3 +1,5 @@
+import ssl
+
 import httpx
 import pytest
 
@@ -34,6 +36,69 @@ async def test_run(
     assert response.status_code == 204
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "tls_client_certificate, expected_common_name",
+    [
+        ("test common name", "test common name"),
+        (' \\,+"<>;=\000\n\r ', 'CN=\\ \\\\\\,\\+\\"\\<\\>\\;\\=\\\x00\\0a\\0d\\ '),
+    ],
+    indirect=["tls_client_certificate"],
+)
+async def test_run_httptools_client_cert(
+    tls_ca_ssl_context,
+    tls_certificate_server_cert_path,
+    tls_certificate_private_key_path,
+    tls_ca_certificate_pem_path,
+    tls_client_certificate_pem_path,
+    expected_common_name,
+):
+    async def app(scope, receive, send):
+        assert scope["type"] == "http"
+        assert expected_common_name in scope["extensions"]["tls"]["client_cert_name"]
+        await send({"type": "http.response.start", "status": 204, "headers": []})
+        await send({"type": "http.response.body", "body": b"", "more_body": False})
+
+    config = Config(
+        app=app,
+        loop="asyncio",
+        http="httptools",
+        limit_max_requests=1,
+        ssl_keyfile=tls_certificate_private_key_path,
+        ssl_certfile=tls_certificate_server_cert_path,
+        ssl_ca_certs=tls_ca_certificate_pem_path,
+        ssl_cert_reqs=ssl.CERT_REQUIRED,
+    )
+    async with run_server(config):
+        async with httpx.AsyncClient(verify=tls_ca_ssl_context, cert=tls_client_certificate_pem_path) as client:
+            response = await client.get("https://127.0.0.1:8000")
+    assert response.status_code == 204
+
+
+@pytest.mark.anyio
+async def test_run_h11_client_cert(
+    tls_ca_ssl_context,
+    tls_ca_certificate_pem_path,
+    tls_certificate_server_cert_path,
+    tls_certificate_private_key_path,
+    tls_client_certificate_pem_path,
+):
+    config = Config(
+        app=app,
+        loop="asyncio",
+        http="h11",
+        limit_max_requests=1,
+        ssl_keyfile=tls_certificate_private_key_path,
+        ssl_certfile=tls_certificate_server_cert_path,
+        ssl_ca_certs=tls_ca_certificate_pem_path,
+        ssl_cert_reqs=ssl.CERT_REQUIRED,
+    )
+    async with run_server(config):
+        async with httpx.AsyncClient(verify=tls_ca_ssl_context, cert=tls_client_certificate_pem_path) as client:
+            response = await client.get("https://127.0.0.1:8000")
+    assert response.status_code == 204
+
+
 @pytest.mark.anyio
 async def test_run_chain(
     tls_ca_ssl_context,

tools/generate_tls_const.py

@@ -0,0 +1,49 @@
+import pprint
+import xml.etree.ElementTree as ET
+import subprocess
+
+import httpx
+
+GENERATED_FILENAME = "uvicorn/protocols/http/tls_const.py"
+
+TLS_PARAMETERS_URL = "https://www.iana.org/assignments/tls-parameters/tls-parameters.xml"
+NAMESPACES = {"iana": "http://www.iana.org/assignments"}
+TLS_CIPHER_SUITES_XPATH = './/iana:registry[@id="tls-parameters-4"]/iana:record'
+
+content = httpx.get(TLS_PARAMETERS_URL).content
+root = ET.fromstring(content)
+
+tls_cipher_suites = {}
+
+for record in root.findall(TLS_CIPHER_SUITES_XPATH, NAMESPACES):
+    cipher = record.find("iana:description", NAMESPACES).text
+    if cipher == "Unassigned":
+        continue
+    if cipher == "Reserved":
+        continue
+
+    value = record.find("iana:value", NAMESPACES).text
+    if "-" in value:
+        continue
+
+    vs = [int(v, 16) for v in value.split(",")]
+    code = (vs[0] << 8) + vs[1]
+    tls_cipher_suites[cipher] = code
+
+
+GENERATED_SOURCE = f"""
+# generated by tools/generate_tls_const.py
+
+from __future__ import annotations
+
+from typing import Final
+
+TLS_CIPHER_SUITES: Final[dict[str, int]] = {pprint.pformat(tls_cipher_suites)}
+"""
+
+
+with open(GENERATED_FILENAME, "wt") as fp:
+    fp.write(GENERATED_SOURCE)
+
+subprocess.run(["ruff", "format", GENERATED_FILENAME])
+subprocess.run(["ruff", "check", "--fix", GENERATED_FILENAME])

uvicorn/config.py

@@ -259,6 +259,7 @@ def __init__(
         self.callback_notify = callback_notify
         self.ssl_keyfile = ssl_keyfile
         self.ssl_certfile = ssl_certfile
+        self.ssl_cert_pem: str | None = None
         self.ssl_keyfile_password = ssl_keyfile_password
         self.ssl_version = ssl_version
         self.ssl_cert_reqs = ssl_cert_reqs
@@ -406,6 +407,8 @@ def load(self) -> None:
                 ca_certs=self.ssl_ca_certs,
                 ciphers=self.ssl_ciphers,
             )
+            with open(self.ssl_certfile) as file:
+                self.ssl_cert_pem = file.read()
         else:
             self.ssl = None
 

uvicorn/protocols/http/h11_impl.py

@@ -20,8 +20,20 @@
 )
 from uvicorn.config import Config
 from uvicorn.logging import TRACE_LOG_LEVEL
-from uvicorn.protocols.http.flow_control import CLOSE_HEADER, HIGH_WATER_LIMIT, FlowControl, service_unavailable
-from uvicorn.protocols.utils import get_client_addr, get_local_addr, get_path_with_query_string, get_remote_addr, is_ssl
+from uvicorn.protocols.http.flow_control import (
+    CLOSE_HEADER,
+    HIGH_WATER_LIMIT,
+    FlowControl,
+    service_unavailable,
+)
+from uvicorn.protocols.utils import (
+    get_client_addr,
+    get_local_addr,
+    get_path_with_query_string,
+    get_remote_addr,
+    get_tls_info,
+    is_ssl,
+)
 from uvicorn.server import ServerState
 
 
@@ -78,6 +90,7 @@ def __init__(
         self.server: tuple[str, int] | None = None
         self.client: tuple[str, int] | None = None
         self.scheme: Literal["http", "https"] | None = None
+        self.tls: dict[object, object] = {}
 
         # Per-request state
         self.scope: HTTPScope = None  # type: ignore[assignment]
@@ -96,6 +109,11 @@ def connection_made(  # type: ignore[override]
         self.client = get_remote_addr(transport)
         self.scheme = "https" if is_ssl(transport) else "http"
 
+        if self.config.is_ssl:
+            self.tls = get_tls_info(transport)
+            if self.tls:
+                self.tls["server_cert"] = self.config.ssl_cert_pem
+
         if self.logger.level <= TRACE_LOG_LEVEL:
             prefix = "%s:%d - " % self.client if self.client else ""
             self.logger.log(TRACE_LOG_LEVEL, "%sHTTP connection made", prefix)
@@ -205,8 +223,12 @@ def handle_events(self) -> None:
                     "query_string": query_string,
                     "headers": self.headers,
                     "state": self.app_state.copy(),
+                    "extensions": {},
                 }
 
+                if self.config.is_ssl:
+                    self.scope["extensions"]["tls"] = self.tls
+
                 upgrade = self._get_upgrade()
                 if upgrade == b"websocket" and self._should_upgrade_to_ws():
                     self.handle_websocket_upgrade(event)

uvicorn/protocols/http/httptools_impl.py

@@ -7,7 +7,12 @@
 import urllib
 from asyncio.events import TimerHandle
 from collections import deque
-from typing import Any, Callable, Literal, cast
+from typing import (
+    Any,
+    Callable,
+    Literal,
+    cast,
+)
 
 import httptools
 
@@ -21,8 +26,20 @@
 )
 from uvicorn.config import Config
 from uvicorn.logging import TRACE_LOG_LEVEL
-from uvicorn.protocols.http.flow_control import CLOSE_HEADER, HIGH_WATER_LIMIT, FlowControl, service_unavailable
-from uvicorn.protocols.utils import get_client_addr, get_local_addr, get_path_with_query_string, get_remote_addr, is_ssl
+from uvicorn.protocols.http.flow_control import (
+    CLOSE_HEADER,
+    HIGH_WATER_LIMIT,
+    FlowControl,
+    service_unavailable,
+)
+from uvicorn.protocols.utils import (
+    get_client_addr,
+    get_local_addr,
+    get_path_with_query_string,
+    get_remote_addr,
+    get_tls_info,
+    is_ssl,
+)
 from uvicorn.server import ServerState
 
 HEADER_RE = re.compile(b'[\x00-\x1f\x7f()<>@,;:[]={} \t\\"]')
@@ -79,6 +96,7 @@ def __init__(
         self.client: tuple[str, int] | None = None
         self.scheme: Literal["http", "https"] | None = None
         self.pipeline: deque[tuple[RequestResponseCycle, ASGI3Application]] = deque()
+        self.tls: dict[object, object] = {}
 
         # Per-request state
         self.scope: HTTPScope = None  # type: ignore[assignment]
@@ -98,6 +116,11 @@ def connection_made(  # type: ignore[override]
         self.client = get_remote_addr(transport)
         self.scheme = "https" if is_ssl(transport) else "http"
 
+        if self.config.is_ssl:
+            self.tls = get_tls_info(transport)
+            if self.tls:
+                self.tls["server_cert"] = self.config.ssl_cert_pem
+
         if self.logger.level <= TRACE_LOG_LEVEL:
             prefix = "%s:%d - " % self.client if self.client else ""
             self.logger.log(TRACE_LOG_LEVEL, "%sHTTP connection made", prefix)
@@ -220,8 +243,12 @@ def on_message_begin(self) -> None:
             "root_path": self.root_path,
             "headers": self.headers,
             "state": self.app_state.copy(),
+            "extensions": {},
         }
 
+        if self.config.is_ssl:
+            self.scope["extensions"]["tls"] = self.tls
+
     # Parser callbacks
     def on_url(self, url: bytes) -> None:
         self.url += url