SQLite数据库专家Skill SQLiteDatabaseExpert

---

name: SQLite 数据库专家 risk_level: 高 description: SQLite 嵌入式数据库开发专家，专注于 Tauri/桌面应用程序的 SQL 注入预防、迁移、FTS 搜索和安全数据处理 version: 1.0.0 author: JARVIS AI 助手 tags: [数据库, sqlite, sql, 嵌入式, 迁移, fts, 安全] model: claude-sonnet-4-5-20250929

SQLite 数据库专家

0. 强制阅读协议

关键: 在实现任何数据库操作之前，您必须阅读相关的参考文件：

参考文件的触发条件

当以下情况时阅读 references/advanced-patterns.md:

• 实现数据库迁移
• 设置全文搜索 (FTS5)
• 设计使用 CTE 或窗口函数的复杂查询
• 实现连接池或 WAL 模式
• 性能优化任务

当以下情况时阅读 references/security-examples.md:

• 编写任何包含用户输入的 SQL 查询
• 实现参数化查询
• 设置数据库加密考虑
• 处理敏感数据存储
• 为数据库操作实现输入验证

---

1. 概述

风险级别: 中等

理由: 桌面应用程序中的 SQLite 数据库本地处理用户数据，如果查询未正确参数化，则存在 SQL 注入风险，并且需要仔细的迁移管理以防止数据丢失。

您是一名 SQLite 嵌入式数据库开发专家，专长于:

• 安全 SQL 模式 使用参数化查询预防 SQL 注入
• 数据库迁移 具有版本控制和回滚功能
• 全文搜索 (FTS5) 用于高效文本搜索
• 性能优化 包括索引、WAL 模式和连接管理
• Rust/Tauri 集成 使用 rusqlite 和 sea-query

核心原则

1. 测试驱动开发优先 - 在实现之前编写测试；使用内存中 SQLite 进行快速测试执行
2. 性能意识 - 使用 WAL 模式、预编译语句、批量操作和适当的索引进行优化
3. 安全第一 - 始终使用参数化查询；切勿拼接用户输入
4. 事务安全 - 将相关操作包装在事务中以实现原子性
5. 迁移纪律 - 版本化所有架构更改并具有回滚能力

主要用例

• 桌面应用程序的本地数据持久化
• 离线优先应用程序数据存储
• 全文搜索实现
• 配置和设置存储
• 缓存和临时数据管理

---

2. 核心职责

2.1 安全优先的数据库操作

1. 始终使用参数化查询 - 切勿将用户输入拼接到 SQL 字符串中
2. 验证所有输入 在数据库操作之前
3. 实现适当的错误处理 不暴露数据库内部细节
4. 使用事务 以确保数据完整性
5. 应用最小权限原则 用于数据库访问

2.2 数据完整性原则

1. 架构版本控制 带有迁移跟踪
2. 外键强制执行 使用 PRAGMA foreign_keys = ON
3. 约束验证 在数据库级别
4. 备份策略 在破坏性操作之前

---

3. 技术基础

3.1 版本推荐

组件	推荐	最低	备注
SQLite	3.45+	3.35	FTS5, JSON 函数
rusqlite	0.31+	0.29	捆绑 SQLite 支持
sea-query	0.30+	0.28	查询构建器
r2d2	0.8+	0.8	连接池

3.2 必需依赖项 (Cargo.toml)

[dependencies]
rusqlite = { version = "0.31", features = ["bundled", "backup", "functions"] }
sea-query = "0.30"
sea-query-rusqlite = "0.5"
r2d2 = "0.8"
r2d2_sqlite = "0.24"

---

4. 实现模式

4.1 数据库初始化

use rusqlite::{Connection, Result};
use std::path::Path;

pub struct Database {
    conn: Connection,
}

impl Database {
    pub fn new(path: &Path) -> Result<Self> {
        let conn = Connection::open(path)?;

        // 启用安全和性能功能
        conn.execute_batch("
            PRAGMA foreign_keys = ON;
            PRAGMA journal_mode = WAL;
            PRAGMA synchronous = NORMAL;
            PRAGMA temp_store = MEMORY;
            PRAGMA mmap_size = 30000000000;
            PRAGMA page_size = 4096;
        ")?;

        Ok(Self { conn })
    }
}

4.2 参数化查询 (关键)

// 正确：参数化查询
pub fn get_user_by_id(&self, user_id: i64) -> Result<Option<User>> {
    let mut stmt = self.conn.prepare(
        "SELECT id, name, email FROM users WHERE id = ?1"
    )?;

    let user = stmt.query_row([user_id], |row| {
        Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
        })
    }).optional()?;

    Ok(user)
}

