一、Web防火墙的核心功能与实现逻辑

Web防火墙（WAF）的核心价值在于构建请求过滤层，通过预设规则拦截恶意请求。Python实现WAF需重点解决三个技术问题：请求解析、规则匹配、响应处理。

1.1 请求解析层设计

基于WSGI协议的中间件模式是Python WAF的典型架构。以Flask框架为例，可通过before_request钩子实现前置拦截：

from flask import Flask, request, jsonify
app = Flask(__name__)
@app.before_request
def waf_middleware():
    # 获取请求元数据
    method = request.method
    path = request.path
    headers = dict(request.headers)
    args = request.args.to_dict()
    form_data = request.form.to_dict() if request.method in ['POST', 'PUT'] else {}
    # 调用规则引擎
    if not rule_engine(method, path, headers, args, form_data):
        return jsonify({"error": "Access Denied"}), 403

1.2 规则引擎实现策略

规则引擎需支持动态规则加载和高效匹配。推荐采用”白名单+黑名单”混合模式：

class RuleEngine:
    def __init__(self):
        self.rules = {
            'ip_blacklist': set(),  # 黑名单IP
            'path_whitelist': {'/login', '/static/'},  # 白名单路径
            'sql_patterns': ['select.*from', 'union.*select'],  # SQL注入特征
            'xss_patterns': ['<script>', 'javascript:']  # XSS特征
        }
    def check_ip(self, client_ip):
        return client_ip in self.rules['ip_blacklist']
    def check_path(self, path):
        return path not in self.rules['path_whitelist']
    def check_params(self, params):
        for key, value in params.items():
            if isinstance(value, str):
                for pattern in self.rules['sql_patterns'] + self.rules['xss_patterns']:
                    if re.search(pattern, value, re.IGNORECASE):
                        return False
        return True

二、关键防护模块实现

2.1 SQL注入防护

采用正则表达式+参数白名单双重验证机制：

import re
def sql_injection_check(params):
    sql_patterns = [
        r'(\%27)|(\')|(\-\-)',  # 单引号及注释
        r'((\%3D)|(=))[^\n]*((\%27)|(\')|(\-\-)|(\%3B)|(;))',  # 等号注入
        r'\w*((\%27)|(\'))((\%6F)|o|(\%4F))((\%72)|r|(\%52))',  # or注入
        r'exec(\s|\+)+(s|x)p\w+'  # 存储过程注入
    ]
    for param_key, param_value in params.items():
        if not isinstance(param_value, str):
            continue
        cleaned_value = param_value.lower()
        for pattern in sql_patterns:
            if re.search(pattern, cleaned_value):
                return False
    return True

2.2 XSS跨站脚本防护

实现三层过滤机制：

def xss_protection(input_data):
    # 第一层：HTML实体编码
    def html_encode(text):
        return text.replace('&', '&amp;').replace('<', '&lt;').replace('>', '&gt;')
    # 第二层：属性值过滤
    def sanitize_attr(attr_value):
        if not attr_value:
            return ''
        # 移除javascript:等危险协议
        cleaned = re.sub(r'^javascript:', '', attr_value, flags=re.IGNORECASE)
        return cleaned.replace('"', '&quot;').replace("'", '&#39;')
    # 第三层：事件处理器过滤
    def remove_events(html):
        event_attrs = ['onerror', 'onload', 'onclick', 'onmouseover']
        for event in event_attrs:
            html = re.sub(r'\s' + event + r'\s*=\s*["\'][^"\']*["\']', '', html, flags=re.IGNORECASE)
        return html
    if isinstance(input_data, dict):
        return {k: xss_protection(v) for k, v in input_data.items()}
    elif isinstance(input_data, str):
        return remove_events(html_encode(input_data))
    else:
        return input_data

2.3 CC攻击防护

实现基于令牌桶算法的速率限制：

