diff options
Diffstat (limited to 'server.c')
| -rw-r--r-- | server.c | 752 |
1 files changed, 752 insertions, 0 deletions
diff --git a/server.c b/server.c new file mode 100644 index 0000000..9c1a710 --- /dev/null +++ b/server.c @@ -0,0 +1,752 @@ +#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; +} |