Python生成腾讯云实时音视频的UserSig签名

Posted 2021-01-11 朱兆筠

1.UserSig 简介

官方文档链接:https://cloud.tencent.com/document/product/647/16790

腾讯云 IM 的前身是 QQ 的即时通讯消息系统，我们将 QQ 的消息模块进行了抽离，变成了适合移动端接入的 IM SDK。将消息后台进行改造，使其脱离对 QQ 号码的绑定，构成了现在的 IM 云通讯后台。

QQ 可以用来收发单聊和群聊的消息，但前提是您必须先登录才能使用。登录 QQ 用的是 QQ 号和密码，登录 IM SDK 也是一样，只是肯定不能再用 QQ 号和密码了，而是使用您指定的用户名（userid）和密码（usersig）。

• 用户名（userid）
  可以是您目前 App 里的用户 ID，比如您有一个用户，他/她的账号 ID 是 27149， 那么您就可以用 27149 作为登录 IM SDK 的 userid。

• 密码（usersig）
  既然您指定了 27149 是您的用户，腾讯云如何才能确认该用户是您认可的合法用户呢？usersig 就是用于解决这个问题的，usersig 本质是对 userid、appid 等信息的

非对称加密非对称加密用的加密密钥和解密密钥是不同的，您的服务器可以持有私钥并对 userid 和 appid 进行非对称加密，加密之后的结果就是 usersig ；而腾讯云同步持有您的公钥，这样一来，腾讯云就可以确认 usersig 是否合法，从而可以确认它是否是由您的服务器签发的

2.生成 UserSig

官网的 SDK 下载区提供了一份计算 UserSig 的简单版源码，有 Java 和 PHP 两个版本的:https://github.com/TencentVideoCloudMLVBDev/usersig_server_source

下面的代码是用Python生成UserSig的方法,用的sanic框架,主要参考:http://bbs.qcloud.com/thread-14366-1-1.html

from sanic.views import HTTPMethodView
from sanic import response
from sanic import Blueprint
import OpenSSL
import base64
import zlib
import json
import time


UserSigList_Blueprint = Blueprint(__name__,url_prefix=‘/api/banpai/v1.2/‘)

#ec格式的私钥
# ecdsa_pri_key ="""
# -----BEGIN EC PRIVATE KEY-----
# MHcCAQEEIP+EunCa1keJ/ymk+MXhyb22+SHGgtdbYmdsBRzkWhohoAoGCCqGSM49
# AwEHoUQDQgAEXFBfDBCbwt0ZjBUMP9jJ/iqAcPeDi1UVzh51GuhGfNkRezQBTK/C
# 4rgi6oH24Asxeo1jnhFywp13qxFVYiKhtA==
# -----END EC PRIVATE KEY-----
# """

ecdsa_pri_key ="""
-----BEGIN EC PRIVATE KEY-----
MHcCAQEEIKLXR3F2OJkMrrzJtzhH52OnK+rVoUA4pcns3XdYoSfDoAoGCCqGSM49
AwEHoUQDQgAEbRGCpFwUNOjIMUHy7lcF3rDqZ0KvJpoua3VAiP3+5kjlYFHzLneg
0olpV4eJDb3RhGZ3bANgoOZdRxtlI1TR/Q==
-----END EC PRIVATE KEY-----
"""

# sdk_appid = 1400138652
sdk_appid = 1400135040


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(b‘+‘, b‘*‘)
    base64_data = base64_data.replace(b‘/‘, b‘-‘)
    base64_data = base64_data.replace(b‘=‘, b‘_‘)
    return base64_data


def base64_decode_url(base64_data):
    base64_data = base64_data.replace(b‘*‘, b‘+‘)
    base64_data = base64_data.replace(b‘-‘, b‘/‘)
    base64_data = base64_data.replace(b‘_‘, b‘=‘)
    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):
        return {"TLS.account_type": "%d" % self.__acctype, "TLS.identifier": "%s" % self.__identifier,
                "TLS.appid_at_3rd": "%s" % self.__appid3rd, "TLS.sdk_appid": "%d" % self.__sdkappid,
                "TLS.expire_after": "%d" % self.__expire, "TLS.version": "%d" % self.__version,
                "TLS.time": "%d" % time.time()}

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

        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.decode(‘utf-8‘)
        json_str = json.dumps(m)
        sig_compressed = zlib.compress(json_str.encode(‘utf-8‘))
        base64_sig = base64_encode_url(sig_compressed)
        return base64_sig

　　下面用文字简要描述下，
1.将用户的信息组装成一个字符串（json格式的，是直接拼装的，因为顺序不能乱），是哪些信息，可以看 __encode_to_fix_str；
2.使用 sha256 将字符串 hash，然后再用私钥签名，一般加密接口都会一把搞定，加密曲线使用的是 secp256k1；
3.把第2步得到的缓冲区进行base64；
4.将所有用户的信息以及第3步得到签名写进一个 json 串，此时可以不论顺序；
5.将 json 进行序列化，再 zlib 压缩，最后 base64（替换了某些字符，具体哪些看代码），出炉。

特别注意，这段代码在 windows 上验证没问题，但是红帽系(rel 和 centos)上不支持我们使用的曲线，list_all_curves 可以打印所有支持的曲线，红帽系的没有 secp256k1。

如果发现系统所带的 openssl 扩展不支持我们选定的曲线，可以参考 http://bbs.qcloud.com/thread-23280-1-1.html。