diff --git a/src/ssl.jl b/src/ssl.jl index 8a8da9e..e1f0bd6 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,16 +455,22 @@ 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) - # 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) + 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()) + finalizer(finalize_stream, x) return x end end @@ -442,6 +485,133 @@ 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 + errormonitor(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`): 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) + 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 + end + finally + Base.iolock_end() + end + 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) @@ -453,8 +623,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 +671,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 +784,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 +877,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 +968,7 @@ function Base.eof(ssl::SSLStream)::Bool 1, peekbytes ) + sendreply!(ssl) if ret == SSL_ERROR_NONE return false elseif ret == SSL_ERROR_WANT_WRITE @@ -803,18 +994,38 @@ function Base.eof(ssl::SSLStream)::Bool return !isopen(ssl) end +# 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) - 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,12 +1033,41 @@ 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. `@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) + 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 + return end """ diff --git a/test/runtests.jl b/test/runtests.jl index f3f16eb..16740e6 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,446 @@ 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 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) + 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 "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 + # 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