fix(bridge): translate GLOB blocked patterns to RE2 (fail-open) + bundle robustness

Addresses the policies/bridge.py findings from review #3137.

- GLOB fail-open: _pattern_to_regex translated GLOB patterns with
  fnmatch.translate, whose output (e.g. "(?s:.*\.exe)\Z") uses Python-only
  syntax that Go RE2 (OPA) rejects, leaving regex.match undefined so the GLOB
  deny silently failed open. Add _glob_to_re2 (translates * / ? / [seq] /
  [!seq] to RE2-safe syntax, anchored whole-string with (?s)^...$).
- Reject non-finite confidence_threshold up front (inf rendered invalid Rego;
  nan was silently dropped); json.dumps uses allow_nan=False.
- Clean up a self-created temp bundle dir if materialization fails.

Adds tests/test_bridge_pattern.py (no agent_os dependency) including an
OPA-backed regression test that compiles the GLOB regex under real RE2 and a
fnmatchcase cross-check across edge cases. Verified with OPA 0.70.0.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VT2CfGX3tmPsQaofuuWwED

Author: Haruka-Kaya

agent-governance-python/agt-policies/src/agt/policies/bridge.py

@@ -46,6 +46,7 @@
 
 import json
 import logging
+import math
 import shutil
 import tempfile
 from pathlib import Path
@@ -95,6 +96,55 @@ def _find_stock_rego_root() -> Path:
     )
 
 
+def _glob_to_re2(value: str) -> str:
+    r"""Translate a shell glob to a Go RE2 regex anchored to the whole string.
+
+    Mirrors :func:`fnmatch.translate` semantics (``*`` / ``?`` / ``[seq]`` /
+    ``[!seq]``) but emits only RE2-compatible syntax. ``fnmatch.translate``
+    cannot be used directly: it produces Python-specific constructs — an inline
+    ``(?s:...)`` flag group and the ``\Z`` end-of-string anchor — that Go RE2
+    (the engine behind OPA's ``agt.patterns``) rejects, leaving the generated
+    deny rule undefined so a ``GLOB`` blocked pattern silently fails open.
+
+    The result is a leading ``(?s)`` dot-all flag plus ``^``/``$`` anchors.
+    Whole-string anchoring preserves v4 glob semantics (the pattern matches the
+    entire policy text) under ``agt.patterns``' unanchored ``regex.match``.
+    """
+    import re
+
+    out: list[str] = []
+    i, n = 0, len(value)
+    while i < n:
+        ch = value[i]
+        i += 1
+        if ch == "*":
+            out.append(".*")
+        elif ch == "?":
+            out.append(".")
+        elif ch == "[":
+            j = i
+            if j < n and value[j] == "!":
+                j += 1
+            if j < n and value[j] == "]":
+                j += 1
+            while j < n and value[j] != "]":
+                j += 1
+            if j >= n:
+                # No closing bracket: treat '[' as a literal.
+                out.append(r"\[")
+            else:
+                inner = value[i:j].replace("\\", r"\\")
+                i = j + 1
+                if inner.startswith("!"):
+                    inner = "^" + inner[1:]
+                elif inner.startswith("^"):
+                    inner = "\\" + inner
+                out.append("[" + inner + "]")
+        else:
+            out.append(re.escape(ch))
+    return "(?s)^" + "".join(out) + "$"
+
+
 def _pattern_to_regex(pattern: Any) -> str:
     """Normalise a v4 ``blocked_patterns`` entry to a Go RE2 regex string.
 
@@ -119,9 +169,7 @@ def _pattern_to_regex(pattern: Any) -> str:
         if kind_name == "REGEX":
             return value
         if kind_name == "GLOB":
-            import fnmatch
-
-            return fnmatch.translate(value)
+            return _glob_to_re2(value)
         raise ValueError(f"unsupported PatternType: {kind!r}")
 
     raise ValueError(f"unsupported blocked_patterns entry: {pattern!r}")
@@ -180,12 +228,15 @@ def _render_rego(
     if budget_thresholds:
         branches.append(
             "v := budgets.deny_if_budget_exceeded("
-            f"{json.dumps(budget_thresholds)})"
+            f"{json.dumps(budget_thresholds, allow_nan=False)})"
         )
 
     if confidence_threshold is not None and confidence_threshold > 0.0:
+        # ``allow_nan=False`` is a defensive backstop; the caller
+        # (governance_to_acs_manifest) rejects non-finite thresholds up front.
         branches.append(
-            f"v := confidence.deny_if_low_confidence({json.dumps(confidence_threshold)})"
+            "v := confidence.deny_if_low_confidence("
+            f"{json.dumps(confidence_threshold, allow_nan=False)})"
         )
 
     if require_human_approval:
@@ -301,49 +352,77 @@ def governance_to_acs_manifest(
         ``policy.allowed_tools``. ``approval`` is set when
         ``require_human_approval`` is true.
     """
