diff --git a/implementations/python/packages/raes_runtime/backend_calls.py b/implementations/python/packages/raes_runtime/backend_calls.py index 1644c6e9b..c10900353 100644 --- a/implementations/python/packages/raes_runtime/backend_calls.py +++ b/implementations/python/packages/raes_runtime/backend_calls.py @@ -22,10 +22,12 @@ sanitize_account_credential_result, value_free_backend_diagnostics, ) +from .backend_plan_conformance import plan_conformance_diagnostics from .backend_realization_authority import ( _apply_authority_diagnostics, _bind_submitted_plan, _RealizationApplyContext, + _supplemental_realization_requirements, ) from .diagnostics import _failure_diagnostic from .evaluation_result_contracts import evaluation_result_contract_diagnostics @@ -183,6 +185,8 @@ def _post_apply_contract_result( baseline_snapshot, information_state_context_resolver=information_state_context_resolver, ) + if not diagnostics and realization.plan is not None: + diagnostics = plan_conformance_diagnostics(result, baseline_snapshot, realization.plan) provenance: tuple[RealizationProvenanceEntry, ...] = () if not diagnostics and realization.plan is not None: diagnostics, provenance = realization_authority_disclosure( @@ -263,23 +267,6 @@ def _sanitize_backend_realization( return sanitized -def _supplemental_realization_requirements( - realization: _RealizationApplyContext, -) -> tuple[CompiledRealizationRequirement, ...]: - """Keep non-registry contracts while the plan owns registry concerns.""" - - if realization.plan is None or not realization.plan.realization_authority: - return realization.requirements - plan_identities = { - (entry.address, entry.field_path, entry.requirement_kind) for entry in realization.plan.realization_authority - } - return tuple( - requirement - for requirement in realization.requirements - if (requirement.address, requirement.field_path, requirement.requirement_kind) not in plan_identities - ) - - def _with_snapshot( result: ApplyResult, snapshot: RuntimeSnapshot, diff --git a/implementations/python/packages/raes_runtime/backend_plan_conformance.py b/implementations/python/packages/raes_runtime/backend_plan_conformance.py new file mode 100644 index 000000000..8ddb1ffcb --- /dev/null +++ b/implementations/python/packages/raes_runtime/backend_plan_conformance.py @@ -0,0 +1,98 @@ +"""Plan-conformance gate for a provisioning backend's realized snapshot. + +The snapshot-contract checks in :mod:`raes_runtime.backend_calls` validate a +returned snapshot against its own shape and the baseline it evolved from; this +module closes the issue-158 falsification matrix by additionally holding the +result to the *submitted plan*. A backend claim is refused when it realizes an +address the plan never authorized, relabels a planned resource's type, files a +planned resource under a different domain, or mutates state it does not +disclose in ``changed_addresses``. +""" + +from __future__ import annotations + +from raes_contracts.diagnostics import Diagnostic, Severity +from raes_contracts.planning import ChangeAction, ProvisioningPlan, RuntimeDomain +from raes_contracts.runtime_state import ApplyResult, RuntimeSnapshot + +_BACKEND_CONTRACT_INVALID = "runtime.backend-contract-invalid" + + +def _conformance_diagnostic(address: str, message: str) -> Diagnostic: + return Diagnostic( + code=_BACKEND_CONTRACT_INVALID, + domain="runtime", + address=address, + message=message, + severity=Severity.ERROR, + ) + + +def _entry_conformance_message( + entry: object, + operation: object, + expected_domain: RuntimeDomain, +) -> str | None: + if entry.resource_type != operation.resource_type: + return ( + f"Backend misreported the resource type at '{entry.address}': the plan requires " + f"'{operation.resource_type}' but the backend claims '{entry.resource_type}'." + ) + if entry.domain is not expected_domain: + claimed = getattr(entry.domain, "value", entry.domain) + return ( + f"Backend filed '{entry.address}' under domain '{claimed}'; the provisioning plan " + f"authorizes '{expected_domain.value}'." + ) + return None + + +def _expected_domain(plan: ProvisioningPlan, address: str) -> RuntimeDomain: + planned_resource = plan.resources.get(address) + return planned_resource.domain if planned_resource is not None else RuntimeDomain.PROVISIONING + + +def plan_conformance_diagnostics( + result: ApplyResult, + baseline_snapshot: RuntimeSnapshot, + plan: ProvisioningPlan, +) -> list[Diagnostic]: + """Refuse realized entries the submitted provisioning plan never authorized. + + Applies to successful claims only: a failed apply already returns a + non-authoritative result, and its partial snapshot is preserved for + forensics rather than being replaced by the baseline. + """ + + if not result.success: + return [] + operations = {operation.address: operation for operation in plan.operations} + diagnostics: list[Diagnostic] = [] + disclosed = set(result.changed_addresses) + for address, entry in result.snapshot.entries.items(): + baseline_entry = baseline_snapshot.entries.get(address) + operation = operations.get(address) + changed = baseline_entry is None or entry != baseline_entry + if operation is None: + if changed: + verb = "realized" if baseline_entry is None else "mutated" + diagnostics.append( + _conformance_diagnostic( + address, + f"Backend {verb} '{address}', an address absent from the provisioning plan.", + ) + ) + continue + if operation.action is ChangeAction.DELETE: + continue + message = _entry_conformance_message(entry, operation, _expected_domain(plan, address)) + if message is not None: + diagnostics.append(_conformance_diagnostic(address, message)) + elif changed and address not in disclosed and operation.action is not ChangeAction.UNCHANGED: + diagnostics.append( + _conformance_diagnostic( + address, + f"Backend changed '{address}' without disclosing it in changed_addresses.", + ) + ) + return diagnostics diff --git a/implementations/python/packages/raes_runtime/backend_realization_authority.py b/implementations/python/packages/raes_runtime/backend_realization_authority.py index d8263c51e..d7d35cda7 100644 --- a/implementations/python/packages/raes_runtime/backend_realization_authority.py +++ b/implementations/python/packages/raes_runtime/backend_realization_authority.py @@ -70,3 +70,20 @@ def _bind_submitted_plan( __all__ = ["_RealizationApplyContext", "_apply_authority_diagnostics", "_bind_submitted_plan"] + + +def _supplemental_realization_requirements( + realization: _RealizationApplyContext, +) -> tuple[CompiledRealizationRequirement, ...]: + """Keep non-registry contracts while the plan owns registry concerns.""" + + if realization.plan is None or not realization.plan.realization_authority: + return realization.requirements + plan_identities = { + (entry.address, entry.field_path, entry.requirement_kind) for entry in realization.plan.realization_authority + } + return tuple( + requirement + for requirement in realization.requirements + if (requirement.address, requirement.field_path, requirement.requirement_kind) not in plan_identities + ) diff --git a/implementations/python/tests/test_issue_158_backend_falsification_matrix.py b/implementations/python/tests/test_issue_158_backend_falsification_matrix.py new file mode 100644 index 000000000..b31566e90 --- /dev/null +++ b/implementations/python/tests/test_issue_158_backend_falsification_matrix.py @@ -0,0 +1,189 @@ +"""Does the backend boundary falsify dishonest realization claims? (issue #158) + +Issue #158 is a falsification protocol, not an implementation design. Its claim +under test: *backend agnosticism is credible only if conformance can reject +dishonest, incomplete, or overbroad backend claims instead of merely validating +happy-path stubs.* + +Method. Each case is the **real** `ReferenceProvisioner` perturbed by exactly one +lie, so a rejection is attributable to that lie rather than to an incomplete +hand-rolled double. `test_honest_baseline_is_accepted` is the control: without a +perturbation the same construction is admitted, which is what makes the other +results meaningful. + +Every fabrication below is refused at the boundary: the runtime returns the +untouched baseline snapshot with ``runtime.backend-contract-invalid``, so no lie +survives into the authoritative snapshot. "Realizes nothing" is refused by the +SEM-218 realization gate; the other four are refused by the plan-conformance +gate (``raes_runtime.backend_plan_conformance``), which holds the returned +snapshot to the submitted plan's addresses, resource types, domains, and +disclosed changes. + +These probes bound the runtime backend boundary (`RuntimeManager.apply`) only. +`run_target_conformance` is not used as the discriminator because with default +arguments it also fails the honest reference backend, for the reason recorded in +issue #663: its default scenario is not universally realizable. +""" + +from __future__ import annotations + +import dataclasses + +from raes import parse_sdl +from raes_contracts.planning import RuntimeDomain +from raes_contracts.runtime_state import ApplyResult +from raes_reference_backend import create_reference_backend_target +from raes_reference_backend.provisioner import ReferenceProvisioner +from raes_runtime.manager import RuntimeManager + +_SCENARIO = """ +name: backend-falsification +nodes: + lab: {type: switch} + web: {type: compute, resources: {ram: 1 gib, cpu: 1}} +infrastructure: + lab: {count: 1, properties: {cidr: 10.0.0.0/24, gateway: 10.0.0.1}} + web: {count: 1, links: [lab]} +""" + +_SMUGGLED_ADDRESS = "provision.node.smuggled" + + +class _RealizesNothing(ReferenceProvisioner): + """Reports success and every address as changed, returning the predecessor.""" + + def apply(self, plan, snapshot): + super().apply(plan, snapshot) + return ApplyResult( + success=True, + snapshot=snapshot, + changed_addresses=[operation.address for operation in plan.operations], + ) + + +class _InventsResource(ReferenceProvisioner): + """Realizes the plan, then adds a resource the scenario never authored.""" + + def apply(self, plan, snapshot): + result = super().apply(plan, snapshot) + if not result.success: + return result + entries = dict(result.snapshot.entries) + entries[_SMUGGLED_ADDRESS] = dataclasses.replace(next(iter(entries.values())), address=_SMUGGLED_ADDRESS) + return dataclasses.replace( + result, + snapshot=result.snapshot.with_entries(entries), + changed_addresses=[*result.changed_addresses, _SMUGGLED_ADDRESS], + ) + + +class _SubstitutesResourceType(ReferenceProvisioner): + """Realizes the planned addresses but relabels a node as a network.""" + + def apply(self, plan, snapshot): + result = super().apply(plan, snapshot) + if not result.success: + return result + entries = dict(result.snapshot.entries) + for address, entry in entries.items(): + if entry.resource_type == "node": + entries[address] = dataclasses.replace(entry, resource_type="network") + break + return dataclasses.replace(result, snapshot=result.snapshot.with_entries(entries)) + + +class _UnderReportsChanges(ReferenceProvisioner): + """Mutates the snapshot but reports no changed addresses.""" + + def apply(self, plan, snapshot): + result = super().apply(plan, snapshot) + if not result.success: + return result + return dataclasses.replace(result, changed_addresses=[]) + + +class _ForgesDomain(ReferenceProvisioner): + """Files provisioning results under the evaluation domain.""" + + def apply(self, plan, snapshot): + result = super().apply(plan, snapshot) + if not result.success: + return result + entries = { + address: dataclasses.replace(entry, domain=RuntimeDomain.EVALUATION) + for address, entry in result.snapshot.entries.items() + } + return dataclasses.replace(result, snapshot=result.snapshot.with_entries(entries)) + + +def _apply_with(provisioner_class=None): + """Apply the scenario through the reference target, optionally perturbed. + + The perturbed provisioner is built from the honest one's driver and + realization envelope, so the only difference from the control is the lie. + """ + + target = create_reference_backend_target() + if provisioner_class is not None: + honest = target.provisioner + target = dataclasses.replace( + target, + provisioner=provisioner_class( + honest._driver, # noqa: SLF001 - perturbing the real provisioner is the method + realization_envelope=honest._realization_envelope, # noqa: SLF001 + ), + ) + manager = RuntimeManager(target) + return manager.apply(manager.plan(parse_sdl(_SCENARIO))) + + +def test_honest_baseline_is_accepted(): + """The control. Without it, a rejection below would prove nothing.""" + + result = _apply_with() + + assert result.success, [diagnostic.message for diagnostic in result.diagnostics] + assert sorted(result.snapshot.entries) == ["provision.network.lab", "provision.node.web"] + + +def test_a_backend_that_realizes_nothing_is_refused(): + """The one fabrication the boundary already refuses.""" + + result = _apply_with(_RealizesNothing) + + assert result.success is False + assert "runtime.backend-contract-invalid" in {diagnostic.code for diagnostic in result.diagnostics} + + +def test_a_backend_cannot_invent_resources_the_scenario_never_authored(): + result = _apply_with(_InventsResource) + + assert result.success is False + assert "runtime.backend-contract-invalid" in {diagnostic.code for diagnostic in result.diagnostics} + assert _SMUGGLED_ADDRESS not in result.snapshot.entries + + +def test_a_backend_cannot_misreport_what_kind_of_resource_it_realized(): + result = _apply_with(_SubstitutesResourceType) + + assert result.success is False + assert "runtime.backend-contract-invalid" in {diagnostic.code for diagnostic in result.diagnostics} + realized = {entry.resource_type for entry in result.snapshot.entries.values()} + assert "node" not in realized or realized == {"network", "node"} + + +def test_a_backend_cannot_mutate_state_while_reporting_no_changes(): + result = _apply_with(_UnderReportsChanges) + + assert result.success is False + assert "runtime.backend-contract-invalid" in {diagnostic.code for diagnostic in result.diagnostics} + assert result.snapshot.entries == {} + + +def test_a_backend_cannot_file_provisioning_results_under_another_domain(): + result = _apply_with(_ForgesDomain) + + assert result.success is False + assert "runtime.backend-contract-invalid" in {diagnostic.code for diagnostic in result.diagnostics} + domains = {getattr(entry.domain, "value", entry.domain) for entry in result.snapshot.entries.values()} + assert "evaluation" not in domains