Package netaddr :: Module fallback |
[hide private]
[frames] | no frames]

# Source Code for Module netaddr.fallback

      1  #!/usr/bin/env python
      2  #-----------------------------------------------------------------------------
      3  #   Copyright (c) 2008-2009, David P. D. Moss. All rights reserved.
      4  #
      5  #   Released under the BSD license. See the LICENSE file for details.
      6  #-----------------------------------------------------------------------------
      7  """
      8  A module containing pure-Python implementations of certain socket module
      9  constants and functions. Useful on some platforms or Python interpreters where
     10  they may be missing for various reasons.
     11  """
     12  from struct import unpack as _unpack, pack as _pack
     13
     14  AF_UNSPEC = 0
     15  AF_INET = 2
     16  AF_INET6 = 10
     17
     18  #-----------------------------------------------------------------------------
     19 -def inet_ntoa(packed_ip):
     20      """
     21      Convert an IP address from 32-bit packed binary format to string format.
     22      """
     23      if len(packed_ip) != 4 or not hasattr(packed_ip, 'split'):
     24          raise ValueError('invalid length of packed IP address string')
     25      return '%d.%d.%d.%d' % _unpack('4B', packed_ip)
     26
     27  #-----------------------------------------------------------------------------
     28 -def inet_aton(ip_string):
     29      """
     30      Convert an IP address in string format (123.45.67.89) to the 32-bit packed
     31      binary format used in low-level network functions.
     32      """
     33      if hasattr(ip_string, 'split'):
     34          invalid_addr = ValueError('illegal IP address string %r' % ip_string)
     35          #   Support for hexadecimal and octal octets.
     36          tokens = []
     37          base = 10
     38          for token in ip_string.split('.'):
     39              if token.startswith('0x'):
     40                  base = 16
     41              elif token.startswith('0') and len(token) > 1:
     42                  base = 8
     43              elif token == '':
     44                  continue
     45              try:
     46                  tokens.append(int(token, base))
     47              except ValueError:
     48                  raise invalid_addr
     49
     50          #   Zero fill missing octets.
     51          num_tokens = len(tokens)
     52          if num_tokens < 4:
     53              fill_tokens = [0] * (4 - num_tokens)
     54              if num_tokens > 1:
     55                  end_token = tokens.pop()
     56                  tokens = tokens + fill_tokens + [end_token]
     57              else:
     58                  tokens = tokens + fill_tokens
     59
     60          #   Pack octets.
     61          if len(tokens) == 4:
     62              words = []
     63              for token in tokens:
     64                  if (token >> 8) != 0:
     65                      raise invalid_addr
     66                  words.append(chr(token))
     67              return ''.join(words)
     68          else:
     69              raise invalid_addr
     70
     71      raise ValueError('argument should be a string, not %s' % type(ip_string))
     72
     73  #-----------------------------------------------------------------------------
     74 -def _compact_ipv6_tokens(tokens):
     75      new_tokens = []
     76
     77      positions = []
     78      within_run = False
     79      start_index = None
     80      num_tokens = 0
     81
     82      #   Discover all runs of zeros.
     83      for idx, token in enumerate(tokens):
     84          if token == '0':
     85              within_run = True
     86              if start_index is None:
     87                  start_index = idx
     88              num_tokens += 1
     89          else:
     90              if num_tokens > 1:
     91                  positions.append((num_tokens, start_index))
     92              within_run = False
     93              start_index = None
     94              num_tokens = 0
     95
     96          new_tokens.append(token)
     97
     98      #   Store any position not saved before loop exit.
     99      if num_tokens > 1:
    100          positions.append((num_tokens, start_index))
    101
    102      #   Replace first longest run with an empty string.
    103      if len(positions) != 0:
    104          #   Locate longest, left-most run of zeros.
    105          positions.sort(lambda x, y: cmp(x[1], y[1]))
    106          best_position = positions[0]
    107          for position in positions:
    108              if position[0] > best_position[0]:
    109                  best_position = position
    110          #   Replace chosen zero run.
    111          (length, start_idx) = best_position
    112          new_tokens = new_tokens[0:start_idx] + [''] + \
    113                       new_tokens[start_idx+length:]
    114
    115          #   Add start and end blanks so join creates '::'.
    116          if new_tokens[0] == '':
    117              new_tokens.insert(0, '')
    118
    119          if new_tokens[-1] == '':
    120              new_tokens.append('')
    121
    122      return new_tokens
    123
    124  #-----------------------------------------------------------------------------
    125 -def inet_ntop(af, packed_ip):
    126      """Convert an packed IP address of the given family to string format."""
    127      if af == AF_INET:
    128          #   IPv4.
    129          return inet_ntoa(packed_ip)
    130      elif af == AF_INET6:
    131          #   IPv6.
    132          if len(packed_ip) != 16 or not hasattr(packed_ip, 'split'):
    133              raise ValueError('invalid length of packed IP address string')
    134
    135          tokens = ['%x' % i for i in _unpack('>8H', packed_ip)]
    136
    137          #   Convert packed address to an integer value.
    138          words = list(_unpack('>8H', packed_ip))
    139          int_val = 0
    140          for i, num in enumerate(reversed(words)):
    141              word = num
    142              word = word << 16 * i
    143              int_val = int_val | word
    144
    145          if 0xffff < int_val <= 0xffffffff or int_val >> 32 == 0xffff:
    146              #   IPv4 compatible / mapped IPv6.
    147              packed_ipv4 = _pack('>2H', *[int(i, 16) for i in tokens[-2:]])
    148              ipv4_str = inet_ntoa(packed_ipv4)
    149              tokens = tokens[0:-2] + [ipv4_str]
    150
    151          return ':'.join(_compact_ipv6_tokens(tokens))
    152      else:
    153          raise ValueError('unknown address family %d' % af)
    154
    155  #-----------------------------------------------------------------------------
    156 -def inet_pton(af, ip_string):
    157      """
    158      Convert an IP address from string format to a packed string suitable for
    159      use with low-level network functions.
    160      """
    161      if af == AF_INET:
    162          #   IPv4.
    163          return inet_aton(ip_string)
    164      elif af == AF_INET6:
    165          invalid_addr = ValueError('illegal IP address string %r' % ip_string)
    166          #   IPv6.
    167          values = []
    168
    169          if not hasattr(ip_string, 'split'):
    170              raise invalid_addr
    171
    172          if '::' in ip_string:
    173              if ip_string == '::':
    174                  #   Unspecified address.
    175                  return '\x00'*16
    176              #   IPv6 compact mode.
    177              try:
    178                  prefix, suffix = ip_string.split('::')
    179              except ValueError:
    180                  raise invalid_addr
    181
    182              l_prefix = []
    183              l_suffix = []
    184
    185              if prefix != '':
    186                  l_prefix = prefix.split(':')
    187
    188              if suffix != '':
    189                  l_suffix = suffix.split(':')
    190
    191              #   IPv6 compact IPv4 compatibility mode.
    192              if len(l_suffix) and '.' in l_suffix[-1]:
    193                  ipv4_str = inet_aton(l_suffix.pop())
    194                  l_suffix.append('%x' % _unpack('>H', ipv4_str[0:2])[0])
    195                  l_suffix.append('%x' % _unpack('>H', ipv4_str[2:4])[0])
    196
    197              token_count = len(l_prefix) + len(l_suffix)
    198
    199              if not 0 <= token_count <= 8 - 1:
    200                  raise invalid_addr
    201
    202              gap_size = 8 - ( len(l_prefix) + len(l_suffix) )
    203
    204              values = [_pack('>H', int(i, 16)) for i in l_prefix] \
    205                     + ['\x00\x00' for i in range(gap_size)] \
    206                     + [_pack('>H', int(i, 16)) for i in l_suffix]
    207              try:
    208                  for token in l_prefix + l_suffix:
    209                      word = int(token, 16)
    210                      if not 0 <= word <= 0xffff:
    211                          raise invalid_addr
    212              except ValueError:
    213                  raise invalid_addr
    214          else:
    215              #   IPv6 verbose mode.
    216              if ':' in ip_string:
    217                  tokens = ip_string.split(':')
    218
    219                  if '.' in ip_string:
    220                      ipv6_prefix = tokens[:-1]
    221                      if ipv6_prefix[:-1] != ['0', '0', '0', '0', '0']:
    222                          raise invalid_addr
    223
    224                      if ipv6_prefix[-1].lower() not in ('0', 'ffff'):
    225                          raise invalid_addr
    226
    227                      #   IPv6 verbose IPv4 compatibility mode.
    228                      if len(tokens) != 7:
    229                          raise invalid_addr
    230
    231                      ipv4_str = inet_aton(tokens.pop())
    232                      tokens.append('%x' % _unpack('>H', ipv4_str[0:2])[0])
    233                      tokens.append('%x' % _unpack('>H', ipv4_str[2:4])[0])
    234
    235                      values = [_pack('>H', int(i, 16)) for i in tokens]
    236                  else:
    237                      #   IPv6 verbose mode.
    238                      if len(tokens) != 8:
    239                          raise invalid_addr
    240                  try:
    241                      tokens = [int(token, 16) for token in tokens]
    242                      for token in tokens:
    243                          if not 0 <= token <= 0xffff:
    244                              raise invalid_addr
    245
    246                  except ValueError:
    247                      raise invalid_addr
    248
    249                  values = [_pack('>H', i) for i in tokens]
    250              else:
    251                  raise invalid_addr
    252
    253          return ''.join(values)
    254      else:
    255          raise ValueError('Unknown address family %d' % af)
    256