From 6c3181341bab092688e3d645f1c6789c38eb2978 Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sat, 26 Sep 2026 01:15:22 +0800 Subject: [PATCH] =?UTF-8?q?feat(routes):=20=E8=BF=81=E7=A7=BB=E8=B4=A6?= =?UTF-8?q?=E5=8F=B7/=E5=85=A8=E5=B1=80=E6=AD=8C=E5=8D=95=E7=AB=AF?= =?UTF-8?q?=E7=82=B9=20+=20=E9=89=B4=E6=9D=83=E4=BE=9D=E8=B5=96=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 迁移路由（/api/v1/*，与 Flask 同端点逐字段对齐）： - account/vip：admin/user 双路径 + 5 分钟缓存（降级结果 30s）， VIP 探测单飞锁与 Flask 同源跨栈共享；未登录可访问（同 Flask） - global/playlists、global/playlist：管理员闸（require_admin，403 对齐） - web/deps.py 新增 require_admin；_resolve_user_cookies 支持可选用户（匿名审计保底） 测试：46/46（新增 10 项鉴权依赖：有效/管理员/封禁/禁用/顶号/伪造 Cookie、 401/403 闸）；ruff All checks passed；pyright 0 errors。 契约奇偶校验：16 项全 OK（新增 account:vip、global:playlists/playlist、 非管理员 403）；uvicorn 实跑双栈 account/vip 均 200、未授权歌单 403。 --- tests/test_deps.py | 90 ++++++++++++++++++++ web/app.py | 2 + web/deps.py | 9 ++ web/routers/account.py | 181 +++++++++++++++++++++++++++++++++++++++++ web/routers/music.py | 7 +- 5 files changed, 287 insertions(+), 2 deletions(-) create mode 100644 tests/test_deps.py create mode 100644 web/routers/account.py diff --git a/tests/test_deps.py b/tests/test_deps.py new file mode 100644 index 0000000..d18ac83 --- /dev/null +++ b/tests/test_deps.py @@ -0,0 +1,90 @@ +# -*- coding: utf-8 -*- +"""FastAPI 鉴权依赖测试：会话解析、封禁/顶号拒绝、登录与管理员闸。""" +import sqlite3 +import time + +import pytest + +import core.db as db +from core import session as cs +from web.deps import SiteUser, current_user, require_admin, require_login +from web.errors import AppError + +SECRET = 'deps-test-secret' + + +@pytest.fixture() +def site(tmp_path, monkeypatch): + path = tmp_path / 'site.db' + conn = sqlite3.connect(str(path)) + conn.executescript(""" + CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT, role TEXT DEFAULT 'user', + status INTEGER DEFAULT 1, login_seq INTEGER DEFAULT 0, + duration_seconds INTEGER DEFAULT 0, ban_reason TEXT DEFAULT ''); + INSERT INTO users (id, username, role, status, login_seq, duration_seconds) + VALUES (1,'qfbsh','admin',1,376,-1), + (26,'normal','user',1,5,97), + (27,'banned','user',2,1,0), + (28,'disabled','user',0,1,0); + """) + conn.commit() + conn.close() + monkeypatch.setattr(db, 'DATA_DIR', tmp_path) + monkeypatch.setattr(db, 'DB_PATH', path) + monkeypatch.setattr('web.deps.secret_key', lambda: SECRET) + return path + + +def _cookie(uid, seq, ts=None): + return cs.dumps(SECRET, {'site_uid': uid, 'site_uid_seq': seq, + 'site_uid_ts': ts or int(time.time())}) + + +async def test_valid_user(site): + u = await current_user(_cookie(26, 5)) + assert u is not None and u.id == 26 and u.username == 'normal' + assert not u.is_admin and not u.unlimited + + +async def test_admin_unlimited(site): + u = await current_user(_cookie(1, 376)) + assert u is not None and u.is_admin and u.unlimited + + +async def test_missing_cookie(site): + assert await current_user(None) is None + + +async def test_banned_rejected(site): + assert await current_user(_cookie(27, 1)) is None + + +async def test_disabled_rejected(site): + assert await current_user(_cookie(28, 1)) is None + + +async def test_stale_login_seq_rejected(site): + """顶号：会话版本与库不一致 → 视为未登录。""" + assert await current_user(_cookie(26, 4)) is None + + +async def test_forged_cookie_rejected(site): + assert await current_user('not-a-valid-cookie') is None + + +async def test_require_login_raises(site): + with pytest.raises(AppError) as ei: + await require_login(None) + assert ei.value.status_code == 401 + + +async def test_require_admin_forbidden_for_user(site): + u = SiteUser(id=26, username='normal', role='user', credits=97) + with pytest.raises(AppError) as ei: + await require_admin(u) + assert ei.value.status_code == 403 + + +async def test_require_admin_ok(site): + u = SiteUser(id=1, username='qfbsh', role='admin', credits=-1) + assert (await require_admin(u)).is_admin diff --git a/web/app.py b/web/app.py index c9bd70a..4ee725a 100644 --- a/web/app.py +++ b/web/app.py @@ -21,6 +21,7 @@ from core.config import APP_VERSION from core.security import init_secret_key from web.errors import install_error_handlers from web.middleware import ResponseHygieneMiddleware +from web.routers import account as account_router from web.routers import music as music_router from web.routers import site as site_router from web.routers import system as system_router @@ -69,6 +70,7 @@ def create_app(serve_flask_fallback: bool = False) -> FastAPI: app.include_router(system_router.router, prefix='/api/v1', tags=['system']) app.include_router(site_router.router, prefix='/api/v1', tags=['site']) app.include_router(music_router.router, prefix='/api/v1', tags=['music']) + app.include_router(account_router.router, prefix='/api/v1', tags=['account']) install_error_handlers(app) diff --git a/web/deps.py b/web/deps.py index 53b1871..de04242 100644 --- a/web/deps.py +++ b/web/deps.py @@ -79,3 +79,12 @@ async def require_login( if user is None: raise AppError('请先登录后再使用解析功能', 401, error_code='unauthorized') return user + + +async def require_admin( + user: Annotated[SiteUser | None, Depends(current_user)], +) -> SiteUser: + """管理员闸：非管理员抛 403（与 Flask admin_required 同形包体）。""" + if user is None or not user.is_admin: + raise AppError('需要管理员权限', 403, error_code='forbidden') + return user diff --git a/web/routers/account.py b/web/routers/account.py new file mode 100644 index 0000000..4ae097d --- /dev/null +++ b/web/routers/account.py @@ -0,0 +1,181 @@ +# -*- coding: utf-8 -*- +"""账号与全局歌单端点（FastAPI 栈）——与 Flask 侧同名端点响应契约逐字段对齐。 + +已迁移： +- POST/GET /api/v1/account/vip ↔ Flask /account/vip（admin/user 双路径 + 缓存） +- POST/GET /api/v1/global/playlists ↔ Flask /global/playlists（管理员） +- POST/GET /api/v1/global/playlist ↔ Flask /global/playlist（管理员） +""" +import hashlib +import time +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, Request + +import music_api +from core.state import VIP_CACHE, VIP_PROBE_LOCK, VIP_USER_CACHE +from music_api import APIException +from web import ratelimit +from web.deps import SiteUser, current_user, require_admin +from web.response import error, success + +router = APIRouter() + + +AdminUser = Annotated[SiteUser, Depends(require_admin)] + +_VIP_NAMES = { + 0: '普通用户', 1: '音乐包', 11: '黑胶VIP', 100: '黑胶VIP', 10: '黑胶SVIP', 110: '黑胶SVIP', +} + +# 探测单飞：与 Flask _VIP_PROBE_LOCK 同源（跨栈共享，避免两栈各自串行 8 级探测） +_VIP_PROCESS_LOCK: Any = VIP_PROBE_LOCK + + +def _vip_name(vip_type: int) -> str: + return _VIP_NAMES.get(vip_type, '普通用户') + + +def _cookies() -> dict[str, str]: + return music_api.load_cookies() + + +async def _detect_vip(cookies: dict[str, str]) -> tuple[int, str, list[str]]: + """检测账号会员类型与可解析音质（与 Flask _detect_vip 同口径）。 + + 单飞锁为 threading.Lock：异步上下文里用 to_thread 有界等待 0.5s， + 拿不到锁的请求直接用账号接口给降级结果（不等 8 级探测）。 + """ + import asyncio + + api = music_api.NeteaseAPI() + acquired = await asyncio.to_thread(_VIP_PROCESS_LOCK.acquire, True, 0.5) + if not acquired: + info = await api.get_account_info(cookies) + if info: + return info.get('vipType', 0), 'account', [] + return 0, 'probe', [] + try: + probe = await api.detect_vip_by_probe(cookies) + levels_ok = probe.get('levels_ok', []) + info = await api.get_account_info(cookies) + if info: + return info.get('vipType', 0), 'account', levels_ok + return probe.get('vip_type', 0), 'probe', levels_ok + finally: + _VIP_PROCESS_LOCK.release() + + +async def _request_data(request: Request) -> dict[str, Any]: + data: dict[str, Any] = {} + if request.method == 'GET': + data.update(dict(request.query_params)) + return data + try: + body = await request.json() + if isinstance(body, dict): + data.update(body) + except Exception: + pass + if not data: + try: + form = await request.form() + data.update(dict(form.items())) + except Exception: + pass + return data + + +@router.api_route('/account/vip', methods=['GET', 'POST']) +async def account_vip( + user: Annotated[SiteUser | None, Depends(current_user)], + request: Request, +): + """当前解析账号的 VIP 类型（admin=全局账号 / user=个人账号）。无需登录。""" + await ratelimit.guard(request, bucket='free', limit=30, window=60) + data = await _request_data(request) + account = data.get('account', 'admin') + api = music_api.NeteaseAPI() + + try: + if account == 'user': + from web.routers.music import _resolve_user_cookies + user_cookies = await _resolve_user_cookies(data, user, request) + if not user_cookies: + return success({'account': 'user', 'logged_in': False, 'vip_type': None, + 'vip_name': '未登录', 'levels_ok': []}, '个人账号未登录') + ck = '|'.join(f'{k}={v}' for k, v in sorted(user_cookies.items())) + ck_hash = hashlib.md5(ck.encode('utf-8')).hexdigest() + cached_user = VIP_USER_CACHE.get(ck_hash) + if cached_user is not None and time.time() - cached_user['ts'] <= cached_user.get('ttl', 300): + return success(dict(cached_user['data'], account='user'), '获取个人账号信息成功（缓存）') + vip_type, method, levels_ok = await _detect_vip(user_cookies) + info = await api.get_account_info(user_cookies) + user_payload = { + 'account': 'user', 'logged_in': True, 'vip_type': vip_type, + 'vip_name': _vip_name(vip_type), 'detected_by': method, 'levels_ok': levels_ok, + 'nickname': (info or {}).get('nickname', ''), + } + # 降级结果（levels_ok 空）短缓存 30s，正常结果 300s（与 Flask 同口径） + ttl = 30 if not levels_ok else 300 + if len(VIP_USER_CACHE) > 500: + VIP_USER_CACHE.clear() + VIP_USER_CACHE[ck_hash] = {'ts': time.time(), 'ttl': ttl, 'data': user_payload} + return success(user_payload, '获取个人账号信息成功') + + cached = VIP_CACHE['data'] + if cached is None or time.time() - VIP_CACHE['ts'] > VIP_CACHE.get('ttl', 300): + admin_cookies = _cookies() + vip_type, method, levels_ok = await _detect_vip(admin_cookies) + info = await api.get_account_info(admin_cookies) + cached = { + 'vip_type': vip_type, 'vip_name': _vip_name(vip_type), + 'detected_by': method, 'levels_ok': levels_ok, + 'nickname': (info or {}).get('nickname', ''), + 'logged_in': info is not None, + 'note': '' if info else '账号接口未返回会员信息，已通过歌曲音质探测接口检测', + } + VIP_CACHE['data'] = cached + VIP_CACHE['ts'] = time.time() + VIP_CACHE['ttl'] = 30 if not levels_ok else 300 + payload = dict(cached) + payload['account'] = 'admin' + payload['nickname'] = cached['nickname'] if (user and user.is_admin) else '' + return success(payload, '获取全局账号信息成功') + except Exception: + return error('服务暂时不可用，请稍后再试', 500) + + +@router.api_route('/global/playlists', methods=['GET', 'POST']) +async def global_playlists(_admin: AdminUser, request: Request): + """全局账号收藏/创建的歌单（含「我喜欢的音乐」）。""" + await ratelimit.guard(request, bucket='free', limit=30, window=60) + api = music_api.NeteaseAPI() + cookies = _cookies() + try: + account = await api.get_account_info(cookies) + if not account or not account.get('userId'): + return error('全局账号未登录或 Cookie 已失效，请先在「管理员设置」扫码登录', 401) + playlists = await api.get_user_playlists(account['userId'], cookies) + return success({'user': account, 'playlists': playlists}, '获取全局账号歌单成功') + except APIException: + return error('服务暂时不可用，请稍后再试', 500) + + +@router.api_route('/global/playlist', methods=['GET', 'POST']) +async def global_playlist(_admin: AdminUser, request: Request): + """全局账号歌单详情（全局 Cookie，支持「我喜欢的音乐」等私有歌单）。""" + await ratelimit.guard(request, bucket='free', limit=30, window=60) + from web.routers.music import _extract_id + data = await _request_data(request) + playlist_id = data.get('id') + if not playlist_id: + return error("参数 'playlist_id' 不能为空", 400) + playlist_id = _extract_id(playlist_id) + if not str(playlist_id).strip().isdigit(): + return error('无效的歌单ID，请输入数字ID或分享链接') + try: + result = await music_api.NeteaseAPI().get_playlist_detail(int(playlist_id), _cookies()) + return success({'playlist': result}, '获取全局账号歌单详情成功') + except APIException: + return error('服务暂时不可用，请稍后再试', 500) diff --git a/web/routers/music.py b/web/routers/music.py index 68d3a6a..16efe61 100644 --- a/web/routers/music.py +++ b/web/routers/music.py @@ -243,7 +243,9 @@ def _global_music_u() -> str: return GLOBAL_MUSIC_U_CACHE['v'] -async def _resolve_user_cookies(data: dict[str, Any], user: SiteUser, request: Request) -> dict[str, str]: +async def _resolve_user_cookies( + data: dict[str, Any], user: 'SiteUser | None', request: Request +) -> dict[str, str]: """解析个人账号 Cookie（浏览器传入的 user_cookie，不落服务器）。 全局账号（cookie.txt）零个人身份权限：浏览器提交的 MUSIC_U 若与全局账号一致 @@ -255,7 +257,8 @@ async def _resolve_user_cookies(data: dict[str, Any], user: SiteUser, request: R if mu and gmu and hmac.compare_digest(mu, gmu): try: import auth_system - await asyncio.to_thread(auth_system.add_log, 'fail', user.username, + await asyncio.to_thread(auth_system.add_log, 'fail', + user.username if user else '匿名', ratelimit.client_ip(request), '全局账号冒用拦截', '浏览器提交的 MUSIC_U 与全局账号一致，已按未登录拒绝') except Exception: