支付系统是电商和源码交易平台的核心命脉，一旦出现安全漏洞，不仅导致资金损失，更可能引发用户信任崩塌。本文基于实战经验，系统梳理 Python 支付系统的安全实践方案，涵盖数据传输加密、签名验签、SQL 注入防护、CSRF/XSS 防御、敏感信息脱敏以及分布式风控机制，每个模块都附有可直接落地的代码示例。

无论你是在开发转卡码平台、源码商城，还是自己的支付中转服务，这些安全实践都能帮助你构建一个可靠、合规的支付系统。

一、数据传输加密：双向 TLS 与敏感字段加密

支付系统涉及用户隐私和交易数据，传输过程中必须加密。最基础的要求是开启 HTTPS（TLS 1.2+），但对于关键敏感字段（如银行卡号、身份证号），建议在应用层做二次加密。

使用 cryptography 实现 AES-256-GCM 加密

# pip install cryptography
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os, base64

def encrypt_sensitive_data(plaintext: str, key: bytes) -> str:
    aesgcm = AESGCM(key)
    nonce = os.urandom(12)  # GCM 模式推荐 12 字节 nonce
    ct = aesgcm.encrypt(nonce, plaintext.encode(), None)
    return base64.b64encode(nonce + ct).decode()

def decrypt_sensitive_data(encrypted: str, key: bytes) -> str:
    raw = base64.b64decode(encrypted)
    nonce, ct = raw[:12], raw[12:]
    aesgcm = AESGCM(key)
    return aesgcm.decrypt(nonce, ct, None).decode()

# 使用示例
KEY = os.urandom(32)  # 生产环境从 KMS 或密钥管理服务获取
enc = encrypt_sensitive_data("622848****1234", KEY)
print(f"加密后: {enc}")
dec = decrypt_sensitive_data(enc, KEY)
print(f"解密后: {dec}")

AES-256-GCM 模式同时提供加密和完整性校验，能有效防止篡改。生产环境中密钥不应硬编码在代码中，建议通过 HashiCorp Vault、AWS KMS 或阿里云 KMS 管理。

二、签名验签机制：防止请求伪造

支付系统中，无论是前端到后端，还是服务端到第三方支付网关，每个请求都需要签名验证。下面是一个标准的 HMAC-SHA256 签名实现：

import hashlib, hmac, json, time

def generate_sign(payload: dict, secret: str) -> str:
    # 按字典序排序参数，保证签名一致性
    sorted_keys = sorted(payload.keys())
    sign_str = "&".join(
        f"{k}={payload[k]}" for k in sorted_keys
    )
    sign_str += f"&key={secret}"
    return hashlib.md5(sign_str.encode()).hexdigest().upper()

def verify_sign(payload: dict, sign: str, secret: str) -> bool:
    expected = generate_sign(payload, secret)
    return hmac.compare_digest(expected, sign)

# 带时间戳防止重放攻击
def build_signed_request(data: dict, secret: str) -> dict:
    data['timestamp'] = int(time.time())
    data['nonce_str'] = os.urandom(8).hex()
    data['sign'] = generate_sign(data, secret)
    return data

关键安全要点：

  • 时间戳防重放：服务端校验时间戳与当前时间差不超过 5 分钟
  • 随机 nonce：同一 nonce 不可重复使用（Redis 记录已用 nonce）
  • 使用 hmac.compare_digest：防止时序攻击（timing attack）
  • 参数排序：按 ASCII 字典序保证签名一致性

三、SQL 注入防护：ORM 与参数化查询

支付订单表是攻击者的首要目标。直接拼接 SQL 字符串是绝对禁止的。以下是正确的做法：

# ❌ 错误：字符串拼接（高危）
cursor.execute(f"SELECT * FROM orders WHERE order_id='{order_id}'")

# ✅ 正确：参数化查询
cursor.execute(
    "SELECT * FROM orders WHERE order_id = %s",
    (order_id,)
)

# 🏆 最佳：使用 SQLAlchemy ORM
from sqlalchemy.orm import Session
from models import Order

def get_order(db: Session, order_id: str) -> Order | None:
    return db.query(Order).filter(
        Order.order_id == order_id
    ).first()

使用 ORM 不仅能防注入，还天然支持事务管理——这对支付系统至关重要：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine(DATABASE_URL, pool_size=10, max_overflow=20)
SessionLocal = sessionmaker(bind=engine)

def create_order(db: Session, user_id: int, amount: int) -> Order:
    try:
        order = Order(
            order_id=generate_order_id(),
            user_id=user_id,
            amount=amount,
            status='pending'
        )
        db.add(order)
        db.commit()
        db.refresh(order)
        return order
    except Exception as e:
        db.rollback()
        raise

对于源码商城的转卡码系统，订单和卡密数据高度敏感，务必将数据库操作全部置于事务中，并设置合理的连接池参数防止连接耗尽。

四、CSRF 与 XSS 防护

CSRF 防护

付款操作、退款操作等必须防止跨站请求伪造。推荐使用 Double Submit Cookie 模式：

import secrets
from flask import request, session, abort

def generate_csrf_token() -> str:
    token = secrets.token_hex(32)
    session['csrf_token'] = token
    return token

