🐍 Python

声明

本 SDK 由 AI 生成，仅供参考使用。建议在实际生产环境使用前进行充分测试。

概述

平台 Python Server SDK 提供了便捷的服务端接入方式，支持签名生成、请求加密、响应验签和解密等核心功能。

功能特性

• 支持 SHA256withRSA 签名算法
• 支持 AES 对称加密/解密
• 灵活的 HTTP 客户端接口设计，支持多种 HTTP 客户端实现
• 自动处理签名和加密逻辑
• 完整的响应验签和解密

环境要求

• Python 3.7+
• pip

安装

安装依赖

pip install requests cryptography

requirements.txt

requests>=2.31.0
cryptography>=41.0.0

快速开始

1. 创建配置

from kudian_sdk import KudianClient, HttpClient
from kudian_sdk.http import RequestsClient

# 创建配置
config = {
    'gateway_url': 'https://pay.kudianvip.com',  # 正式环境
    # 'gateway_url': 'https://pay.test.kudianvip.com',  # 测试环境
    'mch_id': 'your_mch_id',
    'app_id': 'your_app_id',
    'app_private_key': 'your_app_private_key',  # Base64格式
    'app_secret_key': 'your_secret_key',  # AES加密密钥
    'api_public_key': 'your_api_public_key'  # Base64格式
}

# 创建客户端
client = KudianClient(config, RequestsClient())

2. 发起API请求

import json

# 构造业务参数
biz_params = {
    'param1': 'value1',
    'param2': 'value2'
}

# 发送请求
response = client.post('/api/path', biz_params)

# 解析响应
api_response = json.loads(response)

if api_response['code'] == 0:
    print(f"业务数据: {api_response['data']}")
else:
    print(f"请求失败: {api_response['msg']}")

核心 API

KudianClient

主客户端类，提供所有API调用功能。

配置参数

参数	类型	必填	说明
gatewayUrl	str	是	网关地址，正式环境: https://pay.kudianvip.com，测试环境: https://pay.test.kudianvip.com
mchId	str	是	商户编号
appId	str	是	应用ID
appPrivateKey	str	是	应用私钥（Base64编码格式），用于请求签名
appSecretKey	str	是	AES加密密钥，用于请求/响应内容加密
apiPublicKey	str	是	API平台公钥（Base64编码格式），用于响应验签

主要方法

post(api_path: str, biz_params: dict) -> str

发送POST请求到指定API路径。

参数:

• api_path: API路径，如 /api/pay/create
• biz_params: 业务参数字典

返回:

• 自动解密后的业务数据JSON字符串

示例:

params = {
    'order_no': 'ORDER202401010001',
    'amount': 10000,
    'subject': '测试订单'
}

result = client.post('/api/pay/create', params)

verify_notify(notify_params: dict) -> bool

验证回调通知的签名。

参数:

• notify_params: 回调参数字典

返回:

• True: 验签成功
• False: 验签失败

示例:

# 在你的Flask/Django视图函数中
def handle_notify(request):
    params = request.json

    # 验证签名
    if not client.verify_notify(params):
        return 'FAIL'

    # 获取业务数据（已自动解密）
    biz_data = client.decrypt_notify(params['result'])
    print(f"回调数据: {biz_data}")

    # 处理业务逻辑...

    return 'SUCCESS'

完整代码实现

1. crypto_utils.py - 加密工具类

import base64
import hashlib
import json
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend


