25026908423

Committed 27 Apr 2026 07:14PM UTC coverage: 71.865% (-0.3%) from 72.175%

Build # 25026908423

Build Type

push

github

Committed by

daandemeyer

Commit Message

udev: don't assert on worker cap after killing a broken idle worker

manager_can_process_event() considers an event processable if either
there is room below children_max to spawn, or an idle worker exists.
When only the latter holds, event_run() picks the idle worker and
tries device_monitor_send(). If that send fails, event_run() SIGKILLs
the worker, marks it WORKER_KILLED and continues the loop. With no
other idle worker available, it falls through to worker_spawn(),
guarded by:

    assert(hashmap_size(manager->workers) < manager->config.children_max);

The just-killed worker is still in manager->workers until its SIGCHLD
is reaped by on_worker_exit(), so at the cap this assertion trips and
udevd aborts:

    Assertion 'hashmap_size(manager->workers) < manager->config.children_max'
    failed at src/udev/udev-manager.c:635, function event_run(). Aborting.

Instead of asserting, bail out when we are already at the worker
limit. The event remains in EVENT_QUEUED; once the killed worker's
SIGCHLD arrives and frees it from the hashmap, on_post() re-runs
event_queue_start() and the event is retried.

Coverage Stats

1 of 1 new or added line in 1 file covered. (100.0%)

7309 existing lines in 125 files now uncovered.

322519 of 448782 relevant lines covered (71.87%)

1173939.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

12.61

/src/ssh-generator/ssh-proxy.c

/* SPDX-License-Identifier: LGPL-2.1-or-later */

#include <unistd.h>

#include "sd-varlink.h"

#include "alloc-util.h"
#include "fd-util.h"
#include "io-util.h"
#include "iovec-util.h"
#include "log.h"
#include "main-func.h"
#include "path-lookup.h"
#include "path-util.h"
#include "socket-util.h"
#include "stat-util.h"
#include "string-util.h"
#include "strv.h"
#include "time-util.h"

#define HEADER_READ_TIMEOUT_USEC (5 * USEC_PER_SEC)

static int process_vsock_cid(unsigned cid, const char *port) {
        int r;

        assert(cid != VMADDR_CID_ANY);
        assert(port);

        union sockaddr_union sa = {
                .vm.svm_cid = cid,
                .vm.svm_family = AF_VSOCK,
        };

        r = vsock_parse_port(port, &sa.vm.svm_port);
        if (r < 0)
                return log_error_errno(r, "Failed to parse vsock port: %s", port);

        _cleanup_close_ int fd = socket(AF_VSOCK, SOCK_STREAM|SOCK_CLOEXEC, 0);
        if (fd < 0)
                return log_error_errno(errno, "Failed to allocate AF_VSOCK socket: %m");

        if (connect(fd, &sa.sa, sockaddr_len(&sa)) < 0)
                return log_error_errno(errno, "Failed to connect to vsock:%u:%u: %m", sa.vm.svm_cid, sa.vm.svm_port);

        /* OpenSSH wants us to send a single byte along with the file descriptor, hence do so */
        r = send_one_fd_iov(STDOUT_FILENO, fd, &iovec_nul_byte, /* iovlen= */ 1, /* flags= */ 0);
        if (r < 0)
                return log_error_errno(r, "Failed to send socket via STDOUT: %m");

        log_debug("Successfully sent AF_VSOCK socket via STDOUT.");
        return 0;
}

static int process_vsock_string(const char *host, const char *port) {
        unsigned cid;
        int r;

        assert(host);
        assert(port);

        r = vsock_parse_cid(host, &cid);
        if (r < 0)
                return log_error_errno(r, "Failed to parse vsock cid: %s", host);

        if (cid == VMADDR_CID_ANY)
                return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Cannot use VMADDR_CID_ANY to connect to a remote host.");

        return process_vsock_cid(cid, port);
}

static int process_unix(const char *path) {
        int r;

        assert(path);

        /* We assume the path is absolute unless it starts with a dot (or is already explicitly absolute) */
        _cleanup_free_ char *prefixed = NULL;
        if (!STARTSWITH_SET(path, "/", "./")) {
                prefixed = strjoin("/", path);
                if (!prefixed)
                        return log_oom();

                path = prefixed;
        }

        _cleanup_close_ int fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0);
        if (fd < 0)
                return log_error_errno(errno, "Failed to allocate AF_UNIX socket: %m");

        r = connect_unix_path(fd, AT_FDCWD, path);
        if (r < 0)
                return log_error_errno(r, "Failed to connect to AF_UNIX socket %s: %m", path);

        r = send_one_fd_iov(STDOUT_FILENO, fd, &iovec_nul_byte, /* iovlen= */ 1, /* flags= */ 0);
        if (r < 0)
                return log_error_errno(r, "Failed to send socket via STDOUT: %m");

        log_debug("Successfully sent AF_UNIX socket via STDOUT.");
        return 0;
}

