"""記事C 演習スクリプトの自己点検（改訂版 rev2） drill.py の安全性と照合の検出力を、一時的な演習領域（work_selfcheck_<時刻>/）と架空データだけで確かめる。 正常な演習領域（work/）には触れない。故意にデータやファイルを壊す操作は、この一時領域のコピーにだけ行う。 確認項目（GPTレビューの受入条件に対応）: R1-A 正常な復元は内容・関係の照合まで成功し、終了コード 0 R1-B 件名だけ／履歴の操作者だけ／時刻だけ／備考だけを変えても検知する（件数・状態は同じ） R1-C 件数を変える／状態を変える場合もレポート不合格・終了コード非ゼロ R2-A 指紋不一致のバックアップで復元を試みても、既存の復元物が消えない R2-B 障害後に backup を再実行しても、既存の良いバックアップと基準が失われない R2-C 演習領域の外を対象にする操作（cleanup に外部パス）は拒否される。reset は --confirm と目印が要る R2-E〜H（rev2）無関係な既存フォルダを演習領域にしない／掃除スクリプトの単独実行入口なし／ practice.db が領域外へのリンクなら after-backup が更新しない／リンク先の無い復元先リンクを通って外に書かない R2-I（rev2）空の旧形式の目印は削除許可にならない。基点の直下でない領域は拒否 使い方: python3 drill_selfcheck.py 終了コード: すべて期待どおりなら 0、そうでなければ 1 """ import hashlib import json import os import shutil import sqlite3 import subprocess import sys from datetime import datetime, timezone import drill_safety as safety HERE = os.path.dirname(os.path.abspath(__file__)) STAMP = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") WORK = os.path.join(HERE, f"work_selfcheck_{STAMP}") RESULTS = [] def sha256(path): h = hashlib.sha256() with open(path, "rb") as f: h.update(f.read()) return h.hexdigest() def drill(*args, expect=0): env = dict(os.environ, DRILL_WORK_DIR=WORK) p = subprocess.run([sys.executable, os.path.join(HERE, "drill.py"), *args], capture_output=True, text=True, env=env) return p.returncode, p.stdout + p.stderr def record(item, expected, actual, ok): RESULTS.append((item, expected, actual, ok)) print(f"[{'OK ' if ok else 'NG '}] {item}: 期待={expected} 実際={actual}", flush=True) def normal_run(): for c in [["reset", "--confirm"], ["setup"], ["backup"], ["after-backup"], ["break"], ["freeze"], ["rollback-code"], ["restore"]]: code, out = drill(*c) if code != 0: print(out) raise SystemExit(f"正常手順 {c} が失敗した（終了コード {code}）") def restored_db(): return os.path.join(WORK, "restore", "gen-001", "practice_restored.db") def with_modified_restore(label, sql, params=()): """復元結果のコピーを保存 → 復元結果だけを変更 → verify → 元に戻す。終了コードと判定を返す。""" db = restored_db() keep = db + ".keep" shutil.copy2(db, keep) conn = sqlite3.connect(db) with conn: conn.execute(sql, params) conn.close() code, out = drill("verify") shutil.copy2(keep, db) os.remove(keep) detected = ("不一致あり" in out) return code, detected, out def main(): print(f"一時演習領域: {WORK}") normal_run() # R1-A code, out = drill("verify") record("R1-A 正常復元の照合", "すべて一致・終了コード0", f"{'すべて一致' if 'すべて一致' in out else '不一致あり'}・終了コード{code}", code == 0 and "すべて一致" in out) # R1-B 内容だけの変更（件数・状態は不変） cases = [ ("件名だけ変更", "UPDATE requests SET title='SELFCHECK_TITLE' WHERE id='REQ-C01'"), ("履歴の操作者だけ変更", "UPDATE history SET actor_id='selfcheck-actor' WHERE request_id='REQ-C01' AND action='承認'"), ("履歴の時刻だけ変更", "UPDATE history SET at='invalid-time' WHERE request_id='REQ-C01' AND action='承認'"), ("履歴の備考だけ変更", "UPDATE history SET note='changed-note' WHERE request_id='REQ-C03' AND action='差戻し'"), ] for label, sql in cases: code, detected, _ = with_modified_restore(label, sql) record(f"R1-B {label}", "不一致あり・終了コード1", f"{'不一致あり' if detected else 'すべて一致'}・終了コード{code}", detected and code == 1) # R1-C 件数・状態の変更 cases = [ ("履歴を1件削除（件数変更）", "DELETE FROM history WHERE request_id='REQ-C04' AND action='承認'"), ("状態だけ変更", "UPDATE requests SET state='提出済み' WHERE id='REQ-C01'"), ] for label, sql in cases: code, detected, _ = with_modified_restore(label, sql) record(f"R1-C {label}", "不一致あり・終了コード1", f"{'不一致あり' if detected else 'すべて一致'}・終了コード{code}", detected and code == 1) # 復元結果が元に戻っていることを確認 code, out = drill("verify") record("復元結果を元に戻した後の再照合", "すべて一致・終了コード0", f"終了コード{code}", code == 0 and "すべて一致" in out) # R2-A 指紋不一致のバックアップで再復元 → 既存の復元物が無傷 db = restored_db() before_hash = sha256(db) backup_db = os.path.join(WORK, "backup", "gen-001", "practice_backup.db") keep = backup_db + ".keep" shutil.copy2(backup_db, keep) with open(backup_db, "ab") as f: f.write(b"SELFCHECK-CORRUPTION") code, out = drill("restore", "1") shutil.copy2(keep, backup_db) os.remove(keep) after_exists = os.path.exists(db) after_hash = sha256(db) if after_exists else None record("R2-A 指紋不一致で再復元", "中止（非ゼロ）・既存復元物は無傷", f"終了コード{code}・復元物{'あり' if after_exists else 'なし'}・指紋{'同一' if after_hash == before_hash else '変化'}", code != 0 and after_exists and after_hash == before_hash) # 復元先が既にある状態での正常バックアップからの再復元も拒否されること code, out = drill("restore", "1") record("R2-A 復元先が既にある状態で再復元", "拒否（非ゼロ）・既存復元物は無傷", f"終了コード{code}・指紋{'同一' if sha256(db) == before_hash else '変化'}", code != 0 and sha256(db) == before_hash) # R2-B 障害後に backup を再実行 → gen-002 ができ、gen-001 と基準が無傷 m1 = os.path.join(WORK, "backup", "gen-001", "manifest.json") m1_before = open(m1, encoding="utf-8").read() b1_before = sha256(backup_db) code, out = drill("backup") gens = sorted(os.listdir(os.path.join(WORK, "backup"))) m1_after = open(m1, encoding="utf-8").read() base2 = json.load(open(os.path.join(WORK, "backup", "gen-002", "manifest.json"), encoding="utf-8"))["baseline"] if "gen-002" in gens else {} record("R2-B 障害後の backup 再実行", "gen-002 が増え、gen-001 と控えは不変", f"世代={gens}・gen-001控え{'不変' if m1_after == m1_before else '変化'}・gen-001指紋{'不変' if sha256(backup_db) == b1_before else '変化'}・gen-002控え=依頼{base2.get('requests_count')}/履歴{base2.get('history_count')}", code == 0 and gens == ["gen-001", "gen-002"] and m1_after == m1_before and sha256(backup_db) == b1_before) # 世代が複数あるとき、世代を省略した restore/verify は拒否される code, out = drill("verify") record("R2-B 世代複数時の verify（世代省略）", "拒否（非ゼロ）", f"終了コード{code}", code != 0) code, out = drill("verify", "1") record("R2-B 世代を明示した verify gen-001", "すべて一致・終了コード0", f"終了コード{code}", code == 0 and "すべて一致" in out) # R2-C 演習領域の外への操作を拒否 outside = os.path.join(HERE, f"selfcheck_outside_{STAMP}.db") # 演習領域の外だが、このフォルダ内 conn = sqlite3.connect(outside) conn.execute("CREATE TABLE history (request_id TEXT)"); conn.execute("CREATE TABLE requests (id TEXT, state TEXT)") conn.execute("INSERT INTO history VALUES ('x')"); conn.commit(); conn.close() p = subprocess.run([sys.executable, os.path.join(HERE, "cleanup_v1_1.py"), outside, WORK], capture_output=True, text=True) conn = sqlite3.connect(outside); n = conn.execute("SELECT COUNT(*) FROM history").fetchone()[0]; conn.close() record("R2-C cleanup_v1_1 に演習領域の外の DB を渡す（単独実行）", "拒否（非ゼロ）・削除されない", f"終了コード{p.returncode}・履歴{n}件", p.returncode != 0 and n == 1) # R2-F run() を直接呼んでも、演習領域の practice.db 以外は拒否される import cleanup_v1_1 h0 = sha256(outside) try: cleanup_v1_1.run(outside); refused = False except safety.SafetyError: refused = True record("R2-F cleanup_v1_1.run() に演習領域の外の DB を渡す", "拒否（SafetyError）・内容不変", f"{'拒否' if refused else '実行された'}・指紋{'同一' if sha256(outside)==h0 else '変化'}", refused and sha256(outside) == h0) os.remove(outside) code, out = drill("reset") record("R2-C reset を --confirm なしで実行", "拒否（非ゼロ）・領域は残る", f"終了コード{code}・practice.db{'あり' if os.path.exists(os.path.join(WORK,'practice.db')) else 'なし'}", code != 0 and os.path.exists(os.path.join(WORK, "practice.db"))) marker = os.path.join(WORK, ".drill-work") os.remove(marker) code, out = drill("reset", "--confirm") record("R2-C 目印ファイルのない領域を reset --confirm", "拒否（非ゼロ）・領域は残る", f"終了コード{code}・practice.db{'あり' if os.path.exists(os.path.join(WORK,'practice.db')) else 'なし'}", code != 0 and os.path.exists(os.path.join(WORK, "practice.db"))) # R2-I 空の旧形式の目印は無効 open(marker, "w").close() code, out = drill("reset", "--confirm") record("R2-I 空の旧形式の目印で reset --confirm", "拒否（非ゼロ）・領域は残る", f"終了コード{code}・practice.db{'あり' if os.path.exists(os.path.join(WORK,'practice.db')) else 'なし'}", code != 0 and os.path.exists(os.path.join(WORK, "practice.db"))) code, out = drill("rollback-code") record("R2-I 空の旧形式の目印で通常コマンド", "拒否（非ゼロ）", f"終了コード{code}", code != 0) safety.write_marker(WORK) # 正しい目印に戻す env = dict(os.environ, DRILL_WORK_DIR=os.path.join(HERE, "..", "work_not_under_base")) p = subprocess.run([sys.executable, os.path.join(HERE, "drill.py"), "setup"], capture_output=True, text=True, env=env) record("R2-I 基点の直下でない領域で setup", "拒否（非ゼロ）・作成なし", f"終了コード{p.returncode}", p.returncode != 0 and not os.path.lexists(os.path.join(HERE, "..", "work_not_under_base"))) env = dict(os.environ, DRILL_WORK_DIR=os.path.join(HERE, "not_a_work_dir")) p = subprocess.run([sys.executable, os.path.join(HERE, "drill.py"), "setup"], capture_output=True, text=True, env=env) record("R2-C 名前が work で始まらない領域で setup", "拒否（非ゼロ）", f"終了コード{p.returncode}", p.returncode != 0 and not os.path.exists(os.path.join(HERE, "not_a_work_dir"))) # R2-E 無関係なファイルがある既存フォルダを演習領域にしない other = os.path.join(HERE, f"work_selfcheck_unowned_{STAMP}") os.makedirs(other) note = os.path.join(other, "unrelated_notes.txt") open(note, "w", encoding="utf-8").write("SELFCHECK_UNRELATED") h0 = sha256(note) env = dict(os.environ, DRILL_WORK_DIR=other) p1 = subprocess.run([sys.executable, os.path.join(HERE, "drill.py"), "setup"], capture_output=True, text=True, env=env) marker_added = os.path.lexists(os.path.join(other, ".drill-work")) p2 = subprocess.run([sys.executable, os.path.join(HERE, "drill.py"), "reset", "--confirm"], capture_output=True, text=True, env=env) intact = os.path.isfile(note) and sha256(note) == h0 record("R2-E 無関係なファイルのある既存フォルダで setup → reset --confirm", "両方拒否・目印なし・ファイル不変", f"setup終了コード{p1.returncode}・目印{'あり' if marker_added else 'なし'}・reset終了コード{p2.returncode}・ファイル{'不変' if intact else '変化'}", p1.returncode != 0 and not marker_added and p2.returncode != 0 and intact) shutil.rmtree(other) # R2-G practice.db が演習領域の外へのリンクなら after-backup は更新しない db = os.path.join(WORK, "practice.db") ext = os.path.join(HERE, f"selfcheck_external_{STAMP}.db") shutil.move(db, ext) os.symlink(ext, db) h0 = sha256(ext) code, out = drill("after-backup") ext_same = sha256(ext) == h0 os.remove(db); shutil.move(ext, db) record("R2-G practice.db が領域外へのリンクのとき after-backup", "拒否（非ゼロ）・外のDB不変", f"終了コード{code}・外のDB{'不変' if ext_same else '変化'}", code != 0 and ext_same) # R2-H リンク先の無い復元先リンクを通って外に書かない rdir = os.path.join(WORK, "restore", "gen-001") rdb = os.path.join(rdir, "practice_restored.db") keep = rdb + ".keep" shutil.move(rdb, keep) ext_new = os.path.join(HERE, f"selfcheck_new_external_{STAMP}.db") os.symlink(ext_new, rdb) # リンク先はまだ無い code, out = drill("restore", "1") created = os.path.lexists(ext_new) link_kept = os.path.islink(rdb) os.remove(rdb); shutil.move(keep, rdb) if created: os.remove(ext_new) record("R2-H リンク先の無い復元先リンクで restore", "拒否（非ゼロ）・外に作成なし・リンク不変", f"終了コード{code}・外のファイル{'作成' if created else 'なし'}・リンク{'残存' if link_kept else '消失'}", code != 0 and not created and link_kept) # R2-D v1.0 は削除しない code, out = drill("rollback-code") record("R2-D v1.0 は削除処理なし", "「削除処理なし」と表示", "削除処理なし" if "削除処理なし" in out else out[-80:], "削除処理なし" in out) print() print("| 項目 | 期待結果 | 実際の結果 | 判定 |") print("|---|---|---|---|") for item, e, a, ok in RESULTS: print(f"| {item} | {e} | {a} | {'合' if ok else '否'} |") n_ng = sum(1 for r in RESULTS if not r[3]) print(f"\n合計 {len(RESULTS)} 件、期待どおり {len(RESULTS)-n_ng} 件、期待と異なる {n_ng} 件") # 一時領域を片付ける（内容つきの目印が一致する演習領域だけ） if safety.marker_valid(WORK): shutil.rmtree(WORK) print(f"一時演習領域を削除した: {WORK}") sys.exit(0 if n_ng == 0 else 1) if __name__ == "__main__": main()