小鱼开发
95e55293c6
后端安全: - DEBUG 默认 True → False - 彻底移除 AUTH_BYPASS 认证绕过 - 验证码不再明文打印到日志 - 上传接口增加大小限制(500MB/20MB/100MB)与魔数校验 - python-jose → PyJWT, 更新 requirements.lock/uv.lock - Bandit 恢复关键规则(B104/B301/B305/B314/B324/B603/B607) - 修复 5 处 try_except_pass, 15 处加 nosec 注释 - 启用 Bandit pre-commit 钩子 前端安全: - 配置完整 CSP 策略 - 收紧 Capabilities(fs:allow-read-file → $RESOURCE/**) - 移除硬编码 devToken - 清理前端 TODO(美家卡智影命名统一) 部署修复: - docker-compose.prod 增加 alembic 迁移步骤 - api + scheduler 增加 Redis 心跳健康检查 - Nginx 添加安全响应头 - Nginx client_max_body_size 100M → 500M - .env.example 补充 UPLOAD_MAX_* 配置与安全注释 其他: - /voice/upload 合并到 /upload/audio - Rust 上传增加文件大小检查 - 清理 Rust 19 处 println! + 前端 21 处 console.info - 修复 VideoCompose.tsx toast 未导入(已有bug)
395 lines
12 KiB
Python
395 lines
12 KiB
Python
"""
|
|
文件上传 API
|
|
============
|
|
|
|
提供通用文件上传功能，直接上传到七牛云对象存储。
|
|
"""
|
|
|
|
import io
|
|
import logging
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, File, HTTPException, UploadFile
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.config import get_settings
|
|
from app.schemas.common import ApiResponse, success_response
|
|
from app.services.qiniu_service import get_qiniu_service
|
|
|
|
router = APIRouter(prefix="/upload", tags=["Upload"])
|
|
|
|
logger = logging.getLogger(__name__)
|
|
settings = get_settings()
|
|
|
|
|
|
def _validate_file_magic(content: bytes, expected_content_type: str) -> bool:
|
|
"""通过文件头魔数校验文件真实类型，防止 MIME 伪造攻击。"""
|
|
if len(content) < 12:
|
|
return False
|
|
|
|
# 拒绝常见危险文件头
|
|
dangerous_signatures = [
|
|
(b"MZ", "Windows 可执行文件"), # .exe, .dll
|
|
(b"#!", "Shell 脚本"), # bash, python, etc
|
|
(b"PK\x03\x04", "ZIP 压缩包"), # .zip, .jar, .docx
|
|
(b"<?xml", "XML 文件"),
|
|
(b"<html", "HTML 文件"),
|
|
(b"<!DO", "HTML 文档"),
|
|
(b"%PDF", "PDF 文件"),
|
|
]
|
|
for sig, _ in dangerous_signatures:
|
|
if content.startswith(sig):
|
|
return False
|
|
if b"<script" in content[:512].lower():
|
|
return False
|
|
|
|
main_type = expected_content_type.split("/")[0]
|
|
|
|
# 图片校验
|
|
if main_type == "image":
|
|
if content.startswith(b"\xff\xd8\xff"):
|
|
return expected_content_type in ("image/jpeg", "image/jpg")
|
|
if content.startswith(b"\x89PNG\r\n\x1a\n"):
|
|
return expected_content_type == "image/png"
|
|
if content.startswith(b"GIF89a") or content.startswith(b"GIF87a"):
|
|
return expected_content_type == "image/gif"
|
|
if content.startswith(b"RIFF") and content[8:12] == b"WEBP":
|
|
return expected_content_type == "image/webp"
|
|
return False
|
|
|
|
# 视频校验
|
|
if main_type == "video":
|
|
# MP4 / MOV / M4V 等 ISO Base Media File Format
|
|
if content[4:8] == b"ftyp":
|
|
brand = content[8:12]
|
|
if brand in (b"qt ", b"qtw "):
|
|
return expected_content_type in ("video/quicktime",)
|
|
# mp4, isom, avc1, mp41, mp42 等
|
|
return expected_content_type in (
|
|
"video/mp4",
|
|
"video/quicktime",
|
|
)
|
|
if content.startswith(b"RIFF") and content[8:12] == b"AVI ":
|
|
return expected_content_type == "video/x-msvideo"
|
|
if content.startswith(b"\x1aE\xdf\xa3"):
|
|
return expected_content_type == "video/webm"
|
|
return False
|
|
|
|
# 音频校验
|
|
if main_type == "audio":
|
|
if content[:3] == b"ID3" or content[:2] in (
|
|
b"\xff\xfb",
|
|
b"\xff\xf3",
|
|
b"\xff\xf2",
|
|
):
|
|
return expected_content_type in ("audio/mpeg", "audio/mp3")
|
|
if content.startswith(b"RIFF") and content[8:12] == b"WAVE":
|
|
return expected_content_type in ("audio/wav", "audio/x-wav")
|
|
if content.startswith(b"fLaC"):
|
|
return expected_content_type == "audio/flac"
|
|
if content.startswith(b"OggS"):
|
|
return expected_content_type == "audio/ogg"
|
|
# AAC / M4A（也是 ftyp 格式）
|
|
if content[4:8] == b"ftyp":
|
|
brand = content[8:12]
|
|
if brand in (b"M4A ", b"m4a ", b"mp42", b"isom", b"M4P "):
|
|
return expected_content_type in (
|
|
"audio/mp4",
|
|
"audio/aac",
|
|
"audio/m4a",
|
|
)
|
|
return False
|
|
|
|
return False
|
|
|
|
|
|
def _check_upload_file(content: bytes, max_size: int, content_type: str, type_label: str) -> None:
|
|
"""统一校验文件大小和魔数，失败时直接抛 HTTPException。"""
|
|
if len(content) > max_size:
|
|
max_mb = max_size // 1024 // 1024
|
|
raise HTTPException(
|
|
status_code=413,
|
|
detail=f"{type_label}文件大小不能超过 {max_mb}MB",
|
|
)
|
|
if not _validate_file_magic(content, content_type):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"{type_label}文件内容与实际格式不符，可能存在安全风险",
|
|
)
|
|
|
|
|
|
class UploadResponse(BaseModel):
|
|
"""上传响应"""
|
|
|
|
url: str = Field(..., description="七牛云文件 URL")
|
|
key: str = Field(..., description="七牛云文件 key")
|
|
size: int = Field(..., description="文件大小（字节）")
|
|
|
|
|
|
@router.post("/video", response_model=ApiResponse[UploadResponse])
|
|
async def upload_video(
|
|
file: UploadFile = File(..., description="视频文件"),
|
|
):
|
|
"""
|
|
上传视频到七牛云
|
|
|
|
支持格式：mp4, mov, avi, webm
|
|
返回七牛云永久访问 URL。
|
|
"""
|
|
try:
|
|
# 验证文件格式
|
|
allowed_types = {
|
|
"video/mp4",
|
|
"video/quicktime",
|
|
"video/x-msvideo",
|
|
"video/webm",
|
|
}
|
|
content_type = file.content_type or ""
|
|
|
|
# 如果 content_type 为空，尝试从文件名推断
|
|
if not content_type:
|
|
ext = Path(file.filename or "").suffix.lower()
|
|
ext_to_mime = {
|
|
".mp4": "video/mp4",
|
|
".mov": "video/quicktime",
|
|
".avi": "video/x-msvideo",
|
|
".webm": "video/webm",
|
|
}
|
|
content_type = ext_to_mime.get(ext, "")
|
|
|
|
if content_type not in allowed_types:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"不支持的文件格式: {content_type}，请上传 mp4/mov/avi/webm 视频",
|
|
)
|
|
|
|
# 读取文件内容
|
|
content = await file.read()
|
|
if not content:
|
|
raise HTTPException(status_code=400, detail="文件内容为空")
|
|
|
|
# 校验大小和魔数
|
|
_check_upload_file(
|
|
content,
|
|
settings.UPLOAD_MAX_VIDEO_SIZE,
|
|
content_type,
|
|
"视频",
|
|
)
|
|
|
|
# 生成唯一文件名
|
|
ext = Path(file.filename or "video.mp4").suffix or ".mp4"
|
|
unique_name = f"{uuid.uuid4().hex[:16]}{ext}"
|
|
|
|
# 上传到七牛云
|
|
qiniu = get_qiniu_service()
|
|
bucket, domain = qiniu._get_bucket_and_domain("video")
|
|
key = qiniu.generate_key("video", unique_name)
|
|
stream = io.BytesIO(content)
|
|
result = await qiniu.upload_stream_async(
|
|
stream=stream,
|
|
key=key,
|
|
mime_type=content_type or "video/mp4",
|
|
bucket=bucket,
|
|
domain=domain,
|
|
)
|
|
|
|
url = result.get("url")
|
|
key = result.get("key")
|
|
|
|
if not url:
|
|
raise HTTPException(status_code=500, detail="上传到七牛云失败：未返回 URL")
|
|
|
|
logger.info(f"[Upload] 视频上传成功: {url[:80]}..., size={len(content)}")
|
|
|
|
return success_response(
|
|
data=UploadResponse(
|
|
url=url,
|
|
key=key or unique_name,
|
|
size=len(content),
|
|
)
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"[Upload] 视频上传失败: {e}")
|
|
raise HTTPException(status_code=500, detail=f"上传失败: {e}")
|
|
|
|
|
|
@router.post("/image", response_model=ApiResponse[UploadResponse])
|
|
async def upload_image(
|
|
file: UploadFile = File(..., description="图片文件"),
|
|
):
|
|
"""
|
|
上传图片到七牛云
|
|
|
|
支持格式：jpg, png, gif, webp
|
|
"""
|
|
try:
|
|
allowed_types = {
|
|
"image/jpeg",
|
|
"image/png",
|
|
"image/gif",
|
|
"image/webp",
|
|
}
|
|
content_type = file.content_type or ""
|
|
|
|
if not content_type:
|
|
ext = Path(file.filename or "").suffix.lower()
|
|
ext_to_mime = {
|
|
".jpg": "image/jpeg",
|
|
".jpeg": "image/jpeg",
|
|
".png": "image/png",
|
|
".gif": "image/gif",
|
|
".webp": "image/webp",
|
|
}
|
|
content_type = ext_to_mime.get(ext, "")
|
|
|
|
if content_type not in allowed_types:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"不支持的图片格式: {content_type}",
|
|
)
|
|
|
|
content = await file.read()
|
|
if not content:
|
|
raise HTTPException(status_code=400, detail="文件内容为空")
|
|
|
|
# 校验大小和魔数
|
|
_check_upload_file(
|
|
content,
|
|
settings.UPLOAD_MAX_IMAGE_SIZE,
|
|
content_type,
|
|
"图片",
|
|
)
|
|
|
|
ext = Path(file.filename or "image.jpg").suffix or ".jpg"
|
|
unique_name = f"{uuid.uuid4().hex[:16]}{ext}"
|
|
|
|
qiniu = get_qiniu_service()
|
|
bucket, domain = qiniu._get_bucket_and_domain("image")
|
|
key = qiniu.generate_key("image", unique_name)
|
|
stream = io.BytesIO(content)
|
|
result = await qiniu.upload_stream_async(
|
|
stream=stream,
|
|
key=key,
|
|
mime_type=content_type or "image/jpeg",
|
|
bucket=bucket,
|
|
domain=domain,
|
|
)
|
|
|
|
url = result.get("url")
|
|
key = result.get("key")
|
|
|
|
if not url:
|
|
raise HTTPException(status_code=500, detail="上传到七牛云失败：未返回 URL")
|
|
|
|
logger.info(f"[Upload] 图片上传成功: {url[:80]}..., size={len(content)}")
|
|
|
|
return success_response(
|
|
data=UploadResponse(
|
|
url=url,
|
|
key=key or unique_name,
|
|
size=len(content),
|
|
)
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"[Upload] 图片上传失败: {e}")
|
|
raise HTTPException(status_code=500, detail=f"上传失败: {e}")
|
|
|
|
|
|
@router.post("/audio", response_model=ApiResponse[UploadResponse])
|
|
async def upload_audio(
|
|
file: UploadFile = File(..., description="音频文件"),
|
|
):
|
|
"""
|
|
上传音频到七牛云
|
|
|
|
支持格式：mp3, wav, aac, m4a, ogg, flac
|
|
"""
|
|
try:
|
|
allowed_types = {
|
|
"audio/mpeg",
|
|
"audio/mp3",
|
|
"audio/wav",
|
|
"audio/x-wav",
|
|
"audio/aac",
|
|
"audio/mp4",
|
|
"audio/ogg",
|
|
"audio/flac",
|
|
"audio/x-flac",
|
|
}
|
|
content_type = file.content_type or ""
|
|
|
|
if not content_type:
|
|
ext = Path(file.filename or "").suffix.lower()
|
|
ext_to_mime = {
|
|
".mp3": "audio/mpeg",
|
|
".wav": "audio/wav",
|
|
".aac": "audio/aac",
|
|
".m4a": "audio/mp4",
|
|
".ogg": "audio/ogg",
|
|
".flac": "audio/flac",
|
|
}
|
|
content_type = ext_to_mime.get(ext, "")
|
|
|
|
if content_type not in allowed_types:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"不支持的音频格式: {content_type}，请上传 mp3/wav/aac/m4a/ogg/flac",
|
|
)
|
|
|
|
content = await file.read()
|
|
if not content:
|
|
raise HTTPException(status_code=400, detail="文件内容为空")
|
|
|
|
# 校验大小和魔数
|
|
_check_upload_file(
|
|
content,
|
|
settings.UPLOAD_MAX_AUDIO_SIZE,
|
|
content_type,
|
|
"音频",
|
|
)
|
|
|
|
ext = Path(file.filename or "audio.mp3").suffix or ".mp3"
|
|
unique_name = f"{uuid.uuid4().hex[:16]}{ext}"
|
|
|
|
qiniu = get_qiniu_service()
|
|
# 复用视频 bucket（或根据配置使用音频 bucket）
|
|
bucket, domain = qiniu._get_bucket_and_domain("video")
|
|
key = qiniu.generate_key("audio", unique_name)
|
|
stream = io.BytesIO(content)
|
|
result = await qiniu.upload_stream_async(
|
|
stream=stream,
|
|
key=key,
|
|
mime_type=content_type or "audio/mpeg",
|
|
bucket=bucket,
|
|
domain=domain,
|
|
)
|
|
|
|
url = result.get("url")
|
|
key = result.get("key")
|
|
|
|
if not url:
|
|
raise HTTPException(status_code=500, detail="上传到七牛云失败：未返回 URL")
|
|
|
|
logger.info(f"[Upload] 音频上传成功: {url[:80]}..., size={len(content)}")
|
|
|
|
return success_response(
|
|
data=UploadResponse(
|
|
url=url,
|
|
key=key or unique_name,
|
|
size=len(content),
|
|
)
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"[Upload] 音频上传失败: {e}")
|
|
raise HTTPException(status_code=500, detail=f"上传失败: {e}")
|