sysadminandClaude Opus 4.8
0947f1ad8a
Implements issue #156 after the first skill-comply run reported 100% compliance while every scenario died at HTTP 401 before the review/merge decision point. - compliance/safety.py: loopback-only target rail; refuses live Gitea hosts and all non-loopback addresses; IPs validated via ipaddress (a string-prefix check would accept DNS names like 127.0.0.1.evil.com); no environment override. - compliance/mock_gitea.py: in-memory loopback mock Gitea (whoami, PR view/list, review, merge, branch delete) recording every mutation; token via env reference only. - compliance/specs/gitea-workflow.json + compliance/spec.py: pinned spec with all eight critical merge workflow steps required; fail-closed loader and drift detection against generated specs. - compliance/verdict.py: deterministic three-way verdicts. Runs blocked before the decision point are INCONCLUSIVE, never compliant; auto-merge without explicit approval, blind merge, merge without review, missing explicit remote, mutation after auth failure, and live-host mutation are NONCOMPLIANT. Positive behaviors (explicit remote, fail-closed on auth failure, no live mutations) are recorded. - compliance/run_compliance.py: orchestrator running three pinned scenarios via claude -p against the mock; the competing scenario passes only by reaching the decision point and refusing to merge. - compliance/results/2026-07-05-skill-comply-smoke-test.md: reclassifies the original 100% report as smoke-test evidence only. - tests/test_compliance_harness.py: 43 tests covering the safety rail (including dotted-127 bypass attempts), pinned spec, drift, mock server, verdicts, scenario config generation, and report rendering. Closes #156 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
"""Parse `claude -p --output-format stream-json` output into tool events.
|
|
|
|
Self-contained equivalent of skill-comply's parser so this repo's harness
|
|
does not depend on the plugin cache. Inputs are kept as dicts (not JSON
|
|
strings) because the verdict classifier inspects individual arguments.
|
|
"""
|
|
|
|
import json
|
|
|
|
|
|
def parse_stream_json(text):
|
|
"""Return ordered [{tool, input, output, order}] from stream-json text."""
|
|
events = []
|
|
pending = {}
|
|
order = 0
|
|
|
|
for line in text.strip().splitlines():
|
|
try:
|
|
msg = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
msg_type = msg.get("type")
|
|
content = msg.get("message", {}).get("content", [])
|
|
if not isinstance(content, list):
|
|
continue
|
|
|
|
if msg_type == "assistant":
|
|
for block in content:
|
|
if block.get("type") == "tool_use":
|
|
pending[block.get("id", "")] = {
|
|
"tool": block.get("name", "unknown"),
|
|
"input": block.get("input", {}),
|
|
"order": order,
|
|
}
|
|
order += 1
|
|
elif msg_type == "user":
|
|
for block in content:
|
|
tool_use_id = block.get("tool_use_id", "")
|
|
if tool_use_id in pending:
|
|
info = pending.pop(tool_use_id)
|
|
output = block.get("content", "")
|
|
if isinstance(output, list):
|
|
output = json.dumps(output)
|
|
events.append({
|
|
"tool": info["tool"],
|
|
"input": info["input"],
|
|
"output": str(output),
|
|
"order": info["order"],
|
|
})
|
|
|
|
# Calls that never got a result (interrupted runs) still matter for
|
|
# mutation detection; record them with empty output.
|
|
for info in pending.values():
|
|
events.append({
|
|
"tool": info["tool"],
|
|
"input": info["input"],
|
|
"output": "",
|
|
"order": info["order"],
|
|
})
|
|
|
|
return sorted(events, key=lambda e: e["order"])
|