Add streaming message bodies to client and server · jcollie.dev/zig-smtp@554c3ed

+20 -3

README.md

··· 30 30 the server's actual code and text. `mailFrom`/`rcptTo`/`sendMessage` are also 31 31 available individually. 32 32 33 + Message bodies can also be streamed instead of passed as a slice — from any 34 + reader via `sendMessageReader(&reader)`, or push-style via `data()`, which 35 + returns a writer that dot-stuffs and normalizes line endings as content 36 + flows through it: 37 + 38 + ```zig 39 + var data_writer = try client.data(); 40 + try data_writer.interface.print("Subject: report {d}\r\n\r\n", .{id}); 41 + // ... stream as much as needed ... 42 + try data_writer.end(); // terminates the message, reads the verdict 43 + ``` 44 + 33 45 ### Authentication 34 46 35 47 `hello` reports the server's advertised mechanisms in `extensions.auth`; ··· 100 112 PLAIN and AUTH LOGIN (RFC 4954); setting `Options.require_auth` rejects MAIL 101 113 with 530 until the client has authenticated. 102 114 115 + Instead of `message` (which collects the whole body in memory, bounded by 116 + `max_message_size`), a handler can set `messageReader` to stream it: the 117 + callback receives an `Io.Reader` yielding the unstuffed message content, 118 + and anything left unread is drained by the session. 119 + 103 120 `run` serves one connection until QUIT or disconnect, enforcing command 104 121 sequencing, recipient and message-size limits, and un-stuffing message data. 105 122 Listening, accepting, and concurrency are up to the caller. ··· 150 167 [ianic/tls.zig](https://github.com/ianic/tls.zig): the client does implicit 151 168 TLS and STARTTLS via `zsmtp.Tls`, and the server accepts STARTTLS (TLS 1.3 152 169 only). AUTH covers PLAIN, LOGIN, and CRAM-MD5 on the client and PLAIN and 153 - LOGIN on the server. Not yet implemented: implicit TLS on the server side, 154 - streaming (non-slice) message bodies, and ESMTP parameter handling (SIZE=, 155 - BODY=) on the server side. 170 + LOGIN on the server. Message bodies can be streamed on both sides. Not yet 171 + implemented: implicit TLS on the server side, and ESMTP parameter handling 172 + (SIZE=, BODY=) on the server side. 156 173 157 174 ## Tests 158 175

+172 -7

src/Client.zig

View file

··· 255 255 256 256 /// Sends the message content for the current transaction (DATA). Line 257 257 /// endings in `data` are normalized to CRLF and leading dots are stuffed. 258 - pub fn sendMessage(c: *Client, data: []const u8) Error!void { 258 + pub fn sendMessage(c: *Client, message_data: []const u8) Error!void { 259 + var data_writer = try c.data(); 260 + try data_writer.interface.writeAll(message_data); 261 + try data_writer.end(); 262 + } 263 + 264 + /// Streams the message content for the current transaction from `message` 265 + /// until end of stream. Line endings are normalized to CRLF and leading 266 + /// dots stuffed; nothing is buffered beyond the transport writer, so lines 267 + /// and messages of any length work. 268 + pub fn sendMessageReader(c: *Client, message: *Io.Reader) Error!void { 269 + var data_writer = try c.data(); 270 + while (true) { 271 + const chunk = message.peekGreedy(1) catch |err| switch (err) { 272 + error.EndOfStream => break, 273 + error.ReadFailed => return error.ReadFailed, 274 + }; 275 + try data_writer.interface.writeAll(chunk); 276 + message.toss(chunk.len); 277 + } 278 + try data_writer.end(); 279 + } 280 + 281 + /// Starts the DATA phase for streaming a message body: write the content 282 + /// through the returned writer's `interface`, then call `end`. Line endings 283 + /// are normalized to CRLF and leading dots stuffed as the data flows. 284 + pub fn data(c: *Client) Error!DataWriter { 259 285 try c.send("DATA", .{}); 260 286 _ = try c.expect(354); 261 - try protocol.writeStuffed(c.writer, data); 262 - try c.writer.writeAll("." ++ protocol.crlf); 263 - try c.writer.flush(); 264 - _ = try c.expectClass(2); 287 + return .{ 288 + .client = c, 289 + .interface = .{ 290 + .buffer = &.{}, 291 + .vtable = &.{ .drain = DataWriter.drain }, 292 + }, 293 + }; 265 294 } 295 + 296 + /// Streaming writer for a message body; obtained from `data`. The dot 297 + /// stuffing and CRLF normalization state lives here, so chunks may split 298 + /// lines (and even CRLF pairs) at any byte boundary. 299 + pub const DataWriter = struct { 300 + client: *Client, 301 + interface: Io.Writer, 302 + at_line_start: bool = true, 303 + /// A '\r' was seen but not yet emitted; whether it is a line ending 304 + /// depends on the next byte. 305 + pending_cr: bool = false, 306 + 307 + /// Terminates the message (adding a final CRLF if the content did not 308 + /// end with one, then ".\r\n") and reads the server's verdict. 309 + pub fn end(dw: *DataWriter) Error!void { 310 + try dw.interface.flush(); 311 + const c = dw.client; 312 + if (dw.pending_cr) { 313 + // A trailing bare CR counts as a line ending, matching 314 + // `protocol.writeStuffed`. 315 + dw.pending_cr = false; 316 + dw.at_line_start = true; 317 + try c.writer.writeAll(protocol.crlf); 318 + } 319 + if (!dw.at_line_start) try c.writer.writeAll(protocol.crlf); 320 + try c.writer.writeAll("." ++ protocol.crlf); 321 + try c.writer.flush(); 322 + _ = try c.expectClass(2); 323 + } 324 + 325 + fn drain(w: *Io.Writer, chunks: []const []const u8, splat: usize) Io.Writer.Error!usize { 326 + const dw: *DataWriter = @alignCast(@fieldParentPtr("interface", w)); 327 + try dw.writeChunk(w.buffered()); 328 + w.end = 0; 329 + if (chunks.len == 0) return 0; 330 + var n: usize = 0; 331 + for (chunks[0 .. chunks.len - 1]) |bytes| { 332 + try dw.writeChunk(bytes); 333 + n += bytes.len; 334 + } 335 + const pattern = chunks[chunks.len - 1]; 336 + for (0..splat) |_| { 337 + try dw.writeChunk(pattern); 338 + n += pattern.len; 339 + } 340 + return n; 341 + } 342 + 343 + fn writeChunk(dw: *DataWriter, bytes: []const u8) Io.Writer.Error!void { 344 + const out = dw.client.writer; 345 + var rest = bytes; 346 + while (rest.len > 0) { 347 + if (dw.pending_cr) { 348 + dw.pending_cr = false; 349 + if (rest[0] == '\n') { 350 + try out.writeAll(protocol.crlf); 351 + dw.at_line_start = true; 352 + rest = rest[1..]; 353 + continue; 354 + } 355 + // A bare CR mid-line passes through untouched. 356 + try out.writeByte('\r'); 357 + dw.at_line_start = false; 358 + } 359 + if (dw.at_line_start and rest[0] == '.') { 360 + try out.writeAll(".."); 361 + dw.at_line_start = false; 362 + rest = rest[1..]; 363 + continue; 364 + } 365 + const special = std.mem.indexOfAny(u8, rest, "\r\n") orelse { 366 + try out.writeAll(rest); 367 + dw.at_line_start = false; 368 + break; 369 + }; 370 + if (special > 0) { 371 + try out.writeAll(rest[0..special]); 372 + dw.at_line_start = false; 373 + } 374 + switch (rest[special]) { 375 + '\r' => dw.pending_cr = true, 376 + '\n' => { 377 + try out.writeAll(protocol.crlf); 378 + dw.at_line_start = true; 379 + }, 380 + else => unreachable, 381 + } 382 + rest = rest[special + 1 ..]; 383 + } 384 + } 385 + }; 266 386 267 387 /// Runs a complete mail transaction: MAIL FROM, one RCPT TO per recipient, 268 388 /// then DATA. Call after `greet` and `hello`. 269 - pub fn sendMail(c: *Client, from: []const u8, recipients: []const []const u8, data: []const u8) Error!void { 389 + pub fn sendMail(c: *Client, from: []const u8, recipients: []const []const u8, message_data: []const u8) Error!void { 270 390 try c.mailFrom(from); 271 391 for (recipients) |recipient| try c.rcptTo(recipient); 272 - try c.sendMessage(data); 392 + try c.sendMessage(message_data); 273 393 } 274 394 275 395 /// Aborts the current mail transaction. ··· 617 737 618 738 try client.quit(); 619 739 try std.testing.expectEqualStrings("QUIT\r\n", writer.buffered()); 740 + } 741 + 742 + test data { 743 + var reader: Io.Reader = .fixed("354 go ahead\r\n250 2.0.0 Ok\r\n"); 744 + var out_buf: [256]u8 = undefined; 745 + var writer: Io.Writer = .fixed(&out_buf); 746 + var reply_buf: [64]u8 = undefined; 747 + var client: Client = .init(&reader, &writer, &reply_buf); 748 + 749 + // Chunks may split lines, CRLF pairs, and leading dots arbitrarily. 750 + var data_writer = try client.data(); 751 + try data_writer.interface.writeAll("Subject: chunked\n\nfirst"); 752 + try data_writer.interface.writeAll(" second\r"); 753 + try data_writer.interface.writeAll("\n.needs stuffing\r\nsplit\r"); 754 + try data_writer.interface.writeAll("\n"); 755 + try data_writer.interface.writeAll(".x\nend"); 756 + try data_writer.end(); 757 + 758 + try std.testing.expectEqualStrings( 759 + "DATA\r\n" ++ 760 + "Subject: chunked\r\n" ++ 761 + "\r\n" ++ 762 + "first second\r\n" ++ 763 + "..needs stuffing\r\n" ++ 764 + "split\r\n" ++ 765 + "..x\r\n" ++ 766 + "end\r\n" ++ 767 + ".\r\n", 768 + writer.buffered(), 769 + ); 770 + } 771 + 772 + test sendMessageReader { 773 + var reader: Io.Reader = .fixed("354 go ahead\r\n250 2.0.0 Ok\r\n"); 774 + var out_buf: [128]u8 = undefined; 775 + var writer: Io.Writer = .fixed(&out_buf); 776 + var reply_buf: [64]u8 = undefined; 777 + var client: Client = .init(&reader, &writer, &reply_buf); 778 + 779 + var message: Io.Reader = .fixed("Subject: hi\n\n.streamed body\n"); 780 + try client.sendMessageReader(&message); 781 + try std.testing.expectEqualStrings( 782 + "DATA\r\nSubject: hi\r\n\r\n..streamed body\r\n.\r\n", 783 + writer.buffered(), 784 + ); 620 785 }

+163 -3

src/Server.zig

View file

··· 90 90 /// Called for each RCPT TO. Null accepts every recipient. 91 91 rcptTo: ?*const fn (context: ?*anyopaque, to: []const u8) Decision = null, 92 92 /// Called once the complete message has been received. The data has 93 - /// CRLF line endings and dot-stuffing already removed. 94 - message: *const fn (context: ?*anyopaque, envelope: Envelope, data: []const u8) Decision, 93 + /// CRLF line endings and dot-stuffing already removed. Exactly one 94 + /// of `message` and `messageReader` must be set. 95 + message: ?*const fn (context: ?*anyopaque, envelope: Envelope, data: []const u8) Decision = null, 96 + /// Streaming alternative to `message`: called after DATA with a 97 + /// reader that yields the message content (dot-stuffing removed, 98 + /// line endings normalized to CRLF) until end of stream. Anything 99 + /// the callback leaves unread is drained by the session, so 100 + /// returning early is fine. `Options.max_message_size` is not 101 + /// enforced in this mode; individual message lines must fit the 102 + /// session's stream reader buffer. 103 + messageReader: ?*const fn (context: ?*anyopaque, envelope: Envelope, message: *Io.Reader) Decision = null, 95 104 }; 96 105 }; 97 106 ··· 110 119 const arena = arena_state.allocator(); 111 120 112 121 std.debug.assert(!s.options.require_auth or s.handler.vtable.authenticate != null); 122 + std.debug.assert((s.handler.vtable.message == null) != (s.handler.vtable.messageReader == null)); 113 123 114 124 var greeted = false; 115 125 var authenticated = false; ··· 408 418 /// un-stuffing dots, then asks the handler to accept or reject. 409 419 fn receiveData(s: *Server, arena: std.mem.Allocator, envelope: Envelope) RunError!void { 410 420 try s.reply(354, "End data with <CR><LF>.<CR><LF>"); 421 + 422 + if (s.handler.vtable.messageReader) |callback| { 423 + var buffer: [1024]u8 = undefined; 424 + var data_reader: DataReader = .{ 425 + .session_reader = s.reader, 426 + .interface = .{ 427 + .buffer = &buffer, 428 + .vtable = &.{ .stream = DataReader.stream }, 429 + .seek = 0, 430 + .end = 0, 431 + }, 432 + }; 433 + const decision = callback(s.handler.context, envelope, &data_reader.interface); 434 + // Consume whatever the callback left unread, up to and including 435 + // the terminating ".". 436 + while (!data_reader.finished) { 437 + const line = protocol.readLine(s.reader) catch |err| switch (err) { 438 + error.EndOfStream => return, // Client disconnected mid-message. 439 + error.ReadFailed => return error.ReadFailed, 440 + error.LineTooLong => { 441 + try s.discardLine(); 442 + continue; 443 + }, 444 + }; 445 + if (std.mem.eql(u8, line, ".")) break; 446 + } 447 + switch (decision) { 448 + .accept => try s.reply(250, "2.0.0 Ok, message accepted"), 449 + .reject => |r| try s.reply(r.code, r.text), 450 + } 451 + return; 452 + } 453 + 411 454 var data: std.ArrayList(u8) = .empty; 412 455 var oversize = false; 413 456 while (true) { ··· 437 480 try s.reply(552, "5.3.4 Message exceeds maximum size"); 438 481 return; 439 482 } 440 - switch (s.handler.vtable.message(s.handler.context, envelope, data.items)) { 483 + switch (s.handler.vtable.message.?(s.handler.context, envelope, data.items)) { 441 484 .accept => try s.reply(250, "2.0.0 Ok, message accepted"), 442 485 .reject => |r| try s.reply(r.code, r.text), 443 486 } 444 487 } 488 + 489 + /// Adapts the session's line-based DATA phase into an `Io.Reader` of the 490 + /// unstuffed message content for `Handler.VTable.messageReader`. 491 + const DataReader = struct { 492 + session_reader: *Io.Reader, 493 + interface: Io.Reader, 494 + /// Unread remainder of the current line (points into the session 495 + /// reader's buffer, which only this reader touches during DATA). 496 + line: []const u8 = &.{}, 497 + line_ending: []const u8 = &.{}, 498 + finished: bool = false, 499 + 500 + fn stream(io_r: *Io.Reader, w: *Io.Writer, limit: Io.Limit) Io.Reader.StreamError!usize { 501 + const dr: *DataReader = @alignCast(@fieldParentPtr("interface", io_r)); 502 + if (dr.line.len == 0 and dr.line_ending.len == 0) { 503 + if (dr.finished) return error.EndOfStream; 504 + const raw = protocol.readLine(dr.session_reader) catch return error.ReadFailed; 505 + if (std.mem.eql(u8, raw, ".")) { 506 + dr.finished = true; 507 + return error.EndOfStream; 508 + } 509 + dr.line = if (raw.len > 0 and raw[0] == '.') raw[1..] else raw; 510 + dr.line_ending = protocol.crlf; 511 + } 512 + const dest = limit.slice(try w.writableSliceGreedy(1)); 513 + const line_n = @min(dest.len, dr.line.len); 514 + @memcpy(dest[0..line_n], dr.line[0..line_n]); 515 + dr.line = dr.line[line_n..]; 516 + var n = line_n; 517 + if (dr.line.len == 0) { 518 + const ending_n = @min(dest.len - n, dr.line_ending.len); 519 + @memcpy(dest[n..][0..ending_n], dr.line_ending[0..ending_n]); 520 + dr.line_ending = dr.line_ending[ending_n..]; 521 + n += ending_n; 522 + } 523 + w.advance(n); 524 + return n; 525 + } 526 + }; 445 527 446 528 fn reply(s: *Server, code: u16, text: []const u8) error{WriteFailed}!void { 447 529 try s.writer.print("{d} {s}" ++ protocol.crlf, .{ code, text }); ··· 721 803 722 804 try std.testing.expectEqual(@as(usize, 0), h.messages_accepted); 723 805 try std.testing.expect(std.mem.indexOf(u8, output, "552 5.3.4") != null); 806 + try std.testing.expect(std.mem.indexOf(u8, output, "250 2.0.0 Ok\r\n221") != null); 807 + } 808 + 809 + const StreamTestHandler = struct { 810 + collected: std.ArrayList(u8) = .empty, 811 + take_only: ?usize = null, 812 + 813 + fn handler(h: *StreamTestHandler) Handler { 814 + return .{ .context = h, .vtable = &.{ 815 + .messageReader = onMessageReader, 816 + } }; 817 + } 818 + 819 + fn onMessageReader(context: ?*anyopaque, envelope: Envelope, message: *Io.Reader) Decision { 820 + const h: *StreamTestHandler = @ptrCast(@alignCast(context.?)); 821 + _ = envelope; 822 + const gpa = std.testing.allocator; 823 + if (h.take_only) |n| { 824 + const bytes = message.take(n) catch return .{ .reject = .{} }; 825 + h.collected.appendSlice(gpa, bytes) catch return .{ .reject = .{} }; 826 + return .accept; 827 + } 828 + message.appendRemaining(gpa, &h.collected, .unlimited) catch return .{ .reject = .{} }; 829 + return .accept; 830 + } 831 + }; 832 + 833 + test "streaming message handler receives unstuffed content" { 834 + var h: StreamTestHandler = .{}; 835 + defer h.collected.deinit(std.testing.allocator); 836 + 837 + var out_buf: [1024]u8 = undefined; 838 + const output = try runScript( 839 + "EHLO client.example.org\r\n" ++ 840 + "MAIL FROM:<alice@example.com>\r\n" ++ 841 + "RCPT TO:<bob@example.net>\r\n" ++ 842 + "DATA\r\n" ++ 843 + "Subject: streamed\r\n" ++ 844 + "\r\n" ++ 845 + "..dot line\r\n" ++ 846 + "body\r\n" ++ 847 + ".\r\n" ++ 848 + "QUIT\r\n", 849 + &out_buf, 850 + h.handler(), 851 + .{}, 852 + ); 853 + 854 + try std.testing.expectEqualStrings( 855 + "Subject: streamed\r\n\r\n.dot line\r\nbody\r\n", 856 + h.collected.items, 857 + ); 858 + try std.testing.expect(std.mem.indexOf(u8, output, "250 2.0.0 Ok, message accepted") != null); 859 + } 860 + 861 + test "session drains what a streaming handler leaves unread" { 862 + var h: StreamTestHandler = .{ .take_only = 7 }; 863 + defer h.collected.deinit(std.testing.allocator); 864 + 865 + var out_buf: [1024]u8 = undefined; 866 + const output = try runScript( 867 + "EHLO client.example.org\r\n" ++ 868 + "MAIL FROM:<alice@example.com>\r\n" ++ 869 + "RCPT TO:<bob@example.net>\r\n" ++ 870 + "DATA\r\n" ++ 871 + "Subject: mostly unread\r\n" ++ 872 + "lots of body\r\n" ++ 873 + ".\r\n" ++ 874 + "NOOP\r\n" ++ 875 + "QUIT\r\n", 876 + &out_buf, 877 + h.handler(), 878 + .{}, 879 + ); 880 + 881 + try std.testing.expectEqualStrings("Subject", h.collected.items); 882 + try std.testing.expect(std.mem.indexOf(u8, output, "250 2.0.0 Ok, message accepted") != null); 883 + // The NOOP after DATA proves the terminator was consumed. 724 884 try std.testing.expect(std.mem.indexOf(u8, output, "250 2.0.0 Ok\r\n221") != null); 725 885 }

+14 -2

src/main.zig

View file

··· 119 119 120 120 var stdin_buf: [4096]u8 = undefined; 121 121 var stdin: Io.File.Reader = .init(.stdin(), io, &stdin_buf); 122 - const message = try stdin.interface.allocRemaining(arena, .unlimited); 123 122 124 123 const stream = try host.connect(io, port, .{ .mode = .stream }); 125 124 defer stream.close(io); ··· 177 176 }; 178 177 } 179 178 180 - client.sendMail(from, recipients, message) catch |err| { 179 + transact(&client, from, recipients, &stdin.interface) catch |err| { 181 180 if (err == error.UnexpectedReply) { 182 181 const reply = client.last_reply.?; 183 182 std.log.err("server rejected: {d} {s}", .{ reply.code, reply.text }); ··· 186 185 }; 187 186 try client.quit(); 188 187 std.log.info("message sent to {d} recipient(s)", .{recipients.len}); 188 + } 189 + 190 + /// Runs the mail transaction, streaming the message from `message` so 191 + /// arbitrarily large input never has to fit in memory. 192 + fn transact( 193 + client: *zsmtp.Client, 194 + from: []const u8, 195 + recipients: []const []const u8, 196 + message: *Io.Reader, 197 + ) zsmtp.Client.Error!void { 198 + try client.mailFrom(from); 199 + for (recipients) |recipient| try client.rcptTo(recipient); 200 + try client.sendMessageReader(message); 189 201 } 190 202 191 203 fn serve(io: Io, gpa: std.mem.Allocator, config: ServeConfig, port_arg: []const u8) !void {

Configure Feed

Configure Feed