v0.9.2: migrate to octorules.reserved_ips + @idempotent_registration

Drops the local _PRIVATE_SUPERNETS list in favor of octorules.reserved_ips
from core v0.26.0. GA503 (src_ip_ranges) and GA320 (inIpRange() CEL) now
delegate to is_reserved() — same strict-containment semantics, same
messages, same rule IDs.

Replaces the flag+lock idempotent-registration pattern in linter/__init__.py
and _policy_settings.py with @idempotent_registration from core. Minor
safety upgrade: the registration body now runs inside the lock rather
than only the flag-flip.

Adds tests/test_cli_lint.py exercising the octorules lint CLI with a
Google provider config.

Bumps octorules dep >=0.24.0 -> >=0.26.0.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: barnumbirr

CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/).
 
+## [0.9.2] - 2026-04-18
+
+### Changed
+- Minimum ``octorules`` dependency: ``>=0.26.0`` (was ``>=0.24.0``).
+
 ## [0.9.1] - 2026-04-13
 
 ### Fixed

octorules_google/_policy_settings.py

@@ -10,9 +10,10 @@
 """
 
 import logging
-import threading
 from dataclasses import dataclass, field
 
+from octorules.registration import idempotent_registration
+
 log = logging.getLogger(__name__)
 
 _EXT_KEY = "gcloud_armor_policy_settings"
@@ -402,28 +403,19 @@ def format_report(self, plans: list, zone_has_drift: bool, phases_data: list[dic
 # ---------------------------------------------------------------------------
 # Registration
 # ---------------------------------------------------------------------------
-_registered = False
-_register_lock = threading.Lock()
-
-
+@idempotent_registration
 def register_policy_settings() -> None:
     """Register all policy settings hooks with the core extension system."""
-    global _registered
-    with _register_lock:
-        if _registered:
-            return
-
-        from octorules.extensions import (
-            register_apply_extension,
-            register_dump_extension,
-            register_format_extension,
-            register_plan_zone_hook,
-            register_validate_extension,
-        )
-
-        register_plan_zone_hook(_prefetch_policy_settings, _finalize_policy_settings)
-        register_apply_extension(_EXT_KEY, _apply_policy_settings)
-        register_format_extension(_EXT_KEY, PolicySettingsFormatter())
-        register_validate_extension(_validate_policy_settings)
-        register_dump_extension(_dump_policy_settings)
-        _registered = True
+    from octorules.extensions import (
+        register_apply_extension,
+        register_dump_extension,
+        register_format_extension,
+        register_plan_zone_hook,
+        register_validate_extension,
+    )
+
+    register_plan_zone_hook(_prefetch_policy_settings, _finalize_policy_settings)
+    register_apply_extension(_EXT_KEY, _apply_policy_settings)
+    register_format_extension(_EXT_KEY, PolicySettingsFormatter())
+    register_validate_extension(_validate_policy_settings)
+    register_dump_extension(_dump_policy_settings)

octorules_google/linter/__init__.py

@@ -1,28 +1,16 @@
 """Google Cloud Armor linter — registers all GCloud-specific lint rules and plugins."""
 
-import threading
-
-_registered = False
-_register_lock = threading.Lock()
+from octorules.registration import idempotent_registration
 
 
+@idempotent_registration
 def register_google_linter() -> None:
-    """Register the Google Cloud Armor lint plugin and rule definitions.
-
-    Safe to call multiple times — subsequent calls are no-ops.
-    """
-    global _registered
-    with _register_lock:
-        if _registered:
-            return
-
-        from octorules.linter.plugin import LintPlugin, register_linter
-        from octorules.linter.rules.registry import register_rules
-
-        from octorules_google.linter._plugin import GA_RULE_IDS, google_lint
-        from octorules_google.linter._rules import GA_RULE_METAS
+    """Register the Google Cloud Armor lint plugin and rule definitions."""
+    from octorules.linter.plugin import LintPlugin, register_linter
+    from octorules.linter.rules.registry import register_rules
 
-        register_linter(LintPlugin(name="google", lint_fn=google_lint, rule_ids=GA_RULE_IDS))
-        register_rules(GA_RULE_METAS)
+    from octorules_google.linter._plugin import GA_RULE_IDS, google_lint
+    from octorules_google.linter._rules import GA_RULE_METAS
 
-        _registered = True
+    register_linter(LintPlugin(name="google", lint_fn=google_lint, rule_ids=GA_RULE_IDS))
+    register_rules(GA_RULE_METAS)

octorules_google/validate.py

@@ -7,6 +7,7 @@
 
 import celpy
 from octorules.linter.engine import LintResult, Severity, is_always_false, is_always_true
+from octorules.reserved_ips import is_reserved
 
 # Rule IDs emitted by validate_rules() — kept in sync with _rules.py by
 # test_plugin_rule_ids_match_metas.
@@ -381,40 +382,8 @@ def _extract_sensitivity(expr: str, start: int) -> int | None:
     }
 )
 
-# Reserved/bogon networks (RFC 1918, loopback, link-local, etc.) — flagged as
-# likely mistakes in Cloud Armor src_ip_ranges.
-_PRIVATE_SUPERNETS: list[tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, str]] = [
-    # IPv4
-    (ipaddress.ip_network("10.0.0.0/8"), "RFC 1918 private"),
-    (ipaddress.ip_network("172.16.0.0/12"), "RFC 1918 private"),
-    (ipaddress.ip_network("192.168.0.0/16"), "RFC 1918 private"),
-    (ipaddress.ip_network("127.0.0.0/8"), "loopback"),
-    (ipaddress.ip_network("169.254.0.0/16"), "link-local"),
-    (ipaddress.ip_network("100.64.0.0/10"), "CGNAT (RFC 6598)"),
-    (ipaddress.ip_network("0.0.0.0/8"), "this network"),
-    (ipaddress.ip_network("192.0.2.0/24"), "documentation (RFC 5737)"),
-    (ipaddress.ip_network("198.51.100.0/24"), "documentation (RFC 5737)"),
-    (ipaddress.ip_network("203.0.113.0/24"), "documentation (RFC 5737)"),
-    (ipaddress.ip_network("192.0.0.0/24"), "IANA special purpose"),
-    (ipaddress.ip_network("192.88.99.0/24"), "6to4 relay anycast"),
-    (ipaddress.ip_network("198.18.0.0/15"), "benchmark testing (RFC 2544)"),
-    (ipaddress.ip_network("224.0.0.0/4"), "multicast"),
-    (ipaddress.ip_network("240.0.0.0/4"), "reserved for future use"),
-    # IPv6
-    (ipaddress.ip_network("::/128"), "unspecified"),
-    (ipaddress.ip_network("::1/128"), "loopback"),
-    (ipaddress.ip_network("::ffff:0:0/96"), "IPv4-mapped"),
-    (ipaddress.ip_network("64:ff9b::/96"), "NAT64 (RFC 6052)"),
-    (ipaddress.ip_network("100::/64"), "discard (RFC 6666)"),
-    (ipaddress.ip_network("2001:db8::/32"), "documentation (RFC 3849)"),
-    (ipaddress.ip_network("2001::/23"), "IANA special purpose"),
-    (ipaddress.ip_network("2001::/32"), "Teredo"),
-    (ipaddress.ip_network("2002::/16"), "6to4"),
-    (ipaddress.ip_network("fc00::/7"), "unique local"),
-    (ipaddress.ip_network("fe80::/10"), "link-local"),
-    (ipaddress.ip_network("ff00::/8"), "multicast"),
-    (ipaddress.ip_network("::ffff:0:0:0/96"), "IPv4-translated"),
-]
+# Reserved/bogon network detection is provided by octorules.reserved_ips
+# (single source of truth across providers; see core v0.26.0).
 
 
 def validate_rules(rules: list[dict], *, phase: str = "") -> list[LintResult]:
@@ -943,19 +912,18 @@ def _check_cidrs(
             )
 
         # GA503: private/reserved range
-        for private, desc in _PRIVATE_SUPERNETS:
-            if net.version == private.version and net.subnet_of(private):
-                results.append(
-                    _result(
-                        rule_id="GA503",
-                        severity=Severity.WARNING,
-                        message=f"Private/reserved IP range: {cidr} ({desc})",
-                        phase=phase,
-                        ref=ref,
-                        field="match.config.src_ip_ranges",
-                    )
+        desc = is_reserved(cidr)
+        if desc is not None:
+            results.append(
+                _result(
+                    rule_id="GA503",
+                    severity=Severity.WARNING,
+                    message=f"Private/reserved IP range: {cidr} ({desc})",
+                    phase=phase,
+                    ref=ref,
+                    field="match.config.src_ip_ranges",
                 )
-                break
+            )
 
     # GA305: overlapping CIDRs
     for i, (cidr_a, net_a) in enumerate(networks):
@@ -1428,7 +1396,7 @@ def _check_cel_iniprange_cidr(
     for m in _IN_IP_RANGE_RE.finditer(expr):
         cidr = m.group(1)
         try:
-            net = ipaddress.ip_network(cidr, strict=False)
+            ipaddress.ip_network(cidr, strict=False)
         except ValueError as exc:
             results.append(
                 _result(
@@ -1443,19 +1411,18 @@ def _check_cel_iniprange_cidr(
             continue
 
         # GA320: check for private/reserved ranges
-        for private, desc in _PRIVATE_SUPERNETS:
-            if net.version == private.version and net.subnet_of(private):
-                results.append(
-                    _result(
-                        rule_id="GA320",
-                        severity=Severity.WARNING,
-                        message=f"Private/reserved IP range in inIpRange(): {cidr!r} ({desc})",
-                        phase=phase,
-                        ref=ref,
-                        field="match.expr.expression",
-                    )
+        desc = is_reserved(cidr)
+        if desc is not None:
+            results.append(
+                _result(
+                    rule_id="GA320",
+                    severity=Severity.WARNING,
+                    message=f"Private/reserved IP range in inIpRange(): {cidr!r} ({desc})",
+                    phase=phase,
+                    ref=ref,
+                    field="match.expr.expression",
                 )
-                break
+            )
 
 
 def _check_cel_type_mismatch(

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "octorules-google"
-version = "0.9.1"
+version = "0.9.2"
 description = "Google Cloud Armor provider for octorules"
 license = "Apache-2.0"
 requires-python = ">=3.10"
@@ -28,7 +28,7 @@ classifiers = [
     "Topic :: System :: Systems Administration",
 ]
 dependencies = [
-    "octorules>=0.24.0",
+    "octorules>=0.26.0",
     "google-cloud-compute>=1.10,<2.0",
     "cel-python>=0.1",
 ]

tests/test_cli_lint.py

@@ -0,0 +1,126 @@
+"""End-to-end tests for the 'octorules lint' CLI command with the Google provider."""
+
+from pathlib import Path
+
+import pytest
+from octorules.cli import build_parser, cmd_lint, main
+from octorules.config import Config
+
+# Importing the provider module triggers register_google_linter() at
+# module load time, which is what cmd_lint depends on.
+import octorules_google  # noqa: F401
+
+
+@pytest.fixture
+def lint_config(tmp_path):
+    """Minimal config + rules files exercising Google-specific lint paths."""
+    rules_dir = tmp_path / "rules"
+    rules_dir.mkdir()
+
+    # Clean rules — public IPs only.
+    (rules_dir / "clean-policy.yaml").write_text(
+        "gcloud_armor_custom_rules:\n"
+        "  - ref: '1000'\n"
+        "    description: Block public bad IP\n"
+        "    action: deny(403)\n"
+        "    match:\n"
+        "      versioned_expr: SRC_IPS_V1\n"
+        "      config:\n"
+        "        src_ip_ranges:\n"
+        "          - 1.2.3.0/24\n"
+    )
+
+    # Rules with multiple GA violations.
+    (rules_dir / "bad-policy.yaml").write_text(
+        "gcloud_armor_custom_rules:\n"
+        # GA503: reserved IP in src_ip_ranges.
+        "  - ref: '1000'\n"
+        "    description: Accidentally block RFC 1918\n"
+        "    action: deny(403)\n"
+        "    match:\n"
+        "      versioned_expr: SRC_IPS_V1\n"
+        "      config:\n"
+        "        src_ip_ranges:\n"
+        "          - 10.0.0.0/8\n"
+        # GA306: catch-all /0.
+        "  - ref: '1001'\n"
+        "    description: Catch-all\n"
+        "    action: deny(403)\n"
+        "    match:\n"
+        "      versioned_expr: SRC_IPS_V1\n"
+        "      config:\n"
+        "        src_ip_ranges:\n"
+        "          - 0.0.0.0/0\n"
+    )
+
+    config_file = tmp_path / "config.yaml"
+    config_file.write_text(
+        "providers:\n"
+        "  google:\n"
+        "    project: test-project\n"
+        "  rules:\n"
+        "    directory: ./rules\n"
+        "zones:\n"
+        "  clean-policy:\n"
+        "    sources:\n"
+        "      - rules\n"
+        "  bad-policy:\n"
+        "    sources:\n"
+        "      - rules\n"
+    )
+    return Config.from_file(config_file)
+
+
+class TestBuildParser:
+    def test_lint_subcommand_exists(self):
+        parser = build_parser()
+        args = parser.parse_args(["lint"])
+        assert args.command == "lint"
+
+    def test_lint_rule_filter_accepts_ga_codes(self):
+        parser = build_parser()
+        args = parser.parse_args(["lint", "--rule", "GA503", "--rule", "GA306"])
+        assert args.lint_rules == ["GA503", "GA306"]
+
+
+class TestCmdLint:
+    def test_clean_rules_exit_0(self, lint_config):
+        rc = cmd_lint(lint_config, ["clean-policy"])
+        assert rc == 0
+
+    def test_bad_rules_surface_findings(self, lint_config, capsys):
+        cmd_lint(lint_config, ["bad-policy"])
+        captured = capsys.readouterr()
+        assert "GA503" in captured.out
+        assert "GA306" in captured.out
+
+    def test_json_format(self, lint_config, capsys):
+        cmd_lint(lint_config, ["bad-policy"], lint_format="json")
+        captured = capsys.readouterr()
+        assert '"rule_id"' in captured.out
+        assert "GA503" in captured.out
+
+    def test_sarif_format(self, lint_config, capsys):
+        cmd_lint(lint_config, ["bad-policy"], lint_format="sarif")
+        captured = capsys.readouterr()
+        assert '"version": "2.1.0"' in captured.out
+
+    def test_rule_filter_scopes_output(self, lint_config, capsys):
+        cmd_lint(lint_config, ["bad-policy"], lint_rules=["GA503"])
+        captured = capsys.readouterr()
+        assert "GA503" in captured.out
+        assert "GA306" not in captured.out
+
+    def test_output_file(self, lint_config, tmp_path):
+        out_file = str(tmp_path / "lint-report.txt")
+        cmd_lint(lint_config, ["bad-policy"], output_file=out_file)
+        assert Path(out_file).exists()
+        assert "GA" in Path(out_file).read_text()
+
+
+class TestMainLintCommand:
+    def test_main_lint_exits_zero_on_clean(self, lint_config, tmp_path):
+        config_file = tmp_path / "config.yaml"
+        with pytest.raises(SystemExit) as exc_info:
+            main(["--config", str(config_file), "lint", "--zone", "clean-policy"])
+        assert exc_info.value.code == 0