mirror of
https://gh-proxy.org/https://github.com/klinsmaya/SQLBot.git
synced 2026-11-08 01:06:59 +08:00
fit2cloud-chenywandClaude
bbfc4cc8f5
- TokenMiddleware/ResponseMiddleware/audit logger use scope['path'] instead of request.url.path for whitelist checks: a crafted Host header (e.g. x/api/v1/mcp) was concatenated into Starlette's URL and made protected routes match the /mcp* whitelist entry - validateEmbedded rejects admin accounts and non-type-4 apps, so a self-signed token with account=admin can no longer impersonate the administrator - Add HostValidationMiddleware rejecting malformed Host headers at the outermost layer (defense in depth) - Add regression tests for host validation, whitelist matching and source-level guards Co-Authored-By: Claude <[email protected]>
271 lines
8.4 KiB
Python
271 lines
8.4 KiB
Python
"""
|
|
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}$')
|
|
"""),
|
|
"<extracted>",
|
|
"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()
|
|
"""),
|
|
"<extracted>",
|
|
"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"])
|