sleekxmpp/plugins/xep_0047/stanza.py

   1 import re
   2 import base64
   3
   4 from sleekxmpp.util import bytes
   5 from sleekxmpp.exceptions import XMPPError
   6 from sleekxmpp.xmlstream import ElementBase
   7
   8
   9 VALID_B64 = re.compile(r'[A-Za-z0-9\+\/]*=*')
  10
  11
  12 def to_b64(data):
  13     return bytes(base64.b64encode(bytes(data))).decode('utf-8')
  14
  15
  16 def from_b64(data):
  17     return bytes(base64.b64decode(bytes(data)))
  18
  19
  20 class Open(ElementBase):
  21     name = 'open'
  22     namespace = 'http://jabber.org/protocol/ibb'
  23     plugin_attrib = 'ibb_open'
  24     interfaces = set(('block_size', 'sid', 'stanza'))
  25
  26     def get_block_size(self):
  27         return int(self._get_attr('block-size'))
  28
  29     def set_block_size(self, value):
  30         self._set_attr('block-size', str(value))
  31
  32     def del_block_size(self):
  33         self._del_attr('block-size')
  34
  35
  36 class Data(ElementBase):
  37     name = 'data'
  38     namespace = 'http://jabber.org/protocol/ibb'
  39     plugin_attrib = 'ibb_data'
  40     interfaces = set(('seq', 'sid', 'data'))
  41     sub_interfaces = set(['data'])
  42
  43     def get_seq(self):
  44         return int(self._get_attr('seq', '0'))
  45
  46     def set_seq(self, value):
  47         self._set_attr('seq', str(value))
  48
  49     def get_data(self):
  50         b64_data = self.xml.text.strip()
  51         if VALID_B64.match(b64_data).group() == b64_data:
  52             return from_b64(b64_data)
  53         else:
  54             raise XMPPError('not-acceptable')
  55
  56     def set_data(self, value):
  57         self.xml.text = to_b64(value)
  58
  59     def del_data(self):
  60         self.xml.text = ''
  61
  62
  63 class Close(ElementBase):
  64     name = 'close'
  65     namespace = 'http://jabber.org/protocol/ibb'
  66     plugin_attrib = 'ibb_close'
  67     interfaces = set(['sid'])