initial import (Boeing r1752, NRL r878)

daemon/core/misc/ipaddr.py

@@ -0,0 +1,230 @@
+#
+# CORE
+# Copyright (c)2010-2012 the Boeing Company.
+# See the LICENSE file included in this distribution.
+#
+# author: Tom Goff <thomas.goff@boeing.com>
+#
+'''
+ipaddr.py: helper objects for dealing with IPv4/v6 addresses.
+'''
+
+import socket
+import struct
+import random
+
+AF_INET = socket.AF_INET
+AF_INET6 = socket.AF_INET6
+
+class MacAddr(object):
+    def __init__(self, addr):
+        self.addr = addr
+
+    def __str__(self):
+        return ":".join(map(lambda x: ("%02x" % ord(x)), self.addr))
+        
+    def tolinklocal(self):
+        ''' Convert the MAC address to a IPv6 link-local address, using EUI 48
+        to EUI 64 conversion process per RFC 5342.
+        '''
+        if not self.addr:
+            return IPAddr.fromstring("::")
+        tmp = struct.unpack("!Q", '\x00\x00' + self.addr)[0]
+        nic = long(tmp) & 0x000000FFFFFFL
+        oui = long(tmp) & 0xFFFFFF000000L
+        # toggle U/L bit
+        oui ^= 0x020000000000L
+        # append EUI-48 octets
+        oui = (oui << 16) | 0xFFFE000000L
+        return IPAddr(AF_INET6,  struct.pack("!QQ", 0xfe80 << 48, oui | nic))
+
+    @classmethod
+    def fromstring(cls, s):
+        addr = "".join(map(lambda x: chr(int(x, 16)), s.split(":")))
+        return cls(addr)
+
+    @classmethod
+    def random(cls):
+        tmp = random.randint(0, 0xFFFFFF)
+        tmp |= 0x00163E << 24    # use the Xen OID 00:16:3E
+        tmpbytes = struct.pack("!Q", tmp)
+        return cls(tmpbytes[2:])
+
+class IPAddr(object):
+    def __init__(self, af, addr):
+        # check if (af, addr) is valid
+        if not socket.inet_ntop(af, addr):
+            raise ValueError, "invalid af/addr"
+        self.af = af
+        self.addr = addr
+
+    def isIPv4(self):
+        return self.af == AF_INET
+
+    def isIPv6(self):
+        return self.af == AF_INET6
+
+    def __str__(self):
+        return socket.inet_ntop(self.af, self.addr)
+
+    def __eq__(self, other):
+        try:
+            return other.af == self.af and other.addr == self.addr
+        except:
+            return False
+
+    def __add__(self, other):
+        try:
+            carry = int(other)
+        except:
+            return NotImplemented
+        tmp = map(lambda x: ord(x), self.addr)
+        for i in xrange(len(tmp) - 1, -1, -1):
+            x = tmp[i] + carry
+            tmp[i] = x & 0xff
+            carry = x >> 8
+            if carry == 0:
+                break
+        addr = "".join(map(lambda x: chr(x), tmp))
+        return self.__class__(self.af, addr)
+
+    def __sub__(self, other):
+        try:
+            tmp = -int(other)
+        except:
+            return NotImplemented
+        return self.__add__(tmp)
+
+    @classmethod
+    def fromstring(cls, s):
+        for af in AF_INET, AF_INET6:
+            try:
+                return cls(af, socket.inet_pton(af, s))
+            except Exception, e:
+                pass
+        raise e
+    
+    @staticmethod
+    def toint(s):
+        ''' convert IPv4 string to 32-bit integer
+        '''
+        bin = socket.inet_pton(AF_INET, s)
+        return(struct.unpack('!I', bin)[0])
+
+class IPPrefix(object):
+    def __init__(self, af, prefixstr):
+        "prefixstr format: address/prefixlen"
+        tmp = prefixstr.split("/")
+        if len(tmp) > 2:
+            raise ValueError, "invalid prefix: '%s'" % prefixstr
+        self.af = af
+        if self.af == AF_INET:
+            self.addrlen = 32
+        elif self.af == AF_INET6:
+            self.addrlen = 128
+        else:
+            raise ValueError, "invalid address family: '%s'" % self.af
+        if len(tmp) == 2:
+            self.prefixlen = int(tmp[1])
+        else:
+            self.prefixlen = self.addrlen
+        self.prefix = socket.inet_pton(self.af, tmp[0])
+        if self.addrlen > self.prefixlen:
+            addrbits = self.addrlen - self.prefixlen
+            netmask = ((1L << self.prefixlen) - 1) << addrbits
+            prefix = ""
+            for i in xrange(-1, -(addrbits >> 3) - 2, -1):
+                prefix = chr(ord(self.prefix[i]) & (netmask & 0xff)) + prefix
+                netmask >>= 8
+            self.prefix = self.prefix[:i] + prefix
+
+    def __str__(self):
+        return "%s/%s" % (socket.inet_ntop(self.af, self.prefix),
+                          self.prefixlen)
+
+    def __eq__(self, other):
+        try:
+            return other.af == self.af and \
+                other.prefixlen == self.prefixlen and \
+                other.prefix == self.prefix
+        except:
+            return False
+
+    def __add__(self, other):
+        try:
+            tmp = int(other)
+        except:
+            return NotImplemented
+        a = IPAddr(self.af, self.prefix) + \
+            (tmp << (self.addrlen - self.prefixlen))
+        prefixstr = "%s/%s" % (a, self.prefixlen)
+        if self.__class__ == IPPrefix:
+            return self.__class__(self.af, prefixstr)
+        else:
+            return self.__class__(prefixstr)
+
+    def __sub__(self, other):
+        try:
+            tmp = -int(other)
+        except:
+            return NotImplemented
+        return self.__add__(tmp)
+
+    def addr(self, hostid):
+        tmp = int(hostid)
+        if (tmp == 1 or tmp == 0 or tmp == -1) and self.addrlen == self.prefixlen:
+            return IPAddr(self.af, self.prefix)
+        if tmp == 0 or \
+            tmp > (1 << (self.addrlen - self.prefixlen)) - 1 or \
+            (self.af == AF_INET and tmp == (1 << (self.addrlen - self.prefixlen)) - 1):
+            raise ValueError, "invalid hostid for prefix %s: %s" % (self, hostid)
+        addr = ""
+        for i in xrange(-1, -(self.addrlen >> 3) - 1, -1):
+            addr = chr(ord(self.prefix[i]) | (tmp & 0xff)) + addr
+            tmp >>= 8
+            if not tmp:
+                break
+        addr = self.prefix[:i] + addr
+        return IPAddr(self.af, addr)
+
+    def minaddr(self):
+        return self.addr(1)
+
+    def maxaddr(self):
+        if self.af == AF_INET:
+            return self.addr((1 << (self.addrlen - self.prefixlen)) - 2)
+        else:
+            return self.addr((1 << (self.addrlen - self.prefixlen)) - 1)
+
+    def numaddr(self):
+        return max(0, (1 << (self.addrlen - self.prefixlen)) - 2)
+        
+    def prefixstr(self):
+        return "%s" % socket.inet_ntop(self.af, self.prefix)
+    
+    def netmaskstr(self):
+        addrbits = self.addrlen - self.prefixlen
+        netmask = ((1L << self.prefixlen) - 1) << addrbits
+        netmaskbytes = struct.pack("!L",  netmask)
+        return IPAddr(af=AF_INET, addr=netmaskbytes).__str__()
+
+class IPv4Prefix(IPPrefix):
+    def __init__(self, prefixstr):
+        IPPrefix.__init__(self, AF_INET, prefixstr)
+
+class IPv6Prefix(IPPrefix):
+    def __init__(self, prefixstr):
+        IPPrefix.__init__(self, AF_INET6, prefixstr)
+
+def isIPAddress(af, addrstr):
+    try:
+        tmp = socket.inet_pton(af, addrstr)
+        return True
+    except:
+        return False
+
+def isIPv4Address(addrstr):
+    return isIPAddress(AF_INET, addrstr)
+
+def isIPv6Address(addrstr):
+    return isIPAddress(AF_INET6, addrstr)