Archived
snowy-yang
0569b50737
为什么：用 AST 级引用分析（而非文本 grep——同名符号会互相掩盖）确认这些 符号全项目零引用。文本 grep 查不出的原因是它们都有「活着的同名兄弟」： services/downloads.py（共 6 个，93 行） _parse_cookie_str / _global_music_u / _vip_name / _detect_vip / _admin_client / _user_client —— 活的是 core.media.parse_cookie_str、web/routers/music.py、account.py 的版本 随之失效的 import：USER_COOKIE_FILE / GLOBAL_MUSIC_U_CACHE / VIP_PROBE_LOCK / account_info / detect_vip_by_probe / QRLoginClient services/api_service.py APIResponse 类（29 行）—— web/response.py 才是活的响应封装 services/_shared.py _rate_ok —— 逐字复制 web.ratelimit.rate_ok core/utils.py safe_int / safe_console_print —— 活的是 web/request_utils.safe_int core/repo.py all_settings / get_user_by_name —— auth_system.get_user_by_name 才是活的 core/db.py / web/auth_state.py / web/request_utils.py execute_many / _safe_register_credits + secrets_rand / dedup_ids 结果：生产代码死代码候选 13 → 0（余下 118 个全是 pytest 测试函数，属正常）。
156 lines
5.5 KiB
Python
156 lines
5.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""认证的内存态与校验辅助（模块级状态，框架无关）。
|
||
|
||
包含：登录失败锁定、管理员 2FA 票据、发码 IP 限流、用户名/保留名校验。
|
||
这些状态**进程内共享**——登录/注册/2FA 各端点都读写同一份，避免「换个端点绕过锁定」。
|
||
"""
|
||
import secrets
|
||
import time
|
||
|
||
# ---- 登录失败锁定（按用户名维度，10 分钟窗口内 >=5 次临时锁定）----
|
||
_LOGIN_FAILS: dict = {}
|
||
_LOGIN_FAIL_SWEEP_TS = 0.0
|
||
# 本锁定周期已落过审计日志的用户名（解锁或成功登录即重置）
|
||
_LOGIN_FAIL_LOGGED: set = set()
|
||
|
||
# ---- 管理员两步验证票据（内存态）----
|
||
_ADMIN_2FA: dict = {}
|
||
_ADMIN2FA_LAST_SEND: dict = {}
|
||
|
||
|
||
_LOGIN_FAIL_LIMIT = 5
|
||
|
||
|
||
_LOGIN_FAIL_WINDOW = 600
|
||
|
||
|
||
_LOGIN_FAIL_SWEEP_TS = 0.0
|
||
|
||
|
||
_ADMIN_2FA_TTL = 300 # 票据 5 分钟有效
|
||
|
||
|
||
_ADMIN_2FA_MAX_ATTEMPTS = 5 # 邮箱比对 + 验证码合计尝试上限
|
||
|
||
|
||
_ADMIN_2FA_MAX_TICKETS = 200 # 全局活票上限（防内存膨胀，挤掉最旧）
|
||
|
||
|
||
_ADMIN2FA_SEND_GAP = 20 # 同账号两次发码最小间隔（秒）——初版 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()
|
||
# 每管理员仅 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:
|
||
"""锁定周期内的重复被封尝试只落一条 fail 日志，防多 IP 刷审计表；
|
||
解锁（窗口过期或成功登录）后进入新周期，才允许再记一条。"""
|
||
if username in _LOGIN_FAIL_LOGGED:
|
||
return False
|
||
_LOGIN_FAIL_LOGGED.add(username)
|
||
return True
|
||
|
||
|
||
_SEND_CODE_IP = {}
|
||
|
||
|
||
_SEND_CODE_IP_LIMIT = 10
|
||
|
||
|
||
_SEND_CODE_IP_WINDOW = 3600
|
||
|
||
|
||
def _send_code_ip_check(ip: str) -> bool:
|
||
"""返回 True 表示允许发送；超限返回 False（表超限时顺手清理过期项，防无界增长）"""
|
||
now = time.time()
|
||
if len(_SEND_CODE_IP) > 10000:
|
||
stale = [k for k, (_, 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
|
||
|
||
|
||
_RESERVED_USERNAMES = {'admin', 'administrator', 'qfbsh', 'root', 'system', 'moderator', '管理员', '客服'}
|
||
|
||
|
||
def re_username_ok(username: str) -> bool:
|
||
import re as _re
|
||
# 禁纯下划线/下划线开头结尾（"____"、"___admin" 有钓鱼观感风险）
|
||
# 白名单收紧为中文/字母/数字/下划线——`\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
|