SQLite Database Expert

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. TDD 优先——先编写测试再实现；使用内存中的 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)?;

        // Enable security and performance features
        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 参数化查询（重要）

// CORRECT: Parameterized query
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)
}

// CORRECT: Named parameters for clarity
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)
}

// INCORRECT: SQL Injection vulnerability
pub fn get_user_unsafe(&self, user_id: &str) -> Result<Option<User>> {
    // NEVER DO THIS - SQL injection risk
    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()?;

    // Debit from source
    tx.execute(
        "UPDATE accounts SET balance = balance - ?1 WHERE id = ?2",
        [amount, from_id as f64],
    )?;

    // Credit to destination
    tx.execute(
        "UPDATE accounts SET balance = balance + ?1 WHERE id = ?2",
        [amount, to_id as f64],
    )?;

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

4.4 全文搜索（FTS5）

// Create FTS5 virtual table with triggers
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(())
}

// Search with highlighting
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 - 错误配置	中	安全的 PRAGMA、文件权限（600）

5.3 SQL 注入防护

关键规则（参见 references/security-examples.md）：

1. 绝不要使用字符串格式化来编写 SQL 查询
2. 始终使用 ? 位置参数或 :name 命名参数
3. 对动态查询中的列名和表名使用白名单

// Dynamic column selection - SAFE approach
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()); // Table still exists
    }
}

---

7. 实现工作流（TDD）

步骤 1：先编写失败的测试

# tests/test_user_repository.py
import pytest
import sqlite3

@pytest.fixture
def db():
    """In-memory SQLite for fast testing."""
    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("[email protected]", "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 模式

# Good: Enable WAL for concurrent read/write
conn.execute("PRAGMA journal_mode = WAL")
conn.execute("PRAGMA synchronous = NORMAL")
conn.execute("PRAGMA cache_size = -64000")  # 64MB

# Bad: Default DELETE mode blocks reads during writes

模式 2：批量插入

# Good: Single transaction for batch
conn.executemany("INSERT INTO items (name) VALUES (?)", records)
conn.commit()

# Bad: Commit per row (100x slower)
for r in records:
    conn.execute("INSERT INTO items (name) VALUES (?)", (r,))
    conn.commit()

模式 3：连接池

# Good: Reuse connections
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)

# Bad: New connection per query
conn = sqlite3.connect(db_path)  # Expensive!

模式 4：索引优化

# Good: Covering and partial indexes
conn.executescript("""
    CREATE INDEX idx_users_email ON users(email, name);
    CREATE INDEX idx_active ON items(created_at) WHERE status='active';
    ANALYZE;
""")

# Bad: Full table scan on unindexed columns

模式 5：安排 VACUUM

# Good: Maintenance during idle time
def nightly_maintenance(conn):
    conn.execute("PRAGMA optimize")
    freelist = conn.execute("PRAGMA freelist_count").fetchone()[0]
    if freelist > 1000:
        conn.execute("VACUUM")

# Bad: VACUUM during peak usage or never

---

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. 总结

创建 SQLite 实现，使其具备安全性（参数化查询）、可靠性（事务、外键）和高性能（WAL 模式、索引、FTS5）。

安全提示：绝不要将用户输入拼接到 SQL 中。始终使用参数化查询。