# -*- coding: utf-8 -*- """站点认证与管理路由（/auth/* 与 /admin/site/*） 由 main.py 调用 register_auth_routes(app) 挂载。 """ import hmac import secrets import sqlite3 import time from flask import request, send_file import auth_system as auth def _data() -> dict: try: if request.method == 'GET': return dict(request.args) json_data = request.get_json(silent=True) or {} return {**dict(request.form), **json_data} except Exception: return {} # 登录失败锁定（P3-7）：按用户名维度记录失败，10 分钟窗口内 >=5 次临时锁定该用户名。 # 与既有按 IP 10/分限流互补——IP 池轮换爆破同一弱口令账号时仍被挡住 _LOGIN_FAILS: dict = {} _LOGIN_FAIL_LIMIT = 5 _LOGIN_FAIL_WINDOW = 600 # R5-6：清理节流——超限后的全表扫描每 60 秒最多一次，不再每个请求都 O(n)； # 硬上限 20000 条（按插入序砍掉最早的一半），防止节流窗口内被随机用户名撑爆 _LOGIN_FAIL_SWEEP_TS = 0.0 # R5-4：本锁定周期已落过审计日志的用户名（解锁或成功登录即重置，下个周期允许再记一条） _LOGIN_FAIL_LOGGED: set = set() # 第 86 条：管理员两步验证（2FA）——票据表（内存态，单实例部署与全站任务表同口径）。 # ticket -> {user_id, username, email, ts, attempts}；密码验证通过才签发， # Step2 邮箱比对与 Step3 验证码共用 attempts（上限内错满即作废），Step3 成功即销毁。 # 第 87 条：每管理员仅 1 张活票（新票签发即销毁同账号旧票）；_ADMIN2FA_LAST_SEND # 为每账号 admin2fa 发码软冷却（60s，防持密码者借多票持续轰炸绑定邮箱） _ADMIN_2FA: dict = {} _ADMIN2FA_LAST_SEND: dict = {} _ADMIN_2FA_TTL = 300 # 票据 5 分钟有效 _ADMIN_2FA_MAX_ATTEMPTS = 5 # 邮箱比对 + 验证码合计尝试上限 _ADMIN_2FA_MAX_TICKETS = 200 # 全局活票上限（防内存膨胀，挤掉最旧） _ADMIN2FA_SEND_GAP = 20 # 同账号两次发码最小间隔（秒）——第 87 条初版 60s 会卡住合法的背靠背重登（含矩阵双设备全流程），20s 已足以阻断紧密轰炸循环 def _admin2fa_purge() -> None: now = time.time() for t in [t for t, v in _ADMIN_2FA.items() if now - v['ts'] > _ADMIN_2FA_TTL]: _ADMIN_2FA.pop(t, None) def _admin2fa_mask(email: str) -> str: """邮箱掩码（审计日志与前端提示用）：3*****@qq.com 样式""" try: local, dom = str(email).split('@', 1) return (local[0] if local else '*') + '*****@' + dom except Exception: return '*****' def _admin2fa_issue(user) -> str: _admin2fa_purge() # 第 87 条：每管理员仅 1 张活票——新票签发即销毁同账号旧票 for t in [t for t, v in list(_ADMIN_2FA.items()) if v['username'] == user['username']]: _ADMIN_2FA.pop(t, None) if len(_ADMIN_2FA) >= _ADMIN_2FA_MAX_TICKETS: items = list(_ADMIN_2FA.items()) # 快照后再 min，防并发 pop 使迭代失效 if items: oldest = min(items, key=lambda kv: kv[1]['ts'])[0] _ADMIN_2FA.pop(oldest, None) ticket = secrets.token_urlsafe(16) _ADMIN_2FA[ticket] = {'user_id': user['id'], 'username': user['username'], 'email': user['email'], 'ts': int(time.time()), 'attempts': 0} return ticket def _admin2fa_take(ticket: str): """取活票（不销毁）：无票/过期返回 (None, None)；命中返回 (ticket, 数据)。""" _admin2fa_purge() v = _ADMIN_2FA.get(ticket or '') if not v: return None, None return ticket, v def _login_fail_check(username: str) -> bool: """返回 True 表示允许尝试登录；锁定窗口内返回 False""" global _LOGIN_FAIL_SWEEP_TS now = time.time() fails = [t for t in _LOGIN_FAILS.get(username, []) if now - t < _LOGIN_FAIL_WINDOW] _LOGIN_FAILS[username] = fails if len(fails) < _LOGIN_FAIL_LIMIT: _LOGIN_FAIL_LOGGED.discard(username) if len(_LOGIN_FAILS) > 10000 and now - _LOGIN_FAIL_SWEEP_TS > 60: # 容量上限（用户名为键，防无界增长） _LOGIN_FAIL_SWEEP_TS = now stale_users = [u for u, ts in _LOGIN_FAILS.items() if not ts or now - max(ts) > _LOGIN_FAIL_WINDOW] for u in stale_users[:5000]: _LOGIN_FAILS.pop(u, None) if len(_LOGIN_FAILS) > 20000: for u in list(_LOGIN_FAILS.keys())[:len(_LOGIN_FAILS) // 2]: _LOGIN_FAILS.pop(u, None) return len(fails) < _LOGIN_FAIL_LIMIT def _login_fail_record(username: str) -> None: _LOGIN_FAILS.setdefault(username, []).append(time.time()) def _login_fail_clear(username: str) -> None: _LOGIN_FAILS.pop(username, None) _LOGIN_FAIL_LOGGED.discard(username) def _login_fail_locked_log_once(username: str) -> bool: """R5-4：锁定周期内的重复被封尝试只落一条 fail 日志，防多 IP 刷审计表； 解锁（窗口过期或成功登录）后进入新周期，才允许再记一条。""" if username in _LOGIN_FAIL_LOGGED: return False _LOGIN_FAIL_LOGGED.add(username) return True def _actor() -> str: """审计日志/卡密 created_by 的操作者名——跟随站点管理员的真实用户名 （2026-09-22 管理员账号由 admin 更名 qfbsh 后，日志不得再硬编码旧名； 无会话兜底用 system——admin 账号已不存在，不能再出现在新日志里）""" cu = auth.current_user() return cu['username'] if cu else 'system' def _safe_register_credits() -> int: """注册赠送次数（settings.register_credits，默认 3）。 R5 防呆：损坏回退默认；<0 一律按 0——设为 -1 会经 create_user 造出"不限次"新号。""" try: credits = int(auth.get_setting('register_credits', str(auth.DEFAULT_REGISTER_CREDITS)) or 0) except (TypeError, ValueError): return auth.DEFAULT_REGISTER_CREDITS return max(0, credits) # 验证码按 IP 限流（每 IP 每小时 10 次），防止公网环境换邮箱刷爆 SMTP _SEND_CODE_IP = {} _SEND_CODE_IP_LIMIT = 10 _SEND_CODE_IP_WINDOW = 3600 def _send_code_ip_check(ip: str) -> bool: """返回 True 表示允许发送；超限返回 False（S4：表超限时顺手清理过期项，防无界增长）""" now = time.time() if len(_SEND_CODE_IP) > 10000: stale = [k for k, (c, w) in _SEND_CODE_IP.items() if now - w > _SEND_CODE_IP_WINDOW] for k in stale: _SEND_CODE_IP.pop(k, None) cnt, win = _SEND_CODE_IP.get(ip, (0, now)) if now - win > _SEND_CODE_IP_WINDOW: cnt, win = 0, now if cnt >= _SEND_CODE_IP_LIMIT: return False _SEND_CODE_IP[ip] = (cnt + 1, win) return True def _ok(data=None, message: str = 'success'): resp = {'status': 200, 'success': True, 'message': message} if data is not None: resp['data'] = data return resp, 200 def _err(message: str, code: int = 400, error_code: str = ''): resp = {'status': code, 'success': False, 'message': message} if error_code: resp['error_code'] = error_code return resp, code def register_auth_routes(app) -> None: # ==================== 用户侧 ==================== @app.route('/auth/me', methods=['GET', 'POST']) def auth_me(): user = auth.current_user() ann = auth.get_announcement() data = { 'user': auth._user_public(user) if user else None, 'duration_text': auth.format_credits(user['duration_seconds']) if user else '', 'announcement': {'title': ann['title'], 'content': ann['content'], 'popup_mode': ann['popup_mode'], 'updated_at': ann['updated_at']}, # 未配置 SMTP 时前端隐藏注册入口（owner 手动建号模式，不开放邮箱注册） 'smtp_configured': auth.smtp_configured(), # F-6.1（第 79 条）：注册赠送次数动态下发——设置改值后前端文案同步，不再硬编码"3 次" 'register_credits': _safe_register_credits(), } return _ok(data) @app.route('/auth/send_code', methods=['POST']) def auth_send_code(): try: if not _send_code_ip_check(auth._client_ip()): return _err('验证码请求过于频繁，请稍后再试') data = _data() email = str(data.get('email') or '').strip() # 第 85 条：验证码按用途区分（注册/找回），邮件文案与库行用途隔离 purpose = str(data.get('purpose') or 'register').strip() if purpose not in ('register', 'reset'): return _err('验证码用途不正确') auth.send_verify_code(email, purpose) return _ok(message='验证码已发送，请查收邮箱') except ValueError as e: return _err(str(e)) except RuntimeError as e: return _err(str(e)) except Exception: return _err('验证码发送失败，请稍后重试', 500) @app.route('/auth/register', methods=['POST']) def auth_register(): try: data = _data() username = str(data.get('username') or '').strip() password = str(data.get('password') or '') email = auth._normalize_email(str(data.get('email') or '')) code = str(data.get('code') or '').strip() device_id = str(data.get('device_id') or '').strip()[:64] # 第 81 条：注册前必须已同意《用户协议》与《隐私政策》——前端入口闸的 # 服务端兜底（同意状态本身存于用户浏览器，此处为语义性强制） if str(data.get('agree') or '').strip() != '1': return _err('请先阅读并同意《用户协议》与《隐私政策》') if not (3 <= len(username) <= 20 and re_username_ok(username)): return _err('用户名需 3-20 位，仅支持中文、字母、数字、下划线，且不能纯下划线') if len(password) < 8 or len(password) > 128: return _err('密码长度需在 8-128 位之间') if len(email) > 254: return _err('邮箱格式不正确') # R5-P2：查重统一后移到验证码校验通过之后——验证码是“证明你能收这个 # 邮箱的信”的凭证，凭证在手才有资格探测注册状态（堵未认证枚举）； # 并发双提交仍由 UNIQUE 索引兜底 code_ttl = auth.check_verify_code(email, code, 'register') if not code_ttl: return _err('邮箱验证码错误或已过期') if auth.get_user_by_name(username): auth.restore_verify_code(email, code, code_ttl, 'register') return _err('用户名已被注册') # P3-19：邮箱查重（配合库侧唯一索引——索引防并发，这里给出友好提示） if email and auth.get_user_by_email(email): auth.restore_verify_code(email, code, code_ttl, 'register') return _err('该邮箱已被注册，请更换邮箱或联系管理员') # P1-1（第 79 条）：防刷号双闸——① 同设备只送一次注册赠送；② 同 IP 24h # 最多 3 次注册。device_id 是客户端随机串可伪造，故两层都要，叠加 # send_code 10 封/时/IP 与邮箱归一化查重后，刷号产能压到可忽略。 device_used_flag = auth.device_used(device_id) if device_id else False reg_ip = auth._client_ip() if reg_ip and auth.register_ip_exceeded(reg_ip): auth.restore_verify_code(email, code, code_ttl, 'register') auth.add_log('fail', username, reg_ip, '注册', '同 IP 24h 注册次数超限') return _err('当前网络注册过于频繁，请 24 小时后再试或联系管理员') # P3-21 + R5 防呆：赠送次数读取与钳制统一走 _safe_register_credits credits = _safe_register_credits() if credits < 0: credits = 0 # R5 防呆：赠送次数不允许配成 -1（会造出不限次新号） if device_used_flag: # 同设备重复注册：账号仍可创建（家庭共用电脑场景），但不再重复赠送 credits = 0 try: user = auth.create_user(username, password, email=email, credits=credits, device_id=device_id, ip=reg_ip, ua=str(request.headers.get('User-Agent') or ''), register_ip_limit=3, register_device_once=True) except sqlite3.IntegrityError: # 并发双击提交时的兜底（前面已有重名预检）；还原验证码供重试 auth.restore_verify_code(email, code, code_ttl, 'register') return _err('用户名或邮箱已被注册，请换一个再试') except ValueError as e: # 锁内 IP 闸复查（register_ip_limit）等业务性拒绝：友好 400 而非 500 auth.restore_verify_code(email, code, code_ttl, 'register') auth.add_log('fail', username, reg_ip, '注册', str(e)) return _err(str(e)) except Exception: # 建号失败不吞验证码（F14 第二轮补证） auth.restore_verify_code(email, code, code_ttl, 'register') raise auth.login_session(user) fresh = auth.get_user_by_name(username) # 第 100 条（对抗复审）：设备闸在事务内置零时路由层 credits 不回读—— # 按入库实际次数对齐审计日志与 toast 文案（正常路径 min 无操作） credits = min(credits, max(0, int(fresh['duration_seconds'] or 0))) # 第 81 条：注册成功落审计日志（后台日志页可见刷号行为） auth.add_log('success', username, reg_ip, '注册账号', f'赠送 {credits} 次' if credits else '未赠送次数') return _ok({'user': auth._user_public(fresh), 'duration_text': auth.format_credits(fresh['duration_seconds']), 'redirect': 'admin' if fresh['role'] == 'admin' else 'user'}, '注册成功' + (f'，已赠送 {credits} 次下载次数' if credits else '')) except Exception: return _err('注册失败，请稍后重试', 500) @app.route('/auth/trial', methods=['POST']) def auth_trial(): """试用已并入注册：注册即送 N 次（随后台设置联动，第 98 条不再硬编码），无需领取""" return _err(f'注册即送 {_safe_register_credits()} 次免费下载次数，请直接注册账号') @app.route('/auth/login', methods=['POST']) def auth_login(): try: data = _data() username = str(data.get('username') or '').strip() password = str(data.get('password') or '') if len(password) > 128: # 超长密码不做 PBKDF2（CPU 放大器），直接按密码错误拒绝 return _err('用户名或密码错误') # 第 85 条：邮箱登录——含 @ 按归一化邮箱解析出真实账号再验密； # 锁定/日志/状态检查统一用解析出的真实用户名（邮箱大小写与 +tag # 变体归一后，锁定不被变体绕过，日志也不再出现邮箱串）。 # 邮箱不存在与密码错误统一文案，不暴露账号存在性。 if '@' in username: iuser = auth.get_user_by_email(username) lock_name = iuser['username'] if iuser else (auth._normalize_email(username) or username) else: iuser = None lock_name = username # P3-7：用户名维度失败锁定（10 分钟内 5 次失败）；R5-4：日志整个锁定周期只记一条 if not _login_fail_check(lock_name): if _login_fail_locked_log_once(lock_name): auth.add_log('fail', lock_name, auth._client_ip(), '登录', '失败次数过多，临时锁定') return _err('尝试过于频繁，该账号已临时锁定，请 10 分钟后再试') if iuser is not None: user = auth.verify_login(iuser['username'], password) else: user = auth.verify_login(username, password) if not user: _login_fail_record(lock_name) return _err('用户名或密码错误') status = user['status'] if 'status' in user.keys() else 1 if status == 2: reason = user['ban_reason'] if 'ban_reason' in user.keys() else '' auth.add_log('fail', lock_name, auth._client_ip(), '登录', f'封禁账号尝试登录: {reason}') return _err('账号已被封禁' + (f'：{reason}' if reason else '') + '，请联系管理员') if status == 0: return _err('账号已被禁用，请联系管理员') _login_fail_clear(lock_name) # 登录成功清零失败计数 if user['role'] == 'admin': # 第 86 条：管理员不在此建立会话——一律转两步验证通道 # （密码正确才签发票据；未绑定邮箱 fail-closed） admin_email = user['email'] if 'email' in user.keys() else '' if not admin_email: auth.add_log('fail', lock_name, auth._client_ip(), '管理员登录', '未绑定邮箱，两步验证不可用') return _err('管理员账号未绑定邮箱，请联系管理员先完成绑定') ticket = _admin2fa_issue(user) auth.add_log('success', lock_name, auth._client_ip(), '管理员登录', '密码验证通过，待邮箱验证') return _ok({'ticket': ticket, 'email_masked': _admin2fa_mask(admin_email), 'username': user['username'], 'admin_2fa': True}, '密码验证通过，请完成邮箱验证') auth.login_session(user) # IP 归属地异常检测：24h 内 >=3 个不同省份自动封禁（防倒卖） banned, _info = auth.record_login_location(user, auth._client_ip()) if banned: auth.logout_session() return _err('检测到账号在多个省份异常登录，已自动封禁，请联系管理员') fresh = auth.get_user_by_id(user['id']) return _ok({ 'user': auth._user_public(fresh), 'duration_text': auth.format_credits(fresh['duration_seconds']), 'redirect': 'admin' if fresh['role'] == 'admin' else 'user', }, '登录成功') except Exception: return _err('登录失败，请稍后重试', 500) @app.route('/auth/admin_login/email', methods=['POST']) def auth_admin_login_email(): """管理员两步验证 Step2：比对绑定邮箱 → 发送 admin2fa 验证码（第 86 条）""" try: data = _data() ticket, v = _admin2fa_take(str(data.get('ticket') or '')) if not v: return _err('登录会话已过期，请重新开始登录', 401, 'admin_2fa_expired') email_input = auth._normalize_email(str(data.get('email') or '')) # 第 97 条（P3）：第二因子的比对走常时比较，与本文件密钥/令牌口径一致 if (not email_input or not v.get('email') or not hmac.compare_digest(email_input, v['email'])): v['attempts'] += 1 # 第 87 条：失败路径自增（原实现计数器从未生效） auth.add_log('fail', v['username'], auth._client_ip(), '管理员登录', '绑定邮箱不匹配') if v['attempts'] >= _ADMIN_2FA_MAX_ATTEMPTS: _ADMIN_2FA.pop(ticket, None) return _err('错误次数过多，请重新开始登录') return _err(f'绑定邮箱不正确，还可尝试 {_ADMIN_2FA_MAX_ATTEMPTS - v["attempts"]} 次') # 第 87 条：同账号 60s 发码软冷却——票据一次性+单活票后仍防持续轰炸绑定邮箱 last_send = _ADMIN2FA_LAST_SEND.get(v['username'], 0) if time.time() - last_send < _ADMIN2FA_SEND_GAP: return _err('验证码发送过于频繁，请 1 分钟后再试') auth.send_verify_code(v['email'], 'admin2fa') _ADMIN2FA_LAST_SEND[v['username']] = time.time() auth.add_log('success', v['username'], auth._client_ip(), '管理员登录', f'绑定邮箱验证通过，验证码已发送（{_admin2fa_mask(v["email"])}）') return _ok(message='验证码已发送至管理员邮箱，请查收') except ValueError as e: return _err(str(e)) except RuntimeError as e: return _err(str(e)) except Exception: return _err('验证码发送失败，请稍后重试', 500) @app.route('/auth/admin_login/verify', methods=['POST']) def auth_admin_login_verify(): """管理员两步验证 Step3：验证码原子消费 → 建立管理员会话（第 86 条）""" try: data = _data() ticket, v = _admin2fa_take(str(data.get('ticket') or '')) if not v: return _err('登录会话已过期，请重新开始登录', 401, 'admin_2fa_expired') code = str(data.get('code') or '').strip() if not auth.check_verify_code(v['email'], code, 'admin2fa'): v['attempts'] += 1 # 第 87 条：失败路径自增（原实现计数器从未生效） if v['attempts'] >= _ADMIN_2FA_MAX_ATTEMPTS: _ADMIN_2FA.pop(ticket, None) return _err('错误次数过多，请重新开始登录') auth.add_log('fail', v['username'], auth._client_ip(), '管理员登录', '验证码错误') return _err('验证码错误或已过期') # Step3 成功即销毁票据（一次性）；建会话前以库内最新状态复核 _ADMIN_2FA.pop(ticket, None) user = auth.get_user_by_id(v['user_id']) if not user or user['role'] != 'admin': return _err('登录会话已过期，请重新开始登录', 401, 'admin_2fa_expired') status = user['status'] if 'status' in user.keys() else 1 if status == 2: auth.add_log('fail', v['username'], auth._client_ip(), '管理员登录', '封禁账号尝试登录') return _err('账号已被封禁，请联系管理员') if status == 0: return _err('账号已被禁用，请联系管理员') auth.login_session(user) # IP 归属地异常检测：管理员天然豁免省份封禁（record_login_location 内部判定） banned, _info = auth.record_login_location(user, auth._client_ip()) if banned: auth.logout_session() return _err('检测到账号在多个省份异常登录，已自动封禁，请联系管理员') fresh = auth.get_user_by_id(user['id']) auth.add_log('success', v['username'], auth._client_ip(), '管理员登录', '两步验证通过') return _ok({ 'user': auth._user_public(fresh), 'duration_text': auth.format_credits(fresh['duration_seconds']), 'redirect': 'admin', }, '登录成功') except Exception: return _err('登录失败，请稍后重试', 500) @app.route('/auth/logout', methods=['POST']) def auth_logout(): auth.logout_session() return _ok(message='已退出登录') @app.route('/auth/heartbeat', methods=['POST']) def auth_heartbeat(): user = auth.current_user() if not user: return _err('登录已失效', 401, 'unauthorized') fresh = auth.get_user_by_id(user['id']) credits = int(fresh['duration_seconds'] or 0) if fresh['role'] != 'admin' and credits != -1 and credits <= 0: return _err('剩余次数已用完', 401, 'expired') return _ok({'duration_seconds': fresh['duration_seconds'], 'duration_text': auth.format_credits(fresh['duration_seconds'])}) @app.route('/auth/redeem', methods=['POST']) def auth_redeem(): user = auth.current_user() if not user: return _err('请先登录', 401, 'unauthorized') try: code = str(_data().get('code') or '') credits, remaining = auth.redeem_card(code, user['username']) auth.add_log('success', user['username'], auth._client_ip(), '卡密充值', f'{code[:4]}**** · +{credits} 次') return _ok({'credits': credits, 'duration_seconds': remaining, 'duration_text': auth.format_credits(remaining)}, f'充值成功，+{credits} 次') except ValueError as e: auth.add_log('fail', user['username'], auth._client_ip(), '卡密充值', str(e)) return _err(str(e)) except Exception: return _err('充值失败，请稍后重试', 500) @app.route('/auth/ann_seen', methods=['POST']) def auth_ann_seen(): user = auth.current_user() if user: ann = auth.get_announcement() auth._x('UPDATE users SET ann_seen = ? WHERE id = ?', (ann['updated_at'], user['id'])) return _ok() # -------------------- 自助找回密码（第 79 条：复用验证码通道） -------------------- @app.route('/auth/forgot_password', methods=['POST']) def auth_forgot_password(): """自助找回密码：注册邮箱收验证码 → 验码原子消费 → 重置密码并踢下全部会话。 安全口径：响应统一为「验证码错误或已过期」（不区分账号是否存在，防枚举）。""" try: data = _data() # 第 84 条修复：此处原为裸调用 `_normalize_email(...)`——该名称从未定义/ # 导入，NameError 被下方 except 吞成 500，导致本接口自第 79 条诞生起 # 任何输入都返回"重置失败"。改用 auth_system 的实现（邮箱归一化）。 email = auth._normalize_email(str(data.get('email') or '')) code = str(data.get('code') or '').strip() newpwd = str(data.get('password') or '') if len(newpwd) < 8 or len(newpwd) > 128: return _err('新密码长度需在 8-128 位之间') if not email: return _err('缺少邮箱') if not auth.check_verify_code(email, code, 'reset'): return _err('邮箱验证码错误或已过期') user = auth.get_user_by_email(email) if not user: # 验证码已对空邮箱消费掉——统一文案，不暴露账号存在性 return _err('邮箱验证码错误或已过期') # 第 86 条：管理员不走自助找回——放行则邮箱持有者可自助改密， # 改完的密码+邮箱验证码=两因子全握，2FA 形同虚设。统一防枚举文案 if user['role'] == 'admin': return _err('邮箱验证码错误或已过期') auth.reset_password(user['id'], newpwd) auth.add_log('success', user['username'], auth._client_ip(), '自助找回密码', '邮箱验证码重置') return _ok(message='密码已重置，请使用新密码登录（所有设备已下线）') except Exception: return _err('重置失败，请稍后重试', 500) # ==================== 管理后台 ==================== @app.route('/admin/site/users', methods=['POST']) @auth.admin_required def admin_site_users(): return _ok({'users': auth.list_users()}) @app.route('/admin/site/user/delete', methods=['POST']) @auth.admin_required def admin_site_user_delete(): try: uid = int(_data().get('id') or 0) target = auth.get_user_by_id(uid) if not auth.delete_user(uid): return _err('用户不存在或为管理员') auth.add_log('success', _actor(), auth._client_ip(), '删除用户', target['username'] if target else str(uid)) return _ok(message='用户已删除') except Exception as e: return _err(f'删除失败: {e!s}', 500) @app.route('/admin/site/user/duration', methods=['POST']) @auth.admin_required def admin_site_user_duration(): try: data = _data() uid = int(data.get('id') or 0) try: count = int(float(data.get('count') or data.get('minutes') or 0)) except (TypeError, ValueError, OverflowError): return _err('次数必须为数字') if count < -1: return _err('次数不能小于 -1（-1 表示不限次数）') target = auth.get_user_by_id(uid) if not target: return _err('用户不存在') if target['role'] == 'admin': return _err('管理员不限次数') auth.set_user_credits(uid, count) auth.add_log('success', _actor(), auth._client_ip(), '修改用户次数', f'{target["username"]} → {count} 次') return _ok(message=f'已修改为 {count} 次') except Exception as e: return _err(f'修改失败: {e!s}', 500) def _block_admin_target(target) -> tuple | None: """禁止对管理员账号执行危险操作（防自锁）""" if target and target['role'] == 'admin': return _err('不能对管理员账号执行该操作') return None @app.route('/admin/site/user/create', methods=['POST']) @auth.admin_required def admin_site_user_create(): """管理员手动添加用户""" try: data = _data() username = str(data.get('username') or '').strip() password = str(data.get('password') or '') email = auth._normalize_email(str(data.get('email') or '')) try: credits = int(float(data.get('count') or 0)) except (TypeError, ValueError, OverflowError): # OverflowError：count='inf' → int(inf) 溢出（原实现漏捕 → 500） return _err('次数必须为数字') import math as _math if not _math.isfinite(credits): return _err('次数必须为有限数字') if credits < -1: return _err('次数不能小于 -1（-1 表示不限次数）') if not re_username_ok(username): # 第 100 条：管理端为受信方——保留名给明确提示（注册端保持防枚举模糊文案） if username.lower() in _RESERVED_USERNAMES: return _err('该用户名为系统保留名，不可使用') return _err('用户名需 3-20 位，仅支持中文、字母、数字、下划线，且不能纯下划线') if len(password) < 8 or len(password) > 128: return _err('密码长度需在 8-128 位之间') if auth.get_user_by_name(username): return _err('用户名已存在') # R5-13：与注册路径同口径的邮箱查重——否则重复邮箱触发 UNIQUE 约束 # 走兜底 except，给管理员看原始 500 报错 if email and auth.get_user_by_email(email): return _err('该邮箱已被注册，请更换邮箱') auth.create_user(username, password, email=email, credits=credits, ip=auth._client_ip(), ua='admin创建') auth.add_log('success', _actor(), auth._client_ip(), '添加用户', f'{username} · {credits} 次') return _ok(message=f'用户 {username} 创建成功（{credits} 次）') except Exception as e: return _err(f'创建失败: {e!s}', 500) @app.route('/admin/site/user/reset_password', methods=['POST']) @auth.admin_required def admin_site_user_reset_password(): try: data = _data() uid = int(data.get('id') or 0) newpwd = str(data.get('password') or '') if len(newpwd) < 8 or len(newpwd) > 128: # 第 87 条：与注册/找回/建号统一 8 位下限（原 6 位为最后一处口径不一） return _err('新密码长度需在 8-128 位之间') target = auth.get_user_by_id(uid) if not target: return _err('用户不存在') current = auth.current_user() if target['role'] == 'admin': # F02：允许管理员修改自己的密码（需验证旧密码），否则文档要求的改密入口形同虚设 if not current or current['id'] != target['id']: return _err('不能对其他管理员账号执行该操作') oldpwd = str(data.get('old_password') or '') ok, _ = auth._verify_password(target['password_hash'], oldpwd, target['salt']) if not ok: return _err('原密码错误') auth.reset_password(uid, newpwd) auth.add_log('success', current['username'], auth._client_ip(), '修改自己的密码', '管理员') return _ok(message='密码已修改，下次登录请使用新密码（其他设备已全部下线）') auth.reset_password(uid, newpwd) auth.add_log('success', _actor(), auth._client_ip(), '重置密码', target['username']) return _ok(message=f'已重置 {target["username"]} 的密码') except Exception as e: return _err(f'重置失败: {e!s}', 500) @app.route('/admin/site/user/toggle', methods=['POST']) @auth.admin_required def admin_site_user_toggle(): """启用 / 禁用账号""" try: uid = int(_data().get('id') or 0) target = auth.get_user_by_id(uid) if not target: return _err('用户不存在') blocked = _block_admin_target(target) if blocked: return blocked new_status = 0 if (target['status'] if 'status' in target.keys() else 1) else 1 auth.set_user_status(uid, new_status) auth.add_log('success', _actor(), auth._client_ip(), '禁用/启用', f'{target["username"]} → {"启用" if new_status else "禁用"}') return _ok({'status': new_status}, message=f'已{"启用" if new_status else "禁用"} {target["username"]}') except Exception as e: return _err(f'操作失败: {e!s}', 500) @app.route('/admin/site/user/ban', methods=['POST']) @auth.admin_required def admin_user_ban(): """封禁/解封用户（防倒卖：封禁后立即踢下线）""" try: data = _data() uid = int(data.get('id') or 0) action = str(data.get('action') or 'ban') reason = str(data.get('reason') or '') row = auth.get_user_by_id(uid) if not row: return _err('用户不存在') if row['role'] == 'admin': return _err('不能对管理员执行该操作') admin_name = _actor() if action == 'ban': if not auth.ban_user(uid, reason or '管理员手动封禁'): return _err('封禁失败') auth.add_log('success', admin_name, auth._client_ip(), '封禁用户', f'{row["username"]}: {reason or "手动封禁"}') return _ok(message=f'已封禁用户「{row["username"]}」，该账号已强制下线') else: if not auth.unban_user(uid): return _err('解封失败') auth.add_log('success', admin_name, auth._client_ip(), '解封用户', row['username']) return _ok(message=f'已解封用户「{row["username"]}」') except Exception as e: return _err(f'操作失败: {e!s}', 500) @app.route('/admin/site/user/cookie', methods=['POST']) @auth.admin_required def admin_site_user_cookie(): username = str(_data().get('username') or '') user = auth.get_user_by_name(username) if not user: return _err('用户不存在', 404) return _ok({'username': username, 'cookie': user['netease_cookie'] or ''}) @app.route('/admin/site/smtp/get', methods=['POST']) @auth.admin_required def admin_site_smtp_get(): cfg = auth.get_smtp_config() return _ok({'host': cfg['host'], 'port': cfg['port'], 'sender': cfg['sender'], 'configured': auth.smtp_configured()}) @app.route('/admin/site/smtp/save', methods=['POST']) @auth.admin_required def admin_site_smtp_save(): try: data = _data() host = str(data.get('host') or '').strip() try: port = int(data.get('port') or 587) except (TypeError, ValueError): return _err('端口必须为数字') sender = str(data.get('sender') or '').strip() password = str(data.get('password') or '') if not host or not sender: return _err('SMTP 服务器与发件邮箱不能为空') if not password: # 密码留空表示保持原有授权码不变 password = auth.get_smtp_config().get('password', '') auth.save_smtp_config(host, port, sender, password) auth.add_log('success', _actor(), auth._client_ip(), '保存SMTP配置', f'{host}:{port}') return _ok(message='SMTP 配置已保存') except Exception as e: return _err(f'保存失败: {e!s}', 500) @app.route('/admin/site/smtp/test', methods=['POST']) @auth.admin_required def admin_site_smtp_test(): try: to = str(_data().get('to') or '').strip() if not to: return _err('请填写测试收件邮箱') # 收件人地址校验（防 CRLF 头注入；与验证码发送同规则） import re as _re if not _re.match(r'^[\w.%+-]+@[\w.-]+\.[A-Za-z]{2,}$', to): return _err('收件邮箱格式不正确') auth.send_email(to, 'SMTP 测试邮件', '这是一封来自云村小助手的测试邮件，收到即说明 SMTP 配置成功。') auth.add_log('success', _actor(), auth._client_ip(), 'SMTP测试', to) return _ok(message='测试邮件已发送，请查收') except Exception as e: auth.add_log('fail', _actor(), auth._client_ip(), 'SMTP测试', str(e)) return _err(f'发送失败: {e!s}', 500) @app.route('/admin/site/logs', methods=['POST']) @auth.admin_required def admin_site_logs(): """日志分页查询（10 条/页）""" data = _data() log_type = 'fail' if str(data.get('type')) == 'fail' else 'success' try: page = max(1, int(data.get('page') or 1)) except (TypeError, ValueError): page = 1 rows, total = auth.query_logs(log_type, limit=10, offset=(page - 1) * 10) return _ok({'logs': rows, 'type': log_type, 'page': page, 'total': total, 'pages': max(1, (total + 9) // 10)}) @app.route('/admin/site/logs/clear', methods=['POST']) @auth.admin_required def admin_site_logs_clear(): """清除指定类型的全部日志""" log_type = 'fail' if str(_data().get('type')) == 'fail' else 'success' deleted = auth.clear_logs(log_type) auth.add_log('success', _actor(), auth._client_ip(), '清除日志', f'{log_type} · {deleted} 条') return _ok(message=f'已清除 {"失败" if log_type == "fail" else "成功"}日志 {deleted} 条') @app.route('/admin/site/cards/generate', methods=['POST']) @auth.admin_required def admin_site_cards_generate(): try: data = _data() try: count = int(data.get('count') or 1) valid_days = int(data.get('valid_days') or 0) except (TypeError, ValueError): return _err('张数 / 次数 / 有效天数必须为数字') # 第 81 条：显式传 0（JSON 体）不再被 or 链静默替换成默认 5 raw_credits = data.get('credits', data.get('minutes')) try: credits = int(raw_credits) if raw_credits is not None else 5 except (TypeError, ValueError): return _err('张数 / 次数 / 有效天数必须为数字') if credits <= 0: return _err('卡密次数必须大于 0') prefix = str(data.get('prefix') or 'VIP') note = str(data.get('note') or '') codes = auth.generate_cards(count, credits, created_by=_actor(), prefix=prefix, note=note, valid_days=valid_days) auth.add_log('success', _actor(), auth._client_ip(), '生成卡密', f'{count} 张 · {credits} 次/张 · 有效{valid_days or "永久"}天') return _ok({'codes': codes, 'credits': credits, 'count': len(codes)}, f'已生成 {len(codes)} 张卡密') except Exception as e: return _err(f'生成失败: {e!s}', 500) @app.route('/admin/site/cards/list', methods=['POST']) @auth.admin_required def admin_site_cards_list(): # F85：支持 limit/offset 分页与总数返回，较早卡密可翻页查看 data = _data() try: limit = int(data.get('limit') or 100) except (TypeError, ValueError): limit = 100 try: offset = int(data.get('offset') or 0) except (TypeError, ValueError): offset = 0 cards, total = auth.list_cards(limit=limit, offset=offset) return _ok({'cards': cards, 'total': total}) @app.route('/admin/site/cards/disable', methods=['POST']) @auth.admin_required def admin_site_cards_disable(): # 第 81 条：禁用/恢复未使用卡密（可逆操作；已使用卡密无禁用意义） try: data = _data() code = str(data.get('code') or '').strip() if not code: return _err('缺少卡密') disable = str(data.get('disabled') or '').strip().lower() in ('1', 'true') # 第 87 条：兼容布尔/大小写变体 auth.set_card_disabled(code, disable) auth.add_log('success', _actor(), auth._client_ip(), '禁用卡密' if disable else '恢复卡密', code[:4] + '****') return _ok(message=('已禁用 ' if disable else '已恢复 ') + code[:4] + '****') except ValueError as e: return _err(str(e)) except Exception: return _err('操作失败，请稍后重试', 500) @app.route('/admin/site/cards/delete', methods=['POST']) @auth.admin_required def admin_site_cards_delete(): # 第 81 条：删除未使用卡密（已使用卡密保留兑换凭证，按 owner 规则不可删） try: data = _data() code = str(data.get('code') or '').strip() if not code: return _err('缺少卡密') auth.delete_card(code) auth.add_log('success', _actor(), auth._client_ip(), '删除卡密', code[:4] + '****') return _ok(message='已删除 ' + code[:4] + '****') except ValueError as e: return _err(str(e)) except Exception: return _err('操作失败，请稍后重试', 500) @app.route('/admin/site/ann/get', methods=['POST']) @auth.admin_required def admin_site_ann_get(): return _ok(auth.get_announcement()) @app.route('/admin/site/ann/save', methods=['POST']) @auth.admin_required def admin_site_ann_save(): try: data = _data() auth.save_announcement(str(data.get('title') or ''), str(data.get('content') or ''), str(data.get('popup_mode') or 'off')) auth.add_log('success', _actor(), auth._client_ip(), '保存公告', str(data.get('title') or '')) return _ok(message='公告已保存') except Exception as e: return _err(f'保存失败: {e!s}', 500) # -------------------- 播放 API 密钥 -------------------- @app.route('/admin/site/play_key/get', methods=['POST']) @auth.admin_required def admin_site_play_key_get(): """获取当前播放 API 密钥（不存在则自动生成一次，保证弹窗拼 URL 总有可用密钥）""" key = auth.get_setting('playlist_play_key', '') if not key: key = secrets.token_urlsafe(16) auth.set_setting('playlist_play_key', key) return _ok({'key': key}) @app.route('/admin/site/play_key/rotate', methods=['POST']) @auth.admin_required def admin_site_play_key_rotate(): """刷新播放 API 密钥：旧密钥立即失效（已签发的音频地址按 36 分钟自然过期（随交付保留期联动））""" key = secrets.token_urlsafe(16) auth.set_setting('playlist_play_key', key) auth.add_log('success', _actor(), auth._client_ip(), '刷新播放API密钥', '旧密钥已失效') return _ok({'key': key}, '密钥已刷新，旧密钥已全部失效') # -------------------- 数据库管理 -------------------- @app.route('/admin/site/backup/create', methods=['POST']) @auth.admin_required def admin_site_backup_create(): try: name = auth.create_backup() auth.set_setting('last_backup_ts', str(time.time())) auth.add_log('success', _actor(), auth._client_ip(), '数据库备份', name) return _ok({'name': name}, f'备份完成：{name}') except Exception as e: return _err(f'备份失败: {e!s}', 500) @app.route('/admin/site/backup/list', methods=['POST']) @auth.admin_required def admin_site_backup_list(): return _ok({'backups': auth.list_backups(), 'interval_hours': auth.get_setting('backup_interval_hours', '0'), 'db_size': auth.DB_PATH.stat().st_size if auth.DB_PATH.exists() else 0}) @app.route('/admin/site/backup/download', methods=['GET']) @auth.admin_required def admin_site_backup_download(): name = request.args.get('name') or '' p = auth.backup_path(name) if not p: return _err('备份文件不存在', 404) resp = send_file(str(p), as_attachment=True, download_name=name) # 第 97 条（P3）：备份含完整用户库（邮箱+哈希），禁止共享缓存存储 resp.headers['Cache-Control'] = 'no-store' return resp @app.route('/admin/site/backup/upload', methods=['POST']) @auth.admin_required def admin_site_backup_upload(): dest = None try: f = request.files.get('file') if not f or not f.filename: return _err('请选择数据库文件') if not f.filename.endswith('.db'): return _err('仅支持 .db 文件') auth.BACKUP_DIR.mkdir(parents=True, exist_ok=True) # S6：同秒上传自动追加序号，不再互相覆盖 base = 'upload_' + time.strftime('%Y%m%d_%H%M%S') name = base + '.db' k = 1 while (auth.BACKUP_DIR / name).exists(): name = f'{base}-{k}.db' k += 1 dest = auth.BACKUP_DIR / name f.save(str(dest)) # 只读文件头 16 字节做魔数校验（原实现 read_bytes() 会把整个备份读进内存） with dest.open('rb') as hf: header = hf.read(16) if not header.startswith(b'SQLite format 3'): dest.unlink() return _err('不是有效的 SQLite 数据库文件') auth.add_log('success', _actor(), auth._client_ip(), '上传数据库', dest.name) # 第 93 条（P2-3）：上传备份同样纳入配额裁剪（20+5） try: auth.prune_backups(20, 5) except Exception: pass return _ok({'name': dest.name}, '上传成功，可在下方选择该备份进行恢复') except Exception as e: # 第 99 条：上传中途失败（ENOSPC/IO）清理半截文件——残缺 .db 会进 # list_backups 且因 mtime 最新挤掉一份好备份（恢复前校验虽能拒收） try: if dest and dest.exists(): dest.unlink() except Exception: pass return _err(f'上传失败: {e!s}', 500) @app.route('/admin/site/backup/restore', methods=['POST']) @auth.admin_required def admin_site_backup_restore(): try: name = str(_data().get('name') or '') auth.restore_backup(name) auth.add_log('success', _actor(), auth._client_ip(), '恢复数据库', name) return _ok(message='数据库已恢复，请刷新页面') except ValueError as e: # 第 95 条（P3-1）：客户端可纠正错误（备份不存在/校验失败）返回 400， # 与 delete 的 404 口径对齐——原实现一律 500 return _err(f'恢复失败: {e!s}', 400) except Exception as e: auth.add_log('fail', _actor(), auth._client_ip(), '恢复数据库', str(e)) return _err(f'恢复失败: {e!s}', 500) @app.route('/admin/site/backup/delete', methods=['POST']) @auth.admin_required def admin_site_backup_delete(): try: name = str(_data().get('name') or '') if auth.delete_backup(name): return _ok(message='备份已删除') return _err('备份文件不存在', 404) except Exception as e: return _err(f'删除失败: {e!s}', 500) @app.route('/admin/site/backup/auto', methods=['POST']) @auth.admin_required def admin_site_backup_auto(): try: import math as _math try: hours = float(_data().get('hours') or 0) except (TypeError, ValueError): return _err('间隔必须为数字') # F46：NaN/Infinity 一律拒绝，不能持久化后让调度循环静默失效 if not _math.isfinite(hours): return _err('间隔必须为有限数字') if hours < 0: return _err('间隔不能为负') auth.set_setting('backup_interval_hours', str(hours)) return _ok(message=f'定时备份已开启，每 {hours} 小时自动备份' if hours > 0 else '定时备份已关闭') except Exception as e: return _err(f'设置失败: {e!s}', 500) # ============ 辅助 ============ _RESERVED_USERNAMES = {'admin', 'administrator', 'qfbsh', 'root', 'system', 'moderator', '管理员', '客服'} def re_username_ok(username: str) -> bool: import re as _re # P3-10：禁纯下划线/下划线开头结尾（"____"、"___admin" 有钓鱼观感风险） # R5-P3：白名单收紧为中文/字母/数字/下划线——`\w` 会放行西里尔等同形字符 #（"аdmin" 冒充钓鱼）；保留字（admin/qfbsh 等）一并拒绝，防冒充管理入口 return bool(_re.fullmatch(r'[A-Za-z0-9_\u4e00-\u9fa5]{3,20}', username)) and \ not _re.match(r'^_+$', username) and not username.startswith('_') and not username.endswith('_') \ and username.lower() not in _RESERVED_USERNAMES def secrets_rand() -> int: import secrets as _s return _s.randbelow(900000) + 100000