如何用Python构建Web防火墙：代码实现与安全策略详解

一、Web防火墙的核心作用与Python实现优势

Web防火墙（WAF）是保护网站免受SQL注入、XSS攻击、CC攻击等常见威胁的关键防线。相较于传统硬件WAF，Python实现的轻量级WAF具有开发灵活、部署便捷、可定制性强的优势，尤其适合中小型网站或需要快速迭代的场景。

Python的Flask/Django框架可无缝集成WAF逻辑，结合正则表达式、请求头分析、行为模式识别等技术，能有效拦截恶意请求。例如，通过解析User-Agent、Referer等HTTP头信息，可识别自动化扫描工具或爬虫流量。

二、Python WAF的基础架构设计

1. 请求拦截层实现

核心代码示例（基于Flask）：

from flask import Flask, request, jsonify
import re
app = Flask(__name__)
# 定义恶意请求特征库
MALICIOUS_PATTERNS = [
    r'(?i)select\s+.*from',  # SQL注入特征
    r'<script.*?>.*?</script>',  # XSS特征
    r'\.\./'  # 目录遍历特征
]
def check_request(req):
    # 检查URL参数
    for param, value in req.args.items():
        if any(re.search(pattern, str(value)) for pattern in MALICIOUS_PATTERNS):
            return False
    # 检查请求体（JSON/表单）
    if req.is_json:
        data = req.get_json()
        for key, value in data.items():
            if any(re.search(pattern, str(value)) for pattern in MALICIOUS_PATTERNS):
                return False
    return True
@app.before_request
def waf_middleware():
    if not check_request(request):
        return jsonify({"error": "Access denied"}), 403

此代码通过正则匹配请求参数中的恶意模式，拦截包含SQL注入或XSS特征的请求。

2. IP黑名单与速率限制

from collections import defaultdict
import time
BLOCKED_IPS = set()
REQUEST_LOG = defaultdict(list)
def is_ip_blocked(ip):
    return ip in BLOCKED_IPS
def check_rate_limit(ip, limit=100, window=60):
    now = time.time()
    # 清理过期记录
    REQUEST_LOG[ip] = [t for t in REQUEST_LOG[ip] if now - t < window]
    if len(REQUEST_LOG[ip]) >= limit:
        BLOCKED_IPS.add(ip)
        return False
    REQUEST_LOG[ip].append(now)
    return True
@app.before_request
def rate_limit_middleware():
    ip = request.remote_addr
    if is_ip_blocked(ip) or not check_rate_limit(ip):
        return jsonify({"error": "Rate limit exceeded"}), 429

此模块实现IP级速率限制，防止CC攻击，并通过BLOCKED_IPS集合动态封禁异常IP。

三、进阶功能实现

1. 动态规则更新机制

通过配置文件或数据库存储规则，实现热更新：

import json
import os
RULES_FILE = "waf_rules.json"
def load_rules():
    if os.path.exists(RULES_FILE):
        with open(RULES_FILE, "r") as f:
            return json.load(f)
    return {"patterns": MALICIOUS_PATTERNS, "blocked_ips": list(BLOCKED_IPS)}
def save_rules(rules):
    with open(RULES_FILE, "w") as f:
        json.dump(rules, f)
# 示例：添加新规则
def add_rule(pattern):
    rules = load_rules()
    rules["patterns"].append(pattern)
    save_rules(rules)

管理员可通过API或管理界面动态添加攻击特征，无需重启服务。

2. 行为分析引擎

结合请求频率、路径分布等特征识别异常：

from collections import Counter
PATH_COUNTER = Counter()
def analyze_behavior(ip, path):
    PATH_COUNTER[(ip, path)] += 1
    # 检测异常路径访问（如短时间内大量访问管理后台）
    if PATH_COUNTER[(ip, path)] > 50:
        BLOCKED_IPS.add(ip)
        return False
    return True

此模块可识别针对特定路径的暴力攻击。

四、部署与优化建议

1. 性能优化：

   • 使用re.compile预编译正则表达式
   • 对高频规则采用布隆过滤器（Bloom Filter）加速匹配
   • 异步日志记录减少主线程阻塞

2. 误报控制：

   • 引入白名单机制，对信任IP放行
   • 实现人工复核接口，允许管理员解除误拦截

3. 日志与监控：

   import logging
   logging.basicConfig(filename='waf.log', level=logging.INFO)
   def log_attack(ip, request_data):
       logging.warning(f"Attack detected from {ip}: {request_data}")

   记录攻击事件供后续分析。

五、完整实现示例

结合上述模块的完整WAF中间件：

from flask import Flask, request, jsonify
import re
import time
from collections import defaultdict, Counter
app = Flask(__name__)
# 配置
MALICIOUS_PATTERNS = [
    r'(?i)select\s+.*from',
    r'<script.*?>.*?</script>'
]
BLOCKED_IPS = set()
REQUEST_LOG = defaultdict(list)
PATH_COUNTER = Counter()
def check_request(req):
    for param, value in req.args.items():
        if any(re.search(p, str(value)) for p in MALICIOUS_PATTERNS):
            return False
    if req.is_json:
        data = req.get_json()
        for k, v in data.items():
            if any(re.search(p, str(v)) for p in MALICIOUS_PATTERNS):
                return False
    return True
def check_rate_limit(ip, limit=100, window=60):
    now = time.time()
    REQUEST_LOG[ip] = [t for t in REQUEST_LOG[ip] if now - t < window]
    if len(REQUEST_LOG[ip]) >= limit:
        BLOCKED_IPS.add(ip)
        return False
    REQUEST_LOG[ip].append(now)
    return True
def analyze_behavior(ip, path):
    PATH_COUNTER[(ip, path)] += 1
    if PATH_COUNTER[(ip, path)] > 50:
        BLOCKED_IPS.add(ip)
        return False
    return True
@app.before_request
def waf_middleware():
    ip = request.remote_addr
    path = request.path
    if ip in BLOCKED_IPS:
        return jsonify({"error": "Blocked"}), 403
    if not check_rate_limit(ip):
        return jsonify({"error": "Rate limit"}), 429
    if not analyze_behavior(ip, path):
        return jsonify({"error": "Suspicious behavior"}), 403
    if not check_request(request):
        return jsonify({"error": "Malicious content"}), 403

六、总结与扩展方向

Python实现的WAF可通过以下方式增强：

1. 集成机器学习模型进行异常检测
2. 支持与Cloudflare等CDN的API联动
3. 实现规则测试框架，验证规则有效性

实际部署时，建议结合Nginx/Apache的模块化WAF进行分层防护，Python WAF作为应用层补充，形成纵深防御体系。