W

本文仅供合法授权的安全研究与学习使用，请勿用于非法用途。读者应确保行为符合当地法律法规。

前言

Web应用防火墙（Web Application Firewall，简称WAF）是Web安全防御体系的核心组件。它位于Web应用与互联网之间，通过检查HTTP请求的内容来识别和拦截恶意流量，是抵御SQL注入、XSS跨站脚本攻击等OWASP Top 10威胁的第一道防线。

市面上有ModSecurity、阿里云WAF、Cloudflare WAF等成熟产品，但理解WAF的核心工作原理——规则匹配、评分机制、标准化处理——对于安全工程师来说至关重要。无论是自定义WAF规则、调优误报/漏报，还是应对绕过攻击，都需要对WAF底层逻辑有深刻理解。

本文基于一个真实的FastAPI中间件WAF源码，讲解如何用不到150行Python代码实现一个具备规则引擎、评分机制、Fail-Open安全策略的轻量级WAF。

技术背景

WAF的核心检测机制

现代WAF主要采用两种检测方式：

黑名单/规则匹配：预定义已知攻击特征的正则表达式，请求匹配到任何规则即判定为恶意。代表产品：ModSecurity（OWASP CRS规则集）
评分制：为每条规则分配权重（分数），请求累计得分超过阈值才拦截。相比单一规则命中即拦截，评分制能降低误报率，同时能识别组合攻击

本文的WAF采用评分制+黑名单规则的混合模式。

Payload标准化

攻击者常通过编码和变形来绕过WAF检测，例如：

URL编码：UNION%20SELECT 绕过对 UNION SELECT 的直接匹配
大小写混淆：UnIoN sElEcT 绕过大小写敏感的规则
双重编码：%2555NION 绕过单次URL解码

因此WAF在匹配规则前必须对请求进行标准化处理：URL解码、统一小写、拼接各部分内容。本文源码实现了基础的标准化逻辑。

Fail-Open vs Fail-Closed

WAF引擎本身也可能发生异常（如正则编译错误、内存不足）。此时如何决策体现了安全策略：

Fail-Open（故障放行）：WAF异常时放行请求，保证业务可用性，但可能放行攻击
Fail-Closed（故障关闭）：WAF异常时拒绝请求，保证安全性，但可能阻断正常业务

生产环境通常选择Fail-Open，避免WAF故障导致全站不可用。

实现思路

本WAF以FastAPI中间件形式实现，整体架构如下：

HTTP请求 → WAF中间件 → 标准化Payload → 规则引擎匹配 → 评分计算
                                                          ↓
                                              score >= 阈值？
                                           ┌──── 是 ────┐──── 否 ────┐
                                     Fail-Open?      放行到后端
                                  ┌── 是 ──┐── 否 ──┐
                              记录后放行    返回403拦截

关键设计：

规则定义：每条规则包含ID、正则表达式、权重三要素
预编译正则：启动时预编译所有正则，避免运行时重复编译
标准化处理：URL解码 + 小写化 + 拼接路径/查询参数/UA/Body
评分计算：遍历所有规则，命中则累加权重
决策逻辑：总分超过阈值时，根据Fail-Open策略决定是放行还是拦截
日志记录：无论是否拦截，都记录请求信息和匹配结果，用于后续分析和调优

核心代码解析

规则定义与预编译

WAF的检测能力直接由规则集决定。以下代码定义了针对SQL注入和XSS的检测规则：

import re, os, time, json, logging
from urllib.parse import unquote_plus

LOG_PATH = os.getenv("WAF_LOG", "./waf_samples.log")
SCORE_THRESHOLD = int(os.getenv("WAF_BLOCK_SCORE", "100"))  # 默认100：只监控不拦截
FAIL_OPEN = os.getenv("WAF_FAIL_OPEN", "1") == "1"

