jitterbug/server.c

#define _GNU_SOURCE 1

#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#include <string.h>
#include <inttypes.h>
#include "match.h"
#include "try.h"
#include "server.h"
#include "wire.h"
#include "util.h"
#include "log.h"


// https://en.wikipedia.org/wiki/Fowler-Noll-Vo_hash_function#FNV-1a_hash
uint64_t hashmap_hash(const char *bytes, size_t bytes_n, size_t map_len)
{
    uint64_t hash = 0xcbf29ce484222325;

    for (size_t i = 0; i < bytes_n; i++) {
        hash *= 0x100000001b3;
        hash ^= bytes[i];
    }

    return (hash % map_len);
}

void bus_free(bus_t *s)
{
    if (!s) return;
    if (s->sock_fd) close(s->sock_fd);
    // freeing all clients will also free all matches and names
    for (int i = 0; i < BUS_MAX_CLIENTS; i++) {
        bus_client_remove(s, i);
    }
    free(s);
}

bus_t *bus_create(const char *socket_path)
{
    bus_t *s = malloc(sizeof(bus_t));
    if (s == NULL) {
        return NULL;
    }

    s->sock_fd = socket(AF_UNIX, SOCK_STREAM, 0);
    if (s->sock_fd == -1) {
        free(s);
        return NULL;
    }

    struct sockaddr_un name;
    memset(&name, 0, sizeof(name));
    name.sun_family = AF_UNIX;
    strncpy(name.sun_path, socket_path, sizeof(name.sun_path) - 1);

    unlink(socket_path);

    if (bind(s->sock_fd, (const struct sockaddr *)&name, sizeof(name)) == -1) {
        close(s->sock_fd);
        free(s);
        return NULL;
    }
    if (listen(s->sock_fd, BUS_BACKLOG) == -1) {
        close(s->sock_fd);
        free(s);
        return NULL;
    }

    for (int i = 0; i < BUS_MAX_CLIENTS; i++) {
        s->clients[i].fd = -1;
        s->clients[i].owned_name_index = -1;
        s->clients[i].unique_name_index = -1;
        s->clients[i].state = BUS_CLIENT_STATE_NONE;
        s->clients[i].match_count = 0;
        s->fds[i].fd = -1;
        s->fds[i].events = 0;
        s->fds[i].revents = 0;
        for (int j = 0; j < BUS_MAX_MATCH; j++) {
            s->clients[i].matches[j] = NULL;
        }
    }

    for (int i = 0; i < BUS_MAX_NAMES; i++) {
        s->names[i].client_index = -1;
        s->names[i].name = NULL;
    }
    s->names_count = 0;
    s->clients_count = 0;

    s->fds[BUS_MAX_CLIENTS].fd = s->sock_fd;
    s->fds[BUS_MAX_CLIENTS].events = POLLIN;
    s->fd_num = BUS_MAX_CLIENTS + 1;

    return s;
}

int bus_client_add(bus_t *s, int fd)
{
    for (int i = 0; i < BUS_MAX_CLIENTS; i++) {
        if (s->clients[i].fd < 0) {
            s->clients[i].fd = fd;
            s->clients[i].owned_name_index = -1;
            s->clients[i].unique_name_index = -1;
            s->clients[i].state = BUS_CLIENT_STATE_WAIT_AUTH;
            s->fds[i].fd = fd;
            s->fds[i].events = POLLIN;
            s->clients_count++;
            return i;
        }
    }
    return -1;
}

int bus_name_find(bus_t *s, char *name)
{
    int bucket = hashmap_hash(name, strlen(name), BUS_MAX_NAMES);

    for (int i = bucket; i < bucket + 12 && i < BUS_MAX_NAMES; i++) {
        if (s->names[i].name && strcmp(s->names[i].name, name) == 0) {
            return i;
        }
    }

    return -1;
}

bus_client_t *bus_name_find_client(bus_t *s, char *name)
{
    int name_index = bus_name_find(s, name);
    if (name_index < 0) {
        return NULL;
    }
    
    if (s->names[name_index].client_index < 0) {
        return NULL;
    }

    if (s->clients[s->names[name_index].client_index].state != BUS_CLIENT_STATE_READY) {
        return NULL;
    }

    return &s->clients[s->names[name_index].client_index];
}

