sql-linker

Use this skill whenever you need to query, insert, update, or delete database records. Triggers include: (1) querying database data with SELECT statements and parameter binding; (2) inserting or batch-inserting records; (3) updating or deleting records with WHERE conditions; (4) transaction control with begin/commit/rollback; (5) read-only mode enforcement for safe data access; (6) audit trail for identity tracking, SQL logging, and compliance. Also use when the user mentions database operations, SQL queries, CRUD tasks, or any data manipulation involving MySQL, PostgreSQL, or SQLite databases.

Important: All scripts/ paths are relative to this skill directory. Run with: cd {skill_dir} && python scripts/... or use the cwd parameter.

---

SQL-Linker — 双层架构：数据操作层 + 业务层 / Data Ops + Business Layer

---

概述 / Overview

（中文）SQL-Linker 提供跨数据库的 CRUD 操作能力，支持 MySQL、PostgreSQL、SQLite 三种主流数据库。内置审计日志模块，每次操作自动记录操作人身份、IP、SQL 语句、操作时间，确保数据可溯源、安全可控。业务层（db_bridge）负责字段白名单过滤和时间戳自动注入，数据操作层（sql_linker）负责连接管理、CRUD 执行和审计记录，两层严格分离，互不干扰。

（English）SQL-Linker provides cross-database CRUD operations, supporting MySQL, PostgreSQL, and SQLite, with a built-in audit trail module that automatically records operator identity, IP, SQL statements, and timestamps for full traceability and compliance. The business layer (db_bridge) handles field whitelist filtering and automatic timestamp injection, while the data operation layer (sql_linker) manages database connections, CRUD execution, and audit logging. The two layers are strictly separated and independent.

---

核心架构 / Core Architecture

（中文）系统由两层组成，业务层和数据操作层职责分明：

（English）The system consists of two layers with clearly defined responsibilities:

workspace/
└── .sql_linker/                          ← 配置文件根目录
    ├── config_home/
    │   ├── config.yaml                   ← 数据库连接配置
    │   ├── audit_config.json             ← 审计配置
    │   └── extra_tables.json             ← 特权表配置（JSON）
    └── table_home/
        └── table_dictionary.json         ← 主词典（JSON，所有受控表）

skills/sql-linker/scripts/
├── controller_layer/                      ← 数据操作层
│   ├── sql_linker.py                     ← 连接管理 + CRUD 执行 + 审计上下文注入
│   └── sql_audit.py                       ← 审计模块（sql_linker.py 内部使用）
└── service_layer/                        ← 业务层
    └── db_bridge.py                      ← 四层访问控制 + 时间戳注入 + 字段白名单

（中文）业务层（service_layer）：读取 table_dictionary.json，过滤字段，注入时间戳，校验访问权限，调用数据操作层，不直接操作数据库。

（English）Business Layer (service_layer): Reads table_dictionary.json, filters fields, injects timestamps, verifies access rights, and calls the data operation layer. Does not directly access the database.

（中文）数据操作层（controller_layer）：管理数据库连接，执行 CRUD 操作，写入审计日志，处理参数化查询，不处理业务逻辑。

（English）Data Operation Layer (controller_layer): Manages database connections, executes CRUD operations, writes audit logs, handles parameterized queries. Does not process business logic.

---

四层访问模型 / Four-Layer Access Model

（中文）系统通过四层访问模型实现精确的表访问控制：

（English）The system implements precise table access control through a four-layer access model:

SYSTEM (系统表 sql_audit_log)

• Hard-coded: db_bridge.py SYSTEM_TABLES
• SELECT/INSERT: ALLOW
• UPDATE/DELETE: DENY (SystemTableWriteDenied)
• Field whitelist: N/A
• Timestamp injection: N/A
• Audit: Native cursor bypasses db_bridge
• Audit log cannot be tampered by business layer

NORMAL (主词典表格)

• File: table_dictionary.json
• Field whitelist: YES (only fields in table.json)
• Timestamp injection: YES (created_at/updated_at auto-generated)
• Audit: Full
• Ready to use without extra config

PRIVILEGED (特权表格)

• File: extra_tables.json + config.yaml extra_tables_enabled
• Field whitelist: NO (unknown schema, direct DB exposure)
• Timestamp injection: NO
• Audit: Full
• Requires extra_tables.json declaration + config enable

