sqlite-db-truncate

SQLite 截断数据库恢复

本技能介绍了从因二进制截断而损坏的 SQLite 数据库文件中恢复数据的系统方法。它强调在尝试恢复前先了解 SQLite 文件格式，并避免导致多次失败尝试的常见陷阱。

何时使用此技能

适用于以下情况：

• 无法使用标准 sqlite3 命令打开 SQLite 数据库文件
• 数据库错误提示文件损坏或数据格式错误
• 文件大小小于预期（表明可能发生了截断）
• .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]
...

Varint 编码

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

逐步解析并输出调试信息

先完整解析一个单元并验证，再处理所有单元：

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 检查开头几个字节。如果文件以 0x0d 而不是 "SQLite format 3" 开头，则表示文件头缺失。相应调整解析偏移量（无需增加 100 字节的文件头偏移量）。

陷阱 2：反复创建多个脚本

错误做法：通过反复试错创建许多独立的恢复脚本（recover1.py、recover2.py 等）。

预防方法：

• 先仔细阅读十六进制转储，并手动标注结构
• 构建一个带有调试标记的脚本
• 编写代码前先查阅 SQLite 文件格式规范

陷阱 3：读取字符串时超出其边界

错误：读取字符串数据时未检查序列类型长度，导致字符串不正确（例如，得到 "testword052" 而不是 "testword05"）。

预防：始终根据序列类型计算字符串长度：length = (serial_type - 13) // 2。只读取这么多字节。

陷阱 4：生成的代码存在语法错误

错误：运算符中缺少空格，例如将 if48 写成 if 48，或将 12and 写成 12 and。

预防：运行前先验证语法：

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 语法
• 所有多字节整数均使用大端序
• 根据序列类型计算字符串长度
• 通过边界检查处理截断

恢复后：

• 验证单元格数量符合预期
• 如已知字符串模式，则验证其是否符合
• 检查数值范围
• 确认输出格式符合要求