fix: 修复安全问题 - 登录失败返回401 + XSS过滤
All checks were successful
continuous-integration/drone/push Build is passing

- 登录失败返回 HTTP 401 而非 200
- 添加 XSS 输入过滤工具函数
- 课程名称和描述字段添加 XSS 过滤验证器
This commit is contained in:
yuliang_guo
2026-01-31 10:39:07 +08:00
parent 0b7c07eb7f
commit d59a4355a5
3 changed files with 179 additions and 10 deletions

View File

@@ -67,7 +67,7 @@ async def login(
SystemLogCreate(
level="WARNING",
type="security",
message=f"用户 {login_data.username} 登录失败:密码错误",
message=f"用户 {login_data.username} 登录失败:用户名或密码错误",
user=login_data.username,
ip=request.client.host if request.client else None,
path="/api/v1/auth/login",
@@ -75,19 +75,27 @@ async def login(
user_agent=request.headers.get("user-agent")
)
)
# 不返回 401统一返回 HTTP 200 + 业务失败码,便于前端友好提示
logger.warning("login_failed_wrong_credentials", username=login_data.username)
return ResponseModel(
code=400,
message=str(e) or "用户名或密码错误",
data=None,
# 返回 HTTP 401 + 统一错误消息(避免用户枚举)
from fastapi.responses import JSONResponse
return JSONResponse(
status_code=401,
content={
"code": 401,
"message": "用户名或密码错误",
"data": None,
}
)
except Exception as e:
logger.error("login_failed_unexpected", error=str(e))
return ResponseModel(
code=500,
message="登录失败,请稍后重试",
data=None,
from fastapi.responses import JSONResponse
return JSONResponse(
status_code=500,
content={
"code": 500,
"message": "登录失败,请稍后重试",
"data": None,
}
)

View File

@@ -0,0 +1,136 @@
"""
输入清理和XSS防护工具
"""
import re
import html
from typing import Optional
# 危险的HTML标签和属性
DANGEROUS_TAGS = [
'script', 'iframe', 'object', 'embed', 'form', 'input',
'textarea', 'button', 'select', 'style', 'link', 'meta',
'base', 'applet', 'frame', 'frameset', 'layer', 'ilayer',
'bgsound', 'xml', 'blink', 'marquee'
]
DANGEROUS_ATTRS = [
'onclick', 'ondblclick', 'onmousedown', 'onmouseup', 'onmouseover',
'onmousemove', 'onmouseout', 'onkeypress', 'onkeydown', 'onkeyup',
'onload', 'onerror', 'onabort', 'onblur', 'onchange', 'onfocus',
'onreset', 'onsubmit', 'onunload', 'onbeforeunload', 'onresize',
'onscroll', 'ondrag', 'ondragend', 'ondragenter', 'ondragleave',
'ondragover', 'ondragstart', 'ondrop', 'onmousewheel', 'onwheel',
'oncopy', 'oncut', 'onpaste', 'oncontextmenu', 'oninput', 'oninvalid',
'onsearch', 'onselect', 'ontoggle', 'formaction', 'xlink:href'
]
def sanitize_html(text: Optional[str]) -> Optional[str]:
"""
清理HTML内容移除危险标签和属性
Args:
text: 输入文本
Returns:
清理后的文本
"""
if text is None:
return None
if not isinstance(text, str):
return text
result = text
# 移除危险标签
for tag in DANGEROUS_TAGS:
# 移除开标签
pattern = re.compile(rf'<{tag}[^>]*>', re.IGNORECASE)
result = pattern.sub('', result)
# 移除闭标签
pattern = re.compile(rf'</{tag}>', re.IGNORECASE)
result = pattern.sub('', result)
# 移除危险属性
for attr in DANGEROUS_ATTRS:
pattern = re.compile(rf'\s*{attr}\s*=\s*["\'][^"\']*["\']', re.IGNORECASE)
result = pattern.sub('', result)
# 也处理没有引号的情况
pattern = re.compile(rf'\s*{attr}\s*=\s*\S+', re.IGNORECASE)
result = pattern.sub('', result)
# 移除 javascript: 协议
pattern = re.compile(r'javascript\s*:', re.IGNORECASE)
result = pattern.sub('', result)
# 移除 data: 协议(可能包含恶意代码)
pattern = re.compile(r'data\s*:\s*text/html', re.IGNORECASE)
result = pattern.sub('', result)
# 移除 vbscript: 协议
pattern = re.compile(r'vbscript\s*:', re.IGNORECASE)
result = pattern.sub('', result)
return result
def escape_html(text: Optional[str]) -> Optional[str]:
"""
转义HTML特殊字符
Args:
text: 输入文本
Returns:
转义后的文本
"""
if text is None:
return None
if not isinstance(text, str):
return text
return html.escape(text, quote=True)
def strip_tags(text: Optional[str]) -> Optional[str]:
"""
完全移除所有HTML标签
Args:
text: 输入文本
Returns:
移除标签后的纯文本
"""
if text is None:
return None
if not isinstance(text, str):
return text
# 移除所有HTML标签
clean = re.compile('<[^>]*>')
return clean.sub('', text)
def sanitize_input(text: Optional[str], strict: bool = False) -> Optional[str]:
"""
清理用户输入
Args:
text: 输入文本
strict: 是否使用严格模式完全移除所有HTML标签
Returns:
清理后的文本
"""
if text is None:
return None
if strict:
return strip_tags(text)
else:
return sanitize_html(text)

View File

@@ -8,6 +8,7 @@ from enum import Enum
from pydantic import BaseModel, Field, ConfigDict, field_validator
from app.models.course import CourseStatus, CourseCategory
from app.core.sanitize import sanitize_input
class CourseBase(BaseModel):
@@ -26,6 +27,18 @@ class CourseBase(BaseModel):
is_featured: bool = Field(default=False, description="是否推荐")
allow_download: bool = Field(default=False, description="是否允许下载资料")
@field_validator("name", mode="before")
@classmethod
def sanitize_name(cls, v):
"""清理课程名称中的XSS内容"""
return sanitize_input(v, strict=True) if v else v
@field_validator("description", mode="before")
@classmethod
def sanitize_description(cls, v):
"""清理课程描述中的XSS内容"""
return sanitize_input(v, strict=False) if v else v
@field_validator("category", mode="before")
@classmethod
def normalize_category(cls, v):
@@ -75,6 +88,18 @@ class CourseUpdate(BaseModel):
is_featured: Optional[bool] = Field(None, description="是否推荐")
allow_download: Optional[bool] = Field(None, description="是否允许下载资料")
@field_validator("name", mode="before")
@classmethod
def sanitize_name_update(cls, v):
"""清理课程名称中的XSS内容"""
return sanitize_input(v, strict=True) if v else v
@field_validator("description", mode="before")
@classmethod
def sanitize_description_update(cls, v):
"""清理课程描述中的XSS内容"""
return sanitize_input(v, strict=False) if v else v
@field_validator("category", mode="before")
@classmethod
def normalize_category_update(cls, v):