Re-authenticate clients of old services against the new configuration

After a reload, unless noforce, a client of an old service goes by the
rules of the newest service on its listener, the one accepting there
now. It takes one of that service's maxconn slots, which counts it
there and keeps that service and its configuration alive until the
client is done, and is re-authenticated as its client: access lists,
users, authentication, bandwidth limits and traffic counters come from
the new configuration. A later reload moves it on, giving the slot
back. The client stays its own service's child, which keeps its
filters and TLS state.

It is dropped if its service was removed or is of another type now
(26), or if the newest service has no free slot (27); what the
re-authentication refuses ends it as before. A service without a
listener (UDP, connecting back) keeps its clients as they are.

This restores what force did before the configuration was kept per
service: since then a client compared its version with its own
service's configuration, which never changes, and was never checked
again. Services now take their configuration's version as it is: a
reload numbers the new configuration before reading it, and so does
startup now, where services were numbered one ahead.

A listener now stays with its old service until that service's last
client is gone, those clients looking the newest service up through
it; the socket itself is still closed as soon as nobody accepts on it.
Connection limits are given back in the configuration they were taken
in. PSRV() is the service whose rules apply to a client.

The reload case checks that a download of a user the new configuration
keeps finishes under it, and that one of a user it removes is cut.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/3proxy.c

@@ -558,6 +558,7 @@ int WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, LPWSTR lpCmdLine, int
   _3proxy_mutex_init(&config_mutex);
   _3proxy_mutex_init(&confref_mutex);
   _3proxy_mutex_init(&listener_mutex);
+  reattachfunc = reattach;
   _3proxy_mutex_init(&runningservices_mutex);
   _3proxy_mutex_init(&bandlim_mutex);
   _3proxy_mutex_init(&connlim_mutex);
@@ -587,8 +588,9 @@ int WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, LPWSTR lpCmdLine, int
 
   confbegin();
   initcommands();
-  res = readconfig(fp);
+  /* numbered before it is read, as a reload numbers the new one */
   conf->version++;
+  res = readconfig(fp);
   resolvfunc = conf->resolvfunc;
 
   /* readconfig() names the failing line, which an exit status would cut

src/common.c

@@ -233,6 +233,7 @@ struct extparam confboot = {
    whatever runs meanwhile without a service of its own. */
 struct extparam *conf = &confboot;
 struct extparam *runconf = &confboot;
+int (*reattachfunc)(struct clientparam *param) = NULL;
 _3proxy_mutex_t confref_mutex;
 
 

src/limiter.c