static int skip_ok_port_res(int fd, const char *path, const char *port) {
        struct timeval oldtv;
        socklen_t oldtv_size = sizeof(oldtv);
        if (getsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &oldtv, &oldtv_size) < 0)
                return log_error_errno(errno, "Failed to get socket receive timeout for %s: %m", path);
        if (oldtv_size != sizeof(oldtv))
                return log_error_errno(SYNTHETIC_ERRNO(EIO), "Unexpected size of socket receive timeout for %s", path);
        if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, TIMEVAL_STORE(HEADER_READ_TIMEOUT_USEC), sizeof(struct timeval)) < 0)
                return log_error_errno(errno, "Failed to set socket receive timeout for %s: %m", path);

        char recv_buf[STRLEN("OK 65535\n")];
        size_t bytes_recv = 0, bytes_avail = 0, pos = 0;
        static const char expected_prefix[] = "OK ";

        for (;;) {
                if (pos >= bytes_avail) {
                        assert(bytes_recv <= bytes_avail);
                        if (bytes_avail >= sizeof(recv_buf)) {
                                /*
                                  Full buffer means that we have peeked as many bytes as possible and not seeing the ending \n .
                                  So the server is believed to not send OK PORT response, and we just pass out the socket to ssh client,
                                  and let it handle the connection.

                                  If we have not received any bytes from the socket buffer, we can safely pass out the socket,
                                  since no change has been made to the socket buffer. Otherwise, if some bytes have been received,
                                  the socket buffer has been changed, the only option is to give up and terminate the connection.
                                  Similar logic applies below when we meet other kinds of unexpected responses.
                                */
                                if (bytes_recv == 0) {
                                        log_debug("Received too many bytes while waiting for OK PORT response from %s\n"
                                                  "Assume the multiplexer is not sending OK PORT.",
                                                  path);
                                        goto passout_fd;
                                }
                                return log_error_errno(SYNTHETIC_ERRNO(EIO), "Received too many bytes while waiting for OK PORT response from %s", path);
                        }
                        if (bytes_avail > bytes_recv) {
                                /*
                                  Discard already peeked bytes before peeking more.

                                  XXX: We cannot use SO_RCVLOWAT to set the minimum number of bytes to be peeked to peek entire
                                       OK PORT response at once to prevent changes to the recving buffer, because SO_RCVLOWAT
                                       does not work on unix sockets with recv(..., MSG_PEEK). Also poll() does not help here,
                                       because poll() returns readable as long as there is any data in the socket buffer for
                                       unix sockets, not respecting SO_RCVLOWAT.

                                  XXX: We could have used SO_PEEK_OFF to continuously peek more data without changing the socket
                                       receive buffer, but this function breaks since Linux 4.3 due to a kernel bug, which is fixed
                                       in Linux 6.18 commit 7bf3a476ce43 ("af_unix: Read sk_peek_offset() again after sleeping in
                                       unix_stream_read_generic()."). It is also not possible to detect whether the kernel is
                                       affected by this bug at runtime.

                                  As a result, we have no other choice but to discard already peeked data here.
                                */
                                ssize_t rlen = recv(fd, recv_buf + bytes_recv, bytes_avail - bytes_recv, /* flags= */ 0);
                                if (rlen < 0)
                                        return log_error_errno(errno, "Failed to discard OK PORT response from %s: %m", path);
                                if ((size_t) rlen != bytes_avail - bytes_recv)
                                        return log_error_errno(SYNTHETIC_ERRNO(EIO), "Short read while discarding OK PORT response from %s", path);
                                log_debug("Successfully discarded %zi bytes of response: %.*s", rlen, (int) rlen, recv_buf + bytes_recv);
                                bytes_recv = bytes_avail;
                        }
                        ssize_t len = recv(fd, recv_buf + bytes_avail, sizeof(recv_buf) - bytes_avail, MSG_PEEK);
                        if (len < 0) {
                                if (errno != EAGAIN)
                                        return log_error_errno(errno, "Failed to receive OK from %s: %m", path);
                                if (bytes_recv == 0) {
                                        log_debug("Timeout while waiting for OK PORT response from %s\n"
                                                  "Assume the multiplexer will not send OK PORT.",
                                                  path);
                                        goto passout_fd;
                                }
                                return log_error_errno(SYNTHETIC_ERRNO(EIO), "Timed out to receive OK PORT from %s", path);

                        }
                        if (len == 0) {
                                log_debug("Connection closed while waiting for OK PORT response from %s", path);
                                if (bytes_recv == 0) {
                                        log_debug("No data received, which means the connecting port is not open.");
                                        return log_error_errno(SYNTHETIC_ERRNO(ECONNREFUSED), "Port %s on %s is not open", port, path);
                                }
                                return log_error_errno(SYNTHETIC_ERRNO(EIO), "Connection closed before full OK PORT response received from %s.", path);
                        }
                        bytes_avail += len;
                }
                assert(pos < bytes_avail);
                if (pos < strlen(expected_prefix) && recv_buf[pos] != expected_prefix[pos]) {
                        if (bytes_recv == 0) {
                                log_debug("Received response does not start with expected OK PORT response from %s\n"
                                          "Assume the multiplexer will not send OK PORT.",
                                          path);
                                goto passout_fd;
                        }
                        return log_error_errno(SYNTHETIC_ERRNO(EIO), "Received invalid response while waiting for OK PORT from %s", path);
                }
                if (recv_buf[pos] == '\n') {
                        pos += 1;
                        break;
                }
                pos += 1;
        }

        assert(pos <= sizeof(recv_buf));
        assert(bytes_recv <= pos);
        if (bytes_recv < pos) {
                ssize_t len = recv(fd, recv_buf + bytes_recv, pos - bytes_recv, /* flags= */ 0);
                if (len < 0)
                        return log_error_errno(errno, "Failed to discard OK PORT response from %s: %m", path);
                if ((size_t) len != pos - bytes_recv)
                        return log_error_errno(SYNTHETIC_ERRNO(EIO), "Short read while discarding OK PORT response from %s", path);
                log_debug("Successfully discarded response from %s: %.*s", path, (int) pos, recv_buf);
        }

