#!/usr/bin/env python # -*- coding: Latin1 -*- """ supernetting.py merges a list of subnets to remove subnets which are a part of a greater (least specific) subnet. The script can used to get a smaller routing tables or a smaller firewall rule set. The algorithm is detailed described in L{mergeSubnets} and L{sumSubnets}. Tested at:: - SunOS 5.8; Python 2.2.1 - Linux 2.6.11 on x86; Python 2.3.5 Revision History:: $Log: supernetting.py,v $ Revision 1.3 2005/05/19 18:39:04 carsten - version 1.0 - rename some function to get a more readable name - use internal functions to converts ip addresses between dotted quad and integer numbers - enhance and correct documentation - add option --test to allow a simple function test - run pychecker and fix a lof of messages Revision 1.2 2005/04/17 10:20:21 carsten - version 1.0rc1 - enhance documentation - add command line options - add a detailed usage message - small improvements Revision 1.1 2005/02/20 11:17:38 carsten - initial import to CVS - script merge subnets (supernetting) @note: You can use the same functionality in perl with the NetAddr-IP module from U{http://search.cpan.org/dist/NetAddr-IP/} @todo: add support for iana styled ip addresses @author: Carsten Grohmann @license: GPL version 2.0 or above @version: see L{__version__} @var __author__: author of this script and his email address @var __copyright__: copyright and licence string @var __revision__: internal revision number @var __version__: version number of this script @var __date__: date of the last changes @var mask_hex: used for fast translations beetween CIDR styled subnetmasks and normal styled subnet masks @var lowest_bit: dictionary with the lowest bit of a netmask; it will used to get faster access and better readability of the source code @var verbose: switch to enable or disable additional informations about the merging and reducing of the subnets; see also L{info} """ # $Source: /home/cvs/project_carsten/repository/scripts/supernetting.py,v $ # $Id: supernetting.py,v 1.3 2005/05/19 18:39:04 carsten Exp $ __author__ = "Carsten Grohmann " __copyright__ = "Carsten Grohmann (c) 2005; Licence: GPL V2.0 or above" __revision__ = "$Revision: 1.3 $" __version__ = "1.0" __date__ = "$Date: 2005/05/19 18:39:04 $" import getopt import os import socket import struct import sys # add additional epydoc field "example" __extra_epydoc_fields__ = [('example', 'Example', 'Examples')] # used to get fast translations beetween CIDR styled subnetmasks and normal # styled subnet masks mask_hex = { 0 : '0x00000000L', 1 : '0x80000000L', 2 : '0xC0000000L', 3 : '0xE0000000L', 4 : '0xF0000000L', 5 : '0xF8000000L', 6 : '0xFC000000L', 7 : '0xFE000000L', 8 : '0xFF000000L', 9 : '0xFF800000L', 10 : '0xFFC00000L', 11 : '0xFFE00000L', 12 : '0xFFF00000L', 13 : '0xFFF80000L', 14 : '0xFFFC0000L', 15 : '0xFFFE0000L', 16 : '0xFFFF0000L', 17 : '0xFFFF8000L', 18 : '0xFFFFC000L', 19 : '0xFFFFE000L', 20 : '0xFFFFF000L', 21 : '0xFFFFF800L', 22 : '0xFFFFFC00L', 23 : '0xFFFFFE00L', 24 : '0xFFFFFF00L', 25 : '0xFFFFFF80L', 26 : '0xFFFFFFC0L', 27 : '0xFFFFFFE0L', 28 : '0xFFFFFFF0L', 29 : '0xFFFFFFF8L', 30 : '0xFFFFFFFCL', 31 : '0xFFFFFFFEL', 32 : '0xFFFFFFFFL' } # contains the lowest bit of a 32bit netmask # the netmask is also the key to access the values lowest_bit = { 0 : 1L<<(32-0), 1 : 1L<<(32-1), 2 : 1L<<(32-2), 3 : 1L<<(32-3), 4 : 1L<<(32-4), 5 : 1L<<(32-5), 6 : 1L<<(32-6), 7 : 1L<<(32-7), 8 : 1L<<(32-8), 9 : 1L<<(32-9), 10 : 1L<<(32-10), 11 : 1L<<(32-11), 12 : 1L<<(32-12), 13 : 1L<<(32-13), 14 : 1L<<(32-14), 15 : 1L<<(32-15), 16 : 1L<<(32-16), 17 : 1L<<(32-17), 18 : 1L<<(32-18), 19 : 1L<<(32-19), 20 : 1L<<(32-20), 21 : 1L<<(32-21), 22 : 1L<<(32-22), 23 : 1L<<(32-23), 24 : 1L<<(32-24), 25 : 1L<<(32-25), 26 : 1L<<(32-26), 27 : 1L<<(32-27), 28 : 1L<<(32-28), 29 : 1L<<(32-29), 30 : 1L<<(32-30), 31 : 1L<<(32-31), 32 : 1L<<(32-32) } # switch to enable detailed messages verbose = False def isCIDRSubnet(subnet): """ Checks if the subnet mask a valid dotted quad IP address with a subnet mask in CIDR style @note: This functions doesn't raise exceptions. @return: True or False """ if not isinstance(subnet, str): return False if not subnet: return False try: (address, netmask) = subnet.split("/") (octet1, octet2, octet3, octet4) = address.split(".") octet1 = int(octet1) octet2 = int(octet2) octet3 = int(octet3) octet4 = int(octet4) netmask = int(netmask) if (octet1<0) or (octet1>255) or \ (octet2<0) or (octet2>255) or \ (octet3<0) or (octet3>255) or \ (octet4<0) or (octet4>255) or \ (netmask<0) or (netmask>32): return False return True except: return False def getCIDRNetmask(subnet): """ Return the netmask from the subnet @return: the netmask in cidr style as integer @param subnet: any subnetmask in CIDR style e.g. "1.2.3.4/30" @type subnet: string @note: There is no validity check of the returned value @example: getCIDRNetmask("1.2.3.4/5") returns 5 @raise TypeError: on wrong type of the parameter @raise ValueError: on empty or wrong format of the parameter """ if not isinstance(subnet, str): raise TypeError, "ERROR: Wrong data type given! Expected type: string" elif not subnet: raise ValueError, "ERROR: Wrong value given!" try: return int(subnet.split("/")[1]) except: raise ValueError, "ERROR: Wrong value format!" def getDecimalIP(subnet): """ Returns a decimal IP address from a subnetmask with a decimal IP inside @return: a decimal IP address as long @param subnet: any subnetmask with a decimal IP address e.g. "123456/24" or "1234567/255.255.255.0" @type subnet: string @example: getDecimalIP("123456/24") returns 123456L or getDecimalIP("1234567/255.255.255.0") returns 1234567L @raise TypeError: on wrong type of the parameter @raise ValueError: on empty or wrong format of the parameter """ if not isinstance(subnet, str): raise TypeError, "ERROR: Wrong data type given! Expected type: string" elif not subnet: raise ValueError, "ERROR: Wrong value given!" try: return long(subnet.split("/")[0]) except: raise ValueError, \ "ERROR: Wrong value format! Expected type: subnetaddress" def getIP(subnet): """ Returns the IP address from a subnet @return: a IP address as string @param subnet: every kind of subnetmask @type subnet: string @example: getIP("12345678/23") returns "12345678"; getIP("12.34.56.78/255.255.255.0") returns "12.34.56.78" @raise TypeError: on wrong type of the parameter @raise ValueError: on empty or wrong format of the parameter """ if not isinstance(subnet, str): raise TypeError, "ERROR: Wrong data type given! Expected type: string" elif not subnet: raise ValueError, "ERROR: Wrong value given!" try: return subnet.split("/")[0] except: raise ValueError, \ "ERROR: Wrong value format! Expected type: subnetaddress" def getNetworkAddress(subnet, netmask=0): """ Returns the network address. This function have two imput formats: - One parameter: 1. parameter: subnet addresses in CIDR style with a decimal IP e.g. "12345678/21" - Two parameters: 1. parameter: IP address as decimal number 2. parameter: Subnetmask in CIDR style as decimal number @return: the network address as decimal number @type subnet: string or long @type netmask: int or long @example: getNetworkAddress("3232236289/16") returns 3232235520L @raise TypeError: on wrong type of the parameter @raise ValueError: on empty or wrong format of the parameter """ if isinstance(subnet, str): try: address = long(subnet.split("/")[0]) netmask = int(subnet.split("/")[1]) except: raise ValueError, "ERROR: Wrong value format!" elif isinstance(subnet, long) and isinstance(netmask, (int, long)): address = subnet else: raise TypeError, \ "ERROR: Wrong data type given!" return address & eval(mask_hex[netmask]) def dottedQuadToDecimal(ipaddress): """ Convert a dotted quad IP address in long integer @param ipaddress: IP address in dotted quad style @return: IP address as long integer @type ipaddress: string @example: dottedQuadToDecimal("1.2.3.4") returns 16909060L @raise TypeError: on wrong type of the parameter @raise ValueError: if the parameter ipaddress is not a number """ # raise if no data found if not isinstance(ipaddress, str): raise TypeError, "ERROR: Wrong data type given! Expected type: string" if not ipaddress: raise ValueError, "ERROR: No number given!" try: decimal = ( struct.unpack('!L', socket.inet_aton(ipaddress)))[0] except: raise ValueError, 'ERROR: Invalid IP address "%s" given!' % ipaddress return decimal def decimalToDottedQuad(number): """ Convert a long integer IP address to dotted quad IP address @param number: long integer IP address @return: dotted quad IP address @type number: long integer @example: decimalToDottedQuad(123456789L) will returns "7.91.205.21" @raise TypeError: on wrong type of the parameter @raise ValueError: if the parameter is not a number """ # raise if no data found if isinstance(number, str): try: number = long(number) except: raise TypeError, \ "ERROR: Wrong data type given! Expected type: long" elif not number: raise ValueError, "ERROR: No number given!" return socket.inet_ntoa(struct.pack('!L', number)) def normSubnet(subnet): """ Clear all bits of the host part in a subnet address dotted quad style @type subnet: string @return: normalized subnet address @example: normSubnet("3232236035/24") returns "3232236032/24" @raise TypeError: on wrong type of the parameter @raise ValueError: if the parameter is empty or not a subnet """ if not isinstance(subnet, str): raise TypeError, "ERROR: Wrong data type given! Expected type: string!" elif not subnet: raise ValueError, "ERROR: No subnet given!" # split into separate parts try: cidr = getCIDRNetmask(subnet) address = getDecimalIP(subnet) except: raise ValueError, "ERROR: Invalid subnet %s!" % subnet address_normalize = getNetworkAddress(address, cidr) return "%d/%d" % (address_normalize, cidr) def formatSubnet(subnet): """ Transforms a subnet with a decimal IP into a subnet with a dotted quad IP @param subnet: subnet address @type subnet: string @return: subnet with dotted quad IP @example: formatSubnet("1/28") will return "0.0.0.1/28" @raise ValueError: if the parameter is empty or not a subnet """ if not isinstance(subnet, str): raise TypeError, "ERROR: Wrong data type given! Expected type: string!" elif not subnet: raise ValueError, "ERROR: No subnet given!" try: address = decimalToDottedQuad(getDecimalIP(subnet)) netmask = getCIDRNetmask(subnet) except: raise ValueError, "ERROR: Wrong value format!" return "%s/%d" % (address, netmask) def sumSubnets(subnetlist): """ Summarise a list of subnets. Remove all subnet, which are a member of smaller subnet @param subnetlist: a list of subnets, every subnet have to consist of a long int IP address and a subnetmask in CIDR style @type subnetlist: list or tuple of strings @return: list of subnets @see: L{mergeSubnets} @raise TypeError: if parameter subnetlist is not a list """ # finish if not data if not isinstance(subnetlist, (list, tuple)): raise TypeError, \ "ERROR: Wrong data type given! Expected type: list or tuple!" if not subnetlist: return [] returnlist = {} dellist = [] for subnet in subnetlist: returnlist[subnet] = 1 for subnet_out in returnlist.iterkeys(): for subnet_in in returnlist.iterkeys(): # break if the inner and the outer loop have the same value if subnet_out == subnet_in: break netmask_out = getCIDRNetmask(subnet_out) netmask_in = getCIDRNetmask(subnet_in) # continue with the next interation if the netmask is equal if netmask_out == netmask_in: continue elif netmask_out > netmask_in: netmask_lesser = netmask_in subnet_greater = subnet_out subnet_lesser = subnet_in else: netmask_lesser = netmask_out subnet_greater = subnet_in subnet_lesser = subnet_out netaddress_out = getNetworkAddress(subnet_out) netaddress_in = getNetworkAddress(subnet_in) # The smaller netaddress contains the greater netmask # when this condition are true # 1. the smaller netaddress have to be in the greater netaddress if getNetworkAddress(netaddress_out, netmask_lesser) == \ getNetworkAddress(netaddress_in, netmask_lesser): dellist.append(subnet_greater) # print info message info("Remove subnet %s because it's in %s" % ( formatSubnet(subnet_greater), formatSubnet(subnet_lesser))) # remove all redundant subnets for i in dellist: if returnlist.has_key(i): del returnlist[i] return returnlist.keys() def mergeSubnets(subnetlist): """ Merge subnets (Supernetting) A even networkaddress and the following odd networkaddress with the same netmask can merge to the even networkaddress with a networkmask reduced by 1 bit. This smaller (least specific) subnet contains both parent subnets. Example: The subnets 4.0.0.0/8 and 5.0.0.0/8 can reduce to 4.0.0.0/7. The function checks also if subnets are parts of other subnets and reduces the unnecessary subnet entry. See L{sumSubnets} to get more informations about this step. You get smaller routing or firewall tables thru this both steps. It is possible to optimize this function to get more speed. To optimize it, change the handling of the odd right bit of the subnetmask. @return: list of subnets @type subnetlist: list or tuple of strings @see: L{sumSubnets} @raise TypeError: if parameter subnetlist is not a list """ # finish if not data if not isinstance(subnetlist, (list, tuple)): raise TypeError, \ "ERROR: Wrong data type given! Expected type: list or tuple!" elif not subnetlist: return [] # list for all odd IP addresses addresslist = {} for subnet in subnetlist: # normalize IP address ("cut off" unused bits) and split them subnet = normSubnet(subnet) address = getDecimalIP(subnet) netmask = getCIDRNetmask(subnet) addresslist["%u/%u" % (address, netmask)]=1 # initialize length of lists len_begin = len(addresslist) # initialize len_end have to not equal to len_begin len_end = len_begin+1 while (len_begin != len_end): # memorize number of elements on olist len_begin = len(addresslist) newlist = sumSubnets(addresslist.keys()) addresslist.clear() for i in newlist: addresslist[i] = 1 # create a new sorted list with subnets from the current address list subnetlist = addresslist.keys() subnetlist.sort() # main loop of the merge function # searches two sequently netmasks, a even and the following odd # both will removed and a new netmask one bit shorter will added to # the list of addresses for subnet_current in subnetlist: address_current = getDecimalIP(subnet_current) netmask_current = getCIDRNetmask(subnet_current) # break on odd subnet if address_current & lowest_bit[netmask_current]: continue # set right bit of netaddress to 1 address_odd = address_current | lowest_bit[netmask_current] # search following odd entry and replace even subnet and the # following odd subnet with a one bit smaller subnet subnet_odd = "%d/%d" % (address_odd, netmask_current) subnet_new = "%d/%d" % (address_current, netmask_current-1) if addresslist.has_key(subnet_odd): del addresslist[subnet_odd] del addresslist[subnet_current] addresslist[subnet_new] = 1 # print info message info("Merge subnets %s and %s to %s" % ( formatSubnet(subnet_odd), formatSubnet(subnet_current), formatSubnet(subnet_new))) break # get the current number of elements in list len_end = len(addresslist) # concatenate bot dictionaries to one list returnlist = [] map(lambda x: returnlist.append(x), addresslist.iterkeys()) return returnlist def info(message=""): """ Prints a information The message will print out with "INFO:" as prefix. @raise ValueError: on a empty information @param message: the information message @see: L{verbose} """ global verbose # not verbose? do nothing if not verbose: return # raise on empty error message if not message: raise ValueError, "INTERNAL ERROR: Empty info message occur!" # prints the warning message sys.stdout.write("INFO: %s\n" % message) def warning(message=""): """ Prints a warn message The message will print out with "WARNING:" as prefix. @raise ValueError: on a empty warning message @param message: the warning message """ # raise on empty error message if not message: raise ValueError, "INTERNAL ERROR: Empty warning message occur!" # prints the warning message sys.stderr.write("WARNING: %s\n" % message) def error(message="", returnvalue=1): """ Prints an error message to stderr and leave the script The message will print out with "ERROR:" as prefix @raise ValueError: on a empty error message @param message: the error message @param returnvalue: exit value; default is set to 1; None will continue the script """ # raise on empty error message if not message: raise ValueError, "INTERNAL ERROR: Empty error message occur!" # prints the error message and leave the script sys.stderr.write("ERROR: %s\n" %message) if returnvalue: sys.exit(returnvalue) def usage(rval=0): """ Print usage message and exit @param rval: return value of this script """ print """supernetting.py merges subnets (Supernetting) Usage: supernetting.py [Option1 [Option2]] Options: -i, --in-file input file -o, --out-file output file -v, --verbose shows the merge and reduce process detailed --version shows the version and licence of this script --test tests some internal functions -h, --help shows this message Details: A even networkaddress and the following odd networkaddress with the same netmask can merge to the even networkaddress with a networkmask reduced by 1 bit. Example: The subnets 4.0.0.0/8 and 5.0.0.0/8 can reduce to 4.0.0.0/7. supernetting.py also checks if subnets are parts of other subnets and removes the unnecessary subnet entry from the list. Example: 5.0.0.0/8 is a part of 4.0.0.0/7 and will removed This script helps you to get a smaller routing tables or a smaller firewall rule set. General: The input should have only one address per line. Lines starting with "#", ";" or a whitespace will ignored. All lines with other content as a subnet address will be ignored by printing a warning message. The input can read from a file, specified with "--in-file" or from stdin. The result will print to stdout or a new file specified with "--out-file". You can use "-" as filename to use stdin resp. stdout. Reading from stdin and writing to stdout is the default behaviour so it is not necessary to set both options. Note: supernetting.py doesn't writes in existing files for security reasons! """ sys.exit(rval) def test(): """ Runs some tests to check the correct work of different functions """ # to count the successfull tests successfull = 0 # number of tests to run numberOfTests = 11 # function to check the correct function of this script def runtest(function, result): try: returnvalue = eval(function) except: print "Failed!" return 0 if returnvalue == result: print "OK" return 1 print "Failed!" return 0 print "Test 1: Check decimalToDottedQuad: ........", successfull += runtest('decimalToDottedQuad(123456789L)', "7.91.205.21") print "Test 2: Check dottedQuadToDecimal: ........", successfull += runtest('dottedQuadToDecimal("1.2.3.4")', 16909060L) print "Test 3: Check formatSubnet: ...............", successfull += runtest('formatSubnet("1/28")', "0.0.0.1/28") print "Test 4: Check getCIDRNetmask: .............", successfull += runtest('getCIDRNetmask("1.2.3.4/5")', 5) print "Test 5: Check getDecimalIP: ...............", successfull += runtest('getDecimalIP("123456/24")', 123456L) print "Test 6: Check getIP: ......................", successfull += runtest('getIP("12.34.56.78/28")', "12.34.56.78") print "Test 7: Check getNetworkAddress: ..........", successfull += runtest('getNetworkAddress("3232236289/16")', 3232235520L) print "Test 8: Check isCIDRSubnet: ...............", successfull += runtest('isCIDRSubnet("12.3.4.5/16")', True) print "Test 9: Check mergeSubnets: ...............", successfull += runtest('mergeSubnets(["2/32", "3/32"])', ['2/31']) print "Test 10: Check mergeSubnets: ...............", successfull += runtest('normSubnet("3232236035/24")', '3232236032/24') print "Test 11: Check sumSubnets: .................", successfull += runtest('sumSubnets(("2/32", "3/31"))', ['3/31']) # print test results print "===================================================" print "Tests successfull: %d" % successfull print "Tests failed: %d" % (numberOfTests-successfull) if (successfull != numberOfTests): print "Warning: One or more tests failed!. Please check the script" print " and send a email to the author." sys.exit(0) def main(): """ main loop - get and check the command line options and runs the functions """ global verbose # filename of the input file or "-" for stdin infile = "-" # filename of the output file or "-" for stdin outfile = "-" # list with subnets subnetlist = []; # get command line options try: optlist = getopt.getopt(sys.argv[1:], "hvi:o:", ["in-file=", "out-file=", "test", "verbose", "version", "help"])[0] except getopt.GetoptError, msg: error("Wrong program option: %s" % msg, None) usage(1) except: error("ERROR: Unknown exception raised.", None) print sys.exc_info() sys.exit(1) # process options for option, argument in optlist: if option == "--version": print "%s Version %s" % ("supernetting.py", __version__) print "Author %s" % __author__ print "Copyright %s" % __copyright__ print "Homepage http://www.carstengrohmann.de" print "Last Changes %s" % __date__ sys.exit(0) elif option == "--help" or option == "-h": usage(0) elif option == "--test": test() elif option == "--verbose" or option == "-v": verbose = True elif option == "--in-file" or option =="-i": if argument != "-": if not os.access(argument, os.F_OK): error("Input file %s doesn't exist!" % argument) if not os.access(argument, os.R_OK): error("Input file %s isn't readable!" % argument) infile = argument elif option == "--out-file" or option == "-o": if argument != "-": # check file - it haven't exist if os.access(argument, os.F_OK): error("Output file %s exists. Please remove it!" % argument) outfile = argument # read data from file or stdin if infile != "-": try: filehandle = open(infile, "r") except IOError: error('ERROR: Could not open file "%s"! to read IP addresses' % infile) else: filehandle = sys.stdin # read data from file for line in filehandle: # remove line breaks line = line.rstrip() # skip comments or lines starting with a space or a tab firstcharacter = line[0:1] if (firstcharacter == "#") or (firstcharacter == ";") or \ (firstcharacter == " ") or (firstcharacter == "\t"): continue if isCIDRSubnet(line): try: # convert ip address to a decimal number ipaddress = dottedQuadToDecimal(getIP(line)) netmask = getCIDRNetmask(line) # add ip address and subnet reformated to dictionary subnetlist.append("%d/%d" % (ipaddress, netmask)) except: warning('Skip invalid networkaddress "%s"!' % line) else: warning('Skip invalid networkaddress "%s"!' % line) # close file handle if infile != "-": filehandle.close() # leave script, if subnetlist contains no entries now if not subnetlist: warning("No valid subnets found. Ending program.") sys.exit(0) # merge subnets subnetlist = mergeSubnets(subnetlist) # sort subnet list subnetlist.sort() # write result to output file or stdout # read data from file or stdin if outfile != "-": try: # output file haven't exist if os.access(argument, os.F_OK): error("Output file %s exists. Please remove it!" % argument) filehandle = open(outfile, "w") except IOError: error('ERROR: Could not open file "%s" to write IP addresses!' % outfile) else: filehandle = sys.stdout # print result for subnet in subnetlist: filehandle.write("%s\n" % formatSubnet(subnet)) # close file handle if outfile != "-": filehandle.close() if __name__ == '__main__': main()