BLOCKED (禁用)

• Not in dictionary nor extra_tables
• All operations denied, denial logged
• Cannot access unless added to extra_tables.json

（中文）访问判定流程：提取 SQL 中的表名 → 检查 SYSTEM → 检查主词典（NORMAL）→ 检查 extra_tables（PRIVILEGED）→ 其余 BLOCKED。

（English）Access Decision Flow: Extract table name from SQL → Check SYSTEM → Check main dictionary (NORMAL) → Check extra_tables (PRIVILEGED) → Rest BLOCKED.

---

引导初始化 / Bootstrap

（中文）首次使用或缺少配置文件时，系统自动生成默认模板（幂等操作，不会覆盖已有文件）：

（English）On first use or when config files are missing, the system automatically generates default templates (idempotent, will not overwrite existing files):

from db_bridge import DBBridge

db = DBBridge(user_label="openclaw-control-ui", session_id="agent:hr:main")
created = db.bootstrap()
print(f'Created: {created}')
# ['...\\config.yaml', '...\\audit_config.json', ...]

（中文）自动生成的文件列表：

（English）Auto-generated files:

文件路径	默认内容
.sql_linker/config_home/config.yaml	连接模板（host/port/user 占位，password_env 引用 .env）
.sql_linker/config_home/audit_config.json	审计默认开启，log_select=false
.sql_linker/config_home/extra_tables.json	特权表，默认关闭，max_extra_tables=10
.sql_linker/table_home/table_dictionary.json	带示例表的空模板
.sql_linker/.env	凭据占位符提示

---

配置文件说明 / Config Files

table_home/table_dictionary.json — 主词典 / Main Dictionary

（中文）所有受控业务表必须在主词典中声明，字段白名单仅对 NORMAL 层生效：

（English）All controlled business tables must be declared in the main dictionary. Field whitelist only applies to NORMAL layer:

{
  "version": 1,
  "tables": [
    {
      "table_name": "supplier_table",
      "comment": "供应商信息表",
      "fields": [
        { "name": "id",            "type": "BIGINT",       "pk": true,  "auto": true  },
        { "name": "supplier_code", "type": "VARCHAR(32)",  "pk": false, "auto": false },
        { "name": "supplier_name", "type": "VARCHAR(128)", "pk": false, "auto": false },
        { "name": "short_name",    "type": "VARCHAR(64)",  "pk": false, "auto": false },
        { "name": "supplier_level","type": "VARCHAR(16)",  "pk": false, "auto": false },
        { "name": "contact_person","type": "VARCHAR(64)",  "pk": false, "auto": false },
        { "name": "contact_phone", "type": "VARCHAR(32)",  "pk": false, "auto": false },
        { "name": "contact_email", "type": "VARCHAR(128)", "pk": false, "auto": false },
        { "name": "status",        "type": "VARCHAR(16)",  "pk": false, "auto": false },
        { "name": "created_at",    "type": "DATETIME",     "pk": false, "auto": false },
        { "name": "updated_at",   "type": "DATETIME",     "pk": false, "auto": false }
      ]
    }
  ]
}

config_home/extra_tables.json — 特权表配置 / Privileged Table Config

（中文）词典外表格需通过此配置显式授权，enabled=false 时所有非词典表均 BLOCKED：

（English）Tables outside the dictionary require explicit authorization via this config. When enabled=false, all non-dictionary tables are BLOCKED:

{
  "version": 1,
  "enabled": false,
  "max_extra_tables": 10,
  "tables": [
    { "table_name": "employee_table" },
    { "table_name": "payroll_table" }
  ]
}

字段	说明
enabled	false=关闭词典外访问（默认）/ true=开启特权模式
max_extra_tables	最大允许声明数量，防止配置失控
tables[].table_name	特权表名

config_home/config.yaml — 连接配置 / Connection Config

（中文）数据库连接配置，password 不直接写在文件中，通过 password_env 引用 .env 中的 key：

（English）Database connection config. Password is not written directly; password_env references a key in .env:

type: mysql
host: 127.0.0.1
port: 3306
database: db_dev
user: admin
password_env: MYSQLPW_ENV   # 引用 .env 中的 key（不直接写明文）
read_only: false
max_rows: 1000
timeout: 30
extra_tables_enabled: false   # 默认关闭词典外访问

---

