主题
第 13 章 权限与安全
学习目标：能向架构师解释「PG 的 ROLE 体系为什么把用户和组合并成一个东西」；能独立设计「最小权限」的生产账号矩阵；能用 行级安全 RLS 实现一个干净的多租户 SaaS；能看懂
pg_hba.conf5 列的含义并知道生产应该用scram-sha-256；能在面试中接住「列级权限 / RLS / pg_hba 优先级 / 默认权限」这一连串组合拳。
13.0 导读：为什么把「权限」单列一章？
很多读者用 PG 都是直接拿 postgres 这个超级用户连，写代码、跑业务、改表结构，全干。等到有一天：
- 测试同学误删了一张生产表，发现整个团队都用同一个账号；
- 多租户系统里，租户 A 通过应用 Bug 看到了租户 B 的数据；
- 安全审计要求所有连接必须 SSL，否则不许进数据库；
- 监管要求「只读分析师不许看到用户手机号列」。
这些事情，靠 ORM 在应用层做防御都治标不治本。真正能挡住它们的，是数据库内核的权限模型。PG 在权限这一块的能力，比 MySQL 强出一个时代——不仅有传统的「对象/动作」二维 GRANT，还有自 9.5 起的 行级安全 RLS（Row Level Security），让数据库自己来当「门卫」，应用程序写错也漏不出去。
本章会让你彻底理解：
- ROLE 体系：用户、组、组的组，全是一个东西；
- GRANT/REVOKE 矩阵：13 种对象 × 12 种权限的组合;
- 默认权限 ALTER DEFAULT PRIVILEGES：让以后创建的表也自动授权；
- 行级安全 RLS：多租户隔离、按部门隔离、按数据所有者隔离；
- pg_hba.conf 五列法则：谁能从哪台机器连过来、用什么方式认证；
- SSL/TLS 强制加密：
sslmode四档、自签证书与生产证书； - 密码策略：
scram-sha-256、VALID UNTIL、密码强度； - 审计：
pgaudit扩展、event trigger 自建审计； - 生产最佳实践清单：把以上技术拼成一套可落地的安全规范。
13.1 ROLE 体系：一切都是「角色」
13.1.1 生活类比：通行证 = 工牌 = 门卡
想象一栋写字楼：
- 每个员工有一张 工牌（身份）；
- 工牌可以开门（LOGIN 属性）；
- 几个员工可以共属一个 部门门卡组（GROUP）；
- 门卡组也是「一张卡」，只不过它没法刷电梯下班，只能挂在员工卡上做权限继承；
- HR 的卡可以开「全公司所有门」（SUPERUSER）。
PostgreSQL 9.0 之后，做了一个非常聪明的统一：员工卡和部门门卡，本质都是「卡」。它把这个东西叫作 ROLE（角色）。
📌 与 MySQL 的区别
MySQL 5.x 把「用户」（
'alice'@'%'，用户名 + 主机名是主键）和「角色」（MySQL 8.0 才加的，且独立于用户）分得很死。PG 9.0 起就是「ROLE 一统天下」：可以登录的 ROLE 就是「用户」，不能登录的 ROLE 就是「组/角色」。一个 ROLE 既可以是用户，也可以是别人的组，组也能套娃成另一个组。
13.1.2 创建 ROLE 的全套语法
sql
-- 创建一个能登录的「用户」
CREATE ROLE alice LOGIN PASSWORD 'Alice@2025';
-- 等价的语法糖：CREATE USER 自带 LOGIN
CREATE USER bob PASSWORD 'Bob@2025';
-- 创建一个「组角色」（不能登录，只用于挂权限）
CREATE ROLE devs; -- 等价 NOLOGIN
-- 把 alice、bob 加进 devs 组
GRANT devs TO alice;
GRANT devs TO bob;
-- 移除
REVOKE devs FROM bob;
-- 改密码、改属性
ALTER ROLE alice WITH PASSWORD 'NewPass@2025';
ALTER ROLE alice WITH CONNECTION LIMIT 5; -- 同时只能开 5 条连接
ALTER ROLE alice VALID UNTIL '2026-01-01'; -- 密码到期时间
ALTER ROLE alice WITH NOLOGIN; -- 临时禁用登录跑一下 \du（psql 元命令）：
text
learn_pg=# \du
List of roles
Role name | Attributes
-----------+----------------------------------------------------------
alice | Cannot login + -- 因为前一句 NOLOGIN
bob |
devs | Cannot login
postgres | Superuser, Create role, Create DB, Replication, Bypass RLS13.1.3 ROLE 的「属性」八件套
属性（attribute）是 ROLE 自己的能力，不通过 GRANT 授予，必须在 CREATE/ALTER 里指定：
| 属性 | 缺省 | 含义 | 生产建议 |
|---|---|---|---|
LOGIN / NOLOGIN | NOLOGIN | 能否作为登录身份连接 PG | 用户 = LOGIN；组 = NOLOGIN |
SUPERUSER / NOSUPERUSER | NOSUPERUSER | 是否拥有超级权限（绕过所有检查） | 业务账号绝对不要给 |
CREATEDB / NOCREATEDB | NOCREATEDB | 能否 CREATE DATABASE | 仅 DBA 给 |
CREATEROLE / NOCREATEROLE | NOCREATEROLE | 能否创建/管理其它非 SUPERUSER 角色 | 团队 leader 可酌情给 |
REPLICATION / NOREPLICATION | NOREPLICATION | 能否做物理/逻辑复制连接 | 只给复制账号 |
BYPASSRLS / NOBYPASSRLS | NOBYPASSRLS | 是否绕过行级安全 | 只给做数据修复的临时账号 |
INHERIT / NOINHERIT | INHERIT | 是否自动继承所属组角色的权限 | 默认即可，下面会重点讲 |
CONNECTION LIMIT n | -1 不限 | 同时连接上限 | 应用账号建议设 |
⚠️
INHERIT是最容易踩坑的属性。下一节单独细讲。
13.1.4 INHERIT vs SET ROLE：两种「使用组权限」的姿势
PG 的组继承有两种模式：
模式 1：INHERIT（默认）—— 自动继承
sql
CREATE ROLE devs;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO devs;
CREATE ROLE alice LOGIN PASSWORD 'x' INHERIT; -- 默认就是 INHERIT
GRANT devs TO alice;
-- alice 连上后，立即可以 SELECT，自动拥有 devs 的所有权限模式 2：NOINHERIT —— 必须 SET ROLE 切身份
sql
CREATE ROLE bob LOGIN PASSWORD 'x' NOINHERIT;
GRANT devs TO bob;
-- bob 连上后默认不带 devs 权限，必须主动切：
SET ROLE devs; -- 现在以 devs 身份操作
RESET ROLE; -- 切回 bob怎么选？
- INHERIT：日常开发首选，方便。
- NOINHERIT：安全敏感系统用，可以防止「不小心以 devs 身份做了不该做的操作」，类似 Linux 的
sudo：默认是普通用户，要做特权动作必须显式sudo。
13.1.5 一图看懂 ROLE 关系网
text
┌─────────────┐
│ postgres │ SUPERUSER
└─────────────┘
│
┌──────────────┼──────────────┐
▼ ▼ ▼
┌────────┐ ┌────────┐ ┌──────────┐
│ dba │ │ app_rw │ │ app_ro │ ← 都是 NOLOGIN 的「组角色」
└────────┘ └────────┘ └──────────┘
│ │
┌────┴────┐ │
▼ ▼ ▼
┌──────┐ ┌──────┐ ┌──────────┐
│alice │ │bob │ │analyst_a │ ← 都是 LOGIN 的「用户角色」
└──────┘ └──────┘ └──────────┘13.2 GRANT / REVOKE：权限矩阵
13.2.1 PG 把「权限」拆成两个维度
- 对象类型：DATABASE / SCHEMA / TABLE / SEQUENCE / FUNCTION / TYPE / LANGUAGE / FOREIGN SERVER / ...
- 动作（权限名）：SELECT / INSERT / UPDATE / DELETE / TRUNCATE / REFERENCES / TRIGGER / USAGE / EXECUTE / CONNECT / CREATE / TEMP
不是所有动作都适用所有对象。下面这张「对象 × 权限」矩阵是 DBA 的内功心法：
| 对象 \ 权限 | SELECT | INSERT | UPDATE | DELETE | TRUNCATE | REFERENCES | TRIGGER | USAGE | EXECUTE | CONNECT | CREATE | TEMP |
|---|---|---|---|---|---|---|---|---|---|---|---|---|
| DATABASE | ✅ | ✅ | ✅ | |||||||||
| SCHEMA | ✅ | ✅ | ||||||||||
| TABLE | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | |||||
| SEQUENCE | ✅ | ✅ | ✅ | |||||||||
| FUNCTION | ✅ | |||||||||||
| TYPE/DOMAIN | ✅ | |||||||||||
| LANGUAGE | ✅ |
记住三个高频的「容易忘」点：
- 要用 SCHEMA 里的表，必须先
GRANT USAGE ON SCHEMA xx（光给表 SELECT 不够）。 SERIAL自增列实际是个 SEQUENCE，给写权限时记得GRANT USAGE ON SEQUENCE，否则 INSERT 会因为nextval()拿不到序列报错。- DATABASE 的
CONNECT权限：默认所有人都能连任意库（PUBLIC），生产建议REVOKE CONNECT ON DATABASE learn_pg FROM PUBLIC然后白名单授权。
13.2.2 GRANT 全语法速查
sql
-- 给 alice 单表 SELECT
GRANT SELECT ON TABLE orders TO alice;
-- 一次性给一个 schema 下所有现有表
GRANT SELECT ON ALL TABLES IN SCHEMA public TO alice;
-- 列级权限：仅允许查 id, name 两列
GRANT SELECT (id, name) ON users TO analyst;
-- 同时给读 + 写
GRANT SELECT, INSERT, UPDATE, DELETE ON orders TO app_rw;
-- 给序列（SERIAL/IDENTITY 列依赖序列）
GRANT USAGE, SELECT ON SEQUENCE orders_id_seq TO app_rw;
-- 给函数执行权限
GRANT EXECUTE ON FUNCTION calc_tax(numeric) TO app_rw;
-- 给一组角色（一次给一组人）
GRANT SELECT ON orders TO devs;
-- 带 WITH GRANT OPTION：alice 还能继续把这个权限授予别人
GRANT SELECT ON orders TO alice WITH GRANT OPTION;
-- 收回
REVOKE SELECT ON orders FROM alice;
-- 级联收回（连同 alice 转授给别人的也一起收）
REVOKE SELECT ON orders FROM alice CASCADE;实操一段 psql：
text
learn_pg=# \z orders
Access privileges
Schema | Name | Type | Access privileges | Column privileges | Policies
--------+--------+-------+---------------------------------+-------------------+----------
public | orders | table | postgres=arwdDxt/postgres +| |
| | | alice=r/postgres +| |
| | | app_rw=arwd/postgres | |arwdDxtRX 缩写解读（这是面试题）：
| 字母 | 全称 | 中文 |
|---|---|---|
a | INSERT | 增 |
r | SELECT (read) | 查 |
w | UPDATE (write) | 改 |
d | DELETE | 删 |
D | TRUNCATE | 清空 |
x | REFERENCES | 外键引用 |
t | TRIGGER | 触发器 |
X | EXECUTE | 函数执行 |
U | USAGE | 使用 schema/序列 |
C | CREATE | 建对象 |
T | TEMP | 临时表 |
c | CONNECT | 连库 |
grantee=权限/grantor 表示「grantee 拥有这些权限，权限是 grantor 授予的」。空 grantee 表示 PUBLIC。
13.2.3 所有权 vs 权限：所有者隐含「全部权限」
对象的 owner 隐含拥有该对象的所有权限，不需要 GRANT。但要注意：
\z里 owner 自己的权限默认不显示（除非有人 REVOKE 然后重新 GRANT 过）。- owner 还有「修改对象结构」的能力（
DROP / ALTER），这不属于 GRANT 体系。 - 想转移所有权：
ALTER TABLE orders OWNER TO devs;。 - 想让组成员「集体拥有」一个对象：让对象的 owner 设为组角色（比如
devs），组里的 alice/bob 通过INHERIT都能改这张表。
⚠️ 生产坑：很多团队让 alice 直接建表，结果 alice 离职后，这张表只有 alice 和 superuser 能 ALTER，其他同事改不了。正确姿势：让 alice
SET ROLE devs;后再建表，对象 owner 就是devs，全组都能管。
13.3 默认权限 ALTER DEFAULT PRIVILEGES：给「未来的对象」也授权
13.3.1 痛点
sql
-- DBA 给 ro 组授权了所有现有表的 SELECT
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ro;
-- 一周后，开发新建了一张表
CREATE TABLE new_orders (...);
-- ro 用户想 SELECT new_orders？
SELECT * FROM new_orders;
-- ERROR: permission denied for table new_ordersGRANT ON ALL TABLES 只授权当时存在的表，对将来新建的表无效。一种笨方法是每次建表都手动 GRANT，但显然不优雅。
13.3.2 解法：ALTER DEFAULT PRIVILEGES
sql
-- 含义：以后由 alice 在 public schema 下创建的所有 TABLE，
-- 自动给 ro 组 SELECT 权限
ALTER DEFAULT PRIVILEGES
FOR ROLE alice
IN SCHEMA public
GRANT SELECT ON TABLES TO ro;记住四个 限定范围 关键字：
| 关键字 | 含义 | 不指定时 |
|---|---|---|
FOR ROLE | 限定「由谁创建」的对象 | 默认是当前用户 |
IN SCHEMA | 限定哪个 schema 下 | 默认所有 schema |
ON TABLES | 限定对象类型 | 必须显式指定（TABLES/SEQUENCES/FUNCTIONS/TYPES/SCHEMAS） |
TO role | 授予给谁 | 必填 |
13.3.3 一份「只读账号 ro」的标准化授权脚本
sql
-- 1. 创建只读组
CREATE ROLE ro NOLOGIN;
CREATE USER analyst PASSWORD 'x' IN ROLE ro;
-- 2. 允许连库
GRANT CONNECT ON DATABASE learn_pg TO ro;
-- 3. 允许使用 schema
GRANT USAGE ON SCHEMA public TO ro;
-- 4. 已有对象立即授权
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ro;
GRANT SELECT ON ALL SEQUENCES IN SCHEMA public TO ro;
-- 5. ★ 未来新建对象自动授权 ★
ALTER DEFAULT PRIVILEGES IN SCHEMA public
GRANT SELECT ON TABLES TO ro;
ALTER DEFAULT PRIVILEGES IN SCHEMA public
GRANT SELECT ON SEQUENCES TO ro;注意：ALTER DEFAULT PRIVILEGES 是「按对象创建者」记录的。如果 alice 和 bob 都建表，得各自跑一遍（或省略 FOR ROLE 让规则属于「当前角色」自己）。最稳妥是 统一让一个组角色（如 app_owner）拥有所有业务表：
sql
ALTER DEFAULT PRIVILEGES FOR ROLE app_owner IN SCHEMA public
GRANT SELECT ON TABLES TO ro;📌 与 MySQL 的区别：MySQL 没有「未来对象」自动授权机制，新建库/表后必须手动 GRANT。一些 DBA 用脚本巡检 + 自动 GRANT 来弥补，PG 在数据库内核层面就解决了。
13.4 行级安全 RLS（Row Level Security）
RLS 是 PG 9.5 引入的功能，允许在「表里每一行」上挂一个过滤条件。对于多租户 SaaS、按部门隔离、按数据所有者隔离的场景，是 杀手级功能。
13.4.1 生活类比：图书馆的「专属阅览室」
想象一个图书馆，所有书都在同一个大书架上（同一张表），但你只有「儿童区」的借书卡（你的 ROLE）。每次你走到书架前，门卫先看你卡上的颜色（current_user），自动只让你看到儿童区的书。其他大人想看儿童区的书也行，但他们看不到「教师专属书架」。这个门卫就是 RLS Policy，不是你写在应用层的 WHERE owner_id = $1，而是数据库自己挂在表上的过滤条件，应用怎么写 SQL 都绕不开。
13.4.2 一个完整的多租户 SaaS 例子
假设我们做一个 多公司协作 SaaS，所有公司的工单都在同一张表 ch13_tickets，靠 tenant_id 区分。要求：
- 每家公司只能看到/改自己的数据；
- 应用通过
SET LOCAL app.tenant_id = '<id>'在事务里告诉数据库「我现在代表谁」； - 哪怕程序员手抖写了
SELECT * FROM ch13_tickets，也只会返回当前租户的数据。
Step 1：建表（注意 tenant_id 列）
sql
CREATE TABLE ch13_tickets (
id BIGSERIAL PRIMARY KEY,
tenant_id INT NOT NULL,
title TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'open',
created_at TIMESTAMPTZ DEFAULT now()
);Step 2：开启 RLS
sql
ALTER TABLE ch13_tickets ENABLE ROW LEVEL SECURITY;⚠️ 开启之后 没有任何 policy 的话，普通用户读这张表会一行都看不到（owner 和 superuser 例外）。所以必须紧接着写 policy。
Step 3：写 policy
sql
-- 读策略：只能看到自己 tenant 的行
CREATE POLICY ch13_tenant_iso_select ON ch13_tickets
FOR SELECT
USING (tenant_id = current_setting('app.tenant_id')::int);
-- 写策略：只能给自己 tenant 写入
CREATE POLICY ch13_tenant_iso_insert ON ch13_tickets
FOR INSERT
WITH CHECK (tenant_id = current_setting('app.tenant_id')::int);
-- 改/删策略：只能改/删自己的
CREATE POLICY ch13_tenant_iso_update ON ch13_tickets
FOR UPDATE
USING (tenant_id = current_setting('app.tenant_id')::int)
WITH CHECK (tenant_id = current_setting('app.tenant_id')::int);
CREATE POLICY ch13_tenant_iso_delete ON ch13_tickets
FOR DELETE
USING (tenant_id = current_setting('app.tenant_id')::int);USING 和 WITH CHECK 的区别（面试常考）：
| 子句 | 时机 | 含义 |
|---|---|---|
USING | 读取 / 更新 / 删除时 | 这一行 是否对当前用户可见 |
WITH CHECK | 插入 / 更新后 | 这一行 是否被允许写入 |
简单记：
USING是「门卫看进来的」，WITH CHECK是「门卫看出去的」。
Step 4：应用如何使用
python
# 每次 web 请求，进事务后立即设置租户上下文
cur.execute("BEGIN")
cur.execute("SET LOCAL app.tenant_id = %s", (current_tenant_id,))
# 后续所有 SQL，无论怎么写，都只能看到本租户数据
cur.execute("SELECT * FROM ch13_tickets") # 自动加了 tenant_id = N 的过滤SET LOCAL 而不是 SET，含义是「只在当前事务内生效」，事务一结束自动失效，避免连接复用时上一个租户的设置漏给下一个请求。
13.4.3 RLS 流程图
13.4.4 RLS 的进阶用法
(1) FOR ALL —— 一条策略覆盖所有动作
sql
CREATE POLICY ch13_tenant_all ON ch13_tickets
FOR ALL
USING (tenant_id = current_setting('app.tenant_id')::int)
WITH CHECK (tenant_id = current_setting('app.tenant_id')::int);(2) TO role —— 仅对特定角色生效
sql
CREATE POLICY ch13_admin_full_access ON ch13_tickets
FOR ALL
TO admin -- 仅对 admin 角色应用此策略
USING (true) -- 看所有行
WITH CHECK (true);(3) PERMISSIVE vs RESTRICTIVE
- 默认是
PERMISSIVE：多个 policy 之间是 OR 关系，任何一个匹配即放行； RESTRICTIVE：和其它 policy 是 AND 关系，所有 RESTRICTIVE 都要满足，常用于「黑名单」。
sql
CREATE POLICY ch13_block_archived ON ch13_tickets
AS RESTRICTIVE
FOR ALL
USING (status <> 'archived');
-- 哪怕之前的 PERMISSIVE 让你看到 archived 行，
-- 这条 RESTRICTIVE 也会把它拦下来(4) 强制 owner 也走 RLS：FORCE ROW LEVEL SECURITY
默认 owner 和 superuser 都绕过 RLS。如果连 owner 也想被 RLS 限制（防止程序用 owner 账号连）：
sql
ALTER TABLE ch13_tickets FORCE ROW LEVEL SECURITY;(5) BYPASSRLS 角色
sql
ALTER ROLE backup_user BYPASSRLS;
-- backup_user 即使不是 superuser，也能绕过 RLS 看全表（用于 pg_dump）13.4.5 RLS 的性能开销
RLS 本质是重写 SQL，把 policy 当成 WHERE 条件追加。所以：
- 一定要在 policy 引用的列上建索引（如
tenant_id）； - 复杂 policy（带子查询）可能阻碍优化器，必要时用函数
STABLE标记； - 大量 policy 链式叠加会增加 plan 复杂度，单表 policy 别超过 5~6 条。
📌 与 MySQL 的区别：MySQL 没有原生 RLS，多租户场景必须靠应用层
WHERE条件，或用 VIEW + DEFINER 凑出类似效果，但都不如 PG RLS 干净彻底。一旦应用代码哪天忘了加 WHERE，MySQL 就出数据隔离事故；PG 的 RLS 会在内核层兜底。
13.5 认证与连接：pg_hba.conf 五列法则
13.5.1 它是干嘛的？
pg_hba.conf 全称 Host-Based Authentication，决定「谁、从哪台机器、连哪个库、用什么身份、用什么方式认证」。它是 PG 的「门禁规则表」，比 GRANT 更靠前——你密码再对，pg_hba 不放行也连不进来。
文件位置：PGDATA/pg_hba.conf（容器里通常 /var/lib/postgresql/data/pg_hba.conf），改完要 pg_ctl reload 或在数据库内 SELECT pg_reload_conf(); 才生效。
13.5.2 五列结构
text
# TYPE DATABASE USER ADDRESS METHOD
local all postgres peer
host all all 127.0.0.1/32 scram-sha-256
host all all ::1/128 scram-sha-256
host learn_pg alice 10.0.0.0/8 scram-sha-256
hostssl all all 0.0.0.0/0 scram-sha-256
hostnossl all all 0.0.0.0/0 reject五列含义：
| # | 列 | 取值示例 | 含义 |
|---|---|---|---|
| 1 | TYPE | local / host / hostssl / hostnossl | 本地 socket / TCP / 仅 SSL / 仅非 SSL |
| 2 | DATABASE | all / learn_pg / db1,db2 / replication | 适用哪个库 |
| 3 | USER | all / alice / +devs | 适用哪个用户/组（前缀 + 表示组成员） |
| 4 | ADDRESS | 127.0.0.1/32 / 10.0.0.0/8 / samenet | 来源 IP 段 |
| 5 | METHOD | trust / reject / md5 / scram-sha-256 / cert / peer / ident / ldap / pam / ... | 认证方式 |
匹配规则（极其重要）：
- 从上往下逐行匹配；
- 第一条匹配的规则就是最终结果，后面的不再看；
- 如果 method 是
reject，立刻拒绝； - 如果 method 是
trust，任何密码都能进（仅本地调试用）； - 如果一行都不匹配，默认拒绝。
13.5.3 常用 METHOD 一览
| METHOD | 安全性 | 适用场景 | 说明 |
|---|---|---|---|
trust | ⚠️ 极低 | 仅本机 docker 容器内调试 | 不验密码，谁来都让进 |
reject | — | 显式拒绝某些来源 | 立即拒绝 |
md5 | 低 | 老版本兼容 | MD5 哈希，已有彩虹表破解风险，不推荐 |
scram-sha-256 | 高 | 生产推荐 | PG 10+ 默认推荐，需要客户端支持 |
password | 低 | 仅 SSL 内传明文 | 几乎不用，明文密码 |
cert | 高 | 双向 TLS、零密码 | 客户端必须有受信证书 |
peer | 中 | 本地 unix socket | 用 OS 用户名作为 PG 用户名 |
ident | 中 | TCP 但内网可信 | 通过 ident 协议查对方 OS 用户 |
ldap | 中 | 大公司统一登录 | 接公司 AD/LDAP |
pam | 中 | Linux PAM 体系 | 集成 OS 认证模块 |
13.5.4 「为什么我配了密码就是连不上？」常见诊断
text
# 错误信息: psql: error: connection to server ... failed:
# FATAL: no pg_hba.conf entry for host "1.2.3.4", user "alice"→ pg_hba 没匹配到，加一行 host 规则。
text
# FATAL: password authentication failed for user "alice"→ pg_hba 规则匹配了，但密码不对；或者用了 md5 但密码是 scram-sha-256 加密的（升级时常见，重设密码即可）。
text
# FATAL: SSL required→ pg_hba 配了 hostssl，但客户端用的非 SSL 连接，加 sslmode=require。
13.5.5 PG 14+ 的默认推荐：scram-sha-256
bash
# postgresql.conf
password_encryption = scram-sha-256
# pg_hba.conf
host all all 0.0.0.0/0 scram-sha-256升级注意：老用户的密码哈希仍是 md5，需要让他们重新设置一次密码：
sql
ALTER ROLE alice PASSWORD 'NewSecret@2025';
-- 重设后会用当前 password_encryption 的算法（scram-sha-256）存储13.6 SSL/TLS：传输加密
13.6.1 服务端启用
ini
# postgresql.conf
ssl = on
ssl_cert_file = '/etc/postgres/server.crt'
ssl_key_file = '/etc/postgres/server.key'
ssl_ca_file = '/etc/postgres/root.crt' -- 可选：双向认证用证书可以用 OpenSSL 自签（测试），生产用内部 CA 或公网 CA。
13.6.2 客户端 sslmode 四档
| sslmode | 含义 | 中间人风险 |
|---|---|---|
disable | 不用 SSL | — |
allow | 服务端要求才用 | — |
prefer（默认） | 优先 SSL，不行就降级明文 | ⚠️ 仍可能被降级攻击 |
require | 必须 SSL，但不验证证书 | 中间人可伪装 |
verify-ca | SSL + 校验证书是 CA 颁发 | 主机名仍可能被伪造 |
verify-full | SSL + 校验 CA + 校验主机名 | ✅ 最安全 |
python
# psycopg v3 连接字符串
conn = psycopg.connect(
"host=db.prod.example.com port=5432 dbname=learn_pg "
"user=app sslmode=verify-full sslrootcert=/etc/ssl/ca.crt"
)📌 生产强烈建议
verify-full。require只能挡被动嗅探，挡不住主动 MITM。
13.7 密码策略与口令安全
13.7.1 加密算法
sql
SHOW password_encryption;
-- scram-sha-256绝对不要用 md5 了。如果你的库还是 md5，迁移步骤：
sql
SET password_encryption = 'scram-sha-256';
ALTER ROLE alice PASSWORD 'newsecret'; -- 触发用新算法重新加密13.7.2 过期时间
sql
ALTER ROLE alice VALID UNTIL '2026-01-01 00:00:00';
-- 到期后无法登录，必须 DBA 续期或改密码13.7.3 强密码：扩展 passwordcheck
sql
-- 在 postgresql.conf 加载
shared_preload_libraries = 'passwordcheck'
-- 重启后生效，会拒绝过短/含用户名/全数字的弱密码13.7.4 连接数限制
sql
ALTER ROLE alice CONNECTION LIMIT 5; -- 防止某账号占满连接池13.8 审计：知道谁在做什么
13.8.1 自带的 log 级审计（轻量）
ini
# postgresql.conf
log_statement = 'ddl' -- 仅 DDL，或 'mod' / 'all'
log_connections = on
log_disconnections = on
log_line_prefix = '%m [%p] %u@%d ' -- 时间、PID、用户、库13.8.2 重武器：pgaudit 扩展
sql
CREATE EXTENSION pgaudit;
-- 在 postgresql.conf
-- shared_preload_libraries = 'pgaudit'
-- pgaudit.log = 'write,ddl,role'pgaudit 比 log_statement 强在：
- 区分 DDL / WRITE / READ / ROLE / FUNCTION 等类别；
- 能针对单个用户/单张表开启；
- 支持「对象级审计」：指定 audit role + GRANT 给它要审计的对象。
13.8.3 自建审计：用触发器
参见第 12 章，可以为业务关键表挂 AFTER 触发器把变更写入 audit 表。
13.9 生产最佳实践清单
把前面的内容拼成一个 可落地的安全规范：
text
[网络层]
├── 数据库不暴露公网，VPC 内访问
├── pg_hba 仅放行业务 IP 段
└── 强制 ssl=on，应用 sslmode=verify-full
[账号层]
├── 一应用一账号（app_rw / app_ro）
├── 业务账号绝不用 SUPERUSER
├── DBA 操作账号独立，开 MFA
└── 复制账号独立、且 NOSUPERUSER
[权限层]
├── 按 schema 拆业务（payment / user / order）
├── REVOKE CONNECT FROM PUBLIC，白名单授权
├── REVOKE ALL ON SCHEMA public FROM PUBLIC
├── 用组角色管权限，用户挂在组下
└── ALTER DEFAULT PRIVILEGES 兜底未来对象
[数据层]
├── 多租户开 RLS，FORCE ROW LEVEL SECURITY
├── 敏感列考虑列级 GRANT 或 pgcrypto 加密
└── 大表 owner 设为组角色，避免离职断层
[认证层]
├── password_encryption = scram-sha-256
├── 高敏感账号 VALID UNTIL 定期到期
└── 加载 passwordcheck，禁弱密码
[审计层]
├── log_connections / log_disconnections = on
├── log_statement = 'ddl' （生产少 'all'，量大）
└── 关键表用 pgaudit 或触发器写 audit log13.10 与 MySQL 的对比速查
| 主题 | MySQL | PostgreSQL |
|---|---|---|
| 用户与组 | 用户 = 用户名+主机；角色 8.0+ 才支持，独立体系 | ROLE 一统，用户和组都是 ROLE |
| 密码哈希 | caching_sha2_password（8.0 默认） | scram-sha-256（PG 14+ 推荐） |
| 主机白名单 | 'user'@'host' 写在用户身份里 | 集中在 pg_hba.conf |
| 列级权限 | 支持 | 支持，且更细：SELECT (col) |
| 行级安全 | 无原生，靠 VIEW/应用层 | RLS Policy 内核级 |
| 默认权限 | 无「未来对象自动授权」 | ALTER DEFAULT PRIVILEGES |
| SSL | 类似 sslmode，叫 --ssl-mode=VERIFY_IDENTITY | sslmode=verify-full |
| 审计 | MySQL Enterprise Audit / 第三方 plugin | pgaudit 扩展 |
| 角色继承 | SET ROLE 切换 | INHERIT 自动 / NOINHERIT 显式 SET ROLE |
13.11 小结
- ROLE = 用户 ∪ 组，用 LOGIN 区分两者；属性中 SUPERUSER / CREATEDB / BYPASSRLS / INHERIT 必须熟记。
- GRANT 是「对象 × 动作」二维矩阵，列级权限和 SCHEMA
USAGE是新手最容易漏的。 ALTER DEFAULT PRIVILEGES解决「未来对象」自动授权，是只读账号方案的关键。- RLS 是 PG 多租户系统的杀手锏，
USING控读，WITH CHECK控写，FORCE ROW LEVEL SECURITY让 owner 也守规矩。 pg_hba.conf五列法则、自上而下匹配、生产用scram-sha-256。sslmode=verify-full+ 内部 CA 是传输加密的标配。- 审计：先开
log_*三件套，关键表上pgaudit。
🎮 配套演示
用浏览器打开
./13_security/demo.html，跟着可视化动画再走一遍本章核心概念。配套代码在
./13_security/code/，每个脚本都可以独立python xxx.py运行，先跑init.sql准备数据。
13.12 面试高频题
Q1. PG 的 ROLE / USER / GROUP 是什么关系？(⭐⭐)
考察点：基础概念，PG 9.0+ 的「角色统一」设计。
答案：
PostgreSQL 9.0 之后，用户和组在内核里被统一成 ROLE。一个 ROLE 既可以是「能登录的用户」，也可以是「不能登录的组」，区分点是 LOGIN 属性：
CREATE ROLE alice LOGIN PASSWORD 'x';等价于「用户」；CREATE ROLE devs;（不带 LOGIN，默认 NOLOGIN）等价于「组」；CREATE USER只是CREATE ROLE ... LOGIN的语法糖，背后存的都是pg_authid这一张表。
这个统一设计带来 三大好处：
- 嵌套自由：组可以再属于另一个组（
GRANT devs TO seniors;），形成多级权限继承； - 「以组身份建对象」：用户先
SET ROLE devs;再建表，对象 owner 就是 devs，全组都能管，离职无影响； INHERIT控制权限继承方式：默认自动继承所属组权限；改成NOINHERIT后，必须SET ROLE显式切换，相当于内置的sudo，更安全。
与 MySQL 对比：MySQL 5.x 完全没有角色概念，直到 8.0 才加上，且角色和用户是两套体系（角色不能登录、用户独立存在），不像 PG 这样优雅统一。
易错点：以为 CREATE GROUP 还有用——这是 PG 9.0 之前的语法，已经废弃但语法上仍兼容，等价于 CREATE ROLE ... NOLOGIN，不要在新代码中使用。
Q2. pg_hba.conf 里 trust / md5 / scram-sha-256 / cert 的区别？(⭐⭐⭐)
考察点：认证方式、安全推荐、生产规范。
答案：
pg_hba.conf 的 METHOD 列决定客户端连接时如何被认证：
| 方式 | 工作原理 | 安全性 | 适用 |
|---|---|---|---|
trust | 不认证，匹配到的连接直接放行 | ⚠️ 极低 | 仅 docker 容器内或单机调试 |
md5 | 客户端用 MD5 哈希密码（带盐 = 用户名） | 低 | 老版本兼容；MD5 已有破解工具 |
scram-sha-256 | SCRAM 协议，挑战-应答 + SHA256，密码不在网络上传输 | 高 | PG 10+ 生产推荐 |
cert | 双向 TLS：客户端必须出示由服务端信任的 CA 颁发的证书，无密码 | 高 | 零密码方案、机器对机器 |
为什么 PG 14+ 推荐 scram-sha-256：
- SCRAM 是 IETF RFC 5802 标准协议，不依赖单向哈希；
- 即使中间人截获连接全过程，也无法重放或还原密码；
- 服务端只存 SCRAM 验证元数据（盐 + 迭代次数 + StoredKey），DBA 也看不到原密码；
- 自带 channel binding（与 SSL 通道绑定），防降级。
升级踩坑：password_encryption = scram-sha-256 改完后，老用户的密码哈希仍是 md5，必须 ALTER ROLE alice PASSWORD 'newpwd'; 重设一次，新哈希才会以 SCRAM 方式存储；否则把 pg_hba 改成 scram 反而会让老用户登不上。
生产组合建议：hostssl all all 0.0.0.0/0 scram-sha-256 + password_encryption = scram-sha-256 + passwordcheck 扩展强制弱密码拦截。
Q3. 行级安全 RLS 怎么用？多租户场景怎么落地？(⭐⭐⭐⭐)
考察点：RLS 全套语法、应用集成、性能与坑。
答案：
三步走：
- 建表时含租户字段，并
ALTER TABLE ch13_tickets ENABLE ROW LEVEL SECURITY; - 创建 POLICY，用
current_setting()读应用注入的租户 ID：
sql
CREATE POLICY ch13_tenant_iso ON ch13_tickets
FOR ALL
USING (tenant_id = current_setting('app.tenant_id')::int)
WITH CHECK (tenant_id = current_setting('app.tenant_id')::int);- 应用每次进事务后注入：
python
cur.execute("BEGIN")
cur.execute("SET LOCAL app.tenant_id = %s", (tenant_id,))
# 后续所有 SQL 都自动带上 tenant_id 过滤关键细节：
USINGvsWITH CHECK：USING 决定能否「看到」一行（用于 SELECT/UPDATE/DELETE 时的可见性判断）；WITH CHECK 决定能否「写入」一行（INSERT/UPDATE 后新值必须满足）。常见做法是两者都写同一条件，防止把数据「改成」另一个租户。- owner 默认绕过 RLS！如果应用直接用表的 owner 账号连，RLS 形同虚设。生产必须
ALTER TABLE ch13_tickets FORCE ROW LEVEL SECURITY;让 owner 也受约束，并让应用用单独的app_rw账号。 - PERMISSIVE vs RESTRICTIVE：多 PERMISSIVE 之间是 OR；RESTRICTIVE 与所有 policy 是 AND，常用于黑名单（如「不许看 archived 状态」）。
- 性能：policy 中引用的列必须建索引（如
tenant_id），否则每次都全表扫描；policy 表达式应尽量简单，复杂逻辑用 STABLE 函数包装让优化器复用。 SET LOCAL而非SET：连接复用场景下，SET LOCAL在 COMMIT/ROLLBACK 时自动失效，避免「下一个请求继承上个租户身份」的严重事故。BYPASSRLS角色：备份脚本（pg_dump）需要看全表，给备份账号ALTER ROLE backup BYPASSRLS;，比授超管安全。
与 MySQL 对比：MySQL 没有原生 RLS，多租户隔离要么靠应用层 WHERE（漏一次就出事），要么用 VIEW + 不暴露底表，但维护成本高得多。PG RLS 是数据库层兜底，写错 SQL 也漏不出去，是多租户 SaaS 的事实标准。
Q4. 权限粒度可以细到什么程度？列级权限怎么用？(⭐⭐⭐)
考察点：列级 GRANT、合规场景、与 RLS 的搭配。
答案：
PG 的权限粒度可以细到列级：
sql
-- 给 analyst 角色：仅可查 users 表的 id, name 列；不能查 phone, id_card
GRANT SELECT (id, name) ON users TO analyst;
-- 也可以分开授不同列的不同动作
GRANT UPDATE (status) ON orders TO ops; -- 只能改 status，不能改金额典型应用场景：
- 隐私合规：手机号、身份证等敏感列，给分析师只授非敏感列的 SELECT；
- 状态机控制：运营只能改订单状态，不能改金额；
- 审计配合：业务列开 SELECT，审计列（created_by/created_at）只允许 INSERT 不允许 UPDATE。
底层注意：
- 没有显式列权限时，等价于「整张表的权限都没有」；
- 只查未授权的列会报
permission denied for column phone； SELECT *会被拦截，因为*包含未授权列；建议应用代码显式列名；- 列权限和 RLS 是 正交叠加：RLS 决定能看到哪些行，列权限决定每行能看到哪些列。两者结合可以实现「分析师只看自家公司的数据，且看不到手机号」。
进阶：对于真正敏感的字段（比如银行卡号），列权限只能限制 GRANT 体系，不能防 superuser 与 owner。要彻底防泄露应叠加 pgcrypto 加密存储（pgp_sym_encrypt），密钥放 KMS 或 HSM，应用解密时再读。
与 MySQL 对比：MySQL 也支持列级权限（GRANT SELECT (col) ON tbl），但实际生产中用得少；PG 因为有 RLS 配合，列级权限的实用性更高。
Q5. 怎么让一个只读账号「真·只读」？避开 SECURITY DEFINER 函数风险。(⭐⭐⭐)
考察点：默认权限、PUBLIC 收权、SECURITY DEFINER 陷阱。
答案：
新手做的「只读账号」往往不真只读，常见漏洞包括：
- PUBLIC 默认权限没收回，新建表被新账号读到；
- SECURITY DEFINER 函数让只读账号借壳改写数据；
- 未来新建对象 没自动 GRANT，但旧表能改也是只读账号未必只读。
完整脚本：
sql
-- ① 先把 public schema 的默认权限收紧
REVOKE ALL ON SCHEMA public FROM PUBLIC;
REVOKE ALL ON ALL TABLES IN SCHEMA public FROM PUBLIC;
REVOKE ALL ON DATABASE learn_pg FROM PUBLIC;
-- ② 建只读组与用户
CREATE ROLE ro NOLOGIN;
CREATE USER analyst PASSWORD 'x' IN ROLE ro;
-- ③ 精确授权
GRANT CONNECT ON DATABASE learn_pg TO ro;
GRANT USAGE ON SCHEMA public TO ro;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ro;
GRANT SELECT ON ALL SEQUENCES IN SCHEMA public TO ro;
-- ④ 兜底：未来对象也自动只读
ALTER DEFAULT PRIVILEGES IN SCHEMA public
GRANT SELECT ON TABLES TO ro;
ALTER DEFAULT PRIVILEGES IN SCHEMA public
GRANT SELECT ON SEQUENCES TO ro;SECURITY DEFINER 风险：
PG 函数有两种执行身份：
SECURITY INVOKER（默认）：以调用者身份跑，权限受调用者限制；SECURITY DEFINER：以函数所有者身份跑，类似 setuid。
如果一个 owner 是超级用户的函数被 GRANT EXECUTE 给 ro，那么 ro 调用它就能借壳干超级用户的事。防御：
sql
-- 收回 PUBLIC 对所有 SECURITY DEFINER 函数的执行权限
REVOKE EXECUTE ON ALL FUNCTIONS IN SCHEMA public FROM PUBLIC;
-- 函数定义里设置 search_path，避免被 schema 劫持
ALTER FUNCTION risky() SET search_path = pg_catalog, public;最后用一个 验证步骤 自测：以 ro 身份连上后，跑这些应该全 ERROR：
sql
INSERT INTO users VALUES (...); -- ✗ permission denied
UPDATE orders SET status='x'; -- ✗ permission denied
CREATE TABLE foo(id int); -- ✗ permission denied for schema publicQ6. 如何审计 PG 的 DDL / DML 操作？(⭐⭐⭐)
考察点：自带日志、pgaudit 扩展、event trigger 自建审计。
答案：
PG 提供 三档审计，按需选型：
(1) 内置 log（轻量、零依赖）
ini
log_connections = on
log_disconnections = on
log_statement = 'ddl' -- 仅 DDL；可选 'mod'（变更）/ 'all'
log_line_prefix = '%m [%p] %u@%d app=%a '优点：无需扩展、性能影响小。缺点：粒度粗，「all」会把日志写爆，且日志在 OS 文件里，离线分析麻烦。
(2) pgaudit 扩展（推荐生产用）
ini
shared_preload_libraries = 'pgaudit'
pgaudit.log = 'write,ddl,role'
pgaudit.log_relation = on特点：
- 类别精细：READ / WRITE / DDL / ROLE / FUNCTION / MISC；
- 支持 对象级审计（
pgaudit.role+ 给该 role GRANT 要审计的对象）； - 输出到 PG 日志，可对接 ELK/Loki。
(3) Event Trigger + 自建审计表（DDL 审计专用）
sql
CREATE TABLE ddl_audit (
occurred_at TIMESTAMPTZ DEFAULT now(),
username TEXT, command_tag TEXT, object_id TEXT, query TEXT
);
CREATE OR REPLACE FUNCTION audit_ddl() RETURNS event_trigger AS $$
DECLARE r record;
BEGIN
FOR r IN SELECT * FROM pg_event_trigger_ddl_commands() LOOP
INSERT INTO ddl_audit(username, command_tag, object_id, query)
VALUES (current_user, r.command_tag, r.object_identity, current_query());
END LOOP;
END $$ LANGUAGE plpgsql;
CREATE EVENT TRIGGER trg_ddl_audit ON ddl_command_end
EXECUTE FUNCTION audit_ddl();之后任何 DDL（CREATE/ALTER/DROP）都自动落库，可以查询，可以告警。
(4) 表级 DML 审计：用第 12 章的触发器，AFTER INSERT/UPDATE/DELETE 写入审计表，记录 OLD / NEW、操作人、IP。
对比建议：
- 合规场景（金融/医疗）：
pgaudit+ 关键表触发器双保险； - 普通业务：内置 log + Event Trigger 即可；
- 不要把所有 DML 都写审计表，否则写入翻倍。
Q7. PUBLIC 默认权限有哪些坑？为什么生产要先「裸奔收权」？(⭐⭐⭐)
考察点：PUBLIC 角色、默认权限收紧、安全基线。
答案：
PUBLIC 是 PG 内置的「所有角色的隐式父组」，相当于「任何人」。新装的 PG 默认给了 PUBLIC 这些权限：
| 对象 | PUBLIC 默认权限 | 风险 |
|---|---|---|
| 所有 DATABASE | CONNECT, TEMP | 任何能登录的账号都能连任意库、建临时表 |
pg_catalog schema | USAGE | 系统目录可读（必要） |
public schema（PG 14 之前） | CREATE, USAGE | 任何账号都能在 public 下建表，污染严重 |
| 函数 | EXECUTE（非 SECURITY DEFINER） | 调用 SECURITY DEFINER 函数潜在越权 |
🎉 PG 15 起，
publicschema 的CREATE权限默认 REVOKE 了，但很多老库升级上来仍是松的，必须人工收权。
生产「安全基线」三连：
sql
-- 1. 任何账号默认不能连这个库
REVOKE CONNECT ON DATABASE learn_pg FROM PUBLIC;
-- 2. 任何账号默认不能在 public schema 建对象
REVOKE CREATE ON SCHEMA public FROM PUBLIC;
-- 3. 任何账号默认不能调用所有函数
REVOKE EXECUTE ON ALL FUNCTIONS IN SCHEMA public FROM PUBLIC;
ALTER DEFAULT PRIVILEGES IN SCHEMA public
REVOKE EXECUTE ON FUNCTIONS FROM PUBLIC;收权完之后，再按业务白名单逐个 GRANT 给 app_rw、app_ro 等角色。
易错点：
- 容易忘的是
pg_catalog的 USAGE 不能收，否则连\d都跑不了； pg_dump时如果某些对象本身被 PUBLIC GRANT 了，导出会带这些 GRANT，恢复到新库时把不严格的权限带进去——所以 dump 后要校对 ACL；- 改 PUBLIC 权限后，所有现有账号立即受影响，要在维护窗口操作。
🔗 延伸阅读
- 第 12 章 服务器编程：用事件触发器审计 DDL，与本章
pgaudit互为补充。 - 第 14 章 备份与恢复：备份脚本对账号权限的要求，何时给
BYPASSRLS。 - 第 19 章 综合项目：把本章 RLS、最小权限账号矩阵落到一个完整的多租户项目里。
本章完。下一章
14_backup_recovery.md：备份与恢复，物理 vs 逻辑、PITR 时光机、生产备份策略。
🎬 可视化演示
演示加载缓慢或样式异常？点此在新标签页打开 ↗
💻 示例代码
python
"""
第 13 章 · 演示 1：ROLE 创建、组继承、GRANT/REVOKE 全流程
=========================================================
依赖：
pip install "psycopg[binary]>=3.1"
前置：
数据库 learn_pg 已存在，且当前用户是 superuser（postgres）。
建议先 psql -f ../init.sql 跑过初始化，本脚本不依赖 init.sql 中的 ch13_tickets 表，
会创建/销毁自己专属的演示对象。
运行：
python 01_role_grant.py
"""
from __future__ import annotations
import psycopg
DSN = "host=127.0.0.1 port=5432 dbname=learn_pg user=postgres"
def line(title: str) -> None:
print("\n" + "=" * 60)
print(title)
print("=" * 60)
def show_roles(conn: psycopg.Connection, names: list[str]) -> None:
"""打印若干 ROLE 的属性。"""
sql = """
SELECT rolname,
rolsuper, rolcreatedb, rolcreaterole,
rolcanlogin, rolinherit, rolbypassrls,
rolconnlimit, rolvaliduntil
FROM pg_roles
WHERE rolname = ANY(%s)
ORDER BY rolname;
"""
with conn.cursor() as cur:
cur.execute(sql, (names,))
rows = cur.fetchall()
for r in rows:
print(f" {r[0]:<10} super={r[1]} createdb={r[2]} createrole={r[3]} "
f"login={r[4]} inherit={r[5]} bypassrls={r[6]} conn_limit={r[7]}")
def cleanup(conn: psycopg.Connection) -> None:
"""删掉演示用的对象（幂等）。"""
with conn.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS demo_payroll CASCADE;")
for role in ("hr_alice", "hr_bob", "engineer_carol", "hr_team"):
cur.execute(f"DROP OWNED BY {role} CASCADE;") if role_exists(cur, role) else None
cur.execute(f"DROP ROLE IF EXISTS {role};")
conn.commit()
def role_exists(cur: psycopg.Cursor, name: str) -> bool:
cur.execute("SELECT 1 FROM pg_roles WHERE rolname = %s", (name,))
return cur.fetchone() is not None
def step1_create_roles(conn: psycopg.Connection) -> None:
line("Step 1：创建组角色 + 用户角色")
with conn.cursor() as cur:
cur.execute("CREATE ROLE hr_team NOLOGIN;") # 组
cur.execute("CREATE ROLE hr_alice LOGIN PASSWORD 'Hr_Alice@2025' INHERIT;")
cur.execute("CREATE ROLE hr_bob LOGIN PASSWORD 'Hr_Bob@2025' NOINHERIT;")
cur.execute("CREATE ROLE engineer_carol LOGIN PASSWORD 'Eng_Carol@2025';")
cur.execute("GRANT hr_team TO hr_alice, hr_bob;")
conn.commit()
show_roles(conn, ["hr_alice", "hr_bob", "engineer_carol", "hr_team"])
def step2_create_table_and_grant(conn: psycopg.Connection) -> None:
line("Step 2：创建一张 demo_payroll 表，把 SELECT/UPDATE 授权给 hr_team")
with conn.cursor() as cur:
cur.execute("""
CREATE TABLE demo_payroll(
id SERIAL PRIMARY KEY,
emp_name TEXT,
salary NUMERIC(10,2)
);
""")
cur.execute("INSERT INTO demo_payroll(emp_name, salary) VALUES "
"('Alice', 18000), ('Bob', 22000), ('Carol', 35000);")
cur.execute("GRANT SELECT, UPDATE ON demo_payroll TO hr_team;")
# SERIAL 列依赖 sequence
cur.execute("GRANT USAGE, SELECT, UPDATE ON SEQUENCE demo_payroll_id_seq TO hr_team;")
conn.commit()
print(" 表已建好，权限授给 hr_team 组。")
def step3_inherit_vs_noinherit() -> None:
line("Step 3：演示 INHERIT vs NOINHERIT 的差别")
# hr_alice INHERIT，连上来直接能 SELECT
with psycopg.connect("host=127.0.0.1 port=5432 dbname=learn_pg "
"user=hr_alice password=Hr_Alice@2025") as alice:
with alice.cursor() as cur:
cur.execute("SELECT count(*) FROM demo_payroll;")
print(f" hr_alice (INHERIT) SELECT 成功，行数 = {cur.fetchone()[0]}")
# hr_bob NOINHERIT，连上来默认不带 hr_team 权限，必须 SET ROLE
with psycopg.connect("host=127.0.0.1 port=5432 dbname=learn_pg "
"user=hr_bob password=Hr_Bob@2025") as bob:
with bob.cursor() as cur:
try:
cur.execute("SELECT count(*) FROM demo_payroll;")
print(f" hr_bob (NOINHERIT) 直接 SELECT = {cur.fetchone()[0]}（不应该看到这行）")
except psycopg.errors.InsufficientPrivilege as e:
print(f" hr_bob (NOINHERIT) 直接 SELECT → 被拒绝：{type(e).__name__}")
bob.rollback()
cur.execute("SET ROLE hr_team;")
cur.execute("SELECT count(*) FROM demo_payroll;")
print(f" hr_bob 在 SET ROLE hr_team 后 SELECT 成功，行数 = {cur.fetchone()[0]}")
def step4_engineer_denied() -> None:
line("Step 4：未授权账号 engineer_carol 应该被拒绝")
with psycopg.connect("host=127.0.0.1 port=5432 dbname=learn_pg "
"user=engineer_carol password=Eng_Carol@2025") as eng:
with eng.cursor() as cur:
try:
cur.execute("SELECT count(*) FROM demo_payroll;")
except psycopg.errors.InsufficientPrivilege:
print(" engineer_carol SELECT demo_payroll → 被拒绝 ✅")
def step5_show_acl(conn: psycopg.Connection) -> None:
line("Step 5：查看 ACL（access privileges）")
with conn.cursor() as cur:
cur.execute("""
SELECT relname, relacl
FROM pg_class
WHERE relname = 'demo_payroll';
""")
row = cur.fetchone()
print(f" 表 {row[0]} 的 ACL = {row[1]}")
print(" 解读：每条形如 'grantee=权限/grantor'，权限缩写见文档 13.2.2 节")
def main() -> None:
with psycopg.connect(DSN) as conn:
cleanup(conn)
try:
step1_create_roles(conn)
step2_create_table_and_grant(conn)
step3_inherit_vs_noinherit()
step4_engineer_denied()
step5_show_acl(conn)
finally:
line("收尾：清理演示对象")
cleanup(conn)
print(" 已清理。")
if __name__ == "__main__":
main()python
"""
第 13 章 · 演示 2：ALTER DEFAULT PRIVILEGES
============================================
场景：DBA 给 ro 角色授权了「未来所有由 dba_user 创建的表」的 SELECT；
然后 dba_user 新建一张表，验证 ro 不需要再 GRANT 就能读。
依赖：
pip install "psycopg[binary]>=3.1"
需先 psql -f ../init.sql 跑过初始化（init.sql 已设置好 dba_user / analyst
与默认权限规则）。
运行：
python 02_default_privileges.py
"""
from __future__ import annotations
import psycopg
ADMIN_DSN = "host=127.0.0.1 port=5432 dbname=learn_pg user=postgres"
DBA_DSN = "host=127.0.0.1 port=5432 dbname=learn_pg user=dba_user password=Dba@2025"
RO_DSN = "host=127.0.0.1 port=5432 dbname=learn_pg user=analyst password=Analyst@2025"
def line(t: str) -> None:
print("\n" + "=" * 60 + "\n" + t + "\n" + "=" * 60)
def show_default_privs() -> None:
"""展示当前 ALTER DEFAULT PRIVILEGES 规则。"""
sql = """
SELECT pg_get_userbyid(defaclrole) AS for_role,
nspname,
defaclobjtype, -- r=table, S=sequence, f=function...
defaclacl
FROM pg_default_acl d
JOIN pg_namespace n ON n.oid = d.defaclnamespace
ORDER BY for_role, nspname;
"""
with psycopg.connect(ADMIN_DSN) as c, c.cursor() as cur:
cur.execute(sql)
rows = cur.fetchall()
print(" 当前 pg_default_acl 中的规则：")
for r in rows:
kind = {'r': '表', 'S': '序列', 'f': '函数', 'T': '类型'}.get(r[2], r[2])
print(f" FOR ROLE {r[0]:<10} IN SCHEMA {r[1]:<10} ON {kind}: {r[3]}")
def main() -> None:
line("Step 1：查看当前默认权限规则（init.sql 已设置）")
show_default_privs()
# ★ 关键：以 dba_user 身份新建一张表
line("Step 2：dba_user 在 public 下新建一张表 future_table")
with psycopg.connect(DBA_DSN) as dba, dba.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS future_table;")
cur.execute("""
CREATE TABLE future_table(
id BIGSERIAL PRIMARY KEY,
memo TEXT
);
""")
cur.execute("INSERT INTO future_table(memo) VALUES ('hello'), ('world');")
dba.commit()
# 立刻看一下这张表的 ACL，应该已经包含 analyst=r/dba_user
cur.execute("""
SELECT relname, relacl
FROM pg_class
WHERE relname = 'future_table';
""")
row = cur.fetchone()
print(f" 新表 {row[0]} 的 ACL = {row[1]}")
# 用 analyst 连过去，应该可以直接 SELECT，不需要再 GRANT
line("Step 3：analyst 直接 SELECT future_table（验证默认权限生效）")
with psycopg.connect(RO_DSN) as ro, ro.cursor() as cur:
try:
cur.execute("SELECT count(*) FROM future_table;")
print(f" ✅ analyst SELECT 成功，行数 = {cur.fetchone()[0]}")
except psycopg.errors.InsufficientPrivilege as e:
print(f" ❌ analyst SELECT 失败：{e}")
# 进一步：尝试 INSERT，应失败（默认权限只 GRANT SELECT）
line("Step 4：analyst 尝试 INSERT，应被拒绝")
with psycopg.connect(RO_DSN) as ro, ro.cursor() as cur:
try:
cur.execute("INSERT INTO future_table(memo) VALUES ('hack');")
print(" ❌ analyst INSERT 居然成功了，规则有问题！")
except psycopg.errors.InsufficientPrivilege:
print(" ✅ analyst INSERT 被拒绝（默认权限只授了 SELECT）")
ro.rollback()
# 收尾
line("收尾：删除 future_table")
with psycopg.connect(DBA_DSN) as dba, dba.cursor() as cur:
cur.execute("DROP TABLE IF EXISTS future_table;")
dba.commit()
print(" 已清理。")
if __name__ == "__main__":
main()python
"""
第 13 章 · 演示 3：行级安全 RLS · 多租户隔离
==============================================
场景：ch13_tickets 表有 tenant_id 字段，开启 RLS。
用 tenant_a / tenant_b 两个账号分别连进去，验证：
1) 各自只看到自己 tenant 的数据；
2) 不能跨 tenant 写入；
3) RESTRICTIVE 策略屏蔽 archived；
4) dba_user (BYPASSRLS) 看到全部。
依赖：
pip install "psycopg[binary]>=3.1"
必须先 psql -f ../init.sql 跑过初始化。
运行：
python 03_rls_multi_tenant.py
"""
from __future__ import annotations
import psycopg
DSN_BASE = "host=127.0.0.1 port=5432 dbname=learn_pg "
USERS = {
"tenant_a": ("TenantA@2025", 1), # (密码, 租户号)
"tenant_b": ("TenantB@2025", 2),
"dba_user": ("Dba@2025", None), # BYPASSRLS，无需 setting
}
def line(t: str) -> None:
print("\n" + "=" * 60 + "\n" + t + "\n" + "=" * 60)
def query_as(user: str, password: str, tenant: int | None) -> None:
"""以指定用户身份查询 ch13_tickets，可选注入 app.tenant_id。"""
dsn = f"{DSN_BASE}user={user} password={password}"
print(f"\n--- 以 {user} 身份连接 (tenant={tenant}) ---")
with psycopg.connect(dsn) as conn, conn.cursor() as cur:
cur.execute("BEGIN")
if tenant is not None:
cur.execute("SET LOCAL app.tenant_id = %s", (str(tenant),))
cur.execute("SELECT id, tenant_id, title, status FROM ch13_tickets ORDER BY id;")
rows = cur.fetchall()
for r in rows:
print(f" id={r[0]:<3} tenant={r[1]} status={r[2]:<8} title={r[3]}")
if not rows:
print(" (空)")
conn.rollback()
def try_cross_tenant_insert() -> None:
"""tenant_a 试图写入 tenant_id=2 的行，应被 WITH CHECK 拦截。"""
line("Step 2：tenant_a 试图越权插入 tenant_id=2 的行")
dsn = f"{DSN_BASE}user=tenant_a password=TenantA@2025"
with psycopg.connect(dsn) as conn, conn.cursor() as cur:
cur.execute("BEGIN")
cur.execute("SET LOCAL app.tenant_id = '1'") # 自报身份是租户1
try:
cur.execute(
"INSERT INTO ch13_tickets(tenant_id, title) VALUES (2, '我是 A 试图伪装成 B')"
)
print(" ❌ 写入成功了，RLS 失效！请检查 init.sql。")
except psycopg.errors.InsufficientPrivilege as e:
# 注：触发 WITH CHECK 失败时报的是 InsufficientPrivilege 或 CheckViolation
print(f" ✅ 被 WITH CHECK 拦截：{type(e).__name__}: {str(e).splitlines()[0]}")
except Exception as e:
print(f" ✅ 被拦截：{type(e).__name__}: {str(e).splitlines()[0]}")
conn.rollback()
def show_archived_filter() -> None:
"""RESTRICTIVE 策略：ch13_no_archived 应该让 archived 行隐身（即使 owner 也不行）。"""
line("Step 3：验证 RESTRICTIVE 策略 ch13_no_archived（archived 行不可见）")
dsn = f"{DSN_BASE}user=tenant_a password=TenantA@2025"
with psycopg.connect(dsn) as conn, conn.cursor() as cur:
cur.execute("BEGIN")
cur.execute("SET LOCAL app.tenant_id = '1'")
cur.execute("SELECT status, count(*) FROM ch13_tickets GROUP BY status ORDER BY status;")
for r in cur.fetchall():
print(f" status={r[0]:<8} count={r[1]}")
print(" → 应该看不到 archived 状态（init.sql 写入时确实有一行 archived）")
conn.rollback()
def superuser_view() -> None:
line("Step 4：用 superuser (postgres) 连接，确认数据库里实际数据全貌")
dsn = f"{DSN_BASE}user=postgres"
with psycopg.connect(dsn) as conn, conn.cursor() as cur:
# 由于 init.sql 用了 FORCE ROW LEVEL SECURITY，但 superuser 默认 BYPASSRLS
cur.execute("SELECT tenant_id, status, count(*) FROM ch13_tickets "
"GROUP BY tenant_id, status ORDER BY tenant_id, status;")
for r in cur.fetchall():
print(f" tenant={r[0]} status={r[1]:<8} count={r[2]}")
def main() -> None:
line("Step 1：以两个租户身份分别 SELECT，应该只看到自己的行")
query_as("tenant_a", "TenantA@2025", 1)
query_as("tenant_b", "TenantB@2025", 2)
try_cross_tenant_insert()
show_archived_filter()
superuser_view()
line("Step 5：dba_user (BYPASSRLS) 不需要设置 app.tenant_id 也能看全表")
query_as("dba_user", "Dba@2025", None)
if __name__ == "__main__":
main()python
"""
第 13 章 · 演示 4：pg_hba.conf 各 METHOD 差异（说明 + 自检）
==============================================================
本脚本不会修改 pg_hba.conf（那需要 root + reload 服务），而是：
1) 解析当前数据库的 pg_hba_file_rules 视图，把规则可读化打印；
2) 解析 pg_settings 中关键参数（password_encryption / ssl 等）；
3) 对若干 METHOD 给出实操建议与示范的 pg_hba 行；
4) 简单实验：尝试用 sslmode=disable / require / verify-ca 连接，看哪些被拒。
依赖：
pip install "psycopg[binary]>=3.1"
运行：
python 04_pg_hba_demo.py
"""
from __future__ import annotations
import psycopg
DSN = "host=127.0.0.1 port=5432 dbname=learn_pg user=postgres"
EXAMPLES = [
("trust",
"local all postgres trust",
"本地 socket 不验密码，仅适合 docker 容器内、CI 临时调试。生产严禁。"),
("md5",
"host all all 0.0.0.0/0 md5",
"MD5 哈希存储，PG 14+ 不再推荐；存在彩虹表/碰撞风险。"),
("scram-sha-256",
"host all all 0.0.0.0/0 scram-sha-256",
"PG 14+ 默认推荐，挑战-应答协议，密码不上网。"),
("cert",
"hostssl all all 0.0.0.0/0 cert clientcert=verify-full",
"客户端必须出示受信 CA 颁发的证书，可零密码登录，机器对机器最安全。"),
("peer",
"local all all peer map=devmap",
"用 OS 用户名作为 PG 用户名，仅本地 unix socket 有效。"),
("ident",
"host all all 10.0.0.0/8 ident map=devmap",
"通过 ident 协议询问对方 OS 用户，需要客户端机器跑 identd，几乎已废弃。"),
("ldap",
"host all all 0.0.0.0/0 ldap ldapserver=ldap.corp ldapprefix=\"cn=\" ldapsuffix=\",ou=People,dc=corp\"",
"对接公司 AD/LDAP，统一身份。"),
("reject",
"host all all 192.168.99.0/24 reject",
"显式拒绝，匹配后立即 ERROR，不再向下匹配。"),
]
def line(t: str) -> None:
print("\n" + "=" * 60 + "\n" + t + "\n" + "=" * 60)
def show_current_rules(conn: psycopg.Connection) -> None:
"""从 pg_hba_file_rules 视图读取当前生效的规则。"""
sql = """
SELECT line_number, type, database, user_name,
address, netmask, auth_method, options, error
FROM pg_hba_file_rules
ORDER BY line_number;
"""
with conn.cursor() as cur:
cur.execute(sql)
rows = cur.fetchall()
print(f" pg_hba.conf 当前共 {len(rows)} 条规则（按文件顺序匹配）：")
print(f" {'#':<4}{'TYPE':<8}{'DB':<10}{'USER':<10}{'ADDR':<22}{'METHOD':<18}ERROR")
for r in rows:
addr = r[4] or '-'
print(f" {r[0]:<4}{r[1]:<8}{','.join(r[2]):<10}{','.join(r[3]):<10}"
f"{addr:<22}{r[6]:<18}{r[8] or ''}")
def show_security_settings(conn: psycopg.Connection) -> None:
keys = ['password_encryption', 'ssl', 'ssl_cert_file',
'ssl_key_file', 'authentication_timeout',
'log_connections', 'log_disconnections']
sql = "SELECT name, setting FROM pg_settings WHERE name = ANY(%s)"
with conn.cursor() as cur:
cur.execute(sql, (keys,))
rows = cur.fetchall()
print(f" {'参数':<28}值")
for k, v in rows:
print(f" {k:<28}{v}")
def explain_methods() -> None:
print(f" {'METHOD':<16}示例规则")
print(f" {'-'*16}{'-'*60}")
for name, example, desc in EXAMPLES:
print(f" {name:<16}{example}")
print(f" {'':<16}↳ {desc}\n")
def try_sslmodes() -> None:
"""尝试不同 sslmode 连接，观察成功/失败。仅做现象展示。"""
print(" 注意：本地 PG 若未开 ssl，require/verify-* 会失败；这是预期。")
for mode in ("disable", "prefer", "require", "verify-ca", "verify-full"):
dsn = f"host=127.0.0.1 port=5432 dbname=learn_pg user=postgres sslmode={mode}"
try:
with psycopg.connect(dsn, connect_timeout=3) as c, c.cursor() as cur:
cur.execute("SELECT current_setting('ssl')")
ssl_on = cur.fetchone()[0]
# 是否真用 SSL：psycopg v3 暴露 connection.info.ssl_in_use
in_use = c.info.ssl_in_use if hasattr(c.info, "ssl_in_use") else "?"
print(f" sslmode={mode:<12} → ✅ 连上，ssl_in_use={in_use} (server ssl={ssl_on})")
except Exception as e:
print(f" sslmode={mode:<12} → ❌ {type(e).__name__}: {str(e).splitlines()[0]}")
def main() -> None:
with psycopg.connect(DSN) as conn:
line("Step 1：当前 pg_hba.conf 规则一览")
show_current_rules(conn)
line("Step 2：与认证/SSL 相关的 GUC 参数")
show_security_settings(conn)
line("Step 3：常见 METHOD 速查 + 示例")
explain_methods()
line("Step 4：实测不同 sslmode 的连接行为")
try_sslmodes()
line("生产建议总结")
print("""
① 公网/办公网入口的规则一律用 hostssl + scram-sha-256；
② 本地 socket 用 peer + map 文件，禁止 trust；
③ 复制账号单独开 type=replication，IP 限定为对端从库；
④ password_encryption 设 scram-sha-256；老用户必须重设一次密码；
⑤ 修改 pg_hba.conf 后只需 reload，不必 restart：
SELECT pg_reload_conf();
""")
if __name__ == "__main__":
main()markdown
# 第 13 章 配套代码 · 权限与安全
## 准备工作
1. 跑 `psql -h 127.0.0.1 -U postgres -d learn_pg -f ../init.sql` 初始化测试表与角色
2. 安装依赖：`pip install "psycopg[binary]>=3.1"`
3. （可选）通过环境变量覆盖默认连接：`export PG_DSN="host=... port=... dbname=... user=..."`
4. 本章脚本默认以 `postgres` 超级用户连接。运行 03 时还会以 `tenant_a / tenant_b / dba_user` 三个角色登录，密码已写在 `init.sql`。
## 脚本一览（推荐运行顺序）
| 脚本 | 一句话说明 | 关键 PG 特性 |
|------|------------|--------------|
| `01_role_grant.py` | 演示 ROLE 创建、组继承、`GRANT/REVOKE` 全流程，最后清理 | `CREATE ROLE`、`INHERIT`、`SET ROLE` |
| `02_default_privileges.py` | 验证 `ALTER DEFAULT PRIVILEGES` 让未来对象也自动授权 | `ALTER DEFAULT PRIVILEGES`、PUBLIC 收权 |
| `03_rls_multi_tenant.py` | 多租户 RLS 隔离，跨租户写入被 `WITH CHECK` 拦截 | RLS、`USING / WITH CHECK`、`FORCE RLS`、`BYPASSRLS` |
| `04_pg_hba_demo.py` | 解析当前连接的认证元信息，列举 `pg_hba.conf` 优先级与常见报错 | `pg_hba.conf`、`scram-sha-256`、`pg_hba_file_rules` |
## 预期输出
`03_rls_multi_tenant.py` 关键片段（命中 RLS 后只能看到自己 tenant 的工单）：============================================================ Step 1：以两个租户身份分别 SELECT，应该只看到自己的行
--- 以 tenant_a 身份连接 (tenant=1) --- id=1 tenant=1 status=open title=租户A-工单001 id=2 tenant=1 status=closed title=租户A-工单002
--- 以 tenant_b 身份连接 (tenant=2) --- id=4 tenant=2 status=open title=租户B-工单001 ...
注意：`init.sql` 写入了一行 `archived` 工单，但 `ch13_no_archived` 的 RESTRICTIVE 策略会让它对所有人不可见。
## 常见报错与依赖
- `connection refused` → PG 未启动，或 `pg_hba.conf` 没放行本机
- `relation "ch13_tickets" does not exist` → 未运行 `init.sql`
- `password authentication failed for user "tenant_a"` → 改过密码后忘记同步脚本里的密码常量
- `permission denied for table ch13_tickets` → 当前角色不在 `devs` 组里，回到 `init.sql` 检查 `GRANT devs TO ...`
- `04_pg_hba_demo.py` 仅做只读检查，不会改 `pg_hba.conf`；如需修改请直接编辑 `$PGDATA/pg_hba.conf` 后 `SELECT pg_reload_conf();`01_role_grant.py ↗ · 02_default_privileges.py ↗ · 03_rls_multi_tenant.py ↗ · 04_pg_hba_demo.py ↗ · README.md ↗