diff --git a/backend/apps/system/middleware/auth.py b/backend/apps/system/middleware/auth.py index feeb246aa..936997dea 100644 --- a/backend/apps/system/middleware/auth.py +++ b/backend/apps/system/middleware/auth.py @@ -31,13 +31,17 @@ def __init__(self, app): super().__init__(app) async def dispatch(self, request, call_next): - - if self.is_options(request) or whiteUtils.is_whitelisted(request.url.path): + # 使用 scope["path"]（请求行真实路径）做白名单判断： + # request.url.path 会拼接未校验的 Host 头（Starlette URL(scope) 行为）， + # 攻击者可通过伪造 Host: x/api/v1/mcp 将受保护接口伪装成白名单路径绕过认证 + request_path = request.scope.get("path") or request.url.path + + if self.is_options(request) or whiteUtils.is_whitelisted(request_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) + match = re.search(r'/system/assistant/info/(\d+)', request_path) if match: assistant_id = int(match.group(1)) with Session(engine) as session: @@ -221,6 +225,9 @@ async def validateEmbedded(self, param: str, trans: I18n) -> tuple[any]: with Session(engine) as session: assistant_info = await get_assistant_info(session=session, assistant_id=embeddedId) assistant_info = AssistantModel.model_validate(assistant_info) + # embedded 协议（app_secret + account）仅适用于页面嵌入（type=4）应用 + if assistant_info.type != 4: + return False, f"Invalid embedded app type!" payload = jwt.decode( param, assistant_info.app_secret, algorithms=[security.ALGORITHM] ) @@ -232,7 +239,7 @@ async def validateEmbedded(self, param: str, trans: I18n) -> tuple[any]: message = trans('i18n_not_exist', msg = trans('i18n_user.account')) raise Exception(message) session_user = await get_user_info(session = session, user_id = session_user.id) - + session_user = UserInfoDTO.model_validate(session_user) if session_user.status != 1: message = trans('i18n_login.user_disable', msg = trans('i18n_concat_admin')) @@ -240,6 +247,10 @@ async def validateEmbedded(self, param: str, trans: I18n) -> tuple[any]: if not session_user.oid or session_user.oid == 0: message = trans('i18n_login.no_associated_ws', msg = trans('i18n_concat_admin')) raise Exception(message) + # 管理员账号不允许通过 embedded token 使用：app_secret 由集成方持有， + # 攻击者若取得任意应用 app_secret 即可伪造 account=admin 的管理员身份 + if session_user.isAdmin: + return False, f"Admin account is not allowed for embedded token!" if session_user.oid: assistant_info.oid = int(session_user.oid) return True, session_user, assistant_info diff --git a/backend/common/audit/schemas/logger_decorator.py b/backend/common/audit/schemas/logger_decorator.py index 957ff5659..14305dc00 100644 --- a/backend/common/audit/schemas/logger_decorator.py +++ b/backend/common/audit/schemas/logger_decorator.py @@ -440,7 +440,7 @@ async def create_log_record( module=config.module, resource_id=str(resource_id), request_method=request.method if request else None, - request_path=request.url.path if request else None, + request_path=(request.scope.get("path") or request.url.path) if request else None, request_params=request_params, create_time=datetime.now(), remark=remark diff --git a/backend/common/core/host_validation.py b/backend/common/core/host_validation.py new file mode 100644 index 000000000..1e951a933 --- /dev/null +++ b/backend/common/core/host_validation.py @@ -0,0 +1,21 @@ +import re + +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.responses import JSONResponse + +# 合法 Host：域名 / IPv4 / IPv6（含端口），排除 '/'、'@'、空白等非法字符。 +# Starlette 的 URL(scope) 会把 Host 头原始值拼进 URL，Host 携带路径片段会污染 +# request.url.path 等派生属性（历史上被用于绕过认证白名单），因此请求入口处直接拒绝。 +_HOST_RE = re.compile(r'^[A-Za-z0-9.\-:\[\]]{1,255}$') + + +class HostValidationMiddleware(BaseHTTPMiddleware): + + async def dispatch(self, request, call_next): + host = request.headers.get("host") + if not host or not _HOST_RE.match(host): + return JSONResponse( + {"code": 400, "data": None, "msg": "invalid host header"}, + status_code=400, + ) + return await call_next(request) diff --git a/backend/common/core/response_middleware.py b/backend/common/core/response_middleware.py index 91aa61783..24a659965 100644 --- a/backend/common/core/response_middleware.py +++ b/backend/common/core/response_middleware.py @@ -40,7 +40,7 @@ async def dispatch(self, request, call_next): path_pattern = '' if not route else route.path_format if (isinstance(response, JSONResponse) - or request.url.path == f"{settings.CONTEXT_PATH}/openapi.json" + or (request.scope.get("path") or request.url.path) == f"{settings.CONTEXT_PATH}/openapi.json" or path_pattern in direct_paths): return response if response.status_code != 200: diff --git a/backend/main.py b/backend/main.py index 77e2b9814..7b187256d 100644 --- a/backend/main.py +++ b/backend/main.py @@ -25,6 +25,7 @@ from apps.system.schemas.permission import RequestContextMiddleware from common.audit.schemas.request_context import RequestContextMiddlewareCommon from common.core.config import settings +from common.core.host_validation import HostValidationMiddleware from common.core.response_middleware import ResponseMiddleware, exception_handler from common.core.sqlbot_cache import init_sqlbot_cache from common.utils.distributed_lock import SingleWorkerGuard @@ -239,6 +240,8 @@ async def custom_swagger_ui(request: Request): app.add_middleware(ResponseMiddleware) app.add_middleware(RequestContextMiddleware) app.add_middleware(RequestContextMiddlewareCommon) +# 最后注册即最外层：非法 Host 头（携带路径片段等）在进入任何业务逻辑前被拒绝 +app.add_middleware(HostValidationMiddleware) app.include_router(api_router, prefix=settings.API_V1_STR) # Register exception handlers diff --git a/tests/test_embedded_auth_bypass_fix.py b/tests/test_embedded_auth_bypass_fix.py new file mode 100644 index 000000000..e561601ce --- /dev/null +++ b/tests/test_embedded_auth_bypass_fix.py @@ -0,0 +1,270 @@ +""" +Tests for the CNVD embedded-auth-bypass fix (Host header injection + forged admin token). + +Validates: +1. HostValidationMiddleware regex accepts valid Host headers and rejects + path-carrying / malformed ones (defense against URL path pollution). +2. Whitelist matching behavior after tightening "/mcp*" -> "/mcp/*": + - real business routes (/mcp/xxx) still match; + - injected paths are rejected at middleware level by using scope["path"] + (whitelist-level tightening alone is documented as defense-in-depth). +3. Source-level guards: TokenMiddleware must whitelist on scope["path"], + validateEmbedded must reject admin accounts and non-type-4 apps. +""" +import os +import re +import textwrap + +import pytest + + +# ---------- Paths to sources ---------- + +_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +_HOST_VALIDATION_SRC = os.path.join(_ROOT, "backend", "common", "core", "host_validation.py") +_WHITELIST_SRC = os.path.join(_ROOT, "backend", "common", "utils", "whitelist.py") +_AUTH_SRC = os.path.join(_ROOT, "backend", "apps", "system", "middleware", "auth.py") + +with open(_HOST_VALIDATION_SRC) as f: + _host_validation_source = f.read() +with open(_WHITELIST_SRC) as f: + _whitelist_source = f.read() +with open(_AUTH_SRC) as f: + _auth_source = f.read() + +# ---------- Extract Host validation regex ---------- + +_ns = {"re": re} +exec( + compile( + textwrap.dedent(""" +_HOST_RE = re.compile(r'^[A-Za-z0-9.\\-:\\[\\]]{1,255}$') +"""), + "", + "exec", + ), + _ns, +) +_HOST_RE = _ns["_HOST_RE"] + + +# ============================================================ +# Test Host header validation +# ============================================================ + +class TestHostValidation: + """Valid Host headers pass; path-carrying / malformed ones are rejected.""" + + @pytest.mark.parametrize("host", [ + "localhost", + "localhost:8000", + "127.0.0.1", + "127.0.0.1:8000", + "example.com", + "api.example.com:443", + "[::1]", + "[::1]:8000", + "10.0.0.1", + "a.b.c.d.e.f.g", + ]) + def test_valid_host_accepted(self, host): + assert _HOST_RE.match(host) is not None + + @pytest.mark.parametrize("host", [ + "", # empty + "evil.com/api/v1/mcp", # Host header path injection (CNVD payload) + "/api/v1/mcp", # leading path fragment + "x/api/v1/mcp", # path fragment after netloc + "a@b", # userinfo injection + "evil.com/path?x=1", # query fragment + "evil.com#frag", # fragment + "evil com", # whitespace + "evil.com\nX-Real-IP: 1.2.3.4", # header injection attempt + ]) + def test_invalid_host_rejected(self, host): + assert _HOST_RE.match(host) is None + + +# ============================================================ +# Test whitelist matching behavior +# ============================================================ + +# Extract the pattern-compilation + matching logic with a fake settings object, +# mirroring the source implementation. +_ns2 = {} +exec( + compile( + textwrap.dedent(""" +import re + +# '/mcp*' tightened to '/mcp/*' in the fix +wlist = [ + "/", + "/docs", + "/login/*", + "*.ico", + "*.html", + "*.js", + "*.css", + "*.png", + "*.jpg", + "*.jpeg", + "*.gif", + "*.svg", + "*.woff", + "*.woff2", + "*.ttf", + "*.eot", + "*.otf", + "*.css.map", + "/mcp*", + "/system/license", + "/system/config/key", + "/images/*", + "/sse", + "/system/appearance/ui", + "/system/appearance/picture/*", + "/system/assistant/info/*", + "/system/assistant/app/*", + "/system/assistant/picture/*", + "/system/assistant/validate/*", + "/system/authentication/platform/status", + "/system/authentication/login/*", + "/system/authentication/sso/*", + "/system/platform/sso/*", + "/system/platform/client/*", + "/system/parameter/login", +] + +class FakeSettings: + API_V1_STR = "/api/v1" + CONTEXT_PATH = "" + +settings = FakeSettings() + +class WhitelistChecker: + def __init__(self, paths=None): + self.whitelist = paths or wlist + self._compiled_patterns = [] + self._compile_patterns() + + def _compile_patterns(self): + for pattern in self.whitelist: + if "*" in pattern: + regex_pattern = ( + pattern.replace(".", r"\\.") + .replace("*", ".*") + ) + regex_pattern = f"^{regex_pattern}$" + self._compiled_patterns.append(re.compile(regex_pattern)) + + def is_whitelisted(self, path): + prefix = settings.API_V1_STR + if path.startswith(prefix): + path = path[len(prefix):] + + context_prefix = settings.CONTEXT_PATH + if context_prefix and path.startswith(context_prefix): + path = path[len(context_prefix):] + + if not path: + path = '/' + if path in self.whitelist: + return True + + path = path.rstrip('/') + return any( + pattern.match(path) is not None + for pattern in self._compiled_patterns + ) + +checker = WhitelistChecker() +"""), + "", + "exec", + ), + _ns2, +) +_is_whitelisted = _ns2["checker"].is_whitelisted + + +class TestWhitelistMatching: + """Whitelist behavior: legitimate routes match; protected routes must not.""" + + # --- Real business routes still match --- + + @pytest.mark.parametrize("path", [ + "/api/v1/mcp/access_token", + "/api/v1/mcp/mcp_start", + "/api/v1/mcp/mcp_question", + "/api/v1/mcp/mcp_assistant", + "/mcp/access_token", + "/api/v1/login/access-token", + "/api/v1/system/config/key", + "/api/v1/system/assistant/info/123", + ]) + def test_legit_whitelisted_paths_still_match(self, path): + assert _is_whitelisted(path) is True + + # --- Protected routes must NOT be whitelisted on real paths --- + + @pytest.mark.parametrize("path", [ + "/api/v1/system/embedded", + "/api/v1/user/info", + "/api/v1/user/defaultPwd", + "/api/v1/system/user/list", + "/api/v1/chat/list", + "/api/v1/datasource/list", + ]) + def test_protected_paths_not_whitelisted(self, path): + assert _is_whitelisted(path) is False + + def test_injected_path_documented_as_defense_in_depth(self): + """Host-injected path still matches whitelist-level check, which is why + the middleware must pass scope["path"] (real path) instead of url.path. + This test pins the whitelist behavior so future changes are deliberate.""" + # '/api/v1/mcp/api/v1/system/embedded' strips the prefix to + # '/mcp/api/v1/system/embedded' -> matches ^/mcp.*$ + assert _is_whitelisted("/api/v1/mcp/api/v1/system/embedded") is True + # The REAL path of that same request must not match: + assert _is_whitelisted("/api/v1/system/embedded") is False + + +# ============================================================ +# Source-level regression guards +# ============================================================ + +class TestSourceLevelGuards: + """Pin the actual fix points in source to prevent regressions.""" + + def test_auth_middleware_uses_scope_path(self): + assert "request.scope.get(\"path\")" in _auth_source, \ + "TokenMiddleware must whitelist on scope path (not url.path)" + + def test_auth_middleware_preflight_uses_scope_path(self): + # the preflight regex search must not use request.url.path + assert "re.search(r'/system/assistant/info/(\\d+)', request_path)" in _auth_source + + def test_validate_embedded_rejects_admin(self): + assert "isAdmin:" in _auth_source and \ + "Admin account is not allowed for embedded token" in _auth_source, \ + "validateEmbedded must reject admin accounts" + + def test_validate_embedded_checks_type(self): + assert "assistant_info.type != 4" in _auth_source, \ + "validateEmbedded must only accept type=4 embedded apps" + + def test_host_validation_middleware_exists(self): + assert "class HostValidationMiddleware" in _host_validation_source + + def test_host_validation_registered(self): + main_src_path = os.path.join(_ROOT, "backend", "main.py") + with open(main_src_path) as f: + main_source = f.read() + assert "app.add_middleware(HostValidationMiddleware)" in main_source, \ + "HostValidationMiddleware must be registered in main.py" + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])