handle pmuv

Signed-off-by: mohitkhullar <[email protected]>

Author: mohitkhullar

net/net_evbuffer.c

@@ -814,6 +814,7 @@ static void host_connected_info_free(struct host_connected_info *info)
 struct accept_info {
     int fd;
     int secure; /* whether connection is routed from a secure pmux port */
+    int pmuv;
     int badrte;
     struct evbuffer *buf;
     int to_len;
@@ -840,7 +841,7 @@ static void do_read(int, short, void *);
 static void accept_info_free(struct accept_info *);
 
 static void accept_info_new(netinfo_type *netinfo_ptr, struct sockaddr_in *addr, int fd, int secure,
-                                           int badrte)
+                                           int badrte, int pmuv)
 {
     check_base_thd();
     ++pending_connections;
@@ -856,6 +857,7 @@ static void accept_info_new(netinfo_type *netinfo_ptr, struct sockaddr_in *addr,
     a->fd = fd;
     a->secure = secure;
     a->badrte = badrte;
+    a->pmuv = pmuv;
     a->ev = event_new(base, fd, EV_READ, do_read, a);
     event_add(a->ev, NULL);
 }
@@ -2928,6 +2930,75 @@ static int should_reject_request(uint8_t first_byte)
     return check_appsock_limit(pending_connections, is_admin);
 }
 
+/* PMUV REQUEST/RESPONSE */
+enum request { V_WHO = 1, V_NAK = 2, V_ACK = 3 };
+enum response { V_NONE = 0, V_ID = 1 };
+
+static int portmux_server_side_validation(struct accept_info *a, char request) {
+    ssize_t rc;
+    char response;
+    switch(request) {
+        case V_WHO:
+            response = V_ID;
+            rc = write(a->fd, &response, 1);
+            if (rc != 1) {
+                logmsg(LOGMSG_ERROR, "%s:write failure fd:%d rc:%zd (%s)\n", __func__,
+                       a->fd, rc, strerror(errno));
+                return 1;
+            }
+
+            uint32_t size = 0;
+            uint32_t features = htonl(0);
+            /* format the triplet into a string.
+             * snprintf returns the number of characters printed, excluding
+             * the null byte used to end output to strings - we want that too.
+             */
+            char buf[512];
+            netinfo_type *netinfo_ptr = a->netinfo_ptr;
+            uint32_t buflen = snprintf(buf, sizeof(buf), "%s/%s/%s", netinfo_ptr->app,
+                                       netinfo_ptr->service, netinfo_ptr->instance) +
+                              1;
+            if (buflen > sizeof(buf)) {
+                return 1;
+            }
+            evbuffer_drain(a->buf, 1);
+            /* Logic from portmuxuser.c portmux_server_side_validation */
+            size_t total_len = 0;
+            struct evbuffer_iovec iov[3];
+            iov[0].iov_base = (void *)&size;
+            iov[0].iov_len = sizeof(uint32_t);
+            iov[1].iov_base = (void *)&features;
+            iov[1].iov_len = sizeof(uint32_t);
+            iov[2].iov_base = (void *)buf;
+            iov[2].iov_len = buflen;
+            size = iov[1].iov_len + iov[2].iov_len;
+            total_len = size + iov[0].iov_len;
+            size = htonl(size);
+            evbuffer_add_iovec(a->buf, iov, 3);
+            rc = evbuffer_write(a->buf, a->fd);
+            if (rc != total_len) {
+                logmsg(LOGMSG_ERROR, "%s:unable to write fd:%d rc:%zd (%s)\n", __func__,
+                       a->fd, rc, strerror(errno));
+                return 1;
+            }
+            /* Still in pmuv validation mode, don't reset the flag */
+            a->ev = event_new(base, a->fd, EV_READ, do_read, a);
+            event_add(a->ev, NULL);
+            return 0;
+        case V_NAK:
+            /* Validation failed */
+            return 1;
+        case V_ACK:
+            /* Validation done */
+            a->pmuv = 0;
+            a->ev = event_new(base, a->fd, EV_READ, do_read, a);
+            event_add(a->ev, NULL);
+            return 0;
+        default:
+            return 1;
+    }
+}
+
 static void do_read(int fd, short what, void *data)
 {
     check_base_thd();
@@ -2941,6 +3012,18 @@ static void do_read(int fd, short what, void *data)
     }
     uint8_t first_byte;
     evbuffer_copyout(buf, &first_byte, 1);
+    if (a->pmuv) {
+        a->buf = buf;
+        evbuffer_drain(buf, 1);
+        if (portmux_server_side_validation(a, first_byte) != 0) {
+            evbuffer_free(buf);
+            shutdown_close(fd);
+        } else {
+            evbuffer_free(buf);
+            a->buf = NULL;
+        }
+        return;
+    }
     if (first_byte == 0) { /* replication or offloadsql */
         evbuffer_drain(buf, 1);
         a->buf = buf;
@@ -2967,7 +3050,7 @@ static void do_read(int fd, short what, void *data)
     if (badrte) {
         // Failed to handle fd because of badrte
         rem_appsock_connection_evbuffer();
-        accept_info_new(netinfo_ptr, &ss, fd, secure, 1);
+        accept_info_new(netinfo_ptr, &ss, fd, secure, 1, 0);
         return;
     }
     if (handle_appsock(netinfo_ptr, &ss, first_byte, buf, fd) != 0) {
@@ -2977,14 +3060,24 @@ static void do_read(int fd, short what, void *data)
     }
 }
 
