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"""

验证您的电子邮件地址。

感谢您注册我们的账户。为确保当前是您本人操作，请您输入此邮件中提示的验证码以完成账户注册。如您没有注册账户，请放心忽略该信息。

DVSWebSVC-动态验证密码

{code}

(此验证码将在发送后 {VERIFY_CODE_EXPIRE_SECONDS} 秒过期。)

""" 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)