Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

NIST CSF 2.0

Overview

Broken Object Property Level Authorization (BOPLA), classified as API3:2023 in the OWASP API Security Top 10, combines two related vulnerability classes: Excessive Data Exposure (API returning more data than needed) and Mass Assignment (API accepting more data than intended). Even when APIs enforce object-level authorization correctly, they may fail to control which specific properties of an object a user can read or modify. Attackers exploit this by reading sensitive properties from API responses or injecting additional properties into request bodies to modify fields they should not have access to.

When to Use

When investigating security incidents that require detecting broken object property level authorization
When building detection rules or threat hunting queries for this domain
When SOC analysts need structured procedures for this analysis type
When validating security monitoring coverage for related attack techniques

Prerequisites

Target API with endpoints that return or accept object data
API documentation or schema (OpenAPI spec preferred)
Burp Suite or Postman for API request manipulation
Multiple user accounts with different privilege levels
Python 3.8+ with requests library for automated testing
Authorization to perform security testing

Vulnerability Patterns

Excessive Data Exposure

The API returns object properties the client does not need:

// GET /api/v1/users/123
// Response includes sensitive fields the UI doesn't display:
{
  "id": 123,
  "username": "john_doe",
  "email": "john@example.com",
  "name": "John Doe",
  "ssn": "123-45-6789",           // Sensitive - not needed by UI
  "salary": 95000,                 // Sensitive - not needed by UI
  "internal_notes": "VIP client",  // Internal - should not be exposed
  "password_hash": "$2b$12...",    // Critical - never expose
  "role": "admin",                 // May enable privilege discovery
  "created_by": "system_admin",   // Internal metadata
  "credit_card_last4": "4242"     // PCI compliance violation
}

Mass Assignment

The API binds client-supplied data to internal object properties without filtering:

// Normal user update request
PUT /api/v1/users/123
Content-Type: application/json
 
{
  "name": "John Updated",
  "email": "new@example.com",
  "role": "admin",           // Attacker-injected: privilege escalation
  "is_verified": true,       // Attacker-injected: bypass verification
  "discount_rate": 100,      // Attacker-injected: business logic abuse
  "account_balance": 999999  // Attacker-injected: financial fraud
}

Testing Methodology

#!/usr/bin/env python3
"""BOPLA Vulnerability Scanner
 
Tests APIs for Broken Object Property Level Authorization
including Excessive Data Exposure and Mass Assignment.
"""
 
import requests
import json
import sys
from typing import Dict, List, Optional, Set
from dataclasses import dataclass, field
from copy import deepcopy
 
@dataclass
class BOPLAFinding:
    endpoint: str
    method: str
    vulnerability_type: str  # "excessive_exposure" or "mass_assignment"
    severity: str
    property_name: str
    details: str
 
class BOPLAScanner:
    SENSITIVE_PROPERTY_PATTERNS = {
        "critical": [
            "password", "password_hash", "secret", "token", "api_key",
            "private_key", "secret_key", "access_token", "refresh_token",
        ],
        "high": [
            "ssn", "social_security", "tax_id", "credit_card", "card_number",
            "cvv", "bank_account", "routing_number",
        ],
        "medium": [
            "salary", "income", "internal_notes", "admin_notes",
            "created_by", "modified_by", "ip_address", "session_id",
            "role", "permissions", "is_admin", "is_superuser", "privilege",
        ],
        "low": [
            "phone", "address", "date_of_birth", "dob", "age",
            "gender", "ethnicity", "religion",
        ]
    }
 
    MASS_ASSIGNMENT_FIELDS = [
        ("role", "admin"),
        ("is_admin", True),
        ("is_verified", True),
        ("is_active", True),
        ("email_verified", True),
        ("account_type", "premium"),
        ("discount_rate", 100),
        ("credit_limit", 999999),
        ("permissions", ["admin", "write", "delete"]),
        ("account_balance", 999999),
        ("subscription_tier", "enterprise"),
        ("rate_limit", 999999),
    ]
 
    def __init__(self, base_url: str, auth_headers: Dict[str, str]):
        self.base_url = base_url.rstrip('/')
        self.auth_headers = auth_headers
        self.findings: List[BOPLAFinding] = []
 
    def test_excessive_data_exposure(self, endpoint: str,
                                      expected_fields: Set[str]) -> List[BOPLAFinding]:
        """Test if API response contains more fields than expected."""
        findings = []
        url = f"{self.base_url}{endpoint}"
 
        try:
            response = requests.get(url, headers=self.auth_headers, timeout=10)
            if response.status_code != 200:
                return findings
 
            data = response.json()
 
            # Handle both single object and list responses
            objects = data if isinstance(data, list) else [data]
            if isinstance(data, dict) and "data" in data:
                objects = data["data"] if isinstance(data["data"], list) else [data["data"]]
 
            for obj in objects[:5]:  # Check first 5 objects
                if not isinstance(obj, dict):
                    continue
 
                response_fields = set(self._flatten_keys(obj))
                unexpected_fields = response_fields - expected_fields
 
                for field_name in unexpected_fields:
                    severity = self._classify_sensitivity(field_name)
                    if severity:
                        finding = BOPLAFinding(
                            endpoint=endpoint,
                            method="GET",
                            vulnerability_type="excessive_exposure",
                            severity=severity,
                            property_name=field_name,
                            details=f"Unexpected sensitive field '{field_name}' in response"
                        )
                        findings.append(finding)
                        self.findings.append(finding)
 
        except (requests.exceptions.RequestException, json.JSONDecodeError):
            pass
 
        return findings
 
    def test_mass_assignment(self, endpoint: str, method: str = "PUT",
                              original_data: Optional[dict] = None) -> List[BOPLAFinding]:
        """Test if API accepts and processes additional injected properties."""
        findings = []
        url = f"{self.base_url}{endpoint}"
 
        # First, get the current object state
        if original_data is None:
            try:
                response = requests.get(url, headers=self.auth_headers, timeout=10)
                if response.status_code == 200:
                    original_data = response.json()
                else:
                    original_data = {}
            except (requests.exceptions.RequestException, json.JSONDecodeError):
                original_data = {}
 
        # Test each mass assignment field
        for field_name, injected_value in self.MASS_ASSIGNMENT_FIELDS:
            if field_name in original_data:
                # Field exists - test if we can modify it
                original_value = original_data[field_name]
                if original_value == injected_value:
                    continue  # Already has this value
 
            test_data = deepcopy(original_data)
            test_data[field_name] = injected_value
 
            headers = {**self.auth_headers, "Content-Type": "application/json"}
 
            try:
                if method == "PUT":
                    response = requests.put(url, json=test_data,
                                          headers=headers, timeout=10)
                elif method == "PATCH":
                    response = requests.patch(url, json={field_name: injected_value},
                                            headers=headers, timeout=10)
                elif method == "POST":
                    response = requests.post(url, json=test_data,
                                           headers=headers, timeout=10)
 
                if response.status_code in (200, 201, 204):
                    # Verify the field was actually modified
                    verify_response = requests.get(url, headers=self.auth_headers, timeout=10)
                    if verify_response.status_code == 200:
                        updated_data = verify_response.json()
                        if updated_data.get(field_name) == injected_value:
                            finding = BOPLAFinding(
                                endpoint=endpoint,
                                method=method,
                                vulnerability_type="mass_assignment",
                                severity="CRITICAL" if field_name in ["role", "is_admin", "permissions"]
                                         else "HIGH",
                                property_name=field_name,
                                details=f"Successfully injected '{field_name}={injected_value}'"
                            )
                            findings.append(finding)
                            self.findings.append(finding)
 
                            # Restore original value if possible
                            if field_name in original_data:
                                restore_data = {field_name: original_data[field_name]}
                                requests.patch(url, json=restore_data,
                                             headers=headers, timeout=10)
 
            except requests.exceptions.RequestException:
                continue
 
        return findings
 
    def test_graphql_property_exposure(self, graphql_endpoint: str,
                                        query: str) -> List[BOPLAFinding]:
        """Test GraphQL APIs for property-level authorization issues."""
        findings = []
        url = f"{self.base_url}{graphql_endpoint}"
 
        # Introspection query to discover available fields
        introspection = """
        {
          __schema {
            types {
              name
              fields {
                name
                type { name kind }
              }
            }
          }
        }
        """
 
        try:
            response = requests.post(
                url,
                json={"query": introspection},
                headers=self.auth_headers,
                timeout=10
            )
 
            if response.status_code == 200:
                data = response.json()
                if "errors" not in data:
                    finding = BOPLAFinding(
                        endpoint=graphql_endpoint,
                        method="POST",
                        vulnerability_type="excessive_exposure",
                        severity="MEDIUM",
                        property_name="__schema",
                        details="GraphQL introspection enabled - full schema exposed"
                    )
                    findings.append(finding)
                    self.findings.append(finding)
 
        except requests.exceptions.RequestException:
            pass
 
        return findings
 
    def _flatten_keys(self, obj: dict, prefix: str = "") -> List[str]:
        """Recursively flatten nested dictionary keys."""
        keys = []
        for key, value in obj.items():
            full_key = f"{prefix}.{key}" if prefix else key
            keys.append(full_key)
            if isinstance(value, dict):
                keys.extend(self._flatten_keys(value, full_key))
        return keys
 
    def _classify_sensitivity(self, field_name: str) -> Optional[str]:
        """Classify the sensitivity level of a field name."""
        lower_name = field_name.lower().split('.')[-1]
        for severity, patterns in self.SENSITIVE_PROPERTY_PATTERNS.items():
            for pattern in patterns:
                if pattern in lower_name:
                    return severity.upper()
        return None
 
    def generate_report(self) -> dict:
        return {
            "total_findings": len(self.findings),
            "by_type": {
                "excessive_exposure": len([f for f in self.findings
                                          if f.vulnerability_type == "excessive_exposure"]),
                "mass_assignment": len([f for f in self.findings
                                       if f.vulnerability_type == "mass_assignment"]),
            },
            "by_severity": {
                "CRITICAL": len([f for f in self.findings if f.severity == "CRITICAL"]),
                "HIGH": len([f for f in self.findings if f.severity == "HIGH"]),
                "MEDIUM": len([f for f in self.findings if f.severity == "MEDIUM"]),
                "LOW": len([f for f in self.findings if f.severity == "LOW"]),
            },
            "findings": [
                {
                    "endpoint": f.endpoint,
                    "method": f.method,
                    "type": f.vulnerability_type,
                    "severity": f.severity,
                    "property": f.property_name,
                    "details": f.details,
                }
                for f in self.findings
            ]
        }

