From 51002c60836b3407eed5ffaf7afe663863801494 Mon Sep 17 00:00:00 2001 From: krynju Date: Fri, 16 Oct 2026 11:54:26 +0200 Subject: [PATCH 1/2] fix: write the ciphertext to the socket outside ssl.lock (#72) `SSL_write_ex` and `SSL_read_ex` share `ssl.lock`, and the write BIO wrote to the socket from inside the call. A write waiting for a peer that had stopped reading therefore held the lock, and every read on the stream waited behind it: a full-duplex exchange, or a read beside a large write, deadlocked. The write BIO now appends what one of the package's own SSL calls produces to a per-stream buffer (`BIOStreamData`, an `incall` flag set under `ssl.lock` by `@geterror`). The call's task sends the buffer once the lock is released: - a writer (`unsafe_write`, the handshake) with `sendout!`: it takes `sendlock`, swaps the buffer out under `ssl.lock` and writes it, so records leave in the order they were produced, and it returns only once its bytes are on the socket, or with the error that kept them from getting there; - a reader (`unsafe_read`, `eof`) with `sendreply!`: a reply OpenSSL made goes out from a task that only takes a free `sendlock`. Whoever holds the lock re-checks the buffer under `ssl.lock` before releasing it, so the reply is never left behind, and a read never waits behind a parked writer. Writes go to OpenSSL in 64 KiB chunks under a writer lock, so concurrent writes arrive whole. A raw `SSL_*` call made on `ssl.ssl` from outside the package still writes straight to the socket. `close` buffers the close_notify and sends it from a task, closing the socket outright should that take longer than `CLOSE_NOTIFY_GRACE`. With a socket write in flight it makes no close_notify and closes the socket handle outright at once (`jl_forceclose_uv`, as Base does for an unconnected socket), which fails the parked write. Tests: ConcurrentReadWrite (the #72 scenario), FullDuplex, ConcurrentWriters (whole writes), CloseWithParkedWriter, CloseRacingWriter, CloseBehindRawWrite, CloseNotifyEOF, WriteSizes, ReplyBehindSender, LegacyAccept, RawSSLCalls, PeerResetMidRecord. Co-Authored-By: Claude Fable 5.1 --- src/ssl.jl | 284 +++++++++++++++++++++++--- test/runtests.jl | 519 ++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 765 insertions(+), 38 deletions(-) diff --git a/src/ssl.jl b/src/ssl.jl index 8a8da9e..837f6c4 100644 --- a/src/ssl.jl +++ b/src/ssl.jl @@ -32,10 +32,36 @@ end bio_set_read_retry(bio::BIO) = bio_set_flags(bio, BIO_FLAGS_READ | BIO_FLAGS_SHOULD_RETRY) bio_clear_flags(bio::BIO) = bio_set_flags(bio, 0x00) +""" + What the stream BIOs point at: the socket, and the write side's state. + +The write BIO must not write to the socket during one of the package's own SSL calls: +those run under `ssl.lock`, which reads need too, so a socket write that waits for the +peer there stops the reads (#72). While such a call runs (`incall`, set under +`ssl.lock`), the callback appends the ciphertext to `buf` instead, and the call's task +writes it to the socket once it has released the lock (`sendout!`, `sendreply!`). A raw +`SSL_*` call made on `ssl.ssl` from outside the package has `incall` false and writes +straight to the socket. +""" +mutable struct BIOStreamData + io::TCPSocket + # one of the package's SSL calls is running: the write callback buffers + incall::Bool + # ciphertext the calls produced, in order, not yet written to the socket + buf::Vector{UInt8} + # a socket write of taken ciphertext is in progress; set and cleared under `ssl.lock` + inflight::Bool + # ciphertext was dropped or its socket write failed: a writer whose bytes were among + # it must not return as if they were sent + failed::Bool +end + +BIOStreamData(io::TCPSocket) = BIOStreamData(io, false, UInt8[], false, false) + function on_bio_stream_read(bio::BIO, out::Ptr{Cchar}, outlen::Cint) try bio_clear_flags(bio) - io = bio_get_data(bio)::TCPSocket + io = (bio_get_data(bio)::BIOStreamData).io n = bytesavailable(io) if n == 0 bio_set_read_retry(bio) @@ -51,8 +77,19 @@ end function on_bio_stream_write(bio::BIO, in::Ptr{Cchar}, inlen::Cint)::Cint try - io = bio_get_data(bio)::TCPSocket - written = unsafe_write(io, in, inlen) + data = bio_get_data(bio)::BIOStreamData + if data.incall + # produced under `ssl.lock`: buffered, the call's task sends it after + buf = data.buf + n = length(buf) + resize!(buf, n + inlen) + GC.@preserve buf unsafe_copyto!(pointer(buf) + n, Ptr{UInt8}(in), inlen) + return inlen + end + # a raw call from outside the package: straight to the socket. Such a call may + # not overlap the package's own on the same stream (OpenSSL's own rule for one + # SSL object): what the package has buffered would be overtaken on the wire + written = unsafe_write(data.io, in, inlen) return Cint(written) catch e # we don't want to throw a Julia exception from a C callback; -1 fails the @@ -418,13 +455,21 @@ mutable struct SSLStream <: IO peekbuf::Base.RefValue{UInt8} peekbytes::Base.RefValue{Csize_t} closed::Bool + # what the BIOs point at, see `BIOStreamData` + data::BIOStreamData + # held while buffered ciphertext is written to the socket: it leaves in the order + # it was produced, and whoever releases this lock has sent all there was + sendlock::ReentrantLock + # one `unsafe_write` at a time, so that concurrent writes arrive whole + writelock::ReentrantLock function SSLStream(ssl_context::SSLContext, io::TCPSocket) + data = BIOStreamData(io) # Create a read and write BIOs. - bio_read::BIO = BIO(io; finalize=false) - bio_write::BIO = BIO(io; finalize=false) + bio_read::BIO = BIO(data; finalize=false) + bio_write::BIO = BIO(data; finalize=false) ssl = SSL(ssl_context, bio_read, bio_write) - x = new(ssl, ssl_context, bio_read, bio_write, io, ReentrantLock(), ReentrantLock(), Ref{Csize_t}(0), Ref{Csize_t}(0), Ref{UInt8}(0x00), Ref{Csize_t}(0), false) + x = new(ssl, ssl_context, bio_read, bio_write, io, ReentrantLock(), ReentrantLock(), Ref{Csize_t}(0), Ref{Csize_t}(0), Ref{UInt8}(0x00), Ref{Csize_t}(0), false, data, ReentrantLock(), ReentrantLock()) # no close_notify from the finalizer: sending it is a socket write, which can # wait for the peer, and a finalizer may not switch tasks finalizer(x -> close(x, false), x) @@ -442,6 +487,113 @@ Base.isopen(ssl::SSLStream)::Bool = Base.@lock(ssl.lock, !ssl.closed) Base.iswritable(ssl::SSLStream)::Bool = isopen(ssl) && isopen(ssl.io) @noinline throwio(op) = throw(Base.IOError("$op requires ssl to be open", 0)) +# Writes the buffered ciphertext to the socket, with `sendlock` held by the caller, and +# releases the lock. The buffer is taken under `ssl.lock`, so it holds everything the +# calls so far produced, in order; and the lock is released under `ssl.lock` too, once +# the buffer was seen empty: a call that appends after that finds the lock free when it +# tries it (`sendreply!`), one that appended before had its bytes seen here. So a +# reader's reply never waits behind a parked writer, and never gets stuck. A closed +# stream is no exception: what `close` buffered, the close_notify, goes out this way. +function drainlocked!(ssl::SSLStream) + data = ssl.data + while true + out = Base.@lock ssl.lock begin + data.inflight = false + if isempty(data.buf) + unlock(ssl.sendlock) + nothing + else + out = data.buf + data.buf = UInt8[] + data.inflight = true + out + end + end + out === nothing && return + try + write(ssl.io, out) + catch + try + Base.@lock ssl.lock begin + data.inflight = false + data.failed = true + end + # closed before the lock is released: the next sender finds the stream + # closed rather than the dead socket + close(ssl, false) + finally + unlock(ssl.sendlock) + end + rethrow() + end + end +end + +# After one of the package's SSL calls, from a writer or the handshake: sends what it +# produced. Returns once the bytes are on the socket, or throws the error that kept them +# from getting there. Waiting for `sendlock` is waiting for the socket write of the +# records produced before this call's, which this call's records queue behind. +# Not to be called with `ssl.lock` held: the holder of `sendlock` needs that lock to +# finish, so a caller of `write` who holds it would wait for a task that waits for it. +function sendout!(ssl::SSLStream, op::Symbol) + data = ssl.data + lock(ssl.sendlock) + drainlocked!(ssl) + # the lock was free: whoever held it sent this call's bytes with its own, or they + # were dropped when the stream closed, or a socket write of them failed + data.failed && throwio(op) + return +end + +# After one of the package's SSL calls from a reader (`unsafe_read`, `eof`): a reply +# OpenSSL made to the peer, if any, goes out from a task, so that the read does not wait +# behind a writer parked on the socket. If that writer holds `sendlock`, it sends the +# reply before it releases the lock (see `drainlocked!`); only a free lock is taken. +# The task is spawned, not `@async`ed: `@async` pins the calling task, the reader, to its +# thread for good (Julia 1.7+), and the reply does not care which thread sends it. +function sendreply!(ssl::SSLStream) + (Base.@lock ssl.lock isempty(ssl.data.buf)) && return + Threads.@spawn try + trylock(ssl.sendlock) && drainlocked!(ssl) + catch e + e isa Base.IOError || rethrow() + end + return +end + +# Closes the socket handle outright, as `close` does for a socket that never connected, +# rather than shutting it down: the libuv write requests pending on it are cancelled, so +# a write parked on a peer that stopped reading fails instead of waiting forever. +# Goes by the handle, not `io.status`: Base marks the socket closing itself when the +# peer's FIN arrives (Julia 1.6, 1.7) or after an earlier `close(io)`, and in that state +# the handle still waits for its write queue to drain, which a parked write never does. +function forceclose!(io::TCPSocket) + Base.iolock_begin() + try + handle = io.handle + if handle != C_NULL && io.status != Base.StatusClosed && + ccall(:uv_is_closing, Cint, (Ptr{Cvoid},), handle) == 0 + @static if Sys.iswindows() + # libuv's close cancels only the pending writes the calling thread + # issued (`CancelIo`), and waits for the rest before it closes the + # socket: a write parked by a task on another thread would keep it open. + # Cancel every thread's I/O on the socket, once the handle is closing + sock = Ref{Ptr{Cvoid}}(C_NULL) + gotsock = ccall(:uv_fileno, Cint, (Ptr{Cvoid}, Ptr{Ptr{Cvoid}}), handle, sock) == 0 + end + ccall(:jl_forceclose_uv, Cvoid, (Ptr{Cvoid},), handle) + io.status = Base.StatusClosing + @static if Sys.iswindows() + gotsock && ccall((:CancelIoEx, "kernel32"), stdcall, Cint, + (Ptr{Cvoid}, Ptr{Cvoid}), sock[], C_NULL) + end + end + finally + Base.iolock_end() + end + return +end + # this is a macro, but should be a function, but closures are stupid slow # we use this to standardize the error handling for all of the SSL_*_ex functions macro geterror(ssl, op, expr) @@ -453,8 +605,15 @@ macro geterror(ssl, op, expr) $ssl.closed && throwio($op) # clear the current error queue before openssl ccall clear_errors!() - # do the ccall - _ret = $expr + # do the ccall; what it writes is buffered meanwhile (see `BIOStreamData`), + # the caller sends it once the lock is released + _data = $ssl.data + _data.incall = true + _ret = try + $expr + finally + _data.incall = false + end # we want to return one of our SSL return codes, regardless of error # SSL_peek_ex, SSL_write_ex, SSL_connect, and SSL_read_ex all return 1 on success if _ret == 1 @@ -494,29 +653,40 @@ macro geterror(ssl, op, expr) end) end +# what goes to `SSL_write_ex` in one call: bounds the ciphertext buffered under +# `ssl.lock` before it is sent +const SSL_WRITE_CHUNK = UInt(64 * 1024) + function Base.unsafe_write(ssl::SSLStream, in_buffer::Ptr{UInt8}, in_length::UInt) nwritten = 0 # per call, not per stream: the count is read back after `@geterror` released # `ssl.lock`, and another task's `SSL_write_ex` could overwrite a shared one first writebytes = Ref{Csize_t}(0) - while nwritten < in_length - ret = @geterror ssl :unsafe_write ccall( - (:SSL_write_ex, libssl), - Cint, - (SSL, Ptr{Cvoid}, Cint, Ptr{Csize_t}), - ssl.ssl, - in_buffer, - in_length, - writebytes - ) - if ret == SSL_ERROR_NONE - nwritten += writebytes[] - elseif ret == SSL_ERROR_WANT_WRITE - flush(ssl.io) - elseif ret == SSL_ERROR_WANT_READ - # this means write is waiting for more data from the underlying socket - # so call eof on the socket to wait for more bytes to come in - eof(ssl.io) && throw(EOFError()) + # one writer at a time, so that a write arrives whole however many tasks write + Base.@lock ssl.writelock begin + while nwritten < in_length + # without SSL_MODE_ENABLE_PARTIAL_WRITE a call writes all or nothing, so a + # retry after WANT_READ/WANT_WRITE resubmits the same chunk + chunk = min(in_length - nwritten, SSL_WRITE_CHUNK) + ret = @geterror ssl :unsafe_write ccall( + (:SSL_write_ex, libssl), + Cint, + (SSL, Ptr{Cvoid}, Csize_t, Ptr{Csize_t}), + ssl.ssl, + in_buffer + nwritten, + chunk, + writebytes + ) + sendout!(ssl, :write) + if ret == SSL_ERROR_NONE + nwritten += writebytes[] + elseif ret == SSL_ERROR_WANT_WRITE + flush(ssl.io) + elseif ret == SSL_ERROR_WANT_READ + # this means write is waiting for more data from the underlying socket + # so call eof on the socket to wait for more bytes to come in + eof(ssl.io) && throw(EOFError()) + end end end return Base.bitcast(Int, in_length) @@ -596,6 +766,7 @@ end function Sockets.connect(ssl::SSLStream; require_ssl_verification::Bool=true) while true ret = @geterror ssl :connect ssl_connect(ssl.ssl) + sendout!(ssl, :connect) if ret == SSL_ERROR_NONE break elseif ret == SSL_ERROR_WANT_READ @@ -688,6 +859,7 @@ function Base.unsafe_read(ssl::SSLStream, buf::Ptr{UInt8}, nbytes::UInt) nbytes - nread, readbytes ) + sendreply!(ssl) if ret == SSL_ERROR_NONE nread += Base.bitcast(Int, readbytes[]) elseif ret == SSL_ERROR_WANT_READ @@ -778,6 +950,7 @@ function Base.eof(ssl::SSLStream)::Bool 1, peekbytes ) + sendreply!(ssl) if ret == SSL_ERROR_NONE return false elseif ret == SSL_ERROR_WANT_WRITE @@ -806,15 +979,35 @@ end """ Close SSL stream. """ +# how long `close` lets the socket write of what is left to send (an alert from a +# failed call, the close_notify) take before it closes the socket outright: a peer that +# stopped reading would otherwise hold the stream's `sendlock`, and the socket, open +# forever +const CLOSE_NOTIFY_GRACE = 5.0 + function Base.close(ssl::SSLStream, shutdown::Bool=true) - Base.@lock ssl.lock begin + data = ssl.data + inflight = Base.@lock ssl.lock begin ssl.closed && return ssl.closed = true - if shutdown + if data.inflight + # a socket write of taken ciphertext is in progress, and may be parked on a + # peer that stopped reading: nothing more goes out, not even a close_notify, + # and a writer waiting on bytes still buffered is told. The write in + # progress is the sender's own to report + if !isempty(data.buf) + empty!(data.buf) + data.failed = true + end + elseif shutdown + # the close_notify is buffered, and sent below + data.incall = true try ssl_disconnect(ssl.ssl) catch err @debug "SSL disconnect failed" err + finally + data.incall = false end # the close_notify's socket write failing (the peer is gone, or the socket # already closed) fails SSL_shutdown, and OpenSSL 3.2+ records that on the @@ -822,11 +1015,38 @@ function Base.close(ssl::SSLStream, shutdown::Bool=true) clear_errors!() end free(ssl.ssl) + data.inflight end - @async try - Base.close(ssl.io) - catch e - e isa Base.IOError || rethrow() + if inflight + # the parked write could stay so forever: close the socket handle outright, + # which fails it + forceclose!(ssl.io) + else + # what is buffered (the alert a failed call produced, the close_notify), then + # the socket close, from a task: `close` does not wait on the socket. Nor does + # the task wait forever: a peer that has stopped reading gets the socket closed + # outright instead. Spawned rather than `@async`ed, which would pin the closing + # task to its thread (see `sendreply!`) + Threads.@spawn begin + # the timer covers the socket close as well: Base's close waits for the + # socket's pending writes to leave, which a parked raw write never does + grace = Timer(_ -> forceclose!(ssl.io), CLOSE_NOTIFY_GRACE) + try + try + lock(ssl.sendlock) + drainlocked!(ssl) + catch e + e isa Base.IOError || rethrow() + end + try + Base.close(ssl.io) + catch e + e isa Base.IOError || rethrow() + end + finally + close(grace) + end + end end end diff --git a/test/runtests.jl b/test/runtests.jl index f3f16eb..ef945a5 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -68,6 +68,85 @@ function connected_pair(server_ctx, server) return client, fetch(server_task) end +# A finished task's value, or the exception it failed with (not the TaskFailedException +# `fetch` wraps it in), for the caller to assert on. +taskoutcome(task) = istaskfailed(task) ? task.result : fetch(task) + +# `task`'s outcome should it finish within `timeout`, else `placeholder`; nothing waits +# on a task that stalled. +boundedfetch(task, placeholder; timeout=30.0) = + timedwait(() -> istaskdone(task), timeout) === :ok ? taskoutcome(task) : placeholder + +# Bytes handed to libuv that have not left for the kernel yet. +function writequeuesize(io::TCPSocket) + Base.iolock_begin() + try + io.handle == C_NULL && return 0 + return Int(ccall(:uv_stream_get_write_queue_size, Csize_t, (Ptr{Cvoid},), io.handle)) + finally + Base.iolock_end() + end +end + +# Writes to `io` (the stream, or its socket for a raw write) until the peer's receive +# window is full and the writer parks on the socket; returns the task and the count of +# completed writes. Each write is many times what goes +# to OpenSSL in one call, and more than the kernel's send and receive buffers grow to +# (Windows and macOS keep growing them for a peer that does not read): a parked write +# that the buffers could still take would complete of itself. +const PARK_CHUNK = max(32 * 2^20, 3 * Int(OpenSSL.SSL_WRITE_CHUNK)) +function park_writer(client, stop_writing; io=client) + written = Threads.Atomic{Int}(0) + chunk = zeros(UInt8, PARK_CHUNK) + writer = @async try + while !stop_writing[] + write(io, chunk) + Threads.atomic_add!(written, length(chunk)) + end + nothing + catch ex + ex + end + # parked means: bytes in libuv that have not left for the kernel, the same amount + # half a second later, and no write completed meanwhile + parked = timedwait(60.0; pollint=0.5) do + before = (written[], writequeuesize(client.io)) + sleep(0.5) + !istaskdone(writer) && before[2] > 0 && (written[], writequeuesize(client.io)) == before + end + if parked !== :ok + outcome = istaskdone(writer) ? fetch(writer) : "still writing" + close(client, false) + error("the writer did not park on the socket: $outcome") + end + return writer, written +end + +# Whether every run of one writer's byte in `received` is whole writes of that writer long. +function wholewrites(received, chunklen) + i = 1 + while i <= length(received) + j = i + while j <= length(received) && received[j] == received[i] + j += 1 + end + (j - i) % chunklen(Int(received[i])) == 0 || return false + i = j + end + return true +end + +# A client stream connecting to `port` on a task, for a server that drives its side of +# the handshake itself. +function connecting(port) + @async begin + client = OpenSSL.SSLStream(OpenSSL.SSLContext(OpenSSL.TLSClientMethod(), ""), + Sockets.connect(ip"127.0.0.1", port)) + Sockets.connect(client; require_ssl_verification=false) + client + end +end + # Verifies calling into OpenSSL library. @testset "OpenSSL" begin @test OpenSSL.BIO_STREAM_CALLBACKS.x.on_bio_create_ptr != C_NULL @@ -895,9 +974,9 @@ end close(client.io) err = @test_throws Base.IOError write(client, UInt8[1, 2, 3]) @test !isopen(client) - # the message carries OpenSSL's account of the failure, and nothing is left of it on - # the thread's error queue for an unrelated call to report - @test occursin("SSL_ERROR_SYSCALL", sprint(showerror, err.value)) + # the failure is the socket write's own, made after the SSL call: `write` reports it + # and nothing is left on the thread's error queue for an unrelated call to report + @test occursin("closed", sprint(showerror, err.value)) @test ccall((:ERR_peek_error, OpenSSL.libcrypto), Culong, ()) == 0 close(ssl, false) close(server) @@ -951,10 +1030,11 @@ end @testset "ConcurrentWriters" begin # Several tasks writing to one stream, each with its own chunk size: the byte count # `SSL_write_ex` reports has to be the one of this task's call, not of whichever - # task ran last on the stream, or a writer resubmits bytes already sent. + # task ran last on the stream, or a writer resubmits bytes already sent. Every size + # is more than what goes to OpenSSL in one call: a write still arrives in one piece. nwriters = 4 - nchunks = 64 - chunklen(id) = id * 32 * 1024 + nchunks = 16 + chunklen(id) = Int(OpenSSL.SSL_WRITE_CHUNK) + id * 32 * 1024 total = sum(nchunks * chunklen(id) for id in 1:nwriters) server_ctx = selfsigned_server_ctx() @@ -984,6 +1064,8 @@ end for id in 1:nwriters @test count(==(UInt8(id)), received) == nchunks * chunklen(id) end + # and each write in one piece: a run of one writer's byte is whole writes long + @test wholewrites(received, chunklen) else @test false end @@ -991,3 +1073,428 @@ end close(ssl, false) close(server) end + +@testset "ConcurrentReadWrite" begin + # A write that is waiting for the peer must not stop reads on the same stream: + # `SSL_write_ex` and `SSL_read_ex` share `ssl.lock`, so the socket write the write + # BIO does has to happen outside it (#72). + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client, ssl = connected_pair(server_ctx, server) + payload = collect(0x01:0x40) + write(ssl, payload) + # and from here on the server reads nothing, so the client's write parks + + stop_writing = Threads.Atomic{Bool}(false) + writer, _ = park_writer(client, stop_writing) + # the write waits for the socket without holding the lock reads need + @test timedwait(() -> !islocked(client.lock), 5.0) === :ok + + buff = Vector{UInt8}(undef, length(payload)) + reader = @async read!(client, buff) + @test boundedfetch(reader, :stalled) == payload + + # the client still has megabytes parked against a peer that is not reading: its + # close cuts that write off, which fails the writer. On a task: should the close + # wait for that write (as it did before this fix) the test fails, not the suite + stop_writing[] = true + closer = @async close(client) + @test timedwait(() -> istaskdone(closer), 30.0) === :ok + istaskdone(closer) || OpenSSL.forceclose!(client.io) + @test timedwait(() -> istaskdone(writer), 30.0) === :ok + @test istaskdone(writer) && fetch(writer) isa Base.IOError + close(ssl, false) + close(server) +end + +@testset "FullDuplex" begin + # Both ends write more than the socket buffers hold while reading what the other + # writes: with the socket write under `ssl.lock` both writers park and neither reader + # can run, so nothing finishes. + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client, ssl = connected_pair(server_ctx, server) + n = 8 * 2^20 + c2s = rand(UInt8, n) + s2c = rand(UInt8, n) + tasks = [ + @async(write(client, c2s)), + @async(write(ssl, s2c)), + @async(read!(client, Vector{UInt8}(undef, n))), + @async(read!(ssl, Vector{UInt8}(undef, n))), + ] + done = timedwait(() -> all(istaskdone, tasks), 120.0) === :ok + @test done + if done + @test !any(istaskfailed, tasks) + @test fetch(tasks[3]) == s2c + @test fetch(tasks[4]) == c2s + else + # release what stalled rather than leave it to outlast the testset + OpenSSL.forceclose!(client.io) + OpenSSL.forceclose!(ssl.io) + end + close(client) + close(ssl) + close(server) +end + +@testset "CloseWithParkedWriter" begin + # `close` while a write is parked on a peer that stopped reading: the close does not + # wait for that write, which could be forever; it closes the socket outright, which + # fails the writer. + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client, ssl = connected_pair(server_ctx, server) + stop_writing = Threads.Atomic{Bool}(false) + writer, written = park_writer(client, stop_writing) + stop_writing[] = true + t0 = time() + close(client) + @test time() - t0 < 5.0 + @test !isopen(client) + @test timedwait(() -> istaskdone(writer), 30.0) === :ok + @test istaskdone(writer) && fetch(writer) isa Base.IOError + # the server reaches the end of the stream, however much of the write got through + reader = @async try + while !eof(ssl) + readavailable(ssl) + end + :eof + catch ex + ex + end + outcome = boundedfetch(reader, :stalled) + @test outcome !== :stalled + outcome === :stalled && OpenSSL.forceclose!(ssl.io) + close(ssl, false) + close(server) +end + +@testset "CloseRacingWriter" begin + # a write whose socket write completed just as `close` ran: the writer returns + # success, since its bytes did reach the socket, and the peer reads them all + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client, ssl = connected_pair(server_ctx, server) + # one read on the client first, so that the server's session tickets do not sit + # unread in its socket: closing a socket with unread data resets the connection, + # and on Windows the reset discards what the peer has not read yet + write(ssl, UInt8[0x01]) + @test read!(client, Vector{UInt8}(undef, 1)) == UInt8[0x01] + payload = rand(UInt8, 1000) + writer = @async try + write(client, payload) + :ok + catch ex + ex + end + closer = @async close(client) + @test boundedfetch(writer, :stalled) === :ok + @test boundedfetch(closer, :stalled) === nothing + reader = @async read!(ssl, Vector{UInt8}(undef, length(payload))) + @test boundedfetch(reader, :stalled) == payload + close(ssl, false) + close(server) +end + +@testset "CloseBehindRawWrite" begin + # `close` with the socket's send buffer full of a raw write (one not made through the + # stream, so nothing is in flight for `close` to cut): the close_notify's socket + # write would park behind it forever, so after a grace period the socket is closed + # outright, and the stream's `sendlock` is released + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client, ssl = connected_pair(server_ctx, server) + # written until it parks: one write, however large, may not (Windows takes a whole + # 64 MiB with no reader) + stop_writing = Threads.Atomic{Bool}(false) + raw, _ = park_writer(client, stop_writing; io=client.io) + stop_writing[] = true + t0 = time() + close(client) + @test time() - t0 < 1.0 + @test timedwait(() -> istaskdone(raw), OpenSSL.CLOSE_NOTIFY_GRACE + 30.0) === :ok + @test timedwait(() -> !isopen(client.io), 30.0) === :ok + @test timedwait(() -> !islocked(client.sendlock), 30.0) === :ok + close(ssl, false) + close(server) +end + +@testset "CloseBehindRawWriteNoShutdown" begin + # the same, for a close that makes no close_notify (`close(ssl, false)`, which the + # failed-call path and the finalizer use): nothing is buffered to send, but Base's + # own socket close waits behind the parked write, so that too is bounded + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client, ssl = connected_pair(server_ctx, server) + stop_writing = Threads.Atomic{Bool}(false) + raw, _ = park_writer(client, stop_writing; io=client.io) + stop_writing[] = true + t0 = time() + close(client, false) + @test time() - t0 < 1.0 + @test timedwait(() -> istaskdone(raw), OpenSSL.CLOSE_NOTIFY_GRACE + 30.0) === :ok + @test timedwait(() -> !isopen(client.io), 30.0) === :ok + @test timedwait(() -> !islocked(client.sendlock), 30.0) === :ok + close(ssl, false) + close(server) +end + +@testset "CloseNotifyEOF" begin + # the peer's close_notify, sent by `close`, surfaces from `eof` as the "unexpected + # EOF" `IOError` and closes the stream; and it follows the writes made before the + # close, which the peer reads first + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client, ssl = connected_pair(server_ctx, server) + # the client reads too, which takes the server's session tickets off its socket: + # closed with them unread, the kernel may reset the connection rather than end it + # (Windows does), and the peer lose the close_notify to the reset + client_reader = @async try + while !eof(client) + readavailable(client) + end + catch + end + payload = rand(UInt8, 100_000) + write(client, payload) + sleep(0.2) + close(client) + reader = @async begin + received = read!(ssl, Vector{UInt8}(undef, length(payload))) + err = try + eof(ssl) + nothing + catch ex + ex + end + (received, err) + end + outcome = boundedfetch(reader, :stalled) + @test outcome !== :stalled + if outcome !== :stalled + received, err = outcome + @test received == payload + @test err isa Base.IOError + @test occursin("unexpected EOF", sprint(showerror, err)) + @test !isopen(ssl) + else + OpenSSL.forceclose!(ssl.io) + end + @test timedwait(() -> istaskdone(client_reader), 30.0) === :ok + close(ssl) + close(server) +end + +@testset "WriteSizes" begin + # the peer reads every write whole and in order, across sizes around a record's and + # a write of many records + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client, ssl = connected_pair(server_ctx, server) + payloads = [rand(UInt8, n) for n in (1, 100, 16383, 16384, 16385, 50_000, 2^20 + 7, 3)] + expected = reduce(vcat, payloads) + reader = @async read!(ssl, Vector{UInt8}(undef, length(expected))) + for payload in payloads + write(client, payload) + end + @test boundedfetch(reader, :stalled) == expected + # nothing left buffered, and nothing kept + @test isempty(client.data.buf) + close(client) + close(ssl) + close(server) +end + +@testset "ReplyBehindSender" begin + # a reply a read produced goes out from a task when `sendlock` is free; when a sender + # holds the lock, that sender sends it before releasing the lock, so it is not lost + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client, ssl = connected_pair(server_ctx, server) + # the server's socket read raw: what the client's write BIO sent, as it is + marker1 = rand(UInt8, 16) + marker2 = rand(UInt8, 16) + # the lock is free: the reply task sends + got1 = @async read(ssl.io, 16) + Base.@lock client.lock append!(client.data.buf, marker1) + OpenSSL.sendreply!(client) + @test boundedfetch(got1, :stalled) == marker1 + # the lock is taken: the reply task leaves the reply to the holder + got2 = @async read(ssl.io, 16) + lock(client.sendlock) + Base.@lock client.lock append!(client.data.buf, marker2) + OpenSSL.sendreply!(client) + sleep(0.3) + @test !istaskdone(got2) + @test !isempty(client.data.buf) + # what every holder does before releasing the lock + OpenSSL.drainlocked!(client) + @test !islocked(client.sendlock) + @test boundedfetch(got2, :stalled) == marker2 + close(client, false) + close(ssl, false) + close(server) +end + +@testset "LegacyAccept" begin + # `accept` keeps its contract: one round, which does not wait for the peer, so the + # deadline a loop checks between calls holds against a peer that stays silent + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + silent = Sockets.connect(ip"127.0.0.1", port) + ssl = OpenSSL.SSLStream(server_ctx, accept(server)) + # the loop written to `accept`'s contract: one round per call, `OpenSSLError` when + # it needs more bytes, the wait for them and the deadline between calls + function acceptloop(ssl, deadline) + rounds = 0 + while true + try + Sockets.accept(ssl) + return :ok, rounds + catch ex + ex isa OpenSSL.OpenSSLError || rethrow() + rounds += 1 + time() < deadline || return :deadline_expired, rounds + arrival = @async eof(ssl.io) + timedwait(() -> istaskdone(arrival), max(deadline - time(), 0.01)) + end + end + end + outcome, rounds = acceptloop(ssl, time() + 0.5) + @test outcome === :deadline_expired + @test rounds >= 1 + # nothing failed: the loop could have gone on + @test isopen(ssl) + # no close_notify from a handshake in progress, which would queue an error + close(ssl, false) + close(silent) + close(server) +end + +@testset "RawSSLCalls" begin + # a call made on `ssl.ssl` from outside the package writes to the socket from the + # write BIO callback itself, as before: an `SSL_accept` loop of a caller's own (under + # `ssl.lock`, as TLSStreams runs it) completes the handshake, and a raw write lands + # whole among the records of a parked write, nothing lost + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client_task = connecting(port) + ssl = OpenSSL.SSLStream(server_ctx, accept(server)) + deadline = time() + 30.0 + while true + done = Base.@lock ssl.lock begin + try + OpenSSL.ssl_accept(ssl.ssl) + true + catch ex + ex isa OpenSSL.OpenSSLError || rethrow() + false + end + end + done && break + time() < deadline || error("the raw accept loop did not finish") + eof(ssl.io) && error("the peer went away during the raw accept loop") + end + client = fetch(client_task) + write(client, UInt8[1, 2, 3]) + @test read!(ssl, Vector{UInt8}(undef, 3)) == UInt8[1, 2, 3] + + # the server reads nothing for now, so the client's writer parks; a raw write on the + # client then goes straight to the socket, behind the record in libuv + stop_writing = Threads.Atomic{Bool}(false) + parked_writer, written = park_writer(client, stop_writing) + stop_writing[] = true + nzeros = written[] + PARK_CHUNK + marker = fill(UInt8(7), 100) + raw = @async Base.@lock client.lock begin + n = Ref{Csize_t}(0) + r = GC.@preserve marker ccall( + (:SSL_write_ex, OpenSSL.libssl), + Cint, + (OpenSSL.SSL, Ptr{Cvoid}, Csize_t, Ptr{Csize_t}), + client.ssl, pointer(marker), length(marker), n) + (r, Int(n[])) + end + reader = @async read!(ssl, Vector{UInt8}(undef, nzeros + length(marker))) + received = boundedfetch(reader, :stalled; timeout=60.0) + @test received isa Vector{UInt8} + if received isa Vector{UInt8} + @test count(==(0), received) == nzeros + sevens = findall(==(7), received) + @test length(sevens) == length(marker) + # in one piece + @test !isempty(sevens) && sevens == sevens[1]:sevens[1] + length(marker) - 1 + else + OpenSSL.forceclose!(client.io) + end + @test boundedfetch(raw, :stalled) == (1, 100) + @test boundedfetch(parked_writer, :stalled) === nothing + close(client) + close(ssl) + close(server) +end + +@testset "PeerResetMidRecord" begin + # a peer that resets the connection mid-record: `eof` reports the error or the end + # of the stream, and so does the next `eof`, rather than hang + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client, ssl = connected_pair(server_ctx, server) + reader = @async try + eof(ssl) + catch ex + ex + end + # a partial record, then a close with the session tickets the server sent still + # unread, which makes the kernel reset the connection rather than end it + sleep(0.2) + write(client.io, UInt8[0x17, 0x03, 0x03, 0x00, 0x64, 0xaa, 0xbb, 0xcc]) + close(client.io) + outcome = boundedfetch(reader, :stalled) + @test outcome isa Base.IOError || outcome === true + outcome === :stalled && OpenSSL.forceclose!(ssl.io) + again = @async try + eof(ssl) + catch ex + ex + end + outcome = boundedfetch(again, :stalled) + @test outcome isa Base.IOError || outcome === true + outcome === :stalled && OpenSSL.forceclose!(ssl.io) + close(client, false) + close(ssl, false) + close(server) +end + +@testset "FatalAlertReachesPeer" begin + # a forged record fails the server's read; the alert OpenSSL produces while failing + # is buffered, since the call runs under `ssl.lock`, and the `close` that the failure + # makes sends it, so the peer learns why instead of seeing a clean end of stream + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client, ssl = connected_pair(server_ctx, server) + forged = UInt8[0x17, 0x03, 0x03, 0x00, 0x20] + append!(forged, rand(UInt8, 0x20)) + write(client.io, forged) + server_read = @async try + eof(ssl) + catch ex + ex + end + outcome = boundedfetch(server_read, :stalled) + @test outcome isa Base.IOError + @test !isopen(ssl) + client_read = @async try + eof(client) + catch ex + ex + end + outcome = boundedfetch(client_read, :stalled) + @test outcome isa Base.IOError + @test outcome isa Base.IOError && occursin("alert", outcome.msg) + close(client, false) + close(ssl, false) + close(server) +end From 1e3281aadc5373d6ceb7449f7082008daace6ac9 Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Thu, 29 Oct 2026 08:31:04 -0600 Subject: [PATCH 2/2] Close finalized streams without starting a task Julia runs every finalizer at exit, including those of live streams. The finalizer called close(ssl, false), which starts the socket close with Threads.@spawn; another thread then JIT-compiles that task while C exit() tears down LLVM, and the process segfaults or aborts. The finalizer now frees the SSL object and closes the socket handle itself, and leaves a stream whose lock is held to a later collection. close's own task is @async again, as on main: a close just before exit then does not have another thread start it. Both spawned tasks get an errormonitor. On Windows, forceclose! cancels the socket's I/O before the handle is closed, while the handle is still valid. close's docstring is back on the method. Co-Authored-By: Claude Opus 5.5 --- src/ssl.jl | 60 ++++++++++++++++++++++++++++++++---------------- test/runtests.jl | 22 ++++++++++++++++-- 2 files changed, 60 insertions(+), 22 deletions(-) diff --git a/src/ssl.jl b/src/ssl.jl index 837f6c4..e1f0bd6 100644 --- a/src/ssl.jl +++ b/src/ssl.jl @@ -470,9 +470,7 @@ mutable struct SSLStream <: IO bio_write::BIO = BIO(data; finalize=false) ssl = SSL(ssl_context, bio_read, bio_write) x = new(ssl, ssl_context, bio_read, bio_write, io, ReentrantLock(), ReentrantLock(), Ref{Csize_t}(0), Ref{Csize_t}(0), Ref{UInt8}(0x00), Ref{Csize_t}(0), false, data, ReentrantLock(), ReentrantLock()) - # no close_notify from the finalizer: sending it is a socket write, which can - # wait for the peer, and a finalizer may not switch tasks - finalizer(x -> close(x, false), x) + finalizer(finalize_stream, x) return x end end @@ -553,11 +551,11 @@ end # thread for good (Julia 1.7+), and the reply does not care which thread sends it. function sendreply!(ssl::SSLStream) (Base.@lock ssl.lock isempty(ssl.data.buf)) && return - Threads.@spawn try + errormonitor(Threads.@spawn try trylock(ssl.sendlock) && drainlocked!(ssl) catch e e isa Base.IOError || rethrow() - end + end) return end @@ -575,18 +573,17 @@ function forceclose!(io::TCPSocket) ccall(:uv_is_closing, Cint, (Ptr{Cvoid},), handle) == 0 @static if Sys.iswindows() # libuv's close cancels only the pending writes the calling thread - # issued (`CancelIo`), and waits for the rest before it closes the - # socket: a write parked by a task on another thread would keep it open. - # Cancel every thread's I/O on the socket, once the handle is closing + # issued (`CancelIo`): a write parked by a task on another thread would + # keep the socket open. Cancel every thread's I/O on the socket first, + # while the handle is still valid: the close may release it at once sock = Ref{Ptr{Cvoid}}(C_NULL) - gotsock = ccall(:uv_fileno, Cint, (Ptr{Cvoid}, Ptr{Ptr{Cvoid}}), handle, sock) == 0 + if ccall(:uv_fileno, Cint, (Ptr{Cvoid}, Ptr{Ptr{Cvoid}}), handle, sock) == 0 + ccall((:CancelIoEx, "kernel32"), stdcall, Cint, + (Ptr{Cvoid}, Ptr{Cvoid}), sock[], C_NULL) + end end ccall(:jl_forceclose_uv, Cvoid, (Ptr{Cvoid},), handle) io.status = Base.StatusClosing - @static if Sys.iswindows() - gotsock && ccall((:CancelIoEx, "kernel32"), stdcall, Cint, - (Ptr{Cvoid}, Ptr{Cvoid}), sock[], C_NULL) - end end finally Base.iolock_end() @@ -594,6 +591,27 @@ function forceclose!(io::TCPSocket) return end +# The finalizer neither waits nor starts a task, so no close_notify: Julia runs every +# finalizer at exit, also those of live streams, and a task started then can crash the +# process. It frees the SSL object and closes the socket handle outright. A stream +# whose lock is held is finalized at a later collection instead (the manual's pattern +# for finalizers that need a lock). +function finalize_stream(ssl::SSLStream) + if islocked(ssl.lock) || !trylock(ssl.lock) + finalizer(finalize_stream, ssl) + return + end + try + ssl.closed && return + ssl.closed = true + free(ssl.ssl) + finally + unlock(ssl.lock) + end + forceclose!(ssl.io) + return +end + # this is a macro, but should be a function, but closures are stupid slow # we use this to standardize the error handling for all of the SSL_*_ex functions macro geterror(ssl, op, expr) @@ -976,15 +994,15 @@ function Base.eof(ssl::SSLStream)::Bool return !isopen(ssl) end -""" - Close SSL stream. -""" # how long `close` lets the socket write of what is left to send (an alert from a # failed call, the close_notify) take before it closes the socket outright: a peer that # stopped reading would otherwise hold the stream's `sendlock`, and the socket, open # forever const CLOSE_NOTIFY_GRACE = 5.0 +""" + Close SSL stream. +""" function Base.close(ssl::SSLStream, shutdown::Bool=true) data = ssl.data inflight = Base.@lock ssl.lock begin @@ -1025,9 +1043,10 @@ function Base.close(ssl::SSLStream, shutdown::Bool=true) # what is buffered (the alert a failed call produced, the close_notify), then # the socket close, from a task: `close` does not wait on the socket. Nor does # the task wait forever: a peer that has stopped reading gets the socket closed - # outright instead. Spawned rather than `@async`ed, which would pin the closing - # task to its thread (see `sendreply!`) - Threads.@spawn begin + # outright instead. `@async`, unlike `Threads.@spawn`, keeps the task on this + # thread, so a `close` just before `exit` does not have another thread start + # it while the process goes down + errormonitor(@async begin # the timer covers the socket close as well: Base's close waits for the # socket's pending writes to leave, which a parked raw write never does grace = Timer(_ -> forceclose!(ssl.io), CLOSE_NOTIFY_GRACE) @@ -1046,8 +1065,9 @@ function Base.close(ssl::SSLStream, shutdown::Bool=true) finally close(grace) end - end + end) end + return end """ diff --git a/test/runtests.jl b/test/runtests.jl index ef945a5..16740e6 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1223,8 +1223,8 @@ end @testset "CloseBehindRawWriteNoShutdown" begin # the same, for a close that makes no close_notify (`close(ssl, false)`, which the - # failed-call path and the finalizer use): nothing is buffered to send, but Base's - # own socket close waits behind the parked write, so that too is bounded + # failed-call path uses): nothing is buffered to send, but Base's own socket close + # waits behind the parked write, so that too is bounded server_ctx = selfsigned_server_ctx() port, server = Sockets.listenany(ip"127.0.0.1", 20000) client, ssl = connected_pair(server_ctx, server) @@ -1241,6 +1241,24 @@ end close(server) end +@testset "FinalizerClosesInPlace" begin + # the finalizer closes the socket itself, without starting a task (Julia runs + # finalizers at exit, when a task another thread starts can crash the process), and + # leaves a stream whose lock is held to a later collection rather than wait for it + server_ctx = selfsigned_server_ctx() + port, server = Sockets.listenany(ip"127.0.0.1", 20000) + client, ssl = connected_pair(server_ctx, server) + lock(client.lock) + finalize(client) + @test isopen(client) + unlock(client.lock) + finalize(client) + @test !isopen(client) + @test client.io.status in (Base.StatusClosing, Base.StatusClosed) + close(ssl, false) + close(server) +end + @testset "CloseNotifyEOF" begin # the peer's close_notify, sent by `close`, surfaces from `eof` as the "unexpected # EOF" `IOError` and closes the stream; and it follows the writes made before the