SSL Support using rustls

Cargo.toml

@@ -31,9 +31,14 @@ crate-type = ["cdylib", "rlib"]
 
 [dependencies]
 anyhow = "=1.0"
+env_logger = "0.11"
+log = "0.4"
 mio = { version = "=1.0", features = ["net", "os-ext", "os-poll"] }
 papaya = "=0.2"
 pyo3 = { version = "=0.26", features = ["anyhow", "extension-module", "generate-import-lib"] }
+rcgen = "=0.13"
+rustls = { version = "0.23", features = ["std"] }
+rustls-pemfile = "2.0"
 socket2 = { version = "=0.6", features = ["all"] }
 
 [target.'cfg(unix)'.dependencies]

pyproject.toml

@@ -43,6 +43,8 @@ lint = [
 test = [
     'pytest~=8.3',
     'pytest-asyncio~=0.26',
+    'pytest-timeout~=2.4',
+    'requests~=2.32',
 ]
 
 all = [
@@ -105,5 +107,24 @@ known-first-party = ['rloop', 'tests']
 [tool.pytest.ini_options]
 asyncio_mode = 'auto'
 
+log_cli = true
+log_cli_level = "INFO"
+
+doctest_optionflags = [
+    "NORMALIZE_WHITESPACE",
+    "ALLOW_UNICODE",
+    "ALLOW_BYTES",
+    "NUMBER",
+]
+addopts = [
+    "--continue-on-collection-errors",
+    "--ignore-glob=*_BACKUP_*",
+    "-v",
+    "-rxX",
+    "--doctest-modules",
+    "--doctest-ignore-import-errors",
+]
+
+
 [tool.uv]
 package = false

rloop/loop.py

@@ -1,5 +1,6 @@
 import asyncio as __asyncio
 import errno
+import logging
 import os
 import signal
 import socket
@@ -32,6 +33,9 @@
 from .utils import _can_use_pidfd, _HAS_IPv6, _interleave_addrinfos, _ipaddr_info, _noop, _set_reuseport
 
 
+logger = logging.getLogger(__name__)
+
+
 class RLoop(__BaseLoop, __asyncio.AbstractEventLoop):
     def __init__(self):
         super().__init__()
@@ -316,10 +320,6 @@ async def create_connection(
         interleave=None,
         all_errors=False,
     ):
-        # TODO
-        if ssl:
-            raise NotImplementedError
-
         if server_hostname is not None and not ssl:
             raise ValueError('server_hostname is only meaningful with ssl')
 
@@ -385,7 +385,7 @@ async def create_connection(
                         happy_eyeballs_delay,
                         loop=self,
                     )
-                )[0]  # can't use sock, _, _ as it keeks a reference to exceptions
+                )[0]  # can't use sock, _, _ as it keeps a reference to exceptions
 
             if sock is None:
                 exceptions = [exc for sub in exceptions for exc in sub]
@@ -421,16 +421,8 @@ async def create_connection(
         rsock = (sock.fileno(), sock.family)
         sock.detach()
 
-        # TODO: ssl
-        transport, protocol = self._tcp_conn(rsock, protocol_factory)
-        # transport, protocol = await self._create_connection_transport(
-        #     sock,
-        #     protocol_factory,
-        #     ssl,
-        #     server_hostname,
-        #     ssl_handshake_timeout=ssl_handshake_timeout,
-        #     ssl_shutdown_timeout=ssl_shutdown_timeout,
-        # )
+        logger.debug('Creating %s connection', 'SSL' if ssl else 'TCP')
+        transport, protocol = self._tcp_conn(rsock, protocol_factory, ssl, server_hostname if ssl else None)
 
         return transport, protocol
 
@@ -491,10 +483,6 @@ async def create_server(
         ssl_shutdown_timeout=None,
         start_serving=True,
     ):
-        # TODO
-        if ssl:
-            raise NotImplementedError
-
         if isinstance(ssl, bool):
             raise TypeError('ssl argument must be an SSLContext or None')
 
@@ -580,11 +568,12 @@ async def create_server(
             rsocks.append((sock.fileno(), sock.family))
             sock.detach()
 
-        # TODO: ssl
-        # server = self._tcp_server(sockets, rsocks, protocol_factory, backlog,
-        #                 ssl, ssl_handshake_timeout,
-        #                 ssl_shutdown_timeout)
-        server = Server(self._tcp_server(sockets, rsocks, protocol_factory, backlog))
+        if ssl:
+            logger.debug('Creating SSL server')
+            server = Server(self._tcp_server_ssl(sockets, rsocks, protocol_factory, backlog, ssl))
+        else:
+            logger.debug('Creating TCP server')
+            server = Server(self._tcp_server(sockets, rsocks, protocol_factory, backlog))
 
         if start_serving:
             await server.start_serving()

src/event_loop.rs

@@ -346,6 +346,7 @@ impl EventLoop {
                         let guard_poll = self.io.lock().unwrap();
                         _ = guard_poll.registry().reregister(&mut source, token, interests);
                     }
+                    self.wake(); // interest changed
                     return IOHandle::TCPStream(interests);
                 }
                 unreachable!()
@@ -357,6 +358,7 @@ impl EventLoop {
                     let guard_poll = self.io.lock().unwrap();
                     _ = guard_poll.registry().register(&mut source, token, interest);
                 }
+                self.wake(); // interest registered
                 IOHandle::TCPStream(interest)
             },
         );
