Merge pull request #1439 from A-Dawn/dev

feat(WebUI)：增加防爬虫、资产扫描检测中间件与 IP 白名单功能

src/webui/anti_crawler.py

@@ -0,0 +1,773 @@
+"""
+WebUI 防爬虫模块
+提供爬虫检测和阻止功能，保护 WebUI 不被搜索引擎和恶意爬虫访问
+"""
+
+import os
+import time
+import ipaddress
+import re
+from collections import deque
+from typing import Optional
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from starlette.responses import PlainTextResponse
+
+from src.common.logger import get_logger
+
+logger = get_logger("webui.anti_crawler")
+
+# 常见爬虫 User-Agent 列表（使用更精确的关键词，避免误报）
+CRAWLER_USER_AGENTS = {
+    # 搜索引擎爬虫（精确匹配）
+    "googlebot",
+    "bingbot",
+    "baiduspider",
+    "yandexbot",
+    "slurp",  # Yahoo
+    "duckduckbot",
+    "sogou",
+    "exabot",
+    "facebot",
+    "ia_archiver",  # Internet Archive
+    # 通用爬虫（移除过于宽泛的关键词）
+    "crawler",
+    "spider",
+    "scraper",
+    "wget",  # 保留wget，因为通常用于自动化脚本
+    "scrapy",  # 保留scrapy，因为这是爬虫框架
+    # 安全扫描工具（这些是明确的扫描工具）
+    "masscan",
+    "nmap",
+    "nikto",
+    "sqlmap",
+    # 注意：移除了以下过于宽泛的关键词以避免误报：
+    # - "bot" (会误匹配GitHub-Robot等)
+    # - "curl" (正常工具)
+    # - "python-requests" (正常库)
+    # - "httpx" (正常库)
+    # - "aiohttp" (正常库)
+}
+
+# 资产测绘工具 User-Agent 标识
+ASSET_SCANNER_USER_AGENTS = {
+    # 知名资产测绘平台
+    "shodan",
+    "censys",
+    "zoomeye",
+    "fofa",
+    "quake",
+    "hunter",
+    "binaryedge",
+    "onyphe",
+    "securitytrails",
+    "virustotal",
+    "passivetotal",
+    # 安全扫描工具
+    "acunetix",
+    "appscan",
+    "burpsuite",
+    "nessus",
+    "openvas",
+    "qualys",
+    "rapid7",
+    "tenable",
+    "veracode",
+    "zap",
+    "awvs",  # Acunetix Web Vulnerability Scanner
+    "netsparker",
+    "skipfish",
+    "w3af",
+    "arachni",
+    # 其他扫描工具
+    "masscan",
+    "zmap",
+    "nmap",
+    "whatweb",
+    "wpscan",
+    "joomscan",
+    "dnsenum",
+    "subfinder",
+    "amass",
+    "sublist3r",
+    "theharvester",
+}
+
+# 资产测绘工具常用的HTTP头标识
+ASSET_SCANNER_HEADERS = {
+    # 常见的扫描工具自定义头
+    "x-scan": {"shodan", "censys", "zoomeye", "fofa"},
+    "x-scanner": {"nmap", "masscan", "zmap"},
+    "x-probe": {"masscan", "zmap"},
+    # 其他可疑头（移除反向代理标准头）
+    "x-originating-ip": set(),
+    "x-remote-ip": set(),
+    "x-remote-addr": set(),
+    # 注意：移除了以下反向代理标准头以避免误报：
+    # - "x-forwarded-proto" (反向代理标准头)
+    # - "x-real-ip" (反向代理标准头，已在_get_client_ip中使用)
+}
+
+# 仅检查特定HTTP头中的可疑模式（收紧匹配范围）
+# 只检查这些特定头，不检查所有头
+SCANNER_SPECIFIC_HEADERS = {
+    "x-scan",
+    "x-scanner",
+    "x-probe",
+    "x-originating-ip",
+    "x-remote-ip",
+    "x-remote-addr",
+}
+
+# 防爬虫模式配置
+# false: 禁用
+# strict: 严格模式（更严格的检测，更低的频率限制）
+# loose: 宽松模式（较宽松的检测，较高的频率限制）
+# basic: 基础模式（只记录恶意访问，不阻止，不限制请求数，不跟踪IP）
+ANTI_CRAWLER_MODE = os.getenv("WEBUI_ANTI_CRAWLER_MODE", "basic").lower()
+
+# IP白名单配置（从环境变量读取，逗号分隔）
+# 支持格式：
+# - 精确IP：127.0.0.1, 192.168.1.100
+# - CIDR格式：192.168.1.0/24, 172.17.0.0/16 (适用于Docker网络)
+# - 通配符：192.168.*.*, 10.*.*.*, *.*.*.* (匹配所有)
+# - IPv6：::1, 2001:db8::/32
+def _parse_allowed_ips(ip_string: str) -> list:
+    """
+    解析IP白名单字符串，支持精确IP、CIDR格式和通配符
+    
+    Args:
+        ip_string: 逗号分隔的IP字符串
+        
+    Returns:
+        IP白名单列表，每个元素可能是：
+        - ipaddress.IPv4Network/IPv6Network对象（CIDR格式）
+        - ipaddress.IPv4Address/IPv6Address对象（精确IP）
+        - str（通配符模式，已转换为正则表达式）
+    """
+    allowed = []
+    if not ip_string:
+        return allowed
+    
+    for ip_entry in ip_string.split(","):
+        ip_entry = ip_entry.strip()  # 去除空格
+        if not ip_entry:
+            continue
+        
+        # 检查通配符格式（包含*）
+        if "*" in ip_entry:
+            # 处理通配符
+            pattern = _convert_wildcard_to_regex(ip_entry)
+            if pattern:
+                allowed.append(pattern)
+            else:
+                logger.warning(f"无效的通配符IP格式，已忽略: {ip_entry}")
+            continue
+            
+        try:
+            # 尝试解析为CIDR格式（包含/）
+            if "/" in ip_entry:
+                allowed.append(ipaddress.ip_network(ip_entry, strict=False))
+            else:
+                # 精确IP地址
+                allowed.append(ipaddress.ip_address(ip_entry))
+        except (ValueError, AttributeError) as e:
+            logger.warning(f"无效的IP白名单条目，已忽略: {ip_entry} ({e})")
+    
+    return allowed
+
+
+def _convert_wildcard_to_regex(wildcard_pattern: str) -> Optional[str]:
+    """
+    将通配符IP模式转换为正则表达式
+    
+    支持的格式：
+    - 192.168.*.* 或 192.168.*
+    - 10.*.*.* 或 10.*
+    - *.*.*.* 或 *
+    
+    Args:
+        wildcard_pattern: 通配符模式字符串
+        
+    Returns:
+        正则表达式字符串，如果格式无效则返回None
+    """
+    # 去除空格
+    pattern = wildcard_pattern.strip()
+    
+    # 处理单个*（匹配所有）
+    if pattern == "*":
+        return r".*"
+    
+    # 处理IPv4通配符格式
+    # 支持：192.168.*.*, 192.168.*, 10.*.*.*, 10.* 等
+    parts = pattern.split(".")
+    
+    if len(parts) > 4:
+        return None  # IPv4最多4段
+    
+    # 构建正则表达式
+    regex_parts = []
+    for part in parts:
+        part = part.strip()
+        if part == "*":
+            regex_parts.append(r"\d+")  # 匹配任意数字
+        elif part.isdigit():
+            # 验证数字范围（0-255）
+            num = int(part)
+            if 0 <= num <= 255:
+                regex_parts.append(re.escape(part))
+            else:
+                return None  # 无效的数字
+        else:
+            return None  # 无效的格式
+    
+    # 如果部分少于4段，补充.*
+    while len(regex_parts) < 4:
+        regex_parts.append(r"\d+")
+    
+    # 组合成正则表达式
+    regex = r"^" + r"\.".join(regex_parts) + r"$"
+    return regex
+
+ALLOWED_IPS = _parse_allowed_ips(os.getenv("WEBUI_ALLOWED_IPS", ""))
+
+# 信任的代理IP配置（从环境变量读取，逗号分隔）
+# 只有在信任的代理IP下才使用X-Forwarded-For头
+# 默认关闭（空），不信任任何代理
+TRUSTED_PROXIES = _parse_allowed_ips(os.getenv("WEBUI_TRUSTED_PROXIES", ""))
+TRUST_XFF = os.getenv("WEBUI_TRUST_XFF", "false").lower() == "true"
+
+
+def _get_mode_config(mode: str) -> dict:
+    """
+    根据模式获取配置参数
+
+    Args:
+        mode: 防爬虫模式 (false/strict/loose/basic)
+
+    Returns:
+        配置字典，包含所有相关参数
+    """
+    mode = mode.lower()
+    
+    if mode == "false":
+        return {
+            "enabled": False,
+            "rate_limit_window": 60,
+            "rate_limit_max_requests": 1000,  # 禁用时设置很高的值
+            "max_tracked_ips": 0,
+            "check_user_agent": False,
+            "check_asset_scanner": False,
+            "check_rate_limit": False,
+            "block_on_detect": False,  # 不阻止
+        }
+    elif mode == "strict":
+        return {
+            "enabled": True,
+            "rate_limit_window": 60,
+            "rate_limit_max_requests": 15,  # 严格模式：更低的请求数
+            "max_tracked_ips": 20000,
+            "check_user_agent": True,
+            "check_asset_scanner": True,
+            "check_rate_limit": True,
+            "block_on_detect": True,  # 阻止恶意访问
+        }
+    elif mode == "loose":
+        return {
+            "enabled": True,
+            "rate_limit_window": 60,
+            "rate_limit_max_requests": 60,  # 宽松模式：更高的请求数
+            "max_tracked_ips": 5000,
+            "check_user_agent": True,
+            "check_asset_scanner": True,
+            "check_rate_limit": True,
+            "block_on_detect": True,  # 阻止恶意访问
+        }
+    else:  # basic (默认模式)
+        return {
+            "enabled": True,
+            "rate_limit_window": 60,
+            "rate_limit_max_requests": 1000,  # 不限制请求数
+            "max_tracked_ips": 0,  # 不跟踪IP
+            "check_user_agent": True,  # 检测但不阻止
+            "check_asset_scanner": True,  # 检测但不阻止
+            "check_rate_limit": False,  # 不限制请求频率
+            "block_on_detect": False,  # 只记录，不阻止
+        }
+
+
+class AntiCrawlerMiddleware(BaseHTTPMiddleware):
+    """防爬虫中间件"""
+
+    def __init__(self, app, mode: str = "standard"):
+        """
+        初始化防爬虫中间件
+
+        Args:
+            app: FastAPI 应用实例
+            mode: 防爬虫模式 (false/strict/loose/standard)
+        """
+        super().__init__(app)
+        self.mode = mode.lower()
+        # 根据模式获取配置
+        config = _get_mode_config(self.mode)
+        self.enabled = config["enabled"]
+        self.rate_limit_window = config["rate_limit_window"]
+        self.rate_limit_max_requests = config["rate_limit_max_requests"]
+        self.max_tracked_ips = config["max_tracked_ips"]
+        self.check_user_agent = config["check_user_agent"]
+        self.check_asset_scanner = config["check_asset_scanner"]
+        self.check_rate_limit = config["check_rate_limit"]
+        self.block_on_detect = config["block_on_detect"]  # 是否阻止检测到的恶意访问
+        
+        # 用于存储每个IP的请求时间戳（使用deque提高性能）
+        self.request_times: dict[str, deque] = {}
+        # 上次清理时间
+        self.last_cleanup = time.time()
+        # 将关键词列表转换为集合以提高查找性能
+        self.crawler_keywords_set = set(CRAWLER_USER_AGENTS)
+        self.scanner_keywords_set = set(ASSET_SCANNER_USER_AGENTS)
+
+    def _is_crawler_user_agent(self, user_agent: Optional[str]) -> bool:
+        """
+        检测是否为爬虫 User-Agent
+
+        Args:
+            user_agent: User-Agent 字符串
+
+        Returns:
+            如果是爬虫则返回 True
+        """
+        if not user_agent:
+            # 没有 User-Agent 的请求记录日志但不直接阻止
+            # 改为只记录，让频率限制来处理
+            logger.debug("请求缺少User-Agent")
+            return False  # 不再直接阻止无User-Agent的请求
+
+        user_agent_lower = user_agent.lower()
+
+        # 使用集合查找提高性能（检查是否包含爬虫关键词）
+        for crawler_keyword in self.crawler_keywords_set:
+            if crawler_keyword in user_agent_lower:
+                return True
+
+        return False
+
+
+    def _is_asset_scanner_header(self, request: Request) -> bool:
+        """
+        检测是否为资产测绘工具的HTTP头（只检查特定头，收紧匹配）
+
+        Args:
+            request: 请求对象
+
+        Returns:
+            如果检测到资产测绘工具头则返回 True
+        """
+        # 只检查特定的扫描工具头，不检查所有头
+        for header_name, header_value in request.headers.items():
+            header_name_lower = header_name.lower()
+            header_value_lower = header_value.lower() if header_value else ""
+
+            # 检查已知的扫描工具头
+            if header_name_lower in ASSET_SCANNER_HEADERS:
+                # 如果该头有特定的工具集合，检查值是否匹配
+                expected_tools = ASSET_SCANNER_HEADERS[header_name_lower]
+                if expected_tools:
+                    for tool in expected_tools:
+                        if tool in header_value_lower:
+                            return True
+                else:
+                    # 如果没有特定工具集合，只要存在该头就视为可疑
+                    if header_value_lower:
+                        return True
+
+            # 只检查特定头中的可疑模式（收紧匹配）
+            if header_name_lower in SCANNER_SPECIFIC_HEADERS:
+                # 检查头值中是否包含已知扫描工具名称
+                for tool in self.scanner_keywords_set:
+                    if tool in header_value_lower:
+                        return True
+
+        return False
+
+    def _detect_asset_scanner(self, request: Request) -> tuple[bool, Optional[str]]:
+        """
+        检测资产测绘工具
+
+        Args:
+            request: 请求对象
+
+        Returns:
+            (是否检测到, 检测到的工具名称)
+        """
+        user_agent = request.headers.get("User-Agent")
+
+        # 检查 User-Agent（使用集合查找提高性能）
+        if user_agent:
+            user_agent_lower = user_agent.lower()
+            for scanner_keyword in self.scanner_keywords_set:
+                if scanner_keyword in user_agent_lower:
+                    return True, scanner_keyword
+
+        # 检查HTTP头
+        if self._is_asset_scanner_header(request):
+            # 尝试从User-Agent或头中提取工具名称
+            detected_tool = None
+            if user_agent:
+                user_agent_lower = user_agent.lower()
+                for tool in self.scanner_keywords_set:
+                    if tool in user_agent_lower:
+                        detected_tool = tool
+                        break
+
+            # 检查HTTP头中的工具标识（只检查特定头）
+            if not detected_tool:
+                for header_name, header_value in request.headers.items():
+                    header_name_lower = header_name.lower()
+                    if header_name_lower in SCANNER_SPECIFIC_HEADERS:
+                        header_value_lower = (header_value or "").lower()
+                        for tool in self.scanner_keywords_set:
+                            if tool in header_value_lower:
+                                detected_tool = tool
+                                break
+                        if detected_tool:
+                            break
+
+            return True, detected_tool or "unknown_scanner"
+
+        return False, None
+
+    def _check_rate_limit(self, client_ip: str) -> bool:
+        """
+        检查请求频率限制
+
+        Args:
+            client_ip: 客户端IP地址
+
+        Returns:
+            如果超过限制则返回 True（需要阻止）
+        """
+        # 检查IP白名单
+        if self._is_ip_allowed(client_ip):
+            return False
+
+        current_time = time.time()
+
+        # 定期清理过期的请求记录（每5分钟清理一次）
+        if current_time - self.last_cleanup > 300:
+            self._cleanup_old_requests(current_time)
+            self.last_cleanup = current_time
+
+        # 限制跟踪的IP数量，防止内存泄漏
+        if self.max_tracked_ips > 0 and len(self.request_times) > self.max_tracked_ips:
+            # 清理最旧的记录（删除最久未访问的IP）
+            self._cleanup_oldest_ips()
+
+        # 获取或创建该IP的请求时间deque（不使用maxlen，避免限流变松）
+        if client_ip not in self.request_times:
+            self.request_times[client_ip] = deque()
+
+        request_times = self.request_times[client_ip]
+
+        # 移除时间窗口外的请求记录（从左侧弹出过期记录）
+        while request_times and current_time - request_times[0] >= self.rate_limit_window:
+            request_times.popleft()
+
+        # 检查是否超过限制
+        if len(request_times) >= self.rate_limit_max_requests:
+            return True
+
+        # 记录当前请求时间
+        request_times.append(current_time)
+        return False
+
+    def _cleanup_old_requests(self, current_time: float):
+        """清理过期的请求记录（只清理当前需要检查的IP，不全量遍历）"""
+        # 这个方法现在主要用于定期清理，实际清理在_check_rate_limit中按需进行
+        # 清理最久未访问的IP记录
+        if len(self.request_times) > self.max_tracked_ips * 0.8:
+            self._cleanup_oldest_ips()
+
+    def _cleanup_oldest_ips(self):
+        """清理最久未访问的IP记录（全量遍历找真正的oldest）"""
+        if not self.request_times:
+            return
+
+        # 先收集空deque的IP（优先删除）
+        empty_ips = []
+        # 找到最久未访问的IP（最旧时间戳）
+        oldest_ip = None
+        oldest_time = float('inf')
+
+        # 全量遍历找真正的oldest（超限时性能可接受）
+        for ip, times in self.request_times.items():
+            if not times:
+                # 空deque，记录待删除
+                empty_ips.append(ip)
+            else:
+                # 找到最旧的时间戳
+                if times[0] < oldest_time:
+                    oldest_time = times[0]
+                    oldest_ip = ip
+
+        # 先删除空deque的IP
+        for ip in empty_ips:
+            del self.request_times[ip]
+
+        # 如果没有空deque可删除，且仍需要清理，删除最旧的一个IP
+        if not empty_ips and oldest_ip:
+            del self.request_times[oldest_ip]
+
+    def _is_trusted_proxy(self, ip: str) -> bool:
+        """
+        检查IP是否在信任的代理列表中
+
+        Args:
+            ip: IP地址字符串
+
+        Returns:
+            如果是信任的代理则返回 True
+        """
+        if not TRUSTED_PROXIES or ip == "unknown":
+            return False
+        
+        # 检查代理列表中的每个条目
+        for trusted_entry in TRUSTED_PROXIES:
+            # 通配符模式（字符串，正则表达式）
+            if isinstance(trusted_entry, str):
+                try:
+                    if re.match(trusted_entry, ip):
+                        return True
+                except re.error:
+                    continue
+            # CIDR格式（网络对象）
+            elif isinstance(trusted_entry, (ipaddress.IPv4Network, ipaddress.IPv6Network)):
+                try:
+                    client_ip_obj = ipaddress.ip_address(ip)
+                    if client_ip_obj in trusted_entry:
+                        return True
+                except (ValueError, AttributeError):
+                    continue
+            # 精确IP（地址对象）
+            elif isinstance(trusted_entry, (ipaddress.IPv4Address, ipaddress.IPv6Address)):
+                try:
+                    client_ip_obj = ipaddress.ip_address(ip)
+                    if client_ip_obj == trusted_entry:
+                        return True
+                except (ValueError, AttributeError):
+                    continue
+        
+        return False
+
+    def _get_client_ip(self, request: Request) -> str:
+        """
+        获取客户端真实IP地址（带基本验证和代理信任检查）
+
+        Args:
+            request: 请求对象
+
+        Returns:
+            客户端IP地址
+        """
+        # 获取直接连接的客户端IP（用于验证代理）
+        direct_client_ip = None
+        if request.client:
+            direct_client_ip = request.client.host
+
+        # 检查是否信任X-Forwarded-For头
+        # TRUST_XFF 只表示"启用代理解析能力"，但仍要求直连 IP 在 TRUSTED_PROXIES 中
+        use_xff = False
+        if TRUST_XFF and TRUSTED_PROXIES and direct_client_ip:
+            # 只有在启用 TRUST_XFF 且直连 IP 在信任列表中时，才信任 XFF
+            use_xff = self._is_trusted_proxy(direct_client_ip)
+
+        # 如果信任代理，优先从 X-Forwarded-For 获取
+        if use_xff:
+            forwarded_for = request.headers.get("X-Forwarded-For")
+            if forwarded_for:
+                # X-Forwarded-For 可能包含多个IP，取第一个
+                ip = forwarded_for.split(",")[0].strip()
+                # 基本验证IP格式
+                if self._validate_ip(ip):
+                    return ip
+
+        # 从 X-Real-IP 获取（如果信任代理）
+        if use_xff:
+            real_ip = request.headers.get("X-Real-IP")
+            if real_ip:
+                ip = real_ip.strip()
+                if self._validate_ip(ip):
+                    return ip
+
+        # 使用直接连接的客户端IP
+        if direct_client_ip and self._validate_ip(direct_client_ip):
+            return direct_client_ip
+
+        return "unknown"
+
+    def _validate_ip(self, ip: str) -> bool:
+        """
+        验证IP地址格式
+
+        Args:
+            ip: IP地址字符串
+
+        Returns:
+            如果格式有效则返回 True
+        """
+        try:
+            ipaddress.ip_address(ip)
+            return True
+        except (ValueError, AttributeError):
+            return False
+
+    def _is_ip_allowed(self, ip: str) -> bool:
+        """
+        检查IP是否在白名单中（支持精确IP、CIDR格式和通配符）
+
+        Args:
+            ip: 客户端IP地址
+
+        Returns:
+            如果IP在白名单中则返回 True
+        """
+        if not ALLOWED_IPS or ip == "unknown":
+            return False
+        
+        # 检查白名单中的每个条目
+        for allowed_entry in ALLOWED_IPS:
+            # 通配符模式（字符串，正则表达式）
+            if isinstance(allowed_entry, str):
+                try:
+                    if re.match(allowed_entry, ip):
+                        return True
+                except re.error:
+                    # 正则表达式错误，跳过
+                    continue
+            # CIDR格式（网络对象）
+            elif isinstance(allowed_entry, (ipaddress.IPv4Network, ipaddress.IPv6Network)):
+                try:
+                    client_ip_obj = ipaddress.ip_address(ip)
+                    if client_ip_obj in allowed_entry:
+                        return True
+                except (ValueError, AttributeError):
+                    # IP格式无效，跳过
+                    continue
+            # 精确IP（地址对象）
+            elif isinstance(allowed_entry, (ipaddress.IPv4Address, ipaddress.IPv6Address)):
+                try:
+                    client_ip_obj = ipaddress.ip_address(ip)
+                    if client_ip_obj == allowed_entry:
+                        return True
+                except (ValueError, AttributeError):
+                    # IP格式无效，跳过
+                    continue
+        
+        return False
+
+    async def dispatch(self, request: Request, call_next):
+        """
+        处理请求
+
+        Args:
+            request: 请求对象
+            call_next: 下一个中间件或路由处理函数
+
+        Returns:
+            响应对象
+        """
+        # 如果未启用，直接通过
+        if not self.enabled:
+            return await call_next(request)
+
+        # 允许访问 robots.txt（由专门的路由处理）
+        if request.url.path == "/robots.txt":
+            return await call_next(request)
+
+        # 允许访问静态资源（CSS、JS、图片等）
+        # 注意：.json 已移除，避免 API 路径绕过防护
+        # 静态资源只在特定前缀下放行（/static/、/assets/、/dist/）
+        static_extensions = {".css", ".js", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".ico", ".woff", ".woff2", ".ttf", ".eot"}
+        static_prefixes = {"/static/", "/assets/", "/dist/"}
+        
+        # 检查是否是静态资源路径（特定前缀下的静态文件）
+        path = request.url.path
+        is_static_path = any(path.startswith(prefix) for prefix in static_prefixes) and any(path.endswith(ext) for ext in static_extensions)
+        
+        # 也允许根路径下的静态文件（如 /favicon.ico）
+        is_root_static = path.count("/") == 1 and any(path.endswith(ext) for ext in static_extensions)
+        
+        if is_static_path or is_root_static:
+            return await call_next(request)
+
+        # 获取客户端IP（只获取一次，避免重复调用）
+        client_ip = self._get_client_ip(request)
+
+        # 检查IP白名单（优先检查，白名单IP直接通过）
+        if self._is_ip_allowed(client_ip):
+            return await call_next(request)
+
+        # 获取 User-Agent
+        user_agent = request.headers.get("User-Agent")
+
+        # 检测资产测绘工具（优先检测，因为更危险）
+        if self.check_asset_scanner:
+            is_scanner, scanner_name = self._detect_asset_scanner(request)
+            if is_scanner:
+                logger.warning(
+                    f"🚫 检测到资产测绘工具请求 - IP: {client_ip}, 工具: {scanner_name}, "
+                    f"User-Agent: {user_agent}, Path: {request.url.path}"
+                )
+                # 根据配置决定是否阻止
+                if self.block_on_detect:
+                    return PlainTextResponse(
+                        "Access Denied: Asset scanning tools are not allowed",
+                        status_code=403,
+                    )
+
+        # 检测爬虫 User-Agent
+        if self.check_user_agent and self._is_crawler_user_agent(user_agent):
+            logger.warning(
+                f"🚫 检测到爬虫请求 - IP: {client_ip}, User-Agent: {user_agent}, Path: {request.url.path}"
+            )
+            # 根据配置决定是否阻止
+            if self.block_on_detect:
+                return PlainTextResponse(
+                    "Access Denied: Crawlers are not allowed",
+                    status_code=403,
+                )
+
+        # 检查请求频率限制
+        if self.check_rate_limit and self._check_rate_limit(client_ip):
+            logger.warning(
+                f"🚫 请求频率过高 - IP: {client_ip}, User-Agent: {user_agent}, Path: {request.url.path}"
+            )
+            return PlainTextResponse(
+                "Too Many Requests: Rate limit exceeded",
+                status_code=429,
+            )
+
+        # 正常请求，继续处理
+        return await call_next(request)
+
+
+def create_robots_txt_response() -> PlainTextResponse:
+    """
+    创建 robots.txt 响应
+
+    Returns:
+        robots.txt 响应对象
+    """
+    robots_content = """User-agent: *
+Disallow: /
+
+# 禁止所有爬虫访问
+"""
+    return PlainTextResponse(
+        content=robots_content,
+        media_type="text/plain",
+        headers={"Cache-Control": "public, max-age=86400"},  # 缓存24小时
+    )
+

