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