# 规则：(id, regex_str, weight)
RULES = [
    ("sqli_union", r"(?i)\bunion\s+select\b", 50),            # UNION SELECT注入
    ("sqli_information_schema", r"(?i)information_schema", 40), # 信息架构探测
    ("xss_tag", r"(?i)<\s*script\b", 30),                      # <script>标签
    ("xss_event", r"(?i)onerror\s*=|javascript:", 20),        # XSS事件处理器
]

# 预编译正则表达式，提升运行时匹配效率
COMPILED = [(rid, re.compile(pat)) for rid, pat, w in [(r[0], r[1], r[2]) for r in RULES]]
WEIGHTS = {r[0]: r[2] for r in RULES}

每条规则的设计思路：

sqli_union（权重50）：UNION SELECT 是SQL注入中最经典的注入手法，用于从其他表中提取数据。(?i) 使匹配不区分大小写，\b 确保匹配单词边界避免误报
sqli_information_schema（权重40）：information_schema 是MySQL的元数据库，攻击者通过查询它来获取表名和列名，是SQL注入后续利用的标志性特征
xss_tag（权重30）：<script> 标签是XSS攻击的核心载荷，<\s*script\b 可以匹配 <script> 和 < script> 等变体
xss_event（权重20）：onerror= 和 javascript: 是两种常见的XSS向量，通过HTML事件属性或伪协议执行脚本

注意默认阈值设为100，而四条规则的总权重为50+40+30+20=140。这意味着单一规则命中不会触发拦截，需要组合命中（如同时触发UNION SELECT和information_schema，得分为90仍未超阈值；再命中任意XSS规则，总分超过100才会拦截）。这种设计有效降低了误报率。

日志记录器初始化

# 初始化日志记录器，使用JSON格式记录，便于后续分析
logger = logging.getLogger("waf")
logger.setLevel(logging.INFO)
fh = logging.FileHandler(LOG_PATH, encoding="utf8")
formatter = logging.Formatter('%(message)s')  # 直接输出JSON字符串，不加额外格式
fh.setFormatter(formatter)
logger.addHandler(fh)

日志格式设为 %(message)s，直接输出JSON字符串。这样日志文件每行就是一个完整的JSON对象，可以用 jq 等工具快速查询分析。

Payload标准化函数

标准化是WAF反绕过的关键环节：

def normalize_payload(path: str, query: str, ua: str, body: str) -> str:
    """
    对HTTP请求的各个部分进行标准化处理，便于后续检测
    """
    try:
        q = unquote_plus(query or "")  # URL解码查询参数
        b = unquote_plus(body or "")   # URL解码请求体
    except Exception:
        q = query or ""
        b = body or ""
    # 拼接所有部分并统一小写
    combined = " ".join([path or "", q, ua or "", b or ""]).lower()
    return combined

这个函数做了三件事：

URL解码：unquote_plus 将 %20 解码为空格、%3C 解码为 <，确保编码绕过无效
拼接所有输入向量：路径、查询参数、User-Agent、请求体合并为一个字符串，确保攻击载荷无论藏在哪个字段都能被检测到
统一小写：.lower() 使后续正则匹配无需考虑大小写问题

WAF中间件核心逻辑

