v0.9.0: automatic tier detection, thread safety, cleanup

Added:
- Automatic Cloud Armor tier detection — resolve_zone_id inspects
  ddos_protection_config and rule_visibility to classify policies
  as standard, plus, or enterprise.

Fixed:
- recaptcha_options_config validation error message now includes
  zone_name/extension_key prefix.

Changed:
- Policy settings and linter registration now thread-safe.

Removed:
- Unused format_plan/count_changes from PolicySettingsFormatter.

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

Author: barnumbirr

CHANGELOG.md

@@ -5,6 +5,29 @@ 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.0] - 2026-04-10
+
+### Added
+- Automatic Cloud Armor tier detection — `resolve_zone_id` inspects
+  `ddos_protection_config` and
+  `adaptive_protection_config.layer7_ddos_defense_config.rule_visibility` to
+  classify policies as `standard`, `plus`, or `enterprise`. Detected tiers are
+  exposed via `zone_plans` and feed into the core zone plans cache for automatic
+  plan-tier-aware linting (GA501, GA502).
+
+### Fixed
+- `recaptcha_options_config` validation error message now includes the
+  `zone_name/extension_key` prefix consistent with all other validation
+  messages.
+
+### Changed
+- Policy settings and linter registration are now thread-safe
+  (`threading.Lock`).
+
+### Removed
+- Unused `format_plan` and `count_changes` methods from
+  `PolicySettingsFormatter`.
+
 ## [0.8.5] - 2026-04-09
 
 ### Changed

README.md

@@ -70,6 +70,7 @@ Safety thresholds are configured under `safety:` (framework-owned, not forwarded
 |---------|--------|-------|
 | Phase rules (4 phases) | Supported | Security policy rules |
 | Policy settings | Supported | Adaptive protection, DDoS config, default rule action |
+| Automatic tier detection | Supported | Detects `standard`, `plus`, or `enterprise` from policy config (no manual setting needed) |
 | Custom rulesets | Not supported | — |
 | Lists | Not supported | Use inline IP ranges in match config |
 | Page Shield | Not supported | — |

docs/lint.md

@@ -1920,9 +1920,9 @@ gcloud_armor_custom_rules:
 
 **Severity:** WARNING
 
-Google Cloud Armor standard tier limits each security policy to **10 rules** that use `matches()` (regex) in their CEL expression. This check counts regex rules across all phases in the policy and warns when the limit is exceeded.
+Google Cloud Armor standard tier limits each security policy to **10 rules** that use `matches()` (regex) in their CEL expression. This check counts regex rules across all phases in the policy and warns when the limit is exceeded. The tier is auto-detected from the policy's DDoS and adaptive protection configuration (see [GA502](#ga502--rule-count-exceeds-tier-limit) for details); this rule only fires when the detected tier is `standard`.
 
-**Triggers on:** A policy with more than 10 rules using `matches()`.
+**Triggers on:** A standard-tier policy with more than 10 rules using `matches()`.
 
 **Fix:** Reduce the number of regex rules, combine patterns, or upgrade to Cloud Armor Plus/Enterprise which has higher limits.
 
@@ -1940,7 +1940,7 @@ Cloud Armor has per-policy rule count limits that vary by tier:
 | Plus | 512 |
 | Enterprise | 1024 |
 
-This check compares the number of rules in a phase against the configured tier's limit. The tier is determined by the `plan_tier` setting (defaults to "enterprise", the most permissive).
+This check compares the number of rules in a phase against the tier's limit. The tier is auto-detected from the policy's `ddos_protection_config` and `adaptive_protection_config.layer7_ddos_defense_config.rule_visibility` during zone resolution (`standard`, `plus`, or `enterprise`). When detection isn't possible (e.g., the policy lacks these fields), the tier falls back to `enterprise` (the most permissive).
 
 **Triggers on:** A phase with more rules than the tier allows.
 

octorules_google/_policy_settings.py

@@ -266,7 +266,10 @@ def _validate_policy_settings(desired, zone_name, errors, lines):
 
     rc_cfg = settings.get("recaptcha_options_config")
     if rc_cfg is not None and not isinstance(rc_cfg, dict):
-        errors.append(f"recaptcha_options_config must be a mapping, got {type(rc_cfg).__name__}")
+        errors.append(
+            f"  {zone_name}/{_EXT_KEY}: recaptcha_options_config must be a mapping,"
+            f" got {type(rc_cfg).__name__}"
+        )
 
 
 def _dump_policy_settings(scope, provider, out_dir):
@@ -291,27 +294,6 @@ def _dump_policy_settings(scope, provider, out_dir):
 class PolicySettingsFormatter:
     """Formats policy settings diffs for plan output."""
 
-    def format_plan(self, plans: list, zone_name: str) -> list[str]:
-        lines: list[str] = []
-        for plan in plans:
-            if not isinstance(plan, PolicySettingsPlan) or not plan.has_changes:
-                continue
-            for change in plan.changes:
-                if not change.has_changes:
-                    continue
-                lines.append(
-                    f"  {zone_name}/policy_settings.{change.field}:"
-                    f" {change.current!r} -> {change.desired!r}"
-                )
-        return lines
-
-    def count_changes(self, plans: list) -> int:
-        count = 0
-        for plan in plans:
-            if isinstance(plan, PolicySettingsPlan):
-                count += sum(1 for c in plan.changes if c.has_changes)
-        return count
-
     def format_text(self, plans: list, use_color: bool) -> list[str]:
         from octorules._color import Pen
 
@@ -430,18 +412,18 @@ def register_policy_settings() -> None:
     with _register_lock:
         if _registered:
             return
-        _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)
+        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

