288 lines
10 KiB
Python
288 lines
10 KiB
Python
import os
|
|
import random
|
|
from datetime import datetime
|
|
from flask import Flask, request, jsonify, Response
|
|
from flask_cors import CORS
|
|
from cryptography.hazmat.backends import default_backend
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.serialization import (
|
|
Encoding, PrivateFormat, NoEncryption, PublicFormat
|
|
)
|
|
from cryptography.hazmat.primitives.asymmetric import padding
|
|
from cryptography.hazmat.primitives.serialization import load_pem_private_key
|
|
from dateutil.relativedelta import relativedelta
|
|
import pandas as pd
|
|
import hashlib
|
|
import numpy as np
|
|
import base64
|
|
|
|
app = Flask(__name__)
|
|
CORS(app)
|
|
|
|
@app.route('/register', methods=['POST'])
|
|
def register_user():
|
|
print("Received request:", request.json)
|
|
try:
|
|
data = request.get_json()
|
|
dname = str(data.get('dname'))
|
|
info = str(data.get('info'))
|
|
did = str(data.get('did'))
|
|
|
|
current_directory = os.path.dirname(__file__)
|
|
csv_path = os.path.join(current_directory, 'Authorize_DataBase.csv')
|
|
print(f"Checking CSV file at path: {csv_path}")
|
|
|
|
# Check for duplicate DID
|
|
if os.path.exists(csv_path):
|
|
df = pd.read_csv(csv_path)
|
|
# Ensure the DID column is treated as strings
|
|
df['DID'] = df['DID'].astype(str)
|
|
|
|
# Print existing DIDs for debugging
|
|
print("Existing DIDs:", df['DID'].str.strip().str.lower().tolist())
|
|
# Normalize incoming DID
|
|
normalized_did = did.strip().lower()
|
|
if normalized_did in df['DID'].str.strip().str.lower().values:
|
|
return jsonify({'error': '设备已存在，重新输入设备'}), 400
|
|
|
|
|
|
gid = generate_gid()
|
|
private_key, public_key = generate_key_pair(gid)
|
|
timestamp = datetime.now().strftime('%Y-%m-%d ')
|
|
active = 1
|
|
|
|
current_id = get_current_max_id()
|
|
new_id = current_id + 1 if current_id >= 1 else 1
|
|
save_to_excel(new_id, did, dname, gid, public_key, private_key, info, active, timestamp)
|
|
|
|
response_data = {
|
|
'ID': new_id,
|
|
'Active': active,
|
|
'DID': did,
|
|
'DName': dname,
|
|
'GID': gid,
|
|
'PublicKey': public_key,
|
|
'PrivateKey': private_key.private_bytes(
|
|
encoding=Encoding.PEM,
|
|
format=PrivateFormat.PKCS8,
|
|
encryption_algorithm=NoEncryption()
|
|
).decode('utf-8'),
|
|
'Info': info,
|
|
'Timestamp': timestamp
|
|
}
|
|
|
|
json_serializable_data = {k: (str(v) if isinstance(v, (pd.Timestamp, datetime)) else v) for k, v in response_data.items()}
|
|
|
|
return jsonify(json_serializable_data), 200
|
|
except Exception as e:
|
|
print(f"An error occurred: {e}")
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
def generate_gid():
|
|
current_time = datetime.now().strftime('%y%m%d')
|
|
random_suffix = random.randint(100, 999)
|
|
return f"{current_time}{random_suffix}"
|
|
|
|
def generate_key_pair(gid):
|
|
# 使用 GID 作为种子生成随机数
|
|
seed = hash(gid)
|
|
random_generator = random.Random(seed)
|
|
|
|
# 使用 SHA256 生成随机数种子
|
|
sha256_hash = hashlib.sha256()
|
|
sha256_hash.update(gid.encode('utf-8'))
|
|
random_seed = sha256_hash.digest()
|
|
|
|
# 使用随机数种子生成 RSA 密钥对
|
|
# 生成随机数种子
|
|
prng = random.Random(random_seed.hex())
|
|
random.seed(prng.randint(0, 2**128))
|
|
|
|
private_key = rsa.generate_private_key(
|
|
public_exponent=65537,
|
|
key_size=2048, # 增加密钥大小以提高安全性
|
|
backend=default_backend()
|
|
)
|
|
public_key = private_key.public_key().public_bytes(
|
|
encoding=Encoding.PEM,
|
|
format=PublicFormat.SubjectPublicKeyInfo
|
|
).decode('utf-8')
|
|
return private_key, public_key
|
|
|
|
def save_to_excel(id, did, dname, gid, public_key, private_key, info, active, timestamp):
|
|
df = pd.DataFrame({
|
|
'ID': [id],
|
|
'Active': [active],
|
|
'DID': [did],
|
|
'DName': [dname],
|
|
'GID': [gid],
|
|
'PublicKey': [public_key],
|
|
'PrivateKey': [private_key.private_bytes(
|
|
encoding=Encoding.PEM,
|
|
format=PrivateFormat.PKCS8,
|
|
encryption_algorithm=NoEncryption()
|
|
).decode('utf-8')],
|
|
'Info': [info],
|
|
'Timestamp': [timestamp]
|
|
})
|
|
|
|
current_directory = os.path.dirname(__file__)
|
|
csv_path = os.path.join(current_directory, 'Authorize_DataBase.csv')
|
|
|
|
if not os.path.exists(csv_path):
|
|
df.to_csv(csv_path, index=False, mode='w', encoding='utf-8-sig')
|
|
else:
|
|
df.to_csv(csv_path, index=False, mode='a', header=False, encoding='utf-8-sig')
|
|
|
|
def get_current_max_id():
|
|
current_directory = os.path.dirname(__file__)
|
|
csv_path = os.path.join(current_directory, 'Authorize_DataBase.csv')
|
|
if not os.path.exists(csv_path):
|
|
return 0
|
|
|
|
df = pd.read_csv(csv_path)
|
|
if df.empty:
|
|
return 0
|
|
|
|
return int(df['ID'].max())
|
|
|
|
|
|
def get_csv_path():
|
|
return os.path.join(os.path.dirname(__file__), 'signatures.csv')
|
|
@app.route('/authorization', methods=['POST'])
|
|
def authorization():
|
|
print("authorization:", request.json)
|
|
try:
|
|
data = request.get_json()
|
|
did = int(data.get('did')) # 将 did 转换为整数类型
|
|
term = int(data.get('term')) # 将 term 转换为整数类型
|
|
|
|
# 获取当前时间
|
|
start_time = datetime.now()
|
|
end_time = start_time + relativedelta(months=+term)
|
|
|
|
# 查询数据库找到 DID 对应的记录
|
|
current_directory = os.path.dirname(__file__)
|
|
csv_path = os.path.join(current_directory, 'Authorize_DataBase.csv')
|
|
if not os.path.exists(csv_path):
|
|
return jsonify({'error': '设备文件不存在'}), 404
|
|
df = pd.read_csv(csv_path)
|
|
if df.empty:
|
|
return jsonify({'error': '文件为空'}), 404
|
|
|
|
user_data = df[df['DID'] == did]
|
|
if user_data.empty:
|
|
return jsonify({'error': '未找到指定设备号'}), 404
|
|
id = int(user_data.iloc[0]['ID'])
|
|
gid = str(user_data.iloc[0]['GID'])
|
|
private_key_pem = user_data.iloc[0]['PrivateKey']
|
|
|
|
# 构造待签名的消息（只包括时间戳）
|
|
message = f"{start_time.strftime('%Y-%m-%d')}{end_time.strftime('%Y-%m-%d')}"
|
|
|
|
# 使用私钥生成签名
|
|
private_key = serialization.load_pem_private_key(private_key_pem.encode(), password=None)
|
|
signature = private_key.sign(
|
|
message.encode(),
|
|
padding.PSS(
|
|
mgf=padding.MGF1(hashes.SHA256()),
|
|
salt_length=padding.PSS.MAX_LENGTH
|
|
),
|
|
hashes.SHA256()
|
|
)
|
|
|
|
# 直接生成 Base64 编码的签名
|
|
base64_signature = base64.b64encode(signature).decode('utf-8')
|
|
|
|
# 将日期格式化为 YYYY-MM-DD 并进行 Base64 编码
|
|
base64_start_time = base64.b64encode(start_time.strftime('%Y-%m-%d').encode()).decode('utf-8')
|
|
base64_end_time = base64.b64encode(end_time.strftime('%Y-%m-%d').encode()).decode('utf-8')
|
|
|
|
# 构造签名字符串
|
|
formatted_signature = (
|
|
base64_start_time + '&&' +
|
|
base64_end_time + '&&' +
|
|
base64_signature
|
|
)
|
|
|
|
# 更新 response_data
|
|
response_data = {
|
|
'DID': did,
|
|
'Signature': formatted_signature, # 使用 CSV 中的签名
|
|
'Term': term,
|
|
'ValidityPeriod': {
|
|
'Start_Time': start_time.strftime('%Y-%m-%d'),
|
|
'End_Time': end_time.strftime('%Y-%m-%d')
|
|
}
|
|
}
|
|
|
|
print("Base64 Start Time:", base64_start_time)
|
|
print("Base64 End Time:", base64_end_time)
|
|
print("Base64 Signature:", base64_signature)
|
|
print("Formatted Signature:", formatted_signature)
|
|
|
|
signature_csv_path = get_csv_path()
|
|
if not os.path.exists(signature_csv_path):
|
|
# 创建 CSV 文件并写入表头
|
|
with open(signature_csv_path, 'w') as f:
|
|
f.write('DID,Start_Time,End_Time,Signature\n')
|
|
|
|
# 写入签名数据
|
|
with open(signature_csv_path, 'a') as f:
|
|
f.write(f'{did},{start_time.strftime("%Y-%m-%d")},{end_time.strftime("%Y-%m-%d")},{formatted_signature}\n')
|
|
|
|
# 确保所有数据类型都是 JSON 可序列化的
|
|
json_serializable_data = {k: (int(v) if isinstance(v, np.int64) else str(v) if isinstance(v, pd.Timestamp) or isinstance(v, datetime) else v) for k, v in response_data.items()}
|
|
|
|
# 在返回 JSON 响应之前打印完整响应内容
|
|
print(f"Response Data: {json_serializable_data}")
|
|
|
|
return jsonify(json_serializable_data), 200
|
|
|
|
except Exception as e:
|
|
print(f"An error occurred: {e}")
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
|
|
@app.route('/export-csv', methods=['GET'])
|
|
def export_csv():
|
|
csv_path = os.path.join(os.path.dirname(__file__), 'Authorize_DataBase.csv')
|
|
if not os.path.exists(csv_path):
|
|
return jsonify({'error': '文件不存在'}), 404
|
|
|
|
print("Exporting Authorize_DataBase.csv:")
|
|
with open(csv_path, mode='r', encoding='utf-8-sig') as file:
|
|
print(file.read()) # Log the content of the CSV
|
|
|
|
def generate():
|
|
with open(csv_path, mode='r', encoding='utf-8-sig') as file:
|
|
for line in file:
|
|
yield line
|
|
|
|
response = Response(generate(), mimetype="text/csv")
|
|
response.headers["Content-Disposition"] = "attachment; filename=Authorize_DataBase.csv"
|
|
return response
|
|
|
|
@app.route('/authorization-csv', methods=['GET'])
|
|
def export1_csv():
|
|
csv_path = os.path.join(os.path.dirname(__file__), 'signatures.csv')
|
|
if not os.path.exists(csv_path):
|
|
return jsonify({'error': '文件不存在'}), 404
|
|
|
|
print("Exporting signatures.csv:")
|
|
with open(csv_path, mode='r', encoding='utf-8-sig') as file:
|
|
print(file.read()) # Log the content of the CSV
|
|
|
|
def generate():
|
|
with open(csv_path, mode='r', encoding='utf-8-sig') as file:
|
|
for line in file:
|
|
yield line
|
|
|
|
response = Response(generate(), mimetype="text/csv")
|
|
response.headers["Content-Disposition"] = "attachment; filename=signatures.csv"
|
|
return response
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app.run(debug=True, host='0.0.0.0', port=5000) |