SOCKS: Support priority groups and weighted-random selection from

multiple proxes.  Bug 1720

3 files changed, 191 insertions, 86 deletions

diff --git a/doc/doc-txt/ChangeLog b/doc/doc-txt/ChangeLog
index d650fc1c0..a199a8d0d 100644
--- a/doc/doc-txt/ChangeLog
+++ b/doc/doc-txt/ChangeLog
@@ -104,6 +104,11 @@ GF/01 Bug 1715: Fix for race condition in exicyclog, where exim could attempt
 
 JH/20 Bug 1701: For MySQL lookups, support MySQL config file option group names.
 
+JH/21 Bug 1720: Add support for priority groups and weighted-random proxy
+      selection for the EXPERIMENTAL_SOCKS feature, via new per-proxy options
+      "pri" and "weight".  Note that the previous implicit priority given by the
+      list order is no longer honoured.
+
 
 Exim version 4.86
 -----------------
diff --git a/doc/doc-txt/experimental-spec.txt b/doc/doc-txt/experimental-spec.txt
index 74614cd3f..4f9504c05 100644
--- a/doc/doc-txt/experimental-spec.txt
+++ b/doc/doc-txt/experimental-spec.txt
@@ -1135,10 +1135,15 @@ These options are currently defined:
 - "pass" sets the authentication password. Default is empty.
 - "port" sets the tcp port number for the proxy. Default is 1080.
 - "tmo" sets a connection timeout in seconds for this proxy. Default is 5.
-
-Proxies from the list are tried in order until
-one responds.  The timeout for the overall connection
-applies to the set of proxied attempts.
+- "pri" specifies a priority for the server within the list, higher
+  values being tried first. The default priority is 1.
+- "weight" specifies a selection bias. Within a priority set servers
+  are queried in a random fashion, weighted by this value. The default
+  value for selection bias is 1.
+
+Proxies from the list are tried according to their priority
+and weight settings until one responds.  The timeout for the
+overall connection applies to the set of proxied attempts.
 
 If events are used, the remote IP/port during a
 tcp:connect event will be that of the proxy.
diff --git a/src/src/transports/smtp_socks.c b/src/src/transports/smtp_socks.c
index a5c4fde10..6d4e072eb 100644
--- a/src/src/transports/smtp_socks.c
+++ b/src/src/transports/smtp_socks.c
@@ -20,6 +20,8 @@
 /* Defaults */
 #define SOCKS_PORT	1080
 #define SOCKS_TIMEOUT	5
+#define SOCKS_WEIGHT	1
+#define SOCKS_PRIORITY	1
 
 #define AUTH_NONE	0
 #define AUTH_NAME	2		/* user/password per RFC 1929 */
@@ -44,21 +46,29 @@ struct socks_err
 
 typedef struct
   {
+  const uschar *	proxy_host;
   uschar		auth_type;	/* RFC 1928 encoding */
   const uschar *	auth_name;
   const uschar *	auth_pwd;
   short			port;
+  BOOL			is_failed;
   unsigned		timeout;
+  unsigned		weight;
+  unsigned		priority;
   } socks_opts;
 
 static void
 socks_option_defaults(socks_opts * sob)
 {
-sob->auth_type = AUTH_NONE;
-sob->auth_name = US"";
-sob->auth_pwd = US"";
-sob->port = SOCKS_PORT;
-sob->timeout = SOCKS_TIMEOUT;
+sob->proxy_host = NULL;
+sob->auth_type =  AUTH_NONE;
+sob->auth_name =  US"";
+sob->auth_pwd =   US"";
+sob->is_failed =  FALSE;
+sob->port =	  SOCKS_PORT;
+sob->timeout =	  SOCKS_TIMEOUT;
+sob->weight =	  SOCKS_WEIGHT;
+sob->priority =   SOCKS_PRIORITY;
 }
 
 static void
