sapi_posix_backend_netlink.c

Go to the documentation of this file.

#define _POSIX_C_SOURCE 200809L

#include "safeapi/posix_backend/sapi_posix_backend.h"
 
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <netdb.h>
#include <netinet/in.h>
#include <poll.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include "safeapi/oal/memory/sapi_mem_util.h"
#include "safeapi/oal/log/sapi_log.h"
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>

#define POSIX_NETLINK_UDP_HELLO ((uint8_t)0xA5U)
#define POSIX_NETLINK_UDP_HELLO_ACK ((uint8_t)0x5AU)
#define POSIX_NETLINK_UDP_HELLO_PERIOD_MS (20U)
 
typedef struct
{
    int    fd;
    size_t message_size;
} posix_netlink_state_t;
 
/* C99-compatible static assert: posix_netlink_state_t must fit inside
 * sapi_netlink_storage_t's reserved bytes. */
typedef char posix_netlink_storage_fits_[(sizeof(posix_netlink_state_t) <= sizeof(sapi_netlink_storage_t)) ? 1
                                                                                                              : -1];
 
static sapi_status_t set_nonblocking(int fd)
{
    int flags = fcntl(fd, F_GETFL, 0);
 
    if (flags < 0)
    {
        return SAPI_STATUS_INTERNAL_ERROR;
    }
    if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0)
    {
        return SAPI_STATUS_INTERNAL_ERROR;
    }
    return SAPI_STATUS_OK;
}
 
static void compute_deadline(sapi_duration_ms_t timeout_ms, struct timespec *out_deadline)
{
    (void)clock_gettime(CLOCK_MONOTONIC, out_deadline);
    out_deadline->tv_sec += (time_t)(timeout_ms / 1000U);
    out_deadline->tv_nsec += (long)((timeout_ms % 1000U) * 1000000L);
    if (out_deadline->tv_nsec >= 1000000000L)
    {
        out_deadline->tv_sec += 1;
        out_deadline->tv_nsec -= 1000000000L;
    }
}

static int remaining_ms(const struct timespec *deadline)
{
    struct timespec now;
    int64_t delta_ms;
 
    if (clock_gettime(CLOCK_MONOTONIC, &now) != 0)
    {
        return -1;
    }
    delta_ms = ((int64_t)deadline->tv_sec - (int64_t)now.tv_sec) * 1000;
    delta_ms += ((int64_t)deadline->tv_nsec - (int64_t)now.tv_nsec) / 1000000;
    if (delta_ms < 0)
    {
        delta_ms = 0;
    }
    return (int)delta_ms;
}
 
