sd-varlink: drop 'user.varlink' xattr from sockets on protocol upgrade

Once a Varlink connection is upgraded to some other protocol it is not a
Varlink socket anymore, hence remove the 'user.varlink' marker xattr
again, so that tools such as the BPF Varlink monitor don't try to
interpret the upgraded protocol's traffic as Varlink messages.

We do this in varlink_handle_upgrade_fds(), which all four upgrade paths
(sync/async, client/server) go through, and which is only reached once
the last Varlink byte has been written (the upgrade reply has been
flushed on the server side; the client's upgrade request was answered)
and read (no buffered input is left), but before any byte of the
upgraded protocol is read or written by us.

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

@@ -187,6 +187,29 @@ static int mark_varlink_socket(int fd, const char *path, const char *role) {
         return 0;
 }
 
+static int unmark_varlink_socket(int fd) {
+        int r;
+
+        assert(fd >= 0);
+
+        /* Removes the 'user.varlink' xattr again from a connection socket, i.e. when the connection is
+         * upgraded to some other protocol, and hence isn't a Varlink socket anymore. */
+
+        r = socket_xattr_supported();
+        if (r < 0)
+                return r;
+        if (r == 0)
+                return -EOPNOTSUPP;
+
+        r = xremovexattr(fd, /* path= */ NULL, AT_EMPTY_PATH, "user.varlink");
+        if (r == -ENODATA) /* Not marked in the first place? */
+                return 0;
+        if (r < 0)
+                return log_debug_errno(r, "Failed to remove 'user.varlink' xattr: %m");
+
+        return 1;
+}
+
 _public_ int sd_varlink_connect_address(sd_varlink **ret, const char *address) {
         _cleanup_(sd_varlink_unrefp) sd_varlink *v = NULL;
         int r;
@@ -1489,6 +1512,15 @@ static int varlink_handle_upgrade_fds(sd_varlink *v, int *ret_input_fd, int *ret
         _cleanup_close_ int input_fd = TAKE_FD(v->stream.input_fd),
                             output_fd = TAKE_FD(v->stream.output_fd);
 
+        /* We get here only once the last Varlink byte has been written (the reply to the upgrade request
+         * on the server side has been flushed, the upgrade request on the client side has been answered)
+         * and read (the upgrade request on the server side, the reply on the client side), and before any
+         * byte of the upgraded protocol is read or written by us. Hence this is the right moment to drop
+         * the 'user.varlink' marker from the socket(s), as they are not speaking Varlink anymore. */
+        (void) unmark_varlink_socket(input_fd);
+        if (output_fd != input_fd)
+                (void) unmark_varlink_socket(output_fd);
+
         /* Pass the connection fds to the caller, it owns them now. Reset to blocking mode
          * since callers of the upgraded protocol will generally expect normal blocking
          * semantics. For bidirectional sockets (input_fd == output_fd), dup the fd so that

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

@@ -26,6 +26,7 @@
 #include "time-util.h"
 #include "tmpfile-util.h"
 #include "varlink-util.h"
+#include "xattr-util.h"
 
 /* Let's pick some high value, that is higher than the largest listen() backlog, but leaves enough room below
    the typical RLIMIT_NOFILE value of 1024 so that we can process both sides of each socket in our
@@ -1209,12 +1210,26 @@ TEST(notify_then_error) {
         ASSERT_OK(sd_event_loop(e));
 }
 
+static void assert_varlink_xattr(int fd, const char *expected) {
+        _cleanup_free_ char *role = NULL;
+
+        if (socket_xattr_supported() <= 0)
+                return;
+
+        if (expected) {
+                ASSERT_OK(fgetxattr_malloc(fd, "user.varlink", &role, /* ret_size= */ NULL));
+                ASSERT_STREQ(role, expected);
+        } else
+                ASSERT_ERROR(fgetxattr_malloc(fd, "user.varlink", &role, /* ret_size= */ NULL), ENODATA);
+}
+
 static int method_upgrade(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
         _cleanup_close_ int input_fd = -EBADF, output_fd = -EBADF;
         int r;
 
         ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE));
         ASSERT_OK_POSITIVE(sd_varlink_wants_reply(link));
+        assert_varlink_xattr(ASSERT_OK(sd_varlink_get_input_fd(link)), "server");
 
         r = sd_varlink_reply_and_upgrade(link, /* parameters= */ NULL, &input_fd, &output_fd);
         if (r < 0)
@@ -1222,6 +1237,10 @@ static int method_upgrade(sd_varlink *link, sd_json_variant *parameters, sd_varl
 
         ASSERT_OK_ZERO(sd_varlink_wants_reply(link));
 
+        /* The connection doesn't speak Varlink anymore, hence the marker must be gone */
+        assert_varlink_xattr(input_fd, /* expected= */ NULL);
+        assert_varlink_xattr(output_fd, /* expected= */ NULL);
+
         /* After upgrade, do raw I/O: read until the client shuts down its write side (giving us a clean
          * EOF), reverse what we got, and write it back. Use suspending I/O so other fibers (the client)
          * can make progress while we're waiting on the socket. */
@@ -1255,12 +1274,15 @@ static int upgrade_client_fiber(void *arg) {
 
         ASSERT_OK(sd_varlink_connect_address(&c, arg));
         ASSERT_OK(sd_varlink_set_description(c, "upgrade-client"));
+        assert_varlink_xattr(ASSERT_OK(sd_varlink_get_input_fd(c)), "client");
 
         ASSERT_OK(sd_varlink_call_and_upgrade(c, "io.test.Upgrade", /* parameters= */ NULL, &o, &error_id, &input_fd, &output_fd));
         ASSERT_NULL(error_id);
         ASSERT_GE(input_fd, 0);
         ASSERT_GE(output_fd, 0);
         ASSERT_NE(input_fd, output_fd); /* library dups for bidirectional sockets */
+        assert_varlink_xattr(input_fd, /* expected= */ NULL);
+        assert_varlink_xattr(output_fd, /* expected= */ NULL);
 
         /* Send a test string, shut down write side so server sees EOF, then read the reversed reply */
         static const char msg[] = "Hello!";
@@ -1359,6 +1381,8 @@ static int async_upgrade_on_upgrade(sd_varlink *link, int input_fd, int output_f
         ASSERT_GE(output_fd, 0);
         ASSERT_NE(input_fd, output_fd);
         ASSERT_FALSE(sd_varlink_is_connected(link));
+        assert_varlink_xattr(input_fd, /* expected= */ NULL);
+        assert_varlink_xattr(output_fd, /* expected= */ NULL);
 
         t->got_upgrade = true;
         t->input_fd = input_fd;
@@ -1436,6 +1460,7 @@ TEST(upgrade_async) {
         sd_varlink_set_userdata(c, &t);
         ASSERT_OK(sd_varlink_bind_reply(c, async_upgrade_on_reply));
         ASSERT_OK(sd_varlink_bind_upgrade(c, async_upgrade_on_upgrade));
+        assert_varlink_xattr(ASSERT_OK(sd_varlink_get_input_fd(c)), "client");
         ASSERT_OK(sd_varlink_invoke_and_upgrade(c, "io.test.Upgrade", /* parameters= */ NULL));
 
         /* No other calls may be pipelined behind an upgrade request */