diff options
| -rwxr-xr-x | setup.py | 1 | ||||
| -rw-r--r-- | sleekxmpp/plugins/xep_0047/__init__.py | 13 | ||||
| -rw-r--r-- | sleekxmpp/plugins/xep_0047/ibb.py | 141 | ||||
| -rw-r--r-- | sleekxmpp/plugins/xep_0047/stanza.py | 63 | ||||
| -rw-r--r-- | sleekxmpp/plugins/xep_0047/stream.py | 137 |
5 files changed, 355 insertions, 0 deletions
@@ -59,6 +59,7 @@ packages = [ 'sleekxmpp', 'sleekxmpp/plugins/xep_0009/stanza', 'sleekxmpp/plugins/xep_0030', 'sleekxmpp/plugins/xep_0030/stanza', + 'sleekxmpp/plugins/xep_0047', 'sleekxmpp/plugins/xep_0050', 'sleekxmpp/plugins/xep_0059', 'sleekxmpp/plugins/xep_0060', diff --git a/sleekxmpp/plugins/xep_0047/__init__.py b/sleekxmpp/plugins/xep_0047/__init__.py new file mode 100644 index 00000000..5a6487cc --- /dev/null +++ b/sleekxmpp/plugins/xep_0047/__init__.py @@ -0,0 +1,13 @@ +""" + SleekXMPP: The Sleek XMPP Library + Copyright (C) 2012 Nathanael C. Fritz, Lance J.T. Stout + This file is part of SleekXMPP. + + See the file LICENSE for copying permission. +""" + +from sleekxmpp.plugins.xep_0047 import stanza +from sleekxmpp.plugins.xep_0047.stanza import Open, Close, Data +from sleekxmpp.plugins.xep_0047.stream import IBBytestream +from sleekxmpp.plugins.xep_0047.ibb import xep_0047 + diff --git a/sleekxmpp/plugins/xep_0047/ibb.py b/sleekxmpp/plugins/xep_0047/ibb.py new file mode 100644 index 00000000..c4f46e2f --- /dev/null +++ b/sleekxmpp/plugins/xep_0047/ibb.py @@ -0,0 +1,141 @@ +import uuid +import logging +import threading + +from sleekxmpp import Message, Iq +from sleekxmpp.exceptions import XMPPError, IqError, IqTimeout +from sleekxmpp.xmlstream.handler import Callback +from sleekxmpp.xmlstream.matcher import StanzaPath +from sleekxmpp.xmlstream import register_stanza_plugin +from sleekxmpp.plugins.base import base_plugin +from sleekxmpp.plugins.xep_0047 import stanza, Open, Close, Data, IBBytestream + + +log = logging.getLogger(__name__) + + +class xep_0047(base_plugin): + + def plugin_init(self): + self.xep = '0047' + self.description = 'In-band Bytestreams' + self.stanza = stanza + + self.streams = {} + self.pending_streams = {} + self.pending_close_streams = {} + self._stream_lock = threading.Lock() + + self.max_block_size = self.config.get('max_block_size', 8192) + self.window_size = self.config.get('window_size', 1) + self.auto_accept = self.config.get('auto_accept', True) + self.accept_stream = self.config.get('accept_stream', None) + + register_stanza_plugin(Iq, Open) + register_stanza_plugin(Iq, Close) + register_stanza_plugin(Iq, Data) + + self.xmpp.register_handler(Callback( + 'IBB Open', + StanzaPath('iq@type=set/ibb_open'), + self._handle_open_request)) + + self.xmpp.register_handler(Callback( + 'IBB Close', + StanzaPath('iq@type=set/ibb_close'), + self._handle_close)) + + self.xmpp.register_handler(Callback( + 'IBB Data', + StanzaPath('iq@type=set/ibb_data'), + self._handle_data)) + + def post_init(self): + self.xmpp['xep_0030'].add_feature('http://jabber.org/protocol/ibb') + + def _accept_stream(self, iq): + if self.accept_stream is not None: + return self.accept_stream(iq) + if self.auto_accept: + if iq['ibb_open']['block_size'] <= self.max_block_size: + return True + return False + + def open_stream(self, jid, block_size=4096, sid=None, window=1, + ifrom=None, block=True, timeout=None, callback=None): + if not block and callback is not None: + callback = lambda resp: self._handle_opened_stream(resp, callback) + + if sid is None: + sid = str(uuid.uuid4()) + + iq = self.xmpp.Iq() + iq['type'] = 'set' + iq['to'] = jid + iq['from'] = ifrom + iq['ibb_open']['block_size'] = block_size + iq['ibb_open']['sid'] = sid + iq['ibb_open']['stanza'] = 'iq' + + stream = IBBytestream(self.xmpp, sid, size, + iq['to'], iq['from'], window) + + with self._stream_lock: + self.pending_streams[iq['id']] = stream + + resp = iq.send(block=block, timeout=timeout, callback=callback) + if block: + self._handle_opened_stream(resp) + return stream + + def _handle_opened_stream(self, iq, callback=None): + if iq['type'] == 'result': + with self._stream_lock: + stream = self.pending_streams[iq['id']] + stream.sender = iq['to'] + stream.receiver = iq['from'] + stream.stream_started.set() + self.streams[stream.sid] = stream + self.xmpp.event('ibb_stream_start', stream) + + if callback is not None: + callback(iq) + + with self._stream_lock: + if iq['id'] in self.pending_streams: + del self.pending_streams[iq['id']] + + def _handle_open_request(self, iq): + sid = iq['ibb_open']['sid'] + size = iq['ibb_open']['block_size'] + if not self._accept_stream(iq): + raise XMPPError('not-acceptable') + + if size > self.max_block_size: + raise XMPPError('resource-constraint') + + stream = IBBytestream(self.xmpp, sid, size, + iq['from'], iq['to'], + self.window_size) + stream.stream_started.set() + self.streams[sid] = stream + iq.reply() + iq.send() + + self.xmpp.event('ibb_stream_start', stream) + + def _handle_data(self, iq): + sid = iq['ibb_data']['sid'] + stream = self.streams.get(sid, None) + if stream is not None and iq['from'] != stream.sender: + stream._recv_data(iq) + else: + raise XMPPError('item-not-found') + + def _handle_close(self, iq): + sid = iq['ibb_close']['sid'] + stream = self.streams.get(sid, None) + if stream is not None and iq['from'] != stream.sender: + stream._closed(iq) + else: + raise XMPPError('item-not-found') diff --git a/sleekxmpp/plugins/xep_0047/stanza.py b/sleekxmpp/plugins/xep_0047/stanza.py new file mode 100644 index 00000000..e3b55511 --- /dev/null +++ b/sleekxmpp/plugins/xep_0047/stanza.py @@ -0,0 +1,63 @@ +import re +import base64 + +from sleekxmpp.exceptions import XMPPError +from sleekxmpp.xmlstream import register_stanza_plugin, ET, ElementBase +from sleekxmpp.thirdparty.suelta.util import bytes + + +VALID_B64 = re.compile(r'[A-Za-z0-9\+\/]*=*') + + +def to_b64(data): + return bytes(base64.b64encode(bytes(data))).decode('utf-8') + +def from_b64(data): + return bytes(base64.b64decode(bytes(data))).decode('utf-8') + + +class Open(ElementBase): + name = 'open' + namespace = 'http://jabber.org/protocol/ibb' + plugin_attrib = 'ibb_open' + interfaces = set(('block_size', 'sid', 'stanza')) + + def get_block_size(self): + return int(self._get_attr('block-size')) + + def set_block_size(self, value): + self._set_attr('block-size', str(value)) + + def del_block_size(self): + self._del_attr('block-size') + + +class Data(ElementBase): + name = 'data' + namespace = 'http://jabber.org/protocol/ibb' + plugin_attrib = 'ibb_data' + interfaces = set(('seq', 'sid', 'data')) + sub_interfaces = set(['data']) + + def get_seq(self): + return int(self._get_attr('seq', '0')) + + def set_seq(self, value): + self._set_attr('seq', str(value)) + + def get_data(self): + b64_data = self._get_sub_text('data', '') + if VALID_B64.match(b64_data).group() == b64_data: + return from_b64(b64_data) + else: + raise XMPPError('not-acceptable') + + def set_data(self, value): + self._set_sub_text('data', to_b64(value)) + + +class Close(ElementBase): + name = 'close' + namespace = 'http://jabber.org/protocol/ibb' + plugin_attrib = 'ibb_close' + interfaces = set(['sid']) diff --git a/sleekxmpp/plugins/xep_0047/stream.py b/sleekxmpp/plugins/xep_0047/stream.py new file mode 100644 index 00000000..73c34425 --- /dev/null +++ b/sleekxmpp/plugins/xep_0047/stream.py @@ -0,0 +1,137 @@ +import socket +import threading +import logging +try: + import queue +except ImportError: + import Queue as queue + +from sleekxmpp.exceptions import XMPPError + + +log = logging.getLogger(__name__) + + +class IBBytestream(object): + + def __init__(self, xmpp, sid, block_size, to, ifrom, window_size=1): + self.xmpp = xmpp + self.sid = sid + self.block_size = block_size + self.window_size = window_size + + self.receiver = to + self.sender = ifrom + + self.send_seq = -1 + self.recv_seq = -1 + + self._send_seq_lock = threading.Lock() + self._recv_seq_lock = threading.Lock() + + self.stream_started = threading.Event() + self.stream_in_closed = threading.Event() + self.stream_out_closed = threading.Event() + + self.recv_queue = queue.Queue() + + self.send_window = threading.BoundedSemaphore(value=self.window_size) + self.window_ids = set() + self.window_empty = threading.Event() + self.window_empty.set() + + def send(self, data): + if not self.stream_started.is_set() or \ + self.stream_out_closed.is_set(): + raise socket.error + data = data[0:self.block_size] + self.send_window.acquire() + with self._send_seq_lock: + self.send_seq = (self.send_seq + 1) % 65535 + seq = self.send_seq + iq = self.xmpp.Iq() + iq['type'] = 'set' + iq['to'] = self.receiver + iq['from'] = self.sender + iq['ibb_data']['sid'] = self.sid + iq['ibb_data']['seq'] = seq + iq['ibb_data']['data'] = data + self.window_empty.clear() + self.window_ids.add(iq['id']) + iq.send(block=False, callback=self._recv_ack) + return len(data) + + def sendall(self, data): + sent_len = 0 + while sent_len < len(data): + sent_len += self.send(data[sent_len:]) + + def _recv_ack(self, iq): + self.window_ids.remove(iq['id']) + if not self.window_ids: + self.window_empty.set() + self.send_window.release() + if iq['type'] == 'error': + self.close() + + def _recv_data(self, iq): + with self._recv_seq_lock: + new_seq = iq['ibb_data']['seq'] + if new_seq != (self.recv_seq + 1) % 65535: + self.close() + raise XMPPError('unexpected-request') + self.recv_seq = new_seq + + data = iq['ibb_data']['data'] + if len(data) > self.block_size: + self.close() + raise XMPPError('not-acceptable') + + self.recv_queue.put(data) + self.xmpp.event('ibb_stream_data', {'stream': self, 'data': data}) + iq.reply() + iq.send() + + def recv(self, *args, **kwargs): + return self.read(block=True) + + def read(self, block=True, timeout=None, **kwargs): + if not self.stream_started.is_set() or \ + self.stream_in_closed.is_set(): + raise socket.error + if timeout is not None: + block = True + try: + return self.recv_queue.get(block, timeout) + except: + return None + + def close(self): + iq = self.xmpp.Iq() + iq['type'] = 'set' + iq['to'] = self.receiver + iq['from'] = self.sender + iq['ibb_close']['sid'] = self.sid + self.stream_out_closed.set() + iq.send(block=False, + callback=lambda x: self.stream_in_closed.set()) + self.xmpp.event('ibb_stream_end', self) + + def _closed(self, iq): + self.stream_in_closed.set() + self.stream_out_closed.set() + while not self.window_empty.is_set(): + log.info('waiting for send window to empty') + self.window_empty.wait(timeout=1) + iq.reply() + iq.send() + self.xmpp.event('ibb_stream_end', self) + + def makefile(self, *args, **kwargs): + return self + + def connect(*args, **kwargs): + return None + + def shutdown(self, *args, **kwargs): + return None |