diff --git a/README.md b/README.md index 90eaf0d..8f4c084 100644 --- a/README.md +++ b/README.md @@ -127,30 +127,7 @@ resp = await client.chat( 存储上 `tenant_id` 两端都是 `TEXT NOT NULL DEFAULT ''`,`meta` 在 Postgres 是 `JSONB`、在 SQLite 是 `TEXT`;老表要补上这两列(补列是否由库自动执行取决于 `PGW_TELEMETRY_SCHEMA_MODE`,见[遥测表 schema 与升级纪律](#遥测表-schema-与升级纪律)),**补列后老行读出是空串而非 NULL**(NULL 在任何 RLS policy 下都对所有人不可见,空串则可用一条 SQL 审出还有多少行待归属)。 -**库只提供列,不启用 RLS、不建索引。** 要数据库层的强制隔离,以下 DDL 是**下游 DBA 的职责,库不会代劳**;不执行则 `tenant_id` 只是一个可查可过滤的普通列,没有任何数据库层强制。库不代劳的原因是 default-deny:启用 RLS 而没有匹配的 policy = 零行可写且静默不报错,会让非多租户部署的遥测全量写失败。 - -```sql -ALTER TABLE llm_calls ENABLE ROW LEVEL SECURITY; -ALTER TABLE llm_calls FORCE ROW LEVEL SECURITY; -- 属主不豁免 -CREATE POLICY llm_calls_tenant_isolation ON llm_calls TO polygateway_app - USING (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')) - WITH CHECK (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')); -``` - -```sql -CREATE INDEX CONCURRENTLY idx_llm_calls_tenant_created - ON llm_calls (tenant_id, created_at); -``` - -`current_setting(..., true)` 的第二参数令 GUC 未设时返回 NULL 而非抛错,外层 `NULLIF` 把空串归一为 NULL——合起来使**未设租户 = 零行**(fail-closed)而不是全部行。索引列序不可颠倒:启用 RLS 后 policy 给每条查询隐式追加 `tenant_id` 等值谓词,它出现在 100% 的谓词里,必然是前导列。 - -三个陷阱,每一个的失败形态都是**静默的**: - -| 陷阱 | 后果 | -|---|---| -| 表属主默认**豁免** RLS | 只写 `ENABLE` 而漏 `FORCE`,用属主角色连库时隔离形同虚设,且查询一切正常看不出来 | -| 租户上下文必须在**显式事务内**用 `set_config('app.tenant_id', ..., true)` | asyncpg 默认 autocommit,单发 `SET LOCAL` 会当场失效,而 PG **只发 warning 不报错**;表现是 policy 永远拿不到租户 → fail-closed 到零行 | -| policy 必须同时写 `USING` 与 `WITH CHECK` | 只写前者则租户 A 读不到 B 的行,却**能插入标着 B 的行**——污染发生在写入侧,读侧查不出来 | +**库只提供列,不启用 RLS、不建索引。** 数据库层的强制隔离是**下游 DBA 的职责,库不会代劳**;不执行则 `tenant_id` 只是一个可查可过滤的普通列,没有任何数据库层强制。库不代劳的原因是 default-deny:启用 RLS 而没有匹配的 policy = 零行可写且静默不报错,会让非多租户部署的遥测全量写失败。三角色、RLS policy、分区与保留期的完整可执行模板见[生产部署 DDL 模板](#生产部署-ddl-模板postgresql)。 ## 遥测表 schema 与升级纪律 @@ -210,6 +187,176 @@ PG 变体的补列语句带 `ADD COLUMN IF NOT EXISTS`,**整段可重复执行** | 库从不 `SELECT *`,也从不读回这张表的数据 | 库侧根本没有读路径,你加索引、加自己的列、挂 RLS 都影响不到它 | | 写入的冲突处理**不绑定具体约束** | 你可以把 `llm_calls` 建成 `PARTITION BY RANGE (created_at)` 的分区表(此时主键必须是 `(call_id, created_at)`,PG 要求分区表唯一约束含分区键),库的探测、补列与写入照常工作 | +## 生产部署 DDL 模板(PostgreSQL) + +上一节讲的是**库怎么对待这张表**(只探测、只 INSERT、可选建表);本节讲的是**你该把这张表部署成什么样**:谁能读、谁能写、写进去的行能不能被改、存多久。这些库一件都不代劳——它没有、也不该有这些权限。 + + + +模板按下表顺序执行,标识符(角色名、schema、分区月份、密码)按你的环境改;`llm_calls` 一律不写 schema 限定,靠 `search_path` 解析,与库的写入口径一致。 + +| # | 锚点 | 做什么 | +|---|---|---| +| 1 | `roles` | 建三角色并授 schema 级权限 | +| 2 | `table` | 把 `llm_calls` 改造成按 `created_at` 的 RANGE 分区表,属主归 `polygateway_owner` | +| 3 | `partition` | 建一个月分区(生产用 `pg_partman` 自动滚动) | +| 4 | `grants` | 授表级权限并 `REVOKE UPDATE, DELETE` | +| 5 | `immutable` | 触发器兜底(只防误操作) | +| 6 | `rls` | 启用并 `FORCE` RLS + 两条 policy | +| 7 | `index` | `(tenant_id, created_at)` 复合索引 | + +### 1. 三角色 + +| 角色 | 拿到什么 | 谁在用 | +|---|---|---| +| `polygateway_owner` | 表属主:DDL、加分区、删分区 | DBA / 定时任务;**不用它连库跑业务** | +| `polygateway_app` | `INSERT` + 受 RLS 约束的 `SELECT` | 库的连接串用这个 | +| `polygateway_report` | 受 RLS 约束的 `SELECT` | BI、对账、成本报表 | + + + +```sql +CREATE ROLE polygateway_owner NOLOGIN; +CREATE ROLE polygateway_app LOGIN PASSWORD 'CHANGE_ME_APP'; +CREATE ROLE polygateway_report LOGIN PASSWORD 'CHANGE_ME_REPORT'; +GRANT polygateway_owner TO CURRENT_USER; -- 下一块要把表属主改过去,须先成为它的成员 +GRANT USAGE ON SCHEMA public TO polygateway_owner, polygateway_app, polygateway_report; +GRANT CREATE ON SCHEMA public TO polygateway_owner; -- 滚动分区要在该 schema 里建表 +``` + +### 2. 分区表 + +分区表**必须下游先手工建**:库的 `CREATE TABLE` 只会建普通表。列不在这里重抄一份——抄了就会漂移,故先用库自带脚本建出普通表,再原地改造: + +```bash +python -c "import polygateway; print(polygateway.telemetry_schema_sql('postgres'))" \ + | psql "$PGW_TELEMETRY_PG_DSN" +``` + + + +```sql +ALTER TABLE llm_calls RENAME TO llm_calls_seed; -- 上一步建出的普通表当模子 +CREATE TABLE llm_calls ( + LIKE llm_calls_seed INCLUDING DEFAULTS, -- 列/类型/NOT NULL/DEFAULT 全照搬 + PRIMARY KEY (call_id, created_at) -- 分区表的唯一约束必须含分区键 +) PARTITION BY RANGE (created_at); +DROP TABLE llm_calls_seed; +ALTER TABLE llm_calls OWNER TO polygateway_owner; +``` + + + +```sql +CREATE TABLE llm_calls_2026_01 PARTITION OF llm_calls + FOR VALUES FROM ('2026-01-01 00:00:00+00') TO ('2026-02-01 00:00:00+00'); +ALTER TABLE llm_calls_2026_01 OWNER TO polygateway_owner; +``` + +生产不要手工滚月份,交给 [`pg_partman`](https://github.com/pgpartman/pg_partman):5.x 用 `create_parent(p_parent_table := 'public.llm_calls', p_control := 'created_at', p_interval := '1 month')`(4.x 的参数序不同,以你装的版本文档为准),再把 `part_config.retention` 设成 `'6 months'`、`retention_keep_table` 设成 `false`,`run_maintenance_proc()` 就会到期 `DROP` 整个分区。清理必须走 `DETACH`/`DROP PARTITION` 而**不是** `DELETE`——这不是性能偏好,是权限张力的唯一解:下一块要对应用角色 `REVOKE DELETE`,而 `DROP PARTITION` 是属主的 DDL,两者不冲突,`DELETE` 则必然冲突。 + +**分区部署改变了幂等键**,按 `cache_hit` 出报表的下游必须知道:普通表上主键是 `call_id`,分区表上是 `(call_id, created_at)`。库的写入是无冲突目标的 `ON CONFLICT DO NOTHING`,两种表形态都合法;但 `emit_cache_hit` 复用的是响应里的**历史** `call_id`,于是同一次缓存命中的重复回放,在普通表上第二次起被 `DO NOTHING` 吞掉、在分区表上**每次都落一行**(`created_at` 由 `DEFAULT now()` 生成,主键不再重复)。逐次尝试行不受影响(每次尝试都是新 `call_id`)。 + +### 3. 权限与不可变性 + +`llm_calls` 按**不可变审计表**对待:写进去的行谁都不许改、不许删,过期数据靠 `DROP PARTITION` 整块消失。 + + + +```sql +GRANT INSERT, SELECT ON llm_calls TO polygateway_app; +GRANT SELECT ON llm_calls TO polygateway_report; +REVOKE UPDATE, DELETE, TRUNCATE ON llm_calls FROM polygateway_app, polygateway_report; +``` + + + +```sql +CREATE FUNCTION llm_calls_reject_mutation() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + RAISE EXCEPTION 'llm_calls 是不可变审计表,% 被拒绝', TG_OP; +END; +$$; +CREATE TRIGGER llm_calls_immutable BEFORE UPDATE OR DELETE ON llm_calls + FOR EACH ROW EXECUTE FUNCTION llm_calls_reject_mutation(); +``` + +触发器**只防误操作,不防恶意**:表属主可以 `ALTER TABLE llm_calls DISABLE TRIGGER llm_calls_immutable` 把它关掉。真正的强制是上一块的 `REVOKE`——权限检查发生在触发器之前,应用角色连触发器都碰不到。要防属主本人,需要的是数据库之外的手段(WAL 归档、只追加的外部存证),不是本表能解决的。 + +`DROP PARTITION` 与 `DETACH PARTITION` 是 DDL,**不会触发**行级触发器,故保留期清理不受这一块影响。 + +### 4. 行级安全与多租户隔离 + + + +```sql +ALTER TABLE llm_calls ENABLE ROW LEVEL SECURITY; +ALTER TABLE llm_calls FORCE ROW LEVEL SECURITY; -- 属主不豁免 +CREATE POLICY llm_calls_app_write ON llm_calls FOR INSERT TO polygateway_app + WITH CHECK (true); +CREATE POLICY llm_calls_app_read ON llm_calls FOR SELECT TO polygateway_app + USING (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')); +CREATE POLICY llm_calls_report_read ON llm_calls FOR SELECT TO polygateway_report + USING (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')); +``` + + + +```sql +CREATE INDEX idx_llm_calls_tenant_created ON llm_calls (tenant_id, created_at); +``` + +`current_setting(..., true)` 的第二参数令 GUC 未设时返回 NULL 而非抛错,外层 `NULLIF` 把空串归一为 NULL——合起来使**未设租户 = 零行**(fail-closed)而不是全部行。索引列序不可颠倒:启用 RLS 后 policy 给每条查询隐式追加 `tenant_id` 等值谓词,它出现在 100% 的谓词里,必然是前导列。分区表上**不能**用 `CREATE INDEX CONCURRENTLY`(PG 不支持在分区父表上并发建索引);父表此时还没有数据,直接建即可,给已有数据的普通表补索引才需要逐个分区 `CONCURRENTLY`。 + +**写侧 policy 为什么是 `WITH CHECK (true)` 而不是等值比较**:库用一个连接池给**所有**租户写遥测,且从不发 `set_config('app.tenant_id', ...)`(源码里没有这条语句)。把写侧也绑到 GUC 上,库的每一条 `INSERT` 都会被 policy 拒绝——而遥测的失败方向是静默降级,表现是逐行 warning + 整表零行。隔离在这个模型里由**读侧**承担:写入方是库自己(可信),读取方才是要隔离的人。若你的调用点保证每次调用都带 `tenant_id`,可把写侧收紧成 `WITH CHECK (tenant_id <> '')`,代价是漏传 `tenant_id` 的调用点会**丢遥测行**(只留一条 warning)。 + +四个陷阱,每一个的失败形态都是**静默的**: + +| 陷阱 | 后果 | +|---|---| +| 表属主默认**豁免** RLS | 只写 `ENABLE` 而漏 `FORCE`,用属主角色连库时隔离形同虚设,且查询一切正常看不出来 | +| `FORCE` 之后属主自己也被 policy 管 | 模板没给 `polygateway_owner` 任何 policy,故它读不到、也写不进任何行——这是有意的(它只用来做 DDL),但别拿它跑报表 | +| 租户上下文必须在**显式事务内**用 `set_config('app.tenant_id', ..., true)` | asyncpg 默认 autocommit,单发 `SET LOCAL` 会当场失效,而 PG **只发 warning 不报错**;表现是 policy 永远拿不到租户 → fail-closed 到零行 | +| 读侧 policy 漏写 `USING` | `FOR SELECT` 的 policy 只认 `USING`;写成 `WITH CHECK` 不报错也不生效,隔离直接落空 | + +### 5. 库本身需要的最小权限 + +按上面的模板部署后,库的连接串用 `polygateway_app`,它需要的权限恰好是下表这些——多一分都不必给: + +| 库会发的语句 | 需要什么 | +|---|---| +| 连库 | 数据库 `CONNECT` + schema `USAGE` | +| `SELECT to_regclass('llm_calls')`、查 `pg_attribute`(列探测) | 无需额外授权(系统 catalog 默认对 `PUBLIC` 可读) | +| `INSERT INTO llm_calls (...)` | 表 `INSERT`;RLS 打开后还须有一条允许写的 policy | +| `CREATE TABLE IF NOT EXISTS`(**仅当表不存在**) | schema `CREATE`。生产建议**不给**:表由 `owner` 先建好,库探测到表在就不发这条 | +| `ALTER TABLE ADD COLUMN`(**仅 `PGW_TELEMETRY_SCHEMA_MODE=auto`**) | 表**属主**——PG 的 `ALTER TABLE` 只认属主,这一项无法单独 `GRANT`。PG 侧缺省就是 `manual`,补列交给 DBA | + +### 6. 合规下游的推荐配置 + +三件事(截断、保留期、访问控制)要一起上才有意义,故给一份可直接照抄的组合,而不是让你自己拼: + +```dotenv +PGW_TELEMETRY_BACKEND=postgres +PGW_TELEMETRY_PG_DSN=postgresql://polygateway_app:...@db:5432/telemetry +PGW_TELEMETRY_SCHEMA_MODE=manual # PG 侧本就是缺省;写出来是为了不依赖缺省 +PGW_TELEMETRY_TEXT_CAP=2000 # 落库正文的字符上限;不设 = 存全文 +``` + +| 层 | 配置 | +|---|---| +| 正文体量 | `PGW_TELEMETRY_TEXT_CAP=2000`(按需调);超出部分头部硬切并附 `…(略 N 字)` | +| 保留期 | 上面的分区模板 + `pg_partman` 的 `retention`,过期分区整块 `DROP` | +| 访问控制 | 上面的三角色 + `REVOKE UPDATE, DELETE` + `FORCE` RLS | +| 存量兜底 | 已经攒成一张大普通表、来不及改造分区时,用 `tools/telemetry_retention.py`(默认 dry-run,`--apply` 才动手;探测到分区表会直接退出让路给 `DROP PARTITION`) | + +**`PGW_TELEMETRY_TEXT_CAP` 的覆盖面必须说清,否则合规判断会出错。** cap 落在四处:`messages` 里每条消息的字符串 `content`、多模态 content 数组中 `type == "text"` 的 part 的 `text`,以及 `response` 与 `thinking` 两列。消息侧的这个面与缓存摘要函数 `digest_messages` 一致——**只碰 `content`**,消息里别的字段一概不碰。所以调用方自己塞进 `tool_calls.function.arguments`、`name` 等字段的内容**不在覆盖范围内**:开了 cap 不等于表里没有全文残留。另需知道:缺省是**不截断**(存全文),而截断之后遥测不再是可复现重放的证据。 + +### 7. SQLite 侧的保留期 + +SQLite 侧**不建议**对着一个大库文件跑 `DELETE` + `VACUUM`,而应**按天/按实验轮转库文件**——`runs/.db`、`runs/.db` 这样,到期直接删文件。这是三个现有下游(Video-Tree-TRM5 / CHSAnalyzer / dissect)天然就有的形态,比删行省事也安全得多:删文件是 O(1) 且不可能删错行,而 `VACUUM` 会重写整库、期间需要一倍磁盘空间,还会把并发写入方挡在外面。 + +`tools/telemetry_retention.py` 的 SQLite 分支是给**存量场景**兜底的——已经攒成一个大库、来不及改轮转时用它,不是推荐路径。 + ## 错误模型(四分类) 一切失败在 transport 层翻译为四类之一,治理行为由分类决定,业务侧不需要判断状态码: @@ -254,6 +401,7 @@ PG 变体的补列语句带 `ADD COLUMN IF NOT EXISTS`,**整段可重复执行** | `PGW_CACHE_BACKEND` | `none` / `memory` / `redis`;非 `none` 时需 `PGW_CACHE_NAMESPACE` + `PGW_CACHE_TTL_S`(须 > 0) | | `PGW_TELEMETRY_BACKEND` | `none` / `sqlite`(需 `PGW_TELEMETRY_SQLITE_PATH`)/ `postgres`(需 `PGW_TELEMETRY_PG_DSN`) | | `PGW_TELEMETRY_SCHEMA_MODE` | 可选:`auto` / `manual`;**不设则按后端派生**(sqlite→`auto`、postgres→`manual`),显式设置则两侧都可覆盖。决定库是否给已存在的旧表自动 `ALTER` 补列,详见[遥测表 schema 与升级纪律](#遥测表-schema-与升级纪律) | +| `PGW_TELEMETRY_TEXT_CAP` | 可选正整数:遥测落库正文的字符上限(作用于每条消息的文本 `content`、多模态 part 的 `text`、`response`、`thinking`);**不设 = 不截断**,详见[合规下游的推荐配置](#6-合规下游的推荐配置) | | `PGW_PRICING_PATH` / `PGW_STRUCTURED_MAX_RETRIES` / `PGW_LEASE_TTL_S` | 可选:价格表(缺省则成本恒 `None`)/ 结构化重问上限(缺省 2)/ permit 租约秒数(缺省 1500,须 ≥ 最大源 `TIMEOUT_S`) | 两个易被忽略的源级键:`MISSING_DONE` 决定 SSE 缺 `[DONE]` 时的处置(`retry` 默认判瞬时重试 / `salvage` 收下已收内容并把用量可信度降为 `estimated`;零内容恒 `retry`,不受该键影响);`EXTRA_BODY` 是该源**恒定**的采样参数(JSON 对象串,并入请求体,优先级低于 `chat(overlay=...)`),禁用键 `model` / `messages` / `stream` / `stream_options` 配了直接报错,OCR 与 EMBED scope 不消费该键(配了忽略并 warning)。 diff --git a/tests/integration/test_postgres_telemetry.py b/tests/integration/test_postgres_telemetry.py index dd427cd..e301c2d 100644 --- a/tests/integration/test_postgres_telemetry.py +++ b/tests/integration/test_postgres_telemetry.py @@ -14,7 +14,9 @@ import asyncio import json import os import re +from dataclasses import dataclass from datetime import UTC, datetime, timedelta +from pathlib import Path from uuid import uuid4 import pytest @@ -913,3 +915,297 @@ class TestPublishedSchemaScript: await _execute_script(fresh_dsn, script) # 可重复执行: 第二遍不得抛 rerun = [r["column_name"] for r in await _fetch(fresh_dsn, _PHYSICAL_COLUMNS_SQL, schema)] assert rerun == actual # 且第二遍没有偷偷改动表结构 + + +# --------------------------------------------------------------------------- +# issue #12 Task 4: README 的生产部署 DDL 模板,逐条在真实 PG 上执行 +# --------------------------------------------------------------------------- + +# 模板 SQL **只有一份**,在 README 里。测试从 README 解析出来跑,而不是在这里另抄 +# 一份: 抄一份就是两份会各自漂移的东西,而"README 里的 SQL 能跑"这个承诺恰恰只在 +# 同源时才成立(doctest / Rust doc tests / mdbook test 都是这个范式)。 +_README = Path(__file__).resolve().parents[2] / "README.md" + +# 锚点写成 HTML 注释,渲染时不可见,比按章节标题或代码块序号定位稳固得多。 +_TEMPLATE_BLOCK = re.compile(r"\s*\n```sql\n(.*?)\n```", re.DOTALL) + +# 顺序即执行顺序;数量与名字都钉死——解析不到或多出一块必须当场红, +# 绝不能退化成空列表让这条测试变成永远绿的摆设。 +_EXPECTED_TEMPLATE_BLOCKS = ( + "roles", + "table", + "partition", + "grants", + "immutable", + "rls", + "index", +) + +# README 里必须原样保留、由本测试做受控替换的标识符。README 那份是给下游照抄的, +# 故占位符是**合法可执行的具体值**而不是 `` 之类的尖括号洞。 +_TEMPLATE_PLACEHOLDERS = ( + "polygateway_owner", + "polygateway_app", + "polygateway_report", + "CHANGE_ME_APP", + "CHANGE_ME_REPORT", + "SCHEMA public", + "llm_calls_2026_01", + "'2026-01-01 00:00:00+00'", + "'2026-02-01 00:00:00+00'", +) + +# 应用角色在生产里能发的唯一一类写语句(与库的 INSERT 同形,只列 NOT NULL 列) +_TEMPLATE_INSERT = ( + "INSERT INTO llm_calls (call_id, model, provider, source_name, messages, response, " + "prompt_tokens, completion_tokens, usage_source, latency_ms, tenant_id) " + "VALUES ($1, 'm', 'p', 's1', '[]', 'ok', 1, 2, 'measured', 10, $2)" +) + + +def _template_blocks() -> dict[str, str]: + """从 README 解析带锚点的 SQL 块;顺序即文中出现顺序。""" + return dict(_TEMPLATE_BLOCK.findall(_README.read_text(encoding="utf-8"))) + + +@dataclass(frozen=True) +class _TemplateEnv: + """模板部署完成后的现场句柄:三个角色各自的连接串 + 当月分区名。""" + + admin_dsn: str + app_dsn: str + report_dsn: str + schema: str + partition: str + seeded: tuple[str, str] # (tenant-a 的行, tenant-b 的行) + + +def _localize(sql: str, schema: str, roles: dict[str, str], month: datetime) -> str: + """把 README 里给下游照抄的标识符换成本次运行专属的临时对象。 + + 替换规则写在测试里而不是让 README 变得不可直接复制: README 里那份必须是 + 下游 `pip install` 后照抄就能用的,占位符因此都是合法 SQL 值。 + """ + start = month.strftime("%Y-%m-%d %H:%M:%S%z") + end = (month + timedelta(days=32)).replace(day=1).strftime("%Y-%m-%d %H:%M:%S%z") + for placeholder, actual in ( + # 长名在前: 三个角色名互不为前缀,但顺序稳定便于排查 + ("polygateway_owner", roles["owner"]), + ("polygateway_report", roles["report"]), + ("polygateway_app", roles["app"]), + ("CHANGE_ME_APP", _PROBE_PASSWORD), + ("CHANGE_ME_REPORT", _PROBE_PASSWORD), + ("SCHEMA public", f"SCHEMA {schema}"), + ("llm_calls_2026_01", f"llm_calls_{month:%Y_%m}"), + ("'2026-01-01 00:00:00+00'", f"'{start}'"), + ("'2026-02-01 00:00:00+00'", f"'{end}'"), + ): + sql = sql.replace(placeholder, actual) + return sql + + +def _role_dsn(dsn: str, role: str, schema: str) -> str: + low = re.sub(r"//[^@/]+@", f"//{role}:{_PROBE_PASSWORD}@", dsn, count=1) + return _search_path_dsn(low, schema) + + +async def _drop_template_objects(dsn: str, schema: str, roles: dict[str, str]) -> None: + """删净临时 schema 与三个角色(角色是**全局**对象,漏删会跨 run 残留)。""" + import asyncpg + + admin = await asyncpg.connect(dsn, timeout=10) + try: + await admin.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE") + for role in roles.values(): + await admin.execute(f"DROP OWNED BY {role}") + await admin.execute(f"DROP ROLE IF EXISTS {role}") + finally: + await admin.close() + + +@pytest.fixture +async def production_template(dsn): + """在临时 schema + 临时角色上跑完 README 的整套模板,产出可用的三条连接串。 + + 隔离纪律(M4 事故教训)同 `least_privilege_dsn`: 共享的 `public.llm_calls` + 一个字节都不碰,建的 schema / 角色 / 函数 / 分区在 teardown 里删净。 + """ + import asyncpg + + suffix = uuid4().hex[:8] + schema = f"pgwtpl_{suffix}" + roles = { + "owner": f"pgwtpl_owner_{suffix}", + "app": f"pgwtpl_app_{suffix}", + "report": f"pgwtpl_report_{suffix}", + } + month = datetime.now(UTC).replace(day=1, hour=0, minute=0, second=0, microsecond=0) + blocks = _template_blocks() + # 解析不到就地红: 空 dict 会让下面的 for 一句不执行,测试变成"只验证了能连上库" + assert list(blocks) == list(_EXPECTED_TEMPLATE_BLOCKS), ( + f"README 的模板锚点与预期不符: {list(blocks)}" + ) + + seeded = (_cid("tpl-a"), _cid("tpl-b")) + admin_dsn = _search_path_dsn(dsn, schema) + admin = await asyncpg.connect(dsn, timeout=10) + # 权限门放在建任何对象**之前**: `pytest.skip` 抛的是 BaseException, + # 若它在下面的清理块内触发,清理会去 DROP 从未建过的角色而把 skip 盖掉 + can_create = await admin.fetchval( + "SELECT rolcreaterole OR rolsuper FROM pg_roles WHERE rolname = current_user" + ) + if not can_create: + await admin.close() + pytest.skip("当前账号无权建临时角色,跳过生产模板用例") + try: + await admin.execute(f"CREATE SCHEMA {schema}") + await admin.execute(f"SET search_path = {schema}") + # README §2 写明的前置步骤: 先用库自带脚本建出普通表当模子 + await admin.execute(telemetry_schema_sql("postgres")) + for name in _EXPECTED_TEMPLATE_BLOCKS: + await admin.execute(_localize(blocks[name], schema, roles, month)) + # 种两个租户的行(超级用户绕过 RLS,属于布景不属于被测行为) + for call_id, tenant in zip(seeded, ("tenant-a", "tenant-b"), strict=True): + await admin.execute(_TEMPLATE_INSERT, call_id, tenant) + except BaseException: + # 模板 SQL 出错时也必须删净: 建到一半的 schema 会残留一张 llm_calls, + # 而 `TestSchema` 那条按 table_name 查 information_schema 的用例不带 + # schema 过滤,会被残留物在**下一次运行**里以列数不符的形态误伤 + await admin.close() + await _drop_template_objects(dsn, schema, roles) + raise + finally: + if not admin.is_closed(): + await admin.close() + + yield _TemplateEnv( + admin_dsn=admin_dsn, + app_dsn=_role_dsn(dsn, roles["app"], schema), + report_dsn=_role_dsn(dsn, roles["report"], schema), + schema=schema, + partition=f"llm_calls_{month:%Y_%m}", + seeded=seeded, + ) + + admin = await asyncpg.connect(dsn, timeout=10) + try: + await admin.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE") + for role in roles.values(): + await admin.execute(f"DROP OWNED BY {role}") + await admin.execute(f"DROP ROLE IF EXISTS {role}") + finally: + await admin.close() + + +class TestProductionTemplate: + """issue #12: README 的生产部署 DDL 模板必须逐条可执行,且行为与文中描述一致。 + + 模板出错的代价全部落在下游身上(照抄就中招),而人工核对不构成回归保护—— + 改一次 README 就会悄悄失去它。故这里从 README **直接解析** SQL 来执行。 + """ + + def test_readme_exposes_exactly_the_expected_template_blocks(self): + """先钉死解析本身: 锚点没了、改名了、块数变了,这条当场红。 + + 没有它,`production_template` 里解析出空 dict 时下面每条用例都会以 + "表不存在"之类的间接形态失败,真因(README 结构变了)要靠猜。 + """ + blocks = _template_blocks() + assert list(blocks) == list(_EXPECTED_TEMPLATE_BLOCKS) + assert all(sql.strip() for sql in blocks.values()) + joined = "\n".join(blocks.values()) + for placeholder in _TEMPLATE_PLACEHOLDERS: + # 占位符没了 = 受控替换静默失效,测试会去打真实的 polygateway_* 角色 + assert placeholder in joined, f"README 模板缺占位符 {placeholder!r}" + + async def test_app_can_insert_but_cannot_mutate(self, production_template): + """应用角色: INSERT 通过,UPDATE / DELETE 被权限层拒绝(不是被触发器拒)。 + + 权限检查早于行级触发器,故这里拿到的必须是 InsufficientPrivilegeError—— + 若换成触发器的 RaiseError,说明 REVOKE 那一块没生效,而"不可变"就只剩 + 一层属主随手可关的兜底。 + """ + import asyncpg + + env = production_template + conn = await asyncpg.connect(env.app_dsn, timeout=10) + try: + await conn.execute(_TEMPLATE_INSERT, _cid("tpl-app"), "tenant-a") + with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError): + await conn.execute("DELETE FROM llm_calls WHERE call_id = $1", _cid("tpl-app")) + with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError): + await conn.execute("UPDATE llm_calls SET response = 'x'") + finally: + await conn.close() + rows = await _fetch( + env.admin_dsn, "SELECT call_id FROM llm_calls WHERE call_id = $1", _cid("tpl-app") + ) + assert [r["call_id"] for r in rows] == [_cid("tpl-app")] # 写入真落库了 + + async def test_report_can_read_but_cannot_write(self, production_template): + """报表角色: 带租户上下文读得到自己的行,任何写入都被拒。""" + import asyncpg + + env = production_template + conn = await asyncpg.connect(env.report_dsn, timeout=10) + try: + with pytest.raises(asyncpg.exceptions.InsufficientPrivilegeError): + await conn.execute(_TEMPLATE_INSERT, _cid("tpl-rpt"), "tenant-a") + async with conn.transaction(): + await conn.execute("SELECT set_config('app.tenant_id', 'tenant-a', true)") + rows = await conn.fetch("SELECT call_id, tenant_id FROM llm_calls") + assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(env.seeded[0], "tenant-a")] + finally: + await conn.close() + + async def test_reads_are_fail_closed_until_the_tenant_guc_is_set(self, production_template): + """未设 `app.tenant_id` → 零行(fail-closed);设了 → 只看得到本租户。 + + 两个断言缺一不可: 只验"设了能看到自己的"漏掉了 GUC 未设时全表泄露, + 只验"未设是零行"则一条永远返回 false 的 policy 也能通过。 + """ + import asyncpg + + env = production_template + conn = await asyncpg.connect(env.app_dsn, timeout=10) + try: + async with conn.transaction(): + assert await conn.fetch("SELECT call_id FROM llm_calls") == [] + async with conn.transaction(): + await conn.execute("SELECT set_config('app.tenant_id', 'tenant-b', true)") + rows = await conn.fetch("SELECT call_id, tenant_id FROM llm_calls") + assert [(r["call_id"], r["tenant_id"]) for r in rows] == [(env.seeded[1], "tenant-b")] + finally: + await conn.close() + + async def test_rows_land_in_the_current_month_partition(self, production_template): + """分区表写入成功,且行确实落进当月分区(不是落进某个兜底分区)。""" + env = production_template + rows = await _fetch( + env.admin_dsn, + "SELECT tableoid::regclass::text AS part FROM llm_calls WHERE call_id = $1", + env.seeded[0], + ) + assert [r["part"].split(".")[-1] for r in rows] == [env.partition] + + async def test_trigger_blocks_delete_while_drop_partition_still_works( + self, production_template + ): + """兜底触发器拦得住 DELETE(连超级用户也拦),却拦不住 DROP PARTITION。 + + 这正是 README 说"清理只能走 DROP PARTITION 而不是 DELETE"的机械化依据: + 既要对应用角色 REVOKE DELETE、又要能清理过期数据,分区是唯一不冲突的解。 + """ + import asyncpg + + env = production_template + conn = await asyncpg.connect(env.admin_dsn, timeout=10) + try: + with pytest.raises(asyncpg.exceptions.RaiseError) as exc: + await conn.execute("DELETE FROM llm_calls WHERE call_id = $1", env.seeded[0]) + assert "不可变审计表" in str(exc.value) + await conn.execute(f"ALTER TABLE llm_calls DETACH PARTITION {env.partition}") + await conn.execute(f"DROP TABLE {env.partition}") + assert await conn.fetchval("SELECT count(*) FROM llm_calls") == 0 + finally: + await conn.close()