Python集成Paytrail API：HMAC认证下获取支付状态遇401错误排查

实现代码 #

import hmac
import hashlib
import json
import requests
import uuid
from datetime import datetime, timedelta
from typing import Optional
from .base import BasePaymentProvider


class PaytrailPaymentProvider(BasePaymentProvider):
    """
    Paytrail payment provider implementation (v2 HMAC).
    """
    def __init__(self, organization, credentials):
        super().__init__(organization, credentials)
        self.merchant = credentials.credentials["MERCHANT_ID"]
        self.secret   = credentials.credentials["SECRET_KEY"]
        self.test_mode = credentials.credentials.get("test_mode", True)

        if not self.merchant or not self.secret:
            raise RuntimeError("Missing Paytrail MERCHANT_ID or SECRET_KEY in credentials")

        # Always point at the same host
        self.base_url = "https://services.paytrail.com"

    def _generate_auth_headers(self, method: str, body: str = "") -> dict:
        """
        Build Paytrail v2 HMAC headers:
          1) checkout-account, algorithm, method, nonce, timestamp
          2) signature = HMAC_SHA256(secret, "\n".join(sorted checkout lines + body))
        """
        ts    = datetime.utcnow().isoformat(timespec="milliseconds") + "Z"
        nonce = uuid.uuid4().hex

        hdrs = {
            "checkout-account":   self.merchant,
            "checkout-algorithm": "sha256",
            "checkout-method":    method.upper(),
            "checkout-nonce":     nonce,
            "checkout-timestamp": ts,
        }

        # Build the signing string: each header line sorted alphabetically
        lines = [f"{k}:{v}" for k, v in sorted(hdrs.items())]
        lines.append(body)

        signature = hmac.new(
            self.secret.encode("utf-8"),
            "\n".join(lines).encode("utf-8"),
            hashlib.sha256
        ).hexdigest()

        hdrs["checkout-signature"] = signature
        hdrs["Content-Type"]       = "application/json"
        return hdrs

    def initiate_payment(
        self,
        amount: int,
        customer_info: dict,
        order_id: Optional[str]   = None,
        return_url: Optional[str] = None,
        **kwargs
    ) -> dict:
        """
        Create a Paytrail payment via POST /api/v2/payments.
        """
        # 1) Build the payment payload
        stamp = str(uuid.uuid4())
        payload = {
            "stamp":     stamp,
            "reference": stamp,
            "amount":    amount,
            "currency":  customer_info.get("currency", "EUR"),
            "language":  customer_info.get("language", "FI"),
            "items": [{
                "unitPrice":     amount,
                "units":         1,
                "vatPercentage": 24,
                "productCode":   "default",
                "description":   "Payment",
                "deliveryDate":  datetime.utcnow().date().isoformat()
            }],
            "customer": {
                "email": customer_info.get("email", "")
            },
            "redirectUrls": {
                "success": return_url or f"{self.base_url}/success",
                "cancel":  return_url or f"{self.base_url}/cancel"
            }
        }

        # 2) Compact JSON for signing
        body = json.dumps(payload, separators=(",", ":"), sort_keys=True)

        # 3) Generate HMAC headers
        headers = self._generate_auth_headers("POST", body)

        # 4) Debug log
        print("=== PAYTRAIL DEBUG ===")
        print("URL:   ", self.base_url + "/payments")
        print("HEADERS:")
        for k, v in headers.items():
            print(f"  {k}: {v}")
        print("BODY:", body)
        print("=======================")

       # 5) Fire the request and handle errors
        try:
            resp = requests.post(
                f"{self.base_url}/payments",
                headers=headers,
                data=body,
                timeout=10
            )
            # If status is 4xx/5xx, print raw body then raise
            if not resp.ok:
                print(">>> PAYTRAIL RESPONSE BODY:", resp.status_code, resp.text)
                resp.raise_for_status()

            # 6) Parse JSON and return
            data = resp.json()
            return {
                "id":          data["transactionId"],
                "reference":   data["transactionId"],
                "payment_url": data.get("href") or data.get("url"),
                "status":      "INITIATED",
                "amount":      amount,
                "currency":    payload["currency"],
            }

        except requests.exceptions.RequestException as e:
            raise ValueError(f"Payment initiation failed: {str(e)}")


    def get_payment_status(self, reference: str) -> dict:
        """
        Get Paytrail payment status.
        """
        try:
            # Generate headers
            headers = self._generate_auth_headers("GET", "")
            
            # Make API request
            response = requests.get(
                f"{self.base_url}/payments/{reference}",
                headers=headers
            )
            response.raise_for_status()
            
            result = response.json()
            
            # Map Paytrail status to our status
            status_mapping = {
                    "new": "INITIATED",
                    "ok": "PAID",
                    "fail": "FAILED",
                    "pending": "PENDING",
                    "delayed": "PENDING",
        
            }
            
            return {
                "id": reference,
                "status": status_mapping.get(result.get("status", "").lower(), "UNKNOWN"),
                "amount": result.get("amount"),
                "captured": result.get("status", "").lower() == "ok"
            }

        except requests.exceptions.RequestException as e:
            raise ValueError(f"Failed to get payment status: {str(e)}")

问题 #

实现Paytrail支付提供商时，调用API获取支付状态遇到401 Unauthorized错误，需解决以下问题：

1. 此场景下导致401错误的常见原因有哪些？
2. 如何排查HMAC请求头与请求签名？
3. Paytrail API的GET请求是否有特定的额外请求头或步骤？

回答 #

1. 导致401错误的常见原因 #

• 凭据错误：MERCHANT_ID或SECRET_KEY填写错误，或者测试环境凭据调用生产环境（反之亦然）
• HMAC签名生成错误：
  • 请求方法大小写不一致（比如用get而非GET）
  • 时间戳格式不符合要求（必须是UTC毫秒级ISO格式，如2024-05-20T12:34:56.789Z）
  • 签名用的请求体不正确（GET请求需确保body为空字符串，POST需用压缩排序后的JSON）
  • 签名字符串拼接时未按字母顺序排序header键值对
• 请求头缺失或错误：
  • 缺少checkout-account、checkout-algorithm等必填HMAC头
  • Content-Type设置错误（必须为application/json）
• 请求URL错误：GET请求路径中的reference参数不正确，或调用了错误的API版本路径
• 时间同步问题：本地服务器时间与UTC时间偏差过大（超过Paytrail允许的5分钟窗口）

2. 排查HMAC请求头与签名的步骤 #

• 打印完整请求信息：在get_payment_status方法中添加debug日志，输出签名字符串构建过程、所有请求头及生成的签名，示例代码：

  def get_payment_status(self, reference: str) -> dict:
      try:
          # 生成调试用的签名字符串
          ts = datetime.utcnow().isoformat(timespec="milliseconds") + "Z"
          nonce = uuid.uuid4().hex
          debug_hdrs = {
              "checkout-account": self.merchant,
              "checkout-algorithm": "sha256",
              "checkout-method": "GET",
              "checkout-nonce": nonce,
              "checkout-timestamp": ts,
          }
          lines = [f"{k}:{v}" for k, v in sorted(debug_hdrs.items())]
          lines.append("")
          print("=== SIGNING STRING ===")
          print("\n".join(lines))
          
          headers = self._generate_auth_headers("GET", "")
          print("=== GENERATED SIGNATURE ===")
          print(headers["checkout-signature"])
          # 后续请求代码...
  

• 手动验证签名：用生成的签名字符串、SECRET_KEY，通过本地HMAC计算工具或代码，对比代码生成的签名是否一致
• 核对头信息：检查checkout-timestamp是否为UTC时间、格式是否包含毫秒；checkout-method是否为大写GET；所有HMAC相关头是否无拼写错误
• 对比成功请求：如果initiate_payment方法能成功，对比两个方法的签名逻辑和请求头，找出差异点

3. Paytrail API GET请求的特定要求 #

• 请求体必须为空：GET请求不能携带任何body，签名时对应的body参数必须是空字符串
• 路径参数需正确编码：如果reference包含特殊字符，需确保URL编码正确（requests库会自动处理，手动拼接时需注意）
• HMAC头要求与POST一致：必须包含checkout-account、checkout-algorithm、checkout-method、checkout-nonce、checkout-timestamp、checkout-signature头，且checkout-method必须是大写GET
• 无需额外头：除HMAC相关头和Content-Type: application/json外，不需要其他特殊请求头

内容的提问来源于stack exchange，提问作者rachna soni