Mitigation

# Server-side: Explicit property allowlists
class UserSerializer:
    # Only expose these fields - never use to_json() or to_dict()
    PUBLIC_FIELDS = ['id', 'username', 'name', 'avatar_url']
    OWNER_FIELDS = PUBLIC_FIELDS + ['email', 'phone', 'preferences']
    ADMIN_FIELDS = OWNER_FIELDS + ['role', 'created_at', 'last_login']
 
    def serialize(self, user, requesting_user):
        if requesting_user.is_admin:
            fields = self.ADMIN_FIELDS
        elif requesting_user.id == user.id:
            fields = self.OWNER_FIELDS
        else:
            fields = self.PUBLIC_FIELDS
 
        return {field: getattr(user, field) for field in fields}
 
# Mass assignment protection - explicit allowlist for writable fields
WRITABLE_FIELDS = {'name', 'email', 'phone', 'avatar_url', 'preferences'}
 
def update_user(user_id, request_data, requesting_user):
    # Filter out any fields not in the allowlist
    safe_data = {k: v for k, v in request_data.items() if k in WRITABLE_FIELDS}
    # Apply updates only with safe data
    User.objects.filter(id=user_id).update(**safe_data)

References

OWASP API3:2023: https://owasp.org/API-Security/editions/2023/en/0xa3-broken-object-property-level-authorization/
Salt Security BOPLA Analysis: https://salt.security/blog/api3-2023-broken-object-property-level-authorization
Wallarm BOPLA Guide: https://lab.wallarm.com/api32023-broken-object-property-level-authorization/
API Security News BOPLA: https://apisecurity.io/owasp-api-security-top-10/api3-2023-broken-object-property-level-authorization/
CloudDefense BOPLA: https://www.clouddefense.ai/owasp/2023/3

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md2.1 KB