@@ -80,6 +90,10 @@ else if (Ustrncmp(opt, "port=", 5) == 0)
   sob->port = atoi(opt + 5);
 else if (Ustrncmp(opt, "tmo=", 4) == 0)
   sob->timeout = atoi(opt + 4);
+else if (Ustrncmp(opt, "pri=", 4) == 0)
+  sob->priority = atoi(opt + 4);
+else if (Ustrncmp(opt, "weight=", 7) == 0)
+  sob->weight = atoi(opt + 7);
 return;
 }
 
@@ -133,6 +147,61 @@ switch(method)
 
 
 
+/* Find a suitable proxy to use from the list.
+Possible common code with spamd_get_server() ?
+
+Return: index into proxy spec array, or -1
+*/
+
+static int
+socks_get_proxy(socks_opts * proxies, unsigned nproxies)
+{
+unsigned int i;
+socks_opts * sd;
+socks_opts * lim = &proxies[nproxies];
+long rnd, weights;
+unsigned pri;
+static BOOL srandomed = FALSE;
+
+if (nproxies == 1)		/* shortcut, if we have only 1 server */
+  return (proxies[0].is_failed ? -1 : 0);
+
+/* init random */
+if (!srandomed)
+  {
+  struct timeval tv;
+  gettimeofday(&tv, NULL);
+  srandom((unsigned int)(tv.tv_usec/1000));
+  srandomed = TRUE;
+  }
+
+/* scan for highest pri */
+for (pri = 0, sd = proxies; sd < lim; sd++)
+  if (!sd->is_failed && sd->priority > pri)
+    pri = sd->priority;
+
+/* get sum of weights at this pri */
+for (weights = 0, sd = proxies; sd < lim; sd++)
+  if (!sd->is_failed && sd->priority == pri)
+    weights += sd->weight;
+if (weights == 0)       /* all servers failed */
+  return -1;
+
+for (rnd = random() % weights, i = 0; i < nproxies; i++)
+  {
+  sd = &proxies[i];
+  if (!sd->is_failed && sd->priority == pri)
+    if ((rnd -= sd->weight) <= 0)
+      return i;
+  }
+
+log_write(0, LOG_MAIN|LOG_PANIC,
+  "%s unknown error (memory/cpu corruption?)", __FUNCTION__);
+return -1;
+}
+
+
+
 /* Make a connection via a socks proxy
 
 Arguments:
@@ -160,6 +229,10 @@ int fd;
 time_t tmo;
 const uschar * state;
 uschar buf[24];
+socks_opts proxies[32];			/* max #proxies handled */
+unsigned nproxies;
+socks_opts * sob;
+unsigned size;
 
 if (!timeout) timeout = 24*60*60;	/* use 1 day for "indefinite" */
 tmo = time(NULL) + timeout;
@@ -171,64 +244,88 @@ if (!(proxy_list = expand_string(ob->socks_proxy)))
   return -1;
   }
 