int bus_name_add(bus_t *s, char *name, int client_index)
{
    if (bus_name_find(s, name) >= 0) {
        return -1;
    }

    int bucket = hashmap_hash(name, strlen(name), BUS_MAX_NAMES);

    for (int i = bucket; i < bucket + 12 && i < BUS_MAX_NAMES; i++) {
        if (s->names[i].client_index == -1) {
            s->names[i].client_index = client_index;
            s->names[i].name = name;
            s->names_count++;
            return i;
        }
    }

    return -1;
}

int bus_client_match_add(bus_t *s, int client_index, char *match)
{
    bus_client_t *c = &s->clients[client_index];

    for (int i = 0; i < BUS_MAX_MATCH; i++) {
        if (!c->matches[i]) {
            c->matches[i] = match_rule_from_string(match);
            c->match_count++;
            return 0;
        }
    }

    return -1;
}

int bus_client_assign_unique_name(bus_t *s, int i)
{
    bus_client_t *c = &s->clients[i];
    if (c->unique_name_index != -1) {
        return -1;
    }
    uint32_t id = 0;
    FILE *urandom_file = fopen("/dev/urandom", "rb");
    if (!urandom_file) {
        return -1;
    }
    if (fread(&id, 1, sizeof(uint32_t), urandom_file) != sizeof(uint32_t)) {
        return -1;
    }
    fclose(urandom_file);

    char *name = malloc(sizeof(char) * 16);
    if (!name) {
        return -1;
    }

    if (snprintf(name, 16, ":1.%"PRIu32, id) < 0) {
        return -1;
    }

    int name_index = bus_name_add(s, name, i);
    if (name_index < 0) {
        free(name);
        return -1;
    }
    c->unique_name_index = name_index;

    return name_index;
}

int bus_client_assign_own_name(bus_t *s, int i, char *name)
{
    bus_client_t *c = &s->clients[i];
    if (c->owned_name_index != -1 || !name || *name == ':' || *name == '\0') {
        return -1;
    }

    int name_index = bus_name_add(s, name, i);
    if (name_index < 0) {
        return -1;
    }
    c->owned_name_index = name_index;

    return 0;
}

void bus_name_remove(bus_t *s, int i)
{
    if (i >= 0) {
        if (s->names[i].name) {
            free(s->names[i].name);
            s->names[i].name = NULL;
        }
        s->names[i].client_index = -1;
        s->names_count--;
    }
}

void bus_client_remove(bus_t *s, int i)
{
    bus_client_t *c = &s->clients[i];

    if (c->fd >= 0) {
        close(c->fd);
    }
    for (int j = 0; j < BUS_MAX_MATCH; j++) {
        if (c->matches[j]) {
            match_rule_free(c->matches[j]);
            c->matches[j] = NULL;
        }
    }
    bus_name_remove(s, c->unique_name_index);
    bus_name_remove(s, c->owned_name_index);
    c->unique_name_index = -1;
    c->owned_name_index = -1;
    c->match_count = 0;
    c->fd = -1;
    c->state = BUS_CLIENT_STATE_NONE;
    s->fds[i].fd = -1;
    s->fds[i].events = 0;
    s->fds[i].revents = 0;
    s->clients_count--;
}

void bus_client_error(bus_t *s, int i, const char *msg)
{
    WARN("removing client %d due to error: %s\n", i, msg);
    send(s->clients[i].fd, msg, strlen(msg), 0);
    bus_client_remove(s, i);
}

ssize_t bus_client_recv(bus_t *s, int i, void *buf, size_t n)
{
    ssize_t status = recv(s->fds[i].fd, buf, n, 0);
    if (status <= 0) {
        bus_client_remove(s, i);
    }
    return status;
}

