commit b2495e1b580b045ab964c9ecd79f35cac4c0f1ec Author: yumoqing Date: Thu Jun 5 18:08:09 2025 +0800 first commit diff --git a/README.md b/README.md new file mode 100644 index 0000000..cc547b5 --- /dev/null +++ b/README.md @@ -0,0 +1,30 @@ +# 伟士佳杰API工具 + +## 生成签字 + +GET方法 +``` +from weishijiajie.params_entity import ParamEntity +from weishijiajie.sign_utils_mop import SignUtilsMOP + +url = "https://xxxx/api/blade-vstec/kyyOrder/list" +params = { + 'current': "1", + 'size': '10' +} +params_entity = ParamEntity("Sync_KYY_USER_INFO", "1", "form", cwparamsdata) +signUrl = SignUtilsMOP.doSign(url, PublicRsaKey.privateKey, paramEntity) +``` + +POST方法 +``` +from weishijiajie.params_entity import ParamEntity +from weishijiajie.sign_utils_mop import SignUtilsMOP + +"https://xxxx/api/blade-user/syncinfo/user/v1" +params_entity = ParamEntity("Sync_KYY_USER_INFO", "1", "form", None) +signUrl = SignUtilsMOP.doSign(url, PublicRsaKey.privateKey, paramEntity) + +``` + +## diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..bc6d690 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,19 @@ +[project] +name="weishijiajie" +version = "5.2.2" +description = "weishijiajie API utils" +authors = [{ name = "yu moqing", email = "yumoqing@gmail.com" }] +readme = "README.md" +requires-python = ">=3.8" +license = {text = "MIT"} +dependencies = [ + "Crypto" +] + +[project.optional-dependencies] +dev = ["pytest", "black", "mypy"] + +[build-system] +requires = ["setuptools>=61", "wheel"] +build-backend = "setuptools.build_meta" + diff --git a/test/test_base64.py b/test/test_base64.py new file mode 100644 index 0000000..a8db759 --- /dev/null +++ b/test/test_base64.py @@ -0,0 +1,8 @@ +from base64_custom import Base64 +base64 = Base64() +encoded = base64.encode(b'hello world') +print(f'Encoded: {encoded}') # Output: aGVsbG8gd29ybGQ= + +decoded = base64.decode(encoded) +print(f'Decoded: {decoded.decode()}') # Output: hello world + diff --git a/weishijiajie/__init__.py b/weishijiajie/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/weishijiajie/base64_custom.py b/weishijiajie/base64_custom.py new file mode 100644 index 0000000..153e418 --- /dev/null +++ b/weishijiajie/base64_custom.py @@ -0,0 +1,155 @@ +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/param_entity.py b/weishijiajie/param_entity.py new file mode 100644 index 0000000..553bfc0 --- /dev/null +++ b/weishijiajie/param_entity.py @@ -0,0 +1,34 @@ +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 new file mode 100644 index 0000000..06df2fd --- /dev/null +++ b/weishijiajie/rsa_util.py @@ -0,0 +1,63 @@ +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 new file mode 100644 index 0000000..3ee92ea --- /dev/null +++ b/weishijiajie/sign_utils_mop.py @@ -0,0 +1,52 @@ +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 +