Archived
snowy-yang
a2758aa0da
迁移路由（/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 头正确。
71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
# -*- 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'
|