Mercurial > libervia-backend

comparison src/plugins/plugin_xep_0047.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 069ad98b360d
comparison
equal deleted inserted replaced
992:f51a1895275c 993:301b342c697a
16 16
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 logging import debug, info, warning, error 21 from sat.core.log import getLogger
22 log = getLogger(__name__)
22 from twisted.words.protocols.jabber import client as jabber_client, jid 23 from twisted.words.protocols.jabber import client as jabber_client, jid
23 from twisted.words.xish import domish 24 from twisted.words.xish import domish
24 from twisted.internet import reactor 25 from twisted.internet import reactor
25 26
26 from wokkel import disco, iwokkel 27 from wokkel import disco, iwokkel
57 58
58 class XEP_0047(object): 59 class XEP_0047(object):
59 NAMESPACE = NS_IBB 60 NAMESPACE = NS_IBB
60 61
61 def __init__(self, host): 62 def __init__(self, host):
62 info(_("In-Band Bytestreams plugin initialization")) 63 log.info(_("In-Band Bytestreams plugin initialization"))
63 self.host = host 64 self.host = host
64 65
65 def getHandler(self, profile): 66 def getHandler(self, profile):
66 return XEP_0047_handler(self) 67 return XEP_0047_handler(self)
67 68
70 client.xep_0047_current_stream = {} # key: stream_id, value: data(dict) 71 client.xep_0047_current_stream = {} # key: stream_id, value: data(dict)
71 72
72 def _timeOut(self, sid, profile): 73 def _timeOut(self, sid, profile):
73 """Delecte current_stream id, called after timeout 74 """Delecte current_stream id, called after timeout
74 @param id: id of client.xep_0047_current_stream""" 75 @param id: id of client.xep_0047_current_stream"""
75 info(_("In-Band Bytestream: TimeOut reached for id %(sid)s [%(profile)s]") 76 log.info(_("In-Band Bytestream: TimeOut reached for id %(sid)s [%(profile)s]")
76 % {"sid": sid, "profile": profile}) 77 % {"sid": sid, "profile": profile})
77 self._killId(sid, False, "TIMEOUT", profile) 78 self._killId(sid, False, "TIMEOUT", profile)
78 79
79 def _killId(self, sid, success=False, failure_reason="UNKNOWN", profile=None): 80 def _killId(self, sid, success=False, failure_reason="UNKNOWN", profile=None):
80 """Delete an current_stream id, clean up associated observers 81 """Delete an current_stream id, clean up associated observers
81 @param sid: id of client.xep_0047_current_stream""" 82 @param sid: id of client.xep_0047_current_stream"""
82 assert(profile) 83 assert(profile)
83 client = self.host.getClient(profile) 84 client = self.host.getClient(profile)
84 if sid not in client.xep_0047_current_stream: 85 if sid not in client.xep_0047_current_stream:
85 warning(_("kill id called on a non existant id")) 86 log.warning(_("kill id called on a non existant id"))
86 return 87 return
87 if "observer_cb" in client.xep_0047_current_stream[sid]: 88 if "observer_cb" in client.xep_0047_current_stream[sid]:
88 client.xmlstream.removeObserver(client.xep_0047_current_stream[sid]["event_data"], client.xep_0047_current_stream[sid]["observer_cb"]) 89 client.xmlstream.removeObserver(client.xep_0047_current_stream[sid]["event_data"], client.xep_0047_current_stream[sid]["observer_cb"])
89 if client.xep_0047_current_stream[sid]['timer'].active(): 90 if client.xep_0047_current_stream[sid]['timer'].active():
90 client.xep_0047_current_stream[sid]['timer'].cancel() 91 client.xep_0047_current_stream[sid]['timer'].cancel()
132 data["timer"] = reactor.callLater(TIMEOUT, self._timeOut, sid) 133 data["timer"] = reactor.callLater(TIMEOUT, self._timeOut, sid)
133 data["success_cb"] = success_cb 134 data["success_cb"] = success_cb
134 data["failure_cb"] = failure_cb 135 data["failure_cb"] = failure_cb
135 136
136 def streamOpening(self, IQ, profile): 137 def streamOpening(self, IQ, profile):
137 debug(_("IBB stream opening")) 138 log.debug(_("IBB stream opening"))
138 IQ.handled = True 139 IQ.handled = True
139 client = self.host.getClient(profile) 140 client = self.host.getClient(profile)
140 open_elt = IQ.firstChildElement() 141 open_elt = IQ.firstChildElement()
141 block_size = open_elt.getAttribute('block-size') 142 block_size = open_elt.getAttribute('block-size')
142 sid = open_elt.getAttribute('sid') 143 sid = open_elt.getAttribute('sid')
143 stanza = open_elt.getAttribute('stanza', 'iq') 144 stanza = open_elt.getAttribute('stanza', 'iq')
144 if not sid or not block_size or int(block_size) > 65535: 145 if not sid or not block_size or int(block_size) > 65535:
145 warning(_("malformed IBB transfer: %s" % IQ['id'])) 146 log.warning(_("malformed IBB transfer: %s" % IQ['id']))
146 self.sendNotAcceptableError(IQ['id'], IQ['from'], client.xmlstream) 147 self.sendNotAcceptableError(IQ['id'], IQ['from'], client.xmlstream)
147 return 148 return
148 if not sid in client.xep_0047_current_stream: 149 if not sid in client.xep_0047_current_stream:
149 warning(_("Ignoring unexpected IBB transfer: %s" % sid)) 150 log.warning(_("Ignoring unexpected IBB transfer: %s" % sid))
150 self.sendNotAcceptableError(IQ['id'], IQ['from'], client.xmlstream) 151 self.sendNotAcceptableError(IQ['id'], IQ['from'], client.xmlstream)
151 return 152 return
152 if client.xep_0047_current_stream[sid]["from"] != jid.JID(IQ['from']): 153 if client.xep_0047_current_stream[sid]["from"] != jid.JID(IQ['from']):
153 warning(_("sended jid inconsistency (man in the middle attack attempt ?)")) 154 log.warning(_("sended jid inconsistency (man in the middle attack attempt ?)"))
154 self.sendNotAcceptableError(IQ['id'], IQ['from'], client.xmlstream) 155 self.sendNotAcceptableError(IQ['id'], IQ['from'], client.xmlstream)
155 self._killId(sid, False, "PROTOCOL_ERROR", profile=profile) 156 self._killId(sid, False, "PROTOCOL_ERROR", profile=profile)
156 return 157 return
157 158
158 #at this stage, the session looks ok and will be accepted 159 #at this stage, the session looks ok and will be accepted
175 client.xmlstream.send(result) 176 client.xmlstream.send(result)
176 177
177 def streamClosing(self, IQ, profile): 178 def streamClosing(self, IQ, profile):
178 IQ.handled = True 179 IQ.handled = True
179 client = self.host.getClient(profile) 180 client = self.host.getClient(profile)
180 debug(_("IBB stream closing")) 181 log.debug(_("IBB stream closing"))
181 data_elt = IQ.firstChildElement() 182 data_elt = IQ.firstChildElement()
182 sid = data_elt.getAttribute('sid') 183 sid = data_elt.getAttribute('sid')
183 result = domish.Element((None, 'iq')) 184 result = domish.Element((None, 'iq'))
184 result['type'] = 'result' 185 result['type'] = 'result'
185 result['id'] = IQ['id'] 186 result['id'] = IQ['id']
210 @param data_elt: "data" domish element 211 @param data_elt: "data" domish element
211 @return: True if success""" 212 @return: True if success"""
212 client = self.host.getClient(profile) 213 client = self.host.getClient(profile)
213 sid = data_elt.getAttribute('sid') 214 sid = data_elt.getAttribute('sid')
214 if sid not in client.xep_0047_current_stream: 215 if sid not in client.xep_0047_current_stream:
215 error(_("Received data for an unknown session id")) 216 log.error(_("Received data for an unknown session id"))
216 return False 217 return False
217 218
218 from_jid = client.xep_0047_current_stream[sid]["from"] 219 from_jid = client.xep_0047_current_stream[sid]["from"]
219 file_obj = client.xep_0047_current_stream[sid]["file_obj"] 220 file_obj = client.xep_0047_current_stream[sid]["file_obj"]
220 221
221 if stanza_from_jid != from_jid: 222 if stanza_from_jid != from_jid:
222 warning(_("sended jid inconsistency (man in the middle attack attempt ?)")) 223 log.warning(_("sended jid inconsistency (man in the middle attack attempt ?)"))
223 if stanza == 'iq': 224 if stanza == 'iq':
224 self.sendNotAcceptableError(sid, from_jid, client.xmlstream) 225 self.sendNotAcceptableError(sid, from_jid, client.xmlstream)
225 return False 226 return False
226 227
227 client.xep_0047_current_stream[sid]["seq"] += 1 228 client.xep_0047_current_stream[sid]["seq"] += 1
228 if int(data_elt.getAttribute("seq", -1)) != client.xep_0047_current_stream[sid]["seq"]: 229 if int(data_elt.getAttribute("seq", -1)) != client.xep_0047_current_stream[sid]["seq"]:
229 warning(_("Sequence error")) 230 log.warning(_("Sequence error"))
230 if stanza == 'iq': 231 if stanza == 'iq':
231 self.sendNotAcceptableError(data_elt["id"], from_jid, client.xmlstream) 232 self.sendNotAcceptableError(data_elt["id"], from_jid, client.xmlstream)
232 return False 233 return False
233 234
234 #we reset the timeout: 235 #we reset the timeout:
237 #we can now decode the data 238 #we can now decode the data
238 try: 239 try:
239 file_obj.write(base64.b64decode(str(data_elt))) 240 file_obj.write(base64.b64decode(str(data_elt)))
240 except TypeError: 241 except TypeError:
241 #The base64 data is invalid 242 #The base64 data is invalid
242 warning(_("Invalid base64 data")) 243 log.warning(_("Invalid base64 data"))
243 if stanza == 'iq': 244 if stanza == 'iq':
244 self.sendNotAcceptableError(data_elt["id"], from_jid, client.xmlstream) 245 self.sendNotAcceptableError(data_elt["id"], from_jid, client.xmlstream)
245 return False 246 return False
246 return True 247 return True
247 248
268 @param successCb: method to call when stream successfuly finished 269 @param successCb: method to call when stream successfuly finished
269 @param failureCb: method to call when something goes wrong 270 @param failureCb: method to call when something goes wrong
270 @param profile: %(doc_profile)s""" 271 @param profile: %(doc_profile)s"""
271 client = self.host.getClient(profile) 272 client = self.host.getClient(profile)
272 if length is not None: 273 if length is not None:
273 error(_('stream length not managed yet')) 274 log.error(_('stream length not managed yet'))
274 return 275 return
275 data = client.xep_0047_current_stream[sid] = {} 276 data = client.xep_0047_current_stream[sid] = {}
276 data["timer"] = reactor.callLater(TIMEOUT, self._timeOut, sid) 277 data["timer"] = reactor.callLater(TIMEOUT, self._timeOut, sid)
277 data["file_obj"] = file_obj 278 data["file_obj"] = file_obj
278 data["to"] = to_jid 279 data["to"] = to_jid
295 def iqResult(self, sid, seq, length, profile, iq_elt): 296 def iqResult(self, sid, seq, length, profile, iq_elt):
296 """Called when the result of open iq is received""" 297 """Called when the result of open iq is received"""
297 client = self.host.getClient(profile) 298 client = self.host.getClient(profile)
298 data = client.xep_0047_current_stream[sid] 299 data = client.xep_0047_current_stream[sid]
299 if iq_elt["type"] == "error": 300 if iq_elt["type"] == "error":
300 warning(_("Transfer failed")) 301 log.warning(_("Transfer failed"))
301 self.terminateStream(sid, "IQ_ERROR") 302 self.terminateStream(sid, "IQ_ERROR")
302 return 303 return
303 304
304 if data['timer'].active(): 305 if data['timer'].active():
305 data['timer'].cancel() 306 data['timer'].cancel()