qweechat/weechat/protocol.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3 #
   4 # protocol.py - decode binary messages received from WeeChat/relay
   5 #
   6 # Copyright (C) 2011-2014 Sébastien Helleu <flashcode@flashtux.org>
   7 #
   8 # This file is part of QWeeChat, a Qt remote GUI for WeeChat.
   9 #
  10 # QWeeChat is free software; you can redistribute it and/or modify
  11 # it under the terms of the GNU General Public License as published by
  12 # the Free Software Foundation; either version 3 of the License, or
  13 # (at your option) any later version.
  14 #
  15 # QWeeChat is distributed in the hope that it will be useful,
  16 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  17 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  18 # GNU General Public License for more details.
  19 #
  20 # You should have received a copy of the GNU General Public License
  21 # along with QWeeChat.  If not, see <http://www.gnu.org/licenses/>.
  22 #
  23
  24 #
  25 # For info about protocol and format of messages, please read document
  26 # "WeeChat Relay Protocol", available at:  http://weechat.org/doc/
  27 #
  28 # History:
  29 #
  30 # 2011-11-23, Sébastien Helleu <flashcode@flashtux.org>:
  31 #     start dev
  32 #
  33
  34 import collections
  35 import struct
  36 import zlib
  37
  38 if hasattr(collections, 'OrderedDict'):
  39     # python >= 2.7
  40     class WeechatDict(collections.OrderedDict):
  41         def __str__(self):
  42             return '{%s}' % ', '.join(
  43                 ['%s: %s' % (repr(key), repr(self[key])) for key in self])
  44 else:
  45     # python <= 2.6
  46     WeechatDict = dict
  47
  48
  49 class WeechatObject:
  50     def __init__(self, objtype, value, separator='\n'):
  51         self.objtype = objtype
  52         self.value = value
  53         self.separator = separator
  54         self.indent = '  ' if separator == '\n' else ''
  55         self.separator1 = '\n%s' % self.indent if separator == '\n' else ''
  56
  57     def _str_value(self, v):
  58         if type(v) is str and v is not None:
  59             return '\'%s\'' % v
  60         return str(v)
  61
  62     def _str_value_hdata(self):
  63         lines = ['%skeys: %s%s%spath: %s' % (self.separator1,
  64                                              str(self.value['keys']),
  65                                              self.separator,
  66                                              self.indent,
  67                                              str(self.value['path']))]
  68         for i, item in enumerate(self.value['items']):
  69             lines.append('  item %d:%s%s' % (
  70                 (i + 1), self.separator,
  71                 self.separator.join(
  72                     ['%s%s: %s' % (self.indent * 2, key,
  73                                    self._str_value(value))
  74                      for key, value in item.items()])))
  75         return '\n'.join(lines)
  76
  77     def _str_value_infolist(self):
  78         lines = ['%sname: %s' % (self.separator1, self.value['name'])]
  79         for i, item in enumerate(self.value['items']):
  80             lines.append('  item %d:%s%s' % (
  81                 (i + 1), self.separator,
  82                 self.separator.join(
  83                     ['%s%s: %s' % (self.indent * 2, key,
  84                                    self._str_value(value))
  85                      for key, value in item.items()])))
  86         return '\n'.join(lines)
  87
  88     def _str_value_other(self):
  89         return self._str_value(self.value)
  90
  91     def __str__(self):
  92         self._obj_cb = {'hda': self._str_value_hdata,
  93                         'inl': self._str_value_infolist,
  94                         }
  95         return '%s: %s' % (self.objtype,
  96                            self._obj_cb.get(self.objtype,
  97                                             self._str_value_other)())
  98
  99
 100 class WeechatObjects(list):
 101     def __init__(self, separator='\n'):
 102         self.separator = separator
 103
 104     def __str__(self):
 105         return self.separator.join([str(obj) for obj in self])
 106
 107
 108 class WeechatMessage:
 109     def __init__(self, size, size_uncompressed, compression, uncompressed,
 110                  msgid, objects):
 111         self.size = size
 112         self.size_uncompressed = size_uncompressed
 113         self.compression = compression
 114         self.uncompressed = uncompressed
 115         self.msgid = msgid
 116         self.objects = objects
 117
 118     def __str__(self):
 119         if self.compression != 0:
 120             return 'size: %d/%d (%d%%), id=\'%s\', objects:\n%s' % (
 121                 self.size, self.size_uncompressed,
 122                 100 - ((self.size * 100) // self.size_uncompressed),
 123                 self.msgid, self.objects)
 124         else:
 125             return 'size: %d, id=\'%s\', objects:\n%s' % (self.size,
 126                                                           self.msgid,
 127                                                           self.objects)
 128
 129
 130 class Protocol:
 131     """Decode binary message received from WeeChat/relay."""
 132
 133     def __init__(self):
 134         self._obj_cb = {'chr': self._obj_char,
 135                         'int': self._obj_int,
 136                         'lon': self._obj_long,
 137                         'str': self._obj_str,
 138                         'buf': self._obj_buffer,
 139                         'ptr': self._obj_ptr,
 140                         'tim': self._obj_time,
 141                         'htb': self._obj_hashtable,
 142                         'hda': self._obj_hdata,
 143                         'inf': self._obj_info,
 144                         'inl': self._obj_infolist,
 145                         'arr': self._obj_array,
 146                         }
 147
 148     def _obj_type(self):
 149         """Read type in data (3 chars)."""
 150         if len(self.data) < 3:
 151             self.data = ''
 152             return ''
 153         objtype = str(self.data[0:3])
 154         self.data = self.data[3:]
 155         return objtype
 156
 157     def _obj_len_data(self, length_size):
 158         """Read length (1 or 4 bytes), then value with this length."""
 159         if len(self.data) < length_size:
 160             self.data = ''
 161             return None
 162         if length_size == 1:
 163             length = struct.unpack('B', self.data[0:1])[0]
 164             self.data = self.data[1:]
 165         else:
 166             length = self._obj_int()
 167         if length < 0:
 168             return None
 169         if length > 0:
 170             value = self.data[0:length]
 171             self.data = self.data[length:]
 172         else:
 173             value = ''
 174         return value
 175
 176     def _obj_char(self):
 177         """Read a char in data."""
 178         if len(self.data) < 1:
 179             return 0
 180         value = struct.unpack('b', self.data[0:1])[0]
 181         self.data = self.data[1:]
 182         return value
 183
 184     def _obj_int(self):
 185         """Read an integer in data (4 bytes)."""
 186         if len(self.data) < 4:
 187             self.data = ''
 188             return 0
 189         value = struct.unpack('>i', self.data[0:4])[0]
 190         self.data = self.data[4:]
 191         return value
 192
 193     def _obj_long(self):
 194         """Read a long integer in data (length on 1 byte + value as string)."""
 195         value = self._obj_len_data(1)
 196         if value is None:
 197             return None
 198         return int(str(value))
 199
 200     def _obj_str(self):
 201         """Read a string in data (length on 4 bytes + content)."""
 202         value = self._obj_len_data(4)
 203         if value is None:
 204             return None
 205         return str(value)
 206
 207     def _obj_buffer(self):
 208         """Read a buffer in data (length on 4 bytes + data)."""
 209         return self._obj_len_data(4)
 210
 211     def _obj_ptr(self):
 212         """Read a pointer in data (length on 1 byte + value as string)."""
 213         value = self._obj_len_data(1)
 214         if value is None:
 215             return None
 216         return '0x%s' % str(value)
 217
 218     def _obj_time(self):
 219         """Read a time in data (length on 1 byte + value as string)."""
 220         value = self._obj_len_data(1)
 221         if value is None:
 222             return None
 223         return int(str(value))
 224
 225     def _obj_hashtable(self):
 226         """
 227         Read a hashtable in data
 228         (type for keys + type for values + count + items).
 229         """
 230         type_keys = self._obj_type()
 231         type_values = self._obj_type()
 232         count = self._obj_int()
 233         hashtable = WeechatDict()
 234         for i in range(0, count):
 235             key = self._obj_cb[type_keys]()
 236             value = self._obj_cb[type_values]()
 237             hashtable[key] = value
 238         return hashtable
 239
 240     def _obj_hdata(self):
 241         """Read a hdata in data."""
 242         path = self._obj_str()
 243         keys = self._obj_str()
 244         count = self._obj_int()
 245         list_path = path.split('/')
 246         list_keys = keys.split(',')
 247         keys_types = []
 248         dict_keys = WeechatDict()
 249         for key in list_keys:
 250             items = key.split(':')
 251             keys_types.append(items)
 252             dict_keys[items[0]] = items[1]
 253         items = []
 254         for i in range(0, count):
 255             item = WeechatDict()
 256             item['__path'] = []
 257             pointers = []
 258             for p in range(0, len(list_path)):
 259                 pointers.append(self._obj_ptr())
 260             for key, objtype in keys_types:
 261                 item[key] = self._obj_cb[objtype]()
 262             item['__path'] = pointers
 263             items.append(item)
 264         return {'path': list_path,
 265                 'keys': dict_keys,
 266                 'count': count,
 267                 'items': items,
 268                 }
 269
 270     def _obj_info(self):
 271         """Read an info in data."""
 272         name = self._obj_str()
 273         value = self._obj_str()
 274         return (name, value)
 275
 276     def _obj_infolist(self):
 277         """Read an infolist in data."""
 278         name = self._obj_str()
 279         count_items = self._obj_int()
 280         items = []
 281         for i in range(0, count_items):
 282             count_vars = self._obj_int()
 283             variables = WeechatDict()
 284             for v in range(0, count_vars):
 285                 var_name = self._obj_str()
 286                 var_type = self._obj_type()
 287                 var_value = self._obj_cb[var_type]()
 288                 variables[var_name] = var_value
 289             items.append(variables)
 290         return {'name': name, 'items': items}
 291
 292     def _obj_array(self):
 293         """Read an array of values in data."""
 294         type_values = self._obj_type()
 295         count_values = self._obj_int()
 296         values = []
 297         for i in range(0, count_values):
 298             values.append(self._obj_cb[type_values]())
 299         return values
 300
 301     def decode(self, data, separator='\n'):
 302         """Decode binary data and return list of objects."""
 303         self.data = data
 304         size = len(self.data)
 305         size_uncompressed = size
 306         uncompressed = None
 307         # uncompress data (if it is compressed)
 308         compression = struct.unpack('b', self.data[4:5])[0]
 309         if compression:
 310             uncompressed = zlib.decompress(self.data[5:])
 311             size_uncompressed = len(uncompressed) + 5
 312             uncompressed = '%s%s%s' % (struct.pack('>i', size_uncompressed),
 313                                        struct.pack('b', 0), uncompressed)
 314             self.data = uncompressed
 315         else:
 316             uncompressed = self.data[:]
 317         # skip length and compression flag
 318         self.data = self.data[5:]
 319         # read id
 320         msgid = self._obj_str()
 321         if msgid is None:
 322             msgid = ''
 323         # read objects
 324         objects = WeechatObjects(separator=separator)
 325         while len(self.data) > 0:
 326             objtype = self._obj_type()
 327             value = self._obj_cb[objtype]()
 328             objects.append(WeechatObject(objtype, value, separator=separator))
 329         return WeechatMessage(size, size_uncompressed, compression,
 330                               uncompressed, msgid, objects)
 331
 332
 333 def hex_and_ascii(data, bytes_per_line=10):
 334     """Convert a QByteArray to hex + ascii output."""
 335     num_lines = ((len(data) - 1) // bytes_per_line) + 1
 336     if num_lines == 0:
 337         return ''
 338     lines = []
 339     for i in range(0, num_lines):
 340         str_hex = []
 341         str_ascii = []
 342         for char in data[i*bytes_per_line:(i*bytes_per_line)+bytes_per_line]:
 343             byte = struct.unpack('B', char)[0]
 344             str_hex.append('%02X' % int(byte))
 345             if byte >= 32 and byte <= 127:
 346                 str_ascii.append(char)
 347             else:
 348                 str_ascii.append('.')
 349         fmt = '%%-%ds %%s' % ((bytes_per_line * 3) - 1)
 350         lines.append(fmt % (' '.join(str_hex), ''.join(str_ascii)))
 351     return '\n'.join(lines)