diff options
| -rw-r--r-- | src/main.c | 498 | ||||
| -rw-r--r-- | src/proto.c | 519 | ||||
| -rw-r--r-- | src/proto.h | 44 | ||||
| -rw-r--r-- | src/routing.c | 64 | ||||
| -rw-r--r-- | src/routing.h | 3 | ||||
| -rw-r--r-- | test/proto.c | 111 | ||||
| -rw-r--r-- | test/routing.c | 33 |
7 files changed, 787 insertions, 485 deletions
@@ -1,491 +1,51 @@ -#include "routing.h" -#include "benc.h" -#include "query.h" +#include "proto.h" #include "log.h" -#include <errno.h> -#include <sys/stat.h> -#include <fcntl.h> -#include <ctype.h> -#include <assert.h> -#include <stdlib.h> -#include <stdint.h> -#include <limits.h> -#include <unistd.h> -#include <stdbool.h> -#include <string.h> -#include <sys/socket.h> -#include <netdb.h> -#include <arpa/inet.h> -#include <netdb.h> -#include <sys/types.h> -#include <stdio.h> - -#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; \ - }) - - -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); -} - -struct discovery { - uint16_t port; - struct nodeid expected_id; -}; - -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; -}; - -#define UNDEF_ADDR (struct in_addr){0xC0000200} -#define MAX_DISC 32 -struct in_addr addrs[MAX_DISC]; -struct discovery pending_discover[MAX_DISC]; - -typedef void (*cont)(struct nodeid* self, char* packet, size_t packet_len, int socket, struct sockaddr_in* remote, socklen_t remote_len); - -#define MAX_INFLIGHT 32 -bool reqalloc[MAX_INFLIGHT] = { false }; -struct { - struct sockaddr addr; - cont fun; -} requestdata[MAX_INFLIGHT]; - -bool alloc_req(uint16_t* reqId) { - for(size_t i = 0; i < MAX_INFLIGHT; i++) { - if(!reqalloc[i]) { - reqalloc[i] = true; - *reqId = i; - return true; - } - } - return false; -} - -bool find_req(uint32_t transId, uint16_t* reqId) { - *reqId = transId; - return reqalloc[transId]; -} - -void getclient_response(struct nodeid* self, char* packet, size_t packet_len, int socket, struct sockaddr_in* remote, socklen_t remote_len); - -uint8_t rand_byte() { - int limit = RAND_MAX - (RAND_MAX % UINT8_MAX); - int val; - while((val = rand()) > limit); - - return val; -} - -int send_ping(struct nodeid* self, const int sfd, const struct sockaddr* dest_addr, socklen_t 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(); - } - - uint16_t reqId; - if(!alloc_req(&reqId)) { - return ENOBUFS; - } - char buff[128]; - size_t i = 0; - - dbg("Allocating request %d", reqId); - requestdata[reqId].fun = &getclient_response; - requestdata[reqId].addr = *dest_addr; - - int rc = snprintf(buff+i, 128-i, "d1:ad2:id20:"); - if(rc < 0) - return EPERM; - i += rc; - memcpy(buff+i, self, sizeof(struct nodeid)); - i += sizeof(struct nodeid); - rc = snprintf(buff+i, 128-i, "6:target20:"); - if(rc < 0) - return EPERM; - i += rc; - memcpy(buff+i, &target, sizeof(struct nodeid)); - i += sizeof(struct nodeid); - rc = snprintf(buff+i, 128-i, "e1:q9:find_node1:t%d:%d1:y1:qe", (reqId/10)+1, reqId); - if(rc < 0) - return EPERM; - i += rc; - - //now reply the client with the same data - rc = sendto(sfd, buff, i, 0, dest_addr, dest_len); - if (rc == -1) { - return EPERM; // Operation not permitted is used as the default "generic" error - } - - return 0; -} - -void getclient_response(struct nodeid* self, char* packet, size_t packet_len, int socket, struct sockaddr_in* remote, socklen_t remote_len) { - struct benc_node stream[256]; - struct bcursor bcursor; - bcur_open(&bcursor, packet, packet+packet_len, stream, 256); - - if(bcursor.end - bcursor.readhead <= 0) { - err("Reponse too short"); - exit(EXIT_FAILURE); - } - - 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) { - err("Response is not a dict"); - exit(EXIT_FAILURE); - } - 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) { - err("Wrong value type for response"); - exit(EXIT_FAILURE); - } - - // Skip the dict element - bcur_next(&bcursor, 1); - - 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) { - err("Wrong value type for response"); - exit(EXIT_FAILURE); - } - - if((bcursor.readhead->size % 26) != 0) { - err("get_nodes call returned an incorrect nodes array"); - exit(EXIT_FAILURE); - } - - nodes_len = MIN(bcursor.readhead->size/26, 8); - dbg("We have %d (%d/26) nodes", nodes_len, bcursor.readhead->size); - 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); - } - - // Skip the value - bcur_next(&bcursor, 1); - break; - case 1: - // Skip the key - bcur_next(&bcursor, 1); - - if(bcursor.readhead->type != BNT_STRING) { - err("Wrong value type for response"); - exit(EXIT_FAILURE); - } - - if(bcursor.readhead->size != 20) { - err("remote node id was not 20 bytes long"); - exit(EXIT_FAILURE); - } - - memcpy(&id, bcursor.readhead->loc, 20); - - // Skip the value - bcur_next(&bcursor, 1); - break; - } +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"); } } - - // Print the candidates - 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])); - - struct sockaddr_in dest = { - .sin_family = AF_INET, - .sin_addr = ips[i], - .sin_port = ports[i], - }; - - if(routing_interested(&nodes[i])) { - int rc = send_ping(self, socket, (struct sockaddr*)&dest, sizeof(struct sockaddr_in)); - if(rc != 0) { - err("send_ping failed %d", rc); - } - } else { - dbg("Not interested in node"); - } - } - - struct entry* entry; - routing_offer(&id, &entry); } -enum commandType { - CT_QUERY, - CT_RESPONSE, - CT_ERROR, -}; - int main(int argc, char** argv) { - for(int i = 0; i < MAX_DISC; i++) { - // 192.0.2.0 - addrs[i] = UNDEF_ADDR; - } - - struct nodeid self = {.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; - routing_init(&self); - dbgl_id(&self); - - int sfd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); - if(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(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 message outbuff[10] = {0}; - 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) { - char buff[128]; - inet_ntop(cur->ai_family, &((struct sockaddr_in*)cur->ai_addr)->sin_addr, buff, 128); - dbg("IP %s", buff); - - /* size_t i = 0; */ - /* int rc = snprintf(buff+i, 128-i, "d1:ad2:id20:"); */ - /* i += rc; */ - /* memcpy(buff+i, &self, sizeof(struct nodeid)); */ - /* i += sizeof(struct nodeid); */ - /* rc = snprintf(buff+i, 128-i, "e1:q4:ping1:t2:ab1:y1:qe"); */ - /* i += rc; */ + struct dht dht; + dht.self = (struct nodeid){.inner={0x0034048f, 0x08000020, 0x00888880, 0x02008460, 0x0ab00521}}; + struct message* message_cursor = outbuff; + proto_begin(&dht, &message_cursor, outbuff+10); + flush_messages(dht.sfd, outbuff, message_cursor); - send_ping(&self, sfd, cur->ai_addr, cur->ai_addrlen); - } - - freeaddrinfo(res); - //keep listening for data - while(1) { + int rc = 0; + while(rc == 0) { char buff[2049]; printf("Waiting for data..."); fflush(stdout); - - //try to receive some data, this is a blocking call - ssize_t recv_len; - struct sockaddr_in remote; + + // Try to receive some data, this is a blocking call + struct sockaddr_storage remote; socklen_t remote_len = sizeof(remote); - if ((recv_len = recvfrom(sfd, buff, 2048, 0, (struct sockaddr *)&remote, &remote_len)) == -1) { - fatal("recvfrom failed"); - } - if(recv_len >= 2048) { - fatal("Recv buffer too small"); + ssize_t recv_len = recvfrom(dht.sfd, buff, 2048, 0, (struct sockaddr *)&remote, &remote_len); + if(recv_len == -1) { + fatal("RECV failed"); + } else if(recv_len >= 2048) { + fatal("Receive buffer too small"); } // Null terminate the packet buff[recv_len] = '\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"); - } - 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) - continue; - - if(type == CT_RESPONSE) { - uint32_t transaction_number; - - if(!transaction_set) - fatal("No transaction in response"); - - // 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) { - fatal("DISCARD: Transaction id is not a number %.*s", transaction_len, transaction); - } - - uint16_t reqId; - if(!find_req(transaction_number, &reqId)) { - dbg("DISCARD: unknown transaction id %d", transaction); - continue; - } - dbg("Transaction id matches request %d", reqId); - - if(sockaddr_cmp(&requestdata[reqId].addr, (struct sockaddr*)&remote) != 0) { - fatal("Unexpected IP for valid transaction"); - } - - requestdata[reqId].fun(&self, buff, recv_len, sfd, &remote, remote_len); - - reqalloc[reqId] = false; - } else if(type == CT_QUERY) { // Must be a query - if(!query_set) - fatal("No query function in query request"); - if(!transaction_set) - fatal("No transaction in request"); - - assert(strlen(query) == query_len); - - char response[1024]; - char* end = response+sizeof(response)-1; - char* cursor = response; - - int rc = snprintf(cursor, end-cursor , "d1:t%ld:", transaction_len); - if(rc < 0) - return EPERM; - cursor += rc; - memcpy(cursor, transaction, transaction_len); - cursor += transaction_len; - rc = snprintf(cursor, end-cursor, "1:y1:r1:r"); - if(rc < 0) - return EPERM; - cursor += rc; - - rc = handle_request(&self, query, buff, recv_len, &cursor, end-cursor-1); - if(rc != 0) fatal("Error handling request"); - - rc = snprintf(cursor, end-cursor, "e"); - if(rc < 0) - return EPERM; - cursor += rc; - - //now reply the client with the same data - rc = sendto(sfd, buff, cursor-response, 0, (const struct sockaddr*)&remote, remote_len); - if (rc == -1) { - return EPERM; // Operation not permitted is used as the default "generic" error - } - - } + struct message* message_cursor = outbuff; + rc = proto_run(&dht, buff, recv_len, (struct sockaddr_in*)&remote, remote_len, &message_cursor, outbuff+10); + flush_messages(dht.sfd, outbuff, message_cursor); } - close(sfd); + proto_end(&dht); - return 0; + return rc; } diff --git a/src/proto.c b/src/proto.c new file mode 100644 index 0000000..7a7eb73 --- /dev/null +++ b/src/proto.c @@ -0,0 +1,519 @@ +#include "proto.h" + +#include "benc.h" +#include "query.h" +#include "log.h" + +#include <errno.h> +#include <sys/stat.h> +#include <fcntl.h> +#include <ctype.h> +#include <assert.h> +#include <stdlib.h> +#include <math.h> +#include <stdint.h> +#include <limits.h> +#include <unistd.h> +#include <stdbool.h> +#include <string.h> +#include <netdb.h> +#include <netdb.h> +#include <sys/types.h> +#include <stdio.h> + +#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 UNCERTAIN_TIME 10 + + +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; +}; + +void getclient_response(struct dht* dht, struct nodeid* self, char* packet, size_t packet_len, int socket, struct sockaddr* remote, socklen_t remote_len, struct msgbuff* msgbuff); + +uint8_t rand_byte() { + int limit = RAND_MAX - (RAND_MAX % UINT8_MAX); + int val; + while((val = rand()) > limit); + + return val; +} + +int send_ping(struct dht* dht, struct nodeid* self, const int sfd, const struct sockaddr* dest_addr, socklen_t dest_len, struct msgbuff* msgbuff) { + // 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(); + } + + uint16_t reqId; + if(!alloc_req(dht, &reqId)) { + return ENOBUFS; + } + + assert(*msgbuff->messages < msgbuff->messages_end); + struct message* message = *msgbuff->messages; + + memcpy(&message->dest, dest_addr, dest_len); + message->dest_len = dest_len; + char* buff = message->payload; + size_t i = 0; + + dbg("Allocating request %d", reqId); + dht->requestdata[reqId].fun = &getclient_response; + memcpy(&dht->requestdata[reqId].addr, dest_addr, dest_len); + + int rc = snprintf(buff+i, 128-i, "d1:ad2:id20:"); + if(rc < 0) + fatal("Failed to write packet"); + i += rc; + memcpy(buff+i, self, sizeof(struct nodeid)); + i += sizeof(struct nodeid); + rc = snprintf(buff+i, 128-i, "6:target20:"); + if(rc < 0) + fatal("Failed to write packet"); + i += rc; + memcpy(buff+i, &target, sizeof(struct nodeid)); + i += sizeof(struct nodeid); + rc = snprintf(buff+i, 128-i, "e1:q9:find_node1:t%d:%d1:y1:qe", (int)(log10(reqId+1)+1), reqId); + if(rc < 0) + fatal("Failed to write packet"); + i += rc; + + message->payload_len = i; + (*msgbuff->messages)++; + + return 0; +} + +void getclient_response(struct dht* dht, struct nodeid* self, char* packet, size_t packet_len, int socket, struct sockaddr* remote, socklen_t remote_len, struct msgbuff* msgbuff) { + struct benc_node stream[256]; + struct bcursor bcursor; + bcur_open(&bcursor, packet, packet+packet_len, stream, 256); + + if(bcursor.end - bcursor.readhead <= 0) { + err("Reponse too short"); + exit(EXIT_FAILURE); + } + + 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) { + err("Response is not a dict"); + exit(EXIT_FAILURE); + } + 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) { + err("Wrong value type for response"); + exit(EXIT_FAILURE); + } + + // 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); + 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); + 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) { + err("Wrong value type for response"); + exit(EXIT_FAILURE); + } + + if(bcursor.readhead->size != 20) { + err("remote node id was not 20 bytes long"); + exit(EXIT_FAILURE); + } + + memcpy(&id, bcursor.readhead->loc, 20); + + parts++; + + // Skip the value + bcur_next(&bcursor, 1); + break; + } + } + + if(parts < 2) { + fatal("Response didn't contain the expected values"); + } + } + + // Print the candidates + 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])); + + struct sockaddr_in dest = { + .sin_family = AF_INET, + .sin_addr = ips[i], + .sin_port = ports[i], + }; + + if(routing_interested(&nodes[i])) { + int rc = send_ping(dht, self, socket, (struct sockaddr*)&dest, sizeof(struct sockaddr_in), msgbuff); + if(rc != 0) { + err("send_ping failed %d", rc); + } + } else { + dbg("Not interested in node"); + } + } + + struct entry* entry; + 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->last = time(NULL) + UNCERTAIN_TIME; +} + +enum commandType { + CT_QUERY, + CT_RESPONSE, + CT_ERROR, +}; + +void proto_begin(struct dht* dht, struct message** output, const struct message* const output_end) { + routing_flush(); + struct msgbuff msgbuff = { + output, + output_end, + }; + + for(int i = 0; i < MAX_DISC; i++) { + dht->addrs[i] = UNDEF_ADDR; + } + + for(int i = 0; i < MAX_INFLIGHT; i++) { + dht->reqalloc[i] = false; + } + + routing_init(&dht->self); + 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) { + char buff[128]; + inet_ntop(cur->ai_family, &((struct sockaddr_in*)cur->ai_addr)->sin_addr, buff, 128); + dbg("IP %s", buff); + + /* size_t i = 0; */ + /* int rc = snprintf(buff+i, 128-i, "d1:ad2:id20:"); */ + /* i += rc; */ + /* memcpy(buff+i, &self, sizeof(struct nodeid)); */ + /* i += sizeof(struct nodeid); */ + /* rc = snprintf(buff+i, 128-i, "e1:q4:ping1:t2:ab1:y1:qe"); */ + /* i += rc; */ + + send_ping(dht, &dht->self, dht->sfd, cur->ai_addr, cur->ai_addrlen, &msgbuff); + } + + freeaddrinfo(res); +} + +void proto_end(struct dht* dht) { + close(dht->sfd); +} + +int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* remote, socklen_t remote_len, struct message** output, const struct message* const output_end) { + struct msgbuff msgbuff = { + output, + output_end, + }; + + if(recv_len == -1 && errno == ETIMEDOUT) { + time_t now = time(NULL); + struct entry* oldest = NULL; + routing_oldest(&oldest); + while(oldest != NULL) { + if(difftime(oldest->last, now) > 0.0) + break; + dbg("============ Ping uncertain node"); + + 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, &dht->self, dht->sfd, (const struct sockaddr*)&dest, sizeof(dest), &msgbuff); + if(rc != 0) { + fatal("NOPE %d", rc); + } + + oldest->last = now + 30; + routing_oldest(&oldest); + } + + 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"); + } + 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; + } + + if(type == CT_RESPONSE) { + uint32_t transaction_number; + + if(!transaction_set) + fatal("No transaction in response"); + + // 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) { + 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("Transaction id matches request %d", reqId); + + if(sockaddr_cmp((struct sockaddr*)&dht->requestdata[reqId].addr, (struct sockaddr*)remote) != 0) { + err("DISCARD: Unexpected IP for valid transaction"); + return 0; + } + + dht->requestdata[reqId].fun(dht, &dht->self, buff, recv_len, dht->sfd, (struct sockaddr*)remote, remote_len, &msgbuff); + dht->reqalloc[reqId] = false; + } else if(type == CT_QUERY) { // Must be a query + if(!query_set) + fatal("No query function in query request"); + if(!transaction_set) + fatal("No transaction in request"); + + assert(strlen(query) == query_len); + + assert(*msgbuff.messages < msgbuff.messages_end); + struct message* message = *msgbuff.messages; + + char* end = message->payload+128; + char* cursor = message->payload; + + int rc = snprintf(cursor, end-cursor , "d1:t%ld:", transaction_len); + if(rc < 0) + return EPERM; + cursor += rc; + memcpy(cursor, transaction, transaction_len); + cursor += transaction_len; + rc = snprintf(cursor, end-cursor, "1:y1:r1:r"); + if(rc < 0) + return EPERM; + cursor += rc; + + rc = handle_request(&dht->self, query, buff, recv_len, &cursor, end-cursor-1); + if(rc != 0) fatal("Error handling request"); + + rc = snprintf(cursor, end-cursor, "e"); + if(rc < 0) + return EPERM; + cursor += rc; + assert(cursor < end); + + message->payload_len = cursor - message->payload; + + memcpy(&message->dest, remote, remote_len); + message->dest_len = remote_len; + (*msgbuff.messages)++; + } + + return 0; +} diff --git a/src/proto.h b/src/proto.h new file mode 100644 index 0000000..85dd310 --- /dev/null +++ b/src/proto.h @@ -0,0 +1,44 @@ +#pragma once + +#include "routing.h" +#include <sys/socket.h> +#include <arpa/inet.h> + +// 192.0.2.0 +#define UNDEF_ADDR (struct in_addr){0xC0000200} +#define MAX_DISC 32 +#define MAX_INFLIGHT 32 + +struct dht; +struct msgbuff; +typedef void (*cont)(struct dht* dht, struct nodeid* self, char* packet, size_t packet_len, int socket, struct sockaddr* remote, socklen_t remote_len, struct msgbuff* msgbuff); + +struct discovery { + uint16_t port; + struct nodeid expected_id; +}; + +struct dht { + struct nodeid self; + int sfd; + + struct in_addr addrs[MAX_DISC]; + struct discovery pending_discover[MAX_DISC]; + + bool reqalloc[MAX_INFLIGHT]; + struct { + struct sockaddr_storage addr; + cont fun; + } requestdata[MAX_INFLIGHT]; +}; + +struct message { + char payload[128]; + size_t payload_len; + struct sockaddr_storage dest; + socklen_t dest_len; +}; + +void proto_begin(struct dht* dht, 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, struct message** output, const struct message* const output_end); +void proto_end(struct dht* dht); diff --git a/src/routing.c b/src/routing.c index b8ab037..64d5353 100644 --- a/src/routing.c +++ b/src/routing.c @@ -1,5 +1,7 @@ #include "routing.h" +#include "log.h" + #include <assert.h> #include <limits.h> #include <string.h> @@ -41,6 +43,7 @@ struct entry table[ROUTINGSIZE]; void routing_init(struct nodeid* myid) { myID = *myid; + routing_flush(); } void routing_flush() { @@ -48,7 +51,7 @@ void routing_flush() { } // Calculate the common bit prefix between two node ids. -uint8_t prefix(struct nodeid* a, struct nodeid* b) { +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]; @@ -64,13 +67,13 @@ uint8_t prefix(struct nodeid* a, struct nodeid* b) { return c; } -int8_t scan(uint16_t baseIndex, struct nodeid* id) { +static int8_t scan(uint16_t baseIndex, struct nodeid* id) { assert(baseIndex < ROUTINGSIZE - BUCKETSIZE); - int8_t index = -1; + int8_t index = -2; for(size_t i = baseIndex; i < baseIndex + BUCKETSIZE; i++) { if(!table[i].set) { - index = index == -1 ? i - baseIndex : index; + index = index == -2 ? i - baseIndex : index; continue; } @@ -82,6 +85,28 @@ int8_t scan(uint16_t baseIndex, struct nodeid* id) { return index; } +static uint16_t base_bucket(struct nodeid* id) { + uint16_t bucketIndex = prefix(&myID, id); + assert(bucketIndex != IDBITS); + + // If they are sufficiently similar they end up in the final bucket. Clamp the index to ensure. + bucketIndex = bucketIndex > (IDBITS - BUCKETBITS) ? (IDBITS - BUCKETBITS) : bucketIndex; + assert(bucketIndex <= IDBITS - BUCKETBITS); + + return bucketIndex * BUCKETSIZE; +} + +struct entry* routing_get(struct nodeid* id) { + uint16_t baseIndex = base_bucket(id); + for(size_t i = baseIndex; i < baseIndex + BUCKETSIZE; i++) { + if(memcmp(&table[i].id, id, sizeof(struct nodeid)) == 0) { + return &table[i]; + } + } + + return NULL; +} + bool routing_interested(struct nodeid* id) { uint16_t bucketIndex = prefix(&myID, id); // The nodeid is the same as our own @@ -89,14 +114,10 @@ bool routing_interested(struct nodeid* id) { return false; } - // If they are sufficiently similar they end up in the final bucket. Clamp the index to ensure. - bucketIndex = bucketIndex > (IDBITS - BUCKETBITS) ? (IDBITS - BUCKETBITS) : bucketIndex; - assert(bucketIndex <= IDBITS - BUCKETBITS); - - uint16_t baseIndex = bucketIndex * BUCKETSIZE; + uint16_t baseIndex = base_bucket(id); int8_t inBucketIndex = scan(baseIndex, id); - if(inBucketIndex == -1) { + if(inBucketIndex < 0) { // The bucket either already contains the node, or it has no more space return false; } @@ -112,11 +133,7 @@ bool routing_offer(struct nodeid* id, struct entry **dest) { return false; } - // If they are sufficiently similar they end up in the final bucket. Clamp the index to ensure. - bucketIndex = bucketIndex > (IDBITS - BUCKETBITS) ? (IDBITS - BUCKETBITS) : bucketIndex; - assert(bucketIndex <= IDBITS - BUCKETBITS); - - uint16_t baseIndex = bucketIndex * BUCKETSIZE; + uint16_t baseIndex = base_bucket(id); int8_t inBucketIndex = scan(baseIndex, id); if(inBucketIndex == -1) { @@ -181,3 +198,20 @@ size_t routing_closest(struct nodeid* needle, size_t n, struct entry** res) { return read; } +void routing_oldest(struct entry** dest) { + *dest = NULL; + + for(struct entry* entry = table; entry < table+ROUTINGSIZE; entry++){ + if(!entry->set) + continue; + + if(*dest == NULL) { + *dest = entry; + continue; + } + + if(difftime((*dest)->last, entry->last) > 0.0) { + *dest = entry; + } + } +} diff --git a/src/routing.h b/src/routing.h index 5c98f13..7f1d19e 100644 --- a/src/routing.h +++ b/src/routing.h @@ -29,4 +29,7 @@ void routing_init(struct nodeid* myid); void routing_flush(); bool routing_interested(struct nodeid* id); bool routing_offer(struct nodeid* id, struct entry **dest); +void routing_oldest(struct entry** dest); size_t routing_closest(struct nodeid* needle, size_t n, struct entry** res); + +struct entry* routing_get(struct nodeid* id); diff --git a/test/proto.c b/test/proto.c new file mode 100644 index 0000000..3becc98 --- /dev/null +++ b/test/proto.c @@ -0,0 +1,111 @@ +#include "unity.h" +#include "proto.h" + +#include "log.h" + +#include <string.h> + +void test_begin_pings_bootstrap_node() { + struct message outbuff[10] = {0}; + + struct dht dht; + dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; + + struct message* message_cursor = outbuff; + proto_begin(&dht, &message_cursor, outbuff+10); + + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); + TEST_ASSERT_EQUAL(91, outbuff[0].payload_len); + TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:ad2:id20:BBBBBBBBBBBBBBBBBBBB6:target20:", outbuff[0].payload, 43); + TEST_ASSERT_EQUAL_CHAR_ARRAY("e1:q9:find_node1:t1:01:y1:qe", outbuff[0].payload+63, 28); +} + +void test_response_from_initial_probe() { + struct message outbuff[2] = {0}; + + struct sockaddr_storage remote; + socklen_t remote_len; + + struct dht dht; + dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; + { + struct message* message_cursor = outbuff; + proto_begin(&dht, &message_cursor, outbuff+2); + remote_len = outbuff[0].dest_len; + memcpy(&remote, &outbuff[0].dest, remote_len); + } + + char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes26:bbbbbbbbbbbbbbbbbbbb\xFF\xFF\xFF\xFF\x00\x00""ee"; + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, remote_len, &message_cursor, outbuff+2); + + TEST_ASSERT_EQUAL_PTR_MESSAGE(message_cursor, outbuff+1, "Sends one packet"); + TEST_ASSERT_EQUAL(91, outbuff[0].payload_len); + TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:ad2:id20:BBBBBBBBBBBBBBBBBBBB6:target20:", outbuff[0].payload, 43); + TEST_ASSERT_EQUAL_CHAR_ARRAY("e1:q9:find_node1:t1:11:y1:qe", outbuff[0].payload+63, 28); + + // The queried node gets added to the routing table + struct nodeid other = (struct nodeid){.inner={0x61616161, 0x61616161, 0x61616161, 0x61616161, 0x61616161}}; + struct entry* entry = routing_get(&other); + TEST_ASSERT_NOT_NULL(entry); + TEST_ASSERT_EQUAL(((struct sockaddr_in*)&remote)->sin_addr.s_addr, entry->addr.ip); + TEST_ASSERT_EQUAL(((struct sockaddr_in*)&remote)->sin_port, entry->addr.port); +} + +void test_reponse_from_wrong_ip() { + struct message outbuff[2] = {0}; + + struct dht dht; + dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; + { + struct message* message_cursor = outbuff; + proto_begin(&dht, &message_cursor, outbuff+2); + } + + // This is fragile, since the ip of the bootstrap node could change, and we + // look it up from DNS. Although it pretty unlikely that it would change to + // this ip + struct sockaddr_in remote; + remote.sin_family = AF_INET; + inet_pton(AF_INET, "255.255.255.255", &remote.sin_addr.s_addr); + remote.sin_port = htons(6881); + + char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes26:bbbbbbbbbbbbbbbbbbbb\xFF\xFF\xFF\xFF\x00\x00""ee"; + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), &remote, sizeof(remote), &message_cursor, outbuff+2); + + // We shouldn't send any packets, since the response is rejected + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff); + + // Since the ip was wrong we should not have accepted the node into the + // rounting table + struct nodeid other = (struct nodeid){.inner={0x61616161, 0x61616161, 0x61616161, 0x61616161, 0x61616161}}; + struct entry* entry = routing_get(&other); + TEST_ASSERT_NULL(entry); +} + +void test_ping() { + struct message outbuff[2] = {0}; + + struct dht dht; + dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; + { + struct message* message_cursor = outbuff; + proto_begin(&dht, &message_cursor, outbuff+2); + } + + struct sockaddr_in remote; + remote.sin_family = AF_INET; + inet_pton(AF_INET, "255.255.255.255", &remote.sin_addr.s_addr); + remote.sin_port = htons(6881); + + char buff[] = "d1:ad2:id20:abcdefghij0123456789e1:q4:ping1:t2:aa1:y1:qe"; + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), &remote, sizeof(remote), &message_cursor, outbuff+2); + + // We should have sent a response + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); + + TEST_ASSERT_EQUAL(48, outbuff[0].payload_len); + TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:t2:aa1:y1:r1:rd2:id20:BBBBBBBBBBBBBBBBBBBBee", outbuff[0].payload, 47); +} diff --git a/test/routing.c b/test/routing.c index 60a7f03..7905bc0 100644 --- a/test/routing.c +++ b/test/routing.c @@ -6,7 +6,7 @@ struct nodeid self; void setUp() { - self = (struct nodeid){{ 0x00000000, 0x00000000, 0x00000000, 0x00000000, 0x00000000 }}; + self = (struct nodeid){.inner={ 0x00000000, 0x00000000, 0x00000000, 0x00000000, 0x00000000 }}; routing_init(&self); } @@ -142,3 +142,34 @@ void test_not_interested_when_bucket_is_full() { TEST_ASSERT_FALSE_MESSAGE(routing_interested(&new), "Still interested when bucket was full"); } + +void test_lowest_ts_is_oldest() { + routing_flush(); + + // The address we are going to store + struct addr addr = (struct addr){.ip = IP(128,0,0,1), .port = 0}; + + struct nodeid new = self; + // Flip the top bit of the id to go into the low resolution bucket + new.inner_b[0] ^= 0x80; + + // Fill up the bucket with entries + for(uint8_t i = 0; i < 2; i++) { + struct entry* entry; + TEST_ASSERT_TRUE_MESSAGE(routing_offer(&new, &entry), "Did not accept new entry"); + + // Set the entries + entry->id = new; + entry->addr = addr; + // Invert the timestamps to make the last one have lowest timestamp + entry->last = 2-i; + + new.inner_b[19] += 1; + } + + struct entry* dest = NULL; + routing_oldest(&dest); + + TEST_ASSERT_NOT_NULL(dest); + TEST_ASSERT_EQUAL(1, dest->id.inner_b[19]); +} |
