diff --git a/src/boringssl_sys/boringssl.rs b/src/boringssl_sys/boringssl.rs index fcf5ac41b303..762333760ab9 100644 --- a/src/boringssl_sys/boringssl.rs +++ b/src/boringssl_sys/boringssl.rs @@ -813,6 +813,11 @@ pub struct BIO { pub num_write: usize, } +/// `#define BIO_CTRL_FLUSH 11` — the `cmd` of `BIO_flush`. +pub const BIO_CTRL_FLUSH: c_int = 11; +/// `#define BIO_TYPE_SOURCE_SINK 0x0400` — OR-ed into the type of a BIO that ends a chain. +pub const BIO_TYPE_SOURCE_SINK: c_int = 0x0400; + // ═══════════════════════════════════════════════════════════════════════════ // Additional opaque handles // ═══════════════════════════════════════════════════════════════════════════ @@ -971,6 +976,32 @@ unsafe extern "C" { pub fn BIO_new_mem_buf(buf: *const c_void, len: ossl_ssize_t) -> *mut BIO; pub fn BIO_set_mem_eof_return(bio: *mut BIO, eof_value: c_int) -> c_int; + // ── Custom BIOs ────────────────────────────────────────────────────── + pub safe fn BIO_get_new_index() -> c_int; + /// Opaque: set its hooks with `BIO_meth_set_*`, never through the `BIO_METHOD` fields above. + pub fn BIO_meth_new(r#type: c_int, name: *const c_char) -> *mut BIO_METHOD; + pub fn BIO_meth_set_create( + method: *mut BIO_METHOD, + create_func: Option c_int>, + ) -> c_int; + pub fn BIO_meth_set_write( + method: *mut BIO_METHOD, + write_func: Option c_int>, + ) -> c_int; + pub fn BIO_meth_set_read( + method: *mut BIO_METHOD, + read_func: Option c_int>, + ) -> c_int; + pub fn BIO_meth_set_ctrl( + method: *mut BIO_METHOD, + ctrl_func: Option c_long>, + ) -> c_int; + pub fn BIO_set_data(bio: *mut BIO, ptr: *mut c_void); + pub fn BIO_get_data(bio: *mut BIO) -> *mut c_void; + pub fn BIO_set_init(bio: *mut BIO, init: c_int); + pub fn BIO_set_retry_read(bio: *mut BIO); + pub fn BIO_clear_retry_flags(bio: *mut BIO); + // ── RAND ───────────────────────────────────────────────────────────── /// Fills `buf[0..len]` from BoringSSL's thread-local CTR-DRBG and returns 1. /// In the event that sufficient random data can not be obtained, `abort` diff --git a/src/runtime/socket/UpgradedDuplex.rs b/src/runtime/socket/UpgradedDuplex.rs index 88f0656d7e09..63cbb65a0853 100644 --- a/src/runtime/socket/UpgradedDuplex.rs +++ b/src/runtime/socket/UpgradedDuplex.rs @@ -348,10 +348,7 @@ impl UpgradedDuplex { let staged = self.pending_data.replace(Vec::new()); self.reset_timeout(); // Feed in bounded slices rather than one concatenated buffer. Each JS - // chunk was originally delivered on its own; `receive_data` casts the - // length to `c_int` with a panicking `expect`, so handing it the sum of - // every chunk staged in the window would turn a large pre-start burst - // into a process abort. Re-check the engine each round: BoringSSL can + // chunk was originally delivered on its own. Re-check the engine each round: BoringSSL can // re-enter and tear it down partway through, and `teardown()` neuters // in place (frees the SSL, keeps the Option `Some`), so the live // signal is the SSL handle, not the Option. diff --git a/src/runtime/socket/socket_body.rs b/src/runtime/socket/socket_body.rs index 84dfc2309afc..7b97cf3f602d 100644 --- a/src/runtime/socket/socket_body.rs +++ b/src/runtime/socket/socket_body.rs @@ -155,8 +155,7 @@ extern "C" fn select_alpn_callback( // handler, the selection's `toString`, the scope's checkpoint), and // restore it on every path back to BoringSSL — after the scope guard // below has exited, since this guard is declared first. Connected - // usockets only: UpgradedDuplex/Pipe own mem BIOs whose BIO_get_data - // is a BUF_MEM*, not loop_ssl_data. + // usockets only: the BIO of an UpgradedDuplex/Pipe SSL does not hold loop_ssl_data. const LOOP_STATE_SLOTS: usize = 6; // US_SSL_LOOP_STATE_SLOTS debug_assert_eq!( LOOP_STATE_SLOTS as core::ffi::c_int, diff --git a/src/uws/lib.rs b/src/uws/lib.rs index 31ee23f5b94f..1f3ec5ea1b0d 100644 --- a/src/uws/lib.rs +++ b/src/uws/lib.rs @@ -150,21 +150,23 @@ pub use ssl_wrapper::check_server_identity; pub mod ssl_wrapper { use core::cell::{Cell, RefCell}; - use core::ffi::{c_char, c_int, c_void}; + use core::ffi::{c_char, c_int, c_long, c_void}; use core::ptr::NonNull; + use core::sync::atomic::{AtomicPtr, Ordering}; use std::collections::VecDeque; // Re-export the canonical BoringSSL FFI surface; the lower-tier crate now // declares every symbol SSLWrapper needs, so the old local shim is gone. mod boring_sys { pub(super) use bun_boringssl::c::{ - BIO_ctrl_pending, BIO_free, BIO_new, BIO_read, BIO_reset, BIO_s_mem, - BIO_set_mem_eof_return, BIO_write, ERR_clear_error, OwnedSslCtx, SSL, - SSL_CTX_get_verify_mode, SSL_ERROR_SSL, SSL_ERROR_SYSCALL, SSL_ERROR_WANT_READ, - SSL_ERROR_WANT_RENEGOTIATE, SSL_ERROR_WANT_WRITE, SSL_ERROR_ZERO_RETURN, - SSL_RECEIVED_SHUTDOWN, SSL_SESSION, SSL_SESSION_free, SSL_VERIFY_FAIL_IF_NO_PEER_CERT, - SSL_VERIFY_NONE, SSL_VERIFY_PEER, SSL_do_handshake, SSL_free, SSL_get_error, - SSL_get_rbio, SSL_get_shutdown, SSL_get_verify_result, SSL_get_wbio, + BIO, BIO_CTRL_FLUSH, BIO_METHOD, BIO_TYPE_SOURCE_SINK, BIO_clear_retry_flags, + BIO_get_data, BIO_get_new_index, BIO_meth_new, BIO_meth_set_ctrl, BIO_meth_set_read, + BIO_meth_set_write, BIO_new, BIO_set_data, BIO_set_init, BIO_set_retry_read, + ERR_clear_error, OwnedSslCtx, SSL, SSL_CTX_get_verify_mode, SSL_ERROR_SSL, + SSL_ERROR_SYSCALL, SSL_ERROR_WANT_READ, SSL_ERROR_WANT_RENEGOTIATE, + SSL_ERROR_WANT_WRITE, SSL_ERROR_ZERO_RETURN, SSL_RECEIVED_SHUTDOWN, SSL_SESSION, + SSL_SESSION_free, SSL_VERIFY_FAIL_IF_NO_PEER_CERT, SSL_VERIFY_NONE, SSL_VERIFY_PEER, + SSL_do_handshake, SSL_free, SSL_get_error, SSL_get_shutdown, SSL_get_verify_result, SSL_is_init_finished, SSL_new, SSL_pending, SSL_read, SSL_renegotiate, SSL_set_accept_state, SSL_set_bio, SSL_set_connect_state, SSL_set_renegotiate_mode, SSL_set_session_id_context, SSL_set_verify, SSL_set0_verify_cert_store, SSL_shutdown, @@ -172,6 +174,8 @@ pub mod ssl_wrapper { }; } + use bun_core::UnwrapOrOom; + use crate::us_bun_verify_error_t; bun_core::define_scoped_log!(log, SSLWrapper, hidden); @@ -192,21 +196,177 @@ pub mod ssl_wrapper { // the network. write passes the encrypted data that we want to send to // the network. on_close is triggered when we want the network connection // to be closed (remember to flush before closing). - // - // Notes: - // SSL_read() reads unencrypted data which is stored in the input BIO. - // SSL_write() writes unencrypted data into the output BIO. - // BIO_write() writes encrypted data into the input BIO. - // BIO_read() reads encrypted data from the output BIO. - - /// 64kb nice buffer size for SSL reads and writes, should be enough for - /// most cases. In reads we loop until we have no more data to read and in - /// writes we loop until we have no more data to write/backpressure. + + /// 64kb buffer for SSL reads. We loop until there is no more data to read. const BUFFER_SIZE: usize = 65536; - /// Stack scratch shared by `SSL_read` / `BIO_read`. + /// Stack scratch for `SSL_read`. type IoBuffer = bun_core::vec::UninitBuf; + /// A queue that empties with more capacity than this frees its buffer. + const QUEUE_RETAIN_LIMIT: usize = 1024 * 1024; + + /// `SSL3_RT_MAX_PLAIN_LENGTH`: the most plaintext one TLS record carries. + const MAX_RECORD_PLAINTEXT: usize = 16384; + /// `SSL3_RT_HEADER_LENGTH + SSL3_RT_SEND_MAX_ENCRYPTED_OVERHEAD`: the most one record adds. + const MAX_RECORD_OVERHEAD: usize = 5 + 88; + + /// State of the wrapper's BIO. A `BIO_s_mem` moves all unread bytes per read, these do not. + #[derive(Default)] + struct Ciphertext { + /// Sealed by BoringSSL, not yet given to `Handlers::write`, which gets it as one slice. + outgoing: RefCell>, + /// From the peer, not yet read by BoringSSL. + incoming: RefCell>, + } + + // No borrow of a queue is held across an `SSL_*` call or a handler: both can re-enter. + impl Ciphertext { + fn outgoing_len(&self) -> usize { + self.outgoing.borrow().len() + } + + fn incoming_len(&self) -> usize { + self.incoming.borrow().len() + } + + /// Room for every record of one write, so that the queue grows once. + fn reserve_outgoing(&self, plaintext_len: usize) { + let records = plaintext_len / MAX_RECORD_PLAINTEXT + 1; + let sealed_len = plaintext_len + records * MAX_RECORD_OVERHEAD; + self.outgoing + .borrow_mut() + .try_reserve(sealed_len) + .unwrap_or_oom(); + } + + fn push_outgoing(&self, data: &[u8]) { + let mut outgoing = self.outgoing.borrow_mut(); + outgoing.try_reserve(data.len()).unwrap_or_oom(); + outgoing.extend_from_slice(data); + } + + /// The caller owns the bytes, so a write made while it hands them out queues behind them. + fn take_outgoing(&self) -> Vec { + core::mem::take(&mut *self.outgoing.borrow_mut()) + } + + /// Keeps `buffer` for the next write, unless a newer queue exists or it is over the limit. + fn recycle_outgoing(&self, mut buffer: Vec) { + if buffer.capacity() > QUEUE_RETAIN_LIMIT { + return; + } + let mut outgoing = self.outgoing.borrow_mut(); + if outgoing.capacity() == 0 { + buffer.clear(); + *outgoing = buffer; + } + } + + fn discard_outgoing(&self) { + self.outgoing.borrow_mut().clear(); + } + + fn push_incoming(&self, data: &[u8]) { + let mut incoming = self.incoming.borrow_mut(); + incoming.try_reserve(data.len()).unwrap_or_oom(); + incoming.extend(data); + } + + fn free(&self) { + *self.outgoing.borrow_mut() = Vec::new(); + *self.incoming.borrow_mut() = VecDeque::new(); + } + } + + /// `BIO_write` of a wrapper's BIO: BoringSSL queues a sealed record. + unsafe extern "C" fn bio_write( + bio: *mut boring_sys::BIO, + data: *const c_char, + len: c_int, + ) -> c_int { + let Ok(count) = usize::try_from(len) else { + return -1; + }; + // SAFETY: `bio` is a wrapper's live BIO. `init_with_ctx` pointed it at + // the `Ciphertext` of a boxed `Inner`, which outlives the `SSL` that + // owns the BIO. BoringSSL passes `len` readable bytes. + let (ciphertext, data) = unsafe { + boring_sys::BIO_clear_retry_flags(bio); + ( + &*boring_sys::BIO_get_data(bio).cast::(), + core::slice::from_raw_parts(data.cast::(), count), + ) + }; + ciphertext.push_outgoing(data); + len + } + + /// `BIO_read` of a wrapper's BIO: BoringSSL takes the next bytes the peer sent. + unsafe extern "C" fn bio_read( + bio: *mut boring_sys::BIO, + out: *mut c_char, + len: c_int, + ) -> c_int { + // SAFETY: as in `bio_write`. + let ciphertext = unsafe { + boring_sys::BIO_clear_retry_flags(bio); + &*boring_sys::BIO_get_data(bio).cast::() + }; + let mut incoming = ciphertext.incoming.borrow_mut(); + // Only the front run of the ring: BoringSSL reads again for what lies past the wrap. + let (front, _) = incoming.as_slices(); + if front.is_empty() { + // Not the end of the stream: the caller gets `SSL_ERROR_WANT_READ`. + // SAFETY: `bio` is live. + unsafe { boring_sys::BIO_set_retry_read(bio) }; + return -1; + } + let count = front.len().min(usize::try_from(len).unwrap_or(0)); + // SAFETY: BoringSSL passes `len` writable bytes and `count <= len`. The + // queue and BoringSSL's buffer are separate allocations. + unsafe { core::ptr::copy_nonoverlapping(front.as_ptr(), out.cast::(), count) }; + incoming.drain(..count); + if incoming.is_empty() && incoming.capacity() > QUEUE_RETAIN_LIMIT { + *incoming = VecDeque::new(); + } + c_int::try_from(count).unwrap_or(len) + } + + /// BoringSSL flushes the write BIO after each flight and fails the write if this reports 0. + unsafe extern "C" fn bio_ctrl( + _: *mut boring_sys::BIO, + cmd: c_int, + _: c_long, + _: *mut c_void, + ) -> c_long { + c_long::from(cmd == boring_sys::BIO_CTRL_FLUSH) + } + + /// The method of every wrapper's BIO, made on first use and never freed. + fn bio_method() -> *const boring_sys::BIO_METHOD { + static INIT: std::sync::Once = std::sync::Once::new(); + static METHOD: AtomicPtr = AtomicPtr::new(core::ptr::null_mut()); + INIT.call_once(|| { + // A type of its own: `us_ssl_is_socket` in openssl.c tells a socket's BIO by its type. + let r#type = boring_sys::BIO_get_new_index() | boring_sys::BIO_TYPE_SOURCE_SINK; + // SAFETY: the name is a static C string; the hooks match the + // signatures `bio.h` declares. + let method = unsafe { + let method = boring_sys::BIO_meth_new(r#type, c"bun SSLWrapper".as_ptr()); + if method.is_null() { + bun_core::out_of_memory(); + } + boring_sys::BIO_meth_set_write(method, Some(bio_write)); + boring_sys::BIO_meth_set_read(method, Some(bio_read)); + boring_sys::BIO_meth_set_ctrl(method, Some(bio_ctrl)); + method + }; + METHOD.store(method, Ordering::Release); + }); + METHOD.load(Ordering::Acquire) + } + /// Cap on peer-initiated TLS renegotiations per /// [`MAX_RENEGOTIATION_WINDOW`]. Mirrors the `us_reneg_policy` defaults in /// the uSockets C path (openssl.c) and Node's @@ -265,6 +425,8 @@ pub mod ssl_wrapper { pub(crate) renegotiation_count: Cell, pub(crate) renegotiation_window_start: Cell>, traffic: Cell, + /// The BIO of `ssl` points here: see `init_with_ctx`. + ciphertext: Ciphertext, } /// Re-entrancy state of [`SSLWrapper::handle_traffic`]. @@ -513,29 +675,13 @@ pub mod ssl_wrapper { boring_sys::SSL_set_accept_state(ssl.as_ptr()); } } - // SAFETY: BIO_s_mem returns a static method table; BIO_new returns null on OOM. - let input = NonNull::new(unsafe { boring_sys::BIO_new(boring_sys::BIO_s_mem()) }) - .ok_or(InitError::OutOfMemory)?; - // errdefer _ = BoringSSL.BIO_free(input) - let input_guard = scopeguard::guard(input, |bio| { - // SAFETY: bio was created by BIO_new above and not yet transferred to SSL_set_bio. - unsafe { - let _ = boring_sys::BIO_free(bio.as_ptr()); - } - }); - // SAFETY: same as above. - let output = NonNull::new(unsafe { boring_sys::BIO_new(boring_sys::BIO_s_mem()) }) + // SAFETY: the method is live for the process; BIO_new returns null on OOM. + let bio = NonNull::new(unsafe { boring_sys::BIO_new(bio_method()) }) .ok_or(InitError::OutOfMemory)?; - // Set the EOF return value to -1 so that we can detect when the BIO is empty using BIO_ctrl_pending - // SAFETY: input/output are valid BIOs we just created; ssl is valid. - unsafe { - let _ = boring_sys::BIO_set_mem_eof_return(input.as_ptr(), -1); - let _ = boring_sys::BIO_set_mem_eof_return(output.as_ptr(), -1); - // Set the input and output BIOs - boring_sys::SSL_set_bio(ssl.as_ptr(), input.as_ptr(), output.as_ptr()); - } - // Ownership of input/output transferred to ssl via SSL_set_bio; disarm guards. - let _ = scopeguard::ScopeGuard::into_inner(input_guard); + // SAFETY: `ssl` and `bio` are live. Given one BIO for both ends, the + // SSL takes the single reference. It reads and writes nothing until + // `start`, after the BIO is attached to its queues below. + unsafe { boring_sys::SSL_set_bio(ssl.as_ptr(), bio.as_ptr(), bio.as_ptr()) }; let ssl = scopeguard::ScopeGuard::into_inner(ssl_guard); let flags = Flags::default(); @@ -563,10 +709,16 @@ pub mod ssl_wrapper { renegotiation_count: Cell::new(0), renegotiation_window_start: Cell::new(None), traffic: Cell::new(Traffic::Idle), + ciphertext: Ciphertext::default(), }); let this = Self { inner }; - // SAFETY: `ssl` is live; the box outlives it, `deinit` frees `ssl` first. + // SAFETY: `ssl` and its BIO are live. The box outlives them, `deinit` frees them first. unsafe { + boring_sys::BIO_set_data( + bio.as_ptr(), + core::ptr::from_ref(&this.ciphertext).cast_mut().cast(), + ); + boring_sys::BIO_set_init(bio.as_ptr(), 1); us_ssl_set_wrapper( ssl.as_ptr(), core::ptr::from_ref(&this.callbacks).cast_mut().cast(), @@ -783,48 +935,34 @@ pub mod ssl_wrapper { return false; } } - // SSL_shutdown only queues close_notify into the write BIO; nothing - // else pumps it on the memory-BIO paths (duplex / named pipe), so - // drain it now or the peer never sees our shutdown. - let mut buffer = IoBuffer::uninit(); - self.handle_writing(&mut buffer); + // SSL_shutdown only queues close_notify: drain it, or the peer never sees our shutdown. + self.handle_writing(); ret == 1 // truly closed } /// flush buffered data and returns amount of pending data to write pub fn flush(&self) -> usize { - // handle_traffic may trigger a close callback which frees ssl, - // so we must not capture the ssl pointer before calling it. + // handle_traffic may close the wrapper, which frees the queues: this reads 0 then. self.handle_traffic(); - let Some(ssl) = self.ssl.get() else { return 0 }; - // SAFETY: ssl is a live SSL*; SSL_get_wbio returns the BIO bound in init_with_ctx. - unsafe { boring_sys::BIO_ctrl_pending(boring_sys::SSL_get_wbio(ssl.as_ptr())) } + self.ciphertext.outgoing_len() } /// Return if we have pending data to be read or write. Covers both - /// BIOs and `SSL_pending` — decrypted bytes of a partially-returned - /// record are buffered inside the SSL, invisible to either BIO. + /// queues and `SSL_pending`: decrypted bytes of a partly returned record stay in the SSL. pub fn has_pending_data(&self) -> bool { let Some(ssl) = self.ssl.get() else { return false; }; - // SAFETY: ssl is a live SSL*; rbio/wbio bound in init_with_ctx. - unsafe { - boring_sys::SSL_pending(ssl.as_ptr()) > 0 - || boring_sys::BIO_ctrl_pending(boring_sys::SSL_get_wbio(ssl.as_ptr())) > 0 - || boring_sys::BIO_ctrl_pending(boring_sys::SSL_get_rbio(ssl.as_ptr())) > 0 - } + // SAFETY: ssl is a live SSL*. + let decrypted = unsafe { boring_sys::SSL_pending(ssl.as_ptr()) }; + decrypted > 0 + || self.ciphertext.outgoing_len() > 0 + || self.ciphertext.incoming_len() > 0 } - /// Return if we buffered data inside the BIO read buffer, not - /// necessarily will return data to read. This dont reflect - /// SSL_pending(). + /// True if ciphertext from the peer is queued. This does not reflect SSL_pending(). fn has_pending_read(&self) -> bool { - let Some(ssl) = self.ssl.get() else { - return false; - }; - // SAFETY: ssl is a live SSL*. - unsafe { boring_sys::BIO_ctrl_pending(boring_sys::SSL_get_rbio(ssl.as_ptr())) > 0 } + self.ciphertext.incoming_len() > 0 } /// We sent or received a shutdown (closing or closed) @@ -841,24 +979,12 @@ pub mod ssl_wrapper { /// Receive data from the network (encrypted data) pub fn receive_data(&self, data: &[u8]) { - let Some(ssl) = self.ssl.get() else { return }; - - // SAFETY: ssl is a live SSL*; rbio bound in init_with_ctx. - let Some(input) = NonNull::new(unsafe { boring_sys::SSL_get_rbio(ssl.as_ptr()) }) - else { + if self.ssl.get().is_none() { return; - }; - // SAFETY: input is a valid BIO*; data is a valid &[u8] for len bytes. - let written = unsafe { - boring_sys::BIO_write( - input.as_ptr(), - data.as_ptr().cast::(), - c_int::try_from(data.len()).expect("int cast"), - ) - }; - if written > -1 { - self.handle_traffic(); } + // Copied before any handler runs: `data` is only the caller's for this call. + self.ciphertext.push_incoming(data); + self.handle_traffic(); } /// Send data to the network (unencrypted data) @@ -887,14 +1013,12 @@ pub mod ssl_wrapper { self.handle_traffic(); return Err(WriteDataError::WantRead); } + let len = c_int::try_from(data.len()).unwrap_or(c_int::MAX); + self.ciphertext + .reserve_outgoing(usize::try_from(len).unwrap_or(0)); // SAFETY: ssl is a live SSL*; data is a valid &[u8] for len bytes. - let written = unsafe { - boring_sys::SSL_write( - ssl.as_ptr(), - data.as_ptr().cast::(), - c_int::try_from(data.len()).unwrap_or(c_int::MAX), - ) - }; + let written = + unsafe { boring_sys::SSL_write(ssl.as_ptr(), data.as_ptr().cast::(), len) }; if written <= 0 { // SAFETY: ssl is still valid. let err = unsafe { boring_sys::SSL_get_error(ssl.as_ptr(), written) }; @@ -924,9 +1048,10 @@ pub mod ssl_wrapper { pub fn deinit(&self) { self.flags.set_closed_notified(true); if let Some(ssl) = self.ssl.take() { - // SAFETY: ssl was created by SSL_new and is owned by self; SSL_free also frees the input and output BIOs. + // SAFETY: self owns ssl, created by SSL_new; SSL_free also frees its BIO. unsafe { boring_sys::SSL_free(ssl.as_ptr()) }; } + self.ciphertext.free(); self.ctx.set(None); } @@ -1039,12 +1164,8 @@ pub mod ssl_wrapper { && unsafe { boring_sys::SSL_get_verify_result(ssl.as_ptr()) } != 0 { boring_sys::ERR_clear_error(); - // Reset, not only skip the flush below: a re-entered - // `handle_traffic` flushes the write BIO and never gets here. - // SAFETY: wbio is the mem BIO bound in init_with_ctx. - unsafe { - let _ = boring_sys::BIO_reset(boring_sys::SSL_get_wbio(ssl.as_ptr())); - } + // Discard now: a re-entered `handle_traffic` flushes the queue and never gets here. + self.ciphertext.discard_outgoing(); // The peer never gets our Finished, so it cannot read a close_notify. self.flags.set_fatal_error(true); self.flags @@ -1254,39 +1375,19 @@ pub mod ssl_wrapper { true } - fn handle_writing(&self, buffer: &mut IoBuffer) { - let mut read: usize = 0; - loop { - let Some(ssl) = self.ssl.get() else { return }; - // SAFETY: ssl is a live SSL*; wbio bound in init_with_ctx. - let Some(output) = NonNull::new(unsafe { boring_sys::SSL_get_wbio(ssl.as_ptr()) }) - else { + /// One `write` with everything queued. This frame owns it: a nested write queues behind it. + fn handle_writing(&self) { + while self.ssl.get().is_some() { + let pending = self.ciphertext.take_outgoing(); + if pending.is_empty() { + self.ciphertext.recycle_outgoing(pending); return; - }; - // SAFETY: write-only view of the unfilled tail; BIO_read only stores into it. - let available = unsafe { &mut buffer.as_bytes_mut()[read..] }; - // SAFETY: output is a valid BIO*; available is a valid mutable slice. - let just_read = unsafe { - boring_sys::BIO_read( - output.as_ptr(), - available.as_mut_ptr().cast::(), - c_int::try_from(available.len()).expect("int cast"), - ) - }; - if just_read > 0 { - read += usize::try_from(just_read).expect("int cast"); - if read == BUFFER_SIZE { - // SAFETY: the BIO_read calls above wrote `[0..read]` contiguously. - self.trigger_wanna_write_callback(unsafe { buffer.filled(read) }); - read = 0; - } - } else { - break; } - } - if read > 0 { - // SAFETY: the BIO_read calls above wrote `[0..read]` contiguously. - self.trigger_wanna_write_callback(unsafe { buffer.filled(read) }); + self.trigger_wanna_write_callback(&pending); + // The handler may have closed the wrapper, which freed the queues. + if self.ssl.get().is_some() { + self.ciphertext.recycle_outgoing(pending); + } } } @@ -1298,8 +1399,7 @@ pub mod ssl_wrapper { fn handle_traffic(&self) { if self.traffic.get() != Traffic::Idle { log!("handleTraffic re-entered, flushing and deferring to the outer pass"); - let mut buffer = IoBuffer::uninit(); - self.handle_writing(&mut buffer); + self.handle_writing(); self.traffic.set(Traffic::RerunRequested); return; } @@ -1316,13 +1416,13 @@ pub mod ssl_wrapper { fn traffic_pass(&self) { // always handle the handshake first if self.update_handshake_state() { - // shared stack buffer for reading and writing + // send what the handshake queued first + self.handle_writing(); + + // stack buffer for reading // PERF: 64KiB on-stack array — verify stack-size headroom. let mut buffer = IoBuffer::uninit(); - // drain the input BIO first - self.handle_writing(&mut buffer); - - // drain the output BIO in loop, because read can trigger writing and vice versa + // drain the incoming queue in loop, because read can trigger writing and vice versa // Once a callback re-entered, the next pass takes over: the bytes it fed in // may belong to the handshake, which only update_handshake_state reports. while self.traffic.get() == Traffic::Running @@ -1330,7 +1430,7 @@ pub mod ssl_wrapper { && self.handle_reading(&mut buffer) { // read data can trigger writing so we need to handle it - self.handle_writing(&mut buffer); + self.handle_writing(); } if self.traffic.get() != Traffic::Running { return; diff --git a/test/js/bun/http/proxy.test.ts b/test/js/bun/http/proxy.test.ts index 8f5ce175dd25..08e748f4520a 100644 --- a/test/js/bun/http/proxy.test.ts +++ b/test/js/bun/http/proxy.test.ts @@ -1117,6 +1117,64 @@ test("HTTPS over HTTP proxy preserves TLS record order with large bodies", async } }); +// The tunnel's TLS engine queued the ciphertext of a request body in a +// BoringSSL memory BIO. That buffer never shrinks, so a pooled tunnel kept an +// allocation the size of the largest body it ever sent. ASAN keeps freed +// memory resident, so RSS shows nothing there. +test.skipIf(isASAN)("a pooled HTTPS proxy tunnel does not keep the memory of a large request body", async () => { + using origin = Bun.serve({ + port: 0, + tls: tlsCert, + async fetch(req) { + let received = 0; + for await (const chunk of req.body!) received += chunk.byteLength; + return new Response(String(received)); + }, + }); + const MiB = 1024 * 1024; + const bodySize = 64 * MiB; + const bound = bodySize / 2; + const fixture = ` + const url = ${JSON.stringify(origin.url.href)}; + const proxy = ${JSON.stringify(httpProxyServer.url)}; + async function post(size) { + const res = await fetch(url, { method: "POST", body: Buffer.alloc(size, "a"), proxy, tls: { rejectUnauthorized: false } }); + const received = await res.text(); + if (received !== String(size)) throw new Error("the origin received " + received + " of " + size + " bytes"); + } + // The first request opens the tunnel. Every later one reuses it. + await post(1024); + Bun.gc(true); + const before = process.memoryUsage.rss(); + await post(${bodySize}); + // Small requests on the same tunnel, until the upload's memory is released. + let retained = Infinity; + for (let attempt = 0; attempt < 100 && retained > ${bound}; attempt++) { + await post(1024); + Bun.gc(true); + retained = Math.min(retained, process.memoryUsage.rss() - before); + } + console.log(JSON.stringify({ retainedMiB: retained / ${MiB} })); + `; + httpProxyServer.log.length = 0; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", fixture], + env: { ...bunEnv, ...proxyFreeEnv }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + if (exitCode !== 0) console.error("stderr:", stderr); + const connects = httpProxyServer.log.filter(line => line === `CONNECT localhost:${origin.port}`).length; + + // One CONNECT: every request went through the same tunnel, which is still + // pooled when the child reads its RSS. The memory BIO kept more than the body. + expect(stdout).toStartWith("{"); + expect(connects).toBe(1); + expect(JSON.parse(stdout).retainedMiB).toBeLessThan(bound / MiB); + expect(exitCode).toBe(0); +}); + test("HTTPS origin close-delimited body via HTTP proxy does not ECONNRESET", async () => { // Inline raw HTTPS origin: 200 + no Content-Length then close const originServer = tls.createServer( diff --git a/test/js/node/tls/node-tls-connect.test.ts b/test/js/node/tls/node-tls-connect.test.ts index 2db5d230c4b7..772951457f3a 100644 --- a/test/js/node/tls/node-tls-connect.test.ts +++ b/test/js/node/tls/node-tls-connect.test.ts @@ -8,6 +8,7 @@ import { bunRun, tls as COMMON_CERT_, isASAN, + isWindows, nodeExe, rejectUnauthorizedScope, tempDir, @@ -855,6 +856,232 @@ it("a client and a server TLSSocket connected through a synchronous in-memory du }); }); +describe("large writes and reads of a TLSSocket over a Duplex transport", () => { + // The TLS engine behind a Duplex kept its ciphertext in two BoringSSL memory + // BIOs. A memory BIO moves every byte that is still unread to its front after + // each read, so both directions cost quadratic time in what was queued: a + // write was copied out 64 KiB at a time, and a chunk from the transport was + // taken out one TLS record at a time. + const pattern = Buffer.from(Array.from({ length: 251 }, (_, i) => i)); + + // A client and a server TLSSocket joined in memory. `wire` sees every chunk + // of ciphertext the named side writes and returns what to deliver to the peer. + function connectedPair(wire: (from: "client" | "server", chunk: Buffer) => Buffer | null = (_, chunk) => chunk) { + const makeSide = (name: "client" | "server", peer: () => Duplex) => + new Duplex({ + read() {}, + write(chunk: Buffer, _encoding, callback) { + const forwarded = wire(name, chunk); + if (forwarded !== null) peer().push(forwarded); + callback(); + }, + final(callback) { + peer().push(null); + callback(); + }, + }); + const clientSide: Duplex = makeSide("client", () => serverSide); + const serverSide: Duplex = makeSide("server", () => clientSide); + const server = new TLSSocket(serverSide, { isServer: true, secureContext: tls.createSecureContext(COMMON_CERT_) }); + server.on("error", () => {}); + server.on("end", () => server.end()); + const client = tls.connect({ socket: clientSide, rejectUnauthorized: false }); + return { client, server, clientSide }; + } + + // Resolves with the next `length` bytes `socket` emits. + function receive(socket: TLSSocket, length: number) { + const { promise, resolve, reject } = Promise.withResolvers(); + const chunks: Buffer[] = []; + let received = 0; + const onData = (chunk: Buffer) => { + chunks.push(chunk); + received += chunk.length; + if (received < length) return; + socket.off("data", onData); + socket.off("error", reject); + resolve(Buffer.concat(chunks)); + }; + socket.on("data", onData); + socket.once("error", reject); + return promise; + } + + it("one write() reaches the transport in at most two chunks, like node", async () => { + const payload = Buffer.alloc(1024 * 1024, pattern); + const transportWrites: number[] = []; + let recording = false; + const { client, server } = connectedPair((from, chunk) => { + if (recording && from === "client") transportWrites.push(chunk.length); + return chunk; + }); + await once(client, "secureConnect"); + + const received = receive(server, payload.length); + recording = true; + client.write(payload); + const plaintext = await received; + recording = false; + client.end(); + await once(client, "close"); + + expect(plaintext.equals(payload)).toBe(true); + // 64 KiB pieces made this 17 writes. Node gives the transport 2 chunks. + expect(transportWrites.length).toBeLessThanOrEqual(2); + }); + + // process.cpuUsage() advances in scheduler ticks on Windows, about 15.6 ms + // each. Both measurements here are shorter than one tick on a release + // build, so both read 0 and the ratio is NaN. + it.skipIf(isWindows)("one large chunk from the transport costs the same CPU per byte as small chunks", async () => { + const payload = Buffer.alloc(8 * 1024 * 1024, pattern); + let held: Buffer[] | null = null; + const { client, server, clientSide } = connectedPair((from, chunk) => { + if (held === null || from === "client") return chunk; + held.push(chunk); + return null; + }); + await once(client, "secureConnect"); + + // The CPU time this process takes to decrypt `payload`, with its ciphertext + // handed to the client's transport in pieces of `pieceSize` bytes. + async function cpuTimeToReceive(pieceSize: number) { + held = []; + await new Promise((resolve, reject) => server.write(payload, error => (error ? reject(error) : resolve()))); + const ciphertext = Buffer.concat(held); + held = null; + + const received = receive(client, payload.length); + const before = process.cpuUsage(); + for (let offset = 0; offset < ciphertext.length; offset += pieceSize) { + clientSide.push(ciphertext.subarray(offset, offset + pieceSize)); + } + const plaintext = await received; + const { user, system } = process.cpuUsage(before); + expect(plaintext.equals(payload)).toBe(true); + return user + system; + } + + // The best of two runs each, interleaved, so that a busy machine does not + // decide the ratio. + let small = Infinity; + let large = Infinity; + for (let run = 0; run < 2; run++) { + small = Math.min(small, await cpuTimeToReceive(64 * 1024)); + large = Math.min(large, await cpuTimeToReceive(Infinity)); + } + client.end(); + await once(client, "close"); + + // About 1 when the cost is linear. With the memory BIO it was 36 or more. + expect(large / small).toBeLessThan(4); + }); + + it("writes made from 'data' and from the transport's write() keep both streams in order", async () => { + // Both writes are made while the engine is handing ciphertext to the + // transport or plaintext to the socket, and each is larger than one pass of + // either. A TLS record that leaves out of order fails the connection. + const fromServer = Buffer.alloc(320 * 1024, pattern); + const fromData = Buffer.alloc(200 * 1024, pattern.subarray(7)); + const fromTransportWrite = Buffer.alloc(150 * 1024, pattern.subarray(13)); + + let writeFromTransport = false; + const { client, server } = connectedPair((from, chunk) => { + if (from === "client" && writeFromTransport) { + writeFromTransport = false; + client.write(fromTransportWrite); + } + return chunk; + }); + await once(client, "secureConnect"); + + server.once("data", () => server.write(fromServer)); + const atServer = receive(server, 2 + fromData.length + fromTransportWrite.length); + + const chunks: Buffer[] = []; + let depth = 0; + let nestedDataEvents = 0; + let received = 0; + let repliedFromData = false; + const allAtClient = Promise.withResolvers(); + client.on("data", (chunk: Buffer) => { + depth++; + if (depth > 1) nestedDataEvents++; + chunks.push(chunk); + received += chunk.length; + if (!repliedFromData) { + repliedFromData = true; + writeFromTransport = true; + client.write(fromData); + } + if (received >= fromServer.length) allAtClient.resolve(); + depth--; + }); + client.on("error", allAtClient.reject); + client.write("go"); + + const [serverGot] = await Promise.all([atServer, allAtClient.promise]); + client.end(); + await once(client, "close"); + + expect({ + nestedDataEvents, + clientGotAll: Buffer.concat(chunks).equals(fromServer), + serverGotAll: serverGot.equals(Buffer.concat([Buffer.from("go"), fromData, fromTransportWrite])), + }).toEqual({ nestedDataEvents: 0, clientGotAll: true, serverGotAll: true }); + }); + + it("ciphertext that arrives while an earlier chunk is partly decrypted is read in order", async () => { + // The first chunk ends inside a TLS record. The rest arrives in small + // pieces from inside each 'data' event, so it queues behind bytes the + // engine has not read yet, after it has read part of the first chunk. The + // engine's queue is a ring: the record that was cut comes back in two reads. + const payload = Buffer.alloc(1024 * 1024, pattern); + let held: Buffer[] | null = null; + const { client, server, clientSide } = connectedPair((from, chunk) => { + if (held === null || from === "client") return chunk; + held.push(chunk); + return null; + }); + await once(client, "secureConnect"); + + held = []; + await new Promise((resolve, reject) => server.write(payload, error => (error ? reject(error) : resolve()))); + const wire = Buffer.concat(held); + held = null; + + // The middle of the last record that starts in the first two thirds. + let cut = 0; + for (let offset = 0; offset < (wire.length * 2) / 3; ) { + const recordLength = 5 + wire.readUInt16BE(offset + 3); + cut = offset + (recordLength >> 1); + offset += recordLength; + } + const pieces: Buffer[] = []; + for (let offset = cut; offset < wire.length; offset += 7001) pieces.push(wire.subarray(offset, offset + 7001)); + + const received = receive(client, payload.length); + let next = 0; + let pushedFromData = 0; + client.on("data", () => { + for (let pushed = 0; pushed < 4 && next < pieces.length; pushed++, pushedFromData++) { + clientSide.push(pieces[next++]); + } + }); + clientSide.push(wire.subarray(0, cut)); + while (next < pieces.length) clientSide.push(pieces[next++]); + const plaintext = await received; + client.end(); + await once(client, "close"); + + expect({ + somePiecesCameFromData: pushedFromData > 0, + length: plaintext.length, + intact: plaintext.equals(payload), + }).toEqual({ somePiecesCameFromData: true, length: payload.length, intact: true }); + }); +}); + it("the last 'data' event fires before the close_notify reply is written to a duplex transport (tls.connect({ socket }))", async () => { // The peer's last application data and its close_notify reach the engine in // one chunk. The engine used to answer the close_notify before it emitted diff --git a/test/js/node/tls/node-tls-namedpipes.test.ts b/test/js/node/tls/node-tls-namedpipes.test.ts index 674c39ddac01..2a8cf7733c10 100644 --- a/test/js/node/tls/node-tls-namedpipes.test.ts +++ b/test/js/node/tls/node-tls-namedpipes.test.ts @@ -58,6 +58,55 @@ it.if(isWindows)("should work with named pipes and tls", async () => { await expectMaxObjectTypeCount(expect, "TLSSocket", 3); }); +it.if(isWindows)("a write larger than 64 KiB round-trips over a TLS named pipe", async () => { + // The TLS engine over a pipe hands the pipe all the ciphertext of one write + // at once and queues what the pipe delivers. Both directions carry far more + // here than the 64 KiB the engine reads per pass. + const payload = Buffer.alloc(1024 * 1024, Buffer.from(Array.from({ length: 251 }, (_, i) => i))); + // Resolves with the first `payload.length` bytes `socket` emits. + const receive = (socket: ReturnType) => { + const { promise, resolve, reject } = Promise.withResolvers(); + const chunks: Buffer[] = []; + let received = 0; + socket.on("data", (chunk: Buffer) => { + chunks.push(chunk); + received += chunk.length; + if (received >= payload.length) resolve(Buffer.concat(chunks)); + }); + socket.on("error", reject); + return promise; + }; + + const atServer = Promise.withResolvers(); + let client: ReturnType | null = null; + const server = createServer(tls, socket => { + receive(socket).then(all => { + socket.write(all); + atServer.resolve(all); + }, atServer.reject); + }); + server.on("tlsClientError", atServer.reject); + try { + const pipeName = `\\\\.\\pipe\\test\\${randomUUID()}`; + server.listen(pipeName); + await once(server, "listening"); + + const socket = connect({ path: pipeName, ca: tls.cert }); + client = socket; + const atClient = receive(socket); + socket.on("secureConnect", () => socket.write(payload)); + + const [serverGot, clientGot] = await Promise.all([atServer.promise, atClient]); + expect({ serverGotPayload: serverGot.equals(payload), clientGotPayload: clientGot.equals(payload) }).toEqual({ + serverGotPayload: true, + clientGotPayload: true, + }); + } finally { + client?.destroy(); + server.close(); + } +}); + describe.each(["TLSv1.2", "TLSv1.3"] as const)( "%s over a named pipe: write() issued before the handshake completes", version => { diff --git a/test/js/web/websocket/websocket-proxy.test.ts b/test/js/web/websocket/websocket-proxy.test.ts index 415272b4c64b..65f8bf747e7d 100644 --- a/test/js/web/websocket/websocket-proxy.test.ts +++ b/test/js/web/websocket/websocket-proxy.test.ts @@ -289,6 +289,43 @@ describe("WebSocket wss:// through HTTP proxy (TLS tunnel)", () => { gc(); }); + test("a multi-megabyte message round-trips through the TLS tunnel", async () => { + // One frame far larger than the 64 KiB the tunnel's TLS engine reads per + // pass, out and back: every TLS record of it has to cross the proxy in order. + using recorded = await startRecordingProxy(); + const ws = new WebSocket(`wss://127.0.0.1:${wssPort}`, { + proxy: `http://127.0.0.1:${recorded.port}`, + tls: { rejectUnauthorized: false }, + }); + ws.binaryType = "arraybuffer"; + const payload = Buffer.alloc(4 * 1024 * 1024, Buffer.from(Array.from({ length: 251 }, (_, i) => i))); + const echo = Promise.withResolvers(); + const closed = Promise.withResolvers<{ code: number; wasClean: boolean }>(); + ws.addEventListener("open", () => ws.send(payload)); + ws.addEventListener("message", event => { + if (typeof event.data !== "string") echo.resolve(Buffer.from(event.data)); + }); + ws.addEventListener("close", event => { + echo.reject(new Error(`closed before the echo: ${event.code} ${event.reason}`)); + closed.resolve({ code: event.code, wasClean: event.wasClean }); + }); + + const echoedBack = await echo.promise; + ws.close(1000); + expect({ + length: echoedBack.length, + intact: echoedBack.equals(payload), + closed: await closed.promise, + requests: recorded.requests, + }).toEqual({ + length: payload.length, + intact: true, + closed: { code: 1000, wasClean: true }, + requests: [connectRequest(wssPort)], + }); + gc(); + }); + test("server-initiated ping survives through TLS tunnel proxy", async () => { // Regression test: sendPong checked socket.isClosed() on the detached tcp // field instead of using hasTCP(). For wss:// through HTTP proxy, the