From 067200b438e022cef840d0e772db96657e5cafa9 Mon Sep 17 00:00:00 2001 From: moon Date: Sun, 14 Sep 2025 20:27:54 +0330 Subject: stuff --- src/server.c | 443 ++++++++++++++++++++++++++++++++--------------------------- 1 file changed, 238 insertions(+), 205 deletions(-) (limited to 'src/server.c') diff --git a/src/server.c b/src/server.c index 0b7ea87..3f1a9ee 100644 --- a/src/server.c +++ b/src/server.c @@ -4,7 +4,9 @@ #include #include #include +// dependency: zlib #include +// #include "common.c" #include "config.c" #include "firehose.c" @@ -18,26 +20,26 @@ uint16_t current_port = default_port; const uint64_t max_neighbors = 1 << 10; -uint64_t current_neighbor_ptr = 0; -uint64_t neighbor_count = 0; +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; +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]; + byte major_version; + byte minor_version; + byte motd[ 128 ]; uint16_t retention_hours; - byte leading_zeros; - bool accepts_plain; + byte leading_zeros; + bool accepts_plain; } server_stats = { - .major_version = 0, - .minor_version = 1, - .motd = "A moonpaste server.", + .major_version = 0, + .minor_version = 1, + .motd = "A moonpaste server.", .retention_hours = 168, - .leading_zeros = 12, - .accepts_plain = true, + .leading_zeros = 12, + .accepts_plain = true, }; bool add_neighbor(struct relay_neighbor new_neighbor) { @@ -55,15 +57,15 @@ bool add_neighbor(struct relay_neighbor new_neighbor) { uint64_t add_idx = -1; for (uint64_t i = 0; i < max_neighbors; i++) { - if (empty_neighbor_slots[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_storage[ add_idx ] = new_neighbor; + empty_neighbor_slots[ add_idx ] = false; neighbor_count++; pthread_mutex_unlock(&neighbor_storage_mutex); return true; @@ -74,7 +76,7 @@ bool add_neighbor(struct relay_neighbor new_neighbor) { } void* propagate_packet(void* _packet) { - struct data_packet packet = *(struct data_packet*)_packet; + struct data_packet packet = *(struct data_packet*) _packet; if (decide_packet_propogation_func) { if (!decide_packet_propogation_func(packet)) return NULL; @@ -83,35 +85,36 @@ void* propagate_packet(void* _packet) { 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; + 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]; + struct relay_neighbor neighbor = neighbor_storage[ i ]; if (neighbor.active) { - byte* response_data = NULL; - size_t response_len = 0; - int result = -1; + byte* response_data = NULL; + size_t response_len = 0; + int result = -1; if (neighbor.has_ipv4) { - char ip_string[16]; + 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]); + neighbor.v4_addr[ 0 ], + neighbor.v4_addr[ 1 ], + neighbor.v4_addr[ 2 ], + neighbor.v4_addr[ 3 ]); result = study( ip_string, neighbor.port, send_data, send_len, &response_data, &response_len); } else if (neighbor.has_ipv6) { - char ip_string[46]; - int offset = 0; + 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]; + 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) ? ":" : ""); } @@ -132,12 +135,15 @@ void* propagate_packet(void* _packet) { void save_packet(const struct data_packet packet); -#define reply(__len, ...) \ - ({ *out = (byte*)__VA_ARGS__; *out_len = __len; }) +#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); + 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); @@ -148,41 +154,41 @@ void paste_server( return; } - enum client_input input_type = in[0]; + enum client_input input_type = in[ 0 ]; switch (input_type) { case audit_server: { - log(_info, "<%.*s> request type: audit", (int)ip_size, ip_addr); + 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; + err[ 0 ] = error; memcpy(err + 1, "server OOM", 10); reply(11, err); } return; } - output_data[0] = audit_response; + output_data[ 0 ] = audit_response; - struct relay_info info = {0}; + 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.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.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.neighbors = neighbor_count; info.statistics.held_packets = packet_count; for (int i = 0; i < 4; ++i) { - info.statistics.quick_connect[i] = neighbor_storage[i]; + info.statistics.quick_connect[ i ] = neighbor_storage[ i ]; } memcpy(info.motd, server_stats.motd, 128); @@ -199,28 +205,30 @@ void paste_server( } case get_neighbors: { - byte count = in[1]; + byte count = in[ 1 ]; log(_info, "<%.*s> request type: get neighbors, count %i", - (int)ip_size, ip_addr, count); + (int) ip_size, + ip_addr, + count); - size_t output_len = 1 + (RELAY_NEIGHBOR_WIRE_LEN * count); - byte* output_data = calloc(output_len, sizeof(byte)); + 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; + err[ 0 ] = error; memcpy(err + 1, "server OOM", 10); reply(11, err); } return; } - output_data[0] = neighbors_info; + output_data[ 0 ] = neighbors_info; size_t offset = 1; for (byte i = 0; i < count; i++) { - relay_neighbor2s(neighbor_storage[i], output_data + offset); + relay_neighbor2s(neighbor_storage[ i ], output_data + offset); offset += RELAY_NEIGHBOR_WIRE_LEN; } @@ -232,7 +240,7 @@ void paste_server( if (in_len < 1 + 32) { byte* output_data = calloc(1 + 17, sizeof(byte)); if (!output_data) return; - output_data[0] = error; + output_data[ 0 ] = error; memcpy(output_data + 1, "input too small", 17); reply(18, output_data); return; @@ -244,12 +252,14 @@ void paste_server( log(_info, "<%.*s> request type: get packet with hash " hash_fmt, - (int)ip_size, ip_addr, hash_ref(packet_hash)); + (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; + output_data[ 0 ] = error; memcpy(output_data + 1, "invalid hash", 12); reply(13, output_data); return; @@ -259,84 +269,88 @@ void paste_server( if (entry == NULL) { byte* output_data = calloc(1 + 14, sizeof(byte)); if (!output_data) return; - output_data[0] = error; + output_data[ 0 ] = error; memcpy(output_data + 1, "packet not found", 14); reply(15, output_data); return; } - byte decompressed[sizeof(struct data_packet)]; + byte decompressed[ sizeof(struct data_packet) ]; uLongf decompressed_len = sizeof(struct data_packet); - if (uncompress(decompressed, &decompressed_len, entry->compressed_payload, - entry->compressed_size) != Z_OK) { + if (uncompress( + decompressed, + &decompressed_len, + entry->compressed_payload, + entry->compressed_size) + != Z_OK) { byte* output_data = calloc(1 + 22, sizeof(byte)); if (!output_data) return; - output_data[0] = error; + output_data[ 0 ] = error; memcpy(output_data + 1, "decompression failed", 22); reply(23, output_data); return; } - size_t output_len = 1 + PACKET_SIZE; - byte* output_data = calloc(output_len, sizeof(byte)); + 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; + err[ 0 ] = error; memcpy(err + 1, "server OOM", 10); reply(11, err); } return; } - output_data[0] = get_packet_result; - data_packet2s(*(struct data_packet*)decompressed, output_data + 1); + output_data[ 0 ] = get_packet_result; + data_packet2s(*(struct data_packet*) decompressed, output_data + 1); reply(output_len, output_data); return; } case push_packet: { - bool success; + 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)); + (int) ip_size, + ip_addr, + hash_ref(packet.hash)); - size_t output_len = 1 + 1; - byte* output_data = calloc(output_len, sizeof(byte)); + size_t output_len = 1 + 1; + byte* output_data = calloc(output_len, sizeof(byte)); if (!output_data) { return; } - output_data[0] = push_packet_result; + output_data[ 0 ] = push_packet_result; if (!success) { - log(_warn, "<%.*s> malformed data", (int)ip_size, ip_addr); - output_data[1] = false; + log(_warn, "<%.*s> malformed data", (int) ip_size, ip_addr); + output_data[ 1 ] = false; reply(output_len, output_data); return; } - if (modify_packet_func) { - modify_packet_func(&packet); - } + if (modify_packet_func) { modify_packet_func(&packet); } if (!check_leading_zeros(packet.hash, server_stats.leading_zeros)) { log(_warn, "bad leads"); - output_data[1] = false; + 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]; + 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; + 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; + 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( @@ -348,7 +362,7 @@ void paste_server( "bad hash " hash_fmt " vs " hash_fmt, hash_ref(calculated_hash_struct.hash), hash_ref(packet.hash)); - output_data[1] = false; + output_data[ 1 ] = false; reply(output_len, output_data); return; } @@ -356,7 +370,7 @@ void paste_server( 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; + output_data[ 1 ] = false; reply(output_len, output_data); return; } @@ -364,14 +378,14 @@ void paste_server( if (current_time - packet.contents.created_time_ms > server_stats.retention_hours * (60 * 60 * 1000)) { log(_warn, "packet expired"); - output_data[1] = false; + 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; + output_data[ 1 ] = false; reply(output_len, output_data); return; } @@ -379,7 +393,7 @@ void paste_server( bool sig_ok = false; if (!packet.contents.encrypted) { size_t msg_len = CONTENTS_SIZE - sizeof(signature); - sig_ok = verify_raw_signature( + sig_ok = verify_raw_signature( contents_buffer, msg_len, packet.contents.signature, @@ -397,7 +411,7 @@ void paste_server( if (!sig_ok) { log(_warn, "bad signature"); - output_data[1] = false; + output_data[ 1 ] = false; reply(output_len, output_data); return; } @@ -406,11 +420,11 @@ void paste_server( firehose_push(packet.hash, storage_index); if (storage_index != invalid_index) { - output_data[1] = true; + output_data[ 1 ] = true; save_packet(packet); - if (((float)rand() / (float)RAND_MAX) > 0.5) { + if (((float) rand() / (float) RAND_MAX) > 0.5) { struct data_packet* thread_arg = malloc(sizeof(struct data_packet)); if (thread_arg) { *thread_arg = packet; @@ -420,7 +434,7 @@ void paste_server( } } } else { - output_data[1] = false; + output_data[ 1 ] = false; } reply(output_len, output_data); @@ -430,18 +444,18 @@ void paste_server( case firehose: { if (in_len < 1 + 9) { byte* output_data = calloc(1 + 17, sizeof(byte)); - output_data[0] = error; + output_data[ 0 ] = error; memcpy(output_data + 1, "input too small", 17); reply(18, output_data); return; } struct firehose_request req; - bool success; + 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; + output_data[ 0 ] = error; memcpy(output_data + 1, "invalid request", 14); reply(15, output_data); return; @@ -451,38 +465,45 @@ void paste_server( log(_info, "<%.*s> request type: firehose count %i offset %li", - (int)ip_size, ip_addr, req.count, req.offset); + (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); + 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; + 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_index[entries[i].storage_index]; - byte decompressed[sizeof(struct data_packet)]; - uLongf decompressed_len = sizeof(struct data_packet); - if (uncompress(decompressed, &decompressed_len, entry->compressed_payload, - entry->compressed_size) != Z_OK) { + struct data_packet_entry* entry = &packet_index[ entries[ i ].storage_index ]; + byte decompressed[ sizeof(struct data_packet) ]; + uLongf decompressed_len = sizeof(struct data_packet); + if (uncompress( + decompressed, + &decompressed_len, + entry->compressed_payload, + entry->compressed_size) + != Z_OK) { log(_error, "firehose: decompression failed"); continue; } - data_packet2s(*(struct data_packet*)decompressed, output_data + offset); + data_packet2s(*(struct data_packet*) decompressed, output_data + offset); offset += PACKET_SIZE; } @@ -491,33 +512,34 @@ void paste_server( } case register_neighbor: { - bool success; + 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; + 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); + log(_info, "<%.*s> request type: neighbor registration", (int) ip_size, ip_addr); - char ip_string[INET6_ADDRSTRLEN]; + 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]); + 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]; + 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) ? ":" : ""); } } @@ -527,8 +549,8 @@ void paste_server( && audit_result.minor_version == 0) { byte* err_reply = malloc(2); if (err_reply) { - err_reply[0] = error; - err_reply[1] = 0; + err_reply[ 0 ] = error; + err_reply[ 1 ] = 0; } reply(2, err_reply); break; @@ -536,21 +558,21 @@ void paste_server( byte* __reply = malloc(2); if (__reply) { - __reply[0] = register_neighbor_result; - __reply[1] = add_neighbor(neighbor); + __reply[ 0 ] = register_neighbor_result; + __reply[ 1 ] = add_neighbor(neighbor); } reply(2, __reply); } break; case query_packets: { if (in_len != 1 + 32 + 32 + 64 + 2 + 8) { - *out = NULL; + *out = NULL; *out_len = 0; return; } public_key author_enc, author_sign; - signature sig; - size_t off = 1; + signature sig; + size_t off = 1; memcpy(author_enc, in + off, 32); off += 32; memcpy(author_sign, in + off, 32); @@ -561,60 +583,61 @@ void paste_server( off += 2; uint64_t req_offset = read_u64_le(in + off); off += 8; - byte msg[32 + 32 + 2 + 8]; + byte msg[ 32 + 32 + 2 + 8 ]; memcpy(msg, author_enc, 32); memcpy(msg + 32, author_sign, 32); write_u16_le(msg + 64, req_count); write_u64_le(msg + 66, req_offset); if (!verify_raw_signature(msg, sizeof(msg), sig, author_sign)) { - *out = NULL; + *out = NULL; *out_len = 0; return; } if (req_count > 128) req_count = 128; - struct data_packet* packets = malloc((size_t)req_count * sizeof(struct data_packet)); + struct data_packet* packets = malloc((size_t) req_count * sizeof(struct data_packet)); if (!packets) { - *out = NULL; + *out = NULL; *out_len = 0; return; } uint64_t new_offset; - uint16_t actual_count = get_packets_for_author(author_enc, req_offset, req_count, packets, &new_offset); + uint16_t actual_count + = get_packets_for_author(author_enc, req_offset, req_count, packets, &new_offset); uint16_t verified_count = 0; for (uint16_t i = 0; i < actual_count; i++) { - if (memcmp(packets[i].contents.author_signing_pub, author_sign, 32) == 0) { - packets[verified_count++] = packets[i]; + if (memcmp(packets[ i ].contents.author_signing_pub, author_sign, 32) == 0) { + packets[ verified_count++ ] = packets[ i ]; } } - size_t _out_len = 1 + 2 + 8 + (size_t)verified_count * PACKET_SIZE; - byte* _out = malloc(_out_len); + size_t _out_len = 1 + 2 + 8 + (size_t) verified_count * PACKET_SIZE; + byte* _out = malloc(_out_len); if (!_out) { free(packets); - *out = NULL; + *out = NULL; *out_len = 0; return; } - off = 0; - _out[off++] = queried_author_packets; + off = 0; + _out[ off++ ] = queried_author_packets; write_u16_le(_out + off, verified_count); off += 2; write_u64_le(_out + off, new_offset); off += 8; for (uint16_t i = 0; i < verified_count; i++) { - data_packet2s(packets[i], _out + off); + data_packet2s(packets[ i ], _out + off); off += PACKET_SIZE; } free(packets); - *out = _out; + *out = _out; *out_len = _out_len; return; } break; default: { - log(_info, "<%.*s> request type: invalid method", (int)ip_size, ip_addr); + log(_info, "<%.*s> request type: invalid method", (int) ip_size, ip_addr); byte* output_data = calloc(1 + 14, sizeof(byte)); - output_data[0] = error; + output_data[ 0 ] = error; memcpy(output_data + 1, "invalid input", 14); reply(15, output_data); return; @@ -631,9 +654,7 @@ void* alloc_map(uint64_t size) { void* alloc; if (try_hugepages) { alloc = mmap(NULL, size, RW, PA | MAP_HUGE_2MB, -1, 0); - if (alloc == MAP_FAILED) { - goto NO_HP; - } + if (alloc == MAP_FAILED) { goto NO_HP; } } else { NO_HP: alloc = mmap(NULL, size, RW, PA, -1, 0); @@ -647,13 +668,13 @@ void* alloc_map(uint64_t size) { return alloc; } -pthread_t packet_cleanup_thread; +pthread_t packet_cleanup_thread; volatile bool packet_cleanup_thread_running = true; -pthread_t disk_persistence_thread; +pthread_t disk_persistence_thread; volatile bool disk_persistence_thread_running = true; -pthread_t neighbor_check_thread; +pthread_t neighbor_check_thread; volatile bool neighbor_check_thread_running = true; bool file_exists(const char* filename) { return access(filename, F_OK) == 0; } @@ -667,19 +688,21 @@ void save_packet(const struct data_packet packet) { 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)); + 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)); - byte buffer[PACKET_SIZE]; + byte buffer[ PACKET_SIZE ]; data_packet2s(packet, buffer); uLongf comp_bound = compressBound(PACKET_SIZE); - byte comp_buf[comp_bound]; + byte comp_buf[ comp_bound ]; uLongf comp_len = comp_bound; - int zres = compress2(comp_buf, &comp_len, buffer, PACKET_SIZE, Z_BEST_COMPRESSION); + 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; @@ -720,9 +743,9 @@ void save_neighbors() { 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 (!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"); } @@ -738,10 +761,11 @@ void load_neighbors() { size_t count = 0; - byte buffer[RELAY_NEIGHBOR_WIRE_LEN]; + 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); + bool success; + struct relay_neighbor neighbor + = s2relay_neighbor(RELAY_NEIGHBOR_WIRE_LEN, buffer, &success); if (success) { count++; add_neighbor(neighbor); @@ -768,36 +792,38 @@ void* cleanup_packets(void* arg) { 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; + 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_index[i].is_full) { + if (!packet_index[ i ].is_full) { pthread_mutex_unlock(&packet_storage_mutex); continue; } - byte decompressed[sizeof(struct data_packet)]; + byte decompressed[ sizeof(struct data_packet) ]; uLongf decompressed_len = sizeof(struct data_packet); - if (uncompress(decompressed, &decompressed_len, packet_index[i].compressed_payload, - packet_index[i].compressed_size) != Z_OK) { + if (uncompress( + decompressed, + &decompressed_len, + packet_index[ i ].compressed_payload, + packet_index[ i ].compressed_size) + != Z_OK) { log(_error, "cleanup_packets: decompression failed"); pthread_mutex_unlock(&packet_storage_mutex); continue; } - struct data_packet* packet = (struct data_packet*)decompressed; + struct data_packet* packet = (struct data_packet*) decompressed; bool result = decide_packet_cleanup_func ? decide_packet_cleanup_func(*packet) : (current_time_ms - packet->contents.created_time_ms > retention_ms); if (result) { - log(_debug, - "cleaning up old packet hashed " hash_fmt, - hash_ref(packet->hash)); + log(_debug, "cleaning up old packet hashed " hash_fmt, hash_ref(packet->hash)); - char filename[128]; + char filename[ 128 ]; sprintf(filename, "packets/" hash_fmt ".bin", hash_ref(packet->hash)); if (remove(filename) != 0) log(_error, "failed to delete packet %s", filename); @@ -815,17 +841,18 @@ void* cleanup_packets(void* arg) { struct dirent* entry; while ((entry = readdir(dir)) != NULL) { if (entry->d_type == DT_REG) { - char filepath[256]; + 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) - : (stat(filepath, &st) == 0 - && current_time_ms - (uint64_t)st.st_mtime * 1000 > retention_ms); + bool result + = decide_packet_cleanup_func + ? decide_packet_cleanup_func(*new) + : (stat(filepath, &st) == 0 + && current_time_ms - (uint64_t) st.st_mtime * 1000 > retention_ms); if (decide_packet_cleanup_func) { fclose(fd); @@ -850,32 +877,33 @@ void* check_neighbors(void* arg) { 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]; + 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* response_data = NULL; + size_t response_len = 0; + int result = -1; - byte audit_request[1] = {audit_server}; + byte audit_request[ 1 ] = { audit_server }; if (neighbor.has_ipv4) { - char ip_string[16]; + 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]); + neighbor.v4_addr[ 0 ], + neighbor.v4_addr[ 1 ], + neighbor.v4_addr[ 2 ], + neighbor.v4_addr[ 3 ]); result = study( ip_string, neighbor.port, audit_request, 1, &response_data, &response_len); } else if (neighbor.has_ipv6) { - char ip_string[46]; - int offset = 0; + 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]; + uint16_t word + = (neighbor.v6_addr[ j * 2 ] << 8) | neighbor.v6_addr[ j * 2 + 1 ]; offset += snprintf( ip_string + offset, sizeof(ip_string) - offset, @@ -888,8 +916,8 @@ void* check_neighbors(void* arg) { ip_string, neighbor.port, audit_request, 1, &response_data, &response_len); } - if (result != 0 || response_data[0] != audit_response) { - empty_neighbor_slots[i] = true; + if (result != 0 || response_data[ 0 ] != audit_response) { + empty_neighbor_slots[ i ] = true; neighbor_count--; log(_warn, "removed dead neighbor"); } @@ -921,9 +949,12 @@ int main(int argc, char** argv) { 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 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, @@ -956,15 +987,17 @@ int main(int argc, char** argv) { current_port); packet_index = alloc_map(max_storage * sizeof(struct data_packet_entry)); - packet_arena_size = max_storage * (sizeof(uint8_t) + sizeof(uint32_t) + sizeof(struct data_packet)); - packet_arena = alloc_map(packet_arena_size); + packet_arena_size + = max_storage * (sizeof(uint8_t) + sizeof(uint32_t) + sizeof(struct data_packet)); + packet_arena = alloc_map(packet_arena_size); author_storage = alloc_map(max_authors * sizeof(struct author_entry)); - neighbor_storage = alloc_map(max_neighbors * sizeof(struct relay_neighbor)); + 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_index || !packet_arena || !author_storage || !neighbor_storage || !empty_neighbor_slots) { + if (!packet_index || !packet_arena || !author_storage || !neighbor_storage + || !empty_neighbor_slots) { perror("init alloc failed"); return 1; } -- cgit v1.2.3