// 正确：命名参数以提高清晰度
pub fn search_users(&self, name: &str, status: &str) -> Result<Vec<User>> {
    let mut stmt = self.conn.prepare(
        "SELECT id, name, email FROM users
         WHERE name LIKE :name AND status = :status"
    )?;

    let users = stmt.query_map(
        &[(":name", &format!("%{}%", name)), (":status", &status)],
        |row| Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
        })
    )?.collect::<Result<Vec<_>>>()?;

    Ok(users)
}

// 错误：SQL 注入漏洞
pub fn get_user_unsafe(&self, user_id: &str) -> Result<Option<User>> {
    // 切勿这样做 - SQL 注入风险
    let query = format!("SELECT * FROM users WHERE id = {}", user_id);
    // ...
}

4.3 事务管理

pub fn transfer_funds(
    &mut self,
    from_id: i64,
    to_id: i64,
    amount: f64
) -> Result<()> {
    let tx = self.conn.transaction()?;

    // 从源账户扣除
    tx.execute(
        "UPDATE accounts SET balance = balance - ?1 WHERE id = ?2",
        [amount, from_id as f64],
    )?;

    // 添加到目标账户
    tx.execute(
        "UPDATE accounts SET balance = balance + ?1 WHERE id = ?2",
        [amount, to_id as f64],
    )?;

    tx.commit()?;
    Ok(())
}

4.4 全文搜索 (FTS5)

// 创建 FTS5 虚拟表并带触发器
pub fn setup_fts(&self) -> Result<()> {
    self.conn.execute_batch("
        CREATE VIRTUAL TABLE IF NOT EXISTS docs_fts USING fts5(
            title, content, tags, content=documents, content_rowid=id
        );
        CREATE TRIGGER IF NOT EXISTS docs_ai AFTER INSERT ON documents BEGIN
            INSERT INTO docs_fts(rowid, title, content, tags)
            VALUES (new.id, new.title, new.content, new.tags);
        END;
    ")?;
    Ok(())
}

// 搜索并高亮显示
pub fn search_documents(&self, query: &str) -> Result<Vec<Document>> {
    let mut stmt = self.conn.prepare(
        "SELECT d.*, highlight(docs_fts, 1, '<mark>', '</mark>') as snippet
         FROM documents d JOIN docs_fts ON d.id = docs_fts.rowid
         WHERE docs_fts MATCH ?1 ORDER BY rank"
    )?;
    stmt.query_map([query], |row| Ok(Document { /* ... */ }))?.collect()
}

---

5. 安全标准

5.1 关键漏洞

缓解措施: 更新到 SQLite 3.44.0+ 并始终使用参数化查询。

5.2 OWASP 映射

OWASP 类别	风险	关键控制
A03 - 注入	关键	参数化查询，输入验证
A04 - 不安全设计	中等	架构约束，外键
A05 - 错误配置	中等	安全 PRAGMAs，文件权限 (600)

5.3 SQL 注入预防

关键规则 (见 references/security-examples.md):

1. 切勿使用字符串格式化进行 SQL 查询
2. 始终使用 ? 位置参数或 :name 命名参数
3. 动态查询时白名单列/表名

// 动态列选择 - 安全方法
pub fn get_user_fields(&self, user_id: i64, fields: &[&str]) -> Result<HashMap<String, String>> {
    const ALLOWED: &[&str] = &["id", "name", "email", "created_at"];
    let safe_fields: Vec<&str> = fields.iter()
        .filter(|f| ALLOWED.contains(f)).copied().collect();
    if safe_fields.is_empty() { return Err(rusqlite::Error::InvalidQuery); }
    let query = format!("SELECT {} FROM users WHERE id = ?1", safe_fields.join(", "));
    let mut stmt = self.conn.prepare(&query)?;
    // ...
}

---

6. 测试标准

6.1 Rust 测试模式

#[cfg(test)]
mod tests {
    use super::*;
    use rusqlite::Connection;

    fn setup_test_db() -> Database {
        let conn = Connection::open_in_memory().unwrap();
        let db = Database { conn };
        db.run_migrations().unwrap();
        db
    }

    #[test]
    fn test_sql_injection_prevented() {
        let db = setup_test_db();
        let result = db.search_users("'; DROP TABLE users; --", "active");
        assert!(result.is_ok());
        assert!(db.get_user_by_id(1).is_ok()); // 表仍然存在
    }
}

---

7. 实施工作流程 (测试驱动开发)

步骤 1: 首先编写失败测试

# tests/test_user_repository.py
import pytest
import sqlite3

@pytest.fixture
def db():
    """内存中 SQLite 用于快速测试。"""
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    yield conn
    conn.close()

class TestUserRepository:
    def test_create_user_returns_id(self, db):
        repo = UserRepository(db)
        repo.initialize_schema()
        user_id = repo.create_user("test@example.com", "Test User")
        assert user_id > 0

    def test_sql_injection_prevented(self, db):
        repo = UserRepository(db)
        repo.initialize_schema()
        malicious = "'; DROP TABLE users; --"
        user_id = repo.create_user(malicious, "Hacker")
        assert repo.get_by_id(user_id)["email"] == malicious

步骤 2: 实现最小代码以通过

# app/repositories/user.py
class UserRepository:
    def __init__(self, conn):
        self.conn = conn

    def initialize_schema(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                email TEXT NOT NULL UNIQUE,
                name TEXT NOT NULL
            )""")
        self.conn.commit()

    def create_user(self, email: str, name: str) -> int:
        cursor = self.conn.execute(
            "INSERT INTO users (email, name) VALUES (?, ?)", (email, name))
        self.conn.commit()
        return cursor.lastrowid

    def get_by_id(self, user_id: int):
        return self.conn.execute(
            "SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()

步骤 3: 运行验证

pytest tests/test_*_repository.py -v --cov=app/repositories

---

7.1 性能模式

模式 1: WAL 模式

# 好：启用 WAL 以进行并发读写
conn.execute("PRAGMA journal_mode = WAL")
conn.execute("PRAGMA synchronous = NORMAL")
conn.execute("PRAGMA cache_size = -64000")  # 64MB

# 坏：默认 DELETE 模式在写入期间阻塞读取

模式 2: 批量插入

# 好：单一事务用于批量
conn.executemany("INSERT INTO items (name) VALUES (?)", records)
conn.commit()

# 坏：每行提交（慢 100 倍）
for r in records:
    conn.execute("INSERT INTO items (name) VALUES (?)", (r,))
    conn.commit()

模式 3: 连接池

# 好：重用连接
from queue import Queue
class ConnectionPool:
    def __init__(self, db_path, size=5):
        self.pool = Queue(size)
        for _ in range(size):
            conn = sqlite3.connect(db_path, check_same_thread=False)
            conn.execute("PRAGMA journal_mode = WAL")
            self.pool.put(conn)

# 坏：每个查询新连接
conn = sqlite3.connect(db_path)  # 昂贵！

模式 4: 索引优化

# 好：覆盖和部分索引
conn.executescript("""
    CREATE INDEX idx_users_email ON users(email, name);
    CREATE INDEX idx_active ON items(created_at) WHERE status='active';
    ANALYZE;
""")

# 坏：未索引列上的全表扫描

模式 5: VACUUM 调度

# 好：空闲时间维护
def nightly_maintenance(conn):
    conn.execute("PRAGMA optimize")
    freelist = conn.execute("PRAGMA freelist_count").fetchone()[0]
    if freelist > 1000:
        conn.execute("VACUUM")

# 坏：高峰使用期间或从不 VACUUM

---

8. 常见错误

错误	错误做法	正确做法
SQL 注入	format!("...WHERE name = '{}'", input)	"...WHERE name = ?1" 带参数
无事务	单独的 execute 调用	包装在 transaction() + commit() 中
无外键	默认连接	PRAGMA foreign_keys = ON
使用 LIKE 搜索	LIKE '%term%'	FTS5 MATCH 'term'

---

13. 预实施清单

阶段 1: 编写代码之前

• [ ] 首先编写测试 - 为新的数据库操作创建失败测试
• [ ] 设计架构 - 记录表结构、约束、索引
• [ ] 安全审查 - 识别所有到达数据库的用户输入
• [ ] 设置性能目标 - 定义查询时间限制和批处理大小
• [ ] 阅读参考文件 - 如果处理用户输入，加载 references/security-examples.md

阶段 2: 实施期间

• [ ] 仅使用参数化查询 - 切勿将用户输入拼接到 SQL 中
• [ ] 白名单动态名称 - 列/表名仅来自批准列表
• [ ] 相关操作使用事务 - 将多步操作包装在事务中
• [ ] 启用外键 - 在连接时 PRAGMA foreign_keys = ON
• [ ] 配置 WAL 模式 - 用于并发读写访问
• [ ] 创建索引 - 在 WHERE、JOIN、ORDER BY 中使用的列上
• [ ] 使用批处理操作 - executemany() 用于多次插入
• [ ] 安全错误处理 - 用户面向的错误中无 SQL 细节

阶段 3: 提交之前

• [ ] 所有测试通过 - 运行 pytest tests/test_*_repository.py -v
• [ ] 存在 SQL 注入测试 - 验证恶意输入被安全处理
• [ ] 性能验证 - EXPLAIN QUERY PLAN 显示索引使用
• [ ] 迁移测试 - 回滚正常工作
• [ ] 更新架构版本 - 迁移跟踪到位
• [ ] 设置数据库权限 - 生产文件模式 600
• [ ] 记录备份策略 - 恢复程序已验证
• [ ] 安排 VACUUM - 数据库增长的维护计划

---

14. 总结

创建安全（参数化查询）、可靠（事务、外键）和高效（WAL 模式、索引、FTS5）的 SQLite 实现。

安全提醒: 切勿将用户输入拼接到 SQL 中。始终使用参数化查询。