static sapi_status_t open_listen_udp(const sapi_netlink_config_t *config, int *out_fd)
{
    int fd;
    int reuse = 1;
    struct sockaddr_in bind_addr;
    struct sockaddr_in peer_addr;
    struct timespec deadline;
    uint8_t ack_byte = POSIX_NETLINK_UDP_HELLO_ACK;
 
    fd = socket(AF_INET, SOCK_DGRAM, 0);
    if (fd < 0)
    {
        return SAPI_STATUS_INTERNAL_ERROR;
    }
    (void)setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
 
    sapi_mem_set(&bind_addr, 0, sizeof(bind_addr));
    bind_addr.sin_family = AF_INET;
    bind_addr.sin_port = htons(config->port);
    if (config->host == NULL)
    {
        bind_addr.sin_addr.s_addr = INADDR_ANY;
    }
    else if (inet_pton(AF_INET, config->host, &bind_addr.sin_addr) != 1)
    {
        (void)close(fd);
        return SAPI_STATUS_INVALID_PARAM;
    }
 
    if (bind(fd, (struct sockaddr *)&bind_addr, sizeof(bind_addr)) != 0)
    {
        char msg[80];
 
        (void)snprintf(msg, sizeof(msg), "open_listen_udp: bind port %u failed errno=%d", config->port, errno);
        sapi_log_write(SAPI_LOG_LEVEL_ERROR, "NETLINK", msg);
        (void)close(fd);
        return SAPI_STATUS_INTERNAL_ERROR;
    }
    {
        char msg[80];
 
        (void)snprintf(msg, sizeof(msg), "open_listen_udp: bound port %u successfully (fd=%d)", config->port, fd);
        sapi_log_write(SAPI_LOG_LEVEL_INFO, "NETLINK", msg);
    }
    if (set_nonblocking(fd) != SAPI_STATUS_OK)
    {
        (void)close(fd);
        return SAPI_STATUS_INTERNAL_ERROR;
    }
 
    compute_deadline(config->connect_timeout_ms, &deadline);
    for (;;)
    {
        struct pollfd pfd;
        int wait_ms = remaining_ms(&deadline);
        int rc;
        ssize_t n;
        uint8_t hello_byte;
        socklen_t peer_addr_len = sizeof(peer_addr);
 
        pfd.fd = fd;
        pfd.events = POLLIN;
        pfd.revents = 0;
        rc = poll(&pfd, 1, (wait_ms > 0) ? wait_ms : 0);
        if (rc == 0)
        {
            (void)close(fd);
            return SAPI_STATUS_TIMEOUT;
        }
        if (rc < 0)
        {
            if (errno == EINTR)
            {
                if (wait_ms <= 0)
                {
                    (void)close(fd);
                    return SAPI_STATUS_TIMEOUT;
                }
                continue;
            }
            (void)close(fd);
            return SAPI_STATUS_INTERNAL_ERROR;
        }
 
        sapi_mem_set(&peer_addr, 0, sizeof(peer_addr));
        n = recvfrom(fd, &hello_byte, sizeof(hello_byte), 0, (struct sockaddr *)&peer_addr, &peer_addr_len);
        if (n < 0)
        {
            if ((errno == EAGAIN) || (errno == EWOULDBLOCK) || (errno == EINTR))
            {
                continue;
            }
            {
                char msg[64];
 
                (void)snprintf(msg, sizeof(msg), "open_listen_udp: recvfrom failed errno=%d", errno);
                sapi_log_write(SAPI_LOG_LEVEL_ERROR, "NETLINK", msg);
            }
            (void)close(fd);
            return SAPI_STATUS_INTERNAL_ERROR;
        }
        if ((n != (ssize_t)sizeof(hello_byte)) || (hello_byte != POSIX_NETLINK_UDP_HELLO))
        {
            char msg[64];
 
            (void)snprintf(msg, sizeof(msg), "open_listen_udp: unexpected byte %02x, n=%ld", hello_byte, (long)n);
            sapi_log_write(SAPI_LOG_LEVEL_DEBUG, "NETLINK", msg);
            continue; /* not our handshake datagram - keep waiting */
        }
 
        if (connect(fd, (struct sockaddr *)&peer_addr, peer_addr_len) != 0)
        {
            char msg[64];
 
            (void)snprintf(msg, sizeof(msg), "open_listen_udp: connect failed errno=%d", errno);
            sapi_log_write(SAPI_LOG_LEVEL_ERROR, "NETLINK", msg);
            (void)close(fd);
            return SAPI_STATUS_INTERNAL_ERROR;
        }
        sapi_log_write(SAPI_LOG_LEVEL_INFO, "NETLINK", "open_listen_udp: connected to peer, sending ACK");
        break;
    }
 
    if (send(fd, &ack_byte, sizeof(ack_byte), 0) != (ssize_t)sizeof(ack_byte))
    {
        char msg[64];
 
        (void)snprintf(msg, sizeof(msg), "open_listen_udp: send ACK failed errno=%d", errno);
        sapi_log_write(SAPI_LOG_LEVEL_ERROR, "NETLINK", msg);
        (void)close(fd);
        return SAPI_STATUS_INTERNAL_ERROR;
    }
    sapi_log_write(SAPI_LOG_LEVEL_INFO, "NETLINK", "open_listen_udp: ACK sent OK");
 
    /* Deliberately no post-ACK "grace drain" wait here: an earlier version
     * of this function blocked open() for a further fixed window to
     * absorb any HELLO retransmits still in flight from the peer, but
     * that unconditionally delayed every LISTEN-role open() (and, in
     * turn, callers with a tight end-to-end budget - e.g.
     * SAFEAPI_EXAMPLE_AB_CHECKPOINT_MAX_DELAY_MS - built on the old TCP
     * backend's much faster connection establishment). A stray duplicate
     * HELLO that arrives after open() has already returned is instead
     * handled cheaply, on demand, inside backend_receive() below - no
     * fixed delay paid unless one actually shows up. */
    *out_fd = fd;
    return SAPI_STATUS_OK;
}
 
