rwhoisd/Cidr.py

   1 # This file is part of python-rwhoisd
   2 #
   3 # Copyright (C) 2003, 2008 David E. Blacka
   4 #
   5 # This program is free software; you can redistribute it and/or modify
   6 # it under the terms of the GNU General Public License as published by
   7 # the Free Software Foundation; either version 2 of the License, or
   8 # (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful, but
  11 # WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13 # General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License
  16 # along with this program; if not, write to the Free Software
  17 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  18 # USA
  19
  20 import socket, types, copy, bisect, struct
  21 import v6addr
  22
  23 def new(address, netlen = -1):
  24     """Construct either a CidrV4 or CidrV6 object."""
  25
  26     # ints are probably v4 addresses.
  27     if isinstance(address, int):
  28         return CidrV4(address, netlen)
  29     # longs could be v4 addresses, but we will only assume so if the
  30     # value is small.
  31     if isinstance(address, long):
  32         if address <= pow(2, 32):
  33             return CidrV4(address, netlen)
  34         return CidrV6(address, netlen)
  35     # otherwise, a colon in the address is a dead giveaway.
  36     if ":" in address:
  37         return CidrV6(address, netlen)
  38     return CidrV4(address, netlen)
  39
  40 def valid_cidr(address):
  41     """Returns the converted Cidr object if 'address' is valid CIDR
  42     notation, False if not.  For the purposes of this module, valid
  43     CIDR notation consists of a IPv4 or IPv6 address with an optional
  44     trailing "/netlen"."""
  45
  46     if isinstance(address, Cidr): return address
  47     try:
  48         c = new(address)
  49         return c
  50     except socket.error:
  51         return False
  52
  53
  54 class Cidr:
  55     """A class representing a generic CIDRized network value."""
  56
  57     def _initialize(self, address, netlen):
  58         """This a common constructor that is used by the subclasses."""
  59
  60         if isinstance(address, int) or \
  61                 isinstance(address, long) and netlen >= 0:
  62             self.numaddr, self.netlen = address, netlen
  63             self.addr = self._convert_ipaddr(address)
  64             self.calc()
  65             return
  66
  67         if not self.is_valid_cidr(address):
  68             raise socket.error, \
  69                 repr(address) + " is not a valid CIDR representation"
  70
  71         if netlen < 0:
  72             if type(address) == types.StringType:
  73                 if "/" in address:
  74                     self.addr, self.netlen = address.split("/", 1)
  75                 else:
  76                     self.addr, self.netlen = address, self._max_netlen()
  77             elif type(address) == types.TupleType:
  78                 self.addr, self.netlen = address
  79             else:
  80                 raise TypeError, "address must be a string or a tuple"
  81         else:
  82             self.addr, self.netlen = address, netlen
  83
  84
  85         # convert string network lengths to integer
  86         if type(self.netlen) == types.StringType:
  87             self.netlen = int(self.netlen)
  88
  89         self.calc()
  90
  91     def __str__(self):
  92         return self.addr + "/" + str(self.netlen)
  93
  94     def __repr__(self):
  95         return "<" + str(self) + ">"
  96
  97     def __cmp__(self, other):
  98         """One CIDR network block is less than another if the start
  99         address is numerically less or if the block is larger.  That
 100         is, supernets will sort before subnets.  This ordering allows
 101         for an efficient search for subnets of a given network."""
 102         if not isinstance(other, Cidr): return -1
 103         res = self._base_mask(self.numaddr) - other._base_mask(other.numaddr)
 104         if res == 0: res = self.netlen - other.netlen
 105         if res < 0: return -1
 106         if res > 0: return 1
 107         return 0
 108
 109     def calc(self):
 110         """This method should be called after any change to the main
 111         internal state: netlen or numaddr."""
 112
 113         # make sure the network length is valid
 114         if not self.is_valid_netlen(self.netlen):
 115             raise TypeError, "network length must be between 0 and %d" % \
 116                 (self._max_netlen())
 117
 118         # convert the string ipv4 address to a 32bit number
 119         self.numaddr = self._convert_ipstr(self.addr)
 120         # calculate our netmask
 121         self.mask = self._mask(self.netlen)
 122         # force the cidr address into correct masked notation
 123         self.numaddr &= self.mask
 124
 125         # convert the number back to a string to normalize the string
 126         self.addr = self._convert_ipaddr(self.numaddr)
 127
 128     def is_supernet(self, other):
 129         """returns True if the other Cidr object is a supernet (an
 130         enclosing network block) of this one.  A Cidr object is a
 131         supernet of itself."""
 132         return other.numaddr & self.mask == self.numaddr
 133
 134     def is_subnet(self, other):
 135         """returns True if the other Cidr object is a subnet (an
 136         enclosednetwork block) of this one.  A Cidr object is a
 137         subnet of itself."""
 138         return self.numaddr & other.mask == other.numaddr
 139
 140     def netmask(self):
 141         """return the netmask of this Cidr network"""
 142         return self._convert_ipaddr(self.mask)
 143
 144     def length(self):
 145         """return the length (in number of addresses) of this network block"""
 146         return 1 << (self._max_netlen() - self.netlen);
 147
 148     def end(self):
 149         """return the last IP address in this network block"""
 150         return self._convert_ipaddr(self.numaddr + self.length() - 1)
 151
 152     def to_netblock(self):
 153         return (self.addr, self.end())
 154
 155     def clone(self):
 156         # we can get away with a shallow copy (so far)
 157         return copy.copy(self)
 158
 159     def is_ipv6(self):
 160         if isinstance(self, CidrV6): return True
 161         return False
 162
 163     def is_valid_cidr(self, address):
 164         if "/" in address:
 165             addr, netlen = address.split("/", 1)
 166             netlen = int(netlen)
 167         else:
 168             addr, netlen = address, 0
 169         return self._is_valid_address(addr) and self.is_valid_netlen(netlen)
 170
 171     def is_valid_netlen(self, netlen):
 172         if netlen < 0: return False
 173         if netlen > self._max_netlen(): return False
 174         return True
 175
 176
 177 class CidrV4(Cidr):
 178     """A class representing a CIDRized IPv4 network value.
 179
 180     Specifically, it is representing a contiguous IPv4 network block
 181     that can be expressed as a ip-address/network-length pair."""
 182
 183     base_mask = 0xFFFFFFFF
 184     msb_mask  = 0x80000000
 185
 186     def __init__(self, address, netlen = -1):
 187         """This takes either a formatted string in CIDR notation:
 188         (e.g., "127.0.0.1/32"), A tuple consisting of an formatting
 189         string IPv4 address and a numeric network length, or the same
 190         as two arguments."""
 191
 192         self._initialize(address, netlen)
 193
 194     def _is_valid_address(self, address):
 195         """Returns True if the address is a legal IPv4 address."""
 196         try:
 197             self._convert_ipstr(address)
 198             return True
 199         except socket.error:
 200             return False
 201
 202     def _base_mask(self, numaddr):
 203         return numaddr & CidrV4.base_mask
 204
 205     def _max_netlen(self):
 206         return 32
 207
 208     def _convert_ipstr(self, addr):
 209         packed_numaddr = socket.inet_aton(addr)
 210         return struct.unpack("!I", packed_numaddr)[0]
 211
 212     def _convert_ipaddr(self, numaddr):
 213         packed_numaddr = struct.pack("!I", numaddr)
 214         return socket.inet_ntoa(packed_numaddr)
 215
 216     def _mask(self, len):
 217         return self._base_mask(CidrV4.base_mask << (32 - len))
 218
 219 class CidrV6(Cidr):
 220     """A class representing a CIDRized IPv6 network value.
 221
 222     Specifically, it is representing a contiguous IPv6 network block
 223     that can be expressed as a ipv6-address/network-length pair."""
 224
 225     base_mask  = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFL # 128-bits of all ones.
 226     msb_mask   = 0x80000000000000000000000000000000L
 227     lower_mask = 0x0000000000000000FFFFFFFFFFFFFFFFL
 228     upper_mask = 0xFFFFFFFFFFFFFFFF0000000000000000L
 229
 230     def __init__(self, address, netlen = -1):
 231
 232         self._initialize(address, netlen)
 233
 234     def _is_valid_address(self, address):
 235         try:
 236             self._convert_ipstr(address)
 237             return True
 238         except socket.error, e:
 239             return False
 240
 241     def _base_mask(self, numaddr):
 242         return numaddr & CidrV6.base_mask
 243
 244     def _max_netlen(self):
 245         return 128
 246
 247     def _convert_ipstr(self, addr):
 248         packed_numaddr = socket.inet_pton(socket.AF_INET6, addr)
 249         upper, lower = struct.unpack("!QQ", packed_numaddr);
 250         return (upper << 64) | lower
 251
 252     def _convert_ipaddr(self, numaddr):
 253         upper = (numaddr & CidrV6.upper_mask) >> 64;
 254         lower = numaddr & CidrV6.lower_mask;
 255         packed_numaddr = struct.pack("!QQ", upper, lower)
 256         return socket.inet_ntop(socket.AF_INET6, packed_numaddr)
 257
 258     def _mask(self, len):
 259         return self._base_mask(CidrV6.base_mask << (128 - len))
 260
 261
 262 def netblock_to_cidr(start, end):
 263     """Convert an arbitrary network block expressed as a start and end
 264     address (inclusive) into a series of valid CIDR blocks."""
 265
 266     def largest_prefix(length, max_netlen, msb_mask):
 267         # calculates the largest network length (smallest mask length)
 268         # that can fit within the block length.
 269         i = 1; v = length
 270         while i <= max_netlen:
 271             if v & msb_mask: break
 272             i += 1; v <<= 1
 273         return i
 274     def netlen_to_mask(n, max_netlen, base_mask):
 275         # convert the network length into its netmask
 276         return ~((1 << (max_netlen - n)) - 1) & base_mask
 277     def netlen_to_length(n, max_netlen, base_mask):
 278         return 1 << (max_netlen - n) & base_mask
 279
 280     # convert the start and ending addresses of the netblock to Cidr
 281     # object, mostly so we can get the numeric versions of their
 282     # addresses.
 283     cs = valid_cidr(start)
 284     ce = valid_cidr(end)
 285
 286     # if either the start or ending addresses aren't valid addresses,
 287     # quit now.
 288     if not cs or not ce:
 289         return None
 290     # if the start and ending addresses aren't in the same family, quit now
 291     if cs.is_ipv6() != ce.is_ipv6():
 292         return None
 293
 294     max_netlen = cs._max_netlen()
 295     msb_mask = cs.msb_mask
 296     base_mask = cs.base_mask
 297
 298     # calculate the number of IP address in the netblock
 299     block_len = ce.numaddr - cs.numaddr
 300     # calcuate the largest CIDR block size that fits
 301     netlen = largest_prefix(block_len + 1, max_netlen, msb_mask)
 302
 303     res = []; s = cs.numaddr
 304     while block_len > 0:
 305         mask = netlen_to_mask(netlen, max_netlen, base_mask)
 306         # check to see if our current network length is valid
 307         if (s & mask) != s:
 308             # if not, shrink the network block size
 309             netlen += 1
 310             continue
 311         # otherwise, we have a valid CIDR block, so add it to the list
 312         res.append(new(s, netlen))
 313         # and setup for the next round:
 314         cur_len = netlen_to_length(netlen, max_netlen, base_mask)
 315         s         += cur_len
 316         block_len -= cur_len
 317         netlen = largest_prefix(block_len + 1, max_netlen, msb_mask)
 318     return res
 319
 320 # test driver
 321 if __name__ == "__main__":
 322     import sys
 323     a = new("127.00.000.1/24")
 324     b = new("127.0.0.1", 32)
 325     c = new("24.232.119.192", 26)
 326     d = new("24.232.119.0", 24)
 327     e = new("24.224.0.0", 11)
 328     f = new("216.168.111.0/27");
 329     g = new("127.0.0.2/31");
 330     h = new("127.0.0.16/32")
 331     i = new("3ffe:4:201e:beef::0/64");
 332     j = new("2001:3c01::/32")
 333
 334     print f.addr
 335     print j.addr
 336
 337     try:
 338         bad = new("24.261.119.0", 32)
 339     except ValueError, x:
 340         print "error:", x
 341
 342     print "cidr:", a, "num addresses:", a.length(), "ending address", \
 343         a.end(), "netmask", a.netmask()
 344
 345     print "cidr:", j, "num addresses:", j.length(), "ending address", \
 346         j.end(), "netmask", j.netmask()
 347
 348     clist = [a, b, c, d, e, f, g, h, i , j]
 349     print "unsorted list of cidr objects:\n  ", clist
 350
 351
 352     clist.sort()
 353     print "sorted list of cidr object:\n  ", clist
 354
 355     k = new("2001:3c01::1:0", 120)
 356     print "supernet: ", str(j), " supernet of ", str(k), "? ", \
 357         str(j.is_supernet(k))
 358     print "supernet: ", str(k), " supernet of ", str(j), "? ", \
 359         str(k.is_supernet(j))
 360     print "subnet: ", str(j), " subnet of ", str(k), "? ", \
 361         str(j.is_subnet(k))
 362     print "subnet: ", str(k), " subnet of ", str(j), "? ", \
 363         str(k.is_subnet(j))
 364
 365     netblocks = [ ("192.168.10.0", "192.168.10.255"),
 366                   ("192.168.10.0", "192.168.10.63"),
 367                   ("172.16.0.0", "172.16.127.255"),
 368                   ("24.33.41.22", "24.33.41.37"),
 369                   ("196.11.1.0", "196.11.30.255"),
 370                   ("192.247.1.0", "192.247.10.255"),
 371                   ("10.131.43.3", "10.131.44.7"),
 372                   ("3ffe:4:5::", "3ffe:4:5::ffff"),
 373                   ("3ffe:4:5::", "3ffe:4:6::1")]
 374
 375     for start, end in netblocks:
 376         print "netblock %s - %s:" % (start, end)
 377         blocks = netblock_to_cidr(start, end)
 378         print blocks