Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -328,7 +328,10 @@ def _rego_op_clause(operator: str, accessor: str, value: Any) -> Optional[str]:
if operator == "eq":
return f"{indent}{accessor} == {literal}"
if operator == "ne":
return f"{indent}_v := {accessor}\n{indent}_v != null\n{indent}_v != {literal}"
# No `_v != null` guard: a missing field (accessor -> null) IS
# "not equal" to any non-null value, so it must satisfy a deny
# rule built on `ne` rather than fail the rule body (#3297).
return f"{indent}_v := {accessor}\n{indent}_v != {literal}"
if operator == "gt":
return f"{indent}_v := {accessor}\n{indent}_v != null\n{indent}_v > {literal}"
if operator == "lt":
Expand All @@ -340,7 +343,10 @@ def _rego_op_clause(operator: str, accessor: str, value: Any) -> Optional[str]:
if operator == "in":
return f"{indent}_v := {accessor}\n{indent}_v != null\n{indent}_v in {literal}"
if operator == "not_in":
return f"{indent}_v := {accessor}\n{indent}_v != null\n{indent}not _v in {literal}"
# Same reasoning as `ne`: a missing field is not a member of the
# set, so it must satisfy `not_in` rather than fail closed on
# the null guard (#3297).
return f"{indent}_v := {accessor}\n{indent}not _v in {literal}"
if operator == "exists":
return f"{indent}{accessor} != null"
if operator == "contains":
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -238,3 +238,102 @@ def test_tampered_content_hash_denied(tmp_path: Path) -> None:
)
assert result.is_deny
assert result.reason == "deny_unknown_tool_hash"


def test_missing_content_hash_denied(tmp_path: Path) -> None:
"""A call that omits content_hash entirely must not bypass the `ne`
gate. Omission is trivial for a caller to do and must not be
treated as an implicit match (#3297)."""
snap = pre_tool_call_snapshot(
agent_id="coder",
tool_name="execute_code",
args={"code": "print('hello')"},
)
result = run_scenario(
workspace_root=tmp_path,
governance_yaml={"governance.yaml": _content_hash_governance()},
intervention_point="pre_tool_call",
snapshot=snap,
)
assert result.is_deny
assert result.reason == "deny_unknown_tool_hash"


# ── Region allowlist (not_in) ───────────────────────────────────────


def _region_allowlist_governance() -> dict:
"""Deny calls whose declared region falls outside the allowlist. A
call that omits region entirely must not bypass the gate."""
return {
"rules": [
{
"name": "deny_region_not_allowlisted",
"condition": {
"field": "tool_call.args.region",
"operator": "not_in",
"value": ["us-east", "us-west"],
},
"action": "deny",
"priority": 100,
"message": "region is not on the allowlist",
}
],
"tools": {"transfer_data": {"clearance": "restricted"}},
"intervention_points": {
"pre_tool_call": {
"policy_target": "$.tool_call.args",
"policy_target_kind": "tool_args",
"tool_name_from": "$.tool_call.name",
"policy": {"id": "agt_legacy_rules"},
}
},
}


def test_region_in_allowlist_passes(tmp_path: Path) -> None:
snap = pre_tool_call_snapshot(
agent_id="etl",
tool_name="transfer_data",
args={"region": "us-east"},
)
result = run_scenario(
workspace_root=tmp_path,
governance_yaml={"governance.yaml": _region_allowlist_governance()},
intervention_point="pre_tool_call",
snapshot=snap,
)
assert result.is_allow


def test_region_outside_allowlist_denied(tmp_path: Path) -> None:
snap = pre_tool_call_snapshot(
agent_id="etl",
tool_name="transfer_data",
args={"region": "eu-west"},
)
result = run_scenario(
workspace_root=tmp_path,
governance_yaml={"governance.yaml": _region_allowlist_governance()},
intervention_point="pre_tool_call",
snapshot=snap,
)
assert result.is_deny
assert result.reason == "deny_region_not_allowlisted"


def test_region_missing_denied(tmp_path: Path) -> None:
"""Omitting region entirely must not bypass the not_in gate (#3297)."""
snap = pre_tool_call_snapshot(
agent_id="etl",
tool_name="transfer_data",
args={},
)
result = run_scenario(
workspace_root=tmp_path,
governance_yaml={"governance.yaml": _region_allowlist_governance()},
intervention_point="pre_tool_call",
snapshot=snap,
)
assert result.is_deny
assert result.reason == "deny_region_not_allowlisted"
Loading