amandad.c

开源备份软件源码 AMANDA, the Advanced Maryland Automatic Network Disk Archiver, is a backup system that a

    /* see if it's one we allow */
    for (i = 0; i < (int)NSERVICES; i++)
	if (services[i].active == 1 && strcmp(services[i].name, service) == 0)
	    break;
    if (i == (int)NSERVICES) {
	dbprintf(_("%s: invalid service\n"), service);
	pkt_init(&pkt_out, P_NAK, _("ERROR %s: invalid service, add '%s' as argument to amandad\n"), service, service);
	goto send_pkt_out;
    }

    service_path = vstralloc(amlibexecdir, "/", service, versionsuffix(), NULL);
    if (access(service_path, X_OK) < 0) {
	dbprintf(_("can't execute %s: %s\n"), service_path, strerror(errno));
	    pkt_init(&pkt_out, P_NAK,
		     _("ERROR execute access to \"%s\" denied\n"),
		     service_path);
	goto send_pkt_out;
    }

    /* see if its already running */
    for (as = TAILQ_FIRST(&serviceq.tailq); as != NULL;
	as = TAILQ_NEXT(as, tq)) {
	    if (strcmp(as->cmd, service_path) == 0 &&
		strcmp(as->arguments, arguments) == 0) {
		    dbprintf(_("%s %s: already running, acking req\n"),
			service, arguments);
		    pkt_init_empty(&pkt_out, P_ACK);
		    goto send_pkt_out_no_delete;
	    }
    }

    /*
     * create a new service instance, and send the arguments down
     * the request pipe.
     */
    dbprintf(_("creating new service: %s\n%s\n"), service, arguments);
    as = service_new(handle, service_path, arguments);
    if (writebuf(as, arguments, strlen(arguments)) < 0) {
	const char *errmsg = strerror(errno);
	dbprintf(_("error sending arguments to %s: %s\n"), service, errmsg);
	pkt_init(&pkt_out, P_NAK, _("ERROR error writing arguments to %s: %s\n"),
	    service, errmsg);
	goto send_pkt_out;
    }
    aclose(as->reqfd);

    amfree(pktbody);
    amfree(service);
    amfree(service_path);
    amfree(arguments);

    /*
     * Move to the sendack state, and start up the state
     * machine.
     */
    as->state = s_sendack;
    state_machine(as, A_START, NULL);
    return;

badreq:
    pkt_init(&pkt_out, P_NAK, _("ERROR invalid REQ\n"));
    dbprintf(_("received invalid %s packet:\n<<<<<\n%s>>>>>\n\n"),
	pkt_type2str(pkt->type), pkt->body);

send_pkt_out:
    if(as)
	service_delete(as);
send_pkt_out_no_delete:
    amfree(pktbody);
    amfree(service_path);
    amfree(service);
    amfree(arguments);
    do_sendpkt(handle, &pkt_out);
    security_close(handle);
    amfree(pkt_out.body);
}

/*
 * Handles incoming protocol packets.  Routes responses to the proper
 * running service.
 */
static void
state_machine(
    struct active_service *	as,
    action_t			action,
    pkt_t *			pkt)
{
    action_t retaction;
    state_t curstate;
    pkt_t nak;

    amandad_debug(1, _("state_machine: %p entering\n"), as);
    for (;;) {
	curstate = as->state;
	amandad_debug(1, _("state_machine: %p curstate=%s action=%s\n"), as,
			  state2str(curstate), action2str(action));
	retaction = (*curstate)(as, action, pkt);
	amandad_debug(1, _("state_machine: %p curstate=%s returned %s (nextstate=%s)\n"),
			  as, state2str(curstate), action2str(retaction),
			  state2str(as->state));

	switch (retaction) {
	/*
	 * State has queued up and is now blocking on input.
	 */
	case A_PENDING:
	    amandad_debug(1, _("state_machine: %p leaving (A_PENDING)\n"), as);
	    return;

	/*
	 * service has switched states.  Loop.
	 */
	case A_CONTINUE:
	    break;

	/*
	 * state has determined that the packet it received was bogus.
	 * Send a nak, and return.
	 */
	case A_SENDNAK:
	    dbprintf(_("received unexpected %s packet\n"),
		pkt_type2str(pkt->type));
	    dbprintf(_("<<<<<\n%s----\n\n"), pkt->body);
	    pkt_init(&nak, P_NAK, _("ERROR unexpected packet type %s\n"),
		pkt_type2str(pkt->type));
	    do_sendpkt(as->security_handle, &nak);
	    amfree(nak.body);
	    security_recvpkt(as->security_handle, protocol_recv, as, -1);
	    amandad_debug(1, _("state_machine: %p leaving (A_SENDNAK)\n"), as);
	    return;

	/*
	 * Service is done.  Remove it and finish.
	 */
	case A_FINISH:
	    amandad_debug(1, _("state_machine: %p leaving (A_FINISH)\n"), as);
	    service_delete(as);
	    return;

	default:
	    assert(0);
	    break;
	}
    }
    /*NOTREACHED*/
}

