fix: refuse the sandbox rather than quietly running it as the superuser

Both reviews landed on the same line independently. _as_role swaps the
credentials in the DSN with a regex, and when the pattern does not match
it returned the string unchanged. Two shapes miss it: no inline
credentials, and a unix socket URL. Either one is a legal DSN.

What that costs is not a broken test. The sandbox builds, every
assertion still passes, and bare_dsn is now the admin connection, so the
worst-case case runs the real script with --apply as a superuser against
the shared table. The verifier ran that command as a dry run to see what
it would have done: target public.llm_calls, 11 rows to delete. The case
would still have gone red on the exit code, after the rows were gone.

It raises now. There is also a second check that connects and compares
current_user, because a successful string substitution is not the same
as connecting as that role -- PGUSER and friends still override. The
whole design rests on that connection having no grant on the shared
table; a string comparison is too thin a thing to rest it on.

That check has to stay inside the try. Past it the cleanup statements
have already been merged into the fixture-level stack, and unwinding
again runs DROP OWNED BY twice, which has no IF EXISTS.

The catalog probe took any SQL and ran it on the admin connection. The
design claims withholding the DSN makes the boundary structural; that
was only true of the connection string, not of the capability. It takes
SELECT now.

--table's schema half is restricted to plain identifiers. Not a
security fix, since the name goes through a parameter and _quote: the
help text says complex identifiers are unsupported and the code was
accepting them anyway.

tests/integration/conftest.py

@@ -73,8 +73,22 @@ def _with_search_path(dsn: str, schema: str) -> str:
 
 
 def _as_role(dsn: str, role: str) -> str:
-    """把 DSN 的用户名口令段换成沙箱角色的,其余(主机/库/参数)原样保留。"""
-    return re.sub(r"//[^@/]+@", f"//{role}:{_SANDBOX_PASSWORD}@", dsn, count=1)
+    """把 DSN 的用户名口令段换成沙箱角色的,其余(主机/库/参数)原样保留。
+
+    **换不掉就报错,绝不原样返回**: `postgresql://h:5432/db`(口令走 PGPASSWORD /
+    .pgpass / trust)与 `postgresql:///db?host=/var/run/postgresql`(unix socket)
+    都是合法 DSN,却没有可替换的内联凭据段。静默返回原串的后果不是测试报错,而是
+    沙箱以**管理身份**建成、用例照常绿,同时 `bare_dsn` 变成超级用户连接——最坏
+    情况用例会拿它跑真实 `--apply`,删空共享表之后才在退出码断言上红。
+    这正是 P5"严禁默认值掩盖错误"要挡的形态。
+    """
+    swapped, count = re.subn(r"//[^@/]+@", f"//{role}:{_SANDBOX_PASSWORD}@", dsn, count=1)
+    if count != 1:
+        raise RuntimeError(
+            f"DSN 里没有可替换的内联凭据段,沙箱角色 {role} 无法生效,拒绝以管理身份继续。"
+            "请把 PGW_TELEMETRY_PG_DSN 写成 postgresql://<用户>:<口令>@<主机>/<库> 的形态。"
+        )
+    return swapped
 
 
 @pytest.fixture
@@ -89,6 +103,10 @@ async def pg_catalog_probe():
     dsn = _require_admin_dsn()
 
     async def probe(sql: str, *args: object) -> list[tuple]:
+        # 只读校验不是形式主义: 这个闭包持的是管理连接,不设限就等于把"用例够不到
+        # 管理能力"这句话降格成一句 docstring 里的请求。
+        if not sql.lstrip().upper().startswith("SELECT"):
+            raise RuntimeError(f"pg_catalog_probe 只接受 SELECT 语句,收到: {sql[:60]!r}")
         conn = await asyncpg.connect(dsn, timeout=10)
         try:
             return [tuple(r) for r in await conn.fetch(sql, *args)]
@@ -194,18 +212,37 @@ async def pg_sandbox():
                         f"GRANT {', '.join(grants)} ON ALL TABLES IN SCHEMA {schema} TO {role_name}"
                     )
                 # 关键: 绝不 GRANT CREATE ON SCHEMA —— 缺的正是这一项
+
+            used = bare if role_name is not None else admin_dsn
+            sandbox = PgSandbox(
+                schema=schema,
+                role=role_name,
+                dsn=_with_search_path(used, schema),
+                bare_dsn=bare,
+            )
+            if role_name is not None:
+                # 字符串替换成功不等于连上去就是那个角色(PGUSER 等环境变量仍可能
+                # 盖掉 DSN 里的用户名)。这道校验按**实际身份**兜底: 整个设计的价值
+                # 都压在"跑脚本的那个连接对共享表无权"上,不值得只用一次字符串比较
+                # 来担保。它必须留在 try 之内——出了这个块,清理动作已经并进 fixture
+                # 级的栈,再回滚一次就会对同一个角色跑两遍 DROP OWNED BY(它没有
+                # IF EXISTS,第二遍必报错)。
+                conn = await asyncpg.connect(sandbox.dsn, timeout=10)
+                try:
+                    actual = await conn.fetchval("SELECT current_user")
+                finally:
+                    await conn.close()
+                if actual != role_name:
+                    raise RuntimeError(
+                        f"沙箱 DSN 连上去的身份是 {actual!r},不是预期的 {role_name!r};"
+                        "权限边界不成立,拒绝把这个沙箱交出去。"
+                    )
         except BaseException:
             await _unwind(local)
             raise
         cleanups.extend(local)
 