Python API — 业务层 / Business Layer (Recommended)

（中文）使用业务层 API（推荐），完整支持四层访问控制和时间戳自动注入：

（English）Use the business layer API (recommended), with full four-layer access control and automatic timestamp injection:

import sys
sys.path.insert(0, "skills/sql-linker/scripts/service_layer")
from db_bridge import DBBridge

db = DBBridge(
    user_label="openclaw-control-ui",   # ← OpenClaw metadata.label
    session_id="agent:hr:main"          # ← OpenClaw metadata.id
)

INSERT — 时间戳自动注入 / Automatic Timestamp Injection

（中文）INSERT 操作自动生成 created_at 和 updated_at（两者同值），仅写入主词典中声明的字段：

（English）INSERT operations automatically generate created_at and updated_at (same value), writing only fields declared in the main dictionary:

db.insert("supplier_table", {
    "supplier_code": "LX001",
    "supplier_name": "立讯精密",
    "supplier_level": "A",
    "status": "active"
})
# → created_at / updated_at 自动生成，无需手动传入

UPDATE — 时间戳自动刷新 / Automatic Timestamp Refresh

（中文）UPDATE 操作自动刷新 updated_at，created_at 保持不变：

（English）UPDATE operations automatically refresh updated_at, leaving created_at unchanged:

db.update(
    "supplier_table",
    {"supplier_level": "AA"},
    "supplier_code = %s",
    ("LX001",)
)
# → updated_at 自动刷新为当前时间

DELETE — 完整审计 / Full Audit

db.delete("supplier_table", "status = %s", ("inactive",))

SELECT — 参数化防注入 / Parameterized Injection Prevention

rows = db.query(
    "SELECT * FROM supplier_table WHERE status = %s AND supplier_level = %s",
    ("active", "A")
)
for row in rows:
    print(row)

辅助方法 / Helper Methods

db.tables()           # 返回主词典中所有表格名
db.extra_tables()     # 返回当前特权表列表
db.system_tables()    # 返回受保护的系统表列表
db.fields("supplier_table")  # 返回表字段列表（privilege 表返回空）
db.bootstrap()        # 执行引导初始化，返回被创建的文件路径列表

错误处理 / Error Handling

from db_bridge import TableAccessDenied, SystemTableWriteDenied

try:
    db.query("SELECT * FROM unknown_table LIMIT 1")
except TableAccessDenied as e:
    print("访问被拒绝 / Access denied:", e)

try:
    db.update("sql_audit_log", {"status": "tampered"}, "id = %s", (1,))
except SystemTableWriteDenied as e:
    print("系统表写入被拒绝 / System table write denied:", e)

---

Python API — 数据操作层 / Data Operation Layer

（中文）直接使用数据操作层，跳过业务层字段过滤和时间戳注入（适用于高级用户）：

（English）Use the data operation layer directly, bypassing business layer field filtering and timestamp injection (for advanced users):

import sys
sys.path.insert(0, "skills/sql-linker/scripts/controller_layer")
from sql_linker import SQLLinker

linker = SQLLinker()
linker.connect()
linker.set_user_context_auto()   # 必须：注入审计上下文

---

时间戳注入规则 / Timestamp Logic

操作 / Operation	created_at	updated_at	适用层级 / Applicable Layer
INSERT	Auto	Auto (same as created)	NORMAL
UPDATE	Unchanged	Auto-refresh	NORMAL
DELETE	N/A	N/A	NORMAL
PRIVILEGED	N/A	N/A	PRIVILEGED

---

审计日志 / Audit Trail

（中文）配置位置：.sql_linker/config_home/audit_config.json

（English）Config location: .sql_linker/config_home/audit_config.json

{
  "username": "HR",
  "audit": {
    "enabled": true,
    "log_table": "sql_audit_log",
    "log_select": false,
    "mask_values": true
  }
}

（中文）审计记录字段（自动注入，不可为空）：

（English）Audit record fields (automatically injected, must not be empty):

字段	说明	来源
user_name	操作人	audit_config.json username
user_label	来源标签	OPENCLAW_LABEL 环境变量
ip_address	本机局域网 IP	自动检测（UDP/gethostbyname）
session_id	OpenClaw Session Key	OPENCLAW_SESSION 环境变量
operation	操作类型	SELECT / INSERT / UPDATE / DELETE
table_name	目标表	从 SQL 中提取
sql_statement	SQL 语句	参数化脱敏（%s）
rows_affected	影响行数	数据库返回
status	操作状态	SUCCESS / FAILED

