diff --git a/Sources/PerfectSMTP/LocalMTATransport.swift b/Sources/PerfectSMTP/LocalMTATransport.swift index a21718a..d16c0fa 100644 --- a/Sources/PerfectSMTP/LocalMTATransport.swift +++ b/Sources/PerfectSMTP/LocalMTATransport.swift @@ -211,13 +211,8 @@ public actor LocalMTATransport: SMTPTransport { // bridges the whole loop onto a dedicated background // thread instead, off the cooperative pool entirely. let collected = try await Self.runBlocking { - let handle = stderrPipe.fileHandleForReading var collected: [UInt8] = [] - while true { - let chunk = handle.availableData - if chunk.isEmpty { break } - collected.append(contentsOf: chunk) - } + try Self.readToEnd(stderrPipe.fileHandleForReading) { collected.append(contentsOf: $0) } return collected } stderrBox.value = collected @@ -229,11 +224,7 @@ public actor LocalMTATransport: SMTPTransport { // same way an undrained stderr would. Same off-cooperative- // pool bridging as the stderr drain above (FIX #7). _ = try await Self.runBlocking { - let handle = stdoutPipe.fileHandleForReading - while true { - let chunk = handle.availableData - if chunk.isEmpty { break } - } + try Self.readToEnd(stdoutPipe.fileHandleForReading) { _ in } } } @@ -251,6 +242,31 @@ public actor LocalMTATransport: SMTPTransport { } } + /// Reads `handle` to end of file, passing each chunk to `body`. + /// + /// 6.0.1: this used `FileHandle.availableData`, which corelibs + /// Foundation turns into a `fatalError` on any read error, EINTR + /// included, so a signal interrupting the read crashed the process + /// (the test suite hit it on emulated linux/amd64). An interrupted + /// read is now retried, and any other error is thrown. + static func readToEnd(_ handle: FileHandle, _ body: (UnsafeRawBufferPointer) -> Void) throws { + let fd = handle.fileDescriptor + var buffer = [UInt8](repeating: 0, count: 64 * 1024) + while true { + let (count, error) = buffer.withUnsafeMutableBytes { bytes in + let count = read(fd, bytes.baseAddress, bytes.count) + return (count, count < 0 ? errno : 0) + } + if count > 0 { + buffer.withUnsafeBytes { body(UnsafeRawBufferPointer(rebasing: $0.prefix(count))) } + } else if count == 0 { + return + } else if error != EINTR { + throw POSIXError(POSIXErrorCode(rawValue: error) ?? .EIO) + } + } + } + /// Bridges a blocking, synchronous closure onto a dedicated background /// thread (`DispatchQueue.global(qos: .utility)`), never the Swift /// Concurrency cooperative thread pool -- FIX #7's mechanism (see the diff --git a/Sources/PerfectSMTP/MTASTS/NIOMTASTSHTTPFetcher.swift b/Sources/PerfectSMTP/MTASTS/NIOMTASTSHTTPFetcher.swift index 34e36e0..85820e2 100644 --- a/Sources/PerfectSMTP/MTASTS/NIOMTASTSHTTPFetcher.swift +++ b/Sources/PerfectSMTP/MTASTS/NIOMTASTSHTTPFetcher.swift @@ -37,11 +37,20 @@ import NIOTLS /// - Redirects are never followed (RFC 8461 ยง3.3): a 3xx is returned as /// is, and `MTASTSPolicyManager` treats any non-200 as a failed fetch. /// - `timeout` bounds the whole fetch, resolution to last byte -/// (`MTASTSHTTPFetchError.timedOut`). +/// (`MTASTSHTTPFetchError.timedOut`), and cancelling the calling task +/// ends it at once (`CancellationError`); either way the connection is +/// closed. /// - At most `maximumBodySize` body bytes are read. A larger /// `Content-Length` is refused before the body, and the connection is /// closed as soon as the cap is passed /// (`MTASTSHTTPFetchError.responseTooLarge`). +/// - Once a fetch is done (or capped, timed out, refused) it stops reading +/// and closes without waiting for the server's TLS close_notify +/// (`tlsConfiguration.shutdownTimeout` is overridden). The response is +/// still accepted only when complete: by `Content-Length`, the chunked +/// terminator, or, for a body delimited by the close, the server's +/// close_notify (over `http://`, which is for tests and local tooling, +/// a bare TCP close ends such a body). public struct NIOMTASTSHTTPFetcher: MTASTSHTTPFetching { private let group: any EventLoopGroup private let timeout: TimeInterval @@ -80,106 +89,204 @@ public struct NIOMTASTSHTTPFetcher: MTASTSHTTPFetching { let host = rawHost.hasPrefix("[") && rawHost.hasSuffix("]") ? String(rawHost.dropFirst().dropLast()) : rawHost let useTLS = scheme == "https" let port = url.port ?? (useTLS ? 443 : 80) - let basePath = url.path.isEmpty ? "/" : url.path - let path = url.query.map { basePath + "?" + $0 } ?? basePath + // 6.0.1: the request target is the URL's own percent-encoded path + // and query. `url.path` is percent-decoded and drops a trailing + // slash, so `/a%20b/` went out as `/a b` (a malformed request line) + // and `/%3F` as `/?`. The policy manager's fixed + // `/.well-known/mta-sts.txt` was never affected; direct callers were. + guard let path = Self.requestTarget(url) else { throw FetchError.unsupportedURL(url.absoluteString) } let literal = Self.addressLiteral(host) // Host header: name or address, IPv6 in brackets, port if explicit. let hostField = (literal.map { if case .v6 = $0 { return "[\(host)]" } else { return host } } ?? host) + (url.port.map { ":\($0)" } ?? "") let limit = maximumBodySize - let sslContext = useTLS ? try NIOSSLContext(configuration: tlsConfiguration) : nil - let group = self.group + // 6.0.1: when we close (done, capped, timed out, refused), don't + // wait for the peer's close_notify. NIOSSL's clean shutdown keeps + // reading, and discarding, whatever the peer sends for up to + // `shutdownTimeout` (5 s by default): against an endless body that + // ignores our close_notify, 5 s of full-speed reading per capped + // fetch. This only affects our own close, after the response has + // been accepted or refused. + var tls = tlsConfiguration + tls.shutdownTimeout = .milliseconds(1) + let sslContext = useTLS ? try NIOSSLContext(configuration: tls) : nil let connectTimeout = Self.timeAmount(timeout) let resolver = addressResolver let allowPrivate = allowPrivateAddresses - let promise = group.next().makePromise(of: MTASTSHTTPResponse.self) + // One event loop for the whole fetch: its state, every promise and + // every attempt's channel live there. + let loop = group.next() + let promise = loop.makePromise(of: MTASTSHTTPResponse.self) + let abortSignal = loop.makePromise(of: Void.self) + let state = NIOLoopBoundBox.makeBoxSendingValue(FetchState(), as: FetchState.self, eventLoop: loop) - return try await Self.withDeadline(timeout, onTimeout: { promise.fail(MTASTSHTTPFetchError.timedOut) }) { - // Resolve once and vet; the vetted address is what gets dialed. - let resolved: [DNSAddress] - if let literal { - resolved = [literal] - } else { - resolved = try await resolver.resolveAddresses(hostname: host) + // 6.0.1 (review): the deadline used to fail only the response + // promise, so a server that accepted and never finished the TLS + // handshake held the fetch (and its fd) past the deadline, and a + // cancelled caller waited for the server. Now the deadline and + // cancellation abort every stage (resolution, connect, handshake, + // response), and the channel, even one that connects later, is + // closed. + let abort: @Sendable (any Error) -> Void = { error in + loop.execute { + if state.value.aborted == nil { state.value.aborted = error } + abortSignal.fail(error) + promise.fail(error) + if let channel = state.value.channel { Self.abandon(channel) } } - let permitted = allowPrivate ? resolved : resolved.filter(\.isRoutable) - guard !permitted.isEmpty else { - promise.fail(MTASTSHTTPFetchError.noPermittedAddress(host: host)) - throw MTASTSHTTPFetchError.noPermittedAddress(host: host) - } - // One event loop for every attempt, so each attempt's readiness - // promise is created up front and held here rather than looked up - // in a pipeline that may already have been torn down. - let loop = group.next() - func bootstrap(ready: EventLoopPromise) -> ClientBootstrap { - ClientBootstrap(group: loop) - .connectTimeout(connectTimeout) - .channelInitializer { channel in - channel.eventLoop.makeCompletedFuture { - if let sslContext { - let tls = try NIOSSLClientHandler(context: sslContext, serverHostname: literal == nil ? host : nil) - try channel.pipeline.syncOperations.addHandler(tls) - try channel.pipeline.syncOperations.addHandler( - TLSReadyGate(host: literal == nil ? host : nil, ready: ready) - ) + } + let deadline = loop.scheduleTask(in: connectTimeout) { abort(MTASTSHTTPFetchError.timedOut) } + defer { + deadline.cancel() + // Every promise is completed on every path (a no-op if it + // already was): a leaked promise traps in debug builds. + abortSignal.succeed(()) + promise.fail(FetchError.incompleteResponse) + } + let aborted = abortSignal.futureResult + func abortReason() async -> (any Error)? { + try? await loop.submit { state.value.aborted }.get() + } + + return try await withTaskCancellationHandler { + do { + // Resolve once and vet; the vetted address is what gets dialed. + let resolved: [DNSAddress] + if let literal { + resolved = [literal] + } else { + let resolving = loop.makePromise(of: [DNSAddress].self) + let task = Task { + do { resolving.succeed(try await resolver.resolveAddresses(hostname: host)) } catch { resolving.fail(error) } + } + defer { task.cancel() } + resolved = try await Self.unlessAborted(resolving.futureResult, aborted) + } + let permitted = allowPrivate ? resolved : resolved.filter(\.isRoutable) + guard !permitted.isEmpty else { throw MTASTSHTTPFetchError.noPermittedAddress(host: host) } + + func bootstrap(ready: EventLoopPromise) -> ClientBootstrap { + ClientBootstrap(group: loop) + .connectTimeout(connectTimeout) + .channelInitializer { channel in + channel.eventLoop.makeCompletedFuture { + // Recorded before the connect starts (on the + // fetch's loop), so an abort also closes a + // connect still in flight, failing it at once. + if let error = state.value.aborted { throw error } + state.value.channel = channel + if let sslContext { + let tls = try NIOSSLClientHandler(context: sslContext, serverHostname: literal == nil ? host : nil) + try channel.pipeline.syncOperations.addHandler(tls) + try channel.pipeline.syncOperations.addHandler( + TLSReadyGate(host: literal == nil ? host : nil, ready: ready) + ) + } + try channel.pipeline.syncOperations.addHTTPClientHandlers() + try channel.pipeline.syncOperations.addHandler(CappedResponseCollector(limit: limit, promise: promise)) } - try channel.pipeline.syncOperations.addHTTPClientHandlers() - try channel.pipeline.syncOperations.addHandler(CappedResponseCollector(limit: limit, promise: promise)) } + } + var connected: (channel: any Channel, ready: EventLoopPromise)? + var lastError: any Error = MTASTSHTTPFetchError.noPermittedAddress(host: host) + for address in permitted { + // Each attempt's readiness promise is created up front and + // held here rather than looked up in a pipeline that may + // already have been torn down. + let ready = loop.makePromise(of: Void.self) + let attempt = loop.makePromise(of: (any Channel).self) + do { + // A channel that connects after an abort is closed; + // otherwise it is recorded, so an abort closes it. + bootstrap(ready: ready).connect(to: try SocketAddress(ipAddress: address.description, port: port)) + .hop(to: loop).whenComplete { result in + switch result { + case .success(let channel): + if let error = state.value.aborted { + Self.abandon(channel) + attempt.fail(error) + } else { + state.value.channel = channel + attempt.succeed(channel) + } + case .failure(let error): + attempt.fail(error) + } + } + connected = (try await Self.unlessAborted(attempt.futureResult, aborted), ready) + break + } catch { + ready.fail(error) // no-op if the gate already failed it + attempt.fail(error) + if let reason = await abortReason() { throw reason } + lastError = error } - } - var connected: (channel: any Channel, ready: EventLoopPromise)? - var lastError: any Error = MTASTSHTTPFetchError.noPermittedAddress(host: host) - for address in permitted { - let ready = loop.makePromise(of: Void.self) - do { - let channel = try await bootstrap(ready: ready) - .connect(to: SocketAddress(ipAddress: address.description, port: port)).get() - connected = (channel, ready) - break - } catch { - ready.fail(error) // no-op if the gate already failed it - lastError = error } - } - guard let (channel, ready) = connected else { - promise.fail(lastError) - throw lastError - } - // Re-review: nothing is written until the handshake has - // completed and the certificate's DNS name has been checked, so - // a refused server never sees the request. - if useTLS { - do { - try await ready.futureResult.get() - } catch { - channel.close(promise: nil) - promise.fail(error) - throw error + guard let (channel, ready) = connected else { throw lastError } + // Re-review: nothing is written until the handshake has + // completed and the certificate's DNS name has been checked, + // so a refused server never sees the request. + if useTLS { + try await Self.unlessAborted(ready.futureResult, aborted) + } else { + ready.succeed(()) } - } else { - ready.succeed(()) - } - var headers = HTTPHeaders() - headers.add(name: "Host", value: hostField) - headers.add(name: "User-Agent", value: "PerfectSMTP-MTA-STS") - headers.add(name: "Accept", value: "text/plain") - headers.add(name: "Connection", value: "close") - let head = HTTPRequestHead(version: .http1_1, method: .GET, uri: path, headers: headers) - channel.write(HTTPClientRequestPart.head(head), promise: nil) - channel.writeAndFlush(HTTPClientRequestPart.end(nil), promise: nil) - do { + var headers = HTTPHeaders() + headers.add(name: "Host", value: hostField) + headers.add(name: "User-Agent", value: "PerfectSMTP-MTA-STS") + headers.add(name: "Accept", value: "text/plain") + headers.add(name: "Connection", value: "close") + let head = HTTPRequestHead(version: .http1_1, method: .GET, uri: path, headers: headers) + // Written on the loop, and only if nothing has aborted the fetch. + try await loop.submit { + if let error = state.value.aborted { throw error } + channel.write(HTTPClientRequestPart.head(head), promise: nil) + channel.writeAndFlush(HTTPClientRequestPart.end(nil), promise: nil) + }.get() let response = try await promise.futureResult.get() - channel.close(promise: nil) + Self.abandon(channel) return response } catch { - channel.close(promise: nil) - throw error + let reason = await abortReason() ?? error + abort(reason) // fails what is still pending and closes the channel + throw reason } + } onCancel: { + abort(CancellationError()) + } + } + + /// `future`'s result, or the abort's error if the fetch is aborted first. + private static func unlessAborted(_ future: EventLoopFuture, _ aborted: EventLoopFuture) async throws -> T { + let either = aborted.eventLoop.makePromise(of: T.self) + future.hop(to: aborted.eventLoop).whenComplete { either.completeWith($0) } + aborted.whenFailure { either.fail($0) } + return try await either.futureResult.get() + } + + /// Stops reading, then closes: no new reads are started while the + /// close (and the 1 ms TLS shutdown) completes. + static func abandon(_ channel: any Channel) { + channel.eventLoop.execute { + try? channel.syncOptions?.setOption(.autoRead, value: false) + channel.close(promise: nil) } } + /// The origin-form request target: the URL's percent-encoded path + /// (empty becomes `/`) and query, fragment dropped. Nil if the URL has + /// no components or the result holds a space, a control character or a + /// non-ASCII byte (defence in depth: NIO doesn't check the request URI, + /// and older Foundations differ on encoding such characters). + static func requestTarget(_ url: URL) -> String? { + guard let components = URLComponents(url: url, resolvingAgainstBaseURL: false) else { return nil } + let path = components.percentEncodedPath.isEmpty ? "/" : components.percentEncodedPath + let target = components.percentEncodedQuery.map { path + "?" + $0 } ?? path + guard target.utf8.allSatisfy({ $0 > 0x20 && $0 < 0x7F }) else { return nil } + return target + } + /// `host` as an address if it is an IP literal. static func addressLiteral(_ host: String) -> DNSAddress? { guard let address = try? SocketAddress(ipAddress: host, port: 0) else { return nil } @@ -196,22 +303,12 @@ public struct NIOMTASTSHTTPFetcher: MTASTSHTTPFetching { private static func timeAmount(_ seconds: TimeInterval) -> TimeAmount { .nanoseconds(Int64(min(max(0, seconds), 86_400) * 1_000_000_000)) } +} - private static func withDeadline( - _ seconds: TimeInterval, onTimeout: @escaping @Sendable () -> Void, _ operation: @escaping @Sendable () async throws -> T - ) async throws -> T { - try await withThrowingTaskGroup(of: T.self) { race in - race.addTask { try await operation() } - race.addTask { - try await Task.sleep(nanoseconds: UInt64(min(max(0, seconds), 86_400) * 1_000_000_000)) - onTimeout() - throw MTASTSHTTPFetchError.timedOut - } - defer { race.cancelAll() } - guard let result = try await race.next() else { throw MTASTSHTTPFetchError.timedOut } - return result - } - } +/// Per-fetch state, confined to the fetch's event loop. +private struct FetchState: Sendable { + var channel: (any Channel)? + var aborted: (any Error)? } /// Completes `ready` once the TLS handshake is done and, for a host name, @@ -236,6 +333,7 @@ private final class TLSReadyGate: ChannelInboundHandler, @unchecked Sendable { } catch { ready.fail(error) context.fireErrorCaught(error) + try? context.channel.syncOptions?.setOption(.autoRead, value: false) context.close(promise: nil) return } @@ -261,8 +359,10 @@ private final class TLSReadyGate: ChannelInboundHandler, @unchecked Sendable { /// Collects one HTTP response, failing (and closing) as soon as the body /// would pass `limit`. Event-loop confined: all callbacks run on the /// channel's loop, which is why `@unchecked Sendable` holds. -private final class CappedResponseCollector: ChannelInboundHandler, @unchecked Sendable { +private final class CappedResponseCollector: ChannelDuplexHandler, @unchecked Sendable { typealias InboundIn = HTTPClientResponsePart + typealias OutboundIn = HTTPClientRequestPart + typealias OutboundOut = HTTPClientRequestPart private let limit: Int private let promise: EventLoopPromise @@ -310,11 +410,27 @@ private final class CappedResponseCollector: ChannelInboundHandler, @unchecked S finish(context, .failure(error)) } + /// 6.0.1 (review): any close that passes here before the response is + /// complete fails it first. A close that ends through NIOSSL's shutdown + /// timer (now 1 ms) leaves no `uncleanShutdown` error behind, so the + /// decoder would otherwise end a close-delimited body as if the server + /// had. Every close of ours already settles the response first; this + /// covers any other (say, the event-loop group shutting down). + func close(context: ChannelHandlerContext, mode: CloseMode, promise closePromise: EventLoopPromise?) { + if !done { + done = true + body = [] + promise.fail(NIOMTASTSHTTPFetcher.FetchError.incompleteResponse) + } + context.close(mode: mode, promise: closePromise) + } + private func finish(_ context: ChannelHandlerContext, _ result: Result) { guard !done else { return } done = true body = [] promise.completeWith(result) + try? context.channel.syncOptions?.setOption(.autoRead, value: false) context.close(promise: nil) } } diff --git a/Tests/PerfectSMTPTests/LocalMTATransportReadTests.swift b/Tests/PerfectSMTPTests/LocalMTATransportReadTests.swift new file mode 100644 index 0000000..176c0c3 --- /dev/null +++ b/Tests/PerfectSMTPTests/LocalMTATransportReadTests.swift @@ -0,0 +1,67 @@ +// +// LocalMTATransportReadTests.swift +// PerfectSMTPTests +// +// 6.0.1: the sendmail pipes were drained with `FileHandle.availableData`, +// which corelibs Foundation turns into a `fatalError` on any read error, +// EINTR included ("Interrupted system call", FileHandle.swift:231). The +// full suite crashed that way on emulated linux/amd64. Reads are now +// retried when interrupted. +// + +import Foundation +import Testing +@testable import PerfectSMTP + +struct LocalMTATransportReadTests { + /// The reading thread's pthread, handed over once it's running. + private final class ReaderThread: @unchecked Sendable { + let started = DispatchSemaphore(value: 0) + let finished = DispatchSemaphore(value: 0) + var thread: pthread_t? + var collected: [UInt8] = [] + var error: (any Error)? + } + + @Test(.timeLimit(.minutes(1))) + func anInterruptedReadIsRetried() throws { + // A handler without SA_RESTART, so a blocked read() returns EINTR. + // (`siginterrupt` is deprecated in glibc; `signal` sets SA_RESTART + // there, so the flag is cleared through sigaction.) + signal(SIGUSR1) { _ in } + var action = sigaction() + sigaction(SIGUSR1, nil, &action) + action.sa_flags &= ~SA_RESTART + sigaction(SIGUSR1, &action, nil) + let pipe = Pipe() + let reader = ReaderThread() + let handle = pipe.fileHandleForReading + Thread { + // The test runner's threads may start with SIGUSR1 blocked + // (then nothing would be interrupted); make sure this one doesn't. + var unblock = sigset_t() + sigemptyset(&unblock) + sigaddset(&unblock, SIGUSR1) + pthread_sigmask(SIG_UNBLOCK, &unblock, nil) + reader.thread = pthread_self() + reader.started.signal() + do { + try LocalMTATransport.readToEnd(handle) { reader.collected.append(contentsOf: $0) } + } catch { + reader.error = error + } + reader.finished.signal() + }.start() + reader.started.wait() + let thread = try #require(reader.thread) + for _ in 0..<5 { + usleep(20_000) + pthread_kill(thread, SIGUSR1) + } + try pipe.fileHandleForWriting.write(contentsOf: Data("hello".utf8)) + try pipe.fileHandleForWriting.close() + #expect(reader.finished.wait(timeout: .now() + 10) == .success) + #expect(reader.error == nil) + #expect(String(decoding: reader.collected, as: UTF8.self) == "hello") + } +} diff --git a/Tests/PerfectSMTPTests/MTASTS/MTASTSHTTPFetcherTLSShutdownTests.swift b/Tests/PerfectSMTPTests/MTASTS/MTASTSHTTPFetcherTLSShutdownTests.swift new file mode 100644 index 0000000..a175ccd --- /dev/null +++ b/Tests/PerfectSMTPTests/MTASTS/MTASTSHTTPFetcherTLSShutdownTests.swift @@ -0,0 +1,517 @@ +// +// MTASTSHTTPFetcherTLSShutdownTests.swift +// PerfectSMTPTests +// +// 6.0.1: closing a NIOSSL channel starts a clean TLS shutdown that keeps +// reading, and discarding, until the peer's close_notify or NIOSSL's +// `shutdownTimeout` (5 s by default). A policy host that streams an +// endless body and never answers the client's close_notify kept every +// capped (or timed-out, or refused) fetch's connection reading at full +// speed for 5 s: 50 capped fetches on macOS took 15-25 s and moved +// 40-60 GB. The fetcher now stops reading and closes after 1 ms. +// +// The second half pins down that this never lets a cut-off response +// through: a short Content-Length body, an unterminated chunked body, or +// a close-delimited body without the server's close_notify is refused. +// Runs against local TLS servers on 127.0.0.1. +// + +import Foundation +import NIOConcurrencyHelpers +import NIOCore +import NIOHTTP1 +import NIOPosix +import NIOSSL +import NIOTLS +import Testing +@testable import PerfectSMTP + +@Suite(.serialized) +struct MTASTSHTTPFetcherTLSShutdownTests { + // MARK: - The connection is let go at once + + /// The server ignores the client's close_notify and keeps pumping. The + /// connection must be gone well before NIOSSL's 5 s default. + @Test(.timeLimit(.minutes(1)), arguments: [Framing.closeDelimited, .contentLength(1_000_000_000_000), .chunked]) + func aCappedFetchLetsGoOfADeafEndlessServerAtOnce(framing: Framing) async throws { + let server = try await ScriptedTLSServer.start(.init(framing: framing, body: .endless, deafAfter: .request)) + let returned = try await server.expectFetch(throws: MTASTSHTTPFetchError.responseTooLarge(limit: MTASTSPolicyManager.maximumPolicyResponseSizeBytes)) + try await server.expectClosedSoon(after: returned) + try await server.stop() + } + + @Test(.timeLimit(.minutes(1))) + func aTimedOutFetchLetsGoOfADeafServerAtOnce() async throws { + // Headers and a little body, then silence: the fetch runs into its + // own deadline. + let server = try await ScriptedTLSServer.start(.init(framing: .contentLength(1000), body: .bytes(10), deafAfter: .request)) + let returned = try await server.expectFetch(throws: MTASTSHTTPFetchError.timedOut, timeout: 0.5) + try await server.expectClosedSoon(after: returned) + try await server.stop() + } + + @Test(.timeLimit(.minutes(1))) + func aRefusedCertificateLetsGoOfADeafServerAtOnce() async throws { + // An IP-only certificate for a host name: refused after the + // handshake. TLS 1.2, so the server has finished its handshake (and + // gone deaf) before the client's close_notify can arrive; under 1.3 + // it can come in the same read as the client's Finished and be + // answered. + let server = try await ScriptedTLSServer.start(.init( + leaf: .ipOnly, maximumTLSVersion: .tlsv12, framing: .closeDelimited, body: .bytes(0), deafAfter: .handshake + )) + let returned = try await server.expectFetch(throws: CertificateNameError.noDNSName(host: "mta-sts.victim.test"), host: "mta-sts.victim.test") + try await server.expectClosedSoon(after: returned) + try await server.stop() + } + + // MARK: - The deadline and cancellation end every stage + + /// Review: the deadline used to fail only the response, so a server + /// that accepted and never spoke held the fetch in the TLS handshake + /// until the server closed. + @Test(.timeLimit(.minutes(1))) + func theDeadlineEndsAStalledHandshake() async throws { + let server = try await SilentTCPServer.start() + let fetcher = try server.fetcher(timeout: 0.5) + let started = Date() + await #expect(throws: MTASTSHTTPFetchError.timedOut) { _ = try await fetcher.fetch(url: try server.url()) } + let returned = Date() + #expect(returned.timeIntervalSince(started) < 3) + try await server.expectClosedSoon(after: returned) + try await server.stop() + } + + /// Review: cancelling the caller cancelled the deadline too, and the + /// fetch then waited for the server indefinitely. + @Test(.timeLimit(.minutes(1)), arguments: [CancelPoint.handshake, .response]) + func cancellingTheCallerEndsTheFetch(at point: CancelPoint) async throws { + let server: any ClosableTestServer = switch point { + case .handshake: try await SilentTCPServer.start() + case .response: try await ScriptedTLSServer.start(.init(framing: .contentLength(1000), body: .bytes(10))) + } + let fetcher = try server.fetcher(timeout: 30) + let url = try server.url() + let fetch = Task { try await fetcher.fetch(url: url) } + try await Task.sleep(nanoseconds: 300_000_000) + let cancelled = Date() + fetch.cancel() + await #expect(throws: CancellationError.self) { _ = try await fetch.value } + #expect(Date().timeIntervalSince(cancelled) < 2) + try await server.expectClosedSoon(after: cancelled) + try await server.stop() + } + + /// Review: a resolver error left the response promise uncompleted, + /// which traps ("leaking promise") in debug builds. + @Test(.timeLimit(.minutes(1))) + func aResolverErrorIsThrown() async throws { + struct Failing: MTASTSAddressResolving { + struct Failure: Error, Equatable {} + func resolveAddresses(hostname: String) async throws -> [DNSAddress] { throw Failure() } + } + let group = MultiThreadedEventLoopGroup(numberOfThreads: 1) + let fetcher = NIOMTASTSHTTPFetcher(group: group, timeout: 10, addressResolver: Failing()) + await #expect(throws: Failing.Failure()) { + _ = try await fetcher.fetch(url: try #require(URL(string: "https://mta-sts.victim.test/.well-known/mta-sts.txt"))) + } + try await group.shutdownGracefully() + } + + enum CancelPoint: Sendable { case handshake, response } + + // MARK: - A cut-off response is never accepted + + @Test(.timeLimit(.minutes(1)), arguments: [ScriptedTLSServer.Ending.closeNotify, .tcpCloseWithoutCloseNotify]) + func aShortContentLengthBodyIsRefused(ending: ScriptedTLSServer.Ending) async throws { + let server = try await ScriptedTLSServer.start(.init(framing: .contentLength(Self.policy.utf8.count + 10), body: .policy, ending: ending)) + try await server.expectNoResponse(ending) + try await server.stop() + } + + @Test(.timeLimit(.minutes(1)), arguments: [ScriptedTLSServer.Ending.closeNotify, .tcpCloseWithoutCloseNotify]) + func anUnterminatedChunkedBodyIsRefused(ending: ScriptedTLSServer.Ending) async throws { + let server = try await ScriptedTLSServer.start(.init(framing: .chunked, body: .policy, ending: ending)) + try await server.expectNoResponse(ending) + try await server.stop() + } + + /// A body delimited by the close is complete only if the close is the + /// server's: an attacker on the path can end the TCP stream, but can't + /// forge close_notify. + @Test(.timeLimit(.minutes(1))) + func aCloseDelimitedBodyWithoutCloseNotifyIsRefused() async throws { + let server = try await ScriptedTLSServer.start(.init(framing: .closeDelimited, body: .policy, ending: .tcpCloseWithoutCloseNotify)) + try await server.expectNoResponse(.tcpCloseWithoutCloseNotify) + try await server.stop() + } + + @Test(.timeLimit(.minutes(1))) + func aCloseDelimitedBodyEndedByCloseNotifyIsAccepted() async throws { + let server = try await ScriptedTLSServer.start(.init(framing: .closeDelimited, body: .policy, ending: .closeNotify)) + let response = try await server.fetcher().fetch(url: server.url()) + try await server.stop() + #expect(response.statusCode == 200) + #expect(String(decoding: response.body, as: UTF8.self) == Self.policy) + } + + static let policy = "version: STSv1\r\nmode: enforce\r\nmx: mx.victim.test\r\nmax_age: 86400\r\n" + + enum Framing: Sendable, CustomTestStringConvertible { + case closeDelimited, contentLength(Int), chunked + var testDescription: String { + switch self { + case .closeDelimited: "close-delimited" + case .contentLength: "Content-Length" + case .chunked: "chunked" + } + } + } +} + +/// A TLS server on its own thread (an endless pump on the client's loop +/// would compete with the client) that answers the first request by +/// script and records when the connection closed. +final class ScriptedTLSServer: ClosableTestServer { + enum Body: Sendable { case policy, bytes(Int), endless } + enum DeafPoint: Sendable { case handshake, request } + enum Ending: Sendable { case closeNotify, tcpCloseWithoutCloseNotify } + + struct Script: Sendable { + var leaf: STARTTLSCertificateTests.Leaf = .named + var maximumTLSVersion: TLSVersion? = nil + var framing: MTASTSHTTPFetcherTLSShutdownTests.Framing + var body: Body + /// From then on the server reads nothing, so it never sees (or + /// answers) the client's close_notify. + var deafAfter: DeafPoint? = nil + /// How the server ends after writing a finite body; nil leaves it open. + var ending: Ending? = nil + } + + private let group: MultiThreadedEventLoopGroup + private let channel: any Channel + private let clientGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1) + private let closedAt: NIOLockedValueBox + + private init(group: MultiThreadedEventLoopGroup, channel: any Channel, closedAt: NIOLockedValueBox) { + self.group = group + self.channel = channel + self.closedAt = closedAt + } + + static func start(_ script: Script) async throws -> ScriptedTLSServer { + let (certPEM, keyPEM): (String, String) + switch script.leaf { + case .ipOnly: (certPEM, keyPEM) = (TestCertificates.ipOnlyCertificatePEM, TestCertificates.ipOnlyKeyPEM) + case .named: (certPEM, keyPEM) = (TestCertificates.namedCertificatePEM, TestCertificates.namedKeyPEM) + case .dnsOnly: (certPEM, keyPEM) = (TestCertificates.dnsOnlyCertificatePEM, TestCertificates.dnsOnlyKeyPEM) + } + var configuration = TLSConfiguration.makeServerConfiguration( + certificateChain: [.certificate(try NIOSSLCertificate(bytes: Array(certPEM.utf8), format: .pem))], + privateKey: .privateKey(try NIOSSLPrivateKey(bytes: Array(keyPEM.utf8), format: .pem)) + ) + // The server's own clean close (`.closeNotify`) needn't linger. + configuration.shutdownTimeout = .milliseconds(100) + configuration.maximumTLSVersion = script.maximumTLSVersion + let context = try NIOSSLContext(configuration: configuration) + let group = MultiThreadedEventLoopGroup(numberOfThreads: 1) + let closedAt = NIOLockedValueBox(nil) + let channel = try await ServerBootstrap(group: group) + .serverChannelOption(.socketOption(.so_reuseaddr), value: 1) + .childChannelInitializer { channel in + channel.eventLoop.makeCompletedFuture { + try channel.pipeline.syncOperations.addHandler(SocketEnd(closedAt: closedAt)) + try channel.pipeline.syncOperations.addHandler(NIOSSLServerHandler(context: context)) + try channel.pipeline.syncOperations.addHandler(ScriptedResponder(script: script)) + } + } + .bind(host: "127.0.0.1", port: 0).get() + return ScriptedTLSServer(group: group, channel: channel, closedAt: closedAt) + } + + func url(host: String) throws -> URL { + try #require(URL(string: "https://\(host):\(channel.localAddress?.port ?? 0)/.well-known/mta-sts.txt")) + } + + static func trustingTestCA() throws -> TLSConfiguration { + var configuration = TLSConfiguration.makeClientConfiguration() + configuration.trustRoots = .certificates([try NIOSSLCertificate(bytes: Array(TestCertificates.caPEM.utf8), format: .pem)]) + return configuration + } + + func fetcher(timeout: TimeInterval = 10) throws -> NIOMTASTSHTTPFetcher { + NIOMTASTSHTTPFetcher( + group: clientGroup, timeout: timeout, tlsConfiguration: try Self.trustingTestCA(), + addressResolver: LoopbackResolver(), allowPrivateAddresses: true + ) + } + + /// Fetches, expects `expected`, and returns when the fetch returned. + func expectFetch(throws expected: E, timeout: TimeInterval = 10, host: String = "127.0.0.1") async throws -> Date { + let fetcher = try fetcher(timeout: timeout) + let url = try url(host: host) + await #expect(throws: expected) { _ = try await fetcher.fetch(url: url) } + return Date() + } + + /// A cut-off response must fail the fetch, for the reason the ending + /// gives: the HTTP framing is incomplete at the server's close_notify, + /// or the TCP stream ended without one. + func expectNoResponse(_ ending: Ending) async throws { + let fetcher = try fetcher() + let url = try url() + do { + let response = try await fetcher.fetch(url: url) + Issue.record("a cut-off response was accepted: status \(response.statusCode), \(response.body.count) body bytes") + } catch { + switch ending { + case .closeNotify: #expect(error as? HTTPParserError == .invalidEOFState, "\(error)") + case .tcpCloseWithoutCloseNotify: #expect(error as? NIOSSLError == .uncleanShutdown, "\(error)") + } + } + } + + /// The server sees the connection close well before NIOSSL's 5 s + /// default shutdown timeout (2 s leaves room for an emulated machine). + func expectClosedSoon(after returned: Date) async throws { + try await Self.expectClosedSoon(closedAt, after: returned) + } + + static func expectClosedSoon(_ closedAt: NIOLockedValueBox, after returned: Date) async throws { + let deadline = Date().addingTimeInterval(10) + while closedAt.withLockedValue({ $0 }) == nil, Date() < deadline { + try await Task.sleep(nanoseconds: 10_000_000) + } + let closed = try #require(closedAt.withLockedValue { $0 }, "the client never closed the connection") + #expect(closed.timeIntervalSince(returned) < 2, "the connection stayed open \(closed.timeIntervalSince(returned)) s after the fetch returned") + } + + func stop() async throws { + try await channel.close() + try await group.shutdownGracefully() + try await clientGroup.shutdownGracefully() + } +} + +/// A test server whose connection close the tests wait for. +protocol ClosableTestServer: Sendable { + func url(host: String) throws -> URL + func fetcher(timeout: TimeInterval) throws -> NIOMTASTSHTTPFetcher + func expectClosedSoon(after returned: Date) async throws + func stop() async throws +} + +extension ClosableTestServer { + func url() throws -> URL { try url(host: "127.0.0.1") } +} + +/// Accepts TCP connections and never says anything (so a TLS handshake +/// never completes); records when the connection closed. +final class SilentTCPServer: ClosableTestServer { + private let group = MultiThreadedEventLoopGroup(numberOfThreads: 1) + private let clientGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1) + private let closedAt = NIOLockedValueBox(nil) + private let channel: NIOLockedValueBox<(any Channel)?> = NIOLockedValueBox(nil) + + static func start() async throws -> SilentTCPServer { + let server = SilentTCPServer() + let closedAt = server.closedAt + let channel = try await ServerBootstrap(group: server.group) + .serverChannelOption(.socketOption(.so_reuseaddr), value: 1) + .childChannelInitializer { channel in + channel.eventLoop.makeCompletedFuture { + try channel.pipeline.syncOperations.addHandler(SocketEnd(closedAt: closedAt)) + } + } + .bind(host: "127.0.0.1", port: 0).get() + server.channel.withLockedValue { $0 = channel } + return server + } + + func url(host: String) throws -> URL { + let port = channel.withLockedValue { $0?.localAddress?.port } ?? 0 + return try #require(URL(string: "https://\(host):\(port)/.well-known/mta-sts.txt")) + } + + func fetcher(timeout: TimeInterval) throws -> NIOMTASTSHTTPFetcher { + NIOMTASTSHTTPFetcher( + group: clientGroup, timeout: timeout, tlsConfiguration: try ScriptedTLSServer.trustingTestCA(), + addressResolver: LoopbackResolver(), allowPrivateAddresses: true + ) + } + + func expectClosedSoon(after returned: Date) async throws { + try await ScriptedTLSServer.expectClosedSoon(closedAt, after: returned) + } + + func stop() async throws { + try await channel.withLockedValue { $0 }?.close() + try await group.shutdownGracefully() + try await clientGroup.shutdownGracefully() + } +} + +private struct LoopbackResolver: MTASTSAddressResolving { + func resolveAddresses(hostname: String) async throws -> [DNSAddress] { [.v4([127, 0, 0, 1])] } +} + +/// Socket side of the server's TLS handler: records the close, can go +/// deaf (drop everything inbound), and can close the TCP connection +/// without NIOSSL sending close_notify. Event-loop confined. +final class SocketEnd: ChannelInboundHandler, @unchecked Sendable { + typealias InboundIn = ByteBuffer + private let closedAt: NIOLockedValueBox + var deaf = false + + init(closedAt: NIOLockedValueBox) { self.closedAt = closedAt } + + func channelRead(context: ChannelHandlerContext, data: NIOAny) { + if !deaf { context.fireChannelRead(data) } + } + + func channelInactive(context: ChannelHandlerContext) { + closedAt.withLockedValue { $0 = $0 ?? Date() } + context.fireChannelInactive() + } +} + +/// Answers the first request by script. Event-loop confined. +private final class ScriptedResponder: ChannelInboundHandler, @unchecked Sendable { + typealias InboundIn = ByteBuffer + typealias OutboundOut = ByteBuffer + private let script: ScriptedTLSServer.Script + private var started = false + private let filler = ByteBuffer(repeating: UInt8(ascii: "#"), count: 16 * 1024) + + init(script: ScriptedTLSServer.Script) { self.script = script } + + func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) { + if case .handshakeCompleted? = event as? TLSUserEvent, script.deafAfter == .handshake { goDeaf(context) } + context.fireUserInboundEventTriggered(event) + } + + func channelRead(context: ChannelHandlerContext, data: NIOAny) { + guard !started else { return } + started = true + if script.deafAfter == .request { goDeaf(context) } + var head = "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nConnection: close\r\n" + switch script.framing { + case .closeDelimited: break + case .contentLength(let length): head += "Content-Length: \(length)\r\n" + case .chunked: head += "Transfer-Encoding: chunked\r\n" + } + var out = ByteBuffer(string: head + "\r\n") + switch script.body { + case .policy: out.writeImmutableBuffer(encoded(ByteBuffer(string: MTASTSHTTPFetcherTLSShutdownTests.policy))) + case .bytes(let count): out.writeImmutableBuffer(encoded(ByteBuffer(repeating: UInt8(ascii: "#"), count: count))) + case .endless: break + } + let channel = context.channel + let ending = script.ending + // Closing from SocketEnd's context goes straight to the socket, past NIOSSL. + let socketEnd = NIOLoopBound(try? context.pipeline.syncOperations.context(handlerType: SocketEnd.self), eventLoop: context.eventLoop) + context.writeAndFlush(Self.wrapOutboundOut(out)).whenComplete { _ in + switch ending { + case .closeNotify?: channel.close(promise: nil) + case .tcpCloseWithoutCloseNotify?: socketEnd.value?.close(promise: nil) + case nil: break + } + } + if case .endless = script.body { pump(context) } + } + + func channelWritabilityChanged(context: ChannelHandlerContext) { + if started, case .endless = script.body { pump(context) } + context.fireChannelWritabilityChanged() + } + + func errorCaught(context: ChannelHandlerContext, error: any Error) { context.close(promise: nil) } + + private func pump(_ context: ChannelHandlerContext) { + while context.channel.isActive, context.channel.isWritable { + context.writeAndFlush(Self.wrapOutboundOut(encoded(filler)), promise: nil) + } + } + + private func encoded(_ bytes: ByteBuffer) -> ByteBuffer { + guard case .chunked = script.framing else { return bytes } + var chunk = ByteBuffer(string: String(bytes.readableBytes, radix: 16) + "\r\n") + chunk.writeImmutableBuffer(bytes) + chunk.writeString("\r\n") + return chunk + } + + private func goDeaf(_ context: ChannelHandlerContext) { + try? context.pipeline.syncOperations.handler(type: SocketEnd.self).deaf = true + } +} + +/// 6.0.1: the request target is sent exactly as the URL has it. It used to +/// come from `url.path`, which is percent-decoded and loses a trailing +/// slash. Not reachable through `MTASTSPolicyManager` (its path is fixed), +/// only by calling the fetcher directly. +@Suite(.serialized) +struct NIOMTASTSHTTPFetcherRequestTargetTests { + @Test(.timeLimit(.minutes(1)), arguments: [ + ("/.well-known/mta-sts.txt", "/.well-known/mta-sts.txt"), + ("", "/"), + ("/api/", "/api/"), + ("/%3F/?", "/%3F/?"), + ("/a%20b/%2F/caf%C3%A9?x=%26y&z", "/a%20b/%2F/caf%C3%A9?x=%26y&z"), + ("/a;b;c=d/e;f?x;y", "/a;b;c=d/e;f?x;y"), + ("/p?#fragment", "/p?"), + ]) + func theRequestTargetIsSentAsEncoded(path: String, expected: String) async throws { + let group = MultiThreadedEventLoopGroup(numberOfThreads: 1) + let target = NIOLockedValueBox(nil) + let server = try await ServerBootstrap(group: group) + .serverChannelOption(.socketOption(.so_reuseaddr), value: 1) + .childChannelInitializer { channel in + channel.eventLoop.makeCompletedFuture { + try channel.pipeline.syncOperations.addHandler(RequestLineRecorder(target: target)) + } + } + .bind(host: "127.0.0.1", port: 0).get() + let port = try #require(server.localAddress?.port) + let fetcher = NIOMTASTSHTTPFetcher(group: group, timeout: 10, allowPrivateAddresses: true) + let response = try await fetcher.fetch(url: try #require(URL(string: "http://127.0.0.1:\(port)\(path)"))) + try await server.close() + try await group.shutdownGracefully() + #expect(response.statusCode == 200) + #expect(target.withLockedValue { $0 } == expected) + } +} + +extension NIOMTASTSHTTPFetcherRequestTargetTests { + @Test func userinfoAndFragmentsNeverReachTheTarget() throws { + let url = try #require(URL(string: "https://user:p%0D%0Ass@mta-sts.example.com/x%20y?q#f%0D%0A")) + #expect(NIOMTASTSHTTPFetcher.requestTarget(url) == "/x%20y?q") + } +} + +/// Records the request target from the request line and answers 200. +/// Event-loop confined. +private final class RequestLineRecorder: ChannelInboundHandler, @unchecked Sendable { + typealias InboundIn = ByteBuffer + typealias OutboundOut = ByteBuffer + private let target: NIOLockedValueBox + private var received = "" + private var answered = false + + init(target: NIOLockedValueBox) { self.target = target } + + func channelRead(context: ChannelHandlerContext, data: NIOAny) { + guard !answered else { return } + received += String(buffer: Self.unwrapInboundIn(data)) + guard let line = received.components(separatedBy: "\r\n").first, received.contains("\r\n\r\n") else { return } + answered = true + let parts = line.split(separator: " ") + target.withLockedValue { $0 = parts.count == 3 ? String(parts[1]) : "malformed: \(line)" } + let channel = context.channel + let response = "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + context.writeAndFlush(Self.wrapOutboundOut(ByteBuffer(string: response))).whenComplete { _ in channel.close(promise: nil) } + } + + func errorCaught(context: ChannelHandlerContext, error: any Error) { context.close(promise: nil) } +}