From 47d4780eb448f839fc6b0644135395f879080ba4 Mon Sep 17 00:00:00 2001 From: Jesper Jensen Date: Sun, 23 Jul 2023 12:15:34 +0200 Subject: I don't remember --- src/main.c | 109 +++++++- src/main.c.orig | 202 +++++++++++++++ src/peers.c | 110 ++++++++ src/peers.h | 29 +++ src/proto.c | 125 ++++++---- src/proto.c.orig | 718 +++++++++++++++++++++++++++++++++++++++++++++++++++++ src/proto.h | 20 +- src/proto.h.orig | 78 ++++++ src/query.c | 285 ++++++++++++++++++++- src/query.h | 3 +- src/routing.c | 21 +- src/routing.c.orig | 267 ++++++++++++++++++++ src/routing.h | 7 +- test/peers.c | 104 ++++++++ test/proto.c | 112 ++++++++- test/query.c | 84 ++++++- test/routing.c | 8 +- 17 files changed, 2191 insertions(+), 91 deletions(-) create mode 100644 src/main.c.orig create mode 100644 src/peers.c create mode 100644 src/peers.h create mode 100644 src/proto.c.orig create mode 100644 src/proto.h.orig create mode 100644 src/routing.c.orig create mode 100644 test/peers.c diff --git a/src/main.c b/src/main.c index 20cd376..4a28a1d 100644 --- a/src/main.c +++ b/src/main.c @@ -1,9 +1,74 @@ #include "proto.h" +#include "peers.h" #include "log.h" #include #include #include +#include + +static volatile bool killed = false; +void sigint_handler(int sig) { + killed = true; +} + +#define CONF_ENO 1 + +void save_config() { + FILE* config = fopen("conf.dmp", "w"); + if(config == NULL) + fatal("Couldn't open config for writing"); + + if(fwrite(&myID, sizeof(struct nodeid), 1, config) != 1) + fatal("Couldn't write state"); + if(fwrite(table, sizeof(struct entry), table_size, config) != table_size) + fatal("Couldn't write state"); + + long pos = ftell(config); + dbg("Routing stops at 0x%04lX", pos); + + if(fwrite(&peer_table_size, sizeof(peer_table_size), 1, config) != 1) + fatal("Couldn't write peer table size"); + if(fwrite(&peer_table_load, sizeof(peer_table_load), 1, config) != 1) + fatal("Couldn't write peer table load"); + if(fwrite(peer_table, sizeof(struct peer_entry), peer_table_size, config) != peer_table_size) + fatal("Couldn't write peer table"); + + if(fclose(config) != 0) + fatal("Couldn't close config file"); +} + +int read_config() { + FILE* config = fopen("conf.dmp", "r"); + if(config == NULL) + return CONF_ENO; + + if(fread(&myID, sizeof(struct nodeid), 1, config) != 1) + fatal("Couldn't read routing table"); + if(fread(table, sizeof(struct entry), table_size, config) != table_size) + fatal("Couldn't read routing table"); + + if(fread(&peer_table_size, sizeof(peer_table_size), 1, config) != 1) + fatal("Couldn't read peer table"); + if(fread(&peer_table_load, sizeof(peer_table_load), 1, config) != 1) + fatal("Couldn't read peer table"); + + peer_table = malloc(sizeof(struct peer_entry) * peer_table_size); + assert(peer_table != NULL); + + if(fread(peer_table, sizeof(struct peer_entry), peer_table_size, config) != peer_table_size) + fatal("Couldn't read peer table"); + + long pos = ftell(config); + fseek(config, 0, SEEK_END); + if(pos != ftell(config)) + fatal("The config file was too long?"); + + if(fclose(config) != 0) + fatal("Couldn't close config file"); + + return 0; +} void flush_messages(int sfd, struct message* cursor, const struct message* const end) { dbg("Flushing %ld pending messages", end - cursor); @@ -11,7 +76,7 @@ void flush_messages(int sfd, struct message* cursor, const struct message* const //now reply the client with the same data int rc = sendto(sfd, cursor->payload, cursor->payload_len, 0, (const struct sockaddr*)&cursor->dest, cursor->dest_len); if (rc < 0) { - fatal("Failed to send message"); + fatal("Failed to send message %m"); } } } @@ -19,30 +84,43 @@ void flush_messages(int sfd, struct message* cursor, const struct message* const int main(int argc, char** argv) { struct message outbuff[32] = {0}; + struct sigaction sa; + sa.sa_handler = sigint_handler; + sigemptyset(&sa.sa_mask); + sa.sa_flags = SA_RESTART; + + if(sigaction(SIGINT, &sa, NULL) == -1) + fatal("Couldn't set signal handler"); + struct dht dht; - dht.self = (struct nodeid){.inner={0xebe9bbf1, 0x3cdba6b3, 0x993e0c87, 0x900d5e25}}; - routing_init(&dht.self); + { + int rc = read_config(); + if(rc == CONF_ENO) { + myID = (struct nodeid){.inner={0xebe9bbf1, 0x3cdba6b3, 0x993e0c87, 0x900d5e25}}; + routing_flush(); + allocate_hashtable(); + } + + dht.self = myID; + } struct message* message_cursor = outbuff; proto_begin(&dht, time(NULL), &message_cursor, outbuff+32); flush_messages(dht.sfd, outbuff, message_cursor); - char buff_storage[2049]; +#define RECV_BUFF_SIZE 4096 + char buff_storage[RECV_BUFF_SIZE+1]; int rc = 0; - while(rc == 0) { + while(rc == 0 && !killed) { char* buff = buff_storage; - printf("Waiting for data..."); - fflush(stdout); bool timedout = false; time_t next = 0; if(!dht.pause){ - struct entry* oldest; routing_oldest(&oldest); if(oldest != NULL) next = oldest->expire; - } else { dbg("DHT timeout is paused"); } @@ -75,18 +153,21 @@ int main(int argc, char** argv) { socklen_t remote_len = sizeof(remote); ssize_t recv_len; if(!timedout) { - recv_len = recvfrom(dht.sfd, buff, 2048, 0, (struct sockaddr *)&remote, &remote_len); + recv_len = recvfrom(dht.sfd, buff, RECV_BUFF_SIZE, 0, (struct sockaddr *)&remote, &remote_len); if(recv_len == -1) { // This is really strange. The man pages say we should be getting // an ETIMEDOUT here, but instead linux gives us this. if(errno == EAGAIN) { buff = NULL; recv_len = 0; + } else if(errno == EINTR) { + continue; } else { - fatal("RECV failed %d %m", errno, errno); + fatal("RECV failed %d %m", errno); } - } else if(recv_len >= 2048) { - fatal("Receive buffer too small"); + } else if(recv_len >= RECV_BUFF_SIZE) { + dbg("Receive buffer too small"); + continue; } // Null terminate the packet if(buff != NULL) { @@ -104,6 +185,8 @@ int main(int argc, char** argv) { } proto_end(&dht); + dbg("Writing out config"); + save_config(); return rc; } diff --git a/src/main.c.orig b/src/main.c.orig new file mode 100644 index 0000000..95dc2b9 --- /dev/null +++ b/src/main.c.orig @@ -0,0 +1,202 @@ +#include "proto.h" +#include "peers.h" +#include "log.h" + +#include +#include +#include +#include + +static volatile bool killed = false; +void sigint_handler(int sig) { + killed = true; +} + +#define CONF_ENO 1 + +void save_config() { + FILE* config = fopen("conf.dmp", "w"); + if(config == NULL) + fatal("Couldn't open config for writing"); + + if(fwrite(&myID, sizeof(struct nodeid), 1, config) != 1) + fatal("Couldn't write state"); + if(fwrite(table, sizeof(struct entry), table_size, config) != table_size) + fatal("Couldn't write state"); + + long pos = ftell(config); + dbg("Routing stops at 0x%04lX", pos); + + if(fwrite(&peer_table_size, sizeof(peer_table_size), 1, config) != 1) + fatal("Couldn't write peer table size"); + if(fwrite(&peer_table_load, sizeof(peer_table_load), 1, config) != 1) + fatal("Couldn't write peer table load"); + if(fwrite(peer_table, sizeof(struct peer_entry), peer_table_size, config) != peer_table_size) + fatal("Couldn't write peer table"); + + if(fclose(config) != 0) + fatal("Couldn't close config file"); +} + +int read_config() { + FILE* config = fopen("conf.dmp", "r"); + if(config == NULL) + return CONF_ENO; + + if(fread(&myID, sizeof(struct nodeid), 1, config) != 1) + fatal("Couldn't read routing table"); + if(fread(table, sizeof(struct entry), table_size, config) != table_size) + fatal("Couldn't read routing table"); + + if(fread(&peer_table_size, sizeof(peer_table_size), 1, config) != 1) + fatal("Couldn't read peer table"); + if(fread(&peer_table_load, sizeof(peer_table_load), 1, config) != 1) + fatal("Couldn't read peer table"); + + peer_table = malloc(sizeof(struct peer_entry) * peer_table_size); + assert(peer_table != NULL); + + if(fread(peer_table, sizeof(struct peer_entry), peer_table_size, config) != peer_table_size) + fatal("Couldn't read peer table"); + + long pos = ftell(config); + fseek(config, 0, SEEK_END); + if(pos != ftell(config)) + fatal("The config file was too long?"); + + if(fclose(config) != 0) + fatal("Couldn't close config file"); + + return 0; +} + +void flush_messages(int sfd, struct message* cursor, const struct message* const end) { + dbg("Flushing %ld pending messages", end - cursor); + for(; cursor < end; cursor++) { + //now reply the client with the same data + int rc = sendto(sfd, cursor->payload, cursor->payload_len, 0, (const struct sockaddr*)&cursor->dest, cursor->dest_len); + if (rc < 0) { + fatal("Failed to send message %m"); + } + } +} + +int main(int argc, char** argv) { + struct message outbuff[32] = {0}; + + struct sigaction sa; + sa.sa_handler = sigint_handler; + sigemptyset(&sa.sa_mask); + sa.sa_flags = SA_RESTART; + + if(sigaction(SIGINT, &sa, NULL) == -1) + fatal("Couldn't set signal handler"); + + struct dht dht; +<<<<<<< HEAD + dht.self = (struct nodeid){.inner={0xebe9bbf1, 0x3cdba6b3, 0x993e0c87, 0x900d5e25}}; + routing_init(&dht.self); +======= + { + int rc = read_config(); + if(rc == CONF_ENO) { + myID = (struct nodeid){.inner={0xebe9bbf1, 0x3cdba6b3, 0x993e0c87, 0x900d5e25}}; + routing_flush(); + allocate_hashtable(); + } + + dht.self = myID; + } +>>>>>>> 685b13e (I don't remember) + + struct message* message_cursor = outbuff; + proto_begin(&dht, time(NULL), &message_cursor, outbuff+32); + flush_messages(dht.sfd, outbuff, message_cursor); + +<<<<<<< HEAD + char buff_storage[2049]; +======= + +#define RECV_BUFF_SIZE 4096 + char buff_storage[RECV_BUFF_SIZE+1]; +>>>>>>> 685b13e (I don't remember) + int rc = 0; + while(rc == 0 && !killed) { + char* buff = buff_storage; + + bool timedout = false; + time_t next = 0; + if(!dht.pause){ + struct entry* oldest; + routing_oldest(&oldest); + if(oldest != NULL) + next = oldest->expire; + } else { + dbg("DHT timeout is paused"); + } + + for(int i = 0; i < MAX_INFLIGHT; i++) { + if(!dht.reqalloc[i]) + continue; + + time_t timeout = dht.requestdata[i].timeout; + if(next == 0 || (timeout != 0 && difftime(timeout, next) < 0)) + next = timeout; + } + + if(next != 0) { + time_t sleepfor = next - time(NULL); + dbg("Set timeout to %ld", sleepfor); + struct timeval tv = { + .tv_sec = sleepfor, + .tv_usec = 0, + }; + if(tv.tv_sec <= 0) { + timedout = true; + } else { + setsockopt(dht.sfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + } + } + + // Try to receive some data, this is a blocking call + struct sockaddr_storage remote; + socklen_t remote_len = sizeof(remote); + ssize_t recv_len; + if(!timedout) { + recv_len = recvfrom(dht.sfd, buff, RECV_BUFF_SIZE, 0, (struct sockaddr *)&remote, &remote_len); + if(recv_len == -1) { + // This is really strange. The man pages say we should be getting + // an ETIMEDOUT here, but instead linux gives us this. + if(errno == EAGAIN) { + buff = NULL; + recv_len = 0; + } else if(errno == EINTR) { + continue; + } else { + fatal("RECV failed %d %m", errno); + } + } else if(recv_len >= RECV_BUFF_SIZE) { + dbg("Receive buffer too small"); + continue; + } + // Null terminate the packet + if(buff != NULL) { + buff[recv_len] = '\0'; + } + } else { + buff = NULL; + recv_len = 0; + } + + struct message* message_cursor = outbuff; + time_t now = time(NULL); + rc = proto_run(&dht, buff, recv_len, (struct sockaddr_in*)&remote, remote_len, now, &message_cursor, outbuff+10); + flush_messages(dht.sfd, outbuff, message_cursor); + } + + proto_end(&dht); + dbg("Writing out config"); + save_config(); + + return rc; +} diff --git a/src/peers.c b/src/peers.c new file mode 100644 index 0000000..e1825c1 --- /dev/null +++ b/src/peers.c @@ -0,0 +1,110 @@ +#include "peers.h" + +#include "log.h" + +#include +#include +#include + +struct peer_entry* peer_table; +size_t peer_table_size; +size_t peer_table_load; + +int allocate_hashtable() { + peer_table_load = 0; + peer_table_size = 16; + peer_table = calloc(peer_table_size, sizeof(struct peer_entry)); + if(peer_table == NULL) + return PEER_ENOMEM; + + return 0; +} + +static uint64_t hash(struct infohash* key, size_t size) { + uint64_t hash = (key->inner[4] << 4) | key->inner[3]; + return hash % size; +} + +static void find(struct peer_entry* table, size_t size, struct infohash* infohash, struct peer_entry** entry) { + uint64_t key = hash(infohash, size); + do { + *entry = &table[key++]; + key %= size; + } while((*entry)->set && memcmp(&(*entry)->key, infohash, sizeof(struct infohash)) != 0); +} + +static int resize(size_t new_size) { + assert((new_size & (new_size - 1)) == 0); // Power of two + assert(new_size >= peer_table_load); + + struct peer_entry* new_table = calloc(new_size, sizeof(struct peer_entry)); + if(peer_table == NULL) + return PEER_ENOMEM; + + for(size_t i = 0; i < peer_table_size; i++) { + struct peer_entry* entry = &peer_table[i]; + + struct peer_entry* new_entry = NULL; + find(new_table, new_size, &entry->key, &new_entry); + + assert(!new_entry->set); + *new_entry = *entry; + } + + free(peer_table); + + peer_table = new_table; + peer_table_size = new_size; + return 0; +}; + +static double load_factor(size_t size, size_t load) { + return (double)load / (double)size; +} + +uint64_t next_pow2(uint64_t x) { + if(x == 1) return 1; + uint16_t leading = __builtin_clzl(x-1); + return 1 << (64 - leading); +} + +int add_peer(struct infohash* infohash, struct addr* peer) { + assert(peer_table_load < peer_table_size); + + if(load_factor(peer_table_size, peer_table_load + 1) > 0.75) { + int rc = resize(peer_table_size * 2); + if(rc != 0) return rc; + } + + struct peer_entry* entry = NULL; + find(peer_table, peer_table_size, infohash, &entry); + + if(!entry->set) { + entry->set = true; + entry->key = *infohash; + peer_table_load++; + } + + size_t peern = entry->value_len; + if(peern == PEERS_PER_HASH) + return PEER_EFULL; + assert(peern < PEERS_PER_HASH); + entry->value[peern] = *peer; + entry->value_len++; + + return 0; +} + +void get_peers(struct infohash* infohash, struct addr *peers[PEERS_PER_HASH], size_t *peers_len) { + struct peer_entry* entry; + find(peer_table, peer_table_size, infohash, &entry); + + if(!entry->set) { + *peers = NULL; + *peers_len = 0; + return; + } + + *peers = entry->value; + *peers_len = entry->value_len; +} diff --git a/src/peers.h b/src/peers.h new file mode 100644 index 0000000..624cb21 --- /dev/null +++ b/src/peers.h @@ -0,0 +1,29 @@ +#pragma once + +#include "routing.h" + +#define PEER_ENOMEM 1 +#define PEER_EFULL 1 + +struct infohash { + union { + uint32_t inner[5]; + char inner_b[20]; + }; +}; + +#define PEERS_PER_HASH 8 +struct peer_entry { + struct infohash key; + struct addr value[PEERS_PER_HASH]; + size_t value_len; + bool set; +}; + +extern struct peer_entry* peer_table; +extern size_t peer_table_size; +extern size_t peer_table_load; + +int allocate_hashtable(); +int add_peer(struct infohash* key, struct addr* peer); +void get_peers(struct infohash* infohash, struct addr *peers[PEERS_PER_HASH], size_t *peers_len); diff --git a/src/proto.c b/src/proto.c index b518f79..2191d33 100644 --- a/src/proto.c +++ b/src/proto.c @@ -20,6 +20,7 @@ #include #include #include +#include #define MAX(a, b) \ ({ \ @@ -35,6 +36,9 @@ _a < _b ? _a : _b; \ }) +#define CLAMP(a, b, c) \ + MAX(MIN(a, c), b) + void dbgl_id(struct nodeid* id) { for(uint8_t i = 0; i < 5; i++) { @@ -99,15 +103,31 @@ struct msgbuff { PROCESS_REPONSE(getclient_response); PROCESS_TIMEOUT(getclient_timeout); +#if UINT8_MAX > RAND_MAX +#error UINT8_MAX is larger than RAND_MAX +#endif uint8_t rand_byte() { - int limit = RAND_MAX - (RAND_MAX % UINT8_MAX); + int limit = (RAND_MAX / UINT8_MAX)*UINT8_MAX; + int val; + while((val = rand()) >= limit); + + return val % UINT8_MAX; +} + +// Number of nodeid bits +#define IDBITS 160 +#if IDBITS > RAND_MAX +#error IDBITS is larger than RAND_MAX +#endif +uint8_t rand_bucket() { + int limit = (RAND_MAX / IDBITS)*IDBITS; int val; - while((val = rand()) > limit); + while((val = rand()) >= limit); - return val; + return val % IDBITS; } -int write_ping(char* buff, size_t* buff_len, struct nodeid* self, struct nodeid* target, uint16_t tid) { +int write_find_node(char* buff, size_t* buff_len, struct nodeid* self, struct nodeid* target, uint16_t tid) { char* buff_end = buff + *buff_len; int rc = snprintf(buff, buff_end - buff, "d1:ad2:id20:"); @@ -120,9 +140,9 @@ int write_ping(char* buff, size_t* buff_len, struct nodeid* self, struct nodeid* if(rc < 0) fatal("Failed to write packet"); buff += rc; - memcpy(buff, &target, sizeof(struct nodeid)); + memcpy(buff, target, sizeof(struct nodeid)); buff += sizeof(struct nodeid); - rc = snprintf(buff, buff_end - buff, "e1:q9:find_node1:t%d:%d1:y1:qe", (int)(log10(tid+1)+1), tid); + rc = snprintf(buff, buff_end - buff, "e1:q9:find_node1:t%d:%d1:y1:qe", tid == 0 ? 1 : (int)(log10(tid)+1), tid); if(rc < 0) fatal("Failed to write packet"); buff += rc; @@ -131,6 +151,14 @@ int write_ping(char* buff, size_t* buff_len, struct nodeid* self, struct nodeid* return 0; } +struct nodeid random_node() { + struct nodeid target; + for(uint8_t *target_byte = (uint8_t*)⌖ target_byte < ((uint8_t*)&target)+sizeof(target); target_byte++) { + *target_byte = rand_byte(); + } + return target; +} + int send_ping(struct dht* dht, struct nodeid* expected, time_t now, bool node_is_new, const struct sockaddr* dest_addr, socklen_t dest_len, struct msgbuff* msgbuff) { if(*msgbuff->messages >= msgbuff->messages_end) return PROTO_ENOREQ; @@ -140,7 +168,6 @@ int send_ping(struct dht* dht, struct nodeid* expected, time_t now, bool node_is if(!alloc_req(dht, &reqId)) { return PROTO_ENOREQ; } - dbg("Allocating request %d", reqId); memcpy(&message->dest, dest_addr, dest_len); message->dest_len = dest_len; @@ -154,19 +181,14 @@ int send_ping(struct dht* dht, struct nodeid* expected, time_t now, bool node_is dht->requestdata[reqId].fun = &getclient_response; dht->requestdata[reqId].timeout = now + PROTO_TMOUT; - dht->requestdata[reqId].timeout_fun = getclient_timeout; + dht->requestdata[reqId].timeout_fun = &getclient_timeout; memcpy(&dht->requestdata[reqId].addr, dest_addr, dest_len); dht->requestdata[reqId].addr_len = dest_len; - // Generate a random target - struct nodeid target; - for(uint8_t *target_byte = (uint8_t*)⌖ target_byte < ((uint8_t*)&target)+sizeof(target); target_byte++) { - *target_byte = rand_byte(); - } - target = dht->self; + struct nodeid target = random_node(); - message->payload_len = 128; - int rc = write_ping(message->payload, &message->payload_len, &dht->self, &target, reqId); + message->payload_len = sizeof(message->payload); + int rc = write_find_node(message->payload, &message->payload_len, &dht->self, &target, reqId); if(rc != 0) { return rc; } @@ -184,8 +206,6 @@ PROCESS_TIMEOUT(getclient_timeout) { if(cont->ping.is_new) return 0; - dbg("Discarding node that didn't respond"); - routing_remove(&cont->ping.remote_id); return 0; } @@ -199,14 +219,10 @@ PROCESS_TIMEOUT(getclient_timeout) { memcpy(&message->dest, &dht->requestdata[reqId].addr, dht->requestdata[reqId].addr_len); message->dest_len = dht->requestdata[reqId].addr_len; - // Generate a random target - struct nodeid target; - for(uint8_t *target_byte = (uint8_t*)⌖ target_byte < ((uint8_t*)&target)+sizeof(target); target_byte++) { - *target_byte = rand_byte(); - } + struct nodeid target = random_node(); - message->payload_len = 128; - int rc = write_ping(message->payload, &message->payload_len, &dht->self, &target, reqId); + message->payload_len = sizeof(message->payload); + int rc = write_find_node(message->payload, &message->payload_len, &dht->self, &target, reqId); if(rc != 0) { fatal("Can't create ping"); } @@ -267,7 +283,6 @@ PROCESS_REPONSE(getclient_response) { } nodes_len = MIN(bcursor.readhead->size/26, 8); - dbg("Remote gave us %d new candidates", nodes_len); for(int i = 0; i < nodes_len; i++) { memcpy(nodes+i, bcursor.readhead->loc+(26*i), 20); memcpy(ips+i, bcursor.readhead->loc+(26*i)+20, 4); @@ -320,14 +335,18 @@ PROCESS_REPONSE(getclient_response) { } } else { struct entry* entry = routing_get(&id); - assert(entry != NULL); - entry->expire = now + PROTO_UNCTM; + // @CLEANUP: Figure out why this can be null. Is the node getting + // removed while we are waiting for a response? + if(entry != NULL) { + entry->expire = now + PROTO_UNCTM; + } } + uint8_t accepted = 0; // Fan out the search if the results were interesting for(uint8_t i = 0; i < nodes_len; i++) { - dbgl_id(&nodes[i]); - printf("Candidate %s:%d\n", inet_ntoa(ips[i]), ntohs(ports[i])); + // @ROBUST: Some nodes report a bunch of nodes in the same ip. Maybe we + // could check for that here struct sockaddr_in dest = { .sin_family = AF_INET, @@ -336,17 +355,18 @@ PROCESS_REPONSE(getclient_response) { }; if(routing_interested(&nodes[i])) { + accepted++; int rc = send_ping(dht, &nodes[i], now, true, (struct sockaddr*)&dest, sizeof(struct sockaddr_in), msgbuff); if(rc == PROTO_ENOREQ) { return rc; } else if(rc != 0) { fatal("send_ping failed %d", rc); } - } else { - dbg("Not interested in node"); } } + dbg("Node provided %d nodes. %d of them were useful", nodes_len, accepted); + return 0; } @@ -415,7 +435,6 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran // Temporary null terminate the string to parse the number without a copy char* end; - dbg("Transaction %s", transaction); transaction_number = strtol(transaction, &end, 10); if(end != transaction+transaction_len) { @@ -428,7 +447,7 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran err("DISCARD: unknown transaction id %d", transaction_number); return 0; } - dbg("Transaction id matches request %d", reqId); + dbg("Request %d gets a response", reqId); if(sockaddr_cmp((struct sockaddr*)&dht->requestdata[reqId].addr, (struct sockaddr*)remote) != 0) { err("DISCARD: Unexpected IP for valid transaction"); @@ -452,6 +471,7 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran fatal("No query function in query request"); if(transaction == NULL) fatal("No transaction in request"); + assert(transaction_len <= 16); assert(strlen(query) == query_len); @@ -472,8 +492,10 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran fatal("No space for response"); cursor += rc; - rc = handle_request(&dht->self, query, packet, packet_len, &cursor, end-cursor-1); + dbg("===== HANDLE %s ====", query); + rc = handle_request(&dht->self, query, (const struct sockaddr*)remote, remote_len, packet, packet_len, &cursor, end-cursor-1); if(rc == QUERY_EUNK) { + dbg("Unknown method"); // @FRAGILE: @HACK: Static offsets to fiddle with already written // out packet data. Acceptable because this is the uncommon error // case. @@ -503,6 +525,8 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran memcpy(&message->dest, remote, remote_len); message->dest_len = remote_len; (*msgbuff->messages)++; + } else if(type == CT_ERROR) { + fatal("Unhandled error"); } else { fatal("HOW"); } @@ -517,6 +541,7 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* }; if(recv_len == 0 && buff == NULL) { + uint8_t timedout = 0; for(int i = 0; i < MAX_INFLIGHT; i++) { if(!dht->reqalloc[i]) continue; @@ -540,13 +565,15 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* dht->pause = false; } } + if(timedout != 0) { + dbg("processed %d requests that timed out", timedout); + } struct entry* oldest = NULL; routing_oldest(&oldest); while(oldest != NULL) { if(difftime(now, oldest->expire) < 0) break; - dbg("Node becomes uncertain"); struct sockaddr_in dest = {0}; dest.sin_family = AF_INET; @@ -558,7 +585,6 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* return 0; } else if(rc != 0) { fatal("NOPE %d", rc); - return 0; } oldest->expire = 0; @@ -567,12 +593,10 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* return 0; } - printf("Received packet from %s:%d\n", inet_ntoa(remote->sin_addr), ntohs(remote->sin_port)); struct bcursor bcursor; struct benc_node stream[256]; bcur_open(&bcursor, buff, buff+recv_len, stream, 256); - benc_print(bcursor.readhead, bcursor.end - bcursor.readhead); if(bcursor.readhead->type != BNT_DICT) { fatal("First value is not a dict"); @@ -638,30 +662,37 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* assert(rc == 0); { - size_t allocated = 0; + printf("In flight |"); for(int i = 0; i < MAX_INFLIGHT; i++) { if(dht->reqalloc[i]) { - allocated++; + printf("#"); + } else { + printf(" "); } } - - dbg("%ld/%ld requests pending", allocated, MAX_INFLIGHT); + printf("|\n"); } { int filled; int total; -#define LFACLEN 32 +#define LFACLEN 64 double load_factor[LFACLEN] = {0}; routing_status(&filled, &total, load_factor, LFACLEN); dbg("%d/%d nodes in routing table", filled, total); -#define GRAPHY 10 +#define GRAPHY 5 + // @HACK @CLEANUP: I'm pretty zooted right now. I have zero confidence + // that this is correct. It looks allright though. for(int y = 0; y < GRAPHY; y++) { + printf("|"); for(int x = 0; x < LFACLEN; x++) { - if(load_factor[x] > (1.0/GRAPHY) * (GRAPHY-y)) { + double cell_load = CLAMP((load_factor[x] - ((1.0/GRAPHY) * (GRAPHY-y-1))) * GRAPHY, 0, 1); + if(cell_load == 0.0) { + printf(" "); + } else if (cell_load > 1.0 - 1.0/GRAPHY) { printf("#"); } else { - printf(" "); + printf("%d", (int)(cell_load*10)); } } printf("|\n"); diff --git a/src/proto.c.orig b/src/proto.c.orig new file mode 100644 index 0000000..f65a4f4 --- /dev/null +++ b/src/proto.c.orig @@ -0,0 +1,718 @@ +#include "proto.h" + +#include "benc.h" +#include "query.h" +#include "log.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + punt = true; + +#define MAX(a, b) \ + ({ \ + __typeof__ (a) _a = (a); \ + __typeof__ (b) _b = (b); \ + _a > _b ? _a : _b; \ + }) + +#define MIN(a, b) \ + ({ \ + __typeof__ (a) _a = (a); \ + __typeof__ (b) _b = (b); \ + _a < _b ? _a : _b; \ + }) + +#define CLAMP(a, b, c) \ + MAX(MIN(a, c), b) + + +void dbgl_id(struct nodeid* id) { + for(uint8_t i = 0; i < 5; i++) { + fprintf(stderr, "0x%08x ", id->inner[i]); + } + fprintf(stderr, "\n"); + fflush(stderr); +} + +int sockaddr_cmp(struct sockaddr* x, struct sockaddr* y) { +#define CMP(a, b) \ + do { \ + typeof(a) cmp = a - b; \ + if(cmp != 0) return cmp; \ + } while(0) + if (x->sa_family == AF_INET) { + struct sockaddr_in *xin = (void*)x; + struct sockaddr_in *yin = (void*)y; + + CMP(ntohl(xin->sin_addr.s_addr), ntohl(yin->sin_addr.s_addr)); + CMP(ntohs(xin->sin_port), ntohs(yin->sin_port)); + } else if (x->sa_family == AF_INET6) { + struct sockaddr_in6 *xin6 = (void*)x, *yin6 = (void*)y; + int r = memcmp(xin6->sin6_addr.s6_addr, yin6->sin6_addr.s6_addr, sizeof(xin6->sin6_addr.s6_addr)); + if (r != 0) + return r; + CMP(ntohs(xin6->sin6_port), ntohs(yin6->sin6_port)); + CMP(xin6->sin6_flowinfo, yin6->sin6_flowinfo); + CMP(xin6->sin6_scope_id, yin6->sin6_scope_id); + } else { + err("Unsupported sa_family"); + abort(); + } + + return 0; +}; + +bool alloc_req(struct dht* dht, uint16_t* reqId) { + for(size_t i = 0; i < MAX_INFLIGHT; i++) { + if(!dht->reqalloc[i]) { + dht->reqalloc[i] = true; + *reqId = i; + return true; + } + } + return false; +} + +bool find_req(struct dht* dht, uint32_t transId, uint16_t* reqId) { + *reqId = transId; + return dht->reqalloc[transId]; +} + +struct msgbuff { + struct message** messages; + const struct message* const messages_end; +}; + +#define PROTO_EDISC 1 +#define PROTO_ENOREQ 2 + +PROCESS_REPONSE(getclient_response); +PROCESS_TIMEOUT(getclient_timeout); + +#if UINT8_MAX > RAND_MAX +#error UINT8_MAX is larger than RAND_MAX +#endif +uint8_t rand_byte() { + int limit = (RAND_MAX / UINT8_MAX)*UINT8_MAX; + int val; + while((val = rand()) >= limit); + + return val % UINT8_MAX; +} + +// Number of nodeid bits +#define IDBITS 160 +#if IDBITS > RAND_MAX +#error IDBITS is larger than RAND_MAX +#endif +uint8_t rand_bucket() { + int limit = (RAND_MAX / IDBITS)*IDBITS; + int val; + while((val = rand()) >= limit); + + return val % IDBITS; +} + +int write_find_node(char* buff, size_t* buff_len, struct nodeid* self, struct nodeid* target, uint16_t tid) { + char* buff_end = buff + *buff_len; + + int rc = snprintf(buff, buff_end - buff, "d1:ad2:id20:"); + if(rc < 0) + fatal("Failed to write packet"); + buff += rc; + memcpy(buff, self, sizeof(struct nodeid)); + buff += sizeof(struct nodeid); + rc = snprintf(buff, buff_end - buff, "6:target20:"); + if(rc < 0) + fatal("Failed to write packet"); + buff += rc; + memcpy(buff, target, sizeof(struct nodeid)); + buff += sizeof(struct nodeid); + rc = snprintf(buff, buff_end - buff, "e1:q9:find_node1:t%d:%d1:y1:qe", tid == 0 ? 1 : (int)(log10(tid)+1), tid); + if(rc < 0) + fatal("Failed to write packet"); + buff += rc; + + *buff_len = buff - (buff_end - *buff_len); + return 0; +} + +struct nodeid random_node() { + struct nodeid target; + for(uint8_t *target_byte = (uint8_t*)⌖ target_byte < ((uint8_t*)&target)+sizeof(target); target_byte++) { + *target_byte = rand_byte(); + } + return target; +} + +int send_ping(struct dht* dht, struct nodeid* expected, time_t now, bool node_is_new, const struct sockaddr* dest_addr, socklen_t dest_len, struct msgbuff* msgbuff) { + if(*msgbuff->messages >= msgbuff->messages_end) + return PROTO_ENOREQ; + struct message* message = *msgbuff->messages; + + uint16_t reqId; + if(!alloc_req(dht, &reqId)) { + return PROTO_ENOREQ; + } + + memcpy(&message->dest, dest_addr, dest_len); + message->dest_len = dest_len; + + struct ping* data = &dht->requestdata[reqId].cont.ping; + if(!node_is_new) { + data->remote_id = *expected; + } + data->is_new = node_is_new; + data->attempt = 0; + + dht->requestdata[reqId].fun = &getclient_response; + dht->requestdata[reqId].timeout = now + PROTO_TMOUT; + dht->requestdata[reqId].timeout_fun = &getclient_timeout; + memcpy(&dht->requestdata[reqId].addr, dest_addr, dest_len); + dht->requestdata[reqId].addr_len = dest_len; + + struct nodeid target = random_node(); + + message->payload_len = sizeof(message->payload); + int rc = write_find_node(message->payload, &message->payload_len, &dht->self, &target, reqId); + if(rc != 0) { + return rc; + } + (*msgbuff->messages)++; + + return 0; +} + +PROCESS_TIMEOUT(getclient_timeout) { + // @HACK: This really sucks. maybe we should just pass in the request id + size_t reqId = (typeof(dht->requestdata[0])*)((void*)cont - offsetof(typeof(dht->requestdata[0]), cont)) - dht->requestdata; + + if(cont->ping.attempt >= 2) { + dbg("Timing out request %ld after %d attempts", reqId, cont->ping.attempt); + if(cont->ping.is_new) + return 0; + + routing_remove(&cont->ping.remote_id); + return 0; + } + +<<<<<<< HEAD + dbg("Retrying request %ld", reqId); + +======= +>>>>>>> 685b13e (I don't remember) + if(*msgbuff->messages >= msgbuff->messages_end) + return PROTO_ENOREQ; + struct message* message = *msgbuff->messages; + + memcpy(&message->dest, &dht->requestdata[reqId].addr, dht->requestdata[reqId].addr_len); + message->dest_len = dht->requestdata[reqId].addr_len; + + struct nodeid target = random_node(); + + message->payload_len = sizeof(message->payload); + int rc = write_find_node(message->payload, &message->payload_len, &dht->self, &target, reqId); + if(rc != 0) { + fatal("Can't create ping"); + } + (*msgbuff->messages)++; + + dht->requestdata[reqId].timeout = now + PROTO_TMOUT; + cont->ping.attempt++; + return PROTO_EDISC; +} + +PROCESS_REPONSE(getclient_response) { + struct benc_node stream[256]; + struct bcursor bcursor; + bcur_open(&bcursor, packet, packet+packet_len, stream, 256); + + if(bcursor.end - bcursor.readhead <= 0) { + fatal("Response too short"); + } + + struct nodeid id; + uint8_t nodes_len; + struct nodeid nodes[8]; + struct in_addr ips[8]; + uint16_t ports[8]; + + // Read the payload + { + // Check that we have a dict + if(bcursor.readhead->type != BNT_DICT) { + fatal("Response is not a dict"); + } + bcur_next(&bcursor, 1); + + bcur_find_key(&bcursor, (const enum benc_nodetype[]){BNT_STRING}, (const char*[]){"r"}, (const size_t[]){1}, 1); + // Skip the key + bcur_next(&bcursor, 1); + + if(bcursor.readhead->type != BNT_DICT) { + fatal("Wrong value type for response"); + } + + // Skip the dict element + bcur_next(&bcursor, 1); + + uint8_t parts = 0; + while(bcursor.readhead->type != BNT_END) { + switch(bcur_find_key(&bcursor, (const enum benc_nodetype[]){BNT_STRING, BNT_STRING}, (const char*[]){"nodes", "id"}, (const size_t[]){5, 2}, 2)) { + case 0: + // Skip the key + bcur_next(&bcursor, 1); + + if(bcursor.readhead->type != BNT_STRING) { + fatal("Nodes must be a string"); + } + + if((bcursor.readhead->size % 26) != 0) { + fatal("Nodes string value must be a multiple of 26"); + } + + nodes_len = MIN(bcursor.readhead->size/26, 8); + for(int i = 0; i < nodes_len; i++) { + memcpy(nodes+i, bcursor.readhead->loc+(26*i), 20); + memcpy(ips+i, bcursor.readhead->loc+(26*i)+20, 4); + memcpy(ports+i, bcursor.readhead->loc+(26*i)+24, 2); + } + + parts++; + + // Skip the value + bcur_next(&bcursor, 1); + break; + case 1: + // Skip the key + bcur_next(&bcursor, 1); + + if(bcursor.readhead->type != BNT_STRING) { + fatal("Wrong value type for response"); + } + + if(bcursor.readhead->size != 20) { + fatal("remote node id was not 20 bytes long"); + } + + memcpy(&id, bcursor.readhead->loc, 20); + + parts++; + + // Skip the value + bcur_next(&bcursor, 1); + break; + } + } + + if(parts < 2) { + err("Response didn't contain nodes and id"); + return PROTO_EDISC; + } + } + + // The response was good, so save the node + if(cont->ping.is_new) { + struct entry* entry; + if(routing_offer(&id, &entry)) { + struct sockaddr_in* ipv4 = (struct sockaddr_in*)remote; + entry->addr.ip = ipv4->sin_addr.s_addr; + entry->addr.port = ipv4->sin_port; + entry->expire = now + PROTO_UNCTM; + } else { + dbg("We are no longer interested"); + } + } else { + struct entry* entry = routing_get(&id); +<<<<<<< HEAD + assert(entry != NULL); + entry->expire = now + PROTO_UNCTM; +======= + // @CLEANUP: Figure out why this can be null. Is the node getting + // removed while we are waiting for a response? + if(entry != NULL) { + entry->expire = now + PROTO_UNCTM; + } +>>>>>>> 685b13e (I don't remember) + } + + uint8_t accepted = 0; + // Fan out the search if the results were interesting + for(uint8_t i = 0; i < nodes_len; i++) { + // @ROBUST: Some nodes report a bunch of nodes in the same ip. Maybe we + // could check for that here + + struct sockaddr_in dest = { + .sin_family = AF_INET, + .sin_addr = ips[i], + .sin_port = ports[i], + }; + + if(routing_interested(&nodes[i])) { + accepted++; + int rc = send_ping(dht, &nodes[i], now, true, (struct sockaddr*)&dest, sizeof(struct sockaddr_in), msgbuff); + if(rc == PROTO_ENOREQ) { + return rc; + } else if(rc != 0) { + fatal("send_ping failed %d", rc); + } + } + } + + dbg("Node provided %d nodes. %d of them were useful", nodes_len, accepted); + + return 0; +} + +enum commandType { + CT_QUERY, + CT_RESPONSE, + CT_ERROR, +}; + +void proto_begin(struct dht* dht, time_t now, struct message** output, const struct message* const output_end) { + struct msgbuff msgbuff = { + output, + output_end, + }; + dht->pause = false; + + for(int i = 0; i < MAX_INFLIGHT; i++) { + dht->reqalloc[i] = false; + } + + dbgl_id(&dht->self); + + dht->sfd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + if(dht->sfd == -1) { + err("Failed creating socket"); + exit(1); + } + struct sockaddr_in bindAddr = {0}; + bindAddr.sin_family = AF_INET; + bindAddr.sin_port = htons(6881); + bindAddr.sin_addr.s_addr = htonl(INADDR_ANY); + bind(dht->sfd, (struct sockaddr*)&bindAddr, sizeof(struct sockaddr_in)); + + struct addrinfo hints = {0}; + hints.ai_family = AF_INET; + hints.ai_socktype = SOCK_DGRAM; + hints.ai_protocol = IPPROTO_UDP; + hints.ai_flags = AI_NUMERICSERV; + + struct addrinfo* res; + int rc = getaddrinfo("router.bittorrent.com", "6881", &hints, &res); + if(rc != 0) { + err("Failed getting the bootstrap ip: %s", gai_strerror(rc)); + exit(EXIT_FAILURE); + } + + for(struct addrinfo* cur = res; cur != NULL; cur = cur->ai_next) { + send_ping(dht, NULL, now, true, cur->ai_addr, cur->ai_addrlen, &msgbuff); + } + + freeaddrinfo(res); +} + +void proto_end(struct dht* dht) { + close(dht->sfd); +} + +int handle_packet(struct dht* dht, time_t now, enum commandType type, char* transaction, size_t transaction_len, char* query, size_t query_len, char* packet, size_t packet_len, struct sockaddr_in* remote, socklen_t remote_len, struct msgbuff* msgbuff) { + if(type == CT_RESPONSE) { + uint32_t transaction_number; + + if(transaction == NULL) { + err("DISCARD: No transaction in response"); + return 0; + } + + // Temporary null terminate the string to parse the number without a copy + char* end; + transaction_number = strtol(transaction, &end, 10); + + if(end != transaction+transaction_len) { + err("DISCARD: Transaction id is not a number %.*s", (int)transaction_len, transaction); + return 0; + } + + uint16_t reqId; + if(!find_req(dht, transaction_number, &reqId)) { + err("DISCARD: unknown transaction id %d", transaction_number); + return 0; + } + dbg("Request %d gets a response", reqId); + + if(sockaddr_cmp((struct sockaddr*)&dht->requestdata[reqId].addr, (struct sockaddr*)remote) != 0) { + err("DISCARD: Unexpected IP for valid transaction"); + return 0; + } + + dht->pause = false; + int rc = dht->requestdata[reqId].fun(dht, now, &dht->requestdata[reqId].cont, packet, packet_len, dht->sfd, (struct sockaddr*)remote, remote_len, msgbuff); + if(rc == PROTO_ENOREQ) { + dht->pause = true; + } else if(rc == PROTO_EDISC) { + return 0; + } + + dht->requestdata[reqId].fun = NULL; + dht->requestdata[reqId].timeout_fun = NULL; + dht->requestdata[reqId].timeout = 0; + dht->reqalloc[reqId] = false; + } else if(type == CT_QUERY) { // Must be a query + if(query == NULL) + fatal("No query function in query request"); + if(transaction == NULL) + fatal("No transaction in request"); + assert(transaction_len <= 16); + + assert(strlen(query) == query_len); + + assert(*msgbuff->messages < msgbuff->messages_end); + struct message* message = *msgbuff->messages; + + char* end = message->payload+sizeof(message->payload); + char* cursor = message->payload; + + int rc = snprintf(cursor, end-cursor , "d1:t%ld:", transaction_len); + if(rc < 0) + fatal("No space for response"); + cursor += rc; + memcpy(cursor, transaction, transaction_len); + cursor += transaction_len; + rc = snprintf(cursor, end-cursor, "1:y1:r1:r"); + if(rc < 0) + fatal("No space for response"); + cursor += rc; + + dbg("===== HANDLE %s ====", query); + rc = handle_request(&dht->self, query, (const struct sockaddr*)remote, remote_len, packet, packet_len, &cursor, end-cursor-1); + if(rc == QUERY_EUNK) { + dbg("Unknown method"); + // @FRAGILE: @HACK: Static offsets to fiddle with already written + // out packet data. Acceptable because this is the uncommon error + // case. + // The y key should have value e + *(cursor-4) = 'e'; + // The r key is called e for errors + *(cursor-1) = 'e'; + + // Now create the payload + rc = snprintf(cursor, end-cursor, "li204e14:Unknown Methode"); + if(rc < 0) + fatal("No space for response"); + cursor += rc; + assert(cursor < end); + + // Use the normal finalize flow + } else if(rc != 0) fatal("Error handling request"); + + rc = snprintf(cursor, end-cursor, "e"); + if(rc < 0) + fatal("No space for response"); + cursor += rc; + assert(cursor < end); + + message->payload_len = cursor - message->payload; + + memcpy(&message->dest, remote, remote_len); + message->dest_len = remote_len; + (*msgbuff->messages)++; + } else if(type == CT_ERROR) { + fatal("Unhandled error"); + } else { + fatal("HOW"); + } + + return 0; +} + +int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* remote, socklen_t remote_len, time_t now, struct message** output, const struct message* const output_end) { + struct msgbuff msgbuff = { + output, + output_end, + }; + + if(recv_len == 0 && buff == NULL) { + uint8_t timedout = 0; + for(int i = 0; i < MAX_INFLIGHT; i++) { + if(!dht->reqalloc[i]) + continue; + if(dht->requestdata[i].timeout == 0) + continue; + if(difftime(now, dht->requestdata[i].timeout) < 0) + continue; + + int rc = dht->requestdata[i].timeout_fun(dht, &dht->self, now, &dht->requestdata[i].cont, &msgbuff); + + if(rc == PROTO_ENOREQ) { + dht->pause = true; + return 0; + } + if(rc != PROTO_EDISC) { + dht->requestdata[i].fun = NULL; + dht->requestdata[i].timeout_fun = NULL; + dht->requestdata[i].timeout = 0; + dht->reqalloc[i] = false; + + dht->pause = false; + } + } + if(timedout != 0) { + dbg("processed %d requests that timed out", timedout); + } + + struct entry* oldest = NULL; + routing_oldest(&oldest); + while(oldest != NULL) { + if(difftime(now, oldest->expire) < 0) + break; + + struct sockaddr_in dest = {0}; + dest.sin_family = AF_INET; + dest.sin_addr.s_addr = oldest->addr.ip; + dest.sin_port = oldest->addr.port; + int rc = send_ping(dht, &oldest->id, now, false, (const struct sockaddr*)&dest, sizeof(dest), &msgbuff); + if(rc == PROTO_ENOREQ) { + dht->pause = true; + return 0; + } else if(rc != 0) { + fatal("NOPE %d", rc); + } + + oldest->expire = 0; + routing_oldest(&oldest); + } + + return 0; + } + + struct bcursor bcursor; + struct benc_node stream[256]; + bcur_open(&bcursor, buff, buff+recv_len, stream, 256); + + if(bcursor.readhead->type != BNT_DICT) { + fatal("First value is not a dict"); + } + bcur_next(&bcursor, 1); + + bool discard = false; + enum commandType type; + bool transaction_set = false; + char transaction[64]; + size_t transaction_len; + bool query_set = false; + char query[64]; + size_t query_len; + while(bcursor.readhead->type != BNT_END) { + switch(bcur_find_key(&bcursor, (const enum benc_nodetype[]){BNT_STRING, BNT_STRING, BNT_STRING}, (const char*[]){"y", "t", "q"}, (const size_t[]){1, 1, 1}, 3)) { + case 0: + // Skip the key + bcur_next(&bcursor, 1); + if(*bcursor.readhead->loc == 'r') { + type = CT_RESPONSE; + } else if(*bcursor.readhead->loc == 'q') { + type = CT_QUERY; + } else if(*bcursor.readhead->loc == 'e') { + type = CT_ERROR; + } else { + fatal("Unknown command type %c", *bcursor.readhead->loc); + } + // Skip the value + bcur_next(&bcursor, 1); + break; + case 1: { + // Skip the key + bcur_next(&bcursor, 1); + if(bcursor.readhead->size > 64-1) + fatal("Transaction string too long"); + + transaction_set = true; + transaction_len = bcursor.readhead->size; + memcpy(transaction, bcursor.readhead->loc, transaction_len); + transaction[transaction_len] = '\0'; + + // Skip the value + bcur_next(&bcursor, 1); + break; + } + case 2: { + bcur_next(&bcursor, 1); + query_set = true; + query_len = bcursor.readhead->size; + memcpy(query, bcursor.readhead->loc, query_len); + query[query_len] = '\0'; + bcur_next(&bcursor, 1); + } + } + } + + if(discard){ + return 0; + } + + int rc = handle_packet(dht, now, type, transaction_set ? transaction : NULL, transaction_len, query_set ? query : NULL, query_len, buff, recv_len, remote, remote_len, &msgbuff); + assert(rc == 0); + + { + printf("In flight |"); + for(int i = 0; i < MAX_INFLIGHT; i++) { + if(dht->reqalloc[i]) { + printf("#"); + } else { + printf(" "); + } + } +<<<<<<< HEAD + + dbg("%ld/%ld requests pending", allocated, MAX_INFLIGHT); +======= + printf("|\n"); +>>>>>>> 685b13e (I don't remember) + } + { + int filled; + int total; +#define LFACLEN 64 + double load_factor[LFACLEN] = {0}; + routing_status(&filled, &total, load_factor, LFACLEN); + dbg("%d/%d nodes in routing table", filled, total); + +#define GRAPHY 5 + // @HACK @CLEANUP: I'm pretty zooted right now. I have zero confidence + // that this is correct. It looks allright though. + for(int y = 0; y < GRAPHY; y++) { + printf("|"); + for(int x = 0; x < LFACLEN; x++) { + double cell_load = CLAMP((load_factor[x] - ((1.0/GRAPHY) * (GRAPHY-y-1))) * GRAPHY, 0, 1); + if(cell_load == 0.0) { + printf(" "); + } else if (cell_load > 1.0 - 1.0/GRAPHY) { + printf("#"); + } else { + printf("%d", (int)(cell_load*10)); + } + } + printf("|\n"); + } +#undef GRAPHY +#undef LFACLEN + } + + return 0; +} diff --git a/src/proto.h b/src/proto.h index 1bdc05d..0ce887e 100644 --- a/src/proto.h +++ b/src/proto.h @@ -26,7 +26,7 @@ struct dht; struct msgbuff; #define PROCESS_REPONSE(NAME) int (NAME)(struct dht* dht, time_t now, union message_cont* cont, char* packet, size_t packet_len, int socket, struct sockaddr* remote, socklen_t remote_len, struct msgbuff* msgbuff) -typedef PROCESS_REPONSE(cont); +typedef PROCESS_REPONSE(resp); #define PROCESS_TIMEOUT(NAME) int (NAME)(struct dht* dht, struct nodeid* self, time_t now, union message_cont* cont, struct msgbuff* msgbuff) typedef PROCESS_TIMEOUT(tmout); @@ -41,15 +41,29 @@ struct dht { struct { struct sockaddr_storage addr; socklen_t addr_len; - cont* fun; + resp* fun; time_t timeout; tmout* timeout_fun; union message_cont cont; } requestdata[MAX_INFLIGHT]; }; +// @HACK: This isn't true anymore +// The longest message we can send is probably a response to find_node which +// consists of: +// d1:t <-- 4 bytes +// <-- 2 bytes +// : <-- 1 byte +// <-- 16 bytes* +// 1:y1:r1:r <-- 9 bytes +// d2:id20: <-- 8 bytes +// <-- 20 bytes +// 5:nodes208: <-- 11 bytes +// <-- 208 bytes +// ee <-- 2 bytes +// TOTAL 281 bytes struct message { - char payload[1024]; + char payload[281]; size_t payload_len; struct sockaddr_storage dest; socklen_t dest_len; diff --git a/src/proto.h.orig b/src/proto.h.orig new file mode 100644 index 0000000..ce2a713 --- /dev/null +++ b/src/proto.h.orig @@ -0,0 +1,78 @@ +#pragma once + +#include "routing.h" +#include +#include + +// 192.0.2.0 +#define UNDEF_ADDR (struct in_addr){0xC0000200} +#define MAX_DISC 32 +#define MAX_INFLIGHT 32 + +#define PROTO_UNCTM 60 +#define PROTO_TMOUT 5 + +struct ping { + struct nodeid remote_id; + int attempt; + bool is_new; +}; + +union message_cont { + struct ping ping; +}; + +struct dht; +struct msgbuff; + +#define PROCESS_REPONSE(NAME) int (NAME)(struct dht* dht, time_t now, union message_cont* cont, char* packet, size_t packet_len, int socket, struct sockaddr* remote, socklen_t remote_len, struct msgbuff* msgbuff) +typedef PROCESS_REPONSE(resp); + +#define PROCESS_TIMEOUT(NAME) int (NAME)(struct dht* dht, struct nodeid* self, time_t now, union message_cont* cont, struct msgbuff* msgbuff) +typedef PROCESS_TIMEOUT(tmout); + +struct dht { + struct nodeid self; + int sfd; + + bool pause; + + bool reqalloc[MAX_INFLIGHT]; + struct { + struct sockaddr_storage addr; + socklen_t addr_len; + resp* fun; + time_t timeout; + tmout* timeout_fun; + union message_cont cont; + } requestdata[MAX_INFLIGHT]; +}; + +// @HACK: This isn't true anymore +// The longest message we can send is probably a response to find_node which +// consists of: +// d1:t <-- 4 bytes +// <-- 2 bytes +// : <-- 1 byte +// <-- 16 bytes* +// 1:y1:r1:r <-- 9 bytes +// d2:id20: <-- 8 bytes +// <-- 20 bytes +// 5:nodes208: <-- 11 bytes +// <-- 208 bytes +// ee <-- 2 bytes +// TOTAL 281 bytes +struct message { +<<<<<<< HEAD + char payload[1024]; +======= + char payload[281]; +>>>>>>> 685b13e (I don't remember) + size_t payload_len; + struct sockaddr_storage dest; + socklen_t dest_len; +}; + +void proto_begin(struct dht* dht, time_t now, struct message** output, const struct message* const output_end); +int proto_run(struct dht* dht, char* buffer, size_t buffer_len, struct sockaddr_in* remote, socklen_t remote_len, time_t now, struct message** output, const struct message* const output_end); +void proto_end(struct dht* dht); diff --git a/src/query.c b/src/query.c index 6012540..6ca6565 100644 --- a/src/query.c +++ b/src/query.c @@ -1,12 +1,14 @@ #include "query.h" #include "benc.h" #include "log.h" +#include "peers.h" #include #include #include +#include -int handle_request(struct nodeid* self, const char* method, const char* packet, size_t packet_len, char** response, size_t response_len) { +int handle_request(struct nodeid* self, const char* method, const struct sockaddr* src, socklen_t src_len, const char* packet, size_t packet_len, char** response, size_t response_len) { if(strcmp(method, "ping") == 0) { struct bcursor bcursor; struct benc_node stream[256]; @@ -173,6 +175,287 @@ int handle_request(struct nodeid* self, const char* method, const char* packet, *response += rc; assert(*response < end); } else if(strcmp(method, "get_peers") == 0) { + struct bcursor bcursor; + struct benc_node stream[256]; + bcur_open(&bcursor, packet, packet+packet_len, stream, 256); + + if(bcursor.readhead->type != BNT_DICT) { + err("Bad query: Packet is not a dict"); + return QUERY_EBADQ; + } + if(bcur_next(&bcursor, 1) < 0) { + err("Bad query: No token after outer dict start"); + return QUERY_EBADQ; + } + + if(bcur_find_key(&bcursor, (const enum benc_nodetype[]){BNT_STRING}, (const char*[]){"a"}, (const size_t[]){1}, 1) != 0) { + err("Bad query: No arguments to request"); + return QUERY_EBADQ; + } + bcur_next(&bcursor, 1); + if(bcursor.readhead->type != BNT_DICT) { + err("Bad query: Wrong value type for request"); + return QUERY_EBADQ; + } + // Skip the dict element + bcur_next(&bcursor, 1); + bool infohash_set = false; + struct infohash infohash; + while(bcursor.readhead->type != BNT_END) { + switch(bcur_find_key(&bcursor, (const enum benc_nodetype[]){BNT_STRING}, (const char*[]){"info_hash"}, (const size_t[]){9}, 1)) { + case 0: + // Skip the key + bcur_next(&bcursor, 1); + + if(bcursor.readhead->type != BNT_STRING) { + err("Bad query: Wrong value type for info_hash"); + return QUERY_EBADQ; + } + + if(bcursor.readhead->size != 20) { + err("Bad query: Incorrect target length"); + return QUERY_EBADQ; + } + + infohash_set = true; + memcpy(&infohash, bcursor.readhead->loc, 20); + + // Skip the value + bcur_next(&bcursor, 1); + break; + } + } + + if(!infohash_set) { + err("info_hash argument not provided"); + return QUERY_EBADQ; + } + + char* end = (*response) + response_len; + + int rc = snprintf(*response, end-*response, "d2:id20:"); + if(rc < 0) + return QUERY_EBADQ; + *response += rc; + assert(*response < end); + + memcpy(*response, self, sizeof(struct nodeid)); + *response += sizeof(struct nodeid); + assert(*response < end); + + rc = snprintf(*response, end-*response, "5:token1:t"); + if(rc < 0) + return QUERY_EBADQ; + *response += rc; + assert(*response < end); + + struct addr* peers; + size_t peers_len; + get_peers(&infohash, &peers, &peers_len); + + if(peers != NULL) { + rc = snprintf(*response, end-*response, "6:valuesl"); + if(rc < 0) + return QUERY_EBADQ; + *response += rc; + assert(*response < end); + + for(int i = 0; i < peers_len; i++) { + *(*response) = '6'; + *(*response+1) = ':'; + (*response) += 2; + memcpy(*response, &peers[i].ip, sizeof(uint32_t)); + *response += sizeof(uint32_t); // 4 + assert(*response < end); + memcpy(*response, &peers[i].port, sizeof(uint16_t)); + *response += sizeof(uint16_t); // 2 + assert(*response < end); + } + + (**response) = 'e'; + (*response)++; + assert(*response < end); + } else { + // If we didn't get any peers we send back the closest nodes + struct entry* closest[8]; + int found = routing_closest((struct nodeid*)&infohash, 8, closest); + + rc = snprintf(*response, end-*response, "5:nodes%d:", found*26); + if(rc < 0) + return QUERY_EBADQ; + *response += rc; + assert(*response < end); + + for(int i = 0; i < found; i++) { + memcpy(*response, &closest[i]->id, sizeof(struct nodeid)); + *response += sizeof(struct nodeid); // 20 + assert(*response < end); + memcpy(*response, &closest[i]->addr.ip, sizeof(uint32_t)); + *response += sizeof(uint32_t); // 4 + assert(*response < end); + memcpy(*response, &closest[i]->addr.port, sizeof(uint16_t)); + *response += sizeof(uint16_t); // 2 + assert(*response < end); + } + } + + rc = snprintf(*response, end-*response, "e"); + if(rc < 0) + return QUERY_EBADQ; + *response += rc; + assert(*response < end); + } else if(strcmp(method, "announce_peer") == 0) { + struct bcursor bcursor; + struct benc_node stream[256]; + bcur_open(&bcursor, packet, packet+packet_len, stream, 256); + + if(bcursor.readhead->type != BNT_DICT) { + err("Bad query: Packet is not a dict"); + return QUERY_EBADQ; + } + if(bcur_next(&bcursor, 1) < 0) { + err("Bad query: No token after outer dict start"); + return QUERY_EBADQ; + } + + if(bcur_find_key(&bcursor, (const enum benc_nodetype[]){BNT_STRING}, (const char*[]){"a"}, (const size_t[]){1}, 1) != 0) { + err("Bad query: No arguments to request"); + return QUERY_EBADQ; + } + bcur_next(&bcursor, 1); + if(bcursor.readhead->type != BNT_DICT) { + err("Bad query: Wrong value type for request"); + return QUERY_EBADQ; + } + // Skip the dict element + bcur_next(&bcursor, 1); + + bool implied_port = false; + + bool infohash_set = false; + struct infohash infohash; + + bool port_set = false; + uint16_t port; + + bool token_set = false; + char token; + + while(bcursor.readhead->type != BNT_END) { + switch(bcur_find_key(&bcursor, (const enum benc_nodetype[]){BNT_STRING, BNT_STRING, BNT_STRING, BNT_STRING}, (const char*[]){"implied_port", "info_hash", "port", "token"}, (const size_t[]){12, 9, 4, 5}, 4)) { + case 0: + // Skip the key + bcur_next(&bcursor, 1); + + if(bcursor.readhead->type != BNT_INT) { + err("Bad query: Wrong value type for implied_port"); + return QUERY_EBADQ; + } + + if(*bcursor.readhead->loc == '0') { + implied_port = false; + } else { + implied_port = true; + } + + // Skip the value + bcur_next(&bcursor, 1); + break; + case 1: + // Skip the key + bcur_next(&bcursor, 1); + + if(bcursor.readhead->type != BNT_STRING) { + err("Bad query: Wrong value type for info_hash"); + return QUERY_EBADQ; + } + + if(bcursor.readhead->size != 20) { + err("Bad query: Incorrect info_hash length"); + return QUERY_EBADQ; + } + + infohash_set = true; + memcpy(&infohash, bcursor.readhead->loc, 20); + + bcur_next(&bcursor, 1); + break; + case 2: + // Skip the key + bcur_next(&bcursor, 1); + + if(bcursor.readhead->type != BNT_INT) { + err("Bad query: Wrong value type for port"); + return QUERY_EBADQ; + } + + port_set = true; + port = strtol(bcursor.readhead->loc, NULL, 10); + + bcur_next(&bcursor, 1); + break; + case 3: + // Skip the key + bcur_next(&bcursor, 1); + + if(bcursor.readhead->type != BNT_STRING) { + err("Bad query: Wrong value type for token"); + return QUERY_EBADQ; + } + + if(bcursor.readhead->size != 1) { + err("Bad query: Incorrect token length"); + return QUERY_EBADQ; + } + + token_set = true; + token = *bcursor.readhead->loc; + + bcur_next(&bcursor, 1); + break; + } + } + + if(!infohash_set || (!implied_port && !port_set) || !token_set) { + err("Missing argument to query"); + return QUERY_EBADQ; + } + + if(token != 't') { + err("Invalid token"); + return QUERY_EBADQ; + } + + { + struct sockaddr_in* ipv4 = (struct sockaddr_in*)src; + struct addr src_addr; + src_addr.ip = ipv4->sin_addr.s_addr; + + src_addr.port = ipv4->sin_port; + if(!implied_port) { + src_addr.port = htons(port); + } + add_peer(&infohash, &src_addr); + } + + // Write out the response + char* end = (*response) + response_len; + + int rc = snprintf(*response, end-*response, "d2:id20:"); + if(rc < 0) + return QUERY_EBADQ; + *response += rc; + assert(*response < end); + + memcpy(*response, self, sizeof(struct nodeid)); + *response += sizeof(struct nodeid); + assert(*response < end); + + (**response) = 'e'; + (*response)++; + assert(*response < end); + + return 0; } else { return QUERY_EUNK; } diff --git a/src/query.h b/src/query.h index c1977b3..2a58e25 100644 --- a/src/query.h +++ b/src/query.h @@ -3,8 +3,9 @@ #include "routing.h" #include +#include #define QUERY_EBADQ 1 #define QUERY_EUNK 2 -int handle_request(struct nodeid* self, const char* method, const char* packet, size_t packet_len, char** response, size_t response_len); +int handle_request(struct nodeid* self, const char* method, const struct sockaddr* src, socklen_t src_len, const char* packet, size_t packet_len, char** response, size_t response_len); diff --git a/src/routing.c b/src/routing.c index 313a7cc..4851dda 100644 --- a/src/routing.c +++ b/src/routing.c @@ -7,6 +7,7 @@ #include #include #include +#include // The DHT routing table has a keyspace of 0 -- 2^160 split into buckets of 8. // When a bucket becomes full, we split it in half. As we further expand the @@ -53,14 +54,18 @@ void routing_flush() { static uint8_t prefix(struct nodeid* a, struct nodeid* b) { uint8_t c = 0; for(uint8_t i = 0; i < 5; i++) { - uint32_t word = a->inner[i] ^ b->inner[i]; - - // This word is different, find the location of the difference - if(word != 0) - return c + __builtin_clz(word); - - // This word is completely the same - c += sizeof(word) * CHAR_BIT; + // Since the nodeids are stored in host byteorder in the words we have + // to make sure they're big endian before doing the prefix match, + // otherwise we end up with prefix matching that's different from the + // rest of the network + uint32_t word = htonl(a->inner[i]) ^ htonl(b->inner[i]); + + // This word is different, find the location of the difference + if (word != 0) + return c + __builtin_clz(word); + + // This word is completely the same + c += sizeof(word) * CHAR_BIT; } return c; diff --git a/src/routing.c.orig b/src/routing.c.orig new file mode 100644 index 0000000..596548a --- /dev/null +++ b/src/routing.c.orig @@ -0,0 +1,267 @@ +#include "routing.h" + +#include "log.h" + +#include +#include +#include +#include +#include + +// The DHT routing table has a keyspace of 0 -- 2^160 split into buckets of 8. +// When a bucket becomes full, we split it in half. As we further expand the +// routing table we only continue to split the buckets on the side we fall on. +// +// Initially, this may sound like a binary tree (because we split it in two), +// but looking at it as a flat array leads to some interesting intuitions. +// Since we only expand one half of the "tree", the total size is bounded by +// the depth of the tree log2(2^160) == 160. +// +// As a flat array we notice the intrinsic properties of the routing table. +// With a bucket size of 8, the routing table contains 160 * 8 == 1280 nodes. +// As the node ids get less similar to our own our grouping of them becomes +// less detailed. While the bucket we are in contains node very close to us, +// the nodes furthest away from us are grouped in buckets with nodes they +// barely resemble. +// +// +----------------------------+ +// | n1 | n2 | n3 | ... | n1280 | +// +----------------------------+ +// More Less +// <--------Similarity--------> +// <----------Detail----------> +// + +<<<<<<< HEAD +struct table { + struct nodeid myID; + struct entry table[RT_SIZE]; +}; + +struct table* pTable; +======= +#include + +#define IDBITS 160 +#define BUCKETSIZE 8 +// The 3 here is log2(BUCKETSIZE), since the final bucket will contain all those combinations +#define BUCKETBITS 3 +#define ROUTINGSIZE (IDBITS * BUCKETSIZE) + +struct nodeid myID; +struct entry table[ROUTINGSIZE]; +int table_size = ROUTINGSIZE; +>>>>>>> 685b13e (I don't remember) + +void routing_init(struct nodeid* myid) { + pTable = malloc(sizeof(struct table)); + pTable->myID = *myid; + routing_flush(); +} + +void routing_flush() { + memset(pTable->table, 0, sizeof(pTable->table)); +} + +// Calculate the common bit prefix between two node ids. +static uint8_t prefix(struct nodeid* a, struct nodeid* b) { + uint8_t c = 0; + for(uint8_t i = 0; i < 5; i++) { + // Since the nodeids are stored in host byteorder in the words we have + // to make sure they're big endian before doing the prefix match, + // otherwise we end up with prefix matching that's different from the + // rest of the network + uint32_t word = htonl(a->inner[i]) ^ htonl(b->inner[i]); + + // This word is different, find the location of the difference + if (word != 0) + return c + __builtin_clz(word); + + // This word is completely the same + c += sizeof(word) * CHAR_BIT; + } + + return c; +} + +static int8_t scan(uint16_t baseIndex, struct nodeid* id) { + assert(baseIndex < RT_SIZE - RT_BSIZE); + int8_t index = -2; + + for(size_t i = baseIndex; i < baseIndex + RT_BSIZE; i++) { + if(!pTable->table[i].set) { + index = index == -2 ? i - baseIndex : index; + continue; + } + + if(memcmp(&pTable->table[i].id, id, sizeof(struct nodeid)) == 0) { + return -1; + } + } + + return index; +} + +static uint16_t base_bucket(struct nodeid* id) { + uint16_t bucketIndex = prefix(&pTable->myID, id); + assert(bucketIndex != RT_IDBITS); + + // If they are sufficiently similar they end up in the final bucket. Clamp the index to ensure. + bucketIndex = bucketIndex > (RT_IDBITS - RT_BBITS) ? (RT_IDBITS - RT_BBITS) : bucketIndex; + assert(bucketIndex <= RT_IDBITS - RT_BBITS); + + return bucketIndex * RT_BSIZE; +} + +struct entry* routing_get(struct nodeid* id) { + uint16_t baseIndex = base_bucket(id); + for(size_t i = baseIndex; i < baseIndex + RT_BSIZE; i++) { + if(!pTable->table[i].set) continue; + + if(memcmp(&pTable->table[i].id, id, sizeof(struct nodeid)) == 0) { + return &pTable->table[i]; + } + } + + return NULL; +} + +void routing_remove(struct nodeid* id) { + struct entry* entry = routing_get(id); + + entry->set = false; +} + +bool routing_interested(struct nodeid* id) { + uint16_t bucketIndex = prefix(&pTable->myID, id); + // The nodeid is the same as our own + if(bucketIndex == RT_IDBITS) { + return false; + } + + uint16_t baseIndex = base_bucket(id); + int8_t inBucketIndex = scan(baseIndex, id); + + if(inBucketIndex < 0) { + // The bucket either already contains the node, or it has no more space + return false; + } + + return true; +} + +// Offer the routing table a new node +bool routing_offer(struct nodeid* id, struct entry **dest) { + uint16_t bucketIndex = prefix(&pTable->myID, id); + // The nodeid is the same as our own + if(bucketIndex == RT_IDBITS) { + return false; + } + + uint16_t baseIndex = base_bucket(id); + int8_t inBucketIndex = scan(baseIndex, id); + + if(inBucketIndex < 0) { + // The bucket either already contains the node, or it has no more space + return false; + } + + struct entry* entry = &pTable->table[baseIndex + inBucketIndex]; + entry->set = true; + entry->id = *id; + + *dest = entry; + return true; +} + +struct item { + struct nodeid distance; + bool set; + uint16_t index; +}; +int compareItem(const void* a_v, const void* b_v) { + struct item* a = (struct item*)a_v; + struct item* b = (struct item*)b_v; + + // If either of the two are not set, the one that is set comes before the + // one that isn't. + if(!a->set || !b->set) return b->set - a->set; + + return memcmp(&a->distance, &b->distance, sizeof(struct nodeid)); +} + +size_t routing_closest(struct nodeid* needle, size_t n, struct entry** res) { + assert(n <= RT_SIZE); + static struct item items[RT_SIZE] = {0}; + for(uint16_t i = 0; i < RT_SIZE; i++) { + items[i].index = i; + } + + { + struct item* item; + struct entry* entry; + for(item = &items[0], entry = &pTable->table[0]; item < &items[RT_SIZE] && entry < &pTable->table[RT_SIZE]; item++, entry++){ + item->set = entry->set; + for(uint8_t j = 0; j < 5; j++) { + item->distance.inner[j] = entry->id.inner[j] ^ needle->inner[j]; + } + } + } + + // @PERFORMANCE: There's an algorithm known as quickselect which can select + // the top k elements from a list while only doing a partial sort. + // I imagine that would be more efficient than this full sort. + qsort(items, RT_SIZE, sizeof(struct item), compareItem); + + size_t read; + for(read = 0; read < n; read++) { + if(!items[read].set) + break; + res[read] = &pTable->table[items[read].index]; + } + + return read; +} + +void routing_oldest(struct entry** dest) { + *dest = NULL; + + for(struct entry* entry = pTable->table; entry < pTable->table+RT_SIZE; entry++){ + if(!entry->set) + continue; + + if(entry->expire == 0) + continue; + + if(*dest == NULL) { + *dest = entry; + continue; + } + + if(difftime((*dest)->expire, entry->expire) > 0.0) { + *dest = entry; + } + } +} + +void routing_status(int* filled, int* size, double* load_factor, size_t load_factor_len) { + *size = RT_SIZE; + + *filled = 0; + for(size_t i = 0; i < RT_SIZE; i++) { + if(pTable->table[i].set) + (*filled)++; + } + + int per_bucket = RT_SIZE / load_factor_len; + int overflow = RT_SIZE % load_factor_len; + struct entry* table_cursor = pTable->table; + for(int i = 0; i < load_factor_len; i++) { + int is_overflow = i < overflow; + for(int j = 0; j < per_bucket + is_overflow; j++) { + load_factor[i] += table_cursor->set; + table_cursor++; + } + load_factor[i] /= per_bucket + is_overflow; + } +} diff --git a/src/routing.h b/src/routing.h index f4164fe..bbb4c24 100644 --- a/src/routing.h +++ b/src/routing.h @@ -27,10 +27,15 @@ struct nodeid { struct entry { bool set; struct nodeid id; - time_t expire; struct addr addr; + + time_t expire; }; +extern struct nodeid myID; +extern struct entry table[]; +extern int table_size; + void routing_init(struct nodeid* myid); void routing_flush(); bool routing_interested(struct nodeid* id); diff --git a/test/peers.c b/test/peers.c new file mode 100644 index 0000000..1e9a454 --- /dev/null +++ b/test/peers.c @@ -0,0 +1,104 @@ +#include "unity.h" +#include "peers.h" +#include "log.h" + +#include +#include + +#define IP(a, b, c, d) htonl(a << 24 | b << 16 | c << 8 | d) + +void test_add_single_peer() { + allocate_hashtable(); + + struct infohash sometorrent = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; + struct addr addr = (struct addr){.ip = IP(128,0,0,1), .port = 0}; + + int rc = add_peer(&sometorrent, &addr); + TEST_ASSERT_EQUAL(0, rc); +} + +void test_add_9_peers_same_torrent() { + allocate_hashtable(); + + struct infohash sometorrent = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; + struct addr addr = (struct addr){.ip = IP(128,0,0,1), .port = 0}; + + int rc = add_peer(&sometorrent, &addr); + TEST_ASSERT_EQUAL(0, rc); + rc = add_peer(&sometorrent, &addr); + TEST_ASSERT_EQUAL(0, rc); + rc = add_peer(&sometorrent, &addr); + TEST_ASSERT_EQUAL(0, rc); + rc = add_peer(&sometorrent, &addr); + TEST_ASSERT_EQUAL(0, rc); + rc = add_peer(&sometorrent, &addr); + TEST_ASSERT_EQUAL(0, rc); + rc = add_peer(&sometorrent, &addr); + TEST_ASSERT_EQUAL(0, rc); + rc = add_peer(&sometorrent, &addr); + TEST_ASSERT_EQUAL(0, rc); + rc = add_peer(&sometorrent, &addr); + TEST_ASSERT_EQUAL(0, rc); + // There's room for 8 peers per infohash + rc = add_peer(&sometorrent, &addr); + TEST_ASSERT_EQUAL(PEER_EFULL, rc); +} + +void test_peers_for_only_torrent() { + allocate_hashtable(); + + struct infohash sometorrent = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; + struct infohash othertorrent = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00522}}; + struct addr someaddr = (struct addr){.ip = IP(128,0,0,1), .port = 0}; + struct addr otheraddr = (struct addr){.ip = IP(128,0,0,1), .port = 1}; + + int rc = add_peer(&sometorrent, &someaddr); + TEST_ASSERT_EQUAL(0, rc); + rc = add_peer(&othertorrent, &otheraddr); + TEST_ASSERT_EQUAL(0, rc); + + struct addr* found; + size_t found_len; + + get_peers(&sometorrent, &found, &found_len); + TEST_ASSERT_EQUAL(1, found_len); + TEST_ASSERT_EQUAL_MEMORY(&someaddr, &found[0], sizeof(struct addr)); + + get_peers(&othertorrent, &found, &found_len); + TEST_ASSERT_EQUAL(1, found_len); + TEST_ASSERT_EQUAL_MEMORY(&otheraddr, &found[0], sizeof(struct addr)); +} + +void test_have_no_peers() { + allocate_hashtable(); + + struct infohash sometorrent = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; + + struct addr* found; + size_t found_len; + get_peers(&sometorrent, &found, &found_len); + TEST_ASSERT_NULL(found); + TEST_ASSERT_EQUAL(0, found_len); +} + +void test_grows() { + allocate_hashtable(); + + struct addr addr = (struct addr){.ip = IP(128,0,0,1), .port = 0}; + + for(int i = 0; i < 17; i++) { // Peer table size + 1 to make sure it HAS to grow + struct infohash sometorrent = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, i}}; + int rc = add_peer(&sometorrent, &addr); + TEST_ASSERT_EQUAL(0, rc); + } + + { + // We can find a peer after the resize again + struct infohash sometorrent = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x00000001}}; + struct addr* found; + size_t found_len; + get_peers(&sometorrent, &found, &found_len); + TEST_ASSERT_NOT_NULL(found); + TEST_ASSERT_EQUAL(1, found_len); + } +} diff --git a/test/proto.c b/test/proto.c index 60fa1cd..49a77b9 100644 --- a/test/proto.c +++ b/test/proto.c @@ -1,10 +1,13 @@ #include "unity.h" #include "proto.h" +#include "peers.h" #include "log.h" #include +#define IP(a, b, c, d) htonl(a << 24 | b << 16 | c << 8 | d) + void test_begin_pings_bootstrap_node() { struct message outbuff[10] = {0}; @@ -222,7 +225,7 @@ void test_remove_from_routing_after_3_retries() { { // The node responds // We return no new nodes to stop any new pings from going out - char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes0:""ee"; + char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes0:ee"; struct message* message_cursor = outbuff; proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, remote_len, now, &message_cursor, outbuff+2); } @@ -258,7 +261,7 @@ void test_remove_from_routing_after_3_retries() { struct message* message_cursor = outbuff; proto_run(&dht, NULL, 0, (struct sockaddr_in*)NULL, 0, now, &message_cursor, outbuff+2); - TEST_ASSERT_EQUAL_PTR_MESSAGE(message_cursor, outbuff, "The timeout should send a message"); + TEST_ASSERT_EQUAL_PTR_MESSAGE(message_cursor, outbuff, "The timeout should send not a message"); entry = routing_get(&other); TEST_ASSERT_NULL(entry); } @@ -286,7 +289,7 @@ void test_ping_node_when_uncertain() { { // The node responds // We return no new nodes to stop any new pings from going out - char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes0:""ee"; + char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes0:ee"; struct message* message_cursor = outbuff; proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, remote_len, now, &message_cursor, outbuff+2); } @@ -309,7 +312,7 @@ void test_ping_node_when_uncertain() { now += 5; { // The node responds - char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes0:""ee"; + char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes0:ee"; struct message* message_cursor = outbuff; proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, remote_len, now, &message_cursor, outbuff+2); @@ -343,7 +346,7 @@ void test_query_find_node() { { // The node responds // We return no new nodes to stop any new pings from going out - char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes0:""ee"; + char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes0:ee"; struct message* message_cursor = outbuff; proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, remote_len, now, &message_cursor, outbuff+2); } @@ -370,3 +373,102 @@ void test_query_find_node() { TEST_ASSERT_EQUAL_CHAR_ARRAY("ee", outbuff[0].payload+81, 2); } } + +void test_query_get_peers_have_one() { + struct message outbuff[10] = {0}; + time_t now = 0; + + struct sockaddr_storage remote; + socklen_t remote_len; + + allocate_hashtable(); + + struct dht dht; + dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; + routing_init(&dht.self); + + { + struct message* message_cursor = outbuff; + proto_begin(&dht, 0, &message_cursor, outbuff+2); + remote_len = outbuff[0].dest_len; + memcpy(&remote, &outbuff[0].dest, remote_len); + } + now += 10; + + { + // The node responds + // We return no new nodes to stop any new pings from going out + char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes0:ee"; + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, remote_len, now, &message_cursor, outbuff+2); + } + now += 1; + + // Some other node then asks for peers + { + struct sockaddr_in other; + other.sin_family = AF_INET; + inet_pton(AF_INET, "128.0.0.1", &other.sin_addr.s_addr); + other.sin_port = htons(9090); + + // Node asks for peers to get token + char buff[] = "d1:ad2:id20:abcdefghij01234567899:info_hash20:aaaaaaaaaaaaaaaaaaaae1:q9:get_peers1:t2:aa1:y1:qe"; + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), &other, sizeof(other), 0, &message_cursor, outbuff+2); + + // We should have sent a response + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); + + TEST_ASSERT_EQUAL(93, outbuff[0].payload_len); + char* cursor = outbuff[0].payload; + TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:t2:aa1:y1:r1:rd2:id20:BBBBBBBBBBBBBBBBBBBB5:token1:t5:nodes26:aaaaaaaaaaaaaaaaaaaa", cursor, 85); + cursor+=85; + TEST_ASSERT_EQUAL_MEMORY(&((struct sockaddr_in*)&remote)->sin_addr.s_addr, cursor, 4); + cursor+=4; + TEST_ASSERT_EQUAL_MEMORY(&((struct sockaddr_in*)&remote)->sin_port, cursor, 2); + cursor+=2; + TEST_ASSERT_EQUAL_CHAR_ARRAY("ee", cursor, 2); + cursor+=2; + } + now += 1; + + // Using the token from before that node then announces that it's a peer + { + struct sockaddr_in other; + other.sin_family = AF_INET; + inet_pton(AF_INET, "128.0.0.1", &other.sin_addr.s_addr); + other.sin_port = htons(9090); + + // Node announces that it's a peer for that torrent + char buff[] = "d1:ad2:id20:abcdefghij012345678912:implied_porti1e9:info_hash20:aaaaaaaaaaaaaaaaaaaa4:porti1337e5:token1:te1:q13:announce_peer1:t2:aa1:y1:qe"; + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&other, sizeof(other), now, &message_cursor, outbuff+2); + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); + + TEST_ASSERT_EQUAL(47, outbuff[0].payload_len); + char* cursor = outbuff[0].payload; + TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:t2:aa1:y1:r1:rd2:id20:BBBBBBBBBBBBBBBBBBBBee", cursor, 47); + cursor+=47; + } + now += 1; + + // A third node should now get that peer when asking + { + struct sockaddr_in other; + other.sin_family = AF_INET; + inet_pton(AF_INET, "255.255.255.255", &other.sin_addr.s_addr); + other.sin_port = htons(6881); + + char buff[] = "d1:ad2:id20:abcdefghij01234567899:info_hash20:aaaaaaaaaaaaaaaaaaaae1:q9:get_peers1:t2:aa1:y1:qe"; + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), &other, sizeof(other), 0, &message_cursor, outbuff+2); + + // We should have sent a response + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); + + TEST_ASSERT_EQUAL(75, outbuff[0].payload_len); + char* cursor = outbuff[0].payload; + TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:t2:aa1:y1:r1:rd2:id20:BBBBBBBBBBBBBBBBBBBB5:token1:t6:valuesl6:\x80\x00\x00\x01\x23\x82""eee", cursor, 75); + cursor+=75; + } +} diff --git a/test/query.c b/test/query.c index 5744fba..a3d3bb9 100644 --- a/test/query.c +++ b/test/query.c @@ -3,8 +3,18 @@ #include "query.h" #include +#include + +#define IP(a, b, c, d) htonl(a << 24 | b << 16 | c << 8 | d) void test_malformed_empty() { + + struct sockaddr_in src = { + .sin_family = AF_INET, + .sin_addr.s_addr = IP(255, 0, 0, 1), + .sin_port = 6881, + }; + struct nodeid self = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; char* packet = ""; size_t packet_len = 0; @@ -13,12 +23,19 @@ void test_malformed_empty() { char* response_cursor = response; char* response_end = response + sizeof(response); - int rc = handle_request(&self, "ping", packet, packet_len, &response_cursor, response_end-response_cursor); + int rc = handle_request(&self, "ping", (struct sockaddr*)&src, sizeof(src), packet, packet_len, &response_cursor, response_end-response_cursor); TEST_ASSERT_EQUAL(QUERY_EBADQ, rc); } void test_malformed_only_dict_start() { + + struct sockaddr_in src = { + .sin_family = AF_INET, + .sin_addr.s_addr = IP(255, 0, 0, 1), + .sin_port = 6881, + }; + struct nodeid self = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; char* packet = "d"; size_t packet_len = 1; @@ -27,12 +44,19 @@ void test_malformed_only_dict_start() { char* response_cursor = response; char* response_end = response + sizeof(response); - int rc = handle_request(&self, "ping", packet, packet_len, &response_cursor, response_end-response_cursor); + int rc = handle_request(&self, "ping", (struct sockaddr*)&src, sizeof(src), packet, packet_len, &response_cursor, response_end-response_cursor); TEST_ASSERT_EQUAL(QUERY_EBADQ, rc); } void test_malformed_empty_args_key() { + + struct sockaddr_in src = { + .sin_family = AF_INET, + .sin_addr.s_addr = IP(255, 0, 0, 1), + .sin_port = 6881, + }; + struct nodeid self = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; char* packet = "d1:ae"; size_t packet_len = strlen(packet); @@ -41,12 +65,19 @@ void test_malformed_empty_args_key() { char* response_cursor = response; char* response_end = response + sizeof(response); - int rc = handle_request(&self, "ping", packet, packet_len, &response_cursor, response_end-response_cursor); + int rc = handle_request(&self, "ping", (struct sockaddr*)&src, sizeof(src), packet, packet_len, &response_cursor, response_end-response_cursor); TEST_ASSERT_EQUAL(QUERY_EBADQ, rc); } void test_malformed_wrong_args_type() { + + struct sockaddr_in src = { + .sin_family = AF_INET, + .sin_addr.s_addr = IP(255, 0, 0, 1), + .sin_port = 6881, + }; + struct nodeid self = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; char* packet = "d1:a1:re"; size_t packet_len = strlen(packet); @@ -55,12 +86,19 @@ void test_malformed_wrong_args_type() { char* response_cursor = response; char* response_end = response + sizeof(response); - int rc = handle_request(&self, "ping", packet, packet_len, &response_cursor, response_end-response_cursor); + int rc = handle_request(&self, "ping", (struct sockaddr*)&src, sizeof(src), packet, packet_len, &response_cursor, response_end-response_cursor); TEST_ASSERT_EQUAL(QUERY_EBADQ, rc); } void test_malformed_empty_args() { + + struct sockaddr_in src = { + .sin_family = AF_INET, + .sin_addr.s_addr = IP(255, 0, 0, 1), + .sin_port = 6881, + }; + struct nodeid self = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; char* packet = "d1:adee"; size_t packet_len = strlen(packet); @@ -69,12 +107,19 @@ void test_malformed_empty_args() { char* response_cursor = response; char* response_end = response + sizeof(response); - int rc = handle_request(&self, "ping", packet, packet_len, &response_cursor, response_end-response_cursor); + int rc = handle_request(&self, "ping", (struct sockaddr*)&src, sizeof(src), packet, packet_len, &response_cursor, response_end-response_cursor); TEST_ASSERT_EQUAL(QUERY_EBADQ, rc); } void test_malformed_wrong_id_arg_type() { + + struct sockaddr_in src = { + .sin_family = AF_INET, + .sin_addr.s_addr = IP(255, 0, 0, 1), + .sin_port = 6881, + }; + struct nodeid self = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; char* packet = "d1:ad2:idi1eee"; size_t packet_len = strlen(packet); @@ -83,12 +128,19 @@ void test_malformed_wrong_id_arg_type() { char* response_cursor = response; char* response_end = response + sizeof(response); - int rc = handle_request(&self, "ping", packet, packet_len, &response_cursor, response_end-response_cursor); + int rc = handle_request(&self, "ping", (struct sockaddr*)&src, sizeof(src), packet, packet_len, &response_cursor, response_end-response_cursor); TEST_ASSERT_EQUAL(QUERY_EBADQ, rc); } void test_malformed_wrong_id_length() { + + struct sockaddr_in src = { + .sin_family = AF_INET, + .sin_addr.s_addr = IP(255, 0, 0, 1), + .sin_port = 6881, + }; + struct nodeid self = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; char* packet = "d1:ad2:id19:aaaaaaaaaaaaaaaaaaaee"; size_t packet_len = strlen(packet); @@ -97,12 +149,19 @@ void test_malformed_wrong_id_length() { char* response_cursor = response; char* response_end = response + sizeof(response); - int rc = handle_request(&self, "ping", packet, packet_len, &response_cursor, response_end-response_cursor); + int rc = handle_request(&self, "ping", (struct sockaddr*)&src, sizeof(src), packet, packet_len, &response_cursor, response_end-response_cursor); TEST_ASSERT_EQUAL(QUERY_EBADQ, rc); } void test_ping() { + + struct sockaddr_in src = { + .sin_family = AF_INET, + .sin_addr.s_addr = IP(255, 0, 0, 1), + .sin_port = 6881, + }; + struct nodeid self = {.inner_b={"aaaaaaaaaaaaaaaaaaab"}}; char* packet = "d1:ad2:id20:aaaaaaaaaaaaaaaaaaaaee"; size_t packet_len = strlen(packet); @@ -111,7 +170,7 @@ void test_ping() { char* response_cursor = response; char* response_end = response + sizeof(response); - int rc = handle_request(&self, "ping", packet, packet_len, &response_cursor, response_end-response_cursor); + int rc = handle_request(&self, "ping", (struct sockaddr*)&src, sizeof(src), packet, packet_len, &response_cursor, response_end-response_cursor); TEST_ASSERT_EQUAL(0, rc); TEST_ASSERT_EQUAL(29, response_cursor - response); @@ -119,6 +178,13 @@ void test_ping() { } void test_bad_method() { + + struct sockaddr_in src = { + .sin_family = AF_INET, + .sin_addr.s_addr = IP(255, 0, 0, 1), + .sin_port = 6881, + }; + struct nodeid self = {.inner_b={"aaaaaaaaaaaaaaaaaaab"}}; char* packet = "de"; size_t packet_len = strlen(packet); @@ -127,7 +193,7 @@ void test_bad_method() { char* response_cursor = response; char* response_end = response + sizeof(response); - int rc = handle_request(&self, "someWrongMethod", packet, packet_len, &response_cursor, response_end-response_cursor); + int rc = handle_request(&self, "someWrongMethod", (struct sockaddr*)&src, sizeof(src), packet, packet_len, &response_cursor, response_end-response_cursor); TEST_ASSERT_EQUAL(QUERY_EUNK, rc); } diff --git a/test/routing.c b/test/routing.c index ad6f2c3..4613848 100644 --- a/test/routing.c +++ b/test/routing.c @@ -1,7 +1,9 @@ #include "unity.h" #include "routing.h" -#define IP(a, b, c, d) (a << 24 | b << 16 | c << 8 | d) +#include + +#define IP(a, b, c, d) htonl(a << 24 | b << 16 | c << 8 | d) struct nodeid self; @@ -229,7 +231,7 @@ void test_close_to_self_load_factor() { // Make a nodeid that is one bit different struct nodeid new = self; - new.inner[4] += 1; + new.inner[4] += ntohl(1); struct entry* entry; TEST_ASSERT_TRUE_MESSAGE(routing_offer(&new, &entry), "Did not accept new entry"); @@ -260,7 +262,7 @@ void test_far_from_self_load_factor() { struct nodeid new = self; // Flip top bit to make it very dissimilar - new.inner[0] ^= 0x80000000; + new.inner[0] ^= ntohl(0x80000000); struct entry* entry; TEST_ASSERT_TRUE_MESSAGE(routing_offer(&new, &entry), "Did not accept new entry"); -- cgit v1.2.3