鲁班 - 系统设计大全

60.6 数据安全与权限

一句话

数据权限的本质是让查询引擎替你做过滤，而不是「让应用记得加 WHERE」；表权限只挡住入口，行/列权限才决定数据出得去多少。

权限有六个层次，不是一层

层 控制什么 实现位置 典型绕过
存储 / 文件 能不能直接读对象 / 文件 bucket policy、KMS、临时凭证 预签名 URL 外泄、bucket 公开
Catalog / 库 能不能建表、删表、切换库 Metastore / Unity / Glue / Ranger 有库权限就 CREATE TABLE AS SELECT 读全库
表 能不能 SELECT / INSERT / ALTER 引擎 ACL、catalog grant 表权限给了，文件仍可直读
列 能不能看这一列、看到的是不是明文 列级 grant、masking policy、视图 用 SELECT * 或函数包一层绕过
行 能看哪些行 row filter、安全视图、RLS 应用层漏写 WHERE、参数可篡改
导出 / 共享 能不能下载、外发、共享 导出审批、水印、共享 recipient 有查询权就有批量导出权

🔑 表权限回答「能不能进这个房间」，行/列权限回答「进去后能看哪几个抽屉」。只做表权限的系统，一次越权就是整表泄露。

模型：RBAC + 数据范围，标签驱动策略

权限模型本身的主场在 38.3 授权 (RBAC / ABAC / 权限模型) 与 49.3 权限中心与权限矩阵，这里只讲数据平台的特殊性：

特殊性 后果 做法
资源是「表 + 列 + 行」 功能权限不够，必须数据权限 四维矩阵：主体 × 资源 × 操作 × 范围
数据有分级 同一张表不同列保护强度不同 分级标签驱动 mask / 审批 / 加密
查询是 Ad-hoc 无法为每个查询预先授权 引擎级策略 + 安全视图
分析要聚合 聚合结果可能反推个体 小分组抑制、差分隐私（见 38.5 数据分级与脱敏）
多租户共享表 行级隔离必须强制 tenant_id 来自会话，不接受请求参数

⚠️ 「角色名判定」是数据权限腐化的起点：if role == 'analyst' 写死在代码里，新增一个「外部审计」角色就要改 N 个服务。判定点应该是 can(subject, action, resource)，变化落在策略数据里。

实现位置：五种做法，安全性差一个数量级

做法 安全性 性能 可维护 绕过风险 适用
应用层拼 WHERE 低 好 差（N 处复制） 漏一处、注入、聚合/Join 绕过 临时脚本，不用于生产
数据库安全视图 中高 中 中 用户仍能访问基表就失效 单库、只读分析
引擎 row filter / column mask 高 中（谓词下推） 好（集中策略） 引擎漏洞、策略未覆盖 Hive/Spark/Trino 生产首选
湖仓 catalog 策略 高 好（存储层裁剪） 好 直读对象存储则失效 湖仓（UC / Lake Formation）
查询网关统一改写 高 差（多一跳） 中 网关被绕过 多引擎、统一审计

这类「拼接 + 未校验」的组合也是 SQL 注入的温床，攻击面见 38.6 常见攻击 (注入 / XSS / CSRF / SSRF / 越权 / 重放)。应用层拼接为什么不可靠？看下面的实测：租户条件 tenant_id='t1' 后面直接拼用户输入，传入 amount_cents > 0 OR 1=1，4 行全部返回——功能权限没出错，行权限被一个 OR 穿透。

代码：结构化查询 + 策略编译

#!/usr/bin/env python3
"""行级 + 列级权限：结构化查询 + 策略编译（SQLite，无第三方依赖）。"""
import sqlite3
from dataclasses import dataclass, field

DDL = """
CREATE TABLE orders (
  order_id TEXT, tenant_id TEXT, user_id TEXT, amount_cents INTEGER,
  phone TEXT, status TEXT, dt TEXT
);
INSERT INTO orders VALUES
 ('o1','t1','u1',10000,'13800000001','paid','2026-10-01'),
 ('o2','t1','u2',20000,'13800000002','paid','2026-10-01'),
 ('o3','t2','u3',30000,'13900000003','paid','2026-10-02'),
 ('o4','t2','u4',40000,'13900000004','refunded','2026-10-02');
"""

MASK = "substr(phone,1,3) || '****' || substr(phone,8,4)"
LAST4 = "'*******' || substr(phone,8,4)"

# 策略：行范围 + 允许的列 + 脱敏表达式
POLICIES = {
    "analyst": {"row": "tenant", "cols": {"order_id", "dt", "amount_cents", "status"},
                "masks": {"phone": MASK}},
    "support": {"row": "tenant", "cols": {"order_id", "dt", "status", "phone"},
                "masks": {"phone": LAST4}},
    "finance": {"row": "tenant", "cols": {"order_id", "dt", "amount_cents", "status"},
                "masks": {"phone": "NULL"}},
    "admin":   {"row": "all", "cols": {"order_id", "tenant_id", "dt", "amount_cents", "status"},
                "masks": {"phone": MASK}},
}

EXPR = {"phone": MASK, "order_id": "order_id", "tenant_id": "tenant_id", "dt": "dt",
        "amount_cents": "amount_cents", "status": "status"}


@dataclass
class Query:
    columns: list[str]
    filters: list[tuple[str, str, object]] = field(default_factory=list)
    limit: int = 100


class Denied(Exception):
    pass


class Engine:
    def __init__(self, con):
        self.con = con
        self.audit = []

    def compile(self, role, session_tenant, q: Query):
        p = POLICIES.get(role)
        if not p:
            raise Denied(f"未知角色 {role}")
        if not q.columns:
            raise Denied("必须显式指定列（禁止 SELECT *）")
        for c in q.columns:
            if c not in p["cols"] and c not in p["masks"]:
                raise Denied(f"角色 {role} 无权访问列 {c}")
        for c, _, _ in q.filters:
            # 只允许在「未脱敏的可见列」上过滤：按手机号等值过滤会成为探测预言机
            if c not in p["cols"]:
                raise Denied(f"角色 {role} 无权按列 {c} 过滤")

        select = [p["masks"].get(c, EXPR[c]) + f" AS {c}" for c in q.columns]
        sql = f"SELECT {', '.join(select)} FROM orders"
        where, params = [], []
        if p["row"] == "tenant":
            where.append("tenant_id = ?")
            params.append(session_tenant)          # 租户来自会话，不接受调用方传参
            for c, op, v in q.filters:
                if op not in ("=", ">", "<", ">=", "<="):
                    raise Denied(f"不支持的操作符 {op}")
                where.append(f"{c} {op} ?")
                params.append(v)
        else:
            for c, op, v in q.filters:
                if op not in ("=", ">", "<", ">=", "<="):
                    raise Denied(f"不支持的操作符 {op}")
                where.append(f"{c} {op} ?")
                params.append(v)
        if where:
            sql += " WHERE " + " AND ".join(where)
        sql += " LIMIT ?"
        params.append(min(q.limit, 1000))
        return sql, params

    def run(self, role, session_tenant, q: Query):
        sql, params = self.compile(role, session_tenant, q)
        rows = self.con.execute(sql, params).fetchall()
        self.audit.append({"role": role, "tenant": session_tenant,
                           "rows": len(rows), "cols": q.columns, "sql": sql})
        return rows


