jnk/Send.cpp

   1 #include "Send.hpp"
   2
   3 #include <random>
   4
   5 #include "IP4.hpp"
   6 #include "IP6.hpp"
   7 #include "imemstream.hpp"
   8 #include "message.hpp"
   9 #include "osutil.hpp"
  10
  11 #include <gflags/gflags.h>
  12
  13 // This needs to be at least the length of each string it's trying to match.
  14 DEFINE_uint64(pbfr_size, 4 * 1024, "parser buffer size");
  15
  16 DEFINE_bool(use_esmtp, true, "use ESMTP (EHLO)");
  17
  18 DEFINE_string(local_address, "", "local address to bind");
  19
  20 #include <boost/algorithm/string/case_conv.hpp>
  21
  22 #include <tao/pegtl.hpp>
  23 #include <tao/pegtl/contrib/abnf.hpp>
  24
  25 using namespace tao::pegtl;
  26 using namespace tao::pegtl::abnf;
  27
  28 using namespace std::string_literals;
  29
  30 using std::begin;
  31 using std::end;
  32
  33 // clang-format off
  34
  35 namespace chars {
  36 struct tail : range<'\x80', '\xBF'> {};
  37
  38 struct ch_1 : range<'\x00', '\x7F'> {};
  39
  40 struct ch_2 : seq<range<'\xC2', '\xDF'>, tail> {};
  41
  42 struct ch_3 : sor<seq<one<'\xE0'>, range<'\xA0', '\xBF'>, tail>,
  43                   seq<range<'\xE1', '\xEC'>, rep<2, tail>>,
  44                   seq<one<'\xED'>, range<'\x80', '\x9F'>, tail>,
  45                   seq<range<'\xEE', '\xEF'>, rep<2, tail>>> {};
  46
  47 struct ch_4 : sor<seq<one<'\xF0'>, range<'\x90', '\xBF'>, rep<2, tail>>,
  48                   seq<range<'\xF1', '\xF3'>, rep<3, tail>>,
  49                   seq<one<'\xF4'>, range<'\x80', '\x8F'>, rep<2, tail>>> {};
  50
  51 struct u8char : sor<ch_1, ch_2, ch_3, ch_4> {};
  52
  53 struct non_ascii : sor<ch_2, ch_3, ch_4> {};
  54
  55 struct ascii_only : seq<star<ch_1>, eof> {};
  56
  57 struct utf8_only : seq<star<u8char>, eof> {};
  58 }
  59
  60 namespace SMTP {
  61
  62 // clang-format off
  63
  64 using dot = one<'.'>;
  65 using colon = one<':'>;
  66 using dash = one<'-'>;
  67
  68 struct u_let_dig : sor<ALPHA, DIGIT, chars::non_ascii> {};
  69
  70 struct u_ldh_tail : star<sor<seq<plus<one<'-'>>, u_let_dig>, u_let_dig>> {};
  71
  72 struct u_label : seq<u_let_dig, u_ldh_tail> {};
  73
  74 struct let_dig : sor<ALPHA, DIGIT> {};
  75
  76 struct ldh_tail : star<sor<seq<plus<one<'-'>>, let_dig>, let_dig>> {};
  77
  78 struct ldh_str : seq<let_dig, ldh_tail> {};
  79
  80 struct label : ldh_str {};
  81
  82 struct sub_domain : sor<label, u_label> {};
  83
  84 struct domain : list<sub_domain, dot> {};
  85
  86 struct dec_octet : sor<seq<string<'2','5'>, range<'0','5'>>,
  87                        seq<one<'2'>, range<'0','4'>, DIGIT>,
  88                        seq<range<'0', '1'>, rep<2, DIGIT>>,
  89                        rep_min_max<1, 2, DIGIT>> {};
  90
  91 struct IPv4_address_literal
  92 : seq<dec_octet, dot, dec_octet, dot, dec_octet, dot, dec_octet> {};
  93
  94 struct h16 : rep_min_max<1, 4, HEXDIG> {};
  95
  96 struct ls32 : sor<seq<h16, colon, h16>, IPv4_address_literal> {};
  97
  98 struct dcolon : two<':'> {};
  99
 100 struct IPv6address : sor<seq<                                          rep<6, h16, colon>, ls32>,
 101                          seq<                                  dcolon, rep<5, h16, colon>, ls32>,
 102                          seq<opt<h16                        >, dcolon, rep<4, h16, colon>, ls32>,
 103                          seq<opt<h16,     opt<   colon, h16>>, dcolon, rep<3, h16, colon>, ls32>,
 104                          seq<opt<h16, rep_opt<2, colon, h16>>, dcolon, rep<2, h16, colon>, ls32>,
 105                          seq<opt<h16, rep_opt<3, colon, h16>>, dcolon,        h16, colon,  ls32>,
 106                          seq<opt<h16, rep_opt<4, colon, h16>>, dcolon,                     ls32>,
 107                          seq<opt<h16, rep_opt<5, colon, h16>>, dcolon,                      h16>,
 108                          seq<opt<h16, rep_opt<6, colon, h16>>, dcolon                          >> {};
 109
 110 struct IPv6_address_literal : seq<TAO_PEGTL_ISTRING("IPv6:"), IPv6address> {};
 111
 112 struct dcontent : ranges<33, 90, 94, 126> {};
 113
 114 struct standardized_tag : ldh_str {};
 115
 116 struct general_address_literal : seq<standardized_tag, colon, plus<dcontent>> {};
 117
 118 // See rfc 5321 Section 4.1.3
 119 struct address_literal : seq<one<'['>,
 120                              sor<IPv4_address_literal,
 121                                  IPv6_address_literal,
 122                                  general_address_literal>,
 123                              one<']'>> {};
 124
 125
 126 struct qtextSMTP : sor<ranges<32, 33, 35, 91, 93, 126>, chars::non_ascii> {};
 127 struct graphic : range<32, 126> {};
 128 struct quoted_pairSMTP : seq<one<'\\'>, graphic> {};
 129 struct qcontentSMTP : sor<qtextSMTP, quoted_pairSMTP> {};
 130
 131 // excluded from atext: "(),.@[]"
 132 struct atext : sor<ALPHA, DIGIT,
 133                    one<'!', '#',
 134                        '$', '%',
 135                        '&', '\'',
 136                        '*', '+',
 137                        '-', '/',
 138                        '=', '?',
 139                        '^', '_',
 140                        '`', '{',
 141                        '|', '}',
 142                        '~'>,
 143                    chars::non_ascii> {};
 144 struct atom : plus<atext> {};
 145 struct dot_string : list<atom, dot> {};
 146 struct quoted_string : seq<one<'"'>, star<qcontentSMTP>, one<'"'>> {};
 147 struct local_part : sor<dot_string, quoted_string> {};
 148 struct non_local_part : sor<domain, address_literal> {};
 149 struct mailbox : seq<local_part, one<'@'>, non_local_part> {};
 150
 151 struct at_domain : seq<one<'@'>, domain> {};
 152
 153 struct a_d_l : list<at_domain, one<','>> {};
 154
 155 struct path : seq<opt<seq<a_d_l, colon>>, mailbox> {};
 156
 157 struct path_only : seq<path, eof> {};
 158
 159 // textstring     = 1*(%d09 / %d32-126) ; HT, SP, Printable US-ASCII
 160
 161 // Although not explicit in the grammar of RFC-6531, in practice UTF-8
 162 // is used in the replys.
 163
 164 // struct textstring : plus<sor<one<9>, range<32, 126>>> {};
 165
 166   struct textstring : plus<sor<one<9>, range<32, 126>, chars::non_ascii>> {};
 167
 168 struct server_id : sor<domain, address_literal> {};
 169
 170 // Greeting       = ( "220 " (Domain / address-literal) [ SP textstring ] CRLF )
 171 //                  /
 172 //                  ( "220-" (Domain / address-literal) [ SP textstring ] CRLF
 173 //                 *( "220-" [ textstring ] CRLF )
 174 //                    "220 " [ textstring ] CRLF )
 175
 176 struct greeting_ok
 177 : sor<seq<TAO_PEGTL_ISTRING("220 "), server_id, opt<textstring>, CRLF>,
 178       seq<TAO_PEGTL_ISTRING("220-"), server_id, opt<textstring>, CRLF,
 179  star<seq<TAO_PEGTL_ISTRING("220-"), opt<textstring>, CRLF>>,
 180       seq<TAO_PEGTL_ISTRING("220 "), opt<textstring>, CRLF>>> {};
 181
 182 // Reply-code     = %x32-35 %x30-35 %x30-39
 183
 184 struct reply_code
 185 : seq<range<0x32, 0x35>, range<0x30, 0x35>, range<0x30, 0x39>> {};
 186
 187 // Reply-line     = *( Reply-code "-" [ textstring ] CRLF )
 188 //                     Reply-code  [ SP textstring ] CRLF
 189
 190 struct reply_lines
 191 : seq<star<seq<reply_code, one<'-'>, opt<textstring>, CRLF>>,
 192            seq<reply_code, opt<seq<SP, textstring>>, CRLF>> {};
 193
 194 struct greeting
 195   : sor<greeting_ok, reply_lines> {};
 196
 197 // ehlo-greet     = 1*(%d0-9 / %d11-12 / %d14-127)
 198 //                    ; string of any characters other than CR or LF
 199
 200 struct ehlo_greet : plus<ranges<0, 9, 11, 12, 14, 127>> {};
 201
 202 // ehlo-keyword   = (ALPHA / DIGIT) *(ALPHA / DIGIT / "-")
 203 //                  ; additional syntax of ehlo-params depends on
 204 //                  ; ehlo-keyword
 205
 206 // The '.' we also allow in ehlo-keyword since it has been seen in the
 207 // wild at least at 263.net.
 208
 209 struct ehlo_keyword : seq<sor<ALPHA, DIGIT>, star<sor<ALPHA, DIGIT, dash, dot>>> {};
 210
 211 // ehlo-param     = 1*(%d33-126)
 212 //                  ; any CHAR excluding <SP> and all
 213 //                  ; control characters (US-ASCII 0-31 and 127
 214 //                  ; inclusive)
 215
 216 struct ehlo_param : plus<range<33, 126>> {};
 217
 218 // ehlo-line      = ehlo-keyword *( SP ehlo-param )
 219
 220 // The AUTH= thing is so common with some servers (postfix) that I
 221 // guess we have to accept it.
 222
 223 struct ehlo_line
 224     : seq<ehlo_keyword, star<seq<sor<SP,one<'='>>, ehlo_param>>> {};
 225
 226 // ehlo-ok-rsp    = ( "250 " Domain [ SP ehlo-greet ] CRLF )
 227 //                  /
 228 //                  ( "250-" Domain [ SP ehlo-greet ] CRLF
 229 //                 *( "250-" ehlo-line CRLF )
 230 //                    "250 " ehlo-line CRLF )
 231
 232 // The last line having the optional ehlo_line is not strictly correct.
 233 // Was added to work with postfix/src/smtpstone/smtp-sink.c.
 234
 235 struct ehlo_ok_rsp
 236 : sor<seq<TAO_PEGTL_ISTRING("250 "), server_id, opt<ehlo_greet>, CRLF>,
 237
 238       seq<TAO_PEGTL_ISTRING("250-"), server_id, opt<ehlo_greet>, CRLF,
 239  star<seq<TAO_PEGTL_ISTRING("250-"), ehlo_line, CRLF>>,
 240       seq<TAO_PEGTL_ISTRING("250 "), opt<ehlo_line>, CRLF>>
 241       > {};
 242
 243 struct ehlo_rsp
 244   : sor<ehlo_ok_rsp, reply_lines> {};
 245
 246 struct helo_ok_rsp
 247   : seq<TAO_PEGTL_ISTRING("250 "), server_id, opt<ehlo_greet>, CRLF> {};
 248
 249 struct auth_login_username
 250     : seq<TAO_PEGTL_STRING("334 VXNlcm5hbWU6"), CRLF> {};
 251
 252 struct auth_login_password
 253     : seq<TAO_PEGTL_STRING("334 UGFzc3dvcmQ6"), CRLF> {};
 254
 255 // clang-format on
 256
 257 template <typename Rule>
 258 struct inaction : nothing<Rule> {
 259 };
 260
 261 template <typename Rule>
 262 struct action : nothing<Rule> {
 263 };
 264
 265 template <>
 266 struct action<server_id> {
 267   template <typename Input>
 268   static void apply(Input const& in, Connection& conn)
 269   {
 270     conn.server_id = in.string();
 271   }
 272 };
 273
 274 template <>
 275 struct action<local_part> {
 276   template <typename Input>
 277   static void apply(Input const& in, Mailbox& mbx)
 278   {
 279     mbx.set_local(in.string());
 280   }
 281 };
 282
 283 template <>
 284 struct action<non_local_part> {
 285   template <typename Input>
 286   static void apply(Input const& in, Mailbox& mbx)
 287   {
 288     mbx.set_domain(in.string());
 289   }
 290 };
 291
 292 template <>
 293 struct action<greeting_ok> {
 294   template <typename Input>
 295   static void apply(Input const& in, Connection& conn)
 296   {
 297     conn.greeting_ok = true;
 298     imemstream  stream{begin(in), size(in)};
 299     std::string line;
 300     while (std::getline(stream, line)) {
 301       LOG(INFO) << "S: " << line;
 302     }
 303   }
 304 };
 305
 306 template <>
 307 struct action<ehlo_ok_rsp> {
 308   template <typename Input>
 309   static void apply(Input const& in, Connection& conn)
 310   {
 311     conn.ehlo_ok = true;
 312     imemstream  stream{begin(in), size(in)};
 313     std::string line;
 314     while (std::getline(stream, line)) {
 315       LOG(INFO) << "S: " << line;
 316     }
 317   }
 318 };
 319
 320 template <>
 321 struct action<ehlo_keyword> {
 322   template <typename Input>
 323   static void apply(Input const& in, Connection& conn)
 324   {
 325     conn.ehlo_keyword = in.string();
 326     boost::to_upper(conn.ehlo_keyword);
 327   }
 328 };
 329
 330 template <>
 331 struct action<ehlo_param> {
 332   template <typename Input>
 333   static void apply(Input const& in, Connection& conn)
 334   {
 335     conn.ehlo_param.push_back(in.string());
 336   }
 337 };
 338
 339 template <>
 340 struct action<ehlo_line> {
 341   template <typename Input>
 342   static void apply(Input const& in, Connection& conn)
 343   {
 344     conn.ehlo_params.emplace(std::move(conn.ehlo_keyword),
 345                              std::move(conn.ehlo_param));
 346   }
 347 };
 348
 349 template <>
 350 struct action<reply_lines> {
 351   template <typename Input>
 352   static void apply(Input const& in, Connection& conn)
 353   {
 354     imemstream  stream{begin(in), size(in)};
 355     std::string line;
 356     while (std::getline(stream, line)) {
 357       LOG(INFO) << "S: " << line;
 358     }
 359   }
 360 };
 361
 362 template <>
 363 struct action<reply_code> {
 364   template <typename Input>
 365   static void apply(Input const& in, Connection& conn)
 366   {
 367     conn.reply_code = in.string();
 368   }
 369 };
 370 } // namespace SMTP
 371
 372 namespace {
 373 bool is_localhost(DNS::RR const& rr)
 374 {
 375   if (std::holds_alternative<DNS::RR_MX>(rr)) {
 376     if (iequal(std::get<DNS::RR_MX>(rr).exchange(), "localhost"))
 377       return true;
 378   }
 379   return false;
 380 }
 381
 382 std::vector<Domain> get_mxs(DNS::Resolver& res, Domain const& domain)
 383 {
 384   auto mxs{std::vector<Domain>{}};
 385
 386   // Non-local part is an address literal.
 387   if (domain.is_address_literal()) {
 388     mxs.emplace_back(domain);
 389     return mxs;
 390   }
 391
 392   // RFC 5321 section 5.1 "Locating the Target Host"
 393
 394   // “The lookup first attempts to locate an MX record associated with
 395   //  the name.  If a CNAME record is found, the resulting name is
 396   //  processed as if it were the initial name.”
 397
 398   // Our (full) resolver will traverse any CNAMEs for us and return
 399   // the CNAME and MX records all together.
 400
 401   auto const& dom = domain.ascii();
 402
 403   auto q{DNS::Query{res, DNS::RR_type::MX, dom}};
 404   auto mx_recs{q.get_records()};
 405
 406   mx_recs.erase(std::remove_if(begin(mx_recs), end(mx_recs), is_localhost),
 407                 end(mx_recs));
 408
 409   auto const nmx =
 410       std::count_if(begin(mx_recs), end(mx_recs), [](auto const& rr) {
 411         return std::holds_alternative<DNS::RR_MX>(rr);
 412       });
 413
 414   if (nmx == 1) {
 415     for (auto const& mx : mx_recs) {
 416       if (std::holds_alternative<DNS::RR_MX>(mx)) {
 417         // RFC 7505 null MX record
 418         if ((std::get<DNS::RR_MX>(mx).preference() == 0) &&
 419             (std::get<DNS::RR_MX>(mx).exchange().empty() ||
 420              (std::get<DNS::RR_MX>(mx).exchange() == "."))) {
 421           LOG(WARNING) << "domain " << dom << " does not accept mail";
 422           return mxs;
 423         }
 424       }
 425     }
 426   }
 427
 428   if (nmx == 0) {
 429     // domain must have address record
 430     mxs.emplace_back(dom);
 431     return mxs;
 432   }
 433
 434   // […] then the sender-SMTP MUST randomize them to spread the load
 435   // across multiple mail exchangers for a specific organization.
 436   std::shuffle(begin(mx_recs), end(mx_recs), std::random_device());
 437   std::sort(begin(mx_recs), end(mx_recs), [](auto const& a, auto const& b) {
 438     if (std::holds_alternative<DNS::RR_MX>(a) &&
 439         std::holds_alternative<DNS::RR_MX>(b)) {
 440       return std::get<DNS::RR_MX>(a).preference() <
 441              std::get<DNS::RR_MX>(b).preference();
 442     }
 443     return false;
 444   });
 445
 446   LOG(INFO) << "MXs for " << domain << " are:";
 447   for (auto const& mx : mx_recs) {
 448     if (std::holds_alternative<DNS::RR_MX>(mx)) {
 449       mxs.emplace_back(std::get<DNS::RR_MX>(mx).exchange());
 450       LOG(INFO) << std::setfill(' ') << std::setw(3)
 451                 << std::get<DNS::RR_MX>(mx).preference() << " "
 452                 << std::get<DNS::RR_MX>(mx).exchange();
 453     }
 454   }
 455
 456   return mxs;
 457 }
 458
 459 int conn(DNS::Resolver& res, Domain const& node, uint16_t port)
 460 {
 461   int fd = socket(AF_INET, SOCK_STREAM, 0);
 462   PCHECK(fd >= 0) << "socket() failed";
 463
 464   if (!FLAGS_local_address.empty()) {
 465     auto loc{sockaddr_in{}};
 466     loc.sin_family = AF_INET;
 467     if (1 != inet_pton(AF_INET, FLAGS_local_address.c_str(),
 468                        reinterpret_cast<void*>(&loc.sin_addr))) {
 469       LOG(FATAL) << "can't interpret " << FLAGS_local_address
 470                  << " as IPv4 address";
 471     }
 472     PCHECK(0 == bind(fd, reinterpret_cast<sockaddr*>(&loc), sizeof(loc)));
 473   }
 474
 475   auto addrs{std::vector<std::string>{}};
 476   if (node.is_address_literal()) {
 477     if (IP4::is_address(node.ascii())) {
 478       addrs.push_back(node.ascii());
 479     }
 480     if (IP4::is_address_literal(node.ascii())) {
 481       auto const addr = IP4::as_address(node.ascii());
 482       addrs.push_back(std::string(addr.data(), addr.length()));
 483     }
 484   }
 485   else {
 486     addrs = res.get_strings(DNS::RR_type::A, node.ascii());
 487   }
 488   for (auto addr : addrs) {
 489     auto in4{sockaddr_in{}};
 490     in4.sin_family = AF_INET;
 491     in4.sin_port   = htons(port);
 492     CHECK_EQ(inet_pton(AF_INET, addr.c_str(),
 493                        reinterpret_cast<void*>(&in4.sin_addr)),
 494              1);
 495     if (connect(fd, reinterpret_cast<const sockaddr*>(&in4), sizeof(in4))) {
 496       PLOG(WARNING) << "connect failed " << addr << ":" << port;
 497       continue;
 498     }
 499
 500     // LOG(INFO) << fd << " connected to " << addr << ":" << port;
 501     return fd;
 502   }
 503
 504   close(fd);
 505   return -1;
 506 }
 507
 508 std::optional<std::unique_ptr<SMTP::Connection>>
 509 open_session(DNS::Resolver& res,
 510              fs::path       config_path,
 511              Domain         sender,
 512              Domain         mx,
 513              char const*    service)
 514 {
 515   auto const port{osutil::get_port(service, "tcp")};
 516
 517   int fd = conn(res, mx, port);
 518   if (fd == -1) {
 519     LOG(WARNING) << mx << " no connection";
 520     return {};
 521   }
 522
 523   // Listen for greeting
 524
 525   auto constexpr read_hook{[]() {}};
 526   auto conn = std::make_unique<SMTP::Connection>(fd, fd, read_hook);
 527
 528   auto in =
 529       istream_input<eol::crlf, 1>{conn->sock.in(), FLAGS_pbfr_size, "session"};
 530   if (!parse<SMTP::greeting, SMTP::action>(in, *conn)) {
 531     LOG(WARNING) << "greeting was unrecognizable";
 532     close(fd);
 533     return {};
 534   }
 535   if (!conn->greeting_ok) {
 536     LOG(WARNING) << "greeting was not in the affirmative";
 537     close(fd);
 538     return {};
 539   }
 540
 541   // EHLO/HELO
 542
 543   auto use_esmtp = FLAGS_use_esmtp;
 544   if (use_esmtp) {
 545     LOG(INFO) << "C: EHLO " << sender.ascii();
 546     conn->sock.out() << "EHLO " << sender.ascii() << "\r\n" << std::flush;
 547     if (!parse<SMTP::ehlo_rsp, SMTP::action>(in, *conn) || !conn->ehlo_ok) {
 548       LOG(WARNING) << "EHLO response was unrecognizable, trying HELO";
 549       use_esmtp = false;
 550     }
 551   }
 552   if (!use_esmtp) {
 553     LOG(INFO) << "C: HELO " << sender.ascii();
 554     conn->sock.out() << "HELO " << sender.ascii() << "\r\n" << std::flush;
 555     if (!parse<SMTP::helo_ok_rsp, SMTP::action>(in, *conn)) {
 556       LOG(WARNING) << "HELO response was unrecognizable";
 557       close(fd);
 558       return {};
 559     }
 560   }
 561
 562   // STARTTLS
 563
 564   if (conn->has_extension("STARTTLS")) {
 565     LOG(INFO) << "C: STARTTLS";
 566     conn->sock.out() << "STARTTLS\r\n" << std::flush;
 567     if (!parse<SMTP::reply_lines, SMTP::action>(in, *conn)) {
 568       LOG(WARNING) << "STARTTLS response was unrecognizable";
 569       close(fd);
 570       return {};
 571     }
 572
 573     DNS::RR_collection tlsa_rrs; // FIXME
 574     if (!conn->sock.starttls_client(config_path, sender.ascii().c_str(),
 575                                     mx.ascii().c_str(), tlsa_rrs, false)) {
 576       LOG(WARNING) << "failed to STARTTLS";
 577       close(fd);
 578       return {};
 579     }
 580   }
 581
 582   return std::optional<std::unique_ptr<SMTP::Connection>>(std::move(conn));
 583 }
 584
 585 bool do_mail_from(SMTP::Connection& conn, Mailbox from)
 586 {
 587   std::ostringstream param_stream;
 588   // param_stream << " SIZE=" << total_size;
 589
 590   if (conn.has_extension("BINARYMIME")) {
 591     param_stream << " BODY=BINARYMIME";
 592   }
 593   else if (conn.has_extension("8BITMIME")) {
 594     param_stream << " BODY=8BITMIME";
 595   }
 596
 597   if (conn.has_extension("SMTPUTF8")) {
 598     param_stream << " SMTPUTF8";
 599   }
 600
 601   auto const param_str = param_stream.str();
 602
 603   LOG(INFO) << "C: MAIL FROM:<" << from << '>' << param_str;
 604   conn.sock.out() << "MAIL FROM:<" << from << '>' << param_str << "\r\n"
 605                   << std::flush;
 606   auto in{istream_input<eol::crlf, 1>{conn.sock.in(), FLAGS_pbfr_size,
 607                                       "mail_from"}};
 608   if (!parse<SMTP::reply_lines, SMTP::action>(in, conn)) {
 609     LOG(ERROR) << "MAIL FROM: reply unparseable";
 610     return false;
 611   }
 612   if (conn.reply_code.at(0) != '2') {
 613     LOG(WARNING) << "MAIL FROM: negative reply " << conn.reply_code;
 614     return false;
 615   }
 616
 617   conn.mail_from = from;
 618   return true;
 619 }
 620
 621 bool do_rcpt_to(SMTP::Connection& conn, Mailbox from, Mailbox to)
 622 {
 623   if (conn.mail_from != from) {
 624     do_mail_from(conn, from);
 625   }
 626
 627   if (std::find(begin(conn.rcpt_to), end(conn.rcpt_to), to) !=
 628       end(conn.rcpt_to)) {
 629     LOG(INFO) << to << " already in recpt_to list of " << conn.server_id;
 630     return true;
 631   }
 632
 633   LOG(INFO) << "C: RCPT TO:<" << to << '>';
 634   conn.sock.out() << "RCPT TO:<" << to << ">\r\n" << std::flush;
 635   auto in{
 636       istream_input<eol::crlf, 1>{conn.sock.in(), FLAGS_pbfr_size, "rcpt_to"}};
 637   if (!parse<SMTP::reply_lines, SMTP::action>(in, conn)) {
 638     LOG(ERROR) << "RCPT TO: reply unparseable";
 639     return false;
 640   }
 641   if (conn.reply_code.at(0) != '2') {
 642     LOG(WARNING) << "RCPT TO: negative reply " << conn.reply_code;
 643     return false;
 644   }
 645
 646   conn.rcpt_to.emplace_back(to);
 647   return true;
 648 }
 649
 650 bool do_data(SMTP::Connection& conn, std::istream& is)
 651 {
 652   LOG(INFO) << "C: DATA";
 653   conn.sock.out() << "DATA\r\n" << std::flush;
 654   auto in{istream_input<eol::crlf, 1>{conn.sock.in(), FLAGS_pbfr_size, "data"}};
 655   if (!parse<SMTP::reply_lines, SMTP::action>(in, conn)) {
 656     LOG(ERROR) << "DATA command reply unparseable";
 657     return false;
 658   }
 659   if (conn.reply_code != "354") {
 660     LOG(ERROR) << "DATA returned " << conn.reply_code;
 661     return false;
 662   }
 663
 664   auto lineno = 0;
 665   auto line{std::string{}};
 666
 667   while (std::getline(is, line)) {
 668     ++lineno;
 669     if (!conn.sock.out().good()) {
 670       conn.sock.log_stats();
 671       LOG(ERROR) << "output no good at line " << lineno;
 672       return false;
 673     }
 674     if (line.length() && (line.at(0) == '.')) {
 675       conn.sock.out() << '.';
 676     }
 677     conn.sock.out() << line;
 678     if (line.back() != '\r') {
 679       LOG(WARNING) << "bare new line in message body at line " << lineno;
 680       conn.sock.out() << '\r';
 681     }
 682     conn.sock.out() << '\n';
 683   }
 684   if (!conn.sock.out().good()) {
 685     LOG(ERROR) << "socket error of some sort after DATA";
 686     return false;
 687   }
 688
 689   // Done!
 690   conn.sock.out() << ".\r\n" << std::flush;
 691
 692   if (!parse<SMTP::reply_lines, SMTP::action>(in, conn)) {
 693     LOG(ERROR) << "DATA reply unparseable";
 694     return false;
 695   }
 696
 697   LOG(INFO) << "reply_code == " << conn.reply_code;
 698   return conn.reply_code.at(0) == '2';
 699 }
 700
 701 bool do_bdat(SMTP::Connection& conn, std::istream& is)
 702 {
 703   auto                  bdat_error = false;
 704   std::streamsize const bfr_size   = 1024 * 1024;
 705   iobuffer<char>        bfr(bfr_size);
 706
 707   auto in =
 708       istream_input<eol::crlf, 1>{conn.sock.in(), FLAGS_pbfr_size, "bdat"};
 709   while (!is.eof()) {
 710     is.read(bfr.data(), bfr_size);
 711     auto const size_read = is.gcount();
 712
 713     conn.sock.out() << "BDAT " << size_read << "\r\n";
 714     LOG(INFO) << "C: BDAT " << size_read;
 715
 716     conn.sock.out().write(bfr.data(), size_read);
 717     conn.sock.out() << std::flush;
 718
 719     if (!parse<SMTP::reply_lines, SMTP::action>(in, conn)) {
 720       LOG(ERROR) << "BDAT reply unparseable";
 721       bdat_error = true;
 722       break;
 723     }
 724     if (conn.reply_code != "250") {
 725       LOG(ERROR) << "BDAT returned " << conn.reply_code;
 726       bdat_error = true;
 727       break;
 728     }
 729   }
 730
 731   conn.sock.out() << "BDAT 0 LAST\r\n" << std::flush;
 732   LOG(INFO) << "C: BDAT 0 LAST";
 733
 734   CHECK((parse<SMTP::reply_lines, SMTP::action>(in, conn)));
 735   if (conn.reply_code != "250") {
 736     LOG(ERROR) << "BDAT 0 LAST returned " << conn.reply_code;
 737     return false;
 738   }
 739
 740   return !bdat_error;
 741 }
 742
 743 bool do_send(SMTP::Connection& conn, std::istream& is)
 744 {
 745   if (conn.has_extension("CHUNKING"))
 746     return do_bdat(conn, is);
 747   return do_data(conn, is);
 748 }
 749
 750 bool do_rset(SMTP::Connection& conn)
 751 {
 752   LOG(INFO) << "C: RSET";
 753   conn.sock.out() << "RSET\r\n" << std::flush;
 754   auto in =
 755       istream_input<eol::crlf, 1>{conn.sock.in(), FLAGS_pbfr_size, "rset"};
 756   return parse<SMTP::reply_lines, SMTP::action>(in, conn);
 757 }
 758
 759 bool do_quit(SMTP::Connection& conn)
 760 {
 761   LOG(INFO) << "C: QUIT";
 762   conn.sock.out() << "QUIT\r\n" << std::flush;
 763   auto in =
 764       istream_input<eol::crlf, 1>{conn.sock.in(), FLAGS_pbfr_size, "quit"};
 765   return parse<SMTP::reply_lines, SMTP::action>(in, conn);
 766 }
 767
 768 } // namespace
 769
 770 Send::Send(fs::path config_path, char const* service)
 771   : config_path_(config_path)
 772   , service_(service)
 773 {
 774 }
 775
 776 bool Send::mail_from(Mailbox const& mailbox)
 777 {
 778   mail_from_ = mailbox;
 779
 780   for (auto& [mx, conn] : exchangers_) {
 781     conn->mail_from.clear();
 782     conn->rcpt_to.clear();
 783   }
 784   return true;
 785 }
 786
 787 bool Send::rcpt_to(DNS::Resolver& res,
 788                    Mailbox const& to,
 789                    std::string&   error_msg)
 790 {
 791   if (mail_from_.empty()) {
 792     LOG(WARNING) << "sequence error, must have MAIL FROM: before RCPT TO:";
 793     return false;
 794   }
 795
 796   // Check for existing receivers entry.
 797   if (auto rec = receivers_.find(to.domain()); rec != receivers_.end()) {
 798     if (auto ex = exchangers_.find(rec->second); ex != exchangers_.end()) {
 799       auto& conn = ex->second;
 800       LOG(INFO) << "### found existing receiver";
 801       LOG(INFO) << "### do_rcpt_to(" << mail_from_ << ", " << to << ");";
 802       return do_rcpt_to(*conn, mail_from_, to);
 803     }
 804     LOG(ERROR) << "found a receiver but not an exchanger "
 805                << "from == " << mail_from_ << "  to == " << to;
 806     return false;
 807   }
 808
 809   // Get a connection to an MX for this domain
 810   std::vector<Domain> mxs = get_mxs(res, to.domain());
 811   CHECK(!mxs.empty());
 812   for (auto& mx : mxs) {
 813     // Check for existing connection.
 814     if (auto ex = exchangers_.find(mx); ex != exchangers_.end()) {
 815       LOG(INFO) << "### found existing connection to " << mx;
 816       receivers_.emplace(to.domain(), mx);
 817       auto& conn = ex->second;
 818       LOG(INFO) << "### do_rcpt_to(" << mail_from_ << ", " << to << ");";
 819       return do_rcpt_to(*conn, mail_from_, to);
 820     }
 821     // Open new connection.
 822     if (auto new_conn = open_session(res, config_path_, mail_from_.domain(), mx,
 823                                      service_.c_str());
 824         new_conn) {
 825       LOG(INFO) << "### opened new connection to " << mx;
 826       exchangers_.emplace(mx, std::move(*new_conn));
 827       auto ex = exchangers_.find(mx);
 828       CHECK(ex != exchangers_.end());
 829       receivers_.emplace(to.domain(), mx);
 830       auto& conn = ex->second;
 831       LOG(INFO) << "### do_rcpt_to(" << mail_from_ << ", " << to << ");";
 832       return do_rcpt_to(*conn, mail_from_, to);
 833     }
 834   }
 835
 836   LOG(WARNING) << "ran out of mail exchangers for " << to;
 837   return false;
 838 }
 839
 840 bool Send::send(std::string_view msg_input)
 841 {
 842   // FIXME this should be done in parallel
 843   for (auto& [dom, conn] : exchangers_) {
 844     if (!conn->rcpt_to.empty()) {
 845
 846       auto is{imemstream{msg_input.data(), msg_input.length()}};
 847       if (!do_send(*conn, is)) {
 848         LOG(WARNING) << "failed to send to " << conn->server_id;
 849         return false;
 850       }
 851     }
 852     else {
 853       LOG(INFO) << "no receivers, skipping MX " << dom;
 854     }
 855     conn->mail_from.clear();
 856     conn->rcpt_to.clear();
 857   }
 858   return true;
 859 }
 860
 861 void Send::rset()
 862 {
 863   for (auto& [dom, conn] : exchangers_) {
 864     if (!conn->rcpt_to.empty()) {
 865       if (!do_rset(*conn)) {
 866         LOG(WARNING) << "failed to rset " << conn->server_id;
 867       }
 868     }
 869     conn->mail_from.clear();
 870     conn->rcpt_to.clear();
 871   }
 872 }
 873
 874 void Send::quit()
 875 {
 876   for (auto& [dom, conn] : exchangers_) {
 877     do_quit(*conn);
 878     conn->sock.close_fds();
 879   }
 880   exchangers_.clear();
 881 }