## filetransfer.py ## ## Copyright (C) 2004 Alexey "Snake" Nezhdanov ## ## This program is free software; you can redistribute it and/or modify ## it under the terms of the GNU General Public License as published by ## the Free Software Foundation; either version 2, or (at your option) ## any later version. ## ## This program is distributed in the hope that it will be useful, ## but WITHOUT ANY WARRANTY; without even the implied warranty of ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ## GNU General Public License for more details. # $Id: filetransfer.py,v 1.6 2004/12/25 20:06:59 snakeru Exp $ """ This module contains IBB class that is the simple implementation of JEP-0047. Note that this is just a transport for data. You have to negotiate data transfer before (via StreamInitiation most probably). Unfortunately SI is not implemented yet. """ from protocol import * from dispatcher import PlugIn import base64 class IBB(PlugIn): """ IBB used to transfer small-sized data chunk over estabilished xmpp connection. Data is split into small blocks (by default 3000 bytes each), encoded as base 64 and sent to another entity that compiles these blocks back into the data chunk. This is very inefficiend but should work under any circumstances. Note that using IBB normally should be the last resort. """ def __init__(self): """ Initialise internal variables. """ PlugIn.__init__(self) self.DBG_LINE='ibb' self._exported_methods=[self.OpenStream] self._streams={} self._ampnode=Node(NS_AMP+' amp',payload=[Node('rule',{'condition':'deliver-at','value':'stored','action':'error'}),Node('rule',{'condition':'match-resource','value':'exact','action':'error'})]) def plugin(self,owner): """ Register handlers for receiving incoming datastreams. Used internally. """ self._owner.RegisterHandlerOnce('iq',self.StreamOpenReplyHandler) # Move to StreamOpen and specify stanza id self._owner.RegisterHandler('iq',self.IqHandler,ns=NS_IBB) self._owner.RegisterHandler('message',self.ReceiveHandler,ns=NS_IBB) def IqHandler(self,conn,stanza): """ Handles streams state change. Used internally. """ typ=stanza.getType() self.DEBUG('IqHandler called typ->%s'%typ,'info') if typ=='set' and stanza.getTag('open',namespace=NS_IBB): self.StreamOpenHandler(conn,stanza) elif typ=='set' and stanza.getTag('close',namespace=NS_IBB): self.StreamCloseHandler(conn,stanza) elif typ=='result': self.StreamCommitHandler(conn,stanza) elif typ=='error': self.StreamOpenReplyHandler(conn,stanza) else: conn.send(Error(stanza,ERR_BAD_REQUEST)) raise NodeProcessed def StreamOpenHandler(self,conn,stanza): """ Handles opening of new incoming stream. Used internally. """ """ <iq type='set' from='romeo@montague.net/orchard' to='juliet@capulet.com/balcony' id='inband_1'> <open sid='mySID' block-size='4096' xmlns='http://jabber.org/protocol/ibb'/> </iq> """ err=None sid,blocksize=stanza.getTagAttr('open','sid'),stanza.getTagAttr('open','block-size') self.DEBUG('StreamOpenHandler called sid->%s blocksize->%s'%(sid,blocksize),'info') try: blocksize=int(blocksize) except: err=ERR_BAD_REQUEST if not sid or not blocksize: err=ERR_BAD_REQUEST elif sid in self._streams.keys(): err=ERR_UNEXPECTED_REQUEST if err: rep=Error(stanza,err) else: self.DEBUG("Opening stream: id %s, block-size %s"%(sid,blocksize),'info') rep=Protocol('iq',stanza.getFrom(),'result',stanza.getTo(),{'id':stanza.getID()}) self._streams[sid]={'direction':'<'+str(stanza.getFrom()),'block-size':blocksize,'fp':open('/tmp/xmpp_file_'+sid,'w'),'seq':0,'syn_id':stanza.getID()} conn.send(rep) def OpenStream(self,sid,to,fp,blocksize=3000): """ Start new stream. You should provide stream id 'sid', the endpoind jid 'to', the file object containing info for send 'fp'. Also the desired blocksize can be specified. Take into account that recommended stanza size is 4k and IBB uses base64 encoding that increases size of data by 1/3.""" if sid in self._streams.keys(): return if not JID(to).getResource(): return self._streams[sid]={'direction':'|>'+to,'block-size':blocksize,'fp':fp,'seq':0} self._owner.RegisterCycleHandler(self.SendHandler) syn=Protocol('iq',to,'set',payload=[Node(NS_IBB+' open',{'sid':sid,'block-size':blocksize})]) self._owner.send(syn) self._streams[sid]['syn_id']=syn.getID() return self._streams[sid] def SendHandler(self,conn): """ Send next portion of data if it is time to do it. Used internally. """ self.DEBUG('SendHandler called','info') for sid in self._streams.keys(): stream=self._streams[sid] if stream['direction'][:2]=='|>': cont=1 elif stream['direction'][0]=='>': chunk=stream['fp'].read(stream['block-size']) if chunk: datanode=Node(NS_IBB+' data',{'sid':sid,'seq':stream['seq']},base64.encodestring(chunk)) stream['seq']+=1 if stream['seq']==65536: stream['seq']=0 conn.send(Protocol('message',stream['direction'][1:],payload=[datanode,self._ampnode])) else: """ notify the other side about stream closing notify the local user about sucessfull send delete the local stream""" conn.send(Protocol('iq',stream['direction'][1:],'set',payload=[Node(NS_IBB+' close',{'sid':sid})])) conn.Event(self.DBG_LINE,'SUCCESSFULL SEND',stream) del self._streams[sid] self._owner.UnregisterCycleHandler(self.SendHandler) """ <message from='romeo@montague.net/orchard' to='juliet@capulet.com/balcony' id='msg1'> <data xmlns='http://jabber.org/protocol/ibb' sid='mySID' seq='0'> qANQR1DBwU4DX7jmYZnncmUQB/9KuKBddzQH+tZ1ZywKK0yHKnq57kWq+RFtQdCJ WpdWpR0uQsuJe7+vh3NWn59/gTc5MDlX8dS9p0ovStmNcyLhxVgmqS8ZKhsblVeu IpQ0JgavABqibJolc3BKrVtVV1igKiX/N7Pi8RtY1K18toaMDhdEfhBRzO/XB0+P AQhYlRjNacGcslkhXqNjK5Va4tuOAPy2n1Q8UUrHbUd0g+xJ9Bm0G0LZXyvCWyKH kuNEHFQiLuCY6Iv0myq6iX6tjuHehZlFSh80b5BVV9tNLwNR5Eqz1klxMhoghJOA </data> <amp xmlns='http://jabber.org/protocol/amp'> <rule condition='deliver-at' value='stored' action='error'/> <rule condition='match-resource' value='exact' action='error'/> </amp> </message> """ def ReceiveHandler(self,conn,stanza): """ Receive next portion of incoming datastream and store it write it to temporary file. Used internally. """ sid,seq,data=stanza.getTagAttr('data','sid'),stanza.getTagAttr('data','seq'),stanza.getTagData('data') self.DEBUG('ReceiveHandler called sid->%s seq->%s'%(sid,seq),'info') try: seq=int(seq); data=base64.decodestring(data) except: seq=''; data='' err=None if not sid in self._streams.keys(): err=ERR_ITEM_NOT_FOUND else: stream=self._streams[sid] if not data: err=ERR_BAD_REQUEST elif seq<>stream['seq']: err=ERR_UNEXPECTED_REQUEST else: self.DEBUG('Successfull receive sid->%s %s+%s bytes'%(sid,stream['fp'].tell(),len(data)),'ok') stream['seq']+=1 stream['fp'].write(data) if err: self.DEBUG('Error on receive: %s'%err,'error') conn.send(Error(Iq(to=stanza.getFrom(),frm=stanza.getTo(),payload=[Node(NS_IBB+' close')]),err,reply=0)) def StreamCloseHandler(self,conn,stanza): """ Handle stream closure due to all data transmitted. Raise xmpppy event specifying successfull data receive. """ sid=stanza.getTagAttr('close','sid') self.DEBUG('StreamCloseHandler called sid->%s'%sid,'info') if sid in self._streams.keys(): conn.send(stanza.buildReply('result')) conn.Event(self.DBG_LINE,'SUCCESSFULL RECEIVE',self._streams[sid]) del self._streams[sid] else: conn.send(Error(stanza,ERR_ITEM_NOT_FOUND)) def StreamBrokenHandler(self,conn,stanza): """ Handle stream closure due to all some error while receiving data. Raise xmpppy event specifying unsuccessfull data receive. """ syn_id=stanza.getID() self.DEBUG('StreamBrokenHandler called syn_id->%s'%syn_id,'info') for sid in self._streams.keys(): stream=self._streams[sid] if stream['syn_id']==syn_id: if stream['direction'][0]=='<': conn.Event(self.DBG_LINE,'ERROR ON RECEIVE',stream) else: conn.Event(self.DBG_LINE,'ERROR ON SEND',stream) del self._streams[sid] def StreamOpenReplyHandler(self,conn,stanza): """ Handle remote side reply about is it agree or not to receive our datastream. Used internally. Raises xmpppy event specfiying if the data transfer is agreed upon.""" syn_id=stanza.getID() self.DEBUG('StreamOpenReplyHandler called syn_id->%s'%syn_id,'info') for sid in self._streams.keys(): stream=self._streams[sid] if stream['syn_id']==syn_id: if stanza.getType()=='error': if stream['direction'][0]=='<': conn.Event(self.DBG_LINE,'ERROR ON RECEIVE',stream) else: conn.Event(self.DBG_LINE,'ERROR ON SEND',stream) del self._streams[sid] elif stanza.getType()=='result': if stream['direction'][0]=='|': stream['direction']=stream['direction'][1:] conn.Event(self.DBG_LINE,'STREAM COMMITTED',stream) else: conn.send(Error(stanza,ERR_UNEXPECTED_REQUEST))