From a9133ef6f3ebbf9e4767708e738906ae40bfad1d Mon Sep 17 00:00:00 2001 From: Hiroshi SHIBATA Date: Tue, 25 Aug 2026 08:00:36 +0900 Subject: [PATCH 1/2] Apply the request deadline to TCP frame reads Requester#request computes a monotonic time limit but only the socket readability wait honoured it, so a peer that sends an incomplete frame and keeps the connection open blocked past the configured timeouts until it closed. The parameter defaults to nil, which keeps the previous blocking behaviour for anyone calling recv_reply the old way. Co-Authored-By: Claude Opus 5 --- lib/resolv.rb | 36 +++++-- test/resolv/test_dns.rb | 205 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 235 insertions(+), 6 deletions(-) diff --git a/lib/resolv.rb b/lib/resolv.rb index 6b58f92..79e51d9 100644 --- a/lib/resolv.rb +++ b/lib/resolv.rb @@ -724,7 +724,7 @@ def request(sender, tout) raise ResolvTimeout end begin - reply, from = recv_reply(select_result[0]) + reply, from = recv_reply(select_result[0], timelimit) rescue Errno::ECONNREFUSED, # GNU/Linux, FreeBSD Errno::ECONNRESET, # Windows EOFError @@ -801,7 +801,7 @@ def lazy_initialize self end - def recv_reply(readable_socks) + def recv_reply(readable_socks, timelimit = nil) lazy_initialize reply, from = readable_socks[0].recvfrom(UDPSize) return reply, [from[3],from[1]] @@ -870,7 +870,7 @@ def lazy_initialize self end - def recv_reply(readable_socks) + def recv_reply(readable_socks, timelimit = nil) lazy_initialize reply = readable_socks[0].recv(UDPSize) return reply, nil @@ -935,11 +935,12 @@ def initialize(host, port=Port) @senders = {} end - def recv_reply(readable_socks) - len_data = readable_socks[0].read(2) + def recv_reply(readable_socks, timelimit = nil) + sock = readable_socks[0] + len_data = read_exactly(sock, 2, timelimit) raise EOFError if len_data.nil? || len_data.bytesize != 2 len = len_data.unpack('n')[0] - reply = @socks[0].read(len) + reply = read_exactly(sock, len, timelimit) raise EOFError if reply.nil? || reply.bytesize != len return reply, nil end @@ -968,6 +969,29 @@ def close DNS.free_request_id(@host, @port, id) } end + + private + + # Read +len+ bytes, giving up with ResolvTimeout once +timelimit+ (a + # CLOCK_MONOTONIC value) has passed. A shorter result means the peer + # closed the connection, which the caller turns into an EOFError. + def read_exactly(sock, len, timelimit) + return sock.read(len) unless timelimit + buf = String.new + while buf.bytesize < len + case chunk = sock.read_nonblock(len - buf.bytesize, exception: false) + when :wait_readable + remaining = timelimit - Process.clock_gettime(Process::CLOCK_MONOTONIC) + raise ResolvTimeout if remaining <= 0 + sock.wait_readable(remaining) or raise ResolvTimeout + when nil + break + else + buf << chunk + end + end + buf + end end ## diff --git a/test/resolv/test_dns.rb b/test/resolv/test_dns.rb index 0b81118..fbd9868 100644 --- a/test/resolv/test_dns.rb +++ b/test/resolv/test_dns.rb @@ -943,4 +943,209 @@ def test_tcp_connection_closed_with_partial_message_body client_thread.join end end + + def accept_within_timeout(t) + Timeout.timeout(EnvUtil.apply_timeout_scale(10)) { t.accept } + end + + # Reads one length prefixed DNS message from +sock+. + def read_framed_query(sock) + len_data = sock.read(2) + flunk('the client closed the connection before sending a query') unless len_data&.bytesize == 2 + Resolv::DNS::Message.decode(sock.read(len_data.unpack('n')[0])) + end + + # Builds the encoded reply answering +query+ with a single A record. + def reply_for_query(query, address) + reply = Resolv::DNS::Message.new(query.id) + reply.qr = 1 + reply.rd = query.rd + reply.ra = 1 + query.each_question do |name, typeclass| + reply.add_question(name, typeclass) + reply.add_answer(name, 3600, Resolv::DNS::Resource::IN::A.new(address)) + end + reply.encode + end + + def framed(encoded) + [encoded.bytesize].pack('n') << encoded + end + + # Runs a TCP server which replies with +reply+ and then keeps the connection + # open until the client side is done, so that only the timeout can end the + # request. + def with_tcp_server_keeping_connection_open(reply) + with_tcp('127.0.0.1', 0) do |t| + _, server_port, _, server_address = t.addr + done = Thread::Queue.new + + server_thread = Thread.new do + ct = accept_within_timeout(t) + begin + ct.recv(512) + ct.write(reply) + done.pop + ensure + ct.close + end + end + + client_thread = Thread.new do + begin + yield server_address, server_port + ensure + done.push(true) + end + end + + assert_join_threads([client_thread, server_thread]) + end + end + + def request_over_tcp(server_address, server_port, tout) + requester = Resolv::DNS::Requester::TCP.new(server_address, server_port) + begin + msg = Resolv::DNS::Message.new + msg.add_question('example.org', Resolv::DNS::Resource::IN::A) + sender = requester.sender(msg, msg) + Timeout.timeout(EnvUtil.apply_timeout_scale(10)) do + requester.request(sender, EnvUtil.apply_timeout_scale(tout)) + end + ensure + requester.close + end + end + + def test_tcp_partial_length_prefix_kept_open + with_tcp_server_keeping_connection_open("\x00") do |server_address, server_port| + assert_raise(Resolv::ResolvTimeout) do + request_over_tcp(server_address, server_port, 0.5) + end + end + end + + def test_tcp_partial_message_body_kept_open + reply = [10].pack('n') << '12345' # 5 bytes of a 10 byte message + with_tcp_server_keeping_connection_open(reply) do |server_address, server_port| + assert_raise(Resolv::ResolvTimeout) do + request_over_tcp(server_address, server_port, 0.5) + end + end + end + + def test_tcp_complete_reply_kept_open + with_tcp('127.0.0.1', 0) do |t| + _, server_port, _, server_address = t.addr + done = Thread::Queue.new + + server_thread = Thread.new do + ct = accept_within_timeout(t) + begin + ct.write(framed(reply_for_query(read_framed_query(ct), '192.0.2.1'))) + done.pop + ensure + ct.close + end + end + + client_thread = Thread.new do + begin + reply, = request_over_tcp(server_address, server_port, 2) + assert_equal(1, reply.answer.length) + assert_equal('192.0.2.1', reply.answer[0][2].address.to_s) + ensure + done.push(true) + end + end + + assert_join_threads([client_thread, server_thread]) + end + end + + def test_tcp_reply_arriving_in_two_chunks + with_tcp('127.0.0.1', 0) do |t| + _, server_port, _, server_address = t.addr + done = Thread::Queue.new + + server_thread = Thread.new do + ct = accept_within_timeout(t) + begin + reply = framed(reply_for_query(read_framed_query(ct), '192.0.2.1')) + ct.write(reply.byteslice(0, 6)) + sleep EnvUtil.apply_timeout_scale(0.2) + ct.write(reply.byteslice(6..-1)) + done.pop + ensure + ct.close + end + end + + client_thread = Thread.new do + begin + reply, = request_over_tcp(server_address, server_port, 2) + assert_equal(1, reply.answer.length) + assert_equal('192.0.2.1', reply.answer[0][2].address.to_s) + ensure + done.push(true) + end + end + + assert_join_threads([client_thread, server_thread]) + end + end + + def test_truncated_tcp_fallback_with_partial_message_body_kept_open + with_udp_and_tcp('127.0.0.1', 0) do |u, t| + _, server_port, _, server_address = u.addr + done = Thread::Queue.new + + client_thread = Thread.new do + begin + dns = Resolv::DNS.new(nameserver_port: [[server_address, server_port]], + raise_timeout_errors: true) + begin + dns.timeouts = EnvUtil.apply_timeout_scale(0.5) + assert_raise(Resolv::ResolvError) do + Timeout.timeout(EnvUtil.apply_timeout_scale(10)) do + dns.getresources('foo.example.org', Resolv::DNS::Resource::IN::A) + end + end + ensure + dns.close + end + ensure + done.push(true) + end + end + + udp_server_thread = Thread.new do + msg, (_, client_port, _, client_address) = + Timeout.timeout(EnvUtil.apply_timeout_scale(10)) { u.recvfrom(4096) } + id, word2, = msg.unpack('nnnnnn') + opcode = (word2 & 0x7800) >> 11 + rd = (word2 & 0x0100) >> 8 + qr = 1 + tc = 1 # ask the client to retry over TCP + ra = 1 + word2 = (qr << 15) | (opcode << 11) | (tc << 9) | (rd << 8) | (ra << 7) + u.send([id, word2, 0, 0, 0, 0].pack('nnnnnn'), 0, client_address, client_port) + end + + tcp_server_thread = Thread.new do + ct = accept_within_timeout(t) + begin + ct.recv(512) + ct.write([10].pack('n') << '12345') # 5 bytes of a 10 byte message + done.pop + ensure + ct.close + end + end + + assert_join_threads([client_thread, udp_server_thread, tcp_server_thread]) + end + end + + end From 791447b93f634638a4d9832ac76e6f40e27d807d Mon Sep 17 00:00:00 2001 From: Hiroshi SHIBATA Date: Tue, 25 Aug 2026 08:01:09 +0900 Subject: [PATCH 2/2] Drop a TCP connection that cannot carry another request Giving up part way through a frame leaves the stream between frame boundaries, so reusing the cached requester made every later attempt read the previous frame's bytes as a length prefix. Asking the requester instead of its class keeps a connection whose stream is still on a boundary, which a plain timeout has no reason to throw away. Co-Authored-By: Claude Opus 5 --- lib/resolv.rb | 40 ++++++++++++++- test/resolv/test_dns.rb | 107 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 146 insertions(+), 1 deletion(-) diff --git a/lib/resolv.rb b/lib/resolv.rb index 79e51d9..03e5e9a 100644 --- a/lib/resolv.rb +++ b/lib/resolv.rb @@ -562,7 +562,21 @@ def fetch_resource(name, typeclass) next if !sender senders[[candidate, requester, nameserver, port]] = sender end - reply, reply_name = requester.request(sender, tout) + begin + reply, reply_name = requester.request(sender, tout) + rescue ResolvTimeout + # Giving up part way through a frame loses stream sync, and a peer + # seen going away leaves the socket dead. Either way the requester + # says so, and the next attempt has to open a fresh connection. A + # timeout with the stream still on a frame boundary keeps it; a + # peer that leaves while nothing is being read goes unnoticed here + # and only shows up when the next request is written. + unless requester.reusable? + requesters.delete([nameserver, port]) + requester.close + end + raise + end case reply.rcode when RCode::NoError if reply.tc == 1 and not Requester::TCP === requester @@ -698,6 +712,12 @@ def initialize @socks = nil end + # Whether another request may be sent over the same transport. Only a + # stream transport can end up in a state that rules this out. + def reusable? + true + end + def request(sender, tout) start = Process.clock_gettime(Process::CLOCK_MONOTONIC) timelimit = start + tout @@ -933,6 +953,11 @@ def initialize(host, port=Port) sock = TCPSocket.new(@host, @port) @socks = [sock] @senders = {} + @reusable = true + end + + def reusable? + @reusable end def recv_reply(readable_socks, timelimit = nil) @@ -942,7 +967,14 @@ def recv_reply(readable_socks, timelimit = nil) len = len_data.unpack('n')[0] reply = read_exactly(sock, len, timelimit) raise EOFError if reply.nil? || reply.bytesize != len + @reusable = true return reply, nil + rescue EOFError, SystemCallError + # Whatever the kernel reported, this socket cannot be trusted for + # another frame. In practice that is the peer closing or resetting; + # the rest is rare enough that erring towards reconnecting is right. + @reusable = false + raise end def sender(msg, data, host=@host, port=@port) @@ -964,6 +996,7 @@ def send end def close + @reusable = false super @senders.each_key {|from,id| DNS.free_request_id(@host, @port, id) @@ -975,6 +1008,10 @@ def close # Read +len+ bytes, giving up with ResolvTimeout once +timelimit+ (a # CLOCK_MONOTONIC value) has passed. A shorter result means the peer # closed the connection, which the caller turns into an EOFError. + # Consuming any byte marks the requester unusable until the whole frame + # has been read, since giving up in between loses frame sync. Without a + # +timelimit+ the read blocks instead and keeps no such mark; that is + # only for a caller still using the one argument form of #recv_reply. def read_exactly(sock, len, timelimit) return sock.read(len) unless timelimit buf = String.new @@ -987,6 +1024,7 @@ def read_exactly(sock, len, timelimit) when nil break else + @reusable = false buf << chunk end end diff --git a/test/resolv/test_dns.rb b/test/resolv/test_dns.rb index fbd9868..b4ef92b 100644 --- a/test/resolv/test_dns.rb +++ b/test/resolv/test_dns.rb @@ -843,6 +843,8 @@ def test_tcp_connection_closed_before_length assert_raise(Resolv::ResolvTimeout) do requester.request(sender, 2) end + # The peer went away, so this connection cannot carry another request. + assert_equal(false, requester.reusable?) ensure requester.close end @@ -1147,5 +1149,110 @@ def test_truncated_tcp_fallback_with_partial_message_body_kept_open end end + # A frame read that gives up part way through a frame leaves the stream + # between frame boundaries, so the retry has to start from a new connection. + def test_truncated_tcp_fallback_retries_on_a_new_connection + with_udp_and_tcp('127.0.0.1', 0) do |u, t| + _, server_port, _, server_address = u.addr + + client_thread = Thread.new do + Resolv::DNS.open(nameserver_port: [[server_address, server_port]], + raise_timeout_errors: true) do |dns| + dns.timeouts = [EnvUtil.apply_timeout_scale(1), + EnvUtil.apply_timeout_scale(3)] + Timeout.timeout(EnvUtil.apply_timeout_scale(20)) do + dns.getresources('foo.example.org', Resolv::DNS::Resource::IN::A) + end + end + end + + udp_server_thread = Thread.new do + msg, (_, client_port, _, client_address) = + Timeout.timeout(EnvUtil.apply_timeout_scale(10)) { u.recvfrom(4096) } + id, word2, = msg.unpack('nnnnnn') + opcode = (word2 & 0x7800) >> 11 + rd = (word2 & 0x0100) >> 8 + qr = 1 + tc = 1 # ask the client to retry over TCP + ra = 1 + word2 = (qr << 15) | (opcode << 11) | (tc << 9) | (rd << 8) | (ra << 7) + u.send([id, word2, 0, 0, 0, 0].pack('nnnnnn'), 0, client_address, client_port) + end + + tcp_server_thread = Thread.new do + partial = accept_within_timeout(t) + begin + read_framed_query(partial) + partial.write([45].pack('n') << 'abcdef') # 6 bytes of a 45 byte message + complete = accept_within_timeout(t) + begin + complete.write(framed(reply_for_query(read_framed_query(complete), '192.0.2.1'))) + ensure + complete.close + end + ensure + partial.close + end + end + result, = assert_join_threads([client_thread, udp_server_thread, tcp_server_thread]) + assert_equal(['192.0.2.1'], result.map {|rr| rr.address.to_s }) + end + end + + # A timeout with no bytes read leaves the stream on a frame boundary, so the + # retry keeps the connection. The reply below only ever reaches the client if + # the second attempt reuses the socket the first one opened. + def test_truncated_tcp_fallback_keeps_the_connection_when_nothing_arrived + with_udp_and_tcp('127.0.0.1', 0) do |u, t| + _, server_port, _, server_address = u.addr + done = Thread::Queue.new + + client_thread = Thread.new do + begin + Resolv::DNS.open(nameserver_port: [[server_address, server_port]], + raise_timeout_errors: true) do |dns| + dns.timeouts = [EnvUtil.apply_timeout_scale(0.5), + EnvUtil.apply_timeout_scale(5)] + Timeout.timeout(EnvUtil.apply_timeout_scale(20)) do + dns.getresources('foo.example.org', Resolv::DNS::Resource::IN::A) + end + end + ensure + done.push(true) + end + end + + udp_server_thread = Thread.new do + msg, (_, client_port, _, client_address) = + Timeout.timeout(EnvUtil.apply_timeout_scale(10)) { u.recvfrom(4096) } + id, word2, = msg.unpack('nnnnnn') + opcode = (word2 & 0x7800) >> 11 + rd = (word2 & 0x0100) >> 8 + qr = 1 + tc = 1 # ask the client to retry over TCP + ra = 1 + word2 = (qr << 15) | (opcode << 11) | (tc << 9) | (rd << 8) | (ra << 7) + u.send([id, word2, 0, 0, 0, 0].pack('nnnnnn'), 0, client_address, client_port) + end + + tcp_server_thread = Thread.new do + ct = accept_within_timeout(t) + begin + query = read_framed_query(ct) + # Stay silent past the first interval, then answer on this connection. + # It has to stay open until the client is done: the retry resends the + # query, and closing with that still unread would discard the reply. + sleep EnvUtil.apply_timeout_scale(1) + ct.write(framed(reply_for_query(query, '192.0.2.1'))) + done.pop + ensure + ct.close + end + end + + result, = assert_join_threads([client_thread, udp_server_thread, tcp_server_thread]) + assert_equal(['192.0.2.1'], result.map {|rr| rr.address.to_s }) + end + end end