Vhost support using multiple TLS certificates

sanic/app.py

@@ -19,7 +19,7 @@
 from inspect import isawaitable
 from pathlib import Path
 from socket import socket
-from ssl import Purpose, SSLContext, create_default_context
+from ssl import SSLContext
 from traceback import format_exc
 from types import SimpleNamespace
 from typing import (
@@ -77,6 +77,7 @@
 from sanic.server.protocols.websocket_protocol import WebSocketProtocol
 from sanic.server.websockets.impl import ConnectionClosed
 from sanic.signals import Signal, SignalRouter
+from sanic.tls import process_to_context
 from sanic.touchup import TouchUp, TouchUpMeta
 
 
@@ -1258,16 +1259,6 @@ def _helper(
         auto_reload=False,
     ):
         """Helper function used by `run` and `create_server`."""
-
-        if isinstance(ssl, dict):
-            # try common aliaseses
-            cert = ssl.get("cert") or ssl.get("certificate")
-            key = ssl.get("key") or ssl.get("keyfile")
-            if cert is None or key is None:
-                raise ValueError("SSLContext or certificate and key required.")
-            context = create_default_context(purpose=Purpose.CLIENT_AUTH)
-            context.load_cert_chain(cert, keyfile=key)
-            ssl = context
         if self.config.PROXIES_COUNT and self.config.PROXIES_COUNT < 0:
             raise ValueError(
                 "PROXIES_COUNT cannot be negative. "
@@ -1277,6 +1268,33 @@ def _helper(
 
         self.error_handler.debug = debug
         self.debug = debug
+        if self.configure_logging and debug:
+            logger.setLevel(logging.DEBUG)
+        if (
+            self.config.LOGO
+            and os.environ.get("SANIC_SERVER_RUNNING") != "true"
+        ):
+            logger.debug(
+                self.config.LOGO
+                if isinstance(self.config.LOGO, str)
+                else BASE_LOGO
+            )
+        # Serve
+        if host and port:
+            proto = "http"
+            if ssl is not None:
+                proto = "https"
+            if unix:
+                logger.info(f"Goin' Fast @ {unix} {proto}://...")
+            else:
+                logger.info(f"Goin' Fast @ {proto}://{host}:{port}")
+
+        debug_mode = "enabled" if self.debug else "disabled"
+        reload_mode = "enabled" if auto_reload else "disabled"
+        logger.debug(f"Sanic auto-reload: {reload_mode}")
+        logger.debug(f"Sanic debug mode: {debug_mode}")
+
+        ssl = process_to_context(ssl)
 
         server_settings = {
             "protocol": protocol,
@@ -1305,37 +1323,9 @@ def _helper(
             listeners = [partial(listener, self) for listener in listeners]
             server_settings[settings_name] = listeners
 
-        if self.configure_logging and debug:
-            logger.setLevel(logging.DEBUG)
-
-        if (
-            self.config.LOGO
-            and os.environ.get("SANIC_SERVER_RUNNING") != "true"
-        ):
-            logger.debug(
-                self.config.LOGO
-                if isinstance(self.config.LOGO, str)
-                else BASE_LOGO
-            )
-
         if run_async:
             server_settings["run_async"] = True
 
-        # Serve
-        if host and port:
-            proto = "http"
-            if ssl is not None:
-                proto = "https"
-            if unix:
-                logger.info(f"Goin' Fast @ {unix} {proto}://...")
-            else:
-                logger.info(f"Goin' Fast @ {proto}://{host}:{port}")
-
-        debug_mode = "enabled" if self.debug else "disabled"
-        reload_mode = "enabled" if auto_reload else "disabled"
-        logger.debug(f"Sanic auto-reload: {reload_mode}")
-        logger.debug(f"Sanic debug mode: {debug_mode}")
-
         return server_settings
 
     def _build_endpoint_name(self, *parts):

sanic/models/server_types.py

@@ -20,6 +20,7 @@ class ConnInfo:
         "peername",
         "server_port",
         "server",
+        "server_name",
         "sockname",
         "ssl",
     )
@@ -31,7 +32,10 @@ def __init__(self, transport: TransportProtocol, unix=None):
         self.server_port = self.client_port = 0
         self.client_ip = ""
         self.sockname = addr = transport.get_extra_info("sockname")
-        self.ssl: bool = bool(transport.get_extra_info("sslcontext"))
+        self.ssl = bool(transport.get_extra_info("sslcontext"))
+        # Server name resolves to "" if not SSL or no SNI
+        sslobj = transport.get_extra_info("ssl_object")
+        self.server_name = getattr(sslobj, "server_name", None) or ""
 
         if isinstance(addr, str):  # UNIX socket
             self.server = unix or addr

sanic/tls.py

@@ -0,0 +1,116 @@
+import os
+import ssl
+
+from sanic.log import logger
+
+
+# Only allow secure ciphers, notably leaving out AES-CBC mode
+# OpenSSL chooses ECDSA or RSA depending on the cert in use
+CIPHERS_TLS12 = [
+    "ECDHE-ECDSA-CHACHA20-POLY1305",
+    "ECDHE-ECDSA-AES256-GCM-SHA384",
+    "ECDHE-ECDSA-AES128-GCM-SHA256",
+    "ECDHE-RSA-CHACHA20-POLY1305",
+    "ECDHE-RSA-AES256-GCM-SHA384",
+    "ECDHE-RSA-AES128-GCM-SHA256",
+]
+
+
+def create_context(certfile=None, keyfile=None):
+    """Create a context with secure crypto and HTTP/1.1 in protocols."""
+    context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
+    context.minimum_version = ssl.TLSVersion.TLSv1_2
+    context.set_ciphers(":".join(CIPHERS_TLS12))
+    context.set_alpn_protocols(["http/1.1"])
+    if certfile and keyfile:
+        context.load_cert_chain(certfile, keyfile)
+    return context
+
+
+def process_to_context(context):
+    """Process app.run ssl argument from easy formats to full SSLContext."""
+    if context is None or isinstance(context, ssl.SSLContext):
+        return context
+    if isinstance(context, dict):
+        # try common aliaseses
+        certfile = context.get("cert") or context.get("certificate")
+        keyfile = context.get("key") or context.get("keyfile")
+        if not certfile or not keyfile:
+            raise ValueError("SSL dict needs filenames for cert and key.")
+        return create_context(certfile, keyfile)
+    if isinstance(context, (list, tuple)):
+        return SSLSelector(*context).context
+    raise ValueError(
+        f"Invalid ssl argument type {type(context)}."
+        " Expecting a list of certdirs, a dict or an SSLContext."
+    )
+
+
+def load_cert_dir(p):
+    if os.path.isfile(p):
+        raise ValueError(f"Certificate folder expected but {p} is a file.")
+    keyfile = os.path.join(p, "privkey.pem")
+    certfile = os.path.join(p, "fullchain.pem")
+    if not os.access(keyfile, os.R_OK):
+        raise Exception(
+            f"Certificate not found or permission denied {keyfile}"
+        )
+    if not os.access(certfile, os.R_OK):
+        raise Exception(
+            f"Certificate not found or permission denied {certfile}"
+        )
+    cert = ssl._ssl._test_decode_cert(certfile)
+    return cert, create_context(certfile, keyfile)
+
+
+class SSLSelector:
+    def __init__(self, *paths):
+        """Automatically select SSL certificate based on the hostname that the
+        client is trying to access, via SSL SNI. Paths to certificate folders
+        with privkey.pem and fullchain.pem in them should be provided, and
+        will be matched in the order given whenever there is a new connection.
+        """
+        self.context = create_context()
+        self.context.sni_callback = self._sni_callback
+        self.names = []
+        self.certs = []
+        for p in paths:
+            cert, ctx = load_cert_dir(p)
+            self.names += [
+                name
+                for t, name in cert["subjectAltName"]
+                if t in ["DNS", "IP Address"]
+            ]
+            self.certs.append((cert, ctx))
+        logger.debug(f"Certificate vhosts: {', '.join(self.names)}")
+
+    def find(self, hostname):
+        """Find the first certificate that matches the given hostname.
+
+        :raises ssl.CertificateError: No matching certificate found.
+        :return: A matching ssl.SSLContext object if found."""
+        if not hostname:
+            raise ssl.CertificateError(
+                "The client provided no SNI to match for certificate."
+            )
+        for cert, ctx in self.certs:
+            try:
+                ssl.match_hostname(cert, hostname)
+                return ctx
+            except ssl.CertificateError:
+                pass
+        raise ssl.CertificateError(
+            f"No certificate found matching hostname {hostname!r}"
+        )
+
+    def _sni_callback(self, sslobj, server_name, ctx):
+        sslobj.server_name = server_name
+        try:
+            sslobj.context = self.find(server_name)
+        except ssl.CertificateError:
+            logger.debug(
+                f"Rejecting TLS connection to unrecognized SNI {server_name!r}"
+            )
+            # This would show ERR_SSL_UNRECOGNIZED_NAME_ALERT on client side if
+            # asyncio/uvloop did proper SSL shutdown. They don't.
+            return ssl.ALERT_DESCRIPTION_UNRECOGNIZED_NAME

tests/test_requests.py

@@ -1202,7 +1202,7 @@ async def handler(request):
             "/test", server_kwargs={"ssl": ssl_dict}
         )
 
-    assert str(excinfo.value) == "SSLContext or certificate and key required."
+    assert str(excinfo.value) == "SSL dict needs filenames for cert and key."
 
 
 def test_form_with_multiple_values(app):