yeepay_yop_download_cert

Instructions

Implementation Reference

• yop_mcp/main.py:215-248 (registration)

  MCP tool registration of 'yeepay_yop_download_cert' via @mcp.tool() decorator, defining schema (algorithm, serial_no, auth_code, private_key, public_key, pwd) and delegating to download_cert()

  @mcp.tool()
  def yeepay_yop_download_cert(  # pylint: disable=too-many-arguments,too-many-positional-arguments
      algorithm: str = "RSA",
      serial_no: str = "",
      auth_code: str = "",
      private_key: str = "",
      public_key: str = "",
      pwd: str = "",
  ) -> Dict[str, Any]:
      """
      根据密钥算法、CFCA证书的序列号、授权码、非对称密钥对（公钥和私钥）、密码，下载该证书，并保存到本地路径
  
      Args:
          algorithm: 密钥算法，可选值为 "RSA" 或 "SM2"，默认为 "RSA"
          serial_no: cfca证书序列号
          auth_code: cfca证书授权码
          private_key: Base64 编码后的私钥字符串
          public_key: Base64 编码后的公钥字符串
          pwd: 密码，长度：12~16位
  
      Returns:
          Dict包含:
          - message: 响应信息
          - pfxCert: 私钥证书路径(.pfx)
          - pubCert: 公钥证书路径(.cer)
      """
      return download_cert(
          algorithm=algorithm,
          serial_no=serial_no,
          auth_code=auth_code,
          private_key=private_key,
          public_key=public_key,
          pwd=pwd,
      )

• tools/cert_utils.py:460-547 (handler)

  Core handler function `download_cert()` that validates inputs, checks key pair match, generates P10 cert request, downloads cert from CFCA API, checks cert-key match, saves .pfx and .cer files

  def download_cert(
      algorithm: str = "RSA",
      serial_no: str = "",
      auth_code: str = "",
      private_key: str = "",
      public_key: str = "",
      pwd: str = "",
  ) -> Dict[str, Any]:
      # 确定密钥类型
      key_type = KeyType.SM2 if algorithm.upper() == "SM2" else KeyType.RSA2048
  
      # 检查输入参数
      check_result = CertUtils.check_input(
          serial_no, auth_code, key_type, private_key, public_key, pwd
      )
      if not check_result.result:
          return {"message": check_result.msg}
  
      # 检查公私钥匹配
      p10_generated = False  # 标记是否已生成P10请求
      try:
          if not p10_generated and not CertUtils.check_key(
              private_key, public_key, key_type
          ):
              return {"message": "商户公私钥不匹配，请重新输入"}
      except Exception as e:
          return {"message": f"密钥解析异常: {str(e)}"}
  
      # 生成证书请求
      if p10_generated:
          cert_req = private_key
      else:
          try:
              cert_req = CertUtils.gen_p10(private_key, public_key, key_type)
          except Exception as e:
              return {"message": f"生成证书请求失败: {str(e)}"}
  
      # 确定证书保存路径
      cert_path = (
          Config.SM2_CERT_SAVE_PATH
          if key_type == KeyType.SM2
          else Config.RSA_CERT_SAVE_PATH
      )
      pri_cert_path = os.path.join(cert_path, f"{serial_no}.pfx")
      pub_cert_path = os.path.join(cert_path, f"{serial_no}.cer")
  
      # 检查证书是否已存在
      if SupportUtil.is_file_exists(pri_cert_path) and SupportUtil.is_file_exists(
          pub_cert_path
      ):
          return {
              "message": "本地证书已存在",
              "pfxCert": pri_cert_path,
              "pubCert": pub_cert_path,
          }
  
      try:
          # 获取证书
          cert: Optional[str] = None
          if SupportUtil.is_file_exists(pub_cert_path):
              cert = SupportUtil.read_file_as_string(pub_cert_path)
          else:
              cert_download_result = CertUtils.download_cert_from_cfca(
                  serial_no, auth_code, cert_req
              )
              if cert_download_result.error_msg:
                  return {"message": cert_download_result.error_msg}
              cert = cert_download_result.cert
  
          # 检查证书与私钥匹配
          if cert and not CertUtils.check_cert(private_key, cert, key_type):
              return {"message": "证书已下载过，且证书与输入的私钥不匹配，请核对"}
  
          # 保存证书
          if cert:
              pub_cert_path = CertUtils.make_pub_cert(cert, serial_no, cert_path)
          if not p10_generated and cert:
              pri_cert_path = CertUtils.make_pfx_cert(
                  private_key, cert, key_type, pwd, serial_no, cert_path
              )
  
          return {
              "message": "CFCA证书激活并下载成功",
              "pfxCert": pri_cert_path,
              "pubCert": pub_cert_path,
          }
      except Exception as e:
          return {"message": f"系统异常，请稍后重试: {str(e)}"}

