Add ssl_context_factory for custom SSLContext configuration (#2920)

docs/deployment/index.md

@@ -225,6 +225,36 @@ It's also possible to use certificates with uvicorn's worker for gunicorn.
 $ gunicorn --keyfile=./key.pem --certfile=./cert.pem -k uvicorn.workers.UvicornWorker main:app
 ```
 
+### Customizing the SSL context
+
+For TLS scenarios that the `--ssl-*` flags don't cover (e.g., mutual TLS, custom `SSLContext.options`, bumping `minimum_version`, loading certificates from memory), pass an `ssl_context_factory` to `uvicorn.run()` or `Config`.
+
+The factory receives the `Config` instance and a `default_ssl_context_factory` callable that builds the standard context from the `ssl_*` settings on `Config`. Use it to start from uvicorn's default and mutate it, or ignore it and build your own context from scratch - the `ssl_*` settings are only consumed by the default factory, so if you don't call it they're effectively unused.
+
+```python
+import ssl
+from collections.abc import Callable
+
+import uvicorn
+from uvicorn.config import Config
+
+
+def ssl_context_factory(config: Config, default_ssl_context_factory: Callable[[], ssl.SSLContext]) -> ssl.SSLContext:
+    context = default_ssl_context_factory()
+    context.minimum_version = ssl.TLSVersion.TLSv1_3
+    return context
+
+
+uvicorn.run(
+    "main:app",
+    ssl_keyfile="key.pem",
+    ssl_certfile="cert.pem",
+    ssl_context_factory=ssl_context_factory,
+)
+```
+
+The factory is called inside each worker process, so it works with `--reload` and `--workers > 1`. The factory itself must be picklable in those modes (a top-level function is fine; lambdas and local closures are not). The `ssl_*` settings on `Config` are only consumed by `default_ssl_context_factory()`; if you build the context yourself without calling it, those settings are ignored.
+
 ## Proxies and Forwarded Headers
 
 When running an application behind one or more proxies, certain information about the request is lost.

docs/settings.md

@@ -138,6 +138,8 @@ The [SSL context](https://docs.python.org/3/library/ssl.html#ssl.SSLContext) can
 
 To understand more about the SSL context options, please refer to the [Python documentation](https://docs.python.org/3/library/ssl.html).
 
+For advanced TLS scenarios that the flags above don't cover (e.g., mutual TLS, certificate pinning, custom `SSLContext.options`), pass an `ssl_context_factory` to `uvicorn.run()` or `Config`. See [Running with HTTPS](deployment/index.md#customizing-the-ssl-context) for details.
+
 ## Resource Limits
 
 * `--limit-concurrency <int>` - Maximum number of concurrent connections or tasks to allow, before issuing HTTP 503 responses. Useful for ensuring known memory usage patterns even under over-resourced loads.

tests/test_ssl.py

@@ -1,9 +1,17 @@
+from __future__ import annotations
+
+import ssl
+from collections.abc import Callable
+from typing import TypeAlias
+
 import httpx
 import pytest
 
 from tests.utils import run_server
 from uvicorn.config import Config
 
+DefaultFactory: TypeAlias = Callable[[], ssl.SSLContext]
+
 
 async def app(scope, receive, send):
     assert scope["type"] == "http"
@@ -92,3 +100,108 @@ async def test_run_password(
         async with httpx.AsyncClient(verify=tls_ca_ssl_context) as client:
             response = await client.get(f"https://127.0.0.1:{unused_tcp_port}")
     assert response.status_code == 204
+
+
+@pytest.mark.anyio
+async def test_run_ssl_context_factory_default(
+    tls_ca_ssl_context: ssl.SSLContext,
+    tls_certificate_server_cert_path: str,
+    tls_certificate_private_key_path: str,
+    unused_tcp_port: int,
+) -> None:
+    """A factory that just delegates to the default factory should produce a working server."""
+
+    def ssl_context_factory(config: Config, default_ssl_context_factory: DefaultFactory) -> ssl.SSLContext:
+        return default_ssl_context_factory()
+
+    config = Config(
+        app=app,
+        loop="asyncio",
+        limit_max_requests=1,
+        ssl_keyfile=tls_certificate_private_key_path,
+        ssl_certfile=tls_certificate_server_cert_path,
+        ssl_context_factory=ssl_context_factory,
+        port=unused_tcp_port,
+    )
+    async with run_server(config):
+        async with httpx.AsyncClient(verify=tls_ca_ssl_context) as client:
+            response = await client.get(f"https://127.0.0.1:{unused_tcp_port}")
+    assert response.status_code == 204
+
+
+@pytest.mark.anyio
+async def test_run_ssl_context_factory_custom(
+    tls_ca_ssl_context: ssl.SSLContext,
+    tls_certificate_server_cert_path: str,
+    tls_certificate_private_key_path: str,
+    unused_tcp_port: int,
+) -> None:
+    """A factory that builds its own SSLContext from scratch should work without ssl_keyfile/ssl_certfile."""
+
+    def ssl_context_factory(config: Config, default_ssl_context_factory: DefaultFactory) -> ssl.SSLContext:
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        ctx.load_cert_chain(tls_certificate_server_cert_path, tls_certificate_private_key_path)
+        return ctx
+
+    config = Config(
+        app=app,
+        loop="asyncio",
+        limit_max_requests=1,
+        ssl_context_factory=ssl_context_factory,
+        port=unused_tcp_port,
+    )
+    async with run_server(config):
+        async with httpx.AsyncClient(verify=tls_ca_ssl_context) as client:
+            response = await client.get(f"https://127.0.0.1:{unused_tcp_port}")
+    assert response.status_code == 204
+
+
+def test_ssl_context_factory_mutates_default(
+    tls_certificate_server_cert_path: str,
+    tls_certificate_private_key_path: str,
+) -> None:
+    """The factory can call the default and mutate the result (e.g., bump TLS minimum version)."""
+
+    def ssl_context_factory(config: Config, default_ssl_context_factory: DefaultFactory) -> ssl.SSLContext:
+        ctx = default_ssl_context_factory()
+        ctx.minimum_version = ssl.TLSVersion.TLSv1_3
+        return ctx
+
+    config = Config(
+        app=app,
+        ssl_keyfile=tls_certificate_private_key_path,
+        ssl_certfile=tls_certificate_server_cert_path,
+        ssl_context_factory=ssl_context_factory,
+    )
+    config.load()
+    assert config.is_ssl
+    assert isinstance(config.ssl, ssl.SSLContext)
+    assert config.ssl.minimum_version == ssl.TLSVersion.TLSv1_3
+
+
+def test_default_ssl_context_factory_requires_ssl_certfile() -> None:
+    """Calling `default_ssl_context_factory()` without `ssl_certfile` raises a clear error."""
+
+    def ssl_context_factory(config: Config, default_ssl_context_factory: DefaultFactory) -> ssl.SSLContext:
+        return default_ssl_context_factory()
+
+    config = Config(app=app, ssl_context_factory=ssl_context_factory)
+    with pytest.raises(RuntimeError, match="requires `ssl_certfile`"):
+        config.load()
+
+
+def test_ssl_context_factory_must_return_ssl_context() -> None:
+    def bad_factory(config: Config, default_ssl_context_factory: DefaultFactory) -> object:
+        return "not an SSLContext"
+
+    config = Config(app=app, ssl_context_factory=bad_factory)  # type: ignore[arg-type]
+    with pytest.raises(TypeError, match="must return an `ssl.SSLContext`"):
+        config.load()
+
+
+def test_is_ssl_true_when_only_factory_set() -> None:
+    def ssl_context_factory(config: Config, default_ssl_context_factory: DefaultFactory) -> ssl.SSLContext:
+        return ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)  # pragma: no cover
+
+    config = Config(app=app, ssl_context_factory=ssl_context_factory)
+    assert config.is_ssl is True

uvicorn/config.py

@@ -225,6 +225,7 @@ class Config:
         ssl_cert_reqs: int = ssl.CERT_NONE,
         ssl_ca_certs: str | os.PathLike[str] | None = None,
         ssl_ciphers: str = "TLSv1",
+        ssl_context_factory: Callable[[Config, Callable[[], ssl.SSLContext]], ssl.SSLContext] | None = None,
         headers: list[tuple[str, str]] | None = None,
         factory: bool = False,
         h11_max_incomplete_event_size: int | None = None,
@@ -272,6 +273,7 @@ class Config:
         self.ssl_cert_reqs = ssl_cert_reqs
         self.ssl_ca_certs = ssl_ca_certs
         self.ssl_ciphers = ssl_ciphers
+        self.ssl_context_factory = ssl_context_factory
         self.headers: list[tuple[str, str]] = headers or []
         self.encoded_headers: list[tuple[bytes, bytes]] = []
         self.factory = factory
@@ -357,7 +359,7 @@ class Config:
 
     @property
     def is_ssl(self) -> bool:
-        return bool(self.ssl_keyfile or self.ssl_certfile)
+        return bool(self.ssl_keyfile or self.ssl_certfile or self.ssl_context_factory)
 
     @property
     def use_subprocess(self) -> bool:
@@ -418,9 +420,32 @@ class Config:
     def load(self) -> None:
         assert not self.loaded
 
-        if self.is_ssl:
+        if self.ssl_context_factory is not None:
+
+            def default_factory() -> ssl.SSLContext:
+                if not self.ssl_certfile:
+                    raise RuntimeError(
+                        "`default_ssl_context_factory()` requires `ssl_certfile` to be set on `Config`. "
+                        "Either pass `ssl_certfile` (and optionally `ssl_keyfile`) or build the `SSLContext` "
+                        "directly inside `ssl_context_factory` without calling the default factory."
+                    )
+                return create_ssl_context(
+                    keyfile=self.ssl_keyfile,
+                    certfile=self.ssl_certfile,
+                    password=self.ssl_keyfile_password,
+                    ssl_version=self.ssl_version,
+                    cert_reqs=self.ssl_cert_reqs,
+                    ca_certs=self.ssl_ca_certs,
+                    ciphers=self.ssl_ciphers,
+                )
+
+            context = self.ssl_context_factory(self, default_factory)
+            if not isinstance(context, ssl.SSLContext):
+                raise TypeError(f"`ssl_context_factory` must return an `ssl.SSLContext`, got {type(context).__name__}")
+            self.ssl: ssl.SSLContext | None = context
+        elif self.is_ssl:
             assert self.ssl_certfile
-            self.ssl: ssl.SSLContext | None = create_ssl_context(
+            self.ssl = create_ssl_context(
                 keyfile=self.ssl_keyfile,
                 certfile=self.ssl_certfile,
                 password=self.ssl_keyfile_password,

uvicorn/main.py

@@ -538,6 +538,7 @@ def run(
     ssl_cert_reqs: int = ssl.CERT_NONE,
     ssl_ca_certs: str | os.PathLike[str] | None = None,
     ssl_ciphers: str = "TLSv1",
+    ssl_context_factory: Callable[[Config, Callable[[], ssl.SSLContext]], ssl.SSLContext] | None = None,
     headers: list[tuple[str, str]] | None = None,
     use_colors: bool | None = None,
     app_dir: str | None = None,
@@ -593,6 +594,7 @@ def run(
         ssl_cert_reqs=ssl_cert_reqs,
         ssl_ca_certs=ssl_ca_certs,
         ssl_ciphers=ssl_ciphers,
+        ssl_context_factory=ssl_context_factory,
         headers=headers,
         use_colors=use_colors,
         factory=factory,