def verify_csrf_token() -> None:
    token = request.headers.get('X-CSRF-Token') \
             or request.form.get('csrf_token')
    if not token or token != session.get('csrf_token'):
        abort(403, "CSRF token 验证失败")

XSS 防护

支付页面的任何用户输入输出都必须转义。使用 Jinja2 模板引擎时，默认开启自动转义，但注意 |safe 过滤器的使用：

# Flask + Jinja2 模板中
{{ user_input }}           # ✅ 自动转义 HTML 特殊字符
{{ user_input|safe }}      # ⚠️ 仅在确信内容安全时使用

# Python 中手动转义
import html
safe_text = html.escape(user_input, quote=True)

五、敏感信息脱敏与日志安全

支付系统的日志可能包含大量敏感信息，如手机号、银行卡号、身份证号。必须对日志中的敏感字段做脱敏处理：

import re

def mask_sensitive(text: str) -> str:
    # 脱敏手机号：138****5678
    text = re.sub(r'(1[3-9]\d)\d{4}(\d{4})', r'\1****\2', text)
    # 脱敏银行卡号：622848******1234
    text = re.sub(r'(\d{6})\d{6,10}(\d{4})', r'\1******\2', text)
    # 脱敏身份证号
    text = re.sub(r'(\d{6})\d{8,10}(\d{4})', r'\1********\2', text)
    # 脱敏支付密码关键字
    text = re.sub(r'(password|pwd|secret|sign_key)=[^&\s]+',
                  r'\1=******', text, flags=re.I)
    return text

# 使用 logging Filter 自动脱敏
import logging

class SensitiveFilter(logging.Filter):
    def filter(self, record):
        if hasattr(record, 'msg') and isinstance(record.msg, str):
            record.msg = mask_sensitive(record.msg)
        return True

logger = logging.getLogger('payment')
logger.addFilter(SensitiveFilter())
logger.info("用户 13812345678 支付成功，卡号 6228481234567890")
# 日志输出：用户 138****5678 支付成功，卡号 622848******7890

六、分布式风控机制

使用 Redis 实现分布式频率限制和风控检测，能有效拦截恶意调用：

import redis.asyncio as aioredis
from datetime import datetime

r = aioredis.Redis.from_url("redis://localhost:6379/0")

async def check_risk(user_id: str, ip: str) -> dict:
    now = int(datetime.utcnow().timestamp())
    window = 60  # 60 秒窗口
    risk_score = 0
    reasons = []

    # 1. 检查 IP 请求频率（每分钟限制 30 次）
    ip_key = f"ratelimit:ip:{ip}:{now // window}"
    ip_count = await r.incr(ip_key)
    await r.expire(ip_key, window + 5)
    if ip_count > 30:
        risk_score += 30
        reasons.append("IP 请求频率过高")

    # 2. 检查用户操作频率
    user_key = f"ratelimit:user:{user_id}:{now // window}"
    user_count = await r.incr(user_key)
    await r.expire(user_key, window + 5)
    if user_count > 10:
        risk_score += 20
        reasons.append("用户操作过于频繁")

    # 3. 检查同一 IP 关联多个用户
    ip_user_key = f"risk:ip_users:{ip}"
    await r.sadd(ip_user_key, user_id)
    await r.expire(ip_user_key, 3600)
    user_count_for_ip = await r.scard(ip_user_key)
    if user_count_for_ip > 5:
        risk_score += 25
        reasons.append("IP 关联用户数异常")

    return {"risk_score": risk_score, "blocked": risk_score > 50, "reasons": reasons}

# 风控中间件
async def risk_middleware(request, call_next):
    user_id = request.headers.get("X-User-Id", "anonymous")
    ip = request.client.host
    result = await check_risk(user_id, ip)
    if result["blocked"]:
        return JSONResponse(
            status_code=429,
            content={"error": "触发风控", "reasons": result["reasons"]}
        )
    return await call_next(request)

七、安全配置清单

最后，整理一份支付系统的安全配置自检清单，建议每次部署前逐项核对：

项目检查内容优先级
HTTPSTLS 1.2+，禁用 SSLv3/TLSv1.0P0
签名验签所有接口强制签名校验P0
防重放时间戳 + nonce 双重校验P0
参数化查询禁止 SQL 字符串拼接P0
CSRF 令牌写操作接口验证 CSRF TokenP1
敏感数据加密数据库存储加密（AES-256-GCM）P1
日志脱敏脱敏后才能写入日志文件P1
频率限制IP + 用户双层限流P1
事务管理订单操作全部事务包裹P1
密钥轮换签名密钥每 90 天轮换P2
错误信息不向用户暴露内部栈信息P2
审计日志所有管理操作留下审计轨迹P2

支付安全是一个持续迭代的过程，没有一劳永逸的方案。建议定期进行安全审计和渗透测试，关注 OWASP Top 10 的最新动态。

推荐工具：我们的 转卡码系统 V3 内置了上述所有安全机制——自动 AES-256 数据加密、HMAC-SHA256 签名验证、Redis 分布式限流以及完整的防注入 ORM 封装，开箱即用。配合 Codex Desktop 进行代码审查，能进一步提升支付系统的安全性。