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

Posted 朱兆筠

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Python生成腾讯云实时音视频的UserSig签名相关的知识,希望对你有一定的参考价值。

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

 












以上是关于Python生成腾讯云实时音视频的UserSig签名的主要内容,如果未能解决你的问题,请参考以下文章

实时音视频技术的演进与应用

摸索在web端使用腾讯云的实时音视频

行业实践-Apache Flink 在移动云实时计算的实践

热榜标题词云实时更新词云上线,给标题起名提供参观建议

错过了太后悔,九大绝招大公开,详解华为低时延技术

回顾 | 阿里云实时计算专场 · 北京站