sd-varlink: use MSG_PEEK for protocol_upgrade connections

When there is a potential protocol upgrade we need to be careful that
we do not read beyond our json message as the custom protocol may be
anything. This was archived via a byte-by-byte read. This is of course
very inefficient. So this commit moves to use MSG_PEEK to find the
boundary of the json message instead. This makes the performance hit
a lot smaller.

Thanks to Lennart for suggesting this.

src/libsystemd/sd-varlink/sd-varlink.c

@@ -844,10 +844,49 @@ static int varlink_write(sd_varlink *v) {
 
 #define VARLINK_FDS_MAX (16U*1024U)
 
+static bool varlink_may_protocol_upgrade(sd_varlink *v) {
+        return v->protocol_upgrade || (v->server && FLAGS_SET(v->server->flags, SD_VARLINK_SERVER_UPGRADABLE));
+}
+
+/* When a protocol upgrade might happen, peek at the socket data to find the \0 message
+ * boundary and return a read size that won't consume past it. This prevents over-reading
+ * raw post-upgrade data into the varlink input buffer. Falls back to byte-by-byte for
+ * non-socket fds where MSG_PEEK is not available. */
+static ssize_t varlink_peek_upgrade_boundary(sd_varlink *v, void *p, size_t rs) {
+        assert(v);
+
+        if (!varlink_may_protocol_upgrade(v))
+                return rs;
+
+        if (v->prefer_read)
+                return 1;
+
+        ssize_t peeked = recv(v->input_fd, p, rs, MSG_PEEK|MSG_DONTWAIT);
+        if (peeked < 0) {
+                if (errno == ENOTSOCK) {
+                        v->prefer_read = true;
+                        return 1;  /* Not a socket, fall back to byte-to-byte */
+                } else if (!ERRNO_IS_TRANSIENT(errno))
+                        return -errno;
+
+                /* Transient error, this should not happen but fall back to byte-to-byte */
+                return 1;
+        }
+        /* EOF, the real recv() will also get it so what we return does not matter */
+        if (peeked == 0)
+                return rs;
+
+        void *nul_chr = memchr(p, 0, peeked);
+        if (nul_chr)
+                return (ssize_t) ((char*) nul_chr - (char*) p) + 1;
+
+        return peeked;
+}
+
 static int varlink_read(sd_varlink *v) {
         struct iovec iov;
         struct msghdr mh;
-        size_t rs;
+        ssize_t rs;
         ssize_t n;
         void *p;
 
@@ -895,11 +934,14 @@ static int varlink_read(sd_varlink *v) {
 
         p = v->input_buffer + v->input_buffer_index + v->input_buffer_size;
 
-        /* When a protocol upgrade is requested we can't consume any post-upgrade data from the socket buffer */
-        if (v->protocol_upgrade)
-                rs = 1;
-        else
-                rs = MALLOC_SIZEOF_SAFE(v->input_buffer) - (v->input_buffer_index + v->input_buffer_size);
+        rs = MALLOC_SIZEOF_SAFE(v->input_buffer) - (v->input_buffer_index + v->input_buffer_size);
+
+        /* When a protocol upgrade is requested we can't consume any post-upgrade data from the socket
+         * buffer. Use MSG_PEEK to find the \0 message boundary and only consume up to it. For non-socket
+         * fds (pipes) MSG_PEEK is not available, so fall back to byte-by-byte reading. */
+        rs = varlink_peek_upgrade_boundary(v, p, rs);
+        if (rs < 0)
+                return varlink_log_errno(v, rs, "Failed to peek upgrade boundary: %m");
 
         if (v->allow_fd_passing_input > 0) {
                 iov = IOVEC_MAKE(p, rs);