2 """Some utitlity functions for use in the apt-p2p program.
4 @var isLocal: a compiled regular expression suitable for testing if an
5 IP address is from a known local or private range
10 from twisted.python import log
11 from twisted.trial import unittest
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]\.[0-9]{1,3}\.[0-9]{1,3})|'+
16 '(172\.0?2[0-9]\.[0-9]{1,3}\.[0-9]{1,3})|'+
17 '(172\.0?3[0-1]\.[0-9]{1,3}\.[0-9]{1,3})|'+
18 '(127\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})$')
20 def findMyIPAddr(addrs, intended_port, local_ok = False):
21 """Find the best IP address to use from a list of possibilities.
23 @param addrs: the list of possible IP addresses
24 @param intended_port: the port that was supposed to be used
25 @param local_ok: whether known local/private IP ranges are allowed
27 @return: the preferred IP address, or None if one couldn't be found
29 log.msg("got addrs: %r" % (addrs,))
32 # Try to find an address using the ifconfig function
34 ifconfig = os.popen("/sbin/ifconfig |/bin/grep inet|"+
35 "/usr/bin/awk '{print $2}' | "+
36 "sed -e s/.*://", "r").read().strip().split('\n')
40 # Get counts for all the non-local addresses returned from ifconfig
43 if local_ok or not isLocal.match(addr):
44 addr_count.setdefault(addr, 0)
47 # If only one was found, use it as a starting point
48 local_addrs = addr_count.keys()
49 if len(local_addrs) == 1:
50 my_addr = local_addrs[0]
51 log.msg('Found remote address from ifconfig: %r' % (my_addr,))
53 # Get counts for all the non-local addresses returned from the DHT
57 if local_ok or not isLocal.match(addr[0]):
58 addr_count.setdefault(addr[0], 0)
59 addr_count[addr[0]] += 1
60 port_count.setdefault(addr[1], 0)
61 port_count[addr[1]] += 1
63 # Find the most popular address
66 for addr in addr_count:
67 if addr_count[addr] > popular_count:
69 popular_count = addr_count[addr]
70 elif addr_count[addr] == popular_count:
71 popular_addr.append(addr)
73 # Find the most popular port
76 for port in port_count:
77 if port_count[port] > popular_count:
79 popular_count = port_count[port]
80 elif port_count[port] == popular_count:
81 popular_port.append(port)
83 # Check to make sure the port isn't being changed
85 if len(port_count.keys()) > 1:
86 log.msg('Problem, multiple ports have been found: %r' % (port_count,))
87 if port not in port_count.keys():
88 log.msg('And none of the ports found match the intended one')
89 elif len(port_count.keys()) == 1:
90 port = port_count.keys()[0]
92 log.msg('Port was not found')
94 # If one is popular, use that address
95 if len(popular_addr) == 1:
96 log.msg('Found popular address: %r' % (popular_addr[0],))
97 if my_addr and my_addr != popular_addr[0]:
98 log.msg('But the popular address does not match: %s != %s' % (popular_addr[0], my_addr))
99 my_addr = popular_addr[0]
100 elif len(popular_addr) > 1:
101 log.msg('Found multiple popular addresses: %r' % (popular_addr,))
102 if my_addr and my_addr not in popular_addr:
103 log.msg('And none of the addresses found match the ifconfig one')
105 log.msg('No non-local addresses found: %r' % (popular_addr,))
108 log.msg("Remote IP Address could not be found for this machine")
112 def ipAddrFromChicken():
113 """Retrieve a possible IP address from the ipchecken website."""
115 ip_search = re.compile('\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}')
117 f = urllib.urlopen("http://www.ipchicken.com")
120 current_ip = ip_search.findall(data)
126 """Extract the contact info from a compact peer representation.
129 @param s: the compact representation
130 @rtype: (C{string}, C{int})
131 @return: the IP address and port number to contact the peer on
132 @raise ValueError: if the compact representation doesn't exist
136 ip = '.'.join([str(ord(i)) for i in s[0:4]])
137 port = (ord(s[4]) << 8) | ord(s[5])
140 def compact(ip, port):
141 """Create a compact representation of peer contact info.
144 @param ip: the IP address of the peer
146 @param port: the port number to contact the peer on
148 @return: the compact representation
149 @raise ValueError: if the compact representation doesn't exist
152 s = ''.join([chr(int(i)) for i in ip.split('.')]) + \
153 chr((port & 0xFF00) >> 8) + chr(port & 0xFF)
159 """Format a byte size for reading by the user.
162 @param s: the number of bytes
164 @return: the formatted size with appropriate units
167 r = str(int(s*1000.0)/1000.0) + 'B'
169 r = str(int(s*100.0)/100.0) + 'B'
171 r = str(int(s*10.0)/10.0) + 'B'
173 r = str(int(s)) + 'B'
175 r = str(int((s/1024.0)*100.0)/100.0) + 'KiB'
177 r = str(int((s/1024.0)*10.0)/10.0) + 'KiB'
179 r = str(int(s/1024)) + 'KiB'
180 elif (s < 10737418L):
181 r = str(int((s/1048576.0)*100.0)/100.0) + 'MiB'
182 elif (s < 107374182L):
183 r = str(int((s/1048576.0)*10.0)/10.0) + 'MiB'
184 elif (s < 1073741824L):
185 r = str(int(s/1048576)) + 'MiB'
186 elif (s < 1099511627776L):
187 r = str(int((s/1073741824.0)*100.0)/100.0) + 'GiB'
189 r = str(int((s/1099511627776.0)*100.0)/100.0) + 'TiB'
192 class TestUtil(unittest.TestCase):
193 """Tests for the utilities."""
199 def test_compact(self):
200 """Make sure compacting is reversed correctly by uncompacting."""
201 d = uncompact(compact(self.ip, self.port))
202 self.failUnlessEqual(d[0], self.ip)
203 self.failUnlessEqual(d[1], self.port)