apt_dht/util.py

   1 ## Copyright 2002-2003 Andrew Loewenstern, All Rights Reserved
   2 # see LICENSE.txt for license information
   3
   4 import os, re
   5
   6 from twisted.python import log
   7 from twisted.trial import unittest
   8
   9 isLocal = re.compile('^(192\.168\.[0-9]{1,3}\.[0-9]{1,3})|'+
  10                      '(10\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})|'+
  11                      '(172\.0?([1][6-9])|([2][0-9])|([3][0-1])\.[0-9]{1,3}\.[0-9]{1,3})|'+
  12                      '(127\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})$')
  13
  14 def findMyIPAddr(addrs, intended_port, local_ok = False):
  15     """Find the best IP address to use from a list of possibilities.
  16
  17     @param addrs: the list of possible IP addresses
  18     @param intended_port: the port that was supposed to be used
  19     @param local_ok: whether known local/private IP ranges are allowed
  20         (defaults to False)
  21     @return: the preferred IP address, or None if one couldn't be found
  22     """
  23     log.msg("got addrs: %r" % (addrs,))
  24     my_addr = None
  25
  26     try:
  27         ifconfig = os.popen("/sbin/ifconfig |/bin/grep inet|"+
  28                             "/usr/bin/awk '{print $2}' | "+
  29                             "sed -e s/.*://", "r").read().strip().split('\n')
  30     except:
  31         ifconfig = []
  32
  33     # Get counts for all the non-local addresses returned
  34     addr_count = {}
  35     for addr in ifconfig:
  36         if local_ok or not isLocal.match(addr):
  37             addr_count.setdefault(addr, 0)
  38             addr_count[addr] += 1
  39
  40     local_addrs = addr_count.keys()
  41     if len(local_addrs) == 1:
  42         my_addr = local_addrs[0]
  43         log.msg('Found remote address from ifconfig: %r' % (my_addr,))
  44
  45     # Get counts for all the non-local addresses returned
  46     addr_count = {}
  47     port_count = {}
  48     for addr in addrs:
  49         if local_ok or not isLocal.match(addr[0]):
  50             addr_count.setdefault(addr[0], 0)
  51             addr_count[addr[0]] += 1
  52             port_count.setdefault(addr[1], 0)
  53             port_count[addr[1]] += 1
  54
  55     # Find the most popular address
  56     popular_addr = []
  57     popular_count = 0
  58     for addr in addr_count:
  59         if addr_count[addr] > popular_count:
  60             popular_addr = [addr]
  61             popular_count = addr_count[addr]
  62         elif addr_count[addr] == popular_count:
  63             popular_addr.append(addr)
  64
  65     # Find the most popular port
  66     popular_port = []
  67     popular_count = 0
  68     for port in port_count:
  69         if port_count[port] > popular_count:
  70             popular_port = [port]
  71             popular_count = port_count[port]
  72         elif port_count[port] == popular_count:
  73             popular_port.append(port)
  74
  75     port = intended_port
  76     if len(port_count.keys()) > 1:
  77         log.msg('Problem, multiple ports have been found: %r' % (port_count,))
  78         if port not in port_count.keys():
  79             log.msg('And none of the ports found match the intended one')
  80     elif len(port_count.keys()) == 1:
  81         port = port_count.keys()[0]
  82     else:
  83         log.msg('Port was not found')
  84
  85     if len(popular_addr) == 1:
  86         log.msg('Found popular address: %r' % (popular_addr[0],))
  87         if my_addr and my_addr != popular_addr[0]:
  88             log.msg('But the popular address does not match: %s != %s' % (popular_addr[0], my_addr))
  89         my_addr = popular_addr[0]
  90     elif len(popular_addr) > 1:
  91         log.msg('Found multiple popular addresses: %r' % (popular_addr,))
  92         if my_addr and my_addr not in popular_addr:
  93             log.msg('And none of the addresses found match the ifconfig one')
  94     else:
  95         log.msg('No non-local addresses found: %r' % (popular_addr,))
  96
  97     if not my_addr:
  98         log.msg("Remote IP Address could not be found for this machine")
  99
 100     return my_addr
 101
 102 def ipAddrFromChicken():
 103     import urllib
 104     ip_search = re.compile('\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}')
 105     try:
 106          f = urllib.urlopen("http://www.ipchicken.com")
 107          data = f.read()
 108          f.close()
 109          current_ip = ip_search.findall(data)
 110          return current_ip
 111     except Exception:
 112          return []
 113
 114 def uncompact(s):
 115     """Extract the contatc info from a compact peer representation.
 116
 117     @type s: C{string}
 118     @param s: the compact representation
 119     @rtype: (C{string}, C{int})
 120     @return: the IP address and port number to contact the peer on
 121     @raise ValueError: if the compact representation doesn't exist
 122     """
 123     if (len(s) != 6):
 124         raise ValueError
 125     ip = '.'.join([str(ord(i)) for i in s[0:4]])
 126     port = (ord(s[4]) << 8) | ord(s[5])
 127     return (ip, port)
 128
 129 def compact(ip, port):
 130     """Create a compact representation of peer contact info.
 131
 132     @type ip: C{string}
 133     @param ip: the IP address of the peer
 134     @type port: C{int}
 135     @param port: the port number to contact the peer on
 136     @rtype: C{string}
 137     @return: the compact representation
 138     @raise ValueError: if the compact representation doesn't exist
 139     """
 140
 141     s = ''.join([chr(int(i)) for i in ip.split('.')]) + \
 142           chr((port & 0xFF00) >> 8) + chr(port & 0xFF)
 143     if len(s) != 6:
 144         raise ValueError
 145     return s
 146
 147 class TestUtil(unittest.TestCase):
 148     """Tests for the utilities."""
 149
 150     timeout = 5
 151     ip = '165.234.1.34'
 152     port = 61234
 153
 154     def test_compact(self):
 155         d = uncompact(compact(self.ip, self.port))
 156         self.failUnlessEqual(d[0], self.ip)
 157         self.failUnlessEqual(d[1], self.port)