Package Products :: Package ZenUtils :: Module IpUtil |
[hide private]
[frames] | no frames]

# Source Code for Module Products.ZenUtils.IpUtil

      1  ###########################################################################
      2  #
      3  # This program is part of Zenoss Core, an open source monitoring platform.
      4  # Copyright (C) 2007, Zenoss Inc.
      5  #
      6  # This program is free software; you can redistribute it and/or modify it
      7  # under the terms of the GNU General Public License version 2 as published by
      8  # the Free Software Foundation.
      9  #
     10  # For complete information please visit: http://www.zenoss.com/oss/
     11  #
     12  ###########################################################################
     13
     14  __doc__="""Util
     15
     16  Utility functions for the Confmon Product
     17
     18  """
     19
     20  import types
     21  import re
     22  import string
     23
     24  from Products.ZenUtils.Exceptions import ZentinelException
     25
     26  from twisted.names.client import lookupPointer
     27
     28 -class IpAddressError(ZentinelException): pass
     29
     30
     31  isip = re.compile("^\d+\.\d+\.\d+\.\d+$").search
     32  """return match if this is an ip."""
     33
     34
     35 -def checkip(ip):
     36      """check that an ip is valid"""
     37      success = True
     38      if ip == '':
     39          success = False
     40      else:
     41          try:
     42              octs = ip.split('.')
     43          except:
     44              raise IpAddressError( '%s is not a dot delimited address' % ip )
     45          if len(octs) != 4:
     46              success = False
     47          else:
     48              for o in octs:
     49                  try:
     50                      if not (0 <= int(o) <= 255):
     51                          success = False
     52                  except:
     53                      success = False
     54      if not success:
     55          raise IpAddressError( "%s is an invalid address" % ip )
     56      return True
     57
     58
     59 -def numbip(ip):
     60      """convert a string ip to number"""
     61      checkip(ip)
     62      octs = ip.split('.')
     63      octs.reverse()
     64      i = 0L
     65      for j in range(len(octs)):
     66          i += (256l ** j) * int(octs[j])
     67      return i
     68
     69  _masks = (
     70      0x000000ffL,
     71      0x0000ff00L,
     72      0x00ff0000L,
     73      0xff000000L,
     74      )
     75
     76
     77 -def ipFromIpMask(ipmask):
     78      """get just the ip from an ip mask pair like 1.1.1.1/24"""
     79      return ipmask.split("/")[0]
     80
     81
     82 -def strip(ip):
     83      """convert a number ip to a string"""
     84      o = []
     85      for i in range(len(_masks)):
     86          t = ip & _masks[i]
     87          s = str(t >> (i*8))
     88          o.append(s)
     89      o.reverse()
     90      return '.'.join(o)
     91
     92
     93 -def hexToBits(hex):
     94      """convert hex number (0xff000000 of netbits to numeric netmask (8)"""
     95      return maskToBits(hexToMask(hex))
     96
     97
     98 -def hexToMask(hex):
     99      '''converts a netmask represented in hex to octets represented in
    100      decimal.  e.g. "0xffffff00" -> "255.255.255.0"'''
    101
    102      if hex.find('x') < 0:
    103          return "255.255.255.255"
    104
    105      hex = list(hex.lower().split('x')[1])
    106      octets = []
    107      while len(hex) > 0:
    108          snippit = list(hex.pop() + hex.pop())
    109          snippit.reverse()
    110          decimal = int(string.join(snippit, ''), 16)
    111          octets.append(str(decimal))
    112
    113      octets.reverse()
    114      return string.join(octets, '.')
    115
    116
    117 -def maskToBits(netmask):
    118      """convert string rep of netmask to number of bits"""
    119      if type(netmask) == types.StringType and netmask.find('.') > -1:
    120          test = 0xffffffffL
    121          if netmask[0]=='0': return 0
    122          masknumb = numbip(netmask)
    123          for i in range(32):
    124              if test == masknumb: return 32-i
    125              test = test - 2 ** i
    126          return None
    127      else:
    128          return int(netmask)
    129
    130
    131 -def bitsToMaskNumb(netbits):
    132      """convert integer number of netbits to string netmask"""
    133      masknumb = 0L
    134      netbits=int(netbits)
    135      for i in range(32-netbits, 32):
    136          masknumb += 2L ** i
    137      return masknumb
    138
    139
    140 -def bitsToMask(netbits):
    141      return strip(bitsToMaskNumb(netbits))
    142
    143
    144 -def getnet(ip, netmask):
    145      """get network address of ip as string netmask is in form 255.255.255.0"""
    146      checkip(ip)
    147      ip = numbip(ip)
    148      if 0 < int(netmask) <= 32:
    149          netmask = bitsToMaskNumb(netmask)
    150      else:
    151          checkip(netmask)
    152          netmask = numbip(netmask)
    153      return ip & netmask
    154
    155
    156 -def getnetstr(ip, netmask):
    157      """return network number as string"""
    158      return strip(getnet(ip, netmask))
    159
    160 -def asyncNameLookup(address, uselibcresolver = True):
    161      if uselibcresolver:
    162          # This is the most reliable way to do a lookup use it
    163          from twisted.internet import threads
    164          import socket
    165          return threads.deferToThread(lambda : socket.gethostbyaddr(address)[0])
    166      else:
    167          # There is a problem with this method because it will ignore /etc/hosts
    168          address = '.'.join(address.split('.')[::-1]) + '.in-addr.arpa'
    169          d = lookupPointer(address, [1,2,4])
    170          def ip(result):
    171              return str(result[0][0].payload.name)
    172          d.addCallback(ip)
    173          return d
    174
    175 -def asyncIpLookup(name):
    176      """
    177      Look up an IP based on the name passed in.  We use gethostbyname to make
    178      sure that we use /etc/hosts as mentioned above.
    179
    180      This hasn't been tested.
    181      """
    182      from twisted.internet import threads
    183      import socket
    184      return threads.deferToThread(lambda : socket.gethostbyname(name))
    185
    186
    187 -class InvalidIPRangeError(Exception):
    188      """
    189      Attempted to parse an invalid IP range.
    190      """
    191
    192 -def parse_iprange(iprange):
    193      """
    194      Turn a string specifying an IP range into a list of IPs.
    195
    196      @param iprange: The range string, in the format '10.0.0.a-b'
    197      @type iprange: str
    198
    199          >>> parse_iprange('10.0.0.1-5')
    200          ['10.0.0.1', '10.0.0.2', '10.0.0.3', '10.0.0.4', '10.0.0.5']
    201          >>> parse_iprange('10.0.0.1')
    202          ['10.0.0.1']
    203          >>> try: parse_iprange('10.0.0.1-2-3')
    204          ... except InvalidIPRangeError: print "Invalid"
    205          Invalid
    206
    207      """
    208      # Get the relevant octet
    209      net, octet = iprange.rsplit('.', 1)
    210      split = octet.split('-')
    211      if len(split) > 2: # Nothing we can do about this
    212          raise InvalidIPRangeError('%s is an invalid IP range.')
    213      elif len(split)==1: # A single IP was passed
    214          return [iprange]
    215      else:
    216          start, end = map(int, split)
    217          return ['%s.%s' % (net, x) for x in xrange(start, end+1)]
    218