1
def build_parser():
2
    p = argparse.ArgumentParser(
3
        prog="run.py",
4
        description="waf-lab —— 纯标准库实现的反向代理式 WAF",
5
        formatter_class=argparse.RawDescriptionHelpFormatter,
6
        epilog="""\
7
示例：
8
  python3 run.py --backend 127.0.0.1:8080
9
  python3 run.py --backend 127.0.0.1:8080 --listen 0.0.0.0:80 --mode audit
10
  python3 run.py --list-rules
11

12
把 WAF 接入自己的站点时，只需要把域名解析或 Nginx 的 proxy_pass
13
指到 WAF 的监听端口，后端地址填真实站点的地址即可 —— 后端本身不用改任何代码。
14
""",
15
    )
16

17
    g = p.add_argument_group("网络")
18
    g.add_argument("--listen", default="127.0.0.1:8888",
19
                   help="WAF 监听地址，默认 127.0.0.1:8888")
20
    g.add_argument("--backend", default="127.0.0.1:8080",
21
                   help="后端（被保护的站点）地址，默认 127.0.0.1:8080")
22
    g.add_argument("--timeout", type=float, default=15.0,
23
                   help="与后端通信的超时秒数，默认 15")
24

25
    g = p.add_argument_group("检测")
26
    g.add_argument("--rules", action="append", default=None,
27
                   help="规则文件或目录，可重复指定。默认读 rules/ 目录下的 default.rules")
28
    g.add_argument("--threshold", type=int, default=100,
29
                   help="拦截阈值（累计风险分），默认 100")
30
    g.add_argument("--mode", choices=("block", "audit"), default="block",
31
                   help="block=命中即拦，audit=只记录不拦（默认 block）")
32

33
    g = p.add_argument_group("速率限制")
34
    g.add_argument("--rate-limit", type=int, default=120,
35
                   help="单 IP 在窗口内允许的请求数，默认 120；设为 0 关闭")
36
    g.add_argument("--rate-window", type=float, default=10.0,
37
                   help="速率窗口秒数，默认 10")
38
    g.add_argument("--rate-score", type=int, default=3,
39
                   help="每超出一次加多少分，默认 3")
40

41
    g = p.add_argument_group("输出")
42
    g.add_argument("--log-dir", default=os.path.join(HERE, "logs"),
43
                   help="审计日志目录，默认 ./logs")
44
    g.add_argument("--quiet", action="store_true",
45
                   help="不打印每个请求的处理结果（日志仍会写文件）")
46
    g.add_argument("--quiet-headers", action="store_true",
47
                   help="阻断响应里不暴露规则编号 —— 生产环境建议开启")
48
    g.add_argument("--no-color", action="store_true", help="关闭终端彩色输出")
49
    g.add_argument("--list-rules", action="store_true",
50
                   help="只打印加载到的规则然后退出")
51

52
    return p

1
def main(argv=None):
2
    args = build_parser().parse_args(argv)
3
    for stream in (sys.stdout, sys.stderr):
4
        try:
5
            stream.reconfigure(encoding="utf-8", line_buffering=True)
6
        except Exception:
7
            pass
8

9
    if args.no_color or not sys.stdout.isatty():
10
        C.disable()
11

12
    # ---- 加载规则 ----
13
    rule_paths = args.rules or [os.path.join(HERE, "rules")]
14
    try:
15
        ruleset = RuleSet.from_files(rule_paths, threshold=args.threshold)
16
    except RuleError as exc:
17
        print("%s[!] 规则加载失败%s：%s" % (C.RED, C.RESET, exc))
18
        return 2
19

20
    if args.list_rules:
21
        print(ruleset.summary())
22
        return 0
23

24
    if not ruleset.detect_rules:
25
        print("%s[!]%s 没有任何检测规则，WAF 会把所有流量原样放行。" % (C.YELLOW, C.RESET))
26
        print("    检查一下 --rules 指向的路径是否正确。")
27

28
    # ---- 组装各层 ----
29
    if args.rate_limit and args.rate_limit > 0:
30
        limiter = RateLimiter(
31
            window=args.rate_window,
32
            limit=args.rate_limit,
33
            score_per_excess=args.rate_score,
34
        )
35
    else:
36
        # 传一个上限极大、永不触发的限流器，等于关闭
37
        limiter = RateLimiter(window=1.0, limit=10 ** 9, score_per_excess=0)
38

39
    audit_log = AuditLog(args.log_dir, echo=False)
40

41
    backend_host, backend_port = parse_hostport(args.backend, 80)
42
    listen_host, listen_port = parse_hostport(args.listen, 8888)
43