-        used = bare if role_name is not None else admin_dsn
-        return PgSandbox(
-            schema=schema,
-            role=role_name,
-            dsn=_with_search_path(used, schema),
-            bare_dsn=bare,
-        )
+        return sandbox
 
     yield make
     await _unwind(cleanups)

tests/integration/test_pg_sandbox.py

@@ -10,6 +10,8 @@ from __future__ import annotations
 
 import pytest
 
+from tests.integration.conftest import _as_role
+
 _DDL = "CREATE TABLE llm_calls (call_id TEXT PRIMARY KEY, created_at TIMESTAMPTZ DEFAULT now())"
 
 
@@ -40,6 +42,42 @@ async def _oid_of_llm_calls(dsn: str) -> int | None:
         await conn.close()
 
 
+class TestRoleDsnConstruction:
+    """凭据替换失败必须**当场报错**,不许退回管理身份(合并前审查的 P1)。
+
+    这条防线的失效形态特别隐蔽: 替换不上时 `re.sub` 原样返回管理连接串,沙箱
+    "看起来"建好了、用例照常绿,而 `bare_dsn` 其实是超级用户——最坏情况用例
+    会拿它跑真实 `--apply`,把共享表删空之后才在 `assert returncode == 2` 上红。
+    行已经没了。设计 §5.1 要的是"越界做不到",不是"越界会被发现"。
+    """
+
+    def test_inline_credentials_are_replaced(self):
+        swapped = _as_role("postgresql://app:secret@h:5432/polygateway", "pgw_r_x")
+
+        assert swapped.startswith("postgresql://pgw_r_x:")
+        assert "app:secret" not in swapped
+
+    @pytest.mark.parametrize(
+        "dsn",
+        [
+            "postgresql://h:5432/polygateway",  # 口令走 PGPASSWORD / .pgpass / trust
+            "postgresql:///polygateway?host=/var/run/postgresql",  # unix socket
+        ],
+    )
+    def test_a_dsn_without_inline_credentials_is_refused(self, dsn):
+        """这两种都是合法 DSN,今天的 .env 恰好不是它们——恰好而已。"""
+        with pytest.raises(RuntimeError, match="沙箱角色"):
+            _as_role(dsn, "pgw_r_x")
+
+
+class TestCatalogProbeIsReadOnly:
+    """探针拿的是管理连接,故它只许查——否则"用例够不到管理能力"就是句空话。"""
+
+    async def test_non_select_statements_are_refused(self, pg_catalog_probe):
+        with pytest.raises(RuntimeError, match="只接受 SELECT"):
+            await pg_catalog_probe("DELETE FROM llm_calls WHERE call_id = 'nope'")
+
+
 class TestSchemaOnlySandbox:
     async def test_table_lands_in_the_sandbox_schema_and_bare_dsn_is_absent(self, pg_sandbox):
         """`role="none"`: 表落在自建 schema 下;不发角色,故没有裸 DSN 可给。"""

tests/unit/test_retention_tool.py

@@ -360,6 +360,59 @@ class TestUsageErrors:
         assert "llm_calls" in result.stderr
 
 
+class TestTableIdentifierWhitelist:
+    """schema 段只收普通标识符: 让 `--help` 说的"不支持复杂标识符"成为事实。
+
+    这不是安全边界(`to_regclass($1)` 参数化 + `_quote` 转义,注入面本就不存在),
+    是**契约边界**: 帮助文本写着不支持,实现却照单全收,受害的是照文档做判断的人。
+    """
+
+    def test_schema_with_a_space_exits_one(self):
+        result = _run(
+            "--backend",
+            "postgres",
+            "--dsn",
+            "postgresql://x/y",
+            "--older-than-days",
+            "7",
+            "--table",
+            "bad schema.llm_calls",
+        )
+
+        assert result.returncode == 1
+        assert "--table" in result.stderr
+
+    def test_schema_with_a_semicolon_exits_one(self):
+        result = _run(
+            "--backend",
+            "postgres",
+            "--dsn",
+            "postgresql://x/y",
+            "--older-than-days",
+            "7",
+            "--table",
+            "a;b.llm_calls",
+        )
+
+        assert result.returncode == 1
+        assert "--table" in result.stderr
+
+    def test_a_plain_identifier_with_underscores_and_digits_is_accepted(self):
+        """收紧不得误伤正常名字: 这条走到连接阶段才失败(退出 2),说明校验放行了。"""
+        result = _run(
+            "--backend",
+            "postgres",
+            "--dsn",
+            "postgresql://127.0.0.1:1/nope",
+            "--older-than-days",
+            "7",
+            "--table",
+            "pgw_s_a1b2c3.llm_calls",
+        )
+
+        assert result.returncode == 2
+
+
 class TestHelp:
     def test_help_names_the_maintenance_role_and_the_recommended_path(self):
         """帮助文本是运维唯一会读的文档,权限口径与"推荐不是 DELETE"必须在里面。"""