Vhost support using multiple TLS certificates (sanic-org#2270)

* Initial support for using multiple SSL certificates.

* Also list IP address subjectAltNames on log.

* Use Python 3.7+ way of specifying TLSv1.2 as the minimum version. Linter fixes.

* isort

* Cleanup, store server name for later use. Add RSA ciphers. Log rejected SNIs.

* Cleanup, linter.

* Alter the order of initial log messages and handling. In particular, enable debug mode early so that debug messages during init can be shown.

* Store server name (SNI) to conn_info.

* Update test with new error message.

* Refactor for readability.

* Cleanup

* Replace old expired test cert with new ones and a script for regenerating them as needed.

* Refactor TLS tests to a separate file.

* Add cryptography to dev deps for rebuilding TLS certs.

* Minor adjustment to messages.

* Tests added for new TLS code.

* Find the correct log row before testing for message. The order was different on CI.

* More log message order fixup. The tests do not account for the logo being printed first.

* Another attempt at log message indexing fixup.

* Major TLS refactoring.

CertSelector now allows dicts and SSLContext within its list.
Server names are stored even when no list is used.
SSLContext.sanic now contains a dict with any setting passed and information extracted from cert.
That information is available on request.conn_info.cert.
Type annotations added.
More tests incl. a handler for faking hostname in tests.

* Remove a problematic logger test that apparently was not adding any coverage or value to anything.

* Revert accidental commit of uvloop disable.

* Typing fixes / refactoring.

* Additional test for cert selection. Certs recreated without DNS:localhost on sanic.example cert.

* Add tests for single certificate path shorthand and SNI information.

* Move TLS dict processing to CertSimple, make the names field optional and use names from the cert if absent.

* Sanic CLI options --tls and --tls-strict-host to use the new features.

* SSL argument typing updated

* Use ValueError for internal message passing to avoid CertificateError's odd message formatting.

* Linter

* Test CLI TLS options.

* Maybe the right codeclimate option now...

* Improved TLS argument help, removed support for combining --cert/--key with --tls.

* Removed support for strict checking without any certs, black forced fscked up formatting.

* Update CLI tests for stricter TLS options.

Co-authored-by: L. Karkkainen <tronic@users.noreply.github.com>
Co-authored-by: Adam Hopkins <admhpkns@gmail.com>

.codeclimate.yml

@@ -21,5 +21,7 @@ checks:
     config:
       threshold: 40
   complex-logic:
+    enabled: false
+  method-complexity:
     config:
       threshold: 10

sanic/__main__.py

@@ -4,7 +4,7 @@
 from argparse import ArgumentParser, RawTextHelpFormatter
 from importlib import import_module
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import Union
 
 from sanic_routing import __version__ as __routing_version__  # type: ignore
 
@@ -79,10 +79,30 @@ def main():
         help="location of unix socket\n ",
     )
     parser.add_argument(
-        "--cert", dest="cert", type=str, help="Location of certificate for SSL"
+        "--cert",
+        dest="cert",
+        type=str,
+        help="Location of fullchain.pem, bundle.crt or equivalent",
     )
     parser.add_argument(
-        "--key", dest="key", type=str, help="location of keyfile for SSL\n "
+        "--key",
+        dest="key",
+        type=str,
+        help="Location of privkey.pem or equivalent .key file",
+    )
+    parser.add_argument(
+        "--tls",
+        metavar="DIR",
+        type=str,
+        action="append",
+        help="TLS certificate folder with fullchain.pem and privkey.pem\n"
+        "May be specified multiple times to choose of multiple certificates",
+    )
+    parser.add_argument(
+        "--tls-strict-host",
+        dest="tlshost",
+        action="store_true",
+        help="Only allow clients that send an SNI matching server certs\n ",
     )
     parser.add_bool_arguments(
         "--access-logs", dest="access_log", help="display access logs"
@@ -126,6 +146,26 @@ def main():
     )
     args = parser.parse_args()
 