class CryptoUtils:
    """加密工具类"""

    @staticmethod
    def aes_encrypt(plain_text: str, key: str) -> str:
        """
        AES加密

        Args:
            plain_text: 明文
            key: 密钥

        Returns:
            Base64编码的密文
        """
        # 确保密钥是32字节（AES-256）
        key_bytes = key.encode('utf-8')[:32].ljust(32, b'\0')

        # AES加密
        cipher = Cipher(
            algorithms.AES(key_bytes),
            modes.ECB(),
            backend=default_backend()
        )
        encryptor = cipher.encryptor()

        # PKCS5填充
        padding_length = 16 - (len(plain_text.encode('utf-8')) % 16)
        padded_data = plain_text.encode('utf-8') + (chr(padding_length) * padding_length).encode('utf-8')

        encrypted = encryptor.update(padded_data) + encryptor.finalize()
        return base64.b64encode(encrypted).decode('utf-8')

    @staticmethod
    def aes_decrypt(cipher_text: str, key: str) -> str:
        """
        AES解密

        Args:
            cipher_text: Base64编码的密文
            key: 密钥

        Returns:
            明文
        """
        # 确保密钥是32字节（AES-256）
        key_bytes = key.encode('utf-8')[:32].ljust(32, b'\0')

        # AES解密
        cipher = Cipher(
            algorithms.AES(key_bytes),
            modes.ECB(),
            backend=default_backend()
        )
        decryptor = cipher.decryptor()

        decoded = base64.b64decode(cipher_text)
        decrypted = decryptor.update(decoded) + decryptor.finalize()

        # 移除PKCS5填充
        padding_length = decrypted[-1]
        return decrypted[:-padding_length].decode('utf-8')

    @staticmethod
    def sign_sha256_with_rsa(message: bytes, private_key_bytes: bytes) -> str:
        """
        SHA256withRSA签名

        Args:
            message: 待签名消息
            private_key_bytes: 私钥字节数组（Base64解码后的DER格式）

        Returns:
            Base64编码的签名
        """
        private_key = serialization.load_der_private_key(
            private_key_bytes,
            password=None,
            backend=default_backend()
        )

        signature = private_key.sign(
            message,
            padding.PKCS1v15(),
            hashes.SHA256()
        )

        return base64.b64encode(signature).decode('utf-8')

    @staticmethod
    def verify_sha256_with_rsa(message: bytes, public_key_bytes: bytes, sign: str) -> bool:
        """
        SHA256withRSA验签

        Args:
            message: 待验签消息
            public_key_bytes: 公钥字节数组（Base64解码后的DER格式）
            sign: Base64编码的签名

        Returns:
            True-验签成功，False-验签失败
        """
        public_key = serialization.load_der_public_key(
            public_key_bytes,
            backend=default_backend()
        )

        try:
            public_key.verify(
                base64.b64decode(sign),
                message,
                padding.PKCS1v15(),
                hashes.SHA256()
            )
            return True
        except Exception:
            return False

2. sign_utils.py - 签名工具类

from typing import Dict


class SignUtils:
    """签名工具类"""

    @staticmethod
    def make_sign_string(params: Dict[str, str]) -> str:
        """
        生成签名字符串
        将参数按ASCII码排序后拼接成 key1=value1&key2=value2 格式

        Args:
            params: 参数字典

        Returns:
            待签名字符串
        """
        # 过滤sign字段和空值
        filtered_params = []
        for key, value in params.items():
            if key != 'sign' and value:
                filtered_params.append(key)

        # 按ASCII码排序
        filtered_params.sort()

        # 拼接字符串
        pairs = [f"{key}={params[key]}" for key in filtered_params]
        return "&".join(pairs)

3. http_client.py - HTTP客户端接口

from abc import ABC, abstractmethod
from typing import Dict, Optional


class HttpClient(ABC):
    """HTTP客户端接口"""

    @abstractmethod
    def post(self, url: str, headers: Dict[str, str], request_body: str) -> str:
        """
        发送POST请求

        Args:
            url: 请求URL
            headers: 请求头
            request_body: 请求体（JSON字符串）

        Returns:
            响应体（JSON字符串）

        Raises:
            Exception: 请求异常
        """
        pass

4. requests_client.py - Requests客户端实现

import requests
from typing import Dict
from kudian_sdk.http import HttpClient


