Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 52 additions & 6 deletions ovk/adapters/authorization/deterministic_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.",
],
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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=[
{
Expand All @@ -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(
Expand Down
13 changes: 13 additions & 0 deletions ovk/adapters/authorization/z3_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
159 changes: 159 additions & 0 deletions ovk/core/deterministic_evaluators.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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)
Expand Down
Loading
Loading