Implement asgiref tls extension

tests/conftest.py

@@ -20,6 +20,13 @@ def tls_certificate(tls_certificate_authority: trustme.CA) -> trustme.LeafCert:
     )
 
 
+@pytest.fixture
+def tls_client_certificate(tls_certificate_authority: trustme.CA) -> trustme.LeafCert:
+    return tls_certificate_authority.issue_cert(
+        "[email protected]", common_name="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:
@@ -59,3 +66,10 @@ def tls_ca_ssl_context(tls_certificate: trustme.LeafCert) -> ssl.SSLContext:
     ssl_ctx = ssl.SSLContext()
     tls_certificate.configure_cert(ssl_ctx)
     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

tests/test_ssl.py

@@ -1,3 +1,5 @@
+import ssl
+
 import httpx
 import pytest
 
@@ -29,6 +31,56 @@ async def test_run(
     assert response.status_code == 204
 
 
+@pytest.mark.asyncio
+async def test_run_httptools_client_cert(
+    tls_ca_ssl_context,
+    tls_ca_certificate_pem_path,
+    tls_ca_certificate_private_key_path,
+    tls_client_certificate_pem_path,
+):
+    config = Config(
+        app=app,
+        loop="asyncio",
+        http="httptools",
+        limit_max_requests=1,
+        ssl_keyfile=tls_ca_certificate_private_key_path,
+        ssl_certfile=tls_ca_certificate_pem_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.asyncio
+async def test_run_h11_client_cert(
+    tls_ca_ssl_context,
+    tls_ca_certificate_pem_path,
+    tls_ca_certificate_private_key_path,
+    tls_client_certificate_pem_path,
+):
+    config = Config(
+        app=app,
+        loop="asyncio",
+        http="h11",
+        limit_max_requests=1,
+        ssl_keyfile=tls_ca_certificate_private_key_path,
+        ssl_certfile=tls_ca_certificate_pem_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.asyncio
 async def test_run_chain(
     tls_ca_ssl_context, tls_certificate_pem_path, tls_ca_certificate_pem_path

uvicorn/config.py

@@ -213,6 +213,7 @@ def __init__(
         self.callback_notify = callback_notify
         self.ssl_keyfile = ssl_keyfile
         self.ssl_certfile = ssl_certfile
+        self.ssl_cert_pem: Optional[str] = None
         self.ssl_keyfile_password = ssl_keyfile_password
         self.ssl_version = ssl_version
         self.ssl_cert_reqs = ssl_cert_reqs
@@ -316,6 +317,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

@@ -18,6 +18,7 @@
     get_local_addr,
     get_path_with_query_string,
     get_remote_addr,
+    get_tls_info,
     is_ssl,
 )
 
@@ -69,6 +70,7 @@ def __init__(
         self.server = None
         self.client = None
         self.scheme = None
+        self.tls = None
 
         # Per-request state
         self.scope = None
@@ -85,6 +87,11 @@ def connection_made(self, transport):
         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 - " % tuple(self.client) if self.client else ""
             self.logger.log(TRACE_LOG_LEVEL, "%sHTTP connection made", prefix)
@@ -171,8 +178,12 @@ def handle_events(self):
                     "raw_path": raw_path,
                     "query_string": query_string,
                     "headers": self.headers,
+                    "extensions": {},
                 }
 
+                if self.config.is_ssl:
+                    self.scope["extensions"]["tls"] = self.tls
+
                 for name, value in self.headers:
                     if name == b"connection":
                         tokens = [token.lower().strip() for token in value.split(b",")]

uvicorn/protocols/http/httptools_impl.py

@@ -19,6 +19,7 @@
     get_local_addr,
     get_path_with_query_string,
     get_remote_addr,
+    get_tls_info,
     is_ssl,
 )
 
@@ -75,6 +76,7 @@ def __init__(
         self.client = None
         self.scheme = None
         self.pipeline = []
+        self.tls = None
 
         # Per-request state
         self.url = None
@@ -93,6 +95,11 @@ def connection_made(self, transport):
         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 - " % tuple(self.client) if self.client else ""
             self.logger.log(TRACE_LOG_LEVEL, "%sHTTP connection made", prefix)
@@ -211,8 +218,12 @@ def on_url(self, url):
             "raw_path": raw_path,
             "query_string": parsed_url.query if parsed_url.query else b"",
             "headers": self.headers,
+            "extensions": {},
         }
 
+        if self.config.is_ssl:
+            self.scope["extensions"]["tls"] = self.tls
+
     def on_header(self, name: bytes, value: bytes):
         name = name.lower()
         if name == b"expect" and value.lower() == b"100-continue":

uvicorn/protocols/utils.py

@@ -1,9 +1,29 @@
 import asyncio
+import ssl
 import urllib.parse
-from typing import Optional, Tuple
+from typing import Any, Dict, Optional, Tuple
 
 from asgiref.typing import WWWScope
 
+RDNS_MAPPING: Dict[str, str] = {
+    "commonName": "CN",
+    "localityName": "L",
+    "stateOrProvinceName": "ST",
+    "organizationName": "O",
+    "organizationalUnitName": "OU",
+    "countryName": "C",
+    "streetAddress": "STREET",
+    "domainComponent": "DC",
+    "userId": "UID",
+}
+
+TLS_VERSION_MAP: Dict[str, int] = {
+    "TLSv1": 0x0301,
+    "TLSv1.1": 0x0302,
+    "TLSv1.2": 0x0303,
+    "TLSv1.3": 0x0304,
+}
+
 
 def get_remote_addr(transport: asyncio.Transport) -> Optional[Tuple[str, int]]:
     socket_info = transport.get_extra_info("socket")
@@ -54,3 +74,55 @@ def get_path_with_query_string(scope: WWWScope) -> str:
             path_with_query_string, scope["query_string"].decode("ascii")
         )
     return path_with_query_string
+
+
+def get_tls_info(transport: asyncio.Transport) -> Optional[Dict]:
+
+    ###
+    # server_cert: Unable to set from transport information
+    # client_cert_chain: Just the peercert, currently no access to the full cert chain
+    # client_cert_name:
+    # client_cert_error: No access to this
+    # tls_version:
+    # cipher_suite: Too hard to convert without direct access to openssl
+    ###
+
+    ssl_info: Dict[str, Any] = {
+        "server_cert": None,
+        "client_cert_chain": [],
+        "client_cert_name": None,
+        "client_cert_error": None,
+        "tls_version": None,
+        "cipher_suite": None,
+    }
+
+    ssl_object = transport.get_extra_info("ssl_object", default=None)
+    peercert = ssl_object.getpeercert()
+
+    if peercert:
+        rdn_strings = []
+        for rdn in peercert["subject"]:
+            rdn_strings.append(
+                "+".join(
+                    [
+                        "%s = %s" % (RDNS_MAPPING[entry[0]], entry[1])
+                        for entry in reversed(rdn)
+                        if entry[0] in RDNS_MAPPING
+                    ]
+                )
+            )
+
+        ssl_info["client_cert_chain"] = [
+            ssl.DER_cert_to_PEM_cert(ssl_object.getpeercert(binary_form=True))
+        ]
+        ssl_info["client_cert_name"] = ", ".join(rdn_strings) if rdn_strings else ""
+        ssl_info["tls_version"] = (
+            TLS_VERSION_MAP[ssl_object.version()]
+            if ssl_object.version() in TLS_VERSION_MAP
+            else None
+        )
+        ssl_info["cipher_suite"] = list(ssl_object.cipher())
+
+        return ssl_info
+
+    return None