int bus_broadcast_message(bus_t *s, wire_message_t *msg, wire_context_t *ctx, char *sender_unique_name, wire_context_t *reply_ctx)
{
    uint32_t body_end = ctx->byte_cursor + msg->body_length;
    if (body_end >= ctx->data_len) {
        return -1;
    }

    int left = s->clients_count;
    for (int i = 0; i < BUS_MAX_CLIENTS && left > 0; i++) {
        if (s->clients[i].match_count <= 0) {
            continue;
        }

        left--;

        bus_client_t *c = &s->clients[i];
        int match_left = c->match_count;
        for (int j = 0; j < BUS_MAX_MATCH && match_left > 0; j++) {
            if (!c->matches[j]) {
                continue;
            }

            match_left--;
            if (match_rule_check(c->matches[j], msg, ctx) >= 0) {
                uint32_t previous_cursor = ctx->byte_cursor;
                TRYST(wire_compose_unicast_reply(reply_ctx, ctx, msg, sender_unique_name));
                TRYST(send(c->fd, reply_ctx->data, reply_ctx->byte_cursor, 0));
                ctx->byte_cursor = previous_cursor;
                memset(reply_ctx->data, 0, reply_ctx->data_len);
                break;
            }
        }
    }

    // We rely on the byte_cursor being at the end of the message upon processing.
    ctx->byte_cursor = body_end;
    
    return 0;
}

int bus_broadcast_signal(bus_t *s, wire_context_t *ctx, wire_message_t *msg)
{
    int left = s->clients_count;
    for (int i = 0; i < BUS_MAX_CLIENTS && left > 0; i++) {
        if (s->clients[i].match_count <= 0) {
            continue;
        }

        left--;

        bus_client_t *c = &s->clients[i];
        int match_left = c->match_count;
        for (int j = 0; j < BUS_MAX_MATCH && match_left > 0; j++) {
            if (!c->matches[j]) {
                continue;
            }

            match_left--;
            if (match_rule_check(c->matches[j], msg, ctx) >= 0) {
                uint32_t previous_cursor = ctx->byte_cursor;
                TRYST(send(c->fd, ctx->data, ctx->byte_cursor, 0));
                ctx->byte_cursor = previous_cursor;
                break;
            }
        }
    }
    return 0;
}

int bus_unicast_message(bus_t *s, wire_message_t *msg, wire_context_t *ctx, char *target_name, char *sender_unique_name, wire_context_t *reply_ctx)
{
    bus_client_t *target = TRYPTR(bus_name_find_client(s, target_name));
    TRYST(wire_compose_unicast_reply(reply_ctx, ctx, msg, sender_unique_name));
    TRYST(send(target->fd, reply_ctx->data, reply_ctx->byte_cursor, 0));
    return 0;
}

#define _signal_begin(M_sig, M_member) \
    uint32_t *signal_body_length = NULL; \
    uint32_t signal_body_start = 0; \
    uint8_t signal_reply_data[8192]; \
    memset(signal_reply_data, 0, 8192); \
    wire_context_t signal_reply_ctx = { \
        .byte_cursor = 0, \
        .data = signal_reply_data, \
        .data_len = 8192, \
    }; \
    wire_message_t signal_reply_msg = {0}; \
    TRYST(wire_compose_signal(&signal_reply_ctx, &signal_reply_msg, (M_sig), (M_member), &signal_body_length)); \
    signal_body_start = signal_reply_ctx.byte_cursor;

#define _signal_end() \
    *signal_body_length = signal_reply_ctx.byte_cursor - signal_body_start; \
    TRYST(bus_broadcast_signal(s, &signal_reply_ctx, &signal_reply_msg)); \


#define _reply_begin(M_sig) \
    uint32_t *body_length = NULL; \
    uint32_t body_start = 0; \
    if (!(msg->flags & DBUS_FLAG_NO_REPLY_EXPECTED)) { \
        TRYST(wire_compose_reply(reply_ctx, msg, (M_sig), &body_length)); \
        body_start = reply_ctx->byte_cursor; \

#define _reply_end() \
        *body_length = reply_ctx->byte_cursor - body_start; \
        if (send(s->fds[i].fd, reply_ctx->data, reply_ctx->byte_cursor, 0) != reply_ctx->byte_cursor) { \
            return -1; \
        } \
        VERBOSE("sent response of %d byte(s)\n", reply_ctx->byte_cursor); \
    } /* if (!(msg.flags & DBUS_FLAG_NO_REPLY_EXPECTED)) */

#define _reply_error(message) \
    do { \
        if (!(msg->flags & DBUS_FLAG_NO_REPLY_EXPECTED)) { \
            TRYST(wire_compose_error(reply_ctx, msg, (message))); \
            if (send(s->fds[i].fd, reply_ctx->data, reply_ctx->byte_cursor, 0) != reply_ctx->byte_cursor) { \
                return -1; \
            } \
        } \
    } while(0) \

int handle_hello(bus_t *s, int i, wire_message_t *msg, wire_context_t *ctx, wire_context_t *reply_ctx) {
    (void)ctx;

    int unique_name_index = TRYST(bus_client_assign_unique_name(s, i));

    _reply_begin("s") {
        TRYPTR(wire_set_string(reply_ctx, s->names[unique_name_index].name));
    } _reply_end()

    _signal_begin("sss", "NameOwnerChanged") {
        /* Name with a new owner */
        TRYPTR(wire_set_string(&signal_reply_ctx, s->names[unique_name_index].name));
        /* Old owner or empty string if none */
        TRYPTR(wire_set_string(&signal_reply_ctx, ""));
        /* New owner or empty string if none */
        TRYPTR(wire_set_string(&signal_reply_ctx, ""));
    } _signal_end();

    VERBOSE("assigned unique name '%s' to connection %d\n", s->names[unique_name_index].name, i);
    return 0;
}

int handle_request_name(bus_t *s, int i, wire_message_t *msg, wire_context_t *ctx, wire_context_t *reply_ctx) {
    if (s->clients[i].unique_name_index < 0) {
        return -1;
    }

    char *name = TRYPTR(wire_get_name_string(ctx));
    TRYPTR(wire_get_u32(ctx)); // unused flags

    char *name_str = TRYPTR(string_dup(name));

    int status_code = DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER;

    if (bus_client_assign_own_name(s, i, name_str) < 0) {
        free(name_str);
        // TODO: report the actual error
        status_code = DBUS_REQUEST_NAME_REPLY_EXISTS;
    }

    _reply_begin("u") {
        TRYPTR(wire_set_u32(reply_ctx, status_code));
    } _reply_end()

    if (status_code == DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER) {
        _signal_begin("sss", "NameOwnerChanged") {
            /* Name with a new owner */
            TRYPTR(wire_set_string(&signal_reply_ctx, name_str));
            /* Old owner or empty string if none */
            TRYPTR(wire_set_string(&signal_reply_ctx, ""));
            /* New owner or empty string if none */
            TRYPTR(wire_set_string(&signal_reply_ctx, s->names[s->clients[i].unique_name_index].name));
        } _signal_end();

        VERBOSE("client '%s' (index=%d) now owns name '%s'\n", s->names[s->clients[i].unique_name_index].name, i, name_str);
    }
    return 0;
}

int handle_get_name_owner(bus_t *s, int i, wire_message_t *msg, wire_context_t *ctx, wire_context_t *reply_ctx) {
    char *name = TRYPTR(wire_get_name_string(ctx));

    bus_client_t *target = bus_name_find_client(s, name);
    if (!target || target->unique_name_index < 0) {
        _reply_error("org.freedesktop.DBus.Error.NameHasNoOwner");
        return 0;
    }

    _reply_begin("s") {
        TRYPTR(wire_set_string(reply_ctx, s->names[target->unique_name_index].name));
    } _reply_end()
    return 0;
}

int handle_name_has_owner(bus_t *s, int i, wire_message_t *msg, wire_context_t *ctx, wire_context_t *reply_ctx) {
    char *name = TRYPTR(wire_get_name_string(ctx));

    bus_client_t *target = bus_name_find_client(s, name);

    _reply_begin("b") {
        TRYPTR(wire_set_u32(reply_ctx, target ? 1 : 0));
    } _reply_end()
    return 0;
}

int handle_list_names(bus_t *s, int i, wire_message_t *msg, wire_context_t *ctx, wire_context_t *reply_ctx) {
    (void)ctx;

    _reply_begin("as") {
        uint32_t *array_length = TRYPTR(wire_set_u32(reply_ctx, 0));

        /* arrays start with the alignment of the type they contain */
        TRYPTR(wire_write_align(reply_ctx, 4));

        uint32_t array_start = reply_ctx->byte_cursor;

        TRYPTR(wire_set_string(reply_ctx, "org.freedesktop.DBus"));

        int left = s->names_count;
        for (int j = 0; j < BUS_MAX_NAMES && left > 0; j++) {
            if (s->names[j].name && s->names[j].client_index >= 0) {
                left--;
                TRYPTR(wire_set_string(reply_ctx, s->names[j].name));
            }
        }

        *array_length = reply_ctx->byte_cursor - array_start;
    } _reply_end()
    return 0;
}