@app.middleware("http")
async def waf_middleware(request: Request, call_next):
    """
    WAF中间件，用于检测和拦截恶意请求
    """
    try:
        # 提取请求体和User-Agent
        body_bytes = await request.body()
        body = body_bytes.decode(errors="ignore")
        ua = request.headers.get("user-agent", "")
        # 标准化payload
        payload = normalize_payload(request.url.path, request.url.query, ua, body)

        # 规则匹配与评分
        score = 0
        matched = []
        for rid, cre in COMPILED:
            if cre.search(payload):
                score += WEIGHTS.get(rid, 10)
                matched.append(rid)

        # 记录请求信息和检测结果（无论是否拦截都记录）
        record = {
            "ts": int(time.time()),
            "src_ip": request.client.host if request.client else None,
            "method": request.method,
            "path": str(request.url.path),
            "query": str(request.url.query),
            "ua": ua,
            "score": score,
            "matched": matched,
            "body_snip": body[:800]  # 截取前800字符，避免日志过大
        }
        logger.info(json.dumps(record, ensure_ascii=False))

        # 决策：是否拦截
        if score >= SCORE_THRESHOLD:
            if FAIL_OPEN:
                # 故障安全：fail-open 时依旧放行，但已记录高优先级日志
                return await call_next(request)
            else:
                return JSONResponse(
                    {"detail": "blocked by waf", "score": score, "matched": matched},
                    status_code=403
                )

    except Exception as e:
        # 引擎异常时的安全策略
        if FAIL_OPEN:
            return await call_next(request)  # 放行
        else:
            return JSONResponse({"detail": "waf internal error"}, status_code=500)

    return await call_next(request)

这段代码体现了WAF的完整决策链路，有几个关键点值得深入理解：

1. 全向量提取：body_bytes = await request.body() 读取完整请求体，加上 path、query、ua，覆盖了HTTP请求中所有可能携带攻击载荷的位置。

2. 记录先于决策：无论最终是否拦截，日志都会记录。这意味着即使WAF处于”只监控”模式（阈值设为100），也能收集到所有匹配到规则的请求，为后续调优提供数据基础。

3. 分层异常处理：

正则匹配阶段的异常被 try-except 包裹
异常处理中再次检查 FAIL_OPEN 策略
确保WAF自身的故障不会导致请求处理中断

4. 响应中携带匹配信息：拦截时返回 {"detail": "blocked by waf", "score": score, "matched": matched}，便于前端和API客户端理解被拦截的原因。

测试端点

WAF源码附带了两个测试端点：

@app.get("/health")
async def health():
    """健康检查端点"""
    return {"status": "ok"}

@app.post("/echo")
async def echo(req: Request):
    """回显端点，用于测试WAF对POST body的检测"""
    body = await req.body()
    return {"echo": body.decode(errors="ignore")[:1000]}

/echo 端点特别有用——它会回显请求体内容，可用于验证WAF是否正确拦截了带有恶意payload的POST请求。

部署与使用方法

安装依赖

1	pip install fastapi uvicorn

启动WAF

# 默认模式（只监控，不拦截）
python3 Demo001.py
# 通过uvicorn运行：
uvicorn Demo001:app --host 0.0.0.0 --port 8000

# 拦截模式（设置较低阈值）
WAF_BLOCK_SCORE=60 WAF_FAIL_OPEN=0 uvicorn Demo001:app --host 0.0.0.0 --port 8000

# 自定义日志路径
WAF_LOG=/var/log/waf.log uvicorn Demo001:app --host 0.0.0.0 --port 8000

验证测试

# 正常请求（不会被拦截）
curl http://localhost:8000/health

# SQL注入测试（触发sqli_union规则，得分50）
curl "http://localhost:8000/?id=1+UNION+SELECT+1"

# XSS测试（触发xss_tag规则，得分30）
curl "http://localhost:8000/?q=<script>alert(1)</script>"

# 组合攻击（触发多条规则，得分高）
curl "http://localhost:8000/?id=1+UNION+SELECT+*+FROM+information_schema.tables" \
     -A "<script>javascript:alert(1)</script>"
# 命中: sqli_union(50) + sqli_information_schema(40) + xss_event(20) = 110分 > 100阈值

# POST body注入测试
curl -X POST http://localhost:8000/echo \
     -d "data=1 UNION SELECT password FROM information_schema.tables"

日志分析