/*
 * This state just sends an ack.  After that, we move to the repwait
 * state to wait for REP data to arrive from the subprocess.
 */
static action_t
s_sendack(
    struct active_service *	as,
    action_t			action,
    pkt_t *			pkt)
{
    pkt_t ack;

    (void)action;	/* Quiet unused parameter warning */
    (void)pkt;		/* Quiet unused parameter warning */

    pkt_init_empty(&ack, P_ACK);
    if (do_sendpkt(as->security_handle, &ack) < 0) {
	dbprintf(_("error sending ACK: %s\n"),
	    security_geterror(as->security_handle));
	amfree(ack.body);
	return (A_FINISH);
    }
    amfree(ack.body);

    /*
     * move to the repwait state
     * Setup a listener for data on the reply fd, but also
     * listen for packets over the wire, as the server may
     * poll us if we take a long time.
     * Setup a timeout that will fire if it takes too long to
     * receive rep data.
     */
    as->state = s_repwait;
    as->ev_repfd = event_register((event_id_t)as->repfd, EV_READFD, repfd_recv, as);
    as->ev_reptimeout = event_register(REP_TIMEOUT, EV_TIME,
	timeout_repfd, as);
    security_recvpkt(as->security_handle, protocol_recv, as, -1);
    return (A_PENDING);
}

/*
 * This is the repwait state.  We have responded to the initial REQ with
 * an ACK, and we are now waiting for the process we spawned to pass us 
 * data to send in a REP.
 */
