apt_dht/apt_dht.py

   1
   2 from binascii import b2a_hex
   3 from urlparse import urlunparse
   4 import os, re
   5
   6 from twisted.internet import defer
   7 from twisted.web2 import server, http, http_headers
   8 from twisted.python import log
   9
  10 from apt_dht_conf import config
  11 from PeerManager import PeerManager
  12 from HTTPServer import TopLevel
  13 from MirrorManager import MirrorManager
  14 from Hash import HashObject
  15
  16 class AptDHT:
  17     def __init__(self, dht):
  18         log.msg('Initializing the main apt_dht application')
  19         self.dht = dht
  20         self.dht.loadConfig(config, config.get('DEFAULT', 'DHT'))
  21         self.dht.join().addCallbacks(self.joinComplete, self.joinError)
  22         self.http_server = TopLevel(config.get('DEFAULT', 'cache_dir'), self)
  23         self.http_site = server.Site(self.http_server)
  24         self.peers = PeerManager()
  25         self.mirrors = MirrorManager(config.get('DEFAULT', 'cache_dir'), self)
  26         self.my_addr = None
  27         self.isLocal = re.compile('^(192\.168\.[0-9]{1,3}\.[0-9]{1,3})|'+
  28                                   '(10\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})|'+
  29                                   '(172\.0?([1][6-9])|([2][0-9])|([3][0-1])\.[0-9]{1,3}\.[0-9]{1,3})|'+
  30                                   '(127\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})$')
  31
  32     def getSite(self):
  33         return self.http_site
  34
  35     def joinComplete(self, addrs):
  36         log.msg("got addrs: %r" % (addrs,))
  37
  38         try:
  39             ifconfig = os.popen("/sbin/ifconfig |/bin/grep inet|"+
  40                                 "/usr/bin/awk '{print $2}' | "+
  41                                 "sed -e s/.*://", "r").read().strip().split('\n')
  42         except:
  43             ifconfig = []
  44
  45         # Get counts for all the non-local addresses returned
  46         addr_count = {}
  47         for addr in ifconfig:
  48             if not self.isLocal.match(addr):
  49                 addr_count.setdefault(addr, 0)
  50                 addr_count[addr] += 1
  51
  52         local_addrs = addr_count.keys()
  53         if len(local_addrs) == 1:
  54             self.my_addr = local_addrs[0]
  55             log.msg('Found remote address from ifconfig: %r' % (self.my_addr,))
  56
  57         # Get counts for all the non-local addresses returned
  58         addr_count = {}
  59         port_count = {}
  60         for addr in addrs:
  61             if not self.isLocal.match(addr[0]):
  62                 addr_count.setdefault(addr[0], 0)
  63                 addr_count[addr[0]] += 1
  64                 port_count.setdefault(addr[1], 0)
  65                 port_count[addr[1]] += 1
  66
  67         # Find the most popular address
  68         popular_addr = []
  69         popular_count = 0
  70         for addr in addr_count:
  71             if addr_count[addr] > popular_count:
  72                 popular_addr = [addr]
  73                 popular_count = addr_count[addr]
  74             elif addr_count[addr] == popular_count:
  75                 popular_addr.append(addr)
  76
  77         # Find the most popular port
  78         popular_port = []
  79         popular_count = 0
  80         for port in port_count:
  81             if port_count[port] > popular_count:
  82                 popular_port = [port]
  83                 popular_count = port_count[port]
  84             elif port_count[port] == popular_count:
  85                 popular_port.append(port)
  86
  87         port = config.getint(config.get('DEFAULT', 'DHT'), 'PORT')
  88         if len(port_count.keys()) > 1:
  89             log.msg('Problem, multiple ports have been found: %r' % (port_count,))
  90             if port not in port_count.keys():
  91                 log.msg('And none of the ports found match the intended one')
  92         elif len(port_count.keys()) == 1:
  93             port = port_count.keys()[0]
  94         else:
  95             log.msg('Port was not found')
  96
  97         if len(popular_addr) == 1:
  98             log.msg('Found popular address: %r' % (popular_addr[0],))
  99             if self.my_addr and self.my_addr != popular_addr[0]:
 100                 log.msg('But the popular address does not match: %s != %s' % (popular_addr[0], self.my_addr))
 101             self.my_addr = popular_addr[0]
 102         elif len(popular_addr) > 1:
 103             log.msg('Found multiple popular addresses: %r' % (popular_addr,))
 104             if self.my_addr and self.my_addr not in popular_addr:
 105                 log.msg('And none of the addresses found match the ifconfig one')
 106         else:
 107             log.msg('No non-local addresses found: %r' % (popular_addr,))
 108
 109         if not self.my_addr:
 110             log.err(RuntimeError("Remote IP Address could not be found for this machine"))
 111
 112     def ipAddrFromChicken(self):
 113         import urllib
 114         ip_search = re.compile('\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}')
 115         try:
 116              f = urllib.urlopen("http://www.ipchicken.com")
 117              data = f.read()
 118              f.close()
 119              current_ip = ip_search.findall(data)
 120              return current_ip
 121         except Exception:
 122              return []
 123
 124     def joinError(self, failure):
 125         log.msg("joining DHT failed miserably")
 126         log.err(failure)
 127
 128     def check_freshness(self, path, modtime, resp):
 129         log.msg('Checking if %s is still fresh' % path)
 130         d = self.peers.get([path], "HEAD", modtime)
 131         d.addCallback(self.check_freshness_done, path, resp)
 132         return d
 133
 134     def check_freshness_done(self, resp, path, orig_resp):
 135         if resp.code == 304:
 136             log.msg('Still fresh, returning: %s' % path)
 137             return orig_resp
 138         else:
 139             log.msg('Stale, need to redownload: %s' % path)
 140             return self.get_resp(path)
 141
 142     def get_resp(self, path):
 143         d = defer.Deferred()
 144
 145         log.msg('Trying to find hash for %s' % path)
 146         findDefer = self.mirrors.findHash(path)
 147
 148         findDefer.addCallbacks(self.findHash_done, self.findHash_error,
 149                                callbackArgs=(path, d), errbackArgs=(path, d))
 150         findDefer.addErrback(log.err)
 151         return d
 152
 153     def findHash_error(self, failure, path, d):
 154         log.err(failure)
 155         self.findHash_done(HashObject(), path, d)
 156
 157     def findHash_done(self, hash, path, d):
 158         if hash.expected() is None:
 159             log.msg('Hash for %s was not found' % path)
 160             self.download_file([path], hash, path, d)
 161         else:
 162             log.msg('Found hash %s for %s' % (hash.hexexpected(), path))
 163             # Lookup hash from DHT
 164             key = hash.normexpected(bits = config.getint(config.get('DEFAULT', 'DHT'), 'HASH_LENGTH'))
 165             lookupDefer = self.dht.getValue(key)
 166             lookupDefer.addCallback(self.lookupHash_done, hash, path, d)
 167
 168     def lookupHash_done(self, locations, hash, path, d):
 169         if not locations:
 170             log.msg('Peers for %s were not found' % path)
 171             self.download_file([path], hash, path, d)
 172         else:
 173             log.msg('Found peers for %s: %r' % (path, locations))
 174             # Download from the found peers
 175             self.download_file(locations, hash, path, d)
 176
 177     def download_file(self, locations, hash, path, d):
 178         getDefer = self.peers.get(locations)
 179         getDefer.addCallback(self.mirrors.save_file, hash, path)
 180         getDefer.addErrback(self.mirrors.save_error, path)
 181         getDefer.addCallbacks(d.callback, d.errback)
 182
 183     def download_complete(self, hash, url, file_path):
 184         assert file_path.startswith(config.get('DEFAULT', 'cache_dir'))
 185         directory = file_path[:len(config.get('DEFAULT', 'cache_dir'))]
 186         url_path = file_path[len(config.get('DEFAULT', 'cache_dir')):]
 187         if url_path[0] == '/':
 188             url_path = url_path[1:]
 189         top_directory = url_path.split('/',1)[0]
 190         url_path = url_path[len(top_directory):]
 191         http_dir = os.path.join(directory, top_directory)
 192         new_top = self.http_server.addDirectory(http_dir)
 193         url_path = '/' + new_top + url_path
 194         log.msg('now avaliable at %s: %s' % (url_path, url))
 195
 196         if self.my_addr:
 197             site = self.my_addr + ':' + str(config.getint('DEFAULT', 'PORT'))
 198             full_path = urlunparse(('http', site, url_path, None, None, None))
 199             key = hash.norm(bits = config.getint(config.get('DEFAULT', 'DHT'), 'HASH_LENGTH'))
 200             storeDefer = self.dht.storeValue(key, full_path)
 201             storeDefer.addCallback(self.store_done, full_path)
 202             storeDefer.addErrback(log.err)
 203
 204     def store_done(self, result, path):
 205         log.msg('Added %s to the DHT: %r' % (path, result))
 206