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. まとめ

安全（パラメーター化クエリ）、信頼性（トランザクション、外部キー）、高パフォーマンス（WAL モード、インデックス、FTS5）を備えた SQLite 実装を作成してください。

セキュリティに関する注意：ユーザー入力を SQL に連結することは絶対にしないでください。必ずパラメーター化クエリを使用してください。