passout_fd:
        if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &oldtv, sizeof(oldtv)) < 0)
                return log_error_errno(errno, "Failed to restore socket receive timeout for %s: %m", path);
        return 0;
}

static int process_vsock_mux(const char *path, const char *port) {
        int r;

        assert(path);
        assert(port);

        /* We assume the path is absolute unless it starts with a dot (or is already explicitly absolute) */
        _cleanup_free_ char *prefixed = NULL;
        if (!STARTSWITH_SET(path, "/", "./")) {
                prefixed = strjoin("/", path);
                if (!prefixed)
                        return log_oom();

                path = prefixed;
        }

        _cleanup_close_ int fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0);
        if (fd < 0)
                return log_error_errno(errno, "Failed to allocate AF_UNIX socket: %m");

        r = connect_unix_path(fd, AT_FDCWD, path);
        if (r < 0)
                return log_error_errno(r, "Failed to connect to AF_UNIX socket %s: %m", path);

        /* Based on the protocol as defined here:
         * https://github.com/cloud-hypervisor/cloud-hypervisor/blob/main/docs/vsock.md
         * https://github.com/firecracker-microvm/firecracker/blob/main/docs/vsock.md */
        _cleanup_free_ char *connect_cmd = NULL;
        connect_cmd = strjoin("CONNECT ", port, "\n");
        if (!connect_cmd)
                return log_oom();

        r = loop_write(fd, connect_cmd, SIZE_MAX);
        if (r < 0)
                return log_error_errno(r, "Failed to send CONNECT to %s:%s: %m", path, port);

        r = skip_ok_port_res(fd, path, port);
        if (r < 0)
                return r;

        r = send_one_fd_iov(STDOUT_FILENO, fd, &iovec_nul_byte, /* iovlen= */ 1, /* flags= */ 0);
        if (r < 0)
                return log_error_errno(r, "Failed to send socket via STDOUT: %m");

        log_debug("Successfully sent AF_UNIX socket via STDOUT.");
        return 0;
}

static int fetch_machine(const char *machine, RuntimeScope scope, sd_json_variant **ret) {
        int r;

        assert(machine);
        assert(ret);

        _cleanup_free_ char *addr = NULL;
        r = runtime_directory_generic(scope, "systemd/machine/io.systemd.Machine", &addr);
        if (r < 0)
                return r;

        _cleanup_(sd_varlink_unrefp) sd_varlink *vl = NULL;
        r = sd_varlink_connect_address(&vl, addr);
        if (r < 0)
                return log_error_errno(r, "Failed to connect to machined on %s: %m", addr);

        sd_json_variant *result = NULL;
        const char *error_id;
        r = sd_varlink_callbo(
                        vl,
                        "io.systemd.Machine.List",
                        &result,
                        &error_id,
                        SD_JSON_BUILD_PAIR_STRING("name", machine));
        if (r < 0)
                return log_error_errno(r, "Failed to issue io.systemd.Machine.List() varlink call: %m");
        if (error_id) {
                if (streq(error_id, "io.systemd.Machine.NoSuchMachine"))
                        return -ESRCH;

                r = sd_varlink_error_to_errno(error_id, result); /* If this is a system errno style error, output it with %m */
                if (r != -EBADR)
                        return log_error_errno(r, "Failed to issue io.systemd.Machine.List() varlink call: %m");

                return log_error_errno(r, "Failed to issue io.systemd.Machine.List() varlink call: %s", error_id);
        }

        /* result is a borrowed reference into the varlink connection's receive buffer. Take a real ref so
         * that it survives the cleanup of vl below. */
        *ret = sd_json_variant_ref(result);
        return 0;
}

