fix: Embedded security vulnerabilities

Author: fit2cloud-chenyw

backend/apps/system/api/assistant.py

@@ -1,6 +1,7 @@
 import json
 import os
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
+from zoneinfo import ZoneInfo
 from typing import List, Optional
 
 from fastapi import APIRouter, Form, HTTPException, Path, Query, Request, Response, UploadFile
@@ -22,30 +23,83 @@
 from common.core.security import create_access_token
 from common.core.sqlbot_cache import clear_cache
 from common.utils.utils import get_origin_from_referer, origin_match_domain
-
 router = APIRouter(tags=["system_assistant"], prefix="/system/assistant")
 from common.audit.models.log_model import OperationType, OperationModules
 from common.audit.schemas.logger_decorator import LogConfig, system_log
-
+from sqlbot_xpack.core import decrypt_embedded_sign
 
 @router.get("/info/{id}", include_in_schema=False)
-async def info(request: Request, response: Response, session: SessionDep, trans: Trans, id: int) -> AssistantModel:
+async def info(request: Request, response: Response, session: SessionDep, trans: Trans, id: int, virtual: Optional[int] = Query(None)):
     if not id:
         raise Exception('miss assistant id')
     db_model = await get_assistant_info(session=session, assistant_id=id)
     if not db_model:
         raise RuntimeError(f"assistant application not exist")
     db_model = AssistantModel.model_validate(db_model)
-    
-    origin = request.headers.get("origin") or get_origin_from_referer(request)
-    if not origin:
-        raise RuntimeError(trans('i18n_embedded.invalid_origin', origin=origin or ''))
-    origin = origin.rstrip('/')
-    if not origin_match_domain(origin, db_model.domain):
-        raise RuntimeError(trans('i18n_embedded.invalid_origin', origin=origin or ''))
-    
+
+    # 校验 SQLBOT-EMBEDDED-SIGN 请求头
+    sign_header = request.headers.get("SQLBOT-EMBEDDED-SIGN")
+    if not sign_header:
+        raise RuntimeError(trans('i18n_embedded.invalid_origin', origin=''))
+
+    sign_data = await decrypt_embedded_sign(sign_header)
+
+    # 校验 assistant_id 与 id 参数一致
+    if str(sign_data.get("assistant_id")) != str(id):
+        raise RuntimeError(trans('i18n_embedded.invalid_origin', origin=''))
+
+    # 校验 target（来源域名）是否合法
+    target = sign_data.get("target", "")
+    if not origin_match_domain(target, db_model.domain):
+        raise RuntimeError(trans('i18n_embedded.invalid_origin', origin=target or ''))
+
+    # 校验 sign_time 是否在 10 秒内
+    sign_time_str = sign_data.get("sign_time", "")
+    sign_time = datetime.fromisoformat(sign_time_str)
+    now_utc = datetime.now(timezone.utc)
+    sign_time_utc = sign_time.astimezone(timezone.utc)
+    if abs((now_utc - sign_time_utc).total_seconds()) > 10:
+        raise RuntimeError(trans('i18n_embedded.invalid_origin', origin=target or ''))
+
+    # 校验是否为真实浏览器请求（非自动化工具）
+    if sign_data.get("webdriver", False):
+        raise RuntimeError(trans('i18n_embedded.invalid_origin', origin=target or ''))
+
+    # 校验 User-Agent 一致性（签名中的 navigator.userAgent 与请求头一致）
+    sign_user_agent = sign_data.get("user_agent", "")
+    request_user_agent = request.headers.get("User-Agent", "")
+    if sign_user_agent != request_user_agent:
+        raise RuntimeError(trans('i18n_embedded.invalid_origin', origin=target or ''))
+
+    # 校验 timezone 与 sign_time 偏移一致性（防时区伪造）
+    tz_name = sign_data.get("timezone", "")
+    tz = ZoneInfo(tz_name)
+    sign_time_naive = sign_time.replace(tzinfo=None)
+    if tz.utcoffset(sign_time_naive) != sign_time.utcoffset():
+        raise RuntimeError(trans('i18n_embedded.invalid_origin', origin=target or ''))
+
+    origin = target.rstrip('/')
+
     response.headers["Access-Control-Allow-Origin"] = origin
-    return db_model
+
+
+    assistant_oid = 1
+    if (db_model.type == 0):
+        configuration = db_model.configuration
+        config_obj = json.loads(configuration) if configuration else {}
+        assistant_oid = config_obj.get('oid', 1)
+
+    access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    assistantDict = {
+        "id": virtual, "account": 'sqlbot-inner-assistant', "oid": assistant_oid, "assistant_id": id
+    }
+    access_token = create_access_token(
+        assistantDict, expires_delta=access_token_expires
+    )
+
+    result = db_model.model_dump()
+    result["token"] = access_token
+    return result
 
 
 @router.get("/app/{appId}", include_in_schema=False)
@@ -67,7 +121,7 @@ async def getApp(request: Request, response: Response, session: SessionDep, tran
     return db_model
 
 
-@router.get("/validator", response_model=AssistantValidator, include_in_schema=False)
+""" @router.get("/validator", response_model=AssistantValidator, include_in_schema=False)
 async def validator(session: SessionDep, id: int, virtual: Optional[int] = Query(None)):
     if not id:
         raise Exception('miss assistant id')
@@ -89,7 +143,7 @@ async def validator(session: SessionDep, id: int, virtual: Optional[int] = Query
     access_token = create_access_token(
         assistantDict, expires_delta=access_token_expires
     )
-    return AssistantValidator(True, True, True, access_token)
+    return AssistantValidator(True, True, True, access_token) """
 
 
 @router.get('/picture/{file_id}', summary=f"{PLACEHOLDER_PREFIX}assistant_picture_api", description=f"{PLACEHOLDER_PREFIX}assistant_picture_api")

backend/apps/system/crud/assistant.py

@@ -22,6 +22,21 @@
 from common.core.response_middleware import ResponseMiddleware
 
 
+def _update_cors_middleware_instance(app: FastAPI, updated_origins: list[str]):
+    """遍历 middleware 栈，找到 CORSMiddleware 实例并更新其 allow_origins。
+
+    仅修改 middleware.kwargs 不会影响已构建的中间件实例，
+    需要直接更新实例的 allow_origins 属性。
+    """
+    stack = getattr(app, 'middleware_stack', None)
+    while stack is not None and hasattr(stack, 'app'):
+        if isinstance(stack, CORSMiddleware):
+            stack.allow_origins = updated_origins
+            return
+        stack = stack.app
+
+
+
 @cache(namespace=CacheNamespace.EMBEDDED_INFO, cacheName=CacheName.ASSISTANT_INFO, keyExpression="assistant_id")
 async def get_assistant_info(*, session: Session, assistant_id: int) -> AssistantModel | None:
     db_model = session.get(AssistantModel, assistant_id)
@@ -98,6 +113,7 @@ def init_dynamic_cors(app: FastAPI):
             updated_origins = list(set(settings.all_cors_origins + unique_domains))
             if cors_middleware:
                 cors_middleware.kwargs['allow_origins'] = updated_origins
+                _update_cors_middleware_instance(app, updated_origins)
             if response_middleware:
                 for instance in ResponseMiddleware.instances:
                     instance.update_allow_origins(updated_origins)

backend/apps/system/crud/assistant_manage.py

@@ -12,6 +12,17 @@
 from common.core.response_middleware import ResponseMiddleware
 
 
+def _update_cors_middleware_instance(app: FastAPI, updated_origins: list[str]):
+    """遍历 middleware 栈，找到 CORSMiddleware 实例并更新其 allow_origins。"""
+    stack = getattr(app, 'middleware_stack', None)
+    while stack is not None and hasattr(stack, 'app'):
+        if isinstance(stack, CORSMiddleware):
+            stack.allow_origins = updated_origins
+            return
+        stack = stack.app
+
+
+
 def dynamic_upgrade_cors(request: Request, session: Session):
     list_result = session.exec(select(AssistantModel).order_by(AssistantModel.create_time)).all()
     seen = set()
@@ -37,6 +48,7 @@ def dynamic_upgrade_cors(request: Request, session: Session):
     updated_origins = list(set(settings.all_cors_origins + unique_domains))
     if cors_middleware:
         cors_middleware.kwargs['allow_origins'] = updated_origins
+        _update_cors_middleware_instance(app, updated_origins)
     if response_middleware:
         for instance in ResponseMiddleware.instances:
             instance.update_allow_origins(updated_origins)

backend/apps/system/middleware/auth.py

@@ -1,23 +1,25 @@
 
 import base64
 import json
+import re
 from typing import Optional
 from fastapi import Request
 from fastapi.responses import JSONResponse
+from starlette.responses import Response
 import jwt
 from sqlmodel import Session
 from starlette.middleware.base import BaseHTTPMiddleware
 from apps.system.crud.apikey_manage import get_api_key
 from apps.system.models.system_model import ApiKeyModel, AssistantModel
-from common.core.db import engine 
+from common.core.db import engine
 from apps.system.crud.assistant import get_assistant_info, get_assistant_user
 from apps.system.crud.user import get_user_by_account, get_user_info
 from apps.system.schemas.system_schema import AssistantHeader, UserInfoDTO
 from common.core import security
 from common.core.config import settings
 from common.core.schemas import TokenPayload
 from common.utils.locale import I18n
-from common.utils.utils import SQLBotLogUtil, get_origin_from_referer
+from common.utils.utils import SQLBotLogUtil, get_origin_from_referer, origin_match_domain
 from common.utils.whitelist import whiteUtils
 from fastapi.security.utils import get_authorization_scheme_param
 from common.core.deps import get_i18n
@@ -31,6 +33,26 @@ def __init__(self, app):
     async def dispatch(self, request, call_next):
 
         if self.is_options(request) or whiteUtils.is_whitelisted(request.url.path):
+            # 动态处理 /system/assistant/info/{id} 的 CORS 预检
+            if request.method == "OPTIONS":
+                origin = request.headers.get("origin", "")
+                if origin:
+                    match = re.search(r'/system/assistant/info/(\d+)', request.url.path)
+                    if match:
+                        assistant_id = int(match.group(1))
+                        with Session(engine) as session:
+                            db_model = session.get(AssistantModel, assistant_id)
+                            if db_model and origin_match_domain(origin, db_model.domain):
+                                return Response(
+                                    status_code=200,
+                                    headers={
+                                        "Access-Control-Allow-Origin": origin,
+                                        "Access-Control-Allow-Methods": "GET, OPTIONS",
+                                        "Access-Control-Allow-Headers": "*",
+                                        "Access-Control-Allow-Credentials": "true",
+                                        "Access-Control-Max-Age": "600",
+                                    },
+                                )
             return await call_next(request)
         assistantTokenKey = settings.ASSISTANT_TOKEN_KEY
         assistantToken = request.headers.get(assistantTokenKey)