int handle_list_activatable_names(bus_t *s, int i, wire_message_t *msg, wire_context_t *ctx, wire_context_t *reply_ctx) {
    (void)ctx;

    STUB("handle_list_activatable_names", "always returns empty array");

    _reply_begin("as") {
        TRYPTR(wire_set_u32(reply_ctx, 0));
        // empty arrays still need to align
        TRYPTR(wire_write_align(reply_ctx, 4));
    } _reply_end()
    return 0;
}

int handle_start_service_by_name(bus_t *s, int i, wire_message_t *msg, wire_context_t *ctx, wire_context_t *reply_ctx) {
    TRYPTR(wire_get_name_string(ctx));

    /* unused flags value */
    TRYPTR(wire_get_u32(ctx));

    STUB("handle_start_service_by_name", "does nothing and returns success");

    _reply_begin("u") {
        TRYPTR(wire_set_u32(reply_ctx, 1));
    } _reply_end()
    return 0;
}

int handle_add_match(bus_t *s, int i, wire_message_t *msg, wire_context_t *ctx, wire_context_t *reply_ctx) {
    char *match = TRYPTR(wire_get_string_check(ctx, 1, MATCH_RULE_MAX));

    VERBOSE("client index %d adding match rule: '%s'\n", i, match);

    TRYST(bus_client_match_add(s, i, match));

    _reply_begin("") {} _reply_end()
    return 0;
}

int handle_remove_match(bus_t *s, int i, wire_message_t *msg, wire_context_t *ctx, wire_context_t *reply_ctx) {
    TRYPTR(wire_get_string_check(ctx, 1, MATCH_RULE_MAX));

    STUB("handle_remove_match", "does nothing and returns success");

    _reply_begin("") {} _reply_end()
    return 0;
}

int handle_get_connection_stats(bus_t *s, int i, wire_message_t *msg, wire_context_t *ctx, wire_context_t *reply_ctx) {
    (void)ctx;

    TRYPTR(wire_get_name_string(ctx));

    STUB("handle_get_connection_stats", "always returns empty array");

    _reply_begin("as") {
        TRYPTR(wire_set_u32(reply_ctx, 0));
        // empty arrays still need to align
        TRYPTR(wire_write_align(reply_ctx, 8));
    } _reply_end()
    return 0;
}

int handle_get_connection_unix_process_id(bus_t *s, int i, wire_message_t *msg, wire_context_t *ctx, wire_context_t *reply_ctx) {
    (void)ctx;

    char *name = TRYPTR(wire_get_name_string(ctx));

    bus_client_t *target = bus_name_find_client(s, name);
    if (!target || target->unique_name_index < 0 || target->fd < 0) {
        _reply_error("org.freedesktop.DBus.Error.NameHasNoOwner");
        return 0;
    }

    struct ucred cred;
    socklen_t len = sizeof(struct ucred);

    if (getsockopt(target->fd, SOL_SOCKET, SO_PEERCRED, &cred, &len) == -1) {
        _reply_error("xyz.hippoz.jitterbug.FailedToGetPID");
        return 0;
    }

    _reply_begin("u") {
        TRYPTR(wire_set_u32(reply_ctx, cred.pid));
    } _reply_end()
    return 0;
}

