1. 程式人生 > >Python 服務端整合 騰訊雲 IM 服務

Python 服務端整合 騰訊雲 IM 服務

最近做的專案,需要接入騰訊雲 IM,翻看了一下文件,iOS、Android 以及 Web 端基本上都有 SDK 可以整合。我使用的服務端是用 Python 寫的,騰訊 IM 暫時還沒有 Python 的官方文件。但是在騰訊雲的官方論壇上找到了解決方法。
服務端最基本的一個需求是:使用接入使用者的 identifier 和應用申請的騰訊雲 appid、私鑰等資訊,通過指定演算法,生成使用者用來登入騰訊雲 IM 的 usersig。

原始碼:

#! /usr/bin/python
# coding:utf-8

import OpenSSL, base64, zlib, json, time

ecdsa_pri_key = """
自己的私鑰
"""

def list_all_curves():
    list = OpenSSL.crypto.get_elliptic_curves()
    for element in list:
        print element

def get_secp256k1():
    print OpenSSL.crypto.get_elliptic_curve('secp256k1');


def base64_encode_url(data):
    base64_data = base64.b64encode(data)
    base64_data = base64_data.replace('+', '*')
    base64_data = base64_data.replace('/', '-')
    base64_data = base64_data.replace('=', '_')
    return base64_data

def base64_decode_url(base64_data):
    base64_data = base64_data.replace('*', '+')
    base64_data = base64_data.replace('-', '/')
    base64_data = base64_data.replace('_', '=')
    raw_data = base64.b64decode(base64_data)
    return raw_data

class TLSSigAPI:
    """"""    
    __acctype = 0
    __identifier = ""
    __appid3rd = ""
    __sdkappid = 0
    __version = 20151204
    __expire = 3600*24*30       # 預設一個月,需要調整請自行修改
    __pri_key = ""
    __pub_key = ""
    _err_msg = "ok"

    def __get_pri_key(self):
        return OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, self.__pri_key);

    def __init__(self, sdkappid, pri_key):
        self.__sdkappid = sdkappid
        self.__pri_key = pri_key

    def __create_dict(self):
        m = {}
        m["TLS.account_type"] = "%d" % self.__acctype
        m["TLS.identifier"] = "%s" % self.__identifier
        m["TLS.appid_at_3rd"] = "%s" % self.__appid3rd
        m["TLS.sdk_appid"] = "%d" % self.__sdkappid
        m["TLS.expire_after"] = "%d" % self.__expire
        m["TLS.version"] = "%d" % self.__version
        m["TLS.time"] = "%d" % time.time()
        return m

    def __encode_to_fix_str(self, m):
        fix_str = "TLS.appid_at_3rd:"+m["TLS.appid_at_3rd"]+"\n" \
                  +"TLS.account_type:"+m["TLS.account_type"]+"\n" \
                  +"TLS.identifier:"+m["TLS.identifier"]+"\n" \
                  +"TLS.sdk_appid:"+m["TLS.sdk_appid"]+"\n" \
                  +"TLS.time:"+m["TLS.time"]+"\n" \
                  +"TLS.expire_after:"+m["TLS.expire_after"]+"\n"
        return fix_str

    def tls_gen_sig(self, identifier):
        self.__identifier = identifier

        m = self.__create_dict()
        fix_str = self.__encode_to_fix_str(m)
        pk_loaded = self.__get_pri_key()
        sig_field = OpenSSL.crypto.sign(pk_loaded, fix_str, "sha256");
        sig_field_base64 = base64.b64encode(sig_field)
        m["TLS.sig"] = sig_field_base64
        json_str = json.dumps(m)
        sig_cmpressed = zlib.compress(json_str)
        base64_sig = base64_encode_url(sig_cmpressed)
        return base64_sig 

def main():
    api = TLSSigAPI(1400001052, ecdsa_pri_key)
    sig = api.tls_gen_sig("xiaojun")
    print sig

if __name__ == "__main__":
    main()

過程:

  1. 將使用者的資訊組裝成一個字串(json格式的,是直接拼裝的,因為順序不能亂),是哪些資訊,可以看 __encode_to_fix_str;
  2. 使用 sha256 將字串 hash,然後再用私鑰簽名,一般加密介面都會一把搞定,加密曲線使用的是 secp256k1
  3. 把第2步得到的緩衝區進行 base64;
  4. 將所有使用者的資訊以及第3步得到簽名寫進一個 json 串,此時可以不論順序;
  5. 將 json 進行序列化,再 zlib 壓縮,最後 base64(替換了某些字元,具體哪些看程式碼)。

需要注意的是,加密曲線使用的是 secp256k1, 有些系統的 openssl 是不支援這個曲線的。檢視系統所支援的所有曲線,可參考 list_all_curves

, 這個方法會打印出所有支援的曲線。如果你的伺服器恰好支援這個曲線,那經過上面的過程就可以正常使用了。

如果你的伺服器不支援 secp256k1,可以使用 python ecdsa 。

  1. python ecdsa 開發庫 下載地址
  2. 使用 pip 安裝:pip install ecdsa

由於 python ecdsa 這個開發庫僅支援 ec 格式的私鑰,從騰訊雲下載的私鑰格式是 pk #8 的格式,需要使用 openssl 命令進行轉換。轉換命令如下:

openssl ec -outform PEM -inform PEM -in private.pem -out private_ec.pem

-in 後面的傳入下載的私鑰 -out 後面是轉換後的私鑰檔案

最終實現程式碼:

#! /usr/bin/python
# coding:utf-8

import OpenSSL, base64, zlib, json, time, hashlib
from ecdsa import SigningKey,util

# 這裡請填寫應用自己的私鑰
ecdsa_pri_key = """
請填上應用自己的私鑰
"""

def base64_encode_url(data):
    base64_data = base64.b64encode(data)
    base64_data = base64_data.replace('+', '*')
    base64_data = base64_data.replace('/', '-')
    base64_data = base64_data.replace('=', '_')
    return base64_data

def base64_decode_url(base64_data):
    base64_data = base64_data.replace('*', '+')
    base64_data = base64_data.replace('-', '/')
    base64_data = base64_data.replace('_', '=')
    raw_data = base64.b64decode(base64_data)
    return raw_data

class TLSSigAPI:
    """"""    
    __acctype = 0
    __identifier = ""
    __appid3rd = ""
    __sdkappid = 0
    __version = 20151204
    __expire = 3600*24*30       # 預設一個月,需要調整請自行修改
    __pri_key = ""
    __pub_key = ""
    _err_msg = "ok"
    

    def __get_pri_key(self):
        return self.__pri_key_loaded

    def __init__(self, sdkappid, pri_key):
        self.__sdkappid = sdkappid
        self.__pri_key = pri_key
        self.__pri_key_loaded = SigningKey.from_pem(self.__pri_key)

    def __create_dict(self):
        m = {}
        m["TLS.account_type"] = "%d" % self.__acctype
        m["TLS.identifier"] = "%s" % self.__identifier
        m["TLS.appid_at_3rd"] = "%s" % self.__appid3rd
        m["TLS.sdk_appid"] = "%d" % self.__sdkappid
        m["TLS.expire_after"] = "%d" % self.__expire
        m["TLS.version"] = "%d" % self.__version
        m["TLS.time"] = "%d" % time.time()
        return m

    def __encode_to_fix_str(self, m):
        fix_str = "TLS.appid_at_3rd:"+m["TLS.appid_at_3rd"]+"\n" \
                  +"TLS.account_type:"+m["TLS.account_type"]+"\n" \
                  +"TLS.identifier:"+m["TLS.identifier"]+"\n" \
                  +"TLS.sdk_appid:"+m["TLS.sdk_appid"]+"\n" \
                  +"TLS.time:"+m["TLS.time"]+"\n" \
                  +"TLS.expire_after:"+m["TLS.expire_after"]+"\n"
        return fix_str

    def tls_gen_sig(self, identifier):
        self.__identifier = identifier
        m = self.__create_dict()
        fix_str = self.__encode_to_fix_str(m)
        pk_loaded = self.__get_pri_key()
        sig_field = pk_loaded.sign(fix_str, hashfunc=hashlib.sha256, sigencode=util.sigencode_der)
        sig_field_base64 = base64.b64encode(sig_field)
        m["TLS.sig"] = sig_field_base64
        json_str = json.dumps(m)
        sig_cmpressed = zlib.compress(json_str)
        base64_sig = base64_encode_url(sig_cmpressed)
        return base64_sig 

def main():
    api = TLSSigAPI(1400001052, ecdsa_pri_key)
    sig = api.tls_gen_sig("xiaojun")
    print sig

if __name__ == "__main__":
    main()

參考連結,騰訊雲官方論壇:
http://bbs.qcloud.com/thread-14366-1-1.html
http://bbs.qcloud.com/thread-23280-1-1.html



作者:莫17
連結:https://www.jianshu.com/p/f7bf639a554e
來源:簡書
簡書著作權歸作者所有,任何形式的轉載都請聯絡作者獲得授權並註明出處。