Gonoptics/Authorization_Client
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

20240926-0002

Browse Source
This commit is contained in:
zjq
2024-09-26 16:16:47 +08:00
parent f67504e3a7
commit 5d3ff7a2b6
4 changed files with 63 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Authorize.csv
+11
View File
@@ -0,0 +1,11 @@
PublicKey
"-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAwBfs3cknIcEBeJTx4OQ6
krG634p0DxWMQL5jqwKY82hD26I8o+Zo4jb4jrziqIlJw0YZfuUe+VA4Xeu+byp5
jpkhwy7cp2yxXTv+AqFszR08hj2Tfnge8wijxXrzg7Qs739H2ad2G0ar2Yd6H7iz
lAzOpitG9ZwlaVJuDIEbXA/07cO0kLAAYRh9sqtrnUhLCKymFwbGap7ob3ZOf8FZ
1l69jLvSQYAVHVWlm3/uwMmKWK0f0uiQJblt624NEEUcznK0gXAAIPKJ1EY1Uc/v
OAitE/QJ/abdKr/OC0FmD9nEy5gDjr43B2tOqit1KMoPjvmX+gEDZIAFi6E1vFpE
EQIDAQAB
-----END PUBLIC KEY-----
"
1 PublicKey
2 -----BEGIN PUBLIC KEY----- MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAwBfs3cknIcEBeJTx4OQ6 krG634p0DxWMQL5jqwKY82hD26I8o+Zo4jb4jrziqIlJw0YZfuUe+VA4Xeu+byp5 jpkhwy7cp2yxXTv+AqFszR08hj2Tfnge8wijxXrzg7Qs739H2ad2G0ar2Yd6H7iz lAzOpitG9ZwlaVJuDIEbXA/07cO0kLAAYRh9sqtrnUhLCKymFwbGap7ob3ZOf8FZ 1l69jLvSQYAVHVWlm3/uwMmKWK0f0uiQJblt624NEEUcznK0gXAAIPKJ1EY1Uc/v OAitE/QJ/abdKr/OC0FmD9nEy5gDjr43B2tOqit1KMoPjvmX+gEDZIAFi6E1vFpE EQIDAQAB -----END PUBLIC KEY-----
README.md
+1 -1
View File
@@ -1,5 +1,5 @@
安装python 包
> pip install Flask pandas cryptography flask-cors
> pip install Flask pandas cryptography flask-cors requests
运行
> python client.py
client.py
+51 -58
View File
@@ -4,56 +4,38 @@ import pandas as pd
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
import base64
import requests
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 get_csv_path():
return os.path.join(os.path.dirname(__file__), 'Authorize.csv')
def read_public_key_from_file(public_key_str):
return public_key_str.encode('utf-8')
def get_public_key():
csv_path = get_csv_path()
if not os.path.exists(csv_path):
return None
df = pd.read_csv(csv_path)
if not df.empty:
return df['PublicKey'].iloc[0].encode('utf-8')
return None
def get_current_time():
return datetime.now()
response = requests.get('http://localhost:12235/current-time')
print(f"API 返回: {response.json()}")
if response.status_code == 200:
time_data = response.json()
return datetime(time_data['year'], time_data['month'], time_data['day'])
else:
raise Exception("请求时间失败")
@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'])
@app.route('/verify', methods=['POST', 'OPTIONS'])
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
if request.method == 'OPTIONS':
return jsonify({'message': 'Options request handled successfully'}), 200
data = request.get_json()
signature_data = data.get('signatureData', '')
@@ -63,6 +45,10 @@ def handle_verification():
if not signature_data:
return jsonify({'error': '缺少授权码'}), 400
public_key_pem = get_public_key()
if public_key_pem is None:
return jsonify({'error': '无法加载公钥'}), 500
is_valid, message, time_range, current_time = verify_signature(public_key_pem, signature_data)
print("公钥:", public_key_pem)
@@ -70,36 +56,42 @@ def handle_verification():
print("从授权码获取的时间范围:", time_range)
print("当前时间:", current_time)
# Format current_time to "YYYY-MM-DD"
current_time_str = current_time.strftime('%Y-%m-%d') if current_time else None
return jsonify({
"valid": is_valid,
"message": message["time_range"],
"current_time": current_time,
"current_time": current_time_str,
"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() # 移除可能的末尾空格
start_timestamp_base64, end_timestamp_base64, signature_base64 = parts
# 解码签名部分
signature_bytes = base64.b64decode(signature_base64)
# 构建用于验证的原始消息
original_message = f"{start_timestamp_str}&&{end_timestamp_str}&&{id}&&{did}&&{gid}".encode('utf-8')
# 解码时间戳部分
start_timestamp_bytes = base64.b64decode(start_timestamp_base64)
end_timestamp_bytes = base64.b64decode(end_timestamp_base64)
# 打印原始消息以便调试
print("Original Message:", original_message)
# 解析时间戳
start_timestamp_str = start_timestamp_bytes.decode('utf-8').strip()
end_timestamp_str = end_timestamp_bytes.decode('utf-8').strip()
# 尝试验证签名
# 重新构建原始消息（只包括时间戳）
original_message = f"{start_timestamp_str}{end_timestamp_str}".encode('utf-8')
# 验证签名
public_key.verify(
signature_bytes,
original_message,
@@ -107,19 +99,20 @@ def verify_signature(public_key_pem, signature_data):
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()
# 时间比较
start_time = datetime.strptime(start_timestamp_str, '%Y-%m-%d')
end_time = datetime.strptime(end_timestamp_str, '%Y-%m-%d')
if start_time <= current_time <= end_time:
return True, {"valid": True, "time_range": f"成功。时间范围: {start_time} 到 {end_time}"}, (start_time, end_time), current_time
return True, {"valid": True, "time_range": f"成功。时间范围: {start_time.year}-{start_time.month}-{start_time.day}到{end_time.year}-{end_time.month}- {end_time.day}"}, (start_time, end_time), current_time
else:
return False, {"valid": False, "time_range": f"时间不在有效范围内。时间范围: {start_time} 到 {end_time}"}, (start_time, end_time), current_time
return False, {"valid": False, "time_range": f"时间不在有效范围内。时间范围: {start_time.year}-{start_time.month}-{start_time.day}到{end_time.year}-{end_time.month}- {end_time.day}"}, (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)
app.run(debug=True, host='0.0.0.0', port=5001)
client_test.py
-125
View File
@@ -1,125 +0,0 @@
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)