path: root/server.c

#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
//
#include "common.c"
#include "omen.c"
#include "security.c"
#include "sha256.c"
//
#include "firehose.c"
#include "packet_storage.c"

pthread_mutex_t log_io_mutex = PTHREAD_MUTEX_INITIALIZER;

enum { _debug = 1, _info = 2, _warn = 3, _error = 4 } log_level = _debug;
const char* log_level_names[] = { "debug", "info", "warn", "error" };
#define log(__level, fmt, ...)                                                                     \
    ({                                                                                             \
        pthread_mutex_lock(&log_io_mutex);                                                         \
        if (__level >= log_level)                                                                  \
            fprintf(stderr, "[%s] " fmt, (#__level) + 1 __VA_OPT__(, ) __VA_ARGS__);               \
        pthread_mutex_unlock(&log_io_mutex);                                                       \
    })

const uint64_t max_neighbors = 1 << 10; // probably more than enough for a lot of folks

uint64_t               current_neighbor_ptr = 0;
uint64_t               neighbor_count       = 0;
struct relay_neighbor* neighbor_storage;
bool*                  empty_neighbor_slots;
pthread_mutex_t        neighbor_storage_mutex = PTHREAD_MUTEX_INITIALIZER;

struct {
    byte major_version;
    byte minor_version;
    byte motd[ 128 ];
    //
    uint16_t retention_hours; // should default to 1w (168h)
    byte     leading_zeros;   // i recommend 10
    bool     accepts_plain;   // for legal reasons
} server_stats = {
    .major_version   = 0,
    .minor_version   = 1,
    .motd            = "A moonpaste server.", // minecraft much?
    .retention_hours = 168,                   // 1 week
    .leading_zeros   = 10,                    // the moon guarantee
    .accepts_plain   = true,                  // probably legal
};

bool add_neighbor(struct relay_neighbor new_neighbor) {
    log(_info, "adding neighbor\n");

    pthread_mutex_lock(&neighbor_storage_mutex);
    if (neighbor_count >= max_neighbors) {
        pthread_mutex_unlock(&neighbor_storage_mutex);
        return false;
    }

    uint64_t add_idx = -1;
    for (uint64_t i = 0; i < max_neighbors; i++) {
        if (empty_neighbor_slots[ i ]) {
            add_idx = i;
            break;
        }
    }

    if (add_idx != -1) {
        neighbor_storage[ add_idx ]     = new_neighbor;
        empty_neighbor_slots[ add_idx ] = false;
        neighbor_count++;
        pthread_mutex_unlock(&neighbor_storage_mutex);
        return true;
    }

    pthread_mutex_unlock(&neighbor_storage_mutex);
    return false;
}

#define hash_fmt                                                                                   \
    "%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x"                             \
    "%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x"

#define hash_ref(_hash)                                                                            \
    _hash[ 0 ], _hash[ 1 ], _hash[ 2 ], _hash[ 3 ], _hash[ 4 ], _hash[ 5 ], _hash[ 6 ],            \
        _hash[ 7 ], _hash[ 8 ], _hash[ 9 ], _hash[ 10 ], _hash[ 11 ], _hash[ 12 ], _hash[ 13 ],    \
        _hash[ 14 ], _hash[ 15 ], _hash[ 16 ], _hash[ 17 ], _hash[ 18 ], _hash[ 19 ], _hash[ 20 ], \
        _hash[ 21 ], _hash[ 22 ], _hash[ 23 ], _hash[ 24 ], _hash[ 25 ], _hash[ 26 ], _hash[ 27 ], \
        _hash[ 28 ], _hash[ 29 ], _hash[ 30 ], _hash[ 31 ]

void* propagate_packet(void* _packet) {
    struct data_packet packet = *(struct data_packet*) _packet;
    log(_debug, "propagating packet " hash_fmt "\n", hash_ref(packet.hash));

    size_t send_len  = 1 + sizeof(struct data_packet);
    byte*  send_data = calloc(send_len, sizeof(byte));
    send_data[ 0 ]   = push_packet;
    data_packet2s(packet, send_data + 1);

    pthread_mutex_lock(&neighbor_storage_mutex);
    for (uint64_t i = 0; i < neighbor_count; i++) {
        struct relay_neighbor neighbor = neighbor_storage[ i ];
        if (neighbor.active) {
            byte*  response_data = NULL;
            size_t response_len  = 0;
            int    result        = -1;

            if (neighbor.has_ipv4) {
                char ip_string[ 16 ];
                sprintf(
                    ip_string,
                    "%d.%d.%d.%d",
                    neighbor.v4_addr[ 0 ],
                    neighbor.v4_addr[ 1 ],
                    neighbor.v4_addr[ 2 ],
                    neighbor.v4_addr[ 3 ]);

                result = study(
                    ip_string, default_port, send_data, send_len, &response_data, &response_len);
            } else if (neighbor.has_ipv6) {
                char ip_string[ 46 ];
                int  offset = 0;
                for (int j = 0; j < 8; j++) {
                    uint16_t word
                        = (neighbor.v6_addr[ j * 2 ] << 8) | neighbor.v6_addr[ j * 2 + 1 ];
                    offset += sprintf(ip_string + offset, "%x%s", word, (j < 7) ? ":" : "");
                }

                result = study(
                    ip_string, default_port, send_data, send_len, &response_data, &response_len);
            }

            if (result != 0) { fprintf(stderr, "failed to send packet to neighbor\n"); }

            if (response_data) { free(response_data); }
        }
    }
    pthread_mutex_unlock(&neighbor_storage_mutex);

    free(send_data);
    free(_packet);
    return NULL;
}

void save_packet(const struct data_packet packet);

#define reply(__len, __stream)                                                                     \
    ({                                                                                             \
        *out     = (byte*) __stream;                                                               \
        *out_len = __len;                                                                          \
    })

void paste_server(
    char* ip_addr, size_t ip_size, byte* in, size_t in_len, byte** out, size_t* out_len) {
    log(_info, "<%.*s> incoming request with size %li\n", (int) ip_size, ip_addr, in_len);

    if (in_len < 8) { // server mandates 8 bytes minimum. pad it if you don't need it!
        reply(17, strdup("\x01input too small"));
        return;
    }

    enum client_input input_type = in[ 0 ];

    switch (input_type) {
        case audit_server: {
            log(_info, "<%.*s> request type: audit\n", (int) ip_size, ip_addr);

            byte* output_data = calloc(1 + 237, sizeof(byte));
            if (!output_data) {
                byte* err = calloc(1 + 14, sizeof(byte));
                if (err) {
                    err[ 0 ] = error;
                    memcpy(err + 1, "server OOM", 10);
                    reply(11, err);
                }
                return;
            }
            output_data[ 0 ] = audit_response;

            struct relay_info info = { 0 };

            pthread_mutex_lock(&neighbor_storage_mutex);
            pthread_mutex_lock(&packet_storage_mutex);

            info.major_version                = server_stats.major_version;
            info.minor_version                = server_stats.minor_version;
            info.capabilities.retention_hours = server_stats.retention_hours;
            info.capabilities.can_store       = packet_count < max_storage - 1;
            info.capabilities.leading_zeros   = server_stats.leading_zeros;
            info.capabilities.accepts_plain   = server_stats.accepts_plain;

            info.statistics.neighbors    = neighbor_count;
            info.statistics.held_packets = packet_count;

            // TODO: for now we return the first 4
            for (int i = 0; i < 4; ++i) {
                info.statistics.quick_connect[ i ] = neighbor_storage[ i ];
            }

            pthread_mutex_unlock(&packet_storage_mutex);
            pthread_mutex_unlock(&neighbor_storage_mutex);

            memcpy(info.motd, server_stats.motd, 128);

            relay_info2s(info, output_data + 1);

            reply(1 + 237, output_data);
            return;
        }

        case get_neighbors: {
            byte count = in[ 1 ];
            log(_info,
                "<%.*s> request type: get neighbors, count %i\n",
                (int) ip_size,
                ip_addr,
                count);

            size_t output_len  = 1 + (RELAY_NEIGHBOR_WIRE_LEN * count);
            byte*  output_data = calloc(output_len, sizeof(byte));
            if (!output_data) {
                byte* err = calloc(1 + 14, sizeof(byte));
                if (err) {
                    err[ 0 ] = error;
                    memcpy(err + 1, "server OOM", 10);
                    reply(11, err);
                }
                return;
            }
            output_data[ 0 ] = neighbors_info;

            size_t offset = 1;

            for (byte i = 0; i < count; i++) {
                // TODO: maybe mutex?
                relay_neighbor2s(neighbor_storage[ i ], output_data + offset);
                offset += RELAY_NEIGHBOR_WIRE_LEN;
            }

            reply(output_len, output_data);
            return;
        }

        case get_packet: {
            if (in_len < 1 + 32) {
                byte* output_data = calloc(1 + 17, sizeof(byte));
                if (!output_data) return;
                output_data[ 0 ] = error;
                memcpy(output_data + 1, "input too small", 17);
                reply(18, output_data);
                return;
            }

            hash packet_hash;
            bool success;
            s2hash(32, in + 1, &success, packet_hash);

            log(_info,
                "<%.*s> request type: get packet with hash " hash_fmt "\n",
                (int) ip_size,
                ip_addr,
                hash_ref(packet_hash));

            if (!success) {
                byte* output_data = calloc(1 + 12, sizeof(byte));
                if (!output_data) return;
                output_data[ 0 ] = error;
                memcpy(output_data + 1, "invalid hash", 12);
                reply(13, output_data);
                return;
            }

            struct data_packet_entry* entry = get_data_packet(packet_hash);
            if (entry == NULL) {
                byte* output_data = calloc(1 + 14, sizeof(byte));
                if (!output_data) return;
                output_data[ 0 ] = error;
                memcpy(output_data + 1, "packet not found", 14);
                reply(15, output_data);
                return;
            }

            size_t output_len  = 1 + PACKET_SIZE;
            byte*  output_data = calloc(output_len, sizeof(byte));
            if (!output_data) {
                byte* err = calloc(1 + 14, sizeof(byte));
                if (err) {
                    err[ 0 ] = error;
                    memcpy(err + 1, "server OOM", 10);
                    reply(11, err);
                }
                return;
            }

            output_data[ 0 ] = get_packet_result;
            data_packet2s(entry->packet, output_data + 1);

            reply(output_len, output_data);
            return;
        }

        case push_packet: {
            bool               success;
            struct data_packet packet = s2data_packet(in_len - 1, in + 1, &success);

            log(_info,
                "<%.*s> request type: push packet with hash " hash_fmt "\n",
                (int) ip_size,
                ip_addr,
                hash_ref(packet.hash));

            size_t output_len  = 1 + 1;
            byte*  output_data = calloc(output_len, sizeof(byte));
            if (!output_data) { return; }
            output_data[ 0 ] = push_packet_result;

            if (!success) {
                log(_warn, "<%.*s> malformed data\n", (int) ip_size, ip_addr);
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            if (!check_leading_zeros(packet.hash, server_stats.leading_zeros)) {
                log(_warn, "bad leads\n");
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            const size_t hash_input_len = 4 + CONTENTS_SIZE;
            byte         hash_input_buffer[ hash_input_len ];
            write_u32_le(hash_input_buffer, packet.nonce);
            byte* contents_buffer = hash_input_buffer + 4;

            contents_buffer[ 0 ] = packet.contents.encrypted ? 1 : 0;
            write_u64_le(contents_buffer + 1, packet.contents.created_time_ms);
            memcpy(contents_buffer + 9, packet.contents.author, 32);
            memcpy(contents_buffer + 41, packet.contents.author_signing_pub, 32);
            contents_buffer[ 73 ] = packet.contents.recipient_count;
            memcpy(contents_buffer + 74, packet.contents.recipients, 32 * MAX_RECIPIENTS);
            memcpy(contents_buffer + 74 + 32 * MAX_RECIPIENTS, packet.contents.data, 65536);
            memcpy(
                contents_buffer + 74 + 32 * MAX_RECIPIENTS + 65536, packet.contents.signature, 64);

            struct sha256 calculated_hash_struct = sha256_hash(hash_input_buffer, hash_input_len);
            if (memcmp(calculated_hash_struct.hash, packet.hash, sizeof(hash)) != 0) {
                log(_warn,
                    "bad hash " hash_fmt " vs " hash_fmt "\n",
                    hash_ref(calculated_hash_struct.hash),
                    hash_ref(packet.hash));
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            uint64_t current_time = get_current_time_ms();
            if (packet.contents.created_time_ms > current_time) {
                log(_warn, "future\n");
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            if (!packet.contents.encrypted && !server_stats.accepts_plain) {
                log(_warn, "plaintext not accepted by server\n");
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            bool sig_ok = false;
            if (!packet.contents.encrypted) {
                size_t msg_len = CONTENTS_SIZE - sizeof(signature);
                sig_ok         = verify_raw_signature(
                    contents_buffer,
                    msg_len,
                    packet.contents.signature,
                    packet.contents.author_signing_pub);
            } else {
                secure_payload sp;
                memcpy(sp.nonce, packet.contents.data, sizeof(sp.nonce));
                memcpy(
                    sp.encrypted_data,
                    packet.contents.data + sizeof(sp.nonce),
                    sizeof(sp.encrypted_data));
                memcpy(sp.signature, packet.contents.signature, sizeof(sp.signature));
                sig_ok = verify_signature(sp, packet.contents.author_signing_pub);
            }

            if (!sig_ok) {
                log(_warn, "bad signature\n");
                output_data[ 1 ] = false;
                reply(output_len, output_data);
                return;
            }

            uint64_t storage_index = set_data_packet(&packet);
            firehose_push(packet.hash, storage_index);

            if (storage_index != minus_one) {
                output_data[ 1 ] = true;

                save_packet(packet);

                struct data_packet* thread_arg = malloc(sizeof(struct data_packet));
                if (thread_arg) {
                    *thread_arg = packet;
                    pthread_t prop_thread;
                    pthread_create(&prop_thread, NULL, propagate_packet, thread_arg);
                    pthread_detach(prop_thread);
                }
            } else {
                output_data[ 1 ] = false;
            }

            reply(output_len, output_data);
            return;
        }

        case firehose: {
            if (in_len < 1 + 9) {
                byte* output_data = calloc(1 + 17, sizeof(byte));
                output_data[ 0 ]  = error;
                memcpy(output_data + 1, "input too small", 17);
                reply(18, output_data);
                return;
            }
            struct firehose_request req;
            bool                    success;
            req = s2firehose_request(in_len - 1, in + 1, &success);

            if (unlikely(req.count > 64)) req.count = 64; // clamp

            log(_info,
                "<%.*s> request type: firehose count %i offset %li\n",
                (int) ip_size,
                ip_addr,
                req.count,
                req.offset);

            if (!success) {
                byte* output_data = calloc(1 + 14, sizeof(byte));
                output_data[ 0 ]  = error;
                memcpy(output_data + 1, "invalid request", 14);
                reply(15, output_data);
                return;
            }

            water_splash entries[ req.count ];
            uint16_t     copied_count = firehose_get_packets(req.offset, req.count, entries);

            size_t total_packets_size = PACKET_SIZE * copied_count;

            size_t output_len  = 1 + 8 + total_packets_size;
            byte*  output_data = calloc(output_len, sizeof(byte));
            output_data[ 0 ]   = firehose_reply;
            size_t offset      = 1;

            uint64_t new_offset     = req.offset + copied_count;
            output_data[ offset++ ] = new_offset & 0xFF;
            output_data[ offset++ ] = (new_offset >> 8) & 0xFF;
            output_data[ offset++ ] = (new_offset >> 16) & 0xFF;
            output_data[ offset++ ] = (new_offset >> 24) & 0xFF;
            output_data[ offset++ ] = (new_offset >> 32) & 0xFF;
            output_data[ offset++ ] = (new_offset >> 40) & 0xFF;
            output_data[ offset++ ] = (new_offset >> 48) & 0xFF;
            output_data[ offset++ ] = (new_offset >> 56) & 0xFF;

            for (int i = 0; i < copied_count; i++) {
                struct data_packet_entry* entry = &packet_storage[ entries[ i ].storage_index ];
                data_packet2s(entry->packet, output_data + offset);
                offset += PACKET_SIZE;
            }

            reply(output_len, output_data);
            return;
        }

        default: {
            log(_info, "<%.*s> request type: invalid method\n", (int) ip_size, ip_addr);

            byte* output_data = calloc(1 + 14, sizeof(byte));
            output_data[ 0 ]  = error;
            memcpy(output_data + 1, "invalid input", 14);
            reply(15, output_data);
            return;
        }
    }
}

bool try_hugepages = true;

void* alloc_map(uint64_t size) {
    static const byte RW = PROT_READ | PROT_WRITE;
    static const byte PA = MAP_PRIVATE | MAP_ANONYMOUS;

    void* alloc;
    if (try_hugepages) {
        alloc = mmap(NULL, size, RW, PA | MAP_HUGE_2MB, -1, 0);
        if (alloc == MAP_FAILED) {
            // alloc failed. maybe the fault of hugepages?
            goto NO_HP;
        }
    } else {
    NO_HP:
        alloc = mmap(NULL, size, RW, PA, -1, 0);
        if (alloc != MAP_FAILED) {
            // hugepages failed, remove flag
            printf("[note] no hugepages support present\n");
            try_hugepages = false;
        }
    }

    if (alloc == MAP_FAILED) return NULL;
    return alloc;
}

pthread_t     packet_cleanup_thread;
volatile bool packet_cleanup_thread_running = true;

pthread_t     disk_persistence_thread;
volatile bool disk_persistence_thread_running = true;

bool file_exists(const char* filename) { return access(filename, F_OK) == 0; }

void save_packet(const struct data_packet packet) {
    log(_debug, "saving packet " hash_fmt " to disk\n", hash_ref(packet.hash));

    mkdir("packets", 0755);

    char filename[ 128 ];
    sprintf(filename, "packets/" hash_fmt ".bin", hash_ref(packet.hash));

    FILE* file = fopen(filename, "wb");
    if (!file) {
        perror("failed to save packet");
        return;
    }

    byte* buffer = malloc(PACKET_SIZE);
    if (!buffer) {
        fclose(file);
        return;
    }

    data_packet2s(packet, buffer);
    size_t written = fwrite(buffer, 1, PACKET_SIZE, file);
    if (written != PACKET_SIZE) { perror("failed to write full packet to disk"); }
    free(buffer);
    fclose(file);
}

void save_neighbors() {
    FILE* file = fopen("neighbors.bin", "wb");
    if (!file) {
        perror("failed to save neighbors");
        return;
    }

    pthread_mutex_lock(&neighbor_storage_mutex);
    for (uint64_t i = 0; i < max_neighbors; i++) {
        if (!empty_neighbor_slots[ i ]) {
            byte buffer[ RELAY_NEIGHBOR_WIRE_LEN ];
            relay_neighbor2s(neighbor_storage[ i ], buffer);
            if (fwrite(buffer, 1, RELAY_NEIGHBOR_WIRE_LEN, file) != RELAY_NEIGHBOR_WIRE_LEN) {
                perror("failed to write neighbor entry");
            }
        }
    }
    pthread_mutex_unlock(&neighbor_storage_mutex);
    fclose(file);
}

void load_neighbors() {
    FILE* file = fopen("neighbors.bin", "rb");
    if (!file) { return; }

    size_t count = 0;

    byte buffer[ RELAY_NEIGHBOR_WIRE_LEN ];
    while (fread(buffer, 1, RELAY_NEIGHBOR_WIRE_LEN, file) == RELAY_NEIGHBOR_WIRE_LEN) {
        bool                  success;
        struct relay_neighbor neighbor
            = s2relay_neighbor(RELAY_NEIGHBOR_WIRE_LEN, buffer, &success);
        if (success) {
            count++;
            add_neighbor(neighbor);
        }
    }
    fclose(file);

    log(_info, "loaded %li neighbors\n", count);
}

void* fs_persistence(void* arg) {
    while (disk_persistence_thread_running) {
        sleep(2 * 60); // 2m
        log(_info, "saving neighbors...\n");
        save_neighbors();
    }

    return NULL;
}

void* cleanup_packets(void* arg) {
    while (packet_cleanup_thread_running) {
        sleep(30 * 60); // 30m

        log(_warn, "running cleanup\n");
        uint64_t current_time_ms = get_current_time_ms();
        uint64_t retention_ms    = (uint64_t) server_stats.retention_hours * 60 * 60 * 1000;

        for (uint64_t i = 0; i < max_storage; i++) {
            pthread_mutex_lock(&packet_storage_mutex);
            if (packet_storage[ i ].is_full
                && current_time_ms - packet_storage[ i ].packet.contents.created_time_ms
                       > retention_ms) {
                log(_debug,
                    "cleaning up old packet hashed " hash_fmt "\n",
                    hash_ref(packet_storage[ i ].packet.hash));

                char filename[ 128 ];
                sprintf(
                    filename,
                    "packets/" hash_fmt ".bin",
                    hash_ref(packet_storage[ i ].packet.hash));

                if (remove(filename) != 0) log(_error, "failed to delete packet %s\n", filename);

                packet_storage[ i ].is_full = false;
                packet_count--;
            }
            pthread_mutex_unlock(&packet_storage_mutex);
        }
    }

    return NULL;
}

pthread_t     neighbor_check_thread;
volatile bool neighbor_check_thread_running = true;

void* check_neighbors(void* arg) {
    sleep(5 * 30);

    while (neighbor_check_thread_running) {
        pthread_mutex_lock(&neighbor_storage_mutex);
        for (uint64_t i = 0; i < max_neighbors; i++) {
            if (!empty_neighbor_slots[ i ]) {
                struct relay_neighbor neighbor = neighbor_storage[ i ];

                byte*  response_data = NULL;
                size_t response_len  = 0;
                int    result        = -1;

                byte audit_request[ 1 ] = { audit_server };

                if (neighbor.has_ipv4) {
                    char ip_string[ 16 ];
                    sprintf(
                        ip_string,
                        "%d.%d.%d.%d",
                        neighbor.v4_addr[ 0 ],
                        neighbor.v4_addr[ 1 ],
                        neighbor.v4_addr[ 2 ],
                        neighbor.v4_addr[ 3 ]);
                    result = study(
                        ip_string, default_port, audit_request, 1, &response_data, &response_len);
                } else if (neighbor.has_ipv6) {
                    char ip_string[ 46 ];
                    int  offset = 0;

                    for (int j = 0; j < 8; j++) {
                        uint16_t word
                            = (neighbor.v6_addr[ j * 2 ] << 8) | neighbor.v6_addr[ j * 2 + 1 ];

                        offset += snprintf(
                            ip_string + offset,
                            sizeof(ip_string) - offset,
                            "%x%s",
                            word,
                            (j < 7) ? ":" : "");
                    }

                    result = study(
                        ip_string, default_port, audit_request, 1, &response_data, &response_len);
                }

                if (result != 0 || response_data[ 0 ] != audit_response) {
                    empty_neighbor_slots[ i ] = true;
                    neighbor_count--;
                    log(_warn, "removed dead neighbor\n");
                }

                if (response_data) { free(response_data); }
            }
        }
        pthread_mutex_unlock(&neighbor_storage_mutex);

        sleep(60 * 5); // 5m
    }
    return NULL;
}

int main() {
    fprintf(
        stderr,
        "-- moonpaste server v%i.%i --\n"
        "port %i\n\n",
        server_stats.major_version,
        server_stats.minor_version,
        default_port);

    // init
    packet_storage = alloc_map(max_storage * sizeof(struct data_packet_entry));

    neighbor_storage     = alloc_map(max_neighbors * sizeof(struct relay_neighbor));
    empty_neighbor_slots = alloc_map(max_neighbors * sizeof(bool));
    memset(empty_neighbor_slots, 1, max_neighbors * sizeof(bool));

    if (!packet_storage || !neighbor_storage || !empty_neighbor_slots) {
        perror("init alloc failed");
        return 1;
    }

    if (pthread_create(&packet_cleanup_thread, NULL, cleanup_packets, NULL) != 0) {
        perror("failed to create packet cleanup thread");
        return 1;
    }

    load_neighbors();

    if (pthread_create(&disk_persistence_thread, NULL, fs_persistence, NULL) != 0) {
        perror("failed to create persistence thread");
        return 1;
    }

    if (pthread_create(&neighbor_check_thread, NULL, check_neighbors, NULL) != 0) {
        perror("failed to create neighbor audit thread");
        return 1;
    }

    firehose_init();

    omen(default_port, paste_server);

    packet_cleanup_thread_running = false;
    pthread_join(packet_cleanup_thread, NULL);

    return 0;
}