163 lines
9.8 KiB
Python
163 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
"""test_p0_trust_boundary.py — the re-review's P0 bypasses must now BLOCK.
|
|
|
|
이 스위트는 '정상 경로가 동작한다'가 아니라 '나쁜 경로가 막힌다'를 증명한다(재리뷰 지적의 핵심).
|
|
각 케이스는 리뷰가 재현한 우회를 그대로 재현하고, 이제 차단됨을 확인한다.
|
|
|
|
P0-1 workspace 미설정 → 운영 훅 fail-closed(exit 2)
|
|
P0-2 context-package 없이/위장 패키지로 Org OS 워커 spawn → 차단
|
|
P0-4a 원장(state/evidence/acceptance/registry) 직접 쓰기·evidence_ledger 수동호출·signoff → 차단
|
|
P0-4b plan 불일치 전이·actor 미지정 전이 → 차단; 승인은 acceptance_log 파생만
|
|
P0-4c 존재하지 않는/미검증 report 를 accepted 로 등록 → 차단
|
|
P0-5 report-type 누락/미지·위장 primary-artifacts → 차단
|
|
P0-6 receipt substring/basename 매칭 제거(정확 일치만)
|
|
"""
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
HOOKS = os.path.join(ROOT, ".claude", "hooks")
|
|
sys.path.insert(0, HOOKS)
|
|
os.environ["CLAUDE_PROJECT_DIR"] = ROOT
|
|
|
|
passed = failed = 0
|
|
|
|
|
|
def check(name, ok):
|
|
global passed, failed
|
|
if ok:
|
|
passed += 1
|
|
print(f" PASS {name}")
|
|
else:
|
|
failed += 1
|
|
print(f" FAIL {name}")
|
|
|
|
|
|
def _run(script, args, env_extra=None, stdin=None):
|
|
e = dict(os.environ)
|
|
e["CLAUDE_PROJECT_DIR"] = ROOT
|
|
e.pop("ORGOS_WORKSPACE", None)
|
|
if env_extra:
|
|
e.update(env_extra)
|
|
return subprocess.run([sys.executable, os.path.join(HOOKS, script)] + args,
|
|
capture_output=True, text=True, input=stdin, env=e)
|
|
|
|
|
|
# ============================================================ P0-1 workspace fail-closed
|
|
print("== P0-1 workspace-unset FAILS CLOSED across operational hooks ==")
|
|
nows = {} # ORGOS_WORKSPACE removed by _run
|
|
# Isolate the "unset" cases from the repository's optional .orgos-workspace
|
|
# pointer. A real workflow may legitimately populate that pointer while this
|
|
# suite is running; using ROOT here would then test "pointer configured", not
|
|
# "workspace unset". Seed only the worker card needed by stop_validate's
|
|
# unregistered-agent classifier.
|
|
clean_project = tempfile.mkdtemp(prefix="company-haness-p0-clean-", dir="/tmp")
|
|
clean_agents = os.path.join(clean_project, ".claude", "agents")
|
|
os.makedirs(clean_agents, exist_ok=True)
|
|
with open(os.path.join(clean_agents, "fam-eng-backend.md"), "w", encoding="utf-8") as fh:
|
|
fh.write("---\nname: fam-eng-backend\n---\n")
|
|
check("state_engine guard (unset ws) -> exit 2",
|
|
_run("state_engine.py", ["guard", "--workflow", "wf", "--to", "build"], nows).returncode == 2)
|
|
check("acceptance_log append (unset ws) -> exit 2",
|
|
_run("acceptance_log.py", ["append", "--report-id", "x", "--decision", "accepted"], nows).returncode == 2)
|
|
check("subagent_register report-producing (unset ws) -> exit 2",
|
|
_run("subagent_register.py", [], {"CLAUDE_PROJECT_DIR": clean_project},
|
|
stdin='{"agent_id":"a","agent_type":"eng-be","workflow_id":"wf"}').returncode == 2)
|
|
check("subagent_register helper (unset ws) -> exit 0 (not over-blocked)",
|
|
_run("subagent_register.py", [], nows,
|
|
stdin='{"agent_id":"h","agent_type":"general-purpose"}').returncode == 0)
|
|
check("stop_validate Org OS agent (unset ws) -> exit 2",
|
|
_run("stop_validate.py", [], {"CLAUDE_PROJECT_DIR": clean_project},
|
|
stdin='{"agent_id":"a","agent_type":"fam-eng-backend"}').returncode == 2)
|
|
check("stop_validate --main (unset ws) -> exit 0 (advisory)",
|
|
_run("stop_validate.py", ["--main"], nows, stdin='{"agent_id":"m"}').returncode == 0)
|
|
|
|
# ============================================================ direct hook API cases
|
|
import guard_tools as G # noqa: E402
|
|
import context_package as CP # noqa: E402
|
|
import validate_report as VR # noqa: E402
|
|
|
|
print("== P0-4a ledgers are a trust boundary (guard_tools) ==")
|
|
LED = [
|
|
("Write", {"file_path": "_sandbox/state/wf/workflow.yaml", "content": "x"}, "Write workflow.yaml"),
|
|
("Write", {"file_path": "_sandbox/evidence/wf/ledger.jsonl", "content": "x"}, "Write evidence ledger"),
|
|
("Edit", {"file_path": "_sandbox/state/acceptance-events.jsonl", "old_string": "a", "new_string": "b"}, "Edit acceptance-events"),
|
|
("Write", {"file_path": "_sandbox/state/wf/human-signoff.jsonl", "content": "x"}, "Write human-signoff"),
|
|
("Bash", {"command": "echo x >" + "> _sandbox/state/wf/state-events.jsonl"}, "append state-events"),
|
|
("Bash", {"command": "python3 -c \"open('_sandbox/evidence/x/ledger.jsonl','a').write('{}')\""}, "python -c open ledger"),
|
|
("Bash", {"command": "echo {} | python3 .claude/hooks/" + "evidence_ledger.py"}, "manual evidence_ledger.py"),
|
|
("Bash", {"command": "python3 .claude/hooks/state_engine.py signoff --workflow wf --stage acceptance --by HUMAN-x"}, "agent signoff"),
|
|
("Bash", {"command": "python3 .claude/hooks/state_engine.py record-human-signoff --workflow wf --stage acceptance --by HUMAN-001"}, "agent signoff alias"),
|
|
("Bash", {"command": "python3 .claude/hooks/state_engine.py review-artifact --workflow wf --report r.report.yaml --decision accepted --reviewer HUMAN-001"}, "agent impersonates human reviewer"),
|
|
("Bash", {"command": "python3 -c \"import acceptance_log as AL; AL.append_event(AL.build_event('x','accepted'))\""}, "low-level acceptance append"),
|
|
]
|
|
for t, ti, label in LED:
|
|
cat, _ = G.check(t, ti)
|
|
check(f"BLOCK {label}", cat is not None)
|
|
# legitimate CLIs still pass
|
|
for t, ti, label in [
|
|
("Bash", {"command": "python3 .claude/hooks/state_engine.py transition --workflow wf --to build --actor OPS-ORCH"}, "state transition CLI"),
|
|
("Bash", {"command": "python3 .claude/hooks/acceptance_log.py append --report-id r --decision accepted"}, "acceptance append CLI"),
|
|
("Bash", {"command": "python3 .claude/hooks/verify_run.py --workflow wf --agent QA --session s --category test --subject unit -- pytest -q"}, "typed verification runner"),
|
|
("Bash", {"command": "python3 -m py_compile .claude/hooks/" + "evidence_ledger.py"}, "py_compile mention"),
|
|
]:
|
|
cat, _ = G.check(t, ti)
|
|
check(f"ALLOW {label}", cat is None)
|
|
|
|
print("== P0-2 context-package spawn gate ==")
|
|
check("Org OS worker spawn WITHOUT package ref -> BLOCK",
|
|
G.check("Task", {"subagent_type": "eng-be", "prompt": "do it"})[0] is not None)
|
|
check("helper spawn (general-purpose) -> ALLOW",
|
|
G.check("Task", {"subagent_type": "general-purpose", "prompt": "search"})[0] is None)
|
|
# placeholder package rejected by strengthened validator
|
|
ph = {"workflow-id": "w", "task-id": "t", "mode": "converge", "tier": "standard",
|
|
"target-role-agent": "fam-eng-backend", "workspace": "_sandbox", "target-repo": "repo",
|
|
"objective": "x", "output-format": "y", "allowed-tools": ["ALL"], "task-boundaries": "b",
|
|
"non-goals": ["n"], "must-read": ["none"], "inherited-decisions": [],
|
|
"acceptance-tests": ["none"], "evidence-plan": ["self-assertion"],
|
|
"expected-output": {}, "token-budget": "unlimited"}
|
|
viol = CP.validate(ph)
|
|
check("placeholder package (none/ALL/unlimited/self-assertion) -> violations",
|
|
len(viol) >= 4)
|
|
check("fake target-role-agent card -> violation",
|
|
any("에이전트 카드" in v or "card" in v.lower() for v in CP.validate(dict(ph, **{"target-role-agent": "not-a-real-agent"}))))
|
|
|
|
print("== P0-5 report validator bypasses ==")
|
|
_hdr = {"bottom-line": "x", "decision-needed": {"needed": False},
|
|
"confidence": {"value": "Med"}, "risks": [], "evidence": [{"grade": "E1", "source-uri": "README.md"}]}
|
|
_idn = {"report-id": "r", "workflow-id": "wf", "role-id": "EXEC-CTO"}
|
|
check("no report-type -> BLOCK", len(VR.validate({"report-header": _hdr})) > 0)
|
|
check("unknown report-type -> BLOCK", len(VR.validate({"report-type": "made-up", **_idn, "report-header": _hdr})) > 0)
|
|
check("build w/ empty kind + null verification + null verification-performed -> BLOCK",
|
|
len(VR.validate({"report-type": "build", **_idn, "report-header": _hdr,
|
|
"primary-artifacts": [{"path": "README.md", "kind": "", "verification": None}],
|
|
"verification-performed": None})) > 0)
|
|
check("E3 self-report cmd (no receipt) + High confidence -> BLOCK (overconfidence)",
|
|
len(VR.validate({"report-type": "work", **_idn, "work-summary": "s",
|
|
"report-header": dict(_hdr, **{"confidence": {"value": "High"},
|
|
"evidence": [{"grade": "E3", "command": "pytest -q", "exit-code": 0}]})})) > 0)
|
|
_abs_company = os.path.join(ROOT, "org-os", "01-company", "company-context.yaml")
|
|
check("E3 abs-path company ref @ High -> capped (BLOCK)",
|
|
len(VR.validate({"report-type": "work", **_idn, "work-summary": "s",
|
|
"report-header": dict(_hdr, **{"confidence": {"value": "High"},
|
|
"evidence": [{"grade": "E3", "source-uri": _abs_company}]})})) > 0)
|
|
check("E3 CLAUDE.md @ High -> capped (BLOCK)",
|
|
len(VR.validate({"report-type": "work", **_idn, "work-summary": "s",
|
|
"report-header": dict(_hdr, **{"confidence": {"value": "High"},
|
|
"evidence": [{"grade": "E3", "source-uri": "CLAUDE.md"}]})})) > 0)
|
|
|
|
print("== P0-6 receipt matching is exact (no substring/basename) ==")
|
|
receipts = [{"command": "python3 -m pytest tests && echo done", "exit_code": 0},
|
|
{"artifact_path": "/tmp/a/result.json", "artifact_sha256": "abc"}]
|
|
check("substring command claim ('echo') no longer matches full receipt",
|
|
VR._cmd_receipt(receipts, "echo") is None)
|
|
check("exact command claim matches",
|
|
VR._cmd_receipt([{"command": "pytest -q", "exit_code": 0}], "pytest -q") is not None)
|
|
check("basename artifact claim (different dir) no longer matches",
|
|
VR._artifact_receipt(receipts, "/different/project/result.json", None) is None)
|
|
|
|
print(f"\n{passed} passed, {failed} failed")
|
|
sys.exit(0 if failed == 0 else 1)
|