Use PIPELINING on both sides · jcollie.dev/zig-smtp@497ee77

+47 -11

README.md

··· 76 76 try data_writer.end(); // terminates the message, reads the verdict 77 77 ``` 78 78 79 + `envelope` sends MAIL FROM and every RCPT TO at once and reads all their 80 + replies, which against a server advertising PIPELINING 81 + ([RFC 2920](https://datatracker.ietf.org/doc/html/rfc2920)) turns an envelope 82 + of *n* recipients from *n*+1 round trips into one. `hello` sets 83 + `client.pipelining` from the EHLO response and `envelope` falls back to 84 + waiting for each reply when it is false, so the result is the same either 85 + way: 86 + 87 + ```zig 88 + var codes: [3]u16 = undefined; 89 + const accepted = try client.envelope(from, recipients, &codes, .{}); 90 + // codes[i] is the RCPT reply code for recipients[i]. 91 + ``` 92 + 93 + A refused recipient is not an error — with several of them the caller is the 94 + one who can say whether what remains is worth sending — so compare `accepted` 95 + against `recipients.len`. `sendMail` makes that decision the strict way: if 96 + any recipient was refused it sends RSET and returns `error.UnexpectedReply` 97 + without delivering to the others. 98 + 99 + DATA is deliberately left out of the group, though RFC 2920 allows it as the 100 + last command of one. Once a server has answered DATA with 354 the transaction 101 + is committed, and the only ways out are to send the message or to send an 102 + empty one to whichever recipients were accepted; stopping the group before 103 + DATA keeps that choice with the caller, and costs one round trip out of the 104 + *n*+1 saved. 105 + 79 106 `mail` and `rcpt` are the parameterized forms of `mailFrom` and `rcptTo`, 80 107 carrying the ESMTP parameters the server advertised — today SMTPUTF8 and the 81 108 DSN set of [RFC 3461](https://datatracker.ietf.org/doc/html/rfc3461): ··· 219 246 body type reach the handler via `Envelope`. Listening, accepting, and 220 247 concurrency are up to the caller. 221 248 249 + The server holds back the replies that RFC 2920 §3.2 permits — RSET, MAIL 250 + FROM and RCPT TO — so that a pipelined group is answered in one write, and 251 + sends everything pending the moment its input is empty. The condition is what 252 + makes that safe rather than a deadlock: a reply is only ever held while there 253 + is another command already waiting to be answered. 254 + 222 255 Setting `Options.protocol = .lmtp` makes the session speak LMTP 223 256 ([RFC 2033](https://datatracker.ietf.org/doc/html/rfc2033)) instead: `LHLO` 224 257 greets and `HELO`/`EHLO` are refused with 500, and the end of a message ··· 326 359 both sides, and the server validates MAIL and RCPT parameters (SIZE=, BODY=, 327 360 and the DSN set RET=, ENVID=, NOTIFY=, ORCPT=). Both sides also speak LMTP, 328 361 where a message ends with one verdict per recipient rather than one for the 329 - message. 362 + message, and both use PIPELINING, which collapses an envelope into a single 363 + round trip. 330 364 331 365 ## Known gaps 332 366 ··· 353 387 354 388 ### Protocol 355 389 356 - - **PIPELINING** ([RFC 2920](https://datatracker.ietf.org/doc/html/rfc2920)) 357 - — advertised and parsed by both sides, used by neither. `sendMail` is 358 - strictly request-response. 359 390 - **BINARYMIME** — CHUNKING is implemented but `BODY=BINARYMIME` is refused, 360 391 which is the other half of 361 392 [RFC 3030](https://datatracker.ietf.org/doc/html/rfc3030). ··· 395 426 396 427 ### Client 397 428 398 - - **`sendMail` is all-or-nothing on recipients** — the first rejected RCPT 399 - aborts the transaction, where `smtplib.sendmail` reports the refused ones 400 - and fails only when every one is refused. 429 + - **`sendMail` is all-or-nothing on recipients** — a refused RCPT abandons 430 + the transaction, where `smtplib.sendmail` delivers to the rest and reports 431 + the refusals. `envelope` gives a caller the per-recipient codes to decide 432 + for itself, but no higher-level call does that decision for it. 401 433 - **No `SIZE=` or `BODY=` on MAIL**, though the client parses both 402 434 capabilities off EHLO; `max_size` in particular is read and never used, so 403 435 nothing checks that a message fits before transmitting it. 404 436 - No MX resolution or connect helper, no 4xx retry or backoff, no connection 405 - reuse helper, no pipelined `sendMail`. 437 + reuse helper. 406 438 407 439 ## Standards 408 440 ··· 430 462 recipient instead of one for the message, after DATA and after `BDAT 431 463 LAST` alike. 432 464 - [RFC 2920](https://datatracker.ietf.org/doc/html/rfc2920) — PIPELINING: 433 - advertised by the server, whose strictly sequential command loop handles 434 - pipelined clients naturally; parsed by the client. 465 + the client sends a whole envelope as one group through `envelope`, and the 466 + server holds back the replies it is allowed to (RSET, MAIL, RCPT) so they 467 + leave together, sending everything pending the moment its input runs dry. 435 468 - [RFC 3207](https://datatracker.ietf.org/doc/html/rfc3207) — STARTTLS: 436 469 client and server, including the mandatory post-handshake state reset. 437 470 - [RFC 8314](https://datatracker.ietf.org/doc/html/rfc8314) — implicit TLS ··· 463 496 The specifications this implementation was written against, and the outside 464 497 work it borrows from, in the RFC citation format so that a reference here 465 498 matches one anywhere else. The **Standards** section above says what is 466 - implemented of each; this one says what each document *is*. 499 + implemented of each; this one says what each document *is*. Every entry is 500 + also filed in the project bibliography, so a citation can be taken from there 501 + rather than composed; the RFCs are keyed by their DOIs (`10.17487/RFC5321` 502 + and so on). 467 503 468 504 - **[RFC1870]** Klensin, J., Freed, N., and K. Moore, "SMTP Service 469 505 Extension for Message Size Declaration", RFC 1870, November 1995,

+313 -22

src/Client.zig

View file

··· 37 37 /// upgrade. A session that speaks TLS from the first byte (port 465) hands 38 38 /// `init` an already-encrypted transport, and sets this itself. 39 39 security: Security = .plaintext, 40 + /// Whether the server advertised PIPELINING 41 + /// ([RFC 2920](https://datatracker.ietf.org/doc/html/rfc2920)), which 42 + /// `envelope` uses to send a whole envelope in one round trip. Set by 43 + /// `hello` from the EHLO response, and cleared by a HELO fallback, since 44 + /// RFC 2920 §3.1 lets a client pipeline only against a server that said it 45 + /// could take it. 46 + pipelining: bool = false, 40 47 /// Which protocol to speak. Set before `hello`; see `Protocol`. (Spelled 41 48 /// `mode` rather than `protocol` only because this file's `protocol` 42 49 /// module import already holds that name in this scope; the server's ··· 201 208 pub fn hello(c: *Client, client_name: []const u8) (Error || ArgumentError)!Extensions { 202 209 if (!protocol.isSafeArgument(client_name)) return error.UnsafeArgument; 203 210 c.accepted_recipients = 0; 211 + c.pipelining = false; 204 212 if (c.mode == .lmtp) { 205 213 // LHLO has EHLO's semantics, and there is no older greeting to fall 206 214 // back to: an LMTP server that will not take LHLO is not one. 207 215 try c.send("LHLO {s}", .{client_name}); 208 - return Extensions.parse(try c.expectClass(2)); 216 + const extensions = Extensions.parse(try c.expectClass(2)); 217 + c.pipelining = extensions.pipelining; 218 + return extensions; 209 219 } 210 220 try c.send("EHLO {s}", .{client_name}); 211 221 const reply = try c.readReply(); 212 - if (reply.isPositiveCompletion()) return Extensions.parse(reply); 222 + if (reply.isPositiveCompletion()) { 223 + const extensions = Extensions.parse(reply); 224 + c.pipelining = extensions.pipelining; 225 + return extensions; 226 + } 213 227 if (reply.code == 500 or reply.code == 502) { 228 + // A server old enough to refuse EHLO has no extensions at all. 214 229 try c.send("HELO {s}", .{client_name}); 215 230 _ = try c.expectClass(2); 216 231 return .{}; ··· 405 420 406 421 /// `mailFrom` with ESMTP parameters. 407 422 pub fn mail(c: *Client, from: []const u8, options: MailOptions) (Error || ArgumentError)!void { 423 + try c.checkMail(from, options); 424 + try c.writeMail(from, options); 425 + try c.writer.flush(); 426 + _ = try c.expectClass(2); 427 + c.accepted_recipients = 0; 428 + } 429 + 430 + /// Everything about a MAIL command that can be refused before it is 431 + /// written. Split out so that a pipelined group can be validated in full 432 + /// before any of it goes on the wire. 433 + fn checkMail(c: *Client, from: []const u8, options: MailOptions) ArgumentError!void { 434 + _ = c; 408 435 if (!protocol.isSafeArgument(from)) return error.UnsafeArgument; 409 436 if (options.envid) |envid| { 410 437 if (protocol.xtextEncodedLen(envid) > protocol.max_envid_len) 411 438 return error.ArgumentTooLong; 412 439 } 440 + } 441 + 442 + /// Writes MAIL without flushing or reading its reply. 443 + fn writeMail(c: *Client, from: []const u8, options: MailOptions) Error!void { 413 444 try c.writer.print("MAIL FROM:<{s}>", .{from}); 414 445 if (options.smtputf8) try c.writer.writeAll(" SMTPUTF8"); 415 446 if (options.ret) |ret| try c.writer.print(" RET={f}", .{ret}); ··· 418 449 try protocol.writeXtext(c.writer, envid); 419 450 } 420 451 try c.writer.writeAll(protocol.crlf); 421 - try c.writer.flush(); 422 - _ = try c.expectClass(2); 423 - c.accepted_recipients = 0; 424 452 } 425 453 426 454 /// Adds a recipient to the current transaction. Returns ··· 432 460 433 461 /// `rcptTo` with ESMTP parameters. 434 462 pub fn rcpt(c: *Client, to: []const u8, options: RcptOptions) (Error || ArgumentError)!void { 463 + const code = try c.rcptCode(to, options); 464 + if (code / 100 != 2) return error.UnexpectedReply; 465 + } 466 + 467 + /// `rcpt`, but a refusal is the returned code rather than an error. The 468 + /// reply is in `last_reply` either way. 469 + fn rcptCode(c: *Client, to: []const u8, options: RcptOptions) (Error || ArgumentError)!u16 { 470 + try c.checkRcpt(to, options); 471 + try c.writeRcpt(to, options); 472 + try c.writer.flush(); 473 + const reply = try c.readReply(); 474 + if (reply.isPositiveCompletion()) c.accepted_recipients += 1; 475 + return reply.code; 476 + } 477 + 478 + /// Everything about a RCPT command that can be refused before it is 479 + /// written; see `checkMail`. 480 + fn checkRcpt(c: *Client, to: []const u8, options: RcptOptions) ArgumentError!void { 481 + _ = c; 435 482 if (!protocol.isSafeArgument(to)) return error.UnsafeArgument; 436 483 if (options.orcpt) |orcpt| { 437 484 if (orcpt.addr_type.len == 0 or !protocol.isSafeArgument(orcpt.addr_type) or ··· 440 487 if (orcpt.addr_type.len + 1 + protocol.xtextEncodedLen(orcpt.address) > protocol.Orcpt.max_len) 441 488 return error.ArgumentTooLong; 442 489 } 490 + } 491 + 492 + /// Writes RCPT without flushing or reading its reply. 493 + fn writeRcpt(c: *Client, to: []const u8, options: RcptOptions) Error!void { 443 494 try c.writer.print("RCPT TO:<{s}>", .{to}); 444 495 if (options.notify) |notify| try c.writer.print(" NOTIFY={f}", .{notify}); 445 496 if (options.orcpt) |orcpt| try c.writer.print(" ORCPT={f}", .{orcpt}); 446 497 try c.writer.writeAll(protocol.crlf); 498 + } 499 + 500 + pub const EnvelopeOptions = struct { 501 + /// Parameters for the MAIL command. 502 + mail: MailOptions = .{}, 503 + /// Parameters applied to every RCPT command. Per-recipient parameters 504 + /// need `rcpt` called individually. 505 + rcpt: RcptOptions = .{}, 506 + }; 507 + 508 + /// Sends MAIL FROM and one RCPT TO per recipient, then reads every reply, 509 + /// and returns how many recipients the server accepted. 510 + /// 511 + /// When the server advertised PIPELINING the commands go out as a single 512 + /// group and their replies are read together, which turns an envelope of 513 + /// *n* recipients from *n*+1 round trips into one. Otherwise each command 514 + /// waits for its own reply, and the result is the same either way. 515 + /// 516 + /// DATA is deliberately not part of the group, though RFC 2920 §3.1 allows 517 + /// it as the last command of one. Once a server has answered DATA with 354 518 + /// the transaction is committed, and a caller that wanted all-or-nothing 519 + /// delivery has no way back: the only ways out of the data phase are to 520 + /// send the message or to send an empty one to whichever recipients *were* 521 + /// accepted. Stopping the group before DATA keeps that decision with the 522 + /// caller, and costs one round trip out of the *n*+1 saved. 523 + /// 524 + /// `codes`, when given, must have room for `recipients.len` entries and 525 + /// receives each RCPT reply code in order. Codes rather than replies 526 + /// because the replies share one buffer: by the time the group has been 527 + /// read, only the last one's text still exists. 528 + /// 529 + /// A refused MAIL FROM is `error.UnexpectedReply`, with the reply in 530 + /// `last_reply` and the rest of the group drained. Refused *recipients* 531 + /// are not an error — with several of them the caller is the one who can 532 + /// say whether what remains is worth sending — so compare the returned 533 + /// count against `recipients.len`. 534 + pub fn envelope( 535 + c: *Client, 536 + from: []const u8, 537 + recipients: []const []const u8, 538 + codes: ?[]u16, 539 + options: EnvelopeOptions, 540 + ) (Error || ArgumentError)!usize { 541 + if (codes) |slice| std.debug.assert(slice.len >= recipients.len); 542 + if (!c.pipelining) { 543 + try c.mail(from, options.mail); 544 + var accepted: usize = 0; 545 + for (recipients, 0..) |recipient, index| { 546 + const code = try c.rcptCode(recipient, options.rcpt); 547 + if (codes) |slice| slice[index] = code; 548 + if (code / 100 == 2) accepted += 1; 549 + } 550 + return accepted; 551 + } 552 + 553 + // Everything is validated before anything is written: a group that 554 + // turned out to be unsendable halfway through would leave the session 555 + // holding a partial command. 556 + try c.checkMail(from, options.mail); 557 + for (recipients) |recipient| try c.checkRcpt(recipient, options.rcpt); 558 + 559 + try c.writeMail(from, options.mail); 560 + for (recipients) |recipient| try c.writeRcpt(recipient, options.rcpt); 447 561 try c.writer.flush(); 448 - _ = try c.expectClass(2); 449 - c.accepted_recipients += 1; 562 + 563 + // RFC 2920 §3.1: every status in the group must be checked, and all of 564 + // them must be read whatever the first one said, or the replies still 565 + // queued would be mistaken for the answers to whatever comes next. 566 + const mail_reply = try c.readReply(); 567 + const mail_ok = mail_reply.isPositiveCompletion(); 568 + if (mail_ok) c.accepted_recipients = 0; 569 + 570 + var accepted: usize = 0; 571 + for (0..recipients.len) |index| { 572 + if (!mail_ok) { 573 + // The MAIL reply is the one worth keeping, so the rest of the 574 + // group is drained without disturbing it. 575 + try c.discardReply(); 576 + if (codes) |slice| slice[index] = 0; 577 + continue; 578 + } 579 + const reply = try c.readReply(); 580 + if (codes) |slice| slice[index] = reply.code; 581 + if (reply.isPositiveCompletion()) { 582 + accepted += 1; 583 + c.accepted_recipients += 1; 584 + } 585 + } 586 + if (!mail_ok) return error.UnexpectedReply; 587 + return accepted; 450 588 } 451 589 452 590 /// Sends the message content for the current transaction (DATA). Line ··· 506 644 /// In LMTP that is one verdict per accepted recipient rather than one 507 645 /// for the message. All of them are read — leaving any unread would 508 646 /// desynchronize the session — and a non-2xx among them becomes 509 - /// `error.RecipientRejected`, which unlike `error.UnexpectedReply` 510 - /// leaves nothing useful in `last_reply`. A caller that needs to know 511 - /// *which* recipients failed, the whole reason for speaking LMTP, 512 - /// wants `endResults`. 647 + /// `error.RecipientRejected`, with that first refusal left in 648 + /// `last_reply`: once one has been read, the rest of the group is 649 + /// drained without disturbing it. Which *recipient* it belonged to is 650 + /// only available from `endResults`, which is the whole reason for 651 + /// speaking LMTP and the way to see every verdict. 513 652 pub fn end(dw: *DataWriter) Error!void { 514 653 var verdicts = try dw.endResults(); 515 654 const per_recipient = verdicts.remaining > 1; 516 - var rejected = false; 517 655 while (try verdicts.next()) |reply| { 518 - if (!reply.isPositiveCompletion()) rejected = true; 656 + if (reply.isPositiveCompletion()) continue; 657 + while (verdicts.remaining > 0) : (verdicts.remaining -= 1) 658 + try dw.client.discardReply(); 659 + // With one reply there was never any ambiguity to begin with. 660 + return if (per_recipient) error.RecipientRejected else error.UnexpectedReply; 519 661 } 520 - if (!rejected) return; 521 - // With one reply there is no ambiguity: `last_reply` holds it. 522 - return if (per_recipient) error.RecipientRejected else error.UnexpectedReply; 523 662 } 524 663 525 664 /// Terminates the message and returns the verdicts to read: one in ··· 682 821 // final dot of DATA gets, so it is read the same way. 683 822 var chunk_results = c.results(); 684 823 const per_recipient = chunk_results.remaining > 1; 685 - var rejected = false; 686 824 while (try chunk_results.next()) |reply| { 687 - if (!reply.isPositiveCompletion()) rejected = true; 825 + if (reply.isPositiveCompletion()) continue; 826 + while (chunk_results.remaining > 0) : (chunk_results.remaining -= 1) 827 + try c.discardReply(); 828 + return if (per_recipient) error.RecipientRejected else error.UnexpectedReply; 688 829 } 689 - if (!rejected) return; 690 - return if (per_recipient) error.RecipientRejected else error.UnexpectedReply; 691 830 } 692 831 693 832 /// Sends the message content for the current transaction as a single BDAT ··· 699 838 /// Runs a complete mail transaction: MAIL FROM, one RCPT TO per recipient, 700 839 /// then DATA. Call after `greet` and `hello`. 701 840 pub fn sendMail(c: *Client, from: []const u8, recipients: []const []const u8, message_data: []const u8) (Error || ArgumentError)!void { 702 - try c.mailFrom(from); 703 - for (recipients) |recipient| try c.rcptTo(recipient); 841 + const accepted = try c.envelope(from, recipients, null, .{}); 842 + if (accepted != recipients.len) { 843 + // All or nothing, so nothing: the envelope is abandoned before DATA 844 + // rather than delivering to the subset that was accepted. A caller 845 + // who wants the subset calls `envelope` and decides for itself. 846 + c.rset() catch {}; 847 + return error.UnexpectedReply; 848 + } 704 849 try c.sendMessage(message_data); 705 850 } 706 851 ··· 731 876 const reply = try Reply.read(c.reader, c.reply_buffer); 732 877 c.last_reply = reply; 733 878 return reply; 879 + } 880 + 881 + /// Reads one reply and throws it away, without touching `reply_buffer`. 882 + /// 883 + /// That is the point of it: the replies to a pipelined group all arrive 884 + /// before any of them can be acted on, and each one read into 885 + /// `reply_buffer` overwrites the last. Draining the rest of a group this 886 + /// way leaves the failing reply that was already read intact in 887 + /// `last_reply`, so an error can still say what went wrong. 888 + fn discardReply(c: *Client) Error!void { 889 + while (true) { 890 + const line = protocol.readLine(c.reader) catch |err| switch (err) { 891 + error.LineTooLong => return error.ReplyTooLong, 892 + error.ReadFailed, error.EndOfStream => |e| return e, 893 + }; 894 + if (line.len < 4) return if (line.len == 3) {} else error.InvalidReply; 895 + // A '-' in the fourth column continues the reply; a space ends it. 896 + switch (line[3]) { 897 + '-' => continue, 898 + ' ' => return, 899 + else => return error.InvalidReply, 900 + } 901 + } 734 902 } 735 903 736 904 fn expect(c: *Client, code: u16) Error!Reply { ··· 919 1087 client.authenticate(.{}, "u", "p"), 920 1088 ); 921 1089 } 1090 + } 1091 + 1092 + test "a pipelined envelope writes the whole group before reading a reply" { 1093 + // The MAIL is refused. A client working one command at a time would 1094 + // stop there; a pipelined one has already sent everything, and that is 1095 + // what makes the difference observable from the wire alone. 1096 + const responses = "250-mx.example.com\r\n250 PIPELINING\r\n" ++ 1097 + "550 5.1.8 Bad sender\r\n250 2.1.5 Ok\r\n250 2.1.5 Ok\r\n"; 1098 + var reader: Io.Reader = .fixed(responses); 1099 + var out_buf: [512]u8 = undefined; 1100 + var writer: Io.Writer = .fixed(&out_buf); 1101 + var reply_buf: [256]u8 = undefined; 1102 + var client: Client = .init(&reader, &writer, &reply_buf); 1103 + 1104 + _ = try client.hello("client.example.org"); 1105 + try std.testing.expect(client.pipelining); 1106 + 1107 + const recipients: []const []const u8 = &.{ "bob@example.net", "carol@example.net" }; 1108 + try std.testing.expectError( 1109 + error.UnexpectedReply, 1110 + client.envelope("alice@example.com", recipients, null, .{}), 1111 + ); 1112 + try std.testing.expectEqualStrings( 1113 + "EHLO client.example.org\r\n" ++ 1114 + "MAIL FROM:<alice@example.com>\r\n" ++ 1115 + "RCPT TO:<bob@example.net>\r\n" ++ 1116 + "RCPT TO:<carol@example.net>\r\n", 1117 + writer.buffered(), 1118 + ); 1119 + // The MAIL reply is the one kept, even though two more were read after 1120 + // it, and the group was drained so the stream is where it should be. 1121 + try std.testing.expectEqual(@as(u16, 550), client.last_reply.?.code); 1122 + try std.testing.expectEqualStrings("5.1.8 Bad sender", client.last_reply.?.text); 1123 + try std.testing.expectEqual(@as(usize, 0), reader.bufferedLen()); 1124 + } 1125 + 1126 + test "without PIPELINING the commands wait for each other" { 1127 + // Same refusal, no PIPELINING advertised: the RCPTs are never sent. 1128 + const responses = "250-mx.example.com\r\n250 8BITMIME\r\n550 5.1.8 Bad sender\r\n"; 1129 + var reader: Io.Reader = .fixed(responses); 1130 + var out_buf: [512]u8 = undefined; 1131 + var writer: Io.Writer = .fixed(&out_buf); 1132 + var reply_buf: [256]u8 = undefined; 1133 + var client: Client = .init(&reader, &writer, &reply_buf); 1134 + 1135 + _ = try client.hello("client.example.org"); 1136 + try std.testing.expect(!client.pipelining); 1137 + 1138 + const recipients: []const []const u8 = &.{ "bob@example.net", "carol@example.net" }; 1139 + try std.testing.expectError( 1140 + error.UnexpectedReply, 1141 + client.envelope("alice@example.com", recipients, null, .{}), 1142 + ); 1143 + try std.testing.expectEqualStrings( 1144 + "EHLO client.example.org\r\nMAIL FROM:<alice@example.com>\r\n", 1145 + writer.buffered(), 1146 + ); 1147 + } 1148 + 1149 + test "envelope reports which recipients were refused" { 1150 + const responses = "250-mx.example.com\r\n250 PIPELINING\r\n" ++ 1151 + "250 2.1.0 Ok\r\n250 2.1.5 Ok\r\n550 5.1.1 No such user\r\n250 2.1.5 Ok\r\n"; 1152 + var reader: Io.Reader = .fixed(responses); 1153 + var out_buf: [512]u8 = undefined; 1154 + var writer: Io.Writer = .fixed(&out_buf); 1155 + var reply_buf: [256]u8 = undefined; 1156 + var client: Client = .init(&reader, &writer, &reply_buf); 1157 + 1158 + _ = try client.hello("client.example.org"); 1159 + const recipients: []const []const u8 = &.{ 1160 + "bob@example.net", 1161 + "nobody@example.net", 1162 + "carol@example.net", 1163 + }; 1164 + var codes: [3]u16 = undefined; 1165 + const accepted = try client.envelope("alice@example.com", recipients, &codes, .{}); 1166 + 1167 + try std.testing.expectEqual(@as(usize, 2), accepted); 1168 + try std.testing.expectEqualSlices(u16, &.{ 250, 550, 250 }, &codes); 1169 + // A refused recipient is not an error here, so the transaction is still 1170 + // open and the client knows how many it may deliver to. 1171 + try std.testing.expectEqual(@as(usize, 2), client.accepted_recipients); 1172 + } 1173 + 1174 + test "the same envelope works the same way without pipelining" { 1175 + const responses = "250 2.1.0 Ok\r\n250 2.1.5 Ok\r\n550 5.1.1 No such user\r\n250 2.1.5 Ok\r\n"; 1176 + var reader: Io.Reader = .fixed(responses); 1177 + var out_buf: [512]u8 = undefined; 1178 + var writer: Io.Writer = .fixed(&out_buf); 1179 + var reply_buf: [256]u8 = undefined; 1180 + var client: Client = .init(&reader, &writer, &reply_buf); 1181 + 1182 + const recipients: []const []const u8 = &.{ 1183 + "bob@example.net", 1184 + "nobody@example.net", 1185 + "carol@example.net", 1186 + }; 1187 + var codes: [3]u16 = undefined; 1188 + const accepted = try client.envelope("alice@example.com", recipients, &codes, .{}); 1189 + try std.testing.expectEqual(@as(usize, 2), accepted); 1190 + try std.testing.expectEqualSlices(u16, &.{ 250, 550, 250 }, &codes); 1191 + } 1192 + 1193 + test "sendMail abandons the transaction rather than deliver to some" { 1194 + const responses = "250-mx.example.com\r\n250 PIPELINING\r\n" ++ 1195 + "250 2.1.0 Ok\r\n250 2.1.5 Ok\r\n550 5.1.1 No such user\r\n" ++ 1196 + "250 2.0.0 Ok\r\n"; // the RSET 1197 + var reader: Io.Reader = .fixed(responses); 1198 + var out_buf: [512]u8 = undefined; 1199 + var writer: Io.Writer = .fixed(&out_buf); 1200 + var reply_buf: [256]u8 = undefined; 1201 + var client: Client = .init(&reader, &writer, &reply_buf); 1202 + 1203 + _ = try client.hello("client.example.org"); 1204 + const recipients: []const []const u8 = &.{ "bob@example.net", "nobody@example.net" }; 1205 + try std.testing.expectError( 1206 + error.UnexpectedReply, 1207 + client.sendMail("alice@example.com", recipients, "hi\r\n"), 1208 + ); 1209 + // DATA was never sent, so nothing reached the recipient that was 1210 + // accepted, and the session was left clean for the next transaction. 1211 + try std.testing.expect(std.mem.indexOf(u8, writer.buffered(), "DATA") == null); 1212 + try std.testing.expect(std.mem.endsWith(u8, writer.buffered(), "RSET\r\n")); 922 1213 } 923 1214 924 1215 test "LMTP greets with LHLO and reads one verdict per recipient" {