# 查看被匹配到规则的请求
cat waf_samples.log | python3 -c "
import json, sys
for line in sys.stdin:
    r = json.loads(line)
    if r['score'] > 0:
        print(f\"[{r['score']}分] {r['method']} {r['path']} matched={r['matched']} ip={r['src_ip']}\")
"

# 统计各类攻击次数
cat waf_samples.log | python3 -c "
import json, sys
from collections import Counter
c = Counter()
for line in sys.stdin:
    r = json.loads(line)
    for rule in r.get('matched', []):
        c[rule] += 1
for rule, count in c.most_common():
    print(f'{rule}: {count}次')
"

生产环境部署建议

在实际部署中，建议将WAF作为反向代理前置：

1	客户端 → Nginx → FastAPI WAF (uvicorn) → 后端应用

或使用Docker部署：

FROM python:3.11-slim
WORKDIR /app
COPY Demo001.py .
RUN pip install fastapi uvicorn
CMD ["uvicorn", "Demo001:app", "--host", "0.0.0.0", "--port", "8000"]

防御效果分析

检测能力

该WAF能有效检测以下攻击：

攻击类型	测试Payload	命中规则	得分
SQL注入	`1 UNION SELECT * FROM users`	sqli_union	50
信息架构探测	`... FROM information_schema.tables`	sqli_information_schema	40
XSS标签注入	`<script>alert(1)</script>`	xss_tag	30
XSS事件注入	`<img onerror=alert(1)>`	xss_event	20
组合攻击	UNION + information_schema + XSS	三条规则	110

安全策略优势

评分制降低误报：单一低权重规则命中不会触发拦截，减少对正常业务的影响
Fail-Open保障可用性：WAF故障或规则异常时不会阻断业务
全向量覆盖：路径、查询参数、User-Agent、请求体均纳入检测范围
环境变量配置：无需修改代码即可调整阈值和安全策略

局限性与改进方向

规则集精简：仅包含4条规则，生产环境需扩展至覆盖OWASP Top 10全量攻击类型
标准化不足：仅做单次URL解码，无法应对双重编码绕过；缺少HTML实体解码、Unicode规范化
无学习能力：纯规则驱动，无法识别未知攻击模式（可考虑集成机器学习异常检测）
性能考量：每个请求都要读取完整body并执行多次正则匹配，在高并发场景下需做性能评估
无IP信誉库：不结合威胁情报，无法对已知恶意IP提高检测敏感度

扩展规则示例

可以按相同格式添加更多规则：

RULES = [
    # ... 原有规则 ...
    ("sqli_or_injection", r"(?i)'\s*or\s*'1'\s*=\s*'1", 45),      # OR '1'='1
    ("sqli_comment", r"(?i)(--|#|/\*)", 15),                        # SQL注释符
    ("xss_img", r"(?i)<\s*img\b[^>]*\bonerror\b", 25),              # <img onerror>
    ("lfi_traversal", r"\.\./|\.\.\\", 35),                         # 目录穿越
    ("rfi_remote", r"(?i)(file|php|http|ftp)://", 20),             # 远程文件包含
    ("cmd_injection", r"(?i)(;\s*(cat|ls|id|whoami)\b|\$\(|`)", 40), # 命令注入
    ("xxe", r"(?i)<!ENTITY\b", 50),                                 # XXE实体
]

总结

本文基于一个真实的FastAPI WAF源码，完整讲解了Web应用防火墙的核心实现原理。从规则定义、正则预编译、Payload标准化、评分计算到Fail-Open安全策略，这不到150行代码涵盖了WAF最核心的工程要素。

理解这个基础实现后，就能更好地使用和调优生产级WAF：知道为什么某条规则会误报（权重过高）、理解双重编码为什么能绕过（标准化不充分）、明白Fail-Open策略的利弊权衡。WAF不是银弹，它是纵深防御体系中的一环——配合蜜罐的主动感知、IP自动封禁的快速响应，才能构建起完整的Web安全防御体系。

对于安全工程师而言，掌握WAF底层原理的价值远超熟练使用某款产品，因为安全对抗的本质是规则与绕过的持续博弈——只有理解了检测机制，才能在攻防对抗中始终领先一步。

WAF基础实现：基于FastAPI中间件的轻量级Web应用防火墙

W

前言