-/* Loop over proxy list, trying in order until one works */
-while ((proxy_spec = string_nextinlist(&proxy_list, &sep, NULL, 0)))
+/* Read proxy list */
+
+for (nproxies = 0;
+        nproxies < nelem(proxies)
+     && (proxy_spec = string_nextinlist(&proxy_list, &sep, NULL, 0));
+     nproxies++)
   {
-  const uschar * proxy_host;
   int subsep = -' ';
-  host_item proxy;
-  int proxy_af;
-  union sockaddr_46 sin;
-  unsigned size;
-  socks_opts sob;
   const uschar * option;
 
-  if (!(proxy_host = string_nextinlist(&proxy_spec, &subsep, NULL, 0)))
+  socks_option_defaults(sob = &proxies[nproxies]);
+
+  if (!(sob->proxy_host = string_nextinlist(&proxy_spec, &subsep, NULL, 0)))
     {
     /* paniclog config error */
     return -1;
     }
 
   /*XXX consider global options eg. "hide socks_password = wibble" on the tpt */
-  socks_option_defaults(&sob);
-
   /* extract any further per-proxy options */
   while ((option = string_nextinlist(&proxy_spec, &subsep, NULL, 0)))
-    socks_option(&sob, option);
+    socks_option(sob, option);
+  }
+
+/* Try proxies until a connection succeeds */
+
+for(;;)
+  {
+  int idx;
+  host_item proxy;
+  int proxy_af;
+
+  if ((idx = socks_get_proxy(proxies, nproxies)) < 0)
+    {
+    HDEBUG(D_transport|D_acl|D_v) debug_printf("  no proxies left\n");
+    errno = EBUSY;
+    return -1;
+    }
+  sob = &proxies[idx];
 
   /* bodge up a host struct for the proxy */
-  proxy.address = proxy_host;
-  proxy_af = Ustrchr(proxy_host, ':') ? AF_INET6 : AF_INET;
-
-  if ((fd = smtp_sock_connect(&proxy, proxy_af, sob.port,
-	      interface, tb, sob.timeout)) < 0)
-    continue;
-
-  /* Do the socks protocol stuff */
-  /* Send method-selection */
-  state = US"method select";
-  HDEBUG(D_transport|D_acl|D_v) debug_printf("  SOCKS>> 05 01 %02x\n", sob.auth_type);
-  buf[0] = 5; buf[1] = 1; buf[2] = sob.auth_type;
-  if (send(fd, buf, 3, 0) < 0)
-    goto snd_err;
-
-  /* expect method response */
-  if (  !fd_ready(fd, tmo-time(NULL))
-     || read(fd, buf, 2) != 2
-     )
-    goto rcv_err;
-  HDEBUG(D_transport|D_acl|D_v)
-    debug_printf("  SOCKS<< %02x %02x\n", buf[0], buf[1]);
-  if (  buf[0] != 5
-     || socks_auth(fd, buf[1], &sob, tmo) != OK
-     )
-    goto proxy_err;
+  proxy.address = sob->proxy_host;
+  proxy_af = Ustrchr(sob->proxy_host, ':') ? AF_INET6 : AF_INET;
 
+  if ((fd = smtp_sock_connect(&proxy, proxy_af, sob->port,
+	      interface, tb, sob->timeout)) >= 0)
+    break;
+
+  log_write(0, LOG_MAIN, "%s: %s", __FUNCTION__, strerror(errno));
+  sob->is_failed = TRUE;
+  }
+
+/* Do the socks protocol stuff */
+/* Send method-selection */
+
+state = US"method select";
+HDEBUG(D_transport|D_acl|D_v) debug_printf("  SOCKS>> 05 01 %02x\n", sob->auth_type);
+buf[0] = 5; buf[1] = 1; buf[2] = sob->auth_type;
+if (send(fd, buf, 3, 0) < 0)
+  goto snd_err;
+
+/* expect method response */
+
+if (  !fd_ready(fd, tmo-time(NULL))
+   || read(fd, buf, 2) != 2
+   )
+  goto rcv_err;
+HDEBUG(D_transport|D_acl|D_v)
+  debug_printf("  SOCKS<< %02x %02x\n", buf[0], buf[1]);
+if (  buf[0] != 5
+   || socks_auth(fd, buf[1], sob, tmo) != OK
+   )
+  goto proxy_err;
+
+  {
+  union sockaddr_46 sin;
   (void) ip_addr(&sin, host_af, host->address, port);
 
   /* send connect (ipver, ipaddr, port) */
+
   buf[0] = 5; buf[1] = 1; buf[2] = 0; buf[3] = host_af == AF_INET6 ? 4 : 1;
-#if HAVE_IPV6
+  #if HAVE_IPV6
   if (host_af == AF_INET6)
     {
     memcpy(buf+4, &sin.v6.sin6_addr,       sizeof(sin.v6.sin6_addr));
@@ -237,54 +334,52 @@ while ((proxy_spec = string_nextinlist(&proxy_list, &sep, NULL, 0)))
     size = 4+sizeof(sin.v6.sin6_addr)+sizeof(sin.v6.sin6_port);
     }
   else
-#endif
+  #endif
     {
     memcpy(buf+4, &sin.v4.sin_addr.s_addr, sizeof(sin.v4.sin_addr.s_addr));
     memcpy(buf+4+sizeof(sin.v4.sin_addr.s_addr),
       &sin.v4.sin_port, sizeof(sin.v4.sin_port));
     size = 4+sizeof(sin.v4.sin_addr.s_addr)+sizeof(sin.v4.sin_port);
     }
+  }
 