API Reference: Detecting Broken Object Property Level Authorization

OWASP API3:2023 Classification

Category	Description
Excessive Data Exposure	API returns more properties than client needs
Mass Assignment	API accepts more properties than intended
CWE-213	Exposure of Sensitive Information Due to Incompatible Policies
CWE-915	Improperly Controlled Modification of Dynamically-Determined Object Attributes

Python requests Library

import requests
 
# GET - test for excessive exposure
resp = requests.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=10)
resp.status_code  # 200, 401, 403
resp.json()       # parsed response body
 
# PUT - test for mass assignment
resp = requests.put(url, json={"role": "admin"}, headers=headers, timeout=10)
 
# PATCH - test for partial mass assignment
resp = requests.patch(url, json={"is_admin": True}, headers=headers, timeout=10)

Sensitive Field Patterns

Severity	Fields
Critical	password, password_hash, secret, token, api_key, private_key
High	ssn, credit_card, card_number, cvv, bank_account
Medium	salary, role, permissions, is_admin, session_id
Low	phone, address, date_of_birth, gender

Mass Assignment Test Payloads

{"role": "admin"}
{"is_admin": true}
{"is_verified": true}
{"account_type": "premium"}
{"discount_rate": 100}
{"permissions": ["admin", "write", "delete"]}

Burp Suite Extensions

# Autorize - test authorization across roles
# Param Miner - discover hidden parameters
# JSON Beautifier - inspect response properties

CLI Usage

python agent.py --base-url https://api.example.com \
  --endpoint /api/v1/users/123 \
  --token "eyJhbGciOiJIUzI1NiJ9..." \
  --expected-fields id username name email \
  --test both --method PUT

Mitigation Patterns

# Allowlist serialization (Django REST Framework)
class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['id', 'username', 'name']  # explicit allowlist
        read_only_fields = ['id', 'role', 'is_admin']

Scripts 1

agent.py6.4 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
"""BOPLA vulnerability scanner for OWASP API3:2023 Broken Object Property Level Authorization.

Tests APIs for excessive data exposure and mass assignment vulnerabilities
by comparing responses against expected field sets and injecting extra properties.
"""

import argparse
import json
import sys
from copy import deepcopy

try:
    import requests
except ImportError:
    print("Install requests: pip install requests")
    sys.exit(1)

SENSITIVE_PATTERNS = {
    "critical": ["password", "password_hash", "secret", "token", "api_key",
                 "private_key", "access_token", "refresh_token"],
    "high": ["ssn", "social_security", "tax_id", "credit_card", "card_number",
             "cvv", "bank_account", "routing_number"],
    "medium": ["salary", "income", "internal_notes", "role", "permissions",
               "is_admin", "is_superuser", "session_id", "ip_address"],
    "low": ["phone", "address", "date_of_birth", "dob", "gender"]
}

MASS_ASSIGNMENT_PAYLOADS = [
    ("role", "admin"), ("is_admin", True), ("is_verified", True),
    ("email_verified", True), ("account_type", "premium"),
    ("discount_rate", 100), ("permissions", ["admin", "write", "delete"]),
    ("account_balance", 999999), ("subscription_tier", "enterprise"),
]


def classify_field(field_name):
    lower = field_name.lower().split(".")[-1]
    for severity, patterns in SENSITIVE_PATTERNS.items():
        for p in patterns:
            if p in lower:
                return severity.upper()
    return None


def flatten_keys(obj, prefix=""):
    keys = []
    for k, v in obj.items():
        full = f"{prefix}.{k}" if prefix else k
        keys.append(full)
        if isinstance(v, dict):
            keys.extend(flatten_keys(v, full))
    return keys


