python/google/appengine/api/remote_socket/_remote_socket_addr.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright 2007 Google Inc.
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #
  17 """IP address helper functions.
  18
  19 This file contains alternative implementations of the following functions
  20 normally provided by the system:
  21
  22   - inet_(ntop|pton)
  23   - inet_(ntoa|aton)
  24   - (hton|ntoh)(sl)
  25 """
  26
  27 import errno
  28 import os
  29 import re
  30 import struct
  31 import sys
  32
  33
  34
  35 from google.appengine.api.remote_socket._remote_socket_error import *
  36
  37
  38 AF_UNSPEC = 0
  39 AF_UNIX = 1
  40 AF_INET = 2
  41 AF_INET6 = 30
  42
  43
  44 def _TypeName(obj):
  45   if obj is None:
  46     return 'None'
  47   return type(obj).__name__
  48
  49
  50 def _LongestRun(seq, value):
  51   """Finds the longest run (i.e. repeated adjacent values) of value in seq.
  52
  53   Args:
  54     seq: A sequence to scan for runs.
  55     value: A value to scan for in seq.
  56
  57   Returns:
  58     A tuple of the offset and run length for the longest run of value within
  59     seq, or (-1, 0) if value does not occur within seq.
  60   """
  61   off = -1
  62   max_run = 0
  63   i = 0
  64   while i < len(seq):
  65     run = 0
  66     while i + run < len(seq) and seq[i+run] == value:
  67       run += 1
  68     if run > max_run:
  69       off = i
  70       max_run = run
  71     i += 1 + run
  72   return off, max_run
  73
  74
  75 def _GetBase(text):
  76   if text[:2].lower() == '0x':
  77     return 16
  78   elif text[:1] == '0':
  79     return 8
  80   return 10
  81
  82
  83 def ntohs(integer):
  84   """ntohs(integer) -> integer
  85
  86   Convert a 16-bit integer from network to host byte order.
  87   """
  88   if sys.byteorder == 'big':
  89     return integer
  90   if not isinstance(integer, (int, long)):
  91     raise TypeError('an integer is required')
  92   if integer < 0:
  93     raise OverflowError("can't convert negative number to unsigned long")
  94   if integer >= (1<<16):
  95     raise OverflowError('signed integer is greater than maximum')
  96   return int(
  97       ((integer&0xff00)>>8)|
  98       ((integer&0x00ff)<<8))
  99
 100
 101 def ntohl(integer):
 102   """ntohl(integer) -> integer
 103
 104   Convert a 32-bit integer from network to host byte order.
 105   """
 106   if sys.byteorder == 'big':
 107     return integer
 108   if not isinstance(integer, (int, long)):
 109     raise TypeError('expected int/long, %s found' % _TypeName(integer))
 110   if integer < 0:
 111     raise OverflowError('can\'t convert negative number to unsigned long')
 112   if integer >= (1<<32):
 113     raise OverflowError('long int larger than 32 bits')
 114   return int(
 115       ((integer&0xff000000)>>24)|
 116       ((integer&0x00ff0000)>>8)|
 117       ((integer&0x0000ff00)<<8)|
 118       ((integer&0x000000ff)<<24))
 119
 120
 121 def htons(integer):
 122   """htons(integer) -> integer
 123
 124   Convert a 16-bit integer from host to network byte order.
 125   """
 126   return ntohs(integer)
 127
 128
 129 def htonl(integer):
 130   """htonl(integer) -> integer
 131
 132   Convert a 32-bit integer from host to network byte order.
 133   """
 134   return ntohl(integer)
 135
 136
 137 def inet_aton(ip_string):
 138   """inet_aton(string) -> packed 32-bit IP representation
 139
 140   Convert an IP address in string format (123.45.67.89) to the 32-bit packed
 141   binary format used in low-level network functions.
 142   """
 143
 144   if not isinstance(ip_string, basestring):
 145     raise error('inet_aton() argument 1 must be string, not %s' %
 146                 _TypeName(ip_string))
 147   try:
 148     ret = 0
 149     bits = 32
 150     parts = [int(s, _GetBase(s)) for s in ip_string.split('.')]
 151     assert len(parts) >= 1 and len(parts) <= 4
 152     for n in parts[:-1]:
 153       assert n >= 0 and n <= 0xff
 154       bits -= 8
 155       ret |= ((n & 0xff) << bits)
 156     assert parts[-1] >= 0 and parts[-1] < (1 << bits)
 157     ret |= parts[-1]
 158     return struct.pack('!L', ret)
 159   except:
 160     raise error('illegal IP address string passed to inet_aton')
 161
 162
 163 def inet_ntoa(packed_ip):
 164   """inet_ntoa(packed_ip) -> ip_address_string
 165
 166   Convert an IP address from 32-bit packed binary format to string format
 167   """
 168
 169   if not isinstance(packed_ip, basestring):
 170     raise TypeError('inet_ntoa() argument 1 must be string or read-only '
 171                     'buffer, not %s' % _TypeName(packed_ip))
 172   if len(packed_ip) != 4:
 173     raise error('packed IP wrong length for inet_ntoa')
 174   return '%u.%u.%u.%u' % struct.unpack('4B', packed_ip)
 175
 176
 177 def inet_pton(af, ip):
 178   """inet_pton(af, ip) -> packed IP address string
 179
 180   Convert an IP address from string format to a packed string suitable
 181   for use with low-level network functions.
 182   """
 183
 184   if not isinstance(af, (int, long)):
 185     raise TypeError('an integer is required')
 186   if not isinstance(ip, basestring):
 187     raise TypeError('inet_pton() argument 2 must be string, not %s' %
 188                     _TypeName(ip))
 189   if af == AF_INET:
 190     parts = ip.split('.')
 191     if len(parts) != 4:
 192       raise error('illegal IP address string passed to inet_pton')
 193     ret = 0
 194     bits = 32
 195     for part in parts:
 196       if not re.match(r'^(0|[1-9]\d*)$', part) or int(part) > 0xff:
 197         raise error('illegal IP address string passed to inet_pton')
 198       bits -= 8
 199       ret |= ((int(part) & 0xff) << bits)
 200     return struct.pack('!L', ret)
 201   elif af == AF_INET6:
 202     parts = ip.split(':')
 203
 204     if '.' in parts[-1]:
 205       ipv4_shorts = struct.unpack('!2H', inet_pton(AF_INET, parts[-1]))
 206       parts[-1:] = [hex(n)[2:] for n in ipv4_shorts]
 207
 208     if '' in parts:
 209       if len(parts) == 1 or len(parts) >= 8:
 210         raise error('illegal IP address string passed to inet_pton')
 211
 212       idx = parts.index('')
 213       count = parts.count('')
 214       pad = ['0']*(count+(8-len(parts)))
 215
 216       if count == len(parts) == 3:
 217         parts = pad
 218       elif count == 2 and parts[0:2] == ['', '']:
 219         parts[0:2] = pad
 220       elif count == 2 and parts[-2:] == ['', '']:
 221         parts[-2:] = pad
 222       elif count == 1:
 223         parts[idx:idx+1] = pad
 224       else:
 225         raise error('illegal IP address string passed to inet_pton')
 226     if (len(parts) != 8 or
 227         [x for x in parts if not re.match(r'^[0-9A-Fa-f]{1,4}$', x)]):
 228       raise error('illegal IP address string passed to inet_pton')
 229     return struct.pack('!8H', *[int(x, 16) for x in parts])
 230   else:
 231     raise error(errno.EAFNOSUPPORT, os.strerror(errno.EAFNOSUPPORT))
 232
 233
 234 def inet_ntop(af, packed_ip):
 235   """inet_ntop(af, packed_ip) -> string formatted IP address
 236
 237   Convert a packed IP address of the given family to string format.
 238   """
 239   if not isinstance(af, (int, long)):
 240     raise TypeError('an integer is required')
 241   if not isinstance(packed_ip, basestring):
 242     raise TypeError('inet_ntop() argument 2 must be string or read-only '
 243                     'buffer, not %s' % _TypeName(packed_ip))
 244   if af == AF_INET:
 245     if len(packed_ip) != 4:
 246       raise ValueError('invalid length of packed IP address string')
 247     return '%u.%u.%u.%u' % struct.unpack('4B', packed_ip)
 248   elif af == AF_INET6:
 249     if len(packed_ip) != 16:
 250       raise ValueError('invalid length of packed IP address string')
 251     parts = [hex(n)[2:] for n in struct.unpack('!8H', packed_ip)]
 252
 253     if (':'.join(parts[:-2]) in ('0:0:0:0:0:0', '0:0:0:0:0:ffff') and
 254         ':'.join(parts) not in ('0:0:0:0:0:0:0:0', '0:0:0:0:0:0:0:1')):
 255       parts[-2:] = [inet_ntop(AF_INET, packed_ip[-4:])]
 256     off, run = _LongestRun(parts, '0')
 257     if run >= 2:
 258       pad = ['']
 259       if off == 0:
 260         pad.append('')
 261       if off + run == len(parts):
 262         pad.append('')
 263       parts[off:off+run] = pad
 264     return ':'.join(parts)
 265   else:
 266     raise ValueError('unknown address family %u' % af)