Azure rule hardening - Part 2 (#163)

* azure.app_service.idle

* azure.app_service_plan.empty

* azure.container_registry.unused

* azure.compute.snapshot.old

Author: sureshcsdp

cleancloud/providers/aws/rules/ai/sagemaker_training_job_long_running.py

@@ -279,7 +279,7 @@ def find_long_running_sagemaker_training_jobs(
 ) -> List[Finding]:
     sagemaker = session.client("sagemaker", region_name=region)
 
-    # Spec §8: paginate WITHOUT StatusEquals — filter InProgress client-side.
+    # Spec 8: paginate WITHOUT StatusEquals — filter InProgress client-side.
     # AWS documents that StatusEquals + MaxResults filters after paging, which can
     # silently miss InProgress jobs when pagination is truncated.
     try:
@@ -360,7 +360,7 @@ def find_long_running_sagemaker_training_jobs(
                 active_training_hours = None
                 elapsed_runtime_hours = nl["job_age_hours"]
 
-            # applicable_runtime_limit_seconds per spec §3
+            # applicable_runtime_limit_seconds per spec 3
             if nd["enable_managed_spot_training"] and nd["max_wait_time_seconds"] is not None:
                 applicable_runtime_limit_seconds = nd["max_wait_time_seconds"]
             elif training_start_time is not None and nd["max_runtime_seconds"] is not None:

cleancloud/providers/aws/rules/ami_old.py

@@ -104,7 +104,7 @@ def _evaluate_ami(
     """
     Evaluate one AMI against spec v4. Returns a Finding or None.
 
-    Mandatory evaluation order (spec §6):
+    Mandatory evaluation order (spec 6):
       1. Parse + normalize
       2. Check deprecation override
       3. Fetch best-effort signals
@@ -181,7 +181,7 @@ def _evaluate_ami(
 
         # ── 4. EXCLUSION_RULES ────────────────────────────────────────────────
         # Rule: recently launched (lastLaunchedTime must exist and be recent).
-        # Missing lastLaunchedTime does NOT trigger exclusion (spec §5.A).
+        # Missing lastLaunchedTime does NOT trigger exclusion (spec 5.A).
         if days_since_launched is not None and days_since_launched < _RECENTLY_ACTIVE_DAYS:
             return None
 
@@ -201,7 +201,7 @@ def _evaluate_ami(
             return None
 
         # Conservative: unknown active-instance state + borderline score → skip.
-        # Absence of instance check ≠ absence of instances (spec §13).
+        # Absence of instance check ≠ absence of instances (spec 13).
         if instance_check_failed and score == 1:
             return None
 
@@ -222,10 +222,10 @@ def _evaluate_ami(
 
         # ── 8. Risk ───────────────────────────────────────────────────────────
         # MEDIUM: non-deprecated + BOTH signals (score 2).
-        # Guardrail: score 2 → risk MUST be >= MEDIUM (spec §9).
+        # Guardrail: score 2 → risk MUST be >= MEDIUM (spec 9).
         risk = RiskLevel.MEDIUM if score == 2 else RiskLevel.LOW
 
-        # Title (spec §16 + §13):
+        # Title (spec 16 + 13):
         # - Never label "Unused" if active instances exist.
         # - Never label "Unused" if active-instance state is UNKNOWN (check failed) —
         #   missing visibility ≠ "no instances" ≠ "unused".
@@ -364,15 +364,15 @@ def _build_evidence(  # noqa: PLR0913
     contextual_downgrade: bool,
 ) -> Evidence:
     """
-    Build Evidence per spec §15.
+    Build Evidence per spec 15.
     All fields must exist; null is allowed but fields must not be omitted.
-    Evaluation path must be exactly "deprecated" or "scored" (spec §17).
+    Evaluation path must be exactly "deprecated" or "scored" (spec 17).
     signals_not_checked: permission/visibility gaps first, then conceptual blind spots.
     """
     signals: List[str] = []
     not_checked: List[str] = []
 
-    # evaluation_path (spec §17)
+    # evaluation_path (spec 17)
     signals.append(f"evaluation_path: {evaluation_path}")
 
     # age + state
@@ -515,7 +515,7 @@ def _check_active_instances(ec2, ami_id: str) -> Tuple[bool, bool]:
     """
     Return (instances_found, check_failed).
     Existence check — MaxResults=5, EC2 filters server-side.
-    CRITICAL (spec §13): check_failed ≠ "no instances" — treat result as UNKNOWN.
+    CRITICAL (spec 13): check_failed ≠ "no instances" — treat result as UNKNOWN.
     """
     try:
         resp = ec2.describe_instances(
@@ -534,7 +534,7 @@ def _check_active_instances(ec2, ami_id: str) -> Tuple[bool, bool]:
 def _build_lt_index(ec2) -> Tuple[Dict[str, List[str]], bool]:
     """
     Build {ami_id: [lt_ids]} by checking $Default + $Latest versions of each LT.
-    Per spec §11: "prefer $Default + $Latest; full traversal optional."
+    Per spec 11: "prefer $Default + $Latest; full traversal optional."
 
     Phase 1: list all LT IDs (with guard at _LT_INDEX_GUARD).
     Phase 2: per-LT, query $Default + $Latest versions only (no pagination needed).

cleancloud/providers/aws/rules/cloudwatch_logs_no_retention.py

@@ -40,13 +40,13 @@
 from cleancloud.core.risk import RiskLevel
 
 # Approximate CloudWatch Logs storage cost per GB-month (us-east-1, 2024).
-# Informational only — must NOT influence detection or confidence (spec §9).
+# Informational only — must NOT influence detection or confidence (spec 9).
 _STORAGE_COST_PER_GB_APPROX = 0.03
 
-# Risk thresholds by stored size (spec §8)
+# Risk thresholds by stored size (spec 8)
 _HIGH_RISK_GB = 1.0  # ≥ 1 GB → HIGH; < 1 GB → MEDIUM; 0 bytes → LOW
 
-# Only these classes are eligible (spec §2). Allowlist — unknown/missing class is NOT in scope.
+# Only these classes are eligible (spec 2). Allowlist — unknown/missing class is NOT in scope.
 _ELIGIBLE_CLASSES = {"STANDARD", "INFREQUENT_ACCESS"}
 
 
@@ -62,16 +62,16 @@ def find_cloudwatch_logs_no_retention(
 
     for page in paginator.paginate():
         for lg in page.get("logGroups", []):
-            # EXCLUSION: malformed record (spec §2)
+            # EXCLUSION: malformed record (spec 2)
             if not lg.get("logGroupName"):
                 continue
 
-            # EXCLUSION: only STANDARD and INFREQUENT_ACCESS are in scope (spec §2, §4A).
+            # EXCLUSION: only STANDARD and INFREQUENT_ACCESS are in scope (spec 2, 4A).
             # DELIVERY is service-managed; unknown/missing class is not eligible.
             if lg.get("logGroupClass") not in _ELIGIBLE_CLASSES:
                 continue
 
-            # EXCLUSION: retention policy is set — key presence check, not value check (spec §4A).
+            # EXCLUSION: retention policy is set — key presence check, not value check (spec 4A).
             # "retentionInDays is not present in the returned log group object" means
             # key absent, not value null. An explicit null would still mean the key was
             # returned and should be treated as set.
@@ -91,22 +91,22 @@ def find_cloudwatch_logs_no_retention(
                 creation_time = None
                 age_days = None
 
-            # storedBytes is a non-billing, eventually-consistent storage metric (spec §3, §9).
+            # storedBytes is a non-billing, eventually-consistent storage metric (spec 3, 9).
             # It must NOT be used as an activity signal.
             stored_bytes: Optional[int] = lg.get("storedBytes")
             stored_gb: Optional[float] = (
                 (stored_bytes / (1024**3)) if stored_bytes is not None else None
             )
 
-            # Risk is proportional to stored size as a proxy for current storage exposure (spec §8)
+            # Risk is proportional to stored size as a proxy for current storage exposure (spec 8)
             if stored_gb is not None and stored_gb >= _HIGH_RISK_GB:
                 risk = RiskLevel.HIGH
             elif stored_bytes is not None and stored_bytes > 0:
                 risk = RiskLevel.MEDIUM
             else:
                 risk = RiskLevel.LOW
 
-            # Cost estimate — informational only (spec §9)
+            # Cost estimate — informational only (spec 9)
             monthly_storage_cost: Optional[float] = None
             if stored_bytes is not None and stored_bytes > 0 and stored_gb is not None:
                 monthly_storage_cost = round(stored_gb * _STORAGE_COST_PER_GB_APPROX, 2)

cleancloud/providers/aws/rules/ebs_snapshot_old.py

@@ -45,7 +45,7 @@
 
 _DEFAULT_MAX_AGE_DAYS: int = 90
 
-# Tag key prefix that indicates explicit AWS Backup management (spec §4, §5A.6).
+# Tag key prefix that indicates explicit AWS Backup management (spec 4, 5A.6).
 # Only aws:backup: is defined by this spec. DLM is not in scope.
 _BACKUP_TAG_PREFIX: str = "aws:backup:"
 
@@ -95,7 +95,7 @@ def _check_external_sharing(ec2, snap_id: str) -> Tuple[bool, bool]:
 
 
 def _is_backup_managed(snap: dict) -> bool:
-    """Return True if the snapshot has an explicit aws:backup: tag (spec §4, §5A.6).
+    """Return True if the snapshot has an explicit aws:backup: tag (spec 4, 5A.6).
 
     Only tag-based detection; full AWS Backup API integration is not in this spec.
     A negative result means UNKNOWN (no tag evidence found), not confirmed non-Backup.
@@ -113,7 +113,7 @@ def find_old_ebs_snapshots(
 ) -> List[Finding]:
     ec2 = session.client("ec2", region_name=region)
 
-    # Build AMI snapshot index before evaluating snapshots (spec §5A.4, §6, §10).
+    # Build AMI snapshot index before evaluating snapshots (spec 5A.4, 6, 10).
     # If this fails, AMI linkage cannot be verified → all candidates are skipped.
     ami_snapshot_ids, ami_index_failed = _build_ami_snapshot_index(ec2)
 
@@ -127,42 +127,42 @@ def find_old_ebs_snapshots(
             snap_id = snap.get("SnapshotId")
             start_time = snap.get("StartTime")
 
-            # EXCLUSION: malformed record (spec §3)
+            # EXCLUSION: malformed record (spec 3)
             if not snap_id or start_time is None:
                 continue
 
-            # EXCLUSION: status != completed (spec §5A.1)
+            # EXCLUSION: status != completed (spec 5A.1)
             if snap.get("State") != "completed":
                 continue
 
-            # EXCLUSION: non-standard storage tier (spec §5A.2)
+            # EXCLUSION: non-standard storage tier (spec 5A.2)
             # StorageTier absent → treated as standard per AWS default.
             storage_tier = snap.get("StorageTier", "standard")
             if storage_tier != "standard":
                 continue
 
-            # EXCLUSION: age threshold (spec §5A.3)
+            # EXCLUSION: age threshold (spec 5A.3)
             age_days = (now - start_time).days
             if age_days < max_age_days:
                 continue
 
-            # EXCLUSION: AMI linkage (spec §5A.4, §10)
+            # EXCLUSION: AMI linkage (spec 5A.4, 10)
             # If the index build failed, AMI linkage cannot be verified → SKIP.
             # Never treat missing visibility as "no AMI links".
             if ami_index_failed:
                 continue
             if snap_id in ami_snapshot_ids:
                 continue
 
-            # EXCLUSION: external sharing (spec §5A.5, §10)
+            # EXCLUSION: external sharing (spec 5A.5, 10)
             # Per-snapshot check. If the check fails → SKIP that snapshot.
             shared_externally, sharing_check_failed = _check_external_sharing(ec2, snap_id)
             if sharing_check_failed:
                 continue
             if shared_externally:
                 continue
 
-            # EXCLUSION: explicit AWS Backup-managed (spec §5A.6)
+            # EXCLUSION: explicit AWS Backup-managed (spec 5A.6)
             # Tag-based heuristic (aws:backup: prefix only). Only explicit True suppresses;
             # unknown (no tag evidence) does not block.
             if _is_backup_managed(snap):
@@ -201,7 +201,7 @@ def find_old_ebs_snapshots(
                     resource_type="aws.ebs.snapshot",
                     resource_id=snap_id,
                     region=region,
-                    estimated_monthly_cost_usd=None,  # spec §9: no cost from volumeSize
+                    estimated_monthly_cost_usd=None,  # spec 9: no cost from volumeSize
                     title="Old EBS snapshot review candidate",
                     summary=(
                         f"EBS snapshot is {age_days} days old "

cleancloud/providers/aws/rules/ebs_unattached.py

@@ -264,7 +264,7 @@ def find_unattached_ebs_volumes(
                     resource_type="aws.ebs.volume",
                     resource_id=v["volume_id"],
                     region=region,
-                    estimated_monthly_cost_usd=None,  # spec §9: flat rate invalid across volume types
+                    estimated_monthly_cost_usd=None,  # spec 9: flat rate invalid across volume types
                     title="Unattached EBS volume review candidate",
                     summary=(
                         f"EBS volume has been unattached for {age_days} days "

cleancloud/providers/aws/rules/ec2_sg_unused.py

@@ -359,7 +359,7 @@ def find_unused_security_groups(
                     "and the default-group exclusion did not match"
                 ),
                 risk=RiskLevel.LOW,
-                confidence=ConfidenceLevel.MEDIUM,  # spec §7: MEDIUM mandatory; HIGH not recommended
+                confidence=ConfidenceLevel.MEDIUM,  # spec 7: MEDIUM mandatory; HIGH not recommended
                 detected_at=now,
                 evidence=evidence,
                 details=details,

cleancloud/providers/aws/rules/ec2_stopped.py

@@ -347,7 +347,7 @@ def find_stopped_ec2_instances(
     now = datetime.now(timezone.utc)
     findings: List[Finding] = []
 
-    # Guard: lookup window must cover at least the threshold (spec §4).
+    # Guard: lookup window must cover at least the threshold (spec 4).
     # A shorter window cannot prove the required stopped duration.
     if cloudtrail_lookup_days < stopped_age_threshold_days:
         return findings

cleancloud/providers/azure/rules/app_gateway_no_backends.py

@@ -44,7 +44,7 @@
 from cleancloud.core.risk import RiskLevel
 
 # ---------------------------------------------------------------------------
-# Module constants  (spec §17)
+# Module constants  (spec 17)
 # ---------------------------------------------------------------------------
 
 _EVALUATION_PATH = "app-gateway-no-backends"
@@ -125,11 +125,11 @@ def _normalize_pool(pool) -> Optional[dict]:
     raw_name = _get_str(pool, "name")
     pool_name = raw_name or _name_from_id(pool_id)
 
-    # backendAddresses — canonical target source (spec §2)
+    # backendAddresses — canonical target source (spec 2)
     backend_addresses = _get_list(pool, "backend_addresses") or _get_list(pool, "backendAddresses")
     backend_addresses = [a for a in backend_addresses if a is not None]
 
-    # backendIPConfigurations — optional legacy/read-only field (spec §2)
+    # backendIPConfigurations — optional legacy/read-only field (spec 2)
     legacy_cfgs = _get_list(pool, "backend_ip_configurations") or _get_list(
         pool, "backendIPConfigurations"
     )
@@ -421,7 +421,7 @@ def _traverse_url_path_map(
             pool_route_refs,
             pool_rule_ids,
             diags,
-            ldp_keyword="LoadDistributionPolicy",  # spec §4 canonical
+            ldp_keyword="LoadDistributionPolicy",  # spec 4 canonical
         )
 
     # Path rules
@@ -627,7 +627,7 @@ def _traverse_gateway(
             or getattr(rule, "loadDistributionPolicy", None)
         )
 
-        # Redirect presence: spec §6 — presence of the field is sufficient; a malformed
+        # Redirect presence: spec 6 — presence of the field is sufficient; a malformed
         # (non-null, non-resolvable) redirectConfiguration ref still counts as present.
         redirect_present = redirect_ref is not None
         if redirect_present and _norm_id(redirect_ref) is None:
@@ -728,7 +728,7 @@ def _traverse_gateway(
     now = datetime.now(timezone.utc)
 
     # Build signals_not_checked: blind-spot strings + structured diagnostic dicts
-    # (spec §5 diagnostics contract requires minimum structured shape)
+    # (spec 5 diagnostics contract requires minimum structured shape)
     def _diag_key(diag) -> tuple:
         if isinstance(diag, dict):
             return (