44
    cfg = Config(
45
        listen_host=listen_host,
46
        listen_port=listen_port,
47
        backend_host=backend_host,
48
        backend_port=backend_port,
49
        ruleset=ruleset,
50
        limiter=limiter,
51
        audit_log=audit_log,
52
        mode=args.mode,
53
        timeout=args.timeout,
54
        verbose=not args.quiet,
55
        expose_rules=not args.quiet_headers,
56
    )
57

58
    try:
59
        serve(cfg)
60
    except OSError as exc:
61
        print("%s[!] 无法监听 %s:%d —— %s" % (
62
            C.RED, C.RESET, listen_host, listen_port, exc))
63
        print("    端口被占用的话，换一个：--listen 127.0.0.1:8889")
64
        return 1
65

66
    return 0
67

68

69
if __name__ == "__main__":
70
    sys.exit(main())

1
def read_line(self, limit=MAX_LINE):
2
        while True:
3
            idx = self.buf.find(b"\n")
4
            if idx >= 0:
5
                line = self.buf[:idx]
6
                self.buf = self.buf[idx + 1:]
7
                if line.endswith(b"\r"):
8
                    line = line[:-1]
9
                return line
10

11
            # 没找到换行符，说明这一行还没读全
12
            if len(self.buf) > limit:
13
                raise ProtocolError("单行超长（超过 %d 字节）" % limit)
14
            if not self._fill():
15
                if self.buf:
16
                    line = self.buf
17
                    self.buf = b""
18
                    return line
19
                return None

1
def read_chunked(self, max_total=MAX_BODY):
2
        out = bytearray()
3
        while True:
4
            line = self.read_line()
5
            if line is None:
6
                raise ProtocolError("chunked 请求体意外结束")
7
            size_part = line.split(b";", 1)[0].strip()
8
            if not size_part:
9
                continue   # 容忍空行
10

11
            try:
12
                size = int(size_part, 16)
13
            except ValueError:
14
                raise ProtocolError("非法的 chunk 长度: %r" % line)
15

16
            if size == 0:
17
                # 结束标记。后面可能还有 trailer 头，一直读到空行为止
18
                while True:
19
                    trailer = self.read_line()
20
                    if trailer is None or trailer == b"":
21
                        break
22
                break
23

24
            if len(out) + size > max_total:
25
                raise ProtocolError("chunked 请求体超过上限")
26

27
            out += self.read_exactly(size)
28

29
            # 数据后面必须跟一个 CRLF（read_line 会返回空串）
30
            ending = self.read_line()
31
            if ending != b"":
32
                raise ProtocolError("chunk 数据后缺少 CRLF")
33

34
        return bytes(out)

1
def read_request(reader, client):
2
    """从字节流读出一个完整请求。
3

4
    返回值：
5
        RawRequest  —— 解析成功
6
        None        —— 连接被对端正常关闭（没有更多请求了）
7

8
    抛异常：
9
        ProtocolError —— 报文非法或超限，调用方应回 400
10
    """
11
    # ---- 1. 请求行 ----
12
    line = reader.read_line()
13
    if line is None:
14
        return None
15

16
    # RFC 允许请求前有若干空行，用来自律跳过（否则会误判成畸形报文）
17
    guard = 0
18
    while line == b"":
19
        line = reader.read_line()
20
        if line is None:
21
            return None
22
        guard += 1
23
        if guard > 8:
24
            raise ProtocolError("请求行前空行过多")
25

26
    # 请求行格式：METHOD SP request-target SP HTTP-version
27
    # 用 split(b" ") 而不是 split()，因为后者会把连续空白折叠 ——
28
    # 而"路径里带空格"是一个值得观察的异常现象，不该被悄悄消化掉
29
    parts = line.split(b" ")
30
    if len(parts) != 3:
31
        raise ProtocolError("非法请求行: %r" % line[:200])
32

33
    method = parts[0].decode("latin-1")
34
    target = parts[1].decode("latin-1")
35
    version = parts[2].decode("latin-1")
36

37
    if not version.upper().startswith("HTTP/"):
38
        raise ProtocolError("非法协议版本: %r" % version)
39

40
    # ---- 2. 请求头 ----
41
    header_list = []
42
    total_bytes = 0
43
    while True:
44
        hline = reader.read_line()
45
        if hline is None:
46
            raise ProtocolError("请求头未结束，连接就断了")
47
        if hline == b"":
48
            break   # 空行 = 头部结束
49

50
        total_bytes += len(hline)
51
        if total_bytes > MAX_HEADER_BYTES:
52
            raise ProtocolError("请求头总大小超限")
53

54
        # obs-fold：以空格/制表符开头的行是上一行的续行。
55
        # RFC 已废弃这种写法，但解析器对它的处理差异正是走私漏洞的来源之一
56
        if hline[:1] in (b" ", b"\t") and header_list:
57
            k, v = header_list[-1]