src/webui/webui_server.py

@@ -22,6 +22,9 @@ class WebUIServer:
         self.app = FastAPI(title="MaiBot WebUI")
         self._server = None
 
+        # 配置防爬虫中间件（需要在CORS之前注册）
+        self._setup_anti_crawler()
+
         # 配置 CORS（支持开发环境跨域请求）
         self._setup_cors()
 
@@ -31,6 +34,9 @@ class WebUIServer:
         # 重要：先注册 API 路由，再设置静态文件
         self._register_api_routes()
         self._setup_static_files()
+        
+        # 注册robots.txt路由
+        self._setup_robots_txt()
 
     def _setup_cors(self):
         """配置 CORS 中间件"""
@@ -92,21 +98,6 @@ class WebUIServer:
             logger.warning("💡 请确认前端已正确构建")
             return
 
-        # robots.txt - 禁止搜索引擎索引
-        @self.app.get("/robots.txt", include_in_schema=False)
-        async def robots_txt():
-            """返回 robots.txt 禁止所有爬虫"""
-            from fastapi.responses import PlainTextResponse
-            content = """User-agent: *
-Disallow: /
-
-# MaiBot Dashboard - 私有管理面板，禁止索引
-"""
-            return PlainTextResponse(
-                content=content,
-                headers={"X-Robots-Tag": "noindex, nofollow, noarchive"}
-            )
-
         # 处理 SPA 路由 - 注意：这个路由优先级最低
         @self.app.get("/{full_path:path}", include_in_schema=False)
         async def serve_spa(full_path: str):
