fix(security): 核心读接口补齐鉴权(P0 — 无 token 可直接拉取业务数据)
现象:不带任何 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,
而不是让内部接口裸奔。
This commit is contained in:
@ -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())
|
||||
|
||||
@ -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(
|
||||
|
||||
@ -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))
|
||||
|
||||
Reference in New Issue
Block a user