class RequestsClient(HttpClient):
    """基于Requests的HTTP客户端实现"""

    def __init__(self, timeout: int = 30):
        """
        初始化客户端

        Args:
            timeout: 超时时间（秒），默认30秒
        """
        self.timeout = timeout

    def post(self, url: str, headers: Dict[str, str], request_body: str) -> str:
        """
        发送POST请求

        Args:
            url: 请求URL
            headers: 请求头
            request_body: 请求体（JSON字符串）

        Returns:
            响应体（JSON字符串）

        Raises:
            Exception: 请求异常
        """
        response = requests.post(
            url,
            headers=headers,
            data=request_body,
            timeout=self.timeout
        )

        response.raise_for_status()
        return response.text

5. client.py - 主客户端类

import base64
import json
import time
import uuid
from typing import Dict
from kudian_sdk.crypto_utils import CryptoUtils
from kudian_sdk.sign_utils import SignUtils
from kudian_sdk.http import HttpClient


class KudianClient:
    """酷点支付SDK主客户端"""

    def __init__(self, config: Dict[str, str], http_client: HttpClient):
        """
        初始化客户端

        Args:
            config: 配置字典
            http_client: HTTP客户端实例
        """
        self.config = config
        self.http_client = http_client

    def post(self, api_path: str, biz_params: Dict) -> str:
        """
        发送POST请求

        Args:
            api_path: API路径
            biz_params: 业务参数

        Returns:
            解密后的业务数据JSON字符串

        Raises:
            Exception: 请求异常
        """
        # 1. 构造完整请求URL
        url = self.config['gateway_url'] + api_path

        # 2. 将业务参数转换为JSON并加密
        biz_json = json.dumps(biz_params, ensure_ascii=False)
        encrypted_content = CryptoUtils.aes_encrypt(biz_json, self.config['app_secret_key'])

        # 3. 构造公共参数
        public_params = {
            'mch_id': self.config['mch_id'],
            'app_id': self.config['app_id'],
            'timestamp': str(int(time.time() * 1000)),
            'nonce_str': self._generate_nonce_str(),
            'sign_type': 'SHA',
            'content': encrypted_content,
            'version': '2.0'
        }

        # 4. 生成签名
        sign_string = SignUtils.make_sign_string(public_params)
        private_key_bytes = base64.b64decode(self.config['app_private_key'])
        sign = CryptoUtils.sign_sha256_with_rsa(
            sign_string.encode('utf-8'),
            private_key_bytes
        )
        public_params['sign'] = sign

        # 5. 发送请求
        request_body = json.dumps(public_params, ensure_ascii=False)
        print(f"请求URL: {url}")
        print(f"请求参数: {request_body}")

        headers = {
            'Content-Type': 'application/json'
        }

        response_body = self.http_client.post(url, headers, request_body)
        print(f"响应结果: {response_body}")

        # 6. 解析响应并验签
        response = json.loads(response_body)
        code = response['code']

        if code == 0:
            # 成功响应，验证签名
            response_sign = response.get('sign')
            verify_params = {
                k: str(v) if v is not None else ''
                for k, v in response.items()
                # 排除sign字段
                if k != 'sign'
            }

            verify_string = SignUtils.make_sign_string(verify_params)
            public_key_bytes = base64.b64decode(self.config['api_public_key'])
            verify_result = CryptoUtils.verify_sha256_with_rsa(
                verify_string.encode('utf-8'),
                public_key_bytes,
                response_sign
            )

            if not verify_result:
                raise Exception("响应签名验证失败")

            # 解密业务数据
            encrypted_result = response.get('result')
            decrypted_data = CryptoUtils.aes_decrypt(
                encrypted_result,
                self.config['app_secret_key']
            )
            print(f"解密后的业务数据: {decrypted_data}")

            return decrypted_data
        else:
            # 失败响应
            msg = response.get('msg')
            raise Exception(f"请求失败: code={code}, msg={msg}")

    def verify_notify(self, notify_params: Dict[str, str]) -> bool:
        """
        验证回调通知签名

        Args:
            notify_params: 回调参数

        Returns:
            True-验签成功，False-验签失败
        """
        sign = notify_params.get('sign')
        if not sign:
            return False

        verify_params = notify_params.copy()
        verify_string = SignUtils.make_sign_string(verify_params)

        public_key_bytes = base64.b64decode(self.config['api_public_key'])
        return CryptoUtils.verify_sha256_with_rsa(
            verify_string.encode('utf-8'),
            public_key_bytes,
            sign
        )

    def decrypt_notify(self, encrypted_result: str) -> str:
        """
        解密回调通知的业务数据

        Args:
            encrypted_result: 加密的业务数据

        Returns:
            解密后的业务数据JSON字符串
        """
        return CryptoUtils.aes_decrypt(
            encrypted_result,
            self.config['app_secret_key']
        )

    def _generate_nonce_str(self) -> str:
        """
        生成随机字符串

        Returns:
            随机字符串
        """
        return uuid.uuid4().hex[:32]