+    # Custom TLS mismatch handling for better diagnostics
+    if (
+        # one of cert/key missing
+        bool(args.cert) != bool(args.key)
+        # new and old style args used together
+        or args.tls
+        and args.cert
+        # strict host checking without certs would always fail
+        or args.tlshost
+        and not args.tls
+        and not args.cert
+    ):
+        parser.print_usage(sys.stderr)
+        error_logger.error(
+            "sanic: error: TLS certificates must be specified by either of:\n"
+            "  --cert certdir/fullchain.pem --key certdir/privkey.pem\n"
+            "  --tls certdir  (equivalent to the above)"
+        )
+        sys.exit(1)
+
     try:
         module_path = os.path.abspath(os.getcwd())
         if module_path not in sys.path:
@@ -155,14 +195,18 @@ def main():
                     f"Perhaps you meant {args.module}.app?"
                 )
 
+        ssl: Union[None, dict, str, list] = []
+        if args.tlshost:
+            ssl.append(None)
         if args.cert is not None or args.key is not None:
-            ssl: Optional[Dict[str, Any]] = {
-                "cert": args.cert,
-                "key": args.key,
-            }
-        else:
+            ssl.append(dict(cert=args.cert, key=args.key))
+        if args.tls:
+            ssl += args.tls
+        if not ssl:
             ssl = None
-
+        elif len(ssl) == 1 and ssl[0] is not None:
+            # Use only one cert, no TLSSelector.
+            ssl = ssl[0]
         kwargs = {
             "host": args.host,
             "port": args.port,

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 (
@@ -78,6 +78,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
 
 
@@ -952,7 +953,7 @@ def run(
         *,
         debug: bool = False,
         auto_reload: Optional[bool] = None,
-        ssl: Union[Dict[str, str], SSLContext, None] = None,
+        ssl: Union[None, SSLContext, dict, str, list, tuple] = None,
         sock: Optional[socket] = None,
         workers: int = 1,
         protocol: Optional[Type[Protocol]] = None,
@@ -979,7 +980,7 @@ def run(
         :type auto_relaod: bool
         :param ssl: SSLContext, or location of certificate and key
                     for SSL encryption of worker(s)
-        :type ssl: SSLContext or dict
+        :type ssl: str, dict, SSLContext or list
         :param sock: Socket for the server to accept connections from
         :type sock: socket
         :param workers: Number of processes received before it is respected
@@ -1089,7 +1090,7 @@ async def create_server(
         port: Optional[int] = None,
         *,
         debug: bool = False,
-        ssl: Union[Dict[str, str], SSLContext, None] = None,
+        ssl: Union[None, SSLContext, dict, str, list, tuple] = None,
         sock: Optional[socket] = None,
         protocol: Type[Protocol] = None,
         backlog: int = 100,
@@ -1281,16 +1282,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. "
@@ -1300,6 +1291,35 @@ 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:
+                # colon(:) is legal for a host only in an ipv6 address
+                display_host = f"[{host}]" if ":" in host else host
+                logger.info(f"Goin' Fast @ {proto}://{display_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,
@@ -1328,39 +1348,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:
-                # colon(:) is legal for a host only in an ipv6 address
-                display_host = f"[{host}]" if ":" in host else host
-                logger.info(f"Goin' Fast @ {proto}://{display_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

@@ -1,4 +1,6 @@
+from ssl import SSLObject
 from types import SimpleNamespace
+from typing import Optional
 
 from sanic.models.protocol_types import TransportProtocol
 
@@ -20,8 +22,10 @@ class ConnInfo:
         "peername",
         "server_port",
         "server",
+        "server_name",
         "sockname",
         "ssl",
+        "cert",
     )
 
     def __init__(self, transport: TransportProtocol, unix=None):
@@ -31,8 +35,16 @@ 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 = False
+        self.server_name = ""
+        self.cert = {}
+        sslobj: Optional[SSLObject] = transport.get_extra_info(
+            "ssl_object"
+        )  # type: ignore
+        if sslobj:
+            self.ssl = True
+            self.server_name = getattr(sslobj, "sanic_server_name", None) or ""
+            self.cert = getattr(sslobj.context, "sanic", {})
         if isinstance(addr, str):  # UNIX socket
             self.server = unix or addr
             return