sqlite-db-truncate

SQLite の切り詰められたデータベースの復元

このスキルでは、バイナリの切り詰めによって破損した SQLite データベースファイルからデータを復元するための体系的な方法を紹介します。復元を試みる前に SQLite のファイル形式を理解することを重視し、何度も失敗を繰り返す原因となるよくある落とし穴を避ける方法も説明します。

このスキルを使用する場面

次のような場合に、このスキルが役立ちます：

• SQLite データベースファイルを標準の sqlite3 コマンドで開けない
• データベースのエラーに、破損または不正なデータが示されている
• ファイルサイズが想定より小さい（切り詰められている可能性がある）
• .recover コマンドなどの標準的な復元ツールが失敗する
• SQLite のページ構造を手作業でバイナリ解析する必要がある

初期評価の手順

復元コードを書く前に、破損したファイルを十分に分析してください。

手順 1：ファイルの特性を調べる

# Check file size and basic properties
ls -lh database.db
file database.db

# Create hex dump for analysis
hexdump -C database.db | head -100

確認すべき主な点：

• ファイルサイズ：SQLite のページサイズは通常 4096 バイトです。サイズがページ境界と一致するか確認してください
• マジックバイト：有効な SQLite ファイルは "SQLite format 3\000"（16 バイト）で始まります
• ヘッダーの次の最初のバイト：ページの種類を示します（0x0d = 実際のデータを含むテーブルリーフページ）

手順 2：破損のパターンを特定する

よくある切り詰めのパターン：

• ヘッダーのみのファイル：100 バイトのヘッダーだけが残っている
• ヘッダーの欠落：ファイルがデータページから始まっている（最初のバイトが 0x0d、0x05、0x0a、または 0x02）
• ページの一部のみ：ファイルがページの途中で終わり、セルの一部が切り詰められている

ファイルに標準の "SQLite format 3" マジックヘッダーがなく、0x0d から始まっている場合、データベースヘッダーのないテーブルリーフページだけが含まれていることを示しています。

手順 3：まず標準ツールを試す

手動解析を行う前に、必ず標準的な復元を試してください。

# Check if sqlite3 can read the file
sqlite3 database.db ".schema" 2>&1
sqlite3 database.db "SELECT * FROM sqlite_master" 2>&1

# Try built-in recovery
sqlite3 database.db ".recover" > recovered.sql 2>&1

# Try integrity check
sqlite3 database.db "PRAGMA integrity_check;"

これらの方法が "database disk image is malformed" などのエラーで失敗した場合は、手動のバイナリ解析に進んでください。

SQLite のページ構造の概要

復元コードを書く前に、ページ構造を理解しておくことが重要です。

テーブルリーフページのレイアウト（ページタイプ 0x0d）

Offset  Size   Description
------  ----   -----------
0       1      Page type (0x0d for table leaf)
1       2      First freeblock offset (big-endian)
3       2      Number of cells on page (big-endian)
5       2      Cell content area start offset (big-endian)
7       1      Fragmented free bytes count
8+      varies Cell pointer array (2 bytes per cell, big-endian)
...            [Gap/free space]
End            Cell data (grows backward from page end)

セル構造

各セルにはデータベースの行が含まれています。

[Payload size: varint]
[Row ID: varint]
[Header size: varint]
[Serial type 1: varint]
[Serial type 2: varint]
...
[Column 1 value]
[Column 2 value]
...

可変長整数のエンコード

SQLite では、可変長整数（varint）が使用されます。

• バイト 1～8：データに 7 ビットを使用し、最上位ビット（0x80）は継続を示します
• バイト 9：8 ビットすべてを使用します（継続はありません）

シリアルタイプ

シリアルタイプは、列データの解釈方法を示します。

タイプ	サイズ	意味
0	0	NULL
1	1	8 ビット符号付き整数
2	2	16 ビットビッグエンディアン符号付き整数
3	3	24 ビットビッグエンディアン符号付き整数
4	4	32 ビットビッグエンディアン符号付き整数
7	8	IEEE 754 64 ビット浮動小数点数（ビッグエンディアン）
8	0	整数定数 0
9	0	整数定数 1
N >= 12、偶数	(N-12)/2	BLOB
N >= 13、奇数	(N-13)/2	テキスト文字列（UTF-8）

例：シリアルタイプ 0x21（33）は、長さが (33-13)/2 = 10 バイトのテキスト文字列です。

復元方法

単一のモジュール式スクリプトを作成する

復元スクリプトを複数に分けて作成するのは避けてください。代わりに、わかりやすいデバッグ出力を含む単一のスクリプトを段階的に構築してください。

import struct
import json

DEBUG = True

def read_varint(data, offset):
    """Read SQLite variable-length integer."""
    value = 0
    for i in range(9):
        if offset + i >= len(data):
            return None, offset
        byte = data[offset + i]
        if i == 8:
            value = (value << 8) | byte
            return value, offset + i + 1
        value = (value << 7) | (byte & 0x7f)
        if (byte & 0x80) == 0:
            return value, offset + i + 1
    return value, offset

def decode_value(data, offset, serial_type):
    """Decode value based on serial type."""
    if serial_type == 0:
        return None, offset
    elif serial_type == 1:
        return struct.unpack('>b', data[offset:offset+1])[0], offset + 1
    elif serial_type == 2:
        return struct.unpack('>h', data[offset:offset+2])[0], offset + 2
    elif serial_type == 4:
        return struct.unpack('>i', data[offset:offset+4])[0], offset + 4
    elif serial_type == 7:
        return struct.unpack('>d', data[offset:offset+8])[0], offset + 8
    elif serial_type == 8:
        return 0, offset
    elif serial_type == 9:
        return 1, offset
    elif serial_type >= 12:
        if serial_type % 2 == 0:
            length = (serial_type - 12) // 2
            return data[offset:offset+length], offset + length
        else:
            length = (serial_type - 13) // 2
            return data[offset:offset+length].decode('utf-8', errors='replace'), offset + length
    return None, offset

