|
|
|
|
@ -0,0 +1,147 @@
|
|
|
|
|
from cryptography.hazmat.backends import default_backend
|
|
|
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
|
|
|
from cryptography.hazmat.primitives import serialization
|
|
|
|
|
from cryptography.hazmat.primitives.asymmetric import padding
|
|
|
|
|
from cryptography.hazmat.primitives import hashes
|
|
|
|
|
import os
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 生成RSA密钥对
|
|
|
|
|
def generate_rsa_key_pair():
|
|
|
|
|
private_key = rsa.generate_private_key(
|
|
|
|
|
public_exponent=65537,
|
|
|
|
|
key_size=2048,
|
|
|
|
|
backend=default_backend()
|
|
|
|
|
)
|
|
|
|
|
public_key = private_key.public_key()
|
|
|
|
|
return private_key, public_key
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 序列化公钥
|
|
|
|
|
def serialize_public_key(public_key):
|
|
|
|
|
return public_key.public_bytes(
|
|
|
|
|
encoding=serialization.Encoding.PEM,
|
|
|
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 序列化私钥
|
|
|
|
|
def serialize_private_key(private_key):
|
|
|
|
|
return private_key.private_bytes(
|
|
|
|
|
encoding=serialization.Encoding.PEM,
|
|
|
|
|
format=serialization.PrivateFormat.PKCS8,
|
|
|
|
|
encryption_algorithm=serialization.NoEncryption()
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 使用RSA公钥加密文件
|
|
|
|
|
def rsa_encrypt_file(public_key, file_path):
|
|
|
|
|
try:
|
|
|
|
|
encrypted_file_path = file_path + '.encrypted'
|
|
|
|
|
with open(file_path, 'rb') as file_to_encrypt, open(encrypted_file_path, 'wb') as encrypted_file:
|
|
|
|
|
while True:
|
|
|
|
|
data = file_to_encrypt.read(245) # 2048 - 245 = 1803 bytes for padding
|
|
|
|
|
if not data:
|
|
|
|
|
break
|
|
|
|
|
# 明确指定填充参数并检查
|
|
|
|
|
padding_params = {
|
|
|
|
|
"mgf": padding.MGF1(algorithm=hashes.SHA256()),
|
|
|
|
|
"algorithm": hashes.SHA256(),
|
|
|
|
|
"label": None
|
|
|
|
|
}
|
|
|
|
|
encrypted_data = public_key.encrypt(
|
|
|
|
|
data,
|
|
|
|
|
padding.OAEP(**padding_params)
|
|
|
|
|
)
|
|
|
|
|
encrypted_file.write(encrypted_data)
|
|
|
|
|
return encrypted_file_path
|
|
|
|
|
except ValueError as e:
|
|
|
|
|
raise ValueError("加密文件时出现错误，可能是读取文件数据或使用公钥加密的过程出现问题，请检查数据及公钥。")
|
|
|
|
|
except Exception as ex:
|
|
|
|
|
raise Exception("加密文件时发生未知错误，可能是文件系统相关问题等，请检查相关配置及环境。")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 使用RSA私钥解密文件
|
|
|
|
|
def rsa_decrypt_file(private_key, encrypted_file_path):
|
|
|
|
|
try:
|
|
|
|
|
decrypted_file_path = os.path.splitext(encrypted_file_path)[0]
|
|
|
|
|
# 创建临时文件用于存储解密过程中的数据
|
|
|
|
|
temp_file_path = decrypted_file_path + '.tmp'
|
|
|
|
|
with open(encrypted_file_path, 'rb') as encrypted_file, open(temp_file_path, 'wb') as decrypted_file:
|
|
|
|
|
encrypted_data = encrypted_file.read(256)
|
|
|
|
|
while encrypted_data:
|
|
|
|
|
try:
|
|
|
|
|
decrypted_data = private_key.decrypt(
|
|
|
|
|
encrypted_data,
|
|
|
|
|
padding.OAEP(
|
|
|
|
|
mgf=padding.MGF1(algorithm=hashes.SHA256()),
|
|
|
|
|
algorithm=hashes.SHA256(),
|
|
|
|
|
label=None
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
decrypted_file.write(decrypted_data)
|
|
|
|
|
except ValueError as inner_e:
|
|
|
|
|
# 更细致地处理解密过程中的ValueError，确保临时文件被正确清理
|
|
|
|
|
if os.path.exists(temp_file_path):
|
|
|
|
|
os.remove(temp_file_path)
|
|
|
|
|
raise ValueError("解密文件时出现错误，可能是读取加密文件数据、使用私钥解密的过程出现问题，请检查数据、私钥及加密文件完整性。") from inner_e
|
|
|
|
|
encrypted_data = encrypted_file.read(256)
|
|
|
|
|
|
|
|
|
|
# 确保文件没有意外关闭，处理最后一块数据（如果最后一块数据小于256字节）
|
|
|
|
|
if encrypted_file.tell() < os.path.getsize(encrypted_file_path):
|
|
|
|
|
remaining_data = encrypted_file.read()
|
|
|
|
|
if remaining_data:
|
|
|
|
|
decrypted_data = private_key.decrypt(
|
|
|
|
|
remaining_data,
|
|
|
|
|
padding.OAEP(
|
|
|
|
|
mgf=padding.MGF1(algorithm=hashes.SHA256()),
|
|
|
|
|
algorithm=hashes.SHA256(),
|
|
|
|
|
label=None
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
decrypted_file.write(decrypted_data)
|
|
|
|
|
|
|
|
|
|
# 判断目标解密文件是否存在，若存在则删除
|
|
|
|
|
if os.path.exists(decrypted_file_path):
|
|
|
|
|
os.remove(decrypted_file_path)
|
|
|
|
|
|
|
|
|
|
# 解密成功，将临时文件重命名为最终的解密文件
|
|
|
|
|
os.rename(temp_file_path, decrypted_file_path)
|
|
|
|
|
return decrypted_file_path
|
|
|
|
|
except ValueError as e:
|
|
|
|
|
# 如果出现错误，删除临时文件（避免残留无用文件）
|
|
|
|
|
temp_file_path = os.path.splitext(encrypted_file_path)[0] + '.tmp'
|
|
|
|
|
if os.path.exists(temp_file_path):
|
|
|
|
|
os.remove(temp_file_path)
|
|
|
|
|
raise ValueError("解密文件时出现错误，可能是读取加密文件数据、使用私钥解密的过程出现问题，请检查数据、私钥及加密文件完整性。")
|
|
|
|
|
except Exception as ex:
|
|
|
|
|
# 如果出现其他未知错误，同样删除临时文件
|
|
|
|
|
temp_file_path = os.path.splitext(encrypted_file_path)[0] + '.tmp'
|
|
|
|
|
if os.path.exists(temp_file_path):
|
|
|
|
|
os.remove(temp_file_path)
|
|
|
|
|
raise Exception("解密文件时发生错误，可能是私钥不正确，文件不存在、损坏或文件系统权限等问题，请检查相关情况。")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
#测试部分
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
|
# 生成密钥对
|
|
|
|
|
private_key, public_key = generate_rsa_key_pair()
|
|
|
|
|
public_key_bytes = serialize_public_key(public_key)
|
|
|
|
|
private_key_bytes = serialize_private_key(private_key)
|
|
|
|
|
|
|
|
|
|
# 待加密的文件路径，这里假设当前目录下有一个名为test.txt的文件，你可以替换成实际文件路径
|
|
|
|
|
file_path = "D:\\作业\\zzz.txt"
|
|
|
|
|
try:
|
|
|
|
|
# 加密文件
|
|
|
|
|
encrypted_file_path = rsa_encrypt_file(public_key, file_path)
|
|
|
|
|
print(f"文件 {file_path} 加密成功，加密后的文件路径为: {encrypted_file_path}")
|
|
|
|
|
|
|
|
|
|
# 解密文件
|
|
|
|
|
decrypted_file_path = rsa_decrypt_file(private_key, encrypted_file_path)
|
|
|
|
|
print(f"文件 {encrypted_file_path} 解密成功，解密后的文件路径为: {decrypted_file_path}")
|
|
|
|
|
except ValueError as ve:
|
|
|
|
|
print(f"出现值错误: {ve}")
|
|
|
|
|
except Exception as e:
|
|
|
|
|
print(f"出现其他错误: {e}")
|
pjbpwnxmf