260 lines
9.6 KiB
Python
260 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
"""PostToolUse hook — append a tool-execution *receipt* to the evidence ledger (C5).
|
|
|
|
이 훅은 Bash/Write/Edit 도구가 실제로 실행된 뒤(PostToolUse) 호출되어, 무엇이
|
|
실제로 돌았는지를 append-only 원장에 기록한다. validator(C6)는 이 원장을 읽어
|
|
에이전트가 report에서 주장한 E4/E5 등급(command+exit-code:0 / 파일 산출)이 실제
|
|
실행에 뒷받침되는지 대조한다 — 자기신고(self-report)를 receipt로 접지시킨다.
|
|
|
|
원장 위치: <evidence_dir>/ledger.jsonl (한 줄 = JSON receipt)
|
|
receipt 필드(C5):
|
|
{tool_use_id, tool_name, ts, cwd, command?, exit_code?, stdout_sha256?,
|
|
artifact_path?, artifact_sha256?}
|
|
- Bash: command / exit_code / stdout_sha256
|
|
- Write/Edit: artifact_path / artifact_sha256 (기록 시점=쓰기 직후, 디스크 실물 해시)
|
|
|
|
입력: Claude Code PostToolUse JSON on stdin
|
|
{tool_name, tool_input, tool_use_id?, cwd?, tool_response?}
|
|
|
|
강건성 계약:
|
|
- 절대 도구 파이프라인을 깨지 않는다. 어떤 예외에도 exit 0.
|
|
- workspace 미설정/디렉터리 부재면 조용히 degrade(로그는 stderr, 그래도 exit 0).
|
|
- ts는 실제 벽시계(UTC) — 이 훅은 일반 OS 프로세스라 현재시각을 쓸 수 있다.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, HERE)
|
|
|
|
ARTIFACT_TOOLS = ("Write", "Edit", "NotebookEdit", "MultiEdit")
|
|
_SECRET_RE = re.compile(
|
|
r"(?i)(--(?:password|token|secret|api-key)|authorization:|bearer)\s*(?:=|\s)\s*([^\s]+)"
|
|
)
|
|
|
|
|
|
def _log(msg):
|
|
try:
|
|
sys.stderr.write(f"[evidence_ledger] {msg}\n")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _sha256_text(s):
|
|
try:
|
|
return hashlib.sha256(str(s).encode("utf-8", "replace")).hexdigest()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _sha256_file(path):
|
|
try:
|
|
h = hashlib.sha256()
|
|
with open(path, "rb") as f:
|
|
for chunk in iter(lambda: f.read(65536), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _redact_command(command):
|
|
value = str(command or "")
|
|
value = _SECRET_RE.sub(lambda match: f"{match.group(1)}=[REDACTED]", value)
|
|
value = re.sub(r"(?i)([?&](?:token|key|secret|password)=)[^&\s]+", r"\1[REDACTED]", value)
|
|
return value
|
|
|
|
|
|
def _ledger_path():
|
|
"""<evidence_dir>/ledger.jsonl. workspace 미설정이면 None(조용히 degrade)."""
|
|
try:
|
|
import _workspace as W # noqa: E402
|
|
ed = W.evidence_dir()
|
|
except Exception as e: # WorkspaceNotSetError 포함
|
|
_log(f"workspace 미해석 — receipt 스킵: {e}")
|
|
return None
|
|
try:
|
|
os.makedirs(ed, exist_ok=True)
|
|
except Exception as e:
|
|
_log(f"evidence_dir 생성 실패 — receipt 스킵: {e}")
|
|
return None
|
|
return os.path.join(ed, "ledger.jsonl")
|
|
|
|
|
|
def _abs(path, cwd):
|
|
if not path:
|
|
return None
|
|
if os.path.isabs(path):
|
|
return path
|
|
for base in (cwd, os.environ.get("CLAUDE_PROJECT_DIR"), os.getcwd()):
|
|
if base:
|
|
cand = os.path.join(base, path)
|
|
if os.path.exists(cand):
|
|
return cand
|
|
# 존재 안 해도 cwd 기준 절대경로는 돌려준다(해시는 실패→None)
|
|
return os.path.join(cwd or os.getcwd(), path)
|
|
|
|
|
|
def _extract_exit(tool_response):
|
|
"""PostToolUse tool_response에서 exit code를 추출. finding P0-6: 어떤 exit 신호도
|
|
해석하지 못하면 **성공(0)으로 위장 기록하지 않고 None(미상)** 을 반환한다 — 예전엔
|
|
signal-less 응답을 0으로 적어, exit-code를 못 읽는 환경에서 자기신고 E4/E5 command
|
|
주장이 '성공 receipt'로 접지되는 우회가 있었다. 명시적 성공 신호(is_error=False)만
|
|
0으로 인정한다.
|
|
|
|
우선순위: 명시 숫자 필드 > interrupted(130) > is_error True(1) > is_error False(0) > None."""
|
|
if isinstance(tool_response, dict):
|
|
for k in ("exit_code", "exitCode", "returncode", "return_code", "code", "status"):
|
|
v = tool_response.get(k)
|
|
if isinstance(v, bool):
|
|
continue
|
|
if isinstance(v, int):
|
|
return v
|
|
if isinstance(v, str) and v.strip().lstrip("-").isdigit():
|
|
return int(v.strip())
|
|
if tool_response.get("interrupted") is True:
|
|
return 130
|
|
ie = tool_response.get("is_error")
|
|
if ie is None:
|
|
ie = tool_response.get("isError")
|
|
if ie is True:
|
|
return 1
|
|
if ie is False:
|
|
return 0 # 명시적 성공 신호만 0
|
|
return None # 미상 — 성공으로 위장하지 않는다
|
|
|
|
|
|
def _stdout_of(tool_response):
|
|
if isinstance(tool_response, str):
|
|
return tool_response
|
|
if isinstance(tool_response, dict):
|
|
for k in ("stdout", "output", "stdoutText", "result"):
|
|
v = tool_response.get(k)
|
|
if isinstance(v, str):
|
|
return v
|
|
content = tool_response.get("content")
|
|
if isinstance(content, list):
|
|
texts = [c.get("text", "") for c in content
|
|
if isinstance(c, dict) and isinstance(c.get("text"), str)]
|
|
if texts:
|
|
return "\n".join(texts)
|
|
return None
|
|
|
|
|
|
def build_receipt(payload):
|
|
tool = payload.get("tool_name") or payload.get("toolName") or ""
|
|
ti = payload.get("tool_input") or payload.get("toolInput") or {}
|
|
if not isinstance(ti, dict):
|
|
ti = {}
|
|
tr = payload.get("tool_response")
|
|
if tr is None:
|
|
tr = payload.get("toolResponse")
|
|
cwd = payload.get("cwd") or os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()
|
|
|
|
# finding P0-6: receipt 를 Claude Code 가 공급하는 실행 컨텍스트(session/agent/workflow/
|
|
# tool_use_id/cwd)에 결속한다 — 오래된 다른 작업·다른 세션의 receipt 재사용을 식별 가능하게.
|
|
receipt = {
|
|
"tool_use_id": (payload.get("tool_use_id") or payload.get("toolUseId")
|
|
or payload.get("id")),
|
|
"tool_name": tool,
|
|
"ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"cwd": cwd,
|
|
}
|
|
receipt["receipt_id"] = receipt.get("tool_use_id")
|
|
_sid = (payload.get("session_id") or payload.get("sessionId")
|
|
or os.environ.get("CLAUDE_SESSION_ID"))
|
|
_aid = (payload.get("agent_id") or payload.get("agentId")
|
|
or os.environ.get("CLAUDE_AGENT_ID"))
|
|
_wid = (payload.get("workflow_id") or os.environ.get("ORGOS_WORKFLOW_ID")
|
|
or os.environ.get("ORGOS_WORKFLOW"))
|
|
for _k, _v in (("session_id", _sid), ("agent_id", _aid), ("workflow_id", _wid)):
|
|
if _v:
|
|
receipt[_k] = _v
|
|
source_revision = payload.get("source_revision_sha256") or os.environ.get("ORGOS_SOURCE_REVISION_SHA256")
|
|
if source_revision:
|
|
receipt["source_revision_sha256"] = source_revision
|
|
|
|
if tool == "Bash":
|
|
command = str(ti.get("command", "")).strip()
|
|
receipt["command"] = _redact_command(command)
|
|
receipt["command_sha256"] = _sha256_text(command)
|
|
receipt["exit_code"] = _extract_exit(tr)
|
|
receipt["receipt_type"] = payload.get("receipt_type") or "command-run"
|
|
if payload.get("assertion_status") in ("passed", "failed"):
|
|
receipt["assertion_status"] = payload.get("assertion_status")
|
|
so = _stdout_of(tr)
|
|
if so is not None:
|
|
receipt["stdout_sha256"] = _sha256_text(so)
|
|
elif tool in ARTIFACT_TOOLS:
|
|
receipt["receipt_type"] = "artifact-write"
|
|
path = ti.get("file_path") or ti.get("notebook_path") or ti.get("path") or ""
|
|
receipt["artifact_path"] = path
|
|
h = _sha256_file(_abs(path, cwd)) if path else None
|
|
if h is None:
|
|
# 디스크 해시 불가 시 입력 콘텐츠로 폴백(Write=content, Edit=new_string 등)
|
|
content = (ti.get("content") if ti.get("content") is not None
|
|
else ti.get("new_string") if ti.get("new_string") is not None
|
|
else ti.get("new_source") if ti.get("new_source") is not None
|
|
else ti.get("new_str"))
|
|
if content is not None:
|
|
h = _sha256_text(content)
|
|
receipt["artifact_sha256"] = h
|
|
return receipt
|
|
|
|
|
|
def main():
|
|
try:
|
|
data = sys.stdin.read()
|
|
except Exception:
|
|
sys.exit(0)
|
|
if not data or not data.strip():
|
|
sys.exit(0)
|
|
try:
|
|
payload = json.loads(data)
|
|
except Exception as e:
|
|
_log(f"malformed PostToolUse JSON — 스킵: {e}")
|
|
sys.exit(0)
|
|
if not isinstance(payload, dict):
|
|
sys.exit(0)
|
|
|
|
tool = payload.get("tool_name") or payload.get("toolName") or ""
|
|
if tool not in ("Bash",) + ARTIFACT_TOOLS:
|
|
sys.exit(0) # 원장 대상 아님(Read/Grep 등) — 조용히 통과
|
|
|
|
try:
|
|
receipt = build_receipt(payload)
|
|
except Exception as e:
|
|
_log(f"receipt 생성 실패 — 스킵: {e}")
|
|
sys.exit(0)
|
|
|
|
lp = _ledger_path()
|
|
if not lp:
|
|
sys.exit(0)
|
|
try:
|
|
with open(lp, "a", encoding="utf-8") as f:
|
|
try:
|
|
import fcntl
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
|
|
except Exception:
|
|
pass
|
|
f.write(json.dumps(receipt, ensure_ascii=False) + "\n")
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
try:
|
|
import fcntl
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
|
|
except Exception:
|
|
pass
|
|
except Exception as e:
|
|
_log(f"원장 append 실패 — 스킵: {e}")
|
|
sys.exit(0)
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|