static sapi_status_t open_connect_udp(const sapi_netlink_config_t *config, int *out_fd)
{
    struct addrinfo hints;
    struct addrinfo *resolved = NULL;
    char port_str[8];
    int fd;
    struct timespec deadline;
    struct timespec next_hello_deadline;
    uint8_t hello_byte = POSIX_NETLINK_UDP_HELLO;
 
    (void)snprintf(port_str, sizeof(port_str), "%u", (unsigned int)config->port);
    {
        char msg[96];
 
        (void)snprintf(msg, sizeof(msg), "open_connect_udp: entry host=%s port=%s", config->host, port_str);
        sapi_log_write(SAPI_LOG_LEVEL_DEBUG, "NETLINK", msg);
    }
 
    sapi_mem_set(&hints, 0, sizeof(hints));
    hints.ai_family = AF_INET;
    hints.ai_socktype = SOCK_DGRAM;
    if (getaddrinfo(config->host, port_str, &hints, &resolved) != 0)
    {
        return SAPI_STATUS_INVALID_PARAM;
    }
 
    fd = socket(resolved->ai_family, resolved->ai_socktype, resolved->ai_protocol);
    if (fd < 0)
    {
        freeaddrinfo(resolved);
        return SAPI_STATUS_INTERNAL_ERROR;
    }
    if (set_nonblocking(fd) != SAPI_STATUS_OK)
    {
        (void)close(fd);
        freeaddrinfo(resolved);
        return SAPI_STATUS_INTERNAL_ERROR;
    }
 
    /* connect() on a UDP socket only fixes the default peer address
     * locally - it performs no network I/O and gives no signal that the
     * peer is actually reachable, unlike TCP's connect(). The HELLO/
     * HELLO_ACK exchange below is what makes SAPI_STATUS_OK meaningful. */
    if (connect(fd, resolved->ai_addr, resolved->ai_addrlen) != 0)
    {
        char msg[64];
 
        (void)snprintf(msg, sizeof(msg), "open_connect_udp: connect failed errno=%d", errno);
        sapi_log_write(SAPI_LOG_LEVEL_ERROR, "NETLINK", msg);
        (void)close(fd);
        freeaddrinfo(resolved);
        return SAPI_STATUS_INTERNAL_ERROR;
    }
    {
        char msg[64];
 
        (void)snprintf(msg, sizeof(msg), "open_connect_udp: connect succeeded (fd=%d)", fd);
        sapi_log_write(SAPI_LOG_LEVEL_INFO, "NETLINK", msg);
    }
    freeaddrinfo(resolved);
 
    compute_deadline(config->connect_timeout_ms, &deadline);
    compute_deadline(0U, &next_hello_deadline); /* send the first HELLO immediately */
    sapi_log_write(SAPI_LOG_LEVEL_DEBUG, "NETLINK", "open_connect_udp: entering loop");
    for (;;)
    {
        struct pollfd pfd;
        int wait_ms = remaining_ms(&deadline);
        int hello_wait_ms;
        int poll_wait_ms;
        int rc;
        ssize_t n;
        uint8_t ack_byte;
 
        /* Deliberately no per-iteration log line here: this loop's own
         * HELLO retry cadence (POSIX_NETLINK_UDP_HELLO_PERIOD_MS, 20ms)
         * means a per-iteration DEBUG line fires roughly 50x/second for
         * as long as the peer isn't up yet - real overhead during a
         * connection-establishment window that later turned out to
         * matter for a known-fragile startup negotiation race
         * (ab_gp_channel_crosscompare.c's NEGOTIATION MISMATCH ->
         * REBOOT path, common_config.h's own ADR-029 comment on it) -
         * only genuinely exceptional outcomes (an unhandled send error
         * below) are worth a log line at this cadence. */
        if (wait_ms <= 0)
        {
            (void)close(fd);
            return SAPI_STATUS_TIMEOUT;
        }
 
        hello_wait_ms = remaining_ms(&next_hello_deadline);
        if (hello_wait_ms <= 0)
        {
            ssize_t sres = send(fd, &hello_byte, sizeof(hello_byte), 0);
 
            if ((sres < 0) && (errno != ECONNREFUSED) &&
                (errno != EAGAIN) && (errno != EWOULDBLOCK) && (errno != ENOTCONN) &&
                (errno != EHOSTUNREACH) && (errno != EADDRNOTAVAIL) && (errno != EPIPE) &&
                (errno != ECONNRESET) && (errno != ENETUNREACH))
            {
                char msg[64];
 
                (void)snprintf(msg, sizeof(msg), "open_connect_udp: unhandled send error errno=%d", errno);
                sapi_log_write(SAPI_LOG_LEVEL_ERROR, "NETLINK", msg);
                (void)close(fd);
                return SAPI_STATUS_INTERNAL_ERROR;
            }
            compute_deadline((sapi_duration_ms_t)POSIX_NETLINK_UDP_HELLO_PERIOD_MS, &next_hello_deadline);
            hello_wait_ms = remaining_ms(&next_hello_deadline);
        }
 
        poll_wait_ms = (hello_wait_ms < wait_ms) ? hello_wait_ms : wait_ms;
        pfd.fd = fd;
        pfd.events = POLLIN;
        pfd.revents = 0;
        rc = poll(&pfd, 1, poll_wait_ms);
        if (rc < 0)
        {
            if (errno == EINTR)
            {
                continue;
            }
            (void)close(fd);
            return SAPI_STATUS_INTERNAL_ERROR;
        }
        if (rc == 0)
        {
            continue; /* hello-resend tick or spurious wake - re-evaluate both deadlines */
        }
 
        n = recv(fd, &ack_byte, sizeof(ack_byte), MSG_PEEK);
        if (n < 0)
        {
            if ((errno == EAGAIN) || (errno == EWOULDBLOCK) || (errno == EINTR) || (errno == ECONNREFUSED) ||
                (errno == ENOTCONN) || (errno == EHOSTUNREACH) || (errno == EPIPE) || (errno == ECONNRESET) ||
                (errno == ENETUNREACH))
            {
                continue; /* including ECONNREFUSED/ENOTCONN/ECONNRESET: peer not up yet */
            }
            {
                char msg[64];
 
                (void)snprintf(msg, sizeof(msg), "open_connect_udp: recv failed errno=%d", errno);
                sapi_log_write(SAPI_LOG_LEVEL_ERROR, "NETLINK", msg);
            }
            (void)close(fd);
            return SAPI_STATUS_INTERNAL_ERROR;
        }
        {
            char msg[64];
 
            (void)snprintf(msg, sizeof(msg), "open_connect_udp: peeked n=%ld byte=%02x", (long)n, ack_byte);
            sapi_log_write(SAPI_LOG_LEVEL_DEBUG, "NETLINK", msg);
        }
        if ((n == (ssize_t)sizeof(ack_byte)) && (ack_byte == POSIX_NETLINK_UDP_HELLO_ACK))
        {
            uint8_t consumed_byte;
 
            (void)recv(fd, &consumed_byte, sizeof(consumed_byte), 0); /* actually consume the ACK */
            sapi_log_write(SAPI_LOG_LEVEL_INFO, "NETLINK", "open_connect_udp: consumed ACK successfully");
            break;
        }
        /* Not our handshake ACK (or a duplicate HELLO we ourselves sent
         * echoing back is impossible - this is a unidirectional wait) -
         * leave it queued for the caller's first real receive() and keep
         * waiting within the same deadline. */
    }
 
    *out_fd = fd;
    return SAPI_STATUS_OK;
}
 
