first commit
This commit is contained in:
commit
b2495e1b58
30
README.md
Normal file
30
README.md
Normal file
@ -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)
|
||||||
|
|
||||||
|
```
|
||||||
|
|
||||||
|
##
|
19
pyproject.toml
Normal file
19
pyproject.toml
Normal file
@ -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"
|
||||||
|
|
8
test/test_base64.py
Normal file
8
test/test_base64.py
Normal file
@ -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
|
||||||
|
|
0
weishijiajie/__init__.py
Normal file
0
weishijiajie/__init__.py
Normal file
155
weishijiajie/base64_custom.py
Normal file
155
weishijiajie/base64_custom.py
Normal file
@ -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"))
|
||||||
|
|
34
weishijiajie/param_entity.py
Normal file
34
weishijiajie/param_entity.py
Normal file
@ -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
|
||||||
|
|
63
weishijiajie/rsa_util.py
Normal file
63
weishijiajie/rsa_util.py
Normal file
@ -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)
|
||||||
|
|
52
weishijiajie/sign_utils_mop.py
Normal file
52
weishijiajie/sign_utils_mop.py
Normal file
@ -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
|
||||||
|
|
Loading…
Reference in New Issue
Block a user