fix(security): harden uploads and wecom binding

backend/app/api/upload.py

@@ -1,12 +1,15 @@
 import uuid
+import io
+import warnings
 from datetime import date
 from pathlib import PurePosixPath
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
+from PIL import Image, UnidentifiedImageError
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.database import get_db
 from app.middleware.auth import get_current_user, require_any_role
-from app.services.minio_client import generate_presigned_upload_url, generate_presigned_download_url
+from app.services.minio_client import generate_presigned_download_url, put_validated_image
 from app.models.visit import Visit
 
 router = APIRouter(prefix="/upload", tags=["Upload"])
@@ -16,16 +19,17 @@ _IMAGE_TYPES = {
     "image/png": "png",
     "image/webp": "webp",
 }
+MAX_IMAGE_BYTES = 10 * 1024 * 1024
+MAX_IMAGE_PIXELS = 25_000_000
 
 
 def validate_image_upload(filename: str, content_type: str) -> str:
-    """Validate client-declared image metadata before issuing a short-lived PUT URL."""
+    """Validate image metadata before accepting bytes for server-side verification."""
-    extension = _IMAGE_TYPES.get(content_type.lower())
+    if content_type.lower() not in _IMAGE_TYPES:
-    if not extension:
         raise HTTPException(status_code=400, detail="Only JPEG, PNG, and WebP images are supported")
     if PurePosixPath(filename).suffix.lower() not in {".jpg", ".jpeg", ".png", ".webp"}:
         raise HTTPException(status_code=400, detail="Invalid image filename extension")
-    return extension
+    return _IMAGE_TYPES[content_type.lower()]
 
 
 def build_object_key(user_id: str, extension: str) -> str:
@@ -44,22 +48,39 @@ def is_owned_upload_key(object_key: str, user_id: str) -> bool:
     )
 
 
