88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
"""上传安全：SVG/WebP 整体禁止上传（存储型 XSS 面 / 无可靠魔数）。"""
|
|
import asyncio
|
|
import json
|
|
|
|
from modules.upload import service as upload_service
|
|
from modules.upload.service import _save_upload, upload_file
|
|
|
|
|
|
class _FakeUpload:
|
|
"""模拟异步分块读取的 UploadFile。"""
|
|
|
|
def __init__(self, filename, data=b''):
|
|
self.filename = filename
|
|
self._data = data
|
|
|
|
async def read(self, size=65536):
|
|
data = self._data
|
|
self._data = b''
|
|
return data
|
|
|
|
|
|
def test_upload_file_rejects_svg(monkeypatch):
|
|
# SVG 被 BANNED_UPLOAD_EXTS 硬拒绝（即使配置放行）
|
|
monkeypatch.setattr(upload_service, 'UPLOAD_ALLOWED_EXTS', '.svg')
|
|
result = asyncio.run(upload_file(_FakeUpload('logo.svg', b'<svg/>')))
|
|
body = json.loads(result.body)
|
|
assert body['ok'] is False
|
|
|
|
|
|
def test_save_upload_rejects_svg_even_if_config_allows():
|
|
# 兜底硬拒绝：即使配置或其他入口放行 .svg，落盘前一律拒绝
|
|
content = b'<svg xmlns="http://www.w3.org/2000/svg"><script>alert(1)</script></svg>'
|
|
result = _save_upload(content, 'a.svg', '.svg')
|
|
body = json.loads(result.body)
|
|
assert body['ok'] is False
|
|
assert 'SVG' in body['msg']
|
|
|
|
|
|
def test_upload_file_rejects_webp(monkeypatch):
|
|
# WebP 被 BANNED_UPLOAD_EXTS 硬拒绝（即使配置放行）
|
|
monkeypatch.setattr(upload_service, 'UPLOAD_ALLOWED_EXTS', '.webp')
|
|
result = asyncio.run(upload_file(_FakeUpload('a.webp', b'RIFF' + b'\x00' * 32)))
|
|
body = json.loads(result.body)
|
|
assert body['ok'] is False
|
|
|
|
|
|
def test_save_upload_rejects_webp_even_if_config_allows():
|
|
# 兜底硬拒绝：webp 魔数仅 RIFF 前缀（WAV/AVI 可冒充），落盘前一律拒绝
|
|
result = _save_upload(b'RIFF' + b'\x00' * 32, 'a.webp', '.webp')
|
|
body = json.loads(result.body)
|
|
assert body['ok'] is False
|
|
assert 'WebP' in body['msg']
|
|
|
|
|
|
def test_upload_file_uses_builtin_set_when_unconfigured(monkeypatch):
|
|
# 方案A：未配置时回落内置智能集合，白名单校验放行内置格式
|
|
from core import R
|
|
called = {}
|
|
|
|
def fake_save(content, original_name, ext):
|
|
called['ext'] = ext
|
|
return R.ok(msg="ok", data={})
|
|
|
|
monkeypatch.setattr(upload_service, 'UPLOAD_ALLOWED_EXTS', '')
|
|
monkeypatch.setattr(upload_service, '_save_upload', fake_save)
|
|
result = asyncio.run(upload_file(_FakeUpload('a.jpg', b'\xff\xd8\xff')))
|
|
body = json.loads(result.body)
|
|
assert body['ok'] is True
|
|
assert called['ext'] == '.jpg'
|
|
|
|
|
|
def test_upload_file_fail_closed_when_config_outside_builtin(monkeypatch):
|
|
# 配置的扩展名全不在内置集合内 → 有效白名单为空 → 拒绝一切上传
|
|
monkeypatch.setattr(upload_service, 'UPLOAD_ALLOWED_EXTS', '.svg,.webp,.exe')
|
|
result = asyncio.run(upload_file(_FakeUpload('a.jpg', b'\xff\xd8\xff')))
|
|
body = json.loads(result.body)
|
|
assert body['ok'] is False
|
|
assert '允许上传范围' in body['msg']
|
|
|
|
|
|
def test_upload_file_rejects_ext_outside_effective_set(monkeypatch):
|
|
# 配置收缩白名单后，内置集合中但不在配置内的扩展名被拒绝
|
|
monkeypatch.setattr(upload_service, 'UPLOAD_ALLOWED_EXTS', '.jpg,.png')
|
|
result = asyncio.run(upload_file(_FakeUpload('a.txt', b'hello')))
|
|
body = json.loads(result.body)
|
|
assert body['ok'] is False
|
|
assert '文件格式' in body['msg']
|