def test_excessive_exposure(base_url, endpoint, expected_fields, headers):
    findings = []
    url = f"{base_url.rstrip('/')}{endpoint}"
    try:
        resp = requests.get(url, headers=headers, timeout=10)
        if resp.status_code != 200:
            return findings
        data = resp.json()
        objects = data if isinstance(data, list) else [data]
        if isinstance(data, dict) and "data" in data:
            inner = data["data"]
            objects = inner if isinstance(inner, list) else [inner]
        for obj in objects[:5]:
            if not isinstance(obj, dict):
                continue
            response_fields = set(flatten_keys(obj))
            unexpected = response_fields - set(expected_fields)
            for field in unexpected:
                sev = classify_field(field)
                if sev:
                    findings.append({
                        "endpoint": endpoint, "method": "GET",
                        "type": "excessive_exposure", "severity": sev,
                        "property": field,
                        "detail": f"Unexpected sensitive field '{field}' in response"
                    })
    except (requests.RequestException, json.JSONDecodeError) as e:
        findings.append({"error": str(e)})
    return findings


def test_mass_assignment(base_url, endpoint, headers, method="PUT"):
    findings = []
    url = f"{base_url.rstrip('/')}{endpoint}"
    try:
        original = requests.get(url, headers=headers, timeout=10)
        original_data = original.json() if original.status_code == 200 else {}
    except (requests.RequestException, json.JSONDecodeError):
        original_data = {}

    for field_name, injected_value in MASS_ASSIGNMENT_PAYLOADS:
        if original_data.get(field_name) == injected_value:
            continue
        test_data = deepcopy(original_data)
        test_data[field_name] = injected_value
        try:
            h = {**headers, "Content-Type": "application/json"}
            if method == "PUT":
                resp = requests.put(url, json=test_data, headers=h, timeout=10)
            elif method == "PATCH":
                resp = requests.patch(url, json={field_name: injected_value}, headers=h, timeout=10)
            else:
                resp = requests.post(url, json=test_data, headers=h, timeout=10)
            if resp.status_code in (200, 201, 204):
                verify = requests.get(url, headers=headers, timeout=10)
                if verify.status_code == 200 and verify.json().get(field_name) == injected_value:
                    sev = "CRITICAL" if field_name in ("role", "is_admin", "permissions") else "HIGH"
                    findings.append({
                        "endpoint": endpoint, "method": method,
                        "type": "mass_assignment", "severity": sev,
                        "property": field_name,
                        "detail": f"Injected {field_name}={injected_value} accepted"
                    })
                    if field_name in original_data:
                        requests.patch(url, json={field_name: original_data[field_name]},
                                       headers=h, timeout=10)
        except requests.RequestException:
            continue
    return findings


def main():
    parser = argparse.ArgumentParser(description="BOPLA Vulnerability Scanner (OWASP API3:2023)")
    parser.add_argument("--base-url", required=True, help="API base URL")
    parser.add_argument("--endpoint", required=True, help="Endpoint to test (e.g. /api/v1/users/1)")
    parser.add_argument("--token", help="Bearer token for Authorization header")
    parser.add_argument("--expected-fields", nargs="+", default=[], help="Expected response fields")
    parser.add_argument("--test", choices=["exposure", "mass_assignment", "both"], default="both")
    parser.add_argument("--method", choices=["PUT", "PATCH", "POST"], default="PUT")
    args = parser.parse_args()

    headers = {}
    if args.token:
        headers["Authorization"] = f"Bearer {args.token}"

    results = {"endpoint": args.endpoint, "findings": []}
    if args.test in ("exposure", "both"):
        results["findings"].extend(
            test_excessive_exposure(args.base_url, args.endpoint, args.expected_fields, headers))
    if args.test in ("mass_assignment", "both"):
        results["findings"].extend(
            test_mass_assignment(args.base_url, args.endpoint, headers, args.method))

    results["total_findings"] = len(results["findings"])
    results["by_severity"] = {}
    for f in results["findings"]:
        sev = f.get("severity", "UNKNOWN")
        results["by_severity"][sev] = results["by_severity"].get(sev, 0) + 1

    print(json.dumps(results, indent=2))


if __name__ == "__main__":
    main()

Keep exploring