static action_t
s_repwait(
    struct active_service *	as,
    action_t			action,
    pkt_t *			pkt)
{
    ssize_t   n;
    char     *repbuf_temp;
    char     *what;
    char     *msg;
    int       code = 0;
    int       t;
    int       pid;
    amwait_t  retstat;

    /*
     * We normally shouldn't receive any packets while waiting
     * for our REP data, but in some cases we do.
     */
    if (action == A_RECVPKT) {
	assert(pkt != NULL);
	/*
	 * Another req for something that's running.  Just send an ACK
	 * and go back and wait for more data.
	 */
	if (pkt->type == P_REQ) {
	    dbprintf(_("received dup P_REQ packet, ACKing it\n"));
	    amfree(as->rep_pkt.body);
	    pkt_init_empty(&as->rep_pkt, P_ACK);
	    do_sendpkt(as->security_handle, &as->rep_pkt);
	    security_recvpkt(as->security_handle, protocol_recv, as, -1);
	    return (A_PENDING);
	}
	/* something unexpected.  Nak it */
	return (A_SENDNAK);
    }

    if (action == A_TIMEOUT) {
	amfree(as->rep_pkt.body);
	pkt_init(&as->rep_pkt, P_NAK, _("ERROR timeout on reply pipe\n"));
	dbprintf(_("%s timed out waiting for REP data\n"), as->cmd);
	do_sendpkt(as->security_handle, &as->rep_pkt);
	return (A_FINISH);
    }

    assert(action == A_RECVREP);
    if(as->bufsize == 0) {
	as->bufsize = NETWORK_BLOCK_BYTES;
	as->repbuf = alloc(as->bufsize);
    }

    do {
	n = read(as->repfd, as->repbuf + as->repbufsize,
		 as->bufsize - as->repbufsize - 1);
    } while ((n < 0) && ((errno == EINTR) || (errno == EAGAIN)));
    if (n < 0) {
	const char *errstr = strerror(errno);
	dbprintf(_("read error on reply pipe: %s\n"), errstr);
	amfree(as->rep_pkt.body);
	pkt_init(&as->rep_pkt, P_NAK, _("ERROR read error on reply pipe: %s\n"),
		 errstr);
	do_sendpkt(as->security_handle, &as->rep_pkt);
	return (A_FINISH);
    }

    /* If end of service, wait for process status */
    if (n == 0) {
	t = 0;
	pid = waitpid(as->pid, &retstat, WNOHANG);
	while (t<5 && pid == 0) {
	    sleep(1);
	    t++;
	    pid = waitpid(as->pid, &retstat, WNOHANG);
	}

	if (pid > 0) {
	    what = NULL;
	    if (! WIFEXITED(retstat)) {
		what = _("signal");
		code = WTERMSIG(retstat);
	    } else if (WEXITSTATUS(retstat) != 0) {
		what = _("code");
		code = WEXITSTATUS(retstat);
	    }
	    if (what) {
		dbprintf(_("service %s failed: pid %u exited with %s %d\n"),
			 (as->cmd)?as->cmd:_("??UNKONWN??"),
			 (unsigned)as->pid,
			 what, code);
		msg = vstrallocf(
		     _("ERROR service %s failed: pid %u exited with %s %d\n"),
		     (as->cmd)?as->cmd:_("??UNKONWN??"), (unsigned)as->pid,
		     what, code);
		if (as->repbufsize + strlen(msg) >= (as->bufsize - 1)) {
			as->bufsize *= 2;
			repbuf_temp = alloc(as->bufsize);
			memcpy(repbuf_temp, as->repbuf, as->repbufsize + 1);
			amfree(as->repbuf);
			as->repbuf = repbuf_temp;
		}
		strcpy(as->repbuf + as->repbufsize, msg);
		as->repbufsize += strlen(msg);
	    }
	}
    }

    /*
     * If we got some data, go back and wait for more, or EOF.  Nul terminate
     * the buffer first.
     */
    as->repbuf[n + as->repbufsize] = '\0';
    if (n > 0) {
	as->repbufsize += n;
	if(as->repbufsize >= (as->bufsize - 1)) {
	    as->bufsize *= 2;
	    repbuf_temp = alloc(as->bufsize);
	    memcpy(repbuf_temp, as->repbuf, as->repbufsize + 1);
	    amfree(as->repbuf);
	    as->repbuf = repbuf_temp;
	}
	else if(as->send_partial_reply) {
	    amfree(as->rep_pkt.body);
	    pkt_init(&as->rep_pkt, P_PREP, "%s", as->repbuf);
	    do_sendpkt(as->security_handle, &as->rep_pkt);
	    amfree(as->rep_pkt.body);
	    pkt_init_empty(&as->rep_pkt, P_REP);
	}
 
	return (A_PENDING);
    }

    /*
     * If we got 0, then we hit EOF.  Process the data and release
     * the timeout.
     */
    assert(n == 0);

    assert(as->ev_repfd != NULL);
    event_release(as->ev_repfd);
    as->ev_repfd = NULL;

    assert(as->ev_reptimeout != NULL);
    event_release(as->ev_reptimeout);
    as->ev_reptimeout = NULL;

    as->state = s_processrep;
    aclose(as->repfd);
    return (A_CONTINUE);
}

/*
 * After we have read in all of the rep data, we process it and send
 * it out as a REP packet.
 */
static action_t
s_processrep(
    struct active_service *	as,
    action_t			action,
    pkt_t *			pkt)
{
    char *tok, *repbuf;

    (void)action;	/* Quiet unused parameter warning */
    (void)pkt;		/* Quiet unused parameter warning */

    /*
     * Copy the rep lines into the outgoing packet.
     *
     * If this line is a CONNECT, translate it
     * Format is "CONNECT <tag> <handle> <tag> <handle> etc...
     * Example:
     *
     *  CONNECT DATA 4 MESG 5 INDEX 6
     *
     * The tags are arbitrary.  The handles are in the DATA_FD pool.
     * We need to map these to security streams and pass them back
     * to the amanda server.  If the handle is -1, then we don't map.
     */
    if (strncmp_const(as->repbuf,"KENCRYPT\n") == 0) {
        amandad_kencrypt = KENCRYPT_WILL_DO;
	repbuf = stralloc(as->repbuf + 9);
    } else {
	repbuf = stralloc(as->repbuf);
    }
    amfree(as->rep_pkt.body);
    pkt_init_empty(&as->rep_pkt, P_REP);
    tok = strtok(repbuf, " ");
    if (tok == NULL)
	goto error;
    if (strcmp(tok, "CONNECT") == 0) {
	char *line, *nextbuf;

	/* Save the entire line */
	line = strtok(NULL, "\n");
	/* Save the buf following the line */
	nextbuf = strtok(NULL, "");

	if (line == NULL || nextbuf == NULL)
	    goto error;

	pkt_cat(&as->rep_pkt, "CONNECT");

	/* loop over the id/handle pairs */
	for (;;) {
	    /* id */
	    tok = strtok(line, " ");
	    line = NULL;	/* keep working from line */
	    if (tok == NULL)
		break;
	    pkt_cat(&as->rep_pkt, " %s", tok);

	    /* handle */
	    tok = strtok(NULL, " \n");
	    if (tok == NULL)
		goto error;
	    /* convert the handle into something the server can process */
	    pkt_cat(&as->rep_pkt, " %d", allocstream(as, atoi(tok)));
	}
	pkt_cat(&as->rep_pkt, "\n%s", nextbuf);
    } else {
error:
	pkt_cat(&as->rep_pkt, "%s", as->repbuf);
    }

    /*
     * We've setup our REP packet in as->rep_pkt.  Now move to the transmission
     * state.
     */
    as->state = s_sendrep;
    as->repretry = getconf_int(CNF_REP_TRIES);
    amfree(repbuf);
    return (A_CONTINUE);
}

