Python论坛  - 讨论区

标题:[zeuux-python] Cocobear用纯Python实现飞信协议

2009年01月03日 星期六 10:00

Jianjun Kong jianjun在zeuux.org
星期六 一月 3 10:00:30 CST 2009

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 = '%s' % (ret,arg,extra)
        if cmd == "INFO":
            self.init('S')
            self.header.insert(3,('N',arg))
            if arg == "GetPersonalInfo":
                body = ''
            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>

[导入自Mailman归档:http://www.zeuux.org/pipermail/zeuux-python]

2009年01月03日 星期六 16:00

Zoom.Quiet zoom.quiet在gmail.com
星期六 一月 3 16:00:55 CST 2009

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 简历直投俺就成;-)

[导入自Mailman归档:http://www.zeuux.org/pipermail/zeuux-python]

2009年01月04日 星期日 08:22

Jianjun Kong jianjun在zeuux.org
星期日 一月 4 08:22:33 CST 2009

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

[导入自Mailman归档:http://www.zeuux.org/pipermail/zeuux-python]

如下红色区域有误,请重新填写。

    你的回复:

    请 登录 后回复。还没有在Zeuux哲思注册吗?现在 注册 !

    Zeuux © 2024

    京ICP备05028076号