关于 audit-db-schema
audit-db-schema 是供数据库及迁移项目团队使用的架构审计技能。它收集架构证据,分类结构与安全风险,并只报告问题,不执行迁移。
主要功能
- 01识别数据库、ORM 和迁移来源
- 02检查列、约束、索引及 RLS
- 03按类别和严重程度分类问题
- 04要求提供证据并复现查询结果
- 05报告问题但不执行迁移
适用场景
- 01检查 Supabase 架构是否缺少 RLS 策略
- 02查找缺失的索引或外键约束
- 03迁移前确定架构问题的优先级
MCP 相关
此技能会在可用时调用 Supabase MCP 工具检查表、执行 SQL 探测并获取安全和性能顾问信息;它不会应用迁移。
安装
复制下面的命令,在终端中运行。
npx skills add https:/ / github.com/ kensaurus/ skills --skill audit-db-schema安装后在 Claude Code 中输入/audit-db-schema确认已加载。
附带文件references/details.md
详细介绍
数据库架构审计技能
自由度:混合 — 步骤 0、1、3 为 [HIGH freedom];步骤 2 和 4 的
MCP/SQL 探测为 [LOW freedom — run exactly](运行查询;不要臆造架构)。
推理方式
- 观察 — 引述列、约束、顾问结果行或查询结果
- 解读 — 写入时、读取时或迁移时会出现什么问题?
- 分类 — 命名 / 类型 / 约束 / 索引 / RLS / 迁移 / 正确
- 严重性 — 公开数据缺少 FK/RLS = P0;类型/索引偏差 = P1;命名 = P2
示例
观察:
orders.user_id是可空的text,没有 FK,也没有索引;rowsecurity = false。 解读: 可以插入孤立行;客户端可以 SELECT 每一笔订单;查询会进行顺序扫描。 分类: 约束 + 索引 + RLS(不是命名小问题)。 严重性: P0 — 公开表没有 RLS,也没有 FK。 发现:orders| RLS+FK | P0 | 启用 RLS +user_id uuid references users(id)+ 索引。
报告前自我检查 [LOW freedom — do not skip]
- 有证据支持 — 查询结果或顾问 URL,而不是“Postgres 通常……”
- 可复现 — 相同 SQL 执行两次;不要引用过时的
list_tables - 严重性有依据 — P0 = 数据丢失、泄露或不受约束的金额类型
- 负责人正确 — 谁能读取哪些内容 →
plan-rls-audit;DELETE/TRUNCATE →plan-data-integrity;RPO →plan-backup-dr - 未应用迁移 — 仅报告发现
步骤 0:自动检测数据库环境
0a. 检测数据库和 ORM
| 信号 | 技术 |
|---|---|
package.json 中有 @supabase/supabase-js | Supabase(Postgres) |
devDependencies 中有 prisma,并且有 prisma/schema.prisma | Prisma ORM |
dependencies 中有 drizzle-orm,并且有 drizzle/ 目录 | Drizzle ORM |
dependencies 中有 sequelize | Sequelize ORM |
requirements 中有 sqlalchemy | SQLAlchemy(Python) |
有 supabase/migrations/*.sql 目录 | Supabase 迁移 |
有 prisma/migrations/ 目录 | Prisma 迁移 |
有 drizzle/migrations/ 或 drizzle/*.sql | Drizzle 迁移 |
0b. 查找 Supabase 项目 ID
从 .env / .env.local 中 SUPABASE_URL 的主机名获取 PROJECT_ID
(https://<ref>.supabase.co;supabase/config.toml 中的 project_id 是
本地名称,不是 ref),并通过 supabase:get_project_url 确认。仅当 MCP
服务公开提供 supabase:list_projects 时才调用它(启动时未指定项目 ref)。
项目范围的服务(此工具包的配置方式)不提供该工具,其工具不接受
project_id;对于未限定范围的服务,请在下面的每次调用中添加
"project_id": "<PROJECT_ID>"。
0c. 检测架构源文件
Glob: **/supabase/migrations/*.sql → Supabase SQL migrations
Glob: **/prisma/schema.prisma → Prisma schema
Glob: **/drizzle/schema.ts → Drizzle schema
Glob: **/src/db/schema.ts → Drizzle alt location
Glob: **/knexfile.* → Knex migrations
Glob: **/alembic/versions/*.py → SQLAlchemy migrations
0d. 记录发现结果
DATABASE ENVIRONMENT:
- Database: [Supabase Postgres / raw Postgres / MySQL / SQLite]
- ORM: [Prisma / Drizzle / Sequelize / none]
- Project ID: [Supabase project ID or N/A]
- Migration tool: [Supabase CLI / Prisma Migrate / Drizzle Kit / Knex]
- Schema files: [list paths]
- Migration count: [N]
步骤 1:研究架构最佳实践
1a. Context7 — ORM 文档
如果使用 Prisma:
context7:resolve-library-id
{
"libraryName": "prisma",
"query": "schema best practices indexes relations"
}
context7:query-docs
{
"libraryId": "<RESOLVED_ID>",
"query": "schema best practices naming conventions indexes onDelete"
}
如果使用 Drizzle,则改为解析 drizzle-orm。
1b. Firecrawl — 当前数据库模式
firecrawl:firecrawl_search
{
"query": "PostgreSQL schema design best practices [current year]",
"limit": 5,
"sources": [{ "type": "web" }]
}
根据检测到的技术栈进行其他搜索:
| 技术栈 | 搜索查询 |
|---|---|
| Supabase | Supabase RLS policies best practices performance [current year] |
| Prisma | Prisma schema design relations indexes best practices [current year] |
| Drizzle | Drizzle ORM schema patterns migrations [current year] |
| 通用 | PostgreSQL indexing strategy production optimization |
抓取最具权威性的结果:
firecrawl:firecrawl_scrape
{
"url": "<BEST_RESULT_URL>",
"formats": ["markdown"],
"onlyMainContent": true
}
1c. 搜索 Supabase 文档
如果使用 Supabase:
supabase:search_docs
{
"graphql_query": "{ searchDocs(query: \"RLS policy performance best practices\", limit: 3) { nodes { title href content } } }"
}
步骤 2:收集完整架构
2a. 列出所有表(Supabase MCP)
supabase:list_tables
{
"schemas": ["public"],
"verbose": true
}
2b. 运行详细审计查询
supabase:execute_sql
{
"query": "SELECT table_name, column_name, data_type, is_nullable, column_default FROM information_schema.columns WHERE table_schema = 'public' ORDER BY table_name, ordinal_position"
}
supabase:execute_sql
{
"query": "SELECT tc.table_name, tc.constraint_name, tc.constraint_type, kcu.column_name, ccu.table_name AS foreign_table FROM information_schema.table_constraints tc JOIN information_schema.key_column_usage kcu ON tc.constraint_name = kcu.constraint_name LEFT JOIN information_schema.constraint_column_usage ccu ON tc.constraint_name = ccu.constraint_name WHERE tc.table_schema = 'public'"
}
2c. 收集索引
supabase:execute_sql
{
"query": "SELECT tablename, indexname, indexdef FROM pg_indexes WHERE schemaname = 'public' ORDER BY tablename"
}
2d. 收集 RLS 状态和策略
supabase:execute_sql
{
"query": "SELECT tablename, rowsecurity FROM pg_tables WHERE schemaname = 'public' ORDER BY tablename"
}
supabase:execute_sql
{
"query": "SELECT schemaname, tablename, policyname, permissive, roles, cmd, qual, with_check FROM pg_policies WHERE schemaname = 'public' ORDER BY tablename"
}
2e. 运行 Supabase 顾问工具
supabase:get_advisors
{
"type": "security"
}
supabase:get_advisors
{
"type": "performance"
}
在最终报告中,将顾问结果中的修复 URL 作为可点击链接列出。
步骤 3:审计类别
3.1 命名约定
| 规则 | 标准 | 检查 |
|---|---|---|
| 表 | snake_case,使用复数形式(users、posts) | 不使用 camelCase,不使用单数形式 |
| 列 | snake_case(created_at、user_id) | 不使用 camelCase |
| 主键 | id | 自身表中不使用 user_id |
| 外键 | {referenced_table_singular}_id(user_id) | 模式一致 |
| 索引 | idx_{table}_{column(s)} | 名称具有描述性 |
| 约束 | {table}_{column}_{type}(users_email_unique) | 名称具有描述性 |
| 枚举 | 类型使用 snake_case,值使用 UPPER_CASE | 大小写格式一致 |
| 布尔列 | 使用 is_ 或 has_ 前缀(is_active、has_access) | 意图明确 |
审计查询:
SELECT table_name FROM information_schema.tables
WHERE table_schema = 'public'
AND (table_name ~ '[A-Z]' OR table_name !~ 's$');
SELECT table_name, column_name FROM information_schema.columns
WHERE table_schema = 'public' AND column_name ~ '[A-Z]';
3.2 数据类型
| 规则 | 标准 |
|---|---|
| 主键 | 使用 uuid 和 gen_random_uuid(),或使用 cuid |
| 时间戳 | timestamptz(不是 timestamp) |
| 金额 | numeric(12,2) 或 bigint(以分为单位)——不要使用 float/real(二进制舍入会导致分值丢失) |
| 电子邮件 | 使用带 CHECK 约束的 text,或使用 citext |
| 状态/枚举 | Postgres enum 类型,或带 CHECK 约束的 text |
| JSON | jsonb(不是 json) |
| 短字符串 | 在 Postgres 中优先使用 text,而非 varchar(n) |
| 布尔值 | 使用带 NOT NULL DEFAULT 的 boolean |
| IP 地址 | inet 类型 |
| 数组 | 适当情况下使用原生 text[]、integer[] |
审计查询:
SELECT table_name, column_name, data_type FROM information_schema.columns
WHERE table_schema = 'public' AND data_type = 'timestamp without time zone';
SELECT table_name, column_name, data_type FROM information_schema.columns
WHERE table_schema = 'public'
AND data_type IN ('real', 'double precision')
AND (column_name LIKE '%price%' OR column_name LIKE '%amount%'
OR column_name LIKE '%cost%' OR column_name LIKE '%balance%');
SELECT table_name, column_name FROM information_schema.columns
WHERE table_schema = 'public' AND data_type = 'json';
3.3 必需列和时间戳
每个表都必须包含:
| 列 | 类型 | 默认值 | 备注 |
|---|---|---|---|
id | uuid | gen_random_uuid() | 主键 |
created_at | timestamptz | now() | NOT NULL |
updated_at | timestamptz | now() | NOT NULL,自动触发器 |
审计查询:
SELECT t.table_name,
EXISTS(SELECT 1 FROM information_schema.columns c WHERE c.table_name = t.table_name AND c.column_name = 'created_at') AS has_created_at,
EXISTS(SELECT 1 FROM information_schema.columns c WHERE c.table_name = t.table_name AND c.column_name = 'updated_at') AS has_updated_at
FROM information_schema.tables t
WHERE t.table_schema = 'public' AND t.table_type = 'BASE TABLE';
SELECT event_object_table, trigger_name FROM information_schema.triggers
WHERE trigger_schema = 'public' AND action_statement LIKE '%updated_at%';
3.4 约束和验证
| 约束 | 必需条件 |
|---|---|
NOT NULL | 除非明确设为可选,否则每列都必须设置 |
UNIQUE | 电子邮件、slug、外部 ID、用户名 |
CHECK | 枚举、范围、格式、正数 |
DEFAULT | 布尔值、时间戳、状态字段 |
FOREIGN KEY | 每个关系列 |
ON DELETE | 自有数据使用 CASCADE,可选引用使用 SET NULL,关键数据使用 RESTRICT |
审计查询:
SELECT table_name, column_name FROM information_schema.columns
WHERE table_schema = 'public' AND column_name LIKE '%_id'
AND is_nullable = 'YES' AND column_name != 'id';
SELECT c.table_name, c.column_name FROM information_schema.columns c
WHERE c.table_schema = 'public' AND c.column_name LIKE '%_id' AND c.column_name != 'id'
AND NOT EXISTS (
SELECT 1 FROM information_schema.key_column_usage kcu
JOIN information_schema.table_constraints tc ON kcu.constraint_name = tc.constraint_name
WHERE tc.constraint_type = 'FOREIGN KEY'
AND kcu.table_name = c.table_name AND kcu.column_name = c.column_name
);
SELECT table_name, column_name FROM information_schema.columns
WHERE table_schema = 'public' AND data_type = 'boolean' AND column_default IS NULL;
3.5 索引
| 规则 | 标准 |
|---|---|
| 外键 | 为每个 FK 列建立索引 |
| 高频查询 | 为 WHERE/ORDER BY 列建立索引 |
| 唯一查找 | 为 email、slug、external_id 建立唯一索引 |
| 复合索引 | 顺序:先等值,再范围,最后排序 |
| RLS 列 | 为 RLS 策略中使用的列建立索引 |
created_at | 为按时间顺序查询建立 DESC 索引 |
| 部分索引 | 使用 WHERE 子句支持子集查询 |
审计查询:
SELECT c.table_name, c.column_name FROM information_schema.columns c
WHERE c.table_schema = 'public' AND c.column_name LIKE '%_id' AND c.column_name != 'id'
AND NOT EXISTS (
SELECT 1 FROM pg_indexes i
WHERE i.schemaname = 'public' AND i.tablename = c.table_name
AND i.indexdef LIKE '%' || c.column_name || '%'
);
SELECT t.table_name, COUNT(i.indexname) as idx_count
FROM information_schema.tables t
LEFT JOIN pg_indexes i ON i.tablename = t.table_name AND i.schemaname = 'public'
WHERE t.table_schema = 'public' AND t.table_type = 'BASE TABLE'
GROUP BY t.table_name HAVING COUNT(i.indexname) <= 1;
3.6 行级安全(Supabase)
| 规则 | 标准 |
|---|---|
| 启用 RLS | 每个 public 表都必须启用 RLS——否则 anon key 可以读取每一行 |
| SELECT 策略 | 每个表都必须有 |
| INSERT 策略 | 使用 WITH CHECK 检查用户所有权 |
| UPDATE 策略 | 使用 USING + WITH CHECK 检查所有权 |
| DELETE 策略 | 使用 USING 检查所有权 |
| Service role | 绕过 RLS(绝不向客户端公开) |
| 性能 | 使用 (select auth.uid()) 子查询模式 |
| 索引 | 为策略中使用的列建立索引 |
审计查询:
SELECT tablename FROM pg_tables WHERE schemaname = 'public' AND rowsecurity = false;
SELECT t.tablename FROM pg_tables t
WHERE t.schemaname = 'public' AND t.rowsecurity = true
AND NOT EXISTS (
SELECT 1 FROM pg_policies p WHERE p.tablename = t.tablename AND p.schemaname = 'public'
);
SELECT tablename, policyname, qual FROM pg_policies
WHERE schemaname = 'public'
AND qual::text LIKE '%auth.uid()%'
AND qual::text NOT LIKE '%(select auth.uid())%';
3.7 关系和规范化
| 规则 | 标准 |
|---|---|
| 最低达到 3NF | 不存在传递依赖 |
| 联结表 | 用于多对多关系(使用 user_roles,而非 JSON 数组) |
| 不重复数据 | 将重复数据规范化到查找表中 |
| 级联规则 | 为每个 FK 关系定义级联规则 |
| 自引用 | 需要时使用 parent_id 模式 |
| 多态 | 避免使用——改用联结表或 STI |
3.8 迁移
| 规则 | 标准 |
|---|---|
| 顺序编号 | 使用时间戳或 0001_、0002_ 前缀 |
| 名称具有描述性 | 使用 0003_add_user_roles.sql,而不是 0003_update.sql |
| 幂等 | 使用 IF NOT EXISTS、IF EXISTS 保护 |
| 不丢失数据 | 提供回滚迁移或回滚计划 |
| 原子性 | 每个迁移只包含一项逻辑变更 |
| 不引入破坏性变更 | 先添加,再回填,最后清理 |
3.9 安全
| 规则 | 标准 |
|---|---|
| 不存储明文机密 | 对密码进行哈希处理,对令牌进行加密 |
| 保护 PII | 识别并保护敏感列 |
| 审计跟踪 | 在敏感表中添加 created_by、updated_by |
| 授权 | 为每个角色分配最小权限 |
| 扩展 | 仅启用必要的扩展 |
| 搜索路径 | 显式指定架构引用 |
审计查询:
SELECT table_name, column_name FROM information_schema.columns
WHERE table_schema = 'public'
AND (column_name LIKE '%password%' OR column_name LIKE '%secret%'
OR column_name LIKE '%token%' OR column_name LIKE '%ssn%'
OR column_name LIKE '%credit_card%');
SELECT grantee, table_name, privilege_type FROM information_schema.table_privileges
WHERE table_schema = 'public' ORDER BY grantee, table_name;
第 4 步:完整架构健康检查(单条查询)
supabase:execute_sql
{
"query": "WITH table_info AS (SELECT t.table_name, EXISTS(SELECT 1 FROM information_schema.columns c WHERE c.table_name = t.table_name AND c.column_name = 'id') AS has_id, EXISTS(SELECT 1 FROM information_schema.columns c WHERE c.table_name = t.table_name AND c.column_name = 'created_at') AS has_created_at, EXISTS(SELECT 1 FROM information_schema.columns c WHERE c.table_name = t.table_name AND c.column_name = 'updated_at') AS has_updated_at, (SELECT rowsecurity FROM pg_tables pt WHERE pt.tablename = t.table_name AND pt.schemaname = 'public') AS rls_enabled, (SELECT COUNT(*) FROM pg_policies p WHERE p.tablename = t.table_name AND p.schemaname = 'public') AS policy_count, (SELECT COUNT(*) FROM pg_indexes i WHERE i.tablename = t.table_name AND i.schemaname = 'public') AS index_count FROM information_schema.tables t WHERE t.table_schema = 'public' AND t.table_type = 'BASE TABLE') SELECT table_name, CASE WHEN has_id THEN 'Y' ELSE 'N' END AS id, CASE WHEN has_created_at THEN 'Y' ELSE 'N' END AS created_at, CASE WHEN has_updated_at THEN 'Y' ELSE 'N' END AS updated_at, CASE WHEN rls_enabled THEN 'Y' ELSE 'N' END AS rls, policy_count AS policies, index_count AS indexes FROM table_info ORDER BY table_name"
}