| name | detecting-broken-object-property-level-authorization |
| description | 检测和测试OWASP API3:2023对象属性级授权失效(BOPLA)漏洞,包括过度数据暴露和批量赋值攻击。
|
| domain | cybersecurity |
| subdomain | api-security |
| tags | ["api-security","bopla","owasp-api3","mass-assignment","excessive-data-exposure","property-level-authorization","api-testing","penetration-testing"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
检测对象属性级授权失效
概述
对象属性级授权失效(Broken Object Property Level Authorization,BOPLA)被OWASP API安全Top 10归类为API3:2023,结合了两类相关漏洞:过度数据暴露(API返回超出所需的数据)和批量赋值(Mass Assignment,API接受超出预期的数据)。即使API正确地执行了对象级授权,也可能无法控制用户对对象特定属性的读写权限。攻击者利用此漏洞从API响应中读取敏感属性,或向请求体中注入额外属性来修改其无权访问的字段。
前置条件
- 目标API需有返回或接受对象数据的端点
- API文档或模式(优先使用OpenAPI规范)
- Burp Suite或Postman用于API请求操控
- 具有不同权限级别的多个用户账户
- Python 3.8+及requests库用于自动化测试
- 已获得执行安全测试的授权
漏洞模式
过度数据暴露
API返回的对象属性超出客户端所需:
{
"id": 123,
"username": "john_doe",
"email": "john@example.com",
"name": "John Doe",
"ssn": "123-45-6789",
"salary": 95000,
"internal_notes": "VIP client",
"password_hash": "$2b$12...",
"role": "admin",
"created_by": "system_admin",
"credit_card_last4": "4242"
}
批量赋值
API未过滤地将客户端提供的数据绑定到内部对象属性:
// 普通用户更新请求
PUT /api/v1/users/123
Content-Type: application/json
{
"name": "John Updated",
"email": "new@example.com",
"role": "admin", // 攻击者注入:权限提升
"is_verified": true, // 攻击者注入:绕过验证
"discount_rate": 100, // 攻击者注入:业务逻辑滥用
"account_balance": 999999 // 攻击者注入:金融欺诈
}
测试方法
"""BOPLA漏洞扫描器
测试API是否存在对象属性级授权失效(BOPLA)漏洞,
包括过度数据暴露和批量赋值。
"""
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", , , ,
, , , ,
, , , , ,
],
: [
, , , , ,
, , ,
]
}
MASS_ASSIGNMENT_FIELDS = [
(, ),
(, ),
(, ),
(, ),
(, ),
(, ),
(, ),
(, ),
(, [, , ]),
(, ),
(, ),
(, ),
]
():
.base_url = base_url.rstrip()
.auth_headers = auth_headers
.findings: [BOPLAFinding] = []
() -> [BOPLAFinding]:
findings = []
url =
:
response = requests.get(url, headers=.auth_headers, timeout=)
response.status_code != :
findings
data = response.json()
objects = data (data, ) [data]
(data, ) data:
objects = data[] (data[], ) [data[]]
obj objects[:]:
(obj, ):
response_fields = (._flatten_keys(obj))
unexpected_fields = response_fields - expected_fields
field_name unexpected_fields:
severity = ._classify_sensitivity(field_name)
severity:
finding = BOPLAFinding(
endpoint=endpoint,
method=,
vulnerability_type=,
severity=severity,
property_name=field_name,
details=
)
findings.append(finding)
.findings.append(finding)
(requests.exceptions.RequestException, json.JSONDecodeError):
findings
() -> [BOPLAFinding]:
findings = []
url =
original_data :
:
response = requests.get(url, headers=.auth_headers, timeout=)
response.status_code == :
original_data = response.json()
:
original_data = {}
(requests.exceptions.RequestException, json.JSONDecodeError):
original_data = {}
field_name, injected_value .MASS_ASSIGNMENT_FIELDS:
field_name original_data:
original_value = original_data[field_name]
original_value == injected_value:
test_data = deepcopy(original_data)
test_data[field_name] = injected_value
headers = {**.auth_headers, : }
:
method == :
response = requests.put(url, json=test_data,
headers=headers, timeout=)
method == :
response = requests.patch(url, json={field_name: injected_value},
headers=headers, timeout=)
method == :
response = requests.post(url, json=test_data,
headers=headers, timeout=)
response.status_code (, , ):
verify_response = requests.get(url, headers=.auth_headers, timeout=)
verify_response.status_code == :
updated_data = verify_response.json()
updated_data.get(field_name) == injected_value:
finding = BOPLAFinding(
endpoint=endpoint,
method=method,
vulnerability_type=,
severity= field_name [, , ]
,
property_name=field_name,
details=
)
findings.append(finding)
.findings.append(finding)
field_name original_data:
restore_data = {field_name: original_data[field_name]}
requests.patch(url, json=restore_data,
headers=headers, timeout=)
requests.exceptions.RequestException:
findings
() -> [BOPLAFinding]:
findings = []
url =
introspection =
:
response = requests.post(
url,
json={: introspection},
headers=.auth_headers,
timeout=
)
response.status_code == :
data = response.json()
data:
finding = BOPLAFinding(
endpoint=graphql_endpoint,
method=,
vulnerability_type=,
severity=,
property_name=,
details=
)
findings.append(finding)
.findings.append(finding)
requests.exceptions.RequestException:
findings
() -> []:
keys = []
key, value obj.items():
full_key = prefix key
keys.append(full_key)
(value, ):
keys.extend(._flatten_keys(value, full_key))
keys
() -> []:
lower_name = field_name.lower().split()[-]
severity, patterns .SENSITIVE_PROPERTY_PATTERNS.items():
pattern patterns:
pattern lower_name:
severity.upper()
() -> :
{
: (.findings),
: {
: ([f f .findings
f.vulnerability_type == ]),
: ([f f .findings
f.vulnerability_type == ]),
},
: {
: ([f f .findings f.severity == ]),
: ([f f .findings f.severity == ]),
: ([f f .findings f.severity == ]),
: ([f f .findings f.severity == ]),
},
: [
{
: f.endpoint,
: f.method,
: f.vulnerability_type,
: f.severity,
: f.property_name,
: f.details,
}
f .findings
]
}
修复措施
class UserSerializer:
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}
WRITABLE_FIELDS = {'name', 'email', 'phone', 'avatar_url', 'preferences'}
def update_user(user_id, request_data, requesting_user):
safe_data = {k: v for k, v in request_data.items() if k in WRITABLE_FIELDS}
User.objects.filter(id=user_id).update(**safe_data)
参考资料