-  state = US"connect";
-  HDEBUG(D_transport|D_acl|D_v)
-    {
-    int i;
-    debug_printf("  SOCKS>>");
-    for (i = 0; i<size; i++) debug_printf(" %02x", buf[i]);
-    debug_printf("\n");
-    }
-  if (send(fd, buf, size, 0) < 0)
-    goto snd_err;
-
-  /* expect conn-reply (success, local(ipver, addr, port))
-  of same length as conn-request, or non-success fail code */
-  if (  !fd_ready(fd, tmo-time(NULL))
-     || (size = read(fd, buf, size)) < 2
-     )
-    goto rcv_err;
-  HDEBUG(D_transport|D_acl|D_v)
-    {
-    int i;
-    debug_printf("  SOCKS>>");
-    for (i = 0; i<size; i++) debug_printf(" %02x", buf[i]);
-    debug_printf("\n");
-    }
-  if (  buf[0] != 5
-     || buf[1] != 0
-     )
-    goto proxy_err;
+state = US"connect";
+HDEBUG(D_transport|D_acl|D_v)
+  {
+  int i;
+  debug_printf("  SOCKS>>");
+  for (i = 0; i<size; i++) debug_printf(" %02x", buf[i]);
+  debug_printf("\n");
+  }
+if (send(fd, buf, size, 0) < 0)
+  goto snd_err;
 
-  /*XXX log proxy outbound addr/port? */
-  HDEBUG(D_transport|D_acl|D_v)
-    debug_printf("  proxy farside local: [%s]:%d\n",
-      host_ntoa(buf[3] == 4 ? AF_INET6 : AF_INET, buf+4, NULL, NULL),
-      ntohs(*((uint16_t *)(buf + (buf[3] == 4 ? 20 : 8)))));
+/* expect conn-reply (success, local(ipver, addr, port))
+of same length as conn-request, or non-success fail code */
 
-  return fd;
+if (  !fd_ready(fd, tmo-time(NULL))
+   || (size = read(fd, buf, size)) < 2
+   )
+  goto rcv_err;
+HDEBUG(D_transport|D_acl|D_v)
+  {
+  int i;
+  debug_printf("  SOCKS>>");
+  for (i = 0; i<size; i++) debug_printf(" %02x", buf[i]);
+  debug_printf("\n");
   }
-
-HDEBUG(D_transport|D_acl|D_v) debug_printf("  no proxies left\n");
-return -1;
+if (  buf[0] != 5
+   || buf[1] != 0
+   )
+  goto proxy_err;
+
+/*XXX log proxy outbound addr/port? */
+HDEBUG(D_transport|D_acl|D_v)
+  debug_printf("  proxy farside local: [%s]:%d\n",
+    host_ntoa(buf[3] == 4 ? AF_INET6 : AF_INET, buf+4, NULL, NULL),
+    ntohs(*((uint16_t *)(buf + (buf[3] == 4 ? 20 : 8)))));
+
+return fd;
 
 snd_err:
   HDEBUG(D_transport|D_acl|D_v) debug_printf("  proxy snd_err %s: %s\n", state, strerror(errno));