Install
openclaw skills install sql-linkerUse this skill whenever you need to query, insert, update, or delete database records. Triggers include: (1) querying database data with SELECT statements an...
openclaw skills install sql-linkerImportant: All
scripts/paths are relative to this skill directory. Run with:cd {skill_dir} && python scripts/...or use thecwdparameter.
Please read this notice before using this skill. / (必读)使用本 skill 前请仔细阅读本声明。
How credentials are resolved (in order of precedence):
| Field in config.yaml | Resolution | Risk Level |
|---|---|---|
password | Direct plaintext in config | ⚠️ Not recommended; commits secret to config file |
password_env | Reads .env file in ~/.sql_linker/ for the named key | Moderate — silently accesses workspace secrets |
password_dpapi | DPAPI-decrypts base64 value using current Windows user credential | Moderate — can recover stored secret at runtime |
⚠️ Disclosure: When
password_envorpassword_dpapiis configured, this skill will automatically read/decrypt credentials at connection time with no additional user prompt. Ensure this skill is invoked only in trusted contexts. Do not setpassword_envpointing to keys used by other projects.
Optional Hardening —
require_explicit_credential_approval: To force explicit confirmation before silent credential loading, setrequire_explicit_credential_approval: trueinaudit_config.json. When enabled, the first connection attempt withpassword_env/password_dpapiwill raisePermissionErroruntil you calldb.explicit_credential_approval()in your code. This prevents accidental silent credential access in untrusted contexts.
Creating a SQLLinker or DBBridge instance does not automatically connect. Connection is deferred until the first actual database call (connect() is called lazily on first query). This avoids premature infrastructure access.
Data Minimization: Audit records collect the minimum identity fields required for compliance traceability:
user_name,user_label,session_id, and optionallyip_address(disabled by default). SQL text is masked before logging. No passwords, personal identity numbers, or business-sensitive fields are intentionally captured.Retention: Audit records are stored in
sql_audit_login the target database. Retention policy is determined by your organization's database retention schedule, not by this skill.Opt-Out: Set
audit: enabled: falseinaudit_config.jsonto disable application-layer audit logging. Database-layer triggers (if any) are independent of this setting.Consent: By using this skill, you consent to having database operation metadata (operator identity, table name, masked SQL, row counts, timestamp, status) recorded in
sql_audit_log. Do not include sensitive personal data (e.g., national ID numbers, passwords, medical info) in SQL query parameters — such values will be masked but still persisted in log records.
Audit Log: Every database operation records the following fields to sql_audit_log:
| Field | Description | Source |
|---|---|---|
user_name | Operator name | Explicit parameter or audit_config.json username |
user_label | Source label | Explicit parameter or OPENCLAW_LABEL env |
ip_address | Client IP | Explicit parameter or LAN IP (only if collect_lan_ip: true) |
session_id | Session identifier | Explicit parameter or OPENCLAW_SESSION env |
sql_statement | Full SQL statement | Parameterized and masked (literals replaced with ?) |
rows_affected | Rows affected | Database return |
status | Operation status | SUCCESS / FAILED |
⚡ Privacy Notice: SQL text is masked before logging — string and numeric literals are replaced with ?. No password or raw PII is intentionally stored. However, the log table itself contains identity metadata; treat it as sensitive. Avoid including sensitive personal data in query parameters.
By default, this skill auto-collects audit context from the following sources:
| Source | Collected Data | How to Disable |
|---|---|---|
OPENCLAW_USER env | Username | Explicitly pass user_name parameter |
OPENCLAW_LABEL env | Source label | Explicitly pass user_label parameter |
OPENCLAW_SESSION env | Session ID | Explicitly pass session_id parameter |
| LAN IP auto-detection | Local LAN IP | Set collect_lan_ip: false in audit_config.json (default: false) |
If you do not want auto-collection, explicitly pass user_label and session_id parameters — this skill will prefer passed values over auto-discovery.
On first use or when config files are missing, this skill auto-creates files under ~/.sql_linker/:
| File | Description |
|---|---|
.env | Credential placeholder (replace with real credentials) |
config_home/config.yaml | Database connection config |
config_home/audit_config.json | Audit configuration |
config_home/extra_tables.json | Privileged table config |
table_home/table_dictionary.json | Main dictionary |
For full manual control, create these files before invoking the skill. bootstrap() is idempotent but prints a safety warning before creating files.
password > password_env > password_dpapi
password: Direct plaintext (not recommended)password_env: Lookup key in .env filepassword_dpapi: Windows DPAPI decryption (Windows only, user-scoped)UPDATE / DELETE operations execute directly and cannot be rolled back. In production, enable read-only mode (read_only: true) for pre-validation.
Version 1.2.2 → 1.2.3 Changes:
bootstrap(dry_run=False) now requires explicit_confirm=True to write files — a BootstrapConfirmationRequired exception is raised otherwise. This prevents accidental configuration persistence in shared workspaces (Finding #1 of the ClawHub security audit, 62% confidence)BootstrapConfirmationRequired raised when bootstrap write is attempted without explicit confirmationDBBridge.bootstrap() forwards the new explicit_confirm parameterVersion 1.1.1 → 1.2.0 Changes:
password_env / password_dpapi auto-resolution and the no-prompt nature of credential loadingSQLLinker/DBBridge instantiation does NOT auto-connectlog_select: false (default false) — SELECT logging only occurs when BOTH audit: enabled: true AND log_select: truecollect_lan_ip defaults to false (was not explicitly defaulted before)session_id / user_label prefer explicitly passed values; no longer auto-read from sessions.json(中文) 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.
(中文) 系统由两层组成,业务层和数据操作层职责分明:
(English) The system consists of two layers with clearly defined responsibilities:
workspace/
└── .sql_linker/ ← Config root
├── config_home/
│ ├── config.yaml ← DB connection config
│ ├── audit_config.json ← Audit config
│ └── extra_tables.json ← Privileged table config (JSON)
└── table_home/
└── table_dictionary.json ← Main dictionary (JSON, all controlled tables)
skills/sql-linker/scripts/
├── controller_layer/ ← Data operation layer
│ ├── sql_linker.py ← Connection management + CRUD execution + audit context injection
│ └── sql_audit.py ← Audit module (used internally by sql_linker.py)
└── service_layer/ ← Business layer
└── db_bridge.py ← Four-layer access control + timestamp injection + field whitelist
(中文) 业务层(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.
(中文) 系统通过四层访问模型实现精确的表访问控制:
(English) The system implements precise table access control through a four-layer access model:
SYSTEM (系统表 sql_audit_log)
NORMAL (主词典表格)
PRIVILEGED (特权表格)
BLOCKED (禁用)
(中文) 访问判定流程:提取 SQL 中的表名 → 检查 SYSTEM → 检查主词典(NORMAL)→ 检查 extra_tables(PRIVILEGED)→ 其余 BLOCKED。
(中文) 首次使用或缺少配置文件时,系统自动生成默认模板(幂等操作,不会覆盖已有文件):
(English) On first use or when config files are missing, the system automatically generates default templates (idempotent, will not overwrite existing files):
⚠️ Bootstrap 自动创建配置:首次使用时会自动在
~/.sql_linker/目录下创建配置文件(.env、config.yaml、audit_config.json等)。如需手动控制,请提前自行创建这些文件。bootstrap()为幂等操作,不会覆盖已有文件,但会创建缺失的文件。
from db_bridge import DBBridge
db = DBBridge(user_label="openclaw-control-ui", session_id="agent:hr:main")
# Preview files to be created (no actual write)
preview = db.bootstrap(dry_run=True)
print(f'Will create: {preview}')
# Execute actual bootstrap
created = db.bootstrap()
print(f'Created: {created}')
# ['...\\config.yaml', '...\\audit_config.json', ...]
(中文) 自动生成的文件列表:
(English) Auto-generated files:
| File Path | Default Content |
|---|---|
.sql_linker/config_home/config.yaml | Connection template (host/port/user placeholders, password_env references .env) |
.sql_linker/config_home/audit_config.json | Audit ON by default, log_select=false, collect_lan_ip=false |
.sql_linker/config_home/extra_tables.json | Privileged tables, disabled by default, max_extra_tables=10 |
.sql_linker/table_home/table_dictionary.json | Empty template with example table |
.sql_linker/.env | Credential placeholder (replace with actual credentials) |
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" }
]
}
| Field | Description |
|---|---|
enabled | false=disable dict-external access (default) / true=enable privileged mode |
max_extra_tables | Max declared tables, prevents config runaway |
tables[].table_name | Privileged 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 # Reference .env key (not plaintext)
# password_dpapi: AQAAANCMnd8BFdERjHoAwE/Cl+sBAAAA... # DPAPI-encrypted password (optional, mutually exclusive with password_env)
read_only: false
max_rows: 1000
timeout: 30
extra_tables_enabled: false # Disabled by default
Security Option — DPAPI Encrypted Password (Optional): If you prefer not to store plaintext passwords in
.env, usepassword_dpapifield instead. Value is a Windows DPAPI-encrypted base64 string, decrypted by current user process only, invalid if migrated to different machine/user.Generation method (run on target machine):
import base64, win32crypt pw = 'YOUR_REAL_PASSWORD_HERE' # ← replace with your actual password enc = win32crypt.CryptProtectData(pw.encode()) print(base64.b64encode(enc).decode())⚠️ Never hard-code real passwords in documentation. Always generate the DPAPI string on the target machine and store only the encrypted form in
config.yaml.
(中文) 使用业务层 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 操作自动生成 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 auto-generated, no manual injection needed
(中文) 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 auto-refreshed to current time
db.delete("supplier_table", "status = %s", ("inactive",))
rows = db.query(
"SELECT * FROM supplier_table WHERE status = %s AND supplier_level = %s",
("active", "A")
)
for row in rows:
print(row)
db.tables() # Return all table names in main dictionary
db.extra_tables() # Return current privileged table list
db.system_tables() # Return protected system table list
db.fields("supplier_table") # Return table field list (privilege table returns empty)
db.bootstrap(dry_run=False) # Execute bootstrap; dry_run=True returns file list without writing
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)
(中文) 直接使用数据操作层,跳过业务层字段过滤和时间戳注入(适用于高级用户):
(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()
# Explicit audit context (preferred over auto-discovery)
linker.set_user_context(user_name="HR", user_label="openclaw-control-ui",
ip_address="", session_id="agent:hr:main")
| 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 |
(中文) 配置位置:.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,
"collect_lan_ip": false
}
}
(中文) 审计记录字段(自动注入,不可为空):
(English) Audit record fields (automatically injected, must not be empty):
| Field | Description | Source |
|---|---|---|
user_name | Operator | audit_config.json username |
user_label | Source label | Explicit or OPENCLAW_LABEL env |
ip_address | Local LAN IP | Explicit or _get_lan_ip() (disabled by default) |
session_id | OpenClaw Session Key | Explicit or OPENCLAW_SESSION env |
operation | Operation type | SELECT / INSERT / UPDATE / DELETE |
table_name | Target table | Extracted from SQL |
sql_statement | SQL statement | Parameterized mask (%s) |
rows_affected | Rows affected | Database return |
status | Operation status | SUCCESS / FAILED |
| type value | Description |
|---|---|
BIGINT | Primary key / auto-increment ID |
VARCHAR(n) | String, max n characters |
TEXT | Long text |
INT | Integer |
DECIMAL(m,n) | Decimal, m total digits, n decimal places |
DATETIME | Date time (YYYY-MM-DD HH:MM:SS) |
DATE | Date |
BOOL | Boolean |
sql-linker adopts application layer + database layer dual-layer audit; any direct-connection bypass of the application layer is still captured.
sql_audit_log is in SYSTEM layer, business layer can SELECT directly:
from datetime import date, timedelta
db = DBBridge(user_label="audit-viewer", session_id="agent:audit")
# Query today's operation records
today = date.today().strftime('%Y-%m-%d')
rows = db.query(
"SELECT log_time, user_name, operation, table_name, sql_statement, rows_affected, status "
"FROM sql_audit_log WHERE DATE(log_time) = %s ORDER BY log_time DESC",
(today,)
)
# Query failed operations in last 7 days
week_ago = (date.today() - timedelta(days=7)).strftime('%Y-%m-%d')
failed = db.query(
"SELECT * FROM sql_audit_log WHERE status = 'FAILED' AND log_time >= %s "
"ORDER BY log_time DESC",
(week_ago,)
)
# Query recent operations for a supplier (fuzzy match)
rows = db.query(
"SELECT log_time, user_name, operation, table_name, sql_statement "
"FROM sql_audit_log WHERE sql_statement LIKE %s ORDER BY log_time DESC LIMIT 20",
('%LX001%',)
)
⚠️
sql_audit_logis a system-protected table. UPDATE/DELETE blocked bySystemTableWriteDenied. Only SELECT is allowed.
sql_linker.py → SQLAudit classsql_audit_log after each CRUD operationdb_bridge onlypymysql / mysql CLI can bypassAFTER INSERT/UPDATE/DELETE triggers on MySQL side, mandatory write to sql_audit_logOn each controlled business table, create one AFTER trigger each for INSERT / UPDATE / DELETE. Example structure:
-- Example using supplier_capa (same for other tables)
CREATE TRIGGER trg_<table>_ai
AFTER INSERT ON <table>
FOR EACH ROW
BEGIN
INSERT INTO sql_audit_log
(log_time, user_name, user_label, ip_address, session_id,
db_type, operation, table_name, sql_statement, rows_affected, status, error_msg)
VALUES
(NOW(), CURRENT_USER(), 'DB_TRIGGER', 'internal', 'DB_TRIGGER',
'mysql', 'INSERT', '<table>',
CONCAT('INSERT id=', NEW.id, ' supplier_code=', NEW.supplier_code),
1, 'SUCCESS', NULL);
END;
Implementation Steps:
sql_audit_log.id is AUTO_INCREMENT (otherwise trigger INSERT fails due to no default id)| Operation Path | App-Layer Audit | Trigger Audit | Conclusion |
|---|---|---|---|
| db_bridge CRUD | ✅ Logged | ✅ Logged | Double guarantee |
| pymysql direct | ❌ Bypassed | ✅ Logged | Trigger fallback |
| mysql CLI direct | ❌ Bypassed | ✅ Logged | Trigger fallback |
| DBA direct op | ❌ Bypassed | ✅ Logged | Trigger fallback |
(中文)
NORMAL tables only write fields declared in table_dictionary.json; illegal fields auto-filter%s + tuple to prevent SQL injectionpassword_env references .env key; password_dpapi (DPAPI encrypted) alternative — neither plaintextdb_bridge + database-layer triggers (created by deployer per actual schema)sql_audit_log prohibits UPDATE/DELETE, SystemTableWriteDenied exception enforced(English)
NORMAL tables only write fields declared in table_dictionary.json; illegal fields are automatically filtered%s + tuple to prevent SQL injectionpassword_env references .env key, not written in config.yamldb_bridge + database-layer triggers (created by deployer per actual schema)sql_audit_log prohibits UPDATE/DELETE, SystemTableWriteDenied exception enforced| Error | Cause | Solution |
|---|---|---|
TableAccessDenied: Table 'xxx' not in dictionary | Table not in dictionary and not authorized | Add to extra_tables.json + enabled:true |
SystemTableWriteDenied: sql_audit_log does not allow UPDATE | Attempt to tamper audit log | Normal interception; if misjudged, contact DBA |
| Audit write failed but data succeeded | Audit and business not in same transaction | Triggers provide fallback; app-layer fix pending |
Access denied for user ... (using password: NO) | password_env / password_dpapi unresolved | Check .env key present or DPAPI-encrypted on this machine |
Config file not found | Config file missing | Call db.bootstrap() to auto-generate, or check .sql_linker/ structure |
Table not found | Table not declared in table_dictionary.json | Add table config in main dictionary |
workspace/
└── .sql_linker/
├── .env ← Sensitive credentials (not migrated with skill)
├── config_home/
│ ├── config.yaml ← Connection config (extra_tables_enabled switch)
│ ├── audit_config.json ← Audit config (collect_lan_ip option)
│ └── extra_tables.json ← Privileged table list (JSON)
└── table_home/
└── table_dictionary.json ← Main dictionary (JSON, all controlled tables)
└── tables[] ← Each table's fields[] whitelist + comment
skills/sql-linker/
├── SKILL.md ← This document
└── scripts/
├── controller_layer/ ← Data operation layer
│ ├── sql_linker.py ← Connection management + CRUD + audit
│ └── sql_audit.py ← Audit module
└── service_layer/ ← Business layer
└── db_bridge.py ← Four-layer access + timestamp + Bootstrap