sangge/algo_present
2
0
Fork 0
Code Issues Pull Requests Activity

移除与算法不相关的文件

Browse Source
This commit is contained in:
sangge-win 2024-04-14 00:01:02 +08:00
parent e0e594592e
commit 2e0b552a50
7 changed files with 0 additions and 2108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

295
DataSearch.py
View File

@ -1,295 +0,0 @@
import re
import pymysql # pylint: disable=e0401 # type: ignore
# 拆分sql对象
def extract_tables(sql_statement):
# 使用正则表达式匹配FROM关键字之后的表名以及逗号分隔的表名：
pattern = r"FROM\s+([\w\s,]+)"
matches = re.findall(pattern, sql_statement, re.IGNORECASE)
# 提取逗号分隔的表名，并按逗号进行分割
tabs = re.split(",", matches[0].strip())
# 处理每个表名，去除空格和其他无关字符
cleaned_tables = []
for tab in tabs:
cleaned_tab = tab.strip()
if " " in cleaned_tab:
cleaned_tab = cleaned_tab.split()[0]
cleaned_tables.append(cleaned_tab)
return cleaned_tables # 返回列表
# 拆分sql
def divide_sql(sql):
"""
例如sql = "SELECT a FROM data1, data2, data3 WHERE a = b ORDER BY misc"
拆分原语句
"""
parts = re.split(r"(?i)from\s", sql) # 拆分"from "【无论大小写】
head = parts[0] + "from " # SELECT a FROM
divide_sqls = []
if re.search(r"where", parts[1], flags=re.IGNORECASE):
data = re.split(r"(?i) where", parts[1]) # 拆分" where"【无论大小写】
tail = " where" + data[1] # WHERE a = b ORDER BY misc
# message_p = "涉及到的数据源有:"
# print(message_p)
# time.sleep(sleep_time)
#
# message_p = data[0]
# print(message_p) # data1, data2, data3
# time.sleep(sleep_time)
divide_providers = data[0].split(", ")
# total = len(divide_providers)
# message_p = "拆分结果如下:"
# print(message_p)
for i in range(len(divide_providers)):
divide_sqls.append(head + divide_providers[i] + tail)
# message_p = str(i + 1) + ":" + divide_sqls[i]
# print(message_p)
else:
data = parts[1] # data1,data2,data3
divide_providers = data.split(",")
for i in range(len(divide_providers)):
divide_sqls.append(head + divide_providers[i])
# message_p = str(i + 1) + ":" + divide_sqls[i]
# print(message_p)
return divide_sqls
class SelectDatabase: # 定义类
def __init__(self, database):
self.database = database # 赋值数据库名称
def ret_hospital(self): # 定义函数，返回字典——医院（HOS）这个表
# 连接服务器Server的数据库:
db = pymysql.connect(
host="localhost",
user="root",
password="111111",
db=f"{self.database}",
charset="utf8",
)
# 使用操作游标:
cursor = db.cursor()
sql = """SELECT * FROM HOS"""
cursor.execute(sql)
results = cursor.fetchall() # 获取查询结果的所有数据
hospital_dict = {} # 创建一个空字典
# 遍历查询结果，将每条消息数据存储到字典中：
for row in results:
hos_id = row[0] # 医院编号(HOS_ID)
hos_name = row[1] # 医院名称(HOS_NAME)
hos_add = row[2] # 医院地址(HOS_ADD)
hos_tel = row[3] # 医院电话(HOS_TEL)
# hospital_dict["医院编号"] = hos_id
# hospital_dict["医院名称"] = hos_name
# hospital_dict["医院地址"] = hos_add
# hospital_dict["医院电话"] = hos_tel
# # 打印字典内容
# print(hospital_dict)
""" 注释的输出形式：
{'医院编号': '001', '医院名称': '极光医院', '医院地址': '千里市广大区极光街道1-1', '医院电话': '023-6296666'}
{'医院编号': '002', '医院名称': '风舱医院', '医院地址': '风火市舱山区飞光街道1-1', '医院电话': '023-6286666'}
"""
# 将每个属性的值存储在对应的列表中
hospital_dict.setdefault("医院编号", []).append(hos_id)
hospital_dict.setdefault("医院名称", []).append(hos_name)
hospital_dict.setdefault("医院地址", []).append(hos_add)
hospital_dict.setdefault("医院电话", []).append(hos_tel)
db.close()
""" 当前返回的字典形式：
{'医院编号': ['001', '002'],
'医院名称': ['极光医院', '风舱医院'],
'医院地址': ['千里市广大区极光街道1-1',
'风火市舱山区飞光街道1-1'],
'医院电话': ['023-6296666', '023-6286666']}
"""
# 返回字典
return hospital_dict
def ret_doctor(self): # 定义函数，返回字典——医生（DOC）这个表
# 连接服务器Server的数据库:
db = pymysql.connect(
host="localhost",
user="root",
password="111111",
db=f"{self.database}",
charset="utf8",
)
# 使用操作游标:
cursor = db.cursor()
sql = """SELECT * FROM DOC"""
cursor.execute(sql)
results = cursor.fetchall() # 获取查询结果的所有数据
doctor_dict = {} # 创建一个空字典
# 遍历查询结果，将每条消息数据存储到字典中：
for row in results:
doc_id = row[0] # 医生编号(DOC_ID)
doc_name = row[1] # 医生名称(DOC_NAME)
doc_tel = row[2] # 医生电话(DOC_TEL)
doc_qua = row[3] # 医院资质(DOC_QUA)
# hospital_dict["医生编号"] = doc_id
# hospital_dict["医生名称"] = doc_name
# hospital_dict["医院地址"] = doc_tel
# hospital_dict["医院电话"] = doc_qua
# # 打印字典内容
# print(doctor_dict)
# 将每个属性的值存储在对应的列表中
doctor_dict.setdefault("医院编号", []).append(doc_id)
doctor_dict.setdefault("医院名称", []).append(doc_name)
doctor_dict.setdefault("医院地址", []).append(doc_tel)
doctor_dict.setdefault("医院电话", []).append(doc_qua)
db.close()
""" 当前返回的字典形式：
{'医院编号': ['001', '002', '003'],
'医院名称': ['神医华佗', '扁鹊', '还医生'],
'医院地址': ['19666666666', '13666666666', '13546981623'],
'医院电话': ['主任医师', '主任医师', '医师']}
"""
# 返回字典
return doctor_dict
def ret_patient(self): # 定义函数，返回字典——病人（PAT）这个表
# 连接服务器Server的数据库:
db = pymysql.connect(
host="localhost",
user="root",
password="111111",
db=f"{self.database}",
charset="utf8",
)
# 使用操作游标:
cursor = db.cursor()
sql = """SELECT * FROM PAT"""
cursor.execute(sql)
results = cursor.fetchall() # 获取查询结果的所有数据
patient_dict = {} # 创建一个空字典
# 遍历查询结果，将每条消息数据存储到字典中：
for row in results:
pat_id = row[0] # 病人编号(PAT_ID)
pat_name = row[1] # 病人姓名(PAT_NAME)
pat_tel = row[2] # 病人电话(PAT_TEL)
# patient_dict["病人编号"] = pat_id
# patient_dict["病人姓名"] = pat_name
# patient_dict["病人电话"] = pat_tel
# # 打印字典内容
# print(patient_dict)
# 将每个属性的值存储在对应的列表中
patient_dict.setdefault("病人编号", []).append(pat_id)
patient_dict.setdefault("病人姓名", []).append(pat_name)
patient_dict.setdefault("病人电话", []).append(pat_tel)
db.close()
""" 当前返回的字典形式：
{'病人编号': ['001', '002', '003', '004'],
'病人姓名': ['曹操', '蔡桓公', '去还', '刘备'],
'病人电话': ['66666666666', '02666666666', '01234567891', '98765432101']}
"""
# 返回字典
return patient_dict
def ret_diagnosis(self): # 定义函数，返回字典——诊断（DIA）这个表
# 连接服务器Server的数据库:
db = pymysql.connect(
host="localhost",
user="root",
password="111111",
db=f"{self.database}",
charset="utf8",
)
# 使用操作游标:
cursor = db.cursor()
sql = """SELECT * FROM DIA"""
cursor.execute(sql)
results = cursor.fetchall() # 获取查询结果的所有数据
diagnosis_dict = {} # 创建一个空字典
# 遍历查询结果，将每条消息数据存储到字典中：
for row in results:
dia_id = row[0] # 诊号(DIA_ID)
doc_id = row[1] # 医生编号(DOC_ID)
pat_id = row[2] # 病人编号(PAT_ID)
time = row[3] # 时间(TIME)
cases = row[4] # 病例(CASES)
symptom = row[5] # 症状(SYMPTOM)
# diagnosis_dict["诊号"] = dia_id
# diagnosis_dict["医生编号"] = doc_id
# diagnosis_dict["病人编号"] = pat_id
# diagnosis_dict["时间"] = time
# diagnosis_dict["病例"] = cases
# diagnosis_dict["症状"] = symptom
# # 打印字典内容
# print(diagnosis_dict)
# 将每个属性的值存储在对应的列表中
diagnosis_dict.setdefault("诊号", []).append(dia_id)
diagnosis_dict.setdefault("医生编号", []).append(doc_id)
diagnosis_dict.setdefault("病人编号", []).append(pat_id)
diagnosis_dict.setdefault("时间", []).append(time)
diagnosis_dict.setdefault("病例", []).append(cases)
diagnosis_dict.setdefault("症状", []).append(symptom)
db.close()
""" 当前返回的字典形式：
{'诊号': ['001', '002', '003', '004', '005'],
'医生编号': ['001', '001', '002', '003', '003'],
'病人编号': ['001', '001', '002', '003', '004'],
'时间': ['2015.03.04', '2015.05.16', '2016.12.30', '2017.01.15', '2017.01.15'],
'病例': ['小感冒', '慢性头痛', '通风', '中风', '脚部内伤'], '症状': ['突然头痛', '头非常痛，不能睡觉', '怕凉', '伤口大量出血，且发烧', '崴脚，走路痛']}
"""
# 返回字典
return diagnosis_dict
def ret_define(self, sql): # 定义函数，返回自定义信息的字典
# 连接服务器Server的数据库:
db = pymysql.connect(
host="localhost",
user="root",
password="111111",
db=f"{self.database}",
charset="utf8",
)
try:
# 使用操作游标:
cursor = db.cursor()
cursor.execute(sql)
results = cursor.fetchall() # 获取查询结果的所有数据
# tables = extract_tables(sql) # 分离出sql语句中的表名
# 遍历查询结果，将每条消息都输出出来：
for row in results:
for i in range(len(row)):
print(f"{row[i]}\t", end="")
print()
except:
print("查询权限不足！或查询语句出错！")
def main(hospital_name):
sql = """SELECT *
FROM HOS, DOC, PAT, DIA
WHERE DOC.DOC_ID = DIA.DOC_ID AND
PAT.PAT_ID = DIA.PAT_ID
"""
hospital_name = hospital_name.replace("\n", "")
if hospital_name == "xx大学附属医院":
database = SelectDatabase("Hospital1")
database.ret_define(sql)
elif hospital_name == "xx阳光社区附属医院":
database = SelectDatabase("Hospital2")
database.ret_define(sql)
else:
print("暂无记录！")

BIN
background.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 171 KiB

471
data_provider.py
View File

@ -1,471 +0,0 @@
import socket
import sys
import threading
import time
import pickle
from io import StringIO
import CoreAlgorithm
import DataSearch
# 审批回执类消息head：01001001
# sql密文类消息head：01001010
# 元数据类消息head：01001100
# 实现GUI界面
import tkinter as tk
from tkinter import *
from tkinter import messagebox, scrolledtext
from PIL import ImageTk
# 美化
from ttkbootstrap import Style # pylint: disable=e0401 # type: ignore
from tkinter import ttk
sleep_time = 1
# 用于生成本地公钥和私钥
def generate_key_pair_data():
public_key_data, private_key_data = CoreAlgorithm.generate_paillier_keypair()
return private_key_data, public_key_data
# 用公钥为本地生成数字证书
def get_certificate(temp_public_key, cert_name):
public_key_str = "\n".join(
temp_public_key[i : i + 60] for i in range(0, len(temp_public_key), 60)
)
pack_public_key = (
"-----BEGIN PUBLIC KEY-----\n" + public_key_str + "\n-----END PUBLIC KEY-----"
)
cert = {"public_key": pack_public_key, "name": "<数据提供方>" + cert_name}
return cert
# 公钥转公钥信息
def key_to_data(key):
# str转bytes
byte = bytes.fromhex(key)
# bytes转PaillierPublicKey
data = pickle.loads(byte)
return data
# 加密字符串
def str_to_encrypt(message, public_data):
# str 转 int
if message.isdigit():
int_message = int(message)
else:
int_message = int.from_bytes(message.encode(), "big")
enc_message = public_data.encrypt(int_message)
print("int_message", int_message)
return enc_message
class MyClient:
def __init__(self):
# 实现GUI主界面框用到的参数
self.root = None
self.data_text = None
self.name_text = None
self.message_text = None
self.window = None
# 初始化最近一次的待处理的sql申请信息
self.recent_message = ""
# 初始化本端名
self.name = ""
# 初始化sql
self.sql = ""
# 准备界面函数
self.root_window()
self.give_name()
# 生成私钥和公钥信息
(
self.private_key_data,
self.public_key_data,
) = generate_key_pair_data() # PaillierPublicKey 类型
# 生成私钥和公钥字符串
self.private_key, self.public_key = self.generate_key_pair()
# 初始化数字证书
self.certificate = {} # 等名字输入了再生成
# 初始化socket
self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 初始化根界面
def root_window(self):
self.root = tk.Tk()
self.root.title("数据提供方智慧医疗SQL查询系统")
self.root.geometry("450x540")
# 增加背景图片
global photo
photo = ImageTk.PhotoImage(file="background.png")
label = Label(self.root, image=photo)
label.pack()
self.create_widgets() # 调用函数创建功能按钮
# 为本端取名
def give_name(self):
self.window = tk.Toplevel()
self.window.title("命名窗口")
self.window.geometry("300x320")
# 初始化信息输入框
self.name_text = tk.StringVar()
tk.Label(self.window, bg="green", text="请为本端取名:").place(x=50, y=100)
input_message = tk.Entry(self.window, textvariable=self.name_text)
input_message.pack()
input_message.place(x=80, y=150)
# 确认按钮及位置
bt_confirm = tk.Button(
self.window, bg="green", text="确认", width=10, command=self.get_name
)
bt_confirm.pack()
bt_confirm.place(x=125, y=220)
# 读取输入框的name，保存在self.name中
def get_name(self):
self.name = self.name_text.get()
messagebox.showinfo("提示", "命名成功！")
self.window.destroy()
self.root.title("<数据提供方>" + self.name + "的智慧医疗系统")
# 安全通信
def safe_connect(self):
if self.certificate != {}:
# 与服务器端地址建立通信, 8080为服务端程序的端口号
self.client.connect(("localhost", 1111))
# 为接收消息函数添加线程
threading.Thread(target=self.get_message).start()
time.sleep(3)
else:
messagebox.showinfo("提示", "请先生成你的公钥证书！")
# 向平台发送本地公钥信息
def send_public_key(self):
print("正在向平台发送本地公钥信息...")
self.client.send(str(self.certificate).encode())
print(self.certificate)
time.sleep(sleep_time)
print("发送成功!")
# 产生公私钥字符串
def generate_key_pair(self):
# Paillier 转 bytes
public_key_bytes = pickle.dumps(self.public_key_data)
private_key_bytes = pickle.dumps(self.private_key_data)
# bytes 转 str
public_key = public_key_bytes.hex()
private_key = private_key_bytes.hex()
return private_key, public_key
# 打印公钥
def print_public_key(self):
print("本地公钥信息为:")
print(self.public_key_data)
print("打印公钥如下:", end="")
public_key_str = "\n".join(
self.public_key[i : i + 60] for i in range(0, len(self.public_key), 60)
)
pack_public_key = (
"\n-----BEGIN PUBLIC KEY-----\n"
+ public_key_str
+ "\n-----END PUBLIC KEY-----\n"
)
print(pack_public_key)
messagebox.showinfo("本地公钥", pack_public_key) # GUI界面显示
# 打印私钥
def print_private_key(self):
print("本地私钥信息为:")
print(self.private_key_data)
private_key_str = "\n".join(
self.private_key[i : i + 60] for i in range(0, len(self.public_key), 60)
)
pack_private_key = (
"-----BEGIN PRIVATE KEY-----\n"
+ private_key_str
+ "\n-----END PRIVATE KEY-----\n"
)
print("打印私钥如下:", end="")
print(pack_private_key)
messagebox.showinfo("本地私钥", pack_private_key) # GUI界面显示
# 打印证书
def print_certificate(self):
if self.name != "":
self.certificate = get_certificate(self.public_key, self.name)
print("本地公钥证书为:")
message = ""
for key, value in self.certificate.items():
print(key, ":\n", value)
message += key + ":\n" + value + "\n"
messagebox.showinfo("本地公钥证书", message)
else:
messagebox.showinfo("提示", "请先为平台命名！")
# 接收消息
def get_message(self):
while True:
# 接收连接结果信息
message = self.client.recv(4096).decode()
print(message)
if message == "安全多方计算平台:单向授权成功！":
self.client.send("请发送平台的完整公钥证书".encode())
while True:
message = self.client.recv(4096).decode() # 公钥字符串比较长，要多一点
if message.startswith("{'public_key':"): # 等待接收平台公钥证书
cert = eval(message)
messagebox.showinfo("连接结果", "与平台连接成功!")
print("接收到的平台的公钥证书如下:")
message = ""
for key, value in cert.items():
if isinstance(value, bytes):
value = value.decode()
print(key, ":\n", value)
message += key + ":\n" + value
messagebox.showinfo("平台的公钥证书", message)
server_public_key = (
cert["public_key"]
.split("-----END PUBLIC KEY-----")[0]
.split("-----BEGIN PUBLIC KEY-----")[1]
)
server_public_key = server_public_key.replace("\n", "")
print("提取到的server_public_key:")
print(server_public_key)
global server_public_key_data
server_public_key_data = key_to_data(server_public_key)
print("对应的server_public_key_data:")
print(server_public_key_data)
messagebox.showinfo("平台的公钥信息", server_public_key_data)
self.client.send("认证成功!".encode())
self.send_public_key() # 单向认证后把自己的公钥证书发给平台实现双向认证
break
else:
print("错误的Message", message)
elif message == "安全多方计算平台:双向授权成功！":
messagebox.showinfo("平台发来的消息", "与平台的双向认证成功！")
self.client.send("请求查询方信息！".encode())
cert = self.client.recv(4096).decode()
cert = eval(cert)
message_p = "接收到一则非平台的公钥证书"
print(message_p)
# messagebox.showinfo("提示", message_p)
message = ""
for key, value in cert.items():
if isinstance(value, bytes):
value = value.decode()
print(key, ":\n", value)
message += key + ":\n" + value + "\n"
# messagebox.showinfo("数据查询方的公钥证书", message)
search_public_key = (
cert["public_key"]
.split("-----END PUBLIC KEY-----")[0]
.split("-----BEGIN PUBLIC KEY-----")[1]
)
search_public_key = search_public_key.replace("\n", "")
print("提取到的search_public_key:")
print(search_public_key)
global search_public_key_data
search_public_key_data = key_to_data(search_public_key)
print("对应的search_public_key_data:")
print(search_public_key_data)
# messagebox.showinfo("数据查询方的公钥信息", search_public_key_data)
elif message.startswith("01001010"):
message = message.split("01001010", 1)[
1
] # [0]是空字符串，已测试。只切一次是防止密文出现和头部一样的信息被误切。
# 使用私钥解密消息,获得sql明文
print("接收到的sql密文:", message)
int_message = int(message)
# Ciphertext转EncryptedNumber
Encrpt_message = CoreAlgorithm.EncryptedNumber(
self.public_key_data, int_message, exponent=0
)
dec_int_message = self.private_key_data.decrypt(Encrpt_message)
dec_message = dec_int_message.to_bytes(
(dec_int_message.bit_length() + 7) // 8, "big"
).decode()
print("解密后的消息为:", dec_message)
self.sql = dec_message.split("||")[2]
print("收到已授权方的sql:", self.sql)
self.design_window()
# --------------------------接收sql消息的窗口设计函数-------------------------
# 接受调用数据的请求
def send_accept_reply(self):
self.client.send(
(
"01001001"
+ str(self.client.getsockname())
+ "||"
+ self.name
+ "同意了你的数据申请。"
).encode()
)
self.window.destroy()
self.window = tk.Toplevel()
self.window.title("命名窗口")
self.window.geometry("300x320")
# 信息输入框
self.data_text = tk.StringVar()
tk.Label(self.window, bg="green", text="请输入要调用的信息:").place(x=50, y=100)
input_message = tk.Entry(self.window, textvariable=self.data_text)
input_message.pack()
input_message.place(x=80, y=150)
# 确认按钮及位置
bt_confirm = tk.Button(
self.window, bg="green", text="确认", width=10, command=self.send_data
)
bt_confirm.pack()
bt_confirm.place(x=125, y=220)
# 发送要调用的信息
def send_data(self):
message = self.data_text.get()
print(message, type(message))
# 加密并封装消息
print("正在封装并加密消息...")
global search_public_key_data
enc_message = str_to_encrypt(message, search_public_key_data).ciphertext()
pack_message = "01001100" + str(enc_message)
print("正在发送消息给平台...")
self.client.send(pack_message.encode())
print("发送成功！")
messagebox.showinfo("提示", "发送成功！")
self.window.destroy()
# 拒绝调用数据的请求
def send_decline_reply(self):
self.client.send(
(
"01001001" + str(self.client.getsockname()) + "||" + "想得美哦!不给(*^▽^*)"
).encode()
)
messagebox.showinfo("提示", "已拒绝！")
self.window.destroy()
# 稍后处理调用数据的请求
def wait_reply(self):
self.window.destroy()
# 接收到sql信息的专属窗口
def design_window(self):
if self.sql != "":
self.window = tk.Toplevel()
self.window.title("平台发来的消息")
self.window.geometry("550x320") # 设定窗口大小
self.recent_message = "收到已授权方的sql:" + self.sql
tk.Label(self.window, bg="green", text=self.recent_message).place(
x=50, y=100
)
# 创建接受按钮
accept_button = tk.Button(
self.window, text="接受", width=15, command=self.send_accept_reply
)
accept_button.place(x=90, y=220)
# 创建拒绝按钮
decline_button = tk.Button(
self.window, text="拒绝", width=15, command=self.send_decline_reply
)
decline_button.place(x=225, y=220)
# 创建稍后处理按钮
decline_button = tk.Button(
self.window, text="稍后处理", width=15, command=self.wait_reply
)
decline_button.place(x=360, y=220)
else:
messagebox.showinfo("提示", "暂无待处理的信息。")
# -----------------------------------------------------------------------
# 调用本地信息库
def search_data(self):
output = StringIO() # 用于保存打印信息
sys.stdout = output # 重定向sys.stdout到StringIO
print(self.name)
DataSearch.main(self.name)
sys.stdout = sys.__stdout__
long_text = output.getvalue()
print(long_text)
# 创建显示窗口
window = tk.Toplevel()
# 创建滚动文本部件
roll = scrolledtext.ScrolledText(window, width=200, height=40)
# 打印超长字符串
roll.insert(tk.END, long_text)
roll.pack()
# --------------------------主界面的按键功能绑定--------------------------------
# 界面按键
def create_widgets(self):
# 创建按钮和标签等部件并使用ttk和Style进行美化
style = Style(theme="darkly") # 指定样式主题
self.root = style.master
button1 = ttk.Button(
self.root,
text="查看我的公钥",
bootstyle="info-outline",
command=self.print_public_key,
width=30,
)
button1.place(x=104, y=310)
button2 = ttk.Button(
self.root,
text="查看我的私钥",
bootstyle="info-outline",
command=self.print_private_key,
width=30,
)
button2.place(x=104, y=348)
button3 = ttk.Button(
self.root,
text="生成我的证书",
bootstyle="info-outline",
command=self.print_certificate,
width=30,
)
button3.place(x=104, y=383)
button4 = ttk.Button(
self.root,
text="获取平台认证",
bootstyle="info-outline",
command=self.safe_connect,
width=30,
)
button4.place(x=104, y=418)
button5 = ttk.Button(
self.root,
text="处理请求信息",
bootstyle="info-outline",
command=self.design_window,
width=30,
)
button5.place(x=104, y=453)
button6 = ttk.Button(
self.root,
text="查看医疗记录",
bootstyle="info-outline",
command=self.search_data,
width=30,
)
button6.place(x=104, y=488)
# GUI界面
def run(self):
self.root.mainloop()
# ----------------------------------------------------主程序----------------------------------------------------
global server_public_key_data, search_public_key_data, photo # 使用全局变量，否则图片可能会被回收，不显示
# GUI界面
app = MyClient()
app.run()

508
data_requirer.py
View File

@ -1,508 +0,0 @@
import socket
import sys
import threading
import time
import pickle
from io import StringIO
import CoreAlgorithm
import DataSearch
# 审批回执类消息head：01001001
# sql密文类消息head：01001010
# 元数据类消息head：01001100
# 实现GUI界面
import tkinter as tk
from tkinter import *
from tkinter import messagebox, scrolledtext
from PIL import ImageTk
# 美化
from ttkbootstrap import Style # pylint: disable=e0401 # type: ignore
from tkinter import ttk
# 用于生成本地公钥和私钥
def generate_key_pair_data():
public_key_data, private_key_data = CoreAlgorithm.generate_paillier_keypair()
return private_key_data, public_key_data
# 用公钥为本地生成数字证书
def get_certificate(temp_public_key, cert_name):
public_key_str = "\n".join(
temp_public_key[i : i + 60] for i in range(0, len(temp_public_key), 60)
)
pack_public_key = (
"-----BEGIN PUBLIC KEY-----\n" + public_key_str + "\n-----END PUBLIC KEY-----"
)
cert = {"public_key": pack_public_key, "name": "<数据查询方>" + cert_name}
return cert
# 公钥转公钥信息
def key_to_data(key):
# str转bytes
byte = bytes.fromhex(key)
# bytes转PaillierPublicKey
data = pickle.loads(byte)
return data
# 加密字符串
def str_to_encrypt(message, public_data):
# str 转 int
if message.isdigit():
int_message = int(message)
else:
int_message = int.from_bytes(message.encode(), "big")
enc_message = public_data.encrypt(int_message)
print("int_message", int_message)
return enc_message
class MyClient:
def __init__(self):
# 实现GUI主界面框用到的参数
self.root = None
self.data_text = None
self.name_text = None
self.recent_message = None
self.window = None
self.sql_text = None
# 初始化本端名
self.name = ""
# 初始化sql
self.sql = ""
# 准备界面函数
self.root_window()
self.give_name()
# 生成私钥和公钥信息
(
self.private_key_data,
self.public_key_data,
) = generate_key_pair_data() # PaillierPublicKey 类型
# 生成私钥和公钥字符串
self.private_key, self.public_key = self.generate_key_pair()
# 获取数字证书
self.certificate = {} # 等名字输入了再生成
# 初始化socket
self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 初始化根界面
def root_window(self):
self.root = tk.Tk()
self.root.title("数据查询方智慧医疗SQL查询系统")
self.root.geometry("450x660")
# 增加背景图片
global photo
photo = ImageTk.PhotoImage(file="background.png")
label = Label(self.root, image=photo)
label.pack()
self.create_widgets() # 调用函数创建功能按钮
# 为本端取名
def give_name(self):
self.window = tk.Toplevel()
self.window.title("命名窗口")
self.window.geometry("300x320")
# 初始化信息输入框
self.name_text = tk.StringVar()
tk.Label(self.window, bg="green", text="请为本端取名:").place(x=50, y=100)
input_message = tk.Entry(self.window, textvariable=self.name_text)
input_message.pack()
input_message.place(x=80, y=150)
# 确认按钮及位置
bt_confirm = tk.Button(
self.window, bg="green", text="确认", width=10, command=self.get_name
)
bt_confirm.pack()
bt_confirm.place(x=125, y=220)
# 读取输入框的name，保存在self.name中
def get_name(self):
self.name = self.name_text.get()
messagebox.showinfo("提示", "命名成功！")
self.window.destroy()
self.root.title("<数据查询方>" + self.name + "的智慧医疗系统")
# 编写sql
def create_sql(self):
self.window = tk.Toplevel() # 建立窗口
self.window.title("编写sql语句") # 为窗口命名
self.window.geometry("580x270") # 设定窗口大小
# 初始化信息输入框
self.sql_text = tk.StringVar()
tk.Label(self.window, bg="green", text="请输入您需要的sql查询:").place(x=70, y=50)
input_message = tk.Entry(self.window, textvariable=self.sql_text, width=60)
input_message.place(x=70, y=100)
# 初始化确认按钮
bt_confirm = tk.Button(
self.window, bg="green", text="确认", width=60, command=self.get_sql
)
bt_confirm.place(x=70, y=150)
# self.sql = input("请编写你的sql:")
# 读取输入框的sql语句，保存在self.sql中
def get_sql(self):
self.sql = self.sql_text.get()
if self.sql != "":
messagebox.showinfo("提示", "编写成功！")
else:
messagebox.showinfo("提示", "编写失败！语句不能为空！")
self.window.destroy()
# 安全通信
def safe_connect(self):
if self.certificate != {}:
# 与服务器端地址建立通信, 8080为服务端程序的端口号
self.client.connect(("localhost", 1111))
# 为接收消息函数添加线程
threading.Thread(target=self.get_message).start()
time.sleep(3)
else:
messagebox.showinfo("提示", "请先生成你的公钥证书！")
# 向平台发送本地公钥信息
def send_public_key(self):
print("正在向平台发送本地公钥信息...")
self.client.send(str(self.certificate).encode())
time.sleep(1)
print("发送成功!")
# 发送sql
def send_sql(self):
if self.sql != "":
print("正在封装并加密消息...")
message = self.deal_sql(self.sql)
print("处理后的message:", message)
print("正在发送消息给平台...")
self.client.send(message.encode())
print("发送成功！")
messagebox.showinfo("提示", "发送成功！") # GUI界面显示
else:
messagebox.showinfo("提示", "请先进行编写sql的操作确认您的需求！") # GUI界面显示
# 加密封装sql语句
def deal_sql(self, context):
message = str(self.client.getsockname()) + "||" + context
global server_public_key_data
enc_message = str_to_encrypt(message, server_public_key_data) # 用平台的公钥进行加密
return "01001010" + str(enc_message.ciphertext()) # int 型
# 产生公私钥字符串
def generate_key_pair(self):
# Paillier 转 bytes
public_key_bytes = pickle.dumps(self.public_key_data)
private_key_bytes = pickle.dumps(self.private_key_data)
# bytes 转 str
public_key = public_key_bytes.hex()
private_key = private_key_bytes.hex()
return private_key, public_key
# 打印公钥
def print_public_key(self):
print("本地公钥信息为:")
print(self.public_key_data)
print("打印公钥如下:")
public_key_str = "\n".join(
self.public_key[i : i + 60] for i in range(0, len(self.public_key), 60)
)
pack_public_key = (
"\n-----BEGIN PUBLIC KEY-----\n"
+ public_key_str
+ "\n-----END PUBLIC KEY-----\n"
)
print(pack_public_key)
# show_list()
messagebox.showinfo("本地公钥", pack_public_key) # GUI界面显示
# 打印私钥
def print_private_key(self):
print("本地私钥信息为:")
print(self.private_key_data)
private_key_str = "\n".join(
self.private_key[i : i + 60] for i in range(0, len(self.public_key), 60)
)
pack_private_key = (
"-----BEGIN PRIVATE KEY-----\n"
+ private_key_str
+ "\n-----END PRIVATE KEY-----\n"
)
print("打印私钥如下:")
print(pack_private_key)
messagebox.showinfo("本地私钥", pack_private_key) # GUI界面显示
# 打印证书
def print_certificate(self):
if self.name != "":
self.certificate = get_certificate(self.public_key, self.name)
print("本地公钥证书为:")
message = ""
for key, value in self.certificate.items():
print(key, ":\n", value)
message += key + ":\n" + value + "\n"
messagebox.showinfo("本地公钥证书", message)
else:
messagebox.showinfo("提示", "请先为平台命名！")
# 接收消息
def get_message(self):
while True:
# 接收连接结果信息
message = self.client.recv(4096).decode()
print(message)
if message == "安全多方计算平台:单向授权成功！":
self.client.send("请发送平台的完整公钥证书".encode())
while True:
message = self.client.recv(4096).decode()
if message.startswith("{'public_key':"):
cert = eval(message)
messagebox.showinfo("连接结果", "与平台连接成功!")
print("接收到的平台的公钥证书如下:")
message = ""
for key, value in cert.items():
if isinstance(value, bytes):
value = value.decode()
print(key, ":\n", value)
message += key + ":\n" + value
messagebox.showinfo("平台的公钥证书", message)
server_public_key = (
cert["public_key"]
.split("-----END PUBLIC KEY-----")[0]
.split("-----BEGIN PUBLIC KEY-----")[1]
)
server_public_key = server_public_key.replace("\n", "")
print("提取到的server_public_key:")
print(server_public_key)
global server_public_key_data
server_public_key_data = key_to_data(server_public_key)
print("对应的server_public_key_data:")
print(server_public_key_data)
messagebox.showinfo("平台的公钥信息", server_public_key_data)
self.client.send("认证成功!".encode())
self.send_public_key() # 单向认证后把自己的公钥证书发给平台实现双向认证
break
else:
print("错误的Message", message)
elif message == "安全多方计算平台:双向授权成功！":
messagebox.showinfo("平台发来的消息", "与平台的双向认证成功！")
elif message.startswith("01001010"):
message = message.split("01001010", 1)[
1
] # [0]是空字符串，已测试。只切一次是防止密文出现和头部一样的信息被误切。
# 使用私钥解密消息,获得sql明文
print("接收到的sql密文:", message)
int_message = int(message)
# Ciphertext转EncryptedNumber
Encrpt_message = CoreAlgorithm.EncryptedNumber(
self.public_key_data, int_message, exponent=0
)
dec_int_message = self.private_key_data.decrypt(Encrpt_message)
dec_message = dec_int_message.to_bytes(
(dec_int_message.bit_length() + 7) // 8, "big"
).decode()
print("解密后的消息为:", dec_message)
self.sql = dec_message.split("||")[2]
print("收到已授权方的sql:", self.sql)
self.design_window()
elif message.startswith("结果:"):
message = message.split("结果:")[1]
if message != "正在等待接收其他信息...":
int_message = int(message)
# Ciphertext转EncryptedNumber
Encrpt_message = CoreAlgorithm.EncryptedNumber(
self.public_key_data, int_message, exponent=0
)
dec_int_message = self.private_key_data.decrypt(Encrpt_message)
print("分析结果:", dec_int_message)
messagebox.showinfo("分析结果", dec_int_message)
else:
messagebox.showinfo("提示", message)
# --------------------------接收sql消息的窗口设计函数-------------------------
# 接受调用数据的请求
def send_accept_reply(self):
self.client.send(
(
"01001001" + str(self.client.getsockname()) + "||安全多方平台同意了你的数据申请。"
).encode()
)
self.window = tk.Toplevel()
self.window.title("命名窗口")
self.window.geometry("300x320")
# 信息输入框
self.data_text = tk.StringVar()
tk.Label(self.window, bg="green", text="请输入要调用的信息:").place(x=50, y=100)
input_message = tk.Entry(self.window, textvariable=self.data_text)
input_message.pack()
input_message.place(x=80, y=150)
# 确认按钮及位置
bt_confirm = tk.Button(
self.window, bg="green", text="确认", width=10, command=self.send_data
)
bt_confirm.pack()
bt_confirm.place(x=125, y=220)
# 发送要调用的信息
def send_data(self):
message = self.name_text.get()
# 加密并封装消息
print("正在封装并加密消息...")
enc_message = str_to_encrypt(message, self.public_key_data).ciphertext()
pack_message = "01001100" + str(enc_message)
print("正在发送消息给平台...")
self.client.send(pack_message.encode())
print("发送成功！")
messagebox.showinfo("提示", "发送成功！")
self.window.destroy()
# 拒绝调用数据的请求
def send_decline_reply(self):
self.client.send(
(
"01001001" + str(self.client.getsockname()) + "||" + "想得美哦!不给(*^▽^*)"
).encode()
)
messagebox.showinfo("提示", "已拒绝！")
self.window.destroy()
# 稍后处理调用数据的请求
def wait_reply(self):
self.window.destroy()
# 接收到sql信息的专属窗口
def design_window(self):
if self.sql != "":
self.window = tk.Toplevel()
self.window.title("平台发来的消息")
self.window.geometry("550x320") # 设定窗口大小
self.recent_message = "收到已授权方的sql:" + self.sql
tk.Label(self.window, bg="green", text=self.recent_message).place(
x=50, y=100
)
# 创建接受按钮
accept_button = tk.Button(
self.window, text="接受", width=15, command=self.send_accept_reply
)
accept_button.place(x=90, y=220)
# 创建拒绝按钮
decline_button = tk.Button(
self.window, text="拒绝", width=15, command=self.send_decline_reply
)
decline_button.place(x=225, y=220)
# 创建稍后处理按钮
decline_button = tk.Button(
self.window, text="稍后处理", width=15, command=self.wait_reply
)
decline_button.place(x=360, y=220)
else:
messagebox.showinfo("提示", "暂无待处理的信息。")
# --------------------------------------------------------------------------
# 调用本地信息库
def search_data(self):
output = StringIO() # 用于保存打印信息
sys.stdout = output # 重定向sys.stdout到StringIO
print(self.name)
DataSearch.main(self.name)
sys.stdout = sys.__stdout__
long_text = output.getvalue()
print(long_text)
# 创建显示窗口
window = tk.Toplevel()
# 创建滚动文本部件
roll = scrolledtext.ScrolledText(window, width=200, height=40)
# 打印超长字符串
roll.insert(tk.END, long_text)
roll.pack()
# --------------------------主界面的按键功能绑定--------------------------------
# 界面按键
def create_widgets(self):
# 创建按钮和标签等部件并使用ttk和Style进行美化
style = Style(theme="darkly") # 指定样式主题
self.root = style.master
button1 = ttk.Button(
self.root,
text="查看我的公钥",
bootstyle="info-outline",
command=self.print_public_key,
width=30,
)
button1.place(x=104, y=360)
button2 = ttk.Button(
self.root,
text="查看我的私钥",
bootstyle="info-outline",
command=self.print_private_key,
width=30,
)
button2.place(x=104, y=398)
button3 = ttk.Button(
self.root,
text="生成我的证书",
bootstyle="info-outline",
command=self.print_certificate,
width=30,
)
button3.place(x=104, y=433)
button4 = ttk.Button(
self.root,
text="获取平台认证",
bootstyle="info-outline",
command=self.safe_connect,
width=30,
)
button4.place(x=104, y=468)
button5 = ttk.Button(
self.root,
text="调用SQL查询",
bootstyle="info-outline",
command=self.create_sql,
width=30,
)
button5.place(x=104, y=503)
button6 = ttk.Button(
self.root,
text="发送SQL请求",
bootstyle="info-outline",
command=self.send_sql,
width=30,
)
button6.place(x=104, y=538)
button7 = ttk.Button(
self.root,
text="处理请求信息",
bootstyle="info-outline",
command=self.design_window,
width=30,
)
button7.place(x=104, y=573)
button8 = ttk.Button(
self.root,
text="查看医疗记录",
bootstyle="info-outline",
command=self.search_data,
width=30,
)
button8.place(x=104, y=608)
# GUI界面
def run(self):
self.root.mainloop()
# ----------------------------------------------------主程序----------------------------------------------------
global server_public_key_data, photo # 使用全局变量，否则图片可能会被回收，不显示
# GUI界面
app = MyClient()
app.run()

10
note.txt
View File

@ -1,10 +0,0 @@
编写的sql格式硬性规定：
含有from，多个对象需要用,分割
要先定好测试案例，sql语句，然后才方便用if/else，跑结果
中央人民医院
xx阳光社区诊所
xx大学附属医院
select count(*) from xx阳光社区诊所, xx大学附属医院 where diag = "cancer";

562
server.py
View File

@ -1,562 +0,0 @@
import socket
import time
import pickle
import wx # pylint: disable=e0401 # type: ignore
import CoreAlgorithm
import threading
import DataSearch
import server_demo
sleep_time = 0.2
# 审批回执类消息head：01001001
# sql密文类消息head：01001010
# 元数据类消息head：01001100
# 用于生成本地公钥和私钥信息
def generate_key_pair_data():
public_key_data, private_key_data = CoreAlgorithm.generate_paillier_keypair()
return private_key_data, public_key_data
# 从公钥证书中提取公钥信息
def get_public_key_data(message):
message = eval(message.replace("\n", ""))
public_key = (
message["public_key"]
.replace("-----BEGIN PUBLIC KEY-----", "")
.replace("-----END PUBLIC KEY-----", "")
) # 分割得到公钥
public_key_bytes = bytes.fromhex(public_key)
public_key_data = pickle.loads(public_key_bytes)
return public_key_data
# 用公钥为本地生成数字证书
def get_certificate(temp_public_key, cert_name):
public_key_str = "\n".join(
temp_public_key[i : i + 60] for i in range(0, len(temp_public_key), 60)
)
pack_public_key = (
"-----BEGIN PUBLIC KEY-----\n" + public_key_str + "\n-----END PUBLIC KEY-----\n"
)
cert = {"public_key": pack_public_key, "name": cert_name}
return cert
# 加密字符串
def str_to_encrypt(message, public_data):
# str 转 int
if message.isdigit():
int_message = int(message)
else:
int_message = int.from_bytes(message.encode(), "big")
enc_message = public_data.encrypt(int_message)
print("int_message", int_message)
return enc_message
class MyServer(server_demo.MyFrame):
def __init__(self, parent):
server_demo.MyFrame.__init__(self, parent)
# 生成私钥和公钥信息
self.private_key_data, self.public_key_data = generate_key_pair_data()
# 生成私钥和公钥字符串
self.private_key, self.public_key = self.generate_key_pair()
# 获取数字证书
self.certificate = get_certificate(self.public_key, "安全多方服务器")
# 初始化当前sql
self.sql = ""
# 初始化sql拆分对象
self.divide_sqls = []
# 初始化sql拆分数据源
self.divide_providers = []
# 记录属于同一个请求的元数据密文
self.datas = []
# 初始化数据查询方的公钥证书为str，在发来过后为其赋值
self.search_cert = ""
# 初始化socket
self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 绑定IP地址和端口
self.server.bind(("localhost", 1111))
# 设置最大监听数
self.server.listen(5)
# 设置一个字典，用来保存每一个客户端的连接和身份信息
self.socket_mapping = {} # temp_socket: [addr, 公钥信息]
# 设置接收的最大字节数
self.maxSize = 4096
# 记录调用方地址
self.source = None
# 记录收集信息的数量
self.flag = 0
# 记录需要收集的信息总量
self.total = 0
# 保存安全多方计算结果
self.result = 0
self.out_look()
# 等待客户端连接
message_p = "等待客户端连接..."
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
# 添加消息记录列
def out_look(self):
self.m_listCtrl1.InsertColumn(0, "消息记录") # 为聊天框添加‘消息记录’列
self.m_listCtrl1.SetColumnWidth(0, 1000)
self.m_listCtrl2.InsertColumn(0, "消息记录") # 为聊天框添加‘消息记录’列
self.m_listCtrl2.SetColumnWidth(0, 1000)
self.m_listCtrl3.InsertColumn(0, "消息记录") # 为聊天框添加‘消息记录’列
self.m_listCtrl3.SetColumnWidth(0, 1000)
# 建立连接，设置线程
def run(self):
while True:
client_socket, addr = self.server.accept()
# 发送信息，提示客户端已成功连接
message_p = "与{0}连接成功！".format(addr)
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
# 将客户端socket等信息存入字典
self.socket_mapping[client_socket] = []
self.socket_mapping[client_socket].append(addr)
message_p = "安全多方计算平台:单向授权成功！"
client_socket.send(message_p.encode())
self.m_listCtrl2.Append([message_p])
time.sleep(sleep_time)
# 创建线程，负责接收客户端信息并转发给其他客户端
threading.Thread(
target=self.recv_from_client, args=(client_socket,)
).start()
# 产生公私钥字符串
def generate_key_pair(self):
# Paillier 转 bytes
public_key_bytes = pickle.dumps(self.public_key_data)
private_key_bytes = pickle.dumps(self.private_key_data)
# bytes 转 str
public_key = public_key_bytes.hex()
private_key = private_key_bytes.hex()
return private_key, public_key
# 接收客户端消息并转发
def recv_from_client(self, client_socket): # client_socket指的是连接到的端口socket
while True:
message = client_socket.recv(self.maxSize).decode("utf-8")
message_p = "接收到来自{0}的message:\n{1}".format(
self.socket_mapping[client_socket][0], message
)
print(message_p)
self.m_listCtrl1.Append([message_p]) # 准备文件传输
if message.startswith("01001001"):
message_p = "正在解析消息内容..."
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
# 去掉审批回执的头部
message = message.split("01001001", 1)[
1
] # [0]是空字符串，已测试。只切一次是防止密文出现和头部一样的信息被误切。
# 认证发送者的合法性
sender = message.split("||")[0]
context = message.split("||")[1]
message_p = "接收到来自{0}的message:\n{1}".format(sender, context)
print(message_p)
self.m_listCtrl1.Append([message_p])
# time.sleep(sleep_time)
self.flag += 1
if context == "想得美哦!不给(*^▽^*)":
self.flag = -9999
elif message.startswith("01001010"):
message_p = "正在解析消息内容..."
print(message_p)
self.m_listCtrl3.Append([message_p])
# time.sleep(sleep_time)
# 认证发送者的合法性
if len(self.socket_mapping[client_socket]) > 1: # 如果发送者的公钥信息已被收集
# 识别明文中一起发送的发送目标 明文应是发送者||发送内容（||时间戳等）,对象用socket表示吧...
message = message.split("01001010", 1)[
1
] # [0]是空字符串，已测试。只切一次是防止密文出现和头部一样的信息被误切。
# 用发送目标之前给的公钥加密明文，得到密文。
# 去掉sql密文的头部
# 使用平台私钥解密消息,获得sql明文
message_p = "接收到的sql密文:" + message
print(message_p)
self.m_listCtrl1.Append([message_p])
time.sleep(sleep_time)
int_message = int(message)
# Ciphertext转EncryptedNumber
Encrpt_message = CoreAlgorithm.EncryptedNumber(
self.public_key_data, int_message, exponent=0
)
dec_int_message = self.private_key_data.decrypt(Encrpt_message)
dec_message = dec_int_message.to_bytes(
(dec_int_message.bit_length() + 7) // 8, "big"
).decode()
message_p = "解密后的消息为:" + dec_message
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
self.source = dec_message.split("||")[0]
self.sql = dec_message.split("||")[1]
message_p = "收到已授权方的sql:" + self.sql
print(message_p)
self.m_listCtrl1.Append([message_p])
time.sleep(sleep_time)
message_p = "待处理的sql语句为:" + self.sql
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
message_p = "正在拆分sql..."
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
self.divide_providers = DataSearch.extract_tables(self.sql)
message_p = "涉及到的数据持有对象有:" + str(self.divide_providers)
print(message_p)
self.m_listCtrl3.Append(([message_p]))
self.divide_sqls = DataSearch.divide_sql(self.sql)
message_p = "正在分别加密和封装sql..."
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
self.pack_sqls()
message_p = "发送成功！"
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
else:
message_p = "非授权对象,禁止访问!"
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
# show_list()
elif message.startswith("01001100"):
message_p = "正在解析消息内容..."
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
# 去掉元数据头部
message = message.split("01001100", 1)[
1
] # [0]是空字符串，已测试。只切一次是防止密文出现和头部一样的信息被误切。
# 把元数据存入本地数据库的临时表里，格式：provider + encode_data
# print("message:", message)
int_message = int(message)
message_p = "收到元数据为:{}".format(int_message)
print(message_p)
self.m_listCtrl1.Append([message_p])
# 根据证书找公钥信息
search_public_key = (
self.search_cert["public_key"]
.split("-----END PUBLIC KEY-----")[0]
.split("-----BEGIN PUBLIC KEY-----")[1]
)
search_public_key = search_public_key.replace("\n", "")
print("提取到的search_public_key:")
print(search_public_key)
# str转bytes
byte = bytes.fromhex(search_public_key)
# bytes转PaillierPublicKey
search_public_key_data = pickle.loads(byte)
print("对应的search_public_key_data:")
print(search_public_key_data)
# int密 -- EncryptedNumber密
Encrpt_message = CoreAlgorithm.EncryptedNumber(
search_public_key_data, int_message, exponent=0
)
self.datas.append(Encrpt_message)
# Ciphertext转EncryptedNumber
# print("int_message:", int_message)
# Encrpt_message = CoreAlgorithm.EncryptedNumber(self.public_key_data, int_message, exponent=0)
# print("Enc:", Encrpt_message)
# dec_int_message = self.private_key_data.decrypt(Encrpt_message)
# print("dec:", dec_int_message)
# dec_message = dec_int_message.to_bytes((dec_int_message.bit_length() + 7) // 8, 'big').decode()
# print("解密后的消息为:", dec_message) # 已测试，说明平台不可解密元数据
self.safe_calculate()
elif message == "请发送平台的完整公钥证书":
message_p = "正在发送证书..."
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
message_p = str(self.certificate)
client_socket.send(message_p.encode()) # 二进制传输
self.m_listCtrl2.Append([message_p])
time.sleep(sleep_time)
message_p = "发送完成！"
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
elif message.startswith("{'public_key':"):
print(message)
print(self.socket_mapping)
print(get_public_key_data(message))
self.socket_mapping[client_socket].append(
get_public_key_data(message)
) # 绑定端口与公钥信息的关系
print(self.socket_mapping)
cert = eval(message)
print(cert, type(cert)) # 字典型
if cert["name"].startswith("<数据查询方>"):
self.search_cert = cert # 字典型
self.socket_mapping[client_socket].append(cert["name"]) # 绑定端口与用户身份的关系
message_p = "接收到一则公钥证书:"
print(message_p)
self.m_listCtrl1.Append([message_p])
time.sleep(sleep_time)
message_p = ""
for key, value in cert.items():
if isinstance(value, bytes):
value = value.decode()
print(key, ":", value)
message_p = key + ":\n" + value
print(message_p)
self.m_listCtrl1.Append([message_p])
time.sleep(sleep_time)
message_p = "发送完成！"
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
message_p = "安全多方计算平台:双向授权成功！"
print(message_p)
self.m_listCtrl2.Append([message_p])
client_socket.send(message_p.encode("utf-8")) # 二进制传输
time.sleep(sleep_time)
message_p = "使用对象表已更新:"
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
print(self.socket_mapping)
self.m_listCtrl3.Append([self.socket_mapping])
elif message == "请求查询方信息！":
if str(self.search_cert) != "":
message_p = "正在发送提供方的证书..."
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
message_p = str(self.search_cert)
client_socket.send(message_p.encode()) # 二进制传输
self.m_listCtrl2.Append([message_p])
time.sleep(sleep_time)
else:
client_socket.send(" ".encode()) # 二进制传输
for key, value in self.socket_mapping.items():
message_p = str(key) + ":" + str(value) + "\n"
print(message_p)
self.m_listCtrl3.Append([message_p])
elif message == "":
pass
elif message == "认证成功!":
pass
else:
message_p = "Message:" + message
print(message_p)
self.m_listCtrl1.Append([message_p])
time.sleep(sleep_time)
message_p = "错误的消息格式，丢弃！"
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
# 打印公钥
def print_public_key(self, event):
message_p = "本地公钥信息为:"
print(message_p)
self.m_listCtrl3.Append([message_p])
message_p = self.public_key_data
print(message_p)
self.m_listCtrl3.Append([message_p])
message_p = "打印公钥如下:"
print(message_p)
self.m_listCtrl3.Append([message_p])
public_key_str = "\n".join(
self.public_key[i : i + 60] for i in range(0, len(self.public_key), 60)
)
pack_public_key = (
"-----BEGIN PUBLIC KEY-----\n"
+ public_key_str
+ "\n-----END PUBLIC KEY-----\n"
)
message_p = pack_public_key
print(message_p)
message = message_p.split("\n") # 设置打印格式，因为显示窗打印不了\n
for i in range(len(message)):
self.m_listCtrl3.Append([message[i]])
# show_list()
# 打印私钥
def print_private_key(self, event):
message_p = "本地私钥信息为:"
print(message_p)
self.m_listCtrl3.Append([message_p])
message_p = self.private_key_data
print(message_p)
self.m_listCtrl3.Append([message_p])
private_key_str = "\n".join(
self.private_key[i : i + 60] for i in range(0, len(self.public_key), 60)
)
pack_private_key = (
"-----BEGIN PRIVATE KEY-----\n"
+ private_key_str
+ "\n-----END PRIVATE KEY-----\n"
)
message_p = "打印私钥如下:"
print(message_p)
self.m_listCtrl3.Append([message_p])
message_p = pack_private_key
print(message_p)
message = message_p.split("\n") # 设置打印格式，因为显示窗打印不了\n
for i in range(len(message)):
self.m_listCtrl3.Append([message[i]])
# 打印证书
def print_certificate(self, event):
message_p = "本地公钥证书为:"
print(message_p)
self.m_listCtrl3.Append([message_p])
for key, value in self.certificate.items():
message_p = key + ":"
print(message_p)
self.m_listCtrl3.Append([message_p])
if key == "public_key":
value = value.split("\n")
for i in range(len(value)):
self.m_listCtrl3.Append([value[i]])
else:
self.m_listCtrl3.Append([value])
# show_list()
# 加密封装sqls并发送
def pack_sqls(self):
for key, value in self.socket_mapping.items():
for i in range(len(self.divide_providers)):
if (
self.divide_providers[i] in value[2]
): # eg: value[2] == "<数据提供方>风舱医院"
for j in range(len(self.divide_sqls)):
if (
self.divide_providers[i] in self.divide_sqls[j]
): # 如果发送目标和信息匹配)
sql = (
str(self.source)
+ "||"
+ str(key.getsockname())
+ "||"
+ self.divide_sqls[i]
)
print(sql)
int_enc_sql = str_to_encrypt(
sql, value[1]
).ciphertext() # 用接收者的公钥加密消息
message_p = "01001010" + str(int_enc_sql)
key.send(message_p.encode())
self.m_listCtrl2.Append([message_p])
message_p = "已将消息{0}发送给{1},其地址为{2}".format(
self.divide_sqls[j],
self.divide_providers[i],
key.getsockname(),
)
print(message_p)
self.m_listCtrl3.Append([message_p])
# 安全算法
def safe_calculate(self):
self.total = len(self.divide_providers)
if self.flag == self.total:
message_p = "正在进行安全多方计算分析..."
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
# #################安全多方计算分析过程##############
if "select count(*) from xx阳光社区诊所, xx大学附属医院" in self.sql:
print(self.datas)
for x in self.datas:
self.result = self.result + x
# EncryptedNumber密 -- int密
self.result = self.result.ciphertext()
# int密 -- str密
self.result = str(self.result)
message = "分析成功!"
print(message)
self.m_listCtrl3.Append([message])
# EncryptedNumber 转 int
# self.result = self.result.ciphertext()
message = "结果:" + str(self.result)
print(message)
self.m_listCtrl3.Append([message])
time.sleep(sleep_time)
message_p = "正在发送结果给申请人..."
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
for key, value in self.socket_mapping.items():
if value[2].startswith("<数据查询方>"):
key.send(message.encode())
# 重置参数
self.total = 0
self.flag = 0
self.result = 0
self.datas = []
elif self.flag < 0:
message_p = "结果:已有数据持有方拒绝了提供消息的请求，安全分析无法进行，分析失败！"
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
for key, value in self.socket_mapping.items():
if value[2].startswith("<数据查询方>"):
key.send(message_p.encode())
# 重置参数
self.total = 0
self.flag = 0
else:
message_p = "结果:正在等待接收其他信息..."
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
for key, value in self.socket_mapping.items():
if value[2].startswith("<数据查询方>"):
key.send(message_p.encode())
message_p = "发送完成！"
print(message_p)
self.m_listCtrl3.Append([message_p])
time.sleep(sleep_time)
# ----------------------------------------------------主程序----------------------------------------------------
app = wx.App()
frame = MyServer(None)
frame.Show(True) # 展示登录页面
threading.Thread(target=frame.run).start() # 在新线程中运行服务器
app.MainLoop()

262
server_demo.py
View File

@ -1,262 +0,0 @@
import wx # pylint: disable=e0401 # type: ignore
import wx.xrc # pylint: disable=e0401 # type: ignore
class MyFrame(wx.Frame):
def __init__(self, parent):
wx.Frame.__init__(
self,
parent,
id=wx.ID_ANY,
title="安全多方服务器平台",
pos=wx.DefaultPosition,
size=wx.Size(444, 749),
style=wx.DEFAULT_FRAME_STYLE | wx.TAB_TRAVERSAL,
)
self.SetSizeHints(wx.DefaultSize, wx.DefaultSize)
self.SetFont(wx.Font(5, 70, 90, 90, False, "宋体"))
self.SetForegroundColour(wx.SystemSettings.GetColour(wx.SYS_COLOUR_WINDOW))
self.SetBackgroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_HIGHLIGHTTEXT)
)
fgSizer1 = wx.FlexGridSizer(0, 2, 0, 0)
fgSizer1.SetFlexibleDirection(wx.BOTH)
fgSizer1.SetNonFlexibleGrowMode(wx.FLEX_GROWMODE_SPECIFIED)
bSizer11 = wx.BoxSizer(wx.VERTICAL)
self.m_bitmap11 = wx.StaticBitmap(
self, wx.ID_ANY, wx.NullBitmap, wx.DefaultPosition, wx.Size(70, 30), 0
)
bSizer11.Add(self.m_bitmap11, 0, wx.ALL, 5)
self.m_staticline211 = wx.StaticLine(
self, wx.ID_ANY, wx.DefaultPosition, wx.Size(-1, 2), wx.LI_HORIZONTAL
)
self.m_staticline211.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_INACTIVECAPTION)
)
self.m_staticline211.SetMaxSize(wx.Size(80, -1))
bSizer11.Add(self.m_staticline211, 0, wx.EXPAND | wx.ALL, 5)
self.m_staticText1 = wx.StaticText(
self, wx.ID_ANY, "接收记录", wx.DefaultPosition, wx.Size(-1, 30), 0
)
self.m_staticText1.Wrap(-1)
self.m_staticText1.SetFont(wx.Font(11, 70, 90, 90, False, "宋体"))
self.m_staticText1.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_WINDOWTEXT)
)
bSizer11.Add(self.m_staticText1, 0, wx.ALL, 5)
self.m_bitmap1 = wx.StaticBitmap(
self, wx.ID_ANY, wx.NullBitmap, wx.DefaultPosition, wx.Size(70, 111), 0
)
bSizer11.Add(self.m_bitmap1, 0, wx.ALL, 5)
self.m_staticline21 = wx.StaticLine(
self, wx.ID_ANY, wx.DefaultPosition, wx.Size(-1, 2), wx.LI_HORIZONTAL
)
self.m_staticline21.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_INACTIVECAPTION)
)
self.m_staticline21.SetMaxSize(wx.Size(80, -1))
bSizer11.Add(self.m_staticline21, 0, wx.EXPAND | wx.ALL, 5)
self.m_staticText11 = wx.StaticText(
self, wx.ID_ANY, "发送记录", wx.DefaultPosition, wx.Size(-1, 30), 0
)
self.m_staticText11.Wrap(-1)
self.m_staticText11.SetFont(wx.Font(11, 70, 90, 90, False, "宋体"))
self.m_staticText11.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_WINDOWTEXT)
)
bSizer11.Add(self.m_staticText11, 0, wx.ALL, 5)
self.m_bitmap12 = wx.StaticBitmap(
self, wx.ID_ANY, wx.NullBitmap, wx.DefaultPosition, wx.Size(70, 111), 0
)
bSizer11.Add(self.m_bitmap12, 0, wx.ALL, 5)
self.m_staticline212 = wx.StaticLine(
self, wx.ID_ANY, wx.DefaultPosition, wx.Size(-1, 2), wx.LI_HORIZONTAL
)
self.m_staticline212.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_INACTIVECAPTION)
)
self.m_staticline212.SetMaxSize(wx.Size(80, -1))
bSizer11.Add(self.m_staticline212, 0, wx.EXPAND | wx.ALL, 5)
self.m_staticText111 = wx.StaticText(
self, wx.ID_ANY, "当前状态", wx.DefaultPosition, wx.Size(-1, 30), 0
)
self.m_staticText111.Wrap(-1)
self.m_staticText111.SetFont(wx.Font(11, 70, 90, 90, False, "宋体"))
self.m_staticText111.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_WINDOWTEXT)
)
bSizer11.Add(self.m_staticText111, 0, wx.ALL, 5)
fgSizer1.Add(bSizer11, 1, wx.EXPAND, 5)
bSizer1 = wx.BoxSizer(wx.HORIZONTAL)
bSizer2 = wx.BoxSizer(wx.VERTICAL)
self.m_staticText = wx.StaticText(
self, wx.ID_ANY, "安全多方平台服务器", wx.DefaultPosition, wx.Size(-1, 30), 0
)
self.m_staticText.Wrap(-1)
self.m_staticText.SetFont(wx.Font(16, 70, 90, 92, False, "宋体"))
self.m_staticText.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_WINDOWTEXT)
)
bSizer2.Add(self.m_staticText, 0, wx.ALL, 5)
self.m_staticline1 = wx.StaticLine(
self, wx.ID_ANY, wx.DefaultPosition, wx.Size(1, 2), wx.LI_HORIZONTAL
)
self.m_staticline1.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_INACTIVECAPTION)
)
bSizer2.Add(self.m_staticline1, 0, wx.EXPAND | wx.ALL, 5)
self.m_listCtrl1 = wx.ListCtrl(
self, wx.ID_ANY, wx.DefaultPosition, wx.Size(360, 150), wx.LC_REPORT
)
self.m_listCtrl1.SetFont(wx.Font(9, 70, 90, 90, False, "宋体"))
self.m_listCtrl1.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_WINDOWTEXT)
)
bSizer2.Add(self.m_listCtrl1, 0, wx.ALL, 5)
self.m_staticline2 = wx.StaticLine(
self, wx.ID_ANY, wx.DefaultPosition, wx.Size(-1, 2), wx.LI_HORIZONTAL
)
self.m_staticline2.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_INACTIVECAPTION)
)
bSizer2.Add(self.m_staticline2, 0, wx.EXPAND | wx.ALL, 5)
self.m_listCtrl2 = wx.ListCtrl(
self, wx.ID_ANY, wx.DefaultPosition, wx.Size(360, 150), wx.LC_REPORT
)
self.m_listCtrl2.SetFont(wx.Font(9, 70, 90, 90, False, "宋体"))
self.m_listCtrl2.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_WINDOWTEXT)
)
bSizer2.Add(self.m_listCtrl2, 0, wx.ALL, 5)
self.m_staticline22 = wx.StaticLine(
self, wx.ID_ANY, wx.DefaultPosition, wx.Size(-1, 2), wx.LI_HORIZONTAL
)
self.m_staticline22.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_INACTIVECAPTION)
)
bSizer2.Add(self.m_staticline22, 0, wx.EXPAND | wx.ALL, 5)
self.m_listCtrl3 = wx.ListCtrl(
self, wx.ID_ANY, wx.DefaultPosition, wx.Size(360, 150), wx.LC_REPORT
)
self.m_listCtrl3.SetFont(wx.Font(9, 70, 90, 90, False, "宋体"))
self.m_listCtrl3.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_WINDOWTEXT)
)
bSizer2.Add(self.m_listCtrl3, 0, wx.ALL, 5)
self.m_staticline221 = wx.StaticLine(
self, wx.ID_ANY, wx.DefaultPosition, wx.Size(-1, 2), wx.LI_HORIZONTAL
)
self.m_staticline221.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_INACTIVECAPTION)
)
bSizer2.Add(self.m_staticline221, 0, wx.EXPAND | wx.ALL, 5)
bSizer12 = wx.BoxSizer(wx.VERTICAL)
bSizer102 = wx.BoxSizer(wx.HORIZONTAL)
self.m_button31 = wx.Button(
self, wx.ID_ANY, "显示本地公钥", wx.DefaultPosition, wx.Size(300, 30), 0
)
self.m_button31.SetFont(wx.Font(11, 70, 90, 90, False, wx.EmptyString))
self.m_button31.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_WINDOWTEXT)
)
bSizer102.Add(self.m_button31, 0, wx.ALIGN_CENTER | wx.ALL, 5)
bSizer12.Add(bSizer102, 1, wx.EXPAND, 5)
bSizer1012 = wx.BoxSizer(wx.HORIZONTAL)
self.m_button41 = wx.Button(
self, wx.ID_ANY, "显示本地私钥", wx.DefaultPosition, wx.Size(300, 30), 0
)
self.m_button41.SetFont(wx.Font(11, 70, 90, 90, False, wx.EmptyString))
self.m_button41.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_WINDOWTEXT)
)
bSizer1012.Add(self.m_button41, 0, wx.ALIGN_CENTER | wx.ALL, 5)
bSizer12.Add(bSizer1012, 1, wx.EXPAND, 5)
bSizer10111 = wx.BoxSizer(wx.HORIZONTAL)
self.m_button51 = wx.Button(
self, wx.ID_ANY, "打印本地证书", wx.DefaultPosition, wx.Size(300, 30), 0
)
self.m_button51.SetFont(wx.Font(11, 70, 90, 90, False, wx.EmptyString))
self.m_button51.SetForegroundColour(
wx.SystemSettings.GetColour(wx.SYS_COLOUR_WINDOWTEXT)
)
bSizer10111.Add(self.m_button51, 0, wx.ALIGN_CENTER | wx.ALL, 5)
bSizer12.Add(bSizer10111, 1, wx.EXPAND, 5)
bSizer2.Add(bSizer12, 1, wx.EXPAND, 5)
bSizer1.Add(bSizer2, 1, wx.EXPAND, 5)
fgSizer1.Add(bSizer1, 1, wx.EXPAND, 5)
self.SetSizer(fgSizer1)
self.Layout()
self.Centre(wx.BOTH)
# Connect Events
self.m_button31.Bind(wx.EVT_BUTTON, self.print_public_key)
self.m_button41.Bind(wx.EVT_BUTTON, self.print_private_key)
self.m_button51.Bind(wx.EVT_BUTTON, self.print_certificate)
def __del__(self):
pass
# Virtual event handlers, override them in your derived class
def print_public_key(self, event):
event.Skip()
def print_private_key(self, event):
event.Skip()
def print_certificate(self, event):
event.Skip()