class DummyExecutor(futures.Executor):
- def submit(
+ def submit( # type: ignore[override]
self, fn: Callable[..., _T], *args: Any, **kwargs: Any
) -> "futures.Future[_T]":
future = futures.Future() # type: futures.Future[_T]
future_set_exc_info(future, sys.exc_info())
return future
- def shutdown(self, wait: bool = True) -> None:
- pass
+ if sys.version_info >= (3, 9):
+
+ def shutdown(self, wait: bool = True, cancel_futures: bool = False) -> None:
+ pass
+
+ else:
+
+ def shutdown(self, wait: bool = True) -> None:
+ pass
dummy_executor = DummyExecutor()
return self.close(exc_info=err)
else:
self._ssl_accepting = False
- if not self._verify_cert(self.socket.getpeercert()):
- self.close()
- return
+ # Prior to the introduction of SNI, this is where we would check
+ # the server's claimed hostname.
+ assert ssl.HAS_SNI
self._finish_ssl_connect()
def _finish_ssl_connect(self) -> None:
self._ssl_connect_future = None
future_set_result_unless_cancelled(future, self)
- def _verify_cert(self, peercert: Any) -> bool:
- """Returns ``True`` if peercert is valid according to the configured
- validation mode and hostname.
-
- The ssl handshake already tested the certificate for a valid
- CA signature; the only thing that remains is to check
- the hostname.
- """
- if isinstance(self._ssl_options, dict):
- verify_mode = self._ssl_options.get("cert_reqs", ssl.CERT_NONE)
- elif isinstance(self._ssl_options, ssl.SSLContext):
- verify_mode = self._ssl_options.verify_mode
- assert verify_mode in (ssl.CERT_NONE, ssl.CERT_REQUIRED, ssl.CERT_OPTIONAL)
- if verify_mode == ssl.CERT_NONE or self._server_hostname is None:
- return True
- cert = self.socket.getpeercert()
- if cert is None and verify_mode == ssl.CERT_REQUIRED:
- gen_log.warning("No SSL certificate given")
- return False
- try:
- ssl.match_hostname(peercert, self._server_hostname)
- except ssl.CertificateError as e:
- gen_log.warning("Invalid SSL certificate: %s" % e)
- return False
- else:
- return True
-
def _handle_read(self) -> None:
if self._ssl_accepting:
self._do_ssl_handshake()
context = ssl_options_to_context(ssl_options, server_side=server_side)
if server_side is None:
server_side = False
- if ssl.HAS_SNI:
- # In python 3.4, wrap_socket only accepts the server_hostname
- # argument if HAS_SNI is true.
- # TODO: add a unittest (python added server-side SNI support in 3.4)
- # In the meantime it can be manually tested with
- # python3 -m tornado.httpclient https://sni.velox.ch
- return context.wrap_socket(
- socket, server_hostname=server_hostname, server_side=server_side, **kwargs
- )
- else:
- return context.wrap_socket(socket, server_side=server_side, **kwargs)
+ assert ssl.HAS_SNI
+ # TODO: add a unittest for hostname validation (python added server-side SNI support in 3.4)
+ # In the meantime it can be manually tested with
+ # python3 -m tornado.httpclient https://sni.velox.ch
+ return context.wrap_socket(
+ socket, server_hostname=server_hostname, server_side=server_side, **kwargs
+ )
# this always happens in tests, so cancel any tasks that are
# still pending by the time we get here.
asyncio_loop = self.io_loop.asyncio_loop # type: ignore
- if hasattr(asyncio, "all_tasks"): # py37
- tasks = asyncio.all_tasks(asyncio_loop) # type: ignore
- else:
- tasks = asyncio.Task.all_tasks(asyncio_loop)
+ tasks = asyncio.all_tasks(asyncio_loop)
# Tasks that are done may still appear here and may contain
# non-cancellation exceptions, so filter them out.
tasks = [t for t in tasks if not t.done()] # type: ignore
# so we have to typecheck both.
mypy --platform linux {posargs:tornado}
mypy --platform windows {posargs:tornado}
+ # We mainly lint on the oldest version of Python we support, since
+ # we're more likely to catch problems of accidentally depending on
+ # something new than of depending on something old and deprecated.
+ # But sometimes something we depend on gets removed so we should also
+ # test the newest version.
+ mypy --platform linux --python-version 3.12 {posargs:tornado}
changedir = {toxinidir}