2009年01月03日 星期六 10:00
Linux下使用飞信有很多方式,可以安装pidgin的插件,也可以安装其他客户端。 pidgin的飞信插件最新是v0.98,可以从sourceforge.net上下载到源代码 ($ cvs -d:pserver:anonymous在fetion.cvs.sourceforge.net:/cvsroot/fetion co fetion)。 不过作者从10月6日好后好像再没有更新过,他最近在开发一个python的独立飞信客户端,等后期再技术回流给插件版本。 还有一个由邓东东主持开发的飞信应用程序开发库LibFetion,其大部分代码使用C编写,支持所有POSIX兼容的操作系统,现在貌似移植到了很多平台。基于这个库,开发了一个linux的客户端,还有其他平台的客户端,最新版本v0.9.1。 下载地址:http://www.libfetion.cn/demoapp_download.html 不过不是开源的 :< 作者封装了一个 .a 的二进制文件,公开接口函数。 ------------------------------------------ 俺们的可可熊(http://cocobear.info),最近用纯Python实现了飞信协议,OPEN SOURCE :) 了解更多: http://cocobear.info/blog/?s=fetion ------------------------------------------ #!/usr/bin/env python # -*- coding: utf-8 -*- #Using GPL v2 #Author: cocobear.cn在gmail.com import urllib2 import urllib import cookielib import sys,re import binascii import hashlib import socket from hashlib import md5 from hashlib import sha1 from uuid import uuid1 FetionVer = "2008" #"SIPP" USED IN HTTP CONNECTION FetionSIPP= "SIPP" FetionNavURL = "nav.fetion.com.cn" FetionConfigURL = "http://nav.fetion.com.cn/nav/getsystemconfig.aspx" FetionConfigXML = """""" FetionLoginXML = """ """ debug = True class PyFetionException(Exception): """Base class for all exceptions raised by this module.""" class PyFetionInfoError(PyFetionException): """Phone number or password incomplete""" class PyFetionResponseException(PyFetionException): """Base class for all exceptions that include SIPC/HTTP error code. """ def __init__(self, code, msg): self.PyFetion_code = code self.PyFetion_error = msg self.args = (code, msg) class PyFetionAuthError(PyFetionResponseException): """Authentication error. Your password error, or your mobile NO. don't support fetion """ class PyFetionRegisterError(PyFetionResponseException): """RegisterError. """ class PyFetionSendError(PyFetionResponseException): """Send SMS error """ class PyFetion(): __config_data = "" __sipc_url = "" __sipc_proxy = "" __sid = "" mobile_no = "" passwd = "" login_type = "" def __init__(self,mobile_no,passwd,login_type="HTTP"): if not passwd or len(mobile_no) != 11: raise PyFetionInfoError(mobile_no,passwd) self.mobile_no = mobile_no self.passwd = passwd self.login_type = login_type self.__get_system_config() self.__set_system_config() def login(self): (self.__ssic,self.__domain) = self.__get_uri() try: self.__register(self.__ssic,self.__domain) except PyFetionRegisterError,e: print "Register Failed!" #这里使用一个status变量作为类的成员,每一种失败后都改变一下这个 pass def get_offline_msg(self): self.__SIPC.get("") def add(self,who): self.__SIPC.get("INFO","AddBuddy",who) response = self.__SIPC.send() code = self.__SIPC.get_code(response) if code == 521: d_print("Aleady added.") elif code == 522: d_print("Mobile NO. Don't Have Fetion") self.__SIPC.get("INFO","AddMobileBuddy",who) response = self.__SIPC.send() def get_personal_info(self): self.__SIPC.get("INFO","GetPersonalInfo") self.__SIPC.send() def get_info(self,who): self.__SIPC.get("INFO","GetContactsInfo",who) response = self.__SIPC.send() return response def get_contact_list(self): self.__SIPC.get("INFO","GetContactList") response = self.__SIPC.send() return response def get_uri(self,who): if who == self.mobile_no: who = self.__uri if not who.startswith("sip"): l = self.get_contact_list() all = re.findall('uri="(.+?)" ',l) #Get uri from contact list, compare one by one #I can't get other more effect way for uri in all: ret = self.get_info(uri) no = re.findall('mobile-no="(.+?)" ',ret) if no: if no[0] == who: d_print(('who',),locals()) who = uri break return who def send_msg(self,to,msg,flag="SENDMSG"): self.__SIPC.get(flag,to,msg) response = self.__SIPC.send() code = self.__SIPC.get_code(response) if code == 280: d_print("Send sms/msg OK!") else: d_print(('code',),locals()) def send_sms(self,msg,to=None,long=False): if not to: to = self.__uri else: to = self.get_uri(to) if long: self.send_msg(to,msg,"SENDCatSMS") else: self.send_msg(to,msg,"SENDSMS") def send_schedule_sms(self,msg,time,to=None): if not to: to = self.__uri else: to = self.get_uri(to) self.__SIPC.get("SSSetScheduleSms",msg,time,to) response = self.__SIPC.send() code = self.__SIPC.get_code(response) if code == 486: d_print("Busy Here") return None if code == 200: id = re.search('id="(\d+)"',response).group(1) d_print(('id',),locals(),"schedule_sms id") return id def __register(self,ssic,domain): self.__SIPC = SIPC(self.__sid,self.__domain,self.passwd,self.login_type,self.__http_tunnel,self.__ssic,self.__sipc_proxy) response = "" for step in range(1,3): self.__SIPC.get("REG",step,response) response = self.__SIPC.send() code = self.__SIPC.get_code(response) if code == 200: d_print("register successful.") else: raise PyFetionRegisterError(code,response) def __http_send(self,url,body="",exheaders="",login=False): headers = { 'User-Agent':'IIC2.0/PC 3.2.0540', } headers.update(exheaders) request = urllib2.Request(url,headers=headers,data=body) try: conn = urllib2.urlopen(request) except urllib2.URLError, e: code = e.code msg = e.read() if code == 401 or code == 404: if login: d_print(('code','text'),locals()) raise PyFetionAuthError(code,msg) return -1 return conn def __get_system_config(self): global FetionConfigURL global FetionConfigXML url = FetionConfigURL body = FetionConfigXML % self.mobile_no d_print(('url','body'),locals()) self.__config_data = self.__http_send(url,body).read() def __set_system_config(self): sipc_url = re.search(" (.*) ",self.__config_data).group(1) sipc_proxy = re.search("(.*) ",self.__config_data).group(1) http_tunnel = re.search("(.*) ",self.__config_data).group(1) d_print(('sipc_url','sipc_proxy','http_tunnel'),locals()) self.__sipc_url = sipc_url self.__sipc_proxy = sipc_proxy self.__http_tunnel= http_tunnel def __get_uri(self): url = self.__sipc_url+"?mobileno="+self.mobile_no+"&pwd;="+self.passwd d_print(('url',),locals()) try: ret = self.__http_send(url,login=True) except PyFetionAuthError,e: d_print(('e',),locals()) print "Your password error, or your mobile NO. don't support fetion" sys.exit(-1) header = str(ret.info()) body = ret.read() ssic = re.search("ssic=(.*);",header).group(1) sid = re.search("sip:(.*)@",body).group(1) uri = re.search('uri="(.*)" mobile-no',body).group(1) status = re.search('user-status="(\d+)"',body).group(1) domain = "fetion.com.cn" d_print(('ssic','sid','uri','status','domain'),locals(),"Get SID OK") self.__sid = sid self.__uri = uri return (ssic,domain) class SIPC(): global FetionVer global FetionSIPP global FetionLoginXML header = "" body = "" content = "" code = '' ver = "SIP-C/2.0" ID = 1 sid = "" domain = "" passwd = "" __http_tunnel = "" def __init__(self,sid,domain,passwd,login_type,http_tunnel,ssic,sipc_proxy): self.sid = sid self.domain = domain self.passwd = passwd self.login_type = login_type self.domain = domain self.sid = sid self.__seq = 1 self.__sipc_proxy = sipc_proxy if self.login_type == "HTTP": self.__http_tunnel = http_tunnel self.__ssic = ssic guid = str(uuid1()) self.__exheaders = { 'Cookie':'ssic=%s' % self.__ssic, 'Content-Type':'application/oct-stream', 'Pragma':'xz4BBcV%s' % guid, } def init(self,type): self.content = '%s %s %s\r\n' % (type,self.domain,self.ver) self.header = [('F',self.sid), ('I',self.ID), ('Q','1 %s' % type), ] def send(self): content = self.content d_print(('content',),locals()) if self.login_type == "HTTP": #First time t SHOULD SET AS 'i' #Otherwise 405 code get if self.__seq == 1: t = 'i' else: t = 's' url = self.__http_tunnel+"?t=%s&i;=%s" % (t,self.__seq) response = self.__http_send(url,content,self.__exheaders).read() self.__seq+=1 response = self.__sendSIPP() #This line will enhance the probablity of success. #Sometimes it will return FetionSIPP twice. #Probably you need add more if response == FetionSIPP: response = self.__sendSIPP() else: if self.__seq == 1: self.__tcp_init() self.__tcp_send(content) response = self.__tcp_recv() d_print(('response',),locals()) self.__seq+=1 code = self.get_code(response) d_print(('code',),locals()) return response def get_code(self,response): try: self.code =int(re.search("%s (\d{3})" % self.ver,response).group(1)) self.msg =re.search("%s \d{3} (.*)\r" % self.ver,response).group(1) d_print(('self.code','self.msg',),locals()) return self.code except AttributeError,e: return None def get(self,cmd,arg,ret="",extra=""): body = ret if cmd == "REG": body = FetionLoginXML self.init('R') if arg == 1: pass if arg == 2: nonce = re.search('nonce="(.*)"',ret).group(1) cnonce = self.__get_cnonce() if FetionVer == "2008": response=self.__get_response_sha1(nonce,cnonce) elif FetionVer == "2006": response=self.__get_response_md5(nonce,cnonce) salt = self.__get_salt() d_print(('nonce','cnonce','response','salt'),locals()) #If this step failed try to uncomment this lines #del self.header[2] #self.header.insert(2,('Q','2 R')) if FetionVer == "2008": self.header.insert(3,('A','Digest algorithm="SHA1-sess",response="%s",cnonce="%s",salt="%s"' % (response,cnonce,salt))) elif FetionVer == "2006": self.header.insert(3,('A','Digest response="%s",cnonce="%s"' % (response,cnonce))) #If register successful 200 code get if arg == 3: return self.code if cmd == "SENDMSG": self.init('M') self.header.insert(3,('T',arg)) self.header.insert(4,('C','text/plain')) self.header.insert(5,('K','SaveHistory')) if cmd == "SENDSMS": self.init('M') self.header.append(('T',arg)) self.header.append(('N','SendSMS')) if cmd == "SENDCatSMS": self.init('M') self.header.append(('T',arg)) self.header.append(('N','SendCatSMS')) if cmd == "SSSetScheduleSms": self.init('S') self.header.insert(3,('N',cmd)) body = '' % (ret,arg,extra) if cmd == "INFO": self.init('S') self.header.insert(3,('N',arg)) if arg == "GetPersonalInfo": body = ' %s ' elif arg == "GetContactList": body = ' ' elif arg == "GetContactsInfo": body = ' ' % ret elif arg == "AddBuddy": body = ' ' % ret elif arg == "AddMobileBuddy": body = ' ' % ret #general SIPC info self.header.append(('L',len(body))) for k in self.header: self.content = self.content + k[0] + ": " + str(k[1]) + "\r\n" self.content+="\r\n" self.content+= body if self.login_type == "HTTP": #IN TCP CONNECTION "SIPP" SHOULD NOT BEEN SEND self.content+= FetionSIPP return self.content def __sendSIPP(self): body = FetionSIPP url = self.__http_tunnel+"?t=s&i;=%s" % self.__seq response = self.__http_send(url,body,self.__exheaders).read() d_print(('response',),locals()) self.__seq+=1 return response def __http_send(self,url,body="",exheaders="",login=False): headers = { 'User-Agent':'IIC2.0/PC 3.2.0540', } headers.update(exheaders) request = urllib2.Request(url,headers=headers,data=body) try: conn = urllib2.urlopen(request) except urllib2.URLError, e: code = e.code msg = e.read() d_print(('code','text'),locals()) if code == 401 or code == 404: if login: raise PyFetionAuthError(code,msg) return -1 return conn def __tcp_init(self): try: self.__sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM) except socket.error,e: s = None print e.read() #Should return -1 NOT just exit sys.exit(-1) (host,port) = tuple(self.__sipc_proxy.split(":")) port = int(port) try: self.__sock.connect((host,port)) except socket.error,e: self.__sock.close() self.__sock = None print e.read() sys.exit(-1) def __tcp_send(self,msg): try: self.__sock.send(msg) except socket.error,e: self.__sock.close() print e.read() sys.exit(-1) def __tcp_recv(self): try: data = self.__sock.recv(4096) except socket.error,e: self.__sock.close() print e.read() sys.exit(-1) return data def __get_salt(self): return self.__hash_passwd()[:8] def __get_cnonce(self): return md5(str(uuid1())).hexdigest().upper() def __get_response_md5(self,nonce,cnonce): #nonce = "3D8348924962579418512B8B3966294E" #cnonce= "9E169DCA9CBD85F1D1A89A893E00917E" key = md5("%s:%s:%s" % (self.sid,self.domain,self.passwd)).digest() h1 = md5("%s:%s:%s" % (key,nonce,cnonce)).hexdigest().upper() h2 = md5("REGISTER:%s" % self.sid).hexdigest().upper() response = md5("%s:%s:%s" % (h1,nonce,h2)).hexdigest().upper() #d_print(('nonce','cnonce','key','h1','h2','response'),locals()) return response def __get_response_sha1(self,nonce,cnonce): #nonce = "3D8348924962579418512B8B3966294E" #cnonce= "9E169DCA9CBD85F1D1A89A893E00917E" hash_passwd = self.__hash_passwd() hash_passwd_str = binascii.unhexlify(hash_passwd[8:]) key = sha1("%s:%s:%s" % (self.sid,self.domain,hash_passwd_str)).digest() h1 = md5("%s:%s:%s" % (key,nonce,cnonce)).hexdigest().upper() h2 = md5("REGISTER:%s" % self.sid).hexdigest().upper() response = md5("%s:%s:%s" % (h1,nonce,h2)).hexdigest().upper() return response def __hash_passwd(self): #salt = '%s%s%s%s' % (chr(0x77), chr(0x7A), chr(0x6D), chr(0x03)) salt = 'wzm\x03' src = salt+sha1(self.passwd).digest() return "777A6D03"+sha1(src).hexdigest().upper() def d_print(vars=(),namespace=[],msg=""): """if only sigle variable use like this ('var',)""" global debug if vars and not namespace and not msg: msg = vars if debug and vars and namespace: for var in vars: if var in namespace: print "[PyFetion]:\033[0;31;48m%s\033[0m" % var, print namespace[var] if debug and msg: print "[PyFetion]:\033[0;31;48m%s\033[0m" % msg def main(argv=None): try: phone = PyFetion("138888888","888888","TCP") except PyFetionInfoError,e: print "corrent your mobile NO. and password" return -1 phone.login() #phone.get_offline_msg() #phone.add("138888888") #phone.get_info() #phone.get_contact_list() #phone.send_sms("Hello, ",long=True) s = "2008-12-31 02:39:00." for i in range(100,500): time = s + str(i) phone.send_schedule_sms("请注意,这个是定时短信",time) #time_format = "%Y-%m-%d %H:%M:%S" #time.strftime(time_format,time.gmtime()) if __name__ == "__main__": sys.exit(main()) -- Jianjun Kong|Happy Hacking Homepage:http://kongove.cn kongjianjun (at) gmail.com -------------- 下一部分 -------------- A non-text attachment was scrubbed... Name: PyFetion.py Type: text/x-python Size: 18613 bytes Desc: 不可用 URL: <http://www.zeuux.org/pipermail/zeuux-python/attachments/20090103/a5c18fd9/attachment-0001.py>
2009年01月03日 星期六 16:00
2009/1/3 Jianjun Kong <jianjun在zeuux.org>: > Linux下使用飞信有很多方式,可以安装pidgin的插件,也可以安装其他客户端。 > > pidgin的飞信插件最新是v0.98,可以从sourceforge.net上下载到源代码 > ($ cvs -d:pserver:anonymous在fetion.cvs.sourceforge.net:/cvsroot/fetion co fetion)。 > 不过作者从10月6日好后好像再没有更新过,他最近在开发一个python的独立飞信客户端,等后期再技术回流给插件版本。 > > 还有一个由邓东东主持开发的飞信应用程序开发库LibFetion,其大部分代码使用C编写,支持所有POSIX兼容的操作系统,现在貌似移植到了很多平台。基于这个库,开发了一个linux的客户端,还有其他平台的客户端,最新版本v0.9.1。 > 下载地址:http://www.libfetion.cn/demoapp_download.html > 不过不是开源的 :< 作者封装了一个 .a 的二进制文件,公开接口函数。 > ------------------------------------------ > > 俺们的可可熊(http://cocobear.info),最近用纯Python实现了飞信协议,OPEN SOURCE :) > 了解更多: http://cocobear.info/blog/?s=fetion 好熊!09 开门第一件好事儿! http://wiki.woodpecker.org.cn/moin/MiscItems/2009-01-03 -- http://zoomquiet.org '''过程改进乃是催生可促生靠谱的人的组织!''' 金山常年招聘Py/C++人才! http://bit.ly/UoTV 简历直投俺就成;-)
2009年01月04日 星期日 08:22
On Sat, Jan 03, 2009 at 04:00:55PM +0800, Zoom.Quiet wrote: >2009/1/3 Jianjun Kong <jianjun在zeuux.org>: >> >> 俺们的可可熊(http://cocobear.info),最近用纯Python实现了飞信协议,OPEN SOURCE :) >> 了解更多: http://cocobear.info/blog/?s=fetion # 支持飞信2006,2008协议(其实就是分别使用MD5、SHA1算法进行登录认证); # 支持HTTP、TCP方式; # 支持给自己手机发短信(这个也是我的主要目的); # 支持直接发送信息到指定手机号(前提是加为好友,好处是直接使用手机号,而不需要知道飞信号); # 支持添加好友 # 支持定时发送信息 # 。。。。 对二次开发很有用 :) >好熊!09 开门第一件好事儿! >http://wiki.woodpecker.org.cn/moin/MiscItems/2009-01-03 -- Jianjun Kong|Happy Hacking Homepage:http://kongove.cn kongjianjun (at) gmail.com
Zeuux © 2024
京ICP备05028076号