+    created_bundle_dir = bundle_dir is None
     bundle_dir = (
         Path(bundle_dir).resolve()
-        if bundle_dir is not None
+        if not created_bundle_dir
         else Path(tempfile.mkdtemp(prefix="agt_bridge_")).resolve()
     )
     bundle_dir.mkdir(parents=True, exist_ok=True)
 
-    stock_root = stock_rego_root or _find_stock_rego_root()
-    for rego_file in stock_root.glob("*.rego"):
-        if rego_file.name.endswith("_test.rego"):
-            continue
-        shutil.copy(rego_file, bundle_dir / rego_file.name)
-
-    blocked_pattern_regexes = [_pattern_to_regex(p) for p in policy.blocked_patterns]
-
-    # AGT-M3 round-2 BLOCK A: ``max_tool_calls=0`` is the v4 sentinel for
-    # "deny every tool call", not "no constraint". Forward 0 through to the
-    # ``budgets.deny_if_budget_exceeded`` helper. The helper compares
-    # ``tool_call_count >= limit`` so with ``limit=0`` and the default
-    # ``tool_call_count=0`` the first call is denied with
-    # ``budget_tool_calls_exceeded``, which preserves the v4 contract end-to-end
-    # for any caller that loads the bridge manifest into AgtRuntime directly
-    # (not just through the AdapterRuntimeBridge host fallback). Previously a
-    # ``GovernancePolicy(max_tool_calls=0, confidence_threshold=0.0)`` slipped
-    # through to the default ``allow`` verdict because the budget rule was
-    # omitted and the fallback ``pre_tool_call`` binding (no ``tool_name_from``)
-    # never tripped any deny rule. Keep ``max_tokens`` at ``> 0`` because the v4
-    # dataclass validation rejects ``max_tokens <= 0`` and there is no v4 wire
-    # value to preserve there.
-    rego_source = _render_rego(
-        package="agt.governance_policy",
-        max_tokens=policy.max_tokens if policy.max_tokens > 0 else None,
-        max_tool_calls=policy.max_tool_calls if policy.max_tool_calls >= 0 else None,
-        confidence_threshold=(
+    try:
+        stock_root = stock_rego_root or _find_stock_rego_root()
+        for rego_file in stock_root.glob("*.rego"):
+            if rego_file.name.endswith("_test.rego"):
+                continue
+            shutil.copy(rego_file, bundle_dir / rego_file.name)
+
+        blocked_pattern_regexes = [
+            _pattern_to_regex(p) for p in policy.blocked_patterns
+        ]
+
+        # A non-finite confidence_threshold is invalid input. ``inf`` would
+        # otherwise slip past the ``> 0`` guard in _render_rego and render
+        # ``deny_if_low_confidence(Infinity)`` (invalid Rego/JSON that fails to
+        # compile); ``nan`` would be silently dropped, so a caller who set a
+        # threshold would get no confidence deny at all. Fail loudly instead.
+        if policy.confidence_threshold is not None and not math.isfinite(
             policy.confidence_threshold
-            if policy.confidence_threshold and policy.confidence_threshold > 0
-            else None
-        ),
-        blocked_patterns=blocked_pattern_regexes,
-        require_human_approval=policy.require_human_approval,
-    )
-    rego_path = bundle_dir / f"{policy_id}.rego"
-    rego_path.write_text(rego_source, encoding="utf-8")
+        ):
+            raise ValueError(
+                "confidence_threshold must be finite, got "
+                f"{policy.confidence_threshold!r}"
+            )
+
+        # AGT-M3 round-2 BLOCK A: ``max_tool_calls=0`` is the v4 sentinel for
+        # "deny every tool call", not "no constraint". Forward 0 through to the
+        # ``budgets.deny_if_budget_exceeded`` helper. The helper compares
+        # ``tool_call_count >= limit`` so with ``limit=0`` and the default
+        # ``tool_call_count=0`` the first call is denied with
+        # ``budget_tool_calls_exceeded``, which preserves the v4 contract
+        # end-to-end for any caller that loads the bridge manifest into
+        # AgtRuntime directly (not just through the AdapterRuntimeBridge host
+        # fallback). Previously a ``GovernancePolicy(max_tool_calls=0,
+        # confidence_threshold=0.0)`` slipped through to the default ``allow``
+        # verdict because the budget rule was omitted and the fallback
+        # ``pre_tool_call`` binding (no ``tool_name_from``) never tripped any
+        # deny rule. Keep ``max_tokens`` at ``> 0`` because the v4 dataclass
+        # validation rejects ``max_tokens <= 0`` and there is no v4 wire value
+        # to preserve there.
+        rego_source = _render_rego(
+            package="agt.governance_policy",
+            max_tokens=policy.max_tokens if policy.max_tokens > 0 else None,
+            max_tool_calls=(
+                policy.max_tool_calls if policy.max_tool_calls >= 0 else None
+            ),
+            confidence_threshold=(
+                policy.confidence_threshold
+                if policy.confidence_threshold and policy.confidence_threshold > 0
+                else None
+            ),
+            blocked_patterns=blocked_pattern_regexes,
+            require_human_approval=policy.require_human_approval,
+        )
+        rego_path = bundle_dir / f"{policy_id}.rego"
+        rego_path.write_text(rego_source, encoding="utf-8")
+    except Exception:
+        # Do not leave a half-built bundle behind (stock libs copied, generated
+        # module missing) when pattern translation, rendering, or a write fails.
+        # Only clean up a directory we created ourselves; a caller-supplied one
+        # is theirs to manage.
+        if created_bundle_dir:
+            shutil.rmtree(bundle_dir, ignore_errors=True)
+        raise
 
     # bind_tools must also cover the case where the policy has a budget
     # or human-approval requirement but no explicit tool allowlist; without

agent-governance-python/agt-policies/tests/test_bridge_pattern.py

@@ -0,0 +1,177 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+"""Unit tests for ``agt.policies.bridge`` pattern translation and bundle safety.
+
+These avoid the v4 ``agent_os`` dependency (unlike ``test_bridge.py``) by
+exercising the bridge's pattern translation directly and by driving
+:func:`governance_to_acs_manifest` with a duck-typed policy fixture. They pin
+the fix for the ``GLOB`` fail-open: ``fnmatch.translate`` emits Python-only
+``(?s:...)`` / ``\\Z`` constructs that Go RE2 (OPA) rejects, so a ``GLOB``
+blocked pattern silently never matched.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import shutil
+import subprocess
+import tempfile
+import types
+from pathlib import Path
+
+import pytest
+
+from agt.policies.bridge import (
+    _glob_to_re2,
+    _pattern_to_regex,
+    governance_to_acs_manifest,
+)
+
+
+def _opa_regex_match(pattern: str, subject: str, opa: str) -> bool | None:
+    """Evaluate ``regex.match(pattern, subject)`` under real OPA / Go RE2.
+
+    Returns the boolean result, or ``None`` when RE2 rejects the pattern
+    (which is exactly what the old ``fnmatch.translate`` output triggered:
+    the deny rule went undefined and a GLOB pattern silently failed open).
+    """
+    rego = "package c\nimport rego.v1\nm := regex.match(input.p, input.s)\n"
+    with tempfile.TemporaryDirectory() as d:
+        rego_path = Path(d) / "c.rego"
+        rego_path.write_text(rego, encoding="utf-8")
+        proc = subprocess.run(
+            [opa, "eval", "--stdin-input", "--data", str(rego_path),
+             "--format", "json", "data.c.m"],
+            input=json.dumps({"p": pattern, "s": subject}),
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+    if proc.returncode != 0:
+        return None
+    try:
+        return json.loads(proc.stdout)["result"][0]["expressions"][0]["value"]
+    except (KeyError, IndexError, ValueError):
+        return None
+
+
+def _glob(value: str) -> tuple[str, types.SimpleNamespace]:
+    return (value, types.SimpleNamespace(name="GLOB"))
+
+
+@pytest.mark.parametrize(
+    "glob, subject, expect",
+    [
+        ("*.exe", "payload.exe", True),
+        ("*.exe", "payload.txt", False),
+        ("secret?.log", "secret1.log", True),
+        ("secret?.log", "secret12.log", False),
+        ("[ab]*.sh", "a_run.sh", True),
+        ("[ab]*.sh", "c_run.sh", False),
+    ],
+)
+def test_glob_to_re2_matches_like_fnmatch(glob: str, subject: str, expect: bool) -> None:
+    assert bool(re.match(_glob_to_re2(glob), subject)) is expect
+
+
+@pytest.mark.parametrize(
+    "glob, subject",
+    [
+        ("[!ab]x", "cx"),          # negated class matches
+        ("[!ab]x", "ax"),          # negated class excludes
+        ("a.c", "a.c"),            # '.' is a literal in a glob (escaped)
+        ("a.c", "axc"),            # ...so it must NOT match any char
+        ("a+b*", "a+bcd"),         # '+' is a literal, escaped
+        ("f[oo", "f[oo"),          # unclosed '[' is a literal
+        ("f[oo", "fXoo"),
+        ("*.tar.gz", "archive.tar.gz"),
+        ("data_?.csv", "data_9.csv"),
+        ("data_?.csv", "data_99.csv"),
+    ],
+)
+def test_glob_to_re2_agrees_with_fnmatchcase(glob: str, subject: str) -> None:
+    # Cross-check the translator against Python's reference matcher (the
+    # case-sensitive variant, matching our whole-string anchoring). RE2-safe
+    # output that still mirrors glob semantics for every edge case.
+    import fnmatch
+
+    assert bool(re.match(_glob_to_re2(glob), subject)) is fnmatch.fnmatchcase(
+        subject, glob
+    )
+
+
+def test_glob_output_is_re2_safe() -> None:
+    # Go RE2 rejects ``\Z`` and the inline ``(?s:...)`` flag group that
+    # ``fnmatch.translate`` produces; the translator must emit neither.
+    rx = _glob_to_re2("*.exe")
+    assert "\\Z" not in rx
+    assert "(?s:" not in rx
+    assert rx.startswith("(?s)^") and rx.endswith("$")
+
+
+def test_pattern_to_regex_dispatches_glob_to_re2() -> None:
+    assert _pattern_to_regex(_glob("*.exe")) == _glob_to_re2("*.exe")
+
+
+@pytest.mark.parametrize(
+    "subject, expect",
+    [("payload.exe", True), ("payload.txt", False), ("dir/app.exe", True)],
+)
+def test_glob_re2_compiles_and_matches_under_opa(subject: str, expect: bool) -> None:
+    """Regression guard: the GLOB regex must compile AND match under Go RE2.
+
+    The previous ``fnmatch.translate`` output (``(?s:.*\\.exe)\\Z``) is rejected
+    by RE2, so ``regex.match`` returned undefined (``None`` here) and the deny
+    rule silently failed open. The existing suite never compiled a GLOB pattern
+    under OPA, which is how the bug slipped through.
+    """
+    opa = shutil.which("opa") or str(Path.home() / ".local" / "bin" / "opa")
+    if not Path(opa).exists():
+        pytest.skip("opa binary required for RE2 compatibility check")
+    rx = _glob_to_re2("*.exe")
+    result = _opa_regex_match(rx, subject, opa)
+    assert result is expect, f"regex.match({rx!r}, {subject!r}) -> {result!r}"
+
+
+def test_substring_and_regex_patterns_unchanged() -> None:
+    assert _pattern_to_regex("a.b/c") == re.escape("a.b/c")
+    regex_entry = ("rm\\s+-rf", types.SimpleNamespace(name="REGEX"))
+    assert _pattern_to_regex(regex_entry) == "rm\\s+-rf"
+
+
+class _Policy:
+    name = "p"
+    version = "1.0.0"
+    max_tokens = 100
+    max_tool_calls = 5
+    allowed_tools: list[str] = []
+    blocked_patterns: list = []
+    require_human_approval = False
+    confidence_threshold = 0.0
+
+
+@pytest.mark.parametrize("bad", [float("inf"), float("nan")])
+def test_non_finite_confidence_threshold_rejected(tmp_path: Path, bad: float) -> None:
+    pol = _Policy()
+    pol.confidence_threshold = bad
+    with pytest.raises(ValueError):
+        governance_to_acs_manifest(
+            pol, bundle_dir=tmp_path / "bundle", stock_rego_root=tmp_path / "stock"
+        )
+
+
+def test_created_bundle_dir_cleaned_up_on_failure(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    target = tmp_path / "agt_bridge_leak"
+    monkeypatch.setattr(
+        "agt.policies.bridge.tempfile.mkdtemp", lambda *a, **k: str(target)
+    )
+    pol = _Policy()
+    # Non-string GLOB value makes _pattern_to_regex raise after the temp dir
+    # was created and stock libs (none here) were processed.
+    pol.blocked_patterns = [(123, types.SimpleNamespace(name="GLOB"))]
+    with pytest.raises(ValueError):
+        governance_to_acs_manifest(pol, stock_rego_root=tmp_path / "stock")
+    assert not target.exists(), "self-created bundle dir must be removed on failure"