签名说明

含MAVEN-JAR包版本
https://oss.tevaupay.com/tevau-nexus-test/Public_catalogue/demo.zip

"""
getSortedString 的排序规则与 UserTestController 里的完全一致（顶层按 key 字母排序，
嵌套对象要按 fastjson 默认 HashMap<String,Object> 的哈希桶顺序重排，而不是原始 JSON 顺序），
这里直接复用同一套实现。
"""

import base64
import hashlib
import json
import random
import string
import urllib.error
import urllib.request
from datetime import datetime


# ---------- DER 解析 + 纯标准库 RSA sign/verify（PKCS#1 v1.5） ----------

def _read_length(data: bytes, pos: int):
  length = data[pos]
  pos += 1
  if length & 0x80:
      num_bytes = length & 0x7F
      length = int.from_bytes(data[pos:pos + num_bytes], "big")
      pos += num_bytes
  return length, pos


def _read_tlv(data: bytes, pos: int):
  tag = data[pos]
  pos += 1
  length, pos = _read_length(data, pos)
  value = data[pos:pos + length]
  pos += length
  return tag, value, pos


def _parse_pkcs8_rsa_private_key(der: bytes):
  """解析 PKCS8 PrivateKeyInfo（DER），取出 RSA 模数 n 和私钥指数 d"""
  _, pkcs8_seq, _ = _read_tlv(der, 0)  # 外层 SEQUENCE
  pos = 0
  _, _version, pos = _read_tlv(pkcs8_seq, pos)            # INTEGER version
  _, _alg_id, pos = _read_tlv(pkcs8_seq, pos)              # AlgorithmIdentifier
  tag, priv_key_octets, pos = _read_tlv(pkcs8_seq, pos)    # OCTET STRING
  if tag != 0x04:
      raise ValueError("expected OCTET STRING for privateKey")

  _, rsa_priv_seq, _ = _read_tlv(priv_key_octets, 0)  # RSAPrivateKey ::= SEQUENCE
  ppos = 0
  _, _rsa_version, ppos = _read_tlv(rsa_priv_seq, ppos)
  _, modulus_bytes, ppos = _read_tlv(rsa_priv_seq, ppos)     # n
  _, _pub_exp_bytes, ppos = _read_tlv(rsa_priv_seq, ppos)    # e（签名不需要）
  _, priv_exp_bytes, ppos = _read_tlv(rsa_priv_seq, ppos)    # d

  n = int.from_bytes(modulus_bytes, "big")
  d = int.from_bytes(priv_exp_bytes, "big")
  return n, d


# SHA-1 的 DigestInfo DER 前缀（PKCS#1 v1.5 签名固定结构的一部分）
_SHA1_DIGEST_INFO_PREFIX = bytes.fromhex("3021300906052b0e03021a05000414")


def rsa_sign(data: str, private_key_b64: str) -> str:
  """对应 Java: RSAUtils.sign(data, privateKey) —— PKCS8 私钥"""
  n, d = _parse_pkcs8_rsa_private_key(base64.b64decode(private_key_b64))
  k = (n.bit_length() + 7) // 8  # RSA 模数字节长度

  digest_info = _SHA1_DIGEST_INFO_PREFIX + hashlib.sha1(data.encode()).digest()
  ps_len = k - 3 - len(digest_info)
  if ps_len < 8:
      raise ValueError("RSA modulus too small for SHA1 PKCS#1 v1.5 padding")
  em = b"\x00\x01" + b"\xff" * ps_len + b"\x00" + digest_info

  m = int.from_bytes(em, "big")
  s = pow(m, d, n)
  return base64.b64encode(s.to_bytes(k, "big")).decode()


# ---------- getSortedString 的等价实现（与 UserTestController 版本一致，直接复用） ----------

class OrderedObj(list):
  """标记这是 JSON 对象（保留原始字段出现顺序），用于和 JSON 数组（普通 list）区分开"""
  pass


def _parse_ordered(json_str: str):
  return json.loads(json_str, object_pairs_hook=OrderedObj, parse_float=str, parse_int=str)


def _java_string_hash(s: str) -> int:
  """复现 java.lang.String.hashCode()：h = 31*h + c，按 32 位位模式环绕"""
  h = 0
  for c in s:
      h = (31 * h + ord(c)) & 0xFFFFFFFF
  return h


def _java_spread_hash(s: str) -> int:
  """复现 java.util.HashMap.hash(key)：h ^ (h >>> 16)"""
  h = _java_string_hash(s)
  return (h ^ (h >> 16)) & 0xFFFFFFFF


