feat(routes): 迁移解析端点 /song /playlist /album + 响应卫生中间件

迁移路由(/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 头正确。
This commit is contained in:
snowy-yang committed 2026-09-26 01:12:54 +08:00
1 parent 2c55b55e03
commit a2758aa0da
5 files changed
+452 -8

No files matched your search

+70
View File
@@ -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'