---

字段类型参考 / Field Type Reference

type 值	说明
BIGINT	主键/自增ID
VARCHAR(n)	字符串，最大 n 字符
TEXT	长文本
INT	整数
DECIMAL(m,n)	小数，m位总长，n位小数
DATETIME	日期时间（YYYY-MM-DD HH:MM:SS）
DATE	日期
BOOL	布尔值

---

安全原则 / Security Principles

（中文）

1. 字段白名单：NORMAL 表只写入 table_dictionary.json 中声明的字段，非法字段自动过滤
2. 四层访问控制：SYSTEM 表（只读+写入审计）/ 词典内（白名单+时间戳）/ 特权（直查）/ 禁用（拒绝）
3. 参数化查询：全部使用 %s + tuple 防止 SQL 注入
4. 敏感凭据分离：password_env 引用 .env，不写在 config.yaml 中
5. 审计完整：所有操作（含拒绝）均记录审计日志
6. 系统表保护：sql_audit_log 禁止 UPDATE/DELETE，SystemTableWriteDenied 异常强制拦截
7. Bootstrap 幂等：配置文件缺失时自动生成，不覆盖已有配置

（English）

1. Field Whitelist: NORMAL tables only write fields declared in table_dictionary.json; illegal fields are automatically filtered
2. Four-Layer Access Control: SYSTEM table (read+audit write) / Normal dictionary (whitelist+timestamp) / Privileged (direct query) / Blocked (denied)
3. Parameterized Queries: All use %s + tuple to prevent SQL injection
4. Sensitive Credential Separation: password_env references .env key, not written in config.yaml
5. Full Audit Trail: All operations (including denials) are recorded in audit logs
6. System Table Protection: sql_audit_log prohibits UPDATE/DELETE, SystemTableWriteDenied exception enforced
7. Idempotent Bootstrap: Missing config files are auto-generated without overwriting existing configs

---

常见错误与解法 / Common Errors and Solutions

错误 / Error	原因 / Cause	解法 / Solution
TableAccessDenied: Table 'xxx' not in dictionary	表格不在词典且未授权 / Table not in dictionary and not authorized	加入 extra_tables.json + enabled:true
SystemTableWriteDenied: sql_audit_log does not allow UPDATE	尝试篡改审计日志 / Attempt to tamper audit log	属正常拦截，如确认误判请通过 DBA 直接操作数据库 / Normal interception; if misjudged, contact DBA
created_at / updated_at 为 NULL	操作非 NORMAL 表格 / Operating on non-NORMAL table	privilege 表格不注入时间戳，属正常行为 / Privilege tables do not inject timestamps, this is normal
Access denied for user ... (using password: NO)	password_env 未解析 / password_env not resolved	sql_linker ≥0.4.0 已修复，从 .env 读取 / Fixed in sql_linker ≥0.4.0, reads from .env
Config file not found	配置文件缺失 / Config file missing	调用 db.bootstrap() 自动生成，或检查 .sql_linker/ 目录结构 / Call db.bootstrap() to auto-generate, or check .sql_linker/ directory structure
Table not found	表未在 table_dictionary.json 中声明 / Table not declared	在主词典中添加表配置 / Add table config in main dictionary

---

目录结构总览 / Directory Structure Overview

workspace/
└── .sql_linker/
    ├── .env                            ← 敏感凭据（不随技能迁移）
    ├── config_home/
    │   ├── config.yaml                 ← 连接配置（extra_tables_enabled 开关）
    │   ├── audit_config.json           ← 审计配置
    │   └── extra_tables.json           ← 特权表列表（JSON）
    └── table_home/
        └── table_dictionary.json       ← 主词典（JSON，所有受控表）
            └── tables[]                 ← 每表 fields[] 白名单 + comment

skills/sql-linker/
├── SKILL.md                           ← 本文档
└── scripts/
    ├── controller_layer/               ← 数据操作层
    │   ├── sql_linker.py              ← 连接管理 + CRUD + 审计
    │   └── sql_audit.py               ← 审计模块
    └── service_layer/                  ← 业务层
        └── db_bridge.py               ← 四层访问 + 时间戳 + Bootstrap