Imported Upstream version 3.2.0
[debian/amanda] / amandad-src / amandad.c
1 /*
2  * Amanda, The Advanced Maryland Automatic Network Disk Archiver
3  * Copyright (c) 1991-1999 University of Maryland at College Park
4  * All Rights Reserved.
5  *
6  * Permission to use, copy, modify, distribute, and sell this software and its
7  * documentation for any purpose is hereby granted without fee, provided that
8  * the above copyright notice appear in all copies and that both that
9  * copyright notice and this permission notice appear in supporting
10  * documentation, and that the name of U.M. not be used in advertising or
11  * publicity pertaining to distribution of the software without specific,
12  * written prior permission.  U.M. makes no representations about the
13  * suitability of this software for any purpose.  It is provided "as is"
14  * without express or implied warranty.
15  *
16  * U.M. DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, INCLUDING ALL
17  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN NO EVENT SHALL U.M.
18  * BE LIABLE FOR ANY SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
19  * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION
20  * OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
21  * CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
22  *
23  * Authors: the Amanda Development Team.  Its members are listed in a
24  * file named AUTHORS, in the root directory of this distribution.
25  */
26
27 /*
28  * $Id: amandad.c,v 1.18 2006/08/21 20:17:09 martinea Exp $
29  *
30  * handle client-host side of Amanda network communications, including
31  * security checks, execution of the proper service, and acking the
32  * master side
33  */
34
35 #include "amanda.h"
36 #include "amandad.h"
37 #include "clock.h"
38 #include "event.h"
39 #include "amfeatures.h"
40 #include "packet.h"
41 #include "version.h"
42 #include "security.h"
43 #include "stream.h"
44 #include "util.h"
45 #include "conffile.h"
46
47 #define REP_TIMEOUT     (6*60*60)       /* secs for service to reply */
48 #define ACK_TIMEOUT     10              /* XXX should be configurable */
49 #define STDERR_PIPE (DATA_FD_COUNT + 1)
50
51 #define amandad_debug(i, ...) do {      \
52         if ((i) <= debug_amandad) {     \
53                 dbprintf(__VA_ARGS__);  \
54         }                               \
55 } while (0)
56
57 /*
58  * These are the actions for entering the state machine
59  */
60 typedef enum { A_START, A_RECVPKT, A_RECVREP, A_PENDING, A_FINISH, A_CONTINUE,
61     A_SENDNAK, A_TIMEOUT } action_t;
62
63 /*
64  * This is a state in the state machine.  It is a function pointer to
65  * the function that actually implements the state.
66  */
67 struct active_service;
68 typedef action_t (*state_t)(struct active_service *, action_t, pkt_t *);
69
70 /* string that we scan for in sendbackup's MESG stream */
71 static const char info_end_str[] = "sendbackup: info end\n";
72 #define INFO_END_LEN (sizeof(info_end_str)-1)
73
74 /* 
75  * Here are the services that we allow.
76  * Must be in the same order as services[].
77  */
78 typedef enum {
79     SERVICE_NOOP,
80     SERVICE_SENDSIZE,
81     SERVICE_SENDBACKUP,
82     SERVICE_SELFCHECK,
83     SERVICE_AMINDEXD,
84     SERVICE_AMIDXTAPED
85 } service_t;
86
87 static struct services {
88     char *name;
89     int  active;
90     service_t service;
91 } services[] = {
92    { "noop", 1, SERVICE_NOOP },
93    { "sendsize", 1, SERVICE_SENDSIZE },
94    { "sendbackup", 1, SERVICE_SENDBACKUP },
95    { "selfcheck", 1, SERVICE_SELFCHECK },
96    { "amindexd", 0, SERVICE_AMINDEXD },
97    { "amidxtaped", 0, SERVICE_AMIDXTAPED }
98 };
99 #define NSERVICES       (int)(sizeof(services) / sizeof(services[0]))
100
101 /*
102  * This structure describes an active running service.
103  *
104  * An active service is something running that we have received
105  * a request for.  This structure holds info on that service, including
106  * file descriptors for data, etc, as well as the security handle
107  * for communications with the amanda server.
108  */
109 struct active_service {
110     service_t service;                  /* service name */
111     char *cmd;                          /* name of command we ran */
112     char *arguments;                    /* arguments we sent it */
113     security_handle_t *security_handle; /* remote server */
114     state_t state;                      /* how far this has progressed */
115     pid_t pid;                          /* pid of subprocess */
116     int send_partial_reply;             /* send PREP packet */
117     int reqfd;                          /* pipe to write requests */
118     int repfd;                          /* pipe to read replies */
119     int errfd;                          /* pipe to read stderr */
120     event_handle_t *ev_repfd;           /* read event handle for repfd */
121     event_handle_t *ev_reptimeout;      /* timeout for rep data */
122     event_handle_t *ev_errfd;           /* read event handle for errfd */
123     pkt_t rep_pkt;                      /* rep packet we're sending out */
124     char *errbuf;                       /* buffer to read the err into */
125     char *repbuf;                       /* buffer to read the rep into */
126     size_t bufsize;                     /* length of repbuf */
127     size_t repbufsize;                  /* length of repbuf */
128     int repretry;                       /* times we'll retry sending the rep */
129     int seen_info_end;                  /* have we seen "sendbackup info end\n"? */
130     char info_end_buf[INFO_END_LEN];    /* last few bytes read, used for scanning for info end */
131
132     /*
133      * General user streams to the process, and their equivalent
134      * network streams.
135      */
136     struct datafd_handle {
137         int fd_read;                    /* pipe to child process */
138         int fd_write;                   /* pipe to child process */
139         event_handle_t *ev_read;        /* it's read event handle */
140         event_handle_t *ev_write;       /* it's write event handle */
141         security_stream_t *netfd;       /* stream to amanda server */
142         struct active_service *as;      /* pointer back to our enclosure */
143     } data[DATA_FD_COUNT];
144     char databuf[NETWORK_BLOCK_BYTES];  /* buffer to relay netfd data in */
145 };
146
147 /*
148  * Queue of outstanding requests that we are running.
149  */
150 GSList *serviceq = NULL;
151
152 static int wait_30s = 1;
153 static int exit_on_qlength = 1;
154 static char *auth = NULL;
155 static kencrypt_type amandad_kencrypt = KENCRYPT_NONE;
156
157 int main(int argc, char **argv);
158
159 static int allocstream(struct active_service *, int);
160 static void exit_check(void *);
161 static void protocol_accept(security_handle_t *, pkt_t *);
162 static void state_machine(struct active_service *, action_t, pkt_t *);
163
164 static action_t s_sendack(struct active_service *, action_t, pkt_t *);
165 static action_t s_repwait(struct active_service *, action_t, pkt_t *);
166 static action_t s_processrep(struct active_service *, action_t, pkt_t *);
167 static action_t s_sendrep(struct active_service *, action_t, pkt_t *);
168 static action_t s_ackwait(struct active_service *, action_t, pkt_t *);
169
170 static void repfd_recv(void *);
171 static void process_errfd(void *cookie);
172 static void errfd_recv(void *);
173 static void timeout_repfd(void *);
174 static void protocol_recv(void *, pkt_t *, security_status_t);
175 static void process_readnetfd(void *);
176 static void process_writenetfd(void *, void *, ssize_t);
177 static struct active_service *service_new(security_handle_t *,
178     const char *, service_t, const char *);
179 static void service_delete(struct active_service *);
180 static int writebuf(struct active_service *, const void *, size_t);
181 static ssize_t do_sendpkt(security_handle_t *handle, pkt_t *pkt);
182 static char *amandad_get_security_conf (char *, void *);
183
184 static const char *state2str(state_t);
185 static const char *action2str(action_t);
186
187 int
188 main(
189     int         argc,
190     char **     argv)
191 {
192     int i, j;
193     int have_services;
194     int in, out;
195     const security_driver_t *secdrv;
196     int no_exit = 0;
197     char *pgm = "amandad";              /* in case argv[0] is not set */
198 #if defined(USE_REUSEADDR)
199     const int on = 1;
200     int r;
201 #endif
202
203     /*
204      * Configure program for internationalization:
205      *   1) Only set the message locale for now.
206      *   2) Set textdomain for all amanda related programs to "amanda"
207      *      We don't want to be forced to support dozens of message catalogs.
208      */  
209     setlocale(LC_MESSAGES, "C");
210     textdomain("amanda"); 
211
212     safe_fd(-1, 0);
213     safe_cd();
214
215     /*
216      * Nexenta needs the SUN_PERSONALITY env variable to be unset, otherwise
217      * the Sun version of tar in /usr/sun/sbin/tar is called instead.
218      *
219      * On other operating systems this will have no effect.
220      */
221
222 #ifdef HAVE_UNSETENV
223     unsetenv("SUN_PERSONALITY");
224 #endif
225
226     /*
227      * When called via inetd, it is not uncommon to forget to put the
228      * argv[0] value on the config line.  On some systems (e.g. Solaris)
229      * this causes argv and/or argv[0] to be NULL, so we have to be
230      * careful getting our name.
231      */
232     if ((argv == NULL) || (argv[0] == NULL)) {
233             pgm = "amandad";            /* in case argv[0] is not set */
234     } else {
235             pgm = basename(argv[0]);    /* Strip of leading path get debug name */
236     }
237     set_pname(pgm);
238     dbopen(DBG_SUBDIR_AMANDAD);
239
240     if(argv == NULL) {
241         error(_("argv == NULL\n"));
242         /*NOTREACHED*/
243     }
244
245     /* Don't die when child closes pipe */
246     signal(SIGPIPE, SIG_IGN);
247
248     /* Parse the configuration; we'll handle errors later */
249     config_init(CONFIG_INIT_CLIENT, NULL);
250
251     if (geteuid() == 0) {
252         check_running_as(RUNNING_AS_ROOT);
253         initgroups(CLIENT_LOGIN, get_client_gid());
254         setgid(get_client_gid());
255         setegid(get_client_gid());
256         seteuid(get_client_uid());
257     } else {
258         check_running_as(RUNNING_AS_CLIENT_LOGIN);
259     }
260
261     add_amanda_log_handler(amanda_log_stderr);
262     add_amanda_log_handler(amanda_log_syslog);
263
264     /*
265      * ad-hoc argument parsing
266      *
267      * We accept        -auth=[authentication type]
268      *                  -no-exit
269      *                  -tcp=[port]
270      *                  -udp=[port]
271      * We also add a list of services that amandad can launch
272      */
273     secdrv = NULL;
274     in = 0; out = 1;            /* default to stdin/stdout */
275     have_services = 0;
276     for (i = 1; i < argc; i++) {
277         /*
278          * Get a driver for a security type specified after -auth=
279          */
280         if (strncmp(argv[i], "-auth=", strlen("-auth=")) == 0) {
281             argv[i] += strlen("-auth=");
282             secdrv = security_getdriver(argv[i]);
283             auth = argv[i];
284             if (secdrv == NULL) {
285                 error(_("no driver for security type '%s'\n"), argv[i]);
286                 /*NOTREACHED*/
287             }
288             continue;
289         }
290
291         /*
292          * If -no-exit is specified, always run even after requests have
293          * been satisfied.
294          */
295         else if (strcmp(argv[i], "-no-exit") == 0) {
296             no_exit = 1;
297             continue;
298         }
299
300         /*
301          * Allow us to directly bind to a udp port for debugging.
302          * This may only apply to some security types.
303          */
304         else if (strncmp(argv[i], "-udp=", strlen("-udp=")) == 0) {
305 #ifdef WORKING_IPV6
306             struct sockaddr_in6 sin;
307 #else
308             struct sockaddr_in sin;
309 #endif
310
311             argv[i] += strlen("-udp=");
312 #ifdef WORKING_IPV6
313             in = out = socket(AF_INET6, SOCK_DGRAM, 0);
314 #else
315             in = out = socket(AF_INET, SOCK_DGRAM, 0);
316 #endif
317             if (in < 0) {
318                 error(_("can't create dgram socket: %s\n"), strerror(errno));
319                 /*NOTREACHED*/
320             }
321 #ifdef USE_REUSEADDR
322             r = setsockopt(in, SOL_SOCKET, SO_REUSEADDR,
323                 (void *)&on, (socklen_t_equiv)sizeof(on));
324             if (r < 0) {
325                 dbprintf(_("amandad: setsockopt(SO_REUSEADDR) failed: %s\n"),
326                           strerror(errno));
327             }
328 #endif
329
330 #ifdef WORKING_IPV6
331             sin.sin6_family = (sa_family_t)AF_INET6;
332             sin.sin6_addr = in6addr_any;
333             sin.sin6_port = (in_port_t)htons((in_port_t)atoi(argv[i]));
334 #else
335             sin.sin_family = (sa_family_t)AF_INET;
336             sin.sin_addr.s_addr = INADDR_ANY;
337             sin.sin_port = (in_port_t)htons((in_port_t)atoi(argv[i]));
338 #endif
339             if (bind(in, (struct sockaddr *)&sin, (socklen_t_equiv)sizeof(sin)) < 0) {
340                 error(_("can't bind to port %d: %s\n"), atoi(argv[i]),
341                     strerror(errno));
342                 /*NOTREACHED*/
343             }
344         }
345         /*
346          * Ditto for tcp ports.
347          */
348         else if (strncmp(argv[i], "-tcp=", strlen("-tcp=")) == 0) {
349 #ifdef WORKING_IPV6
350             struct sockaddr_in6 sin;
351 #else
352             struct sockaddr_in sin;
353 #endif
354             int sock;
355             socklen_t_equiv n;
356
357             argv[i] += strlen("-tcp=");
358 #ifdef WORKING_IPV6
359             sock = socket(AF_INET6, SOCK_STREAM, 0);
360 #else
361             sock = socket(AF_INET, SOCK_STREAM, 0);
362 #endif
363             if (sock < 0) {
364                 error(_("can't create tcp socket: %s\n"), strerror(errno));
365                 /*NOTREACHED*/
366             }
367 #ifdef USE_REUSEADDR
368             r = setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
369                 (void *)&on, (socklen_t_equiv)sizeof(on));
370             if (r < 0) {
371                 dbprintf(_("amandad: setsockopt(SO_REUSEADDR) failed: %s\n"),
372                           strerror(errno));
373             }
374 #endif
375 #ifdef WORKING_IPV6
376             sin.sin6_family = (sa_family_t)AF_INET6;
377             sin.sin6_addr = in6addr_any;
378             sin.sin6_port = (in_port_t)htons((in_port_t)atoi(argv[i]));
379 #else
380             sin.sin_family = (sa_family_t)AF_INET;
381             sin.sin_addr.s_addr = INADDR_ANY;
382             sin.sin_port = (in_port_t)htons((in_port_t)atoi(argv[i]));
383 #endif
384             if (bind(sock, (struct sockaddr *)&sin, (socklen_t_equiv)sizeof(sin)) < 0) {
385                 error(_("can't bind to port %d: %s\n"), atoi(argv[i]),
386                     strerror(errno));
387                 /*NOTREACHED*/
388             }
389             listen(sock, 10);
390             n = (socklen_t_equiv)sizeof(sin);
391             in = out = accept(sock, (struct sockaddr *)&sin, &n);
392         }
393         /*
394          * It must be a service name
395          */
396         else {
397             /* clear all services */
398             if(!have_services) {
399                 for (j = 0; j < (int)NSERVICES; j++)
400                     services[j].active = 0;
401             }
402             have_services = 1;
403
404             if(strcmp(argv[i],"amdump") == 0) {
405                 services[0].active = 1;
406                 services[1].active = 1;
407                 services[2].active = 1;
408                 services[3].active = 1;
409             }
410             else {
411                 for (j = 0; j < (int)NSERVICES; j++)
412                     if (strcmp(services[j].name, argv[i]) == 0)
413                         break;
414                 if (j == (int)NSERVICES) {
415                     dbprintf(_("%s: invalid service\n"), argv[i]);
416                     exit(1);
417                 }
418                 services[j].active = 1;
419             }
420         }
421     }
422
423     /*
424      * If no security type specified, use BSD
425      */
426     if (secdrv == NULL) {
427         secdrv = security_getdriver("BSD");
428         auth = "bsd";
429         if (secdrv == NULL) {
430             error(_("no driver for default security type 'BSD'\n"));
431             /*NOTREACHED*/
432         }
433     }
434
435     if(strcasecmp(auth, "rsh") == 0 ||
436        strcasecmp(auth, "ssh") == 0 ||
437        strcasecmp(auth, "local") == 0 ||
438        strcasecmp(auth, "bsdtcp") == 0) {
439         wait_30s = 0;
440         exit_on_qlength = 1;
441     }
442
443 #ifndef SINGLE_USERID
444     if (geteuid() == 0) {
445         if (strcasecmp(auth, "krb5") != 0) {
446             struct passwd *pwd;
447             /* lookup our local user name */
448             if ((pwd = getpwnam(CLIENT_LOGIN)) == NULL) {
449                 error(_("getpwnam(%s) failed."), CLIENT_LOGIN);
450             }
451
452             if (pwd->pw_uid != 0) {
453                 error(_("'amandad' must be run as user '%s' when using '%s' authentication"),
454                       CLIENT_LOGIN, auth);
455             }
456         }
457     } else {
458         if (strcasecmp(auth, "krb5") == 0) {
459             error(_("'amandad' must be run as user 'root' when using 'krb5' authentication"));
460         }
461     }
462 #endif
463
464     /* initialize */
465
466     startclock();
467
468     dbprintf(_("version %s\n"), VERSION);
469     for (i = 0; version_info[i] != NULL; i++) {
470         dbprintf("    %s", version_info[i]);
471     }
472
473     if (! (argc >= 1 && argv != NULL && argv[0] != NULL)) {
474         dbprintf(_("WARNING: argv[0] not defined: check inetd.conf\n"));
475     }
476
477     /* krb5 require the euid to be 0 */
478     if (strcasecmp(auth, "krb5") == 0) {
479         seteuid((uid_t)0);
480     }
481
482     /*
483      * Schedule to call protocol_accept() when new security handles
484      * are created on stdin.
485      */
486     security_accept(secdrv, amandad_get_security_conf, in, out, protocol_accept, NULL);
487
488     /*
489      * Schedule an event that will try to exit every 30 seconds if there
490      * are no requests outstanding.
491      */
492     if(wait_30s)
493         (void)event_register((event_id_t)30, EV_TIME, exit_check, &no_exit);
494
495     /*
496      * Call event_loop() with an arg of 0, telling it to block until all
497      * events are completed.
498      */
499     event_loop(0);
500
501     close(in);
502     close(out);
503     dbclose();
504     return(0);
505 }
506
507 /*
508  * This runs periodically and checks to see if we have any active services
509  * still running.  If we don't, then we quit.
510  */
511 static void
512 exit_check(
513     void *      cookie)
514 {
515     int no_exit;
516
517     assert(cookie != NULL);
518     no_exit = *(int *)cookie;
519
520     /*
521      * If things are still running, then don't exit.
522      */
523     if (g_slist_length(serviceq) > 0)
524         return;
525
526     /*
527      * If the caller asked us to never exit, then we're done
528      */
529     if (no_exit)
530         return;
531
532     dbclose();
533     exit(0);
534 }
535
536 /*
537  * Handles new incoming protocol handles.  This is a callback for
538  * security_accept(), which gets called when new handles are detected.
539  */
540 static void
541 protocol_accept(
542     security_handle_t * handle,
543     pkt_t *             pkt)
544 {
545     pkt_t pkt_out;
546     GSList *iter;
547     struct active_service *as;
548     char *pktbody, *tok, *service, *arguments;
549     char *service_path = NULL;
550     GSList *errlist = NULL;
551     int i;
552
553     pkt_out.body = NULL;
554
555     /*
556      * If handle is NULL, then the connection is closed.
557      */
558     if(handle == NULL) {
559         return;
560     }
561
562     /*
563      * If we have errors (not warnings) from the config file, let the remote system
564      * know immediately.  Unfortunately, we only get one ERROR line, so if there
565      * are multiple errors, we just show the first.
566      */
567     if (config_errors(&errlist) >= CFGERR_ERRORS) {
568         GSList *iter = errlist;
569         char *errmsg;
570         gboolean multiple_errors = FALSE;
571
572         if (iter) {
573             errmsg = (char *)iter->data;
574             if (iter->next)
575                 multiple_errors = TRUE;
576         } else {
577             errmsg = "(no error message)";
578         }
579
580         pkt_init(&pkt_out, P_NAK, "ERROR %s%s", errmsg,
581             multiple_errors? _(" (additional errors not displayed)"):"");
582         do_sendpkt(handle, &pkt_out);
583         amfree(pkt_out.body);
584         security_close(handle);
585         return;
586     }
587
588     g_debug("authenticated peer name is '%s'", security_get_authenticated_peer_name(handle));
589
590     /*
591      * If pkt is NULL, then there was a problem with the new connection.
592      */
593     if (pkt == NULL) {
594         dbprintf(_("accept error: %s\n"), security_geterror(handle));
595         pkt_init(&pkt_out, P_NAK, "ERROR %s\n", security_geterror(handle));
596         do_sendpkt(handle, &pkt_out);
597         amfree(pkt_out.body);
598         security_close(handle);
599         return;
600     }
601
602     dbprintf(_("accept recv %s pkt:\n<<<<<\n%s>>>>>\n"),
603         pkt_type2str(pkt->type), pkt->body);
604
605     /*
606      * If this is not a REQ packet, just forget about it.
607      */
608     if (pkt->type != P_REQ) {
609         dbprintf(_("received unexpected %s packet:\n<<<<<\n%s>>>>>\n\n"),
610             pkt_type2str(pkt->type), pkt->body);
611         security_close(handle);
612         return;
613     }
614
615     pktbody = service = arguments = NULL;
616     as = NULL;
617
618     /*
619      * Parse out the service and arguments
620      */
621
622     pktbody = stralloc(pkt->body);
623
624     tok = strtok(pktbody, " ");
625     if (tok == NULL)
626         goto badreq;
627     if (strcmp(tok, "SERVICE") != 0)
628         goto badreq;
629
630     tok = strtok(NULL, " \n");
631     if (tok == NULL)
632         goto badreq;
633     service = stralloc(tok);
634
635     /* we call everything else 'arguments' */
636     tok = strtok(NULL, "");
637     if (tok == NULL)
638         goto badreq;
639     arguments = stralloc(tok);
640
641     /* see if it's one we allow */
642     for (i = 0; i < (int)NSERVICES; i++)
643         if (services[i].active == 1 && strcmp(services[i].name, service) == 0)
644             break;
645     if (i == (int)NSERVICES) {
646         dbprintf(_("%s: invalid service\n"), service);
647         pkt_init(&pkt_out, P_NAK, _("ERROR %s: invalid service, add '%s' as argument to amandad\n"), service, service);
648         goto send_pkt_out;
649     }
650
651     service_path = vstralloc(amlibexecdir, "/", service, NULL);
652     if (access(service_path, X_OK) < 0) {
653         dbprintf(_("can't execute %s: %s\n"), service_path, strerror(errno));
654             pkt_init(&pkt_out, P_NAK,
655                      _("ERROR execute access to \"%s\" denied\n"),
656                      service_path);
657         goto send_pkt_out;
658     }
659
660     /* see if its already running */
661     for (iter = serviceq; iter != NULL; iter = g_slist_next(iter)) {
662         as = (struct active_service *)iter->data;
663             if (strcmp(as->cmd, service_path) == 0 &&
664                 strcmp(as->arguments, arguments) == 0) {
665                     dbprintf(_("%s %s: already running, acking req\n"),
666                         service, arguments);
667                     pkt_init_empty(&pkt_out, P_ACK);
668                     goto send_pkt_out_no_delete;
669             }
670     }
671
672     /*
673      * create a new service instance, and send the arguments down
674      * the request pipe.
675      */
676     dbprintf(_("creating new service: %s\n%s\n"), service, arguments);
677     as = service_new(handle, service_path, services[i].service, arguments);
678     if (writebuf(as, arguments, strlen(arguments)) < 0) {
679         const char *errmsg = strerror(errno);
680         dbprintf(_("error sending arguments to %s: %s\n"), service, errmsg);
681         pkt_init(&pkt_out, P_NAK, _("ERROR error writing arguments to %s: %s\n"),
682             service, errmsg);
683         goto send_pkt_out;
684     }
685     aclose(as->reqfd);
686
687     amfree(pktbody);
688     amfree(service);
689     amfree(service_path);
690     amfree(arguments);
691
692     /*
693      * Move to the sendack state, and start up the state
694      * machine.
695      */
696     as->state = s_sendack;
697     state_machine(as, A_START, NULL);
698     return;
699
700 badreq:
701     pkt_init(&pkt_out, P_NAK, _("ERROR invalid REQ\n"));
702     dbprintf(_("received invalid %s packet:\n<<<<<\n%s>>>>>\n\n"),
703         pkt_type2str(pkt->type), pkt->body);
704
705 send_pkt_out:
706     if(as)
707         service_delete(as);
708 send_pkt_out_no_delete:
709     amfree(pktbody);
710     amfree(service_path);
711     amfree(service);
712     amfree(arguments);
713     do_sendpkt(handle, &pkt_out);
714     security_close(handle);
715     amfree(pkt_out.body);
716 }
717
718 /*
719  * Handles incoming protocol packets.  Routes responses to the proper
720  * running service.
721  */
722 static void
723 state_machine(
724     struct active_service *     as,
725     action_t                    action,
726     pkt_t *                     pkt)
727 {
728     action_t retaction;
729     state_t curstate;
730     pkt_t nak;
731
732     amandad_debug(1, _("state_machine: %p entering\n"), as);
733     for (;;) {
734         curstate = as->state;
735         amandad_debug(1, _("state_machine: %p curstate=%s action=%s\n"), as,
736                           state2str(curstate), action2str(action));
737         retaction = (*curstate)(as, action, pkt);
738         amandad_debug(1, _("state_machine: %p curstate=%s returned %s (nextstate=%s)\n"),
739                           as, state2str(curstate), action2str(retaction),
740                           state2str(as->state));
741
742         switch (retaction) {
743         /*
744          * State has queued up and is now blocking on input.
745          */
746         case A_PENDING:
747             amandad_debug(1, _("state_machine: %p leaving (A_PENDING)\n"), as);
748             return;
749
750         /*
751          * service has switched states.  Loop.
752          */
753         case A_CONTINUE:
754             break;
755
756         /*
757          * state has determined that the packet it received was bogus.
758          * Send a nak, and return.
759          */
760         case A_SENDNAK:
761             dbprintf(_("received unexpected %s packet\n"),
762                 pkt_type2str(pkt->type));
763             dbprintf(_("<<<<<\n%s----\n\n"), pkt->body);
764             pkt_init(&nak, P_NAK, _("ERROR unexpected packet type %s\n"),
765                 pkt_type2str(pkt->type));
766             do_sendpkt(as->security_handle, &nak);
767             amfree(nak.body);
768             security_recvpkt(as->security_handle, protocol_recv, as, -1);
769             amandad_debug(1, _("state_machine: %p leaving (A_SENDNAK)\n"), as);
770             return;
771
772         /*
773          * Service is done.  Remove it and finish.
774          */
775         case A_FINISH:
776             amandad_debug(1, _("state_machine: %p leaving (A_FINISH)\n"), as);
777             service_delete(as);
778             return;
779
780         default:
781             assert(0);
782             break;
783         }
784     }
785     /*NOTREACHED*/
786 }
787
788 /*
789  * This state just sends an ack.  After that, we move to the repwait
790  * state to wait for REP data to arrive from the subprocess.
791  */
792 static action_t
793 s_sendack(
794     struct active_service *     as,
795     action_t                    action,
796     pkt_t *                     pkt)
797 {
798     pkt_t ack;
799
800     (void)action;       /* Quiet unused parameter warning */
801     (void)pkt;          /* Quiet unused parameter warning */
802
803     pkt_init_empty(&ack, P_ACK);
804     if (do_sendpkt(as->security_handle, &ack) < 0) {
805         dbprintf(_("error sending ACK: %s\n"),
806             security_geterror(as->security_handle));
807         amfree(ack.body);
808         return (A_FINISH);
809     }
810     amfree(ack.body);
811
812     /*
813      * move to the repwait state
814      * Setup a listener for data on the reply fd, but also
815      * listen for packets over the wire, as the server may
816      * poll us if we take a long time.
817      * Setup a timeout that will fire if it takes too long to
818      * receive rep data.
819      */
820     as->state = s_repwait;
821     as->ev_repfd = event_register((event_id_t)as->repfd, EV_READFD, repfd_recv, as);
822     as->ev_reptimeout = event_register(REP_TIMEOUT, EV_TIME,
823         timeout_repfd, as);
824     as->errbuf = NULL;
825     as->ev_errfd = event_register((event_id_t)as->errfd, EV_READFD, errfd_recv, as);
826     security_recvpkt(as->security_handle, protocol_recv, as, -1);
827     return (A_PENDING);
828 }
829
830 /*
831  * This is the repwait state.  We have responded to the initial REQ with
832  * an ACK, and we are now waiting for the process we spawned to pass us 
833  * data to send in a REP.
834  */
835 static action_t
836 s_repwait(
837     struct active_service *     as,
838     action_t                    action,
839     pkt_t *                     pkt)
840 {
841     ssize_t   n;
842     char     *repbuf_temp;
843     char     *what;
844     char     *msg;
845     int       code = 0;
846     int       pid;
847     amwait_t  retstat;
848
849     /*
850      * We normally shouldn't receive any packets while waiting
851      * for our REP data, but in some cases we do.
852      */
853     if (action == A_RECVPKT) {
854         assert(pkt != NULL);
855         /*
856          * Another req for something that's running.  Just send an ACK
857          * and go back and wait for more data.
858          */
859         if (pkt->type == P_REQ) {
860             dbprintf(_("received dup P_REQ packet, ACKing it\n"));
861             amfree(as->rep_pkt.body);
862             pkt_init_empty(&as->rep_pkt, P_ACK);
863             do_sendpkt(as->security_handle, &as->rep_pkt);
864             security_recvpkt(as->security_handle, protocol_recv, as, -1);
865             return (A_PENDING);
866         }
867         /* something unexpected.  Nak it */
868         return (A_SENDNAK);
869     }
870
871     if (action == A_TIMEOUT) {
872         amfree(as->rep_pkt.body);
873         pkt_init(&as->rep_pkt, P_NAK, _("ERROR timeout on reply pipe\n"));
874         dbprintf(_("%s timed out waiting for REP data\n"), as->cmd);
875         do_sendpkt(as->security_handle, &as->rep_pkt);
876         return (A_FINISH);
877     }
878
879     assert(action == A_RECVREP);
880     if(as->bufsize == 0) {
881         as->bufsize = NETWORK_BLOCK_BYTES;
882         as->repbuf = alloc(as->bufsize);
883     }
884
885     do {
886         n = read(as->repfd, as->repbuf + as->repbufsize,
887                  as->bufsize - as->repbufsize - 1);
888     } while ((n < 0) && ((errno == EINTR) || (errno == EAGAIN)));
889     if (n < 0) {
890         const char *errstr = strerror(errno);
891         dbprintf(_("read error on reply pipe: %s\n"), errstr);
892         amfree(as->rep_pkt.body);
893         pkt_init(&as->rep_pkt, P_NAK, _("ERROR read error on reply pipe: %s\n"),
894                  errstr);
895         do_sendpkt(as->security_handle, &as->rep_pkt);
896         return (A_FINISH);
897     }
898
899     /* If end of service, wait for process status */
900     if (n == 0) {
901         pid = waitpid(as->pid, &retstat, WNOHANG);
902         if (as->service  == SERVICE_NOOP ||
903             as->service  == SERVICE_SENDSIZE ||
904             as->service  == SERVICE_SELFCHECK) {
905             int t = 0;
906             while (t<5 && pid == 0) {
907                 sleep(1);
908                 t++;
909                 pid = waitpid(as->pid, &retstat, WNOHANG);
910             }
911         }
912
913         process_errfd(as);
914
915         if (pid == 0)
916             pid = waitpid(as->pid, &retstat, WNOHANG);
917
918         if (pid > 0) {
919             what = NULL;
920             if (! WIFEXITED(retstat)) {
921                 what = _("signal");
922                 code = WTERMSIG(retstat);
923             } else if (WEXITSTATUS(retstat) != 0) {
924                 what = _("code");
925                 code = WEXITSTATUS(retstat);
926             }
927             if (what) {
928                 dbprintf(_("service %s failed: pid %u exited with %s %d\n"),
929                          (as->cmd)?as->cmd:_("??UNKONWN??"),
930                          (unsigned)as->pid,
931                          what, code);
932                 msg = vstrallocf(
933                      _("ERROR service %s failed: pid %u exited with %s %d\n"),
934                      (as->cmd)?as->cmd:_("??UNKONWN??"), (unsigned)as->pid,
935                      what, code);
936                 if (as->repbufsize + strlen(msg) >= (as->bufsize - 1)) {
937                         as->bufsize *= 2;
938                         repbuf_temp = alloc(as->bufsize);
939                         memcpy(repbuf_temp, as->repbuf, as->repbufsize + 1);
940                         amfree(as->repbuf);
941                         as->repbuf = repbuf_temp;
942                 }
943                 strcpy(as->repbuf + as->repbufsize, msg);
944                 as->repbufsize += strlen(msg);
945                 amfree(msg);
946             }
947         }
948     }
949
950     /*
951      * If we got some data, go back and wait for more, or EOF.  Nul terminate
952      * the buffer first.
953      */
954     as->repbuf[n + as->repbufsize] = '\0';
955     if (n > 0) {
956         as->repbufsize += n;
957         if(as->repbufsize >= (as->bufsize - 1)) {
958             as->bufsize *= 2;
959             repbuf_temp = alloc(as->bufsize);
960             memcpy(repbuf_temp, as->repbuf, as->repbufsize + 1);
961             amfree(as->repbuf);
962             as->repbuf = repbuf_temp;
963         }
964         else if(as->send_partial_reply) {
965             amfree(as->rep_pkt.body);
966             pkt_init(&as->rep_pkt, P_PREP, "%s", as->repbuf);
967             do_sendpkt(as->security_handle, &as->rep_pkt);
968             amfree(as->rep_pkt.body);
969             pkt_init_empty(&as->rep_pkt, P_REP);
970         }
971  
972         return (A_PENDING);
973     }
974
975     /*
976      * If we got 0, then we hit EOF.  Process the data and release
977      * the timeout.
978      */
979     assert(n == 0);
980
981     assert(as->ev_repfd != NULL);
982     event_release(as->ev_repfd);
983     as->ev_repfd = NULL;
984
985     assert(as->ev_reptimeout != NULL);
986     event_release(as->ev_reptimeout);
987     as->ev_reptimeout = NULL;
988
989     as->state = s_processrep;
990     aclose(as->repfd);
991     return (A_CONTINUE);
992 }
993
994 /*
995  * After we have read in all of the rep data, we process it and send
996  * it out as a REP packet.
997  */
998 static action_t
999 s_processrep(
1000     struct active_service *     as,
1001     action_t                    action,
1002     pkt_t *                     pkt)
1003 {
1004     char *tok, *repbuf;
1005
1006     (void)action;       /* Quiet unused parameter warning */
1007     (void)pkt;          /* Quiet unused parameter warning */
1008
1009     /*
1010      * Copy the rep lines into the outgoing packet.
1011      *
1012      * If this line is a CONNECT, translate it
1013      * Format is "CONNECT <tag> <handle> <tag> <handle> etc...
1014      * Example:
1015      *
1016      *  CONNECT DATA 4 MESG 5 INDEX 6
1017      *
1018      * The tags are arbitrary.  The handles are in the DATA_FD pool.
1019      * We need to map these to security streams and pass them back
1020      * to the amanda server.  If the handle is -1, then we don't map.
1021      */
1022     if (strncmp_const(as->repbuf,"KENCRYPT\n") == 0) {
1023         amandad_kencrypt = KENCRYPT_WILL_DO;
1024         repbuf = stralloc(as->repbuf + 9);
1025     } else {
1026         repbuf = stralloc(as->repbuf);
1027     }
1028     amfree(as->rep_pkt.body);
1029     pkt_init_empty(&as->rep_pkt, P_REP);
1030     tok = strtok(repbuf, " ");
1031     if (tok == NULL)
1032         goto error;
1033     if (strcmp(tok, "CONNECT") == 0) {
1034         char *line, *nextbuf;
1035
1036         /* Save the entire line */
1037         line = strtok(NULL, "\n");
1038         /* Save the buf following the line */
1039         nextbuf = strtok(NULL, "");
1040
1041         if (line == NULL || nextbuf == NULL)
1042             goto error;
1043
1044         pkt_cat(&as->rep_pkt, "CONNECT");
1045
1046         /* loop over the id/handle pairs */
1047         for (;;) {
1048             /* id */
1049             tok = strtok(line, " ");
1050             line = NULL;        /* keep working from line */
1051             if (tok == NULL)
1052                 break;
1053             pkt_cat(&as->rep_pkt, " %s", tok);
1054
1055             /* handle */
1056             tok = strtok(NULL, " \n");
1057             if (tok == NULL)
1058                 goto error;
1059             /* convert the handle into something the server can process */
1060             pkt_cat(&as->rep_pkt, " %d", allocstream(as, atoi(tok)));
1061         }
1062         pkt_cat(&as->rep_pkt, "\n%s", nextbuf);
1063     } else {
1064 error:
1065         pkt_cat(&as->rep_pkt, "%s", as->repbuf);
1066     }
1067
1068     /*
1069      * We've setup our REP packet in as->rep_pkt.  Now move to the transmission
1070      * state.
1071      */
1072     as->state = s_sendrep;
1073     as->repretry = getconf_int(CNF_REP_TRIES);
1074     amfree(repbuf);
1075     return (A_CONTINUE);
1076 }
1077
1078 /*
1079  * This is the state where we send the REP we just collected from our child.
1080  */
1081 static action_t
1082 s_sendrep(
1083     struct active_service *     as,
1084     action_t                    action,
1085     pkt_t *                     pkt)
1086 {
1087     (void)action;       /* Quiet unused parameter warning */
1088     (void)pkt;          /* Quiet unused parameter warning */
1089
1090     /*
1091      * Transmit it and move to the ack state.
1092      */
1093     do_sendpkt(as->security_handle, &as->rep_pkt);
1094     security_recvpkt(as->security_handle, protocol_recv, as, ACK_TIMEOUT);
1095     as->state = s_ackwait;
1096     return (A_PENDING);
1097 }
1098
1099 /*
1100  * This is the state in which we wait for the server to ACK the REP
1101  * we just sent it.
1102  */
1103 static action_t
1104 s_ackwait(
1105     struct active_service *     as,
1106     action_t                    action,
1107     pkt_t *                     pkt)
1108 {
1109     struct datafd_handle *dh;
1110     int npipes;
1111
1112     /*
1113      * If we got a timeout, try again, but eventually give up.
1114      */
1115     if (action == A_TIMEOUT) {
1116         if (--as->repretry > 0) {
1117             as->state = s_sendrep;
1118             return (A_CONTINUE);
1119         }
1120         dbprintf(_("timeout waiting for ACK for our REP\n"));
1121         return (A_FINISH);
1122     }
1123     amandad_debug(1, _("received ACK, now opening streams\n"));
1124
1125     assert(action == A_RECVPKT);
1126
1127     if (pkt->type == P_REQ) {
1128         dbprintf(_("received dup P_REQ packet, resending REP\n"));
1129         as->state = s_sendrep;
1130         return (A_CONTINUE);
1131     }
1132
1133     if (pkt->type != P_ACK)
1134         return (A_SENDNAK);
1135
1136     if (amandad_kencrypt == KENCRYPT_WILL_DO) {
1137         amandad_kencrypt = KENCRYPT_YES;
1138     }
1139
1140     /*
1141      * Got the ack, now open the pipes
1142      */
1143     for (dh = &as->data[0]; dh < &as->data[DATA_FD_COUNT]; dh++) {
1144         if (dh->netfd == NULL)
1145             continue;
1146         dbprintf("opening security stream for fd %d\n", (int)(dh - as->data) + DATA_FD_OFFSET);
1147         if (security_stream_accept(dh->netfd) < 0) {
1148             dbprintf(_("stream %td accept failed: %s\n"),
1149                 dh - &as->data[0], security_geterror(as->security_handle));
1150             security_stream_close(dh->netfd);
1151             dh->netfd = NULL;
1152             continue;
1153         }
1154
1155         /* setup an event for reads from it.  As a special case, don't start
1156          * listening on as->data[0] until we read some data on another fd, if
1157          * the service is sendbackup.  This ensures that we send a MESG or 
1158          * INDEX token before any DATA tokens, as dumper assumes. This is a
1159          * hack, if that wasn't already obvious! */
1160         if (dh != &as->data[0] || as->service != SERVICE_SENDBACKUP) {
1161             dh->ev_read = event_register((event_id_t)dh->fd_read, EV_READFD,
1162                                          process_readnetfd, dh);
1163         } else {
1164             amandad_debug(1, "Skipping registration of sendbackup's data FD\n");
1165         }
1166
1167         security_stream_read(dh->netfd, process_writenetfd, dh);
1168
1169     }
1170
1171     /*
1172      * Pipes are open, so auth them.  Count them at the same time.
1173      */
1174     for (npipes = 0, dh = &as->data[0]; dh < &as->data[DATA_FD_COUNT]; dh++) {
1175         if (dh->netfd == NULL)
1176             continue;
1177         if (security_stream_auth(dh->netfd) < 0) {
1178             security_stream_close(dh->netfd);
1179             dh->netfd = NULL;
1180             event_release(dh->ev_read);
1181             event_release(dh->ev_write);
1182             dh->ev_read = NULL;
1183             dh->ev_write = NULL;
1184         } else {
1185             npipes++;
1186         }
1187     }
1188
1189     /*
1190      * If no pipes are open, then we're done.  Otherwise, just start running.
1191      * The event handlers on all of the pipes will take it from here.
1192      */
1193     amandad_debug(1, _("at end of s_ackwait, npipes is %d\n"), npipes);
1194     if (npipes == 0)
1195         return (A_FINISH);
1196     else {
1197         security_close(as->security_handle);
1198         as->security_handle = NULL;
1199         return (A_PENDING);
1200     }
1201 }
1202
1203 /*
1204  * Called when a repfd has received data
1205  */
1206 static void
1207 repfd_recv(
1208     void *      cookie)
1209 {
1210     struct active_service *as = cookie;
1211
1212     assert(as != NULL);
1213     assert(as->ev_repfd != NULL);
1214
1215     state_machine(as, A_RECVREP, NULL);
1216 }
1217
1218 static void
1219 process_errfd(
1220     void *cookie)
1221 {
1222     struct active_service *as = cookie;
1223
1224     /* Process errfd before sending the REP packet */
1225     if (as->ev_errfd) {
1226         SELECT_ARG_TYPE readset;
1227         struct timeval  tv;
1228         int             nfound;
1229
1230         memset(&tv, 0, SIZEOF(tv));
1231         FD_ZERO(&readset);
1232         FD_SET(as->errfd, &readset);
1233         nfound = select(as->errfd+1, &readset, NULL, NULL, &tv);
1234         if (nfound && FD_ISSET(as->errfd, &readset)) {
1235             errfd_recv(as);
1236         }
1237     }
1238 }
1239
1240 /*
1241  * Called when a errfd has received data
1242  */
1243 static void
1244 errfd_recv(
1245     void *      cookie)
1246 {
1247     struct active_service *as = cookie;
1248     char  buf[32769];
1249     int   n;
1250     char *r;
1251
1252     assert(as != NULL);
1253     assert(as->ev_errfd != NULL);
1254
1255     n = read(as->errfd, &buf, 32768);
1256     /* merge buffer */
1257     if (n > 0) {
1258         /* Terminate it with '\0' */
1259         buf[n+1] = '\0';
1260
1261         if (as->errbuf) {
1262             as->errbuf = vstrextend(&as->errbuf, buf, NULL);
1263         } else {
1264             as->errbuf = stralloc(buf);
1265         }
1266     } else if (n == 0) {
1267         event_release(as->ev_errfd);
1268         as->ev_errfd = NULL;
1269     } else { /* n < 0 */
1270         event_release(as->ev_errfd);
1271         as->ev_errfd = NULL;
1272         g_snprintf(buf, 32768,
1273                    "error reading stderr or service: %s\n", strerror(errno));
1274     }
1275
1276     /* for each line terminate by '\n' */
1277     while (as->errbuf != NULL  && (r = strchr(as->errbuf, '\n')) != NULL) {
1278         char *s;
1279
1280         *r = '\0';
1281         s = vstrallocf("ERROR service %s: %s\n",
1282                        services[as->service].name, as->errbuf);
1283
1284         /* Add to repbuf, error message will be in the REP packet if it
1285          * is not already sent
1286          */
1287         n = strlen(s);
1288         if (as->bufsize == 0) {
1289             as->bufsize = NETWORK_BLOCK_BYTES;
1290             as->repbuf = alloc(as->bufsize);
1291         }
1292         while (as->bufsize < as->repbufsize + n) {
1293             char *repbuf_temp;
1294             as->bufsize *= 2;
1295             repbuf_temp = alloc(as->bufsize);
1296             memcpy(repbuf_temp, as->repbuf, as->repbufsize + 1);
1297             amfree(as->repbuf);
1298             as->repbuf = repbuf_temp;
1299         }
1300         memcpy(as->repbuf + as->repbufsize, s, n);
1301         as->repbufsize += n;
1302
1303         dbprintf("%s", s);
1304
1305         /* remove first line from buffer */
1306         r++;
1307         s = stralloc(r);
1308         amfree(as->errbuf);
1309         as->errbuf = s;
1310     }
1311 }
1312
1313 /*
1314  * Called when a repfd has timed out
1315  */
1316 static void
1317 timeout_repfd(
1318     void *      cookie)
1319 {
1320     struct active_service *as = cookie;
1321
1322     assert(as != NULL);
1323     assert(as->ev_reptimeout != NULL);
1324
1325     state_machine(as, A_TIMEOUT, NULL);
1326 }
1327
1328 /*
1329  * Called when a handle has received data
1330  */
1331 static void
1332 protocol_recv(
1333     void *              cookie,
1334     pkt_t *             pkt,
1335     security_status_t   status)
1336 {
1337     struct active_service *as = cookie;
1338
1339     assert(as != NULL);
1340
1341     switch (status) {
1342     case S_OK:
1343         dbprintf(_("received %s pkt:\n<<<<<\n%s>>>>>\n"),
1344             pkt_type2str(pkt->type), pkt->body);
1345         state_machine(as, A_RECVPKT, pkt);
1346         break;
1347     case S_TIMEOUT:
1348         dbprintf(_("timeout\n"));
1349         state_machine(as, A_TIMEOUT, NULL);
1350         break;
1351     case S_ERROR:
1352         dbprintf(_("receive error: %s\n"),
1353             security_geterror(as->security_handle));
1354         break;
1355     }
1356 }
1357
1358 /*
1359  * This is a generic relay function that just reads data from one of
1360  * the process's pipes and passes it up the equivalent security_stream_t
1361  */
1362 static void
1363 process_readnetfd(
1364     void *      cookie)
1365 {
1366     pkt_t nak;
1367     struct datafd_handle *dh = cookie;
1368     struct active_service *as = dh->as;
1369     ssize_t n;
1370
1371     nak.body = NULL;
1372
1373     do {
1374         n = read(dh->fd_read, as->databuf, SIZEOF(as->databuf));
1375     } while ((n < 0) && ((errno == EINTR) || (errno == EAGAIN)));
1376
1377     /*
1378      * Process has died.
1379      */
1380     if (n < 0) {
1381         pkt_init(&nak, P_NAK, _("A ERROR data descriptor %d broken: %s\n"),
1382             dh->fd_read, strerror(errno));
1383         goto sendnak;
1384     }
1385     /*
1386      * Process has closed the pipe.  Just remove this event handler.
1387      * If all pipes are closed, shut down this service.
1388      */
1389     if (n == 0) {
1390         event_release(dh->ev_read);
1391         dh->ev_read = NULL;
1392         if(dh->ev_write == NULL) {
1393             security_stream_close(dh->netfd);
1394             dh->netfd = NULL;
1395         }
1396         for (dh = &as->data[0]; dh < &as->data[DATA_FD_COUNT]; dh++) {
1397             if (dh->netfd != NULL)
1398                 return;
1399         }
1400         service_delete(as);
1401         return;
1402     }
1403
1404     /* Handle the special case of recognizing "sendbackup info end"
1405      * from sendbackup's MESG fd */
1406     if (as->service == SERVICE_SENDBACKUP && !as->seen_info_end && dh == &as->data[1]) {
1407         /* make a buffer containing the combined data from info_end_buf
1408          * and what we've read this time, and search it for info_end_strj
1409          * This includes a NULL byte for strstr's sanity. */
1410         char *combined_buf = malloc(INFO_END_LEN + n + 1);
1411         memcpy(combined_buf, as->info_end_buf, INFO_END_LEN);
1412         memcpy(combined_buf+INFO_END_LEN, as->databuf, n);
1413         combined_buf[INFO_END_LEN+n] = '\0';
1414
1415         as->seen_info_end = (strstr(combined_buf, info_end_str) != NULL);
1416
1417         /* fill info_end_buf from the tail end of combined_buf */
1418         memcpy(as->info_end_buf, combined_buf + n, INFO_END_LEN);
1419         amfree(combined_buf);
1420
1421         /* if we did see info_end_str, start reading the data fd (fd 0) */
1422         if (as->seen_info_end) {
1423             struct datafd_handle *dh = &as->data[0];
1424             amandad_debug(1, "Opening datafd to sendbackup (delayed until sendbackup sent header info)\n");
1425             dh->ev_read = event_register((event_id_t)dh->fd_read, EV_READFD,
1426                                          process_readnetfd, dh);
1427         } else {
1428             amandad_debug(1, "sendbackup header info still not complete\n");
1429         }
1430     }
1431
1432     if (security_stream_write(dh->netfd, as->databuf, (size_t)n) < 0) {
1433         /* stream has croaked */
1434         pkt_init(&nak, P_NAK, _("ERROR write error on stream %d: %s\n"),
1435             security_stream_id(dh->netfd),
1436             security_stream_geterror(dh->netfd));
1437         goto sendnak;
1438     }
1439     return;
1440
1441 sendnak:
1442     do_sendpkt(as->security_handle, &nak);
1443     service_delete(as);
1444     amfree(nak.body);
1445 }
1446
1447 /*
1448  * This is a generic relay function that just read data from one of
1449  * the security_stream_t and passes it up the equivalent process's pipes
1450  */
1451 static void
1452 process_writenetfd(
1453     void *      cookie,
1454     void *      buf,
1455     ssize_t     size)
1456 {
1457     struct datafd_handle *dh;
1458
1459     assert(cookie != NULL);
1460     dh = cookie;
1461
1462     if (dh->fd_write <= 0) {
1463         dbprintf(_("process_writenetfd: dh->fd_write <= 0\n"));
1464     } else if (size > 0) {
1465         full_write(dh->fd_write, buf, (size_t)size);
1466         security_stream_read(dh->netfd, process_writenetfd, dh);
1467     }
1468     else {
1469         aclose(dh->fd_write);
1470     }
1471 }
1472
1473
1474 /*
1475  * Convert a local stream handle (DATA_FD...) into something that
1476  * can be sent to the amanda server.
1477  *
1478  * Returns a number that should be sent to the server in the REP packet.
1479  */
1480 static int
1481 allocstream(
1482     struct active_service *     as,
1483     int                         handle)
1484 {
1485     struct datafd_handle *dh;
1486
1487     /* note that handle is in the range DATA_FD_OFFSET to DATA_FD_COUNT, but
1488      * it is NOT a file descriptor! */
1489
1490     /* if the handle is -1, then we don't bother */
1491     if (handle < 0)
1492         return (-1);
1493
1494     /* make sure the handle's kosher */
1495     if (handle < DATA_FD_OFFSET || handle >= DATA_FD_OFFSET + DATA_FD_COUNT)
1496         return (-1);
1497
1498     /* get a pointer into our handle array */
1499     dh = &as->data[handle - DATA_FD_OFFSET];
1500
1501     /* make sure we're not already using the net handle */
1502     if (dh->netfd != NULL)
1503         return (-1);
1504
1505     /* allocate a stream from the security layer and return */
1506     dh->netfd = security_stream_server(as->security_handle);
1507     if (dh->netfd == NULL) {
1508         dbprintf(_("couldn't open stream to server: %s\n"),
1509             security_geterror(as->security_handle));
1510         return (-1);
1511     }
1512
1513     /*
1514      * convert the stream into a numeric id that can be sent to the
1515      * remote end.
1516      */
1517     return (security_stream_id(dh->netfd));
1518 }
1519
1520 /*
1521  * Create a new service instance
1522  */
1523 static struct active_service *
1524 service_new(
1525     security_handle_t * security_handle,
1526     const char *        cmd,
1527     service_t           service,
1528     const char *        arguments)
1529 {
1530     int i;
1531     int data_read[DATA_FD_COUNT + 2][2];
1532     int data_write[DATA_FD_COUNT + 2][2];
1533     struct active_service *as;
1534     pid_t pid;
1535     int newfd;
1536     char *peer_name;
1537     char *amanda_remote_host_env[2];
1538
1539     assert(security_handle != NULL);
1540     assert(cmd != NULL);
1541     assert(arguments != NULL);
1542
1543     /* a plethora of pipes */
1544     /* data_read[0]                : stdin
1545      * data_write[0]               : stdout
1546      * data_read[1], data_write[1] : first  stream
1547      * data_read[2], data_write[2] : second stream
1548      * data_read[3], data_write[3] : third stream
1549      * data_write[4]               : stderr
1550      */
1551     for (i = 0; i < DATA_FD_COUNT + 1; i++) {
1552         if (pipe(data_read[i]) < 0) {
1553             error(_("pipe: %s\n"), strerror(errno));
1554             /*NOTREACHED*/
1555         }
1556         if (pipe(data_write[i]) < 0) {
1557             error(_("pipe: %s\n"), strerror(errno));
1558             /*NOTREACHED*/
1559         }
1560     }
1561     if (pipe(data_write[STDERR_PIPE]) < 0) {
1562         error(_("pipe: %s\n"), strerror(errno));
1563         /*NOTREACHED*/
1564     }
1565
1566     switch(pid = fork()) {
1567     case -1:
1568         error(_("could not fork service %s: %s\n"), cmd, strerror(errno));
1569         /*NOTREACHED*/
1570     default:
1571         /*
1572          * The parent.  Close the far ends of our pipes and return.
1573          */
1574         as = g_new0(struct active_service, 1);
1575         as->cmd = stralloc(cmd);
1576         as->arguments = stralloc(arguments);
1577         as->security_handle = security_handle;
1578         as->state = NULL;
1579         as->service = service;
1580         as->pid = pid;
1581         as->send_partial_reply = 0;
1582         as->seen_info_end = FALSE;
1583         /* fill in info_end_buf with non-null characters */
1584         memset(as->info_end_buf, '-', sizeof(as->info_end_buf));
1585         if(service == SERVICE_SENDSIZE) {
1586             g_option_t *g_options;
1587             char *option_str, *p;
1588
1589             option_str = stralloc(as->arguments+8);
1590             p = strchr(option_str,'\n');
1591             if(p) *p = '\0';
1592
1593             g_options = parse_g_options(option_str, 1);
1594             if(am_has_feature(g_options->features, fe_partial_estimate)) {
1595                 as->send_partial_reply = 1;
1596             }
1597             free_g_options(g_options);
1598             amfree(option_str);
1599         }
1600
1601         /* write to the request pipe */
1602         aclose(data_read[0][0]);
1603         as->reqfd = data_read[0][1];
1604
1605         /*
1606          * read from the reply pipe
1607          */
1608         as->repfd = data_write[0][0];
1609         aclose(data_write[0][1]);
1610         as->ev_repfd = NULL;
1611         as->repbuf = NULL;
1612         as->repbufsize = 0;
1613         as->bufsize = 0;
1614         as->repretry = 0;
1615         as->rep_pkt.body = NULL;
1616
1617         /*
1618          * read from the stderr pipe
1619          */
1620         as->errfd = data_write[STDERR_PIPE][0];
1621         aclose(data_write[STDERR_PIPE][1]);
1622         as->ev_errfd = NULL;
1623         as->errbuf = NULL;
1624
1625         /*
1626          * read from the rest of the general-use pipes
1627          * (netfds are opened as the client requests them)
1628          */
1629         for (i = 0; i < DATA_FD_COUNT; i++) {
1630             aclose(data_read[i + 1][1]);
1631             aclose(data_write[i + 1][0]);
1632             as->data[i].fd_read = data_read[i + 1][0];
1633             as->data[i].fd_write = data_write[i + 1][1];
1634             as->data[i].ev_read = NULL;
1635             as->data[i].ev_write = NULL;
1636             as->data[i].netfd = NULL;
1637             as->data[i].as = as;
1638         }
1639
1640         /* add it to the service queue */
1641         /* increment the active service count */
1642         serviceq = g_slist_append(serviceq, (gpointer)as);
1643
1644         return (as);
1645     case 0:
1646         /*
1647          * The child.  Put our pipes in their advertised locations
1648          * and start up.
1649          */
1650
1651         /* set up the AMANDA_AUTHENTICATED_PEER env var so child services
1652          * can use it to authenticate */
1653         peer_name = security_get_authenticated_peer_name(security_handle);
1654         amanda_remote_host_env[0] = NULL;
1655         amanda_remote_host_env[1] = NULL;
1656         if (*peer_name) {
1657             amanda_remote_host_env[0] =
1658                 g_strdup_printf("AMANDA_AUTHENTICATED_PEER=%s", peer_name);
1659         }
1660
1661         /*
1662          * The data stream is stdin in the new process
1663          */
1664         if (dup2(data_read[0][0], 0) < 0) {
1665             error(_("dup %d to %d failed: %s\n"), data_read[0][0], 0,
1666                 strerror(errno));
1667             /*NOTREACHED*/
1668         }
1669         aclose(data_read[0][0]);
1670         aclose(data_read[0][1]);
1671
1672         /*
1673          * The reply stream is stdout
1674          */
1675         if (dup2(data_write[0][1], 1) < 0) {
1676             error(_("dup %d to %d failed: %s\n"), data_write[0][1], 1,
1677                 strerror(errno));
1678         }
1679         aclose(data_write[0][0]);
1680         aclose(data_write[0][1]);
1681
1682         for (i = 0; i < DATA_FD_COUNT; i++) {
1683             aclose(data_read[i + 1][0]);
1684             aclose(data_write[i + 1][1]);
1685         }
1686
1687         /*
1688          *  Make sure they are not open in the range DATA_FD_OFFSET to
1689          *      DATA_FD_OFFSET + DATA_FD_COUNT*2 - 1
1690          */
1691         for (i = 0; i < DATA_FD_COUNT; i++) {
1692             while(data_read[i + 1][1] >= DATA_FD_OFFSET &&
1693                   data_read[i + 1][1] <= DATA_FD_OFFSET + DATA_FD_COUNT*2 - 1) {
1694                 newfd = dup(data_read[i + 1][1]);
1695                 if(newfd == -1)
1696                     error(_("Can't dup out off DATA_FD range"));
1697                 data_read[i + 1][1] = newfd;
1698             }
1699             while(data_write[i + 1][0] >= DATA_FD_OFFSET &&
1700                   data_write[i + 1][0] <= DATA_FD_OFFSET + DATA_FD_COUNT*2 - 1) {
1701                 newfd = dup(data_write[i + 1][0]);
1702                 if(newfd == -1)
1703                     error(_("Can't dup out off DATA_FD range"));
1704                 data_write[i + 1][0] = newfd;
1705             }
1706         }
1707         while(data_write[4][0] >= DATA_FD_OFFSET &&
1708               data_write[4][0] <= DATA_FD_OFFSET + DATA_FD_COUNT*2 - 1) {
1709             newfd = dup(data_write[4][0]);
1710             if (newfd == -1)
1711                 error(_("Can't dup out off DATA_FD range"));
1712             data_write[4][0] = newfd;
1713         }
1714         while(data_write[4][1] >= DATA_FD_OFFSET &&
1715               data_write[4][1] <= DATA_FD_OFFSET + DATA_FD_COUNT*2 - 1) {
1716             newfd = dup(data_write[4][1]);
1717             if (newfd == -1)
1718                 error(_("Can't dup out off DATA_FD range"));
1719             data_write[4][1] = newfd;
1720         }
1721
1722         for (i = 0; i < DATA_FD_COUNT*2; i++)
1723             close(DATA_FD_OFFSET + i);
1724
1725         /*
1726          * The rest start at the offset defined in amandad.h, and continue
1727          * through the internal defined.
1728          */
1729         for (i = 0; i < DATA_FD_COUNT; i++) {
1730             if (dup2(data_read[i + 1][1], i*2 + DATA_FD_OFFSET) < 0) {
1731                 error(_("dup %d to %d failed: %s\n"), data_read[i + 1][1],
1732                     i + DATA_FD_OFFSET, strerror(errno));
1733             }
1734             aclose(data_read[i + 1][1]);
1735
1736             if (dup2(data_write[i + 1][0], i*2 + 1 + DATA_FD_OFFSET) < 0) {
1737                 error(_("dup %d to %d failed: %s\n"), data_write[i + 1][0],
1738                     i + DATA_FD_OFFSET, strerror(errno));
1739             }
1740             aclose(data_write[i + 1][0]);
1741         }
1742
1743         /* close all unneeded fd */
1744         close(STDERR_FILENO);
1745         dup2(data_write[STDERR_PIPE][1], 2);
1746         aclose(data_write[STDERR_PIPE][0]);
1747         aclose(data_write[STDERR_PIPE][1]);
1748         safe_fd(DATA_FD_OFFSET, DATA_FD_COUNT*2);
1749
1750         execle(cmd, cmd, "amandad", auth, (char *)NULL, safe_env_full(amanda_remote_host_env));
1751         error(_("could not exec service %s: %s\n"), cmd, strerror(errno));
1752         /*NOTREACHED*/
1753     }
1754     return NULL;
1755 }
1756
1757 /*
1758  * Unallocate a service instance
1759  */
1760 static void
1761 service_delete(
1762     struct active_service *     as)
1763 {
1764     int i;
1765     int   count;
1766     pid_t pid;
1767     struct datafd_handle *dh;
1768
1769     amandad_debug(1, _("closing service: %s\n"),
1770                       (as->cmd)?as->cmd:_("??UNKONWN??"));
1771
1772     assert(as != NULL);
1773
1774     assert(as->cmd != NULL);
1775     amfree(as->cmd);
1776
1777     assert(as->arguments != NULL);
1778     amfree(as->arguments);
1779
1780     if (as->reqfd != -1)
1781         aclose(as->reqfd);
1782     if (as->repfd != -1)
1783         aclose(as->repfd);
1784     if (as->errfd != -1) {
1785         process_errfd(as);
1786         aclose(as->errfd);
1787     }
1788
1789     if (as->ev_repfd != NULL)
1790         event_release(as->ev_repfd);
1791     if (as->ev_reptimeout != NULL)
1792         event_release(as->ev_reptimeout);
1793     if (as->ev_errfd != NULL)
1794         event_release(as->ev_errfd);
1795
1796     for (i = 0; i < DATA_FD_COUNT; i++) {
1797         dh = &as->data[i];
1798
1799         aclose(dh->fd_read);
1800         aclose(dh->fd_write);
1801
1802         if (dh->netfd != NULL)
1803             security_stream_close(dh->netfd);
1804
1805         if (dh->ev_read != NULL)
1806             event_release(dh->ev_read);
1807         if (dh->ev_write != NULL)
1808             event_release(dh->ev_write);
1809     }
1810
1811     if (as->security_handle != NULL)
1812         security_close(as->security_handle);
1813
1814     assert(as->pid > 0);
1815     kill(as->pid, SIGTERM);
1816     pid = waitpid(as->pid, NULL, WNOHANG);
1817     count = 5;
1818     while (pid != as->pid && count > 0) {
1819         count--;
1820         sleep(1);
1821         pid = waitpid(as->pid, NULL, WNOHANG);
1822     }
1823     if (pid != as->pid) {
1824         g_debug("Process %d failed to exit", (int)as->pid);
1825     }
1826
1827     serviceq = g_slist_remove(serviceq, (gpointer)as);
1828
1829     amfree(as->cmd);
1830     amfree(as->arguments);
1831     amfree(as->repbuf);
1832     amfree(as->rep_pkt.body);
1833     amfree(as);
1834
1835     if(exit_on_qlength == 0 && g_slist_length(serviceq) == 0) {
1836         dbclose();
1837         exit(0);
1838     }
1839 }
1840
1841 /*
1842  * Like 'fullwrite', but does the work in a child process so pipelines
1843  * do not hang.
1844  */
1845 static int
1846 writebuf(
1847     struct active_service *     as,
1848     const void *                bufp,
1849     size_t                      size)
1850 {
1851     pid_t pid;
1852     size_t    writesize;
1853
1854     switch (pid=fork()) {
1855     case -1:
1856         break;
1857
1858     default:
1859         waitpid(pid, NULL, WNOHANG);
1860         return 0;                       /* this is the parent */
1861
1862     case 0:                             /* this is the child */
1863         close(as->repfd);
1864         writesize = full_write(as->reqfd, bufp, size);
1865         exit(writesize != size);
1866         /* NOTREACHED */
1867     }
1868     return -1;
1869 }
1870
1871 static ssize_t
1872 do_sendpkt(
1873     security_handle_t * handle,
1874     pkt_t *             pkt)
1875 {
1876     dbprintf(_("sending %s pkt:\n<<<<<\n%s>>>>>\n"),
1877         pkt_type2str(pkt->type), pkt->body);
1878     if (handle)
1879         return security_sendpkt(handle, pkt);
1880     else
1881         return 1;
1882 }
1883
1884 /*
1885  * Convert a state into a string
1886  */
1887 static const char *
1888 state2str(
1889     state_t     state)
1890 {
1891     static const struct {
1892         state_t state;
1893         const char str[13];
1894     } states[] = {
1895 #define X(state)        { state, stringize(state) }
1896         X(s_sendack),
1897         X(s_repwait),
1898         X(s_processrep),
1899         X(s_sendrep),
1900         X(s_ackwait),
1901 #undef X
1902     };
1903     int i;
1904
1905     for (i = 0; i < (int)(sizeof(states) / sizeof(states[0])); i++)
1906         if (state == states[i].state)
1907             return (states[i].str);
1908     return (_("INVALID STATE"));
1909 }
1910
1911 /*
1912  * Convert an action into a string
1913  */
1914 static const char *
1915 action2str(
1916     action_t    action)
1917 {
1918     static const struct {
1919         action_t action;
1920         const char str[12];
1921     } actions[] = {
1922 #define X(action)       { action, stringize(action) }
1923         X(A_START),
1924         X(A_RECVPKT),
1925         X(A_RECVREP),
1926         X(A_PENDING),
1927         X(A_FINISH),
1928         X(A_CONTINUE),
1929         X(A_SENDNAK),
1930         X(A_TIMEOUT),
1931 #undef X
1932     };
1933     int i;
1934
1935     for (i = 0; i < (int)(sizeof(actions) / sizeof(actions[0])); i++)
1936         if (action == actions[i].action)
1937             return (actions[i].str);
1938     return (_("UNKNOWN ACTION"));
1939 }
1940
1941 static char *
1942 amandad_get_security_conf(
1943     char *      string,
1944     void *      arg)
1945 {
1946     (void)arg;      /* Quiet unused parameter warning */
1947
1948     if (!string || !*string)
1949         return(NULL);
1950
1951     if (strcmp(string, "kencrypt")==0) {
1952         if (amandad_kencrypt == KENCRYPT_YES)
1953             return ("yes");
1954         else
1955             return (NULL);
1956     }
1957     return(NULL);
1958 }
1959