Mercurial > libervia-backend

comparison src/plugins/plugin_xep_0096.py @ 538:2c4016921403

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
core, frontends, bridgen plugins: fixed methods which were unproperly managing multi-profiles - added profile argument to askConfirmation, actionResult, actionResultExt, entityDataUpdated, confirmationAnswer, getProgress - core, frontends: fixed calls/signals according to new bridge API - user of proper profile namespace for progression indicators and dialogs - memory: getParam* now return bool when param type is bool - memory: added getStringParam* to return string instead of typed value - core, memory, storage, quick_frontend: getHistory now manage properly multi-profiles - plugins XEP-0047, XEP-0054, XEP-0065, XEP-0077, XEP-0096; multi-profiles proper handling
author Goffi <goffi@goffi.org>
date Sat, 10 Nov 2012 16:38:16 +0100
parents a31abb97310d
children 47e45a577ab7
comparison
equal deleted inserted replaced
537:28cddc96c4ed 538:2c4016921403
19 along with this program. If not, see <http://www.gnu.org/licenses/>. 19 along with this program. If not, see <http://www.gnu.org/licenses/>.
20 """ 20 """
21 21
22 from logging import debug, info, warning, error 22 from logging import debug, info, warning, error
23 from twisted.words.xish import domish 23 from twisted.words.xish import domish
24 from twisted.internet import protocol 24 from twisted.words.protocols.jabber import jid
25 from twisted.words.protocols.jabber import client, jid
26 from twisted.words.protocols.jabber import error as jab_error 25 from twisted.words.protocols.jabber import error as jab_error
27 import os, os.path 26 import os, os.path
28 from twisted.internet import reactor 27 from twisted.internet import reactor
29 import pdb 28 from sat.core.exceptions import ProfileNotInCacheError
30 29
31 from zope.interface import implements 30
32 31 from wokkel import data_form
33 from wokkel import disco, iwokkel, data_form
34 32
35 IQ_SET = '/iq[@type="set"]' 33 IQ_SET = '/iq[@type="set"]'
36 PROFILE_NAME = "file-transfer" 34 PROFILE_NAME = "file-transfer"
37 PROFILE = "http://jabber.org/protocol/si/profile/" + PROFILE_NAME 35 PROFILE = "http://jabber.org/protocol/si/profile/" + PROFILE_NAME
38 36
50 class XEP_0096(): 48 class XEP_0096():
51 49
52 def __init__(self, host): 50 def __init__(self, host):
53 info(_("Plugin XEP_0096 initialization")) 51 info(_("Plugin XEP_0096 initialization"))
54 self.host = host 52 self.host = host
55 self._waiting_for_approval = {} #key = id, value = [transfer data, IdelayedCall Reactor timeout, 53 self.managed_stream_m = [#self.host.plugins["XEP-0065"].NAMESPACE,
56 # current stream method, [failed stream methods], profile]
57 self.managed_stream_m = [self.host.plugins["XEP-0065"].NAMESPACE,
58 self.host.plugins["XEP-0047"].NAMESPACE] #Stream methods managed 54 self.host.plugins["XEP-0047"].NAMESPACE] #Stream methods managed
59 self.host.plugins["XEP-0095"].registerSIProfile(PROFILE_NAME, self.transferRequest) 55 self.host.plugins["XEP-0095"].registerSIProfile(PROFILE_NAME, self.transferRequest)
60 host.bridge.addMethod("sendFile", ".plugin", in_sign='ssa{ss}s', out_sign='s', method=self.sendFile) 56 host.bridge.addMethod("sendFile", ".plugin", in_sign='ssa{ss}s', out_sign='s', method=self.sendFile)
61 57
62 def _kill_id(self, approval_id): 58 def profileConnected(self, profile):
59 client = self.host.getClient(profile)
60 client._xep_0096_waiting_for_approval = {} #key = id, value = [transfer data, IdelayedCall Reactor timeout,
61 # current stream method, [failed stream methods], profile]
62
63 def _kill_id(self, approval_id, profile):
63 """Delete a waiting_for_approval id, called after timeout 64 """Delete a waiting_for_approval id, called after timeout
64 @param approval_id: id of _waiting_for_approval""" 65 @param approval_id: id of _xep_0096_waiting_for_approval"""
65 info(_("SI File Transfer: TimeOut reached for id %s") % approval_id); 66 info(_("SI File Transfer: TimeOut reached for id %s") % approval_id);
66 try: 67 try:
67 del self._waiting_for_approval[approval_id] 68 client = self.host.getClient(profile)
69 del client._xep_0096_waiting_for_approval[approval_id]
68 except KeyError: 70 except KeyError:
69 warning(_("kill id called on a non existant approval id")) 71 warning(_("kill id called on a non existant approval id"))
70 72
71 def transferRequest(self, iq_id, from_jid, si_id, si_mime_type, si_el, profile): 73 def transferRequest(self, iq_id, from_jid, si_id, si_mime_type, si_el, profile):
72 """Called when a file transfer is requested 74 """Called when a file transfer is requested
76 @param si_mime_type: Mime type of the file (or default "application/octet-stream" if unknown) 78 @param si_mime_type: Mime type of the file (or default "application/octet-stream" if unknown)
77 @param si_el: domish.Element of the request 79 @param si_el: domish.Element of the request
78 @param profile: %(doc_profile)s""" 80 @param profile: %(doc_profile)s"""
79 info (_("XEP-0096 file transfer requested")) 81 info (_("XEP-0096 file transfer requested"))
80 debug(si_el.toXml()) 82 debug(si_el.toXml())
83 client = self.host.getClient(profile)
84 if not client:
85 raise ProfileNotInCacheError
81 filename = "" 86 filename = ""
82 file_size = "" 87 file_size = ""
83 file_date = None 88 file_date = None
84 file_hash = None 89 file_hash = None
85 file_desc = "" 90 file_desc = ""
122 self.host.plugins["XEP-0095"].sendBadRequestError(iq_id, from_jid, profile) 127 self.host.plugins["XEP-0095"].sendBadRequestError(iq_id, from_jid, profile)
123 return 128 return
124 129
125 #if we are here, the transfer can start, we just need user's agreement 130 #if we are here, the transfer can start, we just need user's agreement
126 data={ "filename":filename, "id": iq_id, "from":from_jid, "size":file_size, "date":file_date, "hash":file_hash, "desc":file_desc, "can_range": str(can_range) } 131 data={ "filename":filename, "id": iq_id, "from":from_jid, "size":file_size, "date":file_date, "hash":file_hash, "desc":file_desc, "can_range": str(can_range) }
127 self._waiting_for_approval[si_id] = [data, reactor.callLater(300, self._kill_id, si_id), stream_method, [], profile] 132 client._xep_0096_waiting_for_approval[si_id] = [data, reactor.callLater(300, self._kill_id, si_id, profile), stream_method, []]
128 133
129 self.host.askConfirmation(si_id, "FILE_TRANSFER", data, self.confirmationCB) 134 self.host.askConfirmation(si_id, "FILE_TRANSFER", data, self.confirmationCB, profile)
130 135
131 136
132 def _getFileObject(self, dest_path, can_range = False): 137 def _getFileObject(self, dest_path, can_range = False):
133 """Open file, put file pointer to the end if the file if needed 138 """Open file, put file pointer to the end if the file if needed
134 @param dest_path: path of the destination file 139 @param dest_path: path of the destination file
135 @param can_range: True if the file pointer can be moved 140 @param can_range: True if the file pointer can be moved
136 @return: File Object""" 141 @return: File Object"""
137 return open(dest_path, "ab" if can_range else "wb") 142 return open(dest_path, "ab" if can_range else "wb")
138 143
139 def confirmationCB(self, sid, accepted, frontend_data): 144 def confirmationCB(self, sid, accepted, frontend_data, profile):
140 """Called on confirmation answer 145 """Called on confirmation answer
141 @param sid: file transfer session id 146 @param sid: file transfer session id
142 @param accepted: True if file transfer is accepted 147 @param accepted: True if file transfer is accepted
143 @param frontend_data: data sent by frontend""" 148 @param frontend_data: data sent by frontend"""
144 data, timeout, stream_method, failed_methods, profile = self._waiting_for_approval[sid] 149 client = self.host.getClient(profile)
150 if not client:
151 raise ProfileNotInCacheError
152 data, timeout, stream_method, failed_methods = client._xep_0096_waiting_for_approval[sid]
145 can_range = data['can_range'] == "True" 153 can_range = data['can_range'] == "True"
146 range_offset = 0 154 range_offset = 0
147 if accepted: 155 if accepted:
148 if timeout.active(): 156 if timeout.active():
149 timeout.cancel() 157 timeout.cancel()
150 try: 158 try:
151 dest_path = frontend_data['dest_path'] 159 dest_path = frontend_data['dest_path']
152 except KeyError: 160 except KeyError:
153 error(_('dest path not found in frontend_data')) 161 error(_('dest path not found in frontend_data'))
154 del(self._waiting_for_approval[sid]) 162 del(client._xep_0096_waiting_for_approval[sid])
155 return 163 return
156 if stream_method == self.host.plugins["XEP-0065"].NAMESPACE: 164 if stream_method == self.host.plugins["XEP-0065"].NAMESPACE:
157 file_obj = self._getFileObject(dest_path, can_range) 165 file_obj = self._getFileObject(dest_path, can_range)
158 range_offset = file_obj.tell() 166 range_offset = file_obj.tell()
159 self.host.plugins["XEP-0065"].prepareToReceive(jid.JID(data['from']), sid, file_obj, int(data["size"]), self._transferSucceeded, self._transferFailed) 167 self.host.plugins["XEP-0065"].prepareToReceive(jid.JID(data['from']), sid, file_obj, int(data["size"]), self._transferSucceeded, self._transferFailed, profile)
160 elif stream_method == self.host.plugins["XEP-0047"].NAMESPACE: 168 elif stream_method == self.host.plugins["XEP-0047"].NAMESPACE:
161 file_obj = self._getFileObject(dest_path, can_range) 169 file_obj = self._getFileObject(dest_path, can_range)
162 range_offset = file_obj.tell() 170 range_offset = file_obj.tell()
163 self.host.plugins["XEP-0047"].prepareToReceive(jid.JID(data['from']), sid, file_obj, int(data["size"]), self._transferSucceeded, self._transferFailed) 171 self.host.plugins["XEP-0047"].prepareToReceive(jid.JID(data['from']), sid, file_obj, int(data["size"]), self._transferSucceeded, self._transferFailed, profile)
164 else: 172 else:
165 error(_("Unknown stream method, this should not happen at this stage, cancelling transfer")) 173 error(_("Unknown stream method, this should not happen at this stage, cancelling transfer"))
166 del(self._waiting_for_approval[sid]) 174 del(client._xep_0096_waiting_for_approval[sid])
167 return 175 return
168 176
169 #we can send the iq result 177 #we can send the iq result
170 feature_elt = self.host.plugins["XEP-0020"].chooseOption({'stream-method':stream_method}) 178 feature_elt = self.host.plugins["XEP-0020"].chooseOption({'stream-method':stream_method})
171 misc_elts = [] 179 misc_elts = []
177 misc_elts.append(range_elt) 185 misc_elts.append(range_elt)
178 self.host.plugins["XEP-0095"].acceptStream(data["id"], data['from'], feature_elt, misc_elts, profile) 186 self.host.plugins["XEP-0095"].acceptStream(data["id"], data['from'], feature_elt, misc_elts, profile)
179 else: 187 else:
180 debug (_("Transfer [%s] refused"), sid) 188 debug (_("Transfer [%s] refused"), sid)
181 self.host.plugins["XEP-0095"].sendRejectedError (data["id"], data['from'], profile=profile) 189 self.host.plugins["XEP-0095"].sendRejectedError (data["id"], data['from'], profile=profile)
182 del(self._waiting_for_approval[sid]) 190 del(client._xep_0096_waiting_for_approval[sid])
183 191
184 def _transferSucceeded(self, sid, file_obj, stream_method): 192 def _transferSucceeded(self, sid, file_obj, stream_method, profile):
185 """Called by the stream method when transfer successfuly finished 193 """Called by the stream method when transfer successfuly finished
186 @param id: stream id""" 194 @param id: stream id"""
195 client = self.host.getClient(profile)
196 if not client:
197 raise ProfileNotInCacheError
187 file_obj.close() 198 file_obj.close()
188 info(_('Transfer %s successfuly finished') % sid) 199 info(_('Transfer %s successfuly finished') % sid)
189 del(self._waiting_for_approval[sid]) 200 del(client._xep_0096_waiting_for_approval[sid])
190 201
191 def _transferFailed(self, sid, file_obj, stream_method, reason): 202 def _transferFailed(self, sid, file_obj, stream_method, reason, profile):
192 """Called when something went wrong with the transfer 203 """Called when something went wrong with the transfer
193 @param id: stream id 204 @param id: stream id
194 @param reason: can be TIMEOUT, IO_ERROR, PROTOCOL_ERROR""" 205 @param reason: can be TIMEOUT, IO_ERROR, PROTOCOL_ERROR"""
195 data, timeout, stream_method, failed_methods, profile = self._waiting_for_approval[sid] 206 client = self.host.getClient(profile)
196 warning(_('Transfer %(id)s failed with stream method %(s_method)s') % { 'id': sid, 207 if not client:
197 's_method': stream_method }) 208 raise ProfileNotInCacheError
209 data, timeout, stream_method, failed_methods = client._xep_0096_waiting_for_approval[sid]
210 warning(_('Transfer %(id)s failed with stream method %(s_method)s: %(reason)s') % { 'id': sid,
211 's_method': stream_method,
212 'reason': reason})
198 filepath = file_obj.name 213 filepath = file_obj.name
199 file_obj.close() 214 file_obj.close()
200 os.remove(filepath) 215 os.remove(filepath)
201 #TODO: session remenber (within a time limit) when a stream method fail, and avoid that stream method with full jid for the rest of the session 216 #TODO: session remenber (within a time limit) when a stream method fail, and avoid that stream method with full jid for the rest of the session
202 warning(_("All stream methods failed, can't transfer the file")) 217 warning(_("All stream methods failed, can't transfer the file"))
203 del(self._waiting_for_approval[sid]) 218 del(client._xep_0096_waiting_for_approval[sid])
204 219
205 def fileCb(self, profile, filepath, sid, size, IQ): 220 def fileCb(self, filepath, sid, size, profile, IQ):
206 if IQ['type'] == "error": 221 if IQ['type'] == "error":
207 stanza_err = jab_error.exceptionFromStanza(IQ) 222 stanza_err = jab_error.exceptionFromStanza(IQ)
208 if stanza_err.code == '403' and stanza_err.condition == 'forbidden': 223 if stanza_err.code == '403' and stanza_err.condition == 'forbidden':
209 debug(_("File transfer refused by %s") % IQ['from']) 224 debug(_("File transfer refused by %s") % IQ['from'])
210 self.host.bridge.newAlert(_("The contact %s refused your file") % IQ['from'], _("File refused"), "INFO", profile) 225 self.host.bridge.newAlert(_("The contact %s refused your file") % IQ['from'], _("File refused"), "INFO", profile)
250 file_obj.seek(range_offset) 265 file_obj.seek(range_offset)
251 self.host.plugins["XEP-0047"].startStream(file_obj, jid.JID(IQ['from']), sid, range_length, self.sendSuccessCb, self.sendFailureCb, size, profile) 266 self.host.plugins["XEP-0047"].startStream(file_obj, jid.JID(IQ['from']), sid, range_length, self.sendSuccessCb, self.sendFailureCb, size, profile)
252 else: 267 else:
253 warning(_("Invalid stream method received")) 268 warning(_("Invalid stream method received"))
254 269
255 def sendFile(self, to_jid, filepath, data={}, profile_key='@DEFAULT@'): 270 def sendFile(self, to_jid, filepath, data={}, profile_key='@NONE@'):
256 """send a file using XEP-0096 271 """send a file using XEP-0096
257 @to_jid: recipient 272 @to_jid: recipient
258 @filepath: absolute path to the file to send 273 @filepath: absolute path to the file to send
259 @data: dictionnary with the optional following keys: 274 @data: dictionnary with the optional following keys:
260 - "description": description of the file 275 - "description": description of the file
276 file_transfer_elts.append(file_elt) 291 file_transfer_elts.append(file_elt)
277 292
278 file_transfer_elts.append(domish.Element((None,'range'))) 293 file_transfer_elts.append(domish.Element((None,'range')))
279 294
280 sid, offer = self.host.plugins["XEP-0095"].proposeStream(jid.JID(to_jid), PROFILE, feature_elt, file_transfer_elts, profile_key = profile) 295 sid, offer = self.host.plugins["XEP-0095"].proposeStream(jid.JID(to_jid), PROFILE, feature_elt, file_transfer_elts, profile_key = profile)
281 offer.addCallback(self.fileCb, profile, filepath, sid, size) 296 offer.addCallback(self.fileCb, filepath, sid, size, profile)
282 return sid 297 return sid
283 298
284 def sendSuccessCb(self, sid, file_obj, stream_method): 299 def sendSuccessCb(self, sid, file_obj, stream_method, profile):
285 info(_('Transfer %s successfuly finished') % sid) 300 info(_('Transfer %s successfuly finished [%s]') % (sid, profile))
286 file_obj.close() 301 file_obj.close()
287 302
288 def sendFailureCb(self, sid, file_obj, stream_method, reason): 303 def sendFailureCb(self, sid, file_obj, stream_method, reason, profile):
289 file_obj.close() 304 file_obj.close()
290 warning(_('Transfer %(id)s failed with stream method %(s_method)s') % { 'id': sid, "s_method": stream_method }) 305 warning(_('Transfer %(id)s failed with stream method %(s_method)s: %(reason)s [%(profile)s') % { 'id': sid, "s_method": stream_method, 'reason': reason, 'profile': profile })