"""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
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()
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]:
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}"
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 = {}
for field_name, injected_value in self.MASS_ASSIGNMENT_FIELDS:
if field_name in original_data:
original_value = original_data[field_name]
if original_value == injected_value:
continue
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_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)
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 = """
{
__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
]
}
