Files

242 lines
8.7 KiB
Python

"""YYB 付款表单的离线回归测试。"""
from __future__ import annotations
import importlib.util
import sys
import unittest
from pathlib import Path
from urllib.parse import parse_qs
_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "jsdom-pay.py"
_SPEC = importlib.util.spec_from_file_location("yyb_jsdom_pay", _SCRIPT)
assert _SPEC and _SPEC.loader
_MODULE = importlib.util.module_from_spec(_SPEC)
sys.modules[_SPEC.name] = _MODULE
_SPEC.loader.exec_module(_MODULE)
from pyvm.algorithm import derive_key1_from_key16, generate_encrypt_msg_offline
from pyvm.payment_errors import describe_payment_failure
from pyvm.protocol import goods_material_diagnostics, validate_goods_materials
class TestJsdomPay(unittest.TestCase):
def test_build_save_body_has_current_payment_context(self):
body = _MODULE.build_save_body(
{
"token_id": "token",
"transaction_id": "transaction",
"out_trade_no": "trade",
},
{"openid": "openid", "accesstoken": "access-token", "logintype": "WX"},
"web-token",
"anti-token",
"cipher",
"2",
"mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-iap",
600,
)
fields = parse_qs(body, keep_blank_values=True)
self.assertEqual(fields["type"], ["bg"])
self.assertEqual(
fields["pf"], ["mds_myappjp-__mds_myappjp_PC_aW9zd2hpdGVsaX0-iap"]
)
self.assertEqual(fields["zoneid"], ["2"])
self.assertEqual(fields["pay_method"], ["wechat"])
self.assertEqual(fields["wcp"], ["type=CNY&amt=600"])
self.assertEqual(fields["encrypt_way"], ["web_new_encrypt"])
self.assertEqual(fields["webversion"], ["minipayv2"])
def test_qq_payment_uses_qq_oauth_session_fields(self):
body = _MODULE.build_save_body(
{
"token_id": "token",
"transaction_id": "transaction",
"out_trade_no": "trade",
},
{
"openid": "openid",
"accesstoken": "access-token",
"logintype": "QC",
"appid": "102033112",
},
"web-token",
"anti-token",
"cipher",
"1",
"pf",
100,
)
fields = parse_qs(body, keep_blank_values=True)
self.assertEqual(fields["session_id"], ["openid"])
self.assertEqual(fields["session_type"], ["kp_accesstoken"])
self.assertEqual(fields["wx_appid"], [""])
self.assertEqual(fields["offerid_for_qq_appid"], ["102033112"])
self.assertNotIn("qq_appid", fields)
def test_wechat_payment_does_not_include_qq_offer_field(self):
body = _MODULE.build_save_body(
{
"token_id": "token",
"transaction_id": "transaction",
"out_trade_no": "trade",
},
{"openid": "openid", "accesstoken": "access-token", "logintype": "WX"},
"web-token",
"anti-token",
"cipher",
"1",
"pf",
100,
)
fields = parse_qs(body, keep_blank_values=True)
self.assertNotIn("offerid_for_qq_appid", fields)
def test_qq_page_info_matches_the_web_flow(self):
body = _MODULE.build_page_info_body(
{
"token_id": "token",
"transaction_id": "transaction",
"out_trade_no": "trade",
},
{
"openid": "openid",
"accesstoken": "access-token",
"logintype": "QC",
"appid": "102033112",
},
"anti-token",
"1",
"pf",
)
fields = parse_qs(body, keep_blank_values=True)
self.assertEqual(fields["isusempaymode"], ["1"])
self.assertEqual(fields["offerid_for_qq_appid"], ["102033112"])
self.assertNotIn("pay_method", fields)
def test_qq_risk_rejection_is_not_reported_as_qr_error(self):
message = describe_payment_failure(
"QC", "payment", "ret=1099 err_code=1099-1007-0"
)
self.assertIn("风控/限流", message)
self.assertIn("未生成微信付款码", message)
self.assertIn("停止连续重试", message)
def test_wechat_rejection_is_not_labeled_as_risk_control(self):
message = describe_payment_failure(
"WX", "payment", "ret=1099 err_code=1099-1007-0"
)
self.assertIn("服务端拒绝", message)
self.assertNotIn("风控", message)
def test_game_role_error_is_classified_separately(self):
message = describe_payment_failure("QC", "order", "inner=10014 gamerole err")
self.assertIn("角色或区服校验失败", message)
self.assertIn("重新选择角色和区服", message)
def test_qq_encrypt_rand_aligns_plaintext_to_aes_block(self):
params = {
"token_id": "t",
"openid": "o",
"openkey": "k",
"session_id": "openid",
"session_type": "kp_accesstoken",
"zoneid": "1",
"pay_method": "wechat",
"buy_quantity": "1",
"from_h5": "1",
"webversion": "minipayv2",
}
rand_value = _MODULE.make_encrypt_rand(
params, "tdrc_session%3Dpay-test", "1775990000", True
)
plaintext = _MODULE.build_plaintext(
params, "tdrc_session%3Dpay-test", "1775990000", rand_value
)
self.assertEqual(len(plaintext.encode("latin-1")) % 16, 0)
self.assertGreaterEqual(len(rand_value), 8)
def test_wechat_encrypt_rand_preserves_historical_control_byte(self):
rand_value = _MODULE.make_encrypt_rand(
{}, "tdrc_session%3Dpay-test", "1775990000", False
)
self.assertEqual(len(rand_value), 9)
self.assertEqual(rand_value[-1], "\x01")
def test_goods_protocol_template_is_compatible(self):
template = _MODULE.load_template_args()
validate_goods_materials(template, [0] * 59640)
diagnostics = goods_material_diagnostics(_MODULE.ROOT, template, [0] * 59640)
self.assertEqual(diagnostics["goods_xmidas_ops_length"], 59640)
self.assertEqual(diagnostics["args_template_slots"], 18)
self.assertEqual(len(diagnostics["bytecode_sha256_prefix"]), 12)
def test_private_evidence_writer_restricts_permissions(self):
import tempfile
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "evidence.json"
_MODULE.write_private_text(path, "sensitive")
self.assertEqual(path.read_text(encoding="utf-8"), "sensitive")
self.assertEqual(path.stat().st_mode & 0o777, 0o600)
def test_goods_vm_generates_expected_full_length_ciphertext(self):
"""528B 明文应走完 33 个块,输出 1056 个 hex 字符。"""
import json
template = _MODULE.load_template_args()
xmidas = json.loads(
(_MODULE.ROOT / "replay" / "xmidasops.json").read_text(encoding="utf-8")
)
params = {
"token_id": "token",
"openid": "openid",
"openkey": "key",
"session_id": "openid",
"session_type": "kp_accesstoken",
"zoneid": "1",
"pay_method": "wechat",
"buy_quantity": "1",
"mb_pwd": "",
"pay_id": "",
"auth_key": "",
"card_value": "",
"accounttype": "",
"provide_uin": "",
"extend": "",
"from_h5": "1",
"webversion": "minipayv2",
}
timestamp = "1775990000"
fk_extend = "tdrc_session%3Dpay-test"
for length in range(512):
params["extend"] = "x" * length
rand_value = _MODULE.make_encrypt_rand(params, fk_extend, timestamp, True)
plaintext = _MODULE.build_plaintext(
params, fk_extend, timestamp, rand_value
)
if len(plaintext.encode("latin-1")) == 528:
break
else:
self.fail("无法构造 528B goods 测试明文")
key16 = list(range(16))
key1 = derive_key1_from_key16(
key16, [template[i][0] for i in (1, 2, 3, 4)], template[5][0]
)
ciphertext = generate_encrypt_msg_offline(
params,
fk_extend,
timestamp,
rand_value,
key16=key16,
key1=key1,
args_template=template,
xmidas=xmidas,
xmidas_token="A" * 96,
)
self.assertEqual(len(ciphertext), 1056)
if __name__ == "__main__":
unittest.main()