int handle_get_all_match_rules(bus_t *s, int i, wire_message_t *msg, wire_context_t *ctx, wire_context_t *reply_ctx) {
    (void)ctx;
    _reply_begin("a{sas}") {
        uint32_t *outer_array_len = TRYPTR(wire_set_u32(reply_ctx, 0));
        TRYPTR(wire_write_align(reply_ctx, 8)); /* arrays start with the alignment of the type they contain */
        uint32_t outer_array_start = reply_ctx->byte_cursor;

        int left = s->names_count;
        for (int j = 0; j < BUS_MAX_NAMES && left > 0; j++) {
            bus_name_t *n = &s->names[j];
            if (!n->name || n->client_index < 0 || *n->name != ':') {
                continue;
            }
            left--;

            TRYPTR(wire_write_align(reply_ctx, 8)); /* structs always aligned to 8 */
            
            bus_client_t *c = &s->clients[s->names[j].client_index];

            TRYPTR(wire_set_string(reply_ctx, s->names[j].name)); /* unique name */

            /* array of all match rules */
            int match_left = c->match_count;
            uint32_t *name_array_len = TRYPTR(wire_set_u32(reply_ctx, 0));
            TRYPTR(wire_write_align(reply_ctx, 4)); /* arrays start with the alignment of the type they contain */
            uint32_t name_array_start = reply_ctx->byte_cursor;
            for (int k = 0; k < BUS_MAX_MATCH && match_left > 0; k++) {
                if (!c->matches[k]) {
                    continue;
                }
                match_left--;

                TRYPTR(wire_set_string(reply_ctx, c->matches[k]->rule_string));
            }
            *name_array_len = reply_ctx->byte_cursor - name_array_start;
        }

        *outer_array_len = reply_ctx->byte_cursor - outer_array_start;
    } _reply_end()
    return 0;
}


static const bus_method_handler_t bus_method_handlers[] = {
    { "Hello", handle_hello },
    { "RequestName", handle_request_name },
    { "GetNameOwner", handle_get_name_owner },
    { "NameHasOwner", handle_name_has_owner },
    { "ListNames", handle_list_names },
    { "ListActivatableNames", handle_list_activatable_names },
    { "StartServiceByName", handle_start_service_by_name },
    { "AddMatch", handle_add_match },
    { "RemoveMatch", handle_remove_match },
    { "GetConnectionStats", handle_get_connection_stats },
    { "GetConnectionUnixProcessID", handle_get_connection_unix_process_id },
    { "GetAllMatchRules", handle_get_all_match_rules }
};
static const int bus_method_handlers_count = sizeof(bus_method_handlers) / sizeof(bus_method_handlers[0]);


#define _process_message_reply_error(message) \
    do { \
        if (!(msg.flags & DBUS_FLAG_NO_REPLY_EXPECTED)) { \
            TRYST(wire_compose_error(&reply_ctx, &msg, (message))); \
            if (send(s->fds[i].fd, reply_ctx.data, reply_ctx.byte_cursor, 0) != reply_ctx.byte_cursor) { \
                return -1; \
            } \
        } \
    } while(0) \