def main():
    con = sqlite3.connect(":memory:")
    con.executescript(DDL)
    e = Engine(con)

    print("=== A. 应用层拼接（错误示范）===")
    user_input = "amount_cents > 0 OR 1=1"
    leak = con.execute("SELECT order_id, tenant_id FROM orders WHERE tenant_id='t1' AND "
                       + user_input).fetchall()
    print(f"  t1 分析师传入 {user_input!r} -> 返回 {len(leak)} 行：{leak}")

    print("\n=== B. 策略编译：同样的过滤，值走参数绑定 ===")
    rows = e.run("analyst", "t1", Query(["order_id", "amount_cents", "phone"],
                                        [("amount_cents", ">", "0 OR 1=1")]))
    print(f"  返回 {len(rows)} 行（注入串被当成字符串值）")

    print("\n=== 12 条验收用例 ===")
    cases = [
        ("analyst", "t1", Query(["order_id", "amount_cents", "phone"]), 2, "138****0001"),
        ("analyst", "t2", Query(["order_id", "amount_cents", "phone"]), 2, "139****0003"),
        ("analyst", "t1", Query(["order_id"], [("amount_cents", ">", 15000)]), 1, None),
        ("analyst", "t1", Query(["phone"]), 2, "138****0002"),
        ("support", "t1", Query(["order_id", "phone"]), 2, "*******0001"),
        ("finance", "t1", Query(["order_id", "amount_cents", "phone"]), 2, None),
        ("admin",   "t1", Query(["order_id", "tenant_id", "phone"]), 4, "139****0004"),
        ("admin",   "t1", Query(["order_id"], [("status", "=", "paid")]), 3, None),
        ("analyst", "t1", Query(["tenant_id"]), "DENY", None),
        ("analyst", "t1", Query(["*"]), "DENY", None),
        ("analyst", "t1", Query(["order_id"], [("phone", "=", "13800000001")]), "DENY", None),
        ("intern",  "t1", Query(["order_id"]), "DENY", None),
    ]
    passed = 0
    for role, tenant, q, want, phone in cases:
        try:
            got = e.run(role, tenant, q)
            n = len(got)
            ok = (want != "DENY" and n == want and
                  (phone is None or any(r[-1] == phone for r in got)))
            detail = f"{n} 行"
        except Denied as ex:
            ok = want == "DENY"
            detail = f"拒绝：{ex}"
        passed += ok
        print(f"  {'✅' if ok else '❌'} {role:<8} {str(q.columns):<38} 期望 {want} -> {detail}")
    print(f"  通过 {passed}/{len(cases)}")

    print("\n=== finance 角色看到的 phone（无权限列返回 NULL）===")
    fin = e.run("finance", "t1", Query(["order_id", "phone"]))
    print(f"  {[r[1] for r in fin]}")

    print("\n=== 生成的 SQL（analyst）===")
    sql, params = e.compile("analyst", "t1", Query(
        ["order_id", "amount_cents", "phone"], [("amount_cents", ">", 15000)]))
    print("  " + sql)
    print(f"  params={params}")

    print("\n=== 审计日志 ===")
    for a in e.audit[-4:]:
        print(f"  role={a['role']} tenant={a['tenant']} rows={a['rows']} cols={a['cols']}")


main()

本机实测（macOS，Apple M1，Python 3.14.8）：

=== A. 应用层拼接（错误示范）===
  t1 分析师传入 'amount_cents > 0 OR 1=1' -> 返回 4 行：[('o1', 't1'), ('o2', 't1'), ('o3', 't2'), ('o4', 't2')]

=== B. 策略编译：同样的过滤，值走参数绑定 ===
  返回 0 行（注入串被当成字符串值）

=== 12 条验收用例 ===
  ✅ analyst  ['order_id', 'amount_cents', 'phone']  期望 2 -> 2 行
  ✅ analyst  ['order_id', 'amount_cents', 'phone']  期望 2 -> 2 行
  ✅ analyst  ['order_id']                           期望 1 -> 1 行
  ✅ analyst  ['phone']                              期望 2 -> 2 行
  ✅ support  ['order_id', 'phone']                  期望 2 -> 2 行
  ✅ finance  ['order_id', 'amount_cents', 'phone']  期望 2 -> 2 行
  ✅ admin    ['order_id', 'tenant_id', 'phone']     期望 4 -> 4 行
  ✅ admin    ['order_id']                           期望 3 -> 3 行
  ✅ analyst  ['tenant_id']                          期望 DENY -> 拒绝：角色 analyst 无权访问列 tenant_id
  ✅ analyst  ['*']                                  期望 DENY -> 拒绝：角色 analyst 无权访问列 *
  ✅ analyst  ['order_id']                           期望 DENY -> 拒绝：角色 analyst 无权按列 phone 过滤
  ✅ intern   ['order_id']                           期望 DENY -> 拒绝：未知角色 intern
  通过 12/12

=== finance 角色看到的 phone（无权限列返回 NULL）===
  [None, None]

=== 生成的 SQL（analyst）===
  SELECT order_id AS order_id, amount_cents AS amount_cents, substr(phone,1,3) || '****' || substr(phone,8,4) AS phone FROM orders WHERE tenant_id = ? AND amount_cents > ? LIMIT ?
  params=['t1', 15000, 100]

