2022-10-24 10:39:19 +08:00
|
|
|
import os
|
|
|
|
import base64
|
|
|
|
|
|
|
|
from Crypto.PublicKey import RSA
|
|
|
|
from Crypto.Cipher import PKCS1_v1_5 as PKCS1_cipher
|
2022-11-01 01:56:57 +08:00
|
|
|
from fastapi import HTTPException
|
2022-10-24 10:39:19 +08:00
|
|
|
|
|
|
|
|
|
|
|
# 获取密钥
|
|
|
|
def get_key(**kwargs):
|
|
|
|
# 密钥文件
|
|
|
|
key_file = kwargs['key_file']
|
|
|
|
# 密钥文件绝对路径
|
2022-11-07 13:53:42 +08:00
|
|
|
rsa_key_path = os.getcwd().split("User")[0] + "User" + "\\Utils\\EncryptProcess\\" + key_file
|
2022-10-24 10:39:19 +08:00
|
|
|
with open(rsa_key_path) as f:
|
|
|
|
data = f.read()
|
|
|
|
key = RSA.importKey(data)
|
|
|
|
return key
|
|
|
|
|
|
|
|
|
|
|
|
# 用公钥加密
|
|
|
|
def encrypt_data(**kwargs):
|
|
|
|
# 待加密信息
|
|
|
|
msg = kwargs['msg']
|
|
|
|
# 获取密钥
|
|
|
|
public_key = get_key(key_file='rsa_public_key.pem')
|
|
|
|
cipher = PKCS1_cipher.new(public_key)
|
|
|
|
encrypt_text = base64.b64encode(cipher.encrypt(bytes(msg.encode("utf8"))))
|
|
|
|
return encrypt_text.decode('utf-8')
|
|
|
|
|
|
|
|
|
|
|
|
# 用私钥解密
|
|
|
|
def decrypt_data(**kwargs):
|
2022-11-01 01:56:57 +08:00
|
|
|
|
2022-10-24 10:39:19 +08:00
|
|
|
# 待加密信息
|
|
|
|
encrypt_msg = kwargs['encrypt_msg']
|
|
|
|
# 获取密钥
|
|
|
|
private_key = get_key(key_file='rsa_private_key.pem')
|
|
|
|
cipher = PKCS1_cipher.new(private_key)
|
2022-11-01 01:56:57 +08:00
|
|
|
|
|
|
|
try:
|
|
|
|
back_text = cipher.decrypt(base64.b64decode(encrypt_msg), 0)
|
|
|
|
return back_text.decode('utf-8')
|
2022-11-07 13:53:42 +08:00
|
|
|
except Exception:
|
2022-11-01 01:56:57 +08:00
|
|
|
raise HTTPException(
|
|
|
|
status_code=400,
|
|
|
|
detail="Password Invalid"
|
|
|
|
)
|
2022-10-24 10:39:19 +08:00
|
|
|
|
|
|
|
|
|
|
|
# 检验
|
|
|
|
def check_encrypt_decrypt(**kwargs):
|
|
|
|
msg = kwargs['msg']
|
|
|
|
decrypt_msg = decrypt_data(encrypt_msg=kwargs['encrypt_msg'])
|
|
|
|
if msg == decrypt_msg:
|
|
|
|
return True
|
|
|
|
else:
|
|
|
|
return False
|