apt_p2p/util.py

   1
   2 """Some utitlity functions for use in the apt-p2p program.
   3
   4 @var isLocal: a compiled regular expression suitable for testing if an
   5     IP address is from a known local or private range
   6 """
   7
   8 import os, re
   9
  10 from twisted.python import log
  11 from twisted.trial import unittest
  12
  13 isLocal = re.compile('^(192\.168\.[0-9]{1,3}\.[0-9]{1,3})|'+
  14                      '(10\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})|'+
  15                      '(172\.0?([1][6-9])|([2][0-9])|([3][0-1])\.[0-9]{1,3}\.[0-9]{1,3})|'+
  16                      '(127\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})$')
  17
  18 def findMyIPAddr(addrs, intended_port, local_ok = False):
  19     """Find the best IP address to use from a list of possibilities.
  20
  21     @param addrs: the list of possible IP addresses
  22     @param intended_port: the port that was supposed to be used
  23     @param local_ok: whether known local/private IP ranges are allowed
  24         (defaults to False)
  25     @return: the preferred IP address, or None if one couldn't be found
  26     """
  27     log.msg("got addrs: %r" % (addrs,))
  28     my_addr = None
  29
  30     # Try to find an address using the ifconfig function
  31     try:
  32         ifconfig = os.popen("/sbin/ifconfig |/bin/grep inet|"+
  33                             "/usr/bin/awk '{print $2}' | "+
  34                             "sed -e s/.*://", "r").read().strip().split('\n')
  35     except:
  36         ifconfig = []
  37
  38     # Get counts for all the non-local addresses returned from ifconfig
  39     addr_count = {}
  40     for addr in ifconfig:
  41         if local_ok or not isLocal.match(addr):
  42             addr_count.setdefault(addr, 0)
  43             addr_count[addr] += 1
  44
  45     # If only one was found, use it as a starting point
  46     local_addrs = addr_count.keys()
  47     if len(local_addrs) == 1:
  48         my_addr = local_addrs[0]
  49         log.msg('Found remote address from ifconfig: %r' % (my_addr,))
  50
  51     # Get counts for all the non-local addresses returned from the DHT
  52     addr_count = {}
  53     port_count = {}
  54     for addr in addrs:
  55         if local_ok or not isLocal.match(addr[0]):
  56             addr_count.setdefault(addr[0], 0)
  57             addr_count[addr[0]] += 1
  58             port_count.setdefault(addr[1], 0)
  59             port_count[addr[1]] += 1
  60
  61     # Find the most popular address
  62     popular_addr = []
  63     popular_count = 0
  64     for addr in addr_count:
  65         if addr_count[addr] > popular_count:
  66             popular_addr = [addr]
  67             popular_count = addr_count[addr]
  68         elif addr_count[addr] == popular_count:
  69             popular_addr.append(addr)
  70
  71     # Find the most popular port
  72     popular_port = []
  73     popular_count = 0
  74     for port in port_count:
  75         if port_count[port] > popular_count:
  76             popular_port = [port]
  77             popular_count = port_count[port]
  78         elif port_count[port] == popular_count:
  79             popular_port.append(port)
  80
  81     # Check to make sure the port isn't being changed
  82     port = intended_port
  83     if len(port_count.keys()) > 1:
  84         log.msg('Problem, multiple ports have been found: %r' % (port_count,))
  85         if port not in port_count.keys():
  86             log.msg('And none of the ports found match the intended one')
  87     elif len(port_count.keys()) == 1:
  88         port = port_count.keys()[0]
  89     else:
  90         log.msg('Port was not found')
  91
  92     # If one is popular, use that address
  93     if len(popular_addr) == 1:
  94         log.msg('Found popular address: %r' % (popular_addr[0],))
  95         if my_addr and my_addr != popular_addr[0]:
  96             log.msg('But the popular address does not match: %s != %s' % (popular_addr[0], my_addr))
  97         my_addr = popular_addr[0]
  98     elif len(popular_addr) > 1:
  99         log.msg('Found multiple popular addresses: %r' % (popular_addr,))
 100         if my_addr and my_addr not in popular_addr:
 101             log.msg('And none of the addresses found match the ifconfig one')
 102     else:
 103         log.msg('No non-local addresses found: %r' % (popular_addr,))
 104
 105     if not my_addr:
 106         log.msg("Remote IP Address could not be found for this machine")
 107
 108     return my_addr
 109
 110 def ipAddrFromChicken():
 111     """Retrieve a possible IP address from the ipchecken website."""
 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:
 121          return []
 122
 123 def uncompact(s):
 124     """Extract the contact info from a compact peer representation.
 125
 126     @type s: C{string}
 127     @param s: the compact representation
 128     @rtype: (C{string}, C{int})
 129     @return: the IP address and port number to contact the peer on
 130     @raise ValueError: if the compact representation doesn't exist
 131     """
 132     if (len(s) != 6):
 133         raise ValueError
 134     ip = '.'.join([str(ord(i)) for i in s[0:4]])
 135     port = (ord(s[4]) << 8) | ord(s[5])
 136     return (ip, port)
 137
 138 def compact(ip, port):
 139     """Create a compact representation of peer contact info.
 140
 141     @type ip: C{string}
 142     @param ip: the IP address of the peer
 143     @type port: C{int}
 144     @param port: the port number to contact the peer on
 145     @rtype: C{string}
 146     @return: the compact representation
 147     @raise ValueError: if the compact representation doesn't exist
 148     """
 149
 150     s = ''.join([chr(int(i)) for i in ip.split('.')]) + \
 151           chr((port & 0xFF00) >> 8) + chr(port & 0xFF)
 152     if len(s) != 6:
 153         raise ValueError
 154     return s
 155
 156 def byte_format(s):
 157     """Format a byte size for reading by the user.
 158
 159     @type s: C{long}
 160     @param s: the number of bytes
 161     @rtype: C{string}
 162     @return: the formatted size with appropriate units
 163     """
 164     if (s < 1):
 165         r = str(int(s*1000.0)/1000.0) + 'B'
 166     elif (s < 10):
 167         r = str(int(s*100.0)/100.0) + 'B'
 168     elif (s < 102):
 169         r = str(int(s*10.0)/10.0) + 'B'
 170     elif (s < 1024):
 171         r = str(int(s)) + 'B'
 172     elif (s < 10485):
 173         r = str(int((s/1024.0)*100.0)/100.0) + 'KiB'
 174     elif (s < 104857):
 175         r = str(int((s/1024.0)*10.0)/10.0) + 'KiB'
 176     elif (s < 1048576):
 177         r = str(int(s/1024)) + 'KiB'
 178     elif (s < 10737418L):
 179         r = str(int((s/1048576.0)*100.0)/100.0) + 'MiB'
 180     elif (s < 107374182L):
 181         r = str(int((s/1048576.0)*10.0)/10.0) + 'MiB'
 182     elif (s < 1073741824L):
 183         r = str(int(s/1048576)) + 'MiB'
 184     elif (s < 1099511627776L):
 185         r = str(int((s/1073741824.0)*100.0)/100.0) + 'GiB'
 186     else:
 187         r = str(int((s/1099511627776.0)*100.0)/100.0) + 'TiB'
 188     return(r)
 189
 190 class TestUtil(unittest.TestCase):
 191     """Tests for the utilities."""
 192
 193     timeout = 5
 194     ip = '165.234.1.34'
 195     port = 61234
 196
 197     def test_compact(self):
 198         """Make sure compacting is reversed correctly by uncompacting."""
 199         d = uncompact(compact(self.ip, self.port))
 200         self.failUnlessEqual(d[0], self.ip)
 201         self.failUnlessEqual(d[1], self.port)