app/scanner_block_middleware.py

"""扫描器拦截中间件

针对 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)