From 55add91467fc55bf6090254494c6f74f556b96e5 Mon Sep 17 00:00:00 2001 From: dqy <1016751306@qq.com> Date: Mon, 23 Oct 2023 17:12:18 +0800 Subject: [PATCH 1/5] =?UTF-8?q?style:=20=E7=BE=8E=E5=8C=96=E4=BB=A3?= =?UTF-8?q?=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/server.py | 68 ++++++++++++++++++++++++++++++++------------------- 1 file changed, 43 insertions(+), 25 deletions(-) diff --git a/src/server.py b/src/server.py index c4b8fa3..0d0b1a6 100644 --- a/src/server.py +++ b/src/server.py @@ -6,33 +6,40 @@ import sqlite3 import asyncio import time + @asynccontextmanager async def lifespan(app: FastAPI): init() yield clean_env() -app = FastAPI(lifespan = lifespan) + +app = FastAPI(lifespan=lifespan) # 连接到数据库（如果数据库不存在，则会自动创建） -conn = sqlite3.connect('server.db') +conn = sqlite3.connect("server.db") # 创建游标对象，用于执行SQL语句 cursor = conn.cursor() # 创建表: id: int; ip: TEXT -cursor.execute('''CREATE TABLE IF NOT EXISTS nodes ( +cursor.execute( + """CREATE TABLE IF NOT EXISTS nodes ( id INTEGER PRIMARY KEY AUTOINCREMENT, ip TEXT NOT NULL, last_heartbeat INTEGER - )''') + )""" +) + def init(): asyncio.create_task(receive_heartbeat_internal()) + def clean_env(): # 关闭游标和连接 cursor.close() conn.close() + @app.get("/server/show_nodes") async def show_nodes() -> list: nodes_list = [] @@ -43,37 +50,42 @@ async def show_nodes() -> list: nodes_list.append(row) return nodes_list + @app.get("/server/get_node") async def get_node(ip: str) -> int: - ''' - 中心服务器与节点交互, 节点发送ip, 中心服务器接收ip存入数据库并将ip转换为int作为节点id返回给节点 - params: - ip: node ip - return: - id: ip按点分割成四部分, 每部分转二进制后拼接再转十进制作为节点id - ''' + """ + 中心服务器与节点交互, 节点发送ip, 中心服务器接收ip存入数据库并将ip转换为int作为节点id返回给节点 + params: + ip: node ip + return: + id: ip按点分割成四部分, 每部分转二进制后拼接再转十进制作为节点id + """ ip_parts = ip.split(".") ip_int = 0 for i in range(4): ip_int += int(ip_parts[i]) << (24 - (8 * i)) - + # 获取当前时间 current_time = int(time.time()) # 插入数据 - cursor.execute("INSERT INTO nodes (id, ip, last_heartbeat) VALUES (?, ?, ?)", (ip_int, ip, current_time)) + cursor.execute( + "INSERT INTO nodes (id, ip, last_heartbeat) VALUES (?, ?, ?)", + (ip_int, ip, current_time), + ) conn.commit() return ip_int + @app.get("/server/delete_node") async def delete_node(ip: str) -> None: - ''' + """ param: ip: 待删除节点的ip地址 return: None - ''' + """ # 查询要删除的节点 cursor.execute("SELECT * FROM nodes WHERE ip=?", (ip,)) row = cursor.fetchone() @@ -85,12 +97,16 @@ async def delete_node(ip: str) -> None: else: print(f"Node with IP {ip} not found.") + # 接收节点心跳包 @app.post("/server/heartbeat") async def receive_heartbeat(ip: str): - cursor.execute("UPDATE nodes SET last_heartbeat = ? WHERE ip = ?", (time.time(), ip)) - return {"status": "received"} - + cursor.execute( + "UPDATE nodes SET last_heartbeat = ? WHERE ip = ?", (time.time(), ip) + ) + return {"status": "received"} + + async def receive_heartbeat_internal(): while 1: timeout = 70 @@ -99,33 +115,35 @@ async def receive_heartbeat_internal(): # conn.commit() await asyncio.sleep(timeout) + @app.get("/server/send_nodes_list") async def send_nodes_list(count: int) -> list: - ''' + """ 中心服务器与客户端交互, 客户端发送所需节点个数, 中心服务器从数据库中顺序取出节点封装成list格式返回给客户端 - params: - count: 所需节点个数 - return: + params: + count: 所需节点个数 + return: nodes_list: list - ''' + """ nodes_list = [] # 查询数据库中的节点数据 cursor.execute("SELECT * FROM nodes LIMIT ?", (count,)) rows = cursor.fetchall() - for row in rows: id, ip, last_heartbeat = row nodes_list.append(ip) return nodes_list + @app.get("/server/clear_database") async def clear_database() -> None: cursor.execute("DELETE FROM nodes") conn.commit() + if __name__ == "__main__": import uvicorn # pylint: disable=e0401 - uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=True) \ No newline at end of file + uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=True) From a45cd66e9d418dd601226d9dea16c70f3f8d358e Mon Sep 17 00:00:00 2001 From: dqy <1016751306@qq.com> Date: Mon, 23 Oct 2023 17:54:30 +0800 Subject: [PATCH 2/5] modified: src/client.ini modified: src/client.py modified: src/node.py --- src/client.ini | 2 +- src/client.py | 47 ++++++++++++++++++++++-------------------- src/node.py | 56 ++++++++++++++++++++++++-------------------------- 3 files changed, 53 insertions(+), 52 deletions(-) diff --git a/src/client.ini b/src/client.ini index 0725142..23e21d2 100644 --- a/src/client.ini +++ b/src/client.ini @@ -1,3 +1,3 @@ [settings] -server_address = 10.20.127.226:8000 +server_address = 10.20.14.232:8000 version = 1.0 diff --git a/src/client.py b/src/client.py index 6ace678..210965d 100644 --- a/src/client.py +++ b/src/client.py @@ -17,6 +17,7 @@ async def lifespan(app: FastAPI): yield clean_env() + app = FastAPI(lifespan=lifespan) @@ -25,7 +26,7 @@ def init(): init_db() pk, sk = GenerateKeyPair() init_config() - get_node_list(6, server_address) # type: ignore + get_node_list(2, server_address) # type: ignore def init_db(): @@ -93,6 +94,7 @@ class C(BaseModel): Tuple: Tuple[capsule, int] ip: str + # receive messages from node @app.post("/receive_messages") async def receive_messages(message: C): @@ -170,12 +172,9 @@ async def check_merge(db, ct: int, ip: str): # send message to node -async def send_messages(node_ips: tuple[str, ...], - message: bytes, - dest_ip: str, - pk_B: point, - shreshold: int - ): +async def send_messages( + node_ips: tuple[str, ...], message: bytes, dest_ip: str, pk_B: point, shreshold: int +): global pk, sk id_list = [] for node_ip in node_ips: @@ -184,14 +183,14 @@ async def send_messages(node_ips: tuple[str, ...], for i in range(4): id += int(ip_parts[i]) << (24 - (8 * i)) id_list.append(id) - rk_list = GenerateReKey(sk, pk_B, len(node_ips), shreshold, tuple(id_list)) # type: ignore + rk_list = GenerateReKey(sk, pk_B, len(node_ips), shreshold, tuple(id_list)) # type: ignore for i in range(len(node_ips)): url = "http://" + node_ips[i] + ":8001" + "/recieve_message" payload = { "source_ip": local_ip, "dest_ip": dest_ip, "message": message, - "rk": rk_list[i] + "rk": rk_list[i], } response = requests.post(url, json=payload) return 0 @@ -203,6 +202,7 @@ class IP_Message(BaseModel): source_ip: str pk: int + # request message from others @app.post("/request_message") async def request_message(i_m: IP_Message): @@ -212,11 +212,12 @@ async def request_message(i_m: IP_Message): source_ip = get_own_ip() dest_port = "8003" url = "http://" + dest_ip + dest_port + "/recieve_request" - payload = {"dest_ip": dest_ip, - "message_name": message_name, - "source_ip": source_ip, - "pk": pk - } + payload = { + "dest_ip": dest_ip, + "message_name": message_name, + "source_ip": source_ip, + "pk": pk, + } response = requests.post(url, json=payload) if response.status_code == 200: data = response.json() @@ -224,7 +225,7 @@ async def request_message(i_m: IP_Message): threshold = int(data["threshold"]) with sqlite3.connect("client.db") as db: db.execute( - """ + """ INSERT INTO senderinfo (public_key, threshold) VALUES @@ -255,17 +256,20 @@ async def recieve_request(i_m: IP_Message): threshold = random.randrange(1, 6) own_public_key = pk pk_B = i_m.pk - + with sqlite3.connect("client.db") as db: - cursor = db.execute(""" + cursor = db.execute( + """ SELECT nodeip FROM node LIMIT ? - """,(threshold,)) + """, + (threshold,), + ) node_ips = cursor.fetchall() message = b"hello world" + random.randbytes(8) - await send_messages(node_ips, message, dest_ip, pk_B, threshold) # type: ignore - response = {"threshold": threshold,"public_key": own_public_key} + await send_messages(node_ips, message, dest_ip, pk_B, threshold) # type: ignore + response = {"threshold": threshold, "public_key": own_public_key} return response @@ -303,7 +307,6 @@ def get_node_list(count: int, server_addr: str): print("Failed:", response.status_code, response.text) - pk = point sk = int server_address = str @@ -314,4 +317,4 @@ local_ip = get_own_ip() if __name__ == "__main__": import uvicorn # pylint: disable=e0401 - uvicorn.run("client:app", host="0.0.0.0", port=8003, reload=True) + uvicorn.run("client:app", host="0.0.0.0", port=8002, reload=True) diff --git a/src/node.py b/src/node.py index 7178678..2fca97d 100644 --- a/src/node.py +++ b/src/node.py @@ -1,4 +1,4 @@ -from fastapi import FastAPI,Request +from fastapi import FastAPI, Request import requests from contextlib import asynccontextmanager import socket @@ -6,6 +6,7 @@ import asyncio from pydantic import BaseModel from tpre import * + @asynccontextmanager async def lifespan(app: FastAPI): # Load the ML model @@ -14,25 +15,28 @@ async def lifespan(app: FastAPI): # Clean up the ML models and release the resources clear() + app = FastAPI(lifespan=lifespan) -server_address ="http://中心服务器IP地址/server" +server_address = "http://中心服务器IP地址/server" id = 0 -ip = '' -client_ip_src = '' # 发送信息用户的ip -client_ip_des = '' # 接收信息用户的ip -processed_message = () # 重加密后的数据 +ip = "" +client_ip_src = "" # 发送信息用户的ip +client_ip_des = "" # 接收信息用户的ip +processed_message = () # 重加密后的数据 # class C(BaseModel): # Tuple: Tuple[capsule, int] # ip_src: str + # 向中心服务器发送自己的IP地址,并获取自己的id def send_ip(): - url = server_address + '/get_node?ip = ' + ip + url = server_address + "/get_node?ip = " + ip # ip = get_local_ip # type: ignore global id id = requests.get(url) + # 用socket获取本机ip def get_local_ip(): # 创建一个套接字对象 @@ -42,7 +46,7 @@ def get_local_ip(): # 获取本地IP地址 local_ip = s.getsockname()[0] s.close() - global ip + global ip ip = local_ip @@ -51,52 +55,46 @@ def init(): global id send_ip() task = asyncio.create_task(send_heartbeat_internal()) -def clear(): + +def clear(): pass + # 接收用户发来的消息，经过处理之后，再将消息发送给其他用户 + async def send_heartbeat_internal() -> None: while True: # print('successful send my_heart') - global ip - url = server_address + '/get_node?ip = ' + ip + global ip + url = server_address + "/get_node?ip = " + ip folderol = requests.get(url) timeout = 30 # 删除超时的节点（假设你有一个异步的数据库操作函数） await asyncio.sleep(timeout) - -@app.post("/user_src") # 接收用户1发送的信息 +@app.post("/user_src") # 接收用户1发送的信息 async def receive_user_src_message(message: Request): json_data = await message.json() - global client_ip_src,client_ip_des + global client_ip_src, client_ip_des # kfrag , capsule_ct ,client_ip_src , client_ip_des = json_data[] # 看梁俊勇 global processed_message - processed_message = ReEncrypt(kfrag, capsule_ct) + processed_message = ReEncrypt(kfrag, capsule_ct) +def send_user_des_message(): # 发送消息给用户2 + global processed_message, client_ip_src, client_ip_des -def send_user_des_message(): # 发送消息给用户2 - global processed_message,client_ip_src,client_ip_des + data = {"Tuple": processed_message, "ip": client_ip_src} # 类型不匹配 - data = { - "Tuple": processed_message, # 类型不匹配 - "ip": client_ip_src -} - -# 发送 HTTP POST 请求 - response = requests.post("http://"+ client_ip_des + "/receive_messages", json=data) + # 发送 HTTP POST 请求 + response = requests.post("http://" + client_ip_des + "/receive_messages", json=data) print(response) if __name__ == "__main__": import uvicorn # pylint: disable=e0401 - uvicorn.run("node:app", host="0.0.0.0", port=8000, reload=True) - - - - + uvicorn.run("node:app", host="0.0.0.0", port=8001, reload=True) From 2271369d2e70173f868e216528ac62088cdbf367 Mon Sep 17 00:00:00 2001 From: dqy <1016751306@qq.com> Date: Tue, 24 Oct 2023 19:23:34 +0800 Subject: [PATCH 3/5] feat:change serveraddress --- src/node.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/node.py b/src/node.py index 21aad78..8e373c2 100644 --- a/src/node.py +++ b/src/node.py @@ -15,9 +15,9 @@ async def lifespan(app: FastAPI): app = FastAPI(lifespan=lifespan) -server_address = "http://10.20.14.232:8000/server" +server_address = "http://110.41.155.96:8000/server" id = 0 -ip = "10.16.21.163" +ip = "" client_ip_src = "" # 发送信息用户的ip client_ip_des = "" # 接收信息用户的ip processed_message = () # 重加密后的数据 @@ -73,7 +73,7 @@ async def send_heartbeat_internal() -> None: folderol = requests.get(url) except: print("Central server error") - + # 删除超时的节点（假设你有一个异步的数据库操作函数） await asyncio.sleep(timeout) From 452253111b9210c9c2f5733b16141e43a7594e12 Mon Sep 17 00:00:00 2001 From: dqy <1016751306@qq.com> Date: Tue, 24 Oct 2023 20:21:41 +0800 Subject: [PATCH 4/5] test: test --- src/node.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/node.py b/src/node.py index 4ee9231..99ff06c 100644 --- a/src/node.py +++ b/src/node.py @@ -33,7 +33,7 @@ def send_ip(): url = server_address + "/get_node?ip=" + ip # ip = get_local_ip() # type: ignore global id - id = requests.get(url,timeout=3) + id = requests.get(url, timeout=3) # 用环境变量获取本机ip @@ -63,7 +63,7 @@ async def send_heartbeat_internal() -> None: while True: # print('successful send my_heart') try: - folderol = requests.get(url,timeout=3) + folderol = requests.get(url, timeout=3) except: print("Central server error") @@ -108,4 +108,4 @@ async def send_user_des_message(source_ip: str, dest_ip: str, re_message): # if __name__ == "__main__": import uvicorn # pylint: disable=e0401 - uvicorn.run("node:app", host="0.0.0.0", port=8001, reload=True) + uvicorn.run("node:app", host="0.0.0.0", port=8001, reload=False) From 63933a1d19964f3a67f5044191e899f88a1251e8 Mon Sep 17 00:00:00 2001 From: dqy <1016751306@qq.com> Date: Tue, 24 Oct 2023 21:00:19 +0800 Subject: [PATCH 5/5] test: threshold set to 2 --- src/client.py | 29 ++++++++++++++--------------- 1 file changed, 14 insertions(+), 15 deletions(-) diff --git a/src/client.py b/src/client.py index f3e757c..ca8be98 100644 --- a/src/client.py +++ b/src/client.py @@ -96,6 +96,7 @@ class C(BaseModel): Tuple: Tuple[capsule, int] ip: str + # receive messages from nodes @app.post("/receive_messages") async def receive_messages(message: C): @@ -144,7 +145,7 @@ async def receive_messages(message: C): async def check_merge(ct: int, ip: str): global sk, pk, node_response, message with sqlite3.connect("client.db") as db: - # Check if the combination of ct_column and ip_column appears more than once. + # Check if the combination of ct_column and ip_column appears more than once. cursor = db.execute( """ SELECT capsule, ct @@ -166,8 +167,8 @@ async def check_merge(ct: int, ip: str): (ip), ) result = cursor.fetchall() - pk_sender, T = result[0] # result[0] = (pk, threshold) - + pk_sender, T = result[0] # result[0] = (pk, threshold) + if len(cfrag_cts) >= T: cfrags = mergecfrag(cfrag_cts) message = DecryptFrags(sk, pk, pk_sender, cfrags) # type: ignore @@ -180,6 +181,7 @@ async def send_messages( ): global pk, sk id_list = [] + print(node_ips) # calculate id of nodes for node_ip in node_ips: ip_parts = node_ip.split(".") @@ -187,10 +189,10 @@ async def send_messages( for i in range(4): id += int(ip_parts[i]) << (24 - (8 * i)) id_list.append(id) - + # generate rk rk_list = GenerateReKey(sk, pk_B, len(node_ips), shreshold, tuple(id_list)) # type: ignore - + capsule_ct = Encrypt(pk, message) # type: ignore for i in range(len(node_ips)): @@ -255,9 +257,6 @@ async def request_message(i_m: Request_Message): """, (public_key, threshold), ) - - - try: if response.status_code == 200: @@ -282,7 +281,7 @@ async def request_message(i_m: Request_Message): for _ in range(10): if node_response: data = message - + # reset message and node_response message = b"" node_response = False @@ -301,7 +300,8 @@ async def recieve_request(i_m: IP_Message): if source_ip != i_m.dest_ip: return HTTPException(status_code=400, detail="Wrong ip") dest_ip = i_m.source_ip - threshold = random.randrange(1, 2) + # threshold = random.randrange(1, 2) + threshold = 2 own_public_key = pk pk_B = i_m.pk @@ -315,18 +315,17 @@ async def recieve_request(i_m: IP_Message): (threshold,), ) node_ips = cursor.fetchall() - + # message name message = b"hello world" + random.randbytes(8) - + # send message to nodes - await send_messages(tuple(node_ips), message, dest_ip, pk_B, threshold) + await send_messages(tuple(node_ips), message, dest_ip, pk_B, threshold) response = {"threshold": threshold, "public_key": own_public_key} return response def get_own_ip() -> str: - ip = os.environ.get("HOST_IP", "IP not set") return ip @@ -334,7 +333,7 @@ def get_own_ip() -> str: # get node list from central server def get_node_list(count: int, server_addr: str): url = "http://" + server_addr + "/server/send_nodes_list?count=" + str(count) - response = requests.get(url,timeout=3) + response = requests.get(url, timeout=3) # Checking the response if response.status_code == 200: print("Success get node list")