diff --git a/weishijiajie/base64_custom.py b/weishijiajie/base64_custom.py deleted file mode 100644 index 153e418..0000000 --- a/weishijiajie/base64_custom.py +++ /dev/null @@ -1,155 +0,0 @@ -class Base64: - BASELENGTH = 128 - LOOKUPLENGTH = 64 - TWENTYFOURBITGROUP = 24 - EIGHTBIT = 8 - SIXTEENBIT = 16 - FOURBYTE = 4 - SIGN = -128 - PAD = '=' - - def __init__(self): - self.base64Alphabet = [-1] * self.BASELENGTH - self.lookUpBase64Alphabet = [''] * self.LOOKUPLENGTH - self._init_alphabets() - - def _init_alphabets(self): - for i in range(65, 91): # A-Z - self.base64Alphabet[i] = i - 65 - for i in range(97, 123): # a-z - self.base64Alphabet[i] = i - 97 + 26 - for i in range(48, 58): # 0-9 - self.base64Alphabet[i] = i - 48 + 52 - self.base64Alphabet[43] = 62 # '+' - self.base64Alphabet[47] = 63 # '/' - - for i in range(0, 26): - self.lookUpBase64Alphabet[i] = chr(65 + i) - for i in range(26, 52): - self.lookUpBase64Alphabet[i] = chr(97 + i - 26) - for i in range(52, 62): - self.lookUpBase64Alphabet[i] = chr(48 + i - 52) - self.lookUpBase64Alphabet[62] = '+' - self.lookUpBase64Alphabet[63] = '/' - - @staticmethod - def isWhiteSpace(char): - return char in (' ', '\n', '\r', '\t') - - @staticmethod - def isPad(char): - return char == '=' - - def isData(self, char): - return ord(char) < self.BASELENGTH and self.base64Alphabet[ord(char)] != -1 - - def encode(self, binaryData: bytes) -> str: - if binaryData is None: - return None - - lengthDataBits = len(binaryData) * self.EIGHTBIT - fewerThan24bits = len(binaryData) % 3 - numberTriplets = len(binaryData) // 3 - encodedData = [] - - dataIndex = 0 - for _ in range(numberTriplets): - byte1 = binaryData[dataIndex] - byte2 = binaryData[dataIndex + 1] - byte3 = binaryData[dataIndex + 2] - - dataIndex += 3 - - val1 = byte1 >> 2 - val2 = ((byte1 & 0x3) << 4) | (byte2 >> 4) - val3 = ((byte2 & 0xf) << 2) | (byte3 >> 6) - val4 = byte3 & 0x3f - - encodedData.extend([ - self.lookUpBase64Alphabet[val1], - self.lookUpBase64Alphabet[val2], - self.lookUpBase64Alphabet[val3], - self.lookUpBase64Alphabet[val4], - ]) - - if fewerThan24bits == 1: - byte1 = binaryData[dataIndex] - val1 = byte1 >> 2 - val2 = (byte1 & 0x3) << 4 - - encodedData.extend([ - self.lookUpBase64Alphabet[val1], - self.lookUpBase64Alphabet[val2], - self.PAD, - self.PAD - ]) - elif fewerThan24bits == 2: - byte1 = binaryData[dataIndex] - byte2 = binaryData[dataIndex + 1] - - val1 = byte1 >> 2 - val2 = ((byte1 & 0x3) << 4) | (byte2 >> 4) - val3 = (byte2 & 0xf) << 2 - - encodedData.extend([ - self.lookUpBase64Alphabet[val1], - self.lookUpBase64Alphabet[val2], - self.lookUpBase64Alphabet[val3], - self.PAD - ]) - - return ''.join(encodedData) - - def removeWhiteSpace(self, data: str) -> str: - return ''.join([c for c in data if not self.isWhiteSpace(c)]) - - def decode(self, encoded: str) -> bytes: - if encoded is None: - return None - - encoded = self.removeWhiteSpace(encoded) - if len(encoded) % 4 != 0: - return None - - numberQuadruple = len(encoded) // 4 - decodedBytes = bytearray() - - i = 0 - for _ in range(numberQuadruple): - c1 = encoded[i] - c2 = encoded[i + 1] - c3 = encoded[i + 2] - c4 = encoded[i + 3] - i += 4 - - if not (self.isData(c1) and self.isData(c2)): - return None - - b1 = self.base64Alphabet[ord(c1)] - b2 = self.base64Alphabet[ord(c2)] - - b3 = self.base64Alphabet[ord(c3)] if c3 != self.PAD else 0 - b4 = self.base64Alphabet[ord(c4)] if c4 != self.PAD else 0 - - byte1 = (b1 << 2) | (b2 >> 4) - decodedBytes.append(byte1) - - if c3 != self.PAD: - byte2 = ((b2 & 0xf) << 4) | (b3 >> 2) - decodedBytes.append(byte2) - - if c4 != self.PAD: - byte3 = ((b3 & 0x3) << 6) | b4 - decodedBytes.append(byte3) - - return bytes(decodedBytes) - -class Base64Utils: - @staticmethod - def encode(data: bytes) -> str: - return Base64().encode(data).decode("utf-8") - - @staticmethod - def decode(data: str) -> bytes: - return Base64().decode(data.encode("utf-8")) - diff --git a/weishijiajie/jj_utils.py b/weishijiajie/jj_utils.py new file mode 100644 index 0000000..352e882 --- /dev/null +++ b/weishijiajie/jj_utils.py @@ -0,0 +1,163 @@ +import base64 +import json +from appPublic.rsawrap import RSA +from os import urandom +from cryptography.hazmat.primitives import padding +from cryptography.hazmat.primitives.asymmetric import padding as asym_padding +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.backends import default_backend + +def get_rsa_pubkey_from_b64keystring(b64ks): + der_data = base64.b64decode(b64ks) + # 加载公钥 + public_key = serialization.load_der_public_key(der_data, backend=default_backend()) + return public_key + +def pubkey_encode(pubkey, bin): + ciphertext = pubkey.encrypt( + bin, + asym_padding.OAEP( + mgf=asym_padding.MGF1(algorithm=hashes.SHA256()), + algorithm=hashes.SHA256(), + label=None + ) + ) + + # 输出 base64 编码后的密文 + ciphertext_b64 = base64.b64encode(ciphertext).decode() + return ciphertext_b64 + +def get_rsa_prikey_from_b64keystring(b64ks): + der_data = base64.b64decode(b64ks) + private_key = serialization.load_der_private_key(der_data, password=None, backend=default_backend()) + return private_key + +def prikey_decode(prikey, text): + ciphertext = base64.b64decode(text) + plaintext = prikey.decrypt( + ciphertext, + asym_padding.OAEP( + mgf=asym_padding.MGF1(algorithm=hashes.SHA256()), + algorithm=hashes.SHA256(), + label=None + ) + ) + return plaintext + +def generate_key(key_size=32): + return urandom(key_size) # 32 字节 = 256 位 + +def pad(data: bytes) -> bytes: + padder = padding.PKCS7(128).padder() # AES 块大小是 128 bits + return padder.update(data) + padder.finalize() + +def unpad(data: bytes) -> bytes: + unpadder = padding.PKCS7(128).unpadder() + return unpadder.update(data) + unpadder.finalize() + +def aes_encrypt_ecb(key: bytes, plaintext: str) -> bytes: + cipher = Cipher(algorithms.AES(key), modes.ECB(), backend=default_backend()) + encryptor = cipher.encryptor() + padded_data = pad(plaintext.encode()) + return encryptor.update(padded_data) + encryptor.finalize() + +def aes_decrypt_ecb(key: bytes, ciphertext: bytes) -> str: + cipher = Cipher(algorithms.AES(key), modes.ECB(), backend=default_backend()) + decryptor = cipher.decryptor() + padded_plain = decryptor.update(ciphertext) + decryptor.finalize() + return unpad(padded_plain).decode() + +def data_jj_encode(data, pubkey_b64str): + """ return a dict with: + AESKey:aes key encrypted by public key + data: data encrypted by aes key + """ + pubkey = get_rsa_pubkey_from_b64keystring(pubkey_b64str) + aes_key = generate_key() + print(f'{aes_key=}') + aes_key_cyber = pubkey_encode(pubkey, aes_key) + + jd = json.dumps(data) + data_cyber = aes_encrypt_ecb(aes_key, jd) + data_cyber_b64 = base64.b64encode(data_cyber).decode('utf-8') + return { + 'AESKey': aes_key_cyber, + 'data': data_cyber_b64 + } + +def data_jj_decode(cdata, prikey_b64str): + prikey = get_rsa_prikey_from_b64keystring(prikey_str) + aes_key = prikey_decode(prikey, cdata['AESKey']) + print(f'{aes_key=}') + js_cyber = base64.b64decode(cdata['data']) + js = aes_decrypt_ecb(aes_key, js_cyber) + d = json.loads(js) + return d + +if __name__ == '__main__': + # 示例 + import sys + + key = generate_key() + if len(sys.argv) > 1: + key = base64.b64decode(sys.argv[1]) + key_b64 = base64.b64encode(key) + + plaintext = "互信是合作与发展的基础" + ciphertext = aes_encrypt_ecb(key, plaintext) + cb64 = base64.b64encode(ciphertext) + print("原文:", plaintext, "密文(base64ed):", cb64, "KEY:", base64.b64encode(key)) + decrypted = aes_decrypt_ecb(key, ciphertext) + print("解密结果:", decrypted) + d = { + "field1":"19ygf23", + "filed2": 2, + "field3": "r32g" + } + pubkey_str="MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA3LWvOQCufDQpf9PsOrBF" + \ + "SsjPqIg/x1YZpQ4MksZs4BwHWjHT9PHoJ88OZl2cXmyaY6Hde2KXG1v873lI70/2" + \ + "oUizX8Fyl0TTliAQ/p/kSsmgHfzhY+KOZUQ5HwXwnjFKv1q+Y0PpseNPFFHGiB1H" + \ + "/6p8kJeV61GN00etwvAvDdfZKm9W6t1ByD9Bsr3/VSHJtv1DqVdgmVMxnI7W4mx6" + \ + "UmseGmywngLvBHXI0+MWRK1RszCMISA6SsKTlT9IdT/r64vLht+NImEY6KPwODpu" + \ + "6vBUSyRqbGVitZ6PPIPwzLkB8LZ69eTjK5SMXP6RD7kHio+vb4ljHWXzDfRH9HXq" + \ + "FQIDAQAB" + + prikey_str = ( + "MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQDcta85AK58NCl/" + "0+w6sEVKyM+oiD/HVhmlDgySxmzgHAdaMdP08egnzw5mXZxebJpjod17YpcbW/zv" + "eUjvT/ahSLNfwXKXRNOWIBD+n+RKyaAd/OFj4o5lRDkfBfCeMUq/Wr5jQ+mx408U" + "UcaIHUf/qnyQl5XrUY3TR63C8C8N19kqb1bq3UHIP0Gyvf9VIcm2/UOpV2CZUzGc" + "jtbibHpSax4abLCeAu8EdcjT4xZErVGzMIwhIDpKwpOVP0h1P+vri8uG340iYRjo" + "o/A4Om7q8FRLJGpsZWK1no88g/DMuQHwtnr15OMrlIxc/pEPuQeKj69viWMdZfMN" + "9Ef0deoVAgMBAAECggEAAWQPt58+ej/l5w1R1sjXlvIPBPPe2wW64aRmBeLEOuFS" + "Voor2MZRfTa7039RhfLPQkN53tYvN5frzyKHiCKL3lZT8w2ICFYBs1d+Y+xkBuig" + "XrACR5vc49x7M6vY6VZML0qGGqAJ7/oLvIgLZT4x/vV3FX4vDq2T5Dn2aV0UHtUz" + "ZSwJHJLTqIyHSSzAJetW8cEoEE1u3g7Wu3K2+7Z7nRqQzXnpC3+4wNGV+C7/NYWr" + "8VEf8ygRT0Znom0DIWHnAS4PCJFYefkQfxaDqAA3JSUj3BrbnFJzC4+sSPSmPueL" + "6HWubeZyIiUud7h5mGZS/VNylCp/92gvumJ8YeKA4QKBgQDvbXa0TFXgLNaunf6T" + "PK7dFSZE2gIGyDhRNsAWvDXQITmfQDe5cwI1++tMnsULDPhgEmfORHep00L8JIv4" + "uUArKQ9miI3rfrD2NAY5SbqUqtZA9Q5FtfyouTkKWaEvOt2jAYLTX1m+EYtk4GYE" + "X36VrigU7NhuVQlQM+6M2c5BbQKBgQDr/Iyaza0MdD3oL2EBELVVVEukBs/6Dghp" + "NS6PiRny+KznlPYJgowIqN+JXgxxyq+yaPSWBFF+5gJv39+YofHgUkDBlVQVWtRq" + "g6QSWdBX24I1RsPhCW6yioSbbmTNAPCP5iiQ6a50slMoKs+FU+t7YCmtWH4LxmhU" + "ZVtqSxMKSQKBgQDFFAxmooL5cM8RkrlvBr9E+dizSePrIznAF9oLnMaK2B5vKFP2" + "EWXbe86WFgZJb1au9dd+YLtDDcOQeqDAjfP00up7UqHHG1Scg/wvgbHUF78eXtht" + "pzwewb/i9h4IG2M+fsrqJA1ht11PHqq6pMQ0ftM9TitkvhhZbTXJuKrfDQKBgQCQ" + "W0s/nzWOesUr9V+noSrr6cbjre8njdIDWa8DyuB+yzQp7eub4an72iH0C/4c1doo" + "jx+fbUDVGaq/+Pj0i4xEpsxKHyjybz+lcKW7PAnA1NyP7pwAWx7GRQndcuhip4KA" + "5FBIZUp8xGEuHxIXA0D3ldmo2oswdMx5ZA9JtGcPUQKBgQDtlyIeU8p9wTgCLYn8" + "PVrVMJNiLv4zvJdmQNkBQmNIC67mLUGcZRZRfskk4hYB8sglAGkHZIUBDMFk785+" + "grB5Y0Rn7RAuzVs6uoibwAx4u3rEjENR0gOdyzMATzqjgWL8GnrG5QU6VHRIVwr6" + "ni/+cSjJYn6PolxPdurAZXYhqQ==" + ) + pubkey = get_rsa_pubkey_from_b64keystring(pubkey_str) + prikey = get_rsa_prikey_from_b64keystring(prikey_str) + key_crypt = pubkey_encode(pubkey, key_b64) + key_b64T = prikey_decode(prikey, key_crypt) + print(f"RSA加密{key_b64=}==>{key_crypt=}, \n解密==>{key_b64T=}") + x = data_jj_encode(d, pubkey_str) + d1 = data_jj_decode(x, prikey_str) + print(f'{d=}, {d1=}') diff --git a/weishijiajie/param_entity.py b/weishijiajie/param_entity.py deleted file mode 100644 index 553bfc0..0000000 --- a/weishijiajie/param_entity.py +++ /dev/null @@ -1,34 +0,0 @@ -from typing import Dict, Optional - - -class ParamEntity: - def __init__(self, method: str, status: str, format: str, query_map: Optional[Dict[str, str]] = None): - self.method = method - self.status = status - self.format = format - self.query_map = query_map or {} - - def get_method(self) -> str: - return self.method - - def set_method(self, method: str): - self.method = method - - def get_status(self) -> str: - return self.status - - def set_status(self, status: str): - self.status = status - - def get_format(self) -> str: - return self.format - - def set_format(self, format: str): - self.format = format - - def get_query_map(self) -> Dict[str, str]: - return self.query_map - - def set_query_map(self, query_map: Dict[str, str]): - self.query_map = query_map - diff --git a/weishijiajie/rsa_util.py b/weishijiajie/rsa_util.py deleted file mode 100644 index 06df2fd..0000000 --- a/weishijiajie/rsa_util.py +++ /dev/null @@ -1,63 +0,0 @@ -from weishijiajie.base64_custom import Base64 -import urllib.parse -from Crypto.PublicKey import RSA -from Crypto.Cipher import PKCS1_v1_5 -from Crypto.Signature import pkcs1_15 -from Crypto.Hash import MD5 -from Crypto import Random -import collections -from typing import Dict -from base64_custom import Base64Utils - -class RSAUtil: - @staticmethod - def sign(public_map: Dict[str, object], private_key: str) -> str: - keys = sorted(k for k in public_map if k.lower() != "sign") - values = [f"{k}={public_map[k]}" for k in keys] - public_req_str = "&".join(values) - - key_bytes = Base64Utils.decode(private_key) - private_key_obj = RSA.import_key(key_bytes) - signer = pkcs1_15.new(private_key_obj) - digest = MD5.new(public_req_str.encode()) - signature = signer.sign(digest) - return Base64Utils.encode(signature) - - @staticmethod - def verify(public_map: Dict[str, object], public_key: str) -> bool: - keys = sorted(k for k in public_map if k.lower() != "sign") - values = [f"{k}={public_map[k]}" for k in keys] - public_req_str = "&".join(values) - sign = urllib.parse.unquote(public_map["sign"]) - - key_bytes = Base64Utils.decode(public_key) - public_key_obj = RSA.import_key(key_bytes) - verifier = pkcs1_15.new(public_key_obj) - digest = MD5.new(public_req_str.encode()) - - try: - verifier.verify(digest, Base64Utils.decode(sign)) - return True - except (ValueError, TypeError): - return False - - @staticmethod - def decrypt_by_public_key(encrypted: str, public_key: str) -> str: - encrypted_data = Base64Utils.decode(encrypted) - key_bytes = Base64Utils.decode(public_key) - public_key_obj = RSA.import_key(key_bytes) - cipher = PKCS1_v1_5.new(public_key_obj) - - sentinel = Random.new().read(15) - decrypted_data = cipher.decrypt(encrypted_data, sentinel) - return decrypted_data.decode() - - @staticmethod - def encrypt_by_private_key(source: str, private_key: str) -> str: - data = source.encode() - key_bytes = Base64Utils.decode(private_key) - private_key_obj = RSA.import_key(key_bytes) - cipher = PKCS1_v1_5.new(private_key_obj) - encrypted_data = cipher.encrypt(data) - return Base64Utils.encode(encrypted_data) - diff --git a/weishijiajie/sign_utils_mop.py b/weishijiajie/sign_utils_mop.py deleted file mode 100644 index 3ee92ea..0000000 --- a/weishijiajie/sign_utils_mop.py +++ /dev/null @@ -1,52 +0,0 @@ -import uuid -from urllib.parse import urlencode, quote, urlparse, parse_qs -from weishijiajie.rsa_util import RSAUtil # 假设你之前已转换好的 -from weishijiajie.param_entity import ParamEntity # 假设你之前已转换好的 - - -class SignUtilsMOP: - @staticmethod - def do_sign(servlet_path: str, private_key: str, param_entity: ParamEntity) -> str: - if not servlet_path.strip(): - raise Exception("url path is empty!") - if not private_key.strip(): - raise Exception("privateKey is empty!") - - public_map = { - "method": param_entity.method, - "status": param_entity.status, - "format": param_entity.format, - "flowdId": str(uuid.uuid4()).replace("-", "") - } - - if param_entity.query_map: - public_map.update(param_entity.query_map) - - try: - sign = RSAUtil.sign(public_map, private_key) - public_map["sign"] = quote(sign, encoding="utf-8") - - query_string = "&".join(f"{k}={v}" for k, v in public_map.items()) - return f"{servlet_path}?{query_string}" - except Exception as e: - print("sign failed!") - raise e - - @staticmethod - def verify(sign_url: str, public_key: str) -> bool: - if "?" not in sign_url: - return False - - param_str = sign_url.split("?", 1)[1] - param_map = {} - - for pair in param_str.split("&"): - if pair.strip(): - k, v = pair.split("=", 1) - param_map[k] = v - - try: - return RSAUtil.verify(param_map, public_key) - except Exception: - return False -