ccyj/mimajingsai
1
0
Fork 0
forked from sangge/tpre-python
Code Pull Requests Actions Activity

Merge pull request 'main' (#20) from sangge/mimajingsai:main into main

Browse Source 
Reviewed-on: #20
This commit is contained in:
ccyj 2023-10-23 17:47:52 +08:00
commit 0ce7f79d5f
7 changed files with 228 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
README_en.md
View File

@ -21,6 +21,7 @@ The project uses the Chinese national standard cryptography algorithm to impleme
## Environment Dependencies
### Bare mental version(UNTESTED)
System requirements:
- Linux
- Windows(may need to complie and install gmssl yourself)
@ -30,22 +31,24 @@ The project relies on the following software:
- gmssl
- gmssl-python
### Docker version
docker version:
- Version: 24.0.5
- API version: 1.43
- Go version: go1.20.6
## Installation Steps
### Pre-installation
This project depends on gmssl, so you need to compile it from source first.
Visit [GmSSL](https://github.com/guanzhi/GmSSL) to learn how to install.
Then install essential python libs
```bash
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
```
## Docker Installation
my docker version:
- Version: 24.0.5
- API version: 1.43
- Go version: go1.20.6
### Use base image and build yourself
```bash

2
src/client.ini
View File

@ -1,3 +1,3 @@
[settings]
server_address = "127.0.0.1:8000"
server_address = 10.20.127.226:8000
version = 1.0

204
src/client.py
View File

@ -20,18 +20,15 @@ async def lifespan(app: FastAPI):
app = FastAPI(lifespan=lifespan)
pk = point
sk = int
server_address = str
node_response = False
message = bytes
def init():
global pk, sk, server_address
init_db()
pk, sk = GenerateKeyPair()
# load config from config file
init_config()
# get_node_list(6, server_address) # type: ignore
@ -100,14 +97,13 @@ class C(BaseModel):
Tuple: Tuple[capsule, int]
ip: str
# receive messages from node
# receive messages from nodes
@app.post("/receive_messages")
async def receive_messages(message: C):
"""
receive capsule and ip from nodes
params:
C: capsule and ct
Tuple: capsule and ct
ip: sender ip
return:
status_code
@ -137,7 +133,7 @@ async def receive_messages(message: C):
(C_capsule, C_ct, ip),
)
db.commit()
await check_merge(db, C_ct, ip)
await check_merge(C_ct, ip)
return HTTPException(status_code=200, detail="Message received")
except Exception as e:
print(f"Error occurred: {e}")
@ -146,31 +142,33 @@ async def receive_messages(message: C):
# check record count
async def check_merge(db, ct: int, ip: str):
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.
cursor = db.execute(
"""
SELECT capsule, ct
FROM message
WHERE ct = ? AND senderip = ?
""",
(ct, ip),
)
# [(capsule, ct), ...]
cfrag_cts = cursor.fetchall()
cursor = db.execute(
"""
SELECT capsule, ct
FROM message
WHERE ct = ? AND senderip = ?
""",
(ct, ip),
)
# [(capsule, ct), ...]
cfrag_cts = cursor.fetchall()
# get N
cursor = db.execute(
"""
SELECT publickey, threshold
FROM senderinfo
WHERE senderip = ?
""",
(ip),
)
result = cursor.fetchall()
pk_sender, T = result[0]
# get T
cursor = db.execute(
"""
SELECT publickey, threshold
FROM senderinfo
WHERE senderip = ?
""",
(ip),
)
result = cursor.fetchall()
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
@ -178,7 +176,36 @@ async def check_merge(db, ct: int, ip: str):
# send message to node
def send_message(ip: tuple[str, ...]):
async def send_messages(
node_ips: tuple[str, ...], message: bytes, dest_ip: str, pk_B: point, shreshold: int
):
global pk, sk
id_list = []
# calculate id of nodes
for node_ip in node_ips:
ip_parts = node_ip.split(".")
id = 0
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)):
url = "http://" + node_ips[i] + ":8001" + "/user_src?message"
payload = {
"source_ip": local_ip,
"dest_ip": dest_ip,
"capsule_ct": capsule_ct,
"rk": rk_list[i],
}
response = requests.post(url, json=payload)
if response.status_code == 200:
print(f"send to {node_ips[i]} successful")
return 0
@ -186,42 +213,69 @@ class IP_Message(BaseModel):
dest_ip: str
message_name: str
source_ip: str
pk: int
class Request_Message(BaseModel):
dest_ip: str
message_name: str
# request message from others
@app.post("/request_message")
async def request_message(i_m: IP_Message):
global message, node_response
async def request_message(i_m: Request_Message):
global message, node_response, pk
dest_ip = i_m.dest_ip
# dest_ip = dest_ip.split(":")[0]
message_name = i_m.message_name
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}
response = requests.post(url, json=payload)
if response.status_code == 200:
data = response.json()
public_key = int(data["public_key"])
threshold = int(data["threshold"])
with sqlite3.connect("client.db") as db:
db.execute(
"""
INSERT INTO senderinfo
(public_key, threshold)
VALUES
(?, ?)
""",
(public_key, threshold),
)
url = "http://" + dest_ip + ":" + dest_port + "/recieve_request?i_m"
payload = {
"dest_ip": dest_ip,
"message_name": message_name,
"source_ip": source_ip,
"pk": pk,
}
try:
response = requests.post(url, json=payload)
# wait to recieve message from nodes
except:
print("can't post")
return {"message": "can't post"}
try:
if response.status_code == 200:
data = response.json()
public_key = int(data["public_key"])
threshold = int(data["threshold"])
with sqlite3.connect("client.db") as db:
db.execute(
"""
INSERT INTO senderinfo
(public_key, threshold)
VALUES
(?, ?)
""",
(public_key, threshold),
)
except:
print("Database error")
return {"message": "Database Error"}
# wait 10s to recieve message from nodes
for _ in range(10):
if node_response:
data = message
# reset message and node_response
message = b""
node_response = False
# return message to frontend
return {"message": data}
time.sleep(1)
return {"message": "recieve timeout"}
# recieve request from others
@ -233,8 +287,26 @@ async def recieve_request(i_m: IP_Message):
return HTTPException(status_code=400, detail="Wrong ip")
dest_ip = i_m.source_ip
threshold = random.randrange(1, 6)
public_key = pk
response = {"threshold": threshold,"public_key": public_key}
own_public_key = pk
pk_B = i_m.pk
with sqlite3.connect("client.db") as db:
cursor = db.execute(
"""
SELECT nodeip
FROM node
LIMIT ?
""",
(threshold,),
)
node_ips = cursor.fetchall()
# message name
message = b"hello world" + random.randbytes(8)
# send message to nodes
await send_messages(node_ips, message, dest_ip, pk_B, threshold) # type: ignore
response = {"threshold": threshold, "public_key": own_public_key}
return response
@ -246,22 +318,23 @@ 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"
payload = {"count": count}
response = requests.post(url, json=payload)
url = "http://" + server_addr + "/server/send_nodes_list?count=" + str(count)
response = requests.get(url)
# Checking the response
if response.status_code == 200:
print("Success get node list")
node_ip = response.text
node_ip = eval(node_ip)
print(node_ip)
# insert node ip to database
with sqlite3.connect("client.db") as db:
db.executemany(
"""
INSERT INTO node
nodeip
VALUE (?)
(nodeip)
VALUES (?)
""",
node_ip,
[(ip,) for ip in node_ip],
)
db.commit()
print("Success add node ip")
@ -269,7 +342,14 @@ def get_node_list(count: int, server_addr: str):
print("Failed:", response.status_code, response.text)
pk = point
sk = int
server_address = str
node_response = False
message = bytes
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=8003, reload=True)

23
src/client_cli.py Normal file
View File

@ -0,0 +1,23 @@
import argparse
import requests
def send_post_request(ip_addr, message_name):
url = f"http://localhost:20234/request_message/?i_m"
data = {
"dest_ip": ip_addr,
"message_name": message_name
}
response = requests.post(url, json=data)
return response.text
def main():
parser = argparse.ArgumentParser(description="Send POST request to a specified IP.")
parser.add_argument("ip_addr", help="IP address to send request to.")
parser.add_argument("message_name", help="Message name to send.")
args = parser.parse_args()
response = send_post_request(args.ip_addr, args.message_name)
print(response)
if __name__ == "__main__":
main()

3
src/demo.py
View File

@ -24,7 +24,8 @@ T = 5
# 5
start_time = time.time()
rekeys = GenerateReKey(sk_a, pk_b, N, T)
id_tuple = tuple(range(N))
rekeys = GenerateReKey(sk_a, pk_b, N, T, id_tuple)
end_time = time.time()
elapsed_time = end_time - start_time
print(f"代码块5运行时间:{elapsed_time}秒")

78
src/node.py
View File

@ -1,4 +1,4 @@
from fastapi import FastAPI,Request
from fastapi import FastAPI, Request, HTTPException
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,44 +55,56 @@ 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)
def send_user_des_message(): # 发送消息给用户2
global processed_message,client_ip_src,client_ip_des
data = {
"Tuple": processed_message, # 类型不匹配
"ip": client_ip_src
}
'''
payload = {
"source_ip": local_ip,
"dest_ip": dest_ip,
"capsule_ct": capsule_ct,
"rk": rk_list[i],
}
'''
# 发送 HTTP POST 请求
response = requests.post("http://"+ client_ip_des + "/receive_messages", json=data)
data = await message.json()
source_ip = data.get("source_ip")
dest_ip = data.get("dest_ip")
capsule_ct = data.get("capsule_ct")
rk = data.get("rk")
processed_message = ReEncrypt(rk, capsule_ct)
await send_user_des_message(source_ip, dest_ip, processed_message)
return HTTPException(status_code=200, detail="message recieved")
async def send_user_des_message(source_ip: str, dest_ip: str, re_message): # 发送消息给用户2
data = {"Tuple": re_message, "ip": source_ip} # 类型不匹配
# 发送 HTTP POST 请求
response = requests.post("http://" + dest_ip + "/receive_messages?message", json=data)
print(response)
@ -96,7 +112,3 @@ if __name__ == "__main__":
import uvicorn # pylint: disable=e0401
uvicorn.run("node:app", host="0.0.0.0", port=8000, reload=True)

17
src/server.py
View File

@ -95,20 +95,20 @@ async def receive_heartbeat_internal():
while 1:
timeout = 70
# 删除超时的节点
cursor.execute("DELETE FROM nodes WHERE last_heartbeat < ?", (time.time() - timeout,))
conn.commit()
# cursor.execute("DELETE FROM nodes WHERE last_heartbeat < ?", (time.time() - timeout,))
# conn.commit()
await asyncio.sleep(timeout)
@app.get("/server/send_nodes_list")
async def send_nodes_list(count: int) -> JSONResponse:
async def send_nodes_list(count: int) -> list:
'''
中心服务器与客户端交互, 客户端发送所需节点个数, 中心服务器从数据库中顺序取出节点封装成json格式返回给客户端
中心服务器与客户端交互, 客户端发送所需节点个数, 中心服务器从数据库中顺序取出节点封装成list格式返回给客户端
params:
count: 所需节点个数
return:
JSONResponse: {id: ip,...}
nodes_list: list
'''
nodes_list = {}
nodes_list = []
# 查询数据库中的节点数据
cursor.execute("SELECT * FROM nodes LIMIT ?", (count,))
@ -116,10 +116,9 @@ async def send_nodes_list(count: int) -> JSONResponse:
for row in rows:
id, ip, last_heartbeat = row
nodes_list[id] = ip
nodes_list.append(ip)
json_result = jsonable_encoder(nodes_list)
return JSONResponse(content=json_result)
return nodes_list
@app.get("/server/clear_database")
async def clear_database() -> None: