security: API鉴权补全 + SQL拼接隐患消除

1. materials.py
   - get_material_groups和get_material_items补全Depends(get_current_user)
   - 移除TYPE_FILTER="1=1"死代码及4处f-string SQL拼接
   - 全部SQL改为纯参数化text()查询

2. notifications.py
   - list_notifications废弃user_id查询参数(越权漏洞)
   - user_id强制从JWT Token解析,防止篡改参数偷看他人通知
   - mark_notification_read补全鉴权
This commit is contained in:
2026-08-12 12:03:12 +08:00
parent 69f3e35d14
commit 8c54a38f55
2 changed files with 23 additions and 26 deletions

View File

@ -11,18 +11,26 @@ from app.models.notification import Notification
from app.models.task import Task
from app.models.product import Product
from app.schemas.notification import NotificationResponse, NotificationListResponse
from app.services.auth_service import get_current_user
router = APIRouter(prefix="/notifications", tags=["消息通知"])
@router.get("/", response_model=NotificationListResponse)
async def list_notifications(
user_id: str = Query(..., description="当前用户ID"),
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""获取当前用户的通知列表(按时间倒序)"""
"""
获取当前用户的通知列表(按时间倒序)。
安全user_id 强制从 JWT Token 解析,不接受查询参数,
杜绝通过篡改 user_id 参数越权查看他人通知。
"""
user_id: str = current_user.get("username", "") or current_user.get("sub", "")
# 总数
count_stmt = select(func.count()).select_from(Notification).where(
Notification.user_id == user_id
@ -80,6 +88,7 @@ async def list_notifications(
async def mark_notification_read(
notification_id: str,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""标记单条通知为已读"""
nid = uuid.UUID(notification_id)