import random import time import hashlib import urllib import urllib2 import struct import socket from lib import bencode from lib.subtl.subtl import UdpTrackerClient from lib.PySocks import socks from lib.PySocks import sockshandler import gevent from Plugin import PluginManager from Config import config import util from Debug import Debug class AnnounceError(Exception): pass @PluginManager.acceptPlugins class SiteAnnouncer(object): def __init__(self, site): self.site = site self.stats = {} self.fileserver_port = config.fileserver_port self.peer_id = self.site.connection_server.peer_id self.last_tracker_id = random.randint(0, 10) self.time_last_announce = 0 def getSupportedTrackers(self): trackers = config.trackers if config.disable_udp or config.trackers_proxy != "disable": trackers = [tracker for tracker in trackers if not tracker.startswith("udp://")] if not self.site.connection_server.tor_manager.enabled: trackers = [tracker for tracker in trackers if ".onion" not in tracker] return trackers def getAnnouncingTrackers(self, mode): trackers = self.getSupportedTrackers() if trackers and (mode == "update" or mode == "more"): # Only announce on one tracker, increment the queried tracker id self.last_tracker_id += 1 self.last_tracker_id = self.last_tracker_id % len(trackers) trackers_announcing = [trackers[self.last_tracker_id]] # We only going to use this one else: trackers_announcing = trackers return trackers_announcing def getOpenedServiceTypes(self): back = [] # Type of addresses they can reach me if self.site.connection_server.port_opened and config.trackers_proxy == "disable": back.append("ip4") if self.site.connection_server.tor_manager.start_onions: back.append("onion") return back @util.Noparallel(blocking=False) def announce(self, force=False, mode="start", pex=True): if time.time() < self.time_last_announce + 30 and not force: return # No reannouncing within 30 secs if force: self.site.log.debug("Force reannounce in mode %s" % mode) self.fileserver_port = config.fileserver_port self.time_last_announce = time.time() trackers = self.getAnnouncingTrackers(mode) if config.verbose: self.site.log.debug("Tracker announcing, trackers: %s" % trackers) errors = [] slow = [] s = time.time() threads = [] num_announced = 0 for tracker in trackers: # Start announce threads thread = gevent.spawn(self.announceTracker, tracker, mode=mode) threads.append(thread) thread.tracker = tracker time.sleep(0.01) self.updateWebsocket(trackers="announcing") gevent.joinall(threads, timeout=20) # Wait for announce finish for thread in threads: if thread.value is not False: if thread.value > 1.0: # Takes more than 1 second to announce slow.append("%.2fs %s" % (thread.value, thread.tracker)) num_announced += 1 else: if thread.ready(): errors.append(thread.tracker) else: # Still running slow.append("30s+ %s" % thread.tracker) # Save peers num self.site.settings["peers"] = len(self.site.peers) if len(errors) < len(threads): # At least one tracker finished if len(trackers) == 1: announced_to = trackers[0] else: announced_to = "%s/%s trackers" % (num_announced, len(threads)) if mode != "update" or config.verbose: self.site.log.debug( "Announced in mode %s to %s in %.3fs, errors: %s, slow: %s" % (mode, announced_to, time.time() - s, errors, slow) ) else: if len(threads) > 1: self.site.log.error("Announce to %s trackers in %.3fs, failed" % (num_announced, time.time() - s)) self.updateWebsocket(trackers="announced") if pex: self.updateWebsocket(pex="announcing") if mode == "more": # Need more peers self.announcePex(need_num=10) else: self.announcePex() self.updateWebsocket(pex="announced") def getTrackerHandler(self, protocol): if protocol == "udp": handler = self.announceTrackerUdp elif protocol == "http": handler = self.announceTrackerHttp else: handler = None return handler def announceTracker(self, tracker, mode="start", num_want=10): s = time.time() if "://" not in tracker: self.site.log.warning("Tracker %s error: Invalid address" % tracker) return False protocol, address = tracker.split("://", 1) if tracker not in self.stats: self.stats[tracker] = {"status": "", "num_request": 0, "num_success": 0, "num_error": 0, "time_request": 0} self.stats[tracker]["status"] = "announcing" self.stats[tracker]["time_request"] = time.time() self.stats[tracker]["num_request"] += 1 if config.verbose: self.site.log.debug("Tracker announcing to %s (mode: %s)" % (tracker, mode)) if mode == "update": num_want = 10 else: num_want = 30 handler = self.getTrackerHandler(protocol) error = None try: if handler: peers = handler(address, mode=mode, num_want=num_want) else: raise AnnounceError("Unknown protocol: %s" % protocol) except Exception, err: self.site.log.warning("Tracker %s announce failed: %s" % (tracker, str(err).decode("utf8", "ignore"))) error = err if error: self.stats[tracker]["status"] = "error" self.stats[tracker]["time_status"] = time.time() self.stats[tracker]["last_error"] = str(err).decode("utf8", "ignore") self.stats[tracker]["num_error"] += 1 self.updateWebsocket(tracker="error") return False self.stats[tracker]["status"] = "announced" self.stats[tracker]["time_status"] = time.time() self.stats[tracker]["num_success"] += 1 if peers is None: # No peers returned return time.time() - s # Adding peers added = 0 for peer in peers: if peer["port"] == 1: # Some trackers does not accept port 0, so we send port 1 as not-connectable peer["port"] = 0 if not peer["port"]: continue # Dont add peers with port 0 if self.site.addPeer(peer["addr"], peer["port"], source="tracker"): added += 1 if added: self.site.worker_manager.onPeers() self.site.updateWebsocket(peers_added=added) if config.verbose: self.site.log.debug( "Tracker result: %s://%s (found %s peers, new: %s, total: %s)" % (protocol, address, len(peers), added, len(self.site.peers)) ) return time.time() - s def announceTrackerUdp(self, tracker_address, mode="start", num_want=10): s = time.time() if config.disable_udp: raise AnnounceError("Udp disabled by config") if config.trackers_proxy != "disable": raise AnnounceError("Udp trackers not available with proxies") ip, port = tracker_address.split(":") tracker = UdpTrackerClient(ip, int(port)) if "ip4" in self.getOpenedServiceTypes(): tracker.peer_port = self.fileserver_port else: tracker.peer_port = 0 tracker.connect() tracker.poll_once() tracker.announce(info_hash=hashlib.sha1(self.site.address).hexdigest(), num_want=num_want, left=431102370) back = tracker.poll_once() if not back: raise AnnounceError("No response after %.0fs" % (time.time() - s)) elif type(back) is dict and "response" in back: peers = back["response"]["peers"] else: raise AnnounceError("Invalid response: %r" % back) return peers def httpRequest(self, url): if config.trackers_proxy == "tor": tor_manager = self.site.connection_server.tor_manager handler = sockshandler.SocksiPyHandler(socks.SOCKS5, tor_manager.proxy_ip, tor_manager.proxy_port) opener = urllib2.build_opener(handler) return opener.open(url, timeout=50) else: return urllib2.urlopen(url, timeout=25) def announceTrackerHttp(self, tracker_address, mode="start", num_want=10): if "ip4" in self.getOpenedServiceTypes(): port = self.fileserver_port else: port = 1 params = { 'info_hash': hashlib.sha1(self.site.address).digest(), 'peer_id': self.peer_id, 'port': port, 'uploaded': 0, 'downloaded': 0, 'left': 431102370, 'compact': 1, 'numwant': num_want, 'event': 'started' } url = "http://" + tracker_address + "?" + urllib.urlencode(params) s = time.time() response = None # Load url if config.tor == "always" or config.trackers_proxy != "disable": timeout = 60 else: timeout = 30 with gevent.Timeout(timeout, False): # Make sure of timeout req = self.httpRequest(url) response = req.read() req.fp._sock.recv = None # Hacky avoidance of memory leak for older python versions req.close() req = None if not response: raise AnnounceError("No response after %.0fs" % (time.time() - s)) # Decode peers try: peer_data = bencode.decode(response)["peers"] response = None peer_count = len(peer_data) / 6 peers = [] for peer_offset in xrange(peer_count): off = 6 * peer_offset peer = peer_data[off:off + 6] addr, port = struct.unpack('!LH', peer) peers.append({"addr": socket.inet_ntoa(struct.pack('!L', addr)), "port": port}) except Exception as err: raise AnnounceError("Invalid response: %r (%s)" % (response, err)) return peers @util.Noparallel(blocking=False) def announcePex(self, query_num=2, need_num=5): peers = self.site.getConnectedPeers() if len(peers) == 0: # Wait 3s for connections time.sleep(3) peers = self.site.getConnectedPeers() if len(peers) == 0: # Small number of connected peers for this site, connect to any peers = self.site.peers.values() need_num = 10 random.shuffle(peers) done = 0 total_added = 0 for peer in peers: num_added = peer.pex(need_num=need_num) if num_added is not False: done += 1 total_added += num_added if num_added: self.site.worker_manager.onPeers() self.site.updateWebsocket(peers_added=num_added) if done == query_num: break self.site.log.debug("Pex result: from %s peers got %s new peers." % (done, total_added)) def updateWebsocket(self, **kwargs): if kwargs: param = {"event": kwargs.items()[0]} else: param = None for ws in self.site.websockets: ws.event("announcerChanged", self.site, param)