• tools/cert_utils.py:376-410 (helper)

  CertUtils.download_cert_from_cfca() - calls CFCA API to download the certificate with serialNo, authCode, certReq parameters

  def download_cert_from_cfca(
      serial_no: str, auth_code: str, cert_req: str
  ) -> CertDownloadResult:
      try:
          # 准备请求数据
          param = {
              "serialNo": serial_no,
              "authCode": auth_code,
              "certReq": cert_req,
              "toolsVersion": Config.TOOLS_VERSION,
          }
  
          # 发送请求到CFCA API
          headers = {}
          headers["Authorization"] = "Basic " + base64.b64encode(
              Config.BASIC.encode("utf-8")
          ).decode("utf-8")
  
          response = HttpUtils.get_response(
              Config.CFCA_CERT_DOWNLOAD_URL, param, headers
          )
          map_data = JsonUtils.json_to_pojo(response, dict)
  
          if map_data.get("code") == "000000":
              data_map = map_data.get("data")
              return CertDownloadResult().with_cert(
                  "-----BEGIN CERTIFICATE-----\n"
                  + data_map.get("cert")
                  + "\n-----END CERTIFICATE-----"
              )
          else:
              return CertDownloadResult().with_error_msg(map_data.get("message"))
  
      except Exception as e:
          return CertDownloadResult(error_msg=f"下载证书失败: {str(e)}")

• tools/cert_utils.py:263-297 (helper)

  CertUtils.check_input() - validates input parameters (serial_no, auth_code, key_type, pri_key, pub_key, pwd)

  @staticmethod
  def check_input(
      serial_no: str,
      auth_code: str,
      key_type: KeyType,
      pri_key: str,
      pub_key: str,
      pwd: str,
  ) -> CheckResult:
      """检查输入参数有效性"""
      # 使用key_type参数进行验证
      if key_type not in [KeyType.RSA2048, KeyType.SM2]:
          return CheckResult(False, f"不支持的密钥类型: {key_type}")
  
      if not serial_no:
          return CheckResult(False, "证书序列号不能为空")
      if not auth_code:
          return CheckResult(False, "授权码不能为空")
      if not pri_key:
          return CheckResult(False, "私钥不能为空")
      if not pub_key:
          return CheckResult(False, "公钥不能为空")
  
      # 密码长度检查
      if pwd and (len(pwd) < 12 or len(pwd) > 16):
          return CheckResult(False, "密码长度应为12-16位")
  
      # 检查密钥格式是否为Base64
      try:
          base64.b64decode(pri_key)
          base64.b64decode(pub_key)
      except Exception:
          return CheckResult(False, "密钥格式不正确，应为Base64编码")
  
      return CheckResult(True)

• tools/config.py:4-33 (helper)

  Config class with certificate save paths (RSA_CERT_SAVE_PATH, SM2_CERT_SAVE_PATH) and CFCA API endpoint configuration

  class Config:
      # 证书保存路径
      RSA_CERT_SAVE_PATH = "./certs/rsa/"
      SM2_CERT_SAVE_PATH = "./certs/sm2/"
  
      # API主机地址
      HOST = "https://mp.yeepay.com"
  
      # CFCA API相关配置
      CFCA_CERT_DOWNLOAD_URL = HOST + "/yop-developer-center/apis/cfca/cert/download"
      BASIC = "keytools:keytools"
      TOOLS_VERSION = "mcp"
  
      # QA环境配置
      QA_HOST_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "config")
  
      @classmethod
      def get_cert_path(cls, algorithm: str) -> str:
          """获取证书保存路径"""
          if algorithm.upper() == "RSA":
              return cls.RSA_CERT_SAVE_PATH
          if algorithm.upper() == "SM2":
              return cls.SM2_CERT_SAVE_PATH
          raise ValueError(f"不支持的算法: {algorithm}")
  
      @classmethod
      def is_supported_algorithm(cls, algorithm: str) -> bool:
          """检查是否支持指定算法"""
          return algorithm.upper() in ["RSA", "SM2"]