Chop networker into handy bits; related functions now live in their own file.
[citadel.git] / citadel / modules / network / serv_network.c
1 /*
2  * This module handles shared rooms, inter-Citadel mail, and outbound
3  * mailing list processing.
4  *
5  * Copyright (c) 2000-2011 by the citadel.org team
6  *
7  *  This program is open source software; you can redistribute it and/or modify
8  *  it under the terms of the GNU General Public License as published by
9  *  the Free Software Foundation; either version 3 of the License, or
10  *  (at your option) any later version.
11  *
12  *  This program is distributed in the hope that it will be useful,
13  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
14  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15  *  GNU General Public License for more details.
16  *
17  *  You should have received a copy of the GNU General Public License
18  *  along with this program; if not, write to the Free Software
19  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
20  *
21  * ** NOTE **   A word on the S_NETCONFIGS semaphore:
22  * This is a fairly high-level type of critical section.  It ensures that no
23  * two threads work on the netconfigs files at the same time.  Since we do
24  * so many things inside these, here are the rules:
25  *  1. begin_critical_section(S_NETCONFIGS) *before* begin_ any others.
26  *  2. Do *not* perform any I/O with the client during these sections.
27  *
28  */
29
30 /*
31  * Duration of time (in seconds) after which pending list subscribe/unsubscribe
32  * requests that have not been confirmed will be deleted.
33  */
34 #define EXP     259200  /* three days */
35
36 #include "sysdep.h"
37 #include <stdlib.h>
38 #include <unistd.h>
39 #include <stdio.h>
40 #include <fcntl.h>
41 #include <ctype.h>
42 #include <signal.h>
43 #include <pwd.h>
44 #include <errno.h>
45 #include <sys/stat.h>
46 #include <sys/types.h>
47 #include <dirent.h>
48 #if TIME_WITH_SYS_TIME
49 # include <sys/time.h>
50 # include <time.h>
51 #else
52 # if HAVE_SYS_TIME_H
53 #  include <sys/time.h>
54 # else
55 #  include <time.h>
56 # endif
57 #endif
58 #ifdef HAVE_SYSCALL_H
59 # include <syscall.h>
60 #else 
61 # if HAVE_SYS_SYSCALL_H
62 #  include <sys/syscall.h>
63 # endif
64 #endif
65
66 #include <sys/wait.h>
67 #include <string.h>
68 #include <limits.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 "user_ops.h"
76 #include "database.h"
77 #include "msgbase.h"
78 #include "internet_addressing.h"
79 #include "serv_network.h"
80 #include "clientsocket.h"
81 #include "file_ops.h"
82 #include "citadel_dirs.h"
83 #include "threads.h"
84
85 #ifndef HAVE_SNPRINTF
86 #include "snprintf.h"
87 #endif
88
89 #include "context.h"
90 #include "netconfig.h"
91 #include "netmail.h"
92 #include "ctdl_module.h"
93
94
95
96 /*
97  * When we do network processing, it's accomplished in two passes; one to
98  * gather a list of rooms and one to actually do them.  It's ok that rplist
99  * is global; we have a mutex that keeps it safe.
100  */
101 struct RoomProcList *rplist = NULL;
102
103
104
105
106
107
108
109 /*
110  * Check the use table.  This is a list of messages which have recently
111  * arrived on the system.  It is maintained and queried to prevent the same
112  * message from being entered into the database multiple times if it happens
113  * to arrive multiple times by accident.
114  */
115 int network_usetable(struct CtdlMessage *msg) {
116
117         char msgid[SIZ];
118         struct cdbdata *cdbut;
119         struct UseTable ut;
120
121         /* Bail out if we can't generate a message ID */
122         if (msg == NULL) {
123                 return(0);
124         }
125         if (msg->cm_fields['I'] == NULL) {
126                 return(0);
127         }
128         if (IsEmptyStr(msg->cm_fields['I'])) {
129                 return(0);
130         }
131
132         /* Generate the message ID */
133         strcpy(msgid, msg->cm_fields['I']);
134         if (haschar(msgid, '@') == 0) {
135                 strcat(msgid, "@");
136                 if (msg->cm_fields['N'] != NULL) {
137                         strcat(msgid, msg->cm_fields['N']);
138                 }
139                 else {
140                         return(0);
141                 }
142         }
143
144         cdbut = cdb_fetch(CDB_USETABLE, msgid, strlen(msgid));
145         if (cdbut != NULL) {
146                 cdb_free(cdbut);
147                 syslog(LOG_DEBUG, "network_usetable() : we already have %s\n", msgid);
148                 return(1);
149         }
150
151         /* If we got to this point, it's unique: add it. */
152         strcpy(ut.ut_msgid, msgid);
153         ut.ut_timestamp = time(NULL);
154         cdb_store(CDB_USETABLE, msgid, strlen(msgid), &ut, sizeof(struct UseTable) );
155         return(0);
156 }
157
158
159
160
161
162
163
164
165
166
167 /*
168  * Send the *entire* contents of the current room to one specific network node,
169  * ignoring anything we know about which messages have already undergone
170  * network processing.  This can be used to bring a new node into sync.
171  */
172 int network_sync_to(char *target_node) {
173         SpoolControl sc;
174         int num_spooled = 0;
175         int found_node = 0;
176         char buf[256];
177         char sc_type[256];
178         char sc_node[256];
179         char sc_room[256];
180         char filename[PATH_MAX];
181         FILE *fp;
182
183         /* Grab the configuration line we're looking for */
184         assoc_file_name(filename, sizeof filename, &CC->room, ctdl_netcfg_dir);
185         begin_critical_section(S_NETCONFIGS);
186         fp = fopen(filename, "r");
187         if (fp == NULL) {
188                 end_critical_section(S_NETCONFIGS);
189                 return(-1);
190         }
191         while (fgets(buf, sizeof buf, fp) != NULL) {
192                 buf[strlen(buf)-1] = 0;
193                 extract_token(sc_type, buf, 0, '|', sizeof sc_type);
194                 extract_token(sc_node, buf, 1, '|', sizeof sc_node);
195                 extract_token(sc_room, buf, 2, '|', sizeof sc_room);
196                 if ( (!strcasecmp(sc_type, "ignet_push_share"))
197                    && (!strcasecmp(sc_node, target_node)) ) {
198                         found_node = 1;
199                         
200                         /* Concise syntax because we don't need a full linked-list */
201                         memset(&sc, 0, sizeof(SpoolControl));
202                         sc.ignet_push_shares = (maplist *)
203                                 malloc(sizeof(maplist));
204                         sc.ignet_push_shares->next = NULL;
205                         safestrncpy(sc.ignet_push_shares->remote_nodename,
206                                 sc_node,
207                                 sizeof sc.ignet_push_shares->remote_nodename);
208                         safestrncpy(sc.ignet_push_shares->remote_roomname,
209                                 sc_room,
210                                 sizeof sc.ignet_push_shares->remote_roomname);
211                 }
212         }
213         fclose(fp);
214         end_critical_section(S_NETCONFIGS);
215
216         if (!found_node) return(-1);
217
218         /* Send ALL messages */
219         num_spooled = CtdlForEachMessage(MSGS_ALL, 0L, NULL, NULL, NULL,
220                 network_spool_msg, &sc);
221
222         /* Concise cleanup because we know there's only one node in the sc */
223         free(sc.ignet_push_shares);
224
225         syslog(LOG_NOTICE, "Synchronized %d messages to <%s>\n",
226                 num_spooled, target_node);
227         return(num_spooled);
228 }
229
230
231 /*
232  * Implements the NSYN command
233  */
234 void cmd_nsyn(char *argbuf) {
235         int num_spooled;
236         char target_node[256];
237
238         if (CtdlAccessCheck(ac_aide)) return;
239
240         extract_token(target_node, argbuf, 0, '|', sizeof target_node);
241         num_spooled = network_sync_to(target_node);
242         if (num_spooled >= 0) {
243                 cprintf("%d Spooled %d messages.\n", CIT_OK, num_spooled);
244         }
245         else {
246                 cprintf("%d No such room/node share exists.\n",
247                         ERROR + ROOM_NOT_FOUND);
248         }
249 }
250
251
252
253 /*
254  * Batch up and send all outbound traffic from the current room
255  */
256 void network_queue_room(struct ctdlroom *qrbuf, void *data) {
257         struct RoomProcList *ptr;
258
259         ptr = (struct RoomProcList *) malloc(sizeof (struct RoomProcList));
260         if (ptr == NULL) return;
261
262         safestrncpy(ptr->name, qrbuf->QRname, sizeof ptr->name);
263         begin_critical_section(S_RPLIST);
264         ptr->next = rplist;
265         rplist = ptr;
266         end_critical_section(S_RPLIST);
267 }
268
269 void destroy_network_queue_room(void)
270 {
271         struct RoomProcList *cur, *p;
272         NetMap *nmcur, *nmp;
273
274         cur = rplist;
275         begin_critical_section(S_RPLIST);
276         while (cur != NULL)
277         {
278                 p = cur->next;
279                 free (cur);
280                 cur = p;                
281         }
282         rplist = NULL;
283         end_critical_section(S_RPLIST);
284
285         nmcur = the_netmap;
286         while (nmcur != NULL)
287         {
288                 nmp = nmcur->next;
289                 free (nmcur);
290                 nmcur = nmp;            
291         }
292         the_netmap = NULL;
293         if (working_ignetcfg != NULL)
294                 free (working_ignetcfg);
295         working_ignetcfg = NULL;
296 }
297
298
299
300
301
302 /*
303  * Bounce a message back to the sender
304  */
305 void network_bounce(struct CtdlMessage *msg, char *reason) {
306         char *oldpath = NULL;
307         char buf[SIZ];
308         char bouncesource[SIZ];
309         char recipient[SIZ];
310         struct recptypes *valid = NULL;
311         char force_room[ROOMNAMELEN];
312         static int serialnum = 0;
313         size_t size;
314
315         syslog(LOG_DEBUG, "entering network_bounce()\n");
316
317         if (msg == NULL) return;
318
319         snprintf(bouncesource, sizeof bouncesource, "%s@%s", BOUNCESOURCE, config.c_nodename);
320
321         /* 
322          * Give it a fresh message ID
323          */
324         if (msg->cm_fields['I'] != NULL) {
325                 free(msg->cm_fields['I']);
326         }
327         snprintf(buf, sizeof buf, "%ld.%04lx.%04x@%s",
328                 (long)time(NULL), (long)getpid(), ++serialnum, config.c_fqdn);
329         msg->cm_fields['I'] = strdup(buf);
330
331         /*
332          * FIXME ... right now we're just sending a bounce; we really want to
333          * include the text of the bounced message.
334          */
335         if (msg->cm_fields['M'] != NULL) {
336                 free(msg->cm_fields['M']);
337         }
338         msg->cm_fields['M'] = strdup(reason);
339         msg->cm_format_type = 0;
340
341         /*
342          * Turn the message around
343          */
344         if (msg->cm_fields['R'] == NULL) {
345                 free(msg->cm_fields['R']);
346         }
347
348         if (msg->cm_fields['D'] == NULL) {
349                 free(msg->cm_fields['D']);
350         }
351
352         snprintf(recipient, sizeof recipient, "%s@%s",
353                 msg->cm_fields['A'], msg->cm_fields['N']);
354
355         if (msg->cm_fields['A'] == NULL) {
356                 free(msg->cm_fields['A']);
357         }
358
359         if (msg->cm_fields['N'] == NULL) {
360                 free(msg->cm_fields['N']);
361         }
362
363         if (msg->cm_fields['U'] == NULL) {
364                 free(msg->cm_fields['U']);
365         }
366
367         msg->cm_fields['A'] = strdup(BOUNCESOURCE);
368         msg->cm_fields['N'] = strdup(config.c_nodename);
369         msg->cm_fields['U'] = strdup("Delivery Status Notification (Failure)");
370
371         /* prepend our node to the path */
372         if (msg->cm_fields['P'] != NULL) {
373                 oldpath = msg->cm_fields['P'];
374                 msg->cm_fields['P'] = NULL;
375         }
376         else {
377                 oldpath = strdup("unknown_user");
378         }
379         size = strlen(oldpath) + SIZ;
380         msg->cm_fields['P'] = malloc(size);
381         snprintf(msg->cm_fields['P'], size, "%s!%s", config.c_nodename, oldpath);
382         free(oldpath);
383
384         /* Now submit the message */
385         valid = validate_recipients(recipient, NULL, 0);
386         if (valid != NULL) if (valid->num_error != 0) {
387                 free_recipients(valid);
388                 valid = NULL;
389         }
390         if ( (valid == NULL) || (!strcasecmp(recipient, bouncesource)) ) {
391                 strcpy(force_room, config.c_aideroom);
392         }
393         else {
394                 strcpy(force_room, "");
395         }
396         if ( (valid == NULL) && IsEmptyStr(force_room) ) {
397                 strcpy(force_room, config.c_aideroom);
398         }
399         CtdlSubmitMsg(msg, valid, force_room, 0);
400
401         /* Clean up */
402         if (valid != NULL) free_recipients(valid);
403         CtdlFreeMessage(msg);
404         syslog(LOG_DEBUG, "leaving network_bounce()\n");
405 }
406
407
408
409
410
411
412
413 /*
414  * network_do_queue()
415  * 
416  * Run through the rooms doing various types of network stuff.
417  */
418 void network_do_queue(void) {
419         static int doing_queue = 0;
420         static time_t last_run = 0L;
421         struct RoomProcList *ptr;
422         int full_processing = 1;
423
424         /*
425          * Run the full set of processing tasks no more frequently
426          * than once every n seconds
427          */
428         if ( (time(NULL) - last_run) < config.c_net_freq ) {
429                 full_processing = 0;
430                 syslog(LOG_DEBUG, "Network full processing in %ld seconds.\n",
431                         config.c_net_freq - (time(NULL)- last_run)
432                 );
433         }
434
435         /*
436          * This is a simple concurrency check to make sure only one queue run
437          * is done at a time.  We could do this with a mutex, but since we
438          * don't really require extremely fine granularity here, we'll do it
439          * with a static variable instead.
440          */
441         if (doing_queue) {
442                 return;
443         }
444         doing_queue = 1;
445
446         /* Load the IGnet Configuration into memory */
447         load_working_ignetcfg();
448
449
450         /*
451          * Load the network map and filter list into memory.
452          */
453         read_network_map();
454         filterlist = load_filter_list();
455
456         /* 
457          * Go ahead and run the queue
458          */
459         if (full_processing && !server_shutting_down) {
460                 syslog(LOG_DEBUG, "network: loading outbound queue\n");
461                 CtdlForEachRoom(network_queue_room, NULL);
462         }
463
464         if (rplist != NULL) {
465                 syslog(LOG_DEBUG, "network: running outbound queue\n");
466                 while (rplist != NULL && !server_shutting_down) {
467                         char spoolroomname[ROOMNAMELEN];
468                         safestrncpy(spoolroomname, rplist->name, sizeof spoolroomname);
469                         begin_critical_section(S_RPLIST);
470
471                         /* pop this record off the list */
472                         ptr = rplist;
473                         rplist = rplist->next;
474                         free(ptr);
475
476                         /* invalidate any duplicate entries to prevent double processing */
477                         for (ptr=rplist; ptr!=NULL; ptr=ptr->next) {
478                                 if (!strcasecmp(ptr->name, spoolroomname)) {
479                                         ptr->name[0] = 0;
480                                 }
481                         }
482
483                         end_critical_section(S_RPLIST);
484                         if (spoolroomname[0] != 0) {
485                                 network_spoolout_room(spoolroomname);
486                         }
487                 }
488         }
489
490         /* If there is anything in the inbound queue, process it */
491         if (!server_shutting_down) {
492                 network_do_spoolin();
493         }
494
495         /* Save the network map back to disk */
496         write_network_map();
497
498         /* Free the filter list in memory */
499         free_filter_list(filterlist);
500         filterlist = NULL;
501
502         network_consolidate_spoolout();
503
504         syslog(LOG_DEBUG, "network: queue run completed\n");
505
506         if (full_processing) {
507                 last_run = time(NULL);
508         }
509
510         doing_queue = 0;
511 }
512
513
514
515
516 int network_room_handler (struct ctdlroom *room)
517 {
518         network_queue_room(room, NULL);
519         return 0;
520 }
521
522
523 /*
524  * Module entry point
525  */
526 CTDL_MODULE_INIT(network)
527 {
528         if (!threading)
529         {
530                 CtdlRegisterProtoHook(cmd_nsyn, "NSYN", "Synchronize room to node");
531                 CtdlRegisterRoomHook(network_room_handler);
532                 CtdlRegisterCleanupHook(destroy_network_queue_room);
533                 CtdlRegisterSessionHook(network_do_queue, EVT_TIMER);
534         }
535         return "network";
536 }