=== 审计日志 ===
  role=support tenant=t1 rows=2 cols=['order_id', 'phone']
  role=finance tenant=t1 rows=2 cols=['order_id', 'amount_cents', 'phone']
  role=admin tenant=t1 rows=4 cols=['order_id', 'tenant_id', 'phone']
  role=admin tenant=t1 rows=3 cols=['order_id']

读数：

用例 结果 说明了什么
应用层 OR 1=1 返回 4 行（跨两个租户） 行权限不能靠字符串拼接
参数化注入串 0 行 值被绑定为字符串，不参与 SQL 语法
analyst 查 t2 2 行（只有 t2） 租户来自会话，查谁只能由身份决定
support 看手机号 *******0001 列级脱敏在 SELECT 里下推，不是前端打码
finance 看手机号 NULL 无权列直接不返回，而不是「返回后隐藏」
按手机号过滤 拒绝 掩码列做等值过滤会成为「号码是否存在」的探测预言机
SELECT * 拒绝 星号会让新增敏感列自动泄露

🔑 三个决定值得抄：① 列必须显式声明，禁止 SELECT *（新增列不会自动泄露）；② 脱敏表达式放在 SELECT 里（引擎执行，客户端拿不到明文）；③ 掩码列不可用于过滤（否则可以二分/枚举探测原值）。

动态脱敏与静态脱敏

动态脱敏 静态脱敏
数据落盘 明文（或加密） 已脱敏
实现 视图 / masking policy / UDF ETL 洗数、测试数据生成
可逆 通常不可逆（令牌化除外） 不可逆
适用 生产查询、按角色展示 测试/开发/外部分析
风险 直读存储、导出、日志仍可能泄露 关联能力丢失，分布失真
性能 每行计算，聚合查询有开销 无额外开销

脱敏的强度与坑（哈希可枚举、k-匿名、l-diversity）在 38.5 数据分级与脱敏 有实测，这里只补数据平台特有的三条：

  1. 一致性：同一手机号在订单表、用户表、日志里必须得到同一个掩码，否则无法关联。做法是用带密钥的确定性 HMAC，而不是随机盐或裸 SHA。
  2. 聚合泄露：GROUP BY 部门 后某组只有 1 人，AVG(工资) 就等于个人工资。小分组要抑制（HAVING count(*) >= k）或加噪。
  3. 日志与外发：查询日志、导出文件、BI 截图都是泄露面；脱敏只做在结果集上，等于没做。

湖仓：表权限不等于文件权限

湖仓的权限有两套边界：

边界 控制 风险
catalog / 表 谁能 SELECT 这张表、看哪些列/行 策略没覆盖新表
存储 / 文件 谁能读 S3/OSS 上的 Parquet 有 bucket 读权限就绕过表策略

最常见的越权：表上配了行过滤，但 analyst 的云账号还能直接 aws s3 cp s3://lake/orders/ ... 把 Parquet 拉走（湖仓的存储与表两套边界见 60.3 数据湖与湖仓一体）。防护是关闭直读：查询只能通过引擎，引擎用短期凭证访问存储；bucket 策略拒绝非引擎角色；预签名 URL 短 TTL + 审批 + 水印。

Ranger 的行过滤 + 列掩码（Hive/Spark 插件，JSON 字段示意，未验证）：

{
  "service": "hive",
  "name": "dw_orders_tenant_filter",
  "resources": {
    "database": {"values": ["dw"]},
    "table": {"values": ["orders"]},
    "column": {"values": ["*"]}
  },
  "rowFilterPolicyItems": [{
    "users": ["analyst"],
    "accesses": [{"type": "select", "isAllowed": true}],
    "rowFilterInfo": {"filterExpr": "tenant_id = get_session_tenant()"}
  }],
  "dataMaskPolicyItems": [{
    "users": ["analyst"],
    "accesses": [{"type": "select", "isAllowed": true}],
    "dataMaskInfo": {"dataMaskType": "MASK_SHOW_LAST_4", "valueExpr": ""}
  }]
}

Snowflake 的 masking + row access policy（语法示意，未验证；版本差异以官方文档为准）：

CREATE MASKING POLICY phone_mask AS (val STRING) RETURNS STRING ->
  CASE
    WHEN CURRENT_ROLE() IN ('SUPPORT') THEN '*******' || RIGHT(val, 4)
    WHEN CURRENT_ROLE() IN ('FINANCE') THEN NULL
    ELSE LEFT(val, 3) || '****' || RIGHT(val, 4)
  END;

