diff options
Diffstat (limited to 'src/server.c')
| -rw-r--r-- | src/server.c | 946 |
1 files changed, 946 insertions, 0 deletions
diff --git a/src/server.c b/src/server.c new file mode 100644 index 0000000..a246fe5 --- /dev/null +++ b/src/server.c @@ -0,0 +1,946 @@ +#include <dirent.h> +#include <fcntl.h> +#include <pthread.h> +#include <sys/mman.h> +#include <sys/stat.h> +#include <unistd.h> +// dependency: zlib +#include <zlib.h> +// +#include "common.c" +#include "config.c" +#include "firehose.c" +#include "omen.c" +#include "packet_storage.c" +#include "plugin.c" +#include "security.c" +#include "sha256.c" +// + +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 server_stats { + 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 = 12, // the moon guarantee + .accepts_plain = true, // probably legal +}; + +bool add_neighbor(struct relay_neighbor new_neighbor) { + if (decide_neighbor_registry_func) { + if (!decide_neighbor_registry_func(new_neighbor)) return false; + } + + log(_info, "adding neighbor"); + + 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; + + if (decide_packet_propogation_func) { + if (!decide_packet_propogation_func(packet)) return NULL; + } + + log(_debug, "propagating packet " hash_fmt, hash_ref(packet.hash)); + + size_t send_len = 1 + sizeof(struct data_packet); + byte send_data[ 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) { log(_error, "failed to send packet to neighbor"); } + + if (response_data) { free(response_data); } + } + } + pthread_mutex_unlock(&neighbor_storage_mutex); + + free(_packet); + return NULL; +} + +void save_packet(const struct data_packet packet); + +#define reply(__len, ...) \ + ({ \ + *out = (byte*) __VA_ARGS__; \ + *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", (int) ip_size, ip_addr, in_len); + + if (paste_server_start_func) { + paste_server_start_func(&ip_addr, &ip_size, &in, &in_len, &out, &out_len); + } + + // tip: if you want the server to not mandate the 8 bytes minimum, modify in_len using the + // plugin. + + 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", (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 ]; + } + + memcpy(info.motd, server_stats.motd, 128); + + if (modify_server_stats_func) { modify_server_stats_func(&info); } + + pthread_mutex_unlock(&packet_storage_mutex); + pthread_mutex_unlock(&neighbor_storage_mutex); + + 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", + (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, + (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, + (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", (int) ip_size, ip_addr); + output_data[ 1 ] = false; + reply(output_len, output_data); + return; + } + + if (modify_packet_func) { + // things you can do here: + // - prefer certain packet authors + // - run PoW + // - increase leading zeroes + // - recompute the hash + modify_packet_func(&packet); + } + + if (!check_leading_zeros(packet.hash, server_stats.leading_zeros)) { + log(_warn, "bad leads"); + 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, + 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, "packet timestamp is in the future"); + output_data[ 1 ] = false; + reply(output_len, output_data); + return; + } + + if (current_time - packet.contents.created_time_ms + > server_stats.retention_hours * (60 * 60 * 1000)) { + log(_warn, "packet expired"); + 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"); + 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"); + 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); + + if (((float) rand() / (float) RAND_MAX) > 0.5) { + 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 (!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; + } + + if (unlikely(req.count > 64)) req.count = 64; // clamp + + log(_info, + "<%.*s> request type: firehose count %i offset %li", + (int) ip_size, + ip_addr, + req.count, + req.offset); + + 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; + } + + case register_neighbor: { + bool success; + struct relay_neighbor neighbor = s2relay_neighbor(in_len - 1, in + 1, &success); + + 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; + } + + log(_info, "<%.*s> request type: neighbor registration", (int) ip_size, ip_addr); + + char ip_string[ INET6_ADDRSTRLEN ]; + + if (neighbor.has_ipv4) { + sprintf( + ip_string, + "%d.%d.%d.%d", + neighbor.v4_addr[ 0 ], + neighbor.v4_addr[ 1 ], + neighbor.v4_addr[ 2 ], + neighbor.v4_addr[ 3 ]); + } else if (neighbor.has_ipv6) { + 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) ? ":" : ""); + } + } + + struct relay_info audit_result = run_server_audit(ip_string, default_port); + if (!audit_result.capabilities.can_store && audit_result.major_version == 0 + && audit_result.minor_version == 0) { + // invalid + reply(2, (byte[ 2 ]) { error, 0 }); + break; + } + + reply(2, (byte[ 2 ]) { register_neighbor_result, add_neighbor(neighbor) }); + } break; + + default: { + log(_info, "<%.*s> request type: invalid method", (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"); + 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) { + if (decide_packet_save_to_disk_func) { + if (!decide_packet_save_to_disk_func(packet)) return; + } + + log(_debug, "saving packet " hash_fmt " to disk", hash_ref(packet.hash)); + + mkdir("packets", 0755); + + char filename_tmp[ 256 ]; + char filename_final[ 256 ]; + snprintf( + filename_tmp, sizeof(filename_tmp), "packets/" hash_fmt ".bin.tmp", hash_ref(packet.hash)); + snprintf( + filename_final, sizeof(filename_final), "packets/" hash_fmt ".bin", hash_ref(packet.hash)); + + /* serialize packet into buffer */ + byte buffer[ PACKET_SIZE ]; + data_packet2s(packet, buffer); + + /* compress */ + uLongf comp_bound = compressBound(PACKET_SIZE); + byte comp_buf[ comp_bound ]; + + uLongf comp_len = comp_bound; + int zres = compress2(comp_buf, &comp_len, buffer, PACKET_SIZE, Z_BEST_COMPRESSION); + if (zres != Z_OK) { + log(_error, "save_packet: compress2 failed (%d)", zres); + return; + } + + FILE* f = fopen(filename_tmp, "wb"); + if (!f) { + perror("save_packet: fopen tmp"); + return; + } + + if (fwrite(comp_buf, 1, comp_len, f) != comp_len) { + perror("save_packet: write failed"); + fclose(f); + unlink(filename_tmp); + return; + } + + fflush(f); +#if defined(__unix__) || defined(__APPLE__) + int fd = fileno(f); + if (fd >= 0) fsync(fd); +#endif + fclose(f); + + if (rename(filename_tmp, filename_final) != 0) { + perror("save_packet: rename failed"); + unlink(filename_tmp); + } +} + +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", count); +} + +void* fs_persistence(void* arg) { + while (disk_persistence_thread_running) { + sleep(2 * 60); // 2m + log(_info, "saving neighbors..."); + save_neighbors(); + } + + return NULL; +} + +void* cleanup_packets(void* arg) { + while (packet_cleanup_thread_running) { + sleep(30 * 60); // 30m + + log(_warn, "running cleanup"); + uint64_t current_time_ms = get_current_time_ms(); + uint64_t retention_ms = (uint64_t) server_stats.retention_hours * 60 * 60 * 1000; + + // memory + for (uint64_t i = 0; i < max_storage; i++) { + pthread_mutex_lock(&packet_storage_mutex); + + if (!packet_storage[ i ].is_full) { + pthread_mutex_unlock(&packet_storage_mutex); + continue; + } + + bool result + = decide_packet_cleanup_func + ? decide_packet_cleanup_func(packet_storage[ i ].packet) // delegate to user + : ( // do it natively + current_time_ms - packet_storage[ i ].packet.contents.created_time_ms + > retention_ms); + + if (result) { + log(_debug, + "cleaning up old packet hashed " hash_fmt, + 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", filename); + + packet_storage[ i ].is_full = false; + packet_count--; + } + + pthread_mutex_unlock(&packet_storage_mutex); + } + + // external cleanup + system("find packets/ -mtime +7 -delete"); + + // disk cleanup, tho not sure this is needed + DIR* dir = opendir("packets"); + if (dir) { + struct dirent* entry; + while ((entry = readdir(dir)) != NULL) { + if (entry->d_type == DT_REG) { + char filepath[ 256 ]; + snprintf(filepath, sizeof(filepath), "packets/%s", entry->d_name); + + FILE* fd = decide_packet_cleanup_func ? fopen(filepath, "rb") : NULL; + struct data_packet* new = decide_packet_cleanup_func ? load_packet(fd) : NULL; + + struct stat st; + bool result = decide_packet_cleanup_func + ? decide_packet_cleanup_func(*new) // delegate to user + : ( // do it natively + stat(filepath, &st) == 0 + && current_time_ms - (uint64_t) st.st_mtime * 1000 + > retention_ms); + + if (decide_packet_cleanup_func) { + fclose(fd); + free(new); + } + + if (result) { + if (remove(filepath) != 0) log(_error, "failed to delete %s", filepath); + } + } + } + closedir(dir); + } + } + + 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"); + } + + if (response_data) { free(response_data); } + } + } + pthread_mutex_unlock(&neighbor_storage_mutex); + + sleep(60 * 5); // 5m + } + return NULL; +} + +int main(int argc, char** argv) { + parse_arguments(argc, argv); + + char* __log_level = get_config("log", false); + if (__log_level) { + if (isdigit(*__log_level)) { + int _log_level = atoi(__log_level); + if (_log_level < 1 || _log_level > 4) { + fprintf( + stderr, + "invalid log level %d (1 = debug, 2 = info, 3 = warn, 4 = error)\n", + _log_level); + exit(1); + } + log_level = _log_level; + } else { + if (strcmp(__log_level, "debug") == 0) log_level = 1; + else if (strcmp(__log_level, "info") == 0) + log_level = 2; + else if (strcmp(__log_level, "warn") == 0) + log_level = 3; + else if (strcmp(__log_level, "error") == 0) + log_level = 4; + else { + fprintf( + stderr, + "invalid log level %s (1 = debug, 2 = info, 3 = warn, 4 = error)\n", + __log_level); + exit(1); + } + } + + free(__log_level); + } + + 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(); + + // load the plugins! + char* plugin_path = get_config("plugin", false); + if (plugin_path) { + load_plugins(plugin_path); + free(plugin_path); + } + + omen(default_port, paste_server); + + packet_cleanup_thread_running = false; + pthread_join(packet_cleanup_thread, NULL); + + return 0; +} |