第165回では、稼働中のSQLiteから整合したバックアップを作り、SHA-256とSQLiteの検査結果をマニフェストへ残しました。ただし、バックアップを作成できたことと、必要な状態で復元できることは同じではありません。
今回は、本番データベースを上書きせず、新しい隔離用ディレクトリで復元訓練を行います。コピー後のファイルを再読して計算したSHA-256、SQLiteの整合性、外部キー、PRAGMA user_version、必須テーブル、列、行数範囲を確認します。
実装にはPython 3.10以上の標準ライブラリだけを使います。本番データベースとの切り替えや、アプリケーションからの書き込み試験は対象外です。
今回の復元訓練で確認すること
- 別に保管したSHA-256と、コピー後のファイルを再読して計算した値が一致する
- 復元先を新しく作成し、既存のファイルやディレクトリを上書きしない
integrity_check(1)がokを返すforeign_key_checkで違反が見つからないPRAGMA user_versionが復元ポリシーと一致する- 必要なテーブルと列が存在し、行数が許容範囲内にある
- すべての検査を通過した後で
report.jsonを確定する
| 項目 | 今回の契約 |
|---|---|
| 入力 | SQLiteのバックアップファイル、期待するSHA-256、復元ポリシーJSON |
| 復元先 | 実行前に存在しないディレクトリ |
| 生成物 | recovered.sqliteとreport.json |
| 完了条件 | report.jsonをJSONとして読み取れ、statusがpassedであること |
| 上書き | 既存の復元先は拒否する |
| 対象外 | 本番への切り替え、業務データの意味上の正しさ、添付データベース、外部ファイル、アプリケーション固有の検索結果 |
入力には、SQLiteのバックアップAPIなどで作成した、単独で開けるバックアップファイルを使います。稼働中のメインデータベースだけをファイルコピーすると、WALファイルに残っている変更を含められないことがあります。
復元ポリシーを用意する
次の内容をrecovery-policy.jsonとして保存します。この例は、後ほど作成する動作確認用データベースに対応しています。
{
"version": 1,
"sqlite_user_version": 1,
"tables": {
"review_items": {
"required_columns": [
"batch_id",
"record_id",
"status"
],
"min_rows": 2,
"max_rows": 2
}
}
}min_rowsとmax_rowsは、動作確認用データが2行であるため同じ値にしています。実際のバックアップでは、復旧時点として許容できる範囲を設定します。上限を設けない場合はmax_rowsをnullにします。
このポリシーは必須列を検査しますが、追加の列やテーブルは拒否しません。検査対象には通常列だけでなく、生成列と仮想テーブルの隠し列も含めます。そのため、列一覧の取得には、それらを省略するPRAGMA table_infoではなくPRAGMA table_xinfoを使います。列名は大文字と小文字を含めてポリシーと一致させます。PRAGMA user_versionはSQLiteが自動更新する値ではありません。スキーマ変更時に、データベースとポリシーを同じ値へ更新します。
復元訓練CLIを実装する
次のコードをrestore_drill.pyとして保存します。復元先ディレクトリを作成し、コピー後の候補ファイルを再読してSHA-256を計算してから、読み取り専用のSQLite接続で検査します。
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import sqlite3
import stat
import sys
from pathlib import Path
from typing import Any
CHUNK_SIZE = 1024 * 1024
class DrillError(Exception):
def __init__(self, exit_code: int, kind: str, message: str) -> None:
super().__init__(message)
self.exit_code = exit_code
self.kind = kind
self.message = message
def write_json_line(payload: dict[str, Any], stream: Any) -> None:
print(
json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
file=stream,
)
class JsonArgumentParser(argparse.ArgumentParser):
def error(self, message: str) -> None:
write_json_line(
{"status": "error", "kind": "arguments", "message": message},
sys.stderr,
)
raise SystemExit(2)
def reject_duplicate_keys(
pairs: list[tuple[str, Any]],
) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise ValueError(f"duplicate JSON key: {key}")
result[key] = value
return result
def reject_constant(value: str) -> None:
raise ValueError(f"non-standard JSON number: {value}")
def validate_name(value: Any, label: str) -> str:
if type(value) is not str:
raise DrillError(2, "policy", f"{label} must be a string")
if (
not value
or value != value.strip()
or any(ord(char) < 32 or ord(char) == 127 for char in value)
):
raise DrillError(
2,
"policy",
f"{label} contains an invalid character",
)
return value
def load_policy(path: Path) -> dict[str, Any]:
try:
with path.open("r", encoding="utf-8") as policy_file:
raw = json.load(
policy_file,
object_pairs_hook=reject_duplicate_keys,
parse_constant=reject_constant,
)
except (OSError, UnicodeError, ValueError) as exc:
raise DrillError(
2,
"policy",
f"cannot load policy: {exc}",
) from exc
required_top = {"version", "sqlite_user_version", "tables"}
if type(raw) is not dict or set(raw) != required_top:
raise DrillError(
2,
"policy",
"policy keys must be version, sqlite_user_version, and tables",
)
if type(raw["version"]) is not int or raw["version"] != 1:
raise DrillError(
2,
"policy",
"policy version must be integer 1",
)
user_version = raw["sqlite_user_version"]
if (
type(user_version) is not int
or not 0 <= user_version <= 2_147_483_647
):
raise DrillError(
2,
"policy",
"sqlite_user_version must be an integer from 0 to 2147483647",
)
tables = raw["tables"]
if type(tables) is not dict or not tables:
raise DrillError(
2,
"policy",
"tables must be a non-empty object",
)
required_spec = {"required_columns", "min_rows", "max_rows"}
for raw_table_name, spec in tables.items():
table_name = validate_name(raw_table_name, "table name")
if table_name.lower().startswith("sqlite_"):
raise DrillError(
2,
"policy",
"internal SQLite tables are not allowed",
)
if type(spec) is not dict or set(spec) != required_spec:
raise DrillError(
2,
"policy",
f"{table_name} must define required_columns, min_rows, and max_rows",
)
columns = spec["required_columns"]
if type(columns) is not list or not columns:
raise DrillError(
2,
"policy",
f"{table_name}.required_columns must be a non-empty array",
)
checked_columns = [
validate_name(column, f"{table_name} column")
for column in columns
]
if len(checked_columns) != len(set(checked_columns)):
raise DrillError(
2,
"policy",
f"{table_name}.required_columns contains a duplicate",
)
min_rows = spec["min_rows"]
max_rows = spec["max_rows"]
if type(min_rows) is not int or min_rows < 0:
raise DrillError(
2,
"policy",
f"{table_name}.min_rows must be a non-negative integer",
)
if max_rows is not None and (
type(max_rows) is not int or max_rows < min_rows
):
raise DrillError(
2,
"policy",
f"{table_name}.max_rows must be null or at least min_rows",
)
return raw
def parse_expected_sha256(value: str) -> str:
if re.fullmatch(r"[0-9a-fA-F]{64}", value) is None:
raise DrillError(
2,
"arguments",
"expected SHA-256 must contain exactly 64 hexadecimal characters",
)
return value.lower()
def copy_and_hash_destination(
source: Path,
destination: Path,
) -> tuple[str, int]:
try:
with source.open("rb") as source_file:
source_mode = os.fstat(source_file.fileno()).st_mode
if not stat.S_ISREG(source_mode):
raise DrillError(
2,
"input",
"backup must be a regular file",
)
with destination.open("xb") as destination_file:
while True:
chunk = source_file.read(CHUNK_SIZE)
if not chunk:
break
written = destination_file.write(chunk)
if written != len(chunk):
raise OSError("short write while copying backup")
destination_file.flush()
os.fsync(destination_file.fileno())
except DrillError:
raise
except OSError as exc:
raise DrillError(
5,
"io",
f"backup copy failed: {exc}",
) from exc
digest = hashlib.sha256()
total = 0
try:
with destination.open("rb") as copied_file:
while True:
chunk = copied_file.read(CHUNK_SIZE)
if not chunk:
break
digest.update(chunk)
total += len(chunk)
except OSError as exc:
raise DrillError(
5,
"io",
f"cannot read copied backup: {exc}",
) from exc
if total == 0:
raise DrillError(4, "sqlite_check", "backup file is empty")
return digest.hexdigest(), total
def quote_identifier(name: str) -> str:
return '"' + name.replace('"', '""') + '"'
def validate_database(
path: Path,
policy: dict[str, Any],
) -> dict[str, Any]:
uri = f"{path.resolve().as_uri()}?mode=ro&immutable=1"
try:
connection = sqlite3.connect(uri, uri=True)
try:
connection.execute("PRAGMA foreign_keys = ON")
connection.execute("PRAGMA query_only = ON")
integrity_rows = connection.execute(
"PRAGMA integrity_check(1)"
).fetchall()
if integrity_rows != [("ok",)]:
raise DrillError(
4,
"sqlite_check",
"integrity_check did not return ok",
)
foreign_key_row = connection.execute(
"PRAGMA foreign_key_check"
).fetchone()
if foreign_key_row is not None:
raise DrillError(
4,
"sqlite_check",
"foreign_key_check found a violation",
)
actual_user_version = connection.execute(
"PRAGMA user_version"
).fetchone()[0]
if actual_user_version != policy["sqlite_user_version"]:
raise DrillError(
4,
"sqlite_check",
"SQLite user_version does not match the policy",
)
actual_tables = {
row[0]
for row in connection.execute(
"SELECT name FROM sqlite_schema WHERE type = 'table'"
)
}
table_results: dict[str, Any] = {}
for table_name, spec in policy["tables"].items():
if table_name not in actual_tables:
raise DrillError(
4,
"sqlite_check",
f"required table is missing: {table_name}",
)
table_sql = quote_identifier(table_name)
columns = [
row[1]
for row in connection.execute(
f"PRAGMA table_xinfo({table_sql})"
)
]
missing_columns = [
column
for column in spec["required_columns"]
if column not in columns
]
if missing_columns:
raise DrillError(
4,
"sqlite_check",
f"required column is missing from {table_name}",
)
row_count = connection.execute(
f"SELECT COUNT(*) FROM {table_sql}"
).fetchone()[0]
if row_count < spec["min_rows"]:
raise DrillError(
4,
"sqlite_check",
f"row count is below the minimum for {table_name}",
)
max_rows = spec["max_rows"]
if max_rows is not None and row_count > max_rows:
raise DrillError(
4,
"sqlite_check",
f"row count is above the maximum for {table_name}",
)
table_results[table_name] = {
"row_count": row_count,
"required_columns_checked": len(
spec["required_columns"]
),
}
finally:
connection.close()
except DrillError:
raise
except sqlite3.Error as exc:
raise DrillError(
4,
"sqlite_check",
f"SQLite validation failed: {exc}",
) from exc
return {
"integrity_check": "ok",
"foreign_key_violations": 0,
"sqlite_user_version": actual_user_version,
"tables": table_results,
}
def write_report_candidate(path: Path, payload: dict[str, Any]) -> None:
try:
with path.open("x", encoding="utf-8") as report_file:
json.dump(payload, report_file, ensure_ascii=False, indent=2)
print(file=report_file)
report_file.flush()
os.fsync(report_file.fileno())
except OSError as exc:
raise DrillError(
5,
"io",
f"cannot write report: {exc}",
) from exc
def rename_new(source: Path, target: Path) -> None:
if os.path.lexists(target):
raise DrillError(
5,
"output",
f"internal destination already exists: {target}",
)
try:
source.rename(target)
except OSError as exc:
raise DrillError(
5,
"io",
f"cannot finalize {target}: {exc}",
) from exc
def cleanup_created_output(output_dir: Path, paths: list[Path]) -> None:
for path in paths:
try:
path.unlink()
except FileNotFoundError:
pass
except OSError:
pass
try:
output_dir.rmdir()
except OSError:
pass
def build_parser() -> argparse.ArgumentParser:
parser = JsonArgumentParser(
description="Restore a SQLite backup into an isolated directory."
)
parser.add_argument("--backup", required=True, type=Path)
parser.add_argument("--expected-sha256", required=True)
parser.add_argument("--policy", required=True, type=Path)
parser.add_argument("--output-dir", required=True, type=Path)
return parser
def run(args: argparse.Namespace) -> dict[str, Any]:
expected_sha256 = parse_expected_sha256(args.expected_sha256)
policy = load_policy(args.policy)
if not args.backup.is_file():
raise DrillError(
2,
"input",
"backup does not exist or is not a regular file",
)
if args.output_dir.exists():
raise DrillError(2, "output", "output path already exists")
if not args.output_dir.parent.is_dir():
raise DrillError(
2,
"output",
"output parent does not exist or is not a directory",
)
try:
args.output_dir.mkdir(mode=0o700)
except FileExistsError as exc:
raise DrillError(
2,
"output",
"output path already exists",
) from exc
except OSError as exc:
raise DrillError(
5,
"io",
f"cannot create output directory: {exc}",
) from exc
candidate = args.output_dir / ".recovered.sqlite.part"
recovered = args.output_dir / "recovered.sqlite"
report_candidate = args.output_dir / ".report.json.part"
report_path = args.output_dir / "report.json"
known_paths = [report_path, report_candidate, recovered, candidate]
success = False
try:
actual_sha256, size_bytes = copy_and_hash_destination(
args.backup,
candidate,
)
if actual_sha256 != expected_sha256:
raise DrillError(
3,
"hash_mismatch",
"copied file does not match the expected SHA-256",
)
checks = validate_database(candidate, policy)
report = {
"status": "passed",
"backup_name": args.backup.name,
"output_dir": str(args.output_dir),
"database": str(recovered),
"report": str(report_path),
"sha256": actual_sha256,
"size_bytes": size_bytes,
"checks": checks,
}
rename_new(candidate, recovered)
write_report_candidate(report_candidate, report)
rename_new(report_candidate, report_path)
success = True
return report
finally:
if not success:
cleanup_created_output(args.output_dir, known_paths)
def main() -> int:
args = build_parser().parse_args()
try:
report = run(args)
except DrillError as exc:
write_json_line(
{
"status": "error",
"kind": exc.kind,
"message": exc.message,
},
sys.stderr,
)
return exc.exit_code
except KeyboardInterrupt:
write_json_line(
{
"status": "error",
"kind": "interrupted",
"message": "interrupted by user",
},
sys.stderr,
)
return 130
except Exception:
write_json_line(
{
"status": "error",
"kind": "unexpected",
"message": "unexpected execution error",
},
sys.stderr,
)
return 5
write_json_line(report, sys.stdout)
return 0
if __name__ == "__main__":
raise SystemExit(main())immutable=1は、コピー後の候補ファイルを検査中に変更しないことを前提に使います。SQLiteはこの指定があるファイルのロックと変更検出を省略します。書き込みが続いている本番データベースには使えません。
report.jsonは最後に候補名から確定名へ変更します。途中で停止してrecovered.sqliteだけが残っても、完了レポートがなければ成功として扱いません。
動作確認用のSQLiteを作る
次のコードをmake_demo_backup.pyとして保存します。動作確認専用のSQLiteファイルを新規作成し、サンプルデータを2行登録します。
from __future__ import annotations
import hashlib
import json
import sqlite3
from pathlib import Path
DB_PATH = Path("queue-backup.sqlite")
def file_sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as source:
while True:
chunk = source.read(1024 * 1024)
if not chunk:
break
digest.update(chunk)
return digest.hexdigest()
def main() -> None:
try:
DB_PATH.touch(exist_ok=False)
except FileExistsError as exc:
raise SystemExit("queue-backup.sqlite already exists") from exc
try:
with sqlite3.connect(DB_PATH) as connection:
connection.execute("PRAGMA foreign_keys = ON")
connection.execute("PRAGMA user_version = 1")
connection.execute(
"""
CREATE TABLE review_items (
batch_id TEXT NOT NULL,
record_id TEXT NOT NULL,
status TEXT NOT NULL CHECK (
status IN (
'pending',
'accepted',
'corrected',
'rejected'
)
),
PRIMARY KEY (batch_id, record_id)
)
"""
)
connection.executemany(
"""
INSERT INTO review_items (
batch_id,
record_id,
status
) VALUES (?, ?, ?)
""",
[
("demo-batch", "R-001", "pending"),
("demo-batch", "R-002", "accepted"),
],
)
except BaseException:
DB_PATH.unlink(missing_ok=True)
raise
print(
json.dumps(
{
"backup": str(DB_PATH),
"sha256": file_sha256(DB_PATH),
},
ensure_ascii=False,
separators=(",", ":"),
)
)
if __name__ == "__main__":
main()順番に実行する
以下はBashでの実行例です。最初に動作確認用SQLiteを作り、そのSHA-256をJSONへ保存します。
python make_demo_backup.py > demo-result.json
python -m json.tool demo-result.json続いてJSONからSHA-256を取り出し、復元訓練を実行します。
EXPECTED_SHA256=$(python -c 'import json,pathlib; data=json.loads(pathlib.Path("demo-result.json").read_text(encoding="utf-8")); print(data["sha256"])')
python restore_drill.py --backup queue-backup.sqlite --expected-sha256 "$EXPECTED_SHA256" --policy recovery-policy.json --output-dir drill-20261007成功時の終了コードは0です。標準出力には、実際のハッシュ、サイズ、検査結果を含む1行のJSONが出ます。復元先には次の2ファイルが残ります。
drill-20261007/
├── recovered.sqlite
└── report.jsonレポートを読み直します。
python -m json.tool drill-20261007/report.jsonstatusがpassedであることに加え、sqlite_user_version、テーブルごとのrow_count、検査した必須列数を確認します。
失敗条件と終了コードを確認する
| 終了コード | 条件 | 後続処理 |
|---|---|---|
| 0 | すべての検査に合格 | 隔離環境での読み取り試験へ進める |
| 2 | 引数、ポリシー、入力パス、復元先の問題 | 設定を修正する |
| 3 | コピー後のファイルのSHA-256が期待値と不一致 | バックアップとハッシュの取得元を確認する |
| 4 | SQLite、外部キー、スキーマ、行数の検査に不合格 | 復元物を使用しない |
| 5 | ファイル操作または予期しない実行エラー | 空き容量、権限、パス、実行環境を確認する |
| 130 | Ctrl+Cによる中断 | 残存ディレクトリを確認してから再実行する |
動作確認では、次の条件を個別に試せます。試すたびに新しい復元先を指定します。
- 表示されたSHA-256を1文字変更すると終了コード3になる
- ポリシーの
min_rowsとmax_rowsをともに3へ変更すると終了コード4になる - 成功済みの
drill-20261007を再指定すると終了コード2になる - 存在しない列を
required_columnsへ追加すると終了コード4になる
スクリプトが捕捉した失敗では、自身が作成した既知のファイルと、空になった復元先ディレクトリの削除を試みます。強制終了、停電、ファイルシステム障害、削除権限の問題がある場合は途中生成物が残り得ます。既存の復元先は再利用せず、内容を確認してから別の名前でやり直します。
実データへ適用するときの注意点
- 期待するSHA-256は、バックアップ本体とは別に保護したマニフェストから取得する
- バックアップと期待値を同時に書き換えられる環境では、SHA-256だけを改ざん対策にしない
- 行数範囲は現在の本番件数ではなく、対象バックアップの復旧時点に対して決める
COUNT(*)を使う行数検査は、大きなテーブルでの実行時間を事前に測る- 復元先にはバックアップと同程度以上の空き容量を確保する
mode=0o700だけに依存せず、OSとファイルシステムの権限やACLも確認する- 同じOS利用者による競合操作や、突然の電源断に対するファイルシステム上の完全性までは保証しない
integrity_checkはSQLiteの構造を検査しますが、外部キー違反は検査しません。そのため、foreign_key_checkを別に実行しています。どちらも、レビュー内容の正しさや、必要な時点までデータが含まれていることまでは保証しません。
次の実務作業は、合格したrecovered.sqliteを隔離したアプリケーションへ読み取り専用で接続し、担当者が使う一覧表示や件数集計を確認することです。本番ファイルとの切り替えは、サービス停止、権限、切り戻し条件を含む別の手順として管理します。