octorules_google/linter/__init__.py

@@ -1,6 +1,9 @@
 """Google Cloud Armor linter — registers all GCloud-specific lint rules and plugins."""
 
+import threading
+
 _registered = False
+_register_lock = threading.Lock()
 
 
 def register_google_linter() -> None:
@@ -9,16 +12,17 @@ def register_google_linter() -> None:
     Safe to call multiple times — subsequent calls are no-ops.
     """
     global _registered
-    if _registered:
-        return
+    with _register_lock:
+        if _registered:
+            return
 
-    from octorules.linter.plugin import LintPlugin, register_linter
-    from octorules.linter.rules.registry import register_rules
+        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
+        from octorules_google.linter._plugin import GA_RULE_IDS, google_lint
+        from octorules_google.linter._rules import GA_RULE_METAS
 
-    register_linter(LintPlugin(name="google", lint_fn=google_lint, rule_ids=GA_RULE_IDS))
-    register_rules(GA_RULE_METAS)
+        register_linter(LintPlugin(name="google", lint_fn=google_lint, rule_ids=GA_RULE_IDS))
+        register_rules(GA_RULE_METAS)
 
-    _registered = True
+        _registered = True

octorules_google/provider.py

@@ -8,6 +8,7 @@
 
 import logging
 import os
+import threading
 
 from google.api_core.exceptions import Forbidden, GoogleAPIError, NotFound, Unauthorized
 from google.auth.exceptions import DefaultCredentialsError
@@ -97,6 +98,32 @@ def _guarded():
     )
 
 
+# ---------------------------------------------------------------------------
+# Tier detection
+# ---------------------------------------------------------------------------
+
+
+def _detect_tier(policy: dict) -> str:
+    """Detect the Cloud Armor tier from a SecurityPolicy dict.
+
+    Heuristic:
+    - No ``ddos_protection_config`` or ddos_protection is not ADVANCED/ADVANCED_PREVIEW
+      → ``"standard"``
+    - ADVANCED or ADVANCED_PREVIEW with layer7 rule_visibility ``"PREMIUM"``
+      → ``"enterprise"``
+    - ADVANCED or ADVANCED_PREVIEW otherwise → ``"plus"``
+    """
+    ddos_cfg = policy.get("ddos_protection_config") or {}
+    ddos_protection = ddos_cfg.get("ddos_protection", "")
+    if ddos_protection not in ("ADVANCED", "ADVANCED_PREVIEW"):
+        return "standard"
+    adaptive = policy.get("adaptive_protection_config") or {}
+    layer7 = adaptive.get("layer7_ddos_defense_config") or {}
+    if layer7.get("rule_visibility") == "PREMIUM":
+        return "enterprise"
+    return "plus"
+
+
 # ---------------------------------------------------------------------------
 # Rule classification
 # ---------------------------------------------------------------------------
@@ -216,6 +243,8 @@ def __init__(
             )
         self._max_workers = max_workers
         self._timeout = timeout if timeout is not None else 30.0
+        self._zone_plans: dict[str, str] = {}
+        self._lock = threading.Lock()
 
     # -- Properties --
 
@@ -236,8 +265,8 @@ def account_name(self) -> str | None:
 
     @property
     def zone_plans(self) -> dict[str, str]:
-        """Return empty dict; Cloud Armor has no zone plan tiers."""
-        return {}
+        """Zone tiers detected from policy properties."""
+        return dict(self._zone_plans)
 
     # -- Helpers --
 
@@ -330,17 +359,22 @@ def update_policy_settings(self, scope: Scope, settings: dict) -> None:
     def resolve_zone_id(self, zone_name: str) -> str:
         """Resolve a security policy name to itself (Cloud Armor uses names).
 
