From 7cdfd87aa6184c02e8284b7a2771cd16e4f86516 Mon Sep 17 00:00:00 2001 From: Vladimir Dubrovin <3proxy@3proxy.ru> Date: Sun, 18 Oct 2026 14:57:35 +0300 Subject: [PATCH] UDP services in one thread: udppm and dnspr without client threads A UDP service now runs in its own thread alone (udploop()): it reads the service's socket and the sockets of its sessions towards their destinations, keeps their deadlines, and answers every client from the service's socket. A session is a struct clientparam as a client thread's would be, with a slot of maxchild, among the service's children and counted on its listener, but without a thread. The services supply handlers (struct udpservice, childdef.udp): udppm keeps a session per client address, forwarding its datagrams, behind the headers of a parent chain, from a socket of its own and answering what comes back, until STRING_L of silence or, with -s, the first answer; dnspr makes a session of each query, answered at once from the fake address, the name cache or the resolver, or sent on to the first name server over UDP, whose answer is waited for in the loop up to DNS_TO (a name server over TCP is still waited for in place). Authentication stays in the loop for now; datagrams over a bandwidth limit, or of a client without a free slot, are dropped. On a reload the old service goes on serving the socket until it sees its configuration replaced, then hands its sessions to the listener; the newest service of its kind there, which until then does not read the socket so that no client gets two sessions, adopts them: a slot of its own, the old one given back, its filters and limits, and re-authentication. A session is ended with 26 if the newest service is of another kind, 27 without a slot, or with what the authentication says; sessions nobody is left to adopt are ended by the last service to stop on the listener. On Windows a send on a socket waited while another thread polled it, which delayed udppm's answers by up to a second, and old UDP services did not exit after a reload; with one thread per socket neither happens, and the second socket bound to the service's address (NO_SHARE_UDP_SOCKET) is gone from the code and the build files. udpinit, the global udp_table, udpnext(), udppmchild() and dnsprchild() are gone; udpsockmap() serves SOCKS5 UDP ASSOCIATE only. Co-Authored-By: Claude Opus 5.5 --- CMakeLists.txt | 3 - Makefile.msvc | 2 +- Makefile.watcom | 2 +- Makefile.win | 3 - src/conf.c | 10 +- src/dnspr.c | 386 ++++++++++++++++++++------------------- src/plugins.c | 4 +- src/proxy.h | 61 +++++-- src/proxymain.c | 467 ++++++++++++++++++++++++++++++++++++++++------- src/structures.h | 6 +- src/udppm.c | 138 ++++++++------ src/udpsockmap.c | 6 +- 12 files changed, 746 insertions(+), 342 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 1fa42a1..861d19a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -120,9 +120,6 @@ set(CMAKE_POSITION_INDEPENDENT_CODE ON) if(WIN32) # Windows-specific configuration add_compile_definitions( - # Windows does not take a recv and a send on one socket from two - # threads, so a UDP service answers from a socket of its own - NO_SHARE_UDP_SOCKET WIN32 _WIN32 _MBCS diff --git a/Makefile.msvc b/Makefile.msvc index 0e4bbb5..a948e0c 100644 --- a/Makefile.msvc +++ b/Makefile.msvc @@ -18,7 +18,7 @@ SSL_LIBS = wolfssl.lib SSL_DEFS = /D "WITH_SSL" SSL_LIBS = libcrypto.lib libssl.lib !ENDIF -CFLAGS = /D "WITH_HTTPSRV" /nologo /MT /W3 /Ox /GS /EHs- /GA /GF /D "MSVC" /D "WITH_WSAPOLL" /D "NDEBUG" /D "WIN32" $(SSL_DEFS) /D "WITH_PCRE" /D "WITH_ODBC" /D "_CONSOLE" /D "_MBCS" /D "_WIN32" /D "NO_SHARE_UDP_SOCKET" /Fp"proxy.pch" /FD /c $(BUILDDATE) $(VERSION) +CFLAGS = /D "WITH_HTTPSRV" /nologo /MT /W3 /Ox /GS /EHs- /GA /GF /D "MSVC" /D "WITH_WSAPOLL" /D "NDEBUG" /D "WIN32" $(SSL_DEFS) /D "WITH_PCRE" /D "WITH_ODBC" /D "_CONSOLE" /D "_MBCS" /D "_WIN32" /Fp"proxy.pch" /FD /c $(BUILDDATE) $(VERSION) COUT = /Fo LN = link LDFLAGS = /nologo /subsystem:console /incremental:no diff --git a/Makefile.watcom b/Makefile.watcom index 00df629..dbf0752 100644 --- a/Makefile.watcom +++ b/Makefile.watcom @@ -8,7 +8,7 @@ BUILDDIR = ../bin/ PREFIX = 3proxy_ CRYPT_PREFIX = 3proxy_ CC = cl -CFLAGS = /D "WITH_HTTPSRV" /nologo /Ox /MT /D "NOIPV6" /D "NO_UN" /D "NODEBUG" /D "NORADIUS" /D"WATCOM" /D "NO_SHARE_UDP_SOCKET" /D "MSVC" /D "WIN32" /D "_CONSOLE" /D "_MBCS" /D "_WIN32" /D "PRId64=\"I64d\"" /D "PRIu64=\"I64u\"" /D "SCNu64=\"I64u\"" /D "SCNx64=\"I64x\"" /D "SCNd64=\"I64d\"" /D "PRIx64=\"I64x\"" /c $(VERSION) $(BUILDDATE) +CFLAGS = /D "WITH_HTTPSRV" /nologo /Ox /MT /D "NOIPV6" /D "NO_UN" /D "NODEBUG" /D "NORADIUS" /D"WATCOM" /D "MSVC" /D "WIN32" /D "_CONSOLE" /D "_MBCS" /D "_WIN32" /D "PRId64=\"I64d\"" /D "PRIu64=\"I64u\"" /D "SCNu64=\"I64u\"" /D "SCNx64=\"I64x\"" /D "SCNd64=\"I64d\"" /D "PRIx64=\"I64x\"" /c $(VERSION) $(BUILDDATE) COUT = /Fo LN = link LDFLAGS = /nologo /subsystem:console /incremental:no diff --git a/Makefile.win b/Makefile.win index 77304bf..7ad9197 100644 --- a/Makefile.win +++ b/Makefile.win @@ -11,9 +11,6 @@ CRYPT_PREFIX ?= $(PREFIX) CC ?= gcc CFLAGS ?= -O3 -flto CFLAGS += -fno-strict-aliasing -c -mthreads -DWITH_WSAPOLL -DWITH_ODBC -# Windows does not take a recv and a send on one socket from two threads, -# so a UDP service answers from a socket of its own -CFLAGS += -DNO_SHARE_UDP_SOCKET COUT = -o LN ?= $(CC) LDFLAGS ?= -O3 -flto diff --git a/src/conf.c b/src/conf.c index ea476c8..7b63dd1 100644 --- a/src/conf.c +++ b/src/conf.c @@ -256,6 +256,8 @@ static int h_proxy(int argc, unsigned char ** argv){ ch.argc = argc; ch.argv = argv; + /* set below for a UDP service only */ + childdef.udp = NULL; if(!strcmp((char *)argv[0], "proxy")) { childdef.pf = proxychild; childdef.port = 3128; @@ -331,7 +333,8 @@ static int h_proxy(int argc, unsigned char ** argv){ childdef.helpmessage = ""; } else if(!strcmp((char *)argv[0], "udppm")) { - childdef.pf = udppmchild; + childdef.pf = NULL; + childdef.udp = &udppmservice; childdef.port = 0; childdef.isudp = 1; childdef.service = S_UDPPM; @@ -364,7 +367,8 @@ static int h_proxy(int argc, unsigned char ** argv){ } #endif else if(!strcmp((char *)argv[0], "dnspr")) { - childdef.pf = dnsprchild; + childdef.pf = NULL; + childdef.udp = &dnsprservice; childdef.port = 53; childdef.isudp = 1; childdef.service = S_DNSPR; @@ -883,7 +887,7 @@ struct redirdesc redirs[] = { {R_INTPORT, "intport", NULL}, {R_TLS, "tls", tlsprchild}, {R_HA, "ha", NULL}, - {R_DNS, "dns", dnsprchild}, + {R_DNS, "dns", NULL}, {0, NULL, NULL} }; diff --git a/src/dnspr.c b/src/dnspr.c index 64693ce..44b1ec5 100644 --- a/src/dnspr.c +++ b/src/dnspr.c @@ -11,218 +11,238 @@ #ifndef UDP #define UDP #endif -#define RETURN(xxx) { param->res = xxx; goto CLEANRET; } #define BUFSIZE 16384 +/* dnspr: a query is a session of its own in the service's thread (see + udploop() in proxymain.c). It is answered there and then, from the name + cache, the resolver or a fake address, or sent on to the first name + server, whose answer, or its timeout, ends it. A name server over TCP is + still waited for in the loop. */ -void * dnsprchild(struct clientparam* param) { - uint32_t ip = 0; - unsigned char *bbuf; - unsigned char *buf, *s1, *s2; - char * host = NULL; - unsigned char c; - SASIZETYPE size; - int res, i; - int len; - int semlocked = 1; - unsigned type=0; - uint32_t ttl; - unsigned char addr[16]; -#ifdef _WIN32 - unsigned long ul = 1; -#endif +struct dnsquery { + unsigned type; + uint32_t ip, ttl; + unsigned char addr[16]; + char *host; + unsigned char *query; /* as the client sent it, for a name server */ + int len; +}; +/* The answer to the client, the query in buf made one: ip set, an address + for it; a PTR for the service's own address; else the name not found. */ +static int dnsanswer(struct udploop *u, struct clientparam *param, struct dnsquery *q, unsigned char *buf, int len){ + uint32_t ttl; - if(!(bbuf = malloc(BUFSIZE+2))){ - RETURN (21); - } - buf = bbuf+2; - size = sizeof(param->sinsl); - getsockname(param->srv->srvsock, (struct sockaddr *)¶m->sincl, &size); - i = param->srv->udplen; - if(i > BUFSIZE) i = BUFSIZE; - memcpy(buf, param->srv->udpbuf, i); - _3proxy_sem_unlock(udpinit); - semlocked = 0; -#if defined(_WIN32) && !defined(SHARE_UDP_SOCKET) - if((param->clisock=nssocket(param->srv, ¶m->srv->so, param->sostate, NS_IN, AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == INVALID_SOCKET) { - RETURN(818); - } - ioctlsocket(param->clisock, FIONBIO, &ul); - if(param->srv->so._setsockopt(param->sostate, param->clisock, SOL_SOCKET, SO_REUSEADDR, (char *)&ul, sizeof(int))) {RETURN(820);}; - if(param->srv->so._bind(param->sostate, param->clisock,(struct sockaddr *)¶m->sincl,SASIZE(¶m->sincl))) { - RETURN(822); - } - -#else - /* The reply has to come from the address the query was sent to, which - is the listening socket. */ - param->clisock = param->srv->srvsock; -#endif - - if(i < 0) { - RETURN(813); - } - buf[BUFSIZE - 1] = 0; - if(i<=13 || i>1000){ - RETURN (814); - } - param->operation = DNSRESOLVE; - if((res = (*param->srv->authfunc)(param))) {RETURN(res);} - - if(buf[4]!=0 || buf[5]!=1) RETURN(816); - for(len = 12; len (i-4)) {RETURN(817);} - - host = strdup((char *)buf+13); - if(!host) {RETURN(21);} - - for(s2 = buf + 12; (s1 = (unsigned char *)strchr((char *)s2 + 1, '.')); s2 = s1)*s2 = (unsigned char)((s1 - s2) - 1); - *s2 = (len - (int)(s2 - buf)) - 1; - - type = ((unsigned)buf[len+1])*256 + (unsigned)buf[len+2]; - if(type==0x01 && param->srv->fakeip){ - ip = 1; - ttl = 3600; - *(uint32_t *)addr = param->srv->fakeip; - } -#ifndef NOIPV6 - else if(type==0x1c && *(uint32_t *)param->srv->fakeip6){ - ip = 1; - ttl = 3600; - memcpy(addr, param->srv->fakeip6, 16); - } -#endif - else if((type==0x01 || type==0x1c) && !param->srv->s_option){ - ip = udpresolve((type==0x1c)?AF_INET6:AF_INET, (unsigned char *)host, addr, &ttl, param, 0); - } - - len+=5; - - if(ip){ - buf[2] = 0x85; - buf[3] = 0x80; - buf[6] = 0; - buf[7] = 1; - buf[8] = buf[9] = buf[10] = buf[11] = 0; - memset(buf+len, 0, 16); - buf[len] = 0xc0; - buf[len+1] = 0x0c; - buf[len+3] = type; - buf[len+5] = 1; - ttl = htonl(ttl); - memcpy(buf + len + 6, &ttl, 4); - buf[len+11] = type==1? 4:16; - memcpy(buf+len+12,(void *)&addr,type==1? 4:16); - len+=(type==1?16:28); - } - else if(type == 0x0c) { - unsigned a = 0, b = 0, c = 0, d = 0; - if(sscanf(host, "%u.%u.%u.%u", &a, &b, &c, &d) != 4) a = b = c = d = 0; - ip = htonl((d<<24) ^ (c<<16) ^ (b<<8) ^ a); - if(*SAFAMILY(¶m->sincl) == AF_INET && ip == *(uint32_t *)SAADDR(¶m->sincl)){ + if(q->ip){ buf[2] = 0x85; buf[3] = 0x80; buf[6] = 0; buf[7] = 1; buf[8] = buf[9] = buf[10] = buf[11] = 0; - memset(buf+len, 0, 20); + memset(buf+len, 0, 16); buf[len] = 0xc0; buf[len+1] = 0x0c; - buf[len+3] = 0x0c; + buf[len+3] = q->type; buf[len+5] = 1; - ttl = htonl(3600); + ttl = htonl(q->ttl); memcpy(buf + len + 6, &ttl, 4); - buf[len+11] = 7; - buf[len+12] = 6; - memcpy(buf+len+13,(void *)"3proxy",6); - len+=20; - } - else ip = 0; - } - if(!ip && param->srv->config->numservers){ - if((param->remsock=param->srv->so._socket(param->sostate, SASOCK(¶m->srv->config->nservers[0].addr), param->srv->config->nservers[0].usetcp? SOCK_STREAM:SOCK_DGRAM, param->srv->config->nservers[0].usetcp?IPPROTO_TCP:IPPROTO_UDP)) == INVALID_SOCKET) { - RETURN(818); - } - memset(¶m->sinsl, 0, sizeof(param->sinsl)); - *SAFAMILY(¶m->sinsl) = *SAFAMILY(¶m->srv->config->nservers[0].addr); - if(bindwithrange(param, param->remsock, ¶m->sinsl, param->extport)) { - RETURN(819); - } - param->sinsr = param->srv->config->nservers[0].addr; - if(param->srv->config->nservers[0].usetcp) { - if(connectwithpoll(param, param->remsock,(struct sockaddr *)¶m->sinsr,SASIZE(¶m->sinsr),param->srv->timeouts[CONNECT_TO])) RETURN(830); - buf-=2; - *(unsigned short*)buf = htons(i); - i+=2; + buf[len+11] = q->type==1? 4:16; + memcpy(buf+len+12, q->addr, q->type==1? 4:16); + len += (q->type==1? 16:28); } else { + buf[2] = 0x85; + buf[3] = 0x83; + } + if(udpanswer(u, param, buf, len) != len) return 819; + return q->ip? 0 : 888; +} + +static int dnsprfirst(struct udploop *u, struct udpsession *ss, unsigned char *query, int i){ + struct clientparam *param = ss->param; + struct dnsquery *q; + unsigned char *buf, *s1, *s2; + unsigned char c; + int res, len; + uint32_t ttl, ip = 0; #ifdef _WIN32 - ioctlsocket(param->remsock, FIONBIO, &ul); -#else - fcntl(param->remsock, F_SETFL, O_NONBLOCK | fcntl(param->remsock, F_GETFL)); + unsigned long ul = 1; #endif + + if(!(q = calloc(1, sizeof(*q))) || !(q->query = malloc(BUFSIZE + 2))){ + free(q); + return (param->res = 21); + } + ss->data = q; + if(i > BUFSIZE) i = BUFSIZE; + memcpy(q->query + 2, query, i); + q->len = i; + buf = q->query + 2; + param->sincl = param->srv->intsa; + if(i<=13 || i>1000) return (param->res = 814); + buf[BUFSIZE - 1] = 0; + param->operation = DNSRESOLVE; + if((res = (*param->srv->authfunc)(param))) return (param->res = res); + + if(buf[4]!=0 || buf[5]!=1) return (param->res = 816); + for(len = 12; len (i-4)) return (param->res = 817); + if(!(q->host = strdup((char *)buf+13))) return (param->res = 21); + for(s2 = buf + 12; (s1 = (unsigned char *)strchr((char *)s2 + 1, '.')); s2 = s1)*s2 = (unsigned char)((s1 - s2) - 1); + *s2 = (len - (int)(s2 - buf)) - 1; + + q->type = ((unsigned)buf[len+1])*256 + (unsigned)buf[len+2]; + q->ttl = 3600; + if(q->type==0x01 && param->srv->fakeip){ + ip = 1; + *(uint32_t *)q->addr = param->srv->fakeip; + } +#ifndef NOIPV6 + else if(q->type==0x1c && *(uint32_t *)param->srv->fakeip6){ + ip = 1; + memcpy(q->addr, param->srv->fakeip6, 16); + } +#endif + else if((q->type==0x01 || q->type==0x1c) && !param->srv->s_option){ + ip = udpresolve((q->type==0x1c)?AF_INET6:AF_INET, (unsigned char *)q->host, q->addr, &q->ttl, param, 0); + } + len += 5; + + if(!ip && q->type == 0x0c) { + unsigned a = 0, b = 0, cc = 0, d = 0; + uint32_t ptr; + + if(sscanf(q->host, "%u.%u.%u.%u", &a, &b, &cc, &d) != 4) a = b = cc = d = 0; + ptr = htonl((d<<24) ^ (cc<<16) ^ (b<<8) ^ a); + if(*SAFAMILY(¶m->sincl) == AF_INET && ptr == *(uint32_t *)SAADDR(¶m->sincl)){ + buf[2] = 0x85; + buf[3] = 0x80; + buf[6] = 0; + buf[7] = 1; + buf[8] = buf[9] = buf[10] = buf[11] = 0; + memset(buf+len, 0, 20); + buf[len] = 0xc0; + buf[len+1] = 0x0c; + buf[len+3] = 0x0c; + buf[len+5] = 1; + ttl = htonl(3600); + memcpy(buf + len + 6, &ttl, 4); + buf[len+11] = 7; + buf[len+12] = 6; + memcpy(buf+len+13,(void *)"3proxy",6); + len += 20; + q->ip = 1; + if(udpanswer(u, param, buf, len) != len) return (param->res = 819); + return 1; + } + } + q->ip = ip; + if(ip || !param->srv->config->numservers){ + param->res = dnsanswer(u, param, q, buf, len); + return 1; } - if(socksendto(param, param->remsock, (struct sockaddr *)¶m->sinsr, buf, i, param->srv->timeouts[SINGLEBYTE_L]*1000) != i){ - RETURN(820); - } - param->statscli64 += i; - param->nwrites++; - len = sockrecvfrom(param, param->remsock, (struct sockaddr *)¶m->sinsr, buf, BUFSIZE, param->srv->timeouts[DNS_TO]*1000); - if(len <= 13) { - RETURN(821); - } - param->statssrv64 += len; - param->nreads++; + /* on to the first name server */ + if((param->remsock=param->srv->so._socket(param->sostate, SASOCK(¶m->srv->config->nservers[0].addr), param->srv->config->nservers[0].usetcp? SOCK_STREAM:SOCK_DGRAM, param->srv->config->nservers[0].usetcp?IPPROTO_TCP:IPPROTO_UDP)) == INVALID_SOCKET) + return (param->res = 818); + memset(¶m->sinsl, 0, sizeof(param->sinsl)); + *SAFAMILY(¶m->sinsl) = *SAFAMILY(¶m->srv->config->nservers[0].addr); + if(bindwithrange(param, param->remsock, ¶m->sinsl, param->extport)) return (param->res = 819); + param->sinsr = param->srv->config->nservers[0].addr; if(param->srv->config->nservers[0].usetcp) { unsigned short us; + + if(connectwithpoll(param, param->remsock,(struct sockaddr *)¶m->sinsr,SASIZE(¶m->sinsr),param->srv->timeouts[CONNECT_TO])) return (param->res = 830); + buf -= 2; + *(unsigned short*)buf = htons(i); + i += 2; + if(socksend(param, param->remsock, buf, i, param->srv->timeouts[SINGLEBYTE_L]) != i) return (param->res = 820); + param->statscli64 += i; + param->nwrites++; + len = sockrecvfrom(param, param->remsock, NULL, buf, BUFSIZE, param->srv->timeouts[DNS_TO]*1000); + if(len <= 13) return (param->res = 821); + param->statssrv64 += len; + param->nreads++; us = ntohs(*(unsigned short *)buf); - if(us > 4096) RETURN(833); + if(us > 4096) return (param->res = 833); buf += 2; len -= 2; if(len < us) len += sockgetlinebuf(param, SERVER, buf+len, us - len, 0, param->srv->timeouts[SINGLEBYTE_L]); - if(len != us) RETURN(832); - } - if(buf[6] || buf[7]){ - if(socksendto(param, param->clisock, (struct sockaddr *)¶m->sincr, buf, len, param->srv->timeouts[SINGLEBYTE_L]*1000) != len){ - RETURN(822); + if(len != us) return (param->res = 832); + if(buf[6] || buf[7]){ + if(udpanswer(u, param, buf, len) != len) return (param->res = 822); + return 1; } - RETURN(0); + param->res = dnsanswer(u, param, q, buf, len); + return 1; } - - } - if(!ip) { - buf[2] = 0x85; - buf[3] = 0x83; - } - res = socksendto(param, param->clisock, (struct sockaddr *)¶m->sincr, buf, len, param->srv->timeouts[SINGLEBYTE_L]*1000); - if(res != len){RETURN(819);} - if(!ip) {RETURN(888);} - -CLEANRET: - - if(semlocked) _3proxy_sem_unlock(udpinit); - if(param->res!=813){ - sprintf((char *)buf, "%04x/%s/", - (unsigned)type, - host?host:""); - if((ip && type == 0x01) || type == 0x1c){ - myinet_ntop(type == 0x01? AF_INET:AF_INET6, addr, (char *)buf+strlen((char *)buf), 64); - } - dolog(param, buf); - } - if(bbuf)free(bbuf); - if(host)free(host); -#if !defined(_WIN32) || defined(SHARE_UDP_SOCKET) - /* The socket belongs to the service, so the caller must not close it. */ - param->clisock = INVALID_SOCKET; +#ifdef _WIN32 + ioctlsocket(param->remsock, FIONBIO, &ul); +#else + fcntl(param->remsock, F_SETFL, O_NONBLOCK | fcntl(param->remsock, F_GETFL)); #endif - return (NULL); + if(param->srv->so._sendto(param->sostate, param->remsock, (char *)buf, i, 0, (struct sockaddr *)¶m->sinsr, SASIZE(¶m->sinsr)) != i) + return (param->res = 820); + param->statscli64 += i; + param->nwrites++; + /* its answer is waited for in the loop, as long as DNS_TO */ + ss->idle = 0; + ss->deadline = time(NULL) + param->srv->timeouts[DNS_TO]; + return 0; } +/* The name server's answer: to the client as it is if it has answers, + else as the name not found. */ +static int dnsprserver(struct udploop *u, struct udpsession *ss){ + struct clientparam *param = ss->param; + struct dnsquery *q = ss->data; + int len; + + len = param->srv->so._recvfrom(param->sostate, param->remsock, (char *)u->buf, UDPBUFSIZE, 0, NULL, NULL); + if(len < 0 && (errno == EAGAIN || errno == EINTR)) return 0; + if(len <= 13) return (param->res = 821); + param->statssrv64 += len; + param->nreads++; + if(u->buf[6] || u->buf[7]){ + if(udpanswer(u, param, u->buf, len) != len) return (param->res = 822); + param->res = 0; + return 1; + } + q->ip = 0; + param->res = dnsanswer(u, param, q, u->buf, len); + return 1; +} + +static void dnsprend(struct udploop *u, struct udpsession *ss){ + struct clientparam *param = ss->param; + struct dnsquery *q = ss->data; + char line[256]; + + if(param->res != 813){ + sprintf(line, "%04x/%.128s/", q? q->type : 0, q && q->host? q->host : ""); + if(q && ((q->ip && q->type == 0x01) || q->type == 0x1c)){ + myinet_ntop(q->type == 0x01? AF_INET:AF_INET6, q->addr, line+strlen(line), 64); + } + dolog(param, (unsigned char *)line); + } + if(q){ + free(q->host); + free(q->query); + free(q); + } + ss->data = NULL; +} + +const struct udpservice dnsprservice = { + 0, + dnsprfirst, + NULL, + dnsprserver, + dnsprend, + 821 +}; diff --git a/src/plugins.c b/src/plugins.c index 13678cd..f6ee878 100644 --- a/src/plugins.c +++ b/src/plugins.c @@ -55,12 +55,12 @@ struct symbol symbols[] = { {symbols+28, "de64", (void *) de64}, {symbols+29, "tohex", (void *) tohex}, {symbols+30, "fromhex", (void *) fromhex}, - {symbols+31, "dnspr", (void *) dnsprchild}, + {symbols+31, "dnspr", NULL}, {symbols+32, "pop3p", (void *) pop3pchild}, {symbols+33, "proxy", (void *) proxychild}, {symbols+34, "socks", (void *) sockschild}, {symbols+35, "tcppm", (void *) tcppmchild}, - {symbols+36, "udppm", (void *) udppmchild}, + {symbols+36, "udppm", NULL}, {symbols+38, "ftppr", (void *) ftpprchild}, {symbols+39, "smtpp", (void *) smtppchild}, {symbols+40, "auto", (void *) smtppchild}, diff --git a/src/proxy.h b/src/proxy.h index 66ca224..fac3859 100644 --- a/src/proxy.h +++ b/src/proxy.h @@ -146,20 +146,6 @@ void daemonize(void); #endif #endif -/* A UDP service answers from the address the client sent to, so the reply has - to leave the listening socket. Sharing that socket with the request handler - is the simple way, and what every Unix build does. - - Older Windows cannot have two operations in flight on one socket, so a - build for it binds a second socket to the same address instead, which needs - SO_REUSEADDR on the listening socket as well. A build for those versions - asks for that with NO_SHARE_UDP_SOCKET, as Makefile.watcom does. - */ -#ifndef SHARE_UDP_SOCKET -#ifndef NO_SHARE_UDP_SOCKET -#define SHARE_UDP_SOCKET -#endif -#endif #ifndef _WIN32 size_t threadstacksize(int extra); @@ -250,6 +236,50 @@ extern struct extparam confboot; int sockmap(struct clientparam * param, int timeo, int usesplice); int udpsockmap(struct clientparam * param, int timeo); int udpbind(struct clientparam * param); +int socks5_udp_skip_hdr(unsigned char *buf, int len); + +/* A UDP service runs in one thread, the service's: it reads the service's + socket, from which every answer to a client leaves too, and the sockets + of its sessions towards their destinations, and keeps the time. A + session is a client of the service, a struct clientparam, without a + thread. */ +struct udploop; +struct udpsession { + struct udpsession *next, *prev; + struct clientparam *param; + const struct udpservice *svc; + time_t deadline; /* when it ends unless refreshed */ + int idle; /* seconds of silence it lasts, 0: deadline fixed */ + int keyed; /* found by its client's address */ + int done; /* to be ended, param->res saying how */ + void *data; /* the service's own, freed by it in end() */ +}; +struct udpservice { + /* datagrams from a client are found its session by the client's + address (udppm), or each starts one of its own (dnspr) */ + int keyed; + /* a session's first datagram, in buf, from param->sincr: 0 to keep the + session, nonzero to end it, param->res saying how */ + int (*first)(struct udploop *u, struct udpsession *ss, unsigned char *buf, int len); + /* a later datagram of its client (keyed services) */ + int (*client)(struct udploop *u, struct udpsession *ss, unsigned char *buf, int len); + /* its socket towards the destination is readable */ + int (*server)(struct udploop *u, struct udpsession *ss); + /* it ends: logged, its own data freed */ + void (*end)(struct udploop *u, struct udpsession *ss); + /* param->res of a session whose deadline passed */ + int timeoutres; +}; +struct udploop { + struct srvparam *srv; + struct clientparam *defparam; + /* scratch of UDPBUFSIZE for a handler */ + unsigned char *buf; +}; +/* an answer to the session's client, from the service's socket */ +int udpanswer(struct udploop *u, struct clientparam *param, unsigned char *buf, int len); + +extern const struct udpservice udppmservice, dnsprservice; int socks5_setaddr(int family, int atyp, const unsigned char *addr, PROXYSOCKADDRTYPE *sa); /* which of a service's network namespaces a socket is made in */ #define NS_IN 0 /* -ni: what faces the clients */ @@ -337,7 +367,6 @@ unsigned char* en64 (const unsigned char *in, unsigned char *out, int inlen, int void tohex(unsigned char *in, unsigned char *out, int len); void fromhex(unsigned char *in, unsigned char *out, int len); -extern _3proxy_sem_t udpinit; #ifdef _WIN32 #define _3proxy_sem_init(x, count, maxcount) (((x) = CreateSemaphore(NULL, (count), (maxcount), NULL))? 0 : 1) #define _3proxy_sem_lock(x) WaitForSingleObject(x, INFINITE) @@ -460,7 +489,6 @@ extern char *copyright; #define SERVICES 5 -void * dnsprchild(struct clientparam * param); void * pop3pchild(struct clientparam * param); void * imappchild(struct clientparam * param); void * smtppchild(struct clientparam * param); @@ -468,7 +496,6 @@ void * proxychild(struct clientparam * param); void * sockschild(struct clientparam * param); void * tcppmchild(struct clientparam * param); void * autochild(struct clientparam * param); -void * udppmchild(struct clientparam * param); #ifdef WITH_HTTPSRV int op_admin(struct httpreq *r, const unsigned char *params); int op_admin_counters(struct httpreq *r, const unsigned char *params); diff --git a/src/proxymain.c b/src/proxymain.c index 3c0090c..e991a4b 100644 --- a/src/proxymain.c +++ b/src/proxymain.c @@ -35,9 +35,6 @@ static void serveparam(struct clientparam *param){ int i = -1; #ifdef MODULEMAINFUNC if(makefilters(param->srv, param) > CONTINUE){ -#ifndef NOUDPMAIN - if(param->srv->service == S_UDPPM) _3proxy_sem_unlock(udpinit); -#endif releaseparam(param); return; } @@ -97,7 +94,7 @@ static void serveparam(struct clientparam *param){ #ifndef _WIN32 sigset_t mask; sigfillset(&mask); - if(param->srv->service != S_UDPPM)pthread_sigmask(SIG_SETMASK, &mask, NULL); + pthread_sigmask(SIG_SETMASK, &mask, NULL); #endif #endif @@ -378,6 +375,7 @@ struct srvstart { unsigned char *cbc_string, *cbl_string; FILE *fp; int isudp, iscbc, iscbl; + const struct udpservice *udp; #ifndef _WIN32 pthread_attr_t pa; #endif @@ -484,6 +482,7 @@ static int srvprepare(struct srvstart *s, int argc, char **argv){ #endif srv->pf = childdef.pf; s->isudp = childdef.isudp; + s->udp = childdef.udp; /* Default socket options, which -oc, -os, -ol and -oR change by naming them, NAME=0 to turn one off. @@ -496,17 +495,13 @@ static int srvprepare(struct srvstart *s, int argc, char **argv){ listening port there, and it only allows another local process to bind the same address and port, with undefined behaviour as to which socket receives the connections. Use -olSO_EXCLUSIVEADDRUSE to prevent that - instead. A Windows build which does not share the listening socket is the - exception: its UDP services bind a second socket to the same address to - answer from, and Windows only allows that when both sockets ask for it. + instead. */ (void)getopts("TCP_NODELAY", &srv->clisockopts); (void)getopts("TCP_NODELAY", &srv->srvsockopts); #ifndef _WIN32 (void)getopts("SO_REUSEADDR", &srv->lissockopts); (void)getopts("SO_REUSEADDR", &srv->cbssockopts); -#elif !defined SHARE_UDP_SOCKET - if(s->isudp) (void)getopts("SO_REUSEADDR", &srv->lissockopts); #endif #ifdef SO_REUSEPORT (void)getopts("SO_REUSEPORT", &srv->lissockopts); @@ -514,7 +509,6 @@ static int srvprepare(struct srvstart *s, int argc, char **argv){ #endif #ifndef NOUDPMAIN if(s->isudp) { - if(!udp_table.ihashtable)inithashtable(&udp_table, 64, 256, 65536); srv->udpbuf = malloc(UDPBUFSIZE); srv->udpbuf2 = malloc(UDPBUFSIZE); if(!srv->udpbuf || !srv->udpbuf2) return 2; @@ -916,6 +910,9 @@ struct listener { from the client (%O) */ uint64_t trafin, trafout; _3proxy_mutex_t count_mutex; + /* a UDP service's sessions on a reload, for the newest service on the + listener to adopt */ + struct udpsession *udphand; }; /* A client of srv comes (1) or goes (-1). */ @@ -967,6 +964,9 @@ static struct listener *listeners = NULL; _3proxy_mutex_t listener_mutex; static void listenerstop(struct srvparam *srv); +#ifndef NOUDPMAIN +static void udpendall(struct udpsession *list, int res); +#endif static void takeopts(struct listener *l, struct srvparam *srv); static void listenerrelease(struct srvparam *srv); #endif @@ -1292,40 +1292,383 @@ static int tcpnext(struct srvstart *s, SOCKET *client, int *nlog){ #endif #ifndef NOUDPMAIN -/* The next UDP client: a datagram from a peer with no association yet. One - from a known peer is passed on to it here. A new client is returned with - udpinit held, for its child to read the datagram and release it. */ -static int udpnext(struct srvstart *s){ - struct srvparam *srv = &s->srv; - struct clientparam *defparam = &s->defparam; - struct clientparam *toparam; +/* A UDP service's loop, in the service's thread: the datagrams of its + clients come on the service's socket, those of their destinations on the + sockets of their sessions, and every answer to a client leaves from the + service's socket. No thread for a client. */ +struct udpstate { + struct udploop u; + const struct udpservice *svc; + struct udpsession *head; + int count; + struct hashtable keys; /* a session by its client's address */ + struct pollfd *fds; + struct udpsession **of; /* the session of each of fds, NULL: the service */ + int nalloc; + int nfull; +}; - _3proxy_sem_lock(udpinit); - srv->udplen = sockrecvfrom(NULL, srv->srvsock, (struct sockaddr *)&defparam->sincr, srv->udpbuf, UDPBUFSIZE, 0); - if(srv->udplen <= 0) { - _3proxy_sem_unlock(udpinit); - return NEXT_RETRY; +static void udpkey2hash(const struct hashtable *ht, void *index, uint8_t *hash){ + PROXYSOCKADDRTYPE *sa = (PROXYSOCKADDRTYPE *)index; + uint32_t m1, m2; + + m1 = murmurhash3(SAADDR(sa), SAADDRLEN(sa), 0x3a3a3a3a); + m2 = murmurhash3(SAPORT(sa), 2, m1); + memcpy(hash, &m1, 4); + memcpy(hash+4, &m2, 4); +} + +int udpanswer(struct udploop *u, struct clientparam *param, unsigned char *buf, int len){ + return param->srv->so._sendto(param->sostate, param->srv->srvsock, (char *)buf, len, 0, + (struct sockaddr *)¶m->sincr, SASIZE(¶m->sincr)); +} + +/* A session among its service's children, counted there and, if it is new, + on the listener. */ +static void udpjoin(struct srvparam *srv, struct clientparam *param, int new){ + param->prev = param->next = NULL; + _3proxy_mutex_lock(&srv->counter_mutex); + if(srv->child){ + param->next = srv->child; + srv->child->prev = param; } - if(hashresolv(&udp_table, defparam, &toparam, NULL)) { - int i, len=0; + srv->child = param; + srv->childcount++; +#ifndef STDMAIN + if(new) listenercount(srv, 1); +#endif + _3proxy_mutex_unlock(&srv->counter_mutex); +} - if(!toparam->bandlimfunc || !(*toparam->bandlimfunc)(toparam, 0, srv->udplen)){ - if(toparam->udp_nhops){ - for(i=1; i < toparam->udp_nhops; i++){ - len+=socks5_udp_build_hdr(srv->udpbuf2+len, &toparam->udp_relay[i-1]); - } - len += socks5_udp_build_hdr(srv->udpbuf2+len, &toparam->req); +/* A new client: a slot of maxchild and a param, as a thread for it would + have; NULL without a slot or memory. */ +static struct udpsession *udpnew(struct udpstate *st){ + struct srvparam *srv = st->u.srv; + struct udpsession *ss; + struct clientparam *param = NULL; + + if(_3proxy_sem_timedlock(srv->childslots, 0)){ + if(!srv->silent && !(st->nfull++ % 100)){ + char buf[128]; + + sprintf(buf, "Warning: too many connected clients (%d/%d), datagram dropped", srv->childcount, srv->maxchild); + dolog(st->u.defparam, (unsigned char *)buf); } - memcpy(srv->udpbuf2+len, srv->udpbuf, srv->udplen > UDPBUFSIZE - len?UDPBUFSIZE - len : srv->udplen); - len += srv->udplen > UDPBUFSIZE - len?UDPBUFSIZE - len : srv->udplen; - srv->so._sendto(toparam->sostate, toparam->remsock, (char *)srv->udpbuf2, len, 0, (struct sockaddr *)&toparam->sinsr, SASIZE(&toparam->sinsr)); - toparam->statscli64 += srv->udplen; - toparam->nwrites++; - } - _3proxy_sem_unlock(udpinit); - return NEXT_RETRY; + return NULL; } - return NEXT_CLIENT; + st->nfull = 0; + if(!(ss = calloc(1, sizeof(*ss))) || !(param = malloc(sizeof(*param)))){ + free(ss); + _3proxy_sem_unlock(srv->childslots); + return NULL; + } + memset(param, 0, sizeof(*param)); + memcpy(param, st->u.defparam, offsetof(struct clientparam, filters)); + param->clibuf = param->clibufdata; + param->srvbuf = param->srvbufdata; + param->clibufsize = param->srvbufsize = SRVBUFSIZE(srv); + clearstat(param); + param->clisock = param->remsock = param->ctrlsock = param->ctrlsocksrv = INVALID_SOCKET; + udpjoin(srv, param, 1); + ss->param = param; + ss->svc = st->svc; +#ifdef MODULEMAINFUNC + if(makefilters(srv, param) > CONTINUE){ + freeparam(param); + free(ss); + return NULL; + } +#endif + return ss; +} + +/* A session ends: logged by its service, its sockets closed, its slot + given back. Its own service may be gone after. */ +static void udpend(struct udpsession *ss){ + struct clientparam *param = ss->param; + + if(ss->svc->end) (*ss->svc->end)(NULL, ss); + param->clisock = INVALID_SOCKET; + freeparam(param); + free(ss); +} + +static void udpendall(struct udpsession *list, int res){ + struct udpsession *ss; + + while((ss = list)){ + list = ss->next; + if(!ss->param->res) ss->param->res = res; + udpend(ss); + } +} + +static void udpadd(struct udpstate *st, struct udpsession *ss){ + ss->prev = NULL; + ss->next = st->head; + if(st->head) st->head->prev = ss; + st->head = ss; + st->count++; + ss->keyed = st->svc->keyed; + if(ss->keyed) hashadd(&st->keys, &ss->param->sincr, &ss, MAX_COUNTER_TIME); +} + +static void udpremove(struct udpstate *st, struct udpsession *ss){ + if(ss->prev) ss->prev->next = ss->next; + else st->head = ss->next; + if(ss->next) ss->next->prev = ss->prev; + ss->next = ss->prev = NULL; + st->count--; + if(ss->keyed) hashdelete(&st->keys, &ss->param->sincr); +} + +static void udprefresh(struct udpsession *ss, time_t now){ + if(ss->idle) ss->deadline = now + ss->idle; +} + +#ifndef STDMAIN +/* On a reload: the sessions go to the listener, for the newest service on + it to adopt; if it has none, the last service to stop on it ends them. */ +static void udphandover(struct udpstate *st){ + struct listener *l = st->u.srv->listener; + struct udpsession *ss, *last = NULL; + + for(ss = st->head; ss; ss = ss->next){ + if(ss->keyed) hashdelete(&st->keys, &ss->param->sincr); + ss->keyed = 0; + last = ss; + } + if(!last) return; + if(!l){ + udpendall(st->head, 26); + st->head = NULL; + return; + } + _3proxy_mutex_lock(&listener_mutex); + last->next = l->udphand; + if(l->udphand) l->udphand->prev = last; + l->udphand = st->head; + _3proxy_mutex_unlock(&listener_mutex); + st->head = NULL; + st->count = 0; +} + +/* Sessions of an older service on the listener become this one's: a slot + here, the old one given back, its rules and limits, re-authenticated; 26 + if it is another kind of service, 27 without a slot. */ +static void udpadopt(struct udpstate *st){ + struct srvparam *srv = st->u.srv, *old; + struct listener *l = srv->listener; + struct udpsession *list, *ss; + struct clientparam *param; + int res; + + if(!l || !l->udphand) return; + _3proxy_mutex_lock(&listener_mutex); + list = l->udphand; + l->udphand = NULL; + _3proxy_mutex_unlock(&listener_mutex); + while((ss = list)){ + list = ss->next; + ss->next = ss->prev = NULL; + param = ss->param; + old = param->srv; + if(old == srv) { udpadd(st, ss); continue; } + if(param->service != srv->service){ + param->res = 26; + udpend(ss); + continue; + } + if(_3proxy_sem_timedlock(srv->childslots, 0)){ + param->res = 27; + udpend(ss); + continue; + } + /* what belongs to the old service and configuration let go */ + while(param->nfilters > 0){ + param->nfilters--; + if(param->filters[param->nfilters].filter->filter_clear) + (*param->filters[param->nfilters].filter->filter_clear)(param->filters[param->nfilters].data); + } + if(param->connlim) stopconnlims(param); + param->bandlimfunc = NULL; + param->bandlims[0] = param->bandlimsout[0] = NULL; + param->trafcountfunc = NULL; + param->maxtrafin64 = param->maxtrafout64 = 0; + _3proxy_mutex_lock(&old->counter_mutex); + if(param->prev) param->prev->next = param->next; + else old->child = param->next; + if(param->next) param->next->prev = param->prev; + old->childcount--; + _3proxy_sem_unlock(old->childslots); + _3proxy_mutex_unlock(&old->counter_mutex); + param->srv = srv; + param->sostate = srv->so.state; + param->version = srv->version; + udpjoin(srv, param, 0); + ss->svc = st->svc; +#ifdef MODULEMAINFUNC + if(makefilters(srv, param) > CONTINUE){ + param->res = 26; + udpend(ss); + continue; + } +#endif + if((res = (*srv->authfunc)(param))){ + param->res = res; + udpend(ss); + continue; + } + udpadd(st, ss); + } +} +#endif + +#ifndef STDMAIN +/* After a reload, while an older service of this kind still accepts on the + listener, it goes on serving the service's socket alone, until it hands + its sessions over: a client is not given a session in both. */ +static int udpolder(struct srvparam *srv){ + struct srvparam *o; + int older = 0; + + if(!srv->listener) return 0; + _3proxy_mutex_lock(&listener_mutex); + for(o = srv->listener->owners; o; o = o->lisnext) + if(o != srv && o->version < srv->version && o->service == srv->service) older = 1; + _3proxy_mutex_unlock(&listener_mutex); + return older; +} +#endif + +/* The datagrams waiting on the service's socket: to a session, or a new + one. Some at a time, so that the sessions get their turn. */ +static void udpclients(struct udpstate *st, time_t now){ + struct srvparam *srv = st->u.srv; + struct clientparam *defparam = st->u.defparam; + struct udpsession *ss; + SASIZETYPE size; + int len, n; + + for(n = 0; n < 64; n++){ + size = sizeof(defparam->sincr); + len = srv->so._recvfrom(srv->so.state, srv->srvsock, (char *)srv->udpbuf, UDPBUFSIZE, 0, + (struct sockaddr *)&defparam->sincr, &size); + if(len < 0){ +#ifdef _WIN32 + /* an answer to a client was refused (ICMP): reported here */ + if(WSAGetLastError() == WSAECONNRESET) continue; +#endif + break; + } + if(st->svc->keyed && hashresolv(&st->keys, &defparam->sincr, &ss, NULL)){ + if(ss->done) continue; + udprefresh(ss, now); + if(st->svc->client && (*st->svc->client)(&st->u, ss, srv->udpbuf, len)) ss->done = 1; + continue; + } + if(!(ss = udpnew(st))) continue; + ss->param->sincr = defparam->sincr; + udpadd(st, ss); + if((*st->svc->first)(&st->u, ss, srv->udpbuf, len)) ss->done = 1; + else if(!ss->deadline || ss->idle) udprefresh(ss, now); + } +} + +static int udploop(struct srvstart *s){ + struct udpstate st; + struct srvparam *srv = &s->srv; + struct udpsession *ss, *next; + time_t now; + int i, n, nfds, timeo, res = 0, wait = 0; + + memset(&st, 0, sizeof(st)); + st.u.srv = srv; + st.u.defparam = &s->defparam; + st.u.buf = srv->udpbuf2; + st.svc = s->udp; + st.keys.index2hash_add = udpkey2hash; + st.keys.index2hash_search = udpkey2hash; + st.keys.recsize = sizeof(struct udpsession *); + st.keys.hash_size = 8; + if(st.svc->keyed && inithashtable(&st.keys, 64, 256, 1048576)) return 21; + for(;;){ + if(srv->config->paused != srv->paused) break; +#ifndef STDMAIN + udpadopt(&st); +#endif + if(st.nalloc < 2 * st.count + 1){ + int want = 2 * st.count + 64; + struct pollfd *f = realloc(st.fds, want * sizeof(*f)); + + if(f) st.fds = f; + if(f && (st.of = realloc(st.of, want * sizeof(*st.of)))) st.nalloc = want; + else { res = 21; break; } + } + now = time(NULL); + timeo = 1000; +#ifndef STDMAIN + /* waiting for an older service's sessions: soon looked for again */ + if((wait = udpolder(srv))) timeo = 20; +#endif + nfds = 0; + st.fds[nfds].fd = srv->srvsock; + st.fds[nfds].events = wait? 0 : POLLIN; + st.fds[nfds].revents = 0; + st.of[nfds++] = NULL; + for(ss = st.head; ss; ss = ss->next){ + if(ss->deadline && (ss->deadline - now) * 1000 < timeo) + timeo = ss->deadline > now? (int)(ss->deadline - now) * 1000 : 0; + if(ss->param->remsock != INVALID_SOCKET){ + st.fds[nfds].fd = ss->param->remsock; + st.fds[nfds].events = POLLIN; + st.fds[nfds].revents = 0; + st.of[nfds++] = ss; + } + /* a parent's control connection: its end ends the session */ + if(ss->param->ctrlsocksrv != INVALID_SOCKET && ss->param->ctrlsocksrv != ss->param->remsock){ + st.fds[nfds].fd = ss->param->ctrlsocksrv; + st.fds[nfds].events = POLLIN; + st.fds[nfds].revents = 0; + st.of[nfds++] = ss; + } + } + n = srv->so._poll(srv->so.state, st.fds, nfds, timeo); + if(n < 0 && errno != EINTR && errno != EAGAIN){ + res = 91; + break; + } + now = time(NULL); + for(i = 1; n > 0 && i < nfds; i++){ + ss = st.of[i]; + if(!st.fds[i].revents || ss->done) continue; + if(st.fds[i].fd == ss->param->remsock){ + udprefresh(ss, now); + if((*st.svc->server)(&st.u, ss)) ss->done = 1; + } + else ss->done = 1; + } + if(n > 0 && !wait && st.fds[0].revents) udpclients(&st, now); + for(ss = st.head; ss; ss = next){ + next = ss->next; + if(!ss->done && ss->deadline && ss->deadline <= now){ + ss->done = 1; + ss->param->res = st.svc->timeoutres; + } + if(ss->done){ + udpremove(&st, ss); + udpend(ss); + } + } + } +#ifndef STDMAIN + if(srv->config->paused != srv->paused) udphandover(&st); +#endif + udpendall(st.head, 26); + if(st.svc->keyed) destroyhashtable(&st.keys); + free(st.fds); + free(st.of); + return res; } #endif @@ -1363,12 +1706,7 @@ static int srvspawn(struct srvstart *s, SOCKET client_sock){ reused = newparam != NULL; if(!reused && (!(newparam = malloc (sizeof(*defparam))) || _3proxy_sem_init_timed(newparam->wake, 0, 1))){ if(newparam) free(newparam); - if(!s->isudp) srv->so._closesocket(srv->so.state, client_sock); -#ifndef NOUDPMAIN - else { - _3proxy_sem_unlock(udpinit); - } -#endif + srv->so._closesocket(srv->so.state, client_sock); defparam->res = 21; if(!srv->silent)dolog(defparam, (unsigned char *)"Memory Allocation Failed"); usleep(SLEEPTIME); @@ -1427,11 +1765,6 @@ static int srvspawn(struct srvstart *s, SOCKET client_sock){ newparam->clisock = INVALID_SOCKET; } newparam->srv = NULL; -#ifndef NOUDPMAIN - if(s->isudp){ - _3proxy_sem_unlock(udpinit); - } -#endif freeworker(newparam); } else { @@ -1489,6 +1822,10 @@ static int srvrun(struct srvstart *s){ tpool.services++; tpool.maxchild += srv->maxchild; _3proxy_mutex_unlock(&tpool.mutex); +#ifndef NOUDPMAIN + if(s->isudp) res = udploop(s); + else +#endif for (;;) { for(;;){ /* A connection is only taken with a slot of maxchild in hand, which @@ -1518,12 +1855,10 @@ static int srvrun(struct srvstart *s){ } if((srv->config->paused != srv->paused) || (error < 0)) break; error = 0; -#if defined NOTCPMAIN - next = udpnext(s); -#elif defined NOUDPMAIN +#ifndef NOTCPMAIN next = tcpnext(s, &new_sock, &nlog); #else - next = s->isudp? udpnext(s) : tcpnext(s, &new_sock, &nlog); + next = NEXT_STOP; #endif if(next == NEXT_STOP) break; if(next == NEXT_RETRY) continue; @@ -1851,6 +2186,7 @@ static void listenerstop(struct srvparam *srv){ struct listener *l = srv->listener; struct srvparam **op; SOCKET sock = INVALID_SOCKET; + struct udpsession *hand = NULL; _3proxy_mutex_lock(&listener_mutex); for(op = &l->owners; *op; op = &(*op)->lisnext) if(*op == srv){ @@ -1861,8 +2197,14 @@ static void listenerstop(struct srvparam *srv){ if(!l->owners){ sock = l->sock; l->sock = INVALID_SOCKET; + /* sessions handed over with no service left to adopt them */ + hand = l->udphand; + l->udphand = NULL; } _3proxy_mutex_unlock(&listener_mutex); +#ifndef NOUDPMAIN + udpendall(hand, 26); +#endif if(sock == INVALID_SOCKET) return; so._closesocket(so.state, sock); #ifdef WITH_UN @@ -1963,10 +2305,6 @@ void maindrop(void *service){ #endif -#ifndef NOUDPMAIN -int udpinited = 0; -_3proxy_sem_t udpinit; -#endif void srvinit(struct srvparam * srv, struct clientparam *param){ @@ -2017,12 +2355,6 @@ void srvinit(struct srvparam * srv, struct clientparam *param){ *SAFAMILY(¶m->req) = *SAFAMILY(¶m->sinsl) = *SAFAMILY(¶m->sinsr) = *SAFAMILY(¶m->sincr) = *SAFAMILY(¶m->sincl) = AF_INET; _3proxy_mutex_init(&srv->counter_mutex); (void)_3proxy_sem_init_timed(srv->childslots, srv->maxchild, srv->maxchild); -#ifndef NOUDPMAIN - if(!udpinited){ - (void)_3proxy_sem_init(udpinit, 1, 1); - } - udpinited = 1; -#endif srv->intsa = conf->intsa; srv->extsa = conf->extsa; #ifndef NOIPV6 @@ -2147,13 +2479,6 @@ void srvfree(struct srvparam * srv){ void releaseparam(struct clientparam * param) { if(param->srv){ if(param->srv->so.freefunc) param->srv->so.freefunc(param->sostate); -#ifndef STDMAIN - if(param->srv->service == S_UDPPM){ - _3proxy_mutex_lock(¶m->srv->counter_mutex); - hashdelete(&udp_table, param); - _3proxy_mutex_unlock(¶m->srv->counter_mutex); - } -#endif } if(param->clibuf != param->clibufdata) free(param->clibuf); if(param->srvbuf != param->srvbufdata) free(param->srvbuf); diff --git a/src/structures.h b/src/structures.h index 10a762e..10df1d5 100644 --- a/src/structures.h +++ b/src/structures.h @@ -976,12 +976,17 @@ struct symbol { void * value; }; +struct udpservice; + struct proxydef { PROXYFUNC pf; uint16_t port; int isudp; int service; char * helpmessage; + /* a UDP service: its datagrams are handled in the service's own thread + by these, no thread for a client */ + const struct udpservice *udp; }; extern struct proxydef childdef; @@ -996,7 +1001,6 @@ extern struct hashtable dns_table; extern struct hashtable dns6_table; extern struct hashtable auth_table; extern const struct hashtable pwl_template; -extern struct hashtable udp_table; struct authcache { unsigned char username[64]; diff --git a/src/udppm.c b/src/udppm.c index cbf7e56..082628d 100644 --- a/src/udppm.c +++ b/src/udppm.c @@ -14,30 +14,36 @@ #ifndef UDP #define UDP #endif -#define RETURN(xxx) { param->res = xxx; goto CLEANRET; } +/* udppm: a session per client address, in the service's thread (see + udploop() in proxymain.c). Its datagrams go on to the target from a socket + of the session's own, behind the headers of a parent chain if the ACL + gives one; what comes back is answered from the service's socket. */ -static void udpparam2hash(const struct hashtable *ht, void *index, uint8_t *hash){ - struct clientparam *param = (struct clientparam *)index; - uint32_t m1, m2; - m1 = murmurhash3(SAADDR(¶m->srv->intsa), SAADDRLEN(¶m->srv->intsa), 0x3a3a3a3a); - m1 = murmurhash3(SAPORT(¶m->sincr), 2, m1); - m2 = murmurhash3(SAADDR(¶m->sincr), SAADDRLEN(¶m->sincr), m1); - m2 = murmurhash3(SAPORT(¶m->srv->intsa), 2, m2); - memcpy(hash, &m1, 4); - memcpy(hash+4, &m2, 4); +/* A datagram of the client to the target, or to the first relay of a chain + with the headers of the hops after it and of the target. */ +static void udppmforward(struct udploop *u, struct clientparam *param, unsigned char *data, int len){ + int i, off = 0; + + if(param->bandlimfunc && (*param->bandlimfunc)(param, 0, len)) return; + if(param->udp_nhops){ + for(i = 1; i < param->udp_nhops; i++) + off += socks5_udp_build_hdr(u->buf + off, ¶m->udp_relay[i-1]); + off += socks5_udp_build_hdr(u->buf + off, ¶m->req); + } + if(len > UDPBUFSIZE - off) len = UDPBUFSIZE - off; + memcpy(u->buf + off, data, len); + param->srv->so._sendto(param->sostate, param->remsock, (char *)u->buf, off + len, 0, + (struct sockaddr *)¶m->sinsr, SASIZE(¶m->sinsr)); + param->statscli64 += len; + param->nwrites++; } -struct hashtable udp_table = {udpparam2hash, udpparam2hash, sizeof(struct clientparam *), 8}; -int socks5_udp_build_hdr(unsigned char *buf, PROXYSOCKADDRTYPE *addr); - -void * udppmchild(struct clientparam* param) { - int authres; - int i; - int len = 0; - - if(parsehostname((char *)param->srv->target, param, ntohs(param->srv->targetport))) { RETURN(100) } +static int udppmfirst(struct udploop *u, struct udpsession *ss, unsigned char *buf, int len){ + struct clientparam *param = ss->param; + int res; + if(parsehostname((char *)param->srv->target, param, ntohs(param->srv->targetport))) return (param->res = 100); #ifndef NOIPV6 param->sinsl = *SAFAMILY(¶m->req) == AF_INET6? param->srv->extsa6 : param->srv->extsa; #else @@ -46,46 +52,74 @@ void * udppmchild(struct clientparam* param) { *SAPORT(¶m->sinsl) = 0; param->sinsr = param->req; param->operation = UDPASSOC; - authres = (*param->srv->authfunc)(param); - if(authres) { RETURN(authres); } - if((authres = udpbind(param))) { RETURN(authres); } - if(!param->srv->s_option)hashadd(&udp_table, param, ¶m, MAX_COUNTER_TIME); - if(!param->bandlimfunc || !(*param->bandlimfunc)(param, 0, param->srv->udplen)){ - if(param->udp_nhops){ - for(i=1; i < param->udp_nhops; i++){ - len+=socks5_udp_build_hdr(param->srvbuf+len, ¶m->udp_relay[i-1]); - } - len += socks5_udp_build_hdr(param->srvbuf+len, ¶m->req); - } - memcpy(param->srvbuf+len, param->srv->udpbuf, param->srv->udplen > UDPBUFSIZE - len?UDPBUFSIZE - len : param->srv->udplen); - len += param->srv->udplen > UDPBUFSIZE - len?UDPBUFSIZE - len : param->srv->udplen; - param->srv->so._sendto(param->sostate, param->remsock, (char *)param->srvbuf, len, 0, (struct sockaddr *)¶m->sinsr, SASIZE(¶m->sinsr)); - param->statscli64 += param->srvinbuf; - param->nwrites++; - } - _3proxy_sem_unlock(udpinit); - param->srvinbuf = 0; - param->clisock = param->srv->srvsock; - param->waitserver64 = 0x7fffffffffffffff; - param->res = udpsockmap(param, param->srv->timeouts[STRING_L]); - _3proxy_sem_lock(udpinit); - if(!param->srv->s_option)hashdelete(&udp_table, param); - -CLEANRET: - - _3proxy_sem_unlock(udpinit); - dolog(param, NULL); - param->clisock = INVALID_SOCKET; - return (NULL); + if((res = (*param->srv->authfunc)(param))) return (param->res = res); + if((res = udpbind(param))) return (param->res = res); + ss->idle = param->srv->timeouts[STRING_L]; + udppmforward(u, param, buf, len); + return 0; } +static int udppmclient(struct udploop *u, struct udpsession *ss, unsigned char *buf, int len){ + udppmforward(u, ss->param, buf, len); + return 0; +} + +/* What the target, or the relay of a chain, sends back: to the client, the + headers of the chain taken off. */ +static int udppmserver(struct udploop *u, struct udpsession *ss){ + struct clientparam *param = ss->param; + PROXYSOCKADDRTYPE from; + SASIZETYPE size = sizeof(from); + int len, off = 0, k, next; + + len = param->srv->so._recvfrom(param->sostate, param->remsock, (char *)u->buf, UDPBUFSIZE, 0, + (struct sockaddr *)&from, &size); + if(len < 0 && (errno == EAGAIN || errno == EINTR)) return 0; +#ifdef _WIN32 + /* an earlier datagram was refused (ICMP): reported here, not fatal */ + if(len < 0 && WSAGetLastError() == WSAECONNRESET) return 0; +#endif + if(len <= 0) return (param->res = 486); + if(!SAISNULL(¶m->sinsr) && *SAPORT(¶m->sinsr) && + (SAADDRLEN(&from) != SAADDRLEN(¶m->sinsr) || + memcmp(SAADDR(&from), SAADDR(¶m->sinsr), SAADDRLEN(&from)) || + memcmp(SAPORT(&from), SAPORT(¶m->sinsr), 2))) + return 0; + for(k = 0; k < param->udp_nhops; k++){ + if((next = socks5_udp_skip_hdr(u->buf + off, len - off)) < 0) return 0; + off += next; + } + len -= off; + param->statssrv64 += len; + param->nreads++; + if(param->bandlimfunc && (*param->bandlimfunc)(param, len, 0)) return 0; + udpanswer(u, param, u->buf + off, len); + /* -s: one answer to a request */ + if(param->srv->s_option) return 1; + return 0; +} + +static void udppmend(struct udploop *u, struct udpsession *ss){ + dolog(ss->param, NULL); +} + +const struct udpservice udppmservice = { + 1, + udppmfirst, + udppmclient, + udppmserver, + udppmend, + 92 +}; + #ifdef WITHMAIN struct proxydef childdef = { - udppmchild, + NULL, 0, 1, S_UDPPM, - " -s single packet UDP service for request/reply (DNS-like) services\n" + " -s single packet UDP service for request/reply (DNS-like) services\n", + &udppmservice }; #include "proxymain.c" #endif diff --git a/src/udpsockmap.c b/src/udpsockmap.c index 0d7834d..d139577 100644 --- a/src/udpsockmap.c +++ b/src/udpsockmap.c @@ -21,7 +21,7 @@ int socks5_udp_build_hdr(unsigned char *buf, PROXYSOCKADDRTYPE *addr) return 4 + SAADDRLEN(addr) + 2; } -static int socks5_udp_skip_hdr(unsigned char *buf, int len) +int socks5_udp_skip_hdr(unsigned char *buf, int len) { int addr_len; int off; @@ -49,7 +49,6 @@ static int socks5_udp_skip_hdr(unsigned char *buf, int len) * 3 three parent proxies (prepend 2 headers / strip 2 headers) * * param->waitserver64 non-zero: skip client socket polling (server→client only) - * param->srv->s_option non-zero: return after first datagram sent to client * param->ctrlsock TCP control socket from the client; INVALID_SOCKET if none. */ @@ -207,7 +206,6 @@ int udpsockmap(struct clientparam *param, int timeo) sin = cliaddr; nhops = param->udp_nhops; - if(param->srv->service == S_UDPPM) nhops++; udpfds(param, &p); for (;;) { @@ -330,7 +328,6 @@ int udpsockmap(struct clientparam *param, int timeo) ares = udpreconnect(param); if (param->remsock == INVALID_SOCKET) return ares; nhops = param->udp_nhops; - if(param->srv->service == S_UDPPM) nhops++; udpfds(param, &p); } lastres = ares; @@ -413,7 +410,6 @@ int udpsockmap(struct clientparam *param, int timeo) param->srv->so._sendto(param->sostate, param->clisock, (char *)param->srvbuf + sendoff, sendlen, 0, (struct sockaddr *)&sin, SASIZE(&sin)); - if (param->srv->s_option && param->srv->service == S_UDPPM) return 0; } if ((p.ctrl >= 0 && p.fds[p.ctrl].revents) ||