qweechat/weechat/protocol.py

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