デバッグ出力を使って段階的に解析する

すべてのセルを処理する前に、まず 1 つのセルを最後まで解析して検証してください。

def parse_cell(data, cell_offset, debug=DEBUG):
    """Parse a single cell with detailed debug output."""
    if debug:
        print(f"\nParsing cell at offset {cell_offset} (0x{cell_offset:04x})")

    # Read payload size
    payload_size, offset = read_varint(data, cell_offset)
    if debug:
        print(f"  Payload size: {payload_size}")

    # Read row ID
    row_id, offset = read_varint(data, offset)
    if debug:
        print(f"  Row ID: {row_id}")

    # Read header size
    header_size, header_start = read_varint(data, offset)
    if debug:
        print(f"  Header size: {header_size}")

    # Parse serial types
    serial_types = []
    current = header_start
    header_end = offset + header_size
    while current < header_end:
        st, current = read_varint(data, current)
        serial_types.append(st)

    if debug:
        print(f"  Serial types: {serial_types}")

    # Parse values
    values = []
    for st in serial_types:
        val, current = decode_value(data, current, st)
        values.append(val)

    if debug:
        print(f"  Values: {values}")

    return {'row_id': row_id, 'values': values}

よくある落とし穴とその回避方法

落とし穴 1：破損のパターンを理解していない

誤り：ファイルにデータページしか含まれていない可能性があるのに、標準の SQLite ヘッダーがあると思い込むこと。

回避方法：必ず hexdump で最初の数バイトを確認してください。ファイルが "SQLite format 3" ではなく 0x0d から始まっている場合、ヘッダーはありません。それに応じて解析オフセットを調整してください（100 バイトのヘッダー分のオフセットは不要です）。

落とし穴 2：スクリプトを何度も作り直す

誤り：試行錯誤を繰り返し、複数の復元スクリプト（recover1.py、recover2.py など）を別々に作成すること。

回避方法：

• まず 16 進ダンプを十分に読み、構造に手作業で注釈を付けてください
• デバッグフラグを備えた単一のスクリプトを作成してください
• コードを書く前に SQLite のファイル形式の仕様を参照してください

落とし穴 3：文字列の境界を超えて読み取る

間違い：シリアル型の長さを確認せずに文字列データを読み取り、誤った文字列（例：「testword05」ではなく「testword052」）になることです。

防止策：シリアル型から文字列の長さを必ず計算します：length = (serial_type - 13) // 2。そのバイト数だけを正確に読み取ります。

落とし穴 4：生成コードの構文エラー

間違い：if 48 ではなく if48、12 and ではなく 12and のように、演算子の前後にスペースがないことです。

防止策：実行前に構文を検証します：

python3 -m py_compile recovery_script.py

落とし穴 5：バイトオーダーの誤り

間違い：複数バイトの整数をビッグエンディアンではなく、リトルエンディアンで読み取ることです。

防止策：SQLite では、すべての複数バイト整数にビッグエンディアンを使用します。必ず > プレフィックスを付けて struct.unpack('>...', data) を使用します。

落とし穴 6：切り詰めを適切に処理しない

間違い：ファイル末尾で切り詰められたデータに遭遇すると、スクリプトがクラッシュすることです。

防止策：読み取り操作の前に毎回、境界を確認します：

def safe_read(data, offset, length):
    if offset + length > len(data):
        return None
    return data[offset:offset+length]

検証戦略

ステップ 1：セル数を検証する

ページヘッダー（オフセット 3-4）に記録されたセル数と、実際に見つかったセル数を比較します。

ステップ 2：データパターンを検証する

想定されるパターン（例：「testwordXY」に一致する単語）がわかっている場合、抽出した文字列がそのパターンに一致することを確認します。

ステップ 3：値の範囲を確認する

抽出した数値が想定される範囲内にあることを確認します。次の点に注意してください：

• 想定外の負の数（符号ビットの解釈）
• 非常に大きな数（バイトオーダーの問題）
• 浮動小数点数の NaN または無限大

ステップ 4：想定される出力形式と比較する

出力を確定する前に、JSON 構造が要件に一致することを確認します：

# Validate output structure
for record in recovered_data:
    assert 'word' in record and 'value' in record
    assert isinstance(record['word'], str)
    assert isinstance(record['value'], (int, float))

出力の生成

復元したデータを、必要な出力仕様に従ってフォーマットします：

def generate_output(recovered_rows, output_path):
    """Format and save recovered data."""
    results = []
    for row in recovered_rows:
        if len(row['values']) >= 2:
            results.append({
                'word': row['values'][0],
                'value': row['values'][1]
            })

    with open(output_path, 'w') as f:
        json.dump(results, f, indent=2)

    print(f"Recovered {len(results)} records to {output_path}")
    return results

要約チェックリスト

復元コードを書く前：

• hexdump でファイルを調べ、破損の範囲を把握した
• ヘッダーが存在するか、欠落しているかを特定した
• まず標準の SQLite ツールを試した
• SQLite ファイル形式の仕様を確認した

実装中：

• デバッグ出力を含む単一のスクリプトを使用している（複数のスクリプトではない）
• 実行前に Python の構文を検証した
• すべての複数バイト整数にビッグエンディアンを使用している
• シリアル型から文字列の長さを計算している
• 境界チェックによって切り詰めを処理している

復元後：

• セル数が想定と一致することを確認した
• 既知の文字列パターンを検証した
• 数値の範囲を確認した
• 出力形式が要件に一致することを確認した