citadel/modules/smtp/serv_smtpqueue.c

   1 /*
   2  * This module is an SMTP and ESMTP implementation for the Citadel system.
   3  * It is compliant with all of the following:
   4  *
   5  * RFC  821 - Simple Mail Transfer Protocol
   6  * RFC  876 - Survey of SMTP Implementations
   7  * RFC 1047 - Duplicate messages and SMTP
   8  * RFC 1652 - 8 bit MIME
   9  * RFC 1869 - Extended Simple Mail Transfer Protocol
  10  * RFC 1870 - SMTP Service Extension for Message Size Declaration
  11  * RFC 2033 - Local Mail Transfer Protocol
  12  * RFC 2197 - SMTP Service Extension for Command Pipelining
  13  * RFC 2476 - Message Submission
  14  * RFC 2487 - SMTP Service Extension for Secure SMTP over TLS
  15  * RFC 2554 - SMTP Service Extension for Authentication
  16  * RFC 2821 - Simple Mail Transfer Protocol
  17  * RFC 2822 - Internet Message Format
  18  * RFC 2920 - SMTP Service Extension for Command Pipelining
  19  *
  20  * The VRFY and EXPN commands have been removed from this implementation
  21  * because nobody uses these commands anymore, except for spammers.
  22  *
  23  * Copyright (c) 1998-2012 by the citadel.org team
  24  *
  25  *  This program is open source software; you can redistribute it and/or modify
  26  *  it under the terms of the GNU General Public License version 3.
  27  *
  28  *
  29  *
  30  *  This program is distributed in the hope that it will be useful,
  31  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  32  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  33  *  GNU General Public License for more details.
  34  *
  35  *
  36  *
  37  *
  38  */
  39
  40 #include "sysdep.h"
  41 #include <stdlib.h>
  42 #include <unistd.h>
  43 #include <stdio.h>
  44 #include <termios.h>
  45 #include <fcntl.h>
  46 #include <signal.h>
  47 #include <pwd.h>
  48 #include <errno.h>
  49 #include <sys/types.h>
  50 #include <syslog.h>
  51
  52 #if TIME_WITH_SYS_TIME
  53 # include <sys/time.h>
  54 # include <time.h>
  55 #else
  56 # if HAVE_SYS_TIME_H
  57 #  include <sys/time.h>
  58 # else
  59 #  include <time.h>
  60 # endif
  61 #endif
  62 #include <sys/wait.h>
  63 #include <ctype.h>
  64 #include <string.h>
  65 #include <limits.h>
  66 #include <sys/socket.h>
  67 #include <netinet/in.h>
  68 #include <arpa/inet.h>
  69 #include <libcitadel.h>
  70 #include "citadel.h"
  71 #include "server.h"
  72 #include "citserver.h"
  73 #include "support.h"
  74 #include "config.h"
  75 #include "control.h"
  76 #include "user_ops.h"
  77 #include "database.h"
  78 #include "msgbase.h"
  79 #include "internet_addressing.h"
  80 #include "genstamp.h"
  81 #include "domain.h"
  82 #include "clientsocket.h"
  83 #include "locate_host.h"
  84 #include "citadel_dirs.h"
  85
  86 #include "ctdl_module.h"
  87
  88 #include "smtpqueue.h"
  89 #include "event_client.h"
  90
  91
  92 struct CitContext smtp_queue_CC;
  93 pthread_mutex_t ActiveQItemsLock;
  94 HashList *ActiveQItems  = NULL;
  95 HashList *QItemHandlers = NULL;
  96 int max_sessions_for_outbound_smtp = 500; /* how many sessions might be active till we stop adding more smtp jobs */
  97 int ndelay_count = 50; /* every n queued messages we will sleep... */
  98 int delay_msec = 5000; /* this many seconds. */
  99
 100 static const long MaxRetry = SMTP_RETRY_INTERVAL * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2 * 2;
 101 int MsgCount            = 0;
 102 int run_queue_now       = 0;    /* Set to 1 to ignore SMTP send retry times */
 103
 104 void smtp_try_one_queue_entry(OneQueItem *MyQItem,
 105                               MailQEntry *MyQEntry,
 106                               StrBuf *MsgText,
 107 /* KeepMsgText allows us to use MsgText as ours. */
 108                               int KeepMsgText,
 109                               int MsgCount,
 110                               ParsedURL *RelayUrls);
 111
 112
 113 void smtp_evq_cleanup(void)
 114 {
 115
 116         pthread_mutex_lock(&ActiveQItemsLock);
 117         DeleteHash(&QItemHandlers);
 118         DeleteHash(&ActiveQItems);
 119         pthread_mutex_unlock(&ActiveQItemsLock);
 120         pthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
 121 /*      citthread_mutex_destroy(&ActiveQItemsLock); TODO */
 122 }
 123
 124 int DecreaseQReference(OneQueItem *MyQItem)
 125 {
 126         int IDestructQueItem;
 127
 128         pthread_mutex_lock(&ActiveQItemsLock);
 129         MyQItem->ActiveDeliveries--;
 130         IDestructQueItem = MyQItem->ActiveDeliveries == 0;
 131         pthread_mutex_unlock(&ActiveQItemsLock);
 132         return IDestructQueItem;
 133 }
 134
 135 void RemoveQItem(OneQueItem *MyQItem)
 136 {
 137         long len;
 138         const char* Key;
 139         void *VData;
 140         HashPos  *It;
 141
 142         pthread_mutex_lock(&ActiveQItemsLock);
 143         It = GetNewHashPos(ActiveQItems, 0);
 144         if (GetHashPosFromKey(ActiveQItems, LKEY(MyQItem->MessageID), It))
 145                 DeleteEntryFromHash(ActiveQItems, It);
 146         else
 147         {
 148                 syslog(LOG_WARNING,
 149                        "SMTP cleanup: unable to find QItem with ID[%ld]",
 150                        MyQItem->MessageID);
 151                 while (GetNextHashPos(ActiveQItems, It, &len, &Key, &VData))
 152                         syslog(LOG_WARNING,
 153                                "SMTP cleanup: have_: ID[%ld]",
 154                                ((OneQueItem *)VData)->MessageID);
 155         }
 156         pthread_mutex_unlock(&ActiveQItemsLock);
 157         DeleteHashPos(&It);
 158 }
 159
 160
 161 void FreeMailQEntry(void *qv)
 162 {
 163         MailQEntry *Q = qv;
 164 /*
 165         syslog(LOG_DEBUG, "---------------%s--------------", __FUNCTION__);
 166         cit_backtrace();
 167 */
 168         FreeStrBuf(&Q->Recipient);
 169         FreeStrBuf(&Q->StatusMessage);
 170
 171         memset(Q, 0, sizeof(MailQEntry));
 172         free(Q);
 173 }
 174 void FreeQueItem(OneQueItem **Item)
 175 {
 176 /*
 177         syslog(LOG_DEBUG, "---------------%s--------------", __FUNCTION__);
 178         cit_backtrace();
 179 */
 180         DeleteHash(&(*Item)->MailQEntries);
 181         FreeStrBuf(&(*Item)->EnvelopeFrom);
 182         FreeStrBuf(&(*Item)->BounceTo);
 183         FreeStrBuf(&(*Item)->SenderRoom);
 184         FreeURL(&(*Item)->URL);
 185         memset(*Item, 0, sizeof(OneQueItem));
 186         free(*Item);
 187         Item = NULL;
 188 }
 189 void HFreeQueItem(void *Item)
 190 {
 191         FreeQueItem((OneQueItem**)&Item);
 192 }
 193
 194 /* inspect recipients with a status of:
 195  * - 0 (no delivery yet attempted)
 196  * - 3/4 (transient errors
 197  *        were experienced and it's time to try again)
 198  */
 199 int CheckQEntryActive(MailQEntry *ThisItem)
 200 {
 201         if ((ThisItem->Status == 0) ||
 202             (ThisItem->Status == 3) ||
 203             (ThisItem->Status == 4))
 204         {
 205                 return 1;
 206         }
 207         else
 208                 return 0;
 209 }
 210 int CheckQEntryIsBounce(MailQEntry *ThisItem)
 211 {
 212         if ((ThisItem->Status == 3) ||
 213             (ThisItem->Status == 4) ||
 214             (ThisItem->Status == 5))
 215         {
 216                 return 1;
 217         }
 218         else
 219                 return 0;
 220 }
 221
 222 int CountActiveQueueEntries(OneQueItem *MyQItem)
 223 {
 224         HashPos  *It;
 225         long len;
 226         long ActiveDeliveries;
 227         const char *Key;
 228         void *vQE;
 229
 230         ActiveDeliveries = 0;
 231         It = GetNewHashPos(MyQItem->MailQEntries, 0);
 232         while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
 233         {
 234                 MailQEntry *ThisItem = vQE;
 235
 236                 if (CheckQEntryActive(ThisItem))
 237                 {
 238                         ActiveDeliveries++;
 239                         ThisItem->Active = 1;
 240                 }
 241                 else
 242                         ThisItem->Active = 0;
 243         }
 244         DeleteHashPos(&It);
 245         return ActiveDeliveries;
 246 }
 247
 248 OneQueItem *DeserializeQueueItem(StrBuf *RawQItem, long QueMsgID)
 249 {
 250         OneQueItem *Item;
 251         const char *pLine = NULL;
 252         StrBuf *Line;
 253         StrBuf *Token;
 254         void *v;
 255
 256         Item = (OneQueItem*)malloc(sizeof(OneQueItem));
 257         memset(Item, 0, sizeof(OneQueItem));
 258         Item->Retry = SMTP_RETRY_INTERVAL;
 259         Item->MessageID = -1;
 260         Item->QueMsgID = QueMsgID;
 261
 262         Token = NewStrBuf();
 263         Line = NewStrBufPlain(NULL, 128);
 264         while (pLine != StrBufNOTNULL) {
 265                 const char *pItemPart = NULL;
 266                 void *vHandler;
 267
 268                 StrBufExtract_NextToken(Line, RawQItem, &pLine, '\n');
 269                 if (StrLength(Line) == 0) continue;
 270                 StrBufExtract_NextToken(Token, Line, &pItemPart, '|');
 271                 if (GetHash(QItemHandlers, SKEY(Token), &vHandler))
 272                 {
 273                         QItemHandler H;
 274                         H = (QItemHandler) vHandler;
 275                         H(Item, Line, &pItemPart);
 276                 }
 277         }
 278         FreeStrBuf(&Line);
 279         FreeStrBuf(&Token);
 280
 281         if (Item->Retry >= MaxRetry)
 282                 Item->FailNow = 1;
 283
 284         pthread_mutex_lock(&ActiveQItemsLock);
 285         if (GetHash(ActiveQItems,
 286                     LKEY(Item->MessageID),
 287                     &v))
 288         {
 289                 /* WHOOPS. somebody else is already working on this. */
 290                 pthread_mutex_unlock(&ActiveQItemsLock);
 291                 FreeQueItem(&Item);
 292                 return NULL;
 293         }
 294         else {
 295                 /* mark our claim on this. */
 296                 Put(ActiveQItems,
 297                     LKEY(Item->MessageID),
 298                     Item,
 299                     HFreeQueItem);
 300                 pthread_mutex_unlock(&ActiveQItemsLock);
 301         }
 302
 303         return Item;
 304 }
 305
 306 StrBuf *SerializeQueueItem(OneQueItem *MyQItem)
 307 {
 308         StrBuf *QMessage;
 309         HashPos  *It;
 310         const char *Key;
 311         long len;
 312         void *vQE;
 313
 314         QMessage = NewStrBufPlain(NULL, SIZ);
 315         StrBufPrintf(QMessage, "Content-type: %s\n", SPOOLMIME);
 316 //      "attempted|%ld\n"  "retry|%ld\n",, (long)time(NULL), (long)retry );
 317         StrBufAppendBufPlain(QMessage, HKEY("\nmsgid|"), 0);
 318         StrBufAppendPrintf(QMessage, "%ld", MyQItem->MessageID);
 319
 320         StrBufAppendBufPlain(QMessage, HKEY("\nsubmitted|"), 0);
 321         StrBufAppendPrintf(QMessage, "%ld", MyQItem->Submitted);
 322
 323         if (StrLength(MyQItem->BounceTo) > 0) {
 324                 StrBufAppendBufPlain(QMessage, HKEY("\nbounceto|"), 0);
 325                 StrBufAppendBuf(QMessage, MyQItem->BounceTo, 0);
 326         }
 327
 328         if (StrLength(MyQItem->EnvelopeFrom) > 0) {
 329                 StrBufAppendBufPlain(QMessage, HKEY("\nenvelope_from|"), 0);
 330                 StrBufAppendBuf(QMessage, MyQItem->EnvelopeFrom, 0);
 331         }
 332
 333         if (StrLength(MyQItem->SenderRoom) > 0) {
 334                 StrBufAppendBufPlain(QMessage, HKEY("\nsource_room|"), 0);
 335                 StrBufAppendBuf(QMessage, MyQItem->SenderRoom, 0);
 336         }
 337
 338         StrBufAppendBufPlain(QMessage, HKEY("\nretry|"), 0);
 339         StrBufAppendPrintf(QMessage, "%ld",
 340                            MyQItem->Retry);
 341
 342         StrBufAppendBufPlain(QMessage, HKEY("\nattempted|"), 0);
 343         StrBufAppendPrintf(QMessage, "%ld",
 344                            ctdl_ev_now() + MyQItem->Retry);
 345
 346         It = GetNewHashPos(MyQItem->MailQEntries, 0);
 347         while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
 348         {
 349                 MailQEntry *ThisItem = vQE;
 350
 351                 StrBufAppendBufPlain(QMessage, HKEY("\nremote|"), 0);
 352                 StrBufAppendBuf(QMessage, ThisItem->Recipient, 0);
 353                 StrBufAppendBufPlain(QMessage, HKEY("|"), 0);
 354                 StrBufAppendPrintf(QMessage, "%d", ThisItem->Status);
 355                 StrBufAppendBufPlain(QMessage, HKEY("|"), 0);
 356                 StrBufAppendBuf(QMessage, ThisItem->StatusMessage, 0);
 357         }
 358         DeleteHashPos(&It);
 359         StrBufAppendBufPlain(QMessage, HKEY("\n"), 0);
 360         return QMessage;
 361 }
 362
 363
 364
 365
 366
 367 void NewMailQEntry(OneQueItem *Item)
 368 {
 369         Item->Current = (MailQEntry*) malloc(sizeof(MailQEntry));
 370         memset(Item->Current, 0, sizeof(MailQEntry));
 371
 372         if (Item->MailQEntries == NULL)
 373                 Item->MailQEntries = NewHash(1, Flathash);
 374         Item->Current->StatusMessage = NewStrBuf();
 375         Item->Current->n = GetCount(Item->MailQEntries);
 376         Put(Item->MailQEntries,
 377             IKEY(Item->Current->n),
 378             Item->Current,
 379             FreeMailQEntry);
 380 }
 381
 382 void QItem_Handle_MsgID(OneQueItem *Item, StrBuf *Line, const char **Pos)
 383 {
 384         Item->MessageID = StrBufExtractNext_long(Line, Pos, '|');
 385 }
 386
 387 void QItem_Handle_EnvelopeFrom(OneQueItem *Item, StrBuf *Line, const char **Pos)
 388 {
 389         if (Item->EnvelopeFrom == NULL)
 390                 Item->EnvelopeFrom = NewStrBufPlain(NULL, StrLength(Line));
 391         StrBufExtract_NextToken(Item->EnvelopeFrom, Line, Pos, '|');
 392 }
 393
 394 void QItem_Handle_BounceTo(OneQueItem *Item, StrBuf *Line, const char **Pos)
 395 {
 396         if (Item->BounceTo == NULL)
 397                 Item->BounceTo = NewStrBufPlain(NULL, StrLength(Line));
 398         StrBufExtract_NextToken(Item->BounceTo, Line, Pos, '|');
 399 }
 400
 401 void QItem_Handle_SenderRoom(OneQueItem *Item, StrBuf *Line, const char **Pos)
 402 {
 403         if (Item->SenderRoom == NULL)
 404                 Item->SenderRoom = NewStrBufPlain(NULL, StrLength(Line));
 405         StrBufExtract_NextToken(Item->SenderRoom, Line, Pos, '|');
 406 }
 407
 408 void QItem_Handle_Recipient(OneQueItem *Item, StrBuf *Line, const char **Pos)
 409 {
 410         if (Item->Current == NULL)
 411                 NewMailQEntry(Item);
 412         if (Item->Current->Recipient == NULL)
 413                 Item->Current->Recipient=NewStrBufPlain(NULL, StrLength(Line));
 414         StrBufExtract_NextToken(Item->Current->Recipient, Line, Pos, '|');
 415         Item->Current->Status = StrBufExtractNext_int(Line, Pos, '|');
 416         StrBufExtract_NextToken(Item->Current->StatusMessage, Line, Pos, '|');
 417         Item->Current = NULL; // TODO: is this always right?
 418 }
 419
 420
 421 void QItem_Handle_retry(OneQueItem *Item, StrBuf *Line, const char **Pos)
 422 {
 423         Item->Retry =
 424                 StrBufExtractNext_int(Line, Pos, '|');
 425         if (Item->Retry == 0)
 426                 Item->Retry = SMTP_RETRY_INTERVAL;
 427         else
 428                 Item->Retry *= 2;
 429 }
 430
 431
 432 void QItem_Handle_Submitted(OneQueItem *Item, StrBuf *Line, const char **Pos)
 433 {
 434         Item->Submitted = atol(*Pos);
 435
 436 }
 437
 438 void QItem_Handle_Attempted(OneQueItem *Item, StrBuf *Line, const char **Pos)
 439 {
 440         Item->ReattemptWhen = StrBufExtractNext_int(Line, Pos, '|');
 441 }
 442
 443
 444
 445 /**
 446  * this one has to have the context for loading the message via the redirect buffer...
 447  */
 448 StrBuf *smtp_load_msg(OneQueItem *MyQItem, int n)
 449 {
 450         CitContext *CCC=CC;
 451         StrBuf *SendMsg;
 452
 453         CCC->redirect_buffer = NewStrBufPlain(NULL, SIZ);
 454         CtdlOutputMsg(MyQItem->MessageID,
 455                       MT_RFC822, HEADERS_ALL,
 456                       0, 1, NULL,
 457                       (ESC_DOT|SUPPRESS_ENV_TO) );
 458
 459         SendMsg = CCC->redirect_buffer;
 460         CCC->redirect_buffer = NULL;
 461         if ((StrLength(SendMsg) > 0) &&
 462             ChrPtr(SendMsg)[StrLength(SendMsg) - 1] != '\n') {
 463                 syslog(LOG_WARNING,
 464                        "SMTP client[%d]: Possible problem: message did not "
 465                        "correctly terminate. (expecting 0x10, got 0x%02x)\n",
 466                        MsgCount, //yes uncool, but best choice here...
 467                        ChrPtr(SendMsg)[StrLength(SendMsg) - 1] );
 468                 StrBufAppendBufPlain(SendMsg, HKEY("\r\n"), 0);
 469         }
 470         return SendMsg;
 471 }
 472
 473
 474
 475 /*
 476  * smtp_do_bounce() is caled by smtp_do_procmsg() to scan a set of delivery
 477  * instructions for "5" codes (permanent fatal errors) and produce/deliver
 478  * a "bounce" message (delivery status notification).
 479  */
 480 void smtpq_do_bounce(OneQueItem *MyQItem, StrBuf *OMsgTxt)
 481 {
 482         static int seq = 0;
 483
 484         struct CtdlMessage *bmsg = NULL;
 485         StrBuf *boundary;
 486         StrBuf *Msg = NULL;
 487         StrBuf *BounceMB;
 488         struct recptypes *valid;
 489         time_t now;
 490
 491         HashPos *It;
 492         void *vQE;
 493         long len;
 494         const char *Key;
 495
 496         int first_attempt = 0;
 497         int successful_bounce = 0;
 498         int num_bounces = 0;
 499         int give_up = 0;
 500
 501         syslog(LOG_DEBUG, "smtp_do_bounce() called\n");
 502
 503         if (MyQItem->SendBounceMail == 0)
 504                 return;
 505
 506         now = ev_time();
 507
 508         if ( (now - MyQItem->Submitted) > SMTP_GIVE_UP ) {
 509                 give_up = 1;
 510         }
 511
 512         if (MyQItem->Retry == SMTP_RETRY_INTERVAL) {
 513                 first_attempt = 1;
 514         }
 515
 516         /*
 517          * Now go through the instructions checking for stuff.
 518          */
 519         Msg = NewStrBufPlain(NULL, 1024);
 520         It = GetNewHashPos(MyQItem->MailQEntries, 0);
 521         while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
 522         {
 523                 MailQEntry *ThisItem = vQE;
 524                 if ((ThisItem->Active && (ThisItem->Status == 5)) || /* failed now? */
 525                     ((give_up == 1) && (ThisItem->Status != 2)) ||
 526                     ((first_attempt == 1) && (ThisItem->Status != 2)))
 527                         /* giving up after failed attempts... */
 528                 {
 529                         ++num_bounces;
 530
 531                         StrBufAppendBuf(Msg, ThisItem->Recipient, 0);
 532                         StrBufAppendBufPlain(Msg, HKEY(": "), 0);
 533                         StrBufAppendBuf(Msg, ThisItem->StatusMessage, 0);
 534                         StrBufAppendBufPlain(Msg, HKEY("\r\n"), 0);
 535                 }
 536         }
 537         DeleteHashPos(&It);
 538
 539         /* Deliver the bounce if there's anything worth mentioning */
 540         syslog(LOG_DEBUG, "num_bounces = %d\n", num_bounces);
 541
 542         if (num_bounces == 0) {
 543                 FreeStrBuf(&Msg);
 544                 return;
 545         }
 546
 547         boundary = NewStrBufPlain(HKEY("=_Citadel_Multipart_"));
 548         StrBufAppendPrintf(boundary,
 549                            "%s_%04x%04x",
 550                            config.c_fqdn,
 551                            getpid(),
 552                            ++seq);
 553
 554         /* Start building our bounce message; go shopping for memory first. */
 555         BounceMB = NewStrBufPlain(
 556                 NULL,
 557                 1024 + /* mime stuff.... */
 558                 StrLength(Msg) +  /* the bounce information... */
 559                 StrLength(OMsgTxt)); /* the original message */
 560         if (BounceMB == NULL) {
 561                 FreeStrBuf(&boundary);
 562                 syslog(LOG_ERR, "Failed to alloc() bounce message.\n");
 563
 564                 return;
 565         }
 566
 567         bmsg = (struct CtdlMessage *) malloc(sizeof(struct CtdlMessage));
 568         if (bmsg == NULL) {
 569                 FreeStrBuf(&boundary);
 570                 FreeStrBuf(&BounceMB);
 571                 syslog(LOG_ERR, "Failed to alloc() bounce message.\n");
 572
 573                 return;
 574         }
 575         memset(bmsg, 0, sizeof(struct CtdlMessage));
 576
 577
 578         StrBufAppendBufPlain(BounceMB, HKEY("Content-type: multipart/mixed; boundary=\""), 0);
 579         StrBufAppendBuf(BounceMB, boundary, 0);
 580         StrBufAppendBufPlain(BounceMB, HKEY("\"\r\n"), 0);
 581         StrBufAppendBufPlain(BounceMB, HKEY("MIME-Version: 1.0\r\n"), 0);
 582         StrBufAppendBufPlain(BounceMB, HKEY("X-Mailer: " CITADEL "\r\n"), 0);
 583         StrBufAppendBufPlain(BounceMB, HKEY("\r\nThis is a multipart message in MIME format.\r\n\r\n"), 0);
 584         StrBufAppendBufPlain(BounceMB, HKEY("--"), 0);
 585         StrBufAppendBuf(BounceMB, boundary, 0);
 586         StrBufAppendBufPlain(BounceMB, HKEY("\r\n"), 0);
 587         StrBufAppendBufPlain(BounceMB, HKEY("Content-type: text/plain\r\n\r\n"), 0);
 588
 589         if (give_up)
 590                 StrBufAppendBufPlain(
 591                         BounceMB,
 592                         HKEY(
 593                                 "A message you sent could not be delivered "
 594                                 "to some or all of its recipients\n"
 595                                 "due to prolonged unavailability "
 596                                 "of its destination(s).\n"
 597                                 "Giving up on the following addresses:\n\n"
 598                                 ), 0);
 599         else
 600                 StrBufAppendBufPlain(
 601                         BounceMB,
 602                         HKEY(
 603                                 "A message you sent could not be delivered "
 604                                 "to some or all of its recipients.\n"
 605                                 "The following addresses "
 606                                 "were undeliverable:\n\n"
 607                                 ), 0);
 608
 609         StrBufAppendBuf(BounceMB, Msg, 0);
 610         FreeStrBuf(&Msg);
 611
 612         if (StrLength(MyQItem->SenderRoom) > 0)
 613         {
 614                 StrBufAppendBufPlain(
 615                         BounceMB,
 616                         HKEY("The message was originaly posted in: "), 0);
 617                 StrBufAppendBuf(BounceMB, MyQItem->SenderRoom, 0);
 618                 StrBufAppendBufPlain(
 619                         BounceMB,
 620                         HKEY("\n"), 0);
 621         }
 622
 623         /* Attach the original message */
 624         StrBufAppendBufPlain(BounceMB, HKEY("--"), 0);
 625         StrBufAppendBuf(BounceMB, boundary, 0);
 626         StrBufAppendBufPlain(BounceMB, HKEY("\r\n"), 0);
 627         StrBufAppendBufPlain(BounceMB,
 628                              HKEY("Content-type: message/rfc822\r\n"), 0);
 629         StrBufAppendBufPlain(BounceMB,
 630                              HKEY("Content-Transfer-Encoding: 7bit\r\n"), 0);
 631         StrBufAppendBufPlain(BounceMB,
 632                              HKEY("Content-Disposition: inline\r\n"), 0);
 633         StrBufAppendBufPlain(BounceMB, HKEY("\r\n"), 0);
 634         StrBufAppendBuf(BounceMB, OMsgTxt, 0);
 635
 636         /* Close the multipart MIME scope */
 637         StrBufAppendBufPlain(BounceMB, HKEY("--"), 0);
 638         StrBufAppendBuf(BounceMB, boundary, 0);
 639         StrBufAppendBufPlain(BounceMB, HKEY("--\r\n"), 0);
 640
 641         bmsg->cm_magic = CTDLMESSAGE_MAGIC;
 642         bmsg->cm_anon_type = MES_NORMAL;
 643         bmsg->cm_format_type = FMT_RFC822;
 644
 645         bmsg->cm_fields['O'] = strdup(MAILROOM);
 646         bmsg->cm_fields['A'] = strdup("Citadel");
 647         bmsg->cm_fields['N'] = strdup(config.c_nodename);
 648         bmsg->cm_fields['U'] = strdup("Delivery Status Notification (Failure)");
 649         bmsg->cm_fields['M'] = SmashStrBuf(&BounceMB);
 650
 651         /* First try the user who sent the message */
 652         if (StrLength(MyQItem->BounceTo) == 0)
 653                 syslog(LOG_ERR, "No bounce address specified\n");
 654         else
 655                 syslog(LOG_DEBUG, "bounce to user? <%s>\n",
 656                        ChrPtr(MyQItem->BounceTo));
 657
 658         /* Can we deliver the bounce to the original sender? */
 659         valid = validate_recipients(ChrPtr(MyQItem->BounceTo), NULL, 0);
 660         if ((valid != NULL) && (valid->num_error == 0)) {
 661                 CtdlSubmitMsg(bmsg, valid, "", QP_EADDR);
 662                 successful_bounce = 1;
 663         }
 664
 665         /* If not, post it in the Aide> room */
 666         if (successful_bounce == 0) {
 667                 CtdlSubmitMsg(bmsg, NULL, config.c_aideroom, QP_EADDR);
 668         }
 669
 670         /* Free up the memory we used */
 671         free_recipients(valid);
 672         FreeStrBuf(&boundary);
 673         CtdlFreeMessage(bmsg);
 674         syslog(LOG_DEBUG, "Done processing bounces\n");
 675 }
 676
 677 /*
 678  * smtp_do_procmsg()
 679  *
 680  * Called by smtp_do_queue() to handle an individual message.
 681  */
 682 void smtp_do_procmsg(long msgnum, void *userdata) {
 683         int mynumsessions = num_sessions;
 684         struct CtdlMessage *msg = NULL;
 685         char *instr = NULL;
 686         StrBuf *PlainQItem;
 687         OneQueItem *MyQItem;
 688         char *pch;
 689         HashPos  *It;
 690         void *vQE;
 691         long len;
 692         const char *Key;
 693         int nRelays = 0;
 694         ParsedURL *RelayUrls = NULL;
 695         int HaveBuffers = 0;
 696         StrBuf *Msg =NULL;
 697
 698         if (mynumsessions > max_sessions_for_outbound_smtp) {
 699                 syslog(LOG_DEBUG,
 700                        "SMTP Queue: skipping because of num jobs %d > %d max_sessions_for_outbound_smtp",
 701                        mynumsessions,
 702                        max_sessions_for_outbound_smtp);
 703         }
 704
 705         syslog(LOG_DEBUG, "SMTP Queue: smtp_do_procmsg(%ld)\n", msgnum);
 706         ///strcpy(envelope_from, "");
 707
 708         msg = CtdlFetchMessage(msgnum, 1);
 709         if (msg == NULL) {
 710                 syslog(LOG_ERR, "SMTP Queue: tried %ld but no such message!\n",
 711                        msgnum);
 712                 return;
 713         }
 714
 715         pch = instr = msg->cm_fields['M'];
 716
 717         /* Strip out the headers (no not amd any other non-instruction) line */
 718         while (pch != NULL) {
 719                 pch = strchr(pch, '\n');
 720                 if ((pch != NULL) && (*(pch + 1) == '\n')) {
 721                         instr = pch + 2;
 722                         pch = NULL;
 723                 }
 724         }
 725         PlainQItem = NewStrBufPlain(instr, -1);
 726         CtdlFreeMessage(msg);
 727         MyQItem = DeserializeQueueItem(PlainQItem, msgnum);
 728         FreeStrBuf(&PlainQItem);
 729
 730         if (MyQItem == NULL) {
 731                 syslog(LOG_ERR,
 732                        "SMTP Queue: Msg No %ld: already in progress!\n",
 733                        msgnum);
 734                 return; /* s.b. else is already processing... */
 735         }
 736
 737         /*
 738          * Postpone delivery if we've already tried recently.
 739          */
 740         if ((MyQItem->ReattemptWhen != 0) &&
 741             (time(NULL) < MyQItem->ReattemptWhen) &&
 742             (run_queue_now == 0))
 743         {
 744                 syslog(LOG_DEBUG, "SMTP client: Retry time not yet reached.\n");
 745
 746                 It = GetNewHashPos(MyQItem->MailQEntries, 0);
 747                 pthread_mutex_lock(&ActiveQItemsLock);
 748                 {
 749                         if (GetHashPosFromKey(ActiveQItems,
 750                                               LKEY(MyQItem->MessageID),
 751                                               It))
 752                         {
 753                                 DeleteEntryFromHash(ActiveQItems, It);
 754                         }
 755                 }
 756                 pthread_mutex_unlock(&ActiveQItemsLock);
 757                 ////FreeQueItem(&MyQItem); TODO: DeleteEntryFromHash frees this?
 758                 DeleteHashPos(&It);
 759                 return;
 760         }
 761
 762         /*
 763          * Bail out if there's no actual message associated with this
 764          */
 765         if (MyQItem->MessageID < 0L) {
 766                 syslog(LOG_ERR, "SMTP Queue: no 'msgid' directive found!\n");
 767                 It = GetNewHashPos(MyQItem->MailQEntries, 0);
 768                 pthread_mutex_lock(&ActiveQItemsLock);
 769                 {
 770                         if (GetHashPosFromKey(ActiveQItems,
 771                                               LKEY(MyQItem->MessageID),
 772                                               It))
 773                         {
 774                                 DeleteEntryFromHash(ActiveQItems, It);
 775                         }
 776                 }
 777                 pthread_mutex_unlock(&ActiveQItemsLock);
 778                 DeleteHashPos(&It);
 779                 ////FreeQueItem(&MyQItem); TODO: DeleteEntryFromHash frees this?
 780                 return;
 781         }
 782
 783         {
 784                 char mxbuf[SIZ];
 785                 ParsedURL **Url = &MyQItem->URL;
 786                 nRelays = get_hosts(mxbuf, "smarthost");
 787                 if (nRelays > 0) {
 788                         StrBuf *All;
 789                         StrBuf *One;
 790                         const char *Pos = NULL;
 791                         All = NewStrBufPlain(mxbuf, -1);
 792                         One = NewStrBufPlain(NULL, StrLength(All) + 1);
 793
 794                         while ((Pos != StrBufNOTNULL) &&
 795                                ((Pos == NULL) ||
 796                                 !IsEmptyStr(Pos)))
 797                         {
 798                                 StrBufExtract_NextToken(One, All, &Pos, '|');
 799                                 if (!ParseURL(Url, One, 25))
 800                                         syslog(LOG_DEBUG,
 801                                                "Failed to parse: %s\n",
 802                                                ChrPtr(One));
 803                                 else {
 804                                         ///if (!Url->IsIP)) // todo dupe me fork ipv6
 805                                         Url = &(*Url)->Next;
 806                                 }
 807                         }
 808                         FreeStrBuf(&All);
 809                         FreeStrBuf(&One);
 810                 }
 811
 812                 Url = &MyQItem->FallBackHost;
 813                 nRelays = get_hosts(mxbuf, "fallbackhost");
 814                 if (nRelays > 0) {
 815                         StrBuf *All;
 816                         StrBuf *One;
 817                         const char *Pos = NULL;
 818                         All = NewStrBufPlain(mxbuf, -1);
 819                         One = NewStrBufPlain(NULL, StrLength(All) + 1);
 820
 821                         while ((Pos != StrBufNOTNULL) &&
 822                                ((Pos == NULL) ||
 823                                 !IsEmptyStr(Pos)))
 824                         {
 825                                 StrBufExtract_NextToken(One, All, &Pos, '|');
 826                                 if (!ParseURL(Url, One, 25))
 827                                         syslog(LOG_DEBUG,
 828                                                "Failed to parse: %s\n",
 829                                                ChrPtr(One));
 830                                 else
 831                                         Url = &(*Url)->Next;
 832                         }
 833                         FreeStrBuf(&All);
 834                         FreeStrBuf(&One);
 835                 }
 836         }
 837
 838         It = GetNewHashPos(MyQItem->MailQEntries, 0);
 839         while (GetNextHashPos(MyQItem->MailQEntries, It, &len, &Key, &vQE))
 840         {
 841                 MailQEntry *ThisItem = vQE;
 842                 syslog(LOG_DEBUG, "SMTP Queue: Task: <%s> %d\n",
 843                        ChrPtr(ThisItem->Recipient),
 844                        ThisItem->Active);
 845         }
 846         DeleteHashPos(&It);
 847
 848         MyQItem->ActiveDeliveries = CountActiveQueueEntries(MyQItem);
 849
 850         /* failsafe against overload:
 851          * will we exceed the limit set?
 852          */
 853         if ((MyQItem->ActiveDeliveries + mynumsessions > max_sessions_for_outbound_smtp) &&
 854             /* if yes, did we reach more than half of the quota? */
 855             ((mynumsessions * 2) > max_sessions_for_outbound_smtp) &&
 856             /* if... would we ever fit into half of the quota?? */
 857             (((MyQItem->ActiveDeliveries * 2)  < max_sessions_for_outbound_smtp)))
 858         {
 859                 /* abort delivery for another time. */
 860                 syslog(LOG_DEBUG,
 861                        "SMTP Queue: skipping because of num jobs %d + %ld > %d max_sessions_for_outbound_smtp",
 862                        mynumsessions,
 863                        MyQItem->ActiveDeliveries,
 864                        max_sessions_for_outbound_smtp);
 865
 866                 FreeQueItem(&MyQItem);
 867
 868                 return;
 869         }
 870
 871
 872         if (MyQItem->ActiveDeliveries > 0)
 873         {
 874                 int nActivated = 0;
 875                 int n = MsgCount++;
 876                 int m = MyQItem->ActiveDeliveries;
 877                 int i = 1;
 878                 Msg = smtp_load_msg(MyQItem, n);
 879                 It = GetNewHashPos(MyQItem->MailQEntries, 0);
 880                 while ((i <= m) &&
 881                        (GetNextHashPos(MyQItem->MailQEntries,
 882                                        It, &len, &Key, &vQE)))
 883                 {
 884                         MailQEntry *ThisItem = vQE;
 885
 886                         if (ThisItem->Active == 1)
 887                         {
 888                                 int KeepBuffers = (i == m);
 889
 890                                 nActivated++;
 891                                 if (nActivated % ndelay_count == 0)
 892                                         usleep(delay_msec);
 893
 894                                 if (i > 1) n = MsgCount++;
 895                                 syslog(LOG_DEBUG,
 896                                        "SMTPQ: Trying <%ld> <%s> %d / %d \n",
 897                                        MyQItem->MessageID,
 898                                        ChrPtr(ThisItem->Recipient),
 899                                        i,
 900                                        m);
 901                                 smtp_try_one_queue_entry(MyQItem,
 902                                                          ThisItem,
 903                                                          Msg,
 904                                                          KeepBuffers,
 905                                                          n,
 906                                                          RelayUrls);
 907
 908                                 if (KeepBuffers) HaveBuffers = 1;
 909
 910                                 i++;
 911                         }
 912                 }
 913                 DeleteHashPos(&It);
 914         }
 915         else
 916         {
 917                 It = GetNewHashPos(MyQItem->MailQEntries, 0);
 918                 pthread_mutex_lock(&ActiveQItemsLock);
 919                 {
 920                         if (GetHashPosFromKey(ActiveQItems,
 921                                               LKEY(MyQItem->MessageID),
 922                                               It))
 923                         {
 924                                 DeleteEntryFromHash(ActiveQItems, It);
 925                         }
 926                         else
 927                         {
 928                                 long len;
 929                                 const char* Key;
 930                                 void *VData;
 931
 932                                 syslog(LOG_WARNING,
 933                                        "SMTP cleanup: unable to find "
 934                                        "QItem with ID[%ld]",
 935                                        MyQItem->MessageID);
 936                                 while (GetNextHashPos(ActiveQItems,
 937                                                       It,
 938                                                       &len,
 939                                                       &Key,
 940                                                       &VData))
 941                                 {
 942                                         syslog(LOG_WARNING,
 943                                                "SMTP cleanup: have: ID[%ld]",
 944                                               ((OneQueItem *)VData)->MessageID);
 945                                 }
 946                         }
 947
 948                 }
 949                 pthread_mutex_unlock(&ActiveQItemsLock);
 950                 DeleteHashPos(&It);
 951                 ////FreeQueItem(&MyQItem); TODO: DeleteEntryFromHash frees this?
 952
 953 // TODO: bounce & delete?
 954
 955         }
 956         if (!HaveBuffers) {
 957                 FreeStrBuf (&Msg);
 958 // TODO : free RelayUrls
 959         }
 960 }
 961
 962
 963
 964 /*
 965  * smtp_queue_thread()
 966  *
 967  * Run through the queue sending out messages.
 968  */
 969 void smtp_do_queue(void) {
 970         static int is_running = 0;
 971         int num_processed = 0;
 972
 973         if (is_running)
 974                 return; /* Concurrency check - only one can run */
 975         is_running = 1;
 976
 977         pthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
 978         syslog(LOG_INFO, "SMTP client: processing outbound queue");
 979
 980         if (CtdlGetRoom(&CC->room, SMTP_SPOOLOUT_ROOM) != 0) {
 981                 syslog(LOG_ERR, "Cannot find room <%s>", SMTP_SPOOLOUT_ROOM);
 982         }
 983         else {
 984                 num_processed = CtdlForEachMessage(MSGS_ALL,
 985                                                    0L,
 986                                                    NULL,
 987                                                    SPOOLMIME,
 988                                                    NULL,
 989                                                    smtp_do_procmsg,
 990                                                    NULL);
 991         }
 992         syslog(LOG_INFO,
 993                "SMTP client: queue run completed; %d messages processed",
 994                num_processed);
 995
 996         run_queue_now = 0;
 997         is_running = 0;
 998 }
 999
1000
1001
1002 /*
1003  * Initialize the SMTP outbound queue
1004  */
1005 void smtp_init_spoolout(void) {
1006         struct ctdlroom qrbuf;
1007
1008         /*
1009          * Create the room.  This will silently fail if the room already
1010          * exists, and that's perfectly ok, because we want it to exist.
1011          */
1012         CtdlCreateRoom(SMTP_SPOOLOUT_ROOM, 3, "", 0, 1, 0, VIEW_QUEUE);
1013
1014         /*
1015          * Make sure it's set to be a "system room" so it doesn't show up
1016          * in the <K>nown rooms list for Aides.
1017          */
1018         if (CtdlGetRoomLock(&qrbuf, SMTP_SPOOLOUT_ROOM) == 0) {
1019                 qrbuf.QRflags2 |= QR2_SYSTEM;
1020                 CtdlPutRoomLock(&qrbuf);
1021         }
1022 }
1023
1024
1025
1026
1027 /*****************************************************************************/
1028 /*                          SMTP UTILITY COMMANDS                            */
1029 /*****************************************************************************/
1030
1031 void cmd_smtp(char *argbuf) {
1032         char cmd[64];
1033         char node[256];
1034         char buf[1024];
1035         int i;
1036         int num_mxhosts;
1037
1038         if (CtdlAccessCheck(ac_aide)) return;
1039
1040         extract_token(cmd, argbuf, 0, '|', sizeof cmd);
1041
1042         if (!strcasecmp(cmd, "mx")) {
1043                 extract_token(node, argbuf, 1, '|', sizeof node);
1044                 num_mxhosts = getmx(buf, node);
1045                 cprintf("%d %d MX hosts listed for %s\n",
1046                         LISTING_FOLLOWS, num_mxhosts, node);
1047                 for (i=0; i<num_mxhosts; ++i) {
1048                         extract_token(node, buf, i, '|', sizeof node);
1049                         cprintf("%s\n", node);
1050                 }
1051                 cprintf("000\n");
1052                 return;
1053         }
1054
1055         else if (!strcasecmp(cmd, "runqueue")) {
1056                 run_queue_now = 1;
1057                 cprintf("%d All outbound SMTP will be retried now.\n", CIT_OK);
1058                 return;
1059         }
1060
1061         else {
1062                 cprintf("%d Invalid command.\n", ERROR + ILLEGAL_VALUE);
1063         }
1064
1065 }
1066
1067
1068 CTDL_MODULE_INIT(smtp_queu)
1069 {
1070         char *pstr;
1071
1072         if (!threading)
1073         {
1074                 pstr = getenv("CITSERVER_n_session_max");
1075                 if ((pstr != NULL) && (*pstr != '\0'))
1076                         max_sessions_for_outbound_smtp = atol(pstr); /* how many sessions might be active till we stop adding more smtp jobs */
1077
1078                 pstr = getenv("CITSERVER_smtp_n_delay_count");
1079                 if ((pstr != NULL) && (*pstr != '\0'))
1080                         ndelay_count = atol(pstr); /* every n queued messages we will sleep... */
1081
1082                 pstr = getenv("CITSERVER_smtp_delay");
1083                 if ((pstr != NULL) && (*pstr != '\0'))
1084                         delay_msec = atol(pstr) * 1000; /* this many seconds. */
1085
1086
1087
1088
1089                 CtdlFillSystemContext(&smtp_queue_CC, "SMTP_Send");
1090                 ActiveQItems = NewHash(1, lFlathash);
1091                 pthread_mutex_init(&ActiveQItemsLock, NULL);
1092
1093                 QItemHandlers = NewHash(0, NULL);
1094
1095                 Put(QItemHandlers, HKEY("msgid"), QItem_Handle_MsgID, reference_free_handler);
1096                 Put(QItemHandlers, HKEY("envelope_from"), QItem_Handle_EnvelopeFrom, reference_free_handler);
1097                 Put(QItemHandlers, HKEY("retry"), QItem_Handle_retry, reference_free_handler);
1098                 Put(QItemHandlers, HKEY("attempted"), QItem_Handle_Attempted, reference_free_handler);
1099                 Put(QItemHandlers, HKEY("remote"), QItem_Handle_Recipient, reference_free_handler);
1100                 Put(QItemHandlers, HKEY("bounceto"), QItem_Handle_BounceTo, reference_free_handler);
1101                 Put(QItemHandlers, HKEY("source_room"), QItem_Handle_SenderRoom, reference_free_handler);
1102                 Put(QItemHandlers, HKEY("submitted"), QItem_Handle_Submitted, reference_free_handler);
1103                 smtp_init_spoolout();
1104
1105                 CtdlRegisterEVCleanupHook(smtp_evq_cleanup);
1106
1107                 CtdlRegisterProtoHook(cmd_smtp, "SMTP", "SMTP utility commands");
1108                 CtdlRegisterSessionHook(smtp_do_queue, EVT_TIMER);
1109         }
1110
1111         /* return our Subversion id for the Log */
1112         return "smtpeventclient";
1113 }