# 不要信任任何租户
> 分离所有者和应用角色，锁定模式，构建快速且适配连接池的多租户行级安全，并加固连接。
---
LLMS 索引： [llms.txt](/llms.txt)
---
多数多租户应用靠反复写上一千遍的一行来隔离租户：`where account_id = ?`。它一直有效，直到有人写出第一千零一条查询却忘了这一条件，某个客户的仪表盘于是显示了另一个客户的数据。这不是性能问题，而是需要通报的数据泄露。
Postgres 可以在应用下面强制执行这条规则，让任何人都无法忘记它。这需要一个不拥有数据表的应用角色、有意授予的权限，以及正确且快速的行级安全策略。每一部分都有可能静默失效，我们会指出具体位置。
## 应用永远不以 postgres 连接 {#the-app-never-connects-as-postgres}
Postgres 只有一种主体：角色。“用户”是带有 `LOGIN` 的角色；“组”是不带该属性、由其他角色成为其成员的角色。
```sql
create role wg_owner nologin; -- owns the schema and tables
create role wg_app login password 'app-secret'; -- the application logs in as this
create role wg_readonly nologin; -- a group for read access
```
角色属于整个集簇，而不是某个数据库内部的对象，所以 `pg_dump` 会跳过它们（[pg_dump 不包含角色；单独转储它们](/32-backups/#pg_dump-leaves-out-roles-dump-them-separately)）。
角色属性是功能强大的开关（`SUPERUSER`、`CREATEROLE`、`CREATEDB`、`REPLICATION`、`BYPASSRLS`），应用角色不应拥有其中任何一个。Postgres 最常见的安全错误，是让应用以 `postgres` 或托管服务的管理员用户连接：每一次 SQL 注入都会变成完全沦陷，而且任何策略都不会生效（超级用户会绕过策略；托管管理员通常拥有数据表，也会因此绕过策略）。
把一个角色授予另一个角色后，被授予者就成为成员，并且默认继承被授予角色的权限。在创建新角色之前，先检查预定义角色，例如用于查看 [观察数据库运行](/31-observability/#basics-chapter-31)中那些视图的 `pg_monitor`，以及 `pg_maintain`（PostgreSQL 17+），它无需拥有数据表就能执行 `vacuum` 和 `reindex`。
> [!DETAILS]- 深入阅读：迁移角色如何借用所有者的权限，而不整天携带这些权限
>
> 从 PostgreSQL 16 开始，一次授权带有三个选项：`inherit`（自动使用该角色的权限）、`set`（可以通过 `set role` 切换到该角色）和 `admin`（可以把它授予别人）。因此，迁移角色可以有意成为所有者：
>
> ```sql
> create role wg_migrator login;
> grant wg_owner to wg_migrator with inherit false, set true;
>
> select r.rolname as role, m.rolname as member,
> a.inherit_option, a.set_option, a.admin_option
> from pg_auth_members a
> join pg_roles r on r.oid = a.roleid
> join pg_roles m on m.oid = a.member
> where m.rolname like 'wg_%';
> ```
>
> ```text
> role | member | inherit_option | set_option | admin_option
> -------------+-------------+----------------+------------+--------------
> wg_owner | wg_migrator | f | t | f
> wg_readonly | wg_analyst | t | t | f
> ```
>
> 作为 `wg_migrator` 时，普通查询会失败，直到它执行 `set role wg_owner`：这正是模式变更所需要的“sudo”行为。（`wg_analyst` 是一个登录角色，通过普通的 `grant wg_readonly to wg_analyst` 成为成员。）其他值得了解的预定义角色还有 `pg_read_all_data` 和 `pg_write_all_data`（PostgreSQL 14+），以及用于取消查询的 `pg_signal_backend`。
>
> > [!EXAMPLE]- 动手验证 · lab `32-security`
> >
> > ```sql
> > select lab.prove(
> > 'PostgreSQL 16+ grants carry inherit, set, and admin options: wg_migrator may set role but does not inherit',
> > (select not a.inherit_option and a.set_option and not a.admin_option
> > from pg_auth_members a
> > join pg_roles r on r.oid = a.roleid
> > join pg_roles m on m.oid = a.member
> > where r.rolname = 'wg_owner' and m.rolname = 'wg_migrator'));
> > ```
> >
> > [完整实验脚本](https://www.flagon.io/books/eight-kilobytes/lab/32-security.sql)
>
> 来自 MySQL 的读者会发现：MySQL 中的账户是 `'user'@'host'` 对，角色（8.0+）则是另一个可以授予账户的对象。在 Postgres 中，登录角色、组角色和所有者都是同一种对象，只是属性不同；主机规则位于 `pg_hba.conf` 中（[Postgres 不是 MySQL](/11-not-mysql/#basics-chapter-11)）。
## 应用永远不拥有自己的数据表 {#the-app-never-owns-its-tables}
所有者可以对自己的数据表执行 `alter`、`drop`，并授予任何权限，而且默认会忽略这些表上的行级安全。处理 Web 请求的进程不该拥有这些能力。因此，至少需要两个角色：
| 角色 | 登录 | 使用者 | 能做什么 |
|---|---|---|---|
| 所有者 / 迁移角色 | 仅迁移时 | 部署流水线 | DDL，拥有所有对象 |
| 应用角色 | 是 | 正在运行的应用 | 对获授权限的数据表执行 `select`、`insert`、`update`、`delete` |
| 只读角色（可选） | 通过成员角色 | 分析人员、BI 工具 | `select` |
**Table 33-1.** 应用永远不拥有自己的数据表
所有者可以是 `nologin`，由迁移角色 `set role` 成为它（切换成它，像 sudo 一样，但只用于模式变更）。应用的连接字符串永远不包含所有者的凭据。
### 访问数据表需要两项授权 {#reaching-a-table-takes-two-grants}
模式上的 `USAGE` 允许角色查看其中的对象；访问数据表则需要 `USAGE` 加上数据表权限：
```sql
create schema app authorization wg_owner;
grant usage on schema app to wg_app, wg_readonly;
```
### 只给所有数据表授权，会漏掉下周的数据表 {#grants-on-all-tables-miss-next-weeks-table}
`grant ... on all tables in schema app` 只覆盖当前已经存在的数据表。下周新建的数据表会返回 `permission denied`，于是有人会用“给所有人授予一切权限”来修复它。默认权限可以填上这个缺口：它们会应用于某个角色之后创建的对象。
```sql
set role wg_owner;
alter default privileges in schema app
grant select, insert, update, delete on tables to wg_app;
alter default privileges in schema app
grant select on tables to wg_readonly;
-- serial columns also need: grant usage on sequences to wg_app
-- (identity columns, like ours, don't)
```
现在，`wg_owner` 在 `app` 中创建的每个数据表都会带有正确的授权。本章创建了其中两张：`app.projects` 和 `app.events`，它们是示例表的副本，保留所有行但减少了列（没有 user 或 owner 列）。应用可以执行 select、insert、update 和 delete，仅此而已：不能 truncate，不能 alter，也不能 drop。
> [!DETAILS]- 深入阅读：阅读 \dp 中的授权
>
> ```text
> => \dp app.events
> Schema | Name | Type | Access privileges | ...
> --------+--------+-------+----------------------------+
> app | events | table | wg_owner=arwdDxtm/wg_owner+|
> | | | wg_app=arwd/wg_owner +|
> | | | wg_readonly=r/wg_owner |
> ```
>
> 这些字母分别表示：`a` 为 insert，`r` 为 select，`w` 为 update，`d` 为 delete，`D` 为 truncate，`x` 为 references，`t` 为 trigger，`m` 为 maintain。应用拥有 `arwd`。
>
> > [!EXAMPLE]- 动手验证 · lab `32-security`
> >
> > ```sql
> > select lab.prove(
> > 'tables wg_owner creates in app arrive with grants: wg_owner=arwdDxtm, wg_app=arwd, wg_readonly=r',
> > (select relacl::text[] from pg_class where oid = 'app.events'::regclass)
> > = array['wg_owner=arwdDxtm/wg_owner', 'wg_app=arwd/wg_owner', 'wg_readonly=r/wg_owner']);
> > ```
> >
> > [完整实验脚本](https://www.flagon.io/books/eight-kilobytes/lab/32-security.sql)
> [!WARNING] 默认权限取决于创建对象的角色
>
> `alter default privileges` 只作用于执行该命令的角色所创建的对象（或者 `for role` 指定的角色所创建的对象）。如果迁移由另一个角色执行，新建对象就不会自动带上这些授权。始终以所有者角色创建对象，并在 psql 中用 `\ddp` 检查。
### 较旧的数据库仍允许任何人在 public 中创建对象 {#older-databases-still-let-anyone-create-in-public}
`search_path`（默认值为 `"$user", public`）决定 Postgres 查找未限定名称的位置。多年来，每个角色都可以在 `public` 中创建对象，这带来了经典攻击：在其中放置一个与特权用户会调用的函数同名的函数，然后等待。PostgreSQL 15 对新数据库取消了这一默认权限；除了数据库所有者之外的所有人都只有 `USAGE`：
```text
=> create table public.scratch (id int); -- as wg_app
ERROR: permission denied for schema public
```
升级或从旧版本恢复的数据库会保留旧授权：如果 `\dn+ public` 显示所有人都有 `=UC`，就执行 `revoke create on schema public from public`。对应用模式，在 SQL 中限定名称（`app.events`），或者为角色设置路径：`alter role wg_app set search_path = app`。
## 策略会让其他租户的行不可见 {#a-policy-makes-other-tenants-rows-invisible}
行级安全（RLS）会把**策略**附加到数据表上：Postgres 将布尔表达式加入每一次查询，因此策略不为真的行对读取不可见，也无法写入。
```sql
set role wg_owner;
alter table app.events enable row level security;
create policy tenant_isolation on app.events
using (account_id = current_setting('app.account_id', true)::bigint);
-- and the same on app.projects
```
租户由自定义设置 `app.account_id` 表示，应用在每个请求中设置它；如果从未设置，`current_setting(..., true)` 会返回 null。没有单独的 `with check` 表达式时，`using` 同时控制读取和写入。
> [!NOTE] 来自 MySQL 的读者
>
> MySQL 没有行级安全；隔离由应用代码实现，或者由视图实现，而对基础表的查询可以绕过视图。这里策略附加在数据表上（[Postgres 不是 MySQL](/11-not-mysql/#basics-chapter-11)）。
在 1,000 个账户共有 2,000,000 个事件的情况下，未设置租户的应用角色什么也看不到：
```text
=> select count(*) from app.events; -- as wg_app, no tenant
count
-------
0
```
`account_id = null` 永远不会为真，所以所有行都会被隐藏。RLS 会在失败时拒绝访问，这正是你需要的失败方式。设置租户 42 后，它能看到自己的 2,084 个事件，看不到其他内容。明确指定租户 7 进行查询会返回零行，向租户 7 写入一行则会失败：
```text
ERROR: new row violates row-level security policy for table "events"
```
> [!EXAMPLE]- 动手验证 · lab `32-security`
>
> ```sql
> set role wg_app;
> select lab.prove(
> 'no tenant set: the app role sees 0 of 2,000,000 events (RLS fails closed)',
> (select count(*) from app.events) = 0);
>
> select set_config('app.account_id', '42', false) \g /dev/null
> select lab.prove(
> 'tenant 42 sees its 2,084 events',
> (select count(*) from app.events) = 2084);
> ```
>
> [完整实验脚本](https://www.flagon.io/books/eight-kilobytes/lab/32-security.sql)
RLS 是检查机制；不可猜测的公开 ID 是检查缺失那一天的备用防线（[计数器是在邀请攻击](/13-keys/#a-counter-is-an-invitation)）。
### 外键会穿过策略查看数据 {#foreign-keys-look-through-policies}
外键检查会绕过 RLS，所以租户 42 可以把事件关联到一个自己看不到的项目。修复方法是采用 [让跨租户引用不可能发生](/13-keys/#make-cross-tenant-references-impossible)中介绍的多租户键设计：把租户放进外键，使用 `(account_id, project_id)` 引用 `projects (account_id, id)`，这样被引用的行必须属于同一个租户。
> [!DETAILS]- 深入阅读：观察外键越过租户边界
>
> 在租户 42 的视角下，项目 1 不可见：
>
> ```text
> => select account_id from app.projects where id = 1;
> (0 rows)
> ```
>
> 但插入却能通过：
>
> ```sql
> begin;
> insert into app.events (account_id, project_id, kind)
> values (42, 1, 'deploy') returning id;
> rollback;
> ```
>
> ```text
> id
> ---------
> 2000002
> ```
>
> > [!EXAMPLE]- 动手验证 · lab `32-security`
> >
> > ```sql
> > begin;
> > insert into app.events (account_id, project_id, kind) values (42, 1, 'deploy');
> > select lab.prove(
> > 'yet it can attach an event to project 1: the foreign key check bypasses RLS',
> > (select count(*) from app.events where project_id = 1) = 1);
> > rollback;
> > -- after adding the composite foreign key, the same insert fails with:
> > -- insert or update on table "events" violates foreign key constraint "events_project_fk"
> > ```
> >
> > [完整实验脚本](https://www.flagon.io/books/eight-kilobytes/lab/32-security.sql)
>
> 把租户放入键之后，同一个插入会失败：
>
> ```sql
> alter table app.projects
> add constraint projects_account_id_id_key unique (account_id, id);
> alter table app.events
> add constraint events_project_fk
> foreign key (account_id, project_id) references app.projects (account_id, id);
> ```
>
> ```text
> ERROR: insert or update on table "events" violates foreign key constraint
> "events_project_fk"
> DETAIL: Key is not present in table "projects".
> ```
### 除非强制执行，否则 RLS 会跳过表所有者 {#rls-skips-the-table-owner-unless-you-force-it}
RLS 默认不会应用于数据表所有者。在设置了相同租户的情况下，所有者查询 `app.events` 会返回全部 2,000,000 行。如果应用以所有者身份连接（ORM 的迁移工具共享同一个连接字符串时很常见），RLS 就会静默失效。执行 `alter table app.events force row level security` 后，所有者只能看到 2,084 行。
对每一张租户数据表都强制实施 RLS。它不会增加查询时的开销，却会约束所有以所有者身份运行的任务。所有者执行 `pg_dump` 时会报错退出，而不是转储出一张不完整的表；所有者在没有设置租户时执行回填 `update`，则会静默地修改零行。超级用户和带有 `BYPASSRLS` 的角色仍会跳过策略，因此要为备份以及有意进行的跨租户工作提供专用的 `BYPASSRLS` 角色，并明确使用这一绕过能力（备份部分见 [pg_dump 不包含角色；单独转储它们](/32-backups/#pg_dump-leaves-out-roles-dump-them-separately)）。
> [!EXAMPLE]- 动手验证 · lab `32-security`
>
> ```sql
> set role wg_owner;
> select lab.prove(
> 'the owner, same tenant setting, sees all 2,000,000 rows: RLS does not apply to owners by default',
> current_setting('app.account_id', true) = '42'
> and (select count(*) from app.events) = 2000000);
>
> alter table app.events force row level security;
> select lab.prove(
> 'after force row level security, the owner sees 2,084',
> (select count(*) from app.events) = 2084);
> ```
>
> [完整实验脚本](https://www.flagon.io/books/eight-kilobytes/lab/32-security.sql)
> [!DANGER] RLS 静默失效的三种方式
>
> 应用以超级用户、带有 `BYPASSRLS` 的角色，或者未设置 `force row level security` 的表所有者身份连接。在这三种情况下，查询都会成功并返回每个租户的行。进行测试：使用应用的准确凭据连接，不设置租户，并确认查询租户数据表时返回零行。
### 连接池会泄露租户 {#pooled-connections-leak-tenants}
连接池会把同一个服务器连接依次交给一个又一个请求（[事务池模式会破坏会话状态](/28-connections/#transaction-pooling-breaks-session-state)）。为某个请求设置的会话租户会在请求结束后继续存在；下一个忘记设置租户的请求就会以之前那个租户的身份运行。即使 RLS 完全开启，也会发生跨租户泄露。
规则是：**在事务内设置租户，并把它限定在该事务内。** `set_config` 的第三个参数就是作用域，`true` 表示“直到提交”，效果类似 `set local`：
```sql
begin;
select set_config('app.account_id', '7', true); -- true = transaction-local
select count(*) from app.projects;
commit;
```
有一个陷阱：提交之后，一个没有设置租户的请求得到的不是零行，而是 `invalid input syntax for type bigint: ""`。自定义设置一旦在会话中设置过，之后会恢复成空字符串，而不是“未设置”。访问仍会在失败时被拒绝，但忘记设置租户会变成一个 500 错误。要把空字符串当作缺失值处理：
```sql
alter policy tenant_isolation on app.events
using (account_id = nullif(current_setting('app.account_id', true), '')::bigint);
```
然后让安全路径成为唯一的路径：由一个应用辅助函数开始事务，从绑定参数设置租户，执行请求，再提交事务。
> [!DETAILS]- 深入阅读：实证空字符串，以及 node-postgres 的租户辅助函数
>
> ```sql
> select current_setting('app.account_id', true) is null as is_null,
> current_setting('app.account_id', true) as val;
> ```
>
> ```text
> is_null | val
> ---------+-----
> f |
> ```
>
> > [!EXAMPLE]- 动手验证 · lab `32-security`
> >
> > ```sql
> > select lab.prove(
> > 'after commit, a transaction-local custom setting reverts to an empty string, not to null',
> > current_setting('app.account_id', true) = '');
> > -- the original policy then fails with: invalid input syntax for type bigint: ""
> > -- after alter policy ... nullif(current_setting('app.account_id', true), '')::bigint:
> > select lab.prove(
> > 'with nullif(..., ''''), an unset tenant is null again and the query returns zero rows',
> > (select count(*) from app.projects) = 0 and (select count(*) from app.events) = 0);
> > ```
> >
> > [完整实验脚本](https://www.flagon.io/books/eight-kilobytes/lab/32-security.sql)
>
> 辅助函数使用 node-postgres：
>
> ```ts
> import { Pool, PoolClient } from "pg";
>
> const pool = new Pool(); // direct, or through PgBouncer in transaction mode
>
> export async function withTenant(
> accountId: string,
> fn: (db: PoolClient) => Promise,
> ): Promise {
> const client = await pool.connect();
> try {
> await client.query("begin");
> // true = transaction-local: gone at commit, never seen by the next request
> await client.query(
> "select set_config('app.account_id', $1, true)",
> [accountId],
> );
> const result = await fn(client);
> await client.query("commit");
> return result;
> } catch (err) {
> await client.query("rollback");
> throw err;
> } finally {
> client.release();
> }
> }
>
> // every query for the request runs inside the tenant's transaction
> const { rows } = await withTenant(session.accountId, (db) =>
> db.query("select id, name from app.projects order by name"),
> );
> ```
>
> 租户是绑定参数，绝不能拼接进 SQL。每条查询都通过 `db` 执行；直接使用 `pool` 的查询没有租户，并且由于策略的存在，什么也看不到。应用代码使用 `set_config(..., true)` 比使用 `SET LOCAL` 更合适，因为 `SET` 不能接受绑定参数。
## RLS 性能：策略就是谓词 {#rls-performance-policies-are-predicates}
策略就是 `where` 子句中的另一个条件（[视图和策略只是文本替换](/21-plans/#views-and-policies-are-text-substitution)），因此和其他条件一样需要索引。没有索引时，应用角色统计租户 42 的部署事件数量会读取 33,309 个页面：执行并行顺序扫描，按 `kind` 和策略进行过滤。每张租户表的主要索引都应以租户列开头：
```sql
create index events_account_id_created_at_idx on app.events (account_id, created_at);
```
现在，同一个统计只读取 2,036 个页面，策略也成了索引条件。查询租户最近的 20 个事件只读取 23 个页面，而且查询本身根本不需要提到租户。
让策略保持简单：策略中的子查询会加入对该表的每一次查询。每个请求只确定一次成员关系，并把结果设置为租户。
> [!DETAILS]- 深入阅读：两份计划，以及为什么 current_setting 在索引条件中只运行一次
>
> 加索引之前：
>
> ```text
> Finalize Aggregate (actual rows=1.00 loops=1)
> Buffers: shared hit=10258 read=23051 written=5
> ...
> -> Parallel Seq Scan on events (actual rows=160.67 loops=3)
> Filter: ((kind = 'deploy'::text) AND (account_id =
> (NULLIF(current_setting('app.account_id'::text, true),
> ...))::bigint))
> Rows Removed by Filter: 666506
> ```
>
> 加索引之后：
>
> ```text
> Aggregate (actual rows=1.00 loops=1)
> Buffers: shared hit=685 read=1351
> -> Bitmap Heap Scan on events (actual rows=482.00 loops=1)
> ...
> -> Bitmap Index Scan on events_account_id_created_at_idx
> (actual rows=2084.00 ...)
> Index Cond: (account_id =
> (NULLIF(current_setting('app.account_id'::text, true),
> ''::text))::bigint)
> ```
>
> `current_setting` 是稳定函数，因此在索引条件中 Postgres 会计算它一次，再用这个值探查索引。在顺序扫描的过滤器中，它会对每一行执行。查询最新 20 个事件时，会在同一个索引上执行 `Index Scan Backward`，`Buffers: shared hit=23`。
>
> > [!EXAMPLE]- 动手验证 · lab `32-security`
> >
> > ```sql
> > select lab.prove(
> > 'with an index leading on the tenant column, the policy becomes an index condition: about 2,036 pages',
> > 'Bitmap Index Scan' = any (lab.nodes($$ select count(*) from app.events where kind = 'deploy' $$))
> > and lab.buffers($$ select count(*) from app.events where kind = 'deploy' $$) between 1900 and 2200);
> > ```
> >
> > [完整实验脚本](https://www.flagon.io/books/eight-kilobytes/lab/32-security.sql)
### 不具备 leakproof 属性的函数会在策略之后运行 {#a-function-that-isnt-leakproof-runs-after-the-policy}
Postgres 不会把每个条件都放在策略之前执行。某个函数的错误消息如果包含输入，就可能泄露你看不到的行；因此，任何没有标记为 `LEAKPROOF`（即承诺永远不暴露输入）的函数都会在策略之后运行，也不能在策略之前驱动索引扫描。text、bigint 和 timestamptz 上的 `=` 与 `<` 是 leakproof 的；`like`、`ilike` 和 jsonb 包含操作（`@>`）则不是。
代价是真实存在的。在 `payload` 上有一个 [GIN](/43-glossary/#gin) 索引时，把租户写进查询的包含搜索读取了 434 个页面。通过策略执行时读取了 2,036 个：GIN 索引不被允许使用，所以规划器取出该租户的全部行再过滤。对最大的客户来说，这意味着每次搜索都要扫描它的全部数据。
修复方法是把这个字段提升为带类型的列，并使用具有 leakproof 属性的比较操作符。一个存储型生成列 `duration_ms int`，在 `(account_id, duration_ms)` 上建立索引后，只需四个页面就能回答同一个搜索。添加存储列会在 `ACCESS EXCLUSIVE` 锁下重写数据表，因此对大表要按任何重写操作来规划（[每次变更都是目录编辑、扫描或重写](/26-migrations/#every-change-is-a-catalog-edit-a-scan-or-a-rewrite)）。
要以应用角色读取执行计划：超级用户根本看不到策略谓词。
> [!DETAILS]- 深入阅读：哪些操作符是 leakproof 的，以及显示 GIN 索引被拒绝的计划
>
> ```sql
> select oprname, oprcode, p.proleakproof
> from pg_operator o join pg_proc p on p.oid = o.oprcode
> where (oprleft, oprright) in (('text'::regtype, 'text'::regtype),
> ('bigint'::regtype, 'bigint'::regtype),
> ('jsonb'::regtype, 'jsonb'::regtype),
> ('timestamptz'::regtype, 'timestamptz'::regtype))
> and oprname in ('=', '<', '@>', '~~', '~~*')
> order by 1, 2;
> ```
>
> ```text
> oprname | oprcode | proleakproof
> ---------+----------------+--------------
> < | int8lt | t
> < | text_lt | t
> < | timestamptz_lt | t
> < | jsonb_lt | f
> = | texteq | t
> = | int8eq | t
> = | timestamptz_eq | t
> = | jsonb_eq | f
> @> | jsonb_contains | f
> ~~ | textlike | f
> ~~* | texticlike | f
> ```
>
> 以绕过 RLS 的 `postgres` 身份，并把租户写进查询：`where account_id = 42 and payload @> '{"duration_ms": 1234}'`：
>
> ```text
> Aggregate (actual rows=1.00 loops=1)
> Buffers: shared hit=434
> -> Bitmap Heap Scan on events (actual rows=1.00 loops=1)
> -> BitmapAnd (actual rows=0.00 loops=1)
> -> Bitmap Index Scan on events_payload_idx (actual rows=403.00 ...)
> -> Bitmap Index Scan on events_account_id_created_at_idx
> (actual rows=2084.00 ...)
> ```
>
> 以应用角色身份通过策略执行：`where payload @> '{"duration_ms": 1234}'`：
>
> ```text
> Aggregate (actual rows=1.00 loops=1)
> Buffers: shared hit=2036
> -> Bitmap Heap Scan on events (actual rows=1.00 loops=1)
> Filter: (payload @> '{"duration_ms": 1234}'::jsonb)
> Rows Removed by Filter: 2083
> -> Bitmap Index Scan on events_account_id_created_at_idx
> (actual rows=2084.00 ...)
> ```
>
> > [!EXAMPLE]- 动手验证 · lab `32-security`
> >
> > ```sql
> > set role wg_app;
> > select lab.prove(
> > 'through the policy, the GIN index is not allowed: the planner fetches all of the tenant''s rows and filters',
> > not (lab.plan($$ select count(*) from app.events where payload @> '{"duration_ms": 1234}' $$)::text
> > like '%events_payload_idx%')
> > and lab.buffers($$ select count(*) from app.events where payload @> '{"duration_ms": 1234}' $$)
> > between 1900 and 2200);
> > ```
> >
> > [完整实验脚本](https://www.flagon.io/books/eight-kilobytes/lab/32-security.sql)
>
> 生成列的修复方法。（另一条出路是把函数标记为 `LEAKPROOF`，但这需要超级用户权限，并且需要验证该函数绝不会抛出依赖输入的错误。）
>
> ```sql
> alter table app.events
> add column duration_ms int
> generated always as ((payload->>'duration_ms')::int) stored;
> create index on app.events (account_id, duration_ms);
> ```
>
> ```text
> Aggregate (actual rows=1.00 loops=1)
> Buffers: shared hit=1 read=3
> -> Index Only Scan using events_account_id_duration_ms_idx on events
> (actual rows=1.00 ...)
> Index Cond: ((account_id =
> (NULLIF(current_setting(...), ''::text))::bigint)
> AND (duration_ms = 1234))
> Heap Fetches: 0
> ```
## SECURITY DEFINER 函数需要固定的 search_path {#security-definer-functions-need-a-pinned-search_path}
`security definer` 函数以其所有者的权限运行，就像 setuid 程序一样：对于“接受这份邀请”这类范围有限的操作很有用，却也是经典的权限提升入口。函数内部的未限定名称会通过调用者的 `search_path` 解析，因此调用者可以把自己的 `count` 放在最前面，让所有者执行它。固定搜索路径：
```sql
set role wg_owner;
create function app.account_event_count(p_account bigint) returns bigint
language sql stable security definer
set search_path = pg_catalog, pg_temp
as $$ select count(*) from app.events where account_id = p_account $$;
revoke execute on function app.account_event_count(bigint) from public;
grant execute on function app.account_event_count(bigint) to wg_app;
```
记住三点：固定 `search_path`，并把 `pg_temp` 放在最后；为每张数据表加上模式限定；撤销 `execute` 权限，使 `public` 不再拥有它（新函数默认对所有人可执行）。
强制 RLS 也会约束所有者，因此租户 42 调用 `app.account_event_count(7)` 得到 0；绕过 RLS 的所有者本来会返回租户 7 的数量。而且 definer 函数永远不会被内联，所以规划器看到的是一个黑盒（[STRICT、SECURITY DEFINER 或 SET 子句会阻止内联](/24-functions/#strict-security-definer-or-a-set-clause-stops-inlining)）。
> [!DETAILS]- 深入阅读：实际运行 search_path 攻击
>
> 实验定义了第二个函数 `app.unpinned_event_count`，它与前者相同，但没有 `set search_path` 子句。然后，以 `wg_app` 身份在自己控制的模式中定义一个 `count(*)`，其状态函数会记录是谁执行了它，并把该模式放在 `pg_catalog` 之前：
>
> ```sql
> create function wg_app.steal(bigint) returns bigint language plpgsql as $$
> begin
> perform set_config('lab.ran_as', current_user, false);
> return $1 + 1;
> end $$;
> create aggregate wg_app.count(*) (
> sfunc = wg_app.steal, stype = bigint, initcond = '0'
> );
> set search_path = wg_app, pg_catalog;
> select app.unpinned_event_count(42);
> ```
>
> 只有当路径没有写出 `pg_catalog` 时，它才会被优先搜索；把它写在后面就会把第一选择交给调用者。未固定路径的函数会以 `wg_owner` 身份运行冒牌函数。固定路径的函数会忽略调用者的路径，运行真正的 `count`，并返回 2,084。
>
> > [!EXAMPLE]- 动手验证 · lab `32-security`
> >
> > ```sql
> > -- the attack: as wg_app, define wg_app.count(*) whose state function records
> > -- current_user, then: set search_path = wg_app, pg_catalog;
> > select app.unpinned_event_count(42) \g /dev/null
> > select lab.prove(
> > 'an unpinned SECURITY DEFINER function ran the caller''s count(*), as wg_owner',
> > current_setting('lab.ran_as') = 'wg_owner');
> >
> > select set_config('lab.ran_as', '', false) \g /dev/null
> > select lab.prove(
> > 'with search_path pinned to pg_catalog, pg_temp, the real count runs and the caller''s code does not',
> > app.account_event_count(42) = 2084 and current_setting('lab.ran_as') = '');
> > ```
> >
> > [完整实验脚本](https://www.flagon.io/books/eight-kilobytes/lab/32-security.sql)
## 使用 scram-sha-256 和 verify-full，切勿使用 trust {#use-scram-sha-256-and-verify-full-never-trust}
上面所有内容都假设连接进来的正是正确角色。`pg_hba.conf` 决定谁可以连接、从哪里连接，以及如何证明身份。服务器会使用第一条同时匹配连接类型、数据库、用户和地址的规则：
```text
# TYPE DATABASE USER ADDRESS METHOD
local all postgres peer
hostssl app wg_app 10.0.0.0/16 scram-sha-256
hostssl app wg_migrator 10.0.5.10/32 scram-sha-256
hostssl replication replicator 10.0.6.0/24 scram-sha-256
host all all all reject
```
`hostssl` 只匹配 TLS 连接，因此未加密连接根本不会匹配。使用 `scram-sha-256`，网络规则中永远不要使用 `trust`，并以明确的 `reject` 结尾。它的地址 `all` 也涵盖 IPv6，而 `0.0.0.0/0` 会漏掉 IPv6。
密码方面：`scram-sha-256` 从 PostgreSQL 14 起就是默认值；从 PostgreSQL 18 起，MD5 已被弃用，设置 MD5 密码时会发出警告。用 `where rolpassword like 'md5%'` 在 `pg_authid` 上找出旧角色并重置它们的密码。
TLS 方面：设置 `ssl = on`，并使用 `sslmode=verify-full` 连接，让客户端检查证书和主机名；`require` 会加密连接，却不能阻止冒充。托管数据库会提供 CA 证书包。
## 从服务器日志开始审计；它不额外收费 {#start-auditing-with-the-server-log-its-free}
锁上的门仍然需要记录是谁打开了它。先从服务器日志开始：`log_statement = 'ddl'` 会记录每个 create、alter 和 drop 以及执行用户；有了所有者与应用角色的分离，日志中的任何 DDL 都来自迁移角色。连接也要记录。不要在繁忙的生产环境设置 `log_statement = 'all'`：日志会迅速膨胀，并充满敏感值。
> [!DETAILS]- 深入阅读：一份日志基线，以及哪些会话确实使用了 TLS
>
> ```ini
> log_connections = 'authentication,authorization' # PostgreSQL 18 takes a list
> log_disconnections = on
> log_statement = 'ddl' # every create/alter/drop, with the user
> log_line_prefix = '%m [%p] %q%u@%d %a '
> ```
>
> 再看 TLS：哪些会话正在使用它：
>
> ```sql
> select a.usename, a.client_addr, s.ssl, s.version
> from pg_stat_ssl s join pg_stat_activity a using (pid)
> where a.backend_type = 'client backend';
> ```
对于合规级审计，使用 `pgaudit`。对于应用历史（“谁修改了这张发票”），在同一个事务中写入审计表（[历史记录也是你要设计的表](/14-relationships/#history-is-a-table-you-design)），并从应用角色撤销对它的 `update` 和 `delete` 权限。
然后在 CI 中检查权限。下面的查询会找出没有启用并强制执行 RLS 的租户数据表：
```sql
select n.nspname, c.relname, c.relrowsecurity, c.relforcerowsecurity
from pg_class c join pg_namespace n on n.oid = c.relnamespace
where c.relkind in ('r', 'p') and n.nspname = 'app'
and not (c.relrowsecurity and c.relforcerowsecurity);
```
在迁移后的测试数据库上运行时，如果新建的租户数据表没有 RLS，它会让构建失败。在本章的设置中，它已经能找到一张：`app.projects` 从未执行 `force`。再增加一项检查，找出具有 `rolsuper`、`rolbypassrls` 或 `rolcreaterole` 的登录角色。
## 本章要点 {#the-short-version}
- 应用以一个不是超级用户、没有 `BYPASSRLS`、也不拥有数据表的角色连接。迁移以单独的所有者角色运行。
- 使用 `alter default privileges`，让新数据表自动得到正确授权；在早于 PostgreSQL 15 创建的数据库中，撤销 `create` 权限在 `public` 模式上的默认授权。
- 对多租户隔离，在每张租户数据表上启用并强制执行行级安全，策略使用 `account_id`（备份和跨租户任务随后需要 `BYPASSRLS` 角色），并添加复合 `(account_id, id)` 外键，使引用无法跨租户。
- 在每个请求的事务内，通过一个辅助函数用 `set_config('app.account_id', $1, true)` 设置租户；在策略中用 `nullif(..., '')` 包住它。
- 策略就是谓词：索引以租户列开头，让策略保持简单，并把你通过策略搜索的 jsonb 字段提升为带类型的列。
- 每个 `security definer` 函数都应设置 `set search_path = pg_catalog, pg_temp`，使用带模式限定的名称，并执行 `revoke execute ... from public`。
- 使用 `hostssl` 和 `scram-sha-256`，把规则写在 `pg_hba.conf` 中；客户端使用 `sslmode=verify-full`，并记录 DDL 和连接。
## 延伸阅读 {#further-reading}
- [Row Security Policies](https://www.postgresql.org/docs/18/ddl-rowsecurity.html) 和 [CREATE POLICY](https://www.postgresql.org/docs/18/sql-createpolicy.html)，PostgreSQL 文档。介绍宽松与限制性策略、按命令区分的策略，以及 `using` 和 `with check` 的确切规则。
- [Privileges](https://www.postgresql.org/docs/18/ddl-priv.html) 和 [ALTER DEFAULT PRIVILEGES](https://www.postgresql.org/docs/18/sql-alterdefaultprivileges.html)，PostgreSQL 文档。完整的权限模型，包括 `\dp` 打印出的字母。
- [CREATE FUNCTION](https://www.postgresql.org/docs/18/sql-createfunction.html)，PostgreSQL 文档。如何安全地编写 SECURITY DEFINER 函数。
- [A Guide to CVE-2018-1058: Protect Your Search Path](https://wiki.postgresql.org/wiki/A_Guide_to_CVE-2018-1058%3A_Protect_Your_Search_Path)，PostgreSQL Wiki。`search_path` 攻击，以及 PostgreSQL 15 为什么改变了 public 模式。
- [Abusing SECURITY DEFINER functions](https://www.cybertec-postgresql.com/en/abusing-security-definer-functions/)，Laurenz Albe，CYBERTEC，2019。通过未固定的 search_path 实现权限提升的完整示例。
- [The pg_hba.conf File](https://www.postgresql.org/docs/18/auth-pg-hba-conf.html)，PostgreSQL 文档。每一种认证方法和匹配规则。
- [pgAudit](https://github.com/pgaudit/pgaudit)，pgAudit 项目。当服务器日志不够用时，执行语句和对象审计。