水平越权在Web应用中的常见成因与代码级防御策略解析

说实话，水平越权（Horizontal Privilege Escalation，简称 HPE）是Web安全圈里那种“听起来高大上，做起来却让人想摔键盘”的问题。它不像SQL注入那样有着明确的 exploit 路径，也不像XSS那样能直接弹窗秀存在感。水平越权更像是一个隐形的漏洞，攻击者不需要提权到管理员，只需要在“同级用户”之间偷偷摸摸地互换身份，就能拿到别人的数据，甚至干点更坏的事。

我记得刚入行那会儿，有一次帮一家创业公司做安全审计。他们的产品挺不错，用户可以在上面存储个人笔记、照片，还能分享给朋友。功能看着挺温馨，结果我在测试的时候，随手改了个请求里的 user_id，好家伙，直接看到了另一个用户的私密照片。那一刻，我深刻体会到：水平越权不是理论上的风险，它是真实存在、随时可能爆发的定时炸弹。

今天，我们就把这个话题掰开揉碎了讲清楚。我会从成因、代码示例、防御策略几个维度，用大白话配合实际代码，让你彻底搞懂这个问题。

一、水平越权到底是什么？

首先，咱们得把概念理清。权限越权分两种：

垂直越权（Vertical Privilege Escalation）：普通用户把自己变成管理员，比如一个普通员工通过漏洞拿到了 admin 权限。
水平越权（Horizontal Privilege Escalation）：用户A在保持自己身份不变的情况下，访问了用户B的数据。

水平越权的核心是“横向窃取”，而不是“纵向提升”。攻击者通常不会去改自己的角色，而是直接操作目标资源，比如修改请求中的 ID，以为服务端会信任这个 ID。

举个例子：

用户 A 的订单 ID 是 ORDER-1001
用户 B 的订单 ID 是 ORDER-1002
如果接口 /api/orders/ORDER-1002 没有校验当前用户是否有权访问这个订单，那么 A 就能直接看到 B 的订单信息。

这就是水平越权。

二、常见成因：为什么会有水平越权？

水平越权几乎都源于同一个根本问题：服务端没有校验当前请求的用户是否有权访问目标资源。

具体来说，常见的成因有以下几种：

1. 依赖前端传来的 ID，而不是服务端会话中的用户身份

很多开发者会这样写代码：

@app.route('/api/orders/<order_id>')
def get_order(order_id):
    # 从 session 里拿到当前登录用户
    current_user = session.get('user')
    
    # 直接根据前端传来的 order_id 查询数据库
    order = Order.query.get(order_id)
    
    # 没有校验这个 order 是否属于 current_user！
    return jsonify(order.to_dict())

你看，这里 current_user 明明已经存在了，但代码完全没用上。攻击者只要把 URL 里的 order_id 改成别人的，就能拿到数据。

2. 使用不安全的 ID 设计

有些系统用自增 ID，比如 user_id 从 1 开始递增。攻击者可以简单地遍历 ID，尝试访问 user_id=2, user_id=3 等等。虽然这不是典型的水平越权（因为没改请求），但它给了攻击者很大的便利去尝试越权访问。

更危险的是，如果 ID 是可预测的（比如基于时间的哈希），攻击者更容易推断出其他用户的资源 ID。

3. 在查询时没有加入用户过滤条件

这是最隐蔽的问题。开发者可能在写 SQL 查询时，忘了加 WHERE user_id = ? 条件：

String sql = "SELECT * FROM orders WHERE order_id = ?";
PreparedStatement stmt = connection.prepareStatement(sql);
stmt.setInt(1, orderId);
ResultSet rs = stmt.executeQuery();

这里只查了 order_id，没有关联当前用户。正确做法应该是：

String sql = "SELECT * FROM orders WHERE order_id = ? AND user_id = ?";
PreparedStatement stmt = connection.prepareStatement(sql);
stmt.setInt(1, orderId);
stmt.setInt(2, currentUserId);  // 必须加入当前用户 ID 过滤
ResultSet rs = stmt.executeQuery();

4. 使用全局变量或静态变量存储用户状态

在一些老旧的代码里，开发者可能会用静态变量来保存当前用户：

public static class GlobalContext
{
    public static int CurrentUserId { get; set; }
}

然后在控制器里直接使用：