static sapi_status_t backend_open(sapi_netlink_storage_t *storage, const sapi_netlink_config_t *config,
                                   sapi_netlink_handle_t *out_handle)
{
    posix_netlink_state_t *state;
    int fd = -1;
    sapi_status_t status;
 
    if ((storage == NULL) || (config == NULL) || (out_handle == NULL))
    {
        return SAPI_STATUS_INVALID_PARAM;
    }
 
    {
        char msg[96];
 
        (void)snprintf(msg, sizeof(msg), "backend_open: role=%d host=%s port=%u", config->role,
                        config->host ? config->host : "NULL", config->port);
        sapi_log_write(SAPI_LOG_LEVEL_DEBUG, "NETLINK", msg);
    }
    status =
        (config->role == SAPI_NETLINK_ROLE_LISTEN) ? open_listen_udp(config, &fd) : open_connect_udp(config, &fd);
    {
        char msg[64];
 
        (void)snprintf(msg, sizeof(msg), "backend_open (role=%d): returned status=%d", config->role, (int)status);
        sapi_log_write(SAPI_LOG_LEVEL_DEBUG, "NETLINK", msg);
    }
    if (status != SAPI_STATUS_OK)
    {
        return status;
    }
 
    state = (posix_netlink_state_t *)(void *)storage;
    state->fd = fd;
    state->message_size = config->message_size;
    *out_handle = (sapi_netlink_handle_t)(void *)storage;
    return SAPI_STATUS_OK;
}
 