/*
 * This is the state where we send the REP we just collected from our child.
 */
static action_t
s_sendrep(
    struct active_service *	as,
    action_t			action,
    pkt_t *			pkt)
{
    (void)action;	/* Quiet unused parameter warning */
    (void)pkt;		/* Quiet unused parameter warning */

    /*
     * Transmit it and move to the ack state.
     */
    do_sendpkt(as->security_handle, &as->rep_pkt);
    security_recvpkt(as->security_handle, protocol_recv, as, ACK_TIMEOUT);
    as->state = s_ackwait;
    return (A_PENDING);
}

/*
 * This is the state in which we wait for the server to ACK the REP
 * we just sent it.
 */
static action_t
s_ackwait(
    struct active_service *	as,
    action_t			action,
    pkt_t *			pkt)
{
    struct datafd_handle *dh;
    int npipes;

    /*
     * If we got a timeout, try again, but eventually give up.
     */
    if (action == A_TIMEOUT) {
	if (--as->repretry > 0) {
	    as->state = s_sendrep;
	    return (A_CONTINUE);
	}
	dbprintf(_("timeout waiting for ACK for our REP\n"));
	return (A_FINISH);
    }
    amandad_debug(1, _("received ACK, now opening streams\n"));

    assert(action == A_RECVPKT);

    if (pkt->type == P_REQ) {
	dbprintf(_("received dup P_REQ packet, resending REP\n"));
	as->state = s_sendrep;
	return (A_CONTINUE);
    }

    if (pkt->type != P_ACK)
	return (A_SENDNAK);

    if (amandad_kencrypt == KENCRYPT_WILL_DO) {
	amandad_kencrypt = KENCRYPT_YES;
    }

    /*
     * Got the ack, now open the pipes
     */
    for (dh = &as->data[0]; dh < &as->data[DATA_FD_COUNT]; dh++) {
	if (dh->netfd == NULL)
	    continue;
	if (security_stream_accept(dh->netfd) < 0) {
	    dbprintf(_("stream %td accept failed: %s\n"),
		dh - &as->data[0], security_geterror(as->security_handle));
	    security_stream_close(dh->netfd);
	    dh->netfd = NULL;
	    continue;
	}
	/* setup an event for reads from it */
	dh->ev_read = event_register((event_id_t)dh->fd_read, EV_READFD,
				     process_readnetfd, dh);

	security_stream_read(dh->netfd, process_writenetfd, dh);

    }

    /*
     * Pipes are open, so auth them.  Count them at the same time.
     */
    for (npipes = 0, dh = &as->data[0]; dh < &as->data[DATA_FD_COUNT]; dh++) {
	if (dh->netfd == NULL)
	    continue;
	if (security_stream_auth(dh->netfd) < 0) {
	    security_stream_close(dh->netfd);
	    dh->netfd = NULL;
	    event_release(dh->ev_read);
	    event_release(dh->ev_write);
	    dh->ev_read = NULL;
	    dh->ev_write = NULL;
	} else {
	    npipes++;
	}
    }

    /*
     * If no pipes are open, then we're done.  Otherwise, just start running.
     * The event handlers on all of the pipes will take it from here.
     */
    amandad_debug(1, _("at end of s_ackwait, npipes is %d\n"), npipes);
    if (npipes == 0)
	return (A_FINISH);
    else {
	security_close(as->security_handle);
	as->security_handle = NULL;
	return (A_PENDING);
    }
}

/*
 * Called when a repfd has received data
 */
static void
repfd_recv(
    void *	cookie)
{
    struct active_service *as = cookie;

    assert(as != NULL);
    assert(as->ev_repfd != NULL);