/* This software was written by Dirk Engling It is considered beerware. Prost. Skol. Cheers or whatever. */ /* This relays sync traffic between trackers, and between multicast domains that have no route to each other, so that no tracker has to reach out to the outside world by itself. Defunc. */ /* System */ #include #include #include #include #include #include #include #include #include /* Libowfat */ #include "io.h" /* Opentracker */ #include "ot_batch.h" #include "ot_fdstate.h" #include "ot_ip.h" #include "ot_mutex.h" #include "ot_random.h" #include "ot_scan.h" #include "ot_socket.h" #include "ot_stats.h" #include "ot_vector.h" #include "trackerlogic.h" #include "ot_livesync.h" ot_ip6 g_serverip; uint16_t g_serverport = 9009; uint32_t g_tracker_id; ot_ip6 groupip_1 = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 224, 0, 23, 5}; int g_self_pipe[2]; /* If you have more than 10 peers, don't use this proxy Use 20 slots for 10 peers to have room for 10 incoming connection slots */ #define MAX_PEERS 20 #define LIVESYNC_INCOMING_BUFFSIZE (256 * 256) #define STREAMSYNC_OUTGOING_BUFFSIZE (256 * 256) #define LIVESYNC_OUTGOING_BUFFSIZE_PEERS 1480 #define LIVESYNC_OUTGOING_WATERMARK_PEERS (sizeof(ot_peer) + sizeof(ot_hash)) #define LIVESYNC_MAXDELAY 15 /* seconds */ /* The amount of time a complete sync cycle should take */ #define OT_SYNC_INTERVAL_MINUTES 2 /* So after each bucket wait 1 / OT_BUCKET_COUNT intervals */ #define OT_SYNC_SLEEP (((OT_SYNC_INTERVAL_MINUTES) * 60 * 1000000) / (OT_BUCKET_COUNT)) enum { OT_SYNC_PEER4, OT_SYNC_PEER6 }; enum { FLAG_SERVERSOCKET = 1 }; /* For incoming packets */ static int g_socket_in = -1; static uint8_t g_inbuffer[LIVESYNC_INCOMING_BUFFSIZE]; /* For outgoing packets */ static int g_socket_out = -1; static uint8_t g_peerbuffer_start[LIVESYNC_OUTGOING_BUFFSIZE_PEERS]; static uint8_t *g_peerbuffer_pos; static uint8_t *g_peerbuffer_highwater = g_peerbuffer_start + LIVESYNC_OUTGOING_BUFFSIZE_PEERS - LIVESYNC_OUTGOING_WATERMARK_PEERS; static ot_time g_next_packet_time; static void *livesync_worker(void *args); static void *streamsync_worker(void *args); static void livesync_proxytell(uint8_t prefix, uint8_t *info_hash, uint8_t *peer); void exerr(const char *message) { fprintf(stderr, "%s\n", message); exit(111); } void stats_issue_event(struct ot_workstruct *ws, ot_status_event event, PROTO_FLAG proto, uintptr_t event_data) { (void)ws; (void)event; (void)proto; (void)event_data; } void livesync_bind_mcast(ot_ip6 ip, uint16_t port) { if (!ip_is_v4mapped(ip)) exerr("v6 mcast support not yet available."); if (g_socket_in != -1) exerr("Error: Livesync listen ip specified twice."); if ((g_socket_in = socket_mcast_join(groupip_1, ip, port)) == -1) exerr("Error: Cant bind live sync incoming socket."); if ((g_socket_out = socket_mcast_bind(ip, port, 1)) == -1) exerr("Error: Cant bind live sync outgoing socket."); } size_t add_peer_to_torrent_proxy(ot_hash hash, ot_peer *peer, size_t peer_size) { int exactmatch; ot_torrent *torrent; ot_peerlist *peer_list; ot_peer *peer_dest; ot_vector *torrents_list = mutex_bucket_lock_by_hash(hash); size_t compare_size = OT_PEER_COMPARE_SIZE_FROM_PEER_SIZE(peer_size); torrent = vector_find_or_insert(torrents_list, (void *)hash, sizeof(ot_torrent), compare_size, &exactmatch); if (!torrent) return -1; if (!exactmatch) { /* Create a new torrent entry, then */ memcpy(torrent->hash, hash, sizeof(ot_hash)); if (!(torrent->peer_list6 = malloc(sizeof(ot_peerlist))) || !(torrent->peer_list4 = malloc(sizeof(ot_peerlist)))) { vector_remove_torrent(torrents_list, torrent); mutex_bucket_unlock_by_hash(hash, 0); return -1; } memset(torrent->peer_list6, 0, sizeof(ot_peerlist)); memset(torrent->peer_list4, 0, sizeof(ot_peerlist)); } peer_list = peer_size == OT_PEER_SIZE6 ? torrent->peer_list6 : torrent->peer_list4; /* Check for peer in torrent */ peer_dest = vector_find_or_insert_peer(&(peer_list->peers), peer, peer_size, &exactmatch); if (!peer_dest) { mutex_bucket_unlock_by_hash(hash, 0); return -1; } /* Tell peer that it's fresh */ OT_PEERTIME(peer, peer_size) = 0; /* If we hadn't had a match create peer there */ if (!exactmatch) { peer_list->peer_count++; if (OT_PEERFLAG_D(peer, peer_size) & PEER_FLAG_SEEDING) peer_list->seed_count++; } memcpy(peer_dest, peer, peer_size); mutex_bucket_unlock_by_hash(hash, 0); return 0; } size_t remove_peer_from_torrent_proxy(ot_hash hash, ot_peer *peer, size_t peer_size) { int exactmatch; ot_vector *torrents_list = mutex_bucket_lock_by_hash(hash); ot_torrent *torrent = vector_binary_search(hash, torrents_list->data, torrents_list->size, sizeof(ot_torrent), OT_HASH_COMPARE_SIZE, &exactmatch); if (exactmatch) { ot_peerlist *peer_list = peer_list = peer_size == OT_PEER_SIZE6 ? torrent->peer_list6 : torrent->peer_list4; switch (vector_remove_peer(&peer_list->peers, peer, peer_size)) { case 2: peer_list->seed_count--; /* Intentional fallthrough */ case 1: peer_list->peer_count--; /* Intentional fallthrough */ default: break; } } mutex_bucket_unlock_by_hash(hash, 0); return 0; } void free_peerlist(ot_peerlist *peer_list) { if (peer_list->peers.data) { if (OT_PEERLIST_HASBUCKETS(peer_list)) { ot_vector *bucket_list = (ot_vector *)(peer_list->peers.data); while (peer_list->peers.size--) free(bucket_list++->data); } free(peer_list->peers.data); } free(peer_list); } static void livesync_handle_peersync(ssize_t datalen, size_t peer_size) { int off = sizeof(g_tracker_id) + sizeof(uint32_t); fprintf(stderr, "."); while ((ssize_t)(off + sizeof(ot_hash) + peer_size) <= datalen) { ot_peer *peer = (ot_peer *)(g_inbuffer + off + sizeof(ot_hash)); ot_hash *hash = (ot_hash *)(g_inbuffer + off); if (OT_PEERFLAG_D(peer, peer_size) & PEER_FLAG_STOPPED) remove_peer_from_torrent_proxy(*hash, peer, peer_size); else add_peer_to_torrent_proxy(*hash, peer, peer_size); off += sizeof(ot_hash) + peer_size; } } int usage(char *self) { fprintf(stderr, "Usage: %s -L -l : -c :\n", self); return 0; } enum { FLAG_OUTGOING = 0x80, FLAG_DISCONNECTED = 0x00, FLAG_CONNECTING = 0x01, FLAG_WAITTRACKERID = 0x02, FLAG_CONNECTED = 0x03, FLAG_MASK = 0x07 }; #define PROXYPEER_NEEDSCONNECT(flag) ((flag) == FLAG_OUTGOING) #define PROXYPEER_ISCONNECTED(flag) (((flag) & FLAG_MASK) == FLAG_CONNECTED) #define PROXYPEER_SETDISCONNECTED(flag) (flag) = (((flag) & FLAG_OUTGOING) | FLAG_DISCONNECTED) #define PROXYPEER_SETCONNECTING(flag) (flag) = (((flag) & FLAG_OUTGOING) | FLAG_CONNECTING) #define PROXYPEER_SETWAITTRACKERID(flag) (flag) = (((flag) & FLAG_OUTGOING) | FLAG_WAITTRACKERID) #define PROXYPEER_SETCONNECTED(flag) (flag) = (((flag) & FLAG_OUTGOING) | FLAG_CONNECTED) typedef struct { int state; /* Whether we want to connect, how far our handshake is, etc. */ ot_ip6 ip; /* The peer to connect to */ uint16_t port; /* The peers port */ uint8_t indata[8192 * 16]; /* Any data not processed yet */ size_t indata_length; /* Length of unprocessed data */ uint32_t tracker_id; /* How the other end greeted */ int64_t fd; /* A file handle, if connected, <= 0 is disconnected (0 initially, -1 else) */ ot_batch outdata; /* The batch containing our sync data */ size_t packet_tcount; /* Number of unprocessed torrents in packet we currently receive */ uint8_t packet_tprefix; /* Prefix byte for all torrents in current packet */ uint8_t packet_type; /* Type of current packet */ uint32_t packet_tid; /* Tracker id for current packet */ } proxy_peer; static void process_indata(proxy_peer *peer); void reset_info_block(proxy_peer *peer) { peer->indata_length = 0; peer->tracker_id = 0; peer->fd = -1; peer->packet_tcount = 0; batch_reset(&peer->outdata); PROXYPEER_SETDISCONNECTED(peer->state); } /* Number of connections to peers * If a peer's IP is set, we try to reconnect, when the connection drops * If we already have a connected tracker_id in our records for an _incoming_ connection, drop it * Multiple connections to/from the same ip are okay, if tracker_id doesn't match * Reconnect attempts occur only twice a minute */ static size_t g_connection_count; static ot_time g_connection_reconn; static proxy_peer g_connections[MAX_PEERS]; static void handle_reconnects(void) { size_t i; for (i = 0; i < g_connection_count; ++i) if (PROXYPEER_NEEDSCONNECT(g_connections[i].state)) { int64_t newfd = socket_tcp_connect(g_serverip, g_serverport, g_connections[i].ip, g_connections[i].port); fprintf(stderr, "(Re)connecting to peer..."); if (newfd == -1) continue; /* No socket for you */ io_fd(newfd); io_wantwrite(newfd); /* So we will be informed when it is connected */ fdstate_open(newfd, g_connections + i, 0); /* Prepare connection info block */ reset_info_block(g_connections + i); g_connections[i].fd = newfd; PROXYPEER_SETCONNECTING(g_connections[i].state); } g_connection_reconn = time(NULL) + 30; } /* Handle incoming connection requests, check against whitelist */ static void handle_accept(int64_t serversocket) { int64_t newfd; ot_ip6 ip; uint16_t port; while ((newfd = socket_accept(serversocket, ip, &port)) != -1) { /* XXX some access control */ /* Put fd into a non-blocking mode */ io_nonblock(newfd); if (!io_fd(newfd)) io_close(newfd); else { /* Find a new home for our incoming connection */ int i; for (i = 0; i < MAX_PEERS; ++i) if (g_connections[i].state == FLAG_DISCONNECTED) break; if (i == MAX_PEERS) { fprintf(stderr, "No room for incoming connection."); close(newfd); continue; } /* Prepare connection info block */ reset_info_block(g_connections + i); PROXYPEER_SETCONNECTING(g_connections[i].state); g_connections[i].port = port; g_connections[i].fd = newfd; fdstate_open(newfd, g_connections + i, 0); /* We expect the connecting side to begin with its tracker_id */ io_wantread(newfd); } } return; } /* New sync data on the stream */ static void handle_read(int64_t peersocket) { int i; int64_t datalen; uint32_t tracker_id; proxy_peer *peer = fdstate_cookie(peersocket); if (!peer) { /* Can't happen ;) */ io_close(peersocket); return; } switch (peer->state & FLAG_MASK) { case FLAG_DISCONNECTED: io_close(peersocket); break; /* Shouldnt happen */ case FLAG_CONNECTING: case FLAG_WAITTRACKERID: /* We want at least the first four bytes to come at once, to avoid keeping extra states (for now) This also catches 0 bytes reads == EOF and negative values, denoting connection errors */ if (io_tryread(peersocket, (void *)&tracker_id, sizeof(tracker_id)) != sizeof(tracker_id)) goto close_socket; /* See, if we already have a connection to that peer */ for (i = 0; i < MAX_PEERS; ++i) if ((g_connections[i].state & FLAG_MASK) == FLAG_CONNECTED && g_connections[i].tracker_id == tracker_id) { fprintf(stderr, "Peer already connected. Closing connection.\n"); goto close_socket; } /* Also no need for soliloquy */ if (tracker_id == g_tracker_id) goto close_socket; /* The new connection is good, send our tracker_id on incoming connections */ if (peer->state == FLAG_CONNECTING) if (io_trywrite(peersocket, (void *)&g_tracker_id, sizeof(g_tracker_id)) != sizeof(g_tracker_id)) goto close_socket; peer->tracker_id = tracker_id; PROXYPEER_SETCONNECTED(peer->state); if (peer->state & FLAG_OUTGOING) fprintf(stderr, "succeeded.\n"); else fprintf(stderr, "Incoming connection successful.\n"); break; close_socket: fprintf(stderr, "Handshake incomplete, closing socket\n"); io_close(peersocket); reset_info_block(peer); break; case FLAG_CONNECTED: /* Here we acutally expect data from peer indata_length should be less than 20+256*7 bytes, for incomplete torrent entries */ datalen = io_tryread(peersocket, (void *)(peer->indata + peer->indata_length), sizeof(peer->indata) - peer->indata_length); if (!datalen || datalen < -1) { fprintf(stderr, "Connection closed by remote peer.\n"); io_close(peersocket); reset_info_block(peer); } else if (datalen > 0) { peer->indata_length += datalen; process_indata(peer); } break; } } /* Can write new sync data to the stream */ static void handle_write(int64_t peersocket) { proxy_peer *peer = fdstate_cookie(peersocket); if (!peer) { /* Can't happen ;) */ io_close(peersocket); return; } switch (peer->state & FLAG_MASK) { case FLAG_DISCONNECTED: default: /* Should not happen */ io_close(peersocket); break; case FLAG_CONNECTING: /* Ensure that the connection is established and handle connection error */ if (peer->state & FLAG_OUTGOING && !socket_is_connected(peersocket)) { fprintf(stderr, "failed\n"); reset_info_block(peer); io_close(peersocket); break; } if (io_trywrite(peersocket, (void *)&g_tracker_id, sizeof(g_tracker_id)) == sizeof(g_tracker_id)) { PROXYPEER_SETWAITTRACKERID(peer->state); io_dontwantwrite(peersocket); io_wantread(peersocket); } else { fprintf(stderr, "Handshake incomplete, closing socket\n"); io_close(peersocket); reset_info_block(peer); } break; case FLAG_CONNECTED: switch (batch_send(peersocket, &peer->outdata)) { case BATCH_DONE: io_dontwantwrite(peersocket); break; case BATCH_FAILED: io_close(peersocket); reset_info_block(peer); break; default: /* Normal operation or eagain */ break; } break; } return; } static void server_mainloop() { int64_t sock; uint32_t packet_type; /* inlined livesync_init() */ memset(g_peerbuffer_start, 0, sizeof(g_peerbuffer_start)); g_peerbuffer_pos = g_peerbuffer_start; memcpy(g_peerbuffer_pos, &g_tracker_id, sizeof(g_tracker_id)); packet_type = htonl(OT_SYNC_PEER); memcpy(g_peerbuffer_pos + sizeof(g_tracker_id), &packet_type, sizeof(packet_type)); g_peerbuffer_pos += sizeof(g_tracker_id) + sizeof(uint32_t); g_next_packet_time = time(NULL) + LIVESYNC_MAXDELAY; while (1) { /* See if we need to connect to anyone */ if (time(NULL) > g_connection_reconn) handle_reconnects(); /* Wait for io events until next approx reconn check time */ io_waituntil2(30 * 1000); /* Loop over readable sockets */ while ((sock = io_canread()) != -1) { const void *cookie = fdstate_cookie(sock); if ((uintptr_t)cookie == FLAG_SERVERSOCKET) handle_accept(sock); else handle_read(sock); } /* Loop over writable sockets */ while ((sock = io_canwrite()) != -1) handle_write(sock); livesync_ticker(); } } static void panic(const char *routine) { fprintf(stderr, "%s: %s\n", routine, strerror(errno)); exit(111); } static int64_t ot_try_bind(ot_ip6 ip, uint16_t port) { int64_t sock = socket_tcp_listen(ip, port, SOCKET_EXCLUSIVE); if (sock == -1) panic("socket_tcp_listen"); if (!io_fd(sock)) panic("io_fd"); fdstate_open(sock, (void *)FLAG_SERVERSOCKET, 0); io_wantread(sock); return sock; } static int scan_ip6_port(const char *src, ot_ip6 ip, uint16_t *port) { const char *s = src; uint64_t value; int off, bracket = 0; while (isspace(*s)) ++s; if (*s == '[') ++s, ++bracket; /* for v6 style notation */ if (!(off = ip_parse(s, ip))) return 0; s += off; if (*s == 0 || isspace(*s)) return s - src; if (*s == ']' && bracket) ++s; if (!ip_is_v4mapped(ip)) { if ((bracket && *(s) != ':') || (*(s) != '.')) return 0; s++; } else { if (*(s++) != ':') return 0; } if (!(off = scan_uint(s, &value, 65535))) return 0; *port = (uint16_t)value; return off + s - src; } int main(int argc, char **argv) { static pthread_t sync_in_thread_id; static pthread_t sync_out_thread_id; ot_ip6 serverip; uint16_t tmpport; int scanon = 1, lbound = 0, sbound = 0; random_init(); if (fdstate_init() == -1) panic("fdstate_init"); g_tracker_id = random_uint32(); while (scanon) { switch (getopt(argc, argv, ":l:c:L:h")) { case -1: scanon = 0; break; case 'l': tmpport = 0; if (!scan_ip6_port(optarg, serverip, &tmpport) || !tmpport) { usage(argv[0]); exit(1); } ot_try_bind(serverip, tmpport); ++sbound; break; case 'c': if (g_connection_count > MAX_PEERS / 2) exerr("Connection limit exceeded.\n"); tmpport = 0; if (!scan_ip6_port(optarg, g_connections[g_connection_count].ip, &g_connections[g_connection_count].port) || !g_connections[g_connection_count].port) { usage(argv[0]); exit(1); } g_connections[g_connection_count++].state = FLAG_OUTGOING; break; case 'L': tmpport = 9696; if (!scan_ip6_port(optarg, serverip, &tmpport) || !tmpport) { usage(argv[0]); exit(1); } livesync_bind_mcast(serverip, tmpport); ++lbound; break; default: case '?': usage(argv[0]); exit(1); } } if (!lbound) exerr("No livesync port bound."); if (!g_connection_count && !sbound) exerr("No streamsync port bound."); pthread_create(&sync_in_thread_id, NULL, livesync_worker, NULL); pthread_create(&sync_out_thread_id, NULL, streamsync_worker, NULL); server_mainloop(); return 0; } static void *streamsync_worker(void *args) { (void)args; while (1) { size_t bucket; /* For each bucket... */ for (bucket = 0; bucket < OT_BUCKET_COUNT; ++bucket) { /* Get exclusive access to that bucket */ ot_vector *torrents_list = mutex_bucket_lock(bucket); size_t tor_offset, count_def = 0, count_one = 0, count_two = 0, count_peers = 0; size_t mem, mem_a = 0, mem_b = 0; uint8_t *ptr = 0, *ptr_a, *ptr_b, *ptr_c; if (!torrents_list->size) goto unlock_continue; /* For each torrent in this bucket.. */ for (tor_offset = 0; tor_offset < torrents_list->size; ++tor_offset) { /* Address torrents members */ ot_peerlist *peer_list = (((ot_torrent *)(torrents_list->data))[tor_offset]).peer_list; switch (peer_list->peer_count) { case 2: count_two++; break; case 1: count_one++; break; case 0: break; default: count_def++; count_peers += peer_list->peer_count; } } /* Maximal memory requirement: max 3 blocks, max torrents * 20 + max peers * 7 */ mem = 3 * (1 + 1 + 2) + (count_one + count_two) * (19 + 1) + count_def * (19 + 8) + (count_one + 2 * count_two + count_peers) * 7; fprintf(stderr, "Mem: %zd\n", mem); ptr = ptr_a = ptr_b = ptr_c = malloc(mem); if (!ptr) goto unlock_continue; if (count_one > 4 || !count_def) { mem_a = 1 + 1 + 2 + count_one * (19 + 7); ptr_b += mem_a; ptr_c += mem_a; ptr_a[0] = 1; /* Offset 0: packet type 1 */ ptr_a[1] = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 1: the shared prefix */ ptr_a[2] = count_one >> 8; ptr_a[3] = count_one & 255; ptr_a += 4; } else count_def += count_one; if (count_two > 4 || !count_def) { mem_b = 1 + 1 + 2 + count_two * (19 + 14); ptr_c += mem_b; ptr_b[0] = 2; /* Offset 0: packet type 2 */ ptr_b[1] = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 1: the shared prefix */ ptr_b[2] = count_two >> 8; ptr_b[3] = count_two & 255; ptr_b += 4; } else count_def += count_two; if (count_def) { ptr_c[0] = 0; /* Offset 0: packet type 0 */ ptr_c[1] = (bucket << 8) >> OT_BUCKET_COUNT_BITS; /* Offset 1: the shared prefix */ ptr_c[2] = count_def >> 8; ptr_c[3] = count_def & 255; ptr_c += 4; } /* For each torrent in this bucket.. */ for (tor_offset = 0; tor_offset < torrents_list->size; ++tor_offset) { /* Address torrents members */ ot_torrent *torrent = ((ot_torrent *)(torrents_list->data)) + tor_offset; ot_peerlist *peer_list = torrent->peer_list; ot_peer *peers = (ot_peer *)(peer_list->peers.data); uint8_t **dst; /* Determine destination slot */ count_peers = peer_list->peer_count; switch (count_peers) { case 0: continue; case 1: dst = mem_a ? &ptr_a : &ptr_c; break; case 2: dst = mem_b ? &ptr_b : &ptr_c; break; default: dst = &ptr_c; break; } /* Copy tail of info_hash, advance pointer */ memcpy(*dst, ((uint8_t *)torrent->hash) + 1, sizeof(ot_hash) - 1); *dst += sizeof(ot_hash) - 1; /* Encode peer count */ if (dst == &ptr_c) while (count_peers) { if (count_peers <= 0x7f) *(*dst)++ = count_peers; else *(*dst)++ = 0x80 | (count_peers & 0x7f); count_peers >>= 7; } /* Copy peers */ count_peers = peer_list->peer_count; while (count_peers--) { memcpy(*dst, peers++, OT_IP_SIZE + 3); *dst += OT_IP_SIZE + 3; } free_peerlist(peer_list); } free(torrents_list->data); memset(torrents_list, 0, sizeof(*torrents_list)); unlock_continue: mutex_bucket_unlock(bucket, 0); if (ptr) { int i; if (ptr_b > ptr_c) ptr_c = ptr_b; if (ptr_a > ptr_c) ptr_c = ptr_a; mem = ptr_c - ptr; for (i = 0; i < MAX_PEERS; ++i) { if (PROXYPEER_ISCONNECTED(g_connections[i].state)) { void *tmp = malloc(mem); if (tmp) { memcpy(tmp, ptr, mem); batch_add(&g_connections[i].outdata, tmp, mem); io_wantwrite(g_connections[i].fd); } } } free(ptr); } usleep(OT_SYNC_SLEEP); } } return 0; } static void livesync_issue_peersync() { socket_mcast_send(g_socket_out, g_peerbuffer_start, (size_t)(g_peerbuffer_pos - g_peerbuffer_start), groupip_1, LIVESYNC_PORT); g_peerbuffer_pos = g_peerbuffer_start + sizeof(g_tracker_id) + sizeof(uint32_t); g_next_packet_time = time(NULL) + LIVESYNC_MAXDELAY; } void livesync_ticker() { /* livesync_issue_peersync sets g_next_packet_time */ if (time(NULL) > g_next_packet_time && g_peerbuffer_pos > g_peerbuffer_start + sizeof(g_tracker_id)) livesync_issue_peersync(); } static void livesync_proxytell(uint8_t prefix, uint8_t *info_hash, uint8_t *peer) { // unsigned int i; *g_peerbuffer_pos = prefix; memcpy(g_peerbuffer_pos + 1, info_hash, sizeof(ot_hash) - 1); memcpy(g_peerbuffer_pos + sizeof(ot_hash), peer, sizeof(ot_peer) - 1); #if 0 /* Dump info_hash */ for( i=0; i= g_peerbuffer_highwater) livesync_issue_peersync(); } static void process_indata(proxy_peer *peer) { size_t consumed, peers; uint8_t *data = peer->indata, *hash; uint8_t *dataend = data + peer->indata_length; while (1) { /* If we're not inside of a packet, make a new one */ if (!peer->packet_tcount) { /* Ensure the header is complete or postpone processing */ if (data + 4 > dataend) break; peer->packet_type = data[0]; peer->packet_tprefix = data[1]; peer->packet_tcount = data[2] * 256 + data[3]; data += 4; printf("type: %hhu, prefix: %02X, torrentcount: %zd\n", peer->packet_type, peer->packet_tprefix, peer->packet_tcount); } /* Ensure size for a minimal torrent block */ if (data + sizeof(ot_hash) + OT_IP_SIZE + 3 > dataend) break; /* Advance pointer to peer count or peers */ hash = data; data += sizeof(ot_hash) - 1; /* Type 0 has peer count encoded before each peers */ peers = peer->packet_type; if (!peers) { int shift = 0; do peers |= (0x7f & *data) << (7 * shift); while (*(data++) & 0x80 && shift++ < 6); } #if 0 printf( "peers: %zd\n", peers ); #endif /* Ensure enough data being read to hold all peers */ if (data + (OT_IP_SIZE + 3) * peers > dataend) { data = hash; break; } while (peers--) { livesync_proxytell(peer->packet_tprefix, hash, data); data += OT_IP_SIZE + 3; } --peer->packet_tcount; } consumed = data - peer->indata; memmove(peer->indata, data, peer->indata_length - consumed); peer->indata_length -= consumed; } static void *livesync_worker(void *args) { (void)args; while (1) { ot_ip6 in_ip; uint16_t in_port; uint32_t packet_type; ssize_t datalen = socket_recv(g_socket_in, g_inbuffer, LIVESYNC_INCOMING_BUFFSIZE, in_ip, &in_port); /* Expect at least tracker id and packet type */ if (datalen <= (ssize_t)(sizeof(g_tracker_id) + sizeof(uint32_t))) continue; if (!memcmp(g_inbuffer, &g_tracker_id, sizeof(g_tracker_id))) { /* drop packet coming from ourselves */ continue; } memcpy(&packet_type, g_inbuffer + sizeof(g_tracker_id), sizeof(packet_type)); switch (ntohl(packet_type)) { case OT_SYNC_PEER4: livesync_handle_peersync(datalen, OT_PEER_SIZE4); break; case OT_SYNC_PEER6: livesync_handle_peersync(datalen, OT_PEER_SIZE6); break; default: // fprintf( stderr, "Received an unknown live sync packet type %u.\n", ntohl( packet_type ) ); break; } } return 0; }