1 """Implements (a subset of) Sun XDR -- eXternal Data Representation.

     2 

     3 See: RFC 1014

     4 

     5 """

     6 

     7 import struct

     8 try:

     9     from cStringIO import StringIO as _StringIO

    10 except ImportError:

    11     from StringIO import StringIO as _StringIO

    12 

    13 __all__ = ["Error", "Packer", "Unpacker", "ConversionError"]

    14 

    15 # exceptions

    16 class Error(Exception):

    17     """Exception class for this module. Use:

    18 

    19     except xdrlib.Error, var:

    20         # var has the Error instance for the exception

    21 

    22     Public ivars:

    23         msg -- contains the message

    24 

    25     """

    26     def __init__(self, msg):

    27         self.msg = msg

    28     def __repr__(self):

    29         return repr(self.msg)

    30     def __str__(self):

    31         return str(self.msg)

    32 

    33 

    34 class ConversionError(Error):

    35     pass

    36 

    37 

    38 

    39 class Packer:

    40     """Pack various data representations into a buffer."""

    41 

    42     def __init__(self):

    43         self.reset()

    44 

    45     def reset(self):

    46         self.__buf = _StringIO()

    47 

    48     def get_buffer(self):

    49         return self.__buf.getvalue()

    50     # backwards compatibility

    51     get_buf = get_buffer

    52 

    53     def pack_uint(self, x):

    54         self.__buf.write(struct.pack('>L', x))

    55 

    56     pack_int = pack_uint

    57     pack_enum = pack_int

    58 

    59     def pack_bool(self, x):

    60         if x: self.__buf.write('\0\0\0\1')

    61         else: self.__buf.write('\0\0\0\0')

    62 

    63     def pack_uhyper(self, x):

    64         self.pack_uint(x>>32 & 0xffffffffL)

    65         self.pack_uint(x & 0xffffffffL)

    66 

    67     pack_hyper = pack_uhyper

    68 

    69     def pack_float(self, x):

    70         try: self.__buf.write(struct.pack('>f', x))

    71         except struct.error, msg:

    72             raise ConversionError, msg

    73 

    74     def pack_double(self, x):

    75         try: self.__buf.write(struct.pack('>d', x))

    76         except struct.error, msg:

    77             raise ConversionError, msg

    78 

    79     def pack_fstring(self, n, s):

    80         if n < 0:

    81             raise ValueError, 'fstring size must be nonnegative'

    82         data = s[:n]

    83         n = ((n+3)//4)*4

    84         data = data + (n - len(data)) * '\0'

    85         self.__buf.write(data)

    86 

    87     pack_fopaque = pack_fstring

    88 

    89     def pack_string(self, s):

    90         n = len(s)

    91         self.pack_uint(n)

    92         self.pack_fstring(n, s)

    93 

    94     pack_opaque = pack_string

    95     pack_bytes = pack_string

    96 

    97     def pack_list(self, list, pack_item):

    98         for item in list:

    99             self.pack_uint(1)

   100             pack_item(item)

   101         self.pack_uint(0)

   102 

   103     def pack_farray(self, n, list, pack_item):

   104         if len(list) != n:

   105             raise ValueError, 'wrong array size'

   106         for item in list:

   107             pack_item(item)

   108 

   109     def pack_array(self, list, pack_item):

   110         n = len(list)

   111         self.pack_uint(n)

   112         self.pack_farray(n, list, pack_item)

   113 

   114 

   115 

   116 class Unpacker:

   117     """Unpacks various data representations from the given buffer."""

   118 

   119     def __init__(self, data):

   120         self.reset(data)

   121 

   122     def reset(self, data):

   123         self.__buf = data

   124         self.__pos = 0

   125 

   126     def get_position(self):

   127         return self.__pos

   128 

   129     def set_position(self, position):

   130         self.__pos = position

   131 

   132     def get_buffer(self):

   133         return self.__buf

   134 

   135     def done(self):

   136         if self.__pos < len(self.__buf):

   137             raise Error('unextracted data remains')

   138 

   139     def unpack_uint(self):

   140         i = self.__pos

   141         self.__pos = j = i+4

   142         data = self.__buf[i:j]

   143         if len(data) < 4:

   144             raise EOFError

   145         x = struct.unpack('>L', data)[0]

   146         try:

   147             return int(x)

   148         except OverflowError:

   149             return x

   150 

   151     def unpack_int(self):

   152         i = self.__pos

   153         self.__pos = j = i+4

   154         data = self.__buf[i:j]

   155         if len(data) < 4:

   156             raise EOFError

   157         return struct.unpack('>l', data)[0]

   158 

   159     unpack_enum = unpack_int

   160 

   161     def unpack_bool(self):

   162         return bool(self.unpack_int())

   163 

   164     def unpack_uhyper(self):

   165         hi = self.unpack_uint()

   166         lo = self.unpack_uint()

   167         return long(hi)<<32 | lo

   168 

   169     def unpack_hyper(self):

   170         x = self.unpack_uhyper()

   171         if x >= 0x8000000000000000L:

   172             x = x - 0x10000000000000000L

   173         return x

   174 

   175     def unpack_float(self):

   176         i = self.__pos

   177         self.__pos = j = i+4

   178         data = self.__buf[i:j]

   179         if len(data) < 4:

   180             raise EOFError

   181         return struct.unpack('>f', data)[0]

   182 

   183     def unpack_double(self):

   184         i = self.__pos

   185         self.__pos = j = i+8

   186         data = self.__buf[i:j]

   187         if len(data) < 8:

   188             raise EOFError

   189         return struct.unpack('>d', data)[0]

   190 

   191     def unpack_fstring(self, n):

   192         if n < 0:

   193             raise ValueError, 'fstring size must be nonnegative'

   194         i = self.__pos

   195         j = i + (n+3)//4*4

   196         if j > len(self.__buf):

   197             raise EOFError

   198         self.__pos = j

   199         return self.__buf[i:i+n]

   200 

   201     unpack_fopaque = unpack_fstring

   202 

   203     def unpack_string(self):

   204         n = self.unpack_uint()

   205         return self.unpack_fstring(n)

   206 

   207     unpack_opaque = unpack_string

   208     unpack_bytes = unpack_string

   209 

   210     def unpack_list(self, unpack_item):

   211         list = []

   212         while 1:

   213             x = self.unpack_uint()

   214             if x == 0: break

   215             if x != 1:

   216                 raise ConversionError, '0 or 1 expected, got %r' % (x,)

   217             item = unpack_item()

   218             list.append(item)

   219         return list

   220 

   221     def unpack_farray(self, n, unpack_item):

   222         list = []

   223         for i in range(n):

   224             list.append(unpack_item())

   225         return list

   226 

   227     def unpack_array(self, unpack_item):

   228         n = self.unpack_uint()

   229         return self.unpack_farray(n, unpack_item)

   230 

   231 

   232 # test suite

   233 def _test():

   234     p = Packer()

   235     packtest = [

   236         (p.pack_uint,    (9,)),

   237         (p.pack_bool,    (True,)),

   238         (p.pack_bool,    (False,)),

   239         (p.pack_uhyper,  (45L,)),

   240         (p.pack_float,   (1.9,)),

   241         (p.pack_double,  (1.9,)),

   242         (p.pack_string,  ('hello world',)),

   243         (p.pack_list,    (range(5), p.pack_uint)),

   244         (p.pack_array,   (['what', 'is', 'hapnin', 'doctor'], p.pack_string)),

   245         ]

   246     succeedlist = [1] * len(packtest)

   247     count = 0

   248     for method, args in packtest:

   249         print 'pack test', count,

   250         try:

   251             method(*args)

   252             print 'succeeded'

   253         except ConversionError, var:

   254             print 'ConversionError:', var.msg

   255             succeedlist[count] = 0

   256         count = count + 1

   257     data = p.get_buffer()

   258     # now verify

   259     up = Unpacker(data)

   260     unpacktest = [

   261         (up.unpack_uint,   (), lambda x: x == 9),

   262         (up.unpack_bool,   (), lambda x: x is True),

   263         (up.unpack_bool,   (), lambda x: x is False),

   264         (up.unpack_uhyper, (), lambda x: x == 45L),

   265         (up.unpack_float,  (), lambda x: 1.89 < x < 1.91),

   266         (up.unpack_double, (), lambda x: 1.89 < x < 1.91),

   267         (up.unpack_string, (), lambda x: x == 'hello world'),

   268         (up.unpack_list,   (up.unpack_uint,), lambda x: x == range(5)),

   269         (up.unpack_array,  (up.unpack_string,),

   270          lambda x: x == ['what', 'is', 'hapnin', 'doctor']),

   271         ]

   272     count = 0

   273     for method, args, pred in unpacktest:

   274         print 'unpack test', count,

   275         try:

   276             if succeedlist[count]:

   277                 x = method(*args)

   278                 print pred(x) and 'succeeded' or 'failed', ':', x

   279             else:

   280                 print 'skipping'

   281         except ConversionError, var:

   282             print 'ConversionError:', var.msg

   283         count = count + 1

   284 

   285 

   286 if __name__ == '__main__':

   287     _test()