可选实现

httpx_client.py - Httpx 客户端实现

import httpx
from typing import Dict
from kudian_sdk.http import HttpClient


class HttpxClient(HttpClient):
    """基于Httpx的HTTP客户端实现（支持异步）"""

    def __init__(self, timeout: int = 30):
        """
        初始化客户端

        Args:
            timeout: 超时时间（秒），默认30秒
        """
        self.timeout = timeout
        self.client = httpx.Client(timeout=timeout)

    def post(self, url: str, headers: Dict[str, str], request_body: str) -> str:
        """
        发送POST请求

        Args:
            url: 请求URL
            headers: 请求头
            request_body: 请求体（JSON字符串）

        Returns:
            响应体（JSON字符串）

        Raises:
            Exception: 请求异常
        """
        response = self.client.post(
            url,
            headers=headers,
            content=request_body.encode('utf-8')
        )

        response.raise_for_status()
        return response.text

    def __del__(self):
        """关闭客户端"""
        if hasattr(self, 'client'):
            self.client.close()

使用示例

完整示例：创建支付订单

import json
from kudian_sdk import KudianClient
from kudian_sdk.http import RequestsClient


def main():
    # 1. 创建配置
    config = {
        'gateway_url': 'https://pay.test.kudianvip.com',
        'mch_id': 'your_mch_id',
        'app_id': 'your_app_id',
        'app_private_key': 'MIIEvQIBADANBgkqhkiG9w0BAQE...',  # Base64格式的私钥
        'api_public_key': 'MIGfMA0GCSqGSIb3DQEBAQUAA...',  # Base64格式的公钥
        'app_secret_key': 'your_secret_key_32_bytes'  # 32字节的AES密钥
    }

    # 2. 创建客户端
    client = KudianClient(config, RequestsClient())

    try:
        # 3. 构造业务参数
        biz_params = {
            'order_no': f"ORDER{int(time.time() * 1000)}",
            'amount': 10000,
            'currency': 'CNY',
            'subject': '测试商品',
            'notify_url': 'https://your-domain.com/notify'
        }

        # 4. 发送请求
        result = client.post('/api/pay/create', biz_params)

        # 5. 解析响应
        data = json.loads(result)

        print("支付订单创建成功！")
        print(f"订单号: {data['order_no']}")
        print(f"支付URL: {data['pay_url']}")

    except Exception as e:
        print(f"订单创建失败: {e}")


if __name__ == '__main__':
    import time
    main()

Flask 回调通知处理示例

from flask import Flask, request, jsonify
from kudian_sdk import KudianClient
from kudian_sdk.http import RequestsClient


app = Flask(__name__)

