From 5d3ff7a2b62359bde36aa53aded6398ddd196cb5 Mon Sep 17 00:00:00 2001 From: zjq <1783142247@qq.com> Date: Thu, 26 Sep 2024 16:16:47 +0800 Subject: [PATCH] 20240926-0002 --- Authorize.csv | 11 +++++ README.md | 2 +- client.py | 109 ++++++++++++++++++++---------------------- client_test.py | 125 ------------------------------------------------- 4 files changed, 63 insertions(+), 184 deletions(-) create mode 100644 Authorize.csv delete mode 100644 client_test.py diff --git a/Authorize.csv b/Authorize.csv new file mode 100644 index 0000000..8f65491 --- /dev/null +++ b/Authorize.csv @@ -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----- +" \ No newline at end of file diff --git a/README.md b/README.md index 1524f30..a5fc7b1 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,5 @@ 安装python 包 -> pip install Flask pandas cryptography flask-cors +> pip install Flask pandas cryptography flask-cors requests 运行 > python client.py \ No newline at end of file diff --git a/client.py b/client.py index 049f147..707c1ef 100644 --- a/client.py +++ b/client.py @@ -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) \ No newline at end of file + app.run(debug=True, host='0.0.0.0', port=5001) diff --git a/client_test.py b/client_test.py deleted file mode 100644 index a12205d..0000000 --- a/client_test.py +++ /dev/null @@ -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) \ No newline at end of file