import json import base64 import random import string import time import sys from Crypto.Cipher import AES, PKCS1_v1_5 from Crypto.PublicKey import RSA from Crypto.Util.Padding import pad, unpad import hashlib import hmac
class CompleteCryptoHandler: def __init__(self, aes_key=None, iv=None): self.fixed_aes_key = aes_key if aes_key else "......3690......" self.fixed_iv = iv if iv else "......0405......" self.public_rsa_key_b64 = "MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBAIVwAZ2E+O7qVmGPrBrVFVfe93V1bkLrV9LTbEA0tqPLgZz......" self.private_rsa_key_b64 = "MIIBVAIBADANBgkqhkiG9w0BAQEFAASCAT4wggE6AgEAAkEAjoS58XGU1pP+SEiJPxe9tjNiTE3orbr5I2fHtWp64p5Xsa14Yw5apqIaBNL6ot7B+FMCN7514vQhdFOs+kKvKQIDAQABAkB08eJmhc4dGYlRpPvB4uhQJu/lLxTb4g3niJQugDZf11aF2R7pBEl3zPDkXbw9KE+DbvIA5U0IkzRpZg46nIAFAiEAy1WTzlY3qxrr+ioeQneFMQW6rJQ6......" self.public_rsa_key = self._load_public_key() self.private_rsa_key = self._load_private_key() print("=" * 60) print(f"固定RSA公钥: {self.public_rsa_key_b64}") print(f"固定RSA私钥: {self.private_rsa_key_b64}") print(f"当前AES密钥: {self.fixed_aes_key}") print(f"当前IV: {self.fixed_iv}") print("AES填充模式: CBC/PKCS7") print("RSA填充模式: PKCS1_v1_5") print("=" * 60) def _load_public_key(self): """加载RSA公钥""" public_key_der = base64.b64decode(self.public_rsa_key_b64) return RSA.import_key(public_key_der) def _load_private_key(self): """加载RSA私钥""" private_key_der = base64.b64decode(self.private_rsa_key_b64) return RSA.import_key(private_key_der) def generate_random_str(self, length=32): """生成随机字符串""" chars = string.ascii_letters + string.digits return ''.join(random.choice(chars) for _ in range(length)) def rsa_encrypt(self, data): """RSA加密 - 使用PKCS1_v1_5填充模式""" try: cipher = PKCS1_v1_5.new(self.public_rsa_key) encrypted_data = cipher.encrypt(data.encode('utf-8')) return base64.b64encode(encrypted_data).decode('utf-8') except ValueError as e: raise ValueError(f"RSA加密失败: {e}. 数据长度可能超过限制") def rsa_decrypt(self, encrypted_data_b64): """RSA解密 - 使用PKCS1_v1_5填充模式""" try: print("🔓 开始RSA解密流程 (PKCS1_v1_5)") print(f"📥 输入Base64编码数据长度: {len(encrypted_data_b64)}") encrypted_data = base64.b64decode(encrypted_data_b64) print(f"✅ Base64解码完成,数据长度: {len(encrypted_data)} 字节") cipher = PKCS1_v1_5.new(self.private_rsa_key) decrypted_data = cipher.decrypt(encrypted_data, None) if decrypted_data is None: raise ValueError("RSA解密失败: 可能是密钥不匹配或数据损坏") print("✅ RSA解密完成") return decrypted_data.decode('utf-8') except base64.binascii.Error as e: raise ValueError(f"Base64解码失败: {e}") except ValueError as e: raise ValueError(f"RSA解密失败: {e}") except Exception as e: raise ValueError(f"解密过程中出错: {e}") def aes_encrypt(self, plaintext, aes_key=None): """AES加密 - CBC模式,PKCS7填充""" if aes_key is None: aes_key = self.fixed_aes_key key_bytes = aes_key.encode('utf-8') iv_bytes = self.fixed_iv.encode('utf-8') plaintext_bytes = plaintext.encode('utf-8') plaintext_bytes = pad(plaintext_bytes, AES.block_size) cipher = AES.new(key_bytes, AES.MODE_CBC, iv_bytes) encrypted_data = cipher.encrypt(plaintext_bytes) return encrypted_data.hex().upper() def aes_decrypt(self, encrypted_data_hex, aes_key=None): """AES解密 - CBC模式,PKCS7填充""" if aes_key is None: aes_key = self.fixed_aes_key key_bytes = aes_key.encode('utf-8') iv_bytes = self.fixed_iv.encode('utf-8') encrypted_data = bytes.fromhex(encrypted_data_hex) cipher = AES.new(key_bytes, AES.MODE_CBC, iv_bytes) decrypted_data = cipher.decrypt(encrypted_data) unpadded_data = unpad(decrypted_data, AES.block_size) return unpadded_data.decode('utf-8') def decrypt_with_fixed_aes(self, encrypted_data_hex, aes_key=None): """使用固定AES密钥解密数据""" if aes_key is None: aes_key = self.fixed_aes_key print("🔓 使用固定AES密钥解密") print("=" * 50) print(f"🔐 当前AES密钥: {aes_key}") print(f"🔐 当前IV: {self.fixed_iv}") print(f"📥 输入加密数据长度: {len(encrypted_data_hex)}") print(f"📥 输入加密数据: {encrypted_data_hex[:50]}..." if len(encrypted_data_hex) > 50 else f"📥 输入加密数据: {encrypted_data_hex}") try: decrypted_data = self.aes_decrypt(encrypted_data_hex, aes_key) print("✅ AES解密完成") print(f"📄 解密后的原始数据: {decrypted_data}") print("=" * 50) try: json_data = json.loads(decrypted_data) print("✅ JSON解析成功:") print(json.dumps(json_data, indent=2, ensure_ascii=False)) return json_data except json.JSONDecodeError: print("⚠️ 解密结果不是有效的JSON格式,返回原始字符串") return decrypted_data except Exception as e: raise ValueError(f"AES解密失败: {e}") def generate_hmac_signature(self, encrypted_aes_key, encrypt_data, nonce_str, timestamp): """生成HMAC-SHA256签名""" param_str = f"aesKey={encrypted_aes_key}&content={encrypt_data}&nonceStr={nonce_str}×tamp={timestamp}" key_bytes = encrypted_aes_key.encode('utf-8') signature = hmac.new( key_bytes, param_str.encode('utf-8'), digestmod=hashlib.sha256 ).hexdigest().upper() print(f"🔍 密钥长度(bytes): {len(key_bytes)}") print(f"✅ HMAC签名生成完成: {signature}") return signature def encrypt_request(self, user_input, aes_key=None): """加密请求数据 - 完整流程""" if aes_key is None: aes_key = self.fixed_aes_key try: json_data = json.loads(user_input) json_str = json.dumps(json_data, ensure_ascii=False, separators=(',', ':')) print("✅ 输入数据验证成功") print(f"原始数据: {json_str}") encrypt_data = self.aes_encrypt(json_str, aes_key) print(f"✅ AES加密完成, encryptData长度: {len(encrypt_data)}") encrypted_aes_key = self.rsa_encrypt(aes_key) print(f"✅ RSA加密AES密钥完成, aesKey: {encrypted_aes_key}") nonce_str = self.generate_random_str(32) timestamp = int(time.time() * 1000) encrypt_type = "AES" sign_type = "SHA256" print(f"📋 生成参数:") print(f" nonceStr: {nonce_str}") print(f" timestamp: {timestamp}") print(f" RSA填充模式: PKCS1_v1_5") print(f" 使用的AES密钥: {aes_key}") sign = self.generate_hmac_signature(encrypted_aes_key, encrypt_data, nonce_str, timestamp) print(f"✅ HMAC签名生成完成: {sign}") encrypted_request = { "sign": sign, "signType": sign_type, "aesKey": encrypted_aes_key, "nonceStr": nonce_str, "timestamp": timestamp, "encryptType": encrypt_type, "encryptData": encrypt_data } return encrypted_request except json.JSONDecodeError: raise ValueError("输入的不是有效JSON格式") except Exception as e: raise ValueError(f"加密过程中出错: {str(e)}") def decrypt_response(self, encrypt_data, encrypted_aes_key): """解密响应数据 - 添加详细的解密过程显示""" try: print("🔓 开始解密响应数据 (PKCS1_v1_5模式)") print("=" * 50) print(f"📥 输入参数:") print(f" encryptData长度: {len(encrypt_data)}") print(f" encrypted_aes_key长度: {len(encrypted_aes_key)}") print(f" encrypted_aes_key: {encrypted_aes_key}") print("=" * 50) print("\n🔑 RSA解密AES密钥过程 (PKCS1_v1_5):") print(f" 输入encrypted_aes_key: {encrypted_aes_key}") decrypted_aes_key = self.rsa_decrypt(encrypted_aes_key) print(f"✅ RSA解密AES密钥完成!") print(f"🔑 解密得到的真实AES密钥: [{decrypted_aes_key}]") print(f"🔑 密钥长度: {len(decrypted_aes_key)} 字符") print(f"🔑 密钥HEX表示: {decrypted_aes_key.encode('utf-8').hex().upper()}") print("=" * 50) print("\n📄 AES解密数据过程:") print(f" 输入encrypt_data长度: {len(encrypt_data)}") print(f" 使用的IV: {self.fixed_iv}") print(f" 使用的AES密钥: {decrypted_aes_key}") decrypted_data_str = self.aes_decrypt(encrypt_data, decrypted_aes_key) print(f"✅ AES解密完成!") print(f"📄 解密后的原始数据: {decrypted_data_str}") print("=" * 50) print("\n🔍 解析JSON数据:") decrypted_data = json.loads(decrypted_data_str) print(f"✅ JSON解析成功!") return decrypted_data except Exception as e: raise ValueError(f"解密过程中出错: {str(e)}")
def get_aes_key(prompt="请输入AES密钥(16位,回车使用默认值1472583690ABCDEF): "): """获取用户输入的AES密钥""" aes_key = input(prompt).strip() if not aes_key: aes_key = "1472583690ABCDEF" print(f"使用默认AES密钥: {aes_key}") else: print(f"使用输入的AES密钥: {aes_key}") return aes_key
def get_iv(prompt="请输入IV(16位,回车使用默认值0102030405060708): "): """获取用户输入的IV""" iv = input(prompt).strip() if not iv: iv = "0102030405060708" print(f"使用默认IV: {iv}") else: print(f"使用输入的IV: {iv}") return iv
def main(): """主函数 - 提供交互式加解密工具""" print("=" * 60) print("基于AES-CBC + RSA(PKCS1_v1_5) + HMAC-SHA256的加解密工具") print("=" * 60) crypto = CompleteCryptoHandler() while True: try: print("\n请选择操作:") print("1. 加密JSON数据(生成请求包)") print("2. 解密密文数据(解密响应包)") print("3. 使用固定AES密钥解密请求包") print("q. 退出") choice = input("\n请输入选择 (1/2/3/q): ").strip().lower() if choice == 'q': print("程序已退出") break elif choice == '1': print("\n【加密模式 - 生成请求包】") aes_key = get_aes_key() iv = get_iv() crypto = CompleteCryptoHandler(aes_key, iv) print("\n示例: {\"configType\":\"mobile\",\"encrypt\":true}") user_input = input("请输入要加密的JSON数据: ").strip() if not user_input: print("错误: 输入不能为空") continue try: result = crypto.encrypt_request(user_input, aes_key) print("\n" + "=" * 40) print("✅ 加密成功 - 请求包格式:") print("=" * 40) print(json.dumps(result, indent=2, ensure_ascii=False)) except Exception as e: print(f"❌ 加密错误: {e}") elif choice == '2': print("\n【解密模式 - 解密响应包】") iv = get_iv() crypto = CompleteCryptoHandler(crypto.fixed_aes_key, iv) encrypt_data = input("请输入encryptData: ").strip() aes_key = input("请输入aesKey: ").strip() if not encrypt_data or not aes_key: print("错误: 参数不能为空") continue try: result = crypto.decrypt_response(encrypt_data, aes_key) print("\n" + "=" * 40) print("✅ 解密成功:") print("=" * 40) if isinstance(result, dict): print(json.dumps(result, indent=2, ensure_ascii=False)) else: print(result) except Exception as e: print(f"❌ 解密错误: {e}") elif choice == '3': print("\n【解密模式 - 使用固定AES密钥解密请求包】") aes_key = get_aes_key() iv = get_iv() crypto = CompleteCryptoHandler(aes_key, iv) print(f"🔐 当前AES密钥: {aes_key}") print(f"🔐 当前IV: {iv}") encrypt_data = input("请输入encryptData: ").strip() if not encrypt_data: print("错误: encryptData不能为空") continue try: result = crypto.decrypt_with_fixed_aes(encrypt_data, aes_key) print("\n" + "=" * 40) print("✅ 使用固定AES密钥解密成功:") print("=" * 40) if isinstance(result, dict): print(json.dumps(result, indent=2, ensure_ascii=False)) else: print(result) except Exception as e: print(f"❌ 解密错误: {e}") else: print("错误: 请输入1、2、3或q") except KeyboardInterrupt: print("\n程序被用户中断") break except Exception as e: print(f"发生未知错误: {e}")
if __name__ == "__main__": main()
|