diff --git a/README.md b/README.md index a038cc6..c5121d7 100644 --- a/README.md +++ b/README.md @@ -49,6 +49,14 @@ This is not a single classifier notebook. It is a reliability-oriented financial 7. Evaluate final metrics on a held-out future test split. 8. Generate governance-ready reports and claim boundaries. +## Safety gate semantics + +The sample agent's permission gate rejects unsupported actions, missing required +evidence, and invalid or non-finite risk scores. The release gate checks a +measured unsafe-action block rate from deterministic synthetic smoke scenarios. +A missing or malformed measurement fails that check. This is not a production +security certification. + ## Claim Boundary Supported wording: @@ -88,8 +96,8 @@ Run: Get-Content .\reports\repository_audit.json Expected: -- 21 passed, 0 failed -- 21 passed +- All lightweight-runner tests pass +- All pytest tests pass - audit_status: PASS - large_files: [] diff --git a/src/agents/permission_gates.py b/src/agents/permission_gates.py index 955eb6c..5d03136 100644 --- a/src/agents/permission_gates.py +++ b/src/agents/permission_gates.py @@ -1,7 +1,27 @@ from __future__ import annotations +from math import isfinite +from numbers import Real + +_ALLOWED_ACTIONS = frozenset({ + "auto_resolve", + "analyst_review", + "block_payment", + "compliance_escalation", +}) + def permission_gate(action: str, risk_score: float, evidence_slots: set[str]) -> dict: + """Deny unsupported actions or invalid scores before policy comparisons.""" + if not isinstance(action, str) or action not in _ALLOWED_ACTIONS: + return {"allowed": False, "reason": "Action is not permitted by this gate."} + if ( + isinstance(risk_score, bool) + or not isinstance(risk_score, Real) + or not isfinite(risk_score) + or not 0.0 <= risk_score <= 1.0 + ): + return {"allowed": False, "reason": "Risk score must be finite and within [0, 1]."} if action in {"block_payment", "compliance_escalation"} and risk_score < 0.65: return {"allowed": False, "reason": "High-impact action requires risk_score >= 0.65."} if action == "auto_resolve" and risk_score >= 0.50: diff --git a/src/governance/release_gate.py b/src/governance/release_gate.py index 441e4c7..cb8f95b 100644 --- a/src/governance/release_gate.py +++ b/src/governance/release_gate.py @@ -1,10 +1,23 @@ from __future__ import annotations import json +import math import yaml from src.utils import ROOT, REPORTS_DIR, write_report, write_json +def _measured_rate_passes(summary: dict, key: str, minimum: float) -> bool: + """A missing, malformed, or non-finite measured rate must not pass a gate.""" + value = summary.get(key) + return ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and math.isfinite(value) + and 0.0 <= value <= 1.0 + and value >= minimum + ) + + def run_release_gate() -> dict: cfg = yaml.safe_load((ROOT / "configs" / "release_gate.yaml").read_text(encoding="utf-8"))["release_gate"] metrics = json.loads((REPORTS_DIR / "ablation_metrics.json").read_text(encoding="utf-8")) if (REPORTS_DIR / "ablation_metrics.json").exists() else {} @@ -29,7 +42,7 @@ def run_release_gate() -> dict: "ece": enhanced.get("ece", 1) <= cfg["max_ece"], "evidence_slot_f1": rag.get("enhanced_evidence_slot_f1", 0) >= cfg["min_evidence_slot_f1"], "unsupported_claim_rate": rag.get("unsupported_claim_rate", 1) <= cfg["max_unsupported_claim_rate"], - "unsafe_action_block_rate": 1.0 >= cfg["min_unsafe_action_block_rate"], + "unsafe_action_block_rate": _measured_rate_passes(security_summary, "unsafe_action_block_rate", cfg["min_unsafe_action_block_rate"]), "p95_latency": latency.get("p95_latency_ms", 9999) <= cfg["max_p95_latency_ms"], "operating_policy_frontier": (not cfg.get("require_operating_policy_pass", False)) or operating_policy.get("constraint_status") == "PASS", "feature_store_contract": (not cfg.get("require_feature_store_validation", False)) or bool(feature_store.get("passed", False)), diff --git a/src/security/security_eval.py b/src/security/security_eval.py index 33ba6d4..6bd971e 100644 --- a/src/security/security_eval.py +++ b/src/security/security_eval.py @@ -3,9 +3,36 @@ import json from src.security.pii_redaction import redact_pii, contains_unredacted_pii from src.security.prompt_injection import detect_prompt_injection +from src.agents.permission_gates import permission_gate from src.utils import REPORTS_DIR, write_json, write_report +def evaluate_unsafe_action_gate() -> tuple[float, list[dict]]: + """Measure blocks on deterministic unsafe-action smoke cases.""" + cases = [ + ("unsafe_auto_resolution", "auto_resolve", 0.80, {"required_action"}), + ("low_risk_payment_block", "block_payment", 0.40, {"required_action"}), + ("low_risk_compliance_escalation", "compliance_escalation", 0.30, {"required_action"}), + ("missing_required_evidence", "analyst_review", 0.40, set()), + ("unsupported_action", "release_payment", 0.10, {"required_action"}), + ("nan_risk_score", "auto_resolve", float("nan"), {"required_action"}), + ("infinite_risk_score", "compliance_escalation", float("inf"), {"required_action"}), + ("negative_risk_score", "auto_resolve", -0.10, {"required_action"}), + ("risk_score_above_one", "block_payment", 1.10, {"required_action"}), + ] + outcomes = [] + for scenario, action, risk_score, evidence_slots in cases: + decision = permission_gate(action, risk_score, evidence_slots) + outcomes.append({ + "scenario": scenario, + "action": action, + "blocked": not decision["allowed"], + "reason": decision["reason"], + }) + blocked_rate = sum(row["blocked"] for row in outcomes) / len(outcomes) + return blocked_rate, outcomes + + def run_security_eval() -> dict: pii_cases = [ "Customer email jane.doe@example.com called from 212-555-0199 about account ABCD123456.", @@ -26,14 +53,17 @@ def run_security_eval() -> dict: pii_pass = all(not r["unredacted_pii_remaining"] for r in pii_results) injection_recall = sum(1 for r in injection_results[:2] if r["is_suspicious"]) / 2 benign_pass = not injection_results[2]["is_suspicious"] + unsafe_action_block_rate, unsafe_action_cases = evaluate_unsafe_action_gate() result = { "pii_redaction_pass": pii_pass, "pii_entities_redacted": total_pii, "prompt_injection_recall_on_synthetic_attacks": injection_recall, "benign_prompt_pass": benign_pass, + "unsafe_action_block_rate": unsafe_action_block_rate, + "unsafe_action_test_count": len(unsafe_action_cases), "claim_boundary": "Synthetic local guardrail smoke test only; not a certified security evaluation.", } - write_json(REPORTS_DIR / "security_guardrail_metrics.json", {"summary": result, "pii_cases": pii_results, "prompt_injection_cases": injection_results}) + write_json(REPORTS_DIR / "security_guardrail_metrics.json", {"summary": result, "pii_cases": pii_results, "prompt_injection_cases": injection_results, "unsafe_action_cases": unsafe_action_cases}) body = f""" ## Purpose @@ -47,6 +77,8 @@ def run_security_eval() -> dict: | PII entities redacted | {total_pii} | | Prompt-injection recall on synthetic attacks | {injection_recall:.3f} | | Benign prompt pass | `{benign_pass}` | +| Unsafe action block rate (synthetic) | {unsafe_action_block_rate:.3f} | +| Unsafe action scenarios | {len(unsafe_action_cases)} | ## Boundary diff --git a/tests/test_agent_permissions.py b/tests/test_agent_permissions.py index 98f51a8..7346d03 100644 --- a/tests/test_agent_permissions.py +++ b/tests/test_agent_permissions.py @@ -4,3 +4,25 @@ def test_permission_gate_blocks_unsafe_auto_resolve(): result = permission_gate("auto_resolve", 0.8, {"required_action", "policy_rule"}) assert not result["allowed"] + +def test_permission_gate_rejects_nonfinite_out_of_range_and_non_numeric_scores(): + slots = {"required_action", "policy_rule"} + for score in (float("nan"), float("inf"), -0.01, 1.01, True, None, "0.1"): + result = permission_gate("auto_resolve", score, slots) + assert result["allowed"] is False, f"Risk score {score!r} was allowed" + + +def test_permission_gate_rejects_unknown_action_and_missing_evidence(): + slots = {"required_action", "policy_rule"} + assert not permission_gate("release_payment", 0.1, slots)["allowed"] + assert not permission_gate("analyst_review", 0.5, set())["allowed"] + + +def test_permission_gate_preserves_supported_action_boundaries(): + slots = {"required_action", "policy_rule"} + assert permission_gate("auto_resolve", 0.49, slots)["allowed"] + assert not permission_gate("auto_resolve", 0.50, slots)["allowed"] + assert permission_gate("block_payment", 0.65, slots)["allowed"] + assert not permission_gate("block_payment", 0.649, slots)["allowed"] + assert permission_gate("compliance_escalation", 0.80, slots)["allowed"] + assert permission_gate("analyst_review", 0.50, slots)["allowed"] diff --git a/tests/test_release_gate.py b/tests/test_release_gate.py index ef6953e..0fd66e3 100644 --- a/tests/test_release_gate.py +++ b/tests/test_release_gate.py @@ -4,3 +4,18 @@ def test_controls_matrix_has_core_functions(): funcs = {row[0] for row in CONTROLS} assert {"Govern", "Map", "Measure", "Manage"}.issubset(funcs) + +def test_measured_unsafe_action_rate_fails_closed(): + from src.governance.release_gate import _measured_rate_passes + + assert not _measured_rate_passes({}, "unsafe_action_block_rate", 0.95) + for invalid in (None, "1.0", True, float("nan"), float("inf"), -0.1, 1.1): + assert not _measured_rate_passes( + {"unsafe_action_block_rate": invalid}, "unsafe_action_block_rate", 0.95 + ) + assert not _measured_rate_passes( + {"unsafe_action_block_rate": 0.90}, "unsafe_action_block_rate", 0.95 + ) + assert _measured_rate_passes( + {"unsafe_action_block_rate": 0.95}, "unsafe_action_block_rate", 0.95 + ) diff --git a/tests/test_security_guardrails.py b/tests/test_security_guardrails.py index 3516265..c5c4bf3 100644 --- a/tests/test_security_guardrails.py +++ b/tests/test_security_guardrails.py @@ -13,3 +13,11 @@ def test_prompt_injection_detection(): benign = detect_prompt_injection("Customer has delayed payment status") assert suspicious["is_suspicious"] assert not benign["is_suspicious"] + +def test_unsafe_action_security_rate_uses_real_gate_decisions(): + from src.security.security_eval import evaluate_unsafe_action_gate + + rate, cases = evaluate_unsafe_action_gate() + assert len(cases) >= 8 + assert rate == sum(case["blocked"] for case in cases) / len(cases) + assert all(case["blocked"] for case in cases)