def _simulate_java_hashmap_order(pairs):
  """
  复现 fastjson 嵌套 JSONObject 默认使用的普通 HashMap<String,Object>（非 LinkedHashMap）
  在 entrySet() 遍历时的顺序：按 key 的 spread hash 分桶，并模拟 JDK8+ HashMap 扩容时
  “低位/高位链表拆分”算法，而不是原始 JSON 字段顺序。
  """
  capacity = 16
  threshold = 12  # capacity * 0.75
  buckets = [[] for _ in range(capacity)]
  size = 0

  for key, value in pairs:
      h = _java_spread_hash(key)
      idx = h & (capacity - 1)
      buckets[idx].append((key, value, h))
      size += 1
      if size > threshold:
          new_cap = capacity * 2
          new_buckets = [[] for _ in range(new_cap)]
          for j in range(capacity):
              chain = buckets[j]
              if not chain:
                  continue
              lo, hi = [], []
              for entry in chain:
                  (lo if entry[2] & capacity == 0 else hi).append(entry)
              new_buckets[j] = lo
              new_buckets[j + capacity] = hi
          buckets = new_buckets
          capacity = new_cap
          threshold = capacity * 3 // 4

  result = []
  for bucket in buckets:
      for key, value, _ in bucket:
          result.append((key, value))
  return result


def _serialize_value(value) -> str:
  if value is None:
      return "null"
  if value is True:
      return "true"
  if value is False:
      return "false"
  if isinstance(value, OrderedObj):
      ordered = _simulate_java_hashmap_order(value)
      return "{" + ",".join(f"{k}:{_serialize_value(v)}" for k, v in ordered) + "}"
  if isinstance(value, list):
      return "[" + ",".join(_serialize_value(e) for e in value) + "]"
  return str(value)


def get_sorted_string(request_body_str: str) -> str:
  """对应 Java: getSortedString —— 按 key 排序拼接 k=v&，去掉 sign 字段和所有双引号"""
  if not request_body_str or not request_body_str.strip():
      return ""

  parsed = _parse_ordered(request_body_str)
  if not isinstance(parsed, OrderedObj):
      raise ValueError("expected top-level JSON object")

  filtered = [(k, v) for k, v in parsed if k != "sign"]
  filtered.sort(key=lambda kv: kv[0])  # 等价于 Java TreeMap<String,Object> 的自然升序

  sb = "".join(f"{k}={_serialize_value(v)}&" for k, v in filtered)
  result = sb.replace('"', "")
  sorted_str = result[:-1] if result else result  # 去掉末尾的 &
  print("排序后的字符串：" + sorted_str)
  return sorted_str


# ---------- 对应 main() 里用到的两个小工具方法 ----------

def get_timestamp() -> str:
  return datetime.now().strftime("%Y%m%d%H%M%S")


def random_number(length: int = 32) -> str:
  characters = string.ascii_uppercase + string.digits
  return "".join(random.choice(characters) for _ in range(length))


def main():
  # 接口文档地址
  post_url = "https://api-test.tevaupay.com/openapi/kyc/submitKycData"
  # 企业KYB时提交的appId
  app_id = "appId"
  # 平台提供的密钥
  private_key = (
      "xxxxx"
  )
  # 平台提供的apiKey
  x_nexus_api_key = "xxxx"
  # 接口定义入参
  body = (
      "{\n"
      '  "userCode": "CA3720688",\n'
      '  "countryArea": "MY",\n'
      '  "firstNameEn": "wilee",\n'
      '  "lastNameEn": "Matthews",\n'
      '  "birthday": "1980-01-01",\n'
      '  "identityCardType": "1",\n'
      '  "identityFrontPicUrl": "https:/xxx.jpg",\n'
      '  "identityBackPicUrl": "https://xxx.jpg",\n'
      '  "identityCard": "888888888",\n'
      '  "identityCardValidityTime": "2099-12-31"\n'
      "}"
  )
  # 接口定义版本号
  versions = "v1"
  # 时间戳
  timestamp = get_timestamp()
  # 随机数
  nonce = random_number()
  data = f"appId={app_id}&nonce={nonce}&timestamp={timestamp}" + get_sorted_string(body)
  # 签名
  sign = rsa_sign(data, private_key)

  print("sign加密前的数据 ==", data)
  print("sign加密后 ==", sign)

  headers = {
      "Content-type": "application/json; charset=utf-8",
      "Accept": "application/json",
      "appId": app_id,
      "timestamp": timestamp,
      "nonce": nonce,
      "x-nexus-api-key": x_nexus_api_key,
      "sign": sign,
      "versions": versions,
  }

  req = urllib.request.Request(post_url, data=body.encode("utf-8"), headers=headers, method="POST")
  try:
      with urllib.request.urlopen(req, timeout=10) as resp:
          resp_body = resp.read().decode("utf-8", errors="replace")
          print(f"Tevau: {post_url} request headers: {headers} \r\n response: {resp_body}")
  except urllib.error.URLError as e:
      print(f"Tevau: {post_url} 异常: {e}")


if __name__ == "__main__":
  main()

PHP包版本
https://oss.tevaupay.com/tevau-nexus-test/Public_catalogue/TevauTestHttpPhpUtils.php