dot upstreams: per-upstream tls_name for sni and cert verification by dns name

src/platform/tls_client.zig

@@ -233,6 +233,18 @@ pub const TlsStream = struct {
         return &self.client.writer;
     }
 
+    /// Pushes buffered plaintext all the way to the socket.
+    ///
+    /// Both flushes are required. `tls.Client.flush` only encrypts what the
+    /// plaintext writer holds into the socket writer's buffer and calls
+    /// `advance` (crypto/tls/Client.zig:999); it never flushes that writer, so
+    /// on its own it leaves the record sitting in this process. A caller that
+    /// then waits for a reply waits until the peer gives up.
+    pub fn flush(self: *TlsStream) std.Io.Writer.Error!void {
+        try self.client.writer.flush();
+        try self.stream_writer.interface.flush();
+    }
+
     /// Sends close_notify and flushes the socket. Does not close the underlying
     /// stream; the caller owns it.
     pub fn close(self: *TlsStream) void {

src/platform/tls_client_integration_test.zig

@@ -108,3 +108,120 @@ test "live DoT handshake against cloudflare-dns.com" {
         },
     }
 }
+
+// ---------------------------------------------------------------------------
+// Hermetic: loopback only, no name resolution, no external host.
+// ---------------------------------------------------------------------------
+
+const tls_server = @import("tls_server.zig");
+
+/// The fixture certificate carries `DNS:localhost`, so this is the only name a
+/// `TlsStream` can verify against it.
+const fixture_host = "localhost";
+
+const echo_message = "nxdns tls flush probe";
+
+/// Reads one message and echoes it back.
+fn echoOnce(
+    gpa: std.mem.Allocator,
+    ctx: *tls_server.ServerContext,
+    io: std.Io,
+    server: *net.Server,
+) anyerror!void {
+    var stream = try server.accept(io);
+    defer stream.close(io);
+
+    var read_buffer: [4096]u8 = undefined;
+    var write_buffer: [4096]u8 = undefined;
+    var server_stream: tls_server.ServerStream = undefined;
+    try server_stream.accept(gpa, ctx, io, &stream, &read_buffer, &write_buffer);
+    defer server_stream.close(gpa);
+
+    var received: [echo_message.len]u8 = undefined;
+    try server_stream.reader().readSliceAll(&received);
+    try server_stream.writer().writeAll(&received);
+    try server_stream.writer().flush();
+}
+
+/// Writes through `TlsStream.flush` and waits for the echo. With only
+/// `client.writer.flush()` the record would never leave this process and this
+/// read would block until the budget expired.
+fn flushAndEcho(io: std.Io, gpa: std.mem.Allocator, address: net.IpAddress) anyerror!void {
+    var stream = try address.connect(io, .{ .mode = .stream });
+    defer stream.close(io);
+
+    var bundle: Certificate.Bundle = .empty;
+    defer bundle.deinit(gpa);
+    var bundle_lock: std.Io.RwLock = .init;
+
+    var read_buffer: [4096]u8 = undefined;
+    var write_buffer: [4096]u8 = undefined;
+    var stream_read_buffer: [tls.Client.min_buffer_len]u8 = undefined;
+    var stream_write_buffer: [tls.Client.min_buffer_len]u8 = undefined;
+
+    var client: tls_client.TlsStream = undefined;
+    // The fixture certificate is self-signed, so the chain cannot verify; the
+    // host name still must match, which is what this test needs it to do.
+    try client.init(io, &stream, &bundle, &bundle_lock, gpa, .{
+        .host = fixture_host,
+        .ca = .insecure_skip_verify,
+        .read_buffer = &read_buffer,
+        .write_buffer = &write_buffer,
+        .stream_read_buffer = &stream_read_buffer,
+        .stream_write_buffer = &stream_write_buffer,
+    });
+    defer client.close();
+
+    try client.writer().writeAll(echo_message);
+    try client.flush();
+
+    var echoed: [echo_message.len]u8 = undefined;
+    try client.reader().readSliceAll(&echoed);
+    try std.testing.expectEqualStrings(echo_message, &echoed);
+}
+
+test "TlsStream.flush puts the record on the wire" {
+    if (!build_options.integration) return error.SkipZigTest;
+
+    const fixtures = @import("test_fixtures");
+    const gpa = std.testing.allocator;
+
+    var threaded: std.Io.Threaded = .init(gpa, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var ctx = try tls_server.ServerContext.init(gpa, fixtures.cert_pem, fixtures.key_pem);
+    defer ctx.deinit(gpa);
+
+    const listen_address: net.IpAddress = .{ .ip4 = .loopback(0) };
+    var server = try listen_address.listen(io, .{ .reuse_address = true });
+    defer server.deinit(io);
+
+    var server_task = try io.concurrent(echoOnce, .{ gpa, &ctx, io, &server });
+
+    // Raced against the budget: the failure this test guards against is a
+    // record that never leaves the process, which without a deadline would
+    // hang the run instead of failing it.
+    var outcomes: [2]Outcome = undefined;
+    var race: std.Io.Select(Outcome) = .init(io, &outcomes);
+    defer race.cancelDiscard();
+    try race.concurrent(.exchange, flushAndEcho, .{ io, gpa, server.socket.address });
+    try race.concurrent(.expiry, expire, .{ io, budget });
+
+    const client_result: anyerror!void = switch (try race.await()) {
+        .exchange => |result| result,
+        .expiry => |result| blk: {
+            try result;
+            break :blk error.TlsEchoTimedOut;
+        },
+    };
+
+    // A client that never connects would leave the server blocked in `accept`.
+    const server_result = if (client_result) |_|
+        server_task.await(io)
+    else |_|
+        server_task.cancel(io);
+
+    try client_result;
+    try server_result;
+}