@@ -135,6 +126,46 @@ Disallow: /
 
         logger.info(f"✅ WebUI 静态文件服务已配置: {static_path}")
 
+    def _setup_anti_crawler(self):
+        """配置防爬虫中间件"""
+        try:
+            from src.webui.anti_crawler import AntiCrawlerMiddleware
+            
+            # 从环境变量读取防爬虫模式（false/strict/loose/basic）
+            anti_crawler_mode = os.getenv("WEBUI_ANTI_CRAWLER_MODE", "basic").lower()
+            
+            # 注意：中间件按注册顺序反向执行，所以先注册的中间件后执行
+            # 我们需要在CORS之前注册，这样防爬虫检查会在CORS之前执行
+            self.app.add_middleware(
+                AntiCrawlerMiddleware,
+                mode=anti_crawler_mode
+            )
+            
+            mode_descriptions = {
+                "false": "已禁用",
+                "strict": "严格模式",
+                "loose": "宽松模式",
+                "basic": "基础模式"
+            }
+            mode_desc = mode_descriptions.get(anti_crawler_mode, "基础模式")
+            logger.info(f"🛡️ 防爬虫中间件已配置: {mode_desc}")
+        except Exception as e:
+            logger.error(f"❌ 配置防爬虫中间件失败: {e}", exc_info=True)
+
+    def _setup_robots_txt(self):
+        """设置robots.txt路由"""
+        try:
+            from src.webui.anti_crawler import create_robots_txt_response
+            
+            @self.app.get("/robots.txt", include_in_schema=False)
+            async def robots_txt():
+                """返回robots.txt，禁止所有爬虫"""
+                return create_robots_txt_response()
+            
+            logger.debug("✅ robots.txt 路由已注册")
+        except Exception as e:
+            logger.error(f"❌ 注册robots.txt路由失败: {e}", exc_info=True)
+
     def _register_api_routes(self):
         """注册所有 WebUI API 路由"""
         try:

template/template.env

@@ -6,4 +6,13 @@ PORT=8000
 WEBUI_ENABLED=true
 WEBUI_MODE=production   # 模式: development(开发) 或 production(生产)
 WEBUI_HOST=0.0.0.0      # WebUI 服务器监听地址
-WEBUI_PORT=8001         # WebUI 服务器端口
+WEBUI_PORT=8001         # WebUI 服务器端口
+
+# 防爬虫配置
+WEBUI_ANTI_CRAWLER_MODE=basic  # 防爬虫模式: false(禁用) / strict(严格) / loose(宽松) / basic(基础-只记录不阻止)
+WEBUI_ALLOWED_IPS=127.0.0.1     # IP白名单（逗号分隔，支持精确IP、CIDR格式和通配符）
+                                # 示例: 127.0.0.1,192.168.1.0/24,172.17.0.0/16
+WEBUI_TRUSTED_PROXIES=          # 信任的代理IP列表（逗号分隔），只有来自这些IP的X-Forwarded-For才被信任
+                                # 示例: 127.0.0.1,192.168.1.1,172.17.0.1
+WEBUI_TRUST_XFF=false           # 是否启用X-Forwarded-For代理解析（默认false）
+                                # 启用后，仍要求直连IP在TRUSTED_PROXIES中才会信任XFF头