@@ -396,16 +398,20 @@ impl EventLoop {
 
     #[inline(always)]
     pub(crate) fn tcp_stream_close(&self, py: Python, fd: usize) {
-        if let Some(transport) = self.tcp_transports.pin().remove(&fd)
-            && let Some(lfd) = transport.borrow(py).lfd
-        {
-            self.tcp_lstreams.pin().get(&lfd).map(|v| v.pin().remove(&fd));
+        if let Some(transport) = self.tcp_transports.pin().remove(&fd) {
+            // ensure TCPTransport::close() called, as it sends TLS close
+            if transport.borrow(py).is_tls() {
+                transport.borrow(py).close(py);
+            }
+            if let Some(lfd) = transport.borrow(py).lfd {
+                self.tcp_lstreams.pin().get(&lfd).map(|v| v.pin().remove(&fd));
+            }
         }
     }
 
     #[inline(always)]
-    pub(crate) fn get_tcp_transport(&self, fd: usize, py: Python) -> Py<TCPTransport> {
-        self.tcp_transports.pin().get(&fd).unwrap().clone_ref(py)
+    pub(crate) fn get_tcp_transport(&self, fd: usize, py: Python) -> Option<Py<TCPTransport>> {
+        self.tcp_transports.pin().get(&fd).map(|t| t.clone_ref(py))
     }
 
     pub(crate) fn with_tcp_listener_streams<T>(&self, fd: usize, func: T)
@@ -1171,10 +1177,19 @@ impl EventLoop {
         py: Python,
         sock: (i32, i32),
         protocol_factory: Py<PyAny>,
+        ssl_context: Option<Py<PyAny>>,
+        server_hostname: Option<String>,
     ) -> PyResult<(Py<TCPTransport>, Py<PyAny>)> {
         let rself = pyself.get();
         let transport = TCPTransport::from_py(py, &pyself, sock, protocol_factory);
         let fd = transport.fd;
+
+        if let (Some(ssl_context), Some(server_hostname)) = (ssl_context, server_hostname) {
+            // Initialize TLS client connection
+            let ssl_config = crate::ssl::create_ssl_client_config_from_context(&ssl_context.bind(py))?;
+            transport.initialize_tls_client(ssl_config, server_hostname);
+        }
+
         let pytransport = Py::new(py, transport)?;
         let proto = TCPTransport::attach(&pytransport, py)?;
         rself.tcp_transports.pin().insert(fd, pytransport.clone_ref(py));
@@ -1198,6 +1213,30 @@ impl EventLoop {
         Py::new(py, server)
     }
 
+    fn _tcp_server_ssl(
+        pyself: Py<Self>,
+        py: Python,
+        socks: Py<PyAny>,
+        rsocks: Vec<(i32, i32)>,
+        protocol_factory: Py<PyAny>,
+        backlog: i32,
+        ssl_context: Py<PyAny>,
+    ) -> PyResult<Py<Server>> {
+        let ssl_config = crate::ssl::create_ssl_config_from_context(&ssl_context.bind(py))?;
+        let mut servers = Vec::new();
+        for (fd, family) in rsocks {
+            servers.push(TCPServer::from_fd_ssl(
+                fd,
+                family,
+                backlog,
+                protocol_factory.clone_ref(py),
+                ssl_config.clone(),
+            ));
+        }
+        let server = Server::tcp(pyself.clone_ref(py), socks, servers);
+        Py::new(py, server)
+    }
+
     fn _tcp_stream_bound(&self, fd: usize) -> bool {
         self.tcp_transports.pin().contains_key(&fd)
     }

src/lib.rs

@@ -8,6 +8,7 @@ mod log;
 mod py;
 mod server;
 mod sock;
+mod ssl;
 mod tcp;
 mod time;
 mod udp;
@@ -24,11 +25,15 @@ pub(crate) fn get_lib_version() -> &'static str {
 
 #[pymodule(gil_used = false)]
 fn _rloop(_py: Python, module: &Bound<PyModule>) -> PyResult<()> {
+    // Configure logs via RUST_LOG= (default to INFO)
+    env_logger::init();
+
     module.add("__version__", get_lib_version())?;
 
     event_loop::init_pymodule(module)?;
     handles::init_pymodule(module)?;
     server::init_pymodule(module)?;
+    ssl::init_pymodule(module)?;
 
     Ok(())
 }