From d54aabd5d0d55331a6d9b74e99ef311c347abb82 Mon Sep 17 00:00:00 2001 From: moon Date: Sun, 14 Sep 2025 15:20:59 +0330 Subject: query packets by author feature --- .gitignore | 4 +- client/3rd-party/client.py | 69 +++++++++++- client/3rd-party/client.ts | 72 ++++++++----- client/client.c | 69 +++++++++++- src/common.c | 37 ++++--- src/packet_storage.c | 255 ++++++++++++++++++++++++++++++++++++++++++--- src/server.c | 76 +++++++++++++- 7 files changed, 521 insertions(+), 61 deletions(-) diff --git a/.gitignore b/.gitignore index aff9426..6ec7adf 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ bin/ neighbors.bin -packets/ \ No newline at end of file + +packets/ +packets_by_author/ \ No newline at end of file diff --git a/client/3rd-party/client.py b/client/3rd-party/client.py index 56a8faa..52675ad 100644 --- a/client/3rd-party/client.py +++ b/client/3rd-party/client.py @@ -411,6 +411,17 @@ class MoonpasteFFI: self.lib.mp_generate_keypair.argtypes = [] self.lib.mp_generate_keypair.restype = KeyPair + # mp_query_author_packets + self.lib.mp_query_author_packets.argtypes = [ + ctypes.c_char_p, # ip + c_int, # port + KeyPair, # author_keys + c_uint16, # count + c_uint64, # offset + POINTER(DataPacketList) # out_list + ] + self.lib.mp_query_author_packets.restype = c_int + class MoonpasteClient: '''High-level Python client for moonpaste''' @@ -680,4 +691,60 @@ class MoonpasteClient: # Push to server success = self.push_packet(ip, port, packet) - return packet, success \ No newline at end of file + return packet, success + + def query_author_packets(self, ip: str, port: int, author_keys: PyKeyPair, count: int = 10, offset: int = 0) -> PyDataPacketList: + '''Query packets for a specific author''' + c_keys = author_keys.to_c() + packet_list = DataPacketList() + + result = self.ffi.lib.mp_query_author_packets( + ip.encode('utf-8'), + port, + c_keys, + count, + offset, + byref(packet_list) + ) + + if result != ClientError.SUCCESS: + raise MoonpasteClientError(ClientError(result)) + + packets = [] + if packet_list.count > 0 and packet_list.packets: + for i in range(packet_list.count): + packets.append(PyDataPacket.from_c(packet_list.packets[i])) + + # Free the allocated memory safely + try: + import ctypes.util + libc_name = ctypes.util.find_library('c') + if libc_name: + libc = ctypes.CDLL(libc_name) + libc.free.argtypes = [c_void_p] + libc.free(cast(packet_list.packets, c_void_p)) + else: + # Platform-specific fallback + try: + if sys.platform.startswith('linux'): + libc = ctypes.CDLL('libc.so.6') + elif sys.platform == 'darwin': + libc = ctypes.CDLL('libc.dylib') + elif sys.platform == 'win32': + libc = ctypes.CDLL('msvcrt.dll') + else: + libc = None + + if libc: + libc.free.argtypes = [c_void_p] + libc.free(cast(packet_list.packets, c_void_p)) + except OSError: + pass # Can't free, but better than crash + except: + pass # Memory will be freed when process exits + + return PyDataPacketList( + packets=packets, + count=packet_list.count, + new_offset=packet_list.new_offset + ) \ No newline at end of file diff --git a/client/3rd-party/client.ts b/client/3rd-party/client.ts index 44f4634..711dc3b 100644 --- a/client/3rd-party/client.ts +++ b/client/3rd-party/client.ts @@ -155,7 +155,11 @@ class MoonpasteFFI { mp_generate_keypair: { args: [], returns: FFIType.ptr - } + }, + mp_query_author_packets: { + args: [FFIType.cstring, FFIType.i32, FFIType.ptr, FFIType.u16, FFIType.u64, FFIType.ptr], + returns: FFIType.i32 + }, }) break @@ -204,7 +208,7 @@ class MoonpasteFFI { try { const ffi = require('ffi-napi') // dependency const ref = require('ref-napi') // dependency - + this.lib = ffi.Library(libPath, { mp_get_server_info: ['int', ['string', 'int', 'pointer']], mp_get_server_neighbors: ['int', ['string', 'int', 'uint8', 'pointer', 'pointer']], @@ -248,9 +252,9 @@ class MoonpasteFFI { async getServerInfo(ip: string, port: number): Promise { const infoBuffer = this.allocateBuffer(1024) const ipBuffer = this.stringToBuffer(ip) - + const result = this.lib.symbols.mp_get_server_info(ipBuffer, port, infoBuffer) - + if (result !== ClientError.Success) { throw new Error(`Failed to get server info: ${result}`) } @@ -262,9 +266,9 @@ class MoonpasteFFI { const neighborsPtr = this.allocateBuffer(8) // pointer const countPtr = this.allocateBuffer(1) // uint8 const ipBuffer = this.stringToBuffer(ip) - + const result = this.lib.symbols.mp_get_server_neighbors(ipBuffer, port, count, neighborsPtr, countPtr) - + if (result !== ClientError.Success) { throw new Error(`Failed to get neighbors: ${result}`) } @@ -276,9 +280,9 @@ class MoonpasteFFI { const successPtr = this.allocateBuffer(1) const ipBuffer = this.stringToBuffer(ip) const packetBuffer = this.serializePacket(packet) - + const result = this.lib.symbols.mp_push_packet(ipBuffer, port, packetBuffer, successPtr) - + if (result !== ClientError.Success) { throw new Error(`Failed to push packet: ${result}`) } @@ -289,9 +293,9 @@ class MoonpasteFFI { async getPacket(ip: string, port: number, hash: Uint8Array): Promise { const packetBuffer = this.allocateBuffer(70000) // approximation const ipBuffer = this.stringToBuffer(ip) - + const result = this.lib.symbols.mp_get_packet(ipBuffer, port, hash, packetBuffer) - + if (result !== ClientError.Success) { throw new Error(`Failed to get packet: ${result}`) } @@ -302,9 +306,9 @@ class MoonpasteFFI { async getFirehosePackets(ip: string, port: number, offset: bigint, count: number): Promise { const listBuffer = this.allocateBuffer(1024) const ipBuffer = this.stringToBuffer(ip) - + const result = this.lib.symbols.mp_get_firehose_packets(ipBuffer, port, offset, count, listBuffer) - + if (result !== ClientError.Success) { throw new Error(`Failed to get firehose packets: ${result}`) } @@ -312,10 +316,24 @@ class MoonpasteFFI { return this.parseDataPacketList(listBuffer) } + async queryAuthorPackets(ip: string, port: number, authorKeys: KeyPair, count: number, offset: bigint): Promise { + const keysBuffer = this.serializeKeyPair(authorKeys) + const listBuffer = this.allocateBuffer(1024) + const ipBuffer = this.stringToBuffer(ip) + + const result = this.lib.symbols.mp_query_author_packets(ipBuffer, port, keysBuffer, count, offset, listBuffer) + + if (result !== ClientError.Success) { + throw new Error(`Failed to query author packets: ${result}`) + } + + return this.parseDataPacketList(listBuffer) + } + createPacketFromBytes(input: Uint8Array, recipientCount: number, recipients: Uint8Array[] | null, authorKeys: KeyPair): DataPacket { const recipientsBuffer = recipients ? this.serializeRecipients(recipients) : null const keysBuffer = this.serializeKeyPair(authorKeys) - + const result = this.lib.symbols.mp_create_packet_from_bytes( input, input.length, @@ -323,7 +341,7 @@ class MoonpasteFFI { recipientsBuffer, keysBuffer ) - + return this.parseDataPacket(result) } @@ -332,7 +350,7 @@ class MoonpasteFFI { const filenameBuffer = this.stringToBuffer(filename) const recipientsBuffer = recipients ? this.serializeRecipients(recipients) : null const keysBuffer = this.serializeKeyPair(authorKeys) - + const result = this.lib.symbols.mp_create_packet_from_file( filenameBuffer, recipientCount, @@ -340,7 +358,7 @@ class MoonpasteFFI { keysBuffer, packetBuffer ) - + if (result !== ClientError.Success) { throw new Error(`Failed to create packet from file: ${result}`) } @@ -365,7 +383,7 @@ class MoonpasteFFI { const majorVersion = view.getUint8(offset++) const minorVersion = view.getUint8(offset++) - + const motdBytes = new Uint8Array(buffer, offset, 128) const motd = new TextDecoder().decode(motdBytes).replace(/\0.*$/, '') offset += 128 @@ -407,11 +425,11 @@ class MoonpasteFFI { private parseRelayNeighborAt(buffer: any, offset: number): RelayNeighbor { const view = new DataView(buffer instanceof ArrayBuffer ? buffer : buffer.buffer) - + const active = view.getUint8(offset) !== 0 const hasIpv4 = view.getUint8(offset + 1) !== 0 const hasIpv6 = view.getUint8(offset + 2) !== 0 - + const v4Addr = new Uint8Array(buffer, offset + 3, 4) const v6Addr = new Uint8Array(buffer, offset + 7, 16) @@ -421,14 +439,14 @@ class MoonpasteFFI { private parseNeighbors(neighborsPtr: any, countPtr: any): RelayNeighbor[] { const count = this.readUint8(countPtr) const neighbors: RelayNeighbor[] = [] - + if (count > 0 && neighborsPtr) { for (let i = 0; i < count; i++) { const neighborOffset = i * 23 // RELAY_NEIGHBOR_WIRE_LEN neighbors.push(this.parseRelayNeighborAt(neighborsPtr, neighborOffset)) } } - + return neighbors } @@ -481,12 +499,12 @@ class MoonpasteFFI { private parseDataPacketList(buffer: any): DataPacketList { const view = new DataView(buffer instanceof ArrayBuffer ? buffer : buffer.buffer) - + const newOffset = view.getBigUint64(0, true) const availableBytes = buffer.byteLength - 8 const packetSize = 4 + 32 + (1 + 8 + 32 + 32 + 1 + (32 * 32) + 65536 + 64) // PACKET_SIZE const packetCount = Math.floor(availableBytes / packetSize) - + const packets: DataPacket[] = [] for (let i = 0; i < packetCount; i++) { const packetOffset = 8 + (i * packetSize) @@ -624,14 +642,14 @@ export class MoonpasteClient { const data = typeof message === 'string' ? new TextEncoder().encode(message) : message const keys = authorKeys || this.generateKeyPair() const recipientCount = recipients ? recipients.length : 0 - + return this.ffi.createPacketFromBytes(data, recipientCount, recipients || null, keys) } async createPacketFromFile(filename: string, recipients?: Uint8Array[], authorKeys?: KeyPair): Promise { const keys = authorKeys || this.generateKeyPair() const recipientCount = recipients ? recipients.length : 0 - + return this.ffi.createPacketFromFile(filename, recipientCount, recipients || null, keys) } @@ -643,6 +661,10 @@ export class MoonpasteClient { return this.ffi.generateKeyPair() } + async queryAuthorPackets(ip: string, port: number, authorKeys: KeyPair, count: number = 10, offset: bigint = 0n): Promise { + return this.ffi.queryAuthorPackets(ip, port, authorKeys, count, offset) + } + // Utility methods static hashToHex(hash: Uint8Array): string { return Array.from(hash).map(b => b.toString(16).padStart(2, '0')).join('') diff --git a/client/client.c b/client/client.c index bac1b6e..155d895 100644 --- a/client/client.c +++ b/client/client.c @@ -525,4 +525,71 @@ struct data_packet mp_run_pow(struct data_packet packet, byte leading_zeros) { return packet; } -keypair mp_generate_keypair(void) { return generate_keypair(); } \ No newline at end of file +keypair mp_generate_keypair(void) { return generate_keypair(); } + +mp_client_error_t mp_query_author_packets( + const char *ip, + int port, + keypair author_keys, + uint16_t count, + uint64_t offset, + struct data_packet_list *out_list) { + byte msg[ 32 + 32 + 2 + 8 ]; + memcpy(msg, author_keys.encryption_pub, 32); + memcpy(msg + 32, author_keys.signing_pub, 32); + write_u16_le(msg + 64, count); + write_u64_le(msg + 66, offset); + byte sig[ 64 ]; + if (crypto_sign_detached(sig, NULL, msg, sizeof(msg), author_keys.signing_priv) != 0) { + return MP_CLIENT_ERROR_BAD_ARGUMENT; + } + const size_t send_len = 1 + 32 + 32 + 64 + 2 + 8; + byte *send_data = malloc(send_len); + if (!send_data) return MP_CLIENT_ERROR_ALLOC_FAILED; + size_t off = 0; + send_data[ off++ ] = query_packets; + memcpy(send_data + off, author_keys.encryption_pub, 32); + off += 32; + memcpy(send_data + off, author_keys.signing_pub, 32); + off += 32; + memcpy(send_data + off, sig, 64); + off += 64; + write_u16_le(send_data + off, count); + off += 2; + write_u64_le(send_data + off, offset); + off += 8; + byte *response = NULL; + size_t resp_len = 0; + mp_client_error_t ret = mp_client_request(ip, port, send_data, send_len, &response, &resp_len); + free(send_data); + if (ret != MP_CLIENT_SUCCESS) { + free(response); + return ret; + } + if (resp_len < 11 || response[ 0 ] != queried_author_packets) { + free(response); + return MP_CLIENT_ERROR_INVALID_RESPONSE; + } + uint16_t resp_count = read_u16_le(response + 1); + uint64_t new_offset = read_u64_le(response + 3); + size_t expected_len = 11 + (size_t) resp_count * PACKET_SIZE; + if (resp_len != expected_len) { + free(response); + return MP_CLIENT_ERROR_INVALID_RESPONSE; + } + out_list->packets = malloc((size_t) resp_count * sizeof(struct data_packet)); + if (!out_list->packets) { + free(response); + return MP_CLIENT_ERROR_ALLOC_FAILED; + } + out_list->count = 0; + out_list->new_offset = new_offset; + for (uint16_t i = 0; i < resp_count; ++i) { + bool success; + struct data_packet packet + = s2data_packet(PACKET_SIZE, response + 11 + (size_t) i * PACKET_SIZE, &success); + if (success) { out_list->packets[ out_list->count++ ] = packet; } + } + free(response); + return MP_CLIENT_SUCCESS; +} \ No newline at end of file diff --git a/src/common.c b/src/common.c index 771c3e6..0fef6da 100644 --- a/src/common.c +++ b/src/common.c @@ -26,27 +26,35 @@ const uint16_t default_port = 8040; enum client_input { // ask server for its information (responds with audit_response) - audit_server = 0x9a, + audit_server = 0x10, // ask server for a random set of neighbors // (input uint8_t neighbor_count, responds with neighbors_info) - get_neighbors = 0x20, + get_neighbors = 0x11, // monitor incoming packets // (input struct firehose_request, responds with firehose_reply) - firehose = 0xf1, + firehose = 0x12, // give a packet to the relay // (input struct data_packet, responds with push_packet_result) - push_packet = 0x88, + push_packet = 0x13, // get a packet from the server // (input hash, responds with get_packet_result) - get_packet = 0x87, + get_packet = 0x14, // register a new neighbor relay to this server // (input struct relay_neighbor, responds with register_neighbor_result) - register_neighbor = 0x73, + register_neighbor = 0x15, + + // query packets for a specific author + // note: to prevent DoS, you must provide a signature by the requested author to be approved. + // the signature needs to be from a struct shaped `public_key + pub_sign_key + u16 count + u64 offset`. + // server must verify the signing key is of the author's pubkey and signing key by looking at the packet's metadata. + // (input public_key, input pub_sign_key, signature, u16 count, u64 offset) + // (responds with queried_author_packets) + query_packets = 0x16 }; enum server_reply { @@ -54,17 +62,19 @@ enum server_reply { error = 0x1, // type `struct relay_info` - audit_response = 0xaf, + audit_response = 0x17, // type `struct relay_neighbor[ neighbor_count ]` - neighbors_info = 0x43, + neighbors_info = 0x18, // type `uint64_t new_offset, struct data_packet[ count ]` - firehose_reply = 0x6c, + firehose_reply = 0x19, // type `bool result_code` - push_packet_result = 0x4a, + push_packet_result = 0x1a, // type `struct data_packet` - get_packet_result = 0x42, + get_packet_result = 0x1b, // type `bool result_code` - register_neighbor_result = 0x6e, + register_neighbor_result = 0x1c, + // type `u16 count, u64 new_offset, struct data_packet[ count ]` + queried_author_packets = 0x1d }; typedef uint8_t get_neighbors_request; @@ -221,7 +231,8 @@ struct relay_info s2relay_info(size_t len, byte *stream, bool *success) { for (int i = 0; i < 4; i++) { bool ok; - out.statistics.quick_connect[ i ] = s2relay_neighbor(1 + 1 + 1 + 4 + 16 + 2, stream + off, &ok); + out.statistics.quick_connect[ i ] + = s2relay_neighbor(1 + 1 + 1 + 4 + 16 + 2, stream + off, &ok); if (!ok) { *success = false; return (struct relay_info) { 0 }; diff --git a/src/packet_storage.c b/src/packet_storage.c index 1094afd..067c742 100644 --- a/src/packet_storage.c +++ b/src/packet_storage.c @@ -1,12 +1,15 @@ +#include #include #include #include +#include // dependency: zlib #include // #include "common.c" const uint64_t max_storage = 1 << 16; +const uint64_t max_authors = 1 << 14; #define hash_fmt \ "%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x" \ @@ -25,11 +28,96 @@ struct data_packet_entry { struct data_packet packet; uint64_t next; uint64_t prev; + uint64_t author_next; + uint64_t author_prev; bool is_full; }* packet_storage; -// TODO: yes, i know it's weird, but this has to be zero -#define minus_one ((uint64_t) 0) +struct author_entry { + public_key author; + uint64_t packet_head; + uint64_t next; + uint64_t prev; + bool is_full; +}* author_storage; + +pthread_mutex_t author_storage_mutex = PTHREAD_MUTEX_INITIALIZER; + +// TODO: set to zero so we can avoid allocating all the requested pages (init to zero) +#define invalid_index ((uint64_t) 0) + +void save_author_packet_hash(const public_key author, const hash packet_hash) { + const char* dir = "packets_by_author"; + mkdir(dir, 0755); + + char author_hex[ 65 ]; + sprintf(author_hex, hash_fmt, hash_ref(author)); + + char filename[ 256 ]; + snprintf(filename, sizeof(filename), "%s/%s.list", dir, author_hex); + + FILE* f = fopen(filename, "ab"); + if (!f) return; + fwrite(packet_hash, 32, 1, f); + fputc('\n', f); + fclose(f); +} + +uint64_t get_author_entry(const public_key auth_key, bool create) { + uint64_t hash_val; + memcpy(&hash_val, auth_key, sizeof(uint64_t)); + uint64_t start_index = hash_val % max_authors; + uint64_t cur_index = start_index; + + pthread_mutex_lock(&author_storage_mutex); + + struct author_entry* current = &author_storage[ cur_index ]; + while (true) { + if (current->is_full && memcmp(current->author, auth_key, 32) == 0) { + pthread_mutex_unlock(&author_storage_mutex); + return cur_index; + } + if (current->next == invalid_index) break; + cur_index = current->next; + current = &author_storage[ cur_index ]; + } + + if (!create) { + pthread_mutex_unlock(&author_storage_mutex); + return invalid_index; + } + + if (!current->is_full) { + memcpy(current->author, auth_key, 32); + current->packet_head = invalid_index; + current->is_full = true; + current->next = invalid_index; + current->prev = invalid_index; + pthread_mutex_unlock(&author_storage_mutex); + return cur_index; + } + + uint64_t probes = 0; + uint64_t lp_idx = (cur_index + 1) % max_authors; + while (probes < max_authors) { + struct author_entry* empty = &author_storage[ lp_idx ]; + if (!empty->is_full) { + memcpy(empty->author, auth_key, 32); + empty->packet_head = invalid_index; + empty->is_full = true; + empty->next = invalid_index; + empty->prev = cur_index; + current->next = lp_idx; + pthread_mutex_unlock(&author_storage_mutex); + return lp_idx; + } + lp_idx = (lp_idx + 1) % max_authors; + probes++; + } + + pthread_mutex_unlock(&author_storage_mutex); + return invalid_index; +} struct data_packet* load_packet(FILE* file) { if (!file) { return NULL; } @@ -79,7 +167,7 @@ struct data_packet_entry* get_data_packet(const hash _hash) { pthread_mutex_lock(&packet_storage_mutex); struct data_packet_entry* current_entry = &packet_storage[ cur_index ]; - while (cur_index != minus_one) { + while (cur_index != invalid_index) { if (current_entry->is_full && memcmp(current_entry->packet.hash, _hash, sizeof(hash)) == 0) { pthread_mutex_unlock(&packet_storage_mutex); @@ -87,7 +175,7 @@ struct data_packet_entry* get_data_packet(const hash _hash) { } cur_index = current_entry->next; - if (cur_index != minus_one) { current_entry = &packet_storage[ cur_index ]; } + if (cur_index != invalid_index) { current_entry = &packet_storage[ cur_index ]; } } pthread_mutex_unlock(&packet_storage_mutex); @@ -99,7 +187,7 @@ struct data_packet_entry* get_data_packet(const hash _hash) { if (packet) { uint64_t add_idx = set_data_packet(packet); - if (add_idx != minus_one) { + if (add_idx != invalid_index) { free(packet); return &packet_storage[ add_idx ]; } else { @@ -113,7 +201,7 @@ struct data_packet_entry* get_data_packet(const hash _hash) { return NULL; } -uint64_t set_data_packet(struct data_packet* new_packet) { +uint64_t __set_data_packet(struct data_packet* new_packet) { pthread_mutex_lock(&packet_storage_mutex); uint64_t hash_val; @@ -131,7 +219,7 @@ uint64_t set_data_packet(struct data_packet* new_packet) { return cur_index; } - if (current_entry->next == minus_one) break; + if (current_entry->next == invalid_index) break; cur_index = current_entry->next; current_entry = &packet_storage[ cur_index ]; } @@ -139,8 +227,8 @@ uint64_t set_data_packet(struct data_packet* new_packet) { if (!current_entry->is_full) { memcpy(¤t_entry->packet, new_packet, sizeof(struct data_packet)); current_entry->is_full = true; - current_entry->next = minus_one; - current_entry->prev = minus_one; + current_entry->next = invalid_index; + current_entry->prev = invalid_index; packet_count++; pthread_mutex_unlock(&packet_storage_mutex); return cur_index; @@ -155,7 +243,7 @@ uint64_t set_data_packet(struct data_packet* new_packet) { if (!empty_slot->is_full) { memcpy(&empty_slot->packet, new_packet, sizeof(struct data_packet)); empty_slot->is_full = true; - empty_slot->next = minus_one; + empty_slot->next = invalid_index; empty_slot->prev = cur_index; current_entry->next = linear_probe_index; packet_count++; @@ -169,7 +257,26 @@ uint64_t set_data_packet(struct data_packet* new_packet) { // table full pthread_mutex_unlock(&packet_storage_mutex); - return minus_one; + return invalid_index; +} + +uint64_t set_data_packet(struct data_packet* new_packet) { + uint64_t result = __set_data_packet(new_packet); + if (result == invalid_index) return invalid_index; + + save_author_packet_hash(new_packet->contents.author, new_packet->hash); + + uint64_t auth_idx = get_author_entry(new_packet->contents.author, true); + if (auth_idx != invalid_index) { + pthread_mutex_lock(&packet_storage_mutex); + uint64_t old_head = author_storage[ auth_idx ].packet_head; + packet_storage[ result ].author_next = old_head; + packet_storage[ result ].author_prev = invalid_index; + if (old_head != invalid_index) packet_storage[ old_head ].author_prev = result; + author_storage[ auth_idx ].packet_head = result; + pthread_mutex_unlock(&packet_storage_mutex); + } + return result; } #define overload __attribute__((overloadable)) @@ -184,7 +291,7 @@ overload void delete_data_packet(hash _hash) { uint64_t start_index = hash_val % max_storage; uint64_t cur_index = start_index; - while (cur_index != minus_one) { + while (cur_index != invalid_index) { struct data_packet_entry* current_entry = &packet_storage[ cur_index ]; if (current_entry->is_full && memcmp(current_entry->packet.hash, _hash, sizeof(hash)) == 0) { @@ -193,7 +300,7 @@ overload void delete_data_packet(hash _hash) { return; } - if (current_entry->next == minus_one) { + if (current_entry->next == invalid_index) { pthread_mutex_unlock(&packet_storage_mutex); return; } @@ -215,19 +322,133 @@ bool _delete_at_index(uint64_t index) { struct data_packet_entry* current_entry = &packet_storage[ index ]; if (!current_entry->is_full) return false; - if (current_entry->prev != minus_one) { + public_key auth_key; + memcpy(auth_key, packet_storage[ index ].packet.contents.author, 32); + uint64_t auth_idx = get_author_entry(auth_key, false); + if (auth_idx != invalid_index) { + if (author_storage[ auth_idx ].packet_head == index) { + author_storage[ auth_idx ].packet_head = packet_storage[ index ].author_next; + } + } + uint64_t a_prev = packet_storage[ index ].author_prev; + uint64_t a_next = packet_storage[ index ].author_next; + if (a_prev != invalid_index) packet_storage[ a_prev ].author_next = a_next; + if (a_next != invalid_index) packet_storage[ a_next ].author_prev = a_prev; + + if (current_entry->prev != invalid_index) { struct data_packet_entry* prev_entry = &packet_storage[ current_entry->prev ]; prev_entry->next = current_entry->next; } - if (current_entry->next != minus_one) { + if (current_entry->next != invalid_index) { struct data_packet_entry* next_entry = &packet_storage[ current_entry->next ]; next_entry->prev = current_entry->prev; } current_entry->is_full = false; - current_entry->next = minus_one; - current_entry->prev = minus_one; + current_entry->next = invalid_index; + current_entry->prev = invalid_index; return true; } + +int get_packets_for_author__cmp(const void* a, const void* b) { + uint64_t ta = ((struct data_packet*) a)->contents.created_time_ms; + uint64_t tb = ((struct data_packet*) b)->contents.created_time_ms; + return (ta > tb) ? -1 : (ta < tb) ? 1 : 0; +} + +uint16_t get_packets_for_author( + const public_key author, + uint64_t offset, + uint16_t max_count, + struct data_packet* out_packets, + uint64_t* out_new_offset) { + uint64_t auth_idx = get_author_entry(author, false); + if (auth_idx == invalid_index) return 0; + + // overfetch + uint64_t fetch_factor = ((offset + (uint64_t) max_count) * 5LLU) / 2; + + pthread_mutex_lock(&packet_storage_mutex); + uint64_t curr = author_storage[ auth_idx ].packet_head; + uint16_t mem_count = 0; + while (curr != invalid_index) { + mem_count++; + curr = packet_storage[ curr ].author_next; + } + + struct data_packet* all_packets + = malloc((mem_count + fetch_factor) * sizeof(struct data_packet)); + if (!all_packets) { + pthread_mutex_unlock(&packet_storage_mutex); + return 0; + } + size_t all_count = 0; + + curr = author_storage[ auth_idx ].packet_head; + while (curr != invalid_index) { + all_packets[ all_count++ ] = packet_storage[ curr ].packet; + curr = packet_storage[ curr ].author_next; + } + pthread_mutex_unlock(&packet_storage_mutex); + + char author_hex[ 65 ]; + sprintf(author_hex, hash_fmt, hash_ref(author)); + + char filename[ 256 ]; + snprintf(filename, sizeof(filename), "packets_by_author/%s.list", author_hex); + + FILE* f = fopen(filename, "rb"); + size_t disk_added = 0; + if (f) { + fseek(f, 0, SEEK_END); + long fsize = ftell(f); + fseek(f, 0, SEEK_SET); + uint64_t total = fsize / 33; + + uint64_t fetch_count = fetch_factor; + if (fetch_count > total) fetch_count = total; + + for (uint64_t i = 0; i < fetch_count; i++) { + long pos = (total - 1 - i) * 33; + fseek(f, pos, SEEK_SET); + + hash phash; + if (fread(phash, 32, 1, f) != 1) break; + if (fgetc(f) != '\n') break; + + bool dupe = false; + for (size_t j = 0; j < all_count; j++) { + if (memcmp(all_packets[ j ].hash, phash, 32) == 0) { + dupe = true; + break; + } + } + if (dupe) continue; + + struct data_packet_entry* ent = get_data_packet(phash); + if (!ent) continue; + + all_packets[ all_count++ ] = ent->packet; + disk_added++; + } + fclose(f); + } + + qsort(all_packets, all_count, sizeof(struct data_packet), get_packets_for_author__cmp); + + if (offset >= all_count) { + free(all_packets); + *out_new_offset = offset; + return 0; + } + + size_t avail = all_count - offset; + uint16_t collected = (avail > max_count) ? max_count : avail; + memcpy(out_packets, all_packets + offset, collected * sizeof(struct data_packet)); + *out_new_offset = offset + collected; + + free(all_packets); + return collected; +} \ No newline at end of file diff --git a/src/server.c b/src/server.c index 98680a3..71722ba 100644 --- a/src/server.c +++ b/src/server.c @@ -217,7 +217,7 @@ void paste_server( relay_info2s(info, output_data + 1); - reply(1 + 237, output_data); + reply(1 + 245, output_data); return; } @@ -428,7 +428,7 @@ void paste_server( uint64_t storage_index = set_data_packet(&packet); firehose_push(packet.hash, storage_index); - if (storage_index != minus_one) { + if (storage_index != invalid_index) { output_data[ 1 ] = true; save_packet(packet); @@ -563,6 +563,75 @@ void paste_server( reply(2, __reply); } break; + case query_packets: { + if (in_len != 1 + 32 + 32 + 64 + 2 + 8) { + *out = NULL; + *out_len = 0; + return; + } + public_key author_enc, author_sign; + signature sig; + size_t off = 1; + memcpy(author_enc, in + off, 32); + off += 32; + memcpy(author_sign, in + off, 32); + off += 32; + memcpy(sig, in + off, 64); + off += 64; + uint16_t req_count = read_u16_le(in + off); + off += 2; + uint64_t req_offset = read_u64_le(in + off); + off += 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_len = 0; + return; + } + if (req_count > 128) req_count = 128; + struct data_packet* packets = malloc((size_t) req_count * sizeof(struct data_packet)); + if (!packets) { + *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 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 ]; + } + } + 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_len = 0; + return; + } + 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); + off += PACKET_SIZE; + } + free(packets); + *out = _out; + *out_len = _out_len; + return; + } break; + default: { log(_info, "<%.*s> request type: invalid method", (int) ip_size, ip_addr); @@ -924,12 +993,13 @@ int main(int argc, char** argv) { // init packet_storage = alloc_map(max_storage * sizeof(struct data_packet_entry)); + author_storage = alloc_map(max_authors * sizeof(struct author_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) { + if (!packet_storage || !author_storage || !neighbor_storage || !empty_neighbor_slots) { perror("init alloc failed"); return 1; } -- cgit v1.2.3