"""扫描器拦截中间件 针对 Hugging Face Spaces 公网部署场景：每天都会被自动化扫描器盯上， 日志里会出现大量 `.env`、`.git/config`、`/actuator/env`、`/proc/self/environ`、 `/wp-config.php` 等敏感文件探测请求。 策略： 1. 路径命中扫描特征（敏感文件/调试接口/框架特定路径）→ 立即拉黑该 IP 5 分钟 2. 已被拉黑的 IP 的所有请求 → 直接返回 403（不再走下游路由，节省资源） 设计要点： - 内存存储黑名单（HF Space 重启会清空，但短期拦截有效） - 仅按 IP 拉黑，不涉及 token/access_key（合法用户带 token 不受影响） - 白名单路径永远不拉黑：/、/health、/v1/*、/admin/*、/u/* 等业务路径 - 不拦截 OPTIONS 预检（CORS 需要） - 不拦截 HF Spaces 的健康检查（10.x 内网 IP） """ from __future__ import annotations import logging import time from typing import Awaitable, Callable from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request from starlette.responses import JSONResponse, Response logger = logging.getLogger(__name__) # 黑名单 TTL（秒）：被拉黑后 5 分钟内所有请求被拒绝 _BLACKLIST_TTL = 300 # 命中即拉黑的扫描特征路径前缀/子串 # 这些路径在合法业务中绝不会出现，命中即确定是扫描器 # NM3 修复： # - 删除与合法路径冲突的条目（/admin/config 可能误判 /admin/configs/*） # - .env 类改为以 /.env 开头匹配，避免误命中 /v1/xtc/environment 这类合法路径 # - 整体匹配仍用子串，但 dispatch 中先 check 白名单优先放行 _SCANNER_PATH_PATTERNS = [ # 敏感配置文件（必须以 / 开头，避免误命中合法路径片段） "/.env", "/.env.local", "/.env.production", "/.env.development", "/.env.backup", "/.env.bak", "/.env.old", "/.env.example", "/.env.swp", "config.php.bak", "wp-config.php", "wp-config.php.bak", "settings.php.bak", "configuration.php", ".streamlit/secrets.toml", ".streamlit/config.toml", "_stcore/host-config", "_stcore/health", # Git/SVN 泄露 ".git/config", ".git/HEAD", ".git/index", ".svn/entries", ".hg/store", # 路径遍历 "../proc/self/environ", "..%252f", "..%2f", "/proc/self/environ", "/proc/self/cmdline", "/proc/self/status", # 框架调试/监控接口 "/actuator/env", "/actuator/configprops", "/actuator/health", "/actuator/mappings", "/actuator/beans", "/__debug__", "/_debug", "/debug/vars", "/debug/pprof", "/metrics", "/prometheus", "/api-docs", "/swagger-ui.html", "/swagger.json", "/redoc", "/swagger", # GraphQL（除非业务用，否则视为扫描） "/graphql", "/api/graphql", # PHP/Java/ASP 调试接口 "/phpinfo.php", "/__phpinfo", "/elmah.axd", "/trace.axd", "/telescope", "/horizon", "/_profiler", # Java/Mongo/Redis 等端口探测 "/server-status", "/server-info", "/internal/config", "/internal/debug", # 各种配置探测（注意：避免与 /admin/api/config 等合法路径冲突） "/config.json", "/config.yaml", "/config.yml", "/config.py", "/configuration", "/settings.json", "/api/keys", "/api/v1/keys", "/api/config", "/api/v1/config", "/api/settings", "/api/env", "/api/v1/env", "/api/credentials", "/api/secrets", "/api/v1/models", # 扫描器常用的 OpenAI 兼容接口探测（我们的在 /v1/models） # 其他常见扫描路径 "/backup/.env", "/backup/config.json", "/.aws/credentials", "/.ssh/id_rsa", # 注：原 "/admin/config" 已移除——会与 /admin/configs/* 等合法路径冲突。 # /admin/api/* 走 _require_admin 鉴权，无需 scanner_block 拦截。 ] # 白名单路径前缀：即使路径命中扫描特征也永远放行 # 这些是合法业务路径，扫描器偶尔也会请求但绝不能误判 # NM3：dispatch 中必须先 check 白名单，否则死代码会导致误拦合法路径 _WHITELIST_PREFIXES = ( "/", # 根路径 "/health", # 健康检查 "/v1/", # OpenAI 兼容 API "/admin", # 后台（带鉴权） "/u/", # 用户文件 API "/uapi/", # 备用 "/favicon.ico", "/robots.txt", ) # 内网 IP 前缀：HF Spaces 的健康检查来自这些 IP，永远不拉黑 _INTERNAL_IP_PREFIXES = ("10.", "172.", "192.168.", "127.") # 内存黑名单：{ip: expire_ts} _ip_blacklist: dict[str, float] = {} def _get_client_ip(request: Request) -> str: """提取真实客户端 IP（尊重 X-Forwarded-For / X-Real-IP）。 NL1：行为与 rate_limit_store.get_client_ip 保持一致， 避免不同中间件记录的 IP 不一致导致日志关联困难。 保留独立实现避免 scanner_block_middleware 反向依赖 services 包。 """ xff = request.headers.get("x-forwarded-for") if xff: # 取第一个（最外层客户端） return xff.split(",")[0].strip() x_real_ip = request.headers.get("x-real-ip") if x_real_ip: return x_real_ip.strip() if request.client: return request.client.host return "" def _is_internal_ip(ip: str) -> bool: if not ip: return False return any(ip.startswith(p) for p in _INTERNAL_IP_PREFIXES) def _is_scanner_path(path: str) -> bool: """路径是否命中扫描特征""" if not path: return False # 路径小写化便于匹配 lower = path.lower() for pat in _SCANNER_PATH_PATTERNS: if pat in lower: return True return False def _is_whitelisted_path(path: str) -> bool: if not path: return True # 精确匹配 / 和白名单前缀 if path == "/": return True for p in _WHITELIST_PREFIXES: if path.startswith(p): return True return False def _blacklist_ip(ip: str, reason: str = "") -> None: if not ip or _is_internal_ip(ip): return expire = time.time() + _BLACKLIST_TTL _ip_blacklist[ip] = expire logger.warning("[scanner_block] IP %s blacklisted for %ds (%s)", ip, _BLACKLIST_TTL, reason) def _is_blacklisted(ip: str) -> bool: if not ip: return False expire = _ip_blacklist.get(ip) if expire is None: return False if time.time() > expire: # 过期，自动移除 _ip_blacklist.pop(ip, None) return False return True def _cleanup_blacklist() -> None: """清理过期黑名单条目（每次检查时调用，避免内存泄漏）""" now = time.time() expired = [ip for ip, exp in _ip_blacklist.items() if now > exp] for ip in expired: _ip_blacklist.pop(ip, None) class ScannerBlockMiddleware(BaseHTTPMiddleware): """扫描器拦截中间件 部署在最外层（CORS 之前），优先拦截明显扫描行为。 """ async def dispatch( self, request: Request, call_next: Callable[[Request], Awaitable[Response]], ) -> Response: # OPTIONS 预检直接放行（CORS 需要） if request.method == "OPTIONS": return await call_next(request) client_ip = _get_client_ip(request) path = request.url.path # 内网 IP 永远不拦（HF Spaces 健康检查来自 10.x） if _is_internal_ip(client_ip): return await call_next(request) # NM3：白名单路径永远放行，且必须在扫描特征检测之前 # 原实现 _is_whitelisted_path 是死代码从未被调用，导致 /admin/api/config # 等合法路径若被加入 _SCANNER_PATH_PATTERNS 会误拦 IP 5 分钟 if _is_whitelisted_path(path): return await call_next(request) # 1. 已被拉黑的 IP：所有请求直接 403 if _is_blacklisted(client_ip): _cleanup_blacklist() return JSONResponse( status_code=403, content={ "ok": False, "error": { "code": "forbidden", "message": "Access denied", "status": 403, } }, ) # 2. 路径命中扫描特征 → 立即拉黑该 IP if _is_scanner_path(path): _blacklist_ip(client_ip, reason=f"path={path}") logger.warning( "[scanner_block] blocked scan attempt ip=%s path=%s method=%s", client_ip, path[:200], request.method, ) return JSONResponse( status_code=403, content={ "ok": False, "error": { "code": "forbidden", "message": "Access denied", "status": 403, } }, ) # 3. 正常请求继续 return await call_next(request)