static int process_machine(const char *machine, const char *port) {
        int r;

        assert(machine);
        assert(port);

        _cleanup_(sd_json_variant_unrefp) sd_json_variant *result = NULL;
        RuntimeScope scope = RUNTIME_SCOPE_USER;
        r = fetch_machine(machine, scope, &result);
        if (r == -ESRCH) {
                scope = RUNTIME_SCOPE_SYSTEM;
                r = fetch_machine(machine, scope, &result);
        }
        if (r < 0)
                return r;

        struct {
                uint32_t cid;
                const char *class;
                const char *service;
        } p = {
                .cid = VMADDR_CID_ANY,
        };

        static const sd_json_dispatch_field dispatch_table[] = {
                { "vSockCid", SD_JSON_VARIANT_UNSIGNED, sd_json_dispatch_uint32,       voffsetof(p, cid),     0                 },
                { "class",    SD_JSON_VARIANT_STRING,   sd_json_dispatch_const_string, voffsetof(p, class),   SD_JSON_MANDATORY },
                { "service",  SD_JSON_VARIANT_STRING,   sd_json_dispatch_const_string, voffsetof(p, service), 0                 },
                {}
        };

        r = sd_json_dispatch(result, dispatch_table, SD_JSON_ALLOW_EXTENSIONS, &p);
        if (r < 0)
                return log_error_errno(r, "Failed to parse Varlink reply: %m");

        if (streq(p.class, "container")) {
                _cleanup_free_ char *path = NULL;

                if (!streq_ptr(p.service, "systemd-nspawn"))
                        return log_error_errno(SYNTHETIC_ERRNO(EMEDIUMTYPE), "Don't know how to SSH into '%s' container %s.", p.service, machine);

                r = runtime_directory_generic(scope, "systemd/nspawn", &path);
                if (r < 0)
                        return log_error_errno(r, "Failed to determine runtime directory: %m");

                if (!path_extend(&path, machine, "unix-export", "ssh"))
                        return log_oom();

                r = is_socket(path);
                if (r < 0)
                        return log_error_errno(r, "Failed to check if '%s' exists and is a socket: %m", path);
                if (r == 0)
                        return log_error_errno(
                                        SYNTHETIC_ERRNO(ENOENT),
                                        "'%s' does not exist or is not a socket, are sshd and systemd-ssh-generator installed and enabled in the container?",
                                        path);

                return process_unix(path);
        }

        if (!streq(p.class, "vm"))
                return log_error_errno(SYNTHETIC_ERRNO(EMEDIUMTYPE), "Don't know how to SSH into machine %s with class '%s'.", machine, p.class);

        if (p.cid == VMADDR_CID_ANY)
                return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Machine %s has no AF_VSOCK CID assigned.", machine);

        return process_vsock_cid(p.cid, port);
}

static char *startswith_sep(const char *s, const char *prefix) {
        const char *p = startswith(s, prefix);

        if (p && IN_SET(*p, '/', '%'))
                return (char*) p + 1;

        return NULL;
}

static int run(int argc, char* argv[]) {

        log_setup();

        if (argc != 3)
                return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Expected two arguments: host and port.");

        const char *host = argv[1], *port = argv[2];

        const char *p = startswith_sep(host, "vsock");
        if (p)
                return process_vsock_string(p, port);

        p = startswith_sep(host, "unix");
        if (p)
                return process_unix(p);

        p = startswith_sep(host, "vsock-mux");
        if (p)
                return process_vsock_mux(p, port);

        p = startswith_sep(host, "machine");
        if (p)
                return process_machine(p, port);

        return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Don't know how to parse host name specification: %s", host);
}

DEFINE_MAIN_FUNCTION(run);