58
            header_list[-1] = (k, v + " " + hline.decode("latin-1").strip())
59
            continue
60

61
        if b":" not in hline:
62
            raise ProtocolError("非法请求头（缺少冒号）: %r" % hline[:200])
63

64
        k, _, v = hline.partition(b":")
65
        header_list.append((k.decode("latin-1"),
66
                            v.decode("latin-1").strip()))
67

68
        if len(header_list) > MAX_HEADER_COUNT:
69
            raise ProtocolError("请求头条数超限")
70

71
    # ---- 3. 请求体 ----
72
    # 这里就是协议层攻击的观察窗口。
73
    # RFC 7230 规定：如果同时出现 Transfer-Encoding 和 Content-Length，
74
    # 必须忽略 Content-Length。但现实中的中间件（代理 / WAF / 后端框架）
75
    # 对"以谁为准"的处理并不一致 —— 这个分歧就是 HTTP 请求走私（smuggling）
76
    # 的成因：让 WAF 按 CL 切分、让后端按 TE 切分，同一段字节被两种方式解读。
77
    #
78
    # 我们不去猜谁对，只需要**把这个分歧记下来**，交给规则去决定要不要拦。
79
    flags = []
80
    cl_raw = _hdr_first(header_list, "content-length")
81
    te_raw = _hdr_first(header_list, "transfer-encoding")
82

83
    body = b""
84
    body_encoding = "identity"
85

86
    if te_raw is not None:
87
        body_encoding = te_raw.strip().lower()
88
        if cl_raw is not None:
89
            flags.append("cl_te_conflict")
90
        if "chunked" in body_encoding:
91
            if body_encoding.replace(" ", "") not in ("chunked", "chunked,identity", "identity,chunked"):
92
                flags.append("odd_transfer_encoding")
93
            body = reader.read_chunked()
94
        # TE 存在但不含 chunked：没有可信的长度信息，按无 body 处理
95
    elif cl_raw is not None:
96
        # 出现多个不同的 Content-Length 同样是走私信号
97
        all_cl = set(v.strip() for v in _hdr_all(header_list, "content-length"))
98
        if len(all_cl) > 1:
99
            flags.append("cl_conflict")
100
        try:
101
            length = int(cl_raw.strip())
102
        except ValueError:
103
            raise ProtocolError("非法 Content-Length: %r" % cl_raw)
104
        if length < 0:
105
            raise ProtocolError("负数 Content-Length: %d" % length)
106
        if length > MAX_BODY:
107
            raise ProtocolError("请求体超过上限（%d > %d）" % (length, MAX_BODY))
108
        body = reader.read_exactly(length)
109

110
    return RawRequest(method, target, version, header_list, body,
111
                      client, body_encoding, flags)

1
line = reader.read_line()
2
    if line is None:
3
        return None
4

5
    # RFC 允许请求前有若干空行，用来自律跳过（否则会误判成畸形报文）
6
    guard = 0
7
    while line == b"":
8
        line = reader.read_line()
9
        if line is None:
10
            return None
11
        guard += 1
12
        if guard > 8:
13
            raise ProtocolError("请求行前空行过多")
14

15
    # 请求行格式：METHOD SP request-target SP HTTP-version
16
    # 用 split(b" ") 而不是 split()，因为后者会把连续空白折叠 ——
17
    # 而"路径里带空格"是一个值得观察的异常现象，不该被悄悄消化掉
18
    parts = line.split(b" ")
19
    if len(parts) != 3:
20
        raise ProtocolError("非法请求行: %r" % line[:200])
21

22
    method = parts[0].decode("latin-1")
23
    target = parts[1].decode("latin-1")
24
    version = parts[2].decode("latin-1")
25

26
    if not version.upper().startswith("HTTP/"):
27
        raise ProtocolError("非法协议版本: %r" % version)

1
header_list = []
2
    total_bytes = 0
3
    while True:
4
        hline = reader.read_line()
5
        if hline is None:
6
            raise ProtocolError("请求头未结束，连接就断了")
7
        if hline == b"":
8
            break   # 空行 = 头部结束
9

10
        total_bytes += len(hline)
11
        if total_bytes > MAX_HEADER_BYTES:
12
            raise ProtocolError("请求头总大小超限")
13

14
        # obs-fold：以空格/制表符开头的行是上一行的续行。
15
        # RFC 已废弃这种写法，但解析器对它的处理差异正是走私漏洞的来源之一
16
        if hline[:1] in (b" ", b"\t") and header_list:
17
            k, v = header_list[-1]
18
            header_list[-1] = (k, v + " " + hline.decode("latin-1").strip())
19
            continue
20

21
        if b":" not in hline:
22
            raise ProtocolError("非法请求头（缺少冒号）: %r" % hline[:200])
