source: trunk/libtransmission/tr-dht.c @ 9522

Last change on this file since 9522 was 9522, checked in by charles, 13 years ago

(trunk) #2222: revert r8967, as foreshadowed in http://trac.transmissionbt.com/ticket/2222#comment:20

File size: 11.8 KB
Line 
1/*
2Copyright (c) 2009 by Juliusz Chroboczek
3
4Permission is hereby granted, free of charge, to any person obtaining a copy
5of this software and associated documentation files (the "Software"), to deal
6in the Software without restriction, including without limitation the rights
7to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
8copies of the Software, and to permit persons to whom the Software is
9furnished to do so, subject to the following conditions:
10
11The above copyright notice and this permission notice shall be included in
12all copies or substantial portions of the Software.
13
14THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
15IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
16FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL THE
17AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
18LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
19OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
20THE SOFTWARE.
21*/
22
23/* ansi */
24#include <errno.h>
25#include <stdio.h>
26
27/* posix */
28#include <netinet/in.h> /* sockaddr_in */
29#include <signal.h> /* sig_atomic_t */
30#include <sys/time.h>
31#include <sys/types.h>
32#include <sys/socket.h> /* socket(), bind() */
33#include <unistd.h> /* close() */
34
35/* third party */
36#include <event.h>
37#include <dht/dht.h>
38
39/* libT */
40#include "transmission.h"
41#include "bencode.h"
42#include "crypto.h"
43#include "net.h"
44#include "peer-mgr.h" /* tr_peerMgrCompactToPex() */
45#include "platform.h" /* tr_threadNew() */
46#include "session.h"
47#include "torrent.h" /* tr_torrentFindFromHash() */
48#include "tr-dht.h"
49#include "trevent.h" /* tr_runInEventThread() */
50#include "utils.h"
51#include "version.h"
52
53static int dht_socket;
54static struct event dht_event;
55static tr_port dht_port;
56static unsigned char myid[20];
57static tr_session *session = NULL;
58
59static void event_callback(int s, short type, void *ignore);
60
61struct bootstrap_closure {
62    tr_session *session;
63    uint8_t *nodes;
64    size_t len;
65};
66
67static void
68dht_bootstrap(void *closure)
69{
70    struct bootstrap_closure *cl = closure;
71    size_t i;
72
73    if(session != cl->session)
74        return;
75
76    for(i = 0; i < cl->len; i += 6)
77    {
78        struct timeval tv;
79        tr_port port;
80        struct tr_address addr;
81        int status;
82
83        memset(&addr, 0, sizeof(addr));
84        addr.type = TR_AF_INET;
85        memcpy(&addr.addr.addr4, &cl->nodes[i], 4);
86        memcpy(&port, &cl->nodes[i + 4], 2);
87        port = ntohs(port);
88        /* There's no race here -- if we uninit between the test and the
89           AddNode, the AddNode will be ignored. */
90        status = tr_dhtStatus(cl->session, NULL);
91        if(status == TR_DHT_STOPPED || status >= TR_DHT_FIREWALLED)
92            break;
93        tr_dhtAddNode(cl->session, &addr, port, 1);
94        tr_timevalSet( &tv, 2 + tr_cryptoWeakRandInt( 5 ), tr_cryptoWeakRandInt( 1000000 ) );
95        select( 0, NULL, NULL, NULL, &tv );
96    }
97    tr_free( cl->nodes );
98    tr_free( closure );
99    tr_ndbg( "DHT", "Finished bootstrapping" );
100}
101
102int
103tr_dhtInit(tr_session *ss, tr_address * tr_addr)
104{
105    struct sockaddr_in sin;
106    tr_benc benc;
107    int rc;
108    tr_bool have_id = FALSE;
109    char * dat_file;
110    uint8_t * nodes = NULL;
111    const uint8_t * raw;
112    size_t len;
113    char v[5];
114
115    if( session ) /* already initialized */
116        return -1;
117
118    dht_port = tr_sessionGetPeerPort(ss);
119    if(dht_port <= 0)
120        return -1;
121
122    tr_ndbg( "DHT", "Initializing DHT" );
123
124    dht_socket = socket(PF_INET, SOCK_DGRAM, 0);
125    if(dht_socket < 0)
126        goto fail;
127
128    memset(&sin, 0, sizeof(sin));
129    sin.sin_family = AF_INET;
130    memcpy(&(sin.sin_addr), &(tr_addr->addr.addr4), sizeof (struct in_addr));
131    sin.sin_port = htons(dht_port);
132    rc = bind(dht_socket, (struct sockaddr*)&sin, sizeof(sin));
133    if(rc < 0)
134        goto fail;
135
136    if( getenv( "TR_DHT_VERBOSE" ) != NULL )
137        dht_debug = stderr;
138
139    dat_file = tr_buildPath( ss->configDir, "dht.dat", NULL );
140    rc = tr_bencLoadFile( &benc, TR_FMT_BENC, dat_file );
141    tr_free( dat_file );
142    if(rc == 0) {
143        if(( have_id = tr_bencDictFindRaw( &benc, "id", &raw, &len ) && len==20 ))
144            memcpy( myid, raw, len );
145        if( tr_bencDictFindRaw( &benc, "nodes", &raw, &len ) && !(len%6) ) {
146            nodes = tr_memdup( raw, len );
147            tr_ninf( "DHT", "Bootstrapping from %d old nodes", (int)(len/6) );
148        }
149        tr_bencFree( &benc );
150    }
151
152    if( have_id )
153        tr_ninf( "DHT", "Reusing old id" );
154    else {
155        /* Note that DHT ids need to be distributed uniformly,
156         * so it should be something truly random. */
157        tr_ninf( "DHT", "Generating new id" );
158        tr_cryptoRandBuf( myid, 20 );
159    }
160
161    v[0] = 'T';
162    v[1] = 'R';
163    v[2] = (SVN_REVISION_NUM >> 8) & 0xFF;
164    v[3] = SVN_REVISION_NUM & 0xFF;
165    rc = dht_init( dht_socket, myid, (const unsigned char*)v );
166    if(rc < 0)
167        goto fail;
168
169    session = ss;
170
171    if(nodes) {
172        struct bootstrap_closure * cl = tr_new( struct bootstrap_closure, 1 );
173        cl->session = session;
174        cl->nodes = nodes;
175        cl->len = len;
176        tr_threadNew( dht_bootstrap, cl );
177    }
178
179    event_set( &dht_event, dht_socket, EV_READ, event_callback, NULL );
180    tr_timerAdd( &dht_event, 0, tr_cryptoWeakRandInt( 1000000 ) );
181
182    tr_ndbg( "DHT", "DHT initialized" );
183
184    return 1;
185
186    fail:
187    {
188        const int save = errno;
189        close(dht_socket);
190        dht_socket = -1;
191        session = NULL;
192        tr_ndbg( "DHT", "DHT initialization failed (errno = %d)", save );
193        errno = save;
194    }
195
196    return -1;
197}
198
199void
200tr_dhtUninit(tr_session *ss)
201{
202    if(session != ss)
203        return;
204
205    tr_ndbg( "DHT", "Uninitializing DHT" );
206
207    event_del(&dht_event);
208
209    /* Since we only save known good nodes, avoid erasing older data if we
210       don't know enough nodes. */
211    if(tr_dhtStatus(ss, NULL) < TR_DHT_FIREWALLED)
212        tr_ninf( "DHT", "Not saving nodes, DHT not ready" );
213    else {
214        tr_benc benc;
215        struct sockaddr_in sins[300];
216        char compact[300 * 6];
217        char *dat_file;
218        int i;
219        int n = dht_get_nodes(sins, 300);
220        int j = 0;
221
222        tr_ninf( "DHT", "Saving %d nodes", n );
223        for( i=0; i<n; ++i ) {
224            memcpy( compact + j, &sins[i].sin_addr, 4 );
225            memcpy( compact + j + 4, &sins[i].sin_port, 2 );
226            j += 6;
227        }
228        tr_bencInitDict( &benc, 2 );
229        tr_bencDictAddRaw( &benc, "id", myid, 20 );
230        tr_bencDictAddRaw( &benc, "nodes", compact, j );
231        dat_file = tr_buildPath( ss->configDir, "dht.dat", NULL );
232        tr_bencToFile( &benc, TR_FMT_BENC, dat_file );
233        tr_bencFree( &benc );
234        tr_free( dat_file );
235    }
236
237    dht_uninit( dht_socket, 0 );
238    tr_netCloseSocket( dht_socket );
239
240    tr_ndbg("DHT", "Done uninitializing DHT");
241
242    session = NULL;
243}
244
245tr_bool
246tr_dhtEnabled( const tr_session * ss )
247{
248    return ss && ( ss == session );
249}
250
251struct getstatus_closure
252{
253    sig_atomic_t status;
254    sig_atomic_t count;
255};
256
257static void
258getstatus( void * closure )
259{
260    struct getstatus_closure * ret = closure;
261    int good, dubious, incoming;
262
263    dht_nodes( &good, &dubious, NULL, &incoming );
264
265    ret->count = good + dubious;
266
267    if( good < 4 || good + dubious <= 8 )
268        ret->status = TR_DHT_BROKEN;
269    else if( good < 40 )
270        ret->status = TR_DHT_POOR;
271    else if( incoming < 8 )
272        ret->status = TR_DHT_FIREWALLED;
273    else
274        ret->status = TR_DHT_GOOD;
275}
276
277int
278tr_dhtStatus( tr_session * ss, int * nodes_return )
279{
280    struct getstatus_closure ret = { -1, - 1 };
281
282    if( !tr_dhtEnabled( ss ) )
283        return TR_DHT_STOPPED;
284
285    tr_runInEventThread( ss, getstatus, &ret );
286    while( ret.status < 0 )
287        tr_wait( 10 /*msec*/ );
288
289    if( nodes_return )
290        *nodes_return = ret.count;
291
292    return ret.status;
293}
294
295tr_port
296tr_dhtPort( const tr_session *ss )
297{
298    return tr_dhtEnabled( ss ) ? dht_port : 0;
299}
300
301int
302tr_dhtAddNode( tr_session       * ss,
303               const tr_address * address,
304               tr_port            port,
305               tr_bool            bootstrap )
306{
307    struct sockaddr_in sin;
308
309    if( !tr_dhtEnabled( ss ) )
310        return 0;
311
312    if( address->type != TR_AF_INET )
313        return 0;
314
315    /* Since we don't want to abuse our bootstrap nodes,
316     * we don't ping them if the DHT is in a good state. */
317    if(bootstrap) {
318        if(tr_dhtStatus(ss, NULL) >= TR_DHT_FIREWALLED)
319            return 0;
320    }
321
322    memset(&sin, 0, sizeof(sin));
323    sin.sin_family = AF_INET;
324    memcpy(&sin.sin_addr, &address->addr.addr4, 4);
325    sin.sin_port = htons(port);
326    dht_ping_node(dht_socket, &sin);
327
328    return 1;
329}
330
331const char *
332tr_dhtPrintableStatus(int status)
333{
334    switch(status) {
335    case TR_DHT_STOPPED: return "stopped";
336    case TR_DHT_BROKEN: return "broken";
337    case TR_DHT_POOR: return "poor";
338    case TR_DHT_FIREWALLED: return "firewalled";
339    case TR_DHT_GOOD: return "good";
340    default: return "???";
341    }
342}
343
344static void
345callback( void *ignore UNUSED, int event,
346          unsigned char *info_hash, void *data, size_t data_len )
347{
348    if( event == DHT_EVENT_VALUES )
349    {
350        tr_torrent *tor;
351        tr_globalLock( session );
352        tor = tr_torrentFindFromHash( session, info_hash );
353        if( tor && tr_torrentAllowsDHT( tor ))
354        {
355            size_t i, n;
356            tr_pex * pex = tr_peerMgrCompactToPex(data, data_len, NULL, 0, &n);
357            for( i=0; i<n; ++i )
358                tr_peerMgrAddPex( tor, TR_PEER_FROM_DHT, pex+i );
359            tr_free(pex);
360            tr_torinf(tor, "Learned %d peers from DHT", (int)n);
361        }
362        tr_globalUnlock( session );
363    }
364    else if( event == DHT_EVENT_SEARCH_DONE )
365    {
366        tr_torrent * tor = tr_torrentFindFromHash( session, info_hash );
367        if( tor ) {
368            tr_torinf(tor, "DHT announce done");
369            tor->dhtAnnounceInProgress = 0;
370        }
371    }
372}
373
374int
375tr_dhtAnnounce(tr_torrent *tor, tr_bool announce)
376{
377    int rc, status, numnodes;
378
379    if( !tr_torrentAllowsDHT( tor ) )
380        return -1;
381
382    status = tr_dhtStatus( tor->session, &numnodes );
383    if(status < TR_DHT_POOR ) {
384        tr_tordbg(tor, "DHT not ready (%s, %d nodes)",
385                  tr_dhtPrintableStatus(status), numnodes);
386        return 0;
387    }
388
389    rc = dht_search( dht_socket, tor->info.hash,
390                     announce ? tr_sessionGetPeerPort(session) : 0,
391                     callback, NULL);
392
393    if( rc >= 1 ) {
394        tr_torinf(tor, "Starting DHT announce (%s, %d nodes)",
395                  tr_dhtPrintableStatus(status), numnodes);
396        tor->dhtAnnounceInProgress = TRUE;
397    } else {
398        tr_torerr(tor, "DHT announce failed, errno = %d (%s, %d nodes)",
399                  errno, tr_dhtPrintableStatus(status), numnodes);
400    }
401
402    return 1;
403}
404
405static void
406event_callback(int s, short type, void *ignore UNUSED )
407{
408    time_t tosleep;
409
410    if( dht_periodic(s, type == EV_READ, &tosleep, callback, NULL) < 0 ) {
411        if(errno == EINTR) {
412            tosleep = 0;
413        } else {
414            tr_nerr("DHT", "dht_periodic failed (errno = %d)", errno);
415            if(errno == EINVAL || errno == EFAULT)
416                    abort();
417            tosleep = 1;
418        }
419    }
420
421    /* Being slightly late is fine,
422       and has the added benefit of adding some jitter. */
423    tr_timerAdd( &dht_event, tosleep, tr_cryptoWeakRandInt( 1000000 ) );
424}
425
426void
427dht_hash(void *hash_return, int hash_size,
428         const void *v1, int len1,
429         const void *v2, int len2,
430         const void *v3, int len3)
431{
432    unsigned char sha1[SHA_DIGEST_LENGTH];
433    tr_sha1( sha1, v1, len1, v2, len2, v3, len3, NULL );
434    memset( hash_return, 0, hash_size );
435    memcpy( hash_return, sha1, MIN( hash_size, SHA_DIGEST_LENGTH ) );
436}
437
438int
439dht_random_bytes( void * buf, size_t size )
440{
441    tr_cryptoRandBuf( buf, size );
442    return size;
443}
Note: See TracBrowser for help on using the repository browser.