#include #include #include #include #include // #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 = 12, // 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, ...) \ ({ \ *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\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 (current_time - packet.contents.created_time_ms > server_stats.retention_hours * (60 * 60 * 1000)) { log(_warn, "expired\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); 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\n", (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\n", (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\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; // memory 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); } // 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); struct stat st; if (stat(filepath, &st) == 0 && current_time_ms - (uint64_t) st.st_mtime * 1000 > retention_ms) { if (remove(filepath) != 0) log(_error, "failed to delete %s\n", 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\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; }