From f6ba2b68bffdd30590ad2eea7bf245b38c2e104b Mon Sep 17 00:00:00 2001 From: HTHou Date: Thu, 16 Jul 2026 11:36:27 +0800 Subject: [PATCH] THRIFT-6097: Add rustls client and server support Client: rs Co-Authored-By: OpenAI Codex (GPT-5) --- .github/workflows/build.yml | 6 +- doc/thrift-threat-model.md | 13 +- lib/rs/Cargo.toml | 3 + lib/rs/Makefile.am | 8 +- lib/rs/README.md | 23 ++- lib/rs/src/server/threaded.rs | 40 +++++ lib/rs/src/transport/mod.rs | 6 + lib/rs/src/transport/shared.rs | 88 ++++++++++ lib/rs/src/transport/tls.rs | 301 +++++++++++++++++++++++++++++++++ lib/rs/tests/shared_channel.rs | 98 +++++++++++ lib/rs/tests/tls.rs | 170 +++++++++++++++++++ test/rs/Cargo.toml | 3 + test/rs/src/bin/test_client.rs | 136 ++++++++++++++- test/rs/src/bin/test_server.rs | 41 ++++- test/tests.json | 1 + 15 files changed, 920 insertions(+), 17 deletions(-) create mode 100644 lib/rs/src/transport/shared.rs create mode 100644 lib/rs/src/transport/tls.rs create mode 100644 lib/rs/tests/shared_channel.rs create mode 100644 lib/rs/tests/tls.rs diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 8867d088fa1..871092765fb 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -616,8 +616,10 @@ jobs: - name: Run make for rust run: make -C lib/rs - - name: Run make check for rust - run: make -C lib/rs check + - name: Run make checks for rust + run: | + make -C lib/rs check + make -C lib/rs check-rustls - name: Run make test for rust run: make -C lib/rs/test check diff --git a/doc/thrift-threat-model.md b/doc/thrift-threat-model.md index add9e9712ab..9a4809579bf 100644 --- a/doc/thrift-threat-model.md +++ b/doc/thrift-threat-model.md @@ -1308,10 +1308,13 @@ The claim is correct for Go: validation is on by default, and the system store i InsecureSkipVerify: true in their *tls.Config, which is outside Thrift's control. --- -Rust — No TLS support +Rust — Validation is controlled by the supplied rustls configuration -The Rust library (lib/rs/src/transport/socket.rs) implements only plain TCP via TcpStream. There is no TSSLSocket type, no TLS transport, and no SSL dependency. The claim is -inapplicable. +When the optional `rustls` feature is enabled, `TTlsClientChannel` requires the caller to supply an `Arc` and a `ServerName`. `connect()` completes the TLS +handshake according to that configuration before returning. Chain validation and server-name verification are controlled by the configuration's verifier. `TTlsServerChannel` +and `TServer::listen_tls()` similarly require an `Arc`; client-certificate authentication is whatever that configuration selects. Thrift does not construct a +default configuration or automatically load system roots, so there is no implicit trust-store fallback. Applications may also supply rustls custom verifiers; their behavior is +outside Thrift's control. --- Summary table @@ -1333,7 +1336,7 @@ Summary table │ Go │ crypto/tls enforces (default │ System root CA pool if RootCAs=nil │ Yes (system pool) │ Caller can bypass by setting InsecureSkipVerify: │ │ │ InsecureSkipVerify=false) │ │ │ true │ ├─────────┼─────────────────────────────────────────┼───────────────────────────────────────────┼────────────────────┼─────────────────────────────────────────────────────┤ -│ Rust │ N/A │ N/A │ N/A │ No TLS transport exists │ +│ Rust │ Caller-configured rustls verifier │ Caller-supplied ClientConfig │ No │ No default config or implicit trust-store loading │ └─────────┴─────────────────────────────────────────┴───────────────────────────────────────────┴────────────────────┴─────────────────────────────────────────────────────┘ ``` @@ -1503,7 +1506,7 @@ not recommended. THRIFT-5926 (crash on None initial DIGEST-MD5 response) is a co mechanism removal. If the crash is reachable pre-authentication, it qualifies as a remotely-triggerable DoS and will be treated accordingly. **Q42.** -Thrift sets no project-wide TLS version floor. Each binding delegates version negotiation to its underlying TLS library (OpenSSL, JSSE, crypto/tls, etc.). +Thrift sets no project-wide TLS version floor. Each binding delegates version negotiation to its underlying TLS library (OpenSSL, JSSE, crypto/tls, rustls, etc.). THRIFT-5743/5876 added TLS 1.3 capability where it was absent. Operators are responsible for configuring a version floor in the underlying library; TLS 1.2 minimum is recommended. A finding that TLS 1.0 or 1.1 is negotiable is a deployment or library-configuration concern, not a Thrift vulnerability. (Note: verify whether the C++ binding explicitly disables SSLv2/v3 via SSL_OP_NO_SSLv2 | SSL_OP_NO_SSLv3; if so, state that floor explicitly for the C++ binding.) diff --git a/lib/rs/Cargo.toml b/lib/rs/Cargo.toml index e6fbe51d692..460ab693a18 100644 --- a/lib/rs/Cargo.toml +++ b/lib/rs/Cargo.toml @@ -19,10 +19,13 @@ uuid = "1" log = {version = "0.4", optional = true} ordered-float = "3.0" threadpool = {version = "1.7", optional = true} +rustls = { version = "0.23.42", default-features = false, features = ["std", "tls12"], optional = true } [features] default = ["server"] server = ["threadpool", "log"] +rustls = ["dep:rustls"] [dev-dependencies] uuid = { version = "1", features = ["v4"] } +rustls = { version = "0.23.42", default-features = false, features = ["ring", "std", "tls12"] } diff --git a/lib/rs/Makefile.am b/lib/rs/Makefile.am index 71e6335496a..ce392c22e26 100644 --- a/lib/rs/Makefile.am +++ b/lib/rs/Makefile.am @@ -41,6 +41,12 @@ all-local: $(CARGO) clippy --all -- -D warnings $(CARGO) build +.PHONY: check-rustls +check-rustls: + $(CARGO) clippy --all --all-features -- -D warnings + $(CARGO) check --no-default-features --features rustls + $(CARGO) test --all-features + clean-local: $(CARGO) clean -$(RM) Cargo.lock @@ -50,6 +56,7 @@ distdir: EXTRA_DIST = \ src \ + tests \ Cargo.toml \ README.md \ release.sh \ @@ -57,4 +64,3 @@ EXTRA_DIST = \ NOTICE \ LICENSE \ RELEASING.md - diff --git a/lib/rs/README.md b/lib/rs/README.md index 51aa63e620f..211b6eb8ecf 100644 --- a/lib/rs/README.md +++ b/lib/rs/README.md @@ -33,14 +33,33 @@ types and services by writing their own code on top. Add `thrift = "x.y.z"` to your `Cargo.toml`, where `x.y.z` is the version of the Thrift compiler you're using. +### TLS + +TLS client and server channels are available through the optional `rustls` +feature: + +```toml +[dependencies] +thrift = { version = "x.y.z", features = ["rustls"] } +rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] } +``` + +Applications construct a rustls `ClientConfig` or `ServerConfig` and pass it to +`TTlsClientChannel::connect` or `TServer::listen_tls`. Thrift does not choose a +crypto provider, load a system trust store, or read certificate files. This +keeps trust anchors, client authentication, protocol versions, and certificate +selection under application control. + ## API Documentation Full [Rustdoc](https://docs.rs/thrift/) ## Compatibility -The Rust library and auto-generated code targets Rust versions 1.28+. -It does not currently use any Rust 2021 features. +The Rust library and auto-generated code target Rust versions 1.65+. The +minimum was raised from Rust 1.28 by THRIFT-5158, when the library and generator +moved to Rust 2021-only output. Enabling the optional `rustls` feature requires +Rust 1.71+. ### Breaking Changes diff --git a/lib/rs/src/server/threaded.rs b/lib/rs/src/server/threaded.rs index ee863952b71..66c72878352 100644 --- a/lib/rs/src/server/threaded.rs +++ b/lib/rs/src/server/threaded.rs @@ -21,6 +21,9 @@ use std::net::{TcpListener, ToSocketAddrs}; use std::sync::Arc; use threadpool::ThreadPool; +#[cfg(feature = "rustls")] +use rustls::ServerConfig; + #[cfg(unix)] use std::os::unix::net::UnixListener; #[cfg(unix)] @@ -29,6 +32,8 @@ use std::path::Path; use crate::protocol::{ TInputProtocol, TInputProtocolFactory, TOutputProtocol, TOutputProtocolFactory, }; +#[cfg(feature = "rustls")] +use crate::transport::TTlsServerChannel; use crate::transport::{TIoChannel, TReadTransportFactory, TTcpChannel, TWriteTransportFactory}; use crate::{ApplicationError, ApplicationErrorKind}; @@ -199,6 +204,41 @@ where })) } + /// Listen for incoming TLS connections on `listen_address`. + /// + /// `config` controls certificate selection, client authentication, crypto + /// provider, and protocol policy. Accepted connections are handed to a + /// worker before the TLS handshake begins, so the accept loop does not + /// block on a peer that connects without completing a handshake. + #[cfg(feature = "rustls")] + pub fn listen_tls( + &mut self, + listen_address: A, + config: Arc, + ) -> crate::Result<()> { + let listener = TcpListener::bind(listen_address)?; + for stream in listener.incoming() { + match stream { + Ok(stream) => { + stream.set_nodelay(true).ok(); + let channel = TTlsServerChannel::with_stream(stream, Arc::clone(&config))?; + self.handle_stream(channel)?; + } + Err(error) => { + warn!( + "failed to accept remote TLS connection with error {:?}", + error + ); + } + } + } + + Err(crate::Error::Application(ApplicationError { + kind: ApplicationErrorKind::Unknown, + message: "aborted TLS listen loop".into(), + })) + } + /// Listen for incoming connections on `listen_path`. /// /// `listen_path` should implement `AsRef` trait. diff --git a/lib/rs/src/transport/mod.rs b/lib/rs/src/transport/mod.rs index 2b5733f490b..9e43ea6ec54 100644 --- a/lib/rs/src/transport/mod.rs +++ b/lib/rs/src/transport/mod.rs @@ -44,7 +44,10 @@ macro_rules! assert_eq_transport_written_bytes { mod buffered; mod framed; mod mem; +mod shared; mod socket; +#[cfg(feature = "rustls")] +mod tls; pub use self::buffered::{ TBufferedReadTransport, TBufferedReadTransportFactory, TBufferedWriteTransport, @@ -55,7 +58,10 @@ pub use self::framed::{ TFramedWriteTransportFactory, }; pub use self::mem::TBufferChannel; +pub use self::shared::TSharedChannel; pub use self::socket::TTcpChannel; +#[cfg(feature = "rustls")] +pub use self::tls::{TTlsClientChannel, TTlsServerChannel}; /// Identifies a transport used by a `TInputProtocol` to receive bytes. pub trait TReadTransport: Read {} diff --git a/lib/rs/src/transport/shared.rs b/lib/rs/src/transport/shared.rs new file mode 100644 index 00000000000..e14ed57d360 --- /dev/null +++ b/lib/rs/src/transport/shared.rs @@ -0,0 +1,88 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::io::{self, Read, Write}; +use std::sync::{Arc, Mutex, MutexGuard}; + +use super::{ReadHalf, TIoChannel, WriteHalf}; + +/// A cloneable channel that serializes access to an underlying I/O stream. +/// +/// This adapter allows a bidirectional stream that cannot be cloned, such as a +/// TLS session, to implement [`TIoChannel`]. Every read, write, and flush holds +/// the same lock for the duration of that operation. It is intended for +/// synchronous request-response traffic, where a caller writes and flushes a +/// complete request before reading its response. +/// +/// The shared lock makes access memory-safe across threads, but it does not +/// provide full-duplex progress: a blocking read holds the lock and prevents a +/// concurrent write until that read finishes. +#[derive(Debug)] +pub struct TSharedChannel { + inner: Arc>, +} + +impl TSharedChannel { + /// Wrap `inner` in a shared channel. + pub fn new(inner: C) -> Self { + Self { + inner: Arc::new(Mutex::new(inner)), + } + } + + // io::Error::other requires Rust 1.74. + #[allow(unknown_lints)] + #[allow(clippy::io_other_error)] + pub(crate) fn lock(&self) -> io::Result> { + self.inner + .lock() + .map_err(|_| io::Error::new(io::ErrorKind::Other, "shared channel lock is poisoned")) + } +} + +impl Clone for TSharedChannel { + fn clone(&self) -> Self { + Self { + inner: Arc::clone(&self.inner), + } + } +} + +impl Read for TSharedChannel { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + self.lock()?.read(buffer) + } +} + +impl Write for TSharedChannel { + fn write(&mut self, buffer: &[u8]) -> io::Result { + self.lock()?.write(buffer) + } + + fn flush(&mut self) -> io::Result<()> { + self.lock()?.flush() + } +} + +impl TIoChannel for TSharedChannel { + fn split(self) -> crate::Result<(ReadHalf, WriteHalf)> + where + Self: Sized, + { + Ok((ReadHalf::new(self.clone()), WriteHalf::new(self))) + } +} diff --git a/lib/rs/src/transport/tls.rs b/lib/rs/src/transport/tls.rs new file mode 100644 index 00000000000..4f6d3417f24 --- /dev/null +++ b/lib/rs/src/transport/tls.rs @@ -0,0 +1,301 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::io::{self, Read, Write}; +use std::net::{Shutdown, TcpStream, ToSocketAddrs}; +use std::ops::{Deref, DerefMut}; +use std::sync::Arc; +use std::time::Duration; + +use rustls::pki_types::ServerName; +use rustls::{ + ClientConfig, ClientConnection, ConnectionCommon, ServerConfig, ServerConnection, SideData, + StreamOwned, +}; + +use super::{ReadHalf, TIoChannel, TSharedChannel, WriteHalf}; +use crate::{new_transport_error, TransportErrorKind}; + +/// A blocking rustls client channel. +/// +/// The caller supplies a fully configured rustls [`ClientConfig`], including +/// its crypto provider, server-certificate verifier, optional client identity, +/// and protocol policy. The TLS handshake completes according to that +/// configuration before [`connect`](Self::connect) returns. +#[derive(Clone, Debug)] +pub struct TTlsClientChannel { + inner: TSharedChannel>, +} + +impl TTlsClientChannel { + /// Connect to `remote_address` and complete a TLS handshake for + /// `server_name`. + pub fn connect( + remote_address: A, + server_name: ServerName<'static>, + config: Arc, + ) -> crate::Result { + let stream = TcpStream::connect(remote_address)?; + stream.set_nodelay(true)?; + Self::with_stream(stream, server_name, config) + } + + /// Wrap an already-connected TCP stream and complete a TLS handshake. + pub fn with_stream( + mut stream: TcpStream, + server_name: ServerName<'static>, + config: Arc, + ) -> crate::Result { + let mut connection = ClientConnection::new(config, server_name).map_err(|error| { + new_transport_error( + TransportErrorKind::Unknown, + format!("cannot create TLS client connection: {error}"), + ) + })?; + connection + .complete_io(&mut stream) + .map_err(|error| handshake_io_error("client", error))?; + if connection.is_handshaking() { + return Err(incomplete_handshake_error("client")); + } + + Ok(Self { + inner: TSharedChannel::new(StreamOwned::new(connection, stream)), + }) + } + + /// Return the read timeout of the underlying TCP stream. + pub fn read_timeout(&self) -> crate::Result> { + read_timeout(&self.inner) + } + + /// Return the write timeout of the underlying TCP stream. + pub fn write_timeout(&self) -> crate::Result> { + write_timeout(&self.inner) + } + + /// Set the read timeout of the underlying TCP stream. + pub fn set_read_timeout(&mut self, timeout: Option) -> crate::Result<()> { + set_read_timeout(&self.inner, timeout) + } + + /// Set the write timeout of the underlying TCP stream. + pub fn set_write_timeout(&mut self, timeout: Option) -> crate::Result<()> { + set_write_timeout(&self.inner, timeout) + } + + /// Set the read and write timeouts of the underlying TCP stream. + pub fn set_timeouts( + &mut self, + read_timeout: Option, + write_timeout: Option, + ) -> crate::Result<()> { + self.set_read_timeout(read_timeout)?; + self.set_write_timeout(write_timeout) + } + + /// Send a TLS close notification and shut down the underlying TCP stream. + pub fn close(&mut self) -> crate::Result<()> { + close(&self.inner) + } +} + +impl Read for TTlsClientChannel { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + self.inner.read(buffer) + } +} + +impl Write for TTlsClientChannel { + fn write(&mut self, buffer: &[u8]) -> io::Result { + self.inner.write(buffer) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +} + +impl TIoChannel for TTlsClientChannel { + fn split(self) -> crate::Result<(ReadHalf, WriteHalf)> + where + Self: Sized, + { + Ok((ReadHalf::new(self.clone()), WriteHalf::new(self))) + } +} + +/// A blocking rustls server channel. +/// +/// Construction does not perform socket I/O. The TLS handshake is completed +/// lazily by the first read or write, allowing a server accept loop to enqueue +/// the connection without waiting on an untrusted peer. +#[derive(Clone, Debug)] +pub struct TTlsServerChannel { + inner: TSharedChannel>, +} + +impl TTlsServerChannel { + /// Wrap an already-connected TCP stream without blocking for a handshake. + pub fn with_stream(stream: TcpStream, config: Arc) -> crate::Result { + let connection = ServerConnection::new(config).map_err(|error| { + new_transport_error( + TransportErrorKind::Unknown, + format!("cannot create TLS server connection: {error}"), + ) + })?; + + Ok(Self { + inner: TSharedChannel::new(StreamOwned::new(connection, stream)), + }) + } + + /// Complete the TLS handshake immediately. + /// + /// If the handshake is already complete this returns without performing + /// socket I/O. If the stream cannot make progress, the connection state is + /// retained so the caller can retry. + pub fn handshake(&mut self) -> crate::Result<()> { + let mut stream = self.inner.lock()?; + let StreamOwned { conn, sock } = &mut *stream; + if !conn.is_handshaking() { + return Ok(()); + } + conn.complete_io(sock) + .map_err(|error| handshake_io_error("server", error))?; + if conn.is_handshaking() { + return Err(incomplete_handshake_error("server")); + } + Ok(()) + } + + /// Return the read timeout of the underlying TCP stream. + pub fn read_timeout(&self) -> crate::Result> { + read_timeout(&self.inner) + } + + /// Return the write timeout of the underlying TCP stream. + pub fn write_timeout(&self) -> crate::Result> { + write_timeout(&self.inner) + } + + /// Set the read timeout of the underlying TCP stream. + pub fn set_read_timeout(&mut self, timeout: Option) -> crate::Result<()> { + set_read_timeout(&self.inner, timeout) + } + + /// Set the write timeout of the underlying TCP stream. + pub fn set_write_timeout(&mut self, timeout: Option) -> crate::Result<()> { + set_write_timeout(&self.inner, timeout) + } + + /// Set the read and write timeouts of the underlying TCP stream. + pub fn set_timeouts( + &mut self, + read_timeout: Option, + write_timeout: Option, + ) -> crate::Result<()> { + self.set_read_timeout(read_timeout)?; + self.set_write_timeout(write_timeout) + } + + /// Send a TLS close notification and shut down the underlying TCP stream. + pub fn close(&mut self) -> crate::Result<()> { + close(&self.inner) + } +} + +impl Read for TTlsServerChannel { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + self.inner.read(buffer) + } +} + +impl Write for TTlsServerChannel { + fn write(&mut self, buffer: &[u8]) -> io::Result { + self.inner.write(buffer) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +} + +impl TIoChannel for TTlsServerChannel { + fn split(self) -> crate::Result<(ReadHalf, WriteHalf)> + where + Self: Sized, + { + Ok((ReadHalf::new(self.clone()), WriteHalf::new(self))) + } +} + +fn incomplete_handshake_error(role: &str) -> crate::Error { + io::Error::new( + io::ErrorKind::WouldBlock, + format!("TLS {role} handshake did not complete"), + ) + .into() +} + +fn handshake_io_error(role: &str, error: io::Error) -> crate::Error { + if error.kind() == io::ErrorKind::WouldBlock { + incomplete_handshake_error(role) + } else { + error.into() + } +} + +fn read_timeout( + channel: &TSharedChannel>, +) -> crate::Result> { + Ok(channel.lock()?.sock.read_timeout()?) +} + +fn write_timeout( + channel: &TSharedChannel>, +) -> crate::Result> { + Ok(channel.lock()?.sock.write_timeout()?) +} + +fn set_read_timeout( + channel: &TSharedChannel>, + timeout: Option, +) -> crate::Result<()> { + channel.lock()?.sock.set_read_timeout(timeout)?; + Ok(()) +} + +fn set_write_timeout( + channel: &TSharedChannel>, + timeout: Option, +) -> crate::Result<()> { + channel.lock()?.sock.set_write_timeout(timeout)?; + Ok(()) +} + +fn close(channel: &TSharedChannel>) -> crate::Result<()> +where + C: Deref> + DerefMut, + S: SideData, +{ + let mut stream = channel.lock()?; + stream.conn.send_close_notify(); + stream.flush()?; + stream.sock.shutdown(Shutdown::Both)?; + Ok(()) +} diff --git a/lib/rs/tests/shared_channel.rs b/lib/rs/tests/shared_channel.rs new file mode 100644 index 00000000000..80651adac61 --- /dev/null +++ b/lib/rs/tests/shared_channel.rs @@ -0,0 +1,98 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::VecDeque; +use std::io::{self, Read, Write}; +use std::panic::{catch_unwind, AssertUnwindSafe}; +use std::sync::{Arc, Mutex}; + +use thrift::transport::{TIoChannel, TSharedChannel}; + +struct TestIo { + readable: VecDeque, + written: Arc>>, +} + +impl Read for TestIo { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + let count = buffer.len().min(self.readable.len()); + for target in &mut buffer[..count] { + *target = self.readable.pop_front().unwrap(); + } + Ok(count) + } +} + +impl Write for TestIo { + fn write(&mut self, buffer: &[u8]) -> io::Result { + self.written.lock().unwrap().extend_from_slice(buffer); + Ok(buffer.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +#[test] +fn split_supports_a_non_cloneable_inner_channel() { + let written = Arc::new(Mutex::new(Vec::new())); + let channel = TSharedChannel::new(TestIo { + readable: VecDeque::from(b"read".to_vec()), + written: Arc::clone(&written), + }); + + let (mut reader, mut writer) = channel.split().unwrap(); + let mut input = [0; 4]; + reader.read_exact(&mut input).unwrap(); + writer.write_all(b"write").unwrap(); + + assert_eq!(&input, b"read"); + assert_eq!(&*written.lock().unwrap(), b"write"); +} + +struct PanickingIo; + +impl Read for PanickingIo { + fn read(&mut self, _buffer: &mut [u8]) -> io::Result { + panic!("poison the shared channel lock") + } +} + +impl Write for PanickingIo { + fn write(&mut self, buffer: &[u8]) -> io::Result { + Ok(buffer.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +#[test] +fn a_poisoned_channel_returns_an_io_error_instead_of_panicking_again() { + let mut channel = TSharedChannel::new(PanickingIo); + let mut reader = channel.clone(); + let panic = catch_unwind(AssertUnwindSafe(|| { + let _ = reader.read(&mut [0]); + })); + assert!(panic.is_err()); + + let error = channel.write(b"still poisoned").unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::Other); + assert_eq!(error.to_string(), "shared channel lock is poisoned"); +} diff --git a/lib/rs/tests/tls.rs b/lib/rs/tests/tls.rs new file mode 100644 index 00000000000..723e8f10577 --- /dev/null +++ b/lib/rs/tests/tls.rs @@ -0,0 +1,170 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#![cfg(feature = "rustls")] + +use std::io::{Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::sync::{mpsc, Arc}; +use std::thread; +use std::time::Duration; + +use rustls::pki_types::ServerName; +use rustls::server::ResolvesServerCertUsingSni; +use rustls::{ClientConfig, RootCertStore, ServerConfig}; +use thrift::transport::{TTlsClientChannel, TTlsServerChannel}; + +fn server_config() -> Arc { + Arc::new( + ServerConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())) + .with_safe_default_protocol_versions() + .unwrap() + .with_no_client_auth() + .with_cert_resolver(Arc::new(ResolvesServerCertUsingSni::new())), + ) +} + +fn client_config() -> Arc { + Arc::new( + ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())) + .with_safe_default_protocol_versions() + .unwrap() + .with_root_certificates(RootCertStore::empty()) + .with_no_client_auth(), + ) +} + +#[test] +fn client_with_nonblocking_stream_rejects_an_incomplete_handshake() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let (accepted_tx, accepted_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let server = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + stream.write_all(&[0x16]).unwrap(); + stream.flush().unwrap(); + accepted_tx.send(()).unwrap(); + release_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + }); + + let stream = TcpStream::connect(address).unwrap(); + accepted_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + let mut prefix = [0]; + assert_eq!(stream.peek(&mut prefix).unwrap(), 1); + assert_eq!(prefix, [0x16]); + stream.set_nonblocking(true).unwrap(); + let result = TTlsClientChannel::with_stream( + stream, + ServerName::try_from("localhost").unwrap(), + client_config(), + ); + + release_tx.send(()).unwrap(); + server.join().unwrap(); + assert_incomplete_handshake_error(result, "client"); +} + +#[test] +fn client_reports_an_incomplete_nonblocking_handshake_without_progress() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let (accepted_tx, accepted_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let server = thread::spawn(move || { + let (_stream, _) = listener.accept().unwrap(); + accepted_tx.send(()).unwrap(); + release_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + }); + + let stream = TcpStream::connect(address).unwrap(); + accepted_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + stream.set_nonblocking(true).unwrap(); + let result = TTlsClientChannel::with_stream( + stream, + ServerName::try_from("localhost").unwrap(), + client_config(), + ); + + release_tx.send(()).unwrap(); + server.join().unwrap(); + assert_incomplete_handshake_error(result, "client"); +} + +#[test] +fn server_handshake_rejects_an_incomplete_nonblocking_handshake() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let mut client = TcpStream::connect(address).unwrap(); + let (stream, _) = listener.accept().unwrap(); + client.write_all(&[0x16]).unwrap(); + client.flush().unwrap(); + let mut prefix = [0]; + assert_eq!(stream.peek(&mut prefix).unwrap(), 1); + assert_eq!(prefix, [0x16]); + stream.set_nonblocking(true).unwrap(); + + let mut channel = TTlsServerChannel::with_stream(stream, server_config()).unwrap(); + assert_incomplete_handshake_error(channel.handshake(), "server"); +} + +#[test] +fn server_reports_an_incomplete_nonblocking_handshake_without_progress() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let _client = TcpStream::connect(address).unwrap(); + let (stream, _) = listener.accept().unwrap(); + stream.set_nonblocking(true).unwrap(); + + let mut channel = TTlsServerChannel::with_stream(stream, server_config()).unwrap(); + assert_incomplete_handshake_error(channel.handshake(), "server"); +} + +#[test] +fn server_wraps_stream_without_waiting_for_a_handshake() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let (wrapped_tx, wrapped_rx) = mpsc::channel(); + let server = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut channel = TTlsServerChannel::with_stream(stream, server_config()).unwrap(); + wrapped_tx.send(()).unwrap(); + let mut byte = [0]; + assert!(channel.read(&mut byte).is_err()); + }); + + let probe = TcpStream::connect(address).unwrap(); + wrapped_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + drop(probe); + server.join().unwrap(); +} + +fn assert_incomplete_handshake_error(result: thrift::Result, role: &str) { + let error = match result { + Ok(_) => panic!("incomplete TLS handshake succeeded"), + Err(error) => error, + }; + match error { + thrift::Error::Transport(error) => { + assert_eq!( + error.message, + format!("TLS {role} handshake did not complete") + ); + } + error => panic!("unexpected error: {error:?}"), + } +} diff --git a/test/rs/Cargo.toml b/test/rs/Cargo.toml index da53b967a0b..c2827526c21 100644 --- a/test/rs/Cargo.toml +++ b/test/rs/Cargo.toml @@ -11,7 +11,10 @@ clap = "~2.33" bitflags = "=1.2" env_logger = "0.8" log = "0.4" +rustls = { version = "0.23.42", default-features = false, features = ["ring", "std", "tls12"] } +rustls-pemfile = "2.2" uuid = "1" [dependencies.thrift] path = "../../lib/rs" +features = ["rustls"] diff --git a/test/rs/src/bin/test_client.rs b/test/rs/src/bin/test_client.rs index 1e4d6bd650d..b328a4a3fde 100644 --- a/test/rs/src/bin/test_client.rs +++ b/test/rs/src/bin/test_client.rs @@ -17,10 +17,16 @@ use clap::{clap_app, value_t}; use log::*; +use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier}; +use rustls::pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime}; +use rustls::sign::{CertifiedKey, SingleCertAndKey}; +use rustls::{ClientConfig, DigitallySignedStruct, Error as RustlsError, SignatureScheme}; use std::collections::{BTreeMap, BTreeSet}; use std::fmt::Debug; +use std::io::Cursor; use std::net::{TcpStream, ToSocketAddrs}; +use std::sync::Arc; #[cfg(unix)] use std::os::unix::net::UnixStream; @@ -33,11 +39,14 @@ use thrift::protocol::{ }; use thrift::transport::{ TBufferedReadTransport, TBufferedWriteTransport, TFramedReadTransport, TFramedWriteTransport, - TIoChannel, TReadTransport, TTcpChannel, TWriteTransport, + TIoChannel, TReadTransport, TTcpChannel, TTlsClientChannel, TWriteTransport, }; use thrift::OrderedFloat; use thrift_test::*; +const CLIENT_CERT: &[u8] = include_bytes!("../../../keys/client.crt"); +const CLIENT_KEY: &[u8] = include_bytes!("../../../keys/client.key"); + type ThriftClientPair = ( ThriftTestSyncClient, Box>, Option, Box>>, @@ -61,7 +70,6 @@ fn run() -> thrift::Result<()> { // unsupported options: // --pipe // --anon-pipes - // --ssl // --threads let matches = clap_app!(rust_test_client => (version: "1.0") @@ -70,6 +78,7 @@ fn run() -> thrift::Result<()> { (@arg host: --host +takes_value "Host on which the Thrift test server is located") (@arg port: --port +takes_value "Port on which the Thrift test server is listening") (@arg domain_socket: --("domain-socket") +takes_value "Unix Domain Socket on which the Thrift test server is listening") + (@arg ssl: --ssl "Use TLS") (@arg protocol: --protocol +takes_value "Thrift protocol implementation to use (\"binary\", \"compact\", \"multi\", \"multic\")") (@arg transport: --transport +takes_value "Thrift transport implementation to use (\"buffered\", \"framed\")") (@arg testloops: -n --testloops +takes_value "Number of times to run tests") @@ -79,6 +88,7 @@ fn run() -> thrift::Result<()> { let host = matches.value_of("host").unwrap_or("127.0.0.1"); let port = value_t!(matches, "port", u16).unwrap_or(9090); let domain_socket = matches.value_of("domain_socket"); + let ssl = matches.is_present("ssl"); let protocol = matches.value_of("protocol").unwrap_or("binary"); let transport = matches.value_of("transport").unwrap_or("buffered"); let testloops = value_t!(matches, "testloops", u8).unwrap_or(1); @@ -87,10 +97,20 @@ fn run() -> thrift::Result<()> { None => { let listen_address = format!("{}:{}", host, port); info!( - "Client binds to {} with {}+{} stack", - listen_address, protocol, transport + "Client connects to {} with {}+{}{} stack", + listen_address, + protocol, + transport, + if ssl { "+tls" } else { "" } ); - bind(listen_address.as_str(), protocol, transport)? + let tls = if ssl { + let server_name = ServerName::try_from(host.to_owned()) + .map_err(|e| format!("invalid TLS server name {host:?}: {e}"))?; + Some((server_name, tls_config())) + } else { + None + }; + bind(listen_address.as_str(), tls, protocol, transport)? } Some(domain_socket) => { info!( @@ -110,9 +130,28 @@ fn run() -> thrift::Result<()> { fn bind( listen_address: A, + tls: Option<(ServerName<'static>, Arc)>, protocol: &str, transport: &str, ) -> Result { + if let Some((server_name, tls_config)) = tls { + let shared_channel = TTlsClientChannel::connect(listen_address, server_name, tls_config)?; + + let second_service_client = if protocol.starts_with("multi") { + let (i_prot, o_prot) = + build(shared_channel.clone(), transport, protocol, "SecondService")?; + Some(SecondServiceSyncClient::new(i_prot, o_prot)) + } else { + None + }; + + let (i_prot, o_prot) = build(shared_channel, transport, protocol, "ThriftTest")?; + return Ok(( + ThriftTestSyncClient::new(i_prot, o_prot), + second_service_client, + )); + } + // create a TCPStream that will be shared by all Thrift clients // service calls from multiple Thrift clients will be interleaved over the same connection // this isn't a problem for us because we're single-threaded and all calls block to completion @@ -136,6 +175,93 @@ fn bind( Ok((thrift_test_client, second_service_client)) } +fn tls_config() -> Arc { + let provider = Arc::new(rustls::crypto::ring::default_provider()); + let verifier = Arc::new(CrossTestServerVerifier(Arc::clone(&provider))); + + // The shared cross-test client certificate is X.509 v1. rustls can send + // it, but its convenience builder cannot pre-parse it to confirm that the + // public key matches. Build the same single-certificate resolver directly; + // a mismatch would still fail the TLS CertificateVerify signature. + let signing_key = provider + .key_provider + .load_private_key(private_key(CLIENT_KEY)) + .unwrap(); + let certified_key = CertifiedKey::new(certificates(CLIENT_CERT), signing_key); + + let config = ClientConfig::builder_with_provider(provider) + .with_safe_default_protocol_versions() + .unwrap() + .dangerous() + .with_custom_certificate_verifier(verifier) + .with_client_cert_resolver(Arc::new(SingleCertAndKey::from(certified_key))); + + Arc::new(config) +} + +/// Disables peer authentication for the cross-test client, matching the test +/// configuration used by other language bindings. TLS handshake signatures +/// are still verified. This is never used by the Thrift runtime library. +#[derive(Debug)] +struct CrossTestServerVerifier(Arc); + +impl ServerCertVerifier for CrossTestServerVerifier { + fn verify_server_cert( + &self, + _end_entity: &CertificateDer<'_>, + _intermediates: &[CertificateDer<'_>], + _server_name: &ServerName<'_>, + _ocsp_response: &[u8], + _now: UnixTime, + ) -> Result { + Ok(ServerCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + message: &[u8], + certificate: &CertificateDer<'_>, + signature: &DigitallySignedStruct, + ) -> Result { + rustls::crypto::verify_tls12_signature( + message, + certificate, + signature, + &self.0.signature_verification_algorithms, + ) + } + + fn verify_tls13_signature( + &self, + message: &[u8], + certificate: &CertificateDer<'_>, + signature: &DigitallySignedStruct, + ) -> Result { + rustls::crypto::verify_tls13_signature( + message, + certificate, + signature, + &self.0.signature_verification_algorithms, + ) + } + + fn supported_verify_schemes(&self) -> Vec { + self.0.signature_verification_algorithms.supported_schemes() + } +} + +fn certificates(pem: &[u8]) -> Vec> { + rustls_pemfile::certs(&mut Cursor::new(pem)) + .collect::, _>>() + .unwrap() +} + +fn private_key(pem: &[u8]) -> PrivateKeyDer<'static> { + rustls_pemfile::private_key(&mut Cursor::new(pem)) + .unwrap() + .unwrap() +} + #[cfg(unix)] fn bind_uds>( domain_socket: P, diff --git a/test/rs/src/bin/test_server.rs b/test/rs/src/bin/test_server.rs index aa8191cf959..a076aa4757f 100644 --- a/test/rs/src/bin/test_server.rs +++ b/test/rs/src/bin/test_server.rs @@ -17,8 +17,12 @@ use clap::{clap_app, value_t}; use log::*; +use rustls::pki_types::{CertificateDer, PrivateKeyDer}; +use rustls::ServerConfig; use std::collections::{BTreeMap, BTreeSet}; +use std::io::Cursor; +use std::sync::Arc; use std::thread; use std::time::Duration; @@ -34,6 +38,9 @@ use thrift::transport::{ use thrift::OrderedFloat; use thrift_test::*; +const SERVER_CERT: &[u8] = include_bytes!("../../../keys/server.crt"); +const SERVER_KEY: &[u8] = include_bytes!("../../../keys/server.key"); + fn main() { env_logger::init(); @@ -51,13 +58,13 @@ fn main() { fn run() -> thrift::Result<()> { // unsupported options: // --pipe - // --ssl let matches = clap_app!(rust_test_client => (version: "1.0") (author: "Apache Thrift Developers ") (about: "Rust Thrift test server") (@arg port: --port +takes_value "port on which the test server listens") (@arg domain_socket: --("domain-socket") +takes_value "Unix Domain Socket on which the test server listens") + (@arg ssl: --ssl "Use TLS") (@arg transport: --transport +takes_value "transport implementation to use (\"buffered\", \"framed\")") (@arg protocol: --protocol +takes_value "protocol implementation to use (\"binary\", \"compact\")") (@arg server_type: --("server-type") +takes_value "type of server instantiated (\"simple\", \"thread-pool\")") @@ -67,6 +74,7 @@ fn run() -> thrift::Result<()> { let port = value_t!(matches, "port", u16).unwrap_or(9090); let domain_socket = matches.value_of("domain_socket"); + let ssl = matches.is_present("ssl"); let transport = matches.value_of("transport").unwrap_or("buffered"); let protocol = matches.value_of("protocol").unwrap_or("binary"); let server_type = matches.value_of("server_type").unwrap_or("thread-pool"); @@ -74,7 +82,11 @@ fn run() -> thrift::Result<()> { let listen_address = format!("127.0.0.1:{}", port); match domain_socket { - None => info!("Server is binding to {}", listen_address), + None => info!( + "Server is binding to {}{}", + listen_address, + if ssl { " with TLS" } else { "" } + ), Some(domain_socket) => info!("Server is binding to {} (UDS)", domain_socket), } @@ -138,6 +150,7 @@ fn run() -> thrift::Result<()> { ); match domain_socket { + None if ssl => server.listen_tls(&listen_address, tls_config()), None => server.listen(&listen_address), Some(domain_socket) => server.listen_uds(domain_socket), } @@ -152,6 +165,7 @@ fn run() -> thrift::Result<()> { ); match domain_socket { + None if ssl => server.listen_tls(&listen_address, tls_config()), None => server.listen(&listen_address), Some(domain_socket) => server.listen_uds(domain_socket), } @@ -162,6 +176,29 @@ fn run() -> thrift::Result<()> { } } +fn tls_config() -> Arc { + Arc::new( + ServerConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())) + .with_safe_default_protocol_versions() + .unwrap() + .with_no_client_auth() + .with_single_cert(certificates(SERVER_CERT), private_key(SERVER_KEY)) + .unwrap(), + ) +} + +fn certificates(pem: &[u8]) -> Vec> { + rustls_pemfile::certs(&mut Cursor::new(pem)) + .collect::, _>>() + .unwrap() +} + +fn private_key(pem: &[u8]) -> PrivateKeyDer<'static> { + rustls_pemfile::private_key(&mut Cursor::new(pem)) + .unwrap() + .unwrap() +} + struct ThriftTestSyncHandlerImpl; impl ThriftTestSyncHandler for ThriftTestSyncHandlerImpl { fn handle_test_void(&self) -> thrift::Result<()> { diff --git a/test/tests.json b/test/tests.json index ac4a56c4630..27174a8dcb4 100644 --- a/test/tests.json +++ b/test/tests.json @@ -893,6 +893,7 @@ }, "sockets": [ "ip", + "ip-ssl", "domain" ], "transports": [