1 /* $NetBSD: qmgr_deliver.c,v 1.3 2022/10/08 16:12:48 christos Exp $ */
2
3 /*++
4 /* NAME
5 /* qmgr_deliver 3
6 /* SUMMARY
7 /* deliver one per-site queue entry to that site
8 /* SYNOPSIS
9 /* #include "qmgr.h"
10 /*
11 /* int qmgr_deliver_concurrency;
12 /*
13 /* int qmgr_deliver(transport, fp)
14 /* QMGR_TRANSPORT *transport;
15 /* VSTREAM *fp;
16 /* DESCRIPTION
17 /* This module implements the client side of the `queue manager
18 /* to delivery agent' protocol. The queue manager uses
19 /* asynchronous I/O so that it can drive multiple delivery
20 /* agents in parallel. Depending on the outcome of a delivery
21 /* attempt, the status of messages, queues and transports is
22 /* updated.
23 /*
24 /* qmgr_deliver_concurrency is a global counter that says how
25 /* many delivery processes are in use. This can be used, for
26 /* example, to control the size of the `active' message queue.
27 /*
28 /* qmgr_deliver() executes when a delivery process announces its
29 /* availability for the named transport. It arranges for delivery
30 /* of a suitable queue entry. The \fIfp\fR argument specifies a
31 /* stream that is connected to a delivery process, or a null
32 /* pointer if the transport accepts no connection. Upon completion
33 /* of delivery (successful or not), the stream is closed, so that the
34 /* delivery process is released.
35 /* DIAGNOSTICS
36 /* LICENSE
37 /* .ad
38 /* .fi
39 /* The Secure Mailer license must be distributed with this software.
40 /* AUTHOR(S)
41 /* Wietse Venema
42 /* IBM T.J. Watson Research
43 /* P.O. Box 704
44 /* Yorktown Heights, NY 10598, USA
45 /*
46 /* Preemptive scheduler enhancements:
47 /* Patrik Rak
48 /* Modra 6
49 /* 155 00, Prague, Czech Republic
50 /*
51 /* Wietse Venema
52 /* Google, Inc.
53 /* 111 8th Avenue
54 /* New York, NY 10011, USA
55 /*--*/
56
57 /* System library. */
58
59 #include <sys_defs.h>
60 #include <time.h>
61 #include <string.h>
62
63 /* Utility library. */
64
65 #include <msg.h>
66 #include <vstring.h>
67 #include <vstream.h>
68 #include <vstring_vstream.h>
69 #include <events.h>
70 #include <iostuff.h>
71 #include <stringops.h>
72 #include <mymalloc.h>
73
74 /* Global library. */
75
76 #include <mail_queue.h>
77 #include <mail_proto.h>
78 #include <recipient_list.h>
79 #include <mail_params.h>
80 #include <deliver_request.h>
81 #include <verp_sender.h>
82 #include <dsn_util.h>
83 #include <dsn_buf.h>
84 #include <dsb_scan.h>
85 #include <rcpt_print.h>
86 #include <smtputf8.h>
87
88 /* Application-specific. */
89
90 #include "qmgr.h"
91
92 /*
93 * Important note on the _transport_rate_delay implementation: after
94 * qmgr_transport_alloc() sets the QMGR_TRANSPORT_STAT_RATE_LOCK flag, all
95 * code paths must directly or indirectly invoke qmgr_transport_unthrottle()
96 * or qmgr_transport_throttle(). Otherwise, transports with non-zero
97 * _transport_rate_delay will become stuck.
98 */
99
100 int qmgr_deliver_concurrency;
101
102 /*
103 * Message delivery status codes.
104 */
105 #define DELIVER_STAT_OK 0 /* all recipients delivered */
106 #define DELIVER_STAT_DEFER 1 /* try some recipients later */
107 #define DELIVER_STAT_CRASH 2 /* mailer internal problem */
108
109 /* qmgr_deliver_initial_reply - retrieve initial delivery process response */
110
qmgr_deliver_initial_reply(VSTREAM * stream)111 static int qmgr_deliver_initial_reply(VSTREAM *stream)
112 {
113 if (peekfd(vstream_fileno(stream)) < 0) {
114 msg_warn("%s: premature disconnect", VSTREAM_PATH(stream));
115 return (DELIVER_STAT_CRASH);
116 } else if (attr_scan(stream, ATTR_FLAG_STRICT,
117 RECV_ATTR_STREQ(MAIL_ATTR_PROTO, MAIL_ATTR_PROTO_DELIVER),
118 ATTR_TYPE_END) != 0) {
119 msg_warn("%s: malformed response", VSTREAM_PATH(stream));
120 return (DELIVER_STAT_DEFER);
121 } else {
122 return (0);
123 }
124 }
125
126 /* qmgr_deliver_final_reply - retrieve final delivery process response */
127
qmgr_deliver_final_reply(VSTREAM * stream,DSN_BUF * dsb)128 static int qmgr_deliver_final_reply(VSTREAM *stream, DSN_BUF *dsb)
129 {
130 int stat;
131
132 if (peekfd(vstream_fileno(stream)) < 0) {
133 msg_warn("%s: premature disconnect", VSTREAM_PATH(stream));
134 return (DELIVER_STAT_CRASH);
135 } else if (attr_scan(stream, ATTR_FLAG_STRICT,
136 RECV_ATTR_FUNC(dsb_scan, (void *) dsb),
137 RECV_ATTR_INT(MAIL_ATTR_STATUS, &stat),
138 ATTR_TYPE_END) != 2) {
139 msg_warn("%s: malformed response", VSTREAM_PATH(stream));
140 return (DELIVER_STAT_CRASH);
141 } else {
142 return (stat ? DELIVER_STAT_DEFER : 0);
143 }
144 }
145
146 /* qmgr_deliver_send_request - send delivery request to delivery process */
147
qmgr_deliver_send_request(QMGR_ENTRY * entry,VSTREAM * stream)148 static int qmgr_deliver_send_request(QMGR_ENTRY *entry, VSTREAM *stream)
149 {
150 RECIPIENT_LIST list = entry->rcpt_list;
151 RECIPIENT *recipient;
152 QMGR_MESSAGE *message = entry->message;
153 VSTRING *sender_buf = 0;
154 MSG_STATS stats;
155 char *sender;
156 int flags;
157 int smtputf8 = message->smtputf8;
158 const char *addr;
159
160 /*
161 * Todo: integrate with code up-stream that builds the delivery request.
162 */
163 for (recipient = list.info; recipient < list.info + list.len; recipient++)
164 if (var_smtputf8_enable && (addr = recipient->address)[0]
165 && !allascii(addr) && valid_utf8_string(addr, strlen(addr))) {
166 smtputf8 |= SMTPUTF8_FLAG_RECIPIENT;
167 if (message->verp_delims)
168 smtputf8 |= SMTPUTF8_FLAG_SENDER;
169 }
170
171 /*
172 * If variable envelope return path is requested, change prefix+@origin
173 * into prefix+user=domain@origin. Note that with VERP there is only one
174 * recipient per delivery.
175 */
176 if (message->verp_delims == 0) {
177 sender = message->sender;
178 } else {
179 sender_buf = vstring_alloc(100);
180 verp_sender(sender_buf, message->verp_delims,
181 message->sender, list.info);
182 sender = vstring_str(sender_buf);
183 }
184
185 flags = message->tflags
186 | entry->queue->dflags
187 | (message->inspect_xport ? DEL_REQ_FLAG_BOUNCE : DEL_REQ_FLAG_DEFLT);
188 (void) QMGR_MSG_STATS(&stats, message);
189 attr_print(stream, ATTR_FLAG_NONE,
190 SEND_ATTR_INT(MAIL_ATTR_FLAGS, flags),
191 SEND_ATTR_STR(MAIL_ATTR_QUEUE, message->queue_name),
192 SEND_ATTR_STR(MAIL_ATTR_QUEUEID, message->queue_id),
193 SEND_ATTR_LONG(MAIL_ATTR_OFFSET, message->data_offset),
194 SEND_ATTR_LONG(MAIL_ATTR_SIZE, message->cont_length),
195 SEND_ATTR_STR(MAIL_ATTR_NEXTHOP, entry->queue->nexthop),
196 SEND_ATTR_STR(MAIL_ATTR_ENCODING, message->encoding),
197 SEND_ATTR_INT(MAIL_ATTR_SMTPUTF8, smtputf8),
198 SEND_ATTR_STR(MAIL_ATTR_SENDER, sender),
199 SEND_ATTR_STR(MAIL_ATTR_DSN_ENVID, message->dsn_envid),
200 SEND_ATTR_INT(MAIL_ATTR_DSN_RET, message->dsn_ret),
201 SEND_ATTR_FUNC(msg_stats_print, (const void *) &stats),
202 /* XXX Should be encapsulated with ATTR_TYPE_FUNC. */
203 SEND_ATTR_STR(MAIL_ATTR_LOG_CLIENT_NAME, message->client_name),
204 SEND_ATTR_STR(MAIL_ATTR_LOG_CLIENT_ADDR, message->client_addr),
205 SEND_ATTR_STR(MAIL_ATTR_LOG_CLIENT_PORT, message->client_port),
206 SEND_ATTR_STR(MAIL_ATTR_LOG_PROTO_NAME, message->client_proto),
207 SEND_ATTR_STR(MAIL_ATTR_LOG_HELO_NAME, message->client_helo),
208 /* XXX Should be encapsulated with ATTR_TYPE_FUNC. */
209 SEND_ATTR_STR(MAIL_ATTR_SASL_METHOD, message->sasl_method),
210 SEND_ATTR_STR(MAIL_ATTR_SASL_USERNAME, message->sasl_username),
211 SEND_ATTR_STR(MAIL_ATTR_SASL_SENDER, message->sasl_sender),
212 /* XXX Ditto if we want to pass TLS certificate info. */
213 SEND_ATTR_STR(MAIL_ATTR_LOG_IDENT, message->log_ident),
214 SEND_ATTR_STR(MAIL_ATTR_RWR_CONTEXT, message->rewrite_context),
215 SEND_ATTR_INT(MAIL_ATTR_RCPT_COUNT, list.len),
216 ATTR_TYPE_END);
217 if (sender_buf != 0)
218 vstring_free(sender_buf);
219 for (recipient = list.info; recipient < list.info + list.len; recipient++)
220 attr_print(stream, ATTR_FLAG_NONE,
221 SEND_ATTR_FUNC(rcpt_print, (const void *) recipient),
222 ATTR_TYPE_END);
223 if (vstream_fflush(stream) != 0) {
224 msg_warn("write to process (%s): %m", entry->queue->transport->name);
225 return (-1);
226 } else {
227 if (msg_verbose)
228 msg_info("qmgr_deliver: site `%s'", entry->queue->name);
229 return (0);
230 }
231 }
232
233 /* qmgr_deliver_abort - transport response watchdog */
234
qmgr_deliver_abort(int unused_event,void * context)235 static void qmgr_deliver_abort(int unused_event, void *context)
236 {
237 QMGR_ENTRY *entry = (QMGR_ENTRY *) context;
238 QMGR_QUEUE *queue = entry->queue;
239 QMGR_TRANSPORT *transport = queue->transport;
240 QMGR_MESSAGE *message = entry->message;
241
242 msg_fatal("%s: timeout receiving delivery status from transport: %s",
243 message->queue_id, transport->name);
244 }
245
246 /* qmgr_deliver_update - process delivery status report */
247
qmgr_deliver_update(int unused_event,void * context)248 static void qmgr_deliver_update(int unused_event, void *context)
249 {
250 QMGR_ENTRY *entry = (QMGR_ENTRY *) context;
251 QMGR_QUEUE *queue = entry->queue;
252 QMGR_TRANSPORT *transport = queue->transport;
253 QMGR_MESSAGE *message = entry->message;
254 static DSN_BUF *dsb;
255 int status;
256
257 /*
258 * Release the delivery agent from a "hot" queue entry.
259 */
260 #define QMGR_DELIVER_RELEASE_AGENT(entry) do { \
261 event_disable_readwrite(vstream_fileno(entry->stream)); \
262 (void) vstream_fclose(entry->stream); \
263 entry->stream = 0; \
264 qmgr_deliver_concurrency--; \
265 } while (0)
266
267 if (dsb == 0)
268 dsb = dsb_create();
269
270 /*
271 * The message transport has responded. Stop the watchdog timer.
272 */
273 event_cancel_timer(qmgr_deliver_abort, context);
274
275 /*
276 * Retrieve the delivery agent status report. The numerical status code
277 * indicates if delivery should be tried again. The reason text is sent
278 * only when a site should be avoided for a while, so that the queue
279 * manager can log why it does not even try to schedule delivery to the
280 * affected recipients.
281 */
282 status = qmgr_deliver_final_reply(entry->stream, dsb);
283
284 /*
285 * The mail delivery process failed for some reason (although delivery
286 * may have been successful). Back off with this transport type for a
287 * while. Dispose of queue entries for this transport that await
288 * selection (the todo lists). Stay away from queue entries that have
289 * been selected (the busy lists), or we would have dangling pointers.
290 * The queue itself won't go away before we dispose of the current queue
291 * entry.
292 */
293 if (status == DELIVER_STAT_CRASH) {
294 message->flags |= DELIVER_STAT_DEFER;
295 #if 0
296 whatsup = concatenate("unknown ", transport->name,
297 " mail transport error", (char *) 0);
298 qmgr_transport_throttle(transport,
299 DSN_SIMPLE(&dsb->dsn, "4.3.0", whatsup));
300 myfree(whatsup);
301 #else
302 qmgr_transport_throttle(transport,
303 DSN_SIMPLE(&dsb->dsn, "4.3.0",
304 "unknown mail transport error"));
305 #endif
306 msg_warn("transport %s failure -- see a previous warning/fatal/panic logfile record for the problem description",
307 transport->name);
308
309 /*
310 * Assume the worst and write a defer logfile record for each
311 * recipient. This omission was already present in the first queue
312 * manager implementation of 199703, and was fixed 200511.
313 *
314 * To avoid the synchronous qmgr_defer_recipient() operation for each
315 * recipient of this queue entry, release the delivery process and
316 * move the entry back to the todo queue. Let qmgr_defer_transport()
317 * log the recipient asynchronously if possible, and get out of here.
318 * Note: if asynchronous logging is not possible,
319 * qmgr_defer_transport() eventually invokes qmgr_entry_done() and
320 * the entry becomes a dangling pointer.
321 */
322 QMGR_DELIVER_RELEASE_AGENT(entry);
323 qmgr_entry_unselect(entry);
324 qmgr_defer_transport(transport, &dsb->dsn);
325 return;
326 }
327
328 /*
329 * This message must be tried again.
330 *
331 * If we have a problem talking to this site, back off with this site for a
332 * while; dispose of queue entries for this site that await selection
333 * (the todo list); stay away from queue entries that have been selected
334 * (the busy list), or we would have dangling pointers. The queue itself
335 * won't go away before we dispose of the current queue entry.
336 *
337 * XXX Caution: DSN_COPY() will panic on empty status or reason.
338 */
339 #define SUSPENDED "delivery temporarily suspended: "
340
341 if (status == DELIVER_STAT_DEFER) {
342 message->flags |= DELIVER_STAT_DEFER;
343 if (VSTRING_LEN(dsb->status)) {
344 /* Sanitize the DSN status/reason from the delivery agent. */
345 if (!dsn_valid(vstring_str(dsb->status)))
346 vstring_strcpy(dsb->status, "4.0.0");
347 if (VSTRING_LEN(dsb->reason) == 0)
348 vstring_strcpy(dsb->reason, "unknown error");
349 vstring_prepend(dsb->reason, SUSPENDED, sizeof(SUSPENDED) - 1);
350 if (QMGR_QUEUE_READY(queue)) {
351 qmgr_queue_throttle(queue, DSN_FROM_DSN_BUF(dsb));
352 if (QMGR_QUEUE_THROTTLED(queue))
353 qmgr_defer_todo(queue, &dsb->dsn);
354 }
355 }
356 }
357
358 /*
359 * No problems detected. Mark the transport and queue as alive. The queue
360 * itself won't go away before we dispose of the current queue entry.
361 */
362 if (status != DELIVER_STAT_CRASH) {
363 qmgr_transport_unthrottle(transport);
364 if (VSTRING_LEN(dsb->reason) == 0)
365 qmgr_queue_unthrottle(queue);
366 }
367
368 /*
369 * Release the delivery process, and give some other queue entry a chance
370 * to be delivered. When all recipients for a message have been tried,
371 * decide what to do next with this message: defer, bounce, delete.
372 */
373 QMGR_DELIVER_RELEASE_AGENT(entry);
374 qmgr_entry_done(entry, QMGR_QUEUE_BUSY);
375 }
376
377 /* qmgr_deliver - deliver one per-site queue entry */
378
qmgr_deliver(QMGR_TRANSPORT * transport,VSTREAM * stream)379 void qmgr_deliver(QMGR_TRANSPORT *transport, VSTREAM *stream)
380 {
381 QMGR_ENTRY *entry;
382 DSN dsn;
383
384 /*
385 * Find out if this delivery process is really available. Once elected,
386 * the delivery process is supposed to express its happiness. If there is
387 * a problem, wipe the pending deliveries for this transport. This
388 * routine runs in response to an external event, so it does not run
389 * while some other queue manipulation is happening.
390 */
391 if (stream == 0 || qmgr_deliver_initial_reply(stream) != 0) {
392 #if 0
393 whatsup = concatenate(transport->name,
394 " mail transport unavailable", (char *) 0);
395 qmgr_transport_throttle(transport,
396 DSN_SIMPLE(&dsn, "4.3.0", whatsup));
397 myfree(whatsup);
398 #else
399 qmgr_transport_throttle(transport,
400 DSN_SIMPLE(&dsn, "4.3.0",
401 "mail transport unavailable"));
402 #endif
403 qmgr_defer_transport(transport, &dsn);
404 if (stream)
405 (void) vstream_fclose(stream);
406 return;
407 }
408
409 /*
410 * Find a suitable queue entry. Things may have changed since this
411 * transport was allocated. If no suitable entry is found,
412 * unceremoniously disconnect from the delivery process. The delivery
413 * agent request reading routine is prepared for the queue manager to
414 * change its mind for no apparent reason.
415 */
416 if ((entry = qmgr_job_entry_select(transport)) == 0) {
417 (void) vstream_fclose(stream);
418 return;
419 }
420
421 /*
422 * Send the queue file info and recipient info to the delivery process.
423 * If there is a problem, wipe the pending deliveries for this transport.
424 * This routine runs in response to an external event, so it does not run
425 * while some other queue manipulation is happening.
426 */
427 if (qmgr_deliver_send_request(entry, stream) < 0) {
428 qmgr_entry_unselect(entry);
429 #if 0
430 whatsup = concatenate(transport->name,
431 " mail transport unavailable", (char *) 0);
432 qmgr_transport_throttle(transport,
433 DSN_SIMPLE(&dsn, "4.3.0", whatsup));
434 myfree(whatsup);
435 #else
436 qmgr_transport_throttle(transport,
437 DSN_SIMPLE(&dsn, "4.3.0",
438 "mail transport unavailable"));
439 #endif
440 qmgr_defer_transport(transport, &dsn);
441 /* warning: entry may be a dangling pointer here */
442 (void) vstream_fclose(stream);
443 return;
444 }
445
446 /*
447 * If we get this far, go wait for the delivery status report.
448 */
449 qmgr_deliver_concurrency++;
450 entry->stream = stream;
451 event_enable_read(vstream_fileno(stream),
452 qmgr_deliver_update, (void *) entry);
453
454 /*
455 * Guard against broken systems.
456 */
457 event_request_timer(qmgr_deliver_abort, (void *) entry, var_daemon_timeout);
458 }
459