ALTER TABLE orders MODIFY COLUMN phone SET MASKING POLICY phone_mask;

CREATE ROW ACCESS POLICY tenant_policy AS (tenant_id STRING) RETURNS BOOLEAN ->
  tenant_id = CURRENT_ROLE();          -- 生产上用会话变量/映射表，别用角色名当租户

ALTER TABLE orders ADD ROW ACCESS POLICY tenant_policy ON (tenant_id);

⚠️ 行过滤谓词会参与分区裁剪。如果策略写成 tenant_id = current_setting('app.tenant')，优化器可能无法下推，导致全表扫描；生产上要把租户列放进分区/排序键，或让引擎支持策略谓词下推。安全策略的性能代价必须提前压测。

权限生命周期：只授不收，等于慢慢全员管理员

阶段 动作 可量化指标
申请 说明用途、范围、有效期，绑定数据分级 无审批单的授权数 = 0
审批 数据 owner + 安全双签（高危数据） 高危授权双签率 100%
授予 按最小范围、最短有效期授予 平均授权范围 / 有效期
使用 记录访问，异常检测 未使用权限占比
复核 每季度由 owner 确认，未确认自动回收 复核完成率 ≥ 95%，回收率 10–30%
回收 离职、转岗、项目结束自动触发 离职账号残留 = 0

🔑 权限的自然演化方向是膨胀：人只会申请，不会主动退。只有定期复核（recertification）能真正减少权限，其他机制都只是在「防止继续增加」。

审计：没有审计的权限等于没有权限

必记字段 用途
主体（用户 / 角色 / 服务） 谁
动作（select / export / grant） 做了什么
资源（库.表.列 / 行范围） 对什么
结果（行数 / 字节数 / 拒绝原因） 看了多少
策略版本 当时生效的是哪条规则
客户端 IP / 应用 从哪来
时间 / 查询指纹 什么时候、什么模式

异常检测的四条规则（简单但有效）：

  1. 跨租户：一个会话在短时间内命中多个 tenant_id；
  2. 批量导出：单次返回行数 / 字节数超过该角色 P99 的 10 倍；
  3. 全表扫描：查询没有分区谓词，且扫描量超过阈值；
  4. 拒绝激增：同一主体连续被拒，可能是试探或配置错误。

审计日志要防篡改（WORM / 独立账号 / 哈希链），保留期按合规要求（常见 6 个月到 3 年）。审计本身的方案见 3.5 审计与取证。

多租户：越权测试要进 CI

测试 做法 通过标准
横向越权 用 t1 会话查 t2 的 ID 0 行或 403
纵向越权 analyst 调 admin 接口 403
参数篡改 改 tenant_id / dept_id 参数 服务端忽略客户端值
掩码绕过 用函数/CAST/LIKE 探测手机号 仍为掩码
导出绕过 分页拉全量、CSV 导出 行数上限 + 审批
直读存储 用云凭证读 Parquet 拒绝

🔑 多租户的底线是：租户 ID 只能来自认证上下文，永远不能来自请求参数。租户隔离的三种模型（独立库 / 共享库 / 混合）见 49.4 多租户权限模型。这一条写进代码评审清单，比任何「注意越权」的培训都有效。

生产落地：指名 → 可抄 → 可验 → 可退

指名：Hadoop/Spark 生态用 Apache Ranger；Trino 用 access control + row filter（或 OPA）；Databricks 用 Unity Catalog 的 row filter / column mask；Snowflake 用 masking policy + row access policy；AWS 湖仓用 Lake Formation LF-Tags。替代品按云选，不要把策略绑死在某个引擎的私有语法里。

可抄（本地可跑的策略编译，见本节代码；产品侧用上面的 Ranger / Snowflake 片段，均未验证）。

可验：本节 12 条用例全绿即最小验收；生产上加两组回归——① 「t1 会话查 t2 数据」必须 0 行；② 「EXPLAIN 看行过滤谓词是否下推」必须出现在扫描节点，否则全表扫描。可退：新策略先 shadow（只记录不放行），对比新旧判定差异；上线后保留 break-glass 角色（双人审批 + 全量审计），策略本身用「关闭开关 + 回滚到上一版」而不是回滚整次发布。

