Mercurial > libervia-backend

comparison src/plugins/plugin_xep_0096.py @ 993:301b342c697a

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
core: use of the new core.log module: /!\ this is a massive refactoring and was largely automated, it probably did bring some bugs /!\
author Goffi <goffi@goffi.org>
date Sat, 19 Apr 2014 19:19:19 +0200
parents c6d8fc63b1db
children 4dbe8e57ff51
comparison
equal deleted inserted replaced
992:f51a1895275c 993:301b342c697a
17 # You should have received a copy of the GNU Affero General Public License 17 # You should have received a copy of the GNU Affero General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>. 18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
19 19
20 from sat.core.i18n import _ 20 from sat.core.i18n import _
21 from sat.core.constants import Const as C 21 from sat.core.constants import Const as C
22 from logging import debug, info, warning, error 22 from sat.core.log import getLogger
23 log = getLogger(__name__)
23 from twisted.words.xish import domish 24 from twisted.words.xish import domish
24 from twisted.words.protocols.jabber import jid 25 from twisted.words.protocols.jabber import jid
25 from twisted.words.protocols.jabber import error as jab_error 26 from twisted.words.protocols import jabber
26 import os 27 import os
27 from twisted.internet import reactor 28 from twisted.internet import reactor
28 29
29 from wokkel import data_form 30 from wokkel import data_form
30 31
45 46
46 47
47 class XEP_0096(object): 48 class XEP_0096(object):
48 49
49 def __init__(self, host): 50 def __init__(self, host):
50 info(_("Plugin XEP_0096 initialization")) 51 log.info(_("Plugin XEP_0096 initialization"))
51 self.host = host 52 self.host = host
52 self.managed_stream_m = [self.host.plugins["XEP-0065"].NAMESPACE, 53 self.managed_stream_m = [self.host.plugins["XEP-0065"].NAMESPACE,
53 self.host.plugins["XEP-0047"].NAMESPACE] # Stream methods managed 54 self.host.plugins["XEP-0047"].NAMESPACE] # Stream methods managed
54 self.host.plugins["XEP-0095"].registerSIProfile(PROFILE_NAME, self.transferRequest) 55 self.host.plugins["XEP-0095"].registerSIProfile(PROFILE_NAME, self.transferRequest)
55 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)
60 # current stream method, [failed stream methods], profile] 61 # current stream method, [failed stream methods], profile]
61 62
62 def _kill_id(self, approval_id, profile): 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 _xep_0096_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 log.info(_("SI File Transfer: TimeOut reached for id %s") % approval_id)
66 try: 67 try:
67 client = self.host.getClient(profile) 68 client = self.host.getClient(profile)
68 del client._xep_0096_waiting_for_approval[approval_id] 69 del client._xep_0096_waiting_for_approval[approval_id]
69 except KeyError: 70 except KeyError:
70 warning(_("kill id called on a non existant approval id")) 71 log.warning(_("kill id called on a non existant approval id"))
71 72
72 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):
73 """Called when a file transfer is requested 74 """Called when a file transfer is requested
74 @param iq_id: id of the iq request 75 @param iq_id: id of the iq request
75 @param from_jid: jid of the sender 76 @param from_jid: jid of the sender
76 @param si_id: Stream Initiation session id 77 @param si_id: Stream Initiation session id
77 @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)
78 @param si_el: domish.Element of the request 79 @param si_el: domish.Element of the request
79 @param profile: %(doc_profile)s""" 80 @param profile: %(doc_profile)s"""
80 info(_("XEP-0096 file transfer requested")) 81 log.info(_("XEP-0096 file transfer requested"))
81 debug(si_el.toXml()) 82 log.debug(si_el.toXml())
82 client = self.host.getClient(profile) 83 client = self.host.getClient(profile)
83 filename = "" 84 filename = ""
84 file_size = "" 85 file_size = ""
85 file_date = None 86 file_date = None
86 file_hash = None 87 file_hash = None
93 file_el = file_elts[0] 94 file_el = file_elts[0]
94 filename = file_el["name"] 95 filename = file_el["name"]
95 file_size = file_el["size"] 96 file_size = file_el["size"]
96 file_date = file_el.getAttribute("date", "") 97 file_date = file_el.getAttribute("date", "")
97 file_hash = file_el.getAttribute("hash", "") 98 file_hash = file_el.getAttribute("hash", "")
98 info(_("File proposed: name=[%(name)s] size=%(size)s") % {'name': filename, 'size': file_size}) 99 log.info(_("File proposed: name=[%(name)s] size=%(size)s") % {'name': filename, 'size': file_size})
99 for file_child_el in file_el.elements(): 100 for file_child_el in file_el.elements():
100 if file_child_el.name == "desc": 101 if file_child_el.name == "desc":
101 file_desc = unicode(file_child_el) 102 file_desc = unicode(file_child_el)
102 elif file_child_el.name == "range": 103 elif file_child_el.name == "range":
103 can_range = True 104 can_range = True
104 else: 105 else:
105 warning(_("No file element found")) 106 log.warning(_("No file element found"))
106 self.host.plugins["XEP-0095"].sendBadRequestError(iq_id, from_jid, profile) 107 self.host.plugins["XEP-0095"].sendBadRequestError(iq_id, from_jid, profile)
107 return 108 return
108 109
109 if feature_elts: 110 if feature_elts:
110 feature_el = feature_elts[0] 111 feature_el = feature_elts[0]
111 data_form.Form.fromElement(feature_el.firstChildElement()) 112 data_form.Form.fromElement(feature_el.firstChildElement())
112 try: 113 try:
113 stream_method = self.host.plugins["XEP-0020"].negociate(feature_el, 'stream-method', self.managed_stream_m) 114 stream_method = self.host.plugins["XEP-0020"].negociate(feature_el, 'stream-method', self.managed_stream_m)
114 except KeyError: 115 except KeyError:
115 warning(_("No stream method found")) 116 log.warning(_("No stream method found"))
116 self.host.plugins["XEP-0095"].sendBadRequestError(iq_id, from_jid, profile) 117 self.host.plugins["XEP-0095"].sendBadRequestError(iq_id, from_jid, profile)
117 return 118 return
118 if not stream_method: 119 if not stream_method:
119 warning(_("Can't find a valid stream method")) 120 log.warning(_("Can't find a valid stream method"))
120 self.host.plugins["XEP-0095"].sendFailedError(iq_id, from_jid, profile) 121 self.host.plugins["XEP-0095"].sendFailedError(iq_id, from_jid, profile)
121 return 122 return
122 else: 123 else:
123 warning(_("No feature element found")) 124 log.warning(_("No feature element found"))
124 self.host.plugins["XEP-0095"].sendBadRequestError(iq_id, from_jid, profile) 125 self.host.plugins["XEP-0095"].sendBadRequestError(iq_id, from_jid, profile)
125 return 126 return
126 127
127 #if we are here, the transfer can start, we just need user's agreement 128 #if we are here, the transfer can start, we just need user's agreement
128 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)} 129 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)}
150 if timeout.active(): 151 if timeout.active():
151 timeout.cancel() 152 timeout.cancel()
152 try: 153 try:
153 dest_path = frontend_data['dest_path'] 154 dest_path = frontend_data['dest_path']
154 except KeyError: 155 except KeyError:
155 error(_('dest path not found in frontend_data')) 156 log.error(_('dest path not found in frontend_data'))
156 del(client._xep_0096_waiting_for_approval[sid]) 157 del(client._xep_0096_waiting_for_approval[sid])
157 return 158 return
158 if stream_method == self.host.plugins["XEP-0065"].NAMESPACE: 159 if stream_method == self.host.plugins["XEP-0065"].NAMESPACE:
159 file_obj = self._getFileObject(dest_path, can_range) 160 file_obj = self._getFileObject(dest_path, can_range)
160 range_offset = file_obj.tell() 161 range_offset = file_obj.tell()
162 elif stream_method == self.host.plugins["XEP-0047"].NAMESPACE: 163 elif stream_method == self.host.plugins["XEP-0047"].NAMESPACE:
163 file_obj = self._getFileObject(dest_path, can_range) 164 file_obj = self._getFileObject(dest_path, can_range)
164 range_offset = file_obj.tell() 165 range_offset = file_obj.tell()
165 self.host.plugins["XEP-0047"].prepareToReceive(jid.JID(data['from']), sid, file_obj, int(data["size"]), self._transferSucceeded, self._transferFailed, profile) 166 self.host.plugins["XEP-0047"].prepareToReceive(jid.JID(data['from']), sid, file_obj, int(data["size"]), self._transferSucceeded, self._transferFailed, profile)
166 else: 167 else:
167 error(_("Unknown stream method, this should not happen at this stage, cancelling transfer")) 168 log.error(_("Unknown stream method, this should not happen at this stage, cancelling transfer"))
168 del(client._xep_0096_waiting_for_approval[sid]) 169 del(client._xep_0096_waiting_for_approval[sid])
169 return 170 return
170 171
171 #we can send the iq result 172 #we can send the iq result
172 feature_elt = self.host.plugins["XEP-0020"].chooseOption({'stream-method': stream_method}) 173 feature_elt = self.host.plugins["XEP-0020"].chooseOption({'stream-method': stream_method})
177 range_elt['offset'] = str(range_offset) 178 range_elt['offset'] = str(range_offset)
178 #TODO: manage range length 179 #TODO: manage range length
179 misc_elts.append(range_elt) 180 misc_elts.append(range_elt)
180 self.host.plugins["XEP-0095"].acceptStream(data["id"], data['from'], feature_elt, misc_elts, profile) 181 self.host.plugins["XEP-0095"].acceptStream(data["id"], data['from'], feature_elt, misc_elts, profile)
181 else: 182 else:
182 debug(_("Transfer [%s] refused"), sid) 183 log.debug(_("Transfer [%s] refused"), sid)
183 self.host.plugins["XEP-0095"].sendRejectedError(data["id"], data['from'], profile=profile) 184 self.host.plugins["XEP-0095"].sendRejectedError(data["id"], data['from'], profile=profile)
184 del(client._xep_0096_waiting_for_approval[sid]) 185 del(client._xep_0096_waiting_for_approval[sid])
185 186
186 def _transferSucceeded(self, sid, file_obj, stream_method, profile): 187 def _transferSucceeded(self, sid, file_obj, stream_method, profile):
187 """Called by the stream method when transfer successfuly finished 188 """Called by the stream method when transfer successfuly finished
188 @param id: stream id""" 189 @param id: stream id"""
189 client = self.host.getClient(profile) 190 client = self.host.getClient(profile)
190 file_obj.close() 191 file_obj.close()
191 info(_('Transfer %s successfuly finished') % sid) 192 log.info(_('Transfer %s successfuly finished') % sid)
192 del(client._xep_0096_waiting_for_approval[sid]) 193 del(client._xep_0096_waiting_for_approval[sid])
193 194
194 def _transferFailed(self, sid, file_obj, stream_method, reason, profile): 195 def _transferFailed(self, sid, file_obj, stream_method, reason, profile):
195 """Called when something went wrong with the transfer 196 """Called when something went wrong with the transfer
196 @param id: stream id 197 @param id: stream id
197 @param reason: can be TIMEOUT, IO_ERROR, PROTOCOL_ERROR""" 198 @param reason: can be TIMEOUT, IO_ERROR, PROTOCOL_ERROR"""
198 client = self.host.getClient(profile) 199 client = self.host.getClient(profile)
199 data, timeout, stream_method, failed_methods = client._xep_0096_waiting_for_approval[sid] 200 data, timeout, stream_method, failed_methods = client._xep_0096_waiting_for_approval[sid]
200 warning(_('Transfer %(id)s failed with stream method %(s_method)s: %(reason)s') % { 201 log.warning(_('Transfer %(id)s failed with stream method %(s_method)s: %(reason)s') % {
201 'id': sid, 202 'id': sid,
202 's_method': stream_method, 203 's_method': stream_method,
203 'reason': reason}) 204 'reason': reason})
204 filepath = file_obj.name 205 filepath = file_obj.name
205 file_obj.close() 206 file_obj.close()
206 os.remove(filepath) 207 os.remove(filepath)
207 #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 208 #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
208 warning(_("All stream methods failed, can't transfer the file")) 209 log.warning(_("All stream methods failed, can't transfer the file"))
209 del(client._xep_0096_waiting_for_approval[sid]) 210 del(client._xep_0096_waiting_for_approval[sid])
210 211
211 def fileCb(self, filepath, sid, size, profile, IQ): 212 def fileCb(self, filepath, sid, size, profile, IQ):
212 if IQ['type'] == "error": 213 if IQ['type'] == "error":
213 stanza_err = jab_error.exceptionFromStanza(IQ) 214 stanza_err = jabber.error.exceptionFromStanza(IQ)
214 if stanza_err.code == '403' and stanza_err.condition == 'forbidden': 215 if stanza_err.code == '403' and stanza_err.condition == 'forbidden':
215 debug(_("File transfer refused by %s") % IQ['from']) 216 log.debug(_("File transfer refused by %s") % IQ['from'])
216 self.host.bridge.newAlert(_("The contact %s refused your file") % IQ['from'], _("File refused"), "INFO", profile) 217 self.host.bridge.newAlert(_("The contact %s refused your file") % IQ['from'], _("File refused"), "INFO", profile)
217 else: 218 else:
218 warning(_("Error during file transfer with %s") % IQ['from']) 219 log.warning(_("Error during file transfer with %s") % IQ['from'])
219 self.host.bridge.newAlert(_("Something went wrong during the file transfer session intialisation with %s") % IQ['from'], _("File transfer error"), "ERROR", profile) 220 self.host.bridge.newAlert(_("Something went wrong during the file transfer session intialisation with %s") % IQ['from'], _("File transfer error"), "ERROR", profile)
220 return 221 return
221 222
222 si_elt = IQ.firstChildElement() 223 si_elt = IQ.firstChildElement()
223 224
224 if IQ['type'] != "result" or not si_elt or si_elt.name != "si": 225 if IQ['type'] != "result" or not si_elt or si_elt.name != "si":
225 error(_("Protocol error during file transfer")) 226 log.error(_("Protocol error during file transfer"))
226 return 227 return
227 228
228 feature_elts = self.host.plugins["XEP-0020"].getFeatureElt(si_elt) 229 feature_elts = self.host.plugins["XEP-0020"].getFeatureElt(si_elt)
229 if not feature_elts: 230 if not feature_elts:
230 warning(_("No feature element")) 231 log.warning(_("No feature element"))
231 return 232 return
232 233
233 choosed_options = self.host.plugins["XEP-0020"].getChoosedOptions(feature_elts[0]) 234 choosed_options = self.host.plugins["XEP-0020"].getChoosedOptions(feature_elts[0])
234 try: 235 try:
235 stream_method = choosed_options["stream-method"] 236 stream_method = choosed_options["stream-method"]
236 except KeyError: 237 except KeyError:
237 warning(_("No stream method choosed")) 238 log.warning(_("No stream method choosed"))
238 return 239 return
239 240
240 range_offset = 0 241 range_offset = 0
241 range_length = None 242 range_length = None
242 range_elts = filter(lambda elt: elt.name == 'range', si_elt.elements()) 243 range_elts = filter(lambda elt: elt.name == 'range', si_elt.elements())
254 file_obj = open(filepath, 'r') 255 file_obj = open(filepath, 'r')
255 if range_offset: 256 if range_offset:
256 file_obj.seek(range_offset) 257 file_obj.seek(range_offset)
257 self.host.plugins["XEP-0047"].startStream(file_obj, jid.JID(IQ['from']), sid, range_length, self.sendSuccessCb, self.sendFailureCb, size, profile) 258 self.host.plugins["XEP-0047"].startStream(file_obj, jid.JID(IQ['from']), sid, range_length, self.sendSuccessCb, self.sendFailureCb, size, profile)
258 else: 259 else:
259 warning(_("Invalid stream method received")) 260 log.warning(_("Invalid stream method received"))
260 261
261 def sendFile(self, to_jid, filepath, data={}, profile_key=C.PROF_KEY_NONE): 262 def sendFile(self, to_jid, filepath, data={}, profile_key=C.PROF_KEY_NONE):
262 """send a file using XEP-0096 263 """send a file using XEP-0096
263 @to_jid: recipient 264 @to_jid: recipient
264 @filepath: absolute path to the file to send 265 @filepath: absolute path to the file to send
267 @param profile_key: %(doc_profile_key)s 268 @param profile_key: %(doc_profile_key)s
268 @return: an unique id to identify the transfer 269 @return: an unique id to identify the transfer
269 """ 270 """
270 profile = self.host.memory.getProfileName(profile_key) 271 profile = self.host.memory.getProfileName(profile_key)
271 if not profile: 272 if not profile:
272 warning(_("Trying to send a file from an unknown profile")) 273 log.warning(_("Trying to send a file from an unknown profile"))
273 return "" 274 return ""
274 feature_elt = self.host.plugins["XEP-0020"].proposeFeatures({'stream-method': self.managed_stream_m}) 275 feature_elt = self.host.plugins["XEP-0020"].proposeFeatures({'stream-method': self.managed_stream_m})
275 276
276 file_transfer_elts = [] 277 file_transfer_elts = []
277 278
278 statinfo = os.stat(filepath) 279 statlog.info = os.stat(filepath)
279 file_elt = domish.Element((PROFILE, 'file')) 280 file_elt = domish.Element((PROFILE, 'file'))
280 file_elt['name'] = os.path.basename(filepath) 281 file_elt['name'] = os.path.basename(filepath)
281 size = file_elt['size'] = str(statinfo.st_size) 282 size = file_elt['size'] = str(statlog.info.st_size)
282 file_transfer_elts.append(file_elt) 283 file_transfer_elts.append(file_elt)
283 284
284 file_transfer_elts.append(domish.Element((None, 'range'))) 285 file_transfer_elts.append(domish.Element((None, 'range')))
285 286
286 sid, offer = self.host.plugins["XEP-0095"].proposeStream(jid.JID(to_jid), PROFILE, feature_elt, file_transfer_elts, profile_key=profile) 287 sid, offer = self.host.plugins["XEP-0095"].proposeStream(jid.JID(to_jid), PROFILE, feature_elt, file_transfer_elts, profile_key=profile)
287 offer.addCallback(self.fileCb, filepath, sid, size, profile) 288 offer.addCallback(self.fileCb, filepath, sid, size, profile)
288 return sid 289 return sid
289 290
290 def sendSuccessCb(self, sid, file_obj, stream_method, profile): 291 def sendSuccessCb(self, sid, file_obj, stream_method, profile):
291 info(_('Transfer %(sid)s successfuly finished [%(profile)s]') 292 log.info(_('Transfer %(sid)s successfuly finished [%(profile)s]')
292 % {"sid": sid, "profile": profile}) 293 % {"sid": sid, "profile": profile})
293 file_obj.close() 294 file_obj.close()
294 295
295 def sendFailureCb(self, sid, file_obj, stream_method, reason, profile): 296 def sendFailureCb(self, sid, file_obj, stream_method, reason, profile):
296 file_obj.close() 297 file_obj.close()
297 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}) 298 log.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})