From 02193a47c9d1d608f3def59ea846d05475989c0b Mon Sep 17 00:00:00 2001 From: AlgoVoi Date: Mon, 10 Aug 2026 18:06:39 +0000 Subject: [PATCH 1/2] fix(agt-policies): fail closed on non-object input.snapshot in migrated Rego The migration tool renders Rego with `default verdict := {"decision": "allow"}` and inlines each rule condition as chained `object.get(input.snapshot, ...)` accessors. When `input.snapshot` is absent, null, or a non-object, every accessor is undefined, no `_match_i` rule fires, and evaluation falls through to the default-allow verdict, so a caller that omits or mistypes the snapshot root is allowed on a deny-side policy (issue #3517). Render a highest-precedence fail-closed guard: `default _snapshot_valid := false` plus `_snapshot_valid if { is_object(input.snapshot) }`, and a `verdict := deny if { not _snapshot_valid }` branch. A defaulted helper is required because an inline `not is_object(input.snapshot)` is itself undefined when the reference is undefined, so the absent-snapshot case would otherwise leak through to default-allow. Every rule branch, including the always-matching fail-closed deny rendered for an unsupported operator, is gated on `_snapshot_valid`, so a rule branch and the guard can never both hold and OPA never raises a complete-rule conflict on `verdict`. Valid object snapshots are unaffected: a well-formed snapshot that matches no rule still falls through to default-allow, and a matching rule still applies. Add opa-backed tests covering absent, null, string, number, and array snapshots (all deny), a valid snapshot without a match (allow), a matching rule (deny), and an unsupported-operator rule combined with a bad snapshot (single deny, no conflict). Reported-by: @MohammadHaroonAbuomar Signed-off-by: AlgoVoi --- .../src/agt/cli/_migrate_resolution/build.py | 31 ++++ .../tests/test_migrate_snapshot_guard.py | 146 ++++++++++++++++++ 2 files changed, 177 insertions(+) create mode 100644 agent-governance-python/agt-policies/tests/test_migrate_snapshot_guard.py diff --git a/agent-governance-python/agt-policies/src/agt/cli/_migrate_resolution/build.py b/agent-governance-python/agt-policies/src/agt/cli/_migrate_resolution/build.py index 7711763c8..3f2490ad8 100644 --- a/agent-governance-python/agt-policies/src/agt/cli/_migrate_resolution/build.py +++ b/agent-governance-python/agt-policies/src/agt/cli/_migrate_resolution/build.py @@ -266,6 +266,7 @@ def _render_rego(rules: list[dict[str, Any]]) -> str: f"\"reason\": \"runtime_error:manifest_invalid\", " f"\"message\": {json.dumps(f'rule {name!r} has {invalid_detail}; fail-closed deny')}}} if {{\n" f" _match_{idx}\n" + f" _snapshot_valid\n" + "".join(f" not _match_{j}\n" for j in range(idx)) + "}" ) @@ -289,6 +290,7 @@ def _render_rego(rules: list[dict[str, Any]]) -> str: branches.append( f"verdict := {verdict_dict} if {{\n" f" _match_{idx}\n" + f" _snapshot_valid\n" f"{previous_negations}" f"}}" ) @@ -301,6 +303,35 @@ def _render_rego(rules: list[dict[str, Any]]) -> str: unsupported_drops, ) + # Fail-closed snapshot guard (issue #3517): when input.snapshot is absent, + # null, or not an object, every field accessor is undefined, no _match_i + # fires, and evaluation would fall through to the default-allow verdict, so a + # caller that omits or mistypes the snapshot root would be allowed on a + # deny-side policy. Snapshot validity is expressed with a defaulted helper: a + # bare `not is_object(input.snapshot)` does not hold when the reference is + # undefined (an inline negation over a builtin with an undefined operand is + # itself undefined), so the absent-snapshot case would leak through to + # default-allow. `default _snapshot_valid := false` makes the undefined case + # explicit. Every rule branch above is gated on `_snapshot_valid` so the + # guard is mutually exclusive with them: a rule branch (including the + # always-matching fail-closed deny for an unsupported operator) and the guard + # can never both hold, or OPA would raise a complete-rule conflict on + # `verdict`. + matchers.insert( + 0, + "default _snapshot_valid := false\n\n" + "_snapshot_valid if {\n is_object(input.snapshot)\n}", + ) + branches.insert( + 0, + "verdict := {\"decision\": \"deny\", " + "\"reason\": \"runtime_error:snapshot_invalid\", " + "\"message\": \"input.snapshot is absent, null, or not an object; " + "fail-closed deny\"} if {\n" + " not _snapshot_valid\n" + "}", + ) + return header + "\n\n".join(matchers) + ("\n\n" if matchers else "") + "\n\n".join(branches) + "\n" diff --git a/agent-governance-python/agt-policies/tests/test_migrate_snapshot_guard.py b/agent-governance-python/agt-policies/tests/test_migrate_snapshot_guard.py new file mode 100644 index 000000000..266c31acc --- /dev/null +++ b/agent-governance-python/agt-policies/tests/test_migrate_snapshot_guard.py @@ -0,0 +1,146 @@ +"""Regression tests for issue #3517. + +``_render_rego`` emits ``default verdict := {"decision": "allow"}``. When +``input.snapshot`` is absent, null, or a non-object, every field accessor is +undefined, no ``_match_i`` rule fires, and evaluation would fall through to the +default-allow verdict, so a caller that omits or mistypes the snapshot root +would be allowed on a deny-side policy. The rendered module must instead fail +closed with a deny whenever the snapshot is not an object, while preserving the +existing behaviour for well-formed snapshots. +""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import tempfile +from typing import Any + +import pytest + +from agt.cli._migrate_resolution.build import _render_rego + + +def _opa_bin() -> str | None: + return os.environ.get("ACS_OPA_PATH") or shutil.which("opa") + + +_OPA = _opa_bin() +_needs_opa = pytest.mark.skipif(_OPA is None, reason="opa binary not available") + +# A representative deny-side rule set: deny when the tool call amount exceeds a +# threshold; a well-formed snapshot that matches nothing falls to default-allow. +_RULES: list[dict[str, Any]] = [ + { + "name": "deny_large_transfers", + "action": "deny", + "message": "amount exceeds limit", + "condition": { + "field": "tool_call.args.amount_usd", + "operator": "gt", + "value": 100, + }, + } +] + + +def _eval_verdict(rego_src: str, input_doc: dict[str, Any]) -> dict[str, Any]: + """Evaluate ``data.agt.legacy.verdict`` for ``input_doc`` via the opa binary.""" + assert _OPA is not None + with tempfile.TemporaryDirectory() as work: + rego_path = os.path.join(work, "agt_legacy.rego") + with open(rego_path, "w", encoding="utf-8") as handle: + handle.write(rego_src) + proc = subprocess.run( + [ + _OPA, + "eval", + "-f", + "json", + "-d", + rego_path, + "--stdin-input", + "data.agt.legacy.verdict", + ], + input=json.dumps(input_doc), + capture_output=True, + text=True, + check=False, + ) + assert proc.returncode == 0, f"opa eval failed: {proc.stderr}\n{proc.stdout}" + payload = json.loads(proc.stdout) + # A complete-rule conflict surfaces here as an ``errors`` key; assert its + # absence so the mutual-exclusivity of the guard is actually exercised. + assert "errors" not in payload, f"opa reported errors: {payload.get('errors')}" + return payload["result"][0]["expressions"][0]["value"] + + +def _verdict(input_doc: dict[str, Any], rules: list[dict[str, Any]] | None = None) -> dict[str, Any]: + return _eval_verdict(_render_rego(_RULES if rules is None else rules), input_doc) + + +@_needs_opa +@pytest.mark.parametrize( + "input_doc", + [ + pytest.param({}, id="absent"), + pytest.param({"snapshot": None}, id="null"), + pytest.param({"snapshot": "not-an-object"}, id="string"), + pytest.param({"snapshot": 42}, id="number"), + pytest.param({"snapshot": ["a", "b"]}, id="array"), + ], +) +def test_non_object_snapshot_fails_closed(input_doc: dict[str, Any]) -> None: + verdict = _verdict(input_doc) + assert verdict["decision"] == "deny" + assert verdict["reason"] == "runtime_error:snapshot_invalid" + + +@_needs_opa +def test_valid_object_snapshot_without_match_still_allows() -> None: + # Existing behaviour for valid inputs is preserved: a well-formed snapshot + # that matches no rule falls through to the default-allow verdict. + verdict = _verdict({"snapshot": {"tool_call": {"args": {"amount_usd": 5}}}}) + assert verdict["decision"] == "allow" + + +@_needs_opa +def test_valid_object_snapshot_matching_rule_denies() -> None: + verdict = _verdict({"snapshot": {"tool_call": {"args": {"amount_usd": 500}}}}) + assert verdict["decision"] == "deny" + assert verdict["reason"] == "deny_large_transfers" + + +@_needs_opa +def test_invalid_rule_and_bad_snapshot_do_not_conflict() -> None: + # An unsupported operator renders an always-matching fail-closed deny branch. + # With a non-object snapshot the snapshot guard also denies; the two must be + # mutually exclusive so OPA does not raise a complete-rule conflict on + # ``verdict``. _eval_verdict asserts no opa errors, so a conflict would fail. + rules = [ + { + "name": "unsupported_op", + "action": "deny", + "message": "x", + "condition": { + "field": "tool_call.args.amount_usd", + "operator": "nonsense_operator", + "value": 1, + }, + } + ] + verdict = _verdict({}, rules=rules) + assert verdict["decision"] == "deny" + + +def test_render_includes_snapshot_guard() -> None: + # Render-level assertion (no opa needed): the guard rule, its negation gate + # on rule branches, and the deny reason are present so evaluation cannot fall + # through to default-allow on a malformed snapshot. + rego = _render_rego(_RULES) + assert "default _snapshot_valid := false" in rego + assert "is_object(input.snapshot)" in rego + assert "not _snapshot_valid" in rego + assert "runtime_error:snapshot_invalid" in rego From f320c9a80b58a91f50179c5cd59a05a9cc4ba65a Mon Sep 17 00:00:00 2001 From: AlgoVoi Date: Wed, 12 Aug 2026 20:23:40 +0100 Subject: [PATCH 2/2] chore(license): add MIT header to the #3517 snapshot-guard test The new regression test for the snapshot-guard fix shipped without the standard license header, which the changed-files header gate rejects. Add the canonical Microsoft MIT header to match every sibling test file. Signed-off-by: AlgoVoi --- .../agt-policies/tests/test_migrate_snapshot_guard.py | 1 + 1 file changed, 1 insertion(+) diff --git a/agent-governance-python/agt-policies/tests/test_migrate_snapshot_guard.py b/agent-governance-python/agt-policies/tests/test_migrate_snapshot_guard.py index 266c31acc..d5ce85da2 100644 --- a/agent-governance-python/agt-policies/tests/test_migrate_snapshot_guard.py +++ b/agent-governance-python/agt-policies/tests/test_migrate_snapshot_guard.py @@ -1,3 +1,4 @@ +# Copyright (c) Microsoft Corporation. Licensed under the MIT License. """Regression tests for issue #3517. ``_render_rego`` emits ``default verdict := {"decision": "allow"}``. When