+static void accept_pmuv(struct evconnlistener *listener, evutil_socket_t fd,
+                      struct sockaddr *addr, int len, void *data)
+{
+    check_base_thd();
+    struct net_info *n = data;
+    netinfo_type *netinfo_ptr = n->netinfo_ptr;
+    netinfo_ptr->num_accepts++;
+    accept_info_new(netinfo_ptr, (struct sockaddr_in *)addr, fd, 0, 0, 1);
+}
+
 static void accept_cb(struct evconnlistener *listener, evutil_socket_t fd,
                       struct sockaddr *addr, int len, void *data)
 {
     check_base_thd();
     struct net_info *n = data;
     netinfo_type *netinfo_ptr = n->netinfo_ptr;
     netinfo_ptr->num_accepts++;
-    accept_info_new(netinfo_ptr, (struct sockaddr_in *)addr, fd, 0, 0);
+    accept_info_new(netinfo_ptr, (struct sockaddr_in *)addr, fd, 0, 0, 0);
 }
 
 static void accept_secure(struct evconnlistener *listener, evutil_socket_t fd, struct sockaddr *addr, int len,
@@ -2994,7 +3087,7 @@ static void accept_secure(struct evconnlistener *listener, evutil_socket_t fd, s
     struct net_info *n = data;
     netinfo_type *netinfo_ptr = n->netinfo_ptr;
     netinfo_ptr->num_accepts++;
-    accept_info_new(netinfo_ptr, (struct sockaddr_in *)addr, fd, 1, 0);
+    accept_info_new(netinfo_ptr, (struct sockaddr_in *)addr, fd, 1, 0, 0);
 }
 
 static void accept_error_cb(struct evconnlistener *listener, void *data)
@@ -3027,7 +3120,7 @@ static void reopen_unix(int fd, struct net_info *n)
      (m)->cmsg_level == SOL_SOCKET && (m)->cmsg_type == SCM_RIGHTS)
 #endif
 
-static int recvfd(int pmux_fd, int *secure)
+static int recvfd(int pmux_fd, int *secure, int *pmuv)
 {
     int newfd = -1;
     char buf[sizeof("pmux") - 1];
@@ -3081,13 +3174,15 @@ static int recvfd(int pmux_fd, int *secure)
     }
     newfd = *(int *)CMSG_DATA(m);
 #   endif
-    if (memcmp(buf, "pmux", sizeof(buf)) != 0 && memcmp(buf, "spmu", sizeof(buf)) != 0) {
+    if (memcmp(buf, "pmux", sizeof(buf)) != 0 && memcmp(buf, "spmu", sizeof(buf)) != 0 &&
+        memcmp(buf, "pmuv", sizeof(buf))) {
         shutdown_close(newfd);
         logmsg(LOGMSG_ERROR, "%s:recvmsg pmux_fd:%d unexpected msg:%.*s\n", __func__,
                pmux_fd, (int)sizeof(buf), buf);
         return -1;
     }
     *secure = (memcmp(buf, "spmu", sizeof(buf)) == 0);
+    *pmuv = (memcmp(buf, "pmuv", sizeof(buf)) == 0);
     return newfd;
 }
 
@@ -3096,14 +3191,21 @@ static void do_recvfd(int pmux_fd, short what, void *data)
     check_base_thd();
     struct net_info *n = data;
     int secure;
-    int newfd = recvfd(pmux_fd, &secure);
+    int pmuv;
+    int newfd = recvfd(pmux_fd, &secure, &pmuv);
     if (newfd == 0) return;
     if (newfd < 0) {
         reopen_unix(pmux_fd, n);
         return;
     }
     make_server_socket(newfd);
-    ssize_t rc = write(newfd, "0\n", 2);
+    ssize_t rc;
+    if (pmuv) {
+        /* Validate */
+        rc = write(newfd, "1\n", 2);
+    } else {
+        rc = write(newfd, "0\n", 2);
+    }
     if (rc != 2) {
         logmsg(LOGMSG_ERROR, "%s:write pmux_fd:%d rc:%zd (%s)\n", __func__, pmux_fd, rc, strerror(errno));
         shutdown_close(newfd);
@@ -3113,10 +3215,12 @@ static void do_recvfd(int pmux_fd, short what, void *data)
     struct sockaddr *addr = (struct sockaddr *)&saddr;
     socklen_t addrlen = sizeof(saddr);
     getpeername(newfd, addr, &addrlen);
-    if (!secure)
-        accept_cb(NULL, newfd, addr, addrlen, n);
-    else
+    if (secure)
         accept_secure(NULL, newfd, addr, addrlen, n);
+    else if (pmuv)
+        accept_pmuv(NULL, newfd, addr, addrlen, n);
+    else
+        accept_cb(NULL, newfd, addr, addrlen, n);
 }
 
 static int process_reg_reply(char *res, struct net_info *n, int unix_fd)