反模式

反模式 后果
应用层拼 WHERE tenant_id=... 漏一处 / 一个 OR 1=1 就跨租户（本节实测 4 行泄露）
允许 SELECT * 新增敏感列自动泄露，策略永远追不上 schema
只做表权限、不做行/列 一次授权等于整表 + 全列
在应用层做掩码，接口返回明文 抓包即拿到全量；日志/导出仍泄露
用角色名做数据范围判定 新角色要改 N 个服务，矩阵画不出来
超管万能且不审计 出事后无法定位，合规直接不通过
表上配了行过滤，对象存储仍可直读 绕过引擎拉 Parquet，策略形同虚设
权限只加不减，从不复核 三年后人人都是管理员；离职账号仍在
用无密钥哈希脱敏手机号 号段空间小，GPU 秒级枚举（见 38.5）

自检

Q1 表权限、列权限、行权限各防什么？为什么只做表权限会出事故？

▸ 答案：表权限防「能不能查这张表」，列权限防「能不能看这一列/看到的是不是明文」，行权限防「能看哪些行」。只做表权限时，任何拿到 SELECT 的人都能看全表所有列与所有租户的行——一次授权就是整表泄露；而数据平台里「分析师需要看订单」通常只应看本租户、不含手机号明文。

Q2 为什么应用层拼 WHERE 不可靠？除了注入，还有哪些绕过方式？

▸ 答案：因为过滤逻辑分散在 N 个接口，漏一处就漏一片；用户输入直接进 SQL 会被 OR 1=1 穿透（本节实测 4 行跨租户）。此外还有：分页/聚合/Join 子查询绕过、导出接口另写一套 SQL、缓存命中他人数据、前端隐藏字段但接口返回全量。可靠做法是把策略放在引擎/网关，应用只传「结构化查询 + 身份」。

Q3 动态脱敏与静态脱敏怎么选？各自最大的风险是什么？

▸ 答案：动态脱敏适合生产查询（按角色实时掩码，数据仍是明文/密文），风险是直读存储、导出、日志仍可能泄露；静态脱敏适合测试/开发/外部分析（落盘即脱敏），风险是关联能力丢失、分布失真、无法回溯。选择判据是「业务是否需要真值/关联」与「是否存在绕过引擎的访问路径」。

Q4 为什么「按手机号等值过滤」要拒绝，即使返回的是掩码？

▸ 答案：因为掩码列上的等值过滤会变成探测预言机：攻击者可以枚举候选号码，观察「有行/无行」来确认某个手机号是否在库中，甚至二分缩小范围。掩码只保护「返回值」，不保护「是否存在」。正确做法是禁止在掩码列上过滤，或只允许基于令牌（token）的精确匹配。

Q5 湖仓里「表权限」和「文件权限」为什么要一起管？

▸ 答案：表权限由 catalog/引擎判定，文件权限由对象存储判定；两者是两套边界。只配表策略，用户仍可用云凭证直接读 Parquet，绕过行过滤与列掩码；只配文件权限，则无法表达「同表不同列/行」的细粒度策略。必须关闭直读、用短期凭证、让引擎成为唯一入口。

Q6 审计日志至少记哪些字段？怎样发现「爬取型越权」？

▸ 答案：主体、角色、动作、资源（库表列/行范围）、返回行数与字节数、拒绝原因、策略版本、客户端 IP、时间与查询指纹。爬取型越权的特征是：短时间访问大量不同资源 ID、返回行数远超该角色基线、跨多个租户、在非工作时间批量导出。用「单会话不同资源数 / 返回行数 / 跨租户数」三个指标做阈值告警。

Q7 多租户越权测试最少要覆盖哪几类？通过标准是什么？

▸ 答案：横向（t1 查 t2）、纵向（低权限调高权限）、参数篡改（改 tenant_id/dept_id）、掩码绕过（函数/CAST/LIKE）、导出绕过（分页/CSV）、直读存储。通过标准：跨租户 0 行或 403；服务端忽略客户端传入的租户/部门参数；掩码在接口与导出里一致；直读对象存储被拒绝。把这些用例放进 CI，每次改权限代码都跑。

关联