|
|
import os
|
|
|
import pickle
|
|
|
import socket
|
|
|
import tkinter as tk
|
|
|
from tkinter import filedialog, ttk, messagebox
|
|
|
from cryptography.hazmat.primitives.asymmetric import rsa, padding
|
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
|
import hashlib
|
|
|
|
|
|
|
|
|
# 生成并保存发送方密钥对
|
|
|
def generate_and_save_sender_keys():
|
|
|
private_key = rsa.generate_private_key(
|
|
|
public_exponent=65537,
|
|
|
key_size=2048,
|
|
|
)
|
|
|
public_key = private_key.public_key()
|
|
|
|
|
|
with open('sender_private_key.pem', 'wb') as f:
|
|
|
f.write(private_key.private_bytes(
|
|
|
encoding=serialization.Encoding.PEM,
|
|
|
format=serialization.PrivateFormat.PKCS8,
|
|
|
encryption_algorithm=serialization.NoEncryption()
|
|
|
))
|
|
|
|
|
|
with open('sender_public_key.pem', 'wb') as f:
|
|
|
f.write(public_key.public_bytes(
|
|
|
encoding=serialization.Encoding.PEM,
|
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
|
|
))
|
|
|
|
|
|
return private_key, public_key
|
|
|
|
|
|
|
|
|
# 加载发送方密钥对
|
|
|
def load_sender_keys():
|
|
|
try:
|
|
|
with open('sender_private_key.pem', 'rb') as f:
|
|
|
private_key = serialization.load_pem_private_key(
|
|
|
f.read(),
|
|
|
password=None
|
|
|
)
|
|
|
with open('sender_public_key.pem', 'rb') as f:
|
|
|
public_key = serialization.load_pem_public_key(f.read())
|
|
|
return private_key, public_key
|
|
|
except FileNotFoundError:
|
|
|
return generate_and_save_sender_keys()
|
|
|
|
|
|
|
|
|
sender_private_key, sender_public_key = load_sender_keys()
|
|
|
|
|
|
# 用于存储接收方公钥的变量
|
|
|
receiver_public_key = None
|
|
|
|
|
|
|
|
|
# 填充数据函数
|
|
|
def pad(data, block_size):
|
|
|
padding_length = block_size - (len(data) % block_size)
|
|
|
if padding_length == 0:
|
|
|
padding_length = block_size
|
|
|
padding = bytes([padding_length]) * padding_length
|
|
|
return data + padding
|
|
|
|
|
|
|
|
|
# 加密文件函数
|
|
|
def encrypt_file(file_path, key, algorithm, mode):
|
|
|
with open(file_path, 'rb') as f:
|
|
|
file_data = f.read()
|
|
|
if algorithm == "AES":
|
|
|
block_size = 16
|
|
|
elif algorithm == "DES":
|
|
|
block_size = 8
|
|
|
else:
|
|
|
raise ValueError("Unsupported algorithm")
|
|
|
file_data = pad(file_data, block_size)
|
|
|
|
|
|
mode_mapping = {
|
|
|
"CBC": modes.CBC,
|
|
|
"CFB": modes.CFB
|
|
|
}
|
|
|
if algorithm == "AES":
|
|
|
iv = os.urandom(16)
|
|
|
elif algorithm == "DES":
|
|
|
iv = os.urandom(8) # 生成8字节的IV用于DES算法
|
|
|
else:
|
|
|
raise ValueError("Unsupported algorithm")
|
|
|
|
|
|
if algorithm == "AES":
|
|
|
cipher = Cipher(algorithms.AES(key), mode_mapping[mode](iv))
|
|
|
elif algorithm == "DES":
|
|
|
cipher = Cipher(algorithms.TripleDES(key), mode_mapping[mode](iv))
|
|
|
else:
|
|
|
raise ValueError("Unsupported algorithm")
|
|
|
|
|
|
encryptor = cipher.encryptor()
|
|
|
encrypted_data = encryptor.update(file_data) + encryptor.finalize()
|
|
|
return encrypted_data, iv
|
|
|
|
|
|
|
|
|
# 计算哈希值函数
|
|
|
def calculate_hash(data):
|
|
|
digest = hashes.Hash(hashes.SHA256())
|
|
|
digest.update(data)
|
|
|
return digest.finalize()
|
|
|
|
|
|
|
|
|
# 对哈希值进行签名函数
|
|
|
def sign_hash(private_key, hash_value):
|
|
|
signature = private_key.sign(
|
|
|
hash_value,
|
|
|
padding.PSS(
|
|
|
mgf=padding.MGF1(hashes.SHA256()),
|
|
|
salt_length=padding.PSS.MAX_LENGTH
|
|
|
),
|
|
|
hashes.SHA256()
|
|
|
)
|
|
|
return signature
|
|
|
|
|
|
|
|
|
# 加密对称密钥函数
|
|
|
def encrypt_symmetric_key(public_key, symmetric_key):
|
|
|
encrypted_key = public_key.encrypt(
|
|
|
symmetric_key,
|
|
|
padding.OAEP(
|
|
|
mgf=padding.MGF1(algorithm=hashes.SHA256()),
|
|
|
algorithm=hashes.SHA256(),
|
|
|
label=None
|
|
|
)
|
|
|
)
|
|
|
return encrypted_key
|
|
|
|
|
|
|
|
|
# 计算文件的 MD5 校验和
|
|
|
def calculate_md5(data):
|
|
|
md5 = hashlib.md5()
|
|
|
md5.update(data)
|
|
|
return md5.hexdigest()
|
|
|
|
|
|
|
|
|
# 发送文件函数
|
|
|
def send_file(server_ip, server_port, package):
|
|
|
try:
|
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
|
s.connect((server_ip, server_port))
|
|
|
s.sendall(pickle.dumps(package))
|
|
|
return True
|
|
|
except Exception as e:
|
|
|
print(f"连接失败: {e}")
|
|
|
return False
|
|
|
|
|
|
|
|
|
# 导入接收方公钥的函数
|
|
|
def import_receiver_public_key():
|
|
|
global receiver_public_key
|
|
|
file_path = filedialog.askopenfilename(filetypes=[("PEM files", "*.pem")])
|
|
|
if file_path:
|
|
|
try:
|
|
|
with open(file_path, 'rb') as f:
|
|
|
receiver_public_key = serialization.load_pem_public_key(f.read())
|
|
|
messagebox.showinfo("提示", "接收方公钥导入成功")
|
|
|
except Exception as e:
|
|
|
messagebox.showerror("错误", f"导入接收方公钥失败: {e}")
|
|
|
|
|
|
|
|
|
# 主函数，用于启动 GUI 界面
|
|
|
def sender_main():
|
|
|
def select_file():
|
|
|
file_path.set(filedialog.askopenfilename())
|
|
|
|
|
|
def send():
|
|
|
global receiver_public_key
|
|
|
file = file_path.get()
|
|
|
algorithm = algorithm_var.get()
|
|
|
mode = mode_var.get()
|
|
|
|
|
|
if not file:
|
|
|
messagebox.showerror("错误", "请选择文件")
|
|
|
return
|
|
|
|
|
|
if algorithm == "AES":
|
|
|
key_size = 32
|
|
|
elif algorithm == "DES":
|
|
|
key_size = 24
|
|
|
else:
|
|
|
print("不支持的算法！")
|
|
|
return
|
|
|
|
|
|
if mode == "CBC":
|
|
|
pass
|
|
|
elif mode == "CFB":
|
|
|
pass
|
|
|
else:
|
|
|
print("不支持的模式！")
|
|
|
return
|
|
|
|
|
|
if not receiver_public_key:
|
|
|
messagebox.showerror("错误", "请导入证书")
|
|
|
return
|
|
|
|
|
|
symmetric_key = os.urandom(key_size)
|
|
|
encrypted_file, iv = encrypt_file(file, symmetric_key, algorithm, mode)
|
|
|
|
|
|
file_hash = calculate_hash(encrypted_file)
|
|
|
signature = sign_hash(sender_private_key, file_hash)
|
|
|
|
|
|
sender_public_pem = sender_public_key.public_bytes(
|
|
|
encoding=serialization.Encoding.PEM,
|
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
|
|
)
|
|
|
|
|
|
encrypted_symmetric_key = encrypt_symmetric_key(receiver_public_key, symmetric_key)
|
|
|
|
|
|
md5_checksum = calculate_md5(encrypted_file)
|
|
|
|
|
|
file_extension = os.path.splitext(file)[1][1:]
|
|
|
# 在封装包中添加发送方公钥
|
|
|
package = {
|
|
|
'encrypted_file': encrypted_file,
|
|
|
'iv': iv,
|
|
|
'encrypted_symmetric_key': encrypted_symmetric_key,
|
|
|
'signature': signature,
|
|
|
'algorithm': algorithm,
|
|
|
'mode': mode,
|
|
|
'md5_checksum': md5_checksum,
|
|
|
'file_extension': file_extension,
|
|
|
'sender_public_key': sender_public_pem
|
|
|
}
|
|
|
|
|
|
if send_file(server_ip.get(), int(server_port.get()), package):
|
|
|
messagebox.showinfo("提示", "发送成功")
|
|
|
else:
|
|
|
messagebox.showerror("错误", "连接失败")
|
|
|
|
|
|
|
|
|
root = tk.Tk()
|
|
|
root.title("发送方")
|
|
|
root.geometry("490x250")
|
|
|
|
|
|
file_path = tk.StringVar()
|
|
|
algorithm_var = tk.StringVar(value="AES")
|
|
|
mode_var = tk.StringVar(value="CBC")
|
|
|
server_ip = tk.StringVar(value="127.0.0.1")
|
|
|
server_port = tk.StringVar(value="12345")
|
|
|
|
|
|
ttk.Label(root, text="选择文件：").grid(row=0, column=0, sticky=tk.W, pady=5)
|
|
|
ttk.Entry(root, textvariable=file_path, width=30).grid(row=0, column=1, pady=5)
|
|
|
ttk.Button(root, text="浏览", command=select_file).grid(row=0, column=2, pady=5)
|
|
|
ttk.Button(root, text="导入证书", command=import_receiver_public_key).grid(row=0, column=3, pady=5)
|
|
|
|
|
|
ttk.Label(root, text="选择加密算法：").grid(row=1, column=0, sticky=tk.W, pady=5)
|
|
|
ttk.Combobox(root, textvariable=algorithm_var, values=["AES", "DES"], state="readonly").grid(row=1, column=1, pady=5)
|
|
|
|
|
|
ttk.Label(root, text="选择加密模式：").grid(row=2, column=0, sticky=tk.W, pady=5)
|
|
|
ttk.Combobox(root, textvariable=mode_var, values=["CBC", "CFB"], state="readonly").grid(row=2, column=1, pady=5)
|
|
|
|
|
|
ttk.Label(root, text="接收方 IP：").grid(row=3, column=0, sticky=tk.W, pady=5)
|
|
|
ttk.Entry(root, textvariable=server_ip).grid(row=3, column=1, pady=5)
|
|
|
|
|
|
ttk.Label(root, text="接收方端口：").grid(row=4, column=0, sticky=tk.W, pady=5)
|
|
|
ttk.Entry(root, textvariable=server_port).grid(row=4, column=1, pady=5)
|
|
|
|
|
|
ttk.Button(root, text="发送文件", command=send).grid(row=5, columnspan=3, pady=20)
|
|
|
|
|
|
root.mainloop()
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
sender_main() |
