381 lines
13 KiB
Python
381 lines
13 KiB
Python
from flask import Flask, request, jsonify
|
||
import secrets
|
||
import string
|
||
import time
|
||
import requests
|
||
import hashlib
|
||
|
||
app = Flask(__name__)
|
||
|
||
# ==================== 配置 ====================
|
||
VERIFY_CODE_EXPIRE_SECONDS = 120 # 验证码过期时间（秒）
|
||
RATE_LIMIT_SECONDS = 60 # 申请间隔限制（秒）
|
||
TOKEN_EXPIRE_SECONDS = 300 # token过期时间（秒）
|
||
|
||
# ==================== AID ====================
|
||
AID = "填写"
|
||
# ==================== 存储 ====================
|
||
COOKIE_STORE = {} # {cookie: {'cudid': xxx, 'ip': xxx, 'created_at': timestamp}}
|
||
CODE_STORE = {} # {cookie: {'code': xxx, 'expire_at': timestamp, 'last_request': timestamp}}
|
||
TOKEN_STORE = {} # {token: {'cookie': xxx, 'cudid': xxx, 'expire_at': timestamp, 'used': False, 'code': xxx}}
|
||
REQUEST_RECORD = {} # {key: timestamp} key可以是cookie或ip
|
||
|
||
# ==================== 工具函数 ====================
|
||
|
||
def generate_cookie():
|
||
"""生成高熵Cookie"""
|
||
return secrets.token_hex(64) + secrets.token_hex(32)
|
||
|
||
def generate_verify_code():
|
||
"""生成10位高熵动态验证密码"""
|
||
chars = string.ascii_letters + string.digits + "!@#$%^&*()_+-=[]{}|;:,.<>?"
|
||
return ''.join(secrets.choice(chars) for _ in range(10))
|
||
|
||
def generate_token():
|
||
"""生成32位一次性token"""
|
||
return secrets.token_hex(32)
|
||
|
||
def build_verify_email_html(code):
|
||
"""构建验证码邮件HTML"""
|
||
return f"""
|
||
<!DOCTYPE html>
|
||
<html>
|
||
<head><meta charset="UTF-8"></head>
|
||
<body style="margin:0;padding:0;background:#f0f2f3;">
|
||
<table cellspacing="0" cellpadding="0" width="100%" align="center" style="max-width:600px;margin:0 auto;background:#fff;font-family:Arial;font-size:14px;color:#444;border-bottom:1px solid #d6d6d6;">
|
||
<tr>
|
||
<td style="padding:25px 35px;">
|
||
<h1 style="font-size:20px;font-weight:bold;margin:0 0 15px;">验证您的电子邮件地址。</h1>
|
||
<p style="margin:0 0 15px;">感谢您注册我们的账户。为确保当前是您本人操作，请您输入此邮件中提示的验证码以完成账户注册。如您没有注册账户，请放心忽略该信息。</p>
|
||
</td>
|
||
</tr>
|
||
<tr>
|
||
<td style="padding:0 35px 25px;text-align:center;">
|
||
<div style="font-weight:bold;padding-bottom:15px;color:#444;">DVSWebSVC-动态验证密码</div>
|
||
<div style="color:#000;font-size:36px;font-weight:bold;padding-bottom:15px;letter-spacing:4px;">{code}</div>
|
||
<div style="color:#666;font-size:13px;">(此验证码将在发送后 {VERIFY_CODE_EXPIRE_SECONDS} 秒过期。)</div>
|
||
</td>
|
||
</tr>
|
||
</table>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
def send_email_via_local(to_email, html_content):
|
||
"""调用本地:5000/send 发送邮件"""
|
||
try:
|
||
login_resp = requests.post(
|
||
"http://localhost:5000/login",
|
||
json={
|
||
"aid": AID,
|
||
"username": "dvsadmin",
|
||
"password": "G7#kLp$QwZ@2xR&yM9!nVb*C5^jHf%T"
|
||
},
|
||
timeout=10
|
||
)
|
||
if login_resp.status_code != 200:
|
||
return False, "登录本地邮件服务失败"
|
||
|
||
result = login_resp.json()
|
||
if result.get('code') != 200:
|
||
return False, f"登录失败: {result.get('message')}"
|
||
|
||
cookie = result.get('cookie')
|
||
if not cookie:
|
||
return False, "获取邮件服务cookie失败"
|
||
|
||
send_resp = requests.post(
|
||
"http://localhost:5000/send",
|
||
json={
|
||
"cookie": cookie,
|
||
"to": to_email,
|
||
"subject": "【DVSWebSVC】验证您的电子邮件地址",
|
||
"content": html_content
|
||
},
|
||
timeout=30
|
||
)
|
||
|
||
if send_resp.status_code == 200:
|
||
result = send_resp.json()
|
||
if result.get('code') == 200:
|
||
return True, "邮件发送成功"
|
||
else:
|
||
return False, result.get('message', '邮件发送失败')
|
||
else:
|
||
return False, f"邮件服务返回错误: {send_resp.status_code}"
|
||
|
||
except requests.exceptions.Timeout:
|
||
return False, "调用邮件服务超时"
|
||
except requests.exceptions.ConnectionError:
|
||
return False, "无法连接到邮件服务 (localhost:5000)"
|
||
except Exception as e:
|
||
return False, f"调用邮件服务异常: {str(e)}"
|
||
|
||
# ==================== API 接口 ====================
|
||
|
||
@app.route('/apply', methods=['POST'])
|
||
def apply_cookie():
|
||
"""
|
||
申请Cookie（只需CUDID）
|
||
请求体: {"cudid": "256位设备唯一标识符"}
|
||
返回: {"code": 200, "cookie": "xxx", "message": "申请成功"}
|
||
"""
|
||
try:
|
||
data = request.get_json() or {}
|
||
cudid = data.get('cudid')
|
||
|
||
# 验证CUDID
|
||
if not cudid:
|
||
return jsonify({"code": 400, "message": "缺少 cudid"}), 400
|
||
|
||
# 验证CUDID是否为256位哈希（64位十六进制字符）
|
||
if len(cudid) != 64:
|
||
return jsonify({"code": 400, "message": "cudid 必须是64位十六进制字符串（256位）"}), 400
|
||
|
||
try:
|
||
int(cudid, 16) # 验证是否为十六进制
|
||
except ValueError:
|
||
return jsonify({"code": 400, "message": "cudid 必须是有效的十六进制字符串"}), 400
|
||
|
||
# 生成永久Cookie
|
||
cookie = generate_cookie()
|
||
|
||
# 存储cookie信息
|
||
COOKIE_STORE[cookie] = {
|
||
'cudid': cudid,
|
||
'ip': request.remote_addr,
|
||
'created_at': time.time()
|
||
}
|
||
|
||
return jsonify({
|
||
"code": 200,
|
||
"message": "申请成功",
|
||
"cookie": cookie
|
||
})
|
||
|
||
except Exception as e:
|
||
return jsonify({"code": 500, "message": f"服务器错误: {str(e)}"}), 500
|
||
|
||
|
||
@app.route('/request_code', methods=['POST'])
|
||
def request_verify_code():
|
||
"""
|
||
申请验证码
|
||
请求体: {"cookie": "xxx", "cudid": "256位设备标识符", "to": "收件邮箱"}
|
||
返回: {"code": 200, "token": "xxx", "message": "验证码已发送"}
|
||
"""
|
||
try:
|
||
data = request.get_json()
|
||
if not data:
|
||
return jsonify({"code": 400, "message": "请求体不能为空"}), 400
|
||
|
||
cookie = data.get('cookie')
|
||
cudid = data.get('cudid')
|
||
to_email = data.get('to')
|
||
|
||
# 参数校验
|
||
if not cookie:
|
||
return jsonify({"code": 400, "message": "缺少 cookie"}), 400
|
||
if not cudid:
|
||
return jsonify({"code": 400, "message": "缺少 cudid"}), 400
|
||
if not to_email:
|
||
return jsonify({"code": 400, "message": "缺少 to"}), 400
|
||
|
||
# 验证CUDID格式
|
||
if len(cudid) != 64:
|
||
return jsonify({"code": 400, "message": "cudid 必须是64位十六进制字符串（256位）"}), 400
|
||
try:
|
||
int(cudid, 16)
|
||
except ValueError:
|
||
return jsonify({"code": 400, "message": "cudid 必须是有效的十六进制字符串"}), 400
|
||
|
||
# 验证Cookie是否存在
|
||
if cookie not in COOKIE_STORE:
|
||
return jsonify({"code": 401, "message": "Cookie无效，请重新申请"}), 401
|
||
|
||
# 验证CUDID是否匹配
|
||
stored = COOKIE_STORE[cookie]
|
||
if stored.get('cudid') != cudid:
|
||
return jsonify({"code": 401, "message": "CUDID不匹配"}), 401
|
||
|
||
# 限流检查（IP或Cookie任一限制）
|
||
ip = request.remote_addr
|
||
now = time.time()
|
||
|
||
# 检查IP限制
|
||
if ip in REQUEST_RECORD:
|
||
if now - REQUEST_RECORD[ip] < RATE_LIMIT_SECONDS:
|
||
wait = int(RATE_LIMIT_SECONDS - (now - REQUEST_RECORD[ip]))
|
||
return jsonify({"code": 429, "message": f"IP限制，请等待 {wait} 秒"}), 429
|
||
|
||
# 检查Cookie限制
|
||
if cookie in REQUEST_RECORD:
|
||
if now - REQUEST_RECORD[cookie] < RATE_LIMIT_SECONDS:
|
||
wait = int(RATE_LIMIT_SECONDS - (now - REQUEST_RECORD[cookie]))
|
||
return jsonify({"code": 429, "message": f"Cookie限制，请等待 {wait} 秒"}), 429
|
||
|
||
# 生成验证码
|
||
code = generate_verify_code()
|
||
|
||
# 构建HTML邮件
|
||
html_content = build_verify_email_html(code)
|
||
|
||
# 发送邮件
|
||
success, msg = send_email_via_local(to_email, html_content)
|
||
if not success:
|
||
return jsonify({"code": 500, "message": f"发送失败: {msg}"}), 500
|
||
|
||
# 记录申请时间（IP和Cookie都记录）
|
||
REQUEST_RECORD[ip] = now
|
||
REQUEST_RECORD[cookie] = now
|
||
|
||
# 存储验证码
|
||
CODE_STORE[cookie] = {
|
||
'code': code,
|
||
'expire_at': now + VERIFY_CODE_EXPIRE_SECONDS,
|
||
'last_request': now
|
||
}
|
||
|
||
# 生成一次性token
|
||
token = generate_token()
|
||
TOKEN_STORE[token] = {
|
||
'cookie': cookie,
|
||
'cudid': cudid,
|
||
'code': code,
|
||
'expire_at': now + TOKEN_EXPIRE_SECONDS,
|
||
'used': False
|
||
}
|
||
|
||
return jsonify({
|
||
"code": 200,
|
||
"message": "验证码已发送",
|
||
"token": token
|
||
})
|
||
|
||
except Exception as e:
|
||
return jsonify({"code": 500, "message": f"服务器错误: {str(e)}"}), 500
|
||
|
||
|
||
@app.route('/verify', methods=['POST'])
|
||
def verify_code():
|
||
"""
|
||
验证验证码
|
||
请求体: {"token": "xxx", "code": "用户输入的验证码"}
|
||
返回: {"code": 200, "message": "验证成功"}
|
||
"""
|
||
try:
|
||
data = request.get_json()
|
||
if not data:
|
||
return jsonify({"code": 400, "message": "请求体不能为空"}), 400
|
||
|
||
token = data.get('token')
|
||
user_code = data.get('code')
|
||
|
||
if not token:
|
||
return jsonify({"code": 400, "message": "缺少 token"}), 400
|
||
if not user_code:
|
||
return jsonify({"code": 400, "message": "缺少 code"}), 400
|
||
|
||
# 验证token是否存在
|
||
if token not in TOKEN_STORE:
|
||
return jsonify({"code": 401, "message": "Token无效"}), 401
|
||
|
||
token_data = TOKEN_STORE[token]
|
||
|
||
# 检查是否已使用
|
||
if token_data.get('used'):
|
||
return jsonify({"code": 401, "message": "Token已使用"}), 401
|
||
|
||
# 检查是否过期
|
||
if time.time() > token_data.get('expire_at', 0):
|
||
TOKEN_STORE[token]['used'] = True
|
||
return jsonify({"code": 401, "message": "Token已过期"}), 401
|
||
|
||
# 验证验证码
|
||
stored_code = token_data.get('code')
|
||
if user_code != stored_code:
|
||
return jsonify({"code": 401, "message": "验证码错误"}), 401
|
||
|
||
# 验证成功，标记token已使用
|
||
TOKEN_STORE[token]['used'] = True
|
||
|
||
return jsonify({
|
||
"code": 200,
|
||
"message": "验证成功"
|
||
})
|
||
|
||
except Exception as e:
|
||
return jsonify({"code": 500, "message": f"服务器错误: {str(e)}"}), 500
|
||
|
||
|
||
@app.route('/check_token', methods=['POST'])
|
||
def check_token():
|
||
"""
|
||
检查token状态
|
||
请求体: {"token": "xxx"}
|
||
返回: {"code": 200, "valid": true/false, "used": true/false, "expired": true/false}
|
||
"""
|
||
try:
|
||
data = request.get_json() or {}
|
||
token = data.get('token')
|
||
|
||
if not token:
|
||
return jsonify({"code": 400, "message": "缺少 token"}), 400
|
||
|
||
if token not in TOKEN_STORE:
|
||
return jsonify({"code": 200, "valid": False, "message": "Token不存在"})
|
||
|
||
token_data = TOKEN_STORE[token]
|
||
is_expired = time.time() > token_data.get('expire_at', 0)
|
||
is_used = token_data.get('used', False)
|
||
|
||
return jsonify({
|
||
"code": 200,
|
||
"valid": not is_expired and not is_used,
|
||
"used": is_used,
|
||
"expired": is_expired
|
||
})
|
||
|
||
except Exception as e:
|
||
return jsonify({"code": 500, "message": f"服务器错误: {str(e)}"}), 500
|
||
|
||
|
||
@app.route('/health', methods=['GET'])
|
||
def health():
|
||
"""健康检查"""
|
||
return jsonify({
|
||
"code": 200,
|
||
"message": "服务运行正常",
|
||
"stats": {
|
||
"cookies": len(COOKIE_STORE),
|
||
"codes": len(CODE_STORE),
|
||
"tokens": len(TOKEN_STORE),
|
||
"records": len(REQUEST_RECORD)
|
||
}
|
||
})
|
||
|
||
|
||
if __name__ == '__main__':
|
||
print("=" * 60)
|
||
print("DVSWebSVC 验证码服务")
|
||
print("=" * 60)
|
||
print(f"验证码有效期: {VERIFY_CODE_EXPIRE_SECONDS} 秒")
|
||
print(f"限流间隔: {RATE_LIMIT_SECONDS} 秒")
|
||
print(f"Token有效期: {TOKEN_EXPIRE_SECONDS} 秒")
|
||
print("=" * 60)
|
||
print("\n📍 申请Cookie: POST /apply")
|
||
print(" 请求: {\"cudid\": \"64位十六进制（256位）\"}")
|
||
print(" 返回: {\"code\": 200, \"cookie\": \"xxx\"}")
|
||
print("\n📍 申请验证码: POST /request_code")
|
||
print(" 请求: {\"cookie\": \"xxx\", \"cudid\": \"xxx\", \"to\": \"邮箱\"}")
|
||
print(" 返回: {\"code\": 200, \"token\": \"xxx\"}")
|
||
print("\n📍 验证验证码: POST /verify")
|
||
print(" 请求: {\"token\": \"xxx\", \"code\": \"验证码\"}")
|
||
print(" 返回: {\"code\": 200, \"message\": \"验证成功\"}")
|
||
print("\n📍 检查Token: POST /check_token")
|
||
print(" 请求: {\"token\": \"xxx\"}")
|
||
print("📍 健康检查: GET /health")
|
||
print("=" * 60)
|
||
|
||
app.run(host='0.0.0.0', port=888, debug=True)
|