gitea-admin/CASAN
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(harness): Git Bash + graceful degradation for Windows agentic bridge

Browse Source 
Wide-deployment Windows path without WSL2. The agentic bridge already runs
on native Python + PowerShell; the only bash dependency is the H4/H2 gate
scripts, which run under Git Bash (Git for Windows) — much lighter than WSL2.

- h4_scan now returns a status (ok|blocked|timeout|unavailable). Timeout stays
  FAIL-CLOSED (block/deny). "unavailable" (no bash / gate missing) DEGRADES the
  turn to observed_only and does NOT block the developer — never silently
  certifies without a working gate.
- bash interpreter is configurable via CASAN_AGENTIC_BASH; gates use it.
- doctor reports bash_available / gates_runnable + a remediation warning, and
  stays green (degraded, not failed) when bash is absent.
- tests: +4 no-bash cases (degrade to observed_only, tool still allowed,
  non-certified finalize, injection still blocked when bash present). 34/34.
- docs: Windows guide + security guide now point to Git Bash, not WSL2, and
  document the timeout-vs-unavailable distinction.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
thanhnv
2026-07-23 20:52:02 +07:00
co-authored by Claude Opus 4.8
parent 4bb184b935
commit f6d28a3163
4 changed files with 141 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/casan-harness/scripts/python/agentic_bridge.py
+82 -23
View File
@@ -34,6 +34,7 @@ import hashlib
import json
import os
import re
import shutil
import subprocess
import sys
import time
@@ -351,6 +352,16 @@ def internal_timeout():
return 8.0
def bash_bin():
# The bash interpreter used to run the H4/H2 gate scripts. On Windows this is
# Git Bash (Git for Windows) or WSL bash; override via CASAN_AGENTIC_BASH.
return os.environ.get("CASAN_AGENTIC_BASH") or os.environ.get("CASAN_BASH_BIN") or "bash"
def bash_available():
return shutil.which(bash_bin()) is not None
def resolve_strength(declared, client):
"""Compute the certification strength, DOWNGRADING (never upgrading)."""
declared = declared if declared in STRENGTH_RANK else "project_hook"
@@ -363,7 +374,9 @@ def resolve_strength(declared, client):
# ─────────────────────────────────────────────────────────────────────────────
# Gate invocation (real H1/H2/H4/H5 controls) with a hard internal timeout.
# Returns (ok: bool, reason: str). On timeout/crash -> (False, reason) fail-closed.
# h4_scan returns a status string (ok|blocked|timeout|unavailable); h2 returns
# (ok, reason). Timeout is fail-closed (block/deny); "unavailable" (no bash) is a
# graceful degradation to observed_only, not a block.
# ─────────────────────────────────────────────────────────────────────────────
def _run_gate(argv, stdin_text=None):
@@ -383,23 +396,36 @@ def _run_gate(argv, stdin_text=None):
def h4_scan(text, mode="input"):
"""H4 security scan on a piece of text via security-check.sh. Returns
(ok, reason). ok == True means not blocked."""
"""H4 security scan on a piece of text via security-check.sh.
Returns a status STRING so callers can distinguish two very different cases:
* "ok" — scanned, not blocked.
* "blocked" — scanned and the gate refused (real policy block).
* "timeout" — the gate did not answer within the internal timeout;
treated as FAIL-CLOSED (block/deny) by callers.
* "unavailable" — the gate could not run at all (no bash / script missing).
This is a deployment reality (e.g. native Windows with no
Git Bash), NOT an attack, so callers DEGRADE the turn to
observed_only rather than hard-blocking every prompt.
"""
script = os.path.join(gates_dir(), "security-check.sh")
if not os.path.exists(script):
# No gate available: fail closed under enforcement, open under observe.
return (enforcement_mode() != "enforce"), "h4_gate_missing"
if not os.path.exists(script) or not bash_available():
return "unavailable"
tmpin = os.path.join(sessions_dir(), ".scan-in-%s" % uuid.uuid4().hex[:8])
tmpout = os.path.join(sessions_dir(), ".scan-out-%s" % uuid.uuid4().hex[:8])
try:
with open(tmpin, "w", encoding="utf-8") as fh:
fh.write(text or "")
rc, _out, err = _run_gate(["bash", script, tmpin, tmpout, mode])
rc, _out, _err = _run_gate([bash_bin(), script, tmpin, tmpout, mode])
if rc == 124:
return False, "h4_internal_timeout"
return "timeout"
if rc == 125:
# Interpreter present per which() but exec still failed — treat as
# unavailable, not a policy block.
return "unavailable"
if rc != 0:
return False, "h4_blocked"
return True, "h4_ok"
return "blocked"
return "ok"
finally:
for p in (tmpin, tmpout):
try:
@@ -411,13 +437,13 @@ def h4_scan(text, mode="input"):
def h2_registry_gate(action, idempotency_key):
"""H2 tool-registry gate for a mapped side-effect action."""
script = os.path.join(gates_dir(), "tool-registry-gate.sh")
if not os.path.exists(script):
if not os.path.exists(script) or not bash_available():
return True, "h2_gate_missing"
env = dict(os.environ)
env["CASAN_IDEMPOTENCY_KEY"] = idempotency_key
try:
proc = subprocess.run(
["bash", script, action],
[bash_bin(), script, action],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=internal_timeout(),
@@ -646,12 +672,14 @@ def op_begin(req):
"project=%s client=%s" % (rec["project_id"], client))
# H4 security admission on the prompt.
ok, reason = h4_scan(req.get("prompt", ""), "input")
if not ok:
h4 = h4_scan(req.get("prompt", ""), "input")
if h4 in ("blocked", "timeout"):
# Real policy block, or fail-closed on timeout (gate exists but did not
# answer in time). Either way the prompt is refused before the model.
reason = "h4_blocked" if h4 == "blocked" else "h4_internal_timeout"
rec["state"] = "Rejected"
add_evidence(rec, "H4", "prompt-admission", "block", reason)
save_admission(rec)
# A rejected turn is closed immediately and never certified.
write_trace_file(rec, False, ["prompt_blocked:%s" % reason])
resp = _base_response("begin", "block", admission_id=admission_id, trace_id=trace_id,
integration_mode=strength, certification_strength=strength,
@@ -659,13 +687,25 @@ def op_begin(req):
resp["warnings"].append("prompt blocked by H4 (%s) — not sent to model per client capability" % reason)
return resp
add_evidence(rec, "H4", "prompt-admission", "allow", "h4_ok")
if h4 == "unavailable":
# Graceful degradation (e.g. native Windows with no Git Bash): the H4
# gate could not run, so we CANNOT certify this turn — downgrade to
# observed_only — but we do NOT block the developer's prompt.
strength = "observed_only"
rec["integration_mode"] = "observed_only"
rec["certification_strength"] = "observed_only"
add_evidence(rec, "H4", "prompt-admission", "degraded", "gate_unavailable_no_bash")
else:
add_evidence(rec, "H4", "prompt-admission", "allow", "h4_ok")
save_admission(rec)
resp = _base_response("begin", "allow", admission_id=admission_id, trace_id=trace_id,
integration_mode=strength, certification_strength=strength,
reason="admitted")
if strength == "observed_only":
if h4 == "unavailable":
resp["warnings"].append("H4 security gate unavailable (no bash on PATH) — "
"turn is observed_only, install Git Bash to enable certification")
elif strength == "observed_only":
resp["warnings"].append("observe mode — this turn is telemetry-only and NOT certified")
resp["context"] = "CASAN admission %s open (strength=%s)" % (admission_id[:8], strength)
return resp
@@ -717,13 +757,20 @@ def op_pre_tool(req):
# H4 scan on the tool input.
tool_input_text = req.get("tool_input")
if tool_input_text is not None:
ok, reason = h4_scan(tool_input_text if isinstance(tool_input_text, str)
else json.dumps(tool_input_text, ensure_ascii=False), "input")
if not ok:
h4 = h4_scan(tool_input_text if isinstance(tool_input_text, str)
else json.dumps(tool_input_text, ensure_ascii=False), "input")
if h4 in ("blocked", "timeout"):
# Real block or fail-closed timeout — deny the side effect.
reason = "h4_blocked" if h4 == "blocked" else "h4_internal_timeout"
add_evidence(rec, "H4", "pre-tool", "deny", "%s:%s" % (tool, reason))
save_admission(rec)
return _base_response("pre-tool", "deny", admission_id=admission_id,
trace_id=rec.get("trace_id"), reason=reason)
if h4 == "unavailable":
# Gate could not run (no bash): the admission gate above already
# governs this side effect, and the turn is non-certifiable, so we
# record the degradation but do not block the developer.
add_evidence(rec, "H4", "pre-tool", "degraded", "%s:gate_unavailable_no_bash" % tool)
# H2 registry gate for mapped side-effect actions. Opt-in via
# CASAN_AGENTIC_H2_REGISTRY=1: the tool-registry is keyed on NAMED CASAN
@@ -958,20 +1005,32 @@ def op_report(args):
# ─────────────────────────────────────────────────────────────────────────────
def op_doctor(_args):
h4_present = os.path.exists(os.path.join(gates_dir(), "security-check.sh"))
bash_ok = bash_available()
checks = {
"bridge_enabled": bridge_enabled(),
"enforcement_mode": enforcement_mode(),
"harness_root": harness_root(),
"state_root": state_root(),
"sessions_dir_writable": os.access(sessions_dir(), os.W_OK),
"h4_gate_present": os.path.exists(os.path.join(gates_dir(), "security-check.sh")),
"bash_bin": bash_bin(),
"bash_available": bash_ok,
"h4_gate_present": h4_present,
"h2_gate_present": os.path.exists(os.path.join(gates_dir(), "tool-registry-gate.sh")),
"gates_runnable": bool(h4_present and bash_ok),
"metrics_log": metrics_log(),
"schema_version": SCHEMA_VERSION,
}
if not bash_ok:
checks["warning"] = ("no bash on PATH — the H4/H2 gates cannot run, so turns "
"degrade to observed_only (never certified). Install Git for "
"Windows (Git Bash) or set CASAN_AGENTIC_BASH to enable "
"certification. The bridge itself still works.")
print(json.dumps(checks, ensure_ascii=False, indent=2))
ok = checks["sessions_dir_writable"] and checks["h4_gate_present"]
return 0 if ok else 1
# Doctor is GREEN as long as the bridge can operate (state writable). Missing
# bash is a warning (degraded), not a hard failure — that is the whole point
# of graceful degradation for wide Windows deployment.
return 0 if checks["sessions_dir_writable"] else 1
# ─────────────────────────────────────────────────────────────────────────────