From 590a3a38d6bd4495a11d89b1f9151bec8c9d80a2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mat=C3=A9o=20H=2E=20Petel?= <113530345+fraware@users.noreply.github.com> Date: Sat, 26 Sep 2026 01:35:41 -0700 Subject: [PATCH 1/3] feat: add typed Assurance IR v1 --- ovk/core/assurance_ir.py | 259 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 259 insertions(+) create mode 100644 ovk/core/assurance_ir.py diff --git a/ovk/core/assurance_ir.py b/ovk/core/assurance_ir.py new file mode 100644 index 00000000..d8fb4c80 --- /dev/null +++ b/ovk/core/assurance_ir.py @@ -0,0 +1,259 @@ +"""Typed semantic interchange between source extractors and OVK obligations. + +A valid Assurance IR is not verification evidence. It records a source-grounded +semantic model, its provenance, coverage, assumptions, and unresolved semantics. +""" + +from __future__ import annotations + +import re +from typing import Any, Literal + +from pydantic import BaseModel, Field, field_validator, model_validator + +from ovk.core.bundle import content_digest +from ovk.core.models import SourceRange, VerificationSubject + + +CoverageStatus = Literal["complete", "partial", "unknown", "inapplicable"] +PrincipalKind = Literal["human", "service", "agent", "anonymous", "unknown"] +ClaimOrigin = Literal["human", "repository", "policy", "imported", "ai_candidate"] +ApprovalStatus = Literal["approved", "candidate", "rejected", "unknown"] +BindingKind = Literal["principal", "effect", "resource"] +BindingRelation = Literal["equal", "derived_equal", "distinct", "unknown"] + +_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]*$") +_EFFECT_RE = re.compile(r"^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)+$") + + +def _id(value: str) -> str: + value = value.strip() + if not _ID_RE.fullmatch(value): + raise ValueError("semantic ids must be non-empty stable identifiers") + return value + + +class SourceProvenance(BaseModel): + extractor_id: str + extractor_version: str + subject: VerificationSubject + source_ranges: list[SourceRange] = Field(default_factory=list) + coverage: CoverageStatus = "unknown" + assumptions: list[str] = Field(default_factory=list) + notes: list[str] = Field(default_factory=list) + + +class PrincipalRef(BaseModel): + principal_id: str + kind: PrincipalKind = "unknown" + expression: str | None = None + attributes: dict[str, Any] = Field(default_factory=dict) + provenance: SourceProvenance + + _validate_id = field_validator("principal_id")(_id) + + +class ResourceRef(BaseModel): + resource_id: str + resource_type: str + expression: str | None = None + tenant_expression: str | None = None + attributes: dict[str, Any] = Field(default_factory=dict) + provenance: SourceProvenance + + _validate_id = field_validator("resource_id")(_id) + + +class EffectRef(BaseModel): + effect_id: str + name: str + operation: str | None = None + attributes: dict[str, Any] = Field(default_factory=dict) + provenance: SourceProvenance + + _validate_id = field_validator("effect_id")(_id) + + @field_validator("name") + @classmethod + def namespaced_effect(cls, value: str) -> str: + value = value.strip() + if not _EFFECT_RE.fullmatch(value): + raise ValueError("effect name must be namespaced, for example billing.invoice.refund") + return value + + +class AuthorizationGuard(BaseModel): + guard_id: str + principal_ref: str + effect_ref: str + resource_ref: str + decision_expression: str | None = None + policy_ref: str | None = None + provenance: SourceProvenance + + _validate_ids = field_validator("guard_id", "principal_ref", "effect_ref", "resource_ref")(_id) + + +class ProtectedEffect(BaseModel): + protected_effect_id: str + principal_ref: str + effect_ref: str + resource_ref: str + sink: str + severity: Literal["low", "medium", "high", "critical"] = "high" + provenance: SourceProvenance + + _validate_ids = field_validator("protected_effect_id", "principal_ref", "effect_ref", "resource_ref")(_id) + + +class BindingConstraint(BaseModel): + binding_id: str + kind: BindingKind + left_ref: str + right_ref: str + relation: BindingRelation = "unknown" + expression: str | None = None + provenance: SourceProvenance + + _validate_ids = field_validator("binding_id", "left_ref", "right_ref")(_id) + + +class PathCondition(BaseModel): + condition_id: str + expression: str + provenance: SourceProvenance + + _validate_id = field_validator("condition_id")(_id) + + +class SemanticPath(BaseModel): + path_id: str + entrypoint: str + protected_effect_ref: str + guard_refs: list[str] = Field(default_factory=list) + condition_refs: list[str] = Field(default_factory=list) + call_chain: list[str] = Field(default_factory=list) + provenance: SourceProvenance + + _validate_ids = field_validator("path_id", "protected_effect_ref")(_id) + + +class AssuranceClaim(BaseModel): + claim_id: str + property_kind: str + statement: str + origin: ClaimOrigin + approval_status: ApprovalStatus = "candidate" + semantic_refs: list[str] = Field(default_factory=list) + acceptable_guarantees: list[str] = Field(default_factory=list) + assumptions: list[str] = Field(default_factory=list) + provenance: SourceProvenance + + _validate_id = field_validator("claim_id")(_id) + + +class AssuranceIR(BaseModel): + schema_version: Literal["ovk.assurance_ir.v1"] = "ovk.assurance_ir.v1" + subject: VerificationSubject + principals: list[PrincipalRef] = Field(default_factory=list) + resources: list[ResourceRef] = Field(default_factory=list) + effects: list[EffectRef] = Field(default_factory=list) + guards: list[AuthorizationGuard] = Field(default_factory=list) + protected_effects: list[ProtectedEffect] = Field(default_factory=list) + bindings: list[BindingConstraint] = Field(default_factory=list) + path_conditions: list[PathCondition] = Field(default_factory=list) + semantic_paths: list[SemanticPath] = Field(default_factory=list) + claims: list[AssuranceClaim] = Field(default_factory=list) + assumptions: list[str] = Field(default_factory=list) + unknowns: list[str] = Field(default_factory=list) + ir_digest: str | None = None + + @model_validator(mode="after") + def references_are_closed(self) -> "AssuranceIR": + typed = { + "principal": {item.principal_id for item in self.principals}, + "resource": {item.resource_id for item in self.resources}, + "effect": {item.effect_id for item in self.effects}, + } + id_lists = { + "principal": [item.principal_id for item in self.principals], + "resource": [item.resource_id for item in self.resources], + "effect": [item.effect_id for item in self.effects], + "guard": [item.guard_id for item in self.guards], + "protected_effect": [item.protected_effect_id for item in self.protected_effects], + "binding": [item.binding_id for item in self.bindings], + "condition": [item.condition_id for item in self.path_conditions], + "path": [item.path_id for item in self.semantic_paths], + "claim": [item.claim_id for item in self.claims], + } + for kind, values in id_lists.items(): + duplicates = sorted({value for value in values if values.count(value) > 1}) + if duplicates: + raise ValueError(f"duplicate {kind} ids: {', '.join(duplicates)}") + + guards = set(id_lists["guard"]) + protected = set(id_lists["protected_effect"]) + conditions = set(id_lists["condition"]) + + for guard in self.guards: + self._require(typed["principal"], guard.principal_ref, guard.guard_id, "principal") + self._require(typed["effect"], guard.effect_ref, guard.guard_id, "effect") + self._require(typed["resource"], guard.resource_ref, guard.guard_id, "resource") + + for item in self.protected_effects: + self._require(typed["principal"], item.principal_ref, item.protected_effect_id, "principal") + self._require(typed["effect"], item.effect_ref, item.protected_effect_id, "effect") + self._require(typed["resource"], item.resource_ref, item.protected_effect_id, "resource") + + for binding in self.bindings: + self._require(typed[binding.kind], binding.left_ref, binding.binding_id, binding.kind) + self._require(typed[binding.kind], binding.right_ref, binding.binding_id, binding.kind) + + for path in self.semantic_paths: + self._require(protected, path.protected_effect_ref, path.path_id, "protected effect") + for ref in path.guard_refs: + self._require(guards, ref, path.path_id, "guard") + for ref in path.condition_refs: + self._require(conditions, ref, path.path_id, "condition") + + all_ids = set().union(*[set(values) for values in id_lists.values()]) + for claim in self.claims: + for ref in claim.semantic_refs: + self._require(all_ids, ref, claim.claim_id, "semantic object") + + if self.ir_digest is not None and self.ir_digest != compute_assurance_ir_digest(self): + raise ValueError("ir_digest does not match canonical Assurance IR contents") + return self + + @staticmethod + def _require(known: set[str], ref: str, owner: str, kind: str) -> None: + if ref not in known: + raise ValueError(f"{owner} references unknown {kind} {ref}") + + +def assurance_ir_digest_input(ir: AssuranceIR) -> dict[str, Any]: + payload = ir.model_dump(mode="json", exclude={"ir_digest"}) + sort_keys = { + "principals": "principal_id", + "resources": "resource_id", + "effects": "effect_id", + "guards": "guard_id", + "protected_effects": "protected_effect_id", + "bindings": "binding_id", + "path_conditions": "condition_id", + "semantic_paths": "path_id", + "claims": "claim_id", + } + for field, key in sort_keys.items(): + payload[field] = sorted(payload[field], key=lambda item: str(item[key])) + payload["assumptions"] = sorted(payload["assumptions"]) + payload["unknowns"] = sorted(payload["unknowns"]) + return payload + + +def compute_assurance_ir_digest(ir: AssuranceIR) -> str: + return content_digest(assurance_ir_digest_input(ir)) + + +def seal_assurance_ir(ir: AssuranceIR) -> AssuranceIR: + return ir.model_copy(update={"ir_digest": compute_assurance_ir_digest(ir)}) From b396a5163fe7db57d1fe29cfc13ede7ecbd1ed3b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mat=C3=A9o=20H=2E=20Petel?= <113530345+fraware@users.noreply.github.com> Date: Sat, 26 Sep 2026 01:36:03 -0700 Subject: [PATCH 2/3] test: cover Assurance IR identity and references --- tests/test_assurance_ir.py | 189 +++++++++++++++++++++++++++++++++++++ 1 file changed, 189 insertions(+) create mode 100644 tests/test_assurance_ir.py diff --git a/tests/test_assurance_ir.py b/tests/test_assurance_ir.py new file mode 100644 index 00000000..824e82f2 --- /dev/null +++ b/tests/test_assurance_ir.py @@ -0,0 +1,189 @@ +"""Assurance IR identity and reference-integrity tests.""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from ovk.core.assurance_ir import ( + AssuranceClaim, + AssuranceIR, + AuthorizationGuard, + BindingConstraint, + EffectRef, + PrincipalRef, + ProtectedEffect, + ResourceRef, + SemanticPath, + SourceProvenance, + compute_assurance_ir_digest, + seal_assurance_ir, +) +from ovk.core.models import SourceRange, VerificationSubject + + +def _subject() -> VerificationSubject: + return VerificationSubject(repo="example/payments", base_sha="base123", head_sha="head456", pull_request=17) + + +def _provenance(path: str, start: int = 1, end: int = 4) -> SourceProvenance: + return SourceProvenance( + extractor_id="authorization.fastapi.semantic_v2", + extractor_version="0.1.0", + subject=_subject(), + source_ranges=[SourceRange(path=path, start_line=start, end_line=end)], + coverage="complete", + ) + + +def _ir() -> AssuranceIR: + principal = PrincipalRef( + principal_id="principal.current_user", + kind="human", + expression="current_user", + provenance=_provenance("app/auth.py"), + ) + authorized_resource = ResourceRef( + resource_id="resource.invoice.authorized", + resource_type="invoice", + expression="invoice", + provenance=_provenance("app/routes/refund.py", 10, 11), + ) + performed_resource = ResourceRef( + resource_id="resource.invoice.performed", + resource_type="invoice", + expression="invoice", + provenance=_provenance("app/routes/refund.py", 12, 13), + ) + effect = EffectRef( + effect_id="effect.invoice.refund", + name="billing.invoice.refund", + operation="refund", + provenance=_provenance("app/routes/refund.py", 8, 13), + ) + guard = AuthorizationGuard( + guard_id="guard.refund", + principal_ref=principal.principal_id, + effect_ref=effect.effect_id, + resource_ref=authorized_resource.resource_id, + decision_expression='authorize(user, "refund", invoice)', + provenance=_provenance("app/routes/refund.py", 10, 10), + ) + protected = ProtectedEffect( + protected_effect_id="protected.refund", + principal_ref=principal.principal_id, + effect_ref=effect.effect_id, + resource_ref=performed_resource.resource_id, + sink="billing.issue_refund", + severity="critical", + provenance=_provenance("app/routes/refund.py", 12, 12), + ) + binding = BindingConstraint( + binding_id="binding.refund.resource", + kind="resource", + left_ref=authorized_resource.resource_id, + right_ref=performed_resource.resource_id, + relation="equal", + provenance=_provenance("app/routes/refund.py", 10, 12), + ) + path = SemanticPath( + path_id="path.refund", + entrypoint="POST /invoices/{invoice_id}/refund", + protected_effect_ref=protected.protected_effect_id, + guard_refs=[guard.guard_id], + call_chain=["refund", "authorize", "issue_refund"], + provenance=_provenance("app/routes/refund.py", 7, 13), + ) + claim = AssuranceClaim( + claim_id="claim.refund.authorization", + property_kind="protected_effect_integrity", + statement="Refunds must use the same principal, effect, and invoice resource that were authorized.", + origin="human", + approval_status="approved", + semantic_refs=[ + principal.principal_id, + effect.effect_id, + authorized_resource.resource_id, + performed_resource.resource_id, + guard.guard_id, + protected.protected_effect_id, + binding.binding_id, + path.path_id, + ], + acceptable_guarantees=["smt_refutation_search"], + provenance=_provenance(".verification/intents/refund.yml"), + ) + return AssuranceIR( + subject=_subject(), + principals=[principal], + resources=[authorized_resource, performed_resource], + effects=[effect], + guards=[guard], + protected_effects=[protected], + bindings=[binding], + semantic_paths=[path], + claims=[claim], + assumptions=["FastAPI dependency injection follows the supported source profile."], + ) + + +def test_assurance_ir_round_trip() -> None: + ir = _ir() + payload = ir.model_dump(mode="json") + assert AssuranceIR.model_validate(payload).model_dump(mode="json") == payload + + +def test_digest_is_order_insensitive_for_set_like_collections() -> None: + ir = _ir() + assert compute_assurance_ir_digest(ir) == compute_assurance_ir_digest( + ir.model_copy(update={"resources": list(reversed(ir.resources))}) + ) + + +def test_digest_preserves_semantic_path_call_order() -> None: + ir = _ir() + path = ir.semantic_paths[0] + changed = ir.model_copy( + update={"semantic_paths": [path.model_copy(update={"call_chain": list(reversed(path.call_chain))})]} + ) + assert compute_assurance_ir_digest(ir) != compute_assurance_ir_digest(changed) + + +def test_sealed_ir_rejects_semantic_tampering() -> None: + sealed = seal_assurance_ir(_ir()) + payload = sealed.model_dump(mode="json") + payload["effects"][0]["name"] = "billing.invoice.delete" + with pytest.raises(ValidationError, match="ir_digest"): + AssuranceIR.model_validate(payload) + + +def test_effect_name_must_be_namespaced() -> None: + with pytest.raises(ValidationError, match="namespaced"): + EffectRef(effect_id="effect.refund", name="refund", provenance=_provenance("app/routes/refund.py")) + + +def test_unknown_guard_reference_is_rejected() -> None: + ir = _ir() + path = ir.semantic_paths[0].model_copy(update={"guard_refs": ["guard.missing"]}) + with pytest.raises(ValidationError, match="unknown guard"): + AssuranceIR.model_validate({**ir.model_dump(mode="json"), "semantic_paths": [path.model_dump(mode="json")]}) + + +def test_binding_reference_must_match_binding_kind() -> None: + ir = _ir() + invalid = BindingConstraint( + binding_id="binding.invalid", + kind="resource", + left_ref=ir.principals[0].principal_id, + right_ref=ir.resources[0].resource_id, + provenance=_provenance("app/routes/refund.py"), + ) + with pytest.raises(ValidationError, match="unknown resource"): + AssuranceIR.model_validate({**ir.model_dump(mode="json"), "bindings": [invalid.model_dump(mode="json")]}) + + +def test_claim_reference_must_exist() -> None: + ir = _ir() + claim = ir.claims[0].model_copy(update={"semantic_refs": ["missing.semantic.object"]}) + with pytest.raises(ValidationError, match="unknown semantic object"): + AssuranceIR.model_validate({**ir.model_dump(mode="json"), "claims": [claim.model_dump(mode="json")]}) From 7ed56e3b8781fcafeef9e7b406cc6d6e4885c34d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mat=C3=A9o=20H=2E=20Petel?= <113530345+fraware@users.noreply.github.com> Date: Sat, 26 Sep 2026 01:37:00 -0700 Subject: [PATCH 3/3] docs: define Assurance IR trust boundary --- docs/ASSURANCE_IR.md | 88 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 88 insertions(+) create mode 100644 docs/ASSURANCE_IR.md diff --git a/docs/ASSURANCE_IR.md b/docs/ASSURANCE_IR.md new file mode 100644 index 00000000..57805e53 --- /dev/null +++ b/docs/ASSURANCE_IR.md @@ -0,0 +1,88 @@ +# Assurance IR v1 + +Assurance IR is OVK's typed semantic interchange layer between source/framework extraction and verification obligations. + +It is deliberately not verification evidence. A valid IR says that an extractor produced a well-formed semantic model with explicit provenance, coverage, assumptions, and unknowns. It does not say the source program satisfies any property. + +## Pipeline position + +~~~text +Repository change + | + v +Source/framework extractor + | + v +Assurance IR + | + v +Verification obligations + | + v +Backend routing and execution + | + v +Evidence / decision / attestation +~~~ + +The layer is additive in v1. Existing lanes do not consume it yet. + +## Core objects + +PrincipalRef identifies a human, service, agent, anonymous actor, or unresolved principal. + +ResourceRef identifies the object acted upon. Resource identity may include a tenant expression. + +EffectRef names a semantic effect such as billing.invoice.refund, identity.user.delete, or network.egress. + +AuthorizationGuard relates a principal, effect, and resource at an authorization decision. + +ProtectedEffect identifies a security-sensitive effect performed by the application. + +BindingConstraint records a semantic relationship between two principals, effects, or resources. This is the basis for later principal/effect/resource binding obligations. + +PathCondition records a source-derived condition relevant to a semantic path. + +SemanticPath records an ordered application path from an entry point to a protected effect, including relevant guards and path conditions. + +AssuranceClaim records a durable candidate or approved semantic claim. Claims carry origin and approval status and remain distinct from verification evidence. + +## Provenance and uncertainty + +Every source-derived object carries SourceProvenance: + +- extractor identity and version; +- repository/base/head identity; +- source ranges; +- abstraction coverage; +- assumptions; +- notes. + +Incomplete extraction must be represented as partial or unknown. Unsupported semantics must not silently disappear from the model. + +## Identity + +compute_assurance_ir_digest content-addresses the semantic snapshot. Set-like collections are sorted by stable IDs. Ordered call chains retain their order because that order may be semantically relevant. + +seal_assurance_ir adds the computed digest to the IR. Parsing a sealed IR with a mismatched digest fails validation. + +## Trust boundary + +Assurance IR v1 does not prove that: + +- the extractor is sound; +- the extractor found every path; +- a guard dominates a protected effect; +- an authorization decision refers to the same principal/effect/resource as the performed operation; +- any backend proof is valid. + +Those are separate obligations. + +The intended next property family is protected-effect integrity: + +~~~text +Performed(principal, effect, resource) + -> Authorized(principal, effect, resource) +~~~ + +Principal binding, effect binding, resource binding, and guard/path obligations should remain independently visible so that unknown remains explicit.