public IActionResult GetOrder(int orderId)
{
    int orderId = GlobalContext.CurrentUserId; // 危险！没有校验订单归属
    var order = _orderService.GetOrder(orderId);
    return Json(order);
}

这种写法在多线程环境下极易出错，而且完全没有做资源归属校验。

5. API 设计缺陷：缺少对象级权限控制

现代 Web 应用大量使用 RESTful API，但有些 API 设计只考虑了资源的路径，没有考虑资源的访问权限。比如：

GET /api/users/123/profile

这个接口返回用户 123 的个人资料。但如果没有任何权限检查，任何登录用户都能访问任意用户的资料。

三、代码级防御策略：如何彻底解决这个问题？

防御水平越权，核心原则就一句话：永远不要信任客户端传来的数据，所有的资源访问都必须与服务端会话中的用户身份绑定。

下面我分几个层面来讲解具体的防御策略。

策略一：在每一处资源访问时，强制校验用户归属

这是最基础也是最重要的一步。无论你在哪个语言、哪个框架，都必须保证：查询资源时，必须带上当前用户的 ID 作为过滤条件。

Python (Flask) 示例

from flask import Flask, session, jsonify
from models import Order

app = Flask(__name__)

@app.route('/api/orders/<int:order_id>')
def get_order(order_id):
    # 1. 确保用户已登录
    if 'user_id' not in session:
        return jsonify({"error": "未登录"}), 401
    
    current_user_id = session['user_id']
    
    # 2. 查询订单时，同时绑定当前用户 ID
    order = Order.query.filter_by(
        id=order_id,
        user_id=current_user_id
    ).first()
    
    # 3. 如果查不到，说明该订单不属于当前用户，拒绝访问
    if not order:
        return jsonify({"error": "无权访问该订单"}), 403
    
    return jsonify(order.to_dict())

注意看 filter_by(id=order_id, user_id=current_user_id) 这一行。这是关键！ 即使攻击者改了 order_id，只要这个订单不属于当前用户，查询结果就是空的。

Java (Spring Boot) 示例

@RestController
@RequestMapping("/api/orders")
public class OrderController {

    @Autowired
    private OrderService orderService;

    @GetMapping("/{orderId}")
    public ResponseEntity<?> getOrder(@PathVariable Long orderId) {
        // 1. 从安全上下文获取当前用户 ID
        Long currentUserId = SecurityContextHolder.getCurrentUserId();
        
        if (currentUserId == null) {
            return ResponseEntity.status(401).body("未登录");
        }
        
        // 2. 查询时同时传入用户 ID
        Order order = orderService.getOrderByIdAndUserId(orderId, currentUserId);
        
        // 3. 如果订单不存在，说明无权访问
        if (order == null) {
            return ResponseEntity.status(403).body("无权访问该订单");
        }
        
        return ResponseEntity.ok(order);
    }
}

对应的 Service 层：

@Service
public class OrderService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    // 查询时强制绑定用户 ID
    public Order getOrderByIdAndUserId(Long orderId, Long userId) {
        return orderRepository.findByIdAndUserId(orderId, userId);
    }
}

对应的 Repository：

@Repository
public interface OrderRepository extends JpaRepository<Order, Long> {
    
    // 关键：查询条件必须包含 userId
    Order findByIdAndUserId(Long id, Long userId);
}

Node.js (Express) 示例

const express = require('express');
const router = express.Router();
const Order = require('../models/Order');

router.get('/api/orders/:orderId', async (req, res) => {
    // 1. 从 session 或 JWT 中获取当前用户 ID
    const currentUserId = req.user.id; // 假设中间件已经解析了用户信息
    
    if (!currentUserId) {
        return res.status(401).json({ error: '未登录' });
    }
    
    try {
        // 2. 查询时绑定用户 ID
        const order = await Order.findOne({
            where: {
                id: req.params.orderId,
                userId: currentUserId
            }
        });
        
        // 3. 如果订单不存在，拒绝访问
        if (!order) {
            return res.status(403).json({ error: '无权访问该订单' });
        }
        
        res.json(order);
    } catch (error) {
        res.status(500).json({ error: '服务器错误' });
    }
});

策略二：使用权限中间件统一拦截

如果你在每个接口都写一遍 user_id 校验，代码会非常冗余，而且容易漏掉。更好的做法是使用中间件或装饰器，统一做权限校验。

Python (Flask) 使用装饰器