# 创建客户端
config = {
    'gateway_url': 'https://pay.kudianvip.com',
    'mch_id': 'your_mch_id',
    'app_id': 'your_app_id',
    'app_private_key': 'your_app_private_key',
    'app_secret_key': 'your_secret_key',
    'api_public_key': 'your_api_public_key'
}
client = KudianClient(config, RequestsClient())


@app.route('/notify/pay', methods=['POST'])
def handle_pay_notify():
    """处理支付回调"""
    try:
        # 1. 验证签名
        params = request.json
        if not client.verify_notify(params):
            print("签名验证失败")
            return 'FAIL', 400

        # 2. 解密业务数据
        biz_data = client.decrypt_notify(params['result'])
        print(f"收到支付回调: {biz_data}")

        # 3. 解析业务数据并处理
        # TODO: 根据业务逻辑处理支付结果
        # 更新订单状态、发货等

        # 4. 返回成功
        return 'SUCCESS'

    except Exception as e:
        print(f"处理回调失败: {e}")
        return 'FAIL', 500


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Django 回调通知处理示例

from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from kudian_sdk import KudianClient
from kudian_sdk.http import RequestsClient
import json


# 创建客户端（建议在settings.py中配置）
config = {
    'gateway_url': 'https://pay.kudianvip.com',
    'mch_id': 'your_mch_id',
    'app_id': 'your_app_id',
    'app_private_key': 'your_app_private_key',
    'app_secret_key': 'your_secret_key',
    'api_public_key': 'your_api_public_key'
}
client = KudianClient(config, RequestsClient())


@csrf_exempt
def handle_pay_notify(request):
    """处理支付回调"""
    try:
        # 1. 解析请求参数
        params = json.loads(request.body)

        # 2. 验证签名
        if not client.verify_notify(params):
            print("签名验证失败")
            return HttpResponse('FAIL', status=400)

        # 3. 解密业务数据
        biz_data = client.decrypt_notify(params['result'])
        print(f"收到支付回调: {biz_data}")

        # 4. 解析业务数据并处理
        # TODO: 根据业务逻辑处理支付结果
        # 更新订单状态、发货等

        # 5. 返回成功
        return HttpResponse('SUCCESS')

    except Exception as e:
        print(f"处理回调失败: {e}")
        return HttpResponse('FAIL', status=500)

异常处理

SDK可能抛出的异常：

异常类型	说明	处理建议
ValueError	配置参数错误	检查配置参数是否完整和正确
Exception	签名验签失败	检查密钥是否正确
Exception	加解密失败	检查appSecretKey是否正确
requests.RequestException	HTTP请求失败	检查网络连接和网关地址

常见问题

1. 如何获取密钥？

• app_private_key: 商户自行生成RSA密钥对，将公钥上传到商户后台
• api_public_key: 从商户后台获取平台公钥
• secret_key: 从商户后台获取或设置

2. 如何生成RSA密钥对？

可以使用OpenSSL命令生成：

# 生成私钥
openssl genrsa -out app_private_key.pem 2048

# 提取公钥
openssl rsa -in app_private_key.pem -pubout -out app_public_key.pem

# 转换为Base64格式
base64 -w 0 app_private_key.pem
base64 -w 0 app_public_key.pem

3. 如何使用自定义HTTP客户端？

SDK支持自定义HTTP客户端实现，只需继承 HttpClient 类：

from kudian_sdk.http import HttpClient
from typing import Dict

# 使用内置的RequestsClient
from kudian_sdk.http import RequestsClient
client1 = KudianClient(config, RequestsClient())

# 自定义实现
class CustomHttpClient(HttpClient):
    def post(self, url: str, headers: Dict[str, str], request_body: str) -> str:
        # 自定义HTTP客户端实现
        # 例如使用其他HTTP库或添加特殊逻辑
        pass

client2 = KudianClient(config, CustomHttpClient())

4. 日志配置

SDK使用Python标准logging记录日志，可以这样配置：

import logging

# 配置日志级别
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# SDK会使用根logger记录日志