static sapi_status_t backend_send(sapi_netlink_handle_t handle, const void *message, size_t message_size,
                                   sapi_duration_ms_t timeout_ms)
{
    posix_netlink_state_t *state = (posix_netlink_state_t *)(void *)handle;
    struct timespec deadline;
 
    if ((state == NULL) || (message == NULL) || (message_size == 0U))
    {
        return SAPI_STATUS_INVALID_PARAM;
    }
 
    compute_deadline(timeout_ms, &deadline);
    for (;;)
    {
        struct pollfd pfd;
        int wait_ms = remaining_ms(&deadline);
        int rc;
        ssize_t n;
 
        if (wait_ms <= 0)
        {
            return SAPI_STATUS_TIMEOUT;
        }
        pfd.fd = state->fd;
        pfd.events = POLLOUT;
        pfd.revents = 0;
        rc = poll(&pfd, 1, wait_ms);
        if (rc == 0)
        {
            return SAPI_STATUS_TIMEOUT;
        }
        if (rc < 0)
        {
            if (errno == EINTR)
            {
                continue;
            }
            return SAPI_STATUS_INTERNAL_ERROR;
        }
 
        n = send(state->fd, message, message_size, 0);
        if (n < 0)
        {
            if ((errno == EAGAIN) || (errno == EWOULDBLOCK) || (errno == EINTR))
            {
                continue;
            }
            if (errno == ECONNREFUSED)
            {
                return SAPI_STATUS_HARDWARE_FAULT;
            }
            return SAPI_STATUS_INTERNAL_ERROR;
        }
        if ((size_t)n != message_size)
        {
            /* A connected SOCK_DGRAM send() is documented atomic-or-fails;
             * a short write here means an unexpected kernel condition, not
             * a partial transfer to resume (there is nothing to loop for,
             * unlike a TCP stream). */
            return SAPI_STATUS_INTERNAL_ERROR;
        }
        return SAPI_STATUS_OK;
    }
}
 
