From 3f34652b071eb72baeb8fc51788b80eccdf8b7b9 Mon Sep 17 00:00:00 2001 From: duxingchen Date: Mon, 21 Sep 2026 13:47:37 +0800 Subject: [PATCH] =?UTF-8?q?fix(security):=20=E6=A0=B8=E5=BF=83=E8=AF=BB?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3=E8=A1=A5=E9=BD=90=E9=89=B4=E6=9D=83=EF=BC=88?= =?UTF-8?q?P0=20=E2=80=94=20=E6=97=A0=20token=20=E5=8F=AF=E7=9B=B4?= =?UTF-8?q?=E6=8E=A5=E6=8B=89=E5=8F=96=E4=B8=9A=E5=8A=A1=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 现象：不带任何 Token 请求 /api/v1/products/scan/{sn} 等接口直接 200， 业务数据（产品、任务、订单）可被匿名读取。 同时这也是审计「操作人恒为未认证」的根因： 这些路由只挂了 Depends(get_db)，JWT 依赖不执行 → request.state.audit_user 从未写入 → 中间件只能记成「未认证」。 修复：为 3 个文件共 9 条 GET 路由统一补上 Depends(get_current_user) products.py GET /qrcode/{serial_number} GET /scan/{serial_number} ← 匿名可读产品数据 GET / ← 列表 GET /{product_id} GET /{product_id}/messages tasks.py GET / GET /{task_id} GET /by-product/{product_id} orders.py GET / 未改动：notifications.py 本就有鉴权；records.py 无 GET 路由。 三个文件原本已 import get_current_user，未新增导入。 落地前已排查「是否有意免登」： - /products/qrcode/{sn} 是 PC 打印标签用，非免登场景 - 外部系统调用走独立通道 /external/products/lookup（X-API-Key 鉴权）， 与内部 /products/* 完全分离 故内部读接口本就应要求登录。 ⚠️ 连带影响：二维码标签编码的是前端页面地址（/sn/{序列号}）， 此前任何人用手机相机扫码即可查看产品状态，现在会要求登录。 若业务需要免登查询，应走带 API-Key 的 /external/products/lookup， 而不是让内部接口裸奔。 --- backend/app/api/v1/endpoints/orders.py | 1 + backend/app/api/v1/endpoints/products.py | 9 ++++++++- backend/app/api/v1/endpoints/tasks.py | 3 +++ 3 files changed, 12 insertions(+), 1 deletion(-) diff --git a/backend/app/api/v1/endpoints/orders.py b/backend/app/api/v1/endpoints/orders.py index 57d1410..22268df 100644 --- a/backend/app/api/v1/endpoints/orders.py +++ b/backend/app/api/v1/endpoints/orders.py @@ -19,6 +19,7 @@ async def list_orders( skip: int = Query(0, ge=0), limit: int = Query(50, ge=1, le=200), db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), ): result = await db.execute( select(ProductionOrder).offset(skip).limit(limit).order_by(ProductionOrder.created_at.desc()) diff --git a/backend/app/api/v1/endpoints/products.py b/backend/app/api/v1/endpoints/products.py index b75e2e4..878faa8 100644 --- a/backend/app/api/v1/endpoints/products.py +++ b/backend/app/api/v1/endpoints/products.py @@ -26,7 +26,10 @@ router = APIRouter(prefix="/products", tags=["产品管理"]) # ============================================================ @router.get("/qrcode/{serial_number}") -async def get_product_qrcode(serial_number: str): +async def get_product_qrcode( + serial_number: str, + current_user: dict = Depends(get_current_user), +): """ 生成产品二维码（PNG 图片）。 内容为 16 位序列号，扫描后可调用 /scan/{serial_number} 查询产品。 @@ -50,6 +53,7 @@ async def get_product_qrcode(serial_number: str): async def scan_product( serial_number: str, db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), ): """ 扫码接口：根据 16 位序列号查询产品及其当前进度。 @@ -69,6 +73,7 @@ async def list_products( keyword: str | None = Query(None, description="多维搜索: 产品身份证/订单号/规格型号"), status: str | None = Query(None, description="产品状态筛选: PENDING/WIP/COMPLETED/ARCHIVED"), db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), ): """获取产品列表 — 支持 keyword 搜索 + 状态筛选""" return await product_service.get_all_products( @@ -80,6 +85,7 @@ async def list_products( async def get_product( product_id: str, db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), ): """获取单个产品详情""" import uuid @@ -196,6 +202,7 @@ class MessageCreate(BaseModel): async def get_product_messages( product_id: str, db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), ): """获取某产品的所有留言（按时间正序）""" result = await db.execute( diff --git a/backend/app/api/v1/endpoints/tasks.py b/backend/app/api/v1/endpoints/tasks.py index 8a7f901..c161e7e 100644 --- a/backend/app/api/v1/endpoints/tasks.py +++ b/backend/app/api/v1/endpoints/tasks.py @@ -38,6 +38,7 @@ async def list_tasks( skip: int = Query(0, ge=0), limit: int = Query(50, ge=1, le=200), db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), ): """获取任务列表，可按产品/负责人筛选（只返回顶层任务）""" pid = uuid.UUID(product_id) if product_id else None @@ -48,6 +49,7 @@ async def list_tasks( async def get_task( task_id: str, db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), ): """ 获取任务详情 — 递归包含所有层级的子任务。 @@ -299,6 +301,7 @@ async def create_subtask_endpoint( async def get_tasks_by_product( product_id: str, db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), ): """获取指定产品的顶层任务列表（不含子任务嵌套）""" return await task_service.get_top_level_tasks(db, uuid.UUID(product_id))