SmsForwarder使用 SM4 加密主动控制接口Python调用案例

cli.py

import base64
import hmac
import time
import json
import urllib.parse
from gmssl.sm4 import CryptSM4, SM4_ENCRYPT, SM4_DECRYPT


class SmsF:
    """
    SmsF class
    文档: https://github.com/pppscn/SmsForwarder/wiki/%E9%99%84%E5%BD%952%EF%BC%9A%E4%B8%BB%E5%8A%A8%E8%AF%B7%E6%B1%82(%E8%BF%9C%E7%A8%8B%E6%8E%A7%E5%88%B6)
    """

    def __init__(self, host: str, key_str: str):
        """
        __init__ 初始化
        :param key_str: APP 生成的秘钥字符串
        :param host: 接口域名
        """
        self.host = host
        self.key_str = key_str
        self.key = bytes.fromhex(key_str)
        self.iv = bytes([3, 5, 6, 9, 6, 9, 5, 9, 3, 5, 6, 9, 6, 9, 5, 9])
        self.sm4 = CryptSM4()

    def _encode(self, _data: dict):
        """
        encode 加密参数
        :param _data: 待加密数据
        :return: 加密后的16 进制字符串
        """
        _data = json.dumps(_data)
        self.sm4.set_key(self.key, SM4_ENCRYPT)
        _sec = self.sm4.crypt_cbc(self.iv, bytes(_data, encoding="utf8"))
        return _sec.hex()

    def _decode(self, _data):
        """
        decode 解码参数
        :param _data: 待解码字符串
        :return: 解码后的字符串
        """
        self.sm4.set_key(self.key, SM4_DECRYPT)
        decrypt_value = self.sm4.crypt_cbc(self.iv, bytes.fromhex(_data))  # bytes类型
        # print("国密解码后数据:", decrypt_value)
        return decrypt_value.decode("utf8")

    def _get_sign(self):
        """
        get_sign 获取签名
        :return: 签名和时间戳
        """
        # 把 timestamp+"\n"+密钥 当做签名字符串，使用 HmacSHA256 算法计算签名，然后进行 Base64 encode，最后再把签名参数再进行urlEncode，得到最终的签名（需要使用UTF-8字符集）
        timestamp = int(round(time.time() * 1000))
        data = str(timestamp) + "\n" + self.key_str
        data_bytes = data.encode('utf-8')
        # 计算HmacSHA256
        signature = hmac.new(self.key, data_bytes, digestmod='sha256').digest()
        # Base64 编码
        base64_signature = base64.b64encode(signature).decode('utf-8')
        # urlEncode
        url_encoded_signature = urllib.parse.quote(base64_signature, safe='')
        return url_encoded_signature, timestamp

    def _get(self, _api: str, _data: dict = None) -> dict:
        """
        _get 发送请求
        :param _api: 接口地址
        :param _data: 请求参数
        :return: 请求结果
        """
        # 处理默认值
        if _data is None:
            _data = {}
        # 获取签名和时间戳
        _sign, _timestamp = self._get_sign()
        # 组装参数
        _param = {
            "data": _data,
            "timestamp": _timestamp,
            "sign": _sign
        }
        _param_enc = self._encode(_param)
        # 请求数据，参数为加密后的参数
        import requests
        _res = requests.post(_api, headers={'content-type': "application/json"}, data=_param_enc)
        # 解码返回结果
        _res_dec = self._decode(_res.text)
        # 转换返回数据为字典
        _resp = json.loads(_res_dec)
        if _resp["code"] != 200:
            raise Exception(_resp["msg"])
        # 返回结果
        return _resp

    def query_config(self):
        """
        query_config 查询配置
        :return: 配置信息
        """
        _api = self.host + "/config/query"
        return self._get(_api)

    def clone_pull(self, _version: int):
        """
        clone_pull 一键换新机
        :param _version: 客户端App版本号（服务端与客户端的版本号必须一致）
        :return: 配置信息
        """
        if _version is None:
            raise Exception("version is None")
        _api = self.host + "/clone/pull"
        _param = {
            "version_code": _version
        }
        return self._get(_api, _param)

    def clone_push(self, _data: dict):
        """
        clone_push 一键换新机
        :param _data: 请求参数(参见https://github.com/pppscn/SmsForwarder/wiki/%E9%99%84%E5%BD%952%EF%BC%9A%E4%B8%BB%E5%8A%A8%E8%AF%B7%E6%B1%82(%E8%BF%9C%E7%A8%8B%E6%8E%A7%E5%88%B6)#212-%E5%AE%A2%E6%88%B7%E7%AB%AF%E5%90%91%E6%9C%8D%E5%8A%A1%E7%AB%AF%E6%8E%A8%E9%80%81%E9%85%8D%E7%BD%AE)
        :return: 是否成功等信息
        """
        _api = self.host + "/clone/push"
        return self._get(_api, _data)

    def sms_send(self, _slot: int, _phone: str, _content: str):
        """
        sms_send 发送短信
        :param _slot: 卡槽(1:卡1,2:卡2)
        :param _phone: 手机号(多个手机号用英文分号[;]分隔)
        :param _content: 短信内容
        :return: 是否成功等信息
        """
        _api = self.host + "/sms/send"
        _param = {
            "sim_slot": _slot,
            "phone_numbers": _phone,
            "msg_content": _content
        }
        return self._get(_api, _param)

    def sms_query(self, _type: int = 1, _page: int = 1, _size: int = 10, _keyword: str = None):
        """
        sms_query 查询短信
        :param _type: 1:收件箱,2:发件箱
        :param _page: 页码
        :param _size: 每页数量
        :param _keyword: 查询关键字
        :return: 短信列表
        """
        _api = self.host + "/sms/query"
        _param = {
            "type": _type,
            "page_num": _page,
            "page_size": _size
        }
        if _keyword is not None:
            _param["keyword"] = _keyword
        return self._get(_api, _param)

    def call_query(self, _type: int = 0, _page: int = 1, _size: int = 10, _phone: str = None):
        """
        call_query 查询通话记录
        :param _type: 通话类型：1=呼入, 2=呼出, 3=未接，0=不筛选（默认）
        :param _page: 页码
        :param _size: 每页数量
        :param _phone: 查询手机号(模糊匹配)
        :return: 通话记录
        """
        _api = self.host + "/call/query"
        _param = {
            "type": _type,
            "page_num": _page,
            "page_size": _size
        }
        if _phone is not None:
            _param["phone_number"] = _phone
        return self._get(_api, _param)

    def contact_query(self, _phone: str = None, _name: str = None):
        """
        contact_query 查询联系人
        :param _phone: 查询手机号(模糊匹配)
        :param _name: 查询姓名(模糊匹配)
        :return: 联系人列表
        """
        _api = self.host + "/contact/query"
        _param = {}
        if _phone is not None:
            _param["phone_number"] = _phone
        if _name is not None:
            _param["name"] = _name
        return self._get(_api, _param)

    def contact_add(self, _name: str, _phone: str):
        """
        contact_add 添加联系人
        :param _name: 姓名
        :param _phone: 手机号，多个用半角分号分隔，例：15888888888;19999999999
        :return: 是否成功等信息
        """
        _api = self.host + "/contact/add"
        _param = {
            "name": _name,
            "phone_number": _phone
        }
        return self._get(_api, _param)

    def battery_query(self):
        """
        battery_query 查询电量
        :return: 电量信息
        """
        _api = self.host + "/battery/query"
        return self._get(_api)

    def wol_send(self, _mac:str, _ip:str = None, _port:int = 9):
        """
        wol_send 发送WOL唤醒包
        :param _mac: 网卡MAC地址
        :param _ip: IP地址
        :param _port: 端口号：7 或 9，默认：9；仅传入ip节点时有效
        :return: 是否成功等信息
        """
        _api = self.host + "/wol/send"
        _param = {
            "mac": _mac
        }
        if _ip is not None:
            _param["ip"] = _ip
        if _port is not None:
            if _ip is None:
                raise Exception("ip is None")
            _param["port"] = _port

        return self._get(_api, _param)

    def location_query(self):
        """
        location_query 查询定位
        :return: 定位信息
        """
        _api = self.host + "/location/query"
        return self._get(_api)

test.py

from cli import SmsF

if __name__ == '__main__':
    _key = "APP上生成的秘钥字符串"
    _host = "调用接口" # 例: https://sms.example.com
    sms = SmsF(_host, _key)
    # 查询电量
    _battery = sms.battery_query()
    print("充电状态: ", _battery["data"]["status"])

使用方式:

1. 下载cli.py文件，放在自己喜欢的地方
2. 引用cli.py文件，然后参考test.py调用即可