yeepay_yop_parse_certificates
Parse certificate files to extract Base64-encoded RSA or SM2 public/private keys. Supports PFX and CER certificates with optional password.
Instructions
根据证书文件解析出Base64编码后的公钥或私钥字符串
Args: algorithm (str): 密钥算法，可选值为 "RSA" 或 "SM2"，默认为 "RSA" pfx_cert (str): 私钥证书（.pfx）文件路径 pub_cert (str): 公钥证书（.cer）文件路径 pwd (str, optional): PFX证书的密码，默认为None
Returns: dict: 包含解析结果的字典，格式如下: { 'message': 响应信息, 'privateKey': Base64编码后的私钥字符串, 'publicKey': Base64编码后的公钥字符串 }
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| algorithm | No | RSA | |
| pfx_cert | No | ||
| pub_cert | No | ||
| pwd | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- yop_mcp/main.py:251-277 (handler)MCP tool handler that registers 'yeepay_yop_parse_certificates' via @mcp.tool() decorator. This is the entry point exposed to the MCP client, delegating to parse_certificates from tools.cert_key_parser.
@mcp.tool() def yeepay_yop_parse_certificates( algorithm: str = "RSA", pfx_cert: Optional[str] = None, pub_cert: Optional[str] = None, pwd: Optional[str] = None, ) -> Dict[str, Any]: """ 根据证书文件解析出Base64编码后的公钥或私钥字符串 Args: algorithm (str): 密钥算法，可选值为 "RSA" 或 "SM2"，默认为 "RSA" pfx_cert (str): 私钥证书（.pfx）文件路径 pub_cert (str): 公钥证书（.cer）文件路径 pwd (str, optional): PFX证书的密码，默认为None Returns: dict: 包含解析结果的字典，格式如下: { 'message': 响应信息, 'privateKey': Base64编码后的私钥字符串, 'publicKey': Base64编码后的公钥字符串 } """ return parse_certificates( algorithm=algorithm, pfx_cert=pfx_cert, pub_cert=pub_cert, pwd=pwd ) - tools/cert_key_parser.py:17-91 (handler)Core implementation of parse_certificates() that orchestrates parsing of PFX (private key) and/or CER (public key) certificate files, returning Base64-encoded key strings. Validates algorithm type (RSA/SM2), checks file existence, delegates to parse_key_from_certificate(), and handles warnings.
def parse_certificates( algorithm: str = "RSA", pfx_cert: Optional[str] = None, pub_cert: Optional[str] = None, pwd: Optional[str] = None, ) -> Dict[str, Any]: result = {"message": "解析成功", "privateKey": None, "publicKey": None} # 验证算法类型 if algorithm not in ["RSA", "SM2"]: result["message"] = f"不支持的算法类型: {algorithm}，仅支持 RSA 和 SM2" return result # 如果两个证书文件都没有提供 if not pfx_cert and not pub_cert: result["message"] = "请至少提供一个证书文件（pfx_cert 或 pub_cert）" return result try: # 处理私钥证书 if pfx_cert and os.path.exists(pfx_cert): pfx_result = parse_key_from_certificate(pfx_cert, pwd) if pfx_result["private_key"]: result["privateKey"] = pfx_result["private_key"] if pfx_result["public_key"] and not result.get("publicKey"): result["publicKey"] = pfx_result["public_key"] # 检查算法类型是否匹配 if pfx_result["key_type"] != algorithm: result["message"] = ( f"警告：PFX证书中检测到的算法类型({pfx_result['key_type']})与指定的算法类型({algorithm})不匹配" ) elif pfx_cert: result["message"] = f"私钥证书文件不存在: {pfx_cert}" return result # 处理公钥证书 if pub_cert and os.path.exists(pub_cert): pub_result = parse_key_from_certificate(pub_cert) if pub_result["public_key"]: result["publicKey"] = pub_result["public_key"] # 检查算法类型是否匹配 if pub_result["key_type"] != algorithm: current_message = result["message"] if current_message and "warning" in current_message: result["message"] = ( current_message + f"，CER证书中检测到的算法类型({pub_result['key_type']})与指定的算法类型({algorithm})不匹配" ) else: result["message"] = ( f"警告：CER证书中检测到的算法类型({pub_result['key_type']})与指定的算法类型({algorithm})不匹配" ) elif pub_cert: current_message = result["message"] if (current_message and "warning" in current_message) or ( pfx_cert and os.path.exists(pfx_cert) ): result["message"] = ( current_message or "" ) + f"，公钥证书文件不存在: {pub_cert}" else: result["message"] = f"公钥证书文件不存在: {pub_cert}" return result # 检查是否至少解析出了一个密钥 if not result["privateKey"] and not result["publicKey"]: result["message"] = "未能从证书中解析出任何密钥" return result except (ValueError, OSError, IOError) as e: result["message"] = f"解析证书失败: {str(e)}" return result - tools/cert_key_parser.py:93-228 (helper)Helper function parse_key_from_certificate() that handles actual file reading and cryptographic parsing of .pfx/.p12 files (using pkcs12.load_key_and_certificates) and .cer/.pem files (using x509 certificate loading). Extracts public/private keys and determines key type (RSA/SM2/EC).
def parse_key_from_certificate( cert_path: str, password: Optional[str] = None ) -> Dict[str, Any]: """ 从证书文件解析出Base64编码的密钥字符串 Args: cert_path (str): 证书文件路径（.pfx 或 .cer 文件） password (str, optional): PFX证书的密码，默认为None Returns: dict: 包含解析结果的字典，格式如下: { 'key_type': 'RSA'|'SM2'|'UNKNOWN', 'public_key': Base64编码的公钥字符串, 'private_key': Base64编码的私钥字符串 (仅当从PFX文件提取时) } Raises: ValueError: 当证书格式不支持或解析失败时 """ # 检查文件是否存在 if not os.path.exists(cert_path): raise ValueError(f"证书文件不存在: {cert_path}") # 获取文件扩展名 _, ext = os.path.splitext(cert_path) ext = ext.lower() result = {"key_type": "UNKNOWN", "public_key": None, "private_key": None} try: # 处理PFX/PKCS12文件 (包含私钥) if ext in (".pfx", ".p12"): # 读取证书文件 with open(cert_path, "rb") as f: pfx_data = f.read() # 解析PFX password_bytes: Optional[bytes] = None if password: password_bytes = password.encode("utf-8") # 解析PFX/PKCS12文件 private_key, certificate, _ = pkcs12.load_key_and_certificates( pfx_data, password_bytes, default_backend() ) # 处理私钥 if private_key: private_bytes = None # 判断密钥类型 if isinstance(private_key, rsa.RSAPrivateKey): result["key_type"] = "RSA" # 获取PKCS8格式的私钥 private_bytes = private_key.private_bytes( encoding=serialization.Encoding.DER, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ) elif isinstance(private_key, ec.EllipticCurvePrivateKey): # 检查是否为SM2 curve = private_key.curve if hasattr(curve, "name") and "sm2" in curve.name.lower(): result["key_type"] = "SM2" else: result["key_type"] = "EC" # 获取PKCS8格式的私钥 private_bytes = private_key.private_bytes( encoding=serialization.Encoding.DER, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ) # Base64编码 if private_bytes: result["private_key"] = base64.b64encode(private_bytes).decode( "ascii" ) # 处理公钥证书 if certificate: public_key = certificate.public_key() # 获取公钥的DER编码 public_bytes = public_key.public_bytes( encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo, ) # Base64编码 result["public_key"] = base64.b64encode(public_bytes).decode("ascii") # 处理证书文件 (通常只有公钥) elif ext in (".cer", ".pem"): # 读取证书文件 with open(cert_path, "rb") as f: cert_data = f.read() # 尝试以PEM格式加载 try: certificate = load_pem_x509_certificate(cert_data, default_backend()) except ValueError: # 尝试以DER格式加载 certificate = load_der_x509_certificate(cert_data, default_backend()) # 提取公钥 public_key = certificate.public_key() # 判断密钥类型 if isinstance(public_key, rsa.RSAPublicKey): result["key_type"] = "RSA" elif isinstance(public_key, ec.EllipticCurvePublicKey): curve = public_key.curve if hasattr(curve, "name") and "sm2" in curve.name.lower(): result["key_type"] = "SM2" else: result["key_type"] = "EC" # 获取公钥的DER编码 public_bytes = public_key.public_bytes( encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo, ) # Base64编码 result["public_key"] = base64.b64encode(public_bytes).decode("ascii") else: raise ValueError(f"不支持的证书格式: {ext}") return result except Exception as e: raise ValueError(f"解析证书失败: {str(e)}") from e - yop_mcp/main.py:6-6 (registration)Import of parse_certificates from tools.cert_key_parser that connects the helper tool to the MCP handler.
from tools.cert_key_parser import parse_certificates - tools/cert_key_parser.py:17-22 (schema)Function signature and type hints for parse_certificates: algorithm (str, default 'RSA'), pfx_cert (Optional[str]), pub_cert (Optional[str]), pwd (Optional[str]), returns Dict[str, Any] with keys 'message', 'privateKey', 'publicKey'.
def parse_certificates( algorithm: str = "RSA", pfx_cert: Optional[str] = None, pub_cert: Optional[str] = None, pwd: Optional[str] = None, ) -> Dict[str, Any]: