怎样实现恶意请求过滤脚本

wen 实用脚本 2026-06-26 32

从零构建Web安全防线

目录导读

为什么需要恶意请求过滤？
恶意请求的常见类型与特征
过滤脚本的核心架构设计
实战：基于Python的过滤脚本实现
集成到Web服务器的三种方式
性能优化与误报率控制
常见问题FAQ

为什么需要恶意请求过滤？

Q：所有网站都会遭受恶意请求攻击吗？
A：是的，根据Cloudflare 2023年的安全报告，即使是日均访问量不足100的小型网站，平均每小时也会收到3-5次自动化扫描或注入尝试,恶意请求过滤脚本是防御的第一道关卡。

怎样实现恶意请求过滤脚本

传统的Web应用防火墙（WAF）虽然功能强大，但存在成本高、配置复杂、容易产生误报等缺点，而一个轻量级的恶意请求过滤脚本，可以针对性地解决90%以上的常见攻击,且完全自主可控。

恶意请求的常见类型与特征

攻击类型	典型特征	危害级别
SQL注入	包含 `' OR 1=1--`、`UNION SELECT`	⚠️ 高
XSS跨站	包含 `<script>`、`onerror=`、`javascript:`	⚠️ 高
路径遍历	包含、、`/etc/passwd`	⚠️ 中
命令注入	包含 `; rm -rf`、`\| whoami`	⚠️ 极高
爬虫/扫描	User-Agent为空或含 `curl`、`python-requests`	⚠️ 低中
参数污染	同一参数出现多次，或参数值异常长	⚠️ 中

Q：为什么不能用正则表达式一次性过滤所有攻击？
A：因为攻击模式在持续进化，例如简单的SQL注入过滤 ' OR，攻击者可以编码为 %27%20OR%201%3D1 或使用注释绕过,过滤脚本需要多层检测机制。

过滤脚本的核心架构设计

一个高效的过滤脚本应该包含四个模块：

[输入层] → [预处理层] → [检测引擎] → [响应层]
   ↓           ↓            ↓           ↓
原始请求    解码/解析   规则匹配  阻断/记录/放行

关键设计原则：

多层解码：URL编码、Unicode编码、Base64、HTML实体编码需要逐层还原
上下文感知：检测<script>标签时，要区分HTML中的正常JavaScript和XSS攻击
频率阈值：对同一IP的请求速率进行监控，超过阈值自动封禁
白名单优先：先放行已知安全路径（如 /.well-known/），再过滤可疑请求

实战：基于Python的过滤脚本实现

以下是一个可直接部署的轻量级过滤脚本框架：

import re
import html
from urllib.parse import unquote, parse_qs
from datetime import datetime, timedelta
from collections import defaultdict
class RequestFilter:
    def __init__(self):
        # 可疑特征库（建议从OWASP Core Rules同步更新）
        self.sql_patterns = [
            r"(\bSELECT\b.*\bFROM\b)",
            r"(\bUNION\b.*\bSELECT\b)",
            r"('|%27)\s*(OR|AND)\s*('|%27|\d+|=)",
        ]
        self.xss_patterns = [
            r"(<script[\s>])",
            r"(on\w+\s*=)",
            r"(javascript\s*:)"
        ]
        # 请求速率限制
        self.visitor_log = defaultdict(list)
        self.rate_limit = 100  # 每分钟最大请求数
    def decode_payload(self, raw_value):
        """逐层解码 payload"""
        value = raw_value
        # 第一层：URL解码
        if '%' in value:
            value = unquote(value)
        # 第二层：HTML实体解码
        if '&' in value:
            value = html.unescape(value)
        # 第三层：Base64解码尝试（如果看起来像base64）
        if re.match(r'^[A-Za-z0-9+/=]{20,}$', value):
            try:
                import base64
                decoded = base64.b64decode(value).decode('utf-8', errors='ignore')
                if re.search(r'(<script|<iframe|union|select)', decoded, re.I):
                    return decoded
            except:
                pass
        return value
    def check_ip_rate(self, ip):
        """IP速率检测"""
        now = datetime.now()
        self.visitor_log[ip] = [t for t in self.visitor_log[ip] if now - t < timedelta(minutes=1)]
        self.visitor_log[ip].append(now)
        return len(self.visitor_log[ip]) > self.rate_limit
    def analyze(self, request_data):
        """主检测方法"""
        # 获取客户端IP
        ip = request_data.get('remote_addr', '0.0.0.0')
        # IP速率检测
        if self.check_ip_rate(ip):
            return {'block': True, 'reason': 'rate_limit', 'ip': ip}
        # 提取所有输入参数
        inputs = []
        # GET参数
        for key, values in request_data.get('get', {}).items():
            for v in values:
                inputs.append(unquote(v))
        # POST参数（假设是字符串）
        body = request_data.get('body', '')
        if body:
            inputs.append(body)
        # Cookie
        for key, value in request_data.get('cookies', {}).items():
            inputs.append(value)
        # 综合检测
        block_reason = None
        for inp in inputs:
            decoded = self.decode_payload(inp)
            # SQL注入检测
            for pattern in self.sql_patterns:
                if re.search(pattern, decoded, re.IGNORECASE):
                    block_reason = 'sql_injection'
                    break
            # XSS检测
            if not block_reason:
                for pattern in self.xss_patterns:
                    if re.search(pattern, decoded, re.IGNORECASE):
                        block_reason = 'xss_attack'
                        break
            if block_reason:
                break
        if block_reason:
            return {'block': True, 'reason': block_reason, 'ip': ip, 'matched': inp}
        else:
            return {'block': False, 'ip': ip}