int bus_client_process_message(bus_t *s, int i, wire_context_t *ctx)
{
    bus_client_t *client = &s->clients[i];

    wire_message_t msg = {0};

    if (wire_parse_message(ctx, &msg) < 0) {
        WARN("parsing failed\n");
        return -1;
    }

    uint32_t body_end = ctx->byte_cursor + msg.body_length;
    if (body_end >= ctx->data_len) {
        return -1;
    }

    wire_message_field_t *destination_field = &msg.fields[DBUS_HEADER_FIELD_DESTINATION];
    wire_message_field_t *member_field = &msg.fields[DBUS_HEADER_FIELD_MEMBER];

    uint8_t reply_data[8192];
    memset(reply_data, 0, 8192);
    wire_context_t reply_ctx = {
        .byte_cursor = 0,
        .data = reply_data,
        .data_len = 8192,
    };

    if (msg.type == DBUS_MESSAGE_METHOD_CALL && destination_field->present && member_field->present && strcmp(destination_field->t.str, "org.freedesktop.DBus") == 0) {
        // method call to the bus
        for (int j = 0; j < bus_method_handlers_count; j++) {
            const bus_method_handler_t *handler = &bus_method_handlers[j];
            if (strcmp(handler->name, member_field->t.str) == 0) {
                TRYST(handler->handler(s, i, &msg, ctx, &reply_ctx));
                goto end_nonfatal;
            }
        }
        WARN("FIXME: daemon method '%s' is not implemented or invalid\n", member_field->t.str);
        _process_message_reply_error("org.freedesktop.DBus.Error.UnknownMethod");
        goto end_nonfatal;
    }

    // message needs to be routed
    // TODO: perform checks here like making sure all method calls have the 'member' field, etc.

    if (client->unique_name_index < 0) {
        return -1;
    }

    if (destination_field->present) {
        if (bus_unicast_message(s, &msg, ctx, destination_field->t.str, s->names[client->unique_name_index].name, &reply_ctx) < 0) {
            _process_message_reply_error("xyz.hippoz.jitterbug.UnicastFailed");
            goto end_nonfatal;
        }
    } else {
        if (bus_broadcast_message(s, &msg, ctx, s->names[client->unique_name_index].name, &reply_ctx) < 0) {
            _process_message_reply_error("xyz.hippoz.jitterbug.BroadcastFailed");
            goto end_nonfatal;
        }
    }

end_nonfatal:
    if (ctx->byte_cursor != body_end) {
        // The code above did not properly consume the entire message while parsing.
        // This is bad, because we expect the byte_cursor after this function to be
        // exactly at the end of a message. We rely on this invariant for message
        // boundries. We can obviously just set the byte_cursor to the body_end
        // that we have calculated and move on, however, not consuming the entire
        // message is nonetheless a bug that should be fixed, or an indicator of
        // another bug or something going wrong.
        // In this case, I've decided to set the byte_cursor to the end of the
        // body and print an error message. In the future, we may want to have
        // separate debug builds, and maybe exit the program if we are in a
        // debug build and we encounter something like this.
        WARN("parsing code did not consume the entire message (expected: %d, actual: %d)\n", body_end, ctx->byte_cursor);
        ctx->byte_cursor = body_end;
    }

    return 0;
}

int bus_client_drain_messages(bus_t *s, int ci, uint8_t *data, uint32_t data_len)
{
    wire_context_t ctx = {
        .byte_cursor = 0,
        .data = data,
        .data_len = data_len
    };

    TRYST(bus_client_process_message(s, ci, &ctx));

    int i = 0;
    for (; i < 10; i++) {
        if (ctx.byte_cursor >= ctx.data_len || ctx.data[ctx.byte_cursor] != 'l') {
            // no more messages
            break;
        }
        ctx.data = ctx.data + ctx.byte_cursor;
        ctx.data_len = ctx.data_len - ctx.byte_cursor;
        ctx.byte_cursor = 0;
        TRYST(bus_client_process_message(s, ci, &ctx));
    }

    VERBOSE("processed %d messages - cursor:%d; data_len:%ld\n", i + 1, ctx.byte_cursor, ctx.data_len);

    return 0;
}