import time
from collections import defaultdict
class RateLimiter:
    def __init__(self, capacity=100, refill_rate=10):
        self.capacity = capacity  # 桶容量
        self.refill_rate = refill_rate  # 每秒补充的令牌数
        self.tokens = capacity
        self.last_refill = time.time()
        self.client_buckets = defaultdict(int)  # 客户端令牌桶
    def _refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now
    def allow_request(self, client_ip, required_tokens=1):
        self._refill()
        # 客户端独立令牌桶
        if self.client_buckets[client_ip] < 0:
            self.client_buckets[client_ip] += required_tokens
            return False
        if self.tokens >= required_tokens:
            self.tokens -= required_tokens
            return True
        else:
            self.client_buckets[client_ip] -= required_tokens
            return False

三、性能优化与部署方案

3.1 异步处理架构

采用asyncio实现非阻塞处理：

import asyncio
from aiohttp import web
async def waf_middleware(request):
    # 异步规则检查
    ip = request.remote
    if await check_ip_blacklist(ip):
        return web.Response(status=403, text="Forbidden")
    # 异步参数检查
    params = {**request.query, **await request.post()}
    if not await async_sql_check(params):
        return web.Response(status=403, text="SQL Injection Detected")
    return await request.app.router.resolve().handle(request)
async def check_ip_blacklist(ip):
    # 模拟异步数据库查询
    await asyncio.sleep(0.01)
    return ip in BLACKLIST

3.2 规则热更新机制

实现基于Redis的规则动态加载：

import redis
import json
class DynamicRuleEngine:
    def __init__(self):
        self.redis = redis.StrictRedis(host='localhost', port=6379, db=0)
        self.local_cache = {}
    def load_rules(self):
        # 从Redis加载规则
        raw_rules = self.redis.get('waf_rules')
        if raw_rules:
            self.local_cache = json.loads(raw_rules)
    def update_rule(self, rule_type, rule_data):
        # 更新Redis中的规则
        self.local_cache[rule_type] = rule_data
        self.redis.set('waf_rules', json.dumps(self.local_cache))
    def get_rule(self, rule_type):
        return self.local_cache.get(rule_type, [])

四、完整实现示例

from flask import Flask, request, jsonify
import re
import time
from collections import defaultdict
app = Flask(__name__)
class WebApplicationFirewall:
    def __init__(self):
        self.ip_blacklist = set()
        self.path_whitelist = {'/login', '/static/'}
        self.sql_patterns = [
            r'(\%27)|(\')|(\-\-)',
            r'((\%3D)|(=))[^\n]*((\%27)|(\')|(\-\-)|(\%3B)|(;))'
        ]
        self.rate_limiters = defaultdict(lambda: RateLimiter(100, 10))
    def check_request(self, request):
        # IP检查
        client_ip = request.remote_addr
        if client_ip in self.ip_blacklist:
            return False, "IP Blacklisted"
        # 路径检查
        path = request.path
        if path not in self.path_whitelist:
            # 参数检查
            params = {**request.args.to_dict(), **request.form.to_dict()}
            if not self._check_params(params):
                return False, "Malicious Parameters Detected"
        # 速率限制
        limiter = self.rate_limiters[client_ip]
        if not limiter.allow_request(client_ip):
            return False, "Rate Limit Exceeded"
        return True, "Allowed"
    def _check_params(self, params):
        for key, value in params.items():
            if isinstance(value, str):
                for pattern in self.sql_patterns:
                    if re.search(pattern, value, re.IGNORECASE):
                        return False
        return True
waf = WebApplicationFirewall()
@app.before_request
def apply_waf():
    allowed, message = waf.check_request(request)
    if not allowed:
        return jsonify({"error": message}), 403
@app.route('/')
def index():
    return "Welcome to Secure Application"
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8000)

五、部署与维护建议

1. 规则更新策略：建议每周更新一次规则库，重大漏洞爆发时立即更新
2. 监控指标：重点关注拦截率（正常请求/恶意请求比例）、误报率、响应延迟
3. 日志分析：实现结构化日志存储，推荐使用ELK（Elasticsearch+Logstash+Kibana）方案
4. 性能基准：在1000RPS压力下，WAF处理延迟应控制在5ms以内

通过上述技术实现，开发者可以构建出满足企业级安全需求的Web防火墙系统。实际部署时建议结合云服务的安全组策略，形成多层次防护体系。