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

Complete CASAN backlog tier 1 controls

Browse Source
This commit is contained in:
thanhnv
2026-07-06 11:31:17 +09:00
parent 571c2b7e80
commit e79d7973fb
22 changed files with 685 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AINative_OKR_CASAN5/.specify/scripts/bash/approval-jwt-mint.py
Executable
+69
View File
@@ -0,0 +1,69 @@
#!/usr/bin/env python3
"""Mint a mock IdP RS256 approval JWT for CASAN tests/dev.
The token is bound to the same high-risk request that governance-check verifies:
sub=<approver>, role=<IdP role>, action, actor, and input_sha256.
"""
import argparse
import base64
import hashlib
import json
import os
import subprocess
import sys
import tempfile
import time
def b64u(data: bytes) -> str:
return base64.urlsafe_b64encode(data).decode().rstrip("=")
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--key", required=True)
ap.add_argument("--sub", required=True)
ap.add_argument("--role", required=True)
ap.add_argument("--action", required=True)
ap.add_argument("--actor", required=True)
ap.add_argument("--input", required=True)
ap.add_argument("--exp-offset", type=int, default=300)
args = ap.parse_args()
with open(args.input, "rb") as f:
input_sha = hashlib.sha256(f.read()).hexdigest()
now = int(time.time())
header = {"alg": "RS256", "typ": "JWT"}
claims = {
"iss": "casan-mock-idp",
"sub": args.sub,
"role": args.role,
"action": args.action,
"actor": args.actor,
"input_sha256": input_sha,
"iat": now,
"exp": now + args.exp_offset,
}
signing_input = ".".join([
b64u(json.dumps(header, separators=(",", ":"), sort_keys=True).encode()),
b64u(json.dumps(claims, separators=(",", ":"), sort_keys=True).encode()),
])
with tempfile.TemporaryDirectory() as td:
msg = os.path.join(td, "msg.txt")
sig = os.path.join(td, "sig.bin")
open(msg, "wb").write(signing_input.encode())
rc = subprocess.run(
["openssl", "dgst", "-sha256", "-sign", args.key, "-out", sig, msg],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
).returncode
if rc != 0:
print("approval-jwt-mint: signing failed", file=sys.stderr)
return 1
token = signing_input + "." + b64u(open(sig, "rb").read())
print(token)
return 0
if __name__ == "__main__":
raise SystemExit(main())
AINative_OKR_CASAN5/.specify/scripts/bash/approval-verify.sh
+82 -7
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env bash
set -uo pipefail
# CASAN H5 — Signed-approval verifier (Approval-identity MVP · C4 / V20).
# CASAN H5 — Signed/JWT approval verifier (Approval-identity MVP · C4 / V20).
#
# Problem: high-risk approval used to trust a plain env var (CASAN_APPROVER=bob) —
# anyone who can set the env can "approve". This binds an approval to a REGISTERED
@@ -13,11 +13,14 @@ set -uo pipefail
#
# Usage:
# approval-verify.sh <action> <actor> <input-file> <approver-id> <sig-file>
# CASAN_APPROVAL_JWT=<rs256-jwt> approval-verify.sh <action> <actor> <input-file> <approver-id> -
# Registry (line format, no yaml dep):
# reviewer <id> <role> <pubkey-file>
# action <action-name|default> <comma,roles>
# Env: CASAN_REVIEWERS_FILE (default governance/reviewers.registry)
# CASAN_REVIEWERS_DIR (default governance/reviewers) — base dir for pubkey-file
# CASAN_APPROVAL_JWT (optional RS256 IdP token)
# CASAN_IDP_PUBLIC_KEY (default central-governance/idp-public.pem)
# Exit: 0 ok (prints "APPROVAL_OK role=<role>"), 3 deny (reason on stderr), 64 usage.
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
@@ -27,7 +30,7 @@ REVIEWERS_FILE="${CASAN_REVIEWERS_FILE:-$GOV_DIR/reviewers.registry}"
REVIEWERS_DIR="${CASAN_REVIEWERS_DIR:-$GOV_DIR/reviewers}"
ACTION="${1:-}"; ACTOR="${2:-}"; INPUT_FILE="${3:-}"; APPROVER="${4:-}"; SIG_FILE="${5:-}"
if [[ -z "$ACTION" || -z "$ACTOR" || -z "$INPUT_FILE" || -z "$APPROVER" || -z "$SIG_FILE" ]]; then
if [[ -z "$ACTION" || -z "$ACTOR" || -z "$INPUT_FILE" || -z "$APPROVER" ]]; then
echo "Usage: approval-verify.sh <action> <actor> <input-file> <approver-id> <sig-file>" >&2
exit 64
fi
@@ -35,7 +38,6 @@ fi
deny() { echo "APPROVAL_DENIED reason=$1 approver=$APPROVER action=$ACTION" >&2; exit 3; }
[[ -f "$INPUT_FILE" ]] || deny "input_file_missing"
[[ -f "$SIG_FILE" ]] || deny "signature_missing"
[[ -f "$REVIEWERS_FILE" ]] || deny "reviewer_registry_missing"
command -v openssl >/dev/null 2>&1 || deny "openssl_unavailable"
@@ -44,15 +46,89 @@ hash_file() {
else shasum -a 256 "$1" | awk '{print $1}'; fi
}
INPUT_SHA="$(hash_file "$INPUT_FILE")"
# Role authorization for this action (fallback to the "default" action policy).
ROLES="$(awk -v a="$ACTION" '$1=="action" && $2==a {print $3; exit}' "$REVIEWERS_FILE")"
[[ -n "$ROLES" ]] || ROLES="$(awk '$1=="action" && $2=="default" {print $3; exit}' "$REVIEWERS_FILE")"
if [[ -n "${CASAN_APPROVAL_JWT:-}" ]]; then
IDP_PUB="${CASAN_IDP_PUBLIC_KEY:-$GOV_DIR/idp-public.pem}"
[[ -f "$IDP_PUB" ]] || deny "idp_pubkey_missing($IDP_PUB)"
JWT_OUT="$(
python3 - "$CASAN_APPROVAL_JWT" "$IDP_PUB" "$APPROVER" "$ROLES" "$ACTION" "$ACTOR" "$INPUT_SHA" <<'PY'
import base64
import json
import os
import subprocess
import sys
import tempfile
import time
jwt, pub, approver, roles_csv, action, actor, input_sha = sys.argv[1:8]
def die(reason):
print(reason, file=sys.stderr)
sys.exit(3)
def b64u_decode(part):
try:
return base64.urlsafe_b64decode(part + "=" * (-len(part) % 4))
except Exception:
die("jwt_base64_invalid")
parts = jwt.split(".")
if len(parts) != 3:
die("jwt_shape_invalid")
header = json.loads(b64u_decode(parts[0]))
claims = json.loads(b64u_decode(parts[1]))
if header.get("alg") != "RS256":
die("jwt_alg_not_allowed")
if int(claims.get("exp", 0)) <= int(time.time()):
die("jwt_expired")
if claims.get("sub") != approver:
die("jwt_sub_mismatch")
role = claims.get("role", "")
roles = [r for r in roles_csv.split(",") if r]
if role not in roles:
die(f"jwt_role_not_authorized(role={role} allowed={roles_csv})")
if claims.get("action") != action:
die("jwt_action_mismatch")
if claims.get("actor") != actor:
die("jwt_actor_mismatch")
if claims.get("input_sha256") != input_sha:
die("jwt_input_hash_mismatch")
sig = b64u_decode(parts[2])
signing_input = ".".join(parts[:2]).encode()
with tempfile.TemporaryDirectory() as td:
sig_path = os.path.join(td, "sig.bin")
msg_path = os.path.join(td, "msg.txt")
open(sig_path, "wb").write(sig)
open(msg_path, "wb").write(signing_input)
rc = subprocess.run(
["openssl", "dgst", "-sha256", "-verify", pub, "-signature", sig_path, msg_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
).returncode
if rc != 0:
die("jwt_signature_invalid")
print(f"OK role={role}")
PY
)" || deny "oidc_jwt_invalid(${JWT_OUT:-see_stderr})"
ROLE="${JWT_OUT#OK role=}"
echo "APPROVAL_OK mechanism=oidc role=$ROLE approver=$APPROVER action=$ACTION"
exit 0
fi
[[ -n "$SIG_FILE" && -f "$SIG_FILE" ]] || deny "signature_missing"
# Reviewer lookup (first matching registered reviewer).
REV_LINE="$(awk -v id="$APPROVER" '$1=="reviewer" && $2==id {print $3" "$4; exit}' "$REVIEWERS_FILE")"
[[ -n "$REV_LINE" ]] || deny "approver_not_registered"
ROLE="${REV_LINE%% *}"
PUB_REL="${REV_LINE##* }"
# Role authorization for this action (fallback to the "default" action policy).
ROLES="$(awk -v a="$ACTION" '$1=="action" && $2==a {print $3; exit}' "$REVIEWERS_FILE")"
[[ -n "$ROLES" ]] || ROLES="$(awk '$1=="action" && $2=="default" {print $3; exit}' "$REVIEWERS_FILE")"
case ",$ROLES," in
*",$ROLE,"*) : ;;
*) deny "approver_role_not_authorized(role=$ROLE action=$ACTION allowed=$ROLES)" ;;
@@ -63,7 +139,6 @@ PUB="$PUB_REL"; [[ "$PUB" = /* ]] || PUB="$REVIEWERS_DIR/$PUB_REL"
[[ -f "$PUB" ]] || deny "approver_pubkey_missing($PUB)"
# Rebuild the exact signed assertion and verify.
INPUT_SHA="$(hash_file "$INPUT_FILE")"
MSG="casan-approval|v1|$ACTION|$ACTOR|$INPUT_SHA|$APPROVER"
TMP="$(mktemp)"; trap 'rm -f "$TMP"' EXIT
printf '%s' "$MSG" > "$TMP"
AINative_OKR_CASAN5/.specify/scripts/bash/evidence-pack-build.py
+24 -1
View File
@@ -18,6 +18,7 @@ skipped (missing evidence => not certified, with the reason recorded).
import hashlib
import json
import os
import subprocess
import sys
@@ -75,6 +76,25 @@ def main():
"judge_gate_tests": os.path.exists(os.path.join(root, ".specify/tests/phase3-judge-gate-tests.sh")),
"note": "judge-gate fail-before/fix cycle proven by phase3-judge-gate-tests.sh",
}
traceability_out = os.path.join(pack_dir, "traceability-matrix.json")
traceability_script = os.path.join(root, ".specify/scripts/bash/traceability-matrix.py")
traceability_rc = 1
if os.path.isfile(traceability_script):
traceability_rc = subprocess.run(
[
sys.executable,
traceability_script,
"--requirements",
os.path.join(root, "docs/input/okr-requirement.md"),
"--map",
os.path.join(root, ".specify/traceability-map.json"),
"--out",
traceability_out,
"--gate",
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
).returncode
# H4 security
sec_rows = read_jsonl(os.path.join(logs, "audit", "security.jsonl"))
@@ -151,6 +171,8 @@ def main():
reasons.append("unresolved_cost_spike")
if len(sec_rows) == 0:
reasons.append("h4_security_not_exercised")
if traceability_rc != 0:
reasons.append("traceability_gate_failed")
fp_ok = bool(fp) and fp.get("within_budget") is True
if fp is None:
reasons.append("benign_fp_report_missing(gate_skipped)")
@@ -162,7 +184,7 @@ def main():
run_summary = {
"run_id": run_id, "pack_version": "1.0-mvp",
"certified": certified, "certification_reasons": reasons or ["all_required_gates_passed"],
"required_gates": ["H4-security", "H5-audit-chain", "H5-telemetry", "H6-cost", "benign-fp-budget"],
"required_gates": ["H3-traceability", "H4-security", "H5-audit-chain", "H5-telemetry", "H6-cost", "benign-fp-budget"],
"harness_reports": sorted(reports.keys()),
}
with open(os.path.join(pack_dir, "run-summary.json"), "w", encoding="utf-8") as f:
@@ -182,6 +204,7 @@ def main():
f"{total_tokens} provider tokens",
f"- H2 tool audit: {reports['h2-tool-audit.json']['records']} records, "
f"chain_ok={reports['h2-tool-audit.json']['chain_ok']}",
f"- H3 traceability: ok={traceability_rc == 0}",
f"- Red-team: {reports['redteam-result.json']['vectors_defined']} vectors "
f"(block_rate={reports['redteam-result.json']['adversarial_block_rate_pct']}%)",
"",
AINative_OKR_CASAN5/.specify/scripts/bash/governance-check.sh
+7 -3
View File
@@ -92,16 +92,20 @@ if [[ "$RISK_LEVEL" == "high" ]]; then
# Approval-identity mode (V20): an env-var approver is NOT enough — the
# reviewer must cryptographically SIGN this exact request and their role must
# be authorized for the action. SoD (actor != approver) still enforced.
if [[ "$APPROVAL_DECISION" == "approve" && -n "$APPROVER" && -n "${CASAN_APPROVAL_SIG:-}" ]]; then
if [[ "$APPROVAL_DECISION" == "approve" && -n "$APPROVER" && ( -n "${CASAN_APPROVAL_SIG:-}" || -n "${CASAN_APPROVAL_JWT:-}" ) ]]; then
if [[ "$APPROVER" == "$ACTOR" ]]; then
APPROVAL_STATUS="separation_of_duties_violation"
DECISION="denied"
REASONS+=("separation-of-duties:actor-equals-approver")
else
AV_RC=0
AV_OUT="$(bash "$SCRIPT_DIR/approval-verify.sh" "$ACTION_NAME" "$ACTOR" "$INPUT_FILE" "$APPROVER" "$CASAN_APPROVAL_SIG" 2>/dev/null)" || AV_RC=$?
AV_OUT="$(bash "$SCRIPT_DIR/approval-verify.sh" "$ACTION_NAME" "$ACTOR" "$INPUT_FILE" "$APPROVER" "${CASAN_APPROVAL_SIG:-"-"}" 2>/dev/null)" || AV_RC=$?
if [[ "$AV_RC" -eq 0 ]]; then
APPROVAL_STATUS="human_approved_signed"
if printf '%s' "$AV_OUT" | grep -q "mechanism=oidc"; then
APPROVAL_STATUS="human_approved_oidc"
else
APPROVAL_STATUS="human_approved_signed"
fi
DECISION="approved"
REASONS+=("signed-approval:${AV_OUT#APPROVAL_OK }")
else
AINative_OKR_CASAN5/.specify/scripts/bash/model-call.py
+12
View File
@@ -23,6 +23,7 @@ Usage:
import argparse
import json
import os
import subprocess
import sys
import time
import urllib.request
@@ -89,6 +90,17 @@ def call_ollama(model_name, prompt, role):
host = os.environ.get("CASAN_OLLAMA_HOST", OLLAMA_HOST)
if host != OLLAMA_HOST:
fail(f"endpoint_not_allowed ollama host={host} (only {OLLAMA_HOST})")
digest_gate = os.path.join(os.path.dirname(__file__), "model-digest-check.sh")
if os.path.isfile(digest_gate):
check = subprocess.run(
["bash", digest_gate, "verify", model_name],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
if check.returncode != 0:
msg = (check.stderr or check.stdout or "model_digest_check_failed").strip()
fail(msg)
url = f"http://{host}/api/generate"
body = {
"model": model_name,
AINative_OKR_CASAN5/.specify/scripts/bash/model-digest-check.sh
+7 -2
View File
@@ -16,8 +16,9 @@ set -uo pipefail
# model-digest-check.sh verify [model] # compare live digest to the pinned one
# model-digest-check.sh show [model]
# Env: CASAN_MODEL (default ornith:9b) · CASAN_MODEL_DIGEST (override) ·
# CASAN_MODEL_DIGEST_PIN (pin file) · OLLAMA_HOST (default 127.0.0.1:11434)
# Exit: 0 match/pinned · 2 MISMATCH (swap detected) · 3 unpinned/undeterminable.
# CASAN_MODEL_DIGEST_PIN (pin file) · CASAN_MODEL_DIGEST_MODE=block|warn ·
# OLLAMA_HOST (default 127.0.0.1:11434)
# Exit: 0 match/pinned/warned · 2 MISMATCH in block mode · 3 unpinned/undeterminable.
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PROJECT_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)"
@@ -60,6 +61,10 @@ case "$CMD" in
echo "MODEL_DIGEST_OK model=$MODEL digest=${D:0:24}…"
exit 0
fi
if [[ "${CASAN_MODEL_DIGEST_MODE:-block}" == "warn" ]]; then
echo "MODEL_DIGEST_WARN model=$MODEL pinned=${P:0:24}… live=${D:0:24}… — model may have been swapped/poisoned" >&2
exit 0
fi
echo "MODEL_DIGEST_MISMATCH model=$MODEL pinned=${P:0:24}… live=${D:0:24}… — model may have been swapped/poisoned" >&2
exit 2
;;
AINative_OKR_CASAN5/.specify/scripts/bash/traceability-matrix.py
Executable
+115
View File
@@ -0,0 +1,115 @@
#!/usr/bin/env python3
"""CASAN Plan-10 traceability matrix generator/gate.
Parses FR-* requirements from docs/input/okr-requirement.md and checks each
requirement has at least one existing code file and one existing test file.
"""
import argparse
import json
import os
import re
import sys
from datetime import datetime, timezone
FR_RE = re.compile(r"\|\s*(FR-\d+)\s*\|\s*([^|]+?)\s*\|")
def project_root() -> str:
return os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
def parse_requirements(path: str):
seen = {}
with open(path, encoding="utf-8") as f:
for line in f:
match = FR_RE.search(line)
if not match:
continue
fr_id, name = match.groups()
if fr_id not in seen:
seen[fr_id] = {"id": fr_id, "name": " ".join(name.split())}
return [seen[k] for k in sorted(seen)]
def existing_files(root: str, values):
present, missing = [], []
for rel in values or []:
if os.path.isfile(os.path.join(root, rel)):
present.append(rel)
else:
missing.append(rel)
return present, missing
def main() -> int:
root = project_root()
ap = argparse.ArgumentParser()
ap.add_argument("--requirements", default=os.path.join(root, "docs/input/okr-requirement.md"))
ap.add_argument("--map", default=os.path.join(root, ".specify/traceability-map.json"))
ap.add_argument("--out", default=os.path.join(root, "docs/output/casan/traceability-matrix.json"))
ap.add_argument("--gate", action="store_true")
args = ap.parse_args()
reqs = parse_requirements(args.requirements)
with open(args.map, encoding="utf-8") as f:
mapping = json.load(f)
rows = []
failures = []
for req in reqs:
entry = mapping.get(req["id"], {})
code, missing_code = existing_files(root, entry.get("code", []))
tests, missing_tests = existing_files(root, entry.get("tests", []))
status = "PASS" if code and tests and not missing_code and not missing_tests else "FAIL"
row = {
"id": req["id"],
"name": req["name"],
"status": status,
"code": code,
"tests": tests,
"missing_code": missing_code,
"missing_tests": missing_tests,
}
rows.append(row)
if status != "PASS":
failures.append(row)
orphan_mappings = sorted(set(mapping) - {r["id"] for r in reqs})
out = {
"generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"requirements_source": os.path.relpath(args.requirements, root),
"mapping_source": os.path.relpath(args.map, root),
"summary": {
"requirements": len(reqs),
"passed": sum(1 for r in rows if r["status"] == "PASS"),
"failed": len(failures),
"orphan_mappings": orphan_mappings,
},
"matrix": rows,
}
os.makedirs(os.path.dirname(args.out), exist_ok=True)
with open(args.out, "w", encoding="utf-8") as f:
json.dump(out, f, indent=2, ensure_ascii=False)
f.write("\n")
if failures:
for row in failures:
print(
f"TRACEABILITY_FAIL {row['id']} code={len(row['code'])} tests={len(row['tests'])} "
f"missing_code={len(row['missing_code'])} missing_tests={len(row['missing_tests'])}",
file=sys.stderr,
)
if orphan_mappings:
print(f"TRACEABILITY_WARN orphan_mappings={','.join(orphan_mappings)}", file=sys.stderr)
print(
f"TRACEABILITY_MATRIX requirements={len(reqs)} pass={out['summary']['passed']} "
f"fail={len(failures)} out={args.out}"
)
if args.gate and failures:
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())