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
|
#!/usr/bin/python
# -*- vim: encoding: utf-8 -*-
# Yves Fischer, xapek.org 2009
import struct
from twisted.words.protocols import oscar
from twisted.internet import reactor, protocol
from twisted.internet.protocol import ClientFactory
CAP_TYPING = '\x56\x3f\xc8\x09\x0b\x6f\x41\xbd\x9f\x79\x42\x26\x09\xdf\xa2\xf3'
class ExtendedBOSConnection( oscar.BOSConnection ):
capabilities = [oscar.CAP_CHAT,CAP_TYPING]
def __init__(self,s,p,**kwargs):
oscar.BOSConnection.__init__(self, s, p,**kwargs)
"""
handles typing SNAC
"""
def oscar_04_14(self,snac):
data = snac[3]
if len(data) == len("\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\t222840035\x00\x02"):
user = data[11:20]
state = data[20:22]
if state == "\x00\x00": #finish
self.receiveTyping(user, "finish")
elif state == "\x00\x01": #idle
self.receiveTyping(user, "idle")
elif state == "\x00\x02": #begin
self.receiveTyping(user, "begin")
else:
print "Unknown state in typing snac\nuser %s\ndata %s\nsnac %s\nstate %X%X" % (user,data,snac,state[0],sate[1])
else:
print "komisches tpying snac"
def receiveTyping(self, user, state):
pass
def initDone( self ):
self.requestSelfInfo()
self.requestSSI().addCallback(self.gotBuddyList)
def shutdown( self ):
pass
def updateBuddy(self, user):
print "updateBuddy %s" % user
def gotBuddyList( self, buddylist ):
self.activateSSI()
self.setProfile("Forget ICQ, MSN, Yahoo and the other shitty protocols! Use XMPP/Jabber!")
self.setIdleTime( 0 )
self.clientReady()
for user in buddylist[0][0].users:
print "icq: Authorize %s" % user.name
self.sendAuthorizationResponse(user.name, True, '')
"""message - utf8 """
def receiveCleanMessage( self, user, message, flags ):
raise NotImplementedError('receiveCleanMessage( self, user, message, flags )')
def _getMessage(self,multiparts):
message = None
if len(multiparts[0]) == 2:
if multiparts[0][1] == "unicode":
message = unicode(multiparts[0][0])
else:
message = multiparts[0][0].decode(multiparts[0][1],'latin1')
else:
message = str(multiparts[0][0])
return message
def receiveMessage( self, user, multiparts, flags ):
print "icq: receiveMessage(%s,%s,%s)" % (user,multiparts,flags)
if flags != None and "auto" in flags:
print "'auto' message, abort"
return
# because i cant receive the "budded added signal" i auth on every message
self.sendAuthorizationResponse(user.name, True, '')
#cleanup message
try:
message = self._getMessage( multiparts )
#filter qip \x00CHAR\x00CHAR - qip sucks at unicode?
message = filter(lambda x: x!=u"\x00", message)
self.receiveCleanMessage( user, message, flags )
except Exception,e:
print "Exception: %s" % e
def chatReceiveMessage( self, chat, user, message ):
self.receiveMessage( self, user, message, [] )
def sendAuthorizationResponse(self, uin, success, responsString):
packet = struct.pack("b", len(uin)) + uin
if success:
packet += struct.pack("b", 1)
else:
packet += struct.pack("b", 0)
packet += struct.pack(">H", len(responsString)) + responsString
self.sendSNACnr(0x13, 0x1a, packet)
def sendMessage(self, user, messageUtf8):
message = messageUtf8.encode('latin1','replace')
oscar.BOSConnection.sendMessage(self,user,message)
def connectionLost(self,*args,**kwargs):
print "Connection lost"
oscar.BOSConnection.connectionLost(self,*args,**kwargs)
class Authenticator( oscar.OscarAuthenticator ):
# BOSClass = ExtendedBOSConnection
def connectionLost(self,reason):
oscar.OscarAuthenticator.connectionLost(self,reason)
self.factory.connectionLost(reason)
class ReconnectOscarFactory(ClientFactory):
def __init__(self, BOSClass, uid, password,icq=1):
self.BOSClass = BOSClass
self.uid = uid
self.password = password
self.icq = icq
def buildProtocol(self, addr):
p = Authenticator(self.uid, self.password,icq=self.icq)
p.factory = self
p.BOSClass = self.BOSClass
return p
def startedConnecting(self, connector):
print 'Started to connect to oscar'
def connectionLost(self,reason):
print "CONECTION LOST - start njew factory"
reactor.callLater(4.0, reactor.connectTCP,
'login.icq.com', 5238, self)
# reactor.connectTCP('login.icq.com', 5238, OscarFactory(self.uin, self.password,icq=1))
|
