apt_dht/apt_dht.py

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