1	/* SPDX-License-Identifier: LGPL-2.1-or-later */
2
3	#include <unistd.h>
4
5	#include "sd-varlink.h"
6
7	#include "alloc-util.h"
8	#include "fd-util.h"
9	#include "io-util.h"
10	#include "iovec-util.h"
11	#include "log.h"
12	#include "main-func.h"
13	#include "path-lookup.h"
14	#include "path-util.h"
15	#include "socket-util.h"
16	#include "stat-util.h"
17	#include "string-util.h"
18	#include "strv.h"
19	#include "time-util.h"
20
21	#define HEADER_READ_TIMEOUT_USEC (5 * USEC_PER_SEC)
22
23	static int process_vsock_cid(unsigned cid, const char *port) {	×
24	int r;	×
25
26	assert(cid != VMADDR_CID_ANY);	×
27	assert(port);	×
28
29	union sockaddr_union sa = {	×
30	.vm.svm_cid = cid,
31	.vm.svm_family = AF_VSOCK,
32	};
33
34	r = vsock_parse_port(port, &sa.vm.svm_port);	×
35	if (r < 0)	×
36	return log_error_errno(r, "Failed to parse vsock port: %s", port);	×
37
38	_cleanup_close_ int fd = socket(AF_VSOCK, SOCK_STREAM\|SOCK_CLOEXEC, 0);	×
39	if (fd < 0)	×
40	return log_error_errno(errno, "Failed to allocate AF_VSOCK socket: %m");	×
41
42	if (connect(fd, &sa.sa, sockaddr_len(&sa)) < 0)	×
43	return log_error_errno(errno, "Failed to connect to vsock:%u:%u: %m", sa.vm.svm_cid, sa.vm.svm_port);	×
44
45	/* OpenSSH wants us to send a single byte along with the file descriptor, hence do so */
46	r = send_one_fd_iov(STDOUT_FILENO, fd, &iovec_nul_byte, /* iovlen= / 1, / flags= */ 0);	×
47	if (r < 0)	×
48	return log_error_errno(r, "Failed to send socket via STDOUT: %m");	×
49
50	log_debug("Successfully sent AF_VSOCK socket via STDOUT.");	×
51	return 0;
52	}
53
54	static int process_vsock_string(const char host, const char port) {	×
55	unsigned cid;	×
56	int r;	×
57
58	assert(host);	×
59	assert(port);	×
60
61	r = vsock_parse_cid(host, &cid);	×
62	if (r < 0)	×
63	return log_error_errno(r, "Failed to parse vsock cid: %s", host);	×
64
65	if (cid == VMADDR_CID_ANY)	×
66	return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Cannot use VMADDR_CID_ANY to connect to a remote host.");	×
67
68	return process_vsock_cid(cid, port);	×
69	}
70
71	static int process_unix(const char *path) {	3✔
72	int r;	3✔
73
74	assert(path);	3✔
75
76	/* We assume the path is absolute unless it starts with a dot (or is already explicitly absolute) */
77	_cleanup_free_ char *prefixed = NULL;	3✔
78	if (!STARTSWITH_SET(path, "/", "./")) {	3✔
79	prefixed = strjoin("/", path);	3✔
80	if (!prefixed)	3✔
81	return log_oom();	×
82
83	path = prefixed;
84	}
85
86	_cleanup_close_ int fd = socket(AF_UNIX, SOCK_STREAM\|SOCK_CLOEXEC, 0);	6✔
87	if (fd < 0)	3✔
88	return log_error_errno(errno, "Failed to allocate AF_UNIX socket: %m");	×
89
90	r = connect_unix_path(fd, AT_FDCWD, path);	3✔
91	if (r < 0)	3✔
92	return log_error_errno(r, "Failed to connect to AF_UNIX socket %s: %m", path);	×
93
94	r = send_one_fd_iov(STDOUT_FILENO, fd, &iovec_nul_byte, /* iovlen= / 1, / flags= */ 0);	3✔
95	if (r < 0)	3✔
96	return log_error_errno(r, "Failed to send socket via STDOUT: %m");	×
97
98	log_debug("Successfully sent AF_UNIX socket via STDOUT.");	3✔
99	return 0;
100	}
101
102	static int skip_ok_port_res(int fd, const char path, const char port) {	×
103	struct timeval oldtv;	×
104	socklen_t oldtv_size = sizeof(oldtv);	×
105	if (getsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &oldtv, &oldtv_size) < 0)	×
106	return log_error_errno(errno, "Failed to get socket receive timeout for %s: %m", path);	×
107	if (oldtv_size != sizeof(oldtv))	×
108	return log_error_errno(SYNTHETIC_ERRNO(EIO), "Unexpected size of socket receive timeout for %s", path);	×
109	if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, TIMEVAL_STORE(HEADER_READ_TIMEOUT_USEC), sizeof(struct timeval)) < 0)	×
110	return log_error_errno(errno, "Failed to set socket receive timeout for %s: %m", path);	×
111
112	char recv_buf[STRLEN("OK 65535\n")];	×
113	size_t bytes_recv = 0, bytes_avail = 0, pos = 0;	×
114	static const char expected_prefix[] = "OK ";	×
115
116	for (;;) {	×
117	if (pos >= bytes_avail) {	×
118	assert(bytes_recv <= bytes_avail);	×
119	if (bytes_avail >= sizeof(recv_buf)) {	×
120	/*
121	Full buffer means that we have peeked as many bytes as possible and not seeing the ending \n .
122	So the server is believed to not send OK PORT response, and we just pass out the socket to ssh client,
123	and let it handle the connection.
124
125	If we have not received any bytes from the socket buffer, we can safely pass out the socket,
126	since no change has been made to the socket buffer. Otherwise, if some bytes have been received,
127	the socket buffer has been changed, the only option is to give up and terminate the connection.
128	Similar logic applies below when we meet other kinds of unexpected responses.
129	*/
130	if (bytes_recv == 0) {	×
131	log_debug("Received too many bytes while waiting for OK PORT response from %s\n"	×
132	"Assume the multiplexer is not sending OK PORT.",
133	path);
134	goto passout_fd;	×
135	}
136	return log_error_errno(SYNTHETIC_ERRNO(EIO), "Received too many bytes while waiting for OK PORT response from %s", path);	×
137	}
138	if (bytes_avail > bytes_recv) {	×
139	/*
140	Discard already peeked bytes before peeking more.
141
142	XXX: We cannot use SO_RCVLOWAT to set the minimum number of bytes to be peeked to peek entire
143	OK PORT response at once to prevent changes to the recving buffer, because SO_RCVLOWAT
144	does not work on unix sockets with recv(..., MSG_PEEK). Also poll() does not help here,
145	because poll() returns readable as long as there is any data in the socket buffer for
146	unix sockets, not respecting SO_RCVLOWAT.
147
148	XXX: We could have used SO_PEEK_OFF to continuously peek more data without changing the socket
149	receive buffer, but this function breaks since Linux 4.3 due to a kernel bug, which is fixed
150	in Linux 6.18 commit 7bf3a476ce43 ("af_unix: Read sk_peek_offset() again after sleeping in
151	unix_stream_read_generic()."). It is also not possible to detect whether the kernel is
152	affected by this bug at runtime.
153
154	As a result, we have no other choice but to discard already peeked data here.
155	*/
156	ssize_t rlen = recv(fd, recv_buf + bytes_recv, bytes_avail - bytes_recv, /* flags= */ 0);	×
157	if (rlen < 0)	×
158	return log_error_errno(errno, "Failed to discard OK PORT response from %s: %m", path);	×
159	if ((size_t) rlen != bytes_avail - bytes_recv)	×
160	return log_error_errno(SYNTHETIC_ERRNO(EIO), "Short read while discarding OK PORT response from %s", path);	×
161	log_debug("Successfully discarded %zi bytes of response: %.*s", rlen, (int) rlen, recv_buf + bytes_recv);	×
162	bytes_recv = bytes_avail;
163	}
164	ssize_t len = recv(fd, recv_buf + bytes_avail, sizeof(recv_buf) - bytes_avail, MSG_PEEK);	×
165	if (len < 0) {	×
166	if (errno != EAGAIN)	×
167	return log_error_errno(errno, "Failed to receive OK from %s: %m", path);	×
168	if (bytes_recv == 0) {	×
169	log_debug("Timeout while waiting for OK PORT response from %s\n"	×
170	"Assume the multiplexer will not send OK PORT.",
171	path);
172	goto passout_fd;	×
173	}
174	return log_error_errno(SYNTHETIC_ERRNO(EIO), "Timed out to receive OK PORT from %s", path);	×
175
176	}
177	if (len == 0) {	×
178	log_debug("Connection closed while waiting for OK PORT response from %s", path);	×
179	if (bytes_recv == 0) {	×
180	log_debug("No data received, which means the connecting port is not open.");	×
181	return log_error_errno(SYNTHETIC_ERRNO(ECONNREFUSED), "Port %s on %s is not open", port, path);	×
182	}
183	return log_error_errno(SYNTHETIC_ERRNO(EIO), "Connection closed before full OK PORT response received from %s.", path);	×
184	}
185	bytes_avail += len;	×
186	}
187	assert(pos < bytes_avail);	×
188	if (pos < strlen(expected_prefix) && recv_buf[pos] != expected_prefix[pos]) {	×
189	if (bytes_recv == 0) {	×
190	log_debug("Received response does not start with expected OK PORT response from %s\n"	×
191	"Assume the multiplexer will not send OK PORT.",
192	path);
193	goto passout_fd;	×
194	}
195	return log_error_errno(SYNTHETIC_ERRNO(EIO), "Received invalid response while waiting for OK PORT from %s", path);	×
196	}
197	if (recv_buf[pos] == '\n') {	×
198	pos += 1;	×
199	break;	×
200	}
201	pos += 1;	×
202	}
203
204	assert(pos <= sizeof(recv_buf));	×
205	assert(bytes_recv <= pos);	×
206	if (bytes_recv < pos) {	×
207	ssize_t len = recv(fd, recv_buf + bytes_recv, pos - bytes_recv, /* flags= */ 0);	×
208	if (len < 0)	×
209	return log_error_errno(errno, "Failed to discard OK PORT response from %s: %m", path);	×
210	if ((size_t) len != pos - bytes_recv)	×
211	return log_error_errno(SYNTHETIC_ERRNO(EIO), "Short read while discarding OK PORT response from %s", path);	×
212	log_debug("Successfully discarded response from %s: %.*s", path, (int) pos, recv_buf);	×
213	}
214
215	passout_fd:	×
216	if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &oldtv, sizeof(oldtv)) < 0)	×
217	return log_error_errno(errno, "Failed to restore socket receive timeout for %s: %m", path);	×
218	return 0;
219	}
220
221	static int process_vsock_mux(const char path, const char port) {	×
222	int r;	×
223
224	assert(path);	×
225	assert(port);	×
226
227	/* We assume the path is absolute unless it starts with a dot (or is already explicitly absolute) */
228	_cleanup_free_ char *prefixed = NULL;	×
229	if (!STARTSWITH_SET(path, "/", "./")) {	×
230	prefixed = strjoin("/", path);	×
231	if (!prefixed)	×
232	return log_oom();	×
233
234	path = prefixed;
235	}
236
237	_cleanup_close_ int fd = socket(AF_UNIX, SOCK_STREAM\|SOCK_CLOEXEC, 0);	×
238	if (fd < 0)	×
239	return log_error_errno(errno, "Failed to allocate AF_UNIX socket: %m");	×
240
241	r = connect_unix_path(fd, AT_FDCWD, path);	×
242	if (r < 0)	×
243	return log_error_errno(r, "Failed to connect to AF_UNIX socket %s: %m", path);	×
244
245	/* Based on the protocol as defined here:
246	* https://github.com/cloud-hypervisor/cloud-hypervisor/blob/main/docs/vsock.md
247	* https://github.com/firecracker-microvm/firecracker/blob/main/docs/vsock.md */
248	_cleanup_free_ char *connect_cmd = NULL;	×
249	connect_cmd = strjoin("CONNECT ", port, "\n");	×
250	if (!connect_cmd)	×
251	return log_oom();	×
252
253	r = loop_write(fd, connect_cmd, SIZE_MAX);	×
254	if (r < 0)	×
255	return log_error_errno(r, "Failed to send CONNECT to %s:%s: %m", path, port);	×
256
257	r = skip_ok_port_res(fd, path, port);	×
258	if (r < 0)	×
259	return r;
260
261	r = send_one_fd_iov(STDOUT_FILENO, fd, &iovec_nul_byte, /* iovlen= / 1, / flags= */ 0);	×
262	if (r < 0)	×
263	return log_error_errno(r, "Failed to send socket via STDOUT: %m");	×
264
265	log_debug("Successfully sent AF_UNIX socket via STDOUT.");	×
266	return 0;
267	}
268
269	static int fetch_machine(const char machine, RuntimeScope scope, sd_json_variant *ret) {	×
270	int r;	×
271
272	assert(machine);	×
273	assert(ret);	×
274
275	_cleanup_free_ char *addr = NULL;	×
276	r = runtime_directory_generic(scope, "systemd/machine/io.systemd.Machine", &addr);	×
277	if (r < 0)	×
278	return r;
279
280	_cleanup_(sd_varlink_unrefp) sd_varlink *vl = NULL;	×
281	r = sd_varlink_connect_address(&vl, addr);	×
282	if (r < 0)	×
283	return log_error_errno(r, "Failed to connect to machined on %s: %m", addr);	×
284
285	sd_json_variant *result = NULL;	×
286	const char *error_id;	×
287	r = sd_varlink_callbo(	×
288	vl,
289	"io.systemd.Machine.List",
290	&result,
291	&error_id,
292	SD_JSON_BUILD_PAIR_STRING("name", machine));
293	if (r < 0)	×
294	return log_error_errno(r, "Failed to issue io.systemd.Machine.List() varlink call: %m");	×
295	if (error_id) {	×
296	if (streq(error_id, "io.systemd.Machine.NoSuchMachine"))	×
297	return -ESRCH;
298
299	r = sd_varlink_error_to_errno(error_id, result); /* If this is a system errno style error, output it with %m */	×
300	if (r != -EBADR)	×
301	return log_error_errno(r, "Failed to issue io.systemd.Machine.List() varlink call: %m");	×
302
303	return log_error_errno(r, "Failed to issue io.systemd.Machine.List() varlink call: %s", error_id);	×
304	}
305
306	/* result is a borrowed reference into the varlink connection's receive buffer. Take a real ref so
307	* that it survives the cleanup of vl below. */
308	*ret = sd_json_variant_ref(result);	×
309	return 0;	×
310	}
311
312	static int process_machine(const char machine, const char port) {	×
313	int r;	×
314
315	assert(machine);	×
316	assert(port);	×
317
318	_cleanup_(sd_json_variant_unrefp) sd_json_variant *result = NULL;	×
319	RuntimeScope scope = RUNTIME_SCOPE_USER;	×
320	r = fetch_machine(machine, scope, &result);	×
321	if (r == -ESRCH) {	×
322	scope = RUNTIME_SCOPE_SYSTEM;	×
323	r = fetch_machine(machine, scope, &result);	×
324	}
325	if (r < 0)	×
326	return r;
327
328	struct {	×
329	uint32_t cid;
330	const char *class;
331	const char *service;
332	} p = {	×
333	.cid = VMADDR_CID_ANY,
334	};
335
336	static const sd_json_dispatch_field dispatch_table[] = {	×
337	{ "vSockCid", SD_JSON_VARIANT_UNSIGNED, sd_json_dispatch_uint32, voffsetof(p, cid), 0 },
338	{ "class", SD_JSON_VARIANT_STRING, sd_json_dispatch_const_string, voffsetof(p, class), SD_JSON_MANDATORY },
339	{ "service", SD_JSON_VARIANT_STRING, sd_json_dispatch_const_string, voffsetof(p, service), 0 },
340	{}
341	};
342
343	r = sd_json_dispatch(result, dispatch_table, SD_JSON_ALLOW_EXTENSIONS, &p);	×
344	if (r < 0)	×
UNCOV 345	return log_error_errno(r, "Failed to parse Varlink reply: %m");	×
346
347	if (streq(p.class, "container")) {	×
UNCOV 348	_cleanup_free_ char *path = NULL;	×
349
350	if (!streq_ptr(p.service, "systemd-nspawn"))	×
UNCOV 351	return log_error_errno(SYNTHETIC_ERRNO(EMEDIUMTYPE), "Don't know how to SSH into '%s' container %s.", p.service, machine);	×
352
353	r = runtime_directory_generic(scope, "systemd/nspawn", &path);	×
354	if (r < 0)	×
UNCOV 355	return log_error_errno(r, "Failed to determine runtime directory: %m");	×
356
357	if (!path_extend(&path, machine, "unix-export", "ssh"))	×
UNCOV 358	return log_oom();	×
359
360	r = is_socket(path);	×
361	if (r < 0)	×
362	return log_error_errno(r, "Failed to check if '%s' exists and is a socket: %m", path);	×
363	if (r == 0)	×
UNCOV 364	return log_error_errno(	×
365	SYNTHETIC_ERRNO(ENOENT),
366	"'%s' does not exist or is not a socket, are sshd and systemd-ssh-generator installed and enabled in the container?",
367	path);
368
UNCOV 369	return process_unix(path);	×
370	}
371
372	if (!streq(p.class, "vm"))	×
UNCOV 373	return log_error_errno(SYNTHETIC_ERRNO(EMEDIUMTYPE), "Don't know how to SSH into machine %s with class '%s'.", machine, p.class);	×
374
375	if (p.cid == VMADDR_CID_ANY)	×
UNCOV 376	return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Machine %s has no AF_VSOCK CID assigned.", machine);	×
377
UNCOV 378	return process_vsock_cid(p.cid, port);	×
379	}
380
381	static char startswith_sep(const char s, const char *prefix) {	6✔
382	const char *p = startswith(s, prefix);	6✔
383
384	if (p && IN_SET(*p, '/', '%'))	6✔
385	return (char*) p + 1;	3✔
386
387	return NULL;
388	}
389
390	static int run(int argc, char* argv[]) {	3✔
391
392	log_setup();	3✔
393
394	if (argc != 3)	3✔
UNCOV 395	return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Expected two arguments: host and port.");	×
396
397	const char host = argv[1], port = argv[2];	3✔
398
399	const char *p = startswith_sep(host, "vsock");	3✔
400	if (p)	3✔
UNCOV 401	return process_vsock_string(p, port);	×
402
403	p = startswith_sep(host, "unix");	3✔
404	if (p)	3✔
405	return process_unix(p);	3✔
406
407	p = startswith_sep(host, "vsock-mux");	×
408	if (p)	×
UNCOV 409	return process_vsock_mux(p, port);	×
410
411	p = startswith_sep(host, "machine");	×
412	if (p)	×
UNCOV 413	return process_machine(p, port);	×
414
UNCOV 415	return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Don't know how to parse host name specification: %s", host);	×
416	}
417
418	DEFINE_MAIN_FUNCTION(run);	3✔

systemd / systemd / 25026908423

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous