20240925-0001
This commit is contained in:
@@ -0,0 +1,5 @@
|
||||
安装python 包
|
||||
> pip install Flask pandas cryptography flask-cors
|
||||
|
||||
运行
|
||||
> python client.py
|
||||
@@ -0,0 +1,125 @@
|
||||
from flask import Flask, request, jsonify
|
||||
import os
|
||||
import pandas as pd
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import padding
|
||||
import base64
|
||||
from datetime import datetime
|
||||
from flask_cors import CORS
|
||||
|
||||
app = Flask(__name__)
|
||||
CORS(app)
|
||||
|
||||
public_key_pem = None
|
||||
did = None
|
||||
gid = None
|
||||
id = None
|
||||
|
||||
def read_public_key_from_file(public_key_str):
|
||||
return public_key_str.encode('utf-8')
|
||||
|
||||
def get_current_time():
|
||||
return datetime.now()
|
||||
|
||||
@app.route('/endpoint', methods=['POST'])
|
||||
def initialize():
|
||||
global public_key_pem, did, gid, id
|
||||
data = request.get_json()
|
||||
serial_number = data.get('serialNumber', '').strip()
|
||||
|
||||
csv_path = os.path.join(os.path.dirname(__file__), 'Authorize_DataBase.csv')
|
||||
|
||||
if not os.path.exists(csv_path):
|
||||
return jsonify({'error': 'Authorize_DataBase.csv 文件不存在'}), 404
|
||||
|
||||
df = pd.read_csv(csv_path)
|
||||
print("读取的设备号:", df['ID'].values)
|
||||
print("请求的设备号:", serial_number)
|
||||
|
||||
if serial_number not in df['ID'].astype(str).values:
|
||||
return jsonify({'error': '设备号未找到'}), 404
|
||||
|
||||
# 获取正确的ID值
|
||||
id = int(serial_number)
|
||||
|
||||
public_key_pem = df.loc[df['ID'].astype(str) == serial_number, 'PublicKey'].values[0].encode('utf-8')
|
||||
did = str(df.loc[df['ID'].astype(str) == serial_number, 'DID'].values[0]) # 将 DID 转换为字符串
|
||||
gid = str(df.loc[df['ID'].astype(str) == serial_number, 'GID'].values[0]) # 将 GID 转换为字符串
|
||||
|
||||
return jsonify({'message': '公钥和相关信息已加载', 'publicKey': public_key_pem.decode('utf-8'), 'DID': did, 'GID': gid})
|
||||
|
||||
@app.route('/verify', methods=['POST'])
|
||||
def handle_verification():
|
||||
global public_key_pem, did, gid, id
|
||||
|
||||
if public_key_pem is None or id is None or did is None or gid is None:
|
||||
return jsonify({'error': '请先初始化公钥和相关信息'}), 400
|
||||
|
||||
data = request.get_json()
|
||||
signature_data = data.get('signatureData', '')
|
||||
|
||||
print("Received signature data:", signature_data)
|
||||
|
||||
if not signature_data:
|
||||
return jsonify({'error': '缺少授权码'}), 400
|
||||
|
||||
is_valid, message, time_range, current_time = verify_signature(public_key_pem, signature_data)
|
||||
|
||||
print("公钥:", public_key_pem)
|
||||
print("验证结果:", is_valid)
|
||||
print("从授权码获取的时间范围:", time_range)
|
||||
print("当前时间:", current_time)
|
||||
|
||||
return jsonify({
|
||||
"valid": is_valid,
|
||||
"message": message["time_range"],
|
||||
"current_time": current_time,
|
||||
"signatureData": signature_data
|
||||
})
|
||||
|
||||
def verify_signature(public_key_pem, signature_data):
|
||||
try:
|
||||
public_key = serialization.load_pem_public_key(public_key_pem)
|
||||
|
||||
parts = signature_data.split('&&')
|
||||
if len(parts) != 3:
|
||||
print("授权码格式错误:", parts)
|
||||
return False, {"valid": False, "time_range": "授权码格式错误"}, None, None
|
||||
|
||||
signature_base64, start_timestamp_base64, end_timestamp_base64 = parts
|
||||
|
||||
start_timestamp_str = base64.b64decode(start_timestamp_base64).decode('utf-8').rstrip() # 移除可能的末尾空格
|
||||
end_timestamp_str = base64.b64decode(end_timestamp_base64).decode('utf-8').rstrip() # 移除可能的末尾空格
|
||||
|
||||
signature_bytes = base64.b64decode(signature_base64)
|
||||
|
||||
# 构建用于验证的原始消息
|
||||
original_message = f"{start_timestamp_str}&&{end_timestamp_str}&&{id}&&{did}&&{gid}".encode('utf-8')
|
||||
|
||||
# 打印原始消息以便调试
|
||||
print("Original Message:", original_message)
|
||||
|
||||
# 尝试验证签名
|
||||
public_key.verify(
|
||||
signature_bytes,
|
||||
original_message,
|
||||
padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH),
|
||||
hashes.SHA256()
|
||||
)
|
||||
|
||||
# 获取时间并进行比较
|
||||
start_time = datetime.strptime(start_timestamp_str, '%Y-%m-%d')
|
||||
end_time = datetime.strptime(end_timestamp_str, '%Y-%m-%d')
|
||||
current_time = get_current_time()
|
||||
|
||||
if start_time <= current_time <= end_time:
|
||||
return True, {"valid": True, "time_range": f"成功。时间范围: {start_time} 到 {end_time}"}, (start_time, end_time), current_time
|
||||
else:
|
||||
return False, {"valid": False, "time_range": f"时间不在有效范围内。时间范围: {start_time} 到 {end_time}"}, (start_time, end_time), current_time
|
||||
except Exception as e:
|
||||
# 打印异常信息以便调试
|
||||
print("发生错误:", str(e))
|
||||
return False, {"valid": False, "time_range": f"发生错误: {str(e)}"}, None, None
|
||||
|
||||
if __name__ == '__main__':
|
||||
app.run(debug=True, host='0.0.0.0', port=5001)
|
||||
+125
@@ -0,0 +1,125 @@
|
||||
from flask import Flask, request, jsonify
|
||||
import os
|
||||
import pandas as pd
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import padding
|
||||
import base64
|
||||
from datetime import datetime
|
||||
from flask_cors import CORS
|
||||
|
||||
app = Flask(__name__)
|
||||
CORS(app)
|
||||
|
||||
public_key_pem = None
|
||||
did = None
|
||||
gid = None
|
||||
id = None
|
||||
|
||||
def read_public_key_from_file(public_key_str):
|
||||
return public_key_str.encode('utf-8')
|
||||
|
||||
def get_current_time():
|
||||
return datetime.now()
|
||||
|
||||
@app.route('/endpoint', methods=['POST'])
|
||||
def initialize():
|
||||
global public_key_pem, did, gid, id
|
||||
data = request.get_json()
|
||||
serial_number = data.get('serialNumber', '').strip()
|
||||
|
||||
csv_path = os.path.join(os.path.dirname(__file__), 'Authorize_DataBase.csv')
|
||||
|
||||
if not os.path.exists(csv_path):
|
||||
return jsonify({'error': 'Authorize_DataBase.csv 文件不存在'}), 404
|
||||
|
||||
df = pd.read_csv(csv_path)
|
||||
print("读取的设备号:", df['ID'].values)
|
||||
print("请求的设备号:", serial_number)
|
||||
|
||||
if serial_number not in df['ID'].astype(str).values:
|
||||
return jsonify({'error': '设备号未找到'}), 404
|
||||
|
||||
# 获取正确的ID值
|
||||
id = int(serial_number)
|
||||
|
||||
public_key_pem = df.loc[df['ID'].astype(str) == serial_number, 'PublicKey'].values[0].encode('utf-8')
|
||||
did = str(df.loc[df['ID'].astype(str) == serial_number, 'DID'].values[0]) # 将 DID 转换为字符串
|
||||
gid = str(df.loc[df['ID'].astype(str) == serial_number, 'GID'].values[0]) # 将 GID 转换为字符串
|
||||
|
||||
return jsonify({'message': '公钥和相关信息已加载', 'publicKey': public_key_pem.decode('utf-8'), 'DID': did, 'GID': gid})
|
||||
|
||||
@app.route('/verify', methods=['POST'])
|
||||
def handle_verification():
|
||||
global public_key_pem, did, gid, id
|
||||
|
||||
if public_key_pem is None or id is None or did is None or gid is None:
|
||||
return jsonify({'error': '请先初始化公钥和相关信息'}), 400
|
||||
|
||||
data = request.get_json()
|
||||
signature_data = data.get('signatureData', '')
|
||||
|
||||
print("Received signature data:", signature_data)
|
||||
|
||||
if not signature_data:
|
||||
return jsonify({'error': '缺少授权码'}), 400
|
||||
|
||||
is_valid, message, time_range, current_time = verify_signature(public_key_pem, signature_data)
|
||||
|
||||
print("公钥:", public_key_pem)
|
||||
print("验证结果:", is_valid)
|
||||
print("从授权码获取的时间范围:", time_range)
|
||||
print("当前时间:", current_time)
|
||||
|
||||
return jsonify({
|
||||
"valid": is_valid,
|
||||
"message": message["time_range"],
|
||||
"current_time": current_time,
|
||||
"signatureData": signature_data
|
||||
})
|
||||
|
||||
def verify_signature(public_key_pem, signature_data):
|
||||
try:
|
||||
public_key = serialization.load_pem_public_key(public_key_pem)
|
||||
|
||||
parts = signature_data.split('&&')
|
||||
if len(parts) != 3:
|
||||
print("授权码格式错误:", parts)
|
||||
return False, {"valid": False, "time_range": "授权码格式错误"}, None, None
|
||||
|
||||
signature_base64, start_timestamp_base64, end_timestamp_base64 = parts
|
||||
|
||||
start_timestamp_str = base64.b64decode(start_timestamp_base64).decode('utf-8')
|
||||
end_timestamp_str = base64.b64decode(end_timestamp_base64).decode('utf-8')
|
||||
|
||||
signature_bytes = base64.b64decode(signature_base64)
|
||||
|
||||
# 构建用于验证的原始消息
|
||||
original_message = f"{start_timestamp_str}&&{end_timestamp_str}&&{id}&&{did}&&{gid}".encode('utf-8')
|
||||
|
||||
# 打印原始消息以便调试
|
||||
print("Original Message:", original_message)
|
||||
|
||||
# 尝试验证签名
|
||||
public_key.verify(
|
||||
signature_bytes,
|
||||
original_message,
|
||||
padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH),
|
||||
hashes.SHA256()
|
||||
)
|
||||
|
||||
# 获取时间并进行比较
|
||||
start_time = datetime.strptime(start_timestamp_str, '%Y-%m-%d %H:%M:%S')
|
||||
end_time = datetime.strptime(end_timestamp_str, '%Y-%m-%d %H:%M:%S')
|
||||
current_time = get_current_time()
|
||||
|
||||
if start_time <= current_time <= end_time:
|
||||
return True, {"valid": True, "time_range": f"成功。时间范围: {start_time} 到 {end_time}"}, (start_time, end_time), current_time
|
||||
else:
|
||||
return False, {"valid": False, "time_range": f"时间不在有效范围内。时间范围: {start_time} 到 {end_time}"}, (start_time, end_time), current_time
|
||||
except Exception as e:
|
||||
# 打印异常信息以便调试
|
||||
print("发生错误:", str(e))
|
||||
return False, {"valid": False, "time_range": f"发生错误: {str(e)}"}, None, None
|
||||
|
||||
if __name__ == '__main__':
|
||||
app.run(debug=True, host='0.0.0.0', port=5001)
|
||||
Reference in New Issue
Block a user