-
Notifications
You must be signed in to change notification settings - Fork 1
/
client.py
372 lines (332 loc) · 14.7 KB
/
client.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
from twisted.words.protocols.jabber import client, jid
from twisted.words.protocols.jabber import xmlstream as jxml
from twisted.words.xish.xpath import XPathQuery
from twisted.words.xish import domish, utility
from twisted.python import log
from utils import presenceStanza, messageStanza, updateAttribsFromDict, FEATURE_NOT_IMPLEMENTED
from twisted.internet import interfaces
import xpaths, namespaces
class Disco:
def __init__(self, entity, category, type):
self._features = [namespaces.DISCO_INFO]
self.identities = [{'category': category, 'type': type}]
self.nodes = {}
self._entity = entity
self._entity.iq_get_hooks.addObserver(xpaths.DISCO_INFO, self.discoInfo)
self._entity.iq_get_hooks.addObserver(xpaths.DISCO_ITEMS, self.discoItems)
def discoInfo(self, iq):
node = iq.query.getAttribute('node')
if node:
if node in self.nodes: # we know about this node
return self.nodes[node].discoInfo(self, iq)
iq.swapAttributeValues('from', 'to')
iq['type'] = 'result'
iq.children = []
q = iq.addElement("query", namespaces.DISCO_INFO)
for i in self.identities:
queryIdent = q.addElement("identity", None, "")
queryIdent['category'] = i['category']
queryIdent['type'] = i['type']
if 'name' in i:
queryIdent['name'] = i['name']
for f in self._features:
queryFeat = q.addElement("feature", None, "")
queryFeat['var'] = f
self._entity._xmlstream.send(iq)
def iteritems(self):
for i in self.features:
yield i
def discoItems(self, iq):
node = iq.query.getAttribute('node')
if node:
if node in self.nodes:
iq.swapAttributeValues('from', 'to')
iq['type'] = 'result'
for i in self.nodes[node].iteritems_for_jid(iq['from']):
thisItem = iq.query.addElement("item")
thisItem['jid'] = self._entity._jid.full()
updateAttribsFromDict(thisItem, i)
self._entity._xmlstream.send(iq)
else:
iq.swapAttributeValues('from', 'to')
iq['type'] = 'error'
e = iq.query.addElement("error")
e['type'] = 'cancel'
e.children = [FEATURE_NOT_IMPLEMENTED]
self._entity._xmlstream.send(iq)
def addNode(self, name, obj):
self.nodes[name] = obj
def addFeature(self, name, obj=None):
self._features.append(name)
class FileSaver:
"Data consumer which writes data to a file"
def __init__(self, filename, size, local_filename):
self.file = None
self.filename = filename
self.local_filename = local_filename
self.size = size
self.written = 0
def registerProducer(self, producer, streaming):
self.producer = producer
self.file = open(self.local_filename, 'w')
def unregisterProducer(self):
self.producer = None
self.close()
def write(self, data):
if self.file:
self.file.write(data)
self.written += len(data)
if self.written >= self.size:
self.close()
self.onCompleted()
def error(self):
self.close()
def close(self):
if self.file:
self.file.close()
self.file = None
self.onCompleted(self)
def onCompleted(self): pass
class IdleMixin:
def setIdleTime(self, idle_seconds=10, away_seconds=30):
self.idle_seconds = idle_seconds
self.away_seconds = away_seconds
self.antiIdleActivity()
def antiIdleActivity(self):
from twisted.internet import reactor
if hasattr(self, "waitIdle") and \
self.waitIdle.active():
self.waitIdle.reset(self.idle_seconds)
else:
self.waitIdle = reactor.callLater(self.idle_seconds, self.away)
#, self.away)
if hasattr(self, "waitAway") and \
self.waitAway.active():
self.waitAway.reset(self.away_seconds)
else:
self.waitAway = reactor.callLater(self.away_seconds, self.extended_away)
self.showText = None # have to set it like this..
self.setStatus() # send presence stanza
def away(self):
self.setStatus(showText='away')
def extended_away(self):
self.setStatus(showText='xa')
# the following class is useful for when you're testing sending/receiving IQs
# to yourself over the same xmlstream..
# twisted's own IQ class fires callbacks as the inital IQ gets sent, and clears the callbacklist
# so that the result IQ response causes an error
class MyIQ(client.IQ):
def send(self, to=None):
if to != None:
self['to'] = to
if self['type'] in ('get', 'set'):
#self._xmlstream.addOnetimeObserver("/iq[@id='%s'][@type!='%s']" % (self['id'], self['type']), self._resultEvent)
self._xmlstream.addOnetimeObserver("/iq[@id='%s']" % self['id'], self._resultEvent)
self._xmlstream.send(self)
def _resultEvent(self, iq):
self.callbacks.callback(iq)
class BasicJabberClient(utility.EventDispatcher):
def __init__(self, myJID, myPassword, identity_type):
utility.EventDispatcher.__init__(self)
self._xmlstream = None
self.iq_get_hooks = utility.EventDispatcher()
self.iq_set_hooks = utility.EventDispatcher()
self.message_hooks = utility.EventDispatcher()
self.disco = Disco(self, 'client', identity_type)
self._jid = myJID
self._password = myPassword
self._factory = client.basicClientFactory(self._jid, self._password)
self._factory.addBootstrap(jxml.STREAM_AUTHD_EVENT, self.streamAuthenticated)
self._factory.addBootstrap(jxml.STREAM_CONNECTED_EVENT, self.streamConnected)
self._factory.addBootstrap(jxml.STREAM_END_EVENT, self.streamEnded)
self._factory.addBootstrap(jxml.STREAM_ERROR_EVENT, self.streamError)
self._factory.addBootstrap(jxml.STREAM_START_EVENT, self.streamStarted)
if hasattr(jxml, "TLS_FAILED_EVENT"):
self._factory.addBootstrap(jxml.TLS_FAILED_EVENT, self.tlsFailed)
self.showText = None
self.statusText = ''
self.priority = 0
def _rawDataIn(self, data):
self.dispatch(data, '//event/RAW_DATA_IN')
def _rawDataOut(self, data):
self.dispatch(data, '//event/RAW_DATA_OUT')
def streamAuthenticated(self, xmlstream):
log.msg("connection authenticated")
self._xmlstream = xmlstream
self._xmlstream.rawDataInFn = self._rawDataIn
self._xmlstream.rawDataOutFn = self._rawDataOut
self._xmlstream.addObserver('/message', self.onMessage)
self._xmlstream.addObserver("/iq[@type='error']", self.onIqError)
self._xmlstream.addObserver("/iq[@type='get']", self.onIqGet)
self._xmlstream.addObserver("/iq[@type='set']", self.onIqSet)
#self.setStatus(statusText="Online")
self.dispatch('ok', '//event/CONNECTION_AUTHENTICATED')
def onIqGet(self, iq):
noMatch = True
if iq.firstChildElement():
elem = iq.firstChildElement()
noMatch = not self.iq_get_hooks.dispatch( iq )
if noMatch:
log.msg("Unsupported IQ.GET: " + iq.toXml())
elem = iq.firstChildElement() or iq
iq.swapAttributeValues('from', 'to')
iq['type'] = 'error'
e = elem.addElement("error")
e['type'] = 'cancel'
e.children = [FEATURE_NOT_IMPLEMENTED]
self._xmlstream.send(iq)
def onIqSet(self, iq):
noMatch = True
if iq.firstChildElement():
elem = iq.firstChildElement()
noMatch = not self.iq_set_hooks.dispatch( iq )
if noMatch:
log.msg("Unsupported IQ.SET: " + iq.toXml())
elem = iq.firstChildElement() or iq
iq.swapAttributeValues('from', 'to')
iq['type'] = 'error'
e = elem.addElement("error")
e['type'] = 'cancel'
e.children = [FEATURE_NOT_IMPLEMENTED]
self._xmlstream.send(iq)
def streamConnected(self, xmlstream): pass
def streamEnded(self, xmlstream): pass
def streamError(self, xmlstream): pass
def streamStarted(self, xmlstream): pass
def tlsFailed(self, xmlstream): pass
def onIqError(self, iq):
log.err("ERROR: " + iq.toXml())
pass
def setStatus(self, statusText=None, showText=None, priority=None):
if statusText:
self.statusText = statusText
if showText:
self.showText = showText
if priority:
self.priority = priority
if self._xmlstream == None:
return
p = presenceStanza(status=self.statusText, show=self.showText, priority=self.priority)
self._xmlstream.send(p)
def onMessage(self, msg):
"Called when we've received a message.. check our hooks list and call anyone interested"
# alert callbacks waiting for any kind of message
self.message_hooks.dispatch(msg, '//event/RECEIVE')
# alert callbacks with xpath selectors
self.message_hooks.dispatch( msg )
def sendMessage(self, to=None, body=None):
self._xmlstream.send(messageStanza(to, body))
def getIQ(self, type="set"):
return MyIQ(self._xmlstream, type)
class RosterJabberClient(BasicJabberClient):
def __init__(self, myJID, myPassword, identity_type):
BasicJabberClient.__init__(self, myJID, myPassword, identity_type)
self.onlineContacts = {}
self.rosterContacts = []
def streamAuthenticated(self, xmlstream):
BasicJabberClient.streamAuthenticated(self, xmlstream)
self.presence_hooks = utility.EventDispatcher()
self._xmlstream.addObserver('/presence', self.onPresence)
self.iq_set_hooks.addObserver("/*[@xmlns='jabber:iq:roster']", self.onRosterIq)
self.presence_hooks.addObserver(xpaths.PRESENCE_SUBSCRIBE, self.onSubscribe)
self.presence_hooks.addObserver(xpaths.PRESENCE_UNSUBSCRIBE, self.onUnSubscribe)
self.setStatus(statusText="Online")
# request roster
iq = client.IQ(self._xmlstream, type='get')
iq.addElement("query", "jabber:iq:roster")
iq.addCallback(self.onReceiveRoster)
iq.send()
def onReceiveRoster(self, iq):
# JRH - Sauce Labs - Bug fix - rosters were not being reset properly.
# If a new copy of the entire roster was resent by the server,
# the list should be reset to zero before adding to it.
self.rosterContacts = []
# JRH - end
items = iq.children[0].children
for i in items:
self.rosterContacts.append(i['jid'])
self.dispatch((),'//event/ROSTER')
def onRosterIq(self, iq):
pass
def onSubscribe(self, presence):
"""Called when someone tries to subscribe to our presence (adds us to their roster)
Override this, and call self.allowSubscribe(presence) to permit this"""
pass
def onUnSubscribe(self, presence):
"""Called when someone removes us from their contact list. Default action is to remove them from ours"""
self.contactUnsubscribed(presence)
def allowSubscribe(self, presence):
"Call this from onSubscribe to allow a contact to subscribe to our presence"
log.msg("Allowing %s to subscribe to our presence" % presence['from'])
presence['to'] = presence['from']
presence['type'] = 'subscribed'
self._xmlstream.send(presence)
# automatically ask for subscription
log.msg("Requesting authorisation from %s" % presence['to'])
new_pre = domish.Element(("jabber:client", "presence"))
new_pre['to'] = presence['to']
new_pre['type'] = 'subscribe'
self._xmlstream.send(new_pre)
self.rosterContacts.append(presence['from']) # should probably wait til we receive confirmation
def contactUnsubscribed(self, presence):
"Call this from OnUnSubscribe to acknowledge and remove contact from our roster"
log.msg("%s removed us from their roster - doing likewise" % presence['from'])
presence.swapAttributeValues('to', 'from')
presence['type'] = 'unsubscribed'
self._xmlstream.send(presence)
# and remove jid from roster
iq = client.IQ(self._xmlstream, type="set")
query = iq.addElement(("jabber:iq:roster", "query"))
query.addElement("item")
query.item['jid'] = presence['to']
query.item['subscription'] = 'remove'
iq.send()
self.rosterContacts.remove(presence['to'])
if presence['to'] in self.onlineContacts:
del self.onlineContacts[presence['to']]
def onPresence(self, p):
if p.getAttribute('type') == 'unavailable':
self.onContactUnavailable(p)
else:
self.onContactAvailable(p)
# JRH - Sauce Labs - enhancement
# Send the status description text along
# with 'show' value (e.g. 'online', 'away', etc.)
if p.status:
status = p.status.__str__()
else:
status = ''
if p.show:
self.dispatch((p['from'], p.show.__str__(), status), '//event/CONTACT_SHOW')
else:
self.dispatch((p['from'], 'online', status), '//event/CONTACT_SHOW')
# JRH - end
self.presence_hooks.dispatch(p)
def onContactUnavailable(self,p):
j = jid.internJID(p['from'])
c = self.onlineContacts.get(j.userhost(), [])
if j.resource in c:
c.remove(j.resource)
self.dispatch(j, '//event/RESOURCE_UNAVAILABLE')
if c == []:
if j.userhost() in self.onlineContacts:
del self.onlineContacts[j.userhost()]
self.dispatch(j.userhost(), '//event/CONTACT_UNAVAILABLE')
else:
self.onlineContacts[j.userhost()] = c
def onContactAvailable(self, p):
j = jid.internJID(p['from'])
c = self.onlineContacts.get(j.userhost(), [])
# JRH - Sauce Labs - Bug fix - The list of xmpp resources were
# growing in size on every update for the contact. Now, we only
# add to the list if it's not already in the list.
if j.resource not in c:
c.append(j.resource)
# JRH - end
self.onlineContacts[j.userhost()] = c
if len(c) == 1:
self.dispatch(j.userhost(), '//event/CONTACT_AVAILABLE')
self.dispatch(j, '//event/RESOURCE_AVAILABLE')
def onContactStatusChange(self, jid, p):
self.dispatch((jid, p), '//event/CONTACT_STATUS_CHANGE')
pass