diff --git a/ovk/core/protected_effect_integrity.py b/ovk/core/protected_effect_integrity.py new file mode 100644 index 00000000..4489ad39 --- /dev/null +++ b/ovk/core/protected_effect_integrity.py @@ -0,0 +1,266 @@ +"""Compile Assurance IR paths into protected-effect integrity obligations. + +This compiler is source-model preserving: it does not claim that source extraction +is sound or that a binding holds. It turns one Assurance IR semantic path into a +backend-neutral obligation that makes principal/effect/resource binding and guard +coverage explicit. +""" + +from __future__ import annotations + +import json +from typing import Any + +from ovk.core.assurance_ir import ( + AssuranceIR, + BindingConstraint, + SourceProvenance, + compute_assurance_ir_digest, +) +from ovk.core.bundle import content_digest +from ovk.core.execution_models import ( + AbstractionCoverage, + MaterialReference, + VerificationObligation, + compute_abstraction_digest, + compute_obligation_id, +) +from ovk.core.models import RiskSeverity, SourceRange + + +COMPILER_ID = "ovk.assurance_ir.protected_effect_integrity.v1" +COMPILER_VERSION = "0.1.0" + + +def _severity(value: str) -> RiskSeverity: + return RiskSeverity(value) + + +def _binding( + ir: AssuranceIR, + *, + kind: str, + left_ref: str, + right_ref: str, +) -> dict[str, Any]: + if left_ref == right_ref: + return { + "kind": kind, + "left_ref": left_ref, + "right_ref": right_ref, + "binding_source": "identity", + "binding_id": None, + "declared_relation": "equal", + "expression": None, + } + + matches: list[BindingConstraint] = [] + for item in ir.bindings: + if item.kind != kind: + continue + if {item.left_ref, item.right_ref} == {left_ref, right_ref}: + matches.append(item) + + if len(matches) == 1: + item = matches[0] + return { + "kind": kind, + "left_ref": left_ref, + "right_ref": right_ref, + "binding_source": "explicit_constraint", + "binding_id": item.binding_id, + "declared_relation": item.relation, + "expression": item.expression, + } + + return { + "kind": kind, + "left_ref": left_ref, + "right_ref": right_ref, + "binding_source": "unresolved", + "binding_id": None, + "declared_relation": "unknown", + "expression": None, + } + + +def _unique_ranges(provenance: list[SourceProvenance]) -> list[SourceRange]: + ranges: dict[tuple[Any, ...], SourceRange] = {} + for item in provenance: + for source_range in item.source_ranges: + key = ( + source_range.path, + source_range.start_line, + source_range.end_line, + source_range.start_column, + source_range.end_column, + ) + ranges[key] = source_range + return [ + ranges[key] + for key in sorted( + ranges, + key=lambda value: tuple("" if item is None else str(item) for item in value), + ) + ] + + +def _coverage(ir: AssuranceIR, provenance: list[SourceProvenance]) -> AbstractionCoverage: + statuses = [item.coverage for item in provenance] + unknowns = list(ir.unknowns) + if not provenance or "unknown" in statuses: + status = "unknown" + confidence = 0.0 + elif unknowns or "partial" in statuses: + status = "partial" + confidence = 0.5 + elif all(item == "complete" for item in statuses): + status = "complete" + confidence = 1.0 + else: + status = "partial" + confidence = 0.5 + + warnings = sorted( + { + text + for item in provenance + for text in [*item.assumptions, *item.notes] + if text + } + ) + return AbstractionCoverage( + status=status, + confidence=confidence, + extracted_elements=len(provenance), + expected_elements=len(provenance) if status == "complete" else None, + unsupported_constructs=sorted(set(unknowns)), + warnings=warnings, + source_ranges=_unique_ranges(provenance), + ) + + +def _ir_material(ir: AssuranceIR) -> MaterialReference: + digest = compute_assurance_ir_digest(ir) + payload = json.dumps( + ir.model_dump(mode="json", exclude={"ir_digest"}), + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return MaterialReference( + material_id=f"assurance-ir-{digest[:16]}", + kind="generated_harness", + uri=f"ovk-material:assurance-ir/{digest}", + sha256=digest, + size_bytes=len(payload), + source_revision=ir.subject.head_sha, + trusted=False, + ) + + +def compile_protected_effect_integrity( + ir: AssuranceIR, + *, + policy_digest: str | None = None, +) -> list[VerificationObligation]: + """Compile one obligation per semantic path ending in a protected effect.""" + + principals = {item.principal_id: item for item in ir.principals} + resources = {item.resource_id: item for item in ir.resources} + effects = {item.effect_id: item for item in ir.effects} + guards = {item.guard_id: item for item in ir.guards} + protected = {item.protected_effect_id: item for item in ir.protected_effects} + ir_digest = compute_assurance_ir_digest(ir) + effective_policy_digest = policy_digest or content_digest( + {"compiler": COMPILER_ID, "property_kind": "protected_effect_integrity"} + ) + + obligations: list[VerificationObligation] = [] + for path in ir.semantic_paths: + sink = protected[path.protected_effect_ref] + path_guards = [guards[ref] for ref in path.guard_refs] + + binding_requirements: list[dict[str, Any]] = [] + for guard in path_guards: + binding_requirements.extend( + [ + _binding( + ir, + kind="principal", + left_ref=guard.principal_ref, + right_ref=sink.principal_ref, + ), + _binding( + ir, + kind="effect", + left_ref=guard.effect_ref, + right_ref=sink.effect_ref, + ), + _binding( + ir, + kind="resource", + left_ref=guard.resource_ref, + right_ref=sink.resource_ref, + ), + ] + ) + + provenance = [ + path.provenance, + sink.provenance, + principals[sink.principal_ref].provenance, + effects[sink.effect_ref].provenance, + resources[sink.resource_ref].provenance, + *[guard.provenance for guard in path_guards], + ] + for guard in path_guards: + provenance.extend( + [ + principals[guard.principal_ref].provenance, + effects[guard.effect_ref].provenance, + resources[guard.resource_ref].provenance, + ] + ) + + abstraction = { + "kind": "protected_effect_integrity", + "source_ir_digest": ir_digest, + "path_id": path.path_id, + "entrypoint": path.entrypoint, + "protected_effect": sink.model_dump(mode="json"), + "guard_requirement": { + "meaning": "Every feasible path to the protected effect must pass an accepted authorization guard.", + "guard_refs": list(path.guard_refs), + }, + "binding_requirements": binding_requirements, + "path_conditions": [ + condition.model_dump(mode="json") + for condition in ir.path_conditions + if condition.condition_id in set(path.condition_refs) + ], + "call_chain": list(path.call_chain), + } + coverage = _coverage(ir, provenance) + provisional = VerificationObligation( + obligation_id="pending", + subject=ir.subject, + intent_id=f"protected-effect-integrity:{sink.protected_effect_id}", + intent_version="0.1.0", + lane="authorization", + property_kind="protected_effect_integrity", + severity=_severity(sink.severity), + compiler_id=COMPILER_ID, + compiler_version=COMPILER_VERSION, + materials=[_ir_material(ir)], + abstraction=abstraction, + abstraction_digest=compute_abstraction_digest(abstraction), + coverage=coverage, + acceptable_guarantees=["smt_refutation_search", "deterministic_witness"], + required_capabilities=["authorization", "protected_effect_integrity"], + policy_digest=effective_policy_digest, + ) + obligations.append( + provisional.model_copy(update={"obligation_id": compute_obligation_id(provisional)}) + ) + + return obligations diff --git a/tests/test_protected_effect_integrity.py b/tests/test_protected_effect_integrity.py new file mode 100644 index 00000000..9f49d969 --- /dev/null +++ b/tests/test_protected_effect_integrity.py @@ -0,0 +1,145 @@ +"""Protected-effect integrity obligation compilation tests.""" + +from __future__ import annotations + +from ovk.core.assurance_ir import ( + AssuranceIR, + AuthorizationGuard, + BindingConstraint, + EffectRef, + PrincipalRef, + ProtectedEffect, + ResourceRef, + SemanticPath, + SourceProvenance, +) +from ovk.core.models import SourceRange, VerificationSubject +from ovk.core.protected_effect_integrity import compile_protected_effect_integrity + + +def _subject() -> VerificationSubject: + return VerificationSubject(repo="example/payments", base_sha="base", head_sha="head") + + +def _p(path: str, *, coverage: str = "complete") -> SourceProvenance: + return SourceProvenance( + extractor_id="authorization.fastapi.semantic_v2", + extractor_version="0.1.0", + subject=_subject(), + source_ranges=[SourceRange(path=path, start_line=1, end_line=4)], + coverage=coverage, + ) + + +def _ir(*, same_resource: bool = True, unknowns: list[str] | None = None) -> AssuranceIR: + principal = PrincipalRef(principal_id="p.user", kind="human", provenance=_p("auth.py")) + effect = EffectRef(effect_id="e.refund", name="billing.invoice.refund", provenance=_p("billing.py")) + authorized = ResourceRef(resource_id="r.authorized", resource_type="invoice", provenance=_p("route.py")) + performed_id = authorized.resource_id if same_resource else "r.performed" + resources = [authorized] + if not same_resource: + resources.append(ResourceRef(resource_id=performed_id, resource_type="invoice", provenance=_p("route.py"))) + + guard = AuthorizationGuard( + guard_id="g.refund", + principal_ref=principal.principal_id, + effect_ref=effect.effect_id, + resource_ref=authorized.resource_id, + provenance=_p("route.py"), + ) + protected = ProtectedEffect( + protected_effect_id="pe.refund", + principal_ref=principal.principal_id, + effect_ref=effect.effect_id, + resource_ref=performed_id, + sink="billing.issue_refund", + severity="critical", + provenance=_p("route.py"), + ) + path = SemanticPath( + path_id="path.refund", + entrypoint="POST /refund", + protected_effect_ref=protected.protected_effect_id, + guard_refs=[guard.guard_id], + call_chain=["handler", "authorize", "issue_refund"], + provenance=_p("route.py"), + ) + bindings = [] + if not same_resource: + bindings = [ + BindingConstraint( + binding_id="b.resource", + kind="resource", + left_ref=authorized.resource_id, + right_ref=performed_id, + relation="unknown", + expression="authorized_invoice == modified_invoice", + provenance=_p("route.py"), + ) + ] + + return AssuranceIR( + subject=_subject(), + principals=[principal], + resources=resources, + effects=[effect], + guards=[guard], + protected_effects=[protected], + bindings=bindings, + semantic_paths=[path], + unknowns=unknowns or [], + ) + + +def test_compiles_one_obligation_per_semantic_path() -> None: + obligations = compile_protected_effect_integrity(_ir()) + assert len(obligations) == 1 + obligation = obligations[0] + assert obligation.property_kind == "protected_effect_integrity" + assert obligation.lane == "authorization" + assert obligation.severity.value == "critical" + assert obligation.abstraction["path_id"] == "path.refund" + assert obligation.materials[0].uri.startswith("ovk-material:assurance-ir/") + + +def test_identity_bindings_are_explicit() -> None: + obligation = compile_protected_effect_integrity(_ir())[0] + requirements = obligation.abstraction["binding_requirements"] + assert {item["kind"] for item in requirements} == {"principal", "effect", "resource"} + assert all(item["binding_source"] == "identity" for item in requirements) + assert all(item["declared_relation"] == "equal" for item in requirements) + + +def test_nonidentical_resource_uses_explicit_binding_constraint() -> None: + obligation = compile_protected_effect_integrity(_ir(same_resource=False))[0] + resource = next( + item for item in obligation.abstraction["binding_requirements"] if item["kind"] == "resource" + ) + assert resource["binding_source"] == "explicit_constraint" + assert resource["binding_id"] == "b.resource" + assert resource["declared_relation"] == "unknown" + assert resource["expression"] == "authorized_invoice == modified_invoice" + + +def test_ir_unknowns_downgrade_coverage() -> None: + obligation = compile_protected_effect_integrity(_ir(unknowns=["dynamic_resource_lookup"]))[0] + assert obligation.coverage.status == "partial" + assert "dynamic_resource_lookup" in obligation.coverage.unsupported_constructs + + +def test_obligation_identity_is_deterministic() -> None: + first = compile_protected_effect_integrity(_ir())[0] + second = compile_protected_effect_integrity(_ir())[0] + assert first.obligation_id == second.obligation_id + assert first.abstraction_digest == second.abstraction_digest + + +def test_semantic_change_changes_obligation_identity() -> None: + base = compile_protected_effect_integrity(_ir())[0] + changed_ir = _ir() + path = changed_ir.semantic_paths[0].model_copy( + update={"call_chain": ["handler", "authorize", "legacy_adapter", "issue_refund"]} + ) + changed_ir = changed_ir.model_copy(update={"semantic_paths": [path]}) + changed = compile_protected_effect_integrity(changed_ir)[0] + assert base.obligation_id != changed.obligation_id