from functools import wraps
from flask import session, jsonify

def require_ownership(resource_model, id_param):
    """
    装饰器：校验当前用户是否拥有指定资源的访问权
    :param resource_model: 资源模型类
    :param id_param: 请求中的 ID 参数名
    """
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if 'user_id' not in session:
                return jsonify({"error": "未登录"}), 401
            
            current_user_id = session['user_id']
            resource_id = kwargs.get(id_param)
            
            # 查询资源，同时绑定用户 ID
            resource = resource_model.query.filter_by(
                id=resource_id,
                user_id=current_user_id
            ).first()
            
            if not resource:
                return jsonify({"error": "无权访问该资源"}), 403
            
            # 将资源对象传递给原函数
            kwargs['resource'] = resource
            return f(*args, **kwargs)
        return decorated_function
    return decorator

# 使用示例
@app.route('/api/orders/<int:order_id>')
@require_ownership(Order, 'order_id')
def get_order(order_id, resource):
    # resource 已经是当前用户有权访问的订单
    return jsonify(resource.to_dict())

这样，你只需要在需要权限校验的接口上加上装饰器，就能统一保证所有资源访问都有正确的权限控制。

Java (Spring Boot) 使用 AOP 或拦截器

@Aspect
@Component
public class OwnershipAspect {

    @Around("@annotation(RequireOwnership)")
    public Object checkOwnership(ProceedingJoinPoint joinPoint) throws Throwable {
        // 1. 获取当前用户 ID
        Long currentUserId = SecurityContextHolder.getCurrentUserId();
        if (currentUserId == null) {
            throw new AuthenticationException("未登录");
        }
        
        // 2. 获取方法参数中的资源 ID
        Object[] args = joinPoint.getArgs();
        Long resourceId = null;
        for (Object arg : args) {
            if (arg instanceof Long) {
                resourceId = (Long) arg;
                break;
            }
        }
        
        if (resourceId == null) {
            throw new IllegalArgumentException("缺少资源 ID");
        }
        
        // 3. 查询资源，绑定用户 ID
        // 这里假设有一个通用的资源查询服务
        boolean hasAccess = ownershipService.checkOwnership(resourceId, currentUserId);
        if (!hasAccess) {
            throw new AccessDeniedException("无权访问该资源");
        }
        
        // 4. 放行请求
        return joinPoint.proceed();
    }
}

使用自定义注解：

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface RequireOwnership {
}

然后在控制器方法上使用：

@RequireOwnership
@GetMapping("/api/orders/{orderId}")
public ResponseEntity<?> getOrder(@PathVariable Long orderId) {
    // 这里不再需要手动校验权限，AOP 已经处理了
    Order order = orderService.getOrder(orderId);
    return ResponseEntity.ok(order);
}

策略三：使用 UUID 或随机 ID，避免 ID 可预测

如果资源 ID 是自增整数，攻击者可以轻易遍历。使用 UUID 或随机生成的 ID 可以大大增加攻击者的猜测难度。

生成 UUID 示例

import uuid

class Order:
    def __init__(self, user_id):
        self.id = str(uuid.uuid4())  # 生成随机 UUID
        self.user_id = user_id
        # ... 其他字段

这样，攻击者就无法通过遍历 1, 2, 3... 来尝试访问其他用户的资源。

Java 中使用 UUID

import java.util.UUID;

public class Order {
    private String id;
    private Long userId;
    
    public Order(Long userId) {
        this.id = UUID.randomUUID().toString();
        this.userId = userId;
    }
    
    // getter and setter...
}

策略四：实施最小权限原则

即使你做完了上述所有校验，也应该遵循最小权限原则：只返回必要的字段，不提供额外的信息。

比如，查询订单时，不要返回订单的完整详情，只返回用户需要知道的部分：

@app.route('/api/orders/<int:order_id>')
@require_ownership(Order, 'order_id')
def get_order(order_id, resource):
    # 只返回必要字段，避免信息泄露
    return jsonify({
        "order_id": resource.id,
        "status": resource.status,
        "total_amount": resource.total_amount
        # 不要返回支付信息、地址等敏感字段
    })

策略五：记录审计日志

对于敏感的资源访问，记录审计日志可以帮助你在发生越权事件时快速定位问题。

import logging

logger = logging.getLogger(__name__)