@@ -15,8 +15,11 @@ int startconnlims (struct clientparam *param){
 	int ret = 0;
 
 	param->connlim = 1;
+	/* released where they were taken, whatever configuration applies by
+	   then: that one stays while the connection does */
+	param->connlimconf = PSRV(param)->config;
 	_3proxy_mutex_lock(&connlim_mutex);
-	for(ce = param->srv->config->connlimiter; ce; ce = ce->next) {
+	for(ce = param->connlimconf->connlimiter; ce; ce = ce->next) {
 		if(ACLmatches(ce->ace, param)){
 			if(ce->ace->action == NOCONNLIM)break;
 			if(!ce->period){
@@ -44,7 +47,7 @@ int startconnlims (struct clientparam *param){
 	}
 	if(ret) {
 		struct connlim * cee;
-		for(cee = param->srv->config->connlimiter; cee != ce; cee = cee->next) {
+		for(cee = param->connlimconf->connlimiter; cee != ce; cee = cee->next) {
 			if(ACLmatches(cee->ace, param) && !cee->period && cee->rating) {
 				cee->rating--;
 			}
@@ -59,7 +62,7 @@ void stopconnlims (struct clientparam *param){
 	struct connlim * ce;
 
 	_3proxy_mutex_lock(&connlim_mutex);
-	for(ce = param->srv->config->connlimiter; ce; ce = ce->next) {
+	for(ce = (param->connlimconf? param->connlimconf : param->srv->config)->connlimiter; ce; ce = ce->next) {
 		if(ACLmatches(ce->ace, param)){
 			if(ce->ace->action == NOCONNLIM)break;
 			if(!ce->period && ce->rating){
@@ -78,28 +81,28 @@ void initbandlims (struct clientparam *param){
 	param->bandlimfunc = NULL;
 	param->bandlims[0] = NULL;
 	param->bandlimsout[0] = NULL;
-	if(!param->srv->config->bandlimfunc || (!param->srv->config->bandlimiter && !param->srv->config->bandlimiterout)) return;
+	if(!PSRV(param)->config->bandlimfunc || (!PSRV(param)->config->bandlimiter && !PSRV(param)->config->bandlimiterout)) return;
-	for(i=0, be = param->srv->config->bandlimiter; be && i<MAXBANDLIMS; be = be->next) {
+	for(i=0, be = PSRV(param)->config->bandlimiter; be && i<MAXBANDLIMS; be = be->next) {
 		if(ACLmatches(be->ace, param)){
 			if(be->ace->action == NOBANDLIM) {
 				break;
 			}
 			param->bandlims[i++] = be;
-			param->bandlimfunc = param->srv->config->bandlimfunc;
+			param->bandlimfunc = PSRV(param)->config->bandlimfunc;
 		}
 	}
 	if(i<MAXBANDLIMS)param->bandlims[i] = NULL;
-	for(i=0, be = param->srv->config->bandlimiterout; be && i<MAXBANDLIMS; be = be->next) {
+	for(i=0, be = PSRV(param)->config->bandlimiterout; be && i<MAXBANDLIMS; be = be->next) {
 		if(ACLmatches(be->ace, param)){
 			if(be->ace->action == NOBANDLIM) {
 				break;
 			}
 			param->bandlimsout[i++] = be;
-			param->bandlimfunc = param->srv->config->bandlimfunc;
+			param->bandlimfunc = PSRV(param)->config->bandlimfunc;
 		}
 	}
 	if(i<MAXBANDLIMS)param->bandlimsout[i] = NULL;
-	param->bandlimver = param->srv->config->bandlimver;
+	param->bandlimver = PSRV(param)->config->bandlimver;
 }
 
 unsigned bandlimitfunc(struct clientparam *param, unsigned nbytesin, unsigned nbytesout){
@@ -125,9 +128,9 @@ unsigned bandlimitfunc(struct clientparam *param, unsigned nbytesin, unsigned nb
 
 	if(!nbytesin && !nbytesout) return 0;
 	_3proxy_mutex_lock(&bandlim_mutex);
-	if(param->bandlimver != param->srv->config->bandlimver){
+	if(param->bandlimver != PSRV(param)->config->bandlimver){
 		initbandlims(param);
-		param->bandlimver = param->srv->config->bandlimver;
+		param->bandlimver = PSRV(param)->config->bandlimver;
 	}
 	for(i=0; nbytesin&& i<MAXBANDLIMS && param->bandlims[i]; i++){
 		if( !param->bandlims[i]->basetime ||
@@ -172,7 +175,7 @@ void trafcountfunc(struct clientparam *param){
 	int countout = 0;
 
 	_3proxy_mutex_lock(&tc_mutex);
-	for(tc = param->srv->config->trafcounter; tc; tc = tc->next) {
+	for(tc = PSRV(param)->config->trafcounter; tc; tc = tc->next) {
 		if(ACLmatches(tc->ace, param)){
 
 			if(tc->ace->action == NOCOUNTIN) {
@@ -188,7 +191,7 @@ void trafcountfunc(struct clientparam *param){
 			tc->v->updated = curtime;
 		}
 	}
-	if(countout) for(tc = param->srv->config->trafcounter; tc; tc = tc->next) {
+	if(countout) for(tc = PSRV(param)->config->trafcounter; tc; tc = tc->next) {
 		if(ACLmatches(tc->ace, param)){
 			if(tc->ace->action == NOCOUNTOUT || tc->ace->action == NOCOUNTALL) break;
 			if(tc->ace->action != COUNTOUT && tc->ace->action != COUNTALL ) {

src/proxy.h

@@ -226,6 +226,9 @@ extern int timetoexit, needreload;
 extern time_t curtime;		/* the clock, kept by the main thread */
 
 extern struct extparam *conf, *runconf;
+/* the service whose rules apply to a client: its own, or after a reload the
+   newest one on its listener */
+#define PSRV(param) ((param)->newsrv? (param)->newsrv : (param)->srv)
 extern _3proxy_mutex_t confref_mutex, listener_mutex;
 struct extparam * confget(void);
 void confhold(struct extparam *c);
@@ -397,6 +400,9 @@ void freeauth(struct auth *);
 void freefilter(struct filter *filter);
 void freeconf(struct extparam *confp);
 void confbegin(void);
+int reattach(struct clientparam *param);
+/* reattach() where configurations are reloaded, NULL elsewhere */
+extern int (*reattachfunc)(struct clientparam *param);
 struct passwords * copypwl (struct passwords *pwl);
 void freepwl(struct passwords *pw);
 int copyfilter(struct filter *, struct srvparam *srv);

src/proxymain.c

@@ -883,13 +883,17 @@ struct listener {
 	PROXYSOCKADDRTYPE sa;
 	char *netns, *device;
 	struct sockoptset opts;
-	struct srvparam *owner;	/* the service accepting on it */
+	struct srvparam *owner;	/* the service accepting on it, if any */
-	int users;		/* the services holding it */
+	int users;		/* the services holding it: the one accepting,
+				   and those which did until their last client
+				   is gone, whose clients look up the newest
+				   service through it after a reload */
 };
 static struct listener *listeners = NULL;
 _3proxy_mutex_t listener_mutex;
 
 static int listenerkeep(struct srvparam *srv, SOCKET sock);
+static void listenerstop(struct srvparam *srv);
 static void listenerrelease(struct srvparam *srv);
 #endif
 
@@ -1543,7 +1547,7 @@ void mainclaim(void *service){
  if(s->isudp || s->iscbc) return;
  _3proxy_mutex_lock(&listener_mutex);
  for(l = listeners; l; l = l->next){
-	if(l->owner && l->owner->config != srv->config && samelistener(l, srv)){
+	if(l->owner && l->sock != INVALID_SOCKET && l->owner->config != srv->config && samelistener(l, srv)){
 		l->owner = srv;
 		l->users++;
 		srv->listener = l;
@@ -1579,8 +1583,29 @@ static int listenerkeep(struct srvparam *srv, SOCKET sock){
  return 0;
 }
 
-/* A service lets go of its listening socket: closed if nobody else holds
+/* A service stops accepting: if it was the one accepting and no service
-   it. */
+   took the socket over, it is closed now, so that connections do not queue
+   up on it. */
+static void listenerstop(struct srvparam *srv){
+ struct listener *l = srv->listener;
+ SOCKET sock = INVALID_SOCKET;
+
+ _3proxy_mutex_lock(&listener_mutex);
+ if(l->owner == srv){
+	l->owner = NULL;
+	sock = l->sock;
+	l->sock = INVALID_SOCKET;
+ }
+ _3proxy_mutex_unlock(&listener_mutex);
+ if(sock == INVALID_SOCKET) return;
+ so._closesocket(so.state, sock);
+#ifdef WITH_UN
+ if(*SAFAMILY(&l->sa) == AF_UNIX && *SAADDR(&l->sa))unlink((char *)SAADDR(&l->sa));
+#endif
+}
+
+/* A service lets go of its listener once its last client is gone; the last
+   to let go frees it. */
 static void listenerrelease(struct srvparam *srv){
  struct listener *l = srv->listener, **lp;
  int last;
@@ -1595,15 +1620,71 @@ static void listenerrelease(struct srvparam *srv){
  }
  _3proxy_mutex_unlock(&listener_mutex);
  if(!last) return;
- so._closesocket(so.state, l->sock);
+ if(l->sock != INVALID_SOCKET){
+	so._closesocket(so.state, l->sock);
 #ifdef WITH_UN
- if(*SAFAMILY(&l->sa) == AF_UNIX && *SAADDR(&l->sa))unlink((char *)SAADDR(&l->sa));
+	if(*SAFAMILY(&l->sa) == AF_UNIX && *SAADDR(&l->sa))unlink((char *)SAADDR(&l->sa));
 #endif
+ }
  if(l->netns) free(l->netns);
  if(l->device) free(l->device);
  free(l);
 }
 
+/* Gives back a client's slot in a service it was not a child of. */
+static void slotrelease(struct srvparam *srv){
+ _3proxy_mutex_lock(&srv->counter_mutex);
+ _3proxy_sem_unlock(srv->childslots);
+ _3proxy_mutex_unlock(&srv->counter_mutex);
+}
+
+/* After a reload, unless noforce: a client goes by the rules of the newest
+   service on its listener, the one now accepting there, re-authenticated
+   as it. It takes one of that service's maxconn slots, which counts it
+   there and keeps the service, and its configuration, until the client is
+   done; a slot it held in a service before is given back. 0 to go on, or
+   the code to end the connection with: 26 if its service was removed, or
+   is another one now, 27 if the newest has no slot free, or what the
+   authentication says. */
+int reattach(struct clientparam *param){
+ struct srvparam *cur = PSRV(param), *newest, *orig;
+ int version = runconf->version, res;
+
+ /* without a listener (UDP, connecting back) nothing replaces it */
+ if(!cur->listener){
+	param->version = version;
+	return 0;
+ }
+ _3proxy_mutex_lock(&listener_mutex);
+ newest = cur->listener->owner;
+ if(newest == cur){
+	/* the old service has not let go of the socket yet */
+	_3proxy_mutex_unlock(&listener_mutex);
+	return 0;
+ }
+ /* param->service: the old service is a zombie by now */
+ if(!newest || newest->service != param->service){
+	_3proxy_mutex_unlock(&listener_mutex);
+	return 26;
+ }
+ /* while it accepts the newest is not freed, nor after it has a slot */
+ if(_3proxy_sem_timedlock(newest->childslots, 0)){
+	_3proxy_mutex_unlock(&listener_mutex);
+	return 27;
+ }
+ _3proxy_mutex_unlock(&listener_mutex);
+ if(param->newsrv) slotrelease(param->newsrv);
+ param->newsrv = newest;
+ param->version = version;
+ /* authenticated as the newest service's client: its access lists, users,
+    limits and counters */
+ orig = param->srv;
+ param->srv = newest;
+ res = (*newest->authfunc)(param);
+ param->srv = orig;
+ return res;
+}
+
 /* A prepared service which could not be started. */
 void maindrop(void *service){
  struct srvstart *s = (struct srvstart *)service;
@@ -1623,7 +1704,7 @@ _3proxy_sem_t udpinit;
 void srvinit(struct srvparam * srv, struct clientparam *param){
 
  memset(srv, 0, sizeof(struct srvparam));
- srv->version = conf->version + 1;
+ srv->version = conf->version;
  srv->paused = conf->paused;
  srv->logfunc = conf->havelog?conf->logfunc:lognone;
  srv->noforce = conf->noforce;
@@ -1717,11 +1798,11 @@ void srvfree(struct srvparam * srv){
  int i;
 
 #ifndef STDMAIN
- /* a listening socket kept as a listener is closed by its last holder,
+ /* a listening socket kept as a listener goes on if a service of a later
-    which may be a service of a later configuration */
+    configuration took it over */
  if(srv->listener){
 	srv->srvsock = INVALID_SOCKET;
-	listenerrelease(srv);
+	listenerstop(srv);
  }
 #endif
  if(srv->srvsock != INVALID_SOCKET) {
@@ -1781,6 +1862,7 @@ void srvfree(struct srvparam * srv){
 #endif
  /* the last service of a configuration no longer in force releases it */
 #ifndef STDMAIN
+ if(srv->listener) listenerrelease(srv);
  if(srv->config){
 	struct extparam *c = srv->config;
 
@@ -1856,6 +1938,11 @@ void releaseparam(struct clientparam * param) {
 			(*param->filters[param->nfilters].filter->filter_clear)(param->filters[param->nfilters].data);
 	}
 	if(param->connlim) stopconnlims(param);
+	/* a slot held in the newest service on the listener after a reload */
+	if(param->newsrv){
+		slotrelease(param->newsrv);
+		param->newsrv = NULL;
+	}
 #endif
 	/* the sockets were closed and the filters cleared with it */
 	param->sostate = NULL;

src/sockmap.c

@@ -268,10 +268,10 @@ log(logbuf);
 		(param->srv->logdumpcli && (param->statscli64 > param->srv->logdumpcli)))
 			dolog(param, NULL);
 
-	if(param->version < param->srv->config->version){
+	/* a reload: the client goes by the newest configuration's rules */
-		if(!param->srv->noforce && (res = (*param->srv->authfunc)(param)) && res != 2) {RETURN(res);}
+	if(param->version < runconf->version){
-		param->paused = param->srv->config->paused;
+		if(param->srv->noforce || !reattachfunc) param->version = runconf->version;
-		param->version = param->srv->config->version;
+		else if((res = (*reattachfunc)(param)) && res != 2) {RETURN(res);}
 	}
 
 	if((param->maxtrafin64 && param->statssrv64 >= param->maxtrafin64) || (param->maxtrafout64 && param->statscli64 >= param->maxtrafout64)){

src/structures.h

@@ -708,6 +708,12 @@ struct clientparam {
 				*prev;
 	void * sostate;
 	struct srvparam *srv;
+	/* After a reload: the newest service on the same listener, one of whose
+	   maxconn slots the client holds; its configuration's rules apply
+	   (PSRV()). */
+	struct srvparam *newsrv;
+	/* the configuration whose connection limits the client counts in */
+	struct extparam *connlimconf;
 	REDIRECTFUNC redirectfunc;
 	BANDLIMFUNC bandlimfunc;
 	TRAFCOUNTFUNC trafcountfunc;

tests/cases/reload.py

@@ -66,34 +66,47 @@ def run(t):
     t.eq(200, as_user("alice", "one"), "after a failed reload alice still gets in")
     t.ne(200, as_user("bob", "two"), "and bob, from the refused file, does not")
 
-    # --- a good file: the new configuration takes over -------------------
+    # --- a good file: the new configuration takes over, and the connections
-    slow = {}
+    # of the old services go by its rules from then on ---------------------
-
+    def download(result):
-    def download():
         # about two seconds: 2 MB in 64 KB blocks, 60 ms apart
         r = t.http(f"http://127.0.0.1:{origin}/data?size=2000000&block=65536&delay=60",
                    proxy=proxy, proxy_auth=("alice", "one"))
-        slow["status"], slow["length"] = r.status, r.length
+        result["status"], result["length"] = r.status, r.length
 
-    th = threading.Thread(target=download)
+    def during_reload(users, port):
-    th.start()
+        result = {}
-    time.sleep(0.3)
+        th = threading.Thread(target=download, args=(result,))
+        th.start()
+        time.sleep(0.3)
+        t.write_config("reload", config(origin, port, adm, users))
+        reload_and_wait()
+        return th, result
 
+    # alice stays: her download goes on under the new service
+    th, kept = during_reload("alice:CL:one bob:CL:two", first)
+    t.eq(200, as_user("bob", "two"), "the new configuration lets bob in")
+    th.join(20)
+    t.eq(2000000, kept.get("length"),
+         "a download of a user the new configuration keeps finishes whole")
+
+    # alice is removed: her download is cut, being re-authenticated
+    th, cut = during_reload("bob:CL:two", first)
+    th.join(20)
+    t.ok("a download of a user the new configuration removes is cut") \
+        if cut.get("length", 0) < 2000000 else \
+        t.fail("a download of a user the new configuration removes is cut",
+               "less than 2000000 bytes", cut.get("length"))
+    t.ne(200, as_user("alice", "one"), "and alice cannot come back")
+
+    # the service moves to another port: the old one closes
     t.write_config("reload", config(origin, second, adm, "bob:CL:two"))
     reload_and_wait()
-    t.ok("the new services start") if t.wait_port(second) else \
+    t.ok("the moved service listens on its new port") if t.wait_port(second) else \
-        t.fail("the new services start", f"port {second} open", "closed")
+        t.fail("the moved service listens on its new port", f"port {second} open", "closed")
-    moved = f"127.0.0.1:{second}"
+    t.eq(200, as_user("bob", "two", f"127.0.0.1:{second}"), "bob gets in there")
-    t.eq(200, as_user("bob", "two", moved), "the new configuration lets bob in")
-    t.ne(200, as_user("alice", "one", moved), "and not alice, who is gone from it")
-
-    # the old service stops accepting, its port closes
     deadline = time.time() + 5
     while time.time() < deadline and t.wait_port(first, timeout=0.2):
         time.sleep(0.2)
-    t.ok("the old service stops listening") if not t.wait_port(first, timeout=0.2) else \
+    t.ok("the old port is closed") if not t.wait_port(first, timeout=0.2) else \
-        t.fail("the old service stops listening", f"port {first} closed", "open")
+        t.fail("the old port is closed", f"port {first} closed", "open")
-
-    th.join(20)
-    t.eq(200, slow.get("status"), "a download through the old service finishes")
-    t.eq(2000000, slow.get("length"), "with the whole answer")