Build a Flask Mock Payment Callback Service with Signature Verification

Objectives

Implement a local mock payment callback service based on Flask that can receive and verify signed callback requests, return standard response formats (e.g., for WeChat/Alipay), and support various scenarios such as success, failure, and timeout.

Required Dependencies

Make sure the following libraries are installed:

pip install flask pytest respx requests

Recommended Directory Structure

payment_automation/
├── mock_server/
│   ├── app.py               # Callback service entry point
│   └── config.py            # Callback configuration (e.g., signature key)
├── testcases/
│   └── test_payment_callback.py  # Test cases
├── utils/
│   └── sign_utils.py       # Signature utility class
├── config/
│   └── payment_config.py   # Payment configuration
└── requirements.txt

Core Code Implementation

1️⃣ Callback Configuration (mock_server/config.py)

# mock_server/config.py
CALLBACK_CONFIG = {
    "host": "127.0.0.1",
    "port": 5000,
    "public_key": "third_party_public_key",  # Third‑party public key for verification
    "success_response": {"return_code": "SUCCESS", "return_msg": "OK"},
    "fail_response": {"return_code": "FAIL", "return_msg": "SIGNATURE_FAILED"}
}

2️⃣ Signature Utility (utils/sign_utils.py)

# utils/sign_utils.py
import hashlib
import hmac
from urllib.parse import quote_plus

def generate_sign(params: dict, secret_key: str) -> str:
    """Generate a signature by sorting keys and concatenating them."""
    sorted_params = "&".join(
        f"{k}={quote_plus(str(v))}" for k, v in sorted(params.items())
    )
    sign = hmac.new(secret_key.encode(), digestmod=hashlib.sha256)
    sign.update(sorted_params.encode())
    return sign.hexdigest()

def verify_sign(received_sign: str, params: dict, secret_key: str) -> bool:
    """Verify that the received signature matches the expected one."""
    expected_sign = generate_sign(params, secret_key)
    return hmac.compare_digest(expected_sign, received_sign)

3️⃣ Mock Callback Service (mock_server/app.py)

# mock_server/app.py
from flask import Flask, request, jsonify
from utils.sign_utils import verify_sign
from mock_server.config import CALLBACK_CONFIG

app = Flask(__name__)
PUBLIC_KEY = CALLBACK_CONFIG["public_key"]
SUCCESS_RESPONSE = CALLBACK_CONFIG["success_response"]
FAIL_RESPONSE = CALLBACK_CONFIG["fail_response"]

@app.route("/wechat/notify", methods=["POST"])
def wechat_notify():
    data = request.json
    sign = data.pop("sign", None)
    if not sign or not verify_sign(sign, data, PUBLIC_KEY):
        return jsonify(FAIL_RESPONSE), 200
    print("Received WeChat payment callback:", data)
    return jsonify(SUCCESS_RESPONSE), 200

@app.route("/alipay/notify", methods=["POST"])
def alipay_notify():
    data = request.form.to_dict()
    sign = data.pop("sign", None)
    if not sign or not verify_sign(sign, data, PUBLIC_KEY):
        return "failure"
    print("Received Alipay payment callback:", data)
    return "success"

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

Startup Commands

cd mock_server
python app.py

Advanced Feature Suggestions

Simulate multiple payment results such as success, failure, and timeout.

Support dynamic return values controlled via URL parameters.

Log each callback request to a database.

Implement asynchronous retry mechanisms to mimic repeated notifications.

Integrate pytest fixtures for automatic start/stop of the mock server.

Testing Example (testcases/test_payment_callback.py)

# testcases/test_payment_callback.py
import requests, json, pytest
from utils.sign_utils import generate_sign
from mock_server.config import CALLBACK_CONFIG

BASE_URL = "http://localhost:5000"
SECRET_KEY = CALLBACK_CONFIG["public_key"]

def send_wechat_notify(order_id: str, status: str = "SUCCESS"):
    data = {
        "transaction_id": "T1234567890",
        "out_trade_no": order_id,
        "return_code": status,
        "total_fee": 9990,  # amount in cents
        "time_end": "20250405120000"
    }
    data["sign"] = generate_sign(data, SECRET_KEY)
    response = requests.post(f"{BASE_URL}/wechat/notify", json=data)
    return response

def test_wechat_callback_success():
    order_id = "ORDER123456"
    response = send_wechat_notify(order_id)
    assert response.status_code == 200
    assert response.json()["return_code"] == "SUCCESS"

def test_wechat_callback_invalid_sign():
    data = {
        "transaction_id": "T1234567890",
        "out_trade_no": "ORDER789012",
        "return_code": "SUCCESS",
        "total_fee": 9990,
        "time_end": "20250405120000",
        "sign": "invalid-sign-here"
    }
    response = requests.post(f"{BASE_URL}/wechat/notify", json=data)
    assert response.status_code == 200
    assert response.json()["return_code"] == "FAIL"

Expose the Service Publicly

Use ngrok or localtunnel to expose the local server: ngrok http 5000 Set the generated URL (e.g., https://xxx.ngrok.io/wechat/notify) as the callback address in your payment platform.

Summary

Secure signature verification using HMAC‑SHA256.

Supports multiple payment platforms (WeChat, Alipay).

Easy to extend with additional channels and edge cases.

Integrates smoothly with pytest for automated assertions.

Public access possible via ngrok.