@app.route('/api/orders/<int:order_id>')
@require_ownership(Order, 'order_id')
def get_order(order_id, resource):
    logger.info(f"用户 {session['user_id']} 访问了订单 {order_id}")
    return jsonify(resource.to_dict())

如果发现某个用户频繁访问不存在的订单 ID，这可能是攻击者在尝试越权。

四、实际案例：一个完整的防御示例

让我们用一个完整的例子来演示如何防御水平越权。假设我们有一个博客系统，用户可以发布文章，其他用户可以查看文章。

数据库设计

CREATE TABLE articles (
    id INT PRIMARY KEY AUTO_INCREMENT,
    title VARCHAR(255) NOT NULL,
    content TEXT,
    user_id INT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (user_id) REFERENCES users(id)
);

Python (Flask) 完整实现

from flask import Flask, session, jsonify, request
from flask_sqlalchemy import SQLAlchemy
import uuid

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog.db'
db = SQLAlchemy(app)

class Article(db.Model):
    id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    title = db.Column(db.String(255), nullable=False)
    content = db.Column(db.Text, nullable=False)
    user_id = db.Column(db.Integer, nullable=False)
    
    def to_dict(self):
        return {
            "id": self.id,
            "title": self.title,
            "content": self.content,
            "user_id": self.user_id,
            "created_at": str(self.created_at)
        }

# 权限校验装饰器
def require_ownership(model, id_param):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if 'user_id' not in session:
                return jsonify({"error": "未登录"}), 401
            
            current_user_id = session['user_id']
            resource_id = kwargs.get(id_param)
            
            # 查询资源，绑定用户 ID
            resource = model.query.filter_by(
                id=resource_id,
                user_id=current_user_id
            ).first()
            
            if not resource:
                return jsonify({"error": "无权访问该文章"}), 403
            
            kwargs['resource'] = resource
            return f(*args, **kwargs)
        return decorated_function
    return decorator

@app.route('/api/articles/<article_id>')
@require_ownership(Article, 'article_id')
def get_article(article_id, resource):
    # 只返回必要字段，不返回 user_id（避免信息泄露）
    return jsonify({
        "id": resource.id,
        "title": resource.title,
        "content": resource.content,
        "created_at": str(resource.created_at)
    })

@app.route('/api/articles', methods=['POST'])
def create_article():
    if 'user_id' not in session:
        return jsonify({"error": "未登录"}), 401
    
    data = request.get_json()
    article = Article(
        title=data['title'],
        content=data['content'],
        user_id=session['user_id']
    )
    db.session.add(article)
    db.session.commit()
    
    return jsonify(article.to_dict()), 201

攻击测试

假设用户 A

正文

水平越权在Web应用中的常见成因与代码级防御策略解析

一、水平越权到底是什么？

二、常见成因：为什么会有水平越权？

1. 依赖前端传来的 ID，而不是服务端会话中的用户身份

2. 使用不安全的 ID 设计

3. 在查询时没有加入用户过滤条件

4. 使用全局变量或静态变量存储用户状态

5. API 设计缺陷：缺少对象级权限控制

三、代码级防御策略：如何彻底解决这个问题？

策略一：在每一处资源访问时，强制校验用户归属

Python (Flask) 示例

Java (Spring Boot) 示例

Node.js (Express) 示例

策略二：使用权限中间件统一拦截

Python (Flask) 使用装饰器

Java (Spring Boot) 使用 AOP 或拦截器

策略三：使用 UUID 或随机 ID，避免 ID 可预测

生成 UUID 示例

Java 中使用 UUID

策略四：实施最小权限原则

策略五：记录审计日志

四、实际案例：一个完整的防御示例

数据库设计

Python (Flask) 完整实现

攻击测试

相关阅读

职场升迁陷阱：水平越权如何影响市场运作与团队和谐

企业如何建立有效的水平越权监控机制，保障信息安全与合规运营

职场晋升难题：水平越权现象揭秘及市场应对策略

如何识别和预防质量管理中的水平越权问题，保障项目顺利进行

企业内部预防水平越权：五大策略保障组织高效运转

揭秘职场越权陷阱：学会辨别与巧妙应对，保护你的权益和职场地位

职场生存指南：警惕水平越权现象，如何避免市场运作受阻

如何避免产品开发中的越权操作，确保项目顺利进行

企业内部防越权攻略：四大步骤守护公司权益

如何避免职场越权导致的市场混乱与损失