某员工修改请求参数就能查看同事薪资深入解析企业系统防御水平越权的5大核心防线
事情是怎么爆出来的
去年年底，有家中型互联网公司的HR部门闹得沸沸扬扬。一个普通运营员工，在午休时闲着没事，打开公司内部的人事系统想查一下自己的考勤记录。系统跳转的时候，他的浏览器地址栏里有一串参数：?emp_id=10086&dept=sales。他随口嘀咕了一句，”这参数看着像是员工ID”，然后出于好奇，就把 10086 改成了隔壁部门同事的工号 10092，刷新页面。
然后他的表情就凝固了。
页面跳转后，他不仅看到了那个同事的基本信息，还看到了薪资明细——底薪、绩效、奖金、补贴，一清二楚。他第一反应是截图保存，然后发了个朋友圈。虽然很快就删了，但截图已经在公司内部传开了。
这件事让公司CTO当场失眠了。一个普通运营员工，没有任何越权操作的技术门槛，仅仅修改了URL里的一个参数，就拿到了本不该看到的敏感数据。而更让人后背发凉的是：这家公司每年的安全预算并不少，买了防火墙、WAF、堡垒机，甚至还请了外部安全团队做过渗透测试，但结果呢？
这个案例不是孤例。据某知名安全研究机构的统计，超过60%的企业系统都存在不同程度的越权漏洞，其中水平越权（Horizontal Privilege Escalation）是最常见也最容易被忽视的类型。员工修改参数就能看同事薪资这种事，在真实的生产环境中屡见不鲜。
今天我们就来把这件事掰开揉碎，不是讲那些教科书式的理论，而是从一个真实发生的场景出发，把这个技术问题的来龙去脉、为什么会出这个问题、以及怎么从根源上堵住，一个一个讲清楚。
先搞明白：越权到底是什么
很多人一听”越权”就觉得是黑客攻击、是黑客帝国那种代码飞舞的画面。其实不是。越权就是一个非常简单、非常朴素的问题：系统没有搞清楚”你是谁”以及”你能看什么”。
水平越权和垂直越权的区别
在技术圈里，越权通常分为两大类：
垂直越权（Vertical Privilege Escalation）——这是”低角色干高角色的事”。比如一个普通用户登录系统后，发现URL里有个参数可以改成 role=admin，然后他就真的获得了管理员权限。这种情况相对少见，因为现代框架很少让角色这样的关键信息直接暴露在参数里。
水平越权（Horizontal Privilege Escalation）——这才是本文讨论的重点。这是”同等角色的人，去干了别人的事”。员工A和员工B都是普通员工，权限等级完全一样。但员工A通过修改请求参数，访问到了员工B的数据。这就是水平越权。
回到开头那个案例：运营员工和 sales 同事在系统里的权限等级是一样的，都是普通员工。但运营员工通过修改 emp_id 参数，访问到了 sales 同事的薪资数据。权限没有升级，但数据越过了边界。这就是典型的水平越权。
为什么水平越权特别容易中招
水平越权之所以泛滥，根源在于开发者的思维惯性。很多系统在写权限校验逻辑的时候，开发者脑子里想的是：”这个接口需要登录，没登录的访问不了。”至于”登录了的人能不能看别人的数据”，往往被当成理所当然的事——因为”都是普通员工，看别人数据也没关系吧”。
但”没关系”只是一个假设，不是一个技术实现。系统不会自动知道谁的数据归谁管，边界必须被显式地写出来。
漏洞是怎么产生的：从代码层面看本质
光讲概念不够，我们把事情还原到代码层面。那个员工能看到同事薪资，系统在后台到底经历了什么？
一个典型的漏洞代码
假设人事系统后端有一个接口，用来查询员工薪资：
GET /api/employee/salary?emp_id=10092
Authorization: Bearer [员工的JWT token]
后端处理的伪代码可能长这样：
@app.route('/api/employee/salary', methods=['GET'])
def get_salary():
# 1. 验证用户是否登录
user = verify_token(request.headers['Authorization'])
if not user:
return jsonify({"error": "未登录"}), 401
# 2. 从请求参数中获取要查询的员工ID
target_emp_id = request.args.get('emp_id')
# 3. 查询数据库并返回结果
salary_data = db.query("SELECT * FROM salary WHERE emp_id = ?", target_emp_id)
return jsonify(salary_data)
看到了吗？整个代码里没有任何一句”判断当前用户是否有权限查看目标员工数据”的逻辑。 只要登录了，传什么 emp_id 就查什么。这就是漏洞的核心所在。
更隐蔽的变种：基于资源ID的越权
有些系统会稍微”聪明”一点，不再直接传员工ID，而是传资源ID：
GET /api/records/salary/SAL-2024-001
然后后端根据资源ID去关联查询：
@app.route('/api/records/salary/<record_id>', methods=['GET'])
def get_salary_by_record(record_id):
user = verify_token(request.headers['Authorization'])
# 根据记录ID查询薪资数据
record = db.query("SELECT * FROM salary_records WHERE record_id = ?", record_id)
# 关联员工信息
emp = db.query("SELECT * FROM employees WHERE emp_id = ?", record.emp_id)
return jsonify({
"salary": record.salary,
"name": emp.name,
"dept": emp.dept
})
表面上看，这次没有直接暴露员工ID，似乎更安全了。但实际上，攻击者只需要知道 record_id 的生成规律（比如按顺序递增），就可以逐个遍历：SAL-2024-001、SAL-2024-002、SAL-2024-003……一样能遍历到所有同事的薪资。
这就是为什么很多系统”做了一堆安全措施，还是被越权了”的根本原因——安全边界没有被正确地定义和实现。
五大核心防线：从根上堵住越权
好，问题说清楚了，漏洞也找到了。接下来就是重点：怎么防？以下这五大防线，是按照防御深度和有效性排序的，层层递进，每一条都需要认真地落下去。
第一道防线：身份认证到授权校验的完整链路
这是最基础也是最重要的一道防线。很多系统的问题出在：认证做足了，授权漏掉了。
认证 vs 授权：两个不同的东西
- 认证（Authentication）：确认”你是谁”。比如用户名密码登录、OAuth令牌验证。
- 授权（Authorization）：确认”你能干什么”。比如你能访问哪些资源、能执行哪些操作。
上面那个漏洞代码里，认证是做了的（验证了JWT），但授权根本没做。这就好比你进了公司大楼，保安验证了你的工牌（认证），但没检查你有没有进特定楼层的权限（授权），然后你就可以随便进任何楼层。
正确的做法：在接口层做资源级授权
@app.route('/api/employee/salary', methods=['GET'])
def get_salary():
# 认证：验证用户身份
user = verify_token(request.headers['Authorization'])
if not user:
return jsonify({"error": "未登录"}), 401
# 授权：验证用户是否有权限访问目标资源
target_emp_id = request.args.get('emp_id')
# 核心检查：当前用户只能查看自己的薪资，或者是有权限的HR
if user.emp_id != target_emp_id and not user.has_role('HR'):
return jsonify({"error": "无权限访问此数据"}), 403
# 权限校验通过后，再查询数据
salary_data = db.query("SELECT * FROM salary WHERE emp_id = ?", target_emp_id)
return jsonify(salary_data)
关键就在这一行：
if user.emp_id != target_emp_id and not user.has_role('HR'):
用当前登录用户的身份，去校验目标资源的所有者是否匹配。 这就是授权校验的核心逻辑。
更规范的做法：用声明式权限注解
对于大型系统，手动在每个接口里写授权逻辑既不高效也不容易维护。更规范的做法是使用声明式的权限注解：
from flask import Flask
from flask_security import auth_required, roles_required, current_user
import functools
def requires_ownership_or_role(resource_owner_field):
"""自定义权限装饰器：要求用户是资源所有者或拥有指定角色"""
def decorator(f):
@functools.wraps(f)
def decorated_function(*args, **kwargs):
target_id = kwargs.get('emp_id') or request.args.get('emp_id')
target_user = db.query("SELECT owner_id FROM salary WHERE emp_id = ?", target_id)
if current_user.emp_id != target_user.owner_id and not current_user.has_role('HR'):
abort(403) # 权限不足
return f(*args, **kwargs)
return decorated_function
return decorator
@app.route('/api/employee/salary/<emp_id>', methods=['GET'])
@auth_required('token')
@requires_ownership_or_role('owner_id')
def get_salary(emp_id):
salary_data = db.query("SELECT * FROM salary WHERE emp_id = ?", emp_id)
return jsonify(salary_data)
这样做的好处是：权限逻辑和业务逻辑分离，权限规则可以集中管理、统一审计。当公司 policy 变化时（比如”部门经理可以看本部门所有人的薪资”），只需要修改权限规则，不用动业务代码。
第二道防线：参数绑定到上下文的一致性校验
第一道防线解决的是”该不该让看”的问题，第二道防线解决的是”参数对不对”的问题。
问题出在哪里
很多系统在接收参数时，直接拿前端传过来的值去查数据库，没有做任何一致性校验。比如：
# 危险的做法：直接用请求参数
target_emp_id = request.args.get('emp_id')
这样做有一个隐含假设：请求参数是可信的。但实际上，请求参数来自用户浏览器，可以被任何人随意修改。
正确的做法：从用户上下文获取目标ID
最安全的做法是，不要用请求参数来确定”查谁的数据”，而是从当前用户的服务端上下文中获取：
@app.route('/api/employee/salary', methods=['GET'])
@auth_required('token')
def get_salary():
# 从JWT token中直接提取当前用户ID，而不是从请求参数中获取
current_user_id = current_user.emp_id # 这个值来自服务端签发的token
# 固定查询当前用户自己的薪资
salary_data = db.query("SELECT * FROM salary WHERE emp_id = ?", current_user_id)
return jsonify(salary_data)
这样即使攻击者在请求参数里传了 ?emp_id=10092，后端也完全忽略这个参数，只查 token 里绑定的用户自己的数据。
当确实需要查询他人数据时
当然，有些场景下用户确实需要查询他人的数据（比如HR查员工信息）。这时候的参数传递必须经过授权校验，而且授权校验必须基于服务端可信上下文，而不是客户端传递的参数：
@app.route('/api/hr/employee/salary', methods=['GET'])
@auth_required('token')
@roles_required('HR') # 必须有HR角色
def get_salary_by_hr():
# 参数：想查谁的薪资
target_emp_id = request.args.get('emp_id')
# 校验：参数是否合法（不是空的、是有效员工ID）
if not target_emp_id or not is_valid_emp_id(target_emp_id):
return jsonify({"error": "无效的员工ID"}), 400
# 查询并返回
salary_data = db.query("SELECT * FROM salary WHERE emp_id = ?", target_emp_id)
return jsonify(salary_data)
关键区别在于：参数的有效性校验在服务端完成，而且参数只用于业务查询，不用于权限判断。权限判断的依据是用户角色（@roles_required('HR')），角色信息来自服务端签发的token，客户端无法伪造。
第三道防线：统一网关层的路由权限拦截
如果一个系统有几十个、上百个接口，在每个接口上都手动写授权逻辑，既不现实也容易遗漏。这时候就需要一道更宏观的防线：在网关层统一做权限拦截。
为什么需要网关层拦截
想象一下，你的系统架构是这样的：
前端 → API网关 → 后端服务集群
API网关是所有请求的入口。在这里做统一的权限校验，相当于在大门上安排了一个安检员，所有进出的人都要经过检查。
网关层权限校验的实现思路
在API网关层，可以基于以下维度做权限拦截：
# 网关权限配置示例（简化表示）
routes:
- path: /api/employee/salary/*
methods: [GET]
auth: required
policy:
# 规则：用户只能访问自己的薪资，或具有HR角色
allow_if:
- claim.emp_id == path.param.emp_id # 自己查自己
- claim.roles contains "HR" # HR查所有人
- path: /api/hr/employee/*
methods: [GET, POST, PUT]
auth: required
policy:
# 规则：必须有HR角色才能访问
allow_if:
- claim.roles contains "HR"
- path: /api/admin/*
methods: [GET, POST, PUT, DELETE]
auth: required
policy:
# 规则：必须有ADMIN角色才能访问
allow_if:
- claim.roles contains "ADMIN"
网关在转发请求到后端服务之前，先读取请求中的用户token，解析出用户身份和角色，然后根据路由配置的策略做判断：
# 网关权限拦截器伪代码
class PermissionInterceptor:
def handle_request(self, request):
# 1. 解析token，获取用户信息
token = request.headers.get('Authorization')
user_info = jwt.decode(token)
# 2. 获取当前请求的路由和HTTP方法
route = request.path
method = request.method
# 3. 查找匹配的权限规则
rule = permission_policy.find(rule_path=route, method=method)
# 4. 执行权限校验
if not self.evaluate_rule(user_info, rule):
return Response(status=403, body={"error": "权限不足"})
# 5. 校验通过，将用户信息注入请求头，转发给后端
request.headers['X-User-ID'] = user_info['emp_id']
request.headers['X-User-Roles'] = ','.join(user_info['roles'])
return self.forward(request)
def evaluate_rule(self, user_info, rule):
for condition in rule.allow_if:
if self.evaluate_condition(user_info, condition):
return True
return False
这样做有几个显著的好处：
- 集中管理：所有权限规则在一个地方配置，方便审计和变更
- 统一拦截：不会有机接口漏掉权限校验
- 后端减负：后端服务不需要重复写权限逻辑，只关注业务
- 可观测性：所有权限拦截事件都可以被记录和监控
网关层拦截的局限性
网关层拦截虽然强大，但它也有局限性：
- 细粒度授权不足：网关层适合做粗粒度的角色权限判断，但对于”同一个HR，只能看自己部门员工数据”这种细粒度规则，网关层处理能力有限
- 依赖路由规范：如果后端接口路径命名不规范，网关的权限规则配置会变得非常复杂
所以，网关层拦截和接口层授权不是二选一的关系，而是互补的关系。网关做第一道粗筛，接口做第二道精筛。
第四道防线：数据库层面的行级权限控制
前面三道防线都是在应用层做的，第四道防线下沉到数据库层，做最后一道安全保障。
为什么要到数据库层再做一次
即使应用层出现了bug、网关配置被误改、或者有人绕过网关直接访问数据库，数据库层的权限控制仍然可以提供最后一道保护。这就像银行的金库——即使有人闯过了门口保安、红外线警报、金属探测门，金库本身的保险柜仍然是一道屏障。
PostgreSQL的RLS（Row Level Security）
以PostgreSQL为例，行级安全策略可以这样配置：
-- 开启薪资表的行级安全
ALTER TABLE salary ENABLE ROW LEVEL SECURITY;
-- 创建策略：用户只能看到自己的薪资记录
CREATE POLICY employee_self_access ON salary
FOR ALL
TO authenticated_users
USING (emp_id = current_setting('app.current_user_id')::integer)
WITH CHECK (emp_id = current_setting('app.current_user_id')::integer);
-- 创建策略：HR可以看到所有薪资记录
CREATE POLICY hr_full_access ON salary
FOR ALL
TO hr_role
USING (true)
WITH CHECK (true);
关键在 current_setting('app.current_user_id') —— 这个值由应用层在建立数据库连接时设置：
# 应用层建立数据库连接时设置当前用户上下文
connection = db.create_connection()
connection.execute("SET app.current_user_id = ?", current_user.emp_id)
这样，即使有SQL注入或者直接访问数据库的情况发生，行级安全策略也会拦截非法的数据访问。数据在哪层被保护，就在哪层形成边界。
MySQL 8.0+ 的类似方案
MySQL虽然没有原生的RLS功能，但可以通过视图和触发器实现类似效果：
-- 创建只允许查看当前用户薪资的视图
CREATE VIEW v_salary_accessible AS
SELECT * FROM salary
WHERE emp_id = @current_user_id;
-- 应用层设置会话变量
SET @current_user_id = 10086;
-- 之后所有查询都通过视图进行，自动带上了权限过滤
SELECT * FROM v_salary_accessible;
虽然不如PostgreSQL的RLS优雅，但原理是一致的。
第五道防线：审计日志与异常行为监控
前面四道防线是”防守”，第五道防线是”监控”。即使前面的防线都被突破了，审计日志和异常行为监控也能做到及时发现、快速响应。
审计日志要记什么
很多系统的日志只记了”谁在什么时候访问了接口”，这是不够的。有效的越权审计日志应该包含：
{
"timestamp": "2024-12-15T10:23:45Z",
"event_type": "AUTHORIZATION_FAILURE",
"user_id": "10086",
"user_role": "OPERATOR",
"requested_resource": "/api/employee/salary",
"requested_target_id": "10092",
"target_owner_id": "10092",
"action": "GET",
"result": "DENIED",
"ip_address": "192.168.1.105",
"user_agent": "Mozilla/5.0 ...",
"reason": "User 10086(OPERATOR) attempted to access salary data owned by 10092 without sufficient privileges"
}
注意几个关键字段：
requested_target_idvstarget_owner_id：通过对比这两个字段，可以快速发现越权尝试result：记录是成功还是被拦截，用于后续分析reason：详细的拒绝原因，方便安全团队排查
异常行为监控
有了审计日志还不够，需要实时监控异常行为：
class AnomalyDetector:
def __init__(self):
self.query_history = {} # 记录每个用户的查询行为
def check_anomaly(self, user_id, target_id, action):
"""检查当前操作是否异常"""
history = self.query_history.get(user_id, [])
# 规则1：短时间内查询大量不同用户的数据
recent_queries = [h for h in history if time_since(h.timestamp) < 300]
unique_targets = set(h.target_id for h in recent_queries)
if len(unique_targets) > 10:
return True, f"用户{user_id}在5分钟内查询了{len(unique_targets)}个不同用户的数据"
# 规则2：查询的数据owner与自己相差很大（可能是遍历攻击）
current_user = get_user_by_id(user_id)
if target_id and is_valid_emp_id(target_id):
target_user = get_user_by_id(target_id)
if target_user and current_user.dept != target_user.dept:
return True, f"用户{user_id}跨部门访问了用户{target_id}的数据"
return False, ""
监控规则可以根据业务场景定制，核心思路是找到”正常用户不会做的事”。
五大防线的协同：一张完整的防御图
单独看每一道防线都有用，但真正的安全来自多层防线的协同。
┌─────────────────────────────┐
│ 第五层：审计与监控 │
│ • 异常行为检测 │
│ • 越权尝试实时告警 │
│ • 安全事件追溯 │
└──────────────┬──────────────┘
│ 发现异常时反向影响上层规则
┌──────────────▼──────────────┐
│ 第四层：数据库行级安全 │
│ • RLS策略 │
│ • 视图权限过滤 │
│ • 直接DB访问的最后一道屏障 │
└──────────────┬──────────────┐
│
┌──────────────▼──────────────┐
│ 第三层：API网关拦截 │
│ • 路由级权限规则 │
│ • 统一认证授权入口 │
│ • 请求量监控 │
└──────────────┬──────────────┘
│
┌──────────────▼──────────────┐
│ 第二层：参数一致性校验 │
│ • 服务端上下文替代请求参数 │
│ • 参数合法性验证 │
│ • 上下文绑定 │
└──────────────┬──────────────┘
│
┌──────────────▼──────────────┐
│ 第一层：接口级授权校验 │
│ • 资源所有权校验 │
│ • 角色权限校验 │
│ • 业务逻辑内的细粒度控制 │
└─────────────────────────────┘
五层防线各司其职，又相互补充：
- 第一层是基础，每个接口都应该有授权校验
- 第二层解决了参数篡改问题，确保”查谁”不是由客户端决定的
- 第三层是宏观防线，统一管控、集中配置
- 第四层是兜底，即使应用层全部失效，数据库层仍然保护数据
- 第五层是检测和响应，发现攻击、追溯攻击路径
那些你以为做了、其实没做的”伪安全”
讲完了正确的做法，再说几种常见的错误做法。这些做法在现实中非常普遍，很多公司以为自己做足了安全，实际上漏洞依然存在。
伪安全一：只做了认证，没做授权
很多系统的权限逻辑是这样的：
@app.route('/api/employee/salary')
def get_salary():
if not is_login(): # 只检查是否登录
return unauthorized()
emp_id = request.args.get('emp_id')
return query_salary(emp_id)
“已登录就能看”——这是最典型的伪安全。登录只证明了你是合法用户，不代表你有权限看所有数据。
伪安全二：用前端控制权限显示
有些系统在前端做了权限控制：
// 前端代码：如果不是HR，隐藏"查看他人薪资"按钮
if (currentUser.role !== 'HR') {
hideButton('view_others_salary');
}
前端隐藏按钮≠后端没有接口。攻击者直接调API就行：
curl -H "Authorization: Bearer <token>" \
"http://api.example.com/api/employee/salary?emp_id=10092"
前端是给用户看的，后端才是真正管数据的。任何权限控制都必须落实到后端，前端控制只能作为辅助。
伪安全三：用UUID替代数字ID，以为就安全了
有些开发者觉得用数字ID太容易被遍历，改用UUID：
# 之前：emp_id=10092（容易被猜到下一个是10093）
# 现在：emp_id=a3f2b1c4-d5e6-7890-abcd-ef1234567890（猜不到）
UUID确实让遍历变得困难，但遍历困难不等于无法访问。如果系统仍然接受请求参数中的UUID来查询数据，攻击者只需要拿到一个有效的UUID（比如从泄露的日志、错误信息、或者前端页面中获取），就可以查询对应员工的所有数据。
UUID解决的是”枚举攻击”的问题，解决不了”越权访问”的问题。这两件事是相关的，但不是一回事。
伪安全四：用隐藏参数做权限控制
# 把用户ID放到请求体中，而不是URL参数
POST /api/employee/salary
{
"emp_id": 10092
}
URL参数和请求体参数，在安全性上没有任何本质区别。攻击者用Postman或者curl都可以发送请求体。把敏感数据从URL移到body，只是换了个地方暴露，安全性没有提升。
一个完整的实战案例：从零搭建防越权系统
光说不练假把式。我们用上面那个薪资系统的例子，从零搭建一个完整的、具备五大防线防护的系统。
第一步：数据库设计
-- 员工表
CREATE TABLE employees (
emp_id VARCHAR(20) PRIMARY KEY,
name VARCHAR(100) NOT NULL,
dept VARCHAR(50) NOT NULL,
role VARCHAR(20) NOT NULL, -- OPERATOR, HR, ADMIN
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 薪资表
CREATE TABLE salaries (
emp_id VARCHAR(20) PRIMARY KEY,
base_salary DECIMAL(10, 2) NOT NULL,
performance_bonus DECIMAL(10, 2) DEFAULT 0,
allowance DECIMAL(10, 2) DEFAULT 0,
total DECIMAL(10, 2) NOT NULL,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (emp_id) REFERENCES employees(emp_id)
);
-- 审计日志表
CREATE TABLE auth_audit_logs (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
user_id VARCHAR(20) NOT NULL,
user_role VARCHAR(20) NOT NULL,
action VARCHAR(50) NOT NULL,
resource_type VARCHAR(50) NOT NULL,
resource_id VARCHAR(50) NOT NULL,
result ENUM('ALLOW', 'DENIED') NOT NULL,
reason TEXT,
ip_address VARCHAR(45),
user_agent VARCHAR(500),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_user_time (user_id, created_at),
INDEX idx_resource (resource_type, resource_id)
);
第二步：API网关层（Python + Flask）
from flask import Flask, request, jsonify, g
from flask_jwt_extended import JWTManager, create_access_token, decode_token
import hashlib
import logging
app = Flask(__name__)
jwt = JWTManager(app)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 权限规则配置
PERMISSION_RULES = {
'/api/employee/salary': {
'methods': ['GET'],
'auth': True,
'policy': 'self_or_hr'
},
'/api/hr/employee/salary': {
'methods': ['GET', 'POST', 'PUT'],
'auth': True,
'policy': 'hr_only'
},
'/api/admin/*': {
'methods': ['GET', 'POST', 'PUT', 'DELETE'],
'auth': True,
'policy': 'admin_only'
}
}
def check_gateway_permission(user_info, path, method):
"""网关层权限检查"""
rule = find_matching_rule(path)
if not rule:
return True, "无匹配规则，放行"
if rule['auth'] and not user_info:
return False, "需要认证但未登录"
if rule['policy'] == 'self_or_hr':
# 允许用户访问自己的数据，HR可以访问所有
if user_info['role'] == 'HR':
return True, "HR角色，允许访问"
# 对于self_or_hr策略，具体资源级别的检查在后端做
# 网关只做角色层面的粗筛
return True, "通过网关角色检查，后端细粒度校验"
if rule['policy'] == 'hr_only' and user_info['role'] != 'HR':
return False, "需要HR角色"
if rule['policy'] == 'admin_only' and user_info['role'] != 'ADMIN':
return False, "需要ADMIN角色"
return True, "网关层检查通过"
def find_matching_rule(path):
"""查找匹配的路由规则"""
for pattern, rule in PERMISSION_RULES.items():
if pattern.endswith('/*'):
prefix = pattern[:-2]
if path.startswith(prefix):
return rule
elif path == pattern:
return rule
return None
@app.before_request
def gateway_auth_check():
"""在请求到达业务逻辑之前，先做网关层权限检查"""
path = request.path
method = request.method
# 获取token中的用户信息
user_info = None
auth_header = request.headers.get('Authorization', '')
if auth_header.startswith('Bearer '):
token = auth_header.replace('Bearer ', '')
try:
decoded = jwt.decode_token(token)
user_info = {
'emp_id': decoded['emp_id'],
'role': decoded['role'],
'name': decoded['name']
}
g.current_user = user_info # 存入请求上下文，供后续使用
except:
user_info = None
# 网关层权限检查
allowed, reason = check_gateway_permission(user_info, path, method)
if not allowed:
log_audit_event(
user_id=user_info['emp_id'] if user_info else 'unknown',
user_role=user_info['role'] if user_info else 'anonymous',
action=f"{method} {path}",
resource_type=path.split('/')[2] if len(path.split('/')) > 2 else 'unknown',
resource_id='gateway_check',
result='DENIED',
reason=reason,
ip=request.remote_addr,
user_agent=request.headers.get('User-Agent', '')
)
return jsonify({"error": "权限不足", "detail": reason}), 403
def log_audit_event(user_id, user_role, action, resource_type, resource_id, result, reason, ip, user_agent):
"""记录审计日志"""
try:
db.execute("""
INSERT INTO auth_audit_logs
(user_id, user_role, action, resource_type, resource_id, result, reason, ip_address, user_agent)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", user_id, user_role, action, resource_type, resource_id, result, reason, ip, user_agent)
db.commit()
except Exception as e:
logger.error(f"审计日志记录失败: {e}")
第三步：业务层接口（带细粒度授权）
from functools import wraps
from flask_jwt_extended import get_jwt_identity
def require_ownership_or_role(required_roles=None):
"""细粒度授权装饰器"""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
current_user = g.current_user
# 先检查角色
if required_roles:
if current_user['role'] not in required_roles:
log_audit_event(
user_id=current_user['emp_id'],
user_role=current_user['role'],
action=f"{request.method} {request.path}",
resource_type='salary',
resource_id=kwargs.get('emp_id', 'unknown'),
result='DENIED',
reason=f"角色{current_user['role']}不在允许列表{required_roles}中",
ip=request.remote_addr,
user_agent=request.headers.get('User-Agent', '')
)
return jsonify({"error": "角色权限不足"}), 403
# 再检查资源所有权（对于需要所有权校验的接口）
target_emp_id = kwargs.get('emp_id') or request.args.get('emp_id')
if target_emp_id and target_emp_id != current_user['emp_id']:
# 目标不是自己，检查是否是HR
if current_user['role'] != 'HR':
log_audit_event(
user_id=current_user['emp_id'],
user_role=current_user['role'],
action=f"{request.method} {request.path}",
resource_type='salary',
resource_id=target_emp_id,
result='DENIED',
reason=f"员工{current_user['emp_id']}尝试访问员工{target_emp_id}的薪资数据",
ip=request.remote_addr,
user_agent=request.headers.get('User-Agent', '')
)
return jsonify({"error": "无权限访问此数据"}), 403
return f(*args, **kwargs)
return decorated_function
return decorator
@app.route('/api/employee/salary', methods=['GET'])
@auth_required('token')
@require_ownership_or_role()
def get_employee_salary():
"""员工查看薪资接口"""
# 关键点：不从请求参数中获取目标ID，而是从用户上下文中获取
target_emp_id = g.current_user['emp_id']
salary = db.query("SELECT * FROM salaries WHERE emp_id = ?", target_emp_id)
if not salary:
return jsonify({"error": "薪资数据不存在"}), 404
log_audit_event(
user_id=g.current_user['emp_id'],
user_role=g.current_user['role'],
action='GET /api/employee/salary',
resource_type='salary',
resource_id=target_emp_id,
result='ALLOW',
reason='',
ip=request.remote_addr,
user_agent=request.headers.get('User-Agent', '')
)
return jsonify(salary)
@app.route('/api/hr/employee/salary/<emp_id>', methods=['GET'])
@auth_required('token')
@require_ownership_or_role(required_roles=['HR'])
def get_salary_by_hr(emp_id):
"""HR查看指定员工薪资"""
salary = db.query("SELECT * FROM salaries WHERE emp_id = ?", emp_id)
if not salary:
return jsonify({"error": "薪资数据不存在"}), 404
log_audit_event(
user_id=g.current_user['emp_id'],
user_role=g.current_user['role'],
action=f'GET /api/hr/employee/salary/{emp_id}',
resource_type='salary',
resource_id=emp_id,
result='ALLOW',
reason='',
ip=request.remote_addr,
user_agent=request.headers.get('User-Agent', '')
)
return jsonify(salary)
第四步：数据库行级安全（PostgreSQL RLS）
-- 在PostgreSQL中启用薪资表的行级安全
ALTER TABLE salaries ENABLE ROW LEVEL SECURITY;
-- 策略1：普通员工只能看到自己的薪资
CREATE POLICY salary_self_access ON salaries
FOR ALL
USING (emp_id = current_setting('app.current_emp_id')::VARCHAR)
WITH CHECK (emp_id = current_setting('app.current_emp_id')::VARCHAR);
-- 策略2：HR角色可以看到所有薪资
CREATE POLICY salary_hr_access ON salaries
FOR ALL
TO hr_role
USING (true)
WITH CHECK (true);
-- 应用层连接数据库时设置当前用户上下文
-- 每个数据库连接建立时执行：
-- SET app.current_emp_id = '<当前登录员工ID>';
第五步：异常行为监控
class AccessAnomalyMonitor:
"""访问异常监控"""
def __init__(self):
self.user_access_history = {} # 用户访问历史
def record_access(self, user_id, target_id, action, result):
"""记录访问行为"""
if user_id not in self.user_access_history:
self.user_access_history[user_id] = []
self.user_access_history[user_id].append({
'target_id': target_id,
'action': action,
'result': result,
'timestamp': datetime.now()
})
# 清理超过1小时的记录
self._clean_old_records(user_id)
def check_anomaly(self, user_id):
"""检查是否有异常行为"""
history = self.user_access_history.get(user_id, [])
if len(history) < 5:
return False, ""
# 规则1：短时间内访问大量不同用户的数据
recent = [h for h in history if (datetime.now() - h['timestamp']).total_seconds() < 300]
unique_targets = set(h['target_id'] for h in recent)
if len(unique_targets) > 10:
return True, f"用户{user_id}在5分钟内访问了{len(unique_targets)}个不同用户的数据"
# 规则2：大量访问被拒绝
denied_count = sum(1 for h in recent if h['result'] == 'DENIED')
if denied_count > 5:
return True, f"用户{user_id}在5分钟内{denied_count}次访问被拒绝"
return False, ""
def _clean_old_records(self, user_id):
"""清理旧记录"""
cutoff = datetime.now() - timedelta(hours=1)
self.user_access_history[user_id] = [
h for h in self.user_access_history[user_id]
if h['timestamp'] > cutoff
]
# 使用示例
monitor = AccessAnomalyMonitor()
@app.route('/api/hr/employee/salary/<emp_id>', methods=['GET'])
@auth_required('token')
@require_ownership_or_role(required_roles=['HR'])
def get_salary_by_hr(emp_id):
current_user = g.current_user
# 检查异常行为
is_anomaly, anomaly_reason = monitor.check_anomaly(current_user['emp_id'])
if is_anomaly:
logger.warning(f"检测到异常访问: {anomaly_reason}")
# 可以触发告警、临时封禁等操作
return jsonify({"error": "检测到异常访问行为，请联系管理员"}), 429
salary = db.query("SELECT * FROM salaries WHERE emp_id = ?", emp_id)
# 记录访问
monitor.record_access(
current_user['emp_id'],
emp_id,
'GET_salary',
'ALLOW' if salary else 'NOT_FOUND'
)
if not salary:
return jsonify({"error": "薪资数据不存在"}), 404
return jsonify(salary)
防越权的安全开发生命周期建议
技术防线部署好了，只是第一步。更重要的是在开发流程中建立防越权的安全意识。
1. 需求阶段：明确数据权限模型
在产品设计阶段，就要明确：
- 哪些数据是敏感数据（薪资、身份证号、健康信息等）
- 谁可以访问这些数据（角色模型）
- 在什么条件下可以访问（所有权、部门归属、特定授权）
把这些写进PRD，而不是开发过程中临时想起来。
2. 设计阶段：权限设计评审
每次新增接口，必须经过权限设计评审。评审 checklist：
□ 这个接口访问的是什么数据？是否敏感？
□ 谁应该能访问这个接口？（角色模型）
□ 是否有资源级别的权限限制？（比如只能看自己的）
□ 参数中是否有可被篡改的标识符？（ID、资源名等）
□ 是否需要在网关层做统一拦截？
□ 是否需要记录审计日志？
□ 是否有异常行为监控的需求？
3. 开发阶段：代码审查重点
代码审查时，重点关注：
- 所有涉及敏感数据的接口，是否有授权校验
- 授权校验的依据是服务端可信上下文，还是客户端传来的参数
- 是否有硬编码的权限判断（比如
if emp_id == 10086这种）
4. 测试阶段：渗透测试覆盖
安全测试中，越权是必测项：
- 登录账号A，访问账号B的资源
- 修改请求参数中的ID，观察是否能访问到其他用户数据
- 尝试遍历资源ID，看是否能枚举出所有数据
- 尝试提升角色（修改token中的role字段）
5. 运维阶段：持续监控
上线后持续监控：
- 审计日志的异常模式
- 越权拦截的频次和趋势
- 权限配置变更的审计
写在最后
回到开头那个案例，运营员工修改参数看同事薪资这件事，本质上不是”技术有多高超的攻击”，而是”系统有多基本的疏忽”。
安全领域有句话：“安全是一个过程，不是一个产品。” 你买再贵的防火墙，如果代码层面的授权逻辑没写好，漏洞依然存在。五大防线不是五份独立的文档，而是一个层层递进的防御体系——每一层都在补上一层的漏洞，每一层都在增加攻击者的成本。
对于企业管理者来说，最重要的是建立一种安全左移的思维：不要等问题爆出来了再修，而是在设计、开发、测试的每一个环节，都把”这个人有没有权限做这件事”这个问题想清楚、写下来、验证通过。
对于开发者来说，最重要的是记住一个简单的原则：永远不要信任客户端传来的任何标识符作为权限判断的依据。 用户ID、资源ID、角色信息——这些都应该来自服务端签发的token或上下文，而不是请求参数。
防越权这件事，看似是技术问题，其实是管理问题。技术防线搭好了，需要制度去落实，需要文化去维护。但无论如何，迈出第一步总是好的——毕竟，那个在午休时间改了一下URL参数的运营员工，可能并不是 malicious 的，他只是好奇。而一个安全的系统，不应该让好奇心成为泄露敏感信息的通道。