Q：为什么这个脚本看起来很简单，它能防御真实攻击吗？
A：这是一个基础框架，生产环境需要根据实际攻击日志不断扩充规则库，我们可以从ModSecurity的CRS（核心规则集）中提取规则,再转化为Python正则表达式。

集成到Web服务器的三种方式

方式1：反向代理模式（推荐）

使用Nginx作为前置代理，通过 lua-resty-waf 或调用外部Python脚本：

location / {
    access_by_lua_file /etc/nginx/filter.lua;
    proxy_pass http://backend;
}

方式2：WSGI中间件（适合Django/Flask）

# 在Django的settings.py中
MIDDLEWARE = [
    'myapp.middleware.MaliciousRequestMiddleware',
    # ...其他中间件
]

方式3：独立守护进程

通过iptables将80端口流量转发到过滤进程,或使用TCP代理工具如socat：

socat TCP-LISTEN:8080,fork,reuseaddr EXEC:"./filter_script.py"

Q：哪种集成方式对性能影响最小？
A：反向代理模式性能最好，用C语言编写的Nginx处理过滤逻辑比Python快10-20倍，如果对性能要求极高,可以考虑使用OpenResty结合Lua实现。

性能优化与误报率控制

性能优化技巧

预编译正则：使用 re.compile() 缓存所有模式
分层过滤：先做轻量级检查（如参数长度>5000字符直接拦截），再做深度检测
LRU缓存：对已经检测过的请求参数进行哈希缓存，避免重复计算
使用C扩展：将正则引擎替换为 regex 库或 hyperscan（Intel开发的高性能正则匹配库）

误报率控制策略

白名单机制：对静态资源（.js、.css、.png）和API路径（/api/v1）放行
评分系统：不是非黑即白，而是给每个请求打分（如可疑度0-100），超过阈值才阻断
历史记录：对阻断的请求进行人工审核，将误报案例加入例外库
动态调整：根据每日误报率自动调整检测阈值

Q：误报会导致正常用户无法访问怎么办？
A：建议采用“记录+质询”模式：第一次发现可疑请求时只记录日志，不阻断；当同一IP累计达到3次可疑行为时，弹出CAPTCHA验证,验证通过则加入临时白名单。

常见问题FAQ

Q1：我的网站已经用了云WAF，还需要本地过滤脚本吗？

A：需要，云WAF主要防御大流量攻击，且无法保护内部网络接口，本地脚本可作为第二层防御，特别是针对0Day攻击（云WAF规则更新前）。

Q2：过滤脚本如何防绕过？

A：采用“护城河”模式：

第一层：字符黑名单（如 <>%）
第二层：语义分析（如检测字符串中是否包含完整的SQL语句结构）
第三层：行为分析（连续快速请求、异常User-Agent等）

Q3：脚本部署后如何验证效果？

A：使用开源工具进行渗透测试：

SQLMap：测试SQL注入防御
XSStrike：测试XSS防御
Nikto：全面Web扫描
自定义脚本：模拟常见攻击payload

Q4：免费的开源过滤脚本推荐？

A：可以参考以下项目（均已修改为域名占位符）：

ModSecurity CRS（规则库）：原规则集可在 coreruleset.org 获取
PyWAF（Python实现）：搜索 pywaf 项目
VeryNginx（Lua实现）：基于OpenResty的轻量级WAF

实现恶意请求过滤脚本并非一次性的工作，而是一个持续演进的过程,建议遵循以下路线图：

第一周：部署基础过滤脚本，开启“仅记录”模式
第二周：分析误报和漏报，优化规则库
第三周：启用阻断模式，同时开启CAPTCHA备用方案
第四周：接入日志分析工具（如ELK），建立自动更新机制

没有任何脚本能100%防御所有攻击，但建立一个可靠的过滤系统，足以让你的网站远离99%的常见威胁，安全是一个旅程,而不是终点。

附录：推荐定期访问以下资源获取最新攻击模式：

OWASP Top 10（每年更新）
各大安全社区（如 securityxploded.com）
GitHub上的WAF规则库（搜索 waf rules）