+118 -3

src/Server.zig

View file

··· 381 381 transaction.smtputf8 = mail_smtputf8; 382 382 transaction.ret = mail_ret; 383 383 transaction.envid = mail_envid; 384 - try s.reply(250, "2.1.0 Ok"); 384 + try s.replyGrouped(250, "2.1.0 Ok"); 385 385 }, 386 386 .rcpt => |args| { 387 387 if (transaction.from == null) { ··· 434 434 recipient.address = try arena.dupe(u8, args.path); 435 435 if (recipient.orcpt) |*orcpt| orcpt.addr_type = try arena.dupe(u8, orcpt.addr_type); 436 436 try transaction.recipients.append(arena, recipient); 437 - try s.reply(250, "2.1.5 Ok"); 437 + try s.replyGrouped(250, "2.1.5 Ok"); 438 438 }, 439 439 .data => { 440 440 if (transaction.recipients.items.len == 0) { ··· 467 467 .rset => { 468 468 transaction.clear(); 469 469 _ = arena_state.reset(.retain_capacity); 470 - try s.reply(250, "2.0.0 Ok"); 470 + try s.replyGrouped(250, "2.0.0 Ok"); 471 471 }, 472 472 .noop => try s.reply(250, "2.0.0 Ok"), 473 473 .vrfy => try s.reply(252, "2.5.2 Cannot VRFY user"), ··· 996 996 return true; 997 997 } 998 998 999 + /// Answers a command that ends a pipelined group, which is every command 1000 + /// RFC 2920 §3.2 names as one whose reply must not be held back: EHLO, 1001 + /// DATA, VRFY, EXPN, TURN, QUIT and NOOP, and anything that went wrong. 999 1002 fn reply(s: *Server, code: u16, text: []const u8) error{WriteFailed}!void { 1000 1003 try s.replyLine(code, text); 1001 1004 try s.writer.flush(); 1005 + } 1006 + 1007 + /// Answers one of the commands that may appear anywhere in a pipelined 1008 + /// group — RSET, MAIL FROM and RCPT TO — by holding the reply back while 1009 + /// the client has already sent more for the server to read. 1010 + /// 1011 + /// RFC 2920 §3.2 asks for exactly this: keep those replies in a buffer so 1012 + /// they go out as a unit, and send everything pending the moment the input 1013 + /// is empty. The condition is what makes it safe rather than a deadlock — 1014 + /// a reply is only ever held while there is another command to answer, so 1015 + /// the client is never left waiting for something still in the buffer. 1016 + fn replyGrouped(s: *Server, code: u16, text: []const u8) error{WriteFailed}!void { 1017 + try s.replyLine(code, text); 1018 + if (s.reader.bufferedLen() == 0) try s.writer.flush(); 1002 1019 } 1003 1020 1004 1021 /// A reply without the flush, for when several are going out together. ··· 1154 1171 return .accept; 1155 1172 } 1156 1173 }; 1174 + 1175 + /// A writer that records where its flush boundaries fell, so that a test 1176 + /// can tell one reply per write from several replies in one. 1177 + const BatchingWriter = struct { 1178 + interface: Io.Writer, 1179 + sink: std.ArrayList(u8) = .empty, 1180 + /// The bytes handed over at each drain — one entry per effective flush. 1181 + batches: std.ArrayList(usize) = .empty, 1182 + 1183 + fn init(buffer: []u8) BatchingWriter { 1184 + return .{ .interface = .{ 1185 + .buffer = buffer, 1186 + .vtable = &.{ .drain = drain }, 1187 + .end = 0, 1188 + } }; 1189 + } 1190 + 1191 + fn deinit(bw: *BatchingWriter) void { 1192 + bw.sink.deinit(std.testing.allocator); 1193 + bw.batches.deinit(std.testing.allocator); 1194 + } 1195 + 1196 + fn drain(w: *Io.Writer, chunks: []const []const u8, splat: usize) Io.Writer.Error!usize { 1197 + const bw: *BatchingWriter = @alignCast(@fieldParentPtr("interface", w)); 1198 + const gpa = std.testing.allocator; 1199 + var handed: usize = w.buffered().len; 1200 + bw.sink.appendSlice(gpa, w.buffered()) catch return error.WriteFailed; 1201 + w.end = 0; 1202 + var n: usize = 0; 1203 + if (chunks.len > 0) { 1204 + for (chunks[0 .. chunks.len - 1]) |bytes| { 1205 + bw.sink.appendSlice(gpa, bytes) catch return error.WriteFailed; 1206 + n += bytes.len; 1207 + } 1208 + const pattern = chunks[chunks.len - 1]; 1209 + for (0..splat) |_| { 1210 + bw.sink.appendSlice(gpa, pattern) catch return error.WriteFailed; 1211 + n += pattern.len; 1212 + } 1213 + } 1214 + handed += n; 1215 + if (handed > 0) bw.batches.append(gpa, handed) catch return error.WriteFailed; 1216 + return n; 1217 + } 1218 + }; 1219 + 1220 + test "replies to a pipelined group go out together and in order" { 1221 + var h: TestHandler = .{}; 1222 + defer h.deinit(); 1223 + 1224 + // One group: MAIL, two RCPTs and DATA, which RFC 2920 §3.1 allows as 1225 + // the last command of one. A fixed reader has the whole session 1226 + // buffered, which is what a client that pipelines looks like. 1227 + var reader: Io.Reader = .fixed("EHLO client.example.org\r\n" ++ 1228 + "MAIL FROM:<alice@example.com>\r\n" ++ 1229 + "RCPT TO:<bob@example.net>\r\n" ++ 1230 + "RCPT TO:<carol@example.net>\r\n" ++ 1231 + "DATA\r\nhi\r\n.\r\nQUIT\r\n"); 1232 + var buffer: [4096]u8 = undefined; 1233 + var bw: BatchingWriter = .init(&buffer); 1234 + defer bw.deinit(); 1235 + 1236 + var session: Server = .init(&reader, &bw.interface, h.handler(), .{ .hostname = "mx.test" }); 1237 + try session.run(std.testing.allocator); 1238 + 1239 + // Order first: every reply is there, once, in the order asked for. 1240 + const out = bw.sink.items; 1241 + const envelope_replies = "250 2.1.0 Ok\r\n250 2.1.5 Ok\r\n250 2.1.5 Ok\r\n" ++ 1242 + "354 End data with <CR><LF>.<CR><LF>\r\n"; 1243 + try std.testing.expect(std.mem.indexOf(u8, out, envelope_replies) != null); 1244 + 1245 + // And batching: the three envelope replies were held back and left 1246 + // with the 354, rather than going out one at a time. There are five 1247 + // replies after the greeting and the EHLO response, and fewer writes. 1248 + // And batching: eight replies left in five writes, because the three 1249 + // envelope replies were held back and went out with the 354 as one. 1250 + // The others are the greeting, the EHLO response, the message verdict 1251 + // and the goodbye — all of which RFC 2920 §3.2 says must not be held. 1252 + try std.testing.expectEqual(@as(usize, 5), bw.batches.items.len); 1253 + try std.testing.expectEqual(envelope_replies.len, bw.batches.items[2]); 1254 + } 1255 + 1256 + test "a held reply is released as soon as there is nothing left to read" { 1257 + var h: TestHandler = .{}; 1258 + defer h.deinit(); 1259 + 1260 + // MAIL alone: its reply may not be held, because nothing follows it in 1261 + // the buffer and the client is waiting for it. 1262 + var reader: Io.Reader = .fixed("EHLO client.example.org\r\nMAIL FROM:<a@example.com>\r\n"); 1263 + var buffer: [4096]u8 = undefined; 1264 + var bw: BatchingWriter = .init(&buffer); 1265 + defer bw.deinit(); 1266 + 1267 + var session: Server = .init(&reader, &bw.interface, h.handler(), .{ .hostname = "mx.test" }); 1268 + try session.run(std.testing.allocator); 1269 + 1270 + try std.testing.expect(std.mem.endsWith(u8, bw.sink.items, "250 2.1.0 Ok\r\n")); 1271 + } 1157 1272 1158 1273 fn runScript(input: []const u8, out_buf: []u8, handler: Handler, options: Options) ![]const u8 { 1159 1274 var reader: Io.Reader = .fixed(input);

Configure Feed

Configure Feed