static sapi_status_t backend_receive(sapi_netlink_handle_t handle, void *out_message, size_t buffer_size,
                                      sapi_duration_ms_t timeout_ms)
{
    posix_netlink_state_t *state = (posix_netlink_state_t *)(void *)handle;
    struct timespec deadline;
 
    if ((state == NULL) || (out_message == NULL) || (buffer_size == 0U))
    {
        return SAPI_STATUS_INVALID_PARAM;
    }
 
    compute_deadline(timeout_ms, &deadline);
    for (;;)
    {
        struct pollfd pfd;
        int wait_ms = remaining_ms(&deadline);
        int rc;
        ssize_t n;
 
        if (wait_ms <= 0)
        {
            return SAPI_STATUS_TIMEOUT;
        }
        pfd.fd = state->fd;
        pfd.events = POLLIN;
        pfd.revents = 0;
        rc = poll(&pfd, 1, wait_ms);
        if (rc == 0)
        {
            return SAPI_STATUS_TIMEOUT;
        }
        if (rc < 0)
        {
            if (errno == EINTR)
            {
                continue;
            }
            return SAPI_STATUS_INTERNAL_ERROR;
        }
 
        /* MSG_TRUNC makes the kernel report the datagram's real length
         * even if it exceeded buffer_size, instead of silently truncating
         * it - required to actually detect an oversized datagram below,
         * not just an undersized one. */
#ifdef __APPLE__
        n = recv(state->fd, out_message, buffer_size, 0);
#else
        n = recv(state->fd, out_message, buffer_size, MSG_TRUNC);
#endif
        if (n < 0)
        {
            if ((errno == EAGAIN) || (errno == EWOULDBLOCK) || (errno == EINTR))
            {
                continue;
            }
            if (errno == ECONNREFUSED)
            {
                return SAPI_STATUS_HARDWARE_FAULT;
            }
            return SAPI_STATUS_INTERNAL_ERROR;
        }
        if ((size_t)n != buffer_size)
        {
            /* A lone HELLO/HELLO_ACK byte that outlived the handshake (a
             * retransmit already in flight when the peer stopped
             * resending, or open()'s own final ACK arriving late) is not
             * a wire violation - silently drop it and keep waiting within
             * the same deadline, matching a plain "nothing new yet" wait.
             * message_size is never 1 byte for any real link in this
             * example (smallest is ANSWER_WIRE_SIZE/M136_WIRE_SIZE at 8),
             * so this can never mask a genuine 1-byte application frame. */
            if ((n == (ssize_t)sizeof(uint8_t)) &&
                ((((const uint8_t *)out_message)[0] == POSIX_NETLINK_UDP_HELLO) ||
                 (((const uint8_t *)out_message)[0] == POSIX_NETLINK_UDP_HELLO_ACK)))
            {
                continue;
            }
            /* Anything else of the wrong length: this link's message_size
             * is fixed by config, so a datagram of any other length -
             * truncated, oversized, or a stray 0-byte datagram - is a
             * genuine wire protocol violation, not a value to silently
             * accept partial-length. */
            return SAPI_STATUS_DATA_CORRUPTION;
        }
        return SAPI_STATUS_OK;
    }
}
 
static sapi_status_t backend_close(sapi_netlink_handle_t handle)
{
    posix_netlink_state_t *state = (posix_netlink_state_t *)(void *)handle;
 
    if (state == NULL)
    {
        return SAPI_STATUS_INVALID_PARAM;
    }
    (void)close(state->fd);
    return SAPI_STATUS_OK;
}
 
static const sapi_netlink_backend_t s_posix_netlink_backend = { backend_open, backend_send, backend_receive,
                                                                  backend_close };
 
const sapi_netlink_backend_t *sapi_posix_backend_netlink(void)
{
    return &s_posix_netlink_backend;
}