fix: seal compiled policy snapshots

Author: rozmiarD

CHANGELOG.md

@@ -8,6 +8,12 @@ experimental and fixture classifications.
 
 ## Unreleased
 
+- Adds a private complete-payload seal to compiled policy packs and requires a
+  stable, detached canonical snapshot before policy evaluation, pack digesting
+  or enforcement admission. Post-compilation mutation of nested condition,
+  metadata, obligation or constraint values now fails closed with the existing
+  `invalid_compiled_policy_pack` reason without changing public records,
+  signatures or digest payloads.
 - Closes the v0.1 policy pack and rule field inventories while preserving only
   the intentional pack `id`, rule `id`, and rule `decision` compatibility
   aliases. The authoring schema and runtime compiler now reject ambiguous dual

docs/POLICY_ENGINE.md

@@ -104,6 +104,15 @@ Compiled rules are sorted by `priority` (lower first).
 The compiler performs exact deterministic analysis only; it does not attempt
 partial-overlap reasoning, subsumption or SAT solving.
 
+Each `CompiledPolicyPack` returned by the compiler records a private,
+process-local seal over its complete compiled payload. Policy evaluation,
+policy-pack digesting and enforcement admission compare the current payload to
+that seal, recompile a detached canonical snapshot and reject instability with
+`invalid_compiled_policy_pack`. This catches post-compilation changes inside
+mutable condition values, pack/control metadata and constraint values. The
+seal is an in-process consistency check, not a signature, authenticity proof
+or defense against a malicious host that deliberately bypasses GovEngine.
+
 ### Typed conditions (`schema_version: v1`)
 
 Policy pack v1 replaces the implicit equality map with an explicit condition

docs/SECURITY_GUARANTEES.md

@@ -10,6 +10,7 @@ do not apply when an in-process host bypasses or modifies GovEngine.
 | --- | --- | --- |
 | Strict bounded input | Shared JSON byte/depth/node/collection/string limits, duplicate-key and non-finite rejection, recursive forbidden keys | `test_api_hardening.py`, `test_security_properties.py`, conformance corpus |
 | Deterministic policy | Typed closed operators, strict operand types, canonical condition order, bounded compilation, deny-first evaluation | `test_policy_conditions.py`, `test_policy_v1_stability.py` |
+| Compiled policy snapshot integrity | Private complete-payload seal plus detached canonical recompilation before evaluation, digesting and enforcement admission | `test_policy_conditions.py`, `test_policy_engine.py`, `test_policy_enforcement.py` |
 | Active policy binding | Digest/epoch/issuer/status/validity checked against host activation port; authorization cannot outlive the activation observed at issuance | `test_governance_decision.py` |
 | Approval is independent | Exact subject binding, trust policy, validity, revocation and host signature verification | `test_governance_request.py`, `test_governance_decision.py` |
 | Scope is not self-authorized | Requested destination is compared with an independent scope policy | `test_scope_capabilities.py`, corpus |
@@ -25,7 +26,7 @@ do not apply when an in-process host bypasses or modifies GovEngine.
 
 | Binding | Complete payload owner | GovEngine behavior | Security meaning |
 | --- | --- | --- | --- |
-| Policy pack digest | GovEngine | Recomputes canonical GovEngine digest | Exact compiled policy content, not issuer identity |
+| Policy pack digest | GovEngine | Validates a sealed detached snapshot, then recomputes the unchanged canonical GovEngine digest | Exact compiled policy content, not issuer identity or producer authenticity |
 | Execution facts digest | GovEngine bounded projection | Recomputes | Exact bounded policy-evaluation facts |
 | Requested scope digest | GovEngine bounded projection | Recomputes | Exact requested namespace/destination metadata |
 | Scope policy digest | GovEngine | Recomputes | Exact independent allow policy |

govengine/policy/compiler/__init__.py

@@ -2,6 +2,7 @@
 
 from dataclasses import dataclass, field
 from datetime import timezone
+from hmac import compare_digest
 import math
 import re
 from typing import Any, Mapping
@@ -271,6 +272,9 @@ class CompiledPolicyPack:
     expires_at: str = ''
     supersedes: tuple[str, ...] = field(default_factory=tuple)
 
+    def __post_init__(self) -> None:
+        object.__setattr__(self, '_integrity_seal', _compiled_policy_pack_seal(self))
+
     def as_dict(self) -> dict[str, Any]:
         payload: dict[str, Any] = {
             'policy_id': self.policy_id,
@@ -435,6 +439,53 @@ def compile_policy_pack(policy_pack: Mapping[str, Any]) -> CompileResult:
     return PolicyCompiler().compile(policy_pack)
 
 
+def _compiled_policy_pack_seal(policy_pack: CompiledPolicyPack) -> str:
+    return govengine_record_digest(
+        policy_pack,
+        record_type='govengine.policy.compiler.CompiledPolicyPackIntegritySeal',
+    )
+
+
+def _validated_compiled_policy_pack_snapshot(
+    policy_pack: CompiledPolicyPack,
+) -> CompiledPolicyPack:
+    if not isinstance(policy_pack, CompiledPolicyPack):
+        raise GovApiError('invalid_compiled_policy_pack')
+    try:
+        initial_seal = getattr(policy_pack, '_integrity_seal', None)
+        if not isinstance(initial_seal, str):
+            raise GovApiError('invalid_compiled_policy_pack')
+        result = PolicyCompiler().compile(policy_pack.as_dict())
+        if not result.ok or result.policy_pack is None:
+            raise GovApiError('invalid_compiled_policy_pack')
+        snapshot = result.policy_pack
+        snapshot_seal = getattr(snapshot, '_integrity_seal', None)
+        if (
+            not isinstance(snapshot_seal, str)
+            or not compare_digest(
+                snapshot_seal,
+                _compiled_policy_pack_seal(snapshot),
+            )
+        ):
+            raise GovApiError('invalid_compiled_policy_pack')
+        final_stored_seal = getattr(policy_pack, '_integrity_seal', None)
+        final_payload_seal = _compiled_policy_pack_seal(policy_pack)
+        if (
+            snapshot != policy_pack
+            or not isinstance(final_stored_seal, str)
+            or not compare_digest(initial_seal, final_stored_seal)
+            or not compare_digest(initial_seal, final_payload_seal)
+        ):
+            raise GovApiError('invalid_compiled_policy_pack')
+    except GovApiError as exc:
+        if exc.reason_code == 'invalid_compiled_policy_pack':
+            raise
+        raise GovApiError('invalid_compiled_policy_pack') from exc
+    except (AttributeError, RecursionError, TypeError, ValueError) as exc:
+        raise GovApiError('invalid_compiled_policy_pack') from exc
+    return snapshot
+
+
 def _reject_conflicts(rules: tuple[PolicyRule, ...]) -> None:
     rule_ids: set[str] = set()
     seen: dict[str, str] = {}

govengine/policy/enforcement.py

@@ -11,7 +11,10 @@
     admission_decision_from_host_gate,
     validate_admission_decision,
 )
-from govengine.policy.compiler import CompiledPolicyPack
+from govengine.policy.compiler import (
+    CompiledPolicyPack,
+    _validated_compiled_policy_pack_snapshot,
+)
 from govengine.policy.model import PolicyVerdict, validate_policy_verdict
 from govengine.signing import govengine_record_digest
 
@@ -151,8 +154,7 @@ def as_dict(self) -> dict[str, Any]:
 
 
 def policy_pack_digest(policy_pack: CompiledPolicyPack) -> str:
-    if not isinstance(policy_pack, CompiledPolicyPack):
-        raise GovApiError("invalid_compiled_policy_pack")
+    policy_pack = _validated_compiled_policy_pack_snapshot(policy_pack)
     return govengine_record_digest(
         policy_pack,
         record_type="govengine.policy.compiler.CompiledPolicyPack",
@@ -317,6 +319,7 @@ def admit_policy_execution(
     policy_pack: CompiledPolicyPack,
     verdict: Mapping[str, Any] | PolicyVerdict,
 ) -> PolicyEnforcementPlan:
+    policy_pack = _validated_compiled_policy_pack_snapshot(policy_pack)
     checked = validate_policy_verdict(verdict)
     blockers = list(checked.blockers)
     reason_code = checked.reason_code or checked.decision

govengine/policy/explain.py

@@ -3,8 +3,11 @@
 from dataclasses import dataclass, field, replace
 from typing import Any, Mapping
 
-from govengine.api import GovApiError
-from govengine.policy.compiler import CompiledPolicyPack, PolicyRule
+from govengine.policy.compiler import (
+    CompiledPolicyPack,
+    PolicyRule,
+    _validated_compiled_policy_pack_snapshot,
+)
 from govengine.policy.enforcement import (
     SUPPORTED_POLICY_CONSTRAINTS,
     SUPPORTED_POLICY_OBLIGATIONS,
@@ -100,8 +103,7 @@ def explain_policy_evaluation(
     context: Mapping[str, Any] | None = None,
 ) -> PolicyEvaluationExplanation:
     checked_request = validate_policy_request(request)
-    if not isinstance(policy_pack, CompiledPolicyPack):
-        raise GovApiError("invalid_compiled_policy_pack")
+    policy_pack = _validated_compiled_policy_pack_snapshot(policy_pack)
     runtime_context = dict(context or {})
     verdict = PolicyEngine().evaluate(
         checked_request,

govengine/policy/runtime/__init__.py

@@ -3,7 +3,12 @@
 from typing import Any, Mapping
 
 from govengine.api import GovApiError
-from govengine.policy.compiler import CompiledPolicyPack, PolicyCondition, PolicyRule
+from govengine.policy.compiler import (
+    CompiledPolicyPack,
+    PolicyCondition,
+    PolicyRule,
+    _validated_compiled_policy_pack_snapshot,
+)
 from govengine.policy.model import PolicyRequest, PolicyVerdict, validate_policy_request, validate_policy_verdict
 
 
@@ -21,8 +26,7 @@ def evaluate(
         context: Mapping[str, Any] | None = None,
     ) -> PolicyVerdict:
         checked_request = validate_policy_request(request)
-        if not isinstance(policy_pack, CompiledPolicyPack):
-            raise GovApiError('invalid_compiled_policy_pack')
+        policy_pack = _validated_compiled_policy_pack_snapshot(policy_pack)
         runtime_context = dict(context or {})
 
         invariant = self._evaluate_invariants(checked_request, policy_pack)

tests/test_governance_request.py

@@ -209,9 +209,15 @@ def _base_request_mapping() -> dict[str, Any]:
 
 def _request_mapping_for_policy_pack(
     policy_pack: CompiledPolicyPack,
+    *,
+    rebound_policy_pack_digest: str | None = None,
 ) -> dict[str, Any]:
     request = _base_request_mapping()
-    compiled_policy_digest = policy_pack_digest(policy_pack)
+    compiled_policy_digest = (
+        rebound_policy_pack_digest
+        if rebound_policy_pack_digest is not None
+        else policy_pack_digest(policy_pack)
+    )
     scope_policy_payload = dict(request['scope_policy_binding'])
     scope_policy_payload['policy_pack_digest'] = compiled_policy_digest
     scope_policy = ScopePolicyBinding.from_mapping(scope_policy_payload)
@@ -222,6 +228,15 @@ def _request_mapping_for_policy_pack(
     return request
 
 
+def _raw_policy_pack_digest_for_mutation_fixture(
+    policy_pack: CompiledPolicyPack,
+) -> str:
+    return govengine_record_digest(
+        policy_pack,
+        record_type='govengine.policy.compiler.CompiledPolicyPack',
+    )
+
+
 def _request_mapping_with_approval(
     *,
     attestation_patch: Mapping[str, Any] | None = None,
@@ -283,7 +298,14 @@ def test_governance_request_rejects_mutated_typed_policy_pack_metadata() -> None
     policy_pack = _compiled_policy()
     assert isinstance(policy_pack.metadata, dict)
     policy_pack.metadata['password'] = 'REDACTED-FIXTURE'
-    request = _request_mapping_for_policy_pack(policy_pack)
+    with pytest.raises(GovApiError, match='invalid_compiled_policy_pack'):
+        policy_pack_digest(policy_pack)
+    request = _request_mapping_for_policy_pack(
+        policy_pack,
+        rebound_policy_pack_digest=_raw_policy_pack_digest_for_mutation_fixture(
+            policy_pack
+        ),
+    )
 
     with pytest.raises(GovApiError, match='forbidden_policy_metadata'):
         GovernanceRequest.from_mapping(request)
@@ -295,7 +317,11 @@ def test_direct_governance_request_rejects_mutated_typed_policy_pack_metadata()
     )
     assert isinstance(request.policy_pack.metadata, dict)
     request.policy_pack.metadata['password'] = 'REDACTED-FIXTURE'
-    compiled_policy_digest = policy_pack_digest(request.policy_pack)
+    with pytest.raises(GovApiError, match='invalid_compiled_policy_pack'):
+        policy_pack_digest(request.policy_pack)
+    compiled_policy_digest = _raw_policy_pack_digest_for_mutation_fixture(
+        request.policy_pack
+    )
     scope_policy_payload = request.scope_policy_binding.as_dict()
     scope_policy_payload['policy_pack_digest'] = compiled_policy_digest
     scope_policy = ScopePolicyBinding.from_mapping(scope_policy_payload)
@@ -314,7 +340,11 @@ def test_subject_digest_rejects_mutated_typed_policy_pack_metadata() -> None:
     request = GovernanceRequest.from_mapping(_base_request_mapping())
     assert isinstance(request.policy_pack.metadata, dict)
     request.policy_pack.metadata['password'] = 'REDACTED-FIXTURE'
-    compiled_policy_digest = policy_pack_digest(request.policy_pack)
+    with pytest.raises(GovApiError, match='invalid_compiled_policy_pack'):
+        policy_pack_digest(request.policy_pack)
+    compiled_policy_digest = _raw_policy_pack_digest_for_mutation_fixture(
+        request.policy_pack
+    )
     scope_policy_payload = request.scope_policy_binding.as_dict()
     scope_policy_payload['policy_pack_digest'] = compiled_policy_digest
     scope_policy = ScopePolicyBinding.from_mapping(scope_policy_payload)
@@ -378,7 +408,12 @@ def test_safe_typed_policy_pack_matches_mapping_request(
 
 def test_governance_request_rejects_directly_constructed_invalid_typed_pack() -> None:
     policy_pack = replace(_compiled_v1_policy(), issuer_ref='')
-    request = _request_mapping_for_policy_pack(policy_pack)
+    request = _request_mapping_for_policy_pack(
+        policy_pack,
+        rebound_policy_pack_digest=_raw_policy_pack_digest_for_mutation_fixture(
+            policy_pack
+        ),
+    )
 
     with pytest.raises(GovApiError, match='missing_policy_issuer_ref'):
         GovernanceRequest.from_mapping(request)

tests/test_policy_conditions.py

@@ -2,7 +2,12 @@
 
 import pytest
 
-from govengine import PolicyCompiler, PolicyEngine, explain_policy_evaluation
+from govengine import (
+    PolicyCompiler,
+    PolicyEngine,
+    evaluate_policy,
+    explain_policy_evaluation,
+)
 from govengine.api import GovApiError
 from govengine.policy.compiler import (
     _POLICY_CONDITION_V1_FIELDS,
@@ -176,6 +181,27 @@ def test_typed_policy_conditions_are_canonical_and_round_trip() -> None:
     ]
 
 
+def test_runtime_rejects_mutated_nested_compiled_condition_value() -> None:
+    pack = _compile(
+        {
+            'path': 'action.labels',
+            'operator': 'in',
+            'value': [{'scope': {'name': 'read'}}],
+        }
+    )
+    value = pack.rules[0].conditions[0].value
+    assert isinstance(value, list)
+    first = value[0]
+    assert isinstance(first, dict)
+    scope = first['scope']
+    assert isinstance(scope, dict)
+    scope['name'] = 'write'
+
+    for consumer in (PolicyEngine().evaluate, evaluate_policy):
+        with pytest.raises(GovApiError, match='invalid_compiled_policy_pack'):
+            consumer(_request(), pack)
+
+
 def test_legacy_equality_map_compiles_to_typed_ast_without_wire_drift() -> None:
     result = PolicyCompiler().compile(
         {