「ゲートに合格した原稿」と「公開された原稿」は同じものか — 検証と公開のあいだに開く隙間

#!/usr/bin/env python3
"""gate_receipt.py — 品質ゲートを実行し、合格時に内容ダイジェスト付きの合格証を発行する。"""
from __future__ import annotations
 
import hashlib
import json
import os
import subprocess
import sys
import time
from pathlib import Path
 
RECEIPT_NAME = ".gate-receipt.json"
 
 
def digest(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()
 
 
def run_gate(script: Path, targets: list[Path]) -> tuple[bool, str]:
    """ゲートを1つ実行する。戻り値は (合格したか, 出力)。"""
    proc = subprocess.run(
        [sys.executable, str(script), *[str(t) for t in targets]],
        capture_output=True,
        text=True,
        timeout=300,
    )
    output = (proc.stdout + proc.stderr).strip()
    return proc.returncode == 0, output
 
 
def issue(repo: Path, gates: list[Path], targets: list[Path]) -> int:
    results = {}
    for gate in gates:
        passed, output = run_gate(gate, targets)
        results[gate.name] = {
            "passed": passed,
            "gate_digest": digest(gate),
            "tail": output.splitlines()[-3:],
        }
        status = "✅" if passed else "❌"
        print(f"{status} {gate.name}")
        if not passed:
            print(output)
 
    if not all(r["passed"] for r in results.values()):
        # 合格証を発行しない。既存の合格証があれば必ず無効化する。
        (repo / RECEIPT_NAME).unlink(missing_ok=True)
        print("合格証は発行されませんでした。公開してはいけません。")
        return 1
 
    receipt = {
        "version": 1,
        "run_id": os.environ.get("PIPELINE_RUN_ID", str(os.getpid())),
        "issued_at_wall": time.time(),
        "issued_at_mono": time.monotonic(),
        "gates": results,
        "artifacts": {
            str(t.relative_to(repo)): digest(t) for t in targets
        },
    }
    (repo / RECEIPT_NAME).write_text(json.dumps(receipt, indent=2, sort_keys=True))
    print(f"合格証を発行しました: {len(receipt['artifacts'])} ファイル")
    return 0
 
 
if __name__ == "__main__":
    repo_root = Path(sys.argv[1]).resolve()
    gate_dir = Path(sys.argv[2]).resolve()
    files = [Path(p).resolve() for p in sys.argv[3:]]
    sys.exit(issue(repo_root, sorted(gate_dir.glob("*_gate.py")), files))

#!/usr/bin/env python3
"""verify_receipt.py — 合格証を検証する。1つでも不一致なら非ゼロで終了する。"""
from __future__ import annotations
 
import json
import sys
import time
from pathlib import Path
 
from gate_receipt import RECEIPT_NAME, digest
 
MAX_AGE_SECONDS = 900  # 合格証の有効期間は15分
 
 
def verify(repo: Path) -> int:
    receipt_path = repo / RECEIPT_NAME
    if not receipt_path.exists():
        print("❌ 合格証がありません。ゲートを実行してください。")
        return 1
 
    receipt = json.loads(receipt_path.read_text())
 
    age = time.monotonic() - receipt["issued_at_mono"]
    if age < 0 or age > MAX_AGE_SECONDS:
        print(f"❌ 合格証が古すぎます（経過 {age:.0f} 秒）。再ゲートしてください。")
        return 1
 
    failures = []
    for rel, expected in receipt["artifacts"].items():
        target = repo / rel
        if not target.exists():
            failures.append(f"{rel}: 消失")
            continue
        actual = digest(target)
        if actual != expected:
            failures.append(f"{rel}: 内容が変わりました")
 
    for name, meta in receipt["gates"].items():
        gate_path = repo / "_gates" / name
        if gate_path.exists() and digest(gate_path) != meta["gate_digest"]:
            failures.append(f"{name}: ゲート自体が変更されました")
 
    if failures:
        print("❌ 合格証と実体が一致しません:")
        for f in failures:
            print(f"  - {f}")
        return 1
 
    print(f"✅ 合格証を検証しました（{len(receipt['artifacts'])} ファイル・経過 {age:.0f} 秒）")
    return 0
 
 
if __name__ == "__main__":
    sys.exit(verify(Path(sys.argv[1]).resolve()))