diff --git a/backend/.env.example b/backend/.env.example index fa424f2..0603085 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -54,3 +54,4 @@ ALIPAY_TRANSFER_API=alipay.fund.trans.uni.transfer ALIPAY_TRANSFER_QUERY_API=alipay.fund.trans.common.query ALIPAY_TRANSFER_SCENE_NAME=佣金报酬 ALIPAY_HTTP_TIMEOUT_SECONDS=30 +ALIPAY_RECONCILE_INTERVAL_SECONDS=30 diff --git a/backend/app/__pycache__/alipay_provider.cpython-313.pyc b/backend/app/__pycache__/alipay_provider.cpython-313.pyc index e16d0d2..9856ee0 100644 Binary files a/backend/app/__pycache__/alipay_provider.cpython-313.pyc and b/backend/app/__pycache__/alipay_provider.cpython-313.pyc differ diff --git a/backend/app/__pycache__/config.cpython-313.pyc b/backend/app/__pycache__/config.cpython-313.pyc index 5f43841..4b2033e 100644 Binary files a/backend/app/__pycache__/config.cpython-313.pyc and b/backend/app/__pycache__/config.cpython-313.pyc differ diff --git a/backend/app/__pycache__/main.cpython-313.pyc b/backend/app/__pycache__/main.cpython-313.pyc index b55e30e..d2a315b 100644 Binary files a/backend/app/__pycache__/main.cpython-313.pyc and b/backend/app/__pycache__/main.cpython-313.pyc differ diff --git a/backend/app/alipay_provider.py b/backend/app/alipay_provider.py index fddede3..b3fa2d8 100644 --- a/backend/app/alipay_provider.py +++ b/backend/app/alipay_provider.py @@ -151,6 +151,58 @@ def _same_rsa_key(private_key: str, public_key: str) -> bool: return (private_public.n, private_public.e) == (public.n, public.e) +def _signed_response_content(response_value, public_key: str, charset: str = "utf-8") -> str: + """Verify a certificate-mode response and return its API response object. + + Old alipay-sdk-python releases include the top-level ``alipay_cert_sn`` + field in the bytes they verify. Alipay signs only the inner ``*_response`` + JSON object, so otherwise a valid code=10000 response is rejected. + """ + from alipay.aop.api.util.SignatureUtils import verify_with_rsa + + response_text = response_value.decode(charset) if isinstance(response_value, bytes) else str(response_value) + marker = re.search(r'"[^"\\]+_response"\s*:\s*', response_text) + if not marker: + raise AlipayTransferError("支付宝返回内容缺少业务响应对象", ambiguous=True) + start = response_text.find("{", marker.end()) + if start < 0: + raise AlipayTransferError("支付宝返回业务响应格式无效", ambiguous=True) + depth = 0 + quoted = False + escaped = False + end = -1 + for index in range(start, len(response_text)): + char = response_text[index] + if quoted: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == '"': + quoted = False + continue + if char == '"': + quoted = True + elif char == "{": + depth += 1 + elif char == "}": + depth -= 1 + if depth == 0: + end = index + 1 + break + if end < 0: + raise AlipayTransferError("支付宝返回业务响应不完整", ambiguous=True) + try: + envelope = json.loads(response_text) + except (TypeError, ValueError) as exc: + raise AlipayTransferError("支付宝返回内容无法解析", ambiguous=True) from exc + signature = str(envelope.get("sign") or "") + response_content = response_text[start:end] + if not signature or not verify_with_rsa(public_key, response_content.encode(charset), signature): + raise AlipayTransferError("支付宝响应验签失败", "ALIPAY_RESPONSE_UNVERIFIED", ambiguous=True) + return response_content + + def _read_key(path_value: str, label: str) -> str: if not path_value: raise AlipayTransferError(f"未配置{label}文件路径") @@ -206,6 +258,16 @@ def _client(): common["alipay_root_cert_sn"] = alipay_root_cert_sn return common + def _DefaultAlipayClient__parse_response(self, response_value): + response_text = response_value.decode(settings.alipay_charset) if isinstance(response_value, bytes) else str(response_value) + if '"alipay_cert_sn"' in response_text: + return _signed_response_content( + response_text, + config.alipay_public_key, + settings.alipay_charset, + ) + return super()._DefaultAlipayClient__parse_response(response_value) + client_class = CertificateAlipayClient else: config.alipay_public_key = _sdk_key(_read_key(settings.alipay_public_key_path, "支付宝公钥"), private=False) diff --git a/backend/app/config.py b/backend/app/config.py index 3c037e6..f9f6240 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -79,6 +79,7 @@ class Settings: alipay_transfer_query_api = os.getenv("ALIPAY_TRANSFER_QUERY_API", "alipay.fund.trans.common.query") alipay_transfer_scene_name = os.getenv("ALIPAY_TRANSFER_SCENE_NAME", "").strip() alipay_http_timeout_seconds = int(os.getenv("ALIPAY_HTTP_TIMEOUT_SECONDS", "30")) + alipay_reconcile_interval_seconds = max(10, int(os.getenv("ALIPAY_RECONCILE_INTERVAL_SECONDS", "30"))) settings = Settings() diff --git a/backend/app/main.py b/backend/app/main.py index 412fc76..ab6caa4 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -12,6 +12,7 @@ import logging import random import re import secrets +import threading import time import urllib.parse import urllib.request @@ -481,6 +482,8 @@ def startup(): connection.execute(text("CREATE UNIQUE INDEX ix_admin_users_answer_public_token ON admin_users (answer_public_token)")) finally: db.close() + if settings.alipay_enabled: + _start_alipay_reconcile_worker() def _migrate_alipay_transfer_attempts(bind): @@ -3578,6 +3581,139 @@ def _snapshot_refund(db, record, kind, user, payer_account="", alipay_trade_no=" )) +def _complete_alipay_refund_order(db: Session, order: AlipayRefundOrder, trade_no: str) -> bool: + """Atomically persist a verified provider success and advance the task.""" + trade_no = str(trade_no or "").strip() + if not trade_no: + return False + if order.target_kind == "ask_batch": + target = db.get(AskAssignmentBatch, order.target_id) + elif order.target_kind == "ask_assignment": + target = db.get(AskAssignment, order.target_id) + elif order.target_kind == "answer_submission": + target = db.get(AnswerSubmission, order.target_id) + else: + return False + if not target: + return False + existing = db.scalar(select(RefundRecord.id).where( + RefundRecord.target_kind == order.target_kind, + RefundRecord.target_id == order.target_id, + )) + if not existing: + db.add(RefundRecord( + target_kind=order.target_kind, + target_id=order.target_id, + owner_id=order.owner_id, + shop_name=order.shop_name or "", + order_no=order.order_no or "", + payee_name=order.payee_name or "", + payee_account=order.account or "", + amount_fen=order.amount_fen, + operator_name=order.operator_name or "系统自动对账", + payer_account=order.payer_account or "", + alipay_trade_no=trade_no, + )) + if order.target_kind == "ask_batch": + target.status = "paid" + target.updated_at = datetime.utcnow() + _sync_ask_task_status(target.assignment.task) + elif order.target_kind == "ask_assignment": + target.status = "已完成" + target.updated_at = datetime.utcnow() + _sync_ask_task_status(target.task) + elif order.target_kind == "answer_submission": + target.status = "已完成" + order.status = "success" + order.alipay_trade_no = trade_no + order.error_code = "" + order.error_message = "" + return True + + +def _reconcile_pending_alipay_orders_once() -> int: + """Query pending/legacy uncertain transfers and finalize verified results.""" + if not settings.alipay_enabled: + return 0 + from .alipay_provider import query + + with SessionLocal() as lookup_db: + order_ids = lookup_db.scalars( + select(AlipayRefundOrder.id).where(or_( + AlipayRefundOrder.status == "pending", + (AlipayRefundOrder.status == "failed") & (AlipayRefundOrder.error_code == "10000"), + )).order_by(AlipayRefundOrder.id).limit(100) + ).all() + completed = 0 + for order_id in order_ids: + with SessionLocal() as db: + order = db.scalar( + select(AlipayRefundOrder) + .where(AlipayRefundOrder.id == order_id) + .with_for_update() + ) + if not order or (order.status != "pending" and order.error_code != "10000"): + continue + try: + result = query(out_biz_no=order.out_biz_no) + except AlipayTransferError as exc: + order.status = "pending" + order.error_code = exc.code[:128] + order.error_message = str(exc)[:500] + db.commit() + logging.getLogger(__name__).warning( + "Alipay automatic reconciliation deferred: out_biz_no=%s code=%s message=%s", + order.out_biz_no, exc.code, exc, + ) + continue + provider_status = str(result.get("status") or "").upper() + if provider_status == "SUCCESS": + trade_no = str(result.get("trade_no") or "").strip() + if trade_no and _complete_alipay_refund_order(db, order, trade_no): + db.commit() + completed += 1 + logging.getLogger(__name__).info( + "Alipay automatic reconciliation completed: out_biz_no=%s trade_no=%s", + order.out_biz_no, trade_no, + ) + else: + order.error_message = "支付宝确认成功，但暂未返回资金流水号" + db.commit() + elif provider_status in {"FAIL", "REFUND"}: + order.status = "failed" + order.error_code = "" + order.error_message = str(result.get("fail_reason") or "支付宝确认转账失败")[:500] + db.commit() + else: + order.status = "pending" + order.error_message = f"支付宝转账状态：{provider_status or '处理中'}" + db.commit() + return completed + + +def _alipay_reconcile_loop() -> None: + interval = settings.alipay_reconcile_interval_seconds + # Give application startup and database migrations time to finish. + time.sleep(min(5, interval)) + while True: + try: + _reconcile_pending_alipay_orders_once() + except Exception: + logging.getLogger(__name__).exception("Alipay automatic reconciliation failed") + time.sleep(interval) + + +def _start_alipay_reconcile_worker() -> None: + if getattr(app.state, "alipay_reconcile_started", False): + return + app.state.alipay_reconcile_started = True + threading.Thread( + target=_alipay_reconcile_loop, + name="alipay-reconcile", + daemon=True, + ).start() + + def _refund_record_query(user, q, start_date, end_date): query = select(RefundRecord) if user.role_code != "super_admin": @@ -3821,10 +3957,8 @@ async def alipay_callback(request: Request, db: Session = Depends(get_db)): trade_no = str(params.get("pay_fund_order_id") or params.get("order_id") or "").strip() if not trade_no: return Response(content="failure", media_type="text/plain", status_code=400) - order.status = "success" - order.alipay_trade_no = trade_no - order.error_code = "" - order.error_message = "" + if not _complete_alipay_refund_order(db, order, trade_no): + return Response(content="failure", media_type="text/plain", status_code=400) elif provider_status in {"FAIL", "REFUND"}: order.status = "failed" order.error_code = str(params.get("error_code") or params.get("sub_order_error_code") or "")[:128] diff --git a/backend/tests/__pycache__/test_alipay_callback.cpython-313-pytest-9.1.1.pyc b/backend/tests/__pycache__/test_alipay_callback.cpython-313-pytest-9.1.1.pyc index 279cb96..4f4d813 100644 Binary files a/backend/tests/__pycache__/test_alipay_callback.cpython-313-pytest-9.1.1.pyc and b/backend/tests/__pycache__/test_alipay_callback.cpython-313-pytest-9.1.1.pyc differ diff --git a/backend/tests/__pycache__/test_alipay_provider.cpython-313-pytest-9.1.1.pyc b/backend/tests/__pycache__/test_alipay_provider.cpython-313-pytest-9.1.1.pyc index 72718e5..b6a2933 100644 Binary files a/backend/tests/__pycache__/test_alipay_provider.cpython-313-pytest-9.1.1.pyc and b/backend/tests/__pycache__/test_alipay_provider.cpython-313-pytest-9.1.1.pyc differ diff --git a/backend/tests/test_alipay_callback.py b/backend/tests/test_alipay_callback.py index 126b4a3..6db0e1a 100644 --- a/backend/tests/test_alipay_callback.py +++ b/backend/tests/test_alipay_callback.py @@ -12,14 +12,29 @@ def test_alipay_callback_reconciles_verified_success(monkeypatch): with TestClient(main.app) as client: with main.SessionLocal() as db: user = db.scalar(main.select(main.AdminUser).where(main.AdminUser.username == "operator")) + answer_user = main.AnswerUser( + public_id=f"U{uuid4().hex[:20]}", phone=f"1{uuid4().int % 10**10:010d}", + auth_token=uuid4().hex, + ) + db.add(answer_user) + db.flush() + submission = main.AnswerSubmission( + submission_no=f"ANS{uuid4().hex[:20]}", user_id=answer_user.id, + owner_id=user.id, shop_name="回调测试店", status="已通过", + ) + db.add(submission) + db.flush() order = main.AlipayRefundOrder( - target_kind="answer_submission", target_id=int(uuid4().int % 1_000_000_000), + target_kind="answer_submission", target_id=submission.id, out_biz_no=out_biz_no, amount_fen=300, account="buyer@example.com", - payee_name="测试收款人", created_by_id=user.id, status="pending", + payee_name="测试收款人", created_by_id=user.id, owner_id=user.id, + order_no=submission.submission_no, shop_name=submission.shop_name, + operator_name=user.display_name, payer_account="公司账户", status="pending", ) db.add(order) db.commit() order_id = order.id + submission_id = submission.id response = client.post("/alipay_callback", data={ "sign": "valid", "out_biz_no": out_biz_no, "status": "SUCCESS", "order_id": "ALI-CALLBACK-001", @@ -30,6 +45,12 @@ def test_alipay_callback_reconciles_verified_success(monkeypatch): order = db.get(main.AlipayRefundOrder, order_id) assert order.status == "success" assert order.alipay_trade_no == "ALI-CALLBACK-001" + assert db.get(main.AnswerSubmission, submission_id).status == "已完成" + refund = db.scalar(main.select(main.RefundRecord).where( + main.RefundRecord.target_kind == "answer_submission", + main.RefundRecord.target_id == submission_id, + )) + assert refund.alipay_trade_no == "ALI-CALLBACK-001" def test_alipay_callback_rejects_invalid_signature(monkeypatch): diff --git a/backend/tests/test_alipay_provider.py b/backend/tests/test_alipay_provider.py index 21b847e..6aebb0a 100644 --- a/backend/tests/test_alipay_provider.py +++ b/backend/tests/test_alipay_provider.py @@ -114,6 +114,32 @@ def test_transfer_keeps_unverified_success_envelope_ambiguous(monkeypatch): assert "勿重复发起" in str(error.value) +def test_certificate_response_verifies_only_inner_response_object(monkeypatch): + captured = {} + + def fake_verify(_public_key, content, signature): + captured["content"] = content + captured["signature"] = signature + return True + + from alipay.aop.api.util import SignatureUtils + monkeypatch.setattr(SignatureUtils, "verify_with_rsa", fake_verify) + envelope = json.dumps({ + "alipay_fund_trans_common_query_response": { + "code": "10000", "msg": "Success", "status": "SUCCESS", + "pay_fund_order_id": "ALI006", + }, + "alipay_cert_sn": "cert-sn", + "sign": "signed-value", + }, ensure_ascii=False, separators=(",", ":")) + + content = alipay_provider._signed_response_content(envelope, "public-key") + + assert json.loads(content)["pay_fund_order_id"] == "ALI006" + assert b"alipay_cert_sn" not in captured["content"] + assert captured["signature"] == "signed-value" + + def test_query_parses_processing_status(monkeypatch): client = CapturingClient({ "code": "10000",