feat: plan 16 P2 batch 2 (SEC-12 drift, SEC-29 audit fail-closed, SEC-30 replay, SEC-15 low)
- SEC-12: drift-detect adds semantic invariants — negation-flip detection (a dropped "not" now FAILS despite high char-similarity) + env must-keep patterns. - SEC-29 (X-05): governance-check audit write fails CLOSED — an unwritable audit log denies the action and empties the output (no unaudited output). - SEC-30 (X-06): approval-verify records a one-time-use nonce (sha of token/sig) and rejects replays (enforced mode / when a nonce ledger is set); dev unchanged. - SEC-15 (low): typosquat distance<=2 with the levenshtein length-sentinel bug fixed (no false positives); tool-exec fails closed with no timeout backend in enforced mode; validate-tool-input now validates nested objects/arrays recursively. Verify: new SEC suites all green via gate, run-casan4 0-FAIL, adversarial 44/44, track-c 29/0, h5-approval 12/0, no regressions. Plan-16 P2 remaining: infra-gated only (SEC-14/22/23/24/25/26). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
thanhnv
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
8c06a55aed
commit
d695a598ee
@@ -38,6 +38,25 @@ fi
|
||||
|
||||
deny() { echo "APPROVAL_DENIED reason=$1 approver=$APPROVER action=$ACTION" >&2; exit 3; }
|
||||
|
||||
# SEC-30 (X-06): one-time-use. A verified approval token (JWT or offline signature)
|
||||
# is valid for its whole exp window, so it could be REPLAYED. Record a per-token
|
||||
# nonce (sha256 of the token/signature) and reject any repeat. Active in enforced
|
||||
# mode or when an explicit nonce ledger is configured (dev default: off).
|
||||
sha_stdin() {
|
||||
if command -v sha256sum >/dev/null 2>&1; then sha256sum | awk '{print $1}'
|
||||
else shasum -a 256 | awk '{print $1}'; fi
|
||||
}
|
||||
record_nonce_or_deny() {
|
||||
local nonce="$1"
|
||||
[[ "${CASAN_PROFILE:-}" == "prod" || -n "${CASAN_APPROVAL_NONCE_FILE:-}" ]] || return 0
|
||||
local ledger="${CASAN_APPROVAL_NONCE_FILE:-$PROJECT_ROOT/.specify/logs/level5/approval-nonces.txt}"
|
||||
mkdir -p "$(dirname "$ledger")" 2>/dev/null || true
|
||||
if [[ -f "$ledger" ]] && grep -qxF "$nonce" "$ledger" 2>/dev/null; then
|
||||
deny "approval_replayed(nonce=${nonce:0:12}…)"
|
||||
fi
|
||||
printf '%s\n' "$nonce" >> "$ledger" || deny "nonce_ledger_unwritable"
|
||||
}
|
||||
|
||||
[[ -f "$INPUT_FILE" ]] || deny "input_file_missing"
|
||||
[[ -f "$REVIEWERS_FILE" ]] || deny "reviewer_registry_missing"
|
||||
command -v openssl >/dev/null 2>&1 || deny "openssl_unavailable"
|
||||
@@ -164,6 +183,7 @@ print(f"OK role={role}")
|
||||
PY
|
||||
)" || deny "oidc_jwt_invalid(${JWT_OUT:-see_stderr})"
|
||||
ROLE="${JWT_OUT#OK role=}"
|
||||
record_nonce_or_deny "$(printf '%s' "$CASAN_APPROVAL_JWT" | sha_stdin)"
|
||||
echo "APPROVAL_OK mechanism=oidc role=$ROLE approver=$APPROVER action=$ACTION"
|
||||
exit 0
|
||||
fi
|
||||
@@ -192,5 +212,6 @@ printf '%s' "$MSG" > "$TMP"
|
||||
openssl dgst -sha256 -verify "$PUB" -signature "$SIG_FILE" "$TMP" >/dev/null 2>&1 \
|
||||
|| deny "approval_signature_invalid"
|
||||
|
||||
record_nonce_or_deny "$(sha_stdin < "$SIG_FILE")"
|
||||
echo "APPROVAL_OK role=$ROLE approver=$APPROVER action=$ACTION"
|
||||
exit 0
|
||||
|
||||
@@ -108,6 +108,10 @@ run "phase-sec10-agent-identity" bash "$TESTS/phase-sec10-tests.sh"
|
||||
run "phase-sec13-ssrf" bash "$TESTS/phase-sec13-tests.sh"
|
||||
run "phase-sec27-log-controlchar" bash "$TESTS/phase-sec27-tests.sh"
|
||||
run "phase-sec28-path-traversal" bash "$TESTS/phase-sec28-tests.sh"
|
||||
run "phase-sec12-drift-invariant" bash "$TESTS/phase-sec12-tests.sh"
|
||||
run "phase-sec29-audit-failclosed" bash "$TESTS/phase-sec29-tests.sh"
|
||||
run "phase-sec30-approval-replay" bash "$TESTS/phase-sec30-tests.sh"
|
||||
run "phase-sec15-low-cluster" bash "$TESTS/phase-sec15-tests.sh"
|
||||
|
||||
# ARCH-02: coverage cannot silently drop; ARCH-01: harness/policy cannot silently
|
||||
# drift. Both SKIP cleanly when no manifest is provisioned (non-strict dev/CI).
|
||||
|
||||
@@ -72,18 +72,52 @@ length_delta = abs(len(candidate) - len(golden)) / max(len(golden), 1)
|
||||
|
||||
status = "pass"
|
||||
action = "allow"
|
||||
reasons = []
|
||||
if similarity < 0.70 or length_delta > 0.50:
|
||||
status = "fail"
|
||||
action = "block_or_fallback"
|
||||
reasons.append("low_similarity_or_length_delta")
|
||||
elif similarity < 0.85 or length_delta > 0.30:
|
||||
status = "warn"
|
||||
action = "require_review"
|
||||
|
||||
# SEC-12: char-similarity alone misses SEMANTIC inversion — dropping a negation
|
||||
# ("must NOT deploy" -> "must deploy") keeps similarity high but flips meaning. A
|
||||
# candidate that removes negation tokens present in the golden is treated as drift.
|
||||
import re as _re
|
||||
NEG = _re.compile(
|
||||
r"\b(not|no|never|cannot|can't|don't|must not|mustn't|deny|denied|reject|disable|"
|
||||
r"disabled|forbid|prohibit|block|blocked|không|đừng|cấm|từ chối)\b",
|
||||
_re.IGNORECASE,
|
||||
)
|
||||
golden_neg = len(NEG.findall(golden))
|
||||
cand_neg = len(NEG.findall(candidate))
|
||||
if golden_neg > cand_neg:
|
||||
# The dangerous case: looks nearly identical but a negation vanished.
|
||||
status, action = "fail", "block_or_fallback"
|
||||
reasons.append(f"negation_dropped(golden={golden_neg},candidate={cand_neg})")
|
||||
|
||||
# must-keep invariants: regex patterns that MUST still appear in the candidate.
|
||||
_mk = os.environ.get("CASAN_DRIFT_MUSTKEEP_FILE", "")
|
||||
missing = []
|
||||
if _mk and os.path.isfile(_mk):
|
||||
for _line in open(_mk, encoding="utf-8", errors="replace"):
|
||||
_pat = _line.strip()
|
||||
if _pat and not _re.search(_pat, candidate):
|
||||
missing.append(_pat)
|
||||
if missing:
|
||||
status, action = "fail", "block_or_fallback"
|
||||
reasons.append(f"must_keep_missing={len(missing)}")
|
||||
|
||||
report = {
|
||||
"timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
||||
"harness": "L5-drift-detection",
|
||||
"status": status,
|
||||
"action": action,
|
||||
"reasons": reasons,
|
||||
"golden_negations": golden_neg,
|
||||
"candidate_negations": cand_neg,
|
||||
"must_keep_missing": missing,
|
||||
"similarity_ratio": round(similarity, 4),
|
||||
"length_delta_ratio": round(length_delta, 4),
|
||||
"golden_hash": hashlib.sha256(golden.encode()).hexdigest(),
|
||||
|
||||
@@ -155,7 +155,10 @@ TRACE_FILE="$TRACE_DIR/governance-$TRACE_ID.json"
|
||||
# raw, so a value containing `"` + newline could inject a SECOND forged audit record
|
||||
# (a fabricated "approved" decision). The record_hash is still computed from
|
||||
# RECORD_CORE above, so verify-audit-chain.sh recomputes and matches unchanged.
|
||||
CASAN_GC_REASONS="$REASONS_JSON" python - "$TRACE_FILE" "$AUDIT_LOG" \
|
||||
# SEC-29 (X-05): the audit write must FAIL CLOSED. If the audit log cannot be
|
||||
# written (disk full, read-only, quota), there must be NO governed action without
|
||||
# its accountability record — deny and empty the output rather than proceed.
|
||||
if ! CASAN_GC_REASONS="$REASONS_JSON" python - "$TRACE_FILE" "$AUDIT_LOG" \
|
||||
"$TIMESTAMP" "$TRACE_ID" "$ACTION_NAME" "$ACTOR" "$RISK_LEVEL" "$DECISION" \
|
||||
"$APPROVAL_STATUS" "$APPROVER" "$INPUT_HASH" "$OUTPUT_HASH" "$PREV_HASH" "$RECORD_HASH" <<'PY'
|
||||
import json, os, sys
|
||||
@@ -180,7 +183,14 @@ with open(audit_log, "a", encoding="utf-8") as f:
|
||||
# Compact separators: the chain line is regex-parsed elsewhere and must match
|
||||
# the original printf format (no space after ':' / ',').
|
||||
f.write(json.dumps(rec, separators=(",", ":")) + "\n")
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
PY
|
||||
then
|
||||
: > "$OUTPUT_FILE" 2>/dev/null || true
|
||||
echo "GOVERNANCE_DENIED trace_id=$TRACE_ID reason=audit_unwritable (fail-closed: no governed action without an audit record)" >&2
|
||||
exit 2
|
||||
fi
|
||||
|
||||
# --- External anchor: cryptographically sign the new chain head ---
|
||||
# A re-forged chain (recomputed hashes) changes the head; without the private
|
||||
|
||||
@@ -69,8 +69,12 @@ def parse_deps(text: str, name: str):
|
||||
|
||||
|
||||
def levenshtein(a: str, b: str) -> int:
|
||||
if abs(len(a) - len(b)) > 1:
|
||||
return 2
|
||||
# SEC-15: return a LARGE sentinel (not 2) when lengths are far apart — the old
|
||||
# sentinel 2 collided with the widened distance<=2 typosquat threshold and
|
||||
# produced false positives (e.g. fastapi vs numpy). For len-diff<=2 the DP below
|
||||
# computes the true edit distance.
|
||||
if abs(len(a) - len(b)) > 2:
|
||||
return 99
|
||||
prev = list(range(len(b) + 1))
|
||||
for i, ca in enumerate(a, 1):
|
||||
row = [i]
|
||||
@@ -106,7 +110,11 @@ def main() -> int:
|
||||
findings.append({"package": name, "reason": "denylisted_or_known_malicious"})
|
||||
continue
|
||||
if name not in known:
|
||||
near = next((k for k in known if levenshtein(name.lower(), k.lower()) == 1), None)
|
||||
# SEC-15: distance==1 missed 2-char typosquats (e.g. reqests/reqeusts of
|
||||
# "requests"). Allow distance<=2 for names long enough that a 2-edit match
|
||||
# is meaningful (short names stay at 1 to avoid false positives).
|
||||
_maxd = 2 if len(name) >= 5 else 1
|
||||
near = next((k for k in known if 1 <= levenshtein(name.lower(), k.lower()) <= _maxd), None)
|
||||
if near:
|
||||
findings.append({"package": name, "reason": "typosquat_of:" + near})
|
||||
continue
|
||||
|
||||
@@ -29,7 +29,13 @@ elif command -v perl >/dev/null 2>&1; then
|
||||
perl -e 'my $t=shift; $SIG{ALRM}=sub{exit 124}; alarm($t); exec @ARGV or exit 127;' "$TIMEOUT" "$@"
|
||||
rc=$?
|
||||
else
|
||||
echo "TOOL_EXEC_NO_TIMEOUT_BACKEND" >&2
|
||||
# SEC-15: no timeout backend. In enforced mode REFUSE (fail-closed) rather than
|
||||
# run a command that could hang the pipeline; dev warns and proceeds.
|
||||
if [[ "${CASAN_PROFILE:-}" == "prod" || "${CASAN_TOOL_EXEC_STRICT:-}" == "1" ]]; then
|
||||
echo "TOOL_EXEC_NO_TIMEOUT_BACKEND refusing (fail-closed: install coreutils timeout or perl)" >&2
|
||||
exit 2
|
||||
fi
|
||||
echo "TOOL_EXEC_NO_TIMEOUT_BACKEND (dev: running without timeout)" >&2
|
||||
"$@"
|
||||
rc=$?
|
||||
fi
|
||||
|
||||
@@ -30,28 +30,39 @@ TYPES = {
|
||||
}
|
||||
errors = []
|
||||
|
||||
if schema.get("type") == "object" and not isinstance(data, dict):
|
||||
errors.append("root: expected object")
|
||||
else:
|
||||
props = schema.get("properties", {})
|
||||
for field in schema.get("required", []):
|
||||
if field not in data:
|
||||
errors.append(f"missing required field: {field}")
|
||||
if schema.get("additionalProperties") is False:
|
||||
for key in data:
|
||||
if key not in props:
|
||||
errors.append(f"unexpected field: {key}")
|
||||
for key, spec in props.items():
|
||||
if key not in data:
|
||||
continue
|
||||
expected = spec.get("type")
|
||||
py = TYPES.get(expected)
|
||||
# bool is a subclass of int — guard so a boolean isn't accepted as integer
|
||||
if py and (not isinstance(data[key], py)
|
||||
or (expected in ("integer", "number") and isinstance(data[key], bool))):
|
||||
errors.append(f"field {key}: expected {expected}")
|
||||
if "enum" in spec and data[key] not in spec["enum"]:
|
||||
errors.append(f"field {key}: not in enum {spec['enum']}")
|
||||
|
||||
# SEC-15: validate RECURSIVELY. Previously only the top level was checked, so a
|
||||
# nested object could smuggle wrong types / unexpected fields past the gate.
|
||||
def validate(schema, data, path):
|
||||
expected = schema.get("type")
|
||||
py = TYPES.get(expected)
|
||||
if py and (not isinstance(data, py)
|
||||
or (expected in ("integer", "number") and isinstance(data, bool))):
|
||||
errors.append(f"{path or 'root'}: expected {expected}")
|
||||
return
|
||||
if "enum" in schema and data not in schema["enum"]:
|
||||
errors.append(f"{path or 'root'}: not in enum {schema['enum']}")
|
||||
|
||||
if expected == "object" and isinstance(data, dict):
|
||||
props = schema.get("properties", {})
|
||||
for field in schema.get("required", []):
|
||||
if field not in data:
|
||||
errors.append(f"{path or 'root'}: missing required field: {field}")
|
||||
if schema.get("additionalProperties") is False:
|
||||
for key in data:
|
||||
if key not in props:
|
||||
errors.append(f"{path or 'root'}: unexpected field: {key}")
|
||||
for key, spec in props.items():
|
||||
if key in data:
|
||||
validate(spec, data[key], f"{path}.{key}" if path else key)
|
||||
elif expected == "array" and isinstance(data, list):
|
||||
item_spec = schema.get("items")
|
||||
if isinstance(item_spec, dict):
|
||||
for i, item in enumerate(data):
|
||||
validate(item_spec, item, f"{path}[{i}]")
|
||||
|
||||
|
||||
validate(schema, data, "")
|
||||
|
||||
if errors:
|
||||
sys.stderr.write("TOOL_INPUT_INVALID " + "; ".join(errors) + "\n")
|
||||
|
||||
Reference in New Issue
Block a user