wsb1224
e3479f0546
# 问题 修复 文件 1 前端构建失败（引号错误） size="small type=" → size="small" type=" PosterHistoryPage.vue 2 migrate_014 ORM vs 缺失列 全部改为原始 SQL，不再引用 ORM 模型 migrate_014.py 3 cleanup 字段名错误 output_path → ppt_path cleanup.py 4 文案生成 case 越权 添加 case.user_id != user_id 校验 poster/service.py 5 存储路径未接通持久化卷 全部改用 get_storage_root()（默认 /app/api/storage/insurance） config.py, ppt/routes.py, poster/service.py, poster/tasks.py 高风险问题修复 # 问题 修复 文件 6 migrate_019 rollback 撤销成功字段 每个 ALTER 后立即 commit，失败只回滚当前语句 migrate_019.py 7 迁移锁 Windows 不兼容 + 句柄未持久化 全局变量保存锁句柄，支持 Windows msvcrt api/insurance/db/__init__.py 8 PDF 校验异常时放行 异常返回 False（文件损坏） security.py 9 健康检查始终返回成功 缺少关键资源时返回 503 + missing 列表 poster/routes.py 10 短密钥掩码泄露原值 ≤4 字符返回 **** ppt_admin_service.py 11 设置无键名白名单 添加 _ALLOWED_SETTING_KEYS 白名单 ppt_admin_service.py 12 容器重启任务永久 stuck 添加 recover_stale_tasks() 启动恢复函数 poster/tasks.py, ppt/parse_worker.py
117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
"""安全工具 — 文件上传校验和 URL 安全校验。"""
|
||
import ipaddress
|
||
import logging
|
||
import os
|
||
import socket
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# PDF 文件头魔数
|
||
PDF_MAGIC = b"%PDF"
|
||
|
||
# 文件大小限制（默认 30MB）
|
||
MAX_FILE_SIZE = int(os.environ.get("MAX_UPLOAD_SIZE_MB", "30")) * 1024 * 1024
|
||
|
||
# PDF 页数限制（默认 200 页）
|
||
MAX_PDF_PAGES = int(os.environ.get("MAX_PDF_PAGES", "200"))
|
||
|
||
# 私网/保留 IP 前缀
|
||
_PRIVATE_NETWORKS = [
|
||
ipaddress.ip_network("127.0.0.0/8"), # loopback
|
||
ipaddress.ip_network("10.0.0.0/8"), # 私网 A
|
||
ipaddress.ip_network("172.16.0.0/12"), # 私网 B
|
||
ipaddress.ip_network("192.168.0.0/16"), # 私网 C
|
||
ipaddress.ip_network("169.254.0.0/16"), # 链路本地
|
||
ipaddress.ip_network("0.0.0.0/8"), # 当前网络
|
||
ipaddress.ip_network("::1/128"), # IPv6 loopback
|
||
ipaddress.ip_network("fc00::/7"), # IPv6 私网
|
||
ipaddress.ip_network("fe80::/10"), # IPv6 链路本地
|
||
]
|
||
|
||
|
||
def validate_pdf_upload(file_storage) -> tuple[bool, str]:
|
||
"""校验上传的 PDF 文件。
|
||
|
||
返回 (is_valid, error_message)。
|
||
"""
|
||
# 1. 检查文件大小
|
||
file_storage.seek(0, os.SEEK_END)
|
||
size = file_storage.tell()
|
||
file_storage.seek(0)
|
||
|
||
if size == 0:
|
||
return False, "文件为空"
|
||
if size > MAX_FILE_SIZE:
|
||
return False, f"文件大小 {size // (1024*1024)}MB 超过限制 {MAX_FILE_SIZE // (1024*1024)}MB"
|
||
|
||
# 2. 检查 PDF 魔数
|
||
header = file_storage.read(8)
|
||
file_storage.seek(0)
|
||
if not header.startswith(PDF_MAGIC):
|
||
return False, "文件不是有效的 PDF 格式"
|
||
|
||
# 3. 检查 PDF 是否加密
|
||
try:
|
||
import pypdf
|
||
reader = pypdf.PdfReader(file_storage)
|
||
if reader.is_encrypted:
|
||
file_storage.seek(0)
|
||
return False, "不支持加密的 PDF 文件，请先解密后重新上传"
|
||
# 4. 检查页数
|
||
if len(reader.pages) > MAX_PDF_PAGES:
|
||
file_storage.seek(0)
|
||
return False, f"PDF 页数 {len(reader.pages)} 超过限制 {MAX_PDF_PAGES} 页"
|
||
file_storage.seek(0)
|
||
except ImportError:
|
||
# pypdf 不可用时跳过加密和页数检查
|
||
file_storage.seek(0)
|
||
except Exception as e:
|
||
logger.warning(f"PDF 解析检查失败: {e}")
|
||
file_storage.seek(0)
|
||
return False, "PDF 文件内容损坏或格式无效"
|
||
|
||
return True, ""
|
||
|
||
|
||
def is_safe_base_url(url: str) -> tuple[bool, str]:
|
||
"""检查 Base URL 是否安全（拒绝私网、环回、元数据地址）。
|
||
|
||
返回 (is_safe, error_message)。
|
||
"""
|
||
if not url:
|
||
return True, ""
|
||
|
||
# 只允许 http/https
|
||
if not url.startswith(("http://", "https://")):
|
||
return False, "仅支持 http/https 协议"
|
||
|
||
# 解析主机名
|
||
from urllib.parse import urlparse
|
||
try:
|
||
parsed = urlparse(url)
|
||
except Exception:
|
||
return False, "URL 格式无效"
|
||
|
||
hostname = parsed.hostname
|
||
if not hostname:
|
||
return False, "URL 缺少主机名"
|
||
|
||
# 拒绝明确的 localhost
|
||
if hostname in ("localhost", "0.0.0.0"):
|
||
return False, f"不允许访问 {hostname}"
|
||
|
||
# DNS 解析后检查 IP
|
||
try:
|
||
for info in socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM):
|
||
ip_str = info[4][0]
|
||
ip = ipaddress.ip_address(ip_str)
|
||
for network in _PRIVATE_NETWORKS:
|
||
if ip in network:
|
||
return False, f"不允许访问私网/保留地址 ({ip_str})"
|
||
except socket.gaierror:
|
||
pass # DNS 解析失败不阻止，让后续调用报错
|
||
except Exception as e:
|
||
logger.warning(f"URL 安全检查异常: {e}")
|
||
|
||
return True, ""
|