無人実行のエージェントが設定ファイルを書き換えたら戻す、小さな番人スクリプトの組み方

#!/usr/bin/env python3
"""無人実行の前後で設定ファイルの指紋を比べ、書き換えがあれば元に戻して止める番人。"""
import argparse
import hashlib
import json
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path
 
GUARD_HOME = Path(os.environ.get("AGENT_GUARD_HOME", Path.home() / ".agent-guard"))
TAMPER_EXIT = 42  # CLI 側の終了コード（構造化エラーの 3 など）と重ならない値
 
 
def fingerprint(path: Path):
    """ファイルなら中身の sha256、ディレクトリなら配下全体の sha256。無ければ None。"""
    if not path.exists():
        return None
    h = hashlib.sha256()
    if path.is_dir():
        for p in sorted(path.rglob("*")):
            if p.is_file():
                h.update(str(p.relative_to(path)).encode())
                h.update(p.read_bytes())
        return "dir:" + h.hexdigest()
    h.update(path.read_bytes())
    return h.hexdigest()
 
 
def load_watch(list_file: str):
    paths = []
    for line in Path(list_file).read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if line and not line.startswith("#"):
            paths.append(Path(os.path.expanduser(line)).resolve())
    return paths
 
 
def snapshot(paths, run_dir: Path):
    """指紋を記録し、元に戻せるよう実物も控えておく。"""
    state = {}
    for i, p in enumerate(paths):
        state[str(p)] = fingerprint(p)
        dest = run_dir / "before" / str(i)
        if p.is_dir():
            shutil.copytree(p, dest)
        elif p.exists():
            dest.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(p, dest)
    (run_dir / "state.json").write_text(json.dumps(state, indent=2, ensure_ascii=False))
    return state
 
 
def restore(paths, state, run_dir: Path, changed):
    """書き換えられた版は quarantine へ退避し、実行前の版を戻す（削除はしない）。"""
    for i, p in enumerate(paths):
        if str(p) not in changed:
            continue
        if p.exists():
            q = run_dir / "quarantine" / str(i)
            q.parent.mkdir(parents=True, exist_ok=True)
            shutil.move(str(p), str(q))
        src = run_dir / "before" / str(i)
        if state[str(p)] is None:
            continue  # 実行前は存在しなかった → 退避だけで元どおり
        if src.is_dir():
            shutil.copytree(src, p)
        else:
            shutil.copy2(src, p)
 
 
def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--watch", required=True, help="見張るパスを1行1つ書いたファイル")
    ap.add_argument("--timeout", type=int, default=3600, help="秒。headless の既定は無制限なので必ず区切る")
    ap.add_argument("cmd", nargs=argparse.REMAINDER, help="-- の後に実行するコマンド")
    a = ap.parse_args()
    cmd = a.cmd[1:] if a.cmd[:1] == ["--"] else a.cmd
    if not cmd:
        ap.error("実行するコマンドがありません")
 
    paths = load_watch(a.watch)
    run_dir = GUARD_HOME / f"{time.strftime('%Y%m%d-%H%M%S')}-{os.getpid()}"
    run_dir.mkdir(parents=True)
    state = snapshot(paths, run_dir)
 
    try:
        rc = subprocess.run(cmd, timeout=a.timeout).returncode
    except subprocess.TimeoutExpired:
        print(f"[guard] {a.timeout} 秒で打ち切りました", file=sys.stderr)
        rc = 124
 
    changed = {p: {"before": b, "after": fingerprint(Path(p))}
               for p, b in state.items() if fingerprint(Path(p)) != b}
    report = {"command": cmd, "exit": rc, "changed": changed}
    (run_dir / "report.json").write_text(json.dumps(report, indent=2, ensure_ascii=False))
 
    if changed:
        restore(paths, state, run_dir, changed)
        for p in changed:
            print(f"[guard] 書き換えを検知し、実行前の版へ戻しました: {p}", file=sys.stderr)
        print(f"[guard] 詳細: {run_dir / 'report.json'}", file=sys.stderr)
        sys.exit(TAMPER_EXIT)
    sys.exit(rc)
 
 
if __name__ == "__main__":
    main()

# ~/.agent-guard/watch.txt
# プロジェクト側の設定ディレクトリ（実際のパスに置き換える）
~/work/wallpaper-batch/<プロジェクト設定ディレクトリ>
# ユーザー側の設定ファイル（実際のパスに置き換える）
~/<ユーザー設定の置き場>/settings.json
# 夜のジョブを起動するスクリプト自身も見張る
~/bin/run-nightly.sh

#!/bin/bash
# ~/bin/run-nightly.sh — cron や launchd から呼ぶ入口
export AGENT_GUARD_HOME="$HOME/.agent-guard"
PROMPT="$(cat "$HOME/jobs/wallpaper-tagging.md")"
 
python3 "$HOME/bin/agent_guard.py" \
  --watch "$AGENT_GUARD_HOME/watch.txt" --timeout 2700 -- \
  agy -p "$PROMPT"
rc=$?
 
case $rc in
  0)   ;;
  42)  echo "$(date '+%F %T') 設定の書き換えを検知 → 次回の実行を止める" >> "$AGENT_GUARD_HOME/alerts.log"
       touch "$AGENT_GUARD_HOME/HOLD" ;;
  124) echo "$(date '+%F %T') 打ち切り" >> "$AGENT_GUARD_HOME/alerts.log" ;;
  *)   echo "$(date '+%F %T') exit $rc" >> "$AGENT_GUARD_HOME/alerts.log" ;;
esac