-        Verifies the policy exists. Raises ConfigError if not found.
+        Verifies the policy exists and detects the Cloud Armor tier.
+        Raises ConfigError if not found.
         """
         try:
-            self._client.get(
+            response = self._client.get(
                 project=self._project,
                 security_policy=zone_name,
                 timeout=self._timeout,
             )
         except NotFound:
             raise ConfigError(f"No security policy found for {zone_name!r}") from None
-        log.debug("Resolved %s -> %s", zone_name, zone_name)
+        policy = to_plain_dict(response)
+        tier = _detect_tier(policy)
+        with self._lock:
+            self._zone_plans[zone_name] = tier
+        log.debug("Resolved %s -> %s (tier=%s)", zone_name, zone_name, tier)
         return zone_name
 
     @_wrap_provider_errors

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "octorules-google"
-version = "0.8.5"
+version = "0.9.0"
 description = "Google Cloud Armor provider for octorules"
 license = "Apache-2.0"
 requires-python = ">=3.10"

tests/test_policy_settings.py

@@ -566,6 +566,7 @@ def test_recaptcha_options_config_non_dict(self):
         errors: list[str] = []
         _validate_policy_settings(desired, "zone", errors, [])
         assert len(errors) == 1
+        assert "zone/gcloud_armor_policy_settings:" in errors[0]
         assert "recaptcha_options_config" in errors[0]
         assert "mapping" in errors[0]
 
@@ -833,46 +834,6 @@ def test_no_drift(self):
         assert phases_data == []
 
 
-# ---------------------------------------------------------------------------
-# Format extension — format_plan and count_changes
-# ---------------------------------------------------------------------------
-class TestFormatPlanAndCount:
-    def test_format_plan(self):
-        fmt = PolicySettingsFormatter()
-        plan = PolicySettingsPlan(
-            changes=[PolicySettingsChange("default_rule_action", "allow", "deny(403)")]
-        )
-        lines = fmt.format_plan([plan], "my-policy")
-        assert len(lines) == 1
-        assert "my-policy" in lines[0]
-        assert "allow" in lines[0]
-        assert "deny(403)" in lines[0]
-
-    def test_count_changes(self):
-        fmt = PolicySettingsFormatter()
-        plan = PolicySettingsPlan(
-            changes=[
-                PolicySettingsChange("default_rule_action", "allow", "deny(403)"),
-                PolicySettingsChange(
-                    "ddos_protection_config",
-                    {"ddos_protection": "STANDARD"},
-                    {"ddos_protection": "STANDARD"},
-                ),  # no change
-                PolicySettingsChange(
-                    "advanced_options_config",
-                    {"json_parsing": "DISABLED"},
-                    {"json_parsing": "STANDARD"},
-                ),
-            ]
-        )
-        assert fmt.count_changes([plan]) == 2
-
-    def test_empty(self):
-        fmt = PolicySettingsFormatter()
-        assert fmt.format_plan([], "z") == []
-        assert fmt.count_changes([]) == 0
-
-
 # ---------------------------------------------------------------------------
 # Provider methods
 # ---------------------------------------------------------------------------