X-Git-Url: http://dolda2000.com/gitweb/?p=doldaconnect.git;a=blobdiff_plain;f=daemon%2Ftransfer.c;h=9c50ed102342e0b7c91739f5b4c4b5b8294f9c10;hp=15bd5d25beecfdb4007ec35ae462b2744c6429b5;hb=1ce5968e27a19937a0e54f118d91e1a22438a0eb;hpb=d3372da97568d5e1f35fa19787c8ec8af93a0435 diff --git a/daemon/transfer.c b/daemon/transfer.c index 15bd5d2..9c50ed1 100644 --- a/daemon/transfer.c +++ b/daemon/transfer.c @@ -1,6 +1,6 @@ /* * Dolda Connect - Modular multiuser Direct Connect-style client - * Copyright (C) 2004 Fredrik Tolf (fredrik@dolda2000.com) + * Copyright (C) 2004 Fredrik Tolf * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by @@ -26,6 +26,7 @@ #include #include #include +#include #ifdef HAVE_CONFIG_H #include @@ -36,17 +37,18 @@ #include "auth.h" #include "transfer.h" #include "module.h" +#include "client.h" static void killfilter(struct transfer *transfer); +unsigned long long bytesupload = 0; +unsigned long long bytesdownload = 0; struct transfer *transfers = NULL; int numtransfers = 0; GCBCHAIN(newtransfercb, struct transfer *); void freetransfer(struct transfer *transfer) { - struct transarg *ta; - if(transfer == transfers) transfers = transfer->next; if(transfer->next != NULL) @@ -59,13 +61,8 @@ void freetransfer(struct transfer *transfer) CBCHAINFREE(transfer, trans_p); CBCHAINFREE(transfer, trans_destroy); CBCHAINFREE(transfer, trans_filterout); - while((ta = transfer->args) != NULL) - { - transfer->args = ta->next; - free(ta->rec); - free(ta->val); - free(ta); - } + while(transfer->args != NULL) + freewcspair(transfer->args, &transfer->args); if(transfer->filter != -1) killfilter(transfer); if(transfer->etimer != NULL) @@ -82,6 +79,10 @@ void freetransfer(struct transfer *transfer) free(transfer->actdesc); if(transfer->filterbuf != NULL) free(transfer->filterbuf); + if(transfer->hash != NULL) + freehash(transfer->hash); + if(transfer->exitstatus != NULL) + free(transfer->exitstatus); if(transfer->localend != NULL) { transfer->localend->readcb = NULL; @@ -125,36 +126,99 @@ struct transfer *newtransfer(void) return(new); } -void transferaddarg(struct transfer *transfer, wchar_t *rec, wchar_t *val) +static void localread(struct socket *sk, struct transfer *transfer) { - struct transarg *ta; + void *buf; + size_t blen; - ta = smalloc(sizeof(*ta)); - ta->rec = swcsdup(rec); - ta->val = swcsdup(val); - ta->next = transfer->args; - transfer->args = ta; + if((transfer->datapipe != NULL) && (sockqueueleft(transfer->datapipe) > 0)) { + buf = sockgetinbuf(sk, &blen); + sockqueue(transfer->datapipe, buf, blen); + } } -void transferattach(struct transfer *transfer, struct transferiface *iface, void *data) +static void dataread(struct socket *sk, struct transfer *transfer) { - if(transfer->iface != NULL) - transferdetach(transfer); - transfer->iface = iface; - transfer->ifacedata = data; + void *buf; + size_t blen; + + if((transfer->localend != NULL) && (sockqueueleft(transfer->localend) > 0)) { + buf = sockgetinbuf(sk, &blen); + sockqueue(transfer->localend, buf, blen); + } +} + +static void localwrite(struct socket *sk, struct transfer *transfer) +{ + if(transfer->datapipe != NULL) + dataread(transfer->datapipe, transfer); +} + +static void datawrite(struct socket *sk, struct transfer *transfer) +{ + if(transfer->localend != NULL) + localread(transfer->localend, transfer); +} + +static void localerr(struct socket *sk, int errno, struct transfer *transfer) +{ + if(transfer->datapipe != NULL) + closesock(transfer->datapipe); +} + +static void dataerr(struct socket *sk, int errno, struct transfer *transfer) +{ + if(transfer->localend != NULL) + closesock(transfer->localend); +} + +void transferattach(struct transfer *transfer, struct socket *dpipe) +{ + transferdetach(transfer); + getsock(transfer->datapipe = dpipe); + dpipe->readcb = (void (*)(struct socket *, void *))dataread; + dpipe->writecb = (void (*)(struct socket *, void *))datawrite; + dpipe->errcb = (void (*)(struct socket *, int, void *))dataerr; + dpipe->data = transfer; } void transferdetach(struct transfer *transfer) { - if(transfer->iface != NULL) + if(transfer->datapipe != NULL) { + transfer->datapipe->readcb = NULL; + transfer->datapipe->writecb = NULL; + transfer->datapipe->errcb = NULL; + closesock(transfer->datapipe); + putsock(transfer->datapipe); + } + transfer->datapipe = NULL; +} + +struct transfer *finddownload(wchar_t *peerid) +{ + struct transfer *transfer; + + for(transfer = transfers; transfer != NULL; transfer = transfer->next) { - transfer->iface->detach(transfer, transfer->ifacedata); - transfer->iface = NULL; - transfer->ifacedata = NULL; + if((transfer->dir == TRNSD_DOWN) && (transfer->datapipe == NULL) && !wcscmp(peerid, transfer->peerid)) + break; + } + return(transfer); +} + +struct transfer *hasupload(struct fnet *fnet, wchar_t *peerid) +{ + struct transfer *transfer; + + for(transfer = transfers; transfer != NULL; transfer = transfer->next) + { + if((transfer->dir == TRNSD_UP) && (transfer->fnet == fnet) && !wcscmp(transfer->peerid, peerid)) + break; } + return(transfer); } -struct transfer *newupload(struct fnetnode *fn, struct fnet *fnet, wchar_t *nickid, struct transferiface *iface, void *data) +struct transfer *newupload(struct fnetnode *fn, struct fnet *fnet, wchar_t *nickid, struct socket *dpipe) { struct transfer *transfer; @@ -168,7 +232,7 @@ struct transfer *newupload(struct fnetnode *fn, struct fnet *fnet, wchar_t *nick transfer->dir = TRNSD_UP; if(fn != NULL) getfnetnode(transfer->fn = fn); - transferattach(transfer, iface, data); + transferattach(transfer, dpipe); linktransfer(transfer); bumptransfer(transfer); return(transfer); @@ -188,8 +252,7 @@ void resettransfer(struct transfer *transfer) { if(transfer->dir == TRNSD_DOWN) { - if(transfer->iface != NULL) - transferdetach(transfer); + transferdetach(transfer); killfilter(transfer); transfersetstate(transfer, TRNS_WAITING); transfersetactivity(transfer, L"reset"); @@ -218,35 +281,16 @@ static void transexpire(int cancelled, struct transfer *transfer) transfer->timeout = 0; } -static void transferread(struct socket *sk, struct transfer *transfer) -{ - if(sockgetdatalen(sk) >= 65536) - sk->ignread = 1; - if((transfer->iface != NULL) && (transfer->iface->gotdata != NULL)) - transfer->iface->gotdata(transfer, transfer->ifacedata); -} - -static void transferwrite(struct socket *sk, struct transfer *transfer) -{ - if((transfer->iface != NULL) && (transfer->iface->wantdata != NULL)) - transfer->iface->wantdata(transfer, transfer->ifacedata); -} - -static void transfererr(struct socket *sk, int errno, struct transfer *transfer) -{ - if((transfer->iface != NULL) && (transfer->iface->endofdata != NULL)) - transfer->iface->endofdata(transfer, transfer->ifacedata); -} - -void transferputdata(struct transfer *transfer, void *buf, size_t size) +static void transferputdata(struct transfer *transfer, void *buf, size_t size) { time(&transfer->activity); sockqueue(transfer->localend, buf, size); transfer->curpos += size; + bytesdownload += size; CBCHAINDOCB(transfer, trans_p, transfer); } -void transferendofdata(struct transfer *transfer) +static void transferendofdata(struct transfer *transfer) { if(transfer->curpos >= transfer->size) { @@ -261,46 +305,55 @@ void transferendofdata(struct transfer *transfer) } } -size_t transferdatasize(struct transfer *transfer) +static ssize_t transferdatasize(struct transfer *transfer) { - return(sockqueuesize(transfer->localend)); + return(sockqueueleft(transfer->localend)); } -void *transfergetdata(struct transfer *transfer, size_t *size) +static void *transfergetdata(struct transfer *transfer, size_t *size) { void *buf; if(transfer->localend == NULL) return(NULL); - transfer->localend->ignread = 0; time(&transfer->activity); if((buf = sockgetinbuf(transfer->localend, size)) == NULL) return(NULL); if((transfer->endpos >= 0) && (transfer->curpos + *size >= transfer->endpos)) { - *size = transfer->endpos - transfer->curpos; - buf = srealloc(buf, *size); + if((*size = transfer->endpos - transfer->curpos) == 0) { + free(buf); + buf = NULL; + } else { + buf = srealloc(buf, *size); + } } transfer->curpos += *size; + bytesupload += *size; CBCHAINDOCB(transfer, trans_p, transfer); return(buf); } -void transferprepul(struct transfer *transfer, size_t size, size_t start, size_t end, struct socket *lesk) +void transferprepul(struct transfer *transfer, off_t size, off_t start, off_t end, struct socket *lesk) { transfersetsize(transfer, size); transfer->curpos = start; transfer->endpos = end; - lesk->ignread = 1; transfersetlocalend(transfer, lesk); } +void transferstartdl(struct transfer *transfer, struct socket *sk) +{ + transfersetstate(transfer, TRNS_MAIN); + socksettos(sk, confgetint("transfer", "dltos")); +} + void transferstartul(struct transfer *transfer, struct socket *sk) { transfersetstate(transfer, TRNS_MAIN); socksettos(sk, confgetint("transfer", "ultos")); if(transfer->localend != NULL) - transfer->localend->ignread = 0; + localread(transfer->localend, transfer); } void transfersetlocalend(struct transfer *transfer, struct socket *sk) @@ -309,19 +362,64 @@ void transfersetlocalend(struct transfer *transfer, struct socket *sk) putsock(transfer->localend); getsock(transfer->localend = sk); sk->data = transfer; - sk->readcb = (void (*)(struct socket *, void *))transferread; - sk->writecb = (void (*)(struct socket *, void *))transferwrite; - sk->errcb = (void (*)(struct socket *, int, void *))transfererr; + sk->readcb = (void (*)(struct socket *, void *))localread; + sk->writecb = (void (*)(struct socket *, void *))localwrite; + sk->errcb = (void (*)(struct socket *, int, void *))localerr; } -void bumptransfer(struct transfer *transfer) +static int tryreq(struct transfer *transfer) { struct fnetnode *fn; struct fnetpeer *peer; + + if((fn = transfer->fn) != NULL) + { + if(fn->state != FNN_EST) + { + transfer->close = 1; + return(1); + } + peer = fnetfindpeer(fn, transfer->peerid); + } else { + peer = NULL; + for(fn = fnetnodes; fn != NULL; fn = fn->next) + { + if((fn->state == FNN_EST) && (fn->fnet == transfer->fnet) && ((peer = fnetfindpeer(fn, transfer->peerid)) != NULL)) + break; + } + } + if(peer != NULL) + { + time(&transfer->lastreq); + return(fn->fnet->reqconn(peer)); + } + return(1); +} + +void trytransferbypeer(struct fnet *fnet, wchar_t *peerid) +{ + struct transfer *transfer; + + for(transfer = transfers; transfer != NULL; transfer = transfer->next) + { + if((transfer->dir == TRNSD_DOWN) && (transfer->state == TRNS_WAITING)) + { + if((transfer->fnet == fnet) && !wcscmp(transfer->peerid, peerid)) + { + if(!tryreq(transfer)) + return; + } + } + } +} + +void bumptransfer(struct transfer *transfer) +{ time_t now; if((now = time(NULL)) < transfer->timeout) { + if(transfer->etimer == NULL) transfer->etimer = timercallback(transfer->timeout, (void (*)(int, void *))transexpire, transfer); return; @@ -331,32 +429,9 @@ void bumptransfer(struct transfer *transfer) switch(transfer->state) { case TRNS_WAITING: - if(transfer->fn != NULL) - { - fn = transfer->fn; - if(fn->state != FNN_EST) - { - transfer->close = 1; - return; - } - peer = fnetfindpeer(fn, transfer->peerid); - } else { - peer = NULL; - for(fn = fnetnodes; fn != NULL; fn = fn->next) - { - if((fn->state == FNN_EST) && (fn->fnet == transfer->fnet) && ((peer = fnetfindpeer(fn, transfer->peerid)) != NULL)) - break; - } - } transfer->etimer = timercallback(transfer->timeout = (time(NULL) + 30), (void (*)(int, void *))transexpire, transfer); if(now - transfer->lastreq > 30) - { - if(peer != NULL) - { - fn->fnet->reqconn(peer); - time(&transfer->lastreq); - } - } + tryreq(transfer); break; case TRNS_HS: if(transfer->dir == TRNSD_UP) @@ -414,7 +489,7 @@ void transfersetnick(struct transfer *transfer, wchar_t *newnick) CBCHAINDOCB(transfer, trans_ac, transfer, L"nick"); } -void transfersetsize(struct transfer *transfer, int newsize) +void transfersetsize(struct transfer *transfer, off_t newsize) { transfer->size = newsize; CBCHAINDOCB(transfer, trans_ac, transfer, L"size"); @@ -434,6 +509,14 @@ void transfersetpath(struct transfer *transfer, wchar_t *path) CBCHAINDOCB(transfer, trans_ac, transfer, L"path"); } +void transfersethash(struct transfer *transfer, struct hash *hash) +{ + if(transfer->hash != NULL) + freehash(transfer->hash); + transfer->hash = hash; + CBCHAINDOCB(transfer, trans_ac, transfer, L"hash"); +} + int slotsleft(void) { struct transfer *transfer; @@ -477,56 +560,15 @@ static void killfilter(struct transfer *transfer) transfer->filterbufsize = transfer->filterbufdata = 0; } -static char *findfilter(struct passwd *pwd) +static void handletranscmd(struct transfer *transfer, wchar_t *cmd, wchar_t *arg) { - char *path, *filtername; - - if((path = sprintf2("%s/.dcdl-filter", pwd->pw_dir)) != NULL) - { - if(!access(path, X_OK)) - return(path); - free(path); - } - if((filtername = icwcstombs(confgetstr("transfer", "filter"), NULL)) == NULL) - { - flog(LOG_WARNING, "could not convert filter name into local charset: %s", strerror(errno)); - } else { - if(strchr(filtername, '/') == NULL) - { - if((path = sprintf2("/etc/%s", filtername)) != NULL) - { - if(!access(path, X_OK)) - { - free(filtername); - return(path); - } - free(path); - } - if((path = sprintf2("/usr/etc/%s", filtername)) != NULL) - { - if(!access(path, X_OK)) - { - free(filtername); - return(path); - } - free(path); - } - if((path = sprintf2("/usr/local/etc/%s", filtername)) != NULL) - { - if(!access(path, X_OK)) - { - free(filtername); - return(path); - } - free(path); - } - } else { - if(!access(filtername, X_OK)) - return(filtername); - } - free(filtername); + if(!wcscmp(cmd, L"status")) { + if(arg == NULL) + arg = L""; + if(transfer->exitstatus != NULL) + free(transfer->exitstatus); + transfer->exitstatus = swcsdup(arg); } - return(NULL); } static void filterread(struct socket *sk, struct transfer *transfer) @@ -539,7 +581,7 @@ static void filterread(struct socket *sk, struct transfer *transfer) return; bufcat(transfer->filterbuf, buf, bufsize); free(buf); - if((p = memchr(transfer->filterbuf, '\n', transfer->filterbufdata)) != NULL) + while((p = memchr(transfer->filterbuf, '\n', transfer->filterbufdata)) != NULL) { *(p++) = 0; if((p2 = strchr(transfer->filterbuf, ' ')) != NULL) @@ -550,8 +592,9 @@ static void filterread(struct socket *sk, struct transfer *transfer) if(p2 != NULL) { if((arg = icmbstowcs(p2, NULL)) == NULL) - flog(LOG_WARNING, "filter sent a string which could not be converted into the local charset: %s: %s", transfer->filterbuf, strerror(errno)); + flog(LOG_WARNING, "filter sent a string which could not be converted into the local charset: %s: %s", p2, strerror(errno)); } + handletranscmd(transfer, cmd, arg); CBCHAINDOCB(transfer, trans_filterout, transfer, cmd, arg); if(arg != NULL) free(arg); @@ -566,6 +609,8 @@ static void filterread(struct socket *sk, struct transfer *transfer) static void filterexit(pid_t pid, int status, void *data) { struct transfer *transfer; + struct fnet *fnet; + wchar_t *peerid; for(transfer = transfers; transfer != NULL; transfer = transfer->next) { @@ -573,12 +618,14 @@ static void filterexit(pid_t pid, int status, void *data) { transfer->filter = -1; killfilter(transfer); + fnet = transfer->fnet; + peerid = swcsdup(transfer->peerid); if(WEXITSTATUS(status)) - { resettransfer(transfer); - } else { + else freetransfer(transfer); - } + trytransferbypeer(fnet, peerid); + free(peerid); break; } } @@ -586,7 +633,7 @@ static void filterexit(pid_t pid, int status, void *data) int forkfilter(struct transfer *transfer) { - char *filtername, *filename, *peerid, *buf; + char *filtername, *filename, *peerid, *buf, *p; wchar_t *wfilename; struct passwd *pwent; pid_t pid; @@ -594,12 +641,10 @@ int forkfilter(struct transfer *transfer) char **argv; size_t argvsize, argvdata; struct socket *insock, *outsock; - struct transarg *ta; + struct wcspair *ta; char *rec, *val; - wfilename = transfer->path; - if(transfer->fnet->filebasename != NULL) - wfilename = transfer->fnet->filebasename(wfilename); + wfilename = fnfilebasename(transfer->path); if(transfer->auth == NULL) { flog(LOG_WARNING, "tried to fork filter for transfer with NULL authhandle (tranfer %i)", transfer->id); @@ -612,7 +657,10 @@ int forkfilter(struct transfer *transfer) errno = EACCES; return(-1); } - if((filtername = findfilter(pwent)) == NULL) + filtername = findfile("dc-filter", pwent->pw_dir, 0); + if(filtername == NULL) + filtername = findfile(icswcstombs(confgetstr("transfer", "filter"), NULL, NULL), NULL, 0); + if(filtername == NULL) { flog(LOG_WARNING, "could not find filter for user %s", pwent->pw_name); errno = ENOENT; @@ -639,6 +687,12 @@ int forkfilter(struct transfer *transfer) peerid = sprintf2("utf8-%s", buf); free(buf); } + for(p = filename; *p; p++) { + if(*p == '/') + *p = '_'; + else if((p == filename) && (*p == '.')) + *p = '_'; + } if((pid = forksess(transfer->owner, transfer->auth, filterexit, NULL, FD_PIPE, 0, O_WRONLY, &inpipe, FD_PIPE, 1, O_RDONLY, &outpipe, FD_FILE, 2, O_RDWR, "/dev/null", FD_END)) < 0) { flog(LOG_WARNING, "could not fork session for filter for transfer %i: %s", transfer->id, strerror(errno)); @@ -648,14 +702,27 @@ int forkfilter(struct transfer *transfer) { argv = NULL; argvsize = argvdata = 0; - buf = sprintf2("%i", transfer->size); + buf = sprintf2("%ji", (intmax_t)transfer->size); addtobuf(argv, filtername); addtobuf(argv, filename); addtobuf(argv, buf); addtobuf(argv, peerid); + if(transfer->hash) + { + if((buf = icwcstombs(unparsehash(transfer->hash), NULL)) != NULL) + { + /* XXX: I am very doubtful of this, but it can just as + * well be argued that all data should be presented as + * key-value pairs. */ + addtobuf(argv, "hash"); + addtobuf(argv, buf); + } else { + flog(LOG_WARNING, "could not convert hash to local charset"); + } + } for(ta = transfer->args; ta != NULL; ta = ta->next) { - if((rec = icwcstombs(ta->rec, NULL)) == NULL) + if((rec = icwcstombs(ta->key, NULL)) == NULL) continue; if((val = icwcstombs(ta->val, NULL)) == NULL) continue; @@ -676,7 +743,7 @@ int forkfilter(struct transfer *transfer) * the fd, and thus it closes it. Until I can find out whyever the * kernel gives a POLLIN on the fd (if I can at all...), I'll just * set ignread on insock for now. */ - insock->ignread = 1; +/* sockblock(insock, 1); */ transfer->filter = pid; transfersetlocalend(transfer, insock); getsock(transfer->filterout = outsock); @@ -694,6 +761,7 @@ static int run(void) { struct transfer *transfer, *next; + /* for(transfer = transfers; transfer != NULL; transfer = transfer->next) { if((transfer->endpos >= 0) && (transfer->state == TRNS_MAIN) && (transfer->localend != NULL) && (transfer->localend->state == SOCK_EST) && (transfer->curpos >= transfer->endpos)) @@ -703,6 +771,7 @@ static int run(void) closesock(transfer->localend); } } + */ for(transfer = transfers; transfer != NULL; transfer = next) { next = transfer->next; @@ -718,10 +787,23 @@ static int run(void) static struct configvar myvars[] = { + /** The maximum number of simultaneously permitted uploads. A + * common hub rule is that you will need at least as many slots as + * the number of hubs to which you are connected. */ {CONF_VAR_INT, "slots", {.num = 3}}, + /** The TOS value to use for upload connections (see the TOS + * VALUES section). */ {CONF_VAR_INT, "ultos", {.num = SOCK_TOS_MAXTP}}, + /** The TOS value to use for download connections (see the TOS + * VALUES section). */ {CONF_VAR_INT, "dltos", {.num = SOCK_TOS_MAXTP}}, + /** The name of the filter script (see the FILES section for + * lookup information). */ {CONF_VAR_STRING, "filter", {.str = L"dc-filter"}}, + /** If true, only one upload is allowed per remote peer. This + * option is still experimental, so it is recommended to leave it + * off. */ + {CONF_VAR_BOOL, "ulquota", {.num = 0}}, {CONF_VAR_END} };