From a2758aa0dac4f154e3b003a9ec7e731278d3b8f6 Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sat, 26 Sep 2026 01:12:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(routes):=20=E8=BF=81=E7=A7=BB=E8=A7=A3?= =?UTF-8?q?=E6=9E=90=E7=AB=AF=E7=82=B9=20/song=20/playlist=20/album=20+=20?= =?UTF-8?q?=E5=93=8D=E5=BA=94=E5=8D=AB=E7=94=9F=E4=B8=AD=E9=97=B4=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 迁移路由（/api/v1/*，与 Flask 同端点逐字段对齐）： - song（type: url/name/lyric/json）、playlist、album 个人账号路径实现全局账号冒用拦截（同 Flask 口径）；json 模式 asyncio.gather 并发 song/url/lyric（歌词失败不阻断，与 Flask 线程池语义一致） - core/media.py：format_file_size/quality_display_name/parse_cookie_str 纯函数复用 web/middleware.py（镜像 Flask after_request）： - 音频直链 → HMAC 签名代理地址（密钥同源，两栈 token 互认） - 安全头 nosniff/DENY/Referrer-Policy/CSP 基线；Server 抹为 NeteaseUrl；429 带 Retry-After - 纯 ASGI 逐段转发（不破坏流式下载）；JSON 200 延迟 start 并**重算 Content-Length** ——修实测 IncompleteRead（原实现转发旧长度 + 新 body） 契约奇偶校验：12 项全 OK（新增 album/playlist/song×3）； token/CDN 主机归一化（均为易变值，非契约漂移）。 验证：pytest 36/36；ruff All checks passed；pyright 0 errors； uvicorn 实跑 /api/v1/song 直链已代理、CSP/Server 头正确。 --- core/media.py | 52 +++++++++ tests/test_middleware.py | 70 ++++++++++++ web/app.py | 18 ++++ web/middleware.py | 97 +++++++++++++++++ web/routers/music.py | 223 +++++++++++++++++++++++++++++++++++++-- 5 files changed, 452 insertions(+), 8 deletions(-) create mode 100644 core/media.py create mode 100644 tests/test_middleware.py create mode 100644 web/middleware.py diff --git a/core/media.py b/core/media.py new file mode 100644 index 0000000..7466f34 --- /dev/null +++ b/core/media.py @@ -0,0 +1,52 @@ +# -*- coding: utf-8 -*- +"""媒体相关的纯函数（文件大小/音质显示名/Cookie 串解析）——框架无关。 + +与 main.MusicAPIService 上的同名方法保持逐字一致的实现， +供 FastAPI 栈直接复用，避免两栈各写一份而漂移。 +""" + + +def format_file_size(size_bytes) -> str: + """格式化文件大小（上游 size 可能为 null/垃圾值：容错返回"大小未知"而非 500）""" + try: + size = float(size_bytes or 0) + except (TypeError, ValueError): + return "大小未知" + if size <= 0: + return "0B" + + units = ["B", "KB", "MB", "GB", "TB"] + unit_index = 0 + while size >= 1024.0 and unit_index < len(units) - 1: + size /= 1024.0 + unit_index += 1 + return f"{size:.2f}{units[unit_index]}" + + +_QUALITY_NAMES = { + 'standard': '标准音质', + 'exhigh': '极高音质', + 'lossless': '无损音质', + 'hires': 'Hi-Res音质', + 'sky': '沉浸环绕声', + 'jyeffect': '高清环绕声', + 'jymaster': '超清母带', + 'dolby': '杜比全景声', +} + + +def quality_display_name(quality: str) -> str: + return _QUALITY_NAMES.get(quality, f"未知音质({quality})") + + +def parse_cookie_str(raw) -> dict[str, str]: + """将 'k=v;k2=v2' 格式的 Cookie 字符串解析为字典""" + cookies: dict[str, str] = {} + if not raw: + return cookies + for part in str(raw).split(';'): + if '=' in part: + k, v = part.strip().split('=', 1) + if k: + cookies[k] = v + return cookies diff --git a/tests/test_middleware.py b/tests/test_middleware.py new file mode 100644 index 0000000..f8dfae1 --- /dev/null +++ b/tests/test_middleware.py @@ -0,0 +1,70 @@ +# -*- coding: utf-8 -*- +"""响应卫生中间件测试：音频直链保护、安全头、Server 指纹。""" +import pytest +from fastapi import FastAPI +from fastapi.responses import JSONResponse +from fastapi.testclient import TestClient + +from core import security +from web.middleware import ResponseHygieneMiddleware + + +@pytest.fixture() +def client(): + security.init_secret_key('test-secret-key') + + app = FastAPI() + app.add_middleware(ResponseHygieneMiddleware) + + @app.get('/audio') + async def audio(): + return JSONResponse({ + 'data': [{'url': 'http://m801.music.126.net/obj/song.mp3', 'id': 1}], + 'nested': {'audioUrl': 'http://m701.music.126.net/x.flac'}, + 'other': 'https://example.com/not-audio.mp3', + }) + + @app.get('/plain') + async def plain(): + return JSONResponse({'url': 'http://m801.music.126.net/obj/song.mp3'}) + + return TestClient(app) + + +def test_audio_url_replaced_with_signed_proxy(client): + r = client.get('/audio') + body = r.json() + assert body['data'][0]['url'].startswith('/audio_stream?token=') + assert body['nested']['audioUrl'].startswith('/audio_stream?token=') + # 非网易音频域名不被替换 + assert body['other'] == 'https://example.com/not-audio.mp3' + + # token 应可被 core.security 校验回原 URL（同一密钥） + token = body['data'][0]['url'].split('token=', 1)[1] + assert security.verify_audio_token(token) == 'http://m801.music.126.net/obj/song.mp3' + + +def test_security_headers_present(client): + r = client.get('/plain') + assert r.headers['x-content-type-options'] == 'nosniff' + assert r.headers['x-frame-options'] == 'DENY' + assert r.headers['referrer-policy'] == 'same-origin' + assert 'content-security-policy' in r.headers + assert r.headers['server'] == 'NeteaseUrl' + + +def test_non_json_body_untouched(client): + """非 JSON 响应不应被中间件改写/缓冲。""" + + security.init_secret_key('k') + app = FastAPI() + app.add_middleware(ResponseHygieneMiddleware) + + @app.get('/text') + async def text(): + from fastapi.responses import PlainTextResponse + return PlainTextResponse('http://m801.music.126.net/x.mp3') + + c = TestClient(app) + r = c.get('/text') + assert r.text == 'http://m801.music.126.net/x.mp3' diff --git a/web/app.py b/web/app.py index 8e45678..c9bd70a 100644 --- a/web/app.py +++ b/web/app.py @@ -18,7 +18,9 @@ from contextlib import asynccontextmanager from fastapi import FastAPI 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 music as music_router from web.routers import site as site_router from web.routers import system as system_router @@ -30,14 +32,27 @@ async def _lifespan(app: FastAPI): yield try: from core import aiohttp_compat, http_client + await http_client.client.close() await aiohttp_compat.close() except Exception: pass +def _init_audio_secret() -> None: + """注入音频签名密钥（与 Flask 同源，令两栈 token 互认）。""" + try: + import auth_system + + init_secret_key(auth_system.get_secret_key()) + except Exception: + pass + + def create_app(serve_flask_fallback: bool = False) -> FastAPI: """创建 FastAPI 应用。""" + _init_audio_secret() + app = FastAPI( title='云村小助手API服务', version=APP_VERSION, @@ -47,6 +62,9 @@ def create_app(serve_flask_fallback: bool = False) -> FastAPI: lifespan=_lifespan, ) + # 响应侧统一处理：安全头 + 音频直链签名保护 + 版本指纹抹除 + app.add_middleware(ResponseHygieneMiddleware) + # 已迁移路由：统一挂 /api/v1（迁移期与 Flask 根路径并存，便于对比） app.include_router(system_router.router, prefix='/api/v1', tags=['system']) app.include_router(site_router.router, prefix='/api/v1', tags=['site']) diff --git a/web/middleware.py b/web/middleware.py new file mode 100644 index 0000000..7c86e48 --- /dev/null +++ b/web/middleware.py @@ -0,0 +1,97 @@ +# -*- coding: utf-8 -*- +"""FastAPI 中间件：镜像 Flask after_request 的响应侧行为。 + +与 main.py `after_request` 对齐的关键点： +1. 音频直链保护——JSON 响应中的网易云音频地址替换为 HMAC 签名代理地址 + （`/audio_stream?token=...`），使直链不可被外站直接盗用；密钥与 Flask 同源， + 故两栈签发的 token 互相可校验。 +2. 零成本安全响应头（nosniff / DENY / Referrer-Policy / CSP 基线）。 +3. 抹去服务器版本指纹（Server 头统一为 NeteaseUrl）。 +4. 429 统一带 Retry-After，供客户端退避。 + +实现注意： +- 用纯 ASGI 中间件逐段转发（不用 BaseHTTPMiddleware，避免它缓冲整段流式响应， + 破坏大 ZIP / 音频流式下载）。 +- 仅对「JSON 且 200」的响应缓冲改写；此时必须延迟发送 response.start，待改写完成 + 后**重算 Content-Length**——否则长度不一致会让客户端 IncompleteRead（实测踩过）。 +""" +import json + +from starlette.types import ASGIApp, Message, Receive, Scope, Send + +from core import security + +_SECURITY_HEADERS = ( + (b'x-content-type-options', b'nosniff'), + (b'x-frame-options', b'DENY'), + (b'referrer-policy', b'same-origin'), + (b'content-security-policy', + b"default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; " + b"img-src 'self' data: http: https:; media-src 'self' blob:; connect-src 'self'; " + b"font-src 'self' data:; object-src 'none'; frame-ancestors 'none'; base-uri 'self'"), +) + + +def _augment_headers(headers: list[tuple[bytes, bytes]], status_code: int) -> list[tuple[bytes, bytes]]: + existing = {k.lower() for k, _ in headers} + out = list(headers) + for hk, hv in _SECURITY_HEADERS: + if hk not in existing: + out.append((hk, hv)) + out.append((b'server', b'NeteaseUrl')) + if status_code == 429 and b'retry-after' not in existing: + out.append((b'retry-after', b'30')) + return out + + +class ResponseHygieneMiddleware: + """响应侧统一处理（安全头 + 音频直链保护 + 版本指纹抹除）。""" + + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope['type'] != 'http': + await self.app(scope, receive, send) + return + + state = {'hold': False, 'start': None, 'chunks': []} + + async def send_wrapper(message: Message) -> None: + mtype = message['type'] + if mtype == 'http.response.start': + status = message['status'] + headers = _augment_headers(list(message.get('headers') or []), status) + ctype = b''.join(v.lower() for k, v in headers if k.lower() == b'content-type') + if status == 200 and b'application/json' in ctype: + # 需改写 body：延迟发送 start，待重算 Content-Length + state['hold'] = True + state['start'] = message + message['headers'] = headers + else: + message['headers'] = headers + await send(message) + elif mtype == 'http.response.body': + if not state['hold']: + await send(message) + return + state['chunks'].append(message.get('body', b'')) + if message.get('more_body'): + return + await self._flush(state, send) + + await self.app(scope, receive, send_wrapper) + + async def _flush(self, state: dict, send: Send) -> None: + raw = b''.join(state['chunks']) + try: + data = json.loads(raw) + if isinstance(data, (dict, list)) and security.protect_audio_urls(data): + raw = json.dumps(data, ensure_ascii=False).encode('utf-8') + except Exception: + pass + headers = [(k, v) for k, v in state['start']['headers'] if k.lower() != b'content-length'] + headers.append((b'content-length', str(len(raw)).encode('ascii'))) + state['start']['headers'] = headers + await send(state['start']) + await send({'type': 'http.response.body', 'body': raw, 'more_body': False}) diff --git a/web/routers/music.py b/web/routers/music.py index a7994dd..68d3a6a 100644 --- a/web/routers/music.py +++ b/web/routers/music.py @@ -1,24 +1,32 @@ # -*- coding: utf-8 -*- -"""音乐查询端点（FastAPI 栈）——与 Flask 侧同名端点响应契约逐字段对齐。 +"""音乐端点（FastAPI 栈）——与 Flask 侧同名端点响应契约逐字段对齐。 -已迁移（免费解析类，需登录，不扣次数）： -- POST/GET /api/v1/toplist ↔ Flask /toplist -- POST/GET /api/v1/artist_songs ↔ Flask /artist_songs -- POST/GET /api/v1/comments ↔ Flask /comments -- POST/GET /api/v1/search ↔ Flask /search +已迁移： +- 免费解析类（需登录，不扣次数）： + POST/GET /api/v1/toplist ↔ Flask /toplist + POST/GET /api/v1/artist_songs ↔ Flask /artist_songs + POST/GET /api/v1/comments ↔ Flask /comments + POST/GET /api/v1/search ↔ Flask /search + POST/GET /api/v1/album ↔ Flask /album + POST/GET /api/v1/playlist ↔ Flask /playlist + POST/GET /api/v1/song ↔ Flask /song（四种 type：url/name/lyric/json） 参数获取沿用 Flask 口径：GET query / POST json / POST form 三路兼容。 上游调用走 music_api 的**异步**方法（原生并发，无同步桥）。 """ import asyncio +import hmac +import logging import re +import time from typing import Annotated, Any from fastapi import APIRouter, Depends, Request import music_api -from core import repo -from core.state import SERVER_NICK_CACHE +from core import media, repo +from core.config import VALID_QUALITY_LEVELS +from core.state import GLOBAL_MUSIC_U_CACHE, REQ_LOG, SERVER_NICK_CACHE from music_api import APIException from web import ratelimit from web.deps import SiteUser, require_login @@ -28,6 +36,10 @@ router = APIRouter() LoginUser = Annotated[SiteUser, Depends(require_login)] +logger = logging.getLogger('web.music') + +_VALID_SONG_TYPES = ['url', 'name', 'lyric', 'json'] + _SERVER_NICK_LOCK: asyncio.Lock | None = None @@ -214,3 +226,198 @@ async def search(user: LoginUser, request: Request): if song.get('name'): song['name'] = await mask_server_nickname(song['name'], user) return success(result, '搜索完成') + + +# ============ 个人/全局账号 Cookie ============ + +def _global_music_u() -> str: + """全局账号 MUSIC_U（5 分钟缓存，与 Flask main._global_music_u 同源同缓存）。""" + now = time.time() + if GLOBAL_MUSIC_U_CACHE['v'] is None or now - GLOBAL_MUSIC_U_CACHE['ts'] > 300: + try: + ck = _cookies() + GLOBAL_MUSIC_U_CACHE['v'] = (ck.get('MUSIC_U') or '').strip() + except Exception: + GLOBAL_MUSIC_U_CACHE['v'] = '' + GLOBAL_MUSIC_U_CACHE['ts'] = now + return GLOBAL_MUSIC_U_CACHE['v'] + + +async def _resolve_user_cookies(data: dict[str, Any], user: SiteUser, request: Request) -> dict[str, str]: + """解析个人账号 Cookie（浏览器传入的 user_cookie，不落服务器）。 + + 全局账号（cookie.txt）零个人身份权限：浏览器提交的 MUSIC_U 若与全局账号一致 + （全局 Cookie 泄露被当个人 Cookie 冒用），一律拒绝并记审计（与 Flask 同口径）。 + """ + ck = media.parse_cookie_str(data.get('user_cookie')) + mu = (ck.get('MUSIC_U') or '').strip() + gmu = _global_music_u() + 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, + ratelimit.client_ip(request), '全局账号冒用拦截', + '浏览器提交的 MUSIC_U 与全局账号一致，已按未登录拒绝') + except Exception: + pass + logger.warning('[全局账号冒用拦截] %s 提交的 MUSIC_U 与全局账号一致，已按未登录处理', + request.url.path) + return {} + return ck + + +def _record_log(action: str, ip: str, song: str) -> None: + """用户解析/下载操作记录（IP）——写入共享 REQ_LOG（与 Flask _record_log 同口径）。""" + REQ_LOG.append({ + 'action': action, + 'ip': ip, + 'song': str(song or '')[:100], + 'time': time.strftime('%Y-%m-%d %H:%M:%S'), + }) + if len(REQ_LOG) > 2000: + del REQ_LOG[:len(REQ_LOG) - 2000] + + +# ============ 迁移端：专辑 / 歌单 / 歌曲 ============ + +@router.api_route('/album', methods=['GET', 'POST']) +async def album(_user: LoginUser, request: Request): + """专辑详情。""" + await ratelimit.guard(request, bucket='free', limit=30, window=60) + data = await _request_data(request) + album_id = data.get('id') + if not album_id: + return error("参数 'album_id' 不能为空", 400) + album_id = _extract_id(album_id) + if not str(album_id).strip().isdigit(): + return error('无效的专辑ID，请输入数字ID或分享链接') + try: + result = await music_api.NeteaseAPI().get_album_detail(int(album_id), _cookies()) + except APIException: + return error('服务暂时不可用，请稍后再试', 500) + return success({'status': 200, 'album': result}, '获取专辑详情成功') + + +@router.api_route('/playlist', methods=['GET', 'POST']) +async def playlist(user: LoginUser, request: Request): + """歌单详情（非管理员遮蔽服务器账号昵称）。""" + await ratelimit.guard(request, bucket='free', limit=30, window=60) + 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()) + except APIException: + return error('服务暂时不可用，请稍后再试', 500) + if isinstance(result, dict): + if result.get('creator'): + result['creator'] = await mask_server_nickname(result['creator'], user) + if result.get('name'): + result['name'] = await mask_server_nickname(result['name'], user) + return success({'status': 'success', 'playlist': result}, '获取歌单详情成功') + + +@router.api_route('/song', methods=['GET', 'POST']) +async def song(user: LoginUser, request: Request): + """歌曲信息（type: url/name/lyric/json）。""" + await ratelimit.guard(request, bucket='free', limit=30, window=60) + data = await _request_data(request) + song_ids = data.get('ids') or data.get('id') + url = data.get('url') + level = data.get('level', 'lossless') + info_type = data.get('type', 'url') + + if not song_ids and not url: + return error("必须提供 'ids'、'id' 或 'url' 参数") + music_id = _extract_id(song_ids or url) + if not str(music_id).strip().isdigit(): + return error('无效的歌曲ID，请输入数字ID或分享链接') + if level not in VALID_QUALITY_LEVELS: + return error(f"无效的音质参数，支持: {', '.join(VALID_QUALITY_LEVELS)}") + if info_type not in _VALID_SONG_TYPES: + return error(f"无效的类型参数，支持: {', '.join(_VALID_SONG_TYPES)}") + + account = data.get('account', 'admin') + if account == 'user': + cookies = await _resolve_user_cookies(data, user, request) + if not cookies: + return error('请先扫码登录个人账号后再使用个人解析', 401) + else: + cookies = _cookies() + + api = music_api.NeteaseAPI() + try: + if info_type == 'url': + result = await api.get_song_url(int(music_id), level, cookies) + if result and result.get('data') and len(result['data']) > 0: + song_data = result['data'][0] or {} + if not song_data: + return error('获取音乐URL失败，可能是版权限制或音质不支持', 404) + return success({ + 'id': song_data.get('id'), + 'url': song_data.get('url'), + 'level': song_data.get('level'), + 'quality_name': media.quality_display_name(song_data.get('level', level)), + 'size': song_data.get('size'), + 'size_formatted': media.format_file_size(song_data.get('size', 0)), + 'type': song_data.get('type'), + 'bitrate': song_data.get('br'), + }, '获取歌曲URL成功') + return error('获取音乐URL失败，可能是版权限制或音质不支持', 404) + + if info_type == 'name': + return success(await api.get_song_detail(int(music_id)), '获取歌曲信息成功') + + if info_type == 'lyric': + return success(await api.get_lyric(int(music_id), cookies), '获取歌词成功') + + # info_type == 'json'：并行取详情/URL，歌词失败不阻断 + song_info, url_info, lyric_info = await asyncio.gather( + api.get_song_detail(int(music_id)), + api.get_song_url(int(music_id), level, cookies), + api.get_lyric(int(music_id), cookies), + return_exceptions=True, + ) + if isinstance(lyric_info, BaseException): + lyric_info = None + if isinstance(url_info, BaseException): + url_info = None + if isinstance(song_info, BaseException) or not song_info or not song_info.get('songs'): + return error('未找到歌曲信息', 404) + + song_data = song_info['songs'][0] + _al = song_data.get('al') or {} + _ar_names = ', '.join(filter(None, ((a or {}).get('name') for a in (song_data.get('ar') or [])))) + response_data = { + 'id': music_id, + 'name': song_data.get('name') or '', + 'ar_name': _ar_names, + 'al_name': _al.get('name') or '', + 'pic': _al.get('picUrl') or '', + 'level': level, + 'lyric': ((lyric_info or {}).get('lrc') or {}).get('lyric') or '', + 'tlyric': ((lyric_info or {}).get('tlyric') or {}).get('lyric') or '', + 'yrc': ((lyric_info or {}).get('yrc') or {}).get('lyric') or '', + } + if url_info and url_info.get('data') and len(url_info['data']) > 0: + url_data = url_info['data'][0] + response_data.update({ + 'url': url_data.get('url', ''), + 'size': media.format_file_size(url_data.get('size', 0)), + 'level': url_data.get('level', level), + }) + else: + response_data.update({'url': '', 'size': '获取失败'}) + + try: + _record_log('解析', ratelimit.client_ip(request), response_data.get('name') or str(music_id)) + except Exception: + pass + return success(response_data, '获取歌曲信息成功') + + except APIException: + return error('服务暂时不可用，请稍后再试', 500)