新增ERP扫码登录小程序后端代理

app/services/erp_login_security.py

@@ -0,0 +1,99 @@
+import hashlib
+import hmac
+import time
+from uuid import uuid4
+
+from fastapi import HTTPException
+
+
+SERVICE_TIMESTAMP_HEADER = "x-service-timestamp"
+SERVICE_NONCE_HEADER = "x-service-nonce"
+SERVICE_SIGNATURE_HEADER = "x-service-signature"
+
+_seen_nonces: dict[str, int] = {}
+
+
+def build_service_signature(
+    method: str,
+    path: str,
+    timestamp: str,
+    nonce: str,
+    body: bytes,
+    secret: str,
+) -> str:
+    message = "\n".join([
+        method.upper(),
+        path,
+        timestamp,
+        nonce,
+        body.decode("utf-8"),
+    ])
+    return hmac.new(secret.encode("utf-8"), message.encode("utf-8"), hashlib.sha256).hexdigest()
+
+
+def signed_headers(method: str, path: str, body: bytes, secret: str) -> dict[str, str]:
+    timestamp = str(int(time.time()))
+    nonce = uuid4().hex
+    signature = build_service_signature(method, path, timestamp, nonce, body, secret)
+    return {
+        SERVICE_TIMESTAMP_HEADER: timestamp,
+        SERVICE_NONCE_HEADER: nonce,
+        SERVICE_SIGNATURE_HEADER: signature,
+    }
+
+
+def _cleanup_seen_nonces(now_ts: int, ttl_seconds: int) -> None:
+    expired = [
+        nonce
+        for nonce, seen_at in _seen_nonces.items()
+        if now_ts - seen_at > ttl_seconds
+    ]
+    for nonce in expired:
+        _seen_nonces.pop(nonce, None)
+
+
+def verify_service_signature(
+    method: str,
+    path: str,
+    timestamp: str | None,
+    nonce: str | None,
+    body: bytes,
+    signature: str | None,
+    secret: str,
+    ttl_seconds: int = 300,
+) -> None:
+    if not secret:
+        raise HTTPException(status_code=500, detail="ERP扫码登录服务密钥未配置")
+
+    try:
+        timestamp_value = int(str(timestamp))
+    except (TypeError, ValueError) as exc:
+        raise HTTPException(status_code=401, detail="ERP服务签名无效") from exc
+
+    now_ts = int(time.time())
+    if abs(now_ts - timestamp_value) > ttl_seconds:
+        raise HTTPException(status_code=401, detail="ERP服务签名已过期")
+
+    _cleanup_seen_nonces(now_ts, ttl_seconds)
+    nonce_value = str(nonce or "")
+    if not nonce_value or nonce_value in _seen_nonces:
+        raise HTTPException(status_code=401, detail="ERP服务签名无效")
+
+    try:
+        expected = build_service_signature(
+            method,
+            path,
+            str(timestamp),
+            nonce_value,
+            body,
+            secret,
+        )
+    except UnicodeDecodeError as exc:
+        raise HTTPException(status_code=401, detail="ERP服务签名无效") from exc
+
+    if not hmac.compare_digest(expected, str(signature or "")):
+        raise HTTPException(status_code=401, detail="ERP服务签名无效")
+
+    # This in-process nonce store only protects one running process. Production
+    # multi-worker deployments should move replay tracking to shared storage.
+    _seen_nonces[nonce_value] = now_ts

app/services/wechat.py

@@ -190,6 +190,24 @@ async def create_device_qrcode(
     scene: str,
     public_base_url: str,
     label: str | None = None,
+) -> str | None:
+    return await create_miniapp_qrcode(
+        key=device_no,
+        page=page,
+        scene=scene,
+        public_base_url=public_base_url,
+        directory="qrcodes",
+        label=label,
+    )
+
+
+async def create_miniapp_qrcode(
+    key: str,
+    page: str,
+    scene: str,
+    public_base_url: str,
+    directory: str,
+    label: str | None = None,
 ) -> str | None:
     token = await get_access_token()
     url = f"https://api.weixin.qq.com/wxa/getwxacodeunlimit?access_token={token}"
@@ -205,12 +223,15 @@ async def create_device_qrcode(
     content_type = response.headers.get("content-type", "")
     if "application/json" in content_type:
         data = response.json()
-        raise RuntimeError(data.get("errmsg") or "生成设备小程序码失败")
+        raise RuntimeError(data.get("errmsg") or "生成小程序码失败")
 
-    qr_dir = settings.upload_path / "qrcodes"
+    qr_dir = settings.upload_path / directory
     qr_dir.mkdir(parents=True, exist_ok=True)
     suffix = _qrcode_file_suffix(content_type)
-    file_name = _qrcode_file_name(device_no, suffix)
+    file_name = _qrcode_file_name(key, suffix)
     file_path = Path(qr_dir) / file_name
-    file_path.write_bytes(_draw_qrcode_label(response.content, suffix, label or device_no))
-    return f"{public_base_url.rstrip('/')}/uploads/qrcodes/{quote(file_name, safe='')}"
+    label_text = label if label is not None else key
+    file_path.write_bytes(_draw_qrcode_label(response.content, suffix, label_text))
+    encoded_directory = quote(directory.strip("/"), safe="")
+    encoded_file_name = quote(file_name, safe="")
+    return f"{public_base_url.rstrip('/')}/uploads/{encoded_directory}/{encoded_file_name}"