Python 服務端整合 騰訊雲 IM 服務
阿新 • • 發佈：2018-11-21
最近做的專案，需要接入騰訊雲 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()
過程：
- 將使用者的資訊組裝成一個字串（json格式的，是直接拼裝的，因為順序不能亂），是哪些資訊，可以看 __encode_to_fix_str；
- 使用 sha256 將字串 hash，然後再用私鑰簽名，一般加密介面都會一把搞定，加密曲線使用的是 secp256k1；
- 把第2步得到的緩衝區進行 base64；
- 將所有使用者的資訊以及第3步得到簽名寫進一個 json 串，此時可以不論順序；
- 將 json 進行序列化，再 zlib 壓縮，最後 base64(替換了某些字元，具體哪些看程式碼)。
需要注意的是，加密曲線使用的是 secp256k1， 有些系統的 openssl 是不支援這個曲線的。檢視系統所支援的所有曲線，可參考 list_all_curves , 這個方法會打印出所有支援的曲線。如果你的伺服器恰好支援這個曲線，那經過上面的過程就可以正常使用了。
如果你的伺服器不支援 secp256k1，可以使用 python ecdsa 。
- python ecdsa 開發庫 下載地址
- 使用 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
來源：簡書
簡書著作權歸作者所有，任何形式的轉載都請聯絡作者獲得授權並註明出處。