邮政开放平台API中SM4报文加密及参数配置代码分享

本分享来自昨天的社区问答，  EMS开放平台API查询物流报错，   和影刀无关，涉及SM4加密，已解决。

SM4是国密对称秘钥加密算法，python中有成熟的库gmssl可供驱使。

邮政API中签名（这里不应该叫签名，就是对报文加密而已）中，将报文后面直接加上秘钥，然后再用这个秘钥做SM4加密，然后将加密结果变换为base64编码，前面再加上|$4|。SM4中使用ECB（电子密码表）模式，PKCS7填充。

邮政开发者后台有个工具箱---签名校验可以用来比对加密结果

主要是这邮政开发API文档看得人头晕，没有整体说明，也没有示例

直接上代码吧，此处接口代码为“040001”（运单轨迹信息获取）。注意测试和生产参数和URL

# 使用提醒:
# 1. xbot包提供软件自动化、数据表格、Excel、日志、AI等功能
# 2. package包提供访问当前应用数据的功能，如获取元素、访问全局变量、获取资源文件等功能
# 3. 当此模块作为流程独立运行时执行main函数
# 4. 可视化流程中可以通过"调用模块"的指令使用此模块

import xbot
from xbot import print, sleep
from .import package
from .package import variables as glv

# -*- coding: utf-8 -*-
"""
影刀可用 · EMS 轨迹查询（POST 表单 + SM4）
修正：只加密 logitcsInterface 值
"""
import json, base64, requests
from datetime import datetime
from gmssl.sm4 import CryptSM4, SM4_ENCRYPT
from gmssl import func

# ========== 最新表格参数，按实际替换 ==========
API_CODE   = "040001"
SENDER_NO  = "11xxxxxxxxx58"
AUTH_KEY   = "SfxXxx9QSxXxxSZk"           # 明文授权码  
SM4_KEY    = "Um1XXXXXXXJmcU5wxxxxcA=="   # 32-byte SM4 密钥（Base64）
TRACK_URL  = "https://api.ems.com.cn/amp-prod-api/f/amp/api/open"
TIMEOUT    = 15
# ================================

def sm4_encrypt_ecb(plaintext: str, b64_key: str) -> str:
    key_bytes = base64.b64decode(b64_key)
    crypt = CryptSM4()
    crypt.set_key(key_bytes, SM4_ENCRYPT)
    padded = func.bytes_to_list(plaintext.encode('utf-8'))
    pad_len = 16 - (len(padded) % 16)
    padded += [pad_len] * pad_len
    ct = crypt.crypt_ecb(padded)
    return '|$4|' + base64.b64encode(func.list_to_bytes(ct)).decode()
    #return base64.b64encode(func.list_to_bytes(ct)).decode()

def query(track_no: str) -> dict:
    log_if = json.dumps({"waybillNo": track_no}, separators=(',', ':'), ensure_ascii=False)

    log_if = log_if + SM4_KEY

    # 1. 只加密业务报文（官方做法）
    cipher_text = sm4_encrypt_ecb(log_if, SM4_KEY)

    # 2. 组装 x-www-form-urlencoded
    form = {
        "apiCode":   API_CODE,
        "senderNo":  SENDER_NO,
        "authorization": AUTH_KEY,
        "timeStamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        #"waybillNo":track_no,
        "logitcsInterface": cipher_text
    }

    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
        "User-Agent":   "Mozilla/5.0"
    }
    rsp = requests.post(TRACK_URL, data=form, headers=headers, timeout=TIMEOUT)
    rsp.raise_for_status()
    return rsp.json()

# 影刀入口
def main_func(track_no: str = "") -> dict:
    if not track_no:
        raise ValueError("快递单号不能为空")
    ret = query(track_no)
    code = ret.get("retCode")
    if code != "00000":
        raise RuntimeError(f"EMS 业务异常[{code}]: {ret.get('retMsg')}")
    traces = ret.get("retBody", {})
    #print(f"查询成功！共 {len(traces)} 条轨迹")
    print(traces)

    return ret

'''
if __name__ == '__main__':
    try:
        main("EAxxxxxxxxxCN")   # 换成真实单号
    except Exception as e:
        print("[ERROR]", e)
'''

def main(args):
    main_func("127xxxxxx4331")
    pass

------------  Enjoy --------------