diff --git a/ovk/adapters/authorization/deterministic_adapter.py b/ovk/adapters/authorization/deterministic_adapter.py index ef200495..acf6b5f7 100644 --- a/ovk/adapters/authorization/deterministic_adapter.py +++ b/ovk/adapters/authorization/deterministic_adapter.py @@ -57,7 +57,7 @@ def manifest(self) -> BackendCapabilityManifest: ), input_languages=["json"], supported_domains=["authorization"], - supported_property_kinds=["access_control", "safety", "invariant"], + supported_property_kinds=["access_control", "safety", "invariant", "protected_effect_integrity"], assumptions=[ "Route reachability abstraction is supplied by the neutral compiler.", ], @@ -87,6 +87,19 @@ def can_handle( estimated_memory_mb=64, reasons=["not an authorization obligation"], ) + if obligation.property_kind not in set(self.manifest().supported_property_kinds): + return BackendCapabilityAssessment( + backend=self.backend_id, + support="unsupported", + score=0.0, + guarantee_type="deterministic_witness", + material_requirements_met=bool(obligation.materials), + coverage_requirements_met=False, + native_available=False, + estimated_wall_time_seconds=1.0, + estimated_memory_mb=64, + reasons=[f"unsupported authorization property kind: {obligation.property_kind}"], + ) denied = set(context.budget.denied_backends if context.budget else []) allowed = set(context.budget.allowed_backends) if context.budget and context.budget.allowed_backends else None if self.backend_id in denied or (allowed is not None and self.backend_id not in allowed): @@ -124,7 +137,12 @@ def compile( routing: RoutingDecision, ) -> BackendObligation: data = _authorization_input(obligation) - payload = {"input": data, "mode": "deterministic"} + payload = { + "input": data, + "mode": "deterministic", + "property_kind": obligation.property_kind, + "coverage": obligation.coverage.model_dump(mode="json"), + } provisional = BackendObligation( backend_obligation_id="pending", obligation_id=obligation.obligation_id, @@ -184,13 +202,30 @@ def normalize( status = VerificationStatus(status_text) except ValueError: status = VerificationStatus.UNKNOWN + is_protected_effect = backend_obligation.payload.get("property_kind") == "protected_effect_integrity" + assumptions = ( + [ + "Result is conditional on the declared protected sinks, authorization-call signatures, " + "principal dependencies, and complete supported source profile." + ] + if is_protected_effect + else ["Deterministic witness translation; no native SMT solver."] + ) + limits = ( + [ + "Protected-effect pass is bounded to the supported source abstraction; " + "unsupported semantics remain unknown." + ] + if is_protected_effect + else ["Weaker than z3-native smt_refutation_search."] + ) return NormalizedBackendResult( attempt_id="pending", backend=self.backend_id, status=status, guarantee_type=backend_obligation.expected_guarantee, - assumptions=["Deterministic witness translation; no native SMT solver."], - limits=["Weaker than z3-native smt_refutation_search."], + assumptions=assumptions, + limits=limits, counterexamples=list(raw.raw_result.get("counterexamples") or raw.raw_result.get("models") or []), generated_artifacts=[ { @@ -203,10 +238,21 @@ def normalize( def explain(self, result: NormalizedBackendResult) -> HumanExplanation: if result.counterexamples: + failure_mode = str(result.counterexamples[0].get("failure_mode", "admin_route_bypass")) + repair_hints = { + "missing_authorization_guard": "Add an approved authorization guard before the protected effect.", + "protected_effect_binding_mismatch": "Align the principal, effect, and resource used for authorization and execution.", + "unresolved_semantic_binding": "Make the authorization-to-effect identity binding explicit or simplify the supported path.", + "incomplete_semantic_coverage": "Resolve or model the unsupported source semantics before enforcing this guarantee.", + "missing_semantic_bindings": "Provide complete principal, effect, and resource bindings.", + } return HumanExplanation( summary=str(result.counterexamples[0].get("summary", "Authorization violation.")), - repair_hint="Restore admin-only protection on the reported route.", - failure_mode=str(result.counterexamples[0].get("failure_mode", "admin_route_bypass")), + repair_hint=repair_hints.get( + failure_mode, + "Restore admin-only protection on the reported route.", + ), + failure_mode=failure_mode, ) if result.status == VerificationStatus.PASS: return HumanExplanation( diff --git a/ovk/adapters/authorization/z3_adapter.py b/ovk/adapters/authorization/z3_adapter.py index c26a81e6..ab85defc 100644 --- a/ovk/adapters/authorization/z3_adapter.py +++ b/ovk/adapters/authorization/z3_adapter.py @@ -91,6 +91,19 @@ def can_handle( estimated_memory_mb=256, reasons=["not an authorization obligation"], ) + if obligation.property_kind not in set(self.manifest().supported_property_kinds): + return BackendCapabilityAssessment( + backend=self.backend_id, + support="unsupported", + score=0.0, + guarantee_type="smt_refutation_search", + material_requirements_met=bool(obligation.materials), + coverage_requirements_met=False, + native_available=z3_available(), + estimated_wall_time_seconds=5.0, + estimated_memory_mb=256, + reasons=[f"unsupported authorization property kind: {obligation.property_kind}"], + ) denied = set(context.budget.denied_backends if context.budget else []) allowed = set(context.budget.allowed_backends) if context.budget and context.budget.allowed_backends else None native = z3_available() diff --git a/ovk/core/deterministic_evaluators.py b/ovk/core/deterministic_evaluators.py index 6b09adec..94df7f39 100644 --- a/ovk/core/deterministic_evaluators.py +++ b/ovk/core/deterministic_evaluators.py @@ -52,6 +52,9 @@ def evaluate_deterministic(evaluator_id: str, payload: dict[str, Any]) -> dict[s def _evaluate_authorization_deterministic(payload: dict[str, Any]) -> dict[str, Any]: + if str(payload.get("property_kind") or "") == "protected_effect_integrity": + return _evaluate_protected_effect_integrity(payload) + data = dict(payload.get("input") or {}) issues = validate_authorization_input(data) if issues: @@ -84,6 +87,162 @@ def _evaluate_authorization_deterministic(payload: dict[str, Any]) -> dict[str, } +def _evaluate_protected_effect_integrity(payload: dict[str, Any]) -> dict[str, Any]: + """Evaluate the bounded protected-effect abstraction. + + PASS is available only for complete source-profile coverage, a preceding + approved guard, and equal principal/effect/resource bindings. Missing + guards are concrete violations in that supported straight-line model. + Partial coverage or unresolved bindings remain UNKNOWN. + """ + data = dict(payload.get("input") or {}) + coverage = dict(payload.get("coverage") or {}) + if data.get("kind") != "protected_effect_integrity": + return { + "termination": "invalid_output", + "exit_code": 1, + "raw_result": { + "status": "unknown", + "reason": "protected-effect abstraction missing or malformed", + "counterexamples": [ + { + "summary": "Protected-effect integrity abstraction is missing or malformed.", + "failure_mode": "invalid_protected_effect_abstraction", + } + ], + }, + } + + if coverage.get("status") != "complete": + unsupported = list(coverage.get("unsupported_constructs") or []) + return { + "termination": "completed", + "exit_code": 0, + "raw_result": { + "status": "unknown", + "reason": "source-profile coverage is not complete", + "counterexamples": [ + { + "summary": "Protected-effect integrity cannot be established with incomplete semantic coverage.", + "failure_mode": "incomplete_semantic_coverage", + "unsupported_constructs": unsupported, + } + ], + }, + } + + guard_requirement = data.get("guard_requirement") + guard_refs = ( + list(guard_requirement.get("guard_refs") or []) + if isinstance(guard_requirement, dict) + else [] + ) + if not guard_refs: + return { + "termination": "completed", + "exit_code": 0, + "raw_result": { + "status": "fail", + "reason": "protected effect has no accepted preceding authorization guard", + "counterexamples": [ + { + "summary": "Protected effect is reachable on the supported path without an accepted authorization guard.", + "failure_mode": "missing_authorization_guard", + "path_id": data.get("path_id"), + "entrypoint": data.get("entrypoint"), + } + ], + }, + } + + raw_bindings = data.get("binding_requirements") + if not isinstance(raw_bindings, list) or not raw_bindings: + return { + "termination": "completed", + "exit_code": 0, + "raw_result": { + "status": "unknown", + "reason": "binding requirements are missing", + "counterexamples": [ + { + "summary": "Principal/effect/resource binding requirements are missing.", + "failure_mode": "missing_semantic_bindings", + } + ], + }, + } + + bindings = [item for item in raw_bindings if isinstance(item, dict)] + required_kinds = {"principal", "effect", "resource"} + present_kinds = {str(item.get("kind")) for item in bindings} + missing_kinds = sorted(required_kinds - present_kinds) + if missing_kinds: + return { + "termination": "completed", + "exit_code": 0, + "raw_result": { + "status": "unknown", + "reason": "binding requirements are incomplete", + "counterexamples": [ + { + "summary": "Protected-effect binding requirements are incomplete.", + "failure_mode": "missing_semantic_bindings", + "missing_kinds": missing_kinds, + } + ], + }, + } + + distinct = [item for item in bindings if item.get("declared_relation") == "distinct"] + if distinct: + item = distinct[0] + return { + "termination": "completed", + "exit_code": 0, + "raw_result": { + "status": "fail", + "reason": "authorization and performed-effect bindings are distinct", + "counterexamples": [ + { + "summary": f"{item.get('kind', 'semantic')} authorized and performed identities are distinct.", + "failure_mode": "protected_effect_binding_mismatch", + "kind": item.get("kind"), + "left_ref": item.get("left_ref"), + "right_ref": item.get("right_ref"), + } + ], + }, + } + + unresolved = [item for item in bindings if item.get("declared_relation") != "equal"] + if unresolved: + return { + "termination": "completed", + "exit_code": 0, + "raw_result": { + "status": "unknown", + "reason": "semantic binding could not be established", + "counterexamples": [ + { + "summary": "Authorization and performed-effect identity could not be established.", + "failure_mode": "unresolved_semantic_binding", + "bindings": unresolved, + } + ], + }, + } + + return { + "termination": "completed", + "exit_code": 0, + "raw_result": { + "status": "pass", + "reason": "complete supported path has an accepted guard with equal principal/effect/resource bindings", + "counterexamples": [], + }, + } + + def _evaluate_self_protection_deterministic(payload: dict[str, Any]) -> dict[str, Any]: data = dict(payload.get("input") or {}) violations = find_self_protection_violations(data) diff --git a/tests/test_protected_effect_evaluator.py b/tests/test_protected_effect_evaluator.py new file mode 100644 index 00000000..b2c83680 --- /dev/null +++ b/tests/test_protected_effect_evaluator.py @@ -0,0 +1,187 @@ +"""End-to-end bounded protected-effect evaluator tests.""" + +from __future__ import annotations + +from types import SimpleNamespace + +from ovk.adapters.authorization.deterministic_adapter import AuthorizationDeterministicAdapter +from ovk.adapters.authorization.z3_adapter import Z3NativeAuthorizationAdapter +from ovk.compilers.authorization.fastapi_semantic import ( + AuthorizationCallSpec, + FastApiSemanticAssuranceCompiler, + FastApiSemanticConfig, + PrincipalDependencySpec, + ProtectedSinkSpec, +) +from ovk.compilers.authorization.material_loader import AuthMaterials +from ovk.core.deterministic_evaluators import evaluate_deterministic +from ovk.core.execution_models import ExecutionContext +from ovk.core.protected_effect_integrity import compile_protected_effect_integrity + + +def _config() -> FastApiSemanticConfig: + return FastApiSemanticConfig( + principal_dependencies=[PrincipalDependencySpec(dependency="current_user")], + authorization_calls=[ + AuthorizationCallSpec( + function="authorize", + principal_arg=0, + effect_arg=1, + resource_arg=2, + resource_type="invoice", + ) + ], + protected_sinks=[ + ProtectedSinkSpec( + function="issue_refund", + effect_name="billing.invoice.refund", + resource_arg=0, + resource_type="invoice", + severity="critical", + ) + ], + ) + + +def _obligation(source: str): + materials = AuthMaterials( + head_files={"app.py": source}, + repo="example/payments", + base_revision="base", + head_revision="head", + ) + ir = FastApiSemanticAssuranceCompiler(_config()).compile(materials) + return compile_protected_effect_integrity(ir)[0] + + +def _evaluate(obligation): + return evaluate_deterministic( + "authorization-deterministic", + { + "input": obligation.abstraction, + "mode": "deterministic", + "property_kind": obligation.property_kind, + "coverage": obligation.coverage.model_dump(mode="json"), + }, + ) + + +def test_complete_linear_path_is_established() -> None: + obligation = _obligation( + """ +from fastapi import APIRouter, Depends +router = APIRouter() + +@router.post("/refund") +def refund(user = Depends(current_user)): + invoice = load_invoice("x") + authorize(user, "billing.invoice.refund", invoice) + issue_refund(invoice) +""" + ) + result = _evaluate(obligation) + assert result["raw_result"]["status"] == "pass" + assert result["raw_result"]["counterexamples"] == [] + + +def test_missing_guard_is_concrete_violation_on_supported_path() -> None: + obligation = _obligation( + """ +from fastapi import APIRouter, Depends +router = APIRouter() + +@router.post("/refund") +def refund(user = Depends(current_user)): + invoice = load_invoice("x") + issue_refund(invoice) +""" + ) + result = _evaluate(obligation) + assert result["raw_result"]["status"] == "fail" + assert result["raw_result"]["counterexamples"][0]["failure_mode"] == "missing_authorization_guard" + + +def test_unresolved_resource_binding_is_unknown_not_pass() -> None: + obligation = _obligation( + """ +from fastapi import APIRouter, Depends +router = APIRouter() + +@router.post("/refund") +def refund(user = Depends(current_user)): + authorized_invoice = load_invoice("a") + performed_invoice = load_invoice("b") + authorize(user, "billing.invoice.refund", authorized_invoice) + issue_refund(performed_invoice) +""" + ) + result = _evaluate(obligation) + assert result["raw_result"]["status"] == "unknown" + assert result["raw_result"]["counterexamples"][0]["failure_mode"] == "unresolved_semantic_binding" + + +def test_partial_control_flow_is_unknown_even_with_matching_names() -> None: + obligation = _obligation( + """ +from fastapi import APIRouter, Depends +router = APIRouter() + +@router.post("/refund") +def refund(user = Depends(current_user)): + invoice = load_invoice("x") + if user.is_finance: + authorize(user, "billing.invoice.refund", invoice) + issue_refund(invoice) +""" + ) + result = _evaluate(obligation) + assert result["raw_result"]["status"] == "unknown" + assert result["raw_result"]["counterexamples"][0]["failure_mode"] == "incomplete_semantic_coverage" + + +def test_backend_routing_matches_declared_property_semantics() -> None: + obligation = _obligation( + """ +from fastapi import APIRouter, Depends +router = APIRouter() + +@router.post("/refund") +def refund(user = Depends(current_user)): + invoice = load_invoice("x") + authorize(user, "billing.invoice.refund", invoice) + issue_refund(invoice) +""" + ) + context = ExecutionContext(subject=obligation.subject) + + deterministic = AuthorizationDeterministicAdapter().can_handle(obligation, context) + z3 = Z3NativeAuthorizationAdapter().can_handle(obligation, context) + + assert deterministic.support == "supported" + assert deterministic.guarantee_type == "deterministic_witness" + assert z3.support == "unsupported" + + +def test_adapter_payload_preserves_property_and_coverage() -> None: + obligation = _obligation( + """ +from fastapi import APIRouter, Depends +router = APIRouter() + +@router.post("/refund") +def refund(user = Depends(current_user)): + invoice = load_invoice("x") + authorize(user, "billing.invoice.refund", invoice) + issue_refund(invoice) +""" + ) + adapter = AuthorizationDeterministicAdapter() + backend_obligation = adapter.compile( + obligation, + SimpleNamespace(routing_id="routing-test"), # type: ignore[arg-type] + ) + + assert "protected_effect_integrity" in adapter.manifest().supported_property_kinds + assert backend_obligation.payload["property_kind"] == "protected_effect_integrity" + assert backend_obligation.payload["coverage"]["status"] == "complete" + assert backend_obligation.payload["input"]["kind"] == "protected_effect_integrity"