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 import requests from datetime import datetime from flask_cors import CORS app = Flask(__name__) CORS(app) def get_csv_path(): return os.path.join(os.path.dirname(__file__), 'Authorize.csv') def get_info_csv_path(): return os.path.join(os.path.dirname(__file__), 'Information.csv') 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(): 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('/check_time', methods=['POST']) def check_time(): current_time = get_current_time() csv_path = get_info_csv_path() if not os.path.exists(csv_path): return jsonify({"valid": False, "message": "信息文件不存在"}), 404 df = pd.read_csv(csv_path) if df.empty: return jsonify({"valid": False, "message": "信息文件为空"}), 404 start_time = datetime.strptime(df['StartTime'].iloc[0], '%Y-%m-%d') end_time = datetime.strptime(df['EndTime'].iloc[0], '%Y-%m-%d') time_range_message = f"有效时间范围: {start_time.strftime('%Y-%m-%d')} 到 {end_time.strftime('%Y-%m-%d')}" if start_time <= current_time <= end_time: days_remaining = (end_time - current_time).days return jsonify({ "valid": True, "message": f"{time_range_message}", "days_remaining": days_remaining }) return jsonify({ "valid": False, "message": f"{time_range_message}" }) @app.route('/navigate', methods=['POST']) def navigate(): current_time = get_current_time() csv_path = get_info_csv_path() if not os.path.exists(csv_path): return jsonify({"valid": False, "message": "信息文件不存在"}), 404 df = pd.read_csv(csv_path) if df.empty: return jsonify({"valid": False, "message": "信息文件为空"}), 404 start_time = datetime.strptime(df['StartTime'].iloc[0], '%Y-%m-%d') end_time = datetime.strptime(df['EndTime'].iloc[0], '%Y-%m-%d') if start_time <= current_time <= end_time: return jsonify({"valid": True, "redirect": "device"}) return jsonify({"valid": False, "redirect": "error"}) @app.route('/verify', methods=['POST', 'OPTIONS']) def handle_verification(): if request.method == 'OPTIONS': return jsonify({'message': 'Options request handled successfully'}), 200 data = request.get_json() signature_data = data.get('signatureData', '') print("Received signature data:", signature_data) 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) print("验证结果:", is_valid) 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 # Save signature and timestamps to Information.csv if valid if is_valid: start_time, end_time = time_range save_to_information_csv(signature_data, start_time, end_time) return jsonify({ "valid": is_valid, "message": message["time_range"], "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 start_timestamp_base64, end_timestamp_base64, signature_base64 = parts # 解码签名部分 signature_bytes = base64.b64decode(signature_base64) # 解码时间戳部分 start_timestamp_bytes = base64.b64decode(start_timestamp_base64) end_timestamp_bytes = base64.b64decode(end_timestamp_base64) # 解析时间戳 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, padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH), hashes.SHA256() ) # 当前时间 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.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.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"请输入正确的授权码"}, None, None def save_to_information_csv(signature_data, start_time, end_time): csv_path = get_info_csv_path() # 创建DataFrame data = { 'SignatureData': [signature_data], 'StartTime': [start_time.strftime('%Y-%m-%d')], 'EndTime': [end_time.strftime('%Y-%m-%d')] } # 如果文件存在，先读取数据 if os.path.exists(csv_path): existing_df = pd.read_csv(csv_path) # 清空文件内容 existing_df.to_csv(csv_path, index=False, header=False) # 将新的数据写入CSV文件 df = pd.DataFrame(data) df.to_csv(csv_path, index=False) if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=5001)