buffix
This commit is contained in:
parent
b2495e1b58
commit
904e2b2bdf
@ -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"))
|
|
||||||
|
|
163
weishijiajie/jj_utils.py
Normal file
163
weishijiajie/jj_utils.py
Normal file
@ -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=}')
|
@ -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
|
|
||||||
|
|
@ -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)
|
|
||||||
|
|
@ -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
|
|
||||||
|
|
Loading…
Reference in New Issue
Block a user