refactor: Fix some type error (#22594)
Signed-off-by: -LAN- <laipz8200@outlook.com>
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
import hashlib
|
||||
from typing import Union
|
||||
|
||||
from Crypto.Cipher import AES
|
||||
from Crypto.PublicKey import RSA
|
||||
@@ -9,7 +10,7 @@ from extensions.ext_storage import storage
|
||||
from libs import gmpy2_pkcs10aep_cipher
|
||||
|
||||
|
||||
def generate_key_pair(tenant_id):
|
||||
def generate_key_pair(tenant_id: str) -> str:
|
||||
private_key = RSA.generate(2048)
|
||||
public_key = private_key.publickey()
|
||||
|
||||
@@ -26,7 +27,7 @@ def generate_key_pair(tenant_id):
|
||||
prefix_hybrid = b"HYBRID:"
|
||||
|
||||
|
||||
def encrypt(text, public_key):
|
||||
def encrypt(text: str, public_key: Union[str, bytes]) -> bytes:
|
||||
if isinstance(public_key, str):
|
||||
public_key = public_key.encode()
|
||||
|
||||
@@ -38,14 +39,14 @@ def encrypt(text, public_key):
|
||||
rsa_key = RSA.import_key(public_key)
|
||||
cipher_rsa = gmpy2_pkcs10aep_cipher.new(rsa_key)
|
||||
|
||||
enc_aes_key = cipher_rsa.encrypt(aes_key)
|
||||
enc_aes_key: bytes = cipher_rsa.encrypt(aes_key)
|
||||
|
||||
encrypted_data = enc_aes_key + cipher_aes.nonce + tag + ciphertext
|
||||
|
||||
return prefix_hybrid + encrypted_data
|
||||
|
||||
|
||||
def get_decrypt_decoding(tenant_id):
|
||||
def get_decrypt_decoding(tenant_id: str) -> tuple[RSA.RsaKey, object]:
|
||||
filepath = "privkeys/{tenant_id}".format(tenant_id=tenant_id) + "/private.pem"
|
||||
|
||||
cache_key = "tenant_privkey:{hash}".format(hash=hashlib.sha3_256(filepath.encode()).hexdigest())
|
||||
@@ -64,7 +65,7 @@ def get_decrypt_decoding(tenant_id):
|
||||
return rsa_key, cipher_rsa
|
||||
|
||||
|
||||
def decrypt_token_with_decoding(encrypted_text, rsa_key, cipher_rsa):
|
||||
def decrypt_token_with_decoding(encrypted_text: bytes, rsa_key: RSA.RsaKey, cipher_rsa) -> str:
|
||||
if encrypted_text.startswith(prefix_hybrid):
|
||||
encrypted_text = encrypted_text[len(prefix_hybrid) :]
|
||||
|
||||
@@ -83,10 +84,10 @@ def decrypt_token_with_decoding(encrypted_text, rsa_key, cipher_rsa):
|
||||
return decrypted_text.decode()
|
||||
|
||||
|
||||
def decrypt(encrypted_text, tenant_id):
|
||||
def decrypt(encrypted_text: bytes, tenant_id: str) -> str:
|
||||
rsa_key, cipher_rsa = get_decrypt_decoding(tenant_id)
|
||||
|
||||
return decrypt_token_with_decoding(encrypted_text, rsa_key, cipher_rsa)
|
||||
return decrypt_token_with_decoding(encrypted_text=encrypted_text, rsa_key=rsa_key, cipher_rsa=cipher_rsa)
|
||||
|
||||
|
||||
class PrivkeyNotFoundError(Exception):
|
||||
|
Reference in New Issue
Block a user