From f2256c71eebe6a8e8a1bc7b0e267900a37264160 Mon Sep 17 00:00:00 2001 From: Dmytro Shteflyuk Date: Thu, 23 Jul 2026 17:44:08 -0400 Subject: [PATCH] THRIFT-6120: Bound TLS accept handshakes Client: rb Co-Authored-By: OpenAI Codex (GPT-5.6) --- .../lib/thrift/server/nonblocking_server.rb | 7 +- .../lib/thrift/server/thread_pool_server.rb | 7 +- .../lib/thrift/transport/ssl_server_socket.rb | 64 ++++- lib/rb/spec/nonblocking_server_spec.rb | 54 +++- lib/rb/spec/server_spec.rb | 17 ++ lib/rb/spec/ssl_server_socket_spec.rb | 239 +++++++++++++++++- 6 files changed, 375 insertions(+), 13 deletions(-) diff --git a/lib/rb/lib/thrift/server/nonblocking_server.rb b/lib/rb/lib/thrift/server/nonblocking_server.rb index 3f248ea0bb8..6378ca33680 100644 --- a/lib/rb/lib/thrift/server/nonblocking_server.rb +++ b/lib/rb/lib/thrift/server/nonblocking_server.rb @@ -53,7 +53,12 @@ def serve break end next if rd.nil? - socket = @server_transport.accept + begin + socket = @server_transport.accept + rescue => e + next if defined?(OpenSSL::SSL::SSLError) && e.is_a?(OpenSSL::SSL::SSLError) + raise + end @logger.debug "Accepted socket: #{socket.inspect}" @io_manager.add_connection socket end diff --git a/lib/rb/lib/thrift/server/thread_pool_server.rb b/lib/rb/lib/thrift/server/thread_pool_server.rb index 3d53e48da9f..296510ec303 100644 --- a/lib/rb/lib/thrift/server/thread_pool_server.rb +++ b/lib/rb/lib/thrift/server/thread_pool_server.rb @@ -49,7 +49,12 @@ def serve Thread.new do begin loop do - client = @server_transport.accept + begin + client = @server_transport.accept + rescue => e + next if defined?(OpenSSL::SSL::SSLError) && e.is_a?(OpenSSL::SSL::SSLError) + raise + end trans = @transport_factory.get_transport(client) prot = @protocol_factory.get_protocol(trans) begin diff --git a/lib/rb/lib/thrift/transport/ssl_server_socket.rb b/lib/rb/lib/thrift/transport/ssl_server_socket.rb index 2ccb9bf40ff..361eacc3c81 100644 --- a/lib/rb/lib/thrift/transport/ssl_server_socket.rb +++ b/lib/rb/lib/thrift/transport/ssl_server_socket.rb @@ -19,6 +19,7 @@ # under the License. # +require 'io/wait' require 'socket' module Thrift @@ -31,12 +32,71 @@ def initialize(host_or_port, port = nil, ssl_context = nil, client_timeout: DEFA attr_accessor :ssl_context def listen - socket = TCPServer.new(@host, @port) - @handle = OpenSSL::SSL::SSLServer.new(socket, @ssl_context) + tcp_server = TCPServer.new(@host, @port) + @handle = OpenSSL::SSL::SSLServer.new(tcp_server, @ssl_context).tap do |server| + server.start_immediately = false + end + end + + def accept + return if @handle.nil? + + transport = nil + ssl_socket = @handle.accept + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + @client_timeout unless @client_timeout.nil? || @client_timeout == 0 + ssl_socket.to_io.setsockopt(::Socket::IPPROTO_TCP, ::Socket::TCP_NODELAY, 1) + accept_ssl_socket(ssl_socket, deadline) + + transport = Socket.new.tap do |accepted_transport| + accepted_transport.timeout = @client_timeout + accepted_transport.handle = ssl_socket + end + ensure + # Thread#kill leaves $! nil, but marks the thread as aborting. + close_accepted_socket(ssl_socket) if transport.nil? || $! || Thread.current.status == "aborting" end def to_s "ssl(#{super.to_s})" end + + private + + def accept_ssl_socket(ssl_socket, deadline) + return ssl_socket.accept unless deadline + + loop do + case ssl_socket.accept_nonblock(exception: false) + when ssl_socket + return ssl_socket + when :wait_readable + wait_for_handshake(ssl_socket, :read, deadline) + when :wait_writable + wait_for_handshake(ssl_socket, :write, deadline) + else + raise TransportException.new(TransportException::NOT_OPEN, "SSL server socket: Unexpected TLS handshake result") + end + end + end + + def wait_for_handshake(ssl_socket, direction, deadline) + remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC) + ready = if remaining > 0 + if direction == :read + ssl_socket.to_io.wait_readable(remaining) + else + ssl_socket.to_io.wait_writable(remaining) + end + end + return if ready + + raise OpenSSL::SSL::SSLError, "SSL server socket: Timed out accepting TLS connection" + end + + def close_accepted_socket(ssl_socket) + ssl_socket&.close + rescue StandardError + nil + end end end diff --git a/lib/rb/spec/nonblocking_server_spec.rb b/lib/rb/spec/nonblocking_server_spec.rb index c8fbe8ef866..7a1bad30292 100644 --- a/lib/rb/spec/nonblocking_server_spec.rb +++ b/lib/rb/spec/nonblocking_server_spec.rb @@ -265,6 +265,33 @@ def setup_client_thread(result) end end + describe "NonblockingServer accept errors" do + it "preserves the original error when OpenSSL is not loaded" do + hide_const("OpenSSL") + error = RuntimeError.new("plain accept failed") + server_transport = double( + "server transport", + :listen => nil, + :closed? => false, + :close => nil + ) + allow(server_transport).to receive(:accept).and_raise(error) + io_manager = double("IOManager", :ensure_closed => nil) + server = Thrift::NonblockingServer.new( + double("processor"), + server_transport, + nil, + nil, + 1, + Logger.new(IO::NULL) + ) + allow(server).to receive(:start_io_manager).and_return(io_manager) + allow(server).to receive(:select).and_return([[server_transport], nil, nil]) + + expect { server.serve }.to raise_error(error) + end + end + describe Thrift::NonblockingServer::IOManager do def build_io_manager logger = Logger.new(IO::NULL) @@ -330,11 +357,18 @@ def build_io_manager end describe "#{Thrift::NonblockingServer} with TLS transport" do + let(:client_timeout) { Thrift::BaseServerTransport::DEFAULT_CLIENT_TIMEOUT } + before(:each) do @port = available_port handler = Handler.new processor = SpecNamespace::NonblockingService::Processor.new(handler) - @transport = Thrift::SSLServerSocket.new('localhost', @port, create_server_ssl_context) + @transport = Thrift::SSLServerSocket.new( + 'localhost', + @port, + create_server_ssl_context, + client_timeout: client_timeout + ) transport_factory = Thrift::FramedTransportFactory.new logger = Logger.new(STDERR) logger.level = Logger::WARN @@ -355,7 +389,7 @@ def build_io_manager after(:each) do @clients.each(&:close) - @server.shutdown if @server + @server.shutdown if @server && @server_thread&.alive? @server_thread.join(2) if @server_thread @transport.close if @transport end @@ -370,6 +404,22 @@ def build_io_manager expect(@server_thread.join(2)).to be_an_instance_of(Thread) end + context "when a TLS handshake times out" do + let(:client_timeout) { 0.1 } + + it "continues accepting connections" do + stalled_client = TCPSocket.new('localhost', @port) + + expect(Timeout.timeout(1) { stalled_client.read(1) }).to be_nil + expect(@server_thread).to be_alive + + client = setup_tls_client + expect(client.greeting(true)).to eq(SpecNamespace::Hello.new) + ensure + stalled_client&.close + end + end + def setup_tls_client transport = Thrift::FramedTransport.new( Thrift::SSLSocket.new('localhost', @port, nil, create_client_ssl_context) diff --git a/lib/rb/spec/server_spec.rb b/lib/rb/spec/server_spec.rb index 2e66ea23194..fc000c6f9d0 100644 --- a/lib/rb/spec/server_spec.rb +++ b/lib/rb/spec/server_spec.rb @@ -234,5 +234,22 @@ expect(@server_trans).to receive(:close) expect { @server.serve }.to(throw_symbol(:stop)) end + + it "should not enqueue TLS accept errors" do + exception_q = @server.instance_variable_get(:@exception_q) + expect(@server_trans).to receive(:listen).ordered + expect(@server_trans).to receive(:accept).ordered.and_raise(OpenSSL::SSL::SSLError) + expect(@server_trans).to receive(:accept).ordered.and_return(@client) + expect(@trans).to receive(:get_transport).once.with(@client).and_return(@trans) + expect(@prot).to receive(:get_protocol).once.with(@trans).and_return(@prot) + allow(Thread).to receive(:new).and_yield + expect(@processor).to receive(:process).once.with(@prot, @prot) { throw :stop } + expect(@trans).to receive(:close).once + expect(@server_trans).to receive(:close) + + catch(:stop) { @server.serve } + + expect(exception_q).to be_empty + end end end diff --git a/lib/rb/spec/ssl_server_socket_spec.rb b/lib/rb/spec/ssl_server_socket_spec.rb index b3d0b9a5d13..64e1cac9a4b 100644 --- a/lib/rb/spec/ssl_server_socket_spec.rb +++ b/lib/rb/spec/ssl_server_socket_spec.rb @@ -27,15 +27,16 @@ @socket = Thrift::SSLServerSocket.new(1234) end - it "should delegate to_io to the underlying SSL server handle" do + it "should delegate to_io to the underlying TCP server handle" do tcp_server = double("TCPServer") ssl_server = double("SSLServer") allow(TCPServer).to receive(:new).with(nil, 1234).and_return(tcp_server) - allow(OpenSSL::SSL::SSLServer).to receive(:new).with(tcp_server, nil).and_return(ssl_server) + expect(OpenSSL::SSL::SSLServer).to receive(:new).with(tcp_server, nil).and_return(ssl_server) + expect(ssl_server).to receive(:start_immediately=).with(false) allow(ssl_server).to receive(:to_io).and_return(tcp_server) - @socket.listen + expect(@socket.listen).to eq(ssl_server) expect(@socket.to_io).to eq(tcp_server) end @@ -62,24 +63,248 @@ it "should apply the client timeout to accepted sockets" do tcp_server = double("TCPServer") ssl_server = double("SSLServer") - sock = double("SSLSocket") + sock = double("TCPSocket") + ssl_sock = double("SSLSocket") allow(TCPServer).to receive(:new).with(nil, 1234).and_return(tcp_server) - allow(OpenSSL::SSL::SSLServer).to receive(:new).with(tcp_server, nil).and_return(ssl_server) - expect(ssl_server).to receive(:accept).and_return(sock) + expect(OpenSSL::SSL::SSLServer).to receive(:new).with(tcp_server, nil).and_return(ssl_server) + expect(ssl_server).to receive(:start_immediately=).with(false) + expect(ssl_server).to receive(:accept).and_return(ssl_sock) + allow(Process).to receive(:clock_gettime).with(Process::CLOCK_MONOTONIC).and_return(10) expect(sock).to receive(:setsockopt).with(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1) + expect(ssl_sock).to receive(:to_io).and_return(sock) + expect(ssl_sock).to receive(:accept_nonblock).with(exception: false).and_return(ssl_sock) trans = double("Socket") expect(Thrift::Socket).to receive(:new).and_return(trans) expect(trans).to receive(:timeout=).with(Thrift::BaseServerTransport::DEFAULT_CLIENT_TIMEOUT) - expect(trans).to receive(:handle=).with(sock) + expect(trans).to receive(:handle=).with(ssl_sock) @socket.listen expect(@socket.accept).to eq(trans) end + it "does not classify an unexpected nonblocking result as a peer TLS error" do + ssl_socket = double("SSLSocket") + expect(ssl_socket).to receive(:accept_nonblock).with(exception: false).and_return(:unexpected) + + expect { @socket.send(:accept_ssl_socket, ssl_socket, Float::INFINITY) }.to raise_error(Thrift::TransportException) do |error| + expect(error.type).to eq(Thrift::TransportException::NOT_OPEN) + expect(error.message).to eq("SSL server socket: Unexpected TLS handshake result") + end + end + it "should provide a reasonable to_s" do expect(@socket.to_s).to eq("ssl(socket(:1234))") end + + it "times out and closes a TCP client that sends no TLS handshake" do + server = build_server(client_timeout: 0.05) + client = TCPSocket.new('127.0.0.1', server.to_io.local_address.ip_port) + accept_thread = capture_accept(server) + + expect(accept_thread.join(1)).not_to be_nil + status, error = accept_thread.value + expect(status).to eq(:error) + expect(error).to be_a(OpenSSL::SSL::SSLError) + expect(error.message).to eq("SSL server socket: Timed out accepting TLS connection") + expect { client.read_nonblock(1) }.to raise_error(EOFError) + ensure + client&.close + server&.close + stop_thread(accept_thread) + end + + it "uses the same timeout for a stalled partial TLS handshake" do + server = build_server(client_timeout: 0.05) + client = TCPSocket.new('127.0.0.1', server.to_io.local_address.ip_port) + client.write("\x16\x03\x01\x00".b) + accept_thread = capture_accept(server) + + expect(accept_thread.join(1)).not_to be_nil + status, error = accept_thread.value + expect(status).to eq(:error) + expect(error).to be_a(OpenSSL::SSL::SSLError) + expect(error.message).to eq("SSL server socket: Timed out accepting TLS connection") + expect { client.read_nonblock(1) }.to raise_error(EOFError) + ensure + client&.close + server&.close + stop_thread(accept_thread) + end + + it "closes an accepted socket when the accepting thread is interrupted" do + server = build_server(client_timeout: 1) + handshake_waiting = Queue.new + wait_for_handshake = server.method(:wait_for_handshake) + server.define_singleton_method(:wait_for_handshake) do |*args| + handshake_waiting << true + wait_for_handshake.call(*args) + end + client = TCPSocket.new('127.0.0.1', server.to_io.local_address.ip_port) + client.write("\x16\x03\x01\x00".b) + accept_thread = Thread.new { server.accept } + + Timeout.timeout(1) { handshake_waiting.pop } + accept_thread.kill + accept_thread.join + + expect(Timeout.timeout(1) { client.read(1) }).to be_nil + ensure + client&.close + server&.close + stop_thread(accept_thread) + end + + it "closes an accepted socket when transport handoff is interrupted" do + server = build_server(client_timeout: 1) + transport_ready = Queue.new + accepted_transport = Thrift::Socket.new + set_handle = accepted_transport.method(:handle=) + accepted_transport.define_singleton_method(:handle=) do |handle| + set_handle.call(handle) + transport_ready << self + Thread.current.kill + end + allow(Thrift::Socket).to receive(:new).and_return(accepted_transport) + accept_thread = Thread.new { server.accept } + tcp_client, ssl_client = connect_ssl_client(server) + + Timeout.timeout(1) { transport_ready.pop } + accept_thread.join + + expect(accepted_transport).not_to be_open + ensure + accepted_transport&.close + ssl_client&.close + tcp_client&.close + server&.close + stop_thread(accept_thread) + end + + it "accepts a completed TLS handshake and preserves the client timeout" do + server = build_server(client_timeout: 1) + accept_thread = capture_accept(server) + tcp_client, ssl_client = connect_ssl_client(server) + + expect(accept_thread.join(1)).not_to be_nil + status, transport = accept_thread.value + expect(status).to eq(:ok) + expect(transport).to be_open + expect(transport.timeout).to eq(1) + ensure + transport&.close + ssl_client&.close + tcp_client&.close + server&.close + stop_thread(accept_thread) + end + + it "supports resumed client-authenticated TLS sessions" do + context = server_context + context.verify_mode = OpenSSL::SSL::VERIFY_PEER | OpenSSL::SSL::VERIFY_FAIL_IF_NO_PEER_CERT + # The legacy client certificate fixture is self-issued; this test exercises + # authenticated session resumption rather than certificate-chain validation. + context.verify_callback = proc { true } + context.min_version = OpenSSL::SSL::TLS1_2_VERSION + context.max_version = OpenSSL::SSL::TLS1_2_VERSION + context.session_cache_mode = OpenSSL::SSL::SSLContext::SESSION_CACHE_SERVER + server = build_server(client_timeout: 1, context: context) + results = Queue.new + accept_thread = Thread.new do + 2.times do + results << [:ok, server.accept] + rescue => error + results << [:error, error] + break + end + end + + client_context = OpenSSL::SSL::SSLContext.new + client_context.verify_mode = OpenSSL::SSL::VERIFY_NONE + client_context.cert = OpenSSL::X509::Certificate.new(File.read(File.join(ssl_keys_dir, 'client.crt'))) + client_context.key = OpenSSL::PKey::RSA.new(File.read(File.join(ssl_keys_dir, 'client.key'))) + client_context.min_version = OpenSSL::SSL::TLS1_2_VERSION + client_context.max_version = OpenSSL::SSL::TLS1_2_VERSION + client_context.session_cache_mode = OpenSSL::SSL::SSLContext::SESSION_CACHE_CLIENT + + first_client = build_ssl_client(server, client_context) + first_client.connect + status, first_transport = results.pop + expect(status).to eq(:ok) + session = first_client.session + first_client.close + first_transport.close + + second_client = build_ssl_client(server, client_context, session) + client_error = begin + second_client.connect + nil + rescue => error + error + end + status, second_transport = results.pop + + expect(status).to eq(:ok), "server accept failed: #{second_transport.inspect}" + expect(client_error).to be_nil + expect(second_client.session_reused?).to be(true) + ensure + first_client&.close + first_transport&.close + second_client&.close + second_transport&.close if status == :ok + server&.close + stop_thread(accept_thread) + end + + def build_server(client_timeout:, context: server_context) + Thrift::SSLServerSocket.new('127.0.0.1', 0, context, client_timeout: client_timeout).tap(&:listen) + end + + def server_context + context = OpenSSL::SSL::SSLContext.new + context.cert = OpenSSL::X509::Certificate.new(File.read(File.join(ssl_keys_dir, 'server.crt'))) + context.key = OpenSSL::PKey::RSA.new(File.read(File.join(ssl_keys_dir, 'server.key'))) + context + end + + def build_ssl_client(server, context, session = nil) + tcp_client = TCPSocket.new('127.0.0.1', server.to_io.local_address.ip_port) + OpenSSL::SSL::SSLSocket.new(tcp_client, context).tap do |ssl_client| + ssl_client.sync_close = true + ssl_client.session = session unless session.nil? + end + end + + def connect_ssl_client(server) + tcp_client = TCPSocket.new('127.0.0.1', server.to_io.local_address.ip_port) + ssl_client = OpenSSL::SSL::SSLSocket.new(tcp_client, OpenSSL::SSL::SSLContext.new) + ssl_client.sync_close = true + ssl_client.connect + [tcp_client, ssl_client] + rescue + ssl_client&.close + tcp_client&.close + raise + end + + def capture_accept(server) + Thread.new do + [:ok, server.accept] + rescue => error + [:error, error] + end + end + + def stop_thread(thread) + return if thread.nil? + + thread.kill + thread.join + end + + def ssl_keys_dir + File.expand_path('../../../test/keys', __dir__) + end end end