instrumentation/next-share/BaseLib/Test/test_proxyservice_as_coord.py

   1 # Written by Arno Bakker, George Milescu
   2 # see LICENSE.txt for license information
   3 #
   4 # Like test_secure_overlay, we start a new python interpreter for each test.
   5 # Although we don't have the singleton problem here, we do need to do this as the
   6 # HTTPServer that MyTracker uses won't relinquish the listen socket, causing
   7 # "address in use" errors in the next test. This is probably due to the fact that
   8 # MyTracker has a thread mixed in, as a listensocket.close() normally releases it
   9 # (according to lsof).
  10 #
  11
  12 import unittest
  13 import os
  14 import sys
  15 import time
  16 import math
  17 from types import ListType
  18 import socket
  19 import hashlib
  20 import tempfile
  21 import string
  22 import random
  23
  24 from BaseLib.Test.test_as_server import TestAsServer
  25 from btconn import BTConnection
  26 from olconn import OLConnection
  27 from BaseLib.Core.RequestPolicy import AllowAllRequestPolicy
  28 from BaseLib.Core.BitTornado.bencode import bencode,bdecode
  29 from BaseLib.Core.BitTornado.bitfield import Bitfield
  30 from BaseLib.Core.BitTornado.BT1.MessageID import *
  31 from BaseLib.Core.BitTornado.BT1.convert import toint
  32 from BaseLib.Core.CacheDB.CacheDBHandler import FriendDBHandler, TorrentDBHandler
  33 from BaseLib.Test.test_connect_overlay import MyTracker
  34
  35 DEBUG=False
  36
  37 class TestProxyServiceAsCoordinator(TestAsServer):
  38     """ This class tests the ProxyService Helper stack. It simulates a coordinator and connects to the
  39     helper instance, sending messages to it and verifying the received responses
  40     """
  41
  42     def setUp(self):
  43         """ override TestAsServer """
  44         TestAsServer.setUp(self)
  45         print >>sys.stderr,"test: Giving MyLaunchMany time to startup"
  46         time.sleep(5)
  47         print >>sys.stderr,"test: MyLaunchMany should have started up"
  48
  49     def setUpPreSession(self):
  50         """ override TestAsServer """
  51         TestAsServer.setUpPreSession(self)
  52
  53         self.setUpMyListenSockets()
  54
  55         # Must be changed in test/extend_hs_dir/proxyservice.test.torrent as well
  56         self.mytrackerport = 4901
  57         self.myid = 'R410-----HgUyPu56789'
  58         self.mytracker = MyTracker(self.mytrackerport,self.myid,'127.0.0.1',self.mylistenport)
  59         self.mytracker.background_serve()
  60
  61         self.myid2 = 'R410-----56789HuGyx0' # used for the coordinator
  62
  63         # Arno, 2009-12-15: Make sure coop downloads have their own destdir
  64         destdir = tempfile.mkdtemp()
  65         self.config.set_download_help_dir(destdir)
  66
  67         # Set the proxyservice to full speed
  68         self.config.set_proxyservice_status(1) #PROXYSERVICE_ON=1
  69
  70     def setUpMyListenSockets(self):
  71         # Start our server side, to which Tribler will try to connect
  72         # coordinator BitTorrent socket (the helper connects to this socket to sent BT messages with pieces requested by the coordinator)
  73         self.mylistenport = 4810
  74         self.myss = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  75         self.myss.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  76         self.myss.bind(('', self.mylistenport))
  77         self.myss.listen(1)
  78
  79         # Leecher socket (the helper connects to this socket to download the pieces requested by the coordinator)
  80         self.mylistenport2 = 3726
  81         self.myss2 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  82         self.myss2.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  83         self.myss2.bind(('', self.mylistenport2))
  84         self.myss2.listen(1)
  85
  86     def setUpPostSession(self):
  87         """ override TestAsServer """
  88         TestAsServer.setUpPostSession(self)
  89
  90         self.mypermid = str(self.my_keypair.pub().get_der())
  91         self.hispermid = str(self.his_keypair.pub().get_der())
  92
  93         # Calculating the infohash for proxyservice.test.torrent
  94         self.torrentfile = os.path.join('extend_hs_dir','proxyservice.test.torrent')
  95
  96         # Read torrent file to calculate the infohash
  97         torrentfile_content = open(self.torrentfile, "rb")
  98         # Decode all the file
  99         metainfo = bdecode(torrentfile_content.read())
 100         # Calculate the torrent length
 101         if "length" in metainfo["info"]:
 102             self.length = metainfo["info"]["length"]
 103         else:
 104             self.length = 0
 105             for f in metainfo["info"]["files"]:
 106                 self.length += f["length"]
 107         # Re-encode only the info section
 108         self.infohash = hashlib.sha1(bencode(metainfo['info'])).digest()
 109         # Store the number of pieces
 110         self.numpieces = int(math.ceil((self.length / metainfo["info"]["piece length"])))
 111         # Close the torrentfile
 112         torrentfile_content.close()
 113
 114         # Add us as friend, so he will accept the ASK_FOR_HELP message
 115         if False:  # TEMP: friendsdb doesn't have an addFriend method
 116 #            friendsdb = FriendDBHandler.getInstance()
 117 #            friendsdb.addFriend(self.mypermid)
 118             pass
 119         else:
 120             # Accept overlay requests from anybody
 121             self.session.set_overlay_request_policy(AllowAllRequestPolicy())
 122
 123         self.session.set_download_states_callback(self.states_callback)
 124         """
 125         statedir=self.session.get_state_dir()
 126         os.system('cp /tmp/Gopher.torrent ' + statedir + '/collected_torrent_files/Gopher.torrent')
 127         """
 128
 129     def tearDown(self):
 130         """ override TestAsServer """
 131         print >> sys.stderr,"test: *** TEARDOWN"
 132         TestAsServer.tearDown(self)
 133         self.mytracker.shutdown()
 134         self.tearDownMyListenSockets()
 135
 136
 137     def tearDownMyListenSockets(self):
 138         self.myss.close()
 139         self.myss2.close()
 140
 141
 142     def states_callback(self,dslist):
 143         print >>sys.stderr,"stats: dslist",len(dslist)
 144         for ds in dslist:
 145             print >>sys.stderr,"stats: coordinator",`ds.get_coopdl_coordinator()`
 146             print >>sys.stderr,"stats: helpers",`ds.get_coopdl_helpers()`
 147         print >>sys.stderr, ""
 148         return (0.5,False)
 149
 150     # Creates dictionary with the correct (good) commands used by the coordinator to test the helper
 151     def get_genresdict(self):
 152         genresdict = {}
 153         genresdict[ASK_FOR_HELP] = (self.create_good_ask_for_help,True)
 154         genresdict[STOP_HELPING] = (self.create_good_stop_helping,True)
 155         genresdict[REQUEST_PIECES] = (self.create_good_request_pieces,True)
 156         genresdict[CANCEL_PIECE] = (self.create_good_cancel_piece,True)
 157         # The helper will request the .torrent metadata
 158         genresdict[METADATA] = (self.create_good_metadata,True)
 159
 160         return genresdict
 161
 162     #
 163     # Good proxy messages
 164     #
 165     def singtest_good_proxy(self):
 166         genresdict = self.get_genresdict()
 167         print >> sys.stderr, "test: good ASK_FOR_HELP"
 168         self._test_proxy(genresdict)
 169
 170
 171     #
 172     # Bad proxy messages
 173     #
 174     def singtest_bad_proxy_ask_for_help(self):
 175         # Get the correct messages used by the coordinator
 176         genresdict = self.get_genresdict()
 177         # Prepare a bad ASK_FOR_HELP message
 178         genresdict[ASK_FOR_HELP] = (self.create_bad_ask_for_help_no_infohash,False)
 179         print >> sys.stderr, "test: bad ask_for_help"
 180         self._test_proxy(genresdict)
 181
 182     def singtest_bad_proxy_metadata_not_bdecodable(self):
 183         # Get the correct messages used by the coordinator
 184         genresdict = self.get_genresdict()
 185         # Prepare a bad METADATA message
 186         genresdict[METADATA] = (self.create_bad_metadata_not_bdecodable,False)
 187         print >> sys.stderr, "test: bad METADATA (not bdecodable)", genresdict[METADATA][0]
 188         self._test_proxy(genresdict)
 189
 190     def singtest_bad_proxy_metadata_not_dict1(self):
 191         # Get the correct messages used by the coordinator
 192         genresdict = self.get_genresdict()
 193         # Prepare a bad METADATA message
 194         genresdict[METADATA] = (self.create_bad_metadata_not_dict1,False)
 195         print >> sys.stderr, "test: bad METADATA (not a dictionary)", genresdict[METADATA][0]
 196         self._test_proxy(genresdict)
 197
 198     def singtest_bad_proxy_metadata_not_dict2(self):
 199         # Get the correct messages used by the coordinator
 200         genresdict = self.get_genresdict()
 201         # Prepare a bad METADATA message
 202         genresdict[METADATA] = (self.create_bad_metadata_not_dict2,False)
 203         print >>sys.stderr,"test: bad METADATA (not a dictionary)",genresdict[METADATA][0]
 204         self._test_proxy(genresdict)
 205
 206     def singtest_bad_2fast_metadata_empty_dict(self):
 207         # Get the correct messages used by the coordinator
 208         genresdict = self.get_genresdict()
 209         # Prepare a bad METADATA message
 210         genresdict[METADATA] = (self.create_bad_metadata_empty_dict,False)
 211         print >>sys.stderr,"test: bad METADATA (empty dictionary)",genresdict[METADATA][0]
 212         self._test_proxy(genresdict)
 213
 214     def singtest_bad_proxy_metadata_wrong_dict_keys(self):
 215         # Get the correct messages used by the coordinator
 216         genresdict = self.get_genresdict()
 217         # Prepare a bad METADATA message
 218         genresdict[METADATA] = (self.create_bad_metadata_wrong_dict_keys,False)
 219         print >>sys.stderr,"test: bad METADATA (wrong keys in dictionary)",genresdict[METADATA][0]
 220         self._test_proxy(genresdict)
 221
 222     def singtest_bad_proxy_metadata_bad_torrent1(self):
 223         # Get the correct messages used by the coordinator
 224         genresdict = self.get_genresdict()
 225         # Prepare a bad METADATA message
 226         genresdict[METADATA] = (self.create_bad_metadata_bad_torrent1,False)
 227         print >>sys.stderr,"test: bad METADATA (wrong metadata field in dictionary)",genresdict[METADATA][0]
 228         self._test_proxy(genresdict)
 229
 230
 231     def singtest_bad_proxy_metadata_bad_torrent2(self):
 232         # Get the correct messages used by the coordinator
 233         genresdict = self.get_genresdict()
 234         # Prepare a bad METADATA message
 235         genresdict[METADATA] = (self.create_bad_metadata_bad_torrent2,False)
 236         print >>sys.stderr,"test: bad METADATA (empty dictionary in metadata filed)",genresdict[METADATA][0]
 237         self._test_proxy(genresdict)
 238
 239     def singtest_bad_proxy_metadata_bad_torrent3(self):
 240         # Get the correct messages used by the coordinator
 241         genresdict = self.get_genresdict()
 242         # Prepare a bad METADATA message
 243         genresdict[METADATA] = (self.create_bad_metadata_bad_torrent3,False)
 244         print >>sys.stderr,"test: bad METADATA (bad metadata field in dictionary)",genresdict[METADATA][0]
 245         self._test_proxy(genresdict)
 246
 247
 248
 249     def _test_proxy(self,genresdict):
 250         """ Send messages to the helper instance and test it.
 251
 252             Testing ASK_FOR_HELP, STOP_HELPING, REQUEST_PIECES, CANCEL_PIECE and METADATA
 253         """
 254         # 1. Establish overlay connection to Tribler
 255         ol_connection = OLConnection(self.my_keypair, 'localhost', self.hisport, mylistenport=self.mylistenport2)
 256
 257         # 2. Send the ASK_FOR_HELP message
 258         (generate_data,sent_good_values) = genresdict[ASK_FOR_HELP]
 259         msg = generate_data()
 260         ol_connection.send(msg)
 261         if sent_good_values:
 262             # Read the helper's response
 263             resp = ol_connection.recv()
 264             # Check the helper's response
 265             # 3. At this point, the helper does not have the .torrent file, so it requests it with a METADATA message
 266             self.assert_(resp[0] == GET_METADATA)
 267             self.check_get_metadata(resp[1:])
 268             print >>sys.stderr,"test: Got GET_METADATA for torrent, good"
 269         else:
 270             # Read the helper's response
 271             resp = ol_connection.recv()
 272             # Check the helper's response
 273             self.assert_(len(resp)==0)
 274             ol_connection.close()
 275             return
 276
 277         # 4. Send METADATA
 278         (generate_data,sent_good_values) = genresdict[METADATA]
 279         msg = generate_data()
 280         ol_connection.send(msg)
 281         if sent_good_values:
 282             # 5. At this point the helper is confirming his availability to help
 283             # Read the helper's response
 284             resp = ol_connection.recv()
 285             # Check the helper's response
 286             self.assert_(resp[0] == JOIN_HELPERS)
 287             self.check_ask_for_help(resp)
 288             print >>sys.stderr,"test: Got JOIN_HELPERS for torrent, good"
 289
 290             # 6. At this point, the helper will contact the tracker and then wait for REQUEST_PIECES messages
 291             # So we send a request pieces message
 292             (generate_data,sent_good_values) = genresdict[REQUEST_PIECES]
 293             msg = generate_data()
 294             ol_connection.send(msg)
 295
 296             # At this point the helper will contact the seeders in the swarm to download the requested piece
 297             # There is only one seeder in the swarm, the coordinator's twin
 298             # 8. Our tracker says there is another peer (also us) on port 4810
 299             # Now accept a connection on that port and pretend we're a seeder
 300             self.myss.settimeout(10.0)
 301             conn, addr = self.myss.accept()
 302             options = '\x00\x00\x00\x00\x00\x00\x00\x00'
 303             s2 = BTConnection('',0,conn,user_option_pattern=options,user_infohash=self.infohash,myid=self.myid)
 304             s2.read_handshake_medium_rare()
 305
 306             # Send a bitfield message to the helper (pretending we are a regular seeder)
 307             b = Bitfield(self.numpieces)
 308             for i in range(self.numpieces):
 309                 b[i] = True
 310             self.assert_(b.complete())
 311             msg = BITFIELD+b.tostring()
 312             s2.send(msg)
 313             msg = UNCHOKE
 314             s2.send(msg)
 315             print >>sys.stderr,"test: Got BT connection to us, as fake seeder, good"
 316         else:
 317             resp = ol_connection.recv()
 318             self.assert_(len(resp)==0)
 319             ol_connection.close()
 320             return
 321
 322         # 7. Accept the data connection the helper wants to establish with us, the coordinator.
 323         # The helper will send via this connection the pieces we request it to download.
 324         self.myss2.settimeout(10.0)
 325         conn, addr = self.myss2.accept()
 326         s3 = BTConnection('',0,conn,user_infohash=self.infohash,myid=self.myid2)
 327         s3.read_handshake_medium_rare()
 328
 329         msg = UNCHOKE
 330         s3.send(msg)
 331         print >>sys.stderr,"test: Got data connection to us, as coordinator, good"
 332
 333         # 9. At this point the helper should sent a PROXY_HAVE message on the overlay connection
 334 #        resp = ol_connection.recv()
 335 #        self.assert_(resp[0] == PROXY_HAVE)
 336 #        print >>sys.stderr,"test: Got PROXY)HAVE, good"
 337
 338         # 10. Await REQUEST on fake seeder
 339         try:
 340             while True:
 341                 s2.s.settimeout(10.0)
 342                 resp = s2.recv()
 343                 self.assert_(len(resp) > 0)
 344                 print "test: Fake seeder got message",getMessageName(resp[0])
 345                 if resp[0] == REQUEST:
 346                     self.check_request(resp[1:])
 347                     print >>sys.stderr,"test: Fake seeder got REQUEST for reserved piece, good"
 348                     break
 349
 350         except socket.timeout:
 351             print >> sys.stderr,"test: Timeout, bad, fake seeder didn't reply with message"
 352             self.assert_(False)
 353
 354         # 11. Sent the helper a STOP_HELPING message
 355         (generate_data,sent_good_values) = genresdict[STOP_HELPING]
 356         msg = generate_data()
 357         ol_connection.send(msg)
 358         # The other side should close the connection, whether the msg was good or bad
 359         resp = ol_connection.recv()
 360         self.assert_(len(resp)==0)
 361         ol_connection.close()
 362
 363     def check_request(self,data):
 364         piece = toint(data[0:4])
 365         self.assert_(piece == 1)
 366
 367     #
 368     # Correct (good) messages used by the coordinator to test the helper
 369     #
 370     def create_good_ask_for_help(self):
 371         """ Create a correctly formatted ASK_FOR_HELP message and return it
 372         """
 373         # Generate a random challenge - random number on 8 bytes (62**8 possible combinations)
 374         chars = string.letters + string.digits #len(chars)=62
 375         challenge = ''
 376         for i in range(8):
 377             challenge = challenge + random.choice(chars)
 378
 379         return ASK_FOR_HELP + self.infohash + bencode(challenge)
 380
 381     def check_ask_for_help(self, data):
 382         """ Check the answer the coordinator got for an ASK_FOR_HELP message
 383         The helper should have sent a JOIN_HELPERS message
 384         """
 385         infohash = data[1:21]
 386         self.assert_(infohash == self.infohash)
 387
 388     #----------
 389
 390     def create_good_stop_helping(self):
 391         return STOP_HELPING + self.infohash
 392
 393     def check_stop_helping(self, data):
 394         """ Check the answer the coordinator got for a STOP_HELPING message
 395         The helper should have sent a RESIGN_AS_HELPER message
 396         """
 397         infohash = data[1:21]
 398         self.assert_(infohash == self.infohash)
 399
 400     #----------
 401
 402     def create_good_request_pieces(self):
 403         # Request piece number 1
 404         piece = 1
 405         return REQUEST_PIECES + self.infohash + bencode(piece)
 406     # The reply for this message is a BT Have message
 407
 408     #----------
 409
 410     def create_good_cancel_piece(self):
 411         # Cancel piece number 1
 412         piece = 1
 413         return CANCEL_PIECE + self.infohash + bencode(piece)
 414     # This message is not supposed to have any reply
 415     # TODO: test the DROPEPD_PIECE message, after implementation
 416
 417     #----------
 418
 419     def create_good_metadata(self):
 420         f = open(self.torrentfile,"rb")
 421         data = f.read()
 422         f.close()
 423
 424         d = self.create_good_metadata_dict(data)
 425         bd = bencode(d)
 426         return METADATA + bd
 427
 428     def create_good_metadata_dict(self,data):
 429         d = {}
 430         d['torrent_hash'] = self.infohash
 431         d['metadata'] = data
 432         d['leecher'] = 1
 433         d['seeder'] = 1
 434         d['last_check_time'] = int(time.time())
 435         d['status'] = 'good'
 436         return d
 437
 438     def check_get_metadata(self,data):
 439         infohash = bdecode(data) # is bencoded for unknown reason, can't change it =))
 440         self.assert_(infohash == self.infohash)
 441
 442     #----------
 443
 444     #
 445     # Incorrect (bad) messages used by the coordinator to test the helper
 446     #
 447     def create_bad_ask_for_help_no_infohash(self):
 448         return ASK_FOR_HELP+"481"
 449
 450     def create_bad_metadata_not_bdecodable(self):
 451         return METADATA+"bla"
 452
 453     def create_bad_metadata_not_dict1(self):
 454         d  = 481
 455         return METADATA+bencode(d)
 456
 457     def create_bad_metadata_not_dict2(self):
 458         d  = []
 459         return METADATA+bencode(d)
 460
 461     def create_bad_metadata_empty_dict(self):
 462         d = {}
 463         return METADATA+bencode(d)
 464
 465     def create_bad_metadata_wrong_dict_keys(self):
 466         d = {}
 467         d['bla1'] = '\x00\x00\x00\x00\x00\x30\x00\x00'
 468         d['bla2'] = '\x00\x00\x00\x00\x00\x30\x00\x00'
 469         return METADATA+bencode(d)
 470
 471     def create_bad_metadata_bad_torrent1(self):
 472         d = self.create_good_metadata_dict(None)
 473         d['metadata'] = '\x12\x34' * 100 # random data
 474         bd = bencode(d)
 475         return METADATA+bd
 476
 477     def create_bad_metadata_bad_torrent2(self):
 478         torrent = {}
 479         data = bencode(torrent)
 480
 481         d = self.create_good_metadata_dict(data)
 482         d['metadata'] = data
 483         bd = bencode(d)
 484         return METADATA+bd
 485
 486
 487     def create_bad_metadata_bad_torrent3(self):
 488         torrent = {'info':481}
 489         data = bencode(torrent)
 490
 491         d = self.create_good_metadata_dict(data)
 492         d['metadata'] = data
 493         bd = bencode(d)
 494         return METADATA+bd
 495
 496
 497
 498 def test_suite():
 499     suite = unittest.TestSuite()
 500     # We should run the tests in a separate Python interpreter to prevent
 501     # problems with our singleton classes, e.g. PeerDB, etc.
 502     if len(sys.argv) != 2:
 503         print "Usage: python test_proxyservice_as_coord.py <method name>"
 504     else:
 505         suite.addTest(TestProxyServiceAsCoordinator(sys.argv[1]))
 506         # DEBUG
 507         print "***"
 508         print "*** Calling TestProxyServiceAsCoordinator with argument " + sys.argv[1]
 509         print "***"
 510
 511     return suite
 512
 513 def main():
 514     unittest.main(defaultTest='test_suite',argv=[sys.argv[0]])
 515
 516 if __name__ == "__main__":
 517     main()