-@router.post("/presigned-url")
+def normalize_image(image_bytes: bytes) -> bytes:
-async def get_presigned_upload_url(
+    """Verify image bytes, enforce a pixel limit, and strip active/metadata content by re-encoding."""
-    filename: str,
+    if not image_bytes or len(image_bytes) > MAX_IMAGE_BYTES:
-    content_type: str = "image/jpeg",
+        raise HTTPException(status_code=413, detail="Image must be between 1 byte and 10 MB")
+    try:
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", Image.DecompressionBombWarning)
+            with Image.open(io.BytesIO(image_bytes)) as image:
+                image.verify()
+            with Image.open(io.BytesIO(image_bytes)) as image:
+                if image.width * image.height > MAX_IMAGE_PIXELS:
+                    raise HTTPException(status_code=413, detail="Image dimensions are too large")
+                normalized = image.convert("RGB")
+                output = io.BytesIO()
+                normalized.save(output, format="JPEG", quality=85, optimize=True)
+                return output.getvalue()
+    except HTTPException:
+        raise
+    except (Image.DecompressionBombError, Image.DecompressionBombWarning, OSError, UnidentifiedImageError, ValueError) as exc:
+        raise HTTPException(status_code=400, detail="Invalid image content") from exc
+
+
+@router.post("/image")
+async def upload_image(
+    file: UploadFile = File(...),
     current_user: dict = Depends(require_any_role),
 ):
-    """Get a presigned PUT URL for direct MinIO upload."""
+    """Validate and normalize an image before persisting it to private object storage."""
-    extension = validate_image_upload(filename, content_type)
+    validate_image_upload(file.filename or "upload.jpg", file.content_type or "")
-    object_key = build_object_key(current_user["user_id"], extension)
+    normalized = normalize_image(await file.read(MAX_IMAGE_BYTES + 1))
-
+    object_key = build_object_key(current_user["user_id"], "jpg")
-    url = generate_presigned_upload_url(object_key)
+    put_validated_image(object_key, normalized)
-
+    return {"object_key": object_key}
-    return {
-        "upload_url": url,
-        "object_key": object_key,
-    }
 
 
 @router.get("/download-url")

backend/app/api/wecom.py

@@ -9,8 +9,9 @@ from fastapi.responses import PlainTextResponse
 from pydantic import BaseModel
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 from app.config import settings
-from app.database import get_db
+from app.database import async_session, get_db
 from app.middleware.auth import get_current_user, require_director
 from app.models.user import User
 from app.services.wecom import wecom_client, store_bind_token, consume_bind_token
@@ -116,21 +117,20 @@ async def wecom_callback_event(request: Request):
     # Triggers for binding
     if (msg_type == "text" and content == "绑定") or (msg_type == "event" and event == "click" and event_key == "BIND_ACCOUNT"):
         _log.info("wecom bind triggered")
-        _handle_bind_request(from_user)
+        await _handle_bind_request(from_user)
         return PlainTextResponse(content="success")
 
     _log.info("wecom message ignored (no matching action)")
     return PlainTextResponse(content="success")
 
 
-def _handle_bind_request(wecom_userid: str):
+async def _handle_bind_request(wecom_userid: str):
     """Generate bind token, store it, and push a bind link to the user."""
     if not wecom_userid:
         return
 
-    import asyncio
+    async with async_session() as db:
-
+        bind_token = await store_bind_token(db, wecom_userid)
-    bind_token = store_bind_token(wecom_userid)
 
     content = (
         f"【账号绑定】\n\n"
@@ -139,16 +139,7 @@ def _handle_bind_request(wecom_userid: str):
         f"绑定后可使用企微一键登录，并接收填报提醒通知。"
     )
 
-    try:
+    await wecom_client.send_text_message([wecom_userid], content)
-        # Must run async in sync context
-        loop = asyncio.get_event_loop()
-    except RuntimeError:
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-
-    loop.run_until_complete(
-        wecom_client.send_text_message([wecom_userid], content)
-    )
 
 
 # ── Bind confirmation (JWT-protected) ──
@@ -163,7 +154,7 @@ async def bind_confirm(
     if not data.token:
         raise HTTPException(status_code=400, detail="token 不能为空")
 
-    wecom_userid = consume_bind_token(data.token)
+    wecom_userid = await consume_bind_token(db, data.token)
     if not wecom_userid:
         raise HTTPException(status_code=400, detail="绑定链接已过期或无效，请重新在企微发送「绑定」")
 
@@ -181,7 +172,11 @@ async def bind_confirm(
     )
     user = result.scalar_one()
     user.wecom_userid = wecom_userid
+    try:
         await db.commit()
+    except IntegrityError as exc:
+        await db.rollback()
+        raise HTTPException(status_code=409, detail="该企业微信已绑定其他账号") from exc
 
     return {"code": 200, "msg": "绑定成功", "wecom_userid": wecom_userid}
 

backend/app/main.py

@@ -51,6 +51,10 @@ async def lifespan(app: FastAPI):
         await conn.run_sync(lambda c: c.exec_driver_sql(
             "ALTER TABLE users ADD COLUMN IF NOT EXISTS color VARCHAR(7)"
         ))
+        await conn.run_sync(lambda c: c.exec_driver_sql(
+            "CREATE UNIQUE INDEX IF NOT EXISTS uq_users_wecom_userid "
+            "ON users (wecom_userid) WHERE wecom_userid IS NOT NULL"
+        ))
         # system_config table for v0.5
         await conn.run_sync(lambda c: c.exec_driver_sql(
             "CREATE TABLE IF NOT EXISTS system_config (key VARCHAR(64) PRIMARY KEY, value TEXT DEFAULT '')"

backend/app/models/__init__.py

@@ -10,6 +10,7 @@ from app.models.daily_note import DailyNote
 from app.models.ai_summary import AISummary
 from app.models.system_config import SystemConfig
 from app.models.leave import Leave
+from app.models.wecom_bind_token import WecomBindToken
 
 __all__ = [
     "User",
@@ -24,4 +25,5 @@ __all__ = [
     "AISummary",
     "SystemConfig",
     "Leave",
+    "WecomBindToken",
 ]

backend/app/models/wecom_bind_token.py

@@ -0,0 +1,18 @@
+from datetime import datetime
+
+from sqlalchemy import DateTime, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.database import Base
+
+
+class WecomBindToken(Base):
+    """A single-use, expiring binding token issued after a verified WeCom callback."""
+
+    __tablename__ = "wecom_bind_tokens"
+
+    token: Mapped[str] = mapped_column(String(64), primary_key=True)
+    wecom_userid: Mapped[str] = mapped_column(String(100), index=True)
+    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
+    consumed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

backend/app/services/minio_client.py

@@ -1,4 +1,5 @@
 from datetime import timedelta
+from io import BytesIO
 from minio import Minio
 from app.config import settings
 
@@ -20,10 +21,16 @@ def get_minio_client() -> Minio:
     return _client
 
 
-def generate_presigned_upload_url(object_key: str, expires: int = 600) -> str:
+def put_validated_image(object_key: str, image_bytes: bytes) -> None:
-    """Generate a presigned PUT URL for direct upload to MinIO."""
+    """Store a normalized image only after the API has validated its content."""
     client = get_minio_client()
-    return client.presigned_put_object(settings.MINIO_BUCKET, object_key, expires=timedelta(seconds=expires))
+    client.put_object(
+        settings.MINIO_BUCKET,
+        object_key,
+        BytesIO(image_bytes),
+        length=len(image_bytes),
+        content_type="image/jpeg",
+    )
 
 
 def generate_presigned_download_url(object_key: str, expires: int = 3600) -> str:

backend/app/services/wecom.py

@@ -1,30 +1,41 @@
 import time
 import uuid
+from datetime import datetime, timedelta, timezone
 import httpx
+from sqlalchemy import delete, update
+from sqlalchemy.ext.asyncio import AsyncSession
 from app.config import settings
+from app.models.wecom_bind_token import WecomBindToken
 
-# In-memory bind token store (TTL 600s).  Replace with Redis if scaling to multiple workers.
+async def store_bind_token(db: AsyncSession, wecom_userid: str, ttl: int = 600) -> str:
-_bind_tokens: dict[str, tuple[str, float]] = {}  # token → (wecom_userid, expires_at)
+    """Persist a one-time bind token so it works across application workers."""
-
-
-def store_bind_token(wecom_userid: str, ttl: int = 600) -> str:
-    """Store a bind token → wecom_userid mapping. Returns the token."""
     token = uuid.uuid4().hex
-    _bind_tokens[token] = (wecom_userid, time.time() + ttl)
+    now = datetime.now(timezone.utc)
-    # Cleanup expired tokens
+    await db.execute(delete(WecomBindToken).where(WecomBindToken.expires_at < now))
-    now = time.time()
+    db.add(WecomBindToken(
-    for k in list(_bind_tokens):
+        token=token,
-        if _bind_tokens[k][1] < now:
+        wecom_userid=wecom_userid,
-            del _bind_tokens[k]
+        expires_at=now + timedelta(seconds=ttl),
+    ))
+    await db.commit()
     return token
 
 
-def consume_bind_token(token: str) -> str | None:
+async def consume_bind_token(db: AsyncSession, token: str) -> str | None:
-    """Lookup and consume a bind token. Returns wecom_userid or None."""
+    """Atomically consume an unexpired binding token and return its WeCom user ID."""
-    entry = _bind_tokens.pop(token, None)
+    now = datetime.now(timezone.utc)
-    if entry and entry[1] > time.time():
+    result = await db.execute(
-        return entry[0]
+        update(WecomBindToken)
-    return None
+        .where(
+            WecomBindToken.token == token,
+            WecomBindToken.consumed_at.is_(None),
+            WecomBindToken.expires_at > now,
+        )
+        .values(consumed_at=now)
+        .returning(WecomBindToken.wecom_userid)
+    )
+    await db.commit()
+    return result.scalar_one_or_none()
 
 
 class WecomClient:

backend/requirements.txt

@@ -13,3 +13,4 @@ openpyxl==3.1.5
 apscheduler==3.11.0
 python-dotenv==1.0.1
 pycryptodome
+Pillow==11.1.0

backend/tests/test_upload_security.py

@@ -1,7 +1,13 @@
-from fastapi import HTTPException
+import asyncio
+from io import BytesIO
+from unittest.mock import AsyncMock, MagicMock
 
-from app.api.upload import build_object_key, is_owned_upload_key, validate_image_upload
+from fastapi import HTTPException
+from PIL import Image
+
+from app.api.upload import build_object_key, is_owned_upload_key, normalize_image, validate_image_upload
 from app.config import settings, validate_security_settings
+from app.services.wecom import consume_bind_token, store_bind_token
 
 
 USER_ID = "a2d4447f-7b67-4304-ad9a-953007e75ef2"
@@ -32,6 +38,46 @@ def test_upload_rejects_an_invalid_filename_extension():
         raise AssertionError("invalid extensions must be rejected")
 
 
+def test_image_content_is_normalized_to_jpeg():
+    source = BytesIO()
+    Image.new("RGBA", (8, 8), color=(255, 0, 0, 128)).save(source, format="PNG")
+
+    normalized = normalize_image(source.getvalue())
+
+    with Image.open(BytesIO(normalized)) as image:
+        assert image.format == "JPEG"
+        assert image.mode == "RGB"
+
+
+def test_image_content_rejects_non_images():
+    try:
+        normalize_image(b"not an image")
+    except HTTPException as exc:
+        assert exc.status_code == 400
+    else:
+        raise AssertionError("non-image bytes must be rejected")
+
+
+def test_wecom_binding_tokens_are_persisted_and_consumed_atomically():
+    session = MagicMock()
+    session.execute = AsyncMock()
+    session.commit = AsyncMock()
+
+    token = asyncio.run(store_bind_token(session, "wecom-user"))
+
+    assert len(token) == 32
+    assert session.add.call_args.args[0].wecom_userid == "wecom-user"
+    session.commit.assert_awaited_once()
+
+    result = MagicMock()
+    result.scalar_one_or_none.return_value = "wecom-user"
+    session.execute = AsyncMock(return_value=result)
+    session.commit = AsyncMock()
+
+    assert asyncio.run(consume_bind_token(session, token)) == "wecom-user"
+    session.commit.assert_awaited_once()
+
+
 def test_production_rejects_placeholder_database_configuration(monkeypatch):
     monkeypatch.setattr(settings, "ENVIRONMENT", "production")
     monkeypatch.setattr(settings, "SECRET_KEY", "a" * 32)

frontend/src/api/upload.ts

@@ -1,20 +1,15 @@
 import api from './index'
-import axios from 'axios'
 
 export const uploadApi = {
-  async getPresignedUrl(filename: string, contentType: string = 'image/jpeg') {
+  async uploadImage(file: File) {
-    return api.post('/upload/presigned-url', null, {
+    const data = new FormData()
-      params: { filename, content_type: contentType },
+    data.append('file', file)
+    return api.post('/upload/image', data, {
+      headers: { 'Content-Type': 'multipart/form-data' },
+      timeout: 60000,
     })
   },
   getDownloadUrl(objectKey: string) {
     return api.get('/upload/download-url', { params: { object_key: objectKey } })
   },
-  // Direct upload to MinIO
-  async uploadFile(uploadUrl: string, file: File) {
-    return axios.put(uploadUrl, file, {
-      headers: { 'Content-Type': file.type },
-      timeout: 60000,
-    })
-  },
 }

frontend/src/views/desktop/ManagerWorkspace.vue

@@ -140,12 +140,13 @@ async function handleDialogPhotoUpload(event: Event) {
     try {
       // Compress before upload to reduce storage & transfer
       const compressed = await compressImage(file, { maxPixels: 1920, quality: 0.8 })
-      // Get presigned URL
+      const data = new FormData()
-      const presignRes = await api.post('/upload/presigned-url', null, { params: { filename: compressed.name, content_type: compressed.type || 'image/jpeg' } })
+      data.append('file', compressed)
-      // Upload directly to MinIO (not through our API)
+      const uploadRes = await api.post('/upload/image', data, {
-      const axios = (await import('axios')).default
+        headers: { 'Content-Type': 'multipart/form-data' },
-      await axios.put(presignRes.data.upload_url, compressed, { headers: { 'Content-Type': compressed.type || 'image/jpeg' } })
+        timeout: 60000,
-      const key = presignRes.data.object_key
+      })
+      const key = uploadRes.data.object_key
       if (!form.value.photos) form.value.photos = []
       form.value.photos.push(key)
       form.value.photos = [...form.value.photos]

frontend/src/views/mobile/VisitForm.vue

@@ -128,8 +128,7 @@ async function handlePhotoUpload(event: Event) {
     try {
       // Compress before upload to reduce storage & transfer
       const compressed = await compressImage(file, { maxPixels: 1920, quality: 0.8 })
-      const res = await uploadApi.getPresignedUrl(compressed.name, compressed.type || 'image/jpeg')
+      const res = await uploadApi.uploadImage(compressed)
-      await uploadApi.uploadFile(res.data.upload_url, compressed)
       uploadedPhotos.value.push(res.data.object_key)
       form.value.photos = [...uploadedPhotos.value]
     } catch (e: any) {

审查报告-中国电信研发规范.md

@@ -14,7 +14,7 @@
 | P1 | 任意登录用户可为任意对象键签发下载 URL（水平越权） | 安全分册 3.3.3.10【强制】（对象操作的 SQL/查询须附带当前用户身份条件） | `backend/app/api/upload.py` 的 `/download-url` 直接接收 `object_key` 并签发 1 小时 URL，未查询照片所属拜访记录、客户权限或对象键前缀。已知或猜测键名的用户可下载其他员工照片。 | 不接受裸 `object_key`；由照片所属业务记录 ID 查询对象，按当前用户角色、所属人和客户授权校验后签发；对象键应使用完整 UUID 前缀并记录访问审计。 |
 | P1 | 系统配置读取接口没有鉴权 | 安全分册 3.3.3.9【强制】（管理 URL 必须校验当前用户权限） | `backend/app/api/system_config.py` 的 `GET /api/system-config` 未注入 `get_current_user` 或角色校验，会返回全部配置，包括可包含业务提示词等内部配置。 | 至少要求已认证用户；建议仅 director 可读管理配置，并对可公开的配置采用明确白名单。 |
 | P1 | 企微绑定流程信任客户端传入的 `wecom_userid`，且写入时未校验唯一性 | 安全分册 2.1.8（避免未经授权访问会话状态）；3.3.3.9【强制】 | `POST /api/auth/bind-wecom` 使用 `req.wecom_userid` 优先于受保护的绑定状态；`services/auth.py` 直接赋值并提交。攻击者可将自己的 Casdoor 账户绑定到任意企微 ID，形成账号关联劫持/覆盖风险。 | 仅接受服务端保存、一次性、短时有效且与发起人绑定的随机令牌；服务端从令牌取企微 ID；数据库加唯一约束，并在绑定/解绑时保留安全审计。 |
-| P2 | 直传上传未实际限制文件类型或验证内容 | 安全分册 3.3.4.9【强制】（扩展名白名单、不可直访目录、系统命名、图片处理、上传日志） | `/presigned-url` 接收但未使用 `filename`、`content_type`；对象固定以 `.jpg` 结尾，预签名 PUT 后服务端不检查 MIME、魔数、大小、恶意内容或上传结果。对象存储由独立端口暴露。 | 采用受控后端上传/隔离桶，限制大小和白名单；校验文件魔数并转码图片，病毒扫描后再移入正式桶；记录上传审计。保留现有的系统生成键名优点。 |
+| P2 | 上传文件缺少服务端内容验证 | 已整改（代码层）：上传接口已迁移为受控后端上传，限制大小与类型、校验图片内容、重编码为 JPEG 后才入对象存储。 | 仍需在生产环境接入恶意内容扫描并验证桶策略。 |
 | P2 | 日志可能泄露用户内容和第三方认证响应 | 安全分册 3.1.1【强制】（日志不得保存口令、密钥和其他敏感数据）；3.3.3.11【强制】（未经验证输入不得写日志） | `backend/app/api/wecom.py` 记录企微发送人和完整消息正文；`services/auth.py` 在认证失败时记录响应正文或完整 token 数据。周报/客户信息可能含个人或经营敏感内容。 | 使用结构化日志、字段白名单和脱敏；禁止记录 token、企微正文及认证响应体；限制日志访问并设置保留期限。 |
 | P2 | 缺少生产部署与变更可追溯材料 | 部署管理分册 2.2.1、2.2.2、2.3、2.5、2.6 | 仓库无上线测试报告、版本—制品—代码版本映射、部署/回退/数据备份方案、审批记录或部署验证用例；`docker-compose.yml` 以 `--reload` 和源码挂载启动后端，不是可审计的生产部署方式。 | 建立发布包：版本 Tag、制品摘要、SBOM、测试报告、部署/回滚/备份方案、审批和验证记录；生产镜像使用固定版本/摘要，不使用开发热重载或源码挂载。 |
 | P2 | 未发现自动化测试、质量扫描或 CI/CD 流水线 | 测试管理分册第 4 章；流水线管理分册 3.2、3.3 | 仓库未发现 `tests`/`test` 目录或测试脚本；前端 `package.json` 只有 dev/build/preview；无 GitHub/GitLab/其他 CI 配置。缺少单元、集成、系统测试及安全/质量扫描的可追溯证据。 | 为关键鉴权、上传、企微绑定和数据隔离补充单元及 API 集成测试；在 MR 和合入时强制执行测试、SAST、依赖/SBOM 扫描、构建和制品上传；设定测试准入/准出与缺陷闭环。 |

整改计划-中国电信研发规范.md

@@ -15,7 +15,8 @@
 | 修复照片下载越权 | 已完成 | 下载 URL 由拜访记录反查照片归属，再按 manager 所属人权限校验 |
 | 关闭不安全企微绑定 | 已完成 | 停用客户端提供 `wecom_userid` 的旧接口，仅保留一次性令牌确认流程 |
 | 管理配置鉴权 | 已完成 | 系统配置列表接口仅允许 director 访问 |
-| 上传与日志基础防护 | 部分完成 | 上传仅接受 JPEG/PNG/WebP 元数据并使用用户隔离的对象键；回调和认证失败日志不再写入正文/token |
+| 上传与日志基础防护 | 已完成（仓库侧） | 上传经后端校验尺寸与图片内容、重编码为 JPEG 后才写入私有对象存储；回调和认证失败日志不再写入正文/token |
+| 企微绑定令牌持久化 | 已完成（仓库侧） | 一次性令牌存入数据库并原子消费；启动时创建企微 ID 的部分唯一索引，防止重复绑定 |
 | 测试与构建流水线 | 已完成 | 增加上传/配置安全测试和 `.gitlab-ci.yml` 的后端测试、前端构建任务 |
 | 清理禁止的二进制文件 | 已完成 | 移除仓库根目录两份 `.xlsx` 模板；系统仍可由导入模板接口动态生成模板 |
 
@@ -25,8 +26,8 @@
 |---|---|---|---|---|
 | P0 | 在密钥管理系统创建并注入数据库、MinIO、JWT、Casdoor、企微和 AI 密钥；轮换历史弱凭据 | 运维/安全 | 上线前 | 密钥轮换记录、部署变量清单、无明文密钥扫描报告 |
 | P0 | 为 PostgreSQL/MinIO 配置私网访问、TLS、最小权限服务账号和备份恢复演练 | 运维/DBA | 上线前 | 网络策略、TLS 证书、账号权限清单、恢复演练记录 |
-| P1 | 将上传改为隔离桶或受控后端流：校验文件魔数、大小、转码、恶意内容扫描后才入正式桶 | 后端/安全 | 5 个工作日 | 自动化测试、扫描日志、隔离策略和抽样验证 |
+| P1 | 在生产环境接入恶意内容扫描，并确认对象存储桶策略不允许匿名读取 | 后端/安全 | 上线前 | 扫描日志、桶策略和抽样验证 |
-| P1 | 在生产数据库为 `users.wecom_userid` 增加唯一约束，并将一次性绑定令牌改为 Redis/数据库持久化存储 | 后端/DBA | 3 个工作日 | 迁移脚本、并发绑定测试、令牌过期与审计记录 |
+| P1 | 在生产数据库验证 `users.wecom_userid` 唯一索引创建成功，并完成并发绑定演练 | 后端/DBA | 上线前 | 数据库迁移日志、并发绑定测试、令牌过期与审计记录 |
 | P1 | 配置保护分支、MR 至少一名评审、禁止直接推送 `main`/`develop` | Git 平台管理员 | 2 个工作日 | 平台截图/导出配置、MR 审计记录 |
 | P2 | 接入 SAST、依赖漏洞/SBOM 和容器镜像扫描，并将高危结果设为流水线阻断条件 | DevSecOps | 5 个工作日 | 流水线报告、阻断规则、漏洞例外审批 |
 | P2 | 补齐鉴权、对象访问、企微绑定、文件上传的 API 集成测试；明确测试准入/准出及缺陷闭环 | 测试/后端 | 5 个工作日 | 测试计划、用例、报告、缺陷清单 |