diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 2539039..a3cf5ac 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -17,7 +17,7 @@ jobs: - '3.3.4' steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v7 - name: Set up Ruby & Rust uses: oxidize-rb/actions/setup-ruby-and-rust@v1 with: diff --git a/Cargo.lock b/Cargo.lock index 0e14e36..0ad3e20 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1,6 +1,6 @@ # This file is automatically @generated by Cargo. # It is not intended for manual editing. -version = 3 +version = 4 [[package]] name = "addr2line" @@ -366,9 +366,9 @@ checksum = "a8d1add55171497b4705a648c6b583acafb01d58050a51727785f0b2c8e0a2b2" [[package]] name = "h2" -version = "0.4.7" +version = "0.4.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ccae279728d634d083c00f6099cb58f01cc99c145b84b8be2f6c74618d79922e" +checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" dependencies = [ "atomic-waker", "bytes", @@ -443,13 +443,14 @@ checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4" [[package]] name = "hyper" -version = "1.6.0" +version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cc2b571658e38e0c01b1fdca3bbbe93c00d3d71693ff2770043f8c29bc7d6f80" +checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498" dependencies = [ + "atomic-waker", "bytes", "futures-channel", - "futures-util", + "futures-core", "h2", "http", "http-body", diff --git a/ext/hyper_ruby/Cargo.toml b/ext/hyper_ruby/Cargo.toml index 3098e90..2844520 100644 --- a/ext/hyper_ruby/Cargo.toml +++ b/ext/hyper_ruby/Cargo.toml @@ -16,7 +16,7 @@ bytes = "1.5" tokio-stream = { version = "0.1", features = ["net"] } crossbeam-channel = "0.5.14" rb-sys = "0.9.110" -hyper = { version = "1.0", features = ["http1", "http2", "server"] } +hyper = { version = "1.10", features = ["http1", "http2", "server"] } hyper-util = { version = "0.1", features = ["tokio", "server", "server-graceful", "server-auto", "http1", "http2"] } http-body-util = "0.1.2" jemallocator = { version = "0.5.4", features = ["disable_initial_exec_tls"] } diff --git a/ext/hyper_ruby/src/lib.rs b/ext/hyper_ruby/src/lib.rs index 54f785c..7a89261 100644 --- a/ext/hyper_ruby/src/lib.rs +++ b/ext/hyper_ruby/src/lib.rs @@ -16,6 +16,7 @@ use tokio::io::{AsyncRead, AsyncWrite}; use std::cell::RefCell; use std::net::SocketAddr; +use std::os::unix::fs::MetadataExt; use std::sync::atomic::{AtomicU64, Ordering}; use tokio::net::{TcpListener, UnixListener}; @@ -27,7 +28,7 @@ use crossbeam_channel; use hyper::service::service_fn; use hyper::{Error, Request as HyperRequest, Response as HyperResponse, StatusCode}; -use hyper::body::Incoming; +use hyper::body::{Body, Incoming}; use hyper_util::rt::TokioIo; use hyper_util::server::conn::auto; use http_body_util::BodyExt; @@ -113,6 +114,9 @@ struct Server { runtime: RefCell>>, shutdown: RefCell>>, total_connections: Arc, + // (dev, ino) of the Unix socket file this server bound, so stop() only removes + // the file if a replacement server hasn't taken over the path in the meantime. + socket_ident: RefCell>, } impl Server { @@ -126,6 +130,7 @@ impl Server { runtime: RefCell::new(None), shutdown: RefCell::new(None), total_connections: Arc::new(AtomicU64::new(0)), + socket_ident: RefCell::new(None), } } @@ -310,32 +315,40 @@ impl Server { // Create the listener with proper error handling let listener = if config.bind_address.starts_with("unix:") { let path = config.bind_address.trim_start_matches("unix:"); - - // Check if the socket file already exists and try to delete it - if std::path::Path::new(path).exists() { - debug!("Unix socket file {} already exists, attempting to remove it", path); - match std::fs::remove_file(path) { - Ok(_) => debug!("Successfully removed existing socket file"), - Err(e) => { - error!("Failed to remove existing Unix socket file {}: {}", path, e); - return Err(MagnusError::new( - magnus::exception::runtime_error(), - format!("Failed to remove existing Unix socket file {}: {}", path, e) - )); - } - } - } - - match UnixListener::bind(path) { - Ok(listener) => Listener::Unix(listener), + + // Bind to a unique temp path and atomically rename over the target, so the + // path always points at a live socket even when a replacement server takes + // over a path an older, still-draining server bound. + static SOCKET_TMP_SEQ: AtomicU64 = AtomicU64::new(0); + let tmp_path = format!("{}.{}.{}.tmp", path, std::process::id(), SOCKET_TMP_SEQ.fetch_add(1, Ordering::Relaxed)); + + let listener = match UnixListener::bind(&tmp_path) { + Ok(listener) => listener, Err(e) => { - error!("Failed to bind to Unix socket {}: {}", path, e); + error!("Failed to bind to Unix socket {}: {}", tmp_path, e); return Err(MagnusError::new( magnus::exception::runtime_error(), - format!("Failed to bind to Unix socket {}: {}", path, e) + format!("Failed to bind to Unix socket {}: {}", tmp_path, e) )); } + }; + + // The socket file's identity survives the rename; stop() compares against it + // so an older server generation never unlinks a newer generation's socket. + let ident = std::fs::symlink_metadata(&tmp_path).ok().map(|m| (m.dev(), m.ino())); + + if let Err(e) = std::fs::rename(&tmp_path, path) { + let _ = std::fs::remove_file(&tmp_path); + error!("Failed to install Unix socket file {}: {}", path, e); + return Err(MagnusError::new( + magnus::exception::runtime_error(), + format!("Failed to install Unix socket file {}: {}", path, e) + )); } + + *self.socket_ident.borrow_mut() = ident; + + Listener::Unix(listener) } else { match config.bind_address.parse::() { Ok(addr) => { @@ -474,9 +487,18 @@ impl Server { let bind_address = self.config.borrow().bind_address.clone(); if bind_address.starts_with("unix:") { let path = bind_address.trim_start_matches("unix:"); - std::fs::remove_file(path).unwrap_or_else(|e| { - warn!("Failed to remove socket file: {:?}", e); - }); + // Only remove the socket file if it's still the one this server bound; a + // replacement server may have taken over the path while we were draining. + match (self.socket_ident.borrow_mut().take(), std::fs::symlink_metadata(path)) { + (Some((dev, ino)), Ok(meta)) if (meta.dev(), meta.ino()) == (dev, ino) => { + std::fs::remove_file(path).unwrap_or_else(|e| { + warn!("Failed to remove socket file: {:?}", e); + }); + } + (Some(_), Ok(_)) => info!("Socket file {} was replaced by another server; leaving it in place", path), + (Some(_), Err(_)) => debug!("Socket file {} already removed", path), + (None, _) => {} + } } Ok(()) @@ -494,7 +516,17 @@ async fn handle_request( debug!("Headers: {:?}", req.headers()); let (parts, body) = req.into_parts(); - + + // Capture the declared body length before consuming the body. Hyper's + // `Incoming::size_hint().exact()` mirrors the inbound Content-Length + // (populated from the header for H2 as well as H1), and is the only + // signal we have here: hyper converts `RST_STREAM(NO_ERROR|CANCEL)` — + // the codes browsers send on navigation cancellation — into a clean + // end-of-body rather than surfacing the reset (see + // hyper-1.6.0/src/body/incoming.rs:~249). We therefore validate the + // declared length against what we actually collected. + let declared_len = body.size_hint().exact(); + // Collect the body with timeout let body_bytes = match timeout( std::time::Duration::from_millis(recv_timeout), @@ -510,9 +542,19 @@ async fn handle_request( return Ok(create_timeout_response()); } }; - + debug!("Collected body size: {}", body_bytes.len()); + if let Some(declared) = declared_len { + if declared != body_bytes.len() as u64 { + warn!( + "Body truncated: declared {} bytes, received {} — likely RST_STREAM(CANCEL); rejecting request", + declared, body_bytes.len() + ); + return Ok(create_bad_request_response("Body length does not match Content-Length")); + } + } + let hyper_request = HyperRequest::from_parts(parts, body_bytes); let is_grpc = grpc::is_grpc_request(&hyper_request); debug!("Is gRPC: {}", is_grpc); @@ -628,11 +670,19 @@ fn create_too_many_requests_response(error_message: &str) -> HyperResponse HyperResponse { + HyperResponse::builder() + .status(StatusCode::BAD_REQUEST) + .header("content-type", "text/plain") + .body(BodyWithTrailers::new(Bytes::from(error_message.to_string()), None)) + .unwrap() +} + #[magnus::init] fn init(ruby: &Ruby) -> Result<(), MagnusError> { let module = ruby.define_module("HyperRuby")?; diff --git a/test/test_h2_stream_reset.rb b/test/test_h2_stream_reset.rb new file mode 100644 index 0000000..14c8429 --- /dev/null +++ b/test/test_h2_stream_reset.rb @@ -0,0 +1,112 @@ +# frozen_string_literal: true + +require "test_helper" +require "httpx" # with_configured_server builds an HTTPX client (unused here) +require "socket" +require "timeout" + +# Regression test for a silent-truncation bug: when an HTTP/2 peer sends +# HEADERS + a partial DATA frame + RST_STREAM (the frame sequence a browser +# produces when it cancels an in-flight request on page navigation), the +# server currently hands the truncated body to the Ruby handler as if the +# request had completed normally. Downstream consumers (Kafka, etc.) then +# see a short body alongside the original Content-Length header, producing +# confusingly "truncated" payloads. +# +# Correct behaviour: a stream that ends via RST_STREAM (not END_STREAM) is +# not a completed request; the handler should not run. +class TestH2StreamReset < HyperRubyTest + PORT = 3010 + PARTIAL_BYTES = 16_384 # default HTTP/2 SETTINGS_MAX_FRAME_SIZE + CLAIMED_CONTENT_LENGTH = 450_403 + + def test_rst_stream_after_partial_data_does_not_invoke_handler + invocations = [] + mutex = Mutex.new + + handler = lambda do |request| + mutex.synchronize do + invocations << { + path: request.path, + body_size: request.body_size, + content_length: request.header("content-length"), + } + end + HyperRuby::Response.new(200, { "Content-Type" => "text/plain" }, "ok") + end + + config = { bind_address: "127.0.0.1:#{PORT}", tokio_threads: 1, recv_timeout: 1_000 } + + with_configured_server(config, handler) do + send_h2_headers_data_rst( + host: "127.0.0.1", + port: PORT, + partial_bytes: PARTIAL_BYTES, + claimed_content_length: CLAIMED_CONTENT_LENGTH, + ) + + # Give the server time to hand off to the worker if it's going to. + sleep 0.2 + end + + mutex.synchronize do + assert_empty invocations, + "handler should not have been invoked for a RST_STREAM'd request, but it ran with: #{invocations.inspect}" + end + end + + private + + H2_PREFACE = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n".b + TYPE_DATA = 0x0 + TYPE_HEADERS = 0x1 + TYPE_RST_STREAM = 0x3 + TYPE_SETTINGS = 0x4 + FLAG_END_HEADERS = 0x4 + FLAG_ACK = 0x1 + RST_CODE_CANCEL = 0x8 # what Chromium sends on navigation + + def h2_frame(type, flags, stream_id, payload) + len = payload.bytesize + [len >> 16 & 0xff, len >> 8 & 0xff, len & 0xff, + type, flags, stream_id & 0x7fffffff].pack("CCCCCN") + payload + end + + # Minimal HPACK: "literal header field, never indexed, new name" with + # 7-bit string lengths. Names and values must fit in 126 bytes — fine for + # everything we send here. + def hpack_literal(name, value) + raise "name too long" if name.bytesize > 126 + raise "value too long" if value.bytesize > 126 + [0x10, name.bytesize].pack("CC") + name.b + + [value.bytesize].pack("C") + value.b + end + + def send_h2_headers_data_rst(host:, port:, partial_bytes:, claimed_content_length:) + sock = TCPSocket.new(host, port) + sock.write(H2_PREFACE) + sock.write(h2_frame(TYPE_SETTINGS, 0, 0, "".b)) + sock.write(h2_frame(TYPE_SETTINGS, FLAG_ACK, 0, "".b)) + + headers = "".b + headers << hpack_literal(":method", "POST") + headers << hpack_literal(":scheme", "http") + headers << hpack_literal(":path", "/rst-truncated") + headers << hpack_literal(":authority", "#{host}:#{port}") + headers << hpack_literal("content-type", "application/octet-stream") + headers << hpack_literal("content-length", claimed_content_length.to_s) + + sock.write(h2_frame(TYPE_HEADERS, FLAG_END_HEADERS, 1, headers)) + sock.write(h2_frame(TYPE_DATA, 0, 1, "X".b * partial_bytes)) + sock.write(h2_frame(TYPE_RST_STREAM, 0, 1, [RST_CODE_CANCEL].pack("N"))) + + # Drain anything the server may have written before we hung up; we don't + # care what it is. + begin + Timeout.timeout(0.5) { sock.read(4096) } + rescue Timeout::Error + end + ensure + sock&.close + end +end diff --git a/test/test_http.rb b/test/test_http.rb index 173a837..867d1d6 100644 --- a/test/test_http.rb +++ b/test/test_http.rb @@ -153,6 +153,65 @@ def test_unix_socket_cleanup end end + def test_unix_socket_takeover_preserves_new_generation_socket + # A replacement server (e.g. a redeployed container) can bind the same path while the + # old server is still draining; the old server's stop must not delete the new socket. + socket_path = "/tmp/hyper_ruby_test_takeover.sock" + File.unlink(socket_path) if File.exist?(socket_path) + + old_server = HyperRuby::Server.new + old_server.configure({ bind_address: "unix:#{socket_path}" }) + old_server.start + + new_server = HyperRuby::Server.new + new_server.configure({ bind_address: "unix:#{socket_path}" }) + new_server.start + + workers = 1.times.map do + Thread.new do + new_server.run_worker { |request| handler_simple(request) } + end + end + + old_server.stop + old_server = nil + assert File.exist?(socket_path), "old server's stop must not remove the new server's socket" + + client = HTTPX.with(transport: "unix", addresses: [socket_path], origin: "http://host") + response = client.get("/") + assert_equal 200, response.status + + new_server.stop + workers.each(&:join) + workers = nil + refute File.exist?(socket_path), "new server's stop should remove the socket it owns" + ensure + old_server.stop if old_server + new_server.stop if new_server && workers + workers&.each(&:join) + File.unlink(socket_path) if File.exist?(socket_path) + end + + def test_unix_socket_stop_leaves_foreign_file + # If something else has replaced our socket file, stop must leave it alone. + socket_path = "/tmp/hyper_ruby_test_foreign.sock" + File.unlink(socket_path) if File.exist?(socket_path) + + server = HyperRuby::Server.new + server.configure({ bind_address: "unix:#{socket_path}" }) + server.start + + File.unlink(socket_path) + FileUtils.touch(socket_path) + + server.stop + server = nil + assert File.exist?(socket_path), "stop must not remove a file it did not bind" + ensure + server.stop if server + File.unlink(socket_path) if File.exist?(socket_path) + end + # This test requires root permissions to create a file that can't be deleted. # Skip it unless we're running with proper permissions. def test_unix_socket_undeletable @@ -171,13 +230,13 @@ def test_unix_socket_undeletable server = HyperRuby::Server.new server.configure({ bind_address: "unix:#{socket_path}" }) - # This should raise an exception about not being able to remove the file + # This should raise an exception about not being able to install the socket file error = assert_raises(RuntimeError) do server.start end - + # Verify the error message - assert_match(/Failed to remove existing Unix socket file/, error.message) + assert_match(/Failed to install Unix socket file/, error.message) ensure # Clean up with sudo system("sudo rm -f #{socket_path}") if File.exist?(socket_path) @@ -201,14 +260,15 @@ def test_unix_socket_directory_error server.start end - # The error is from trying to remove the directory, not from binding - assert_match(/Failed to remove existing Unix socket file/, error.message) - - # It should include something about "Operation not permitted" or similar - assert(error.message.include?("Operation not permitted") || - error.message.include?("Permission denied") || - error.message.include?("not a socket"), - "Error should indicate issue with removing directory: #{error.message}") + # The error is from renaming the bound socket over the directory, not from binding + assert_match(/Failed to install Unix socket file/, error.message) + + # It should include something about the target being a directory or similar + assert(error.message.include?("Is a directory") || + error.message.include?("Operation not permitted") || + error.message.include?("Permission denied") || + error.message.include?("not a socket"), + "Error should indicate issue with replacing directory: #{error.message}") ensure # Clean up FileUtils.rm_rf(socket_dir) if Dir.exist?(socket_dir)