Python Web防火墙核心架构设计

Web防火墙的核心功能在于对HTTP请求进行实时检测与拦截，其架构可分为三层：请求解析层、规则匹配层和响应处理层。Python因其丰富的网络库（如Flask/Django中间件）和字符串处理能力，成为构建轻量级WAF的理想选择。

1. 基础请求拦截框架

使用Flask框架的before_request钩子可实现全局请求拦截：

from flask import Flask, request, abort
app = Flask(__name__)
@app.before_request
def firewall():
    # 获取请求基础信息
    method = request.method
    path = request.path
    headers = dict(request.headers)
    # 基础规则检查（示例）
    if path == "/admin" and "X-Forwarded-For" not in headers:
        abort(403, "Direct admin access blocked")

此代码片段展示了如何通过中间件模式拦截所有请求，并基于路径和头部信息进行初步过滤。实际项目中需扩展为更复杂的规则引擎。

核心防护模块实现

2.1 SQL注入防护

通过正则表达式匹配常见SQL关键字和特殊字符：

import re
SQL_PATTERN = re.compile(
    r"(?i)(?:'|\"|;|\\x27|\\x22|\\x3b|\\x5c)"  # 引号与分号
    r"|(?:union|select|insert|update|delete|drop|truncate|exec|alter)"  # SQL关键字
    r"|(?:\s+or\s+1\s*=\s*1)"  # 逻辑绕过
)
def check_sql_injection(params):
    for key, value in params.items():
        if SQL_PATTERN.search(str(value)):
            return True
    return False

该模块可集成到参数解析流程中，对GET/POST/COOKIE参数进行全面扫描。建议结合白名单机制，对已知安全参数放行。

2.2 XSS跨站脚本防护

采用双重检测策略：黑名单过滤+CSP头控制：

XSS_PATTERN = re.compile(
    r"<script.*?>.*?</script>"  # 脚本标签
    r"|on\w+\s*=\s*['\"]"  # 事件处理器
    r"|javascript\s*:"  # javascript伪协议
)
def sanitize_input(input_str):
    # 黑名单过滤
    if XSS_PATTERN.search(input_str):
        return ""
    # 转义特殊字符
    return input_str.replace("<", "&lt;").replace(">", "&gt;")

同时应在响应头中添加Content-Security-Policy，限制资源加载来源：

@app.after_request
def add_security_headers(response):
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    return response

2.3 IP黑名单与速率限制

使用Redis实现分布式速率限制：

import redis
from time import time
r = redis.Redis(host='localhost', port=6379, db=0)
def check_rate_limit(ip, limit=100, window=60):
    key = f"rate_limit:{ip}"
    current = r.get(key)
    if current and int(current) > limit:
        return False
    # 滑动窗口计数
    r.multi()
    r.incr(key)
    r.expire(key, window)
    r.execute()
    return True

此实现允许每个IP每分钟最多100次请求，超限则返回429状态码。生产环境需结合Nginx等反向代理实现更高效的限流。

高级防护技术

3.1 行为分析模块

通过统计请求特征识别异常：

from collections import defaultdict
REQUEST_PROFILES = defaultdict(lambda: {
    "count": 0,
    "paths": set(),
    "ua_changes": 0
})
def analyze_request(ip, path, user_agent):
    profile = REQUEST_PROFILES[ip]
    profile["count"] += 1
    profile["paths"].add(path)
    # 检测User-Agent频繁变更
    if "user_agent" in profile and profile["user_agent"] != user_agent:
        profile["ua_changes"] += 1
    profile["user_agent"] = user_agent
    # 简单规则：单IP访问过多不同路径可能为扫描
    if len(profile["paths"]) > 50 and profile["count"] < 100:
        return False
    return True

该模块可识别自动化扫描工具的特征行为，需配合持久化存储和定期清理机制。

3.2 动态规则加载

支持从外部文件加载规则，实现零停机更新：

import json
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class RuleLoader(FileSystemEventHandler):
    def __init__(self, waf):
        self.waf = waf
    def on_modified(self, event):
        if event.src_path.endswith("rules.json"):
            with open("rules.json") as f:
                self.waf.update_rules(json.load(f))
# WAF类中添加规则更新方法
class WebFirewall:
    def __init__(self):
        self.rules = []
        self.observer = Observer()
        self.observer.schedule(RuleLoader(self), ".", recursive=False)
        self.observer.start()
    def update_rules(self, new_rules):
        self.rules = new_rules
        # 触发日志记录或通知机制

此实现使用Python的watchdog库监控规则文件变更，适合需要频繁调整防护策略的场景。

部署与优化建议

1. 性能优化：对高频规则进行预编译，使用re.compile()缓存正则对象
2. 日志分析：集成ELK栈记录拦截事件，示例日志格式：

   {
   "timestamp": "2023-07-20T12:00:00Z",
   "client_ip": "192.168.1.100",
   "rule_id": "SQL_INJECTION_001",
   "request_path": "/login",
   "severity": "high"
   }

3. 测试策略：
   • 使用pytest编写单元测试验证规则准确性
   • 通过locust进行压力测试评估性能影响
   • 部署前在测试环境运行OWASP ZAP扫描

完整实现示例

from flask import Flask, request, abort
import re
import json
class WebFirewall:
    def __init__(self):
        self.rules = {
            "sql_injection": re.compile(
                r"(?i)(?:'|\"|;|\\x27|\\x22|\\x3b|\\x5c)"
                r"|(?:union|select|insert|update|delete|drop|truncate|exec|alter)"
            ),
            "xss": re.compile(
                r"<script.*?>.*?</script>"
                r"|on\w+\s*=\s*['\"]"
                r"|javascript\s*:"
            )
        }
    def check_request(self, req):
        # 参数检查
        params = {**req.args.to_dict(), **req.form.to_dict()}
        for key, value in params.items():
            for rule_name, pattern in self.rules.items():
                if pattern.search(str(value)):
                    return f"Potential {rule_name} attack detected"
        # 头部检查
        if "X-Forwarded-For" not in req.headers and req.path.startswith("/admin"):
            return "Direct admin access blocked"
        return None
app = Flask(__name__)
waf = WebFirewall()
@app.before_request
def protect():
    error = waf.check_request(request)
    if error:
        abort(403, description=error)
if __name__ == "__main__":
    app.run(ssl_context='adhoc')  # 生产环境应使用正式证书

总结与扩展方向

本文实现的Python Web防火墙已具备基础防护能力，实际生产环境还需考虑：

1. 集成WAF管理面板，实现规则可视化配置
2. 添加机器学习模块，自动识别新型攻击模式
3. 支持与云服务商API对接，实现自动封禁
4. 开发规则贡献社区，持续完善防护库

开发者可根据具体需求选择实现深度，小型项目可采用本文的轻量级方案，大型系统建议结合专业WAF设备形成纵深防御。