23

24
        k, _, v = hline.partition(b":")
25
        header_list.append((k.decode("latin-1"),
26
                            v.decode("latin-1").strip()))
27

28
        if len(header_list) > MAX_HEADER_COUNT:
29
            raise ProtocolError("请求头条数超限")

1
flags = []
2
    cl_raw = _hdr_first(header_list, "content-length")
3
    te_raw = _hdr_first(header_list, "transfer-encoding")
4

5
    body = b""
6
    body_encoding = "identity"
7

8
    if te_raw is not None:
9
        body_encoding = te_raw.strip().lower()
10
        if cl_raw is not None:
11
            flags.append("cl_te_conflict")
12
        if "chunked" in body_encoding:
13
            if body_encoding.replace(" ", "") not in ("chunked", "chunked,identity", "identity,chunked"):
14
                flags.append("odd_transfer_encoding")
15
            body = reader.read_chunked()
16
        # TE 存在但不含 chunked：没有可信的长度信息，按无 body 处理
17
    elif cl_raw is not None:
18
        # 出现多个不同的 Content-Length 同样是走私信号
19
        all_cl = set(v.strip() for v in _hdr_all(header_list, "content-length"))
20
        if len(all_cl) > 1:
21
            flags.append("cl_conflict")
22
        try:
23
            length = int(cl_raw.strip())
24
        except ValueError:
25
            raise ProtocolError("非法 Content-Length: %r" % cl_raw)
26
        if length < 0:
27
            raise ProtocolError("负数 Content-Length: %d" % length)
28
        if length > MAX_BODY:
29
            raise ProtocolError("请求体超过上限（%d > %d）" % (length, MAX_BODY))
30
        body = reader.read_exactly(length)
31

32
    return RawRequest(method, target, version, header_list, body,
33
                      client, body_encoding, flags)

1
def describe(req):
2
    """把请求渲染成人能读的文本，用于终端观察和排错。"""
3
    lines = []
4
    lines.append("%s %s %s" % (req.method, req.target, req.version))
5
    for k, v in req.header_list:
6
        if k.lower() == "cookie":
7
            v = v[:80] + ("..." if len(v) > 80 else "")
8
        lines.append("  %s: %s" % (k, v))
9
    if req.body:
10
        preview = req.body[:200]
11
        lines.append("  [body %d bytes, encoding=%s]" % (len(req.body), req.body_encoding))
12
        lines.append("  " + repr(preview))
13
    if req.flags:
14
        lines.append("  [协议层标记: %s]" % ", ".join(req.flags))
15
    return "\n".join(lines)

1
s = view_form(view_form(s))
2

3
    def _u(m):
4
        try:
5
            return chr(int(m.group(1), 16))
6
        except (ValueError, OverflowError):
7
            return m.group(0)
8

9
    def _x(m):
10
        try:
11
            return chr(int(m.group(1), 16))
12
        except (ValueError, OverflowError):
13
            return m.group(0)
14

15
    return _HEX_X.sub(_x, _HEX_U.sub(_u, s))

1
    s = view_url2(s)
2
    s = s.replace("\\", "/")
3
    s = _MULTI_SLASH.sub("/", s)
4
    return s
5

6
VIEWS = {
7
    "raw": view_raw,
8
    "url": view_url,
9
    "url2": view_url2,
10
    "url3": view_url3,
11
    "form": view_form,
12
    "utf8": view_utf8,
13
    "uni": view_uni,
14
    "lower": view_lower,
15
    "html": view_html,
16
    "sql": view_sql,
17
    "path": view_path,
18
}
19

20
VIEW_NAMES = tuple(VIEWS.keys())

1
__slots__ = ("target", "text", "_cache")
2

3
    def __init__(self, target, text):
4
        self.target = target        # 片段名，如 "path" / "query.value" / "header.ua"
5
        self.text = text            # 原始文本（未经解码）
6
        self._cache = {}
7

8
    def view(self, name):
9
        """取某个解码视图的文本。未知视图名按 raw 处理（不报错，避免一条规则打挂全流程）。"""
10
        cached = self._cache.get(name)
11
        if cached is not None:
12
            return cached
13

14
        fn = VIEWS.get(name)
15
        if fn is None:
16
            result = self.text
17
        else:
18
            try:
19
                result = fn(self.text)
20
            except Exception:
21
                # 解码失败不该让整个请求处理挂掉 —— 退化成原文继续检测
22
                result = self.text
23

24
        if result is None:
25
            result = ""
26

27
        # 名字带 "lower" 的视图顺便把结果也小写，方便规则只写一遍模式
28
        self._cache[name] = result
29
        return result
30

31
    def __repr__(self):
32
        return "<Payload %s=%r>" % (self.target, self.text[:60])