#define _client_die(m) do { bus_client_error(s,i,m); continue; } while(0)
int bus_turn(bus_t *s)
{
    static const char agree_unix_fd[] = "AGREE_UNIX_FD\r\n";
    static const char auth_ok[] = "OK aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\r\n";
    static const char auth_list[] = "REJECTED EXTERNAL\r\n";
    static const char auth_data[] = "DATA\r\n";
    static const int data_buffer_len = 16384;

    // We can keep a padding of null bytes for the data buffer. In the event that, for
    // example, a string function is called on a char array without a proper ending null
    // byte, we will reach the null bytes here at the end instead, thus preventing a
    // crash and potential corruption. While this is a good "last line of defense"
    // against such issues, it is much more important for these kinds of bugs to not
    // exist in the first place. This mitigation will prevent, for example, ASAN from
    // finding such bugs. It's recommended that you disable this padding outside
    // of production so that you can find these bugs.
    static const int data_buffer_padding = 0;

    TRYST(poll(s->fds, s->fd_num, -1));

    for (int i = 0; i < s->fd_num; i++) {
        int fd = s->fds[i].fd;
        if (s->fds[i].revents & POLLNVAL) {
            ERROR("bus_turn: error: got POLLNVAL for fds[%d]. This is considered a bug in the server implementation.\n", i);
            return -1;
        }
        if (s->fds[i].revents & POLLERR) {
            if (fd == s->sock_fd) {
                ERROR("bus_turn: error: got POLLERR for sock_fd\n");
                return -1;
            }
            bus_client_remove(s, i);
            continue;
        }
        if (s->fds[i].revents & POLLHUP) {
            if (fd == s->sock_fd) {
                ERROR("bus_turn: error: got POLLHUP for sock_fd\n");
                return -1;
            }
            bus_client_remove(s, i);
            continue;
        }
        if (s->fds[i].revents & POLLIN) {
            // file descriptor ready for reading

            if (fd == s->sock_fd) {
                // new connection
                int accepted_fd = TRYST(accept(s->sock_fd, NULL, NULL));

                if (bus_client_add(s, accepted_fd) == -1) {
                    close(accepted_fd);
                }
                continue;
            }

            // We add padding. See above.
            char data[data_buffer_len + data_buffer_padding];
            memset(data, 0, sizeof(data));
            ssize_t bytes = recv(fd, data, data_buffer_len, 0);
            if (bytes <= 0) {
                // error during recv() OR client disconnected, disconnect the client
                // TODO: should we actually do this?
                bus_client_remove(s, i);
                continue;
            }

            VERBOSE("\nrecv: got %zd bytes\n", bytes);

            bus_client_t *c = &s->clients[i];
            switch (c->state) {
                case BUS_CLIENT_STATE_WAIT_AUTH: {
                    // The D-Bus authentication protocol is a simple plain text protocol.
                    // Before the flow of messages can begin, the two applications must authenticate.
                    // SPEC: https://dbus.freedesktop.org/doc/dbus-specification.html#auth-protocol

                    // Immediately after connecting, clients must send a null byte
                    // SPEC: https://dbus.freedesktop.org/doc/dbus-specification.html#auth-nul-byte
                    char *auth_string = data;
                    if (*auth_string == '\0') {
                        auth_string++;
                    }

                    // TODO: the code below is hacky and does not reflect the specification
                    if (strcmp(auth_string, "AUTH\r\n") == 0) {
                        send(fd, auth_list, sizeof(auth_list) - 1, 0);
                    } else if (strcmp(auth_string, "AUTH EXTERNAL 31303030\r\n") == 0 || strcmp(auth_string, "DATA\r\n") == 0) {
                        send(fd, auth_ok, sizeof(auth_ok) - 1, 0);
                        c->state = BUS_CLIENT_STATE_WAIT_BEGIN;
                    } else if (strcmp(auth_string, "AUTH EXTERNAL\r\n") == 0) {
                        send(fd, auth_data, sizeof(auth_data) - 1, 0);
                    }
                } break;
                case BUS_CLIENT_STATE_WAIT_BEGIN: {
                    // Right now, we're expecting the client to either immediately begin the connection,
                    // or to negotiate UNIX file descriptor passing.
                    if (strncmp(data, "BEGIN\r\n", 7) == 0) {
                        c->state = BUS_CLIENT_STATE_READY;
                        // At this point, a D-Bus connection has been established.
                        // The first octet after the \r\n of the BEGIN command is the first octet of the D-Bus communication.
                        // SPEC: https://dbus.freedesktop.org/doc/dbus-specification.html#auth-command-begin
                        char *first_message_begin = data + 7; /* 7 = length of "BEGIN\r\n" */
                        if (*first_message_begin == 'l' || *first_message_begin == 'B') {
                            // This looks like a D-Bus message. Let's process it!
                            if (bus_client_drain_messages(s, i, (uint8_t *)first_message_begin, data_buffer_len - 7) < 0) {
                                _client_die("failed to process message");
                            }
                        }
                    } else if (strcmp(data, "NEGOTIATE_UNIX_FD\r\n") == 0) {
                        // SPEC: https://dbus.freedesktop.org/doc/dbus-specification.html#auth-command-negotiate-unix-fd
                        send(fd, agree_unix_fd, sizeof(agree_unix_fd) - 1, 0);
                    } else {
                        _client_die("bad auth response (expected BEGIN or NEGOTIATE_UNIX_FD)");
                    }
                } break;

                case BUS_CLIENT_STATE_READY: {
                    if (bus_client_drain_messages(s, i, (uint8_t*)data, data_buffer_len) < 0) {
                        _client_die("failed to process message");
                    }
                } break;

                case BUS_CLIENT_STATE_NONE: {} /* through */
                default: {
                    _client_die("bad state");
                } break;
            }
        }
    }

    return 0;
}