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