feat(scanner): migrate storage rules AZ-STOR-001 to 010 to evaluate() (#370)

CHANGELOG.md

@@ -9,6 +9,7 @@ OpenShield uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Added
 
+- Add the evaluate() contract foundation for issue #369: `AzureClient.list_storage_accounts()` and `list_key_vaults()` return `None` when the list call fails, shared `inventory_unavailable()`, `no_resources_found()` and `fail_findings()` helpers with standard reason codes in `scanner/evaluation.py`, and `tests/test_rule_evaluation_contract.py`, which every rule exposing `evaluate()` must pass.
 - Add the remediation approval, idempotency, audit and rescan gate (`api/models/remediation.py`, `docs/remediation-gate.md`) for issue #266: approved-only, allowlisted, at-most-once execution grants with an append-only audit trail and rescan-based verification. Proposal-only; nothing executes a playbook.
 - Add all ten evidence-rich enterprise network and perimeter controls `AZ-NET-018` through `AZ-NET-027` for issue #253, preserving API failures and incomplete data as indeterminate.
 - Branch protection declared as versioned GitHub rulesets for `dev` and `main`, with a scheduled drift audit that retains evidence (#298)
@@ -32,10 +33,13 @@ OpenShield uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Changed
 
+- Migrate the ten storage rules `AZ-STOR-001` to `AZ-STOR-010` to the `evaluate()` coverage contract for issue #370. Each rule now reports PASS, FAIL, UNKNOWN, ERROR or NOT_APPLICABLE for every storage account (per sub-service for `AZ-STOR-004`, per blob container for `AZ-STOR-009`), records the inspected value as evidence, and reports a failed storage inventory as ERROR. Findings are unchanged, except that `AZ-STOR-006` no longer emits a finding with an empty `resource_id` for an account returned without an ID; that account is reported as UNKNOWN.
+- The scan engine runs a rule that exposes `evaluate()` through `evaluate()` only and takes its findings from the FAIL evaluations, instead of also calling `scan()`. A rule whose `evaluate()` crashes, returns malformed data, or reports at least one `ERROR` evaluation (such as `INVENTORY_UNAVAILABLE`) is listed in `failed_rule_ids`; a per-resource gap is `UNKNOWN` and does not list the rule. Items that are not `RuleEvaluation` objects are dropped individually, keeping the rule's valid evaluations and findings, and recorded as one `MALFORMED_EVALUATION` error (#369).
 - Migrate the ISO 27001 mapping pack from ISO/IEC 27001:2013 to ISO/IEC 27001:2022 Annex A (mapping pack 2.0.0) and each rule's own `ISO27001` value, for issue #358. Rules about monitoring coverage use the new A.8.16 and Kubernetes policy governance uses A.8.9. All entries stay `pending_review`. Scans saved with a mapping snapshot keep reporting against the 2013 controls.
 
 ### Fixed
 
+- `AZ-KV-006` reports `ERROR` (`INVENTORY_UNAVAILABLE`) when the Key Vault list call fails, instead of `NOT_APPLICABLE` as if the subscription had no vaults (#369).
 - Learn page statistics are computed at site build time, and the post-merge stats workflow no longer pushes to the protected `dev` branch (it failed on every merge); README count drift is now reported as a warning
 - High-severity CodeQL findings in Python and JavaScript code
 - Security findings identified during Semgrep analysis

api/models/finding.py

@@ -1817,13 +1817,13 @@ def get_compliance_score(self, framework: str, subscription_id: Optional[str] =
                 current["severity"] = severity
                 current["category"] = category
 
-        # Rules the scan engine could not complete for this scan (raised, or
-        # returned malformed data) - recorded by save_scan() alongside the
-        # mapping snapshot. This is a supplementary signal on top of
-        # rule_evaluations: it can only ever make a control's status stricter
-        # (force ERROR), never looser, and an ERROR still counts in the
-        # denominator, so a rule that failed to run can never improve a score
-        # (issue #302/#263).
+        # Rules with no usable evaluator output (raised, returned malformed
+        # data, or returned only ERROR evaluations) - recorded by save_scan()
+        # alongside the mapping snapshot. This is a supplementary signal on
+        # top of rule_evaluations: it can only ever make a control's status
+        # stricter (force ERROR), never looser, and an ERROR still counts in
+        # the denominator, so a rule that failed to run can never improve a
+        # score (issue #302/#263).
         unevaluated_rule_ids = set((snapshot.get("_scan_rule_outcomes") or {}).get("failed_rule_ids") or [])
 
         results = []

docs/adding-a-rule.md

@@ -149,28 +149,61 @@ When a helper returns `None`, skip the resource and log a warning. Never create
 
 ---
 
-## Optional: Reporting Evaluation Coverage (`evaluate()`)
+## Reporting Evaluation Coverage (`evaluate()`)
 
-`scan()` only ever reports violations, so a scan with no findings for your rule is indistinguishable from "everything is compliant," "nothing of this resource type exists," and "the rule errored before it could check anything." A rule can additionally expose:
+`scan()` only ever reports violations, so a scan with no findings for your rule is indistinguishable from "everything is compliant," "nothing of this resource type exists," and "the rule errored before it could check anything." New rules should expose `evaluate()` and keep `scan()` as a thin wrapper over it:
 
 ```python
-from scanner.evaluation import EvaluationStatus, RuleEvaluation, subscription_scope_id
+from scanner.evaluation import (
+    EvaluationStatus,
+    RuleEvaluation,
+    fail_findings,
+    inventory_unavailable,
+    no_resources_found,
+)
+
+RESOURCE_TYPE = "Microsoft.Xxx/yyy"
+
+
+def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]:
+    return fail_findings(evaluate(azure_client, subscription_id))
 
 
 def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]:
     """Report a status for every resource this rule looked at, PASS included."""
+    resources = azure_client.list_xxx()  # None means the list call failed
+    if resources is None:
+        return [inventory_unavailable(RULE_ID, RESOURCE_TYPE, subscription_id)]
+    if not resources:
+        return [no_resources_found(RULE_ID, RESOURCE_TYPE, subscription_id)]
+    ...
 ```
 
-to state a `PASS`/`FAIL`/`UNKNOWN`/`ERROR`/`NOT_APPLICABLE` result per resource instead of only per violation. This is additive: `scan()` keeps working unchanged, and a rule without `evaluate()` still runs, its coverage is just recorded as `UNKNOWN`/`LEGACY_RULE_NOT_MIGRATED` rather than assumed to be a pass.
+When a rule exposes `evaluate()`, the engine runs it **instead of** `scan()` and takes the rule's findings from its `FAIL` evaluations. A rule without `evaluate()` still runs through `scan()`, but its coverage is recorded as `UNKNOWN`/`LEGACY_RULE_NOT_MIGRATED` and every compliance control mapped to it reads `UNKNOWN`. Migration of the existing rules is tracked in #380.
 
-Rules of the contract (see `scanner/evaluation.py` and `scanner/rules/az_kv_006.py` for the reference implementation):
+The scan result's `failed_rule_ids` identifies evaluator crashes, malformed evaluator output, and rules that returned only `ERROR` evaluations (including inventory-wide failures with no usable outcomes). Partial `ERROR` rows do not mark a rule failed when it also returned a usable `PASS`, `FAIL`, `UNKNOWN`, or `NOT_APPLICABLE` outcome; the error rows remain in `evaluations` and still affect compliance roll-up. Thus even all-error per-resource results count as a rule failure because none of the rule's coverage could be used.
 
-- `resource_id` must be a real, non-empty identifier. For a subscription-level result with no single resource to blame, use `subscription_scope_id(subscription_id)`, never `""`.
-- `UNKNOWN`, `ERROR`, and `NOT_APPLICABLE` require a `reason_code` explaining why — never leave one unexplained.
-- A `FAIL` result may attach `finding=` with the same dict shape `scan()` returns; the engine deduplicates it against anything `scan()` already reported for the same `(rule_id, resource_id)`, so implementing both never double-counts.
-- If you can't tell "no resources of this type exist" apart from "the list call failed" (a real gap in some `AzureClient` methods today), report `NOT_APPLICABLE` rather than guessing `PASS`.
+Rules of the contract (see `scanner/evaluation.py`, `scanner/rules/az_kv_006.py` for a single-branch rule, and `scanner/rules/az_stor_*.py` for rules with per-resource lookups, sub-resources and opt-in policy tags):
+
+- Use an inventory call that returns `None` on failure (`list_storage_accounts()`, `list_key_vaults()`, and the `Optional[List]` getters such as `get_managed_clusters()`). A failed inventory must be `ERROR`, never `PASS` or `NOT_APPLICABLE`. If the method you need returns `[]` on failure, add a `list_*()` variant next to it.
+- Every path that would otherwise be a silent `continue` gets an explicit status, using the standard reason codes:
+
+  | Situation | Status | `reason_code` |
+  |---|---|---|
+  | Inventory call failed | `ERROR` | `INVENTORY_UNAVAILABLE` |
+  | Inventory succeeded but is empty | `NOT_APPLICABLE` | `NO_RESOURCES_FOUND` |
+  | Per-resource lookup returned `None` | `UNKNOWN` | `EVIDENCE_UNAVAILABLE` |
+  | Resource lacks a required field | `UNKNOWN` | `MISSING_PROPERTIES` |
+  | Opt-in policy tag not set | `NOT_APPLICABLE` | `POLICY_NOT_REQUIRED` |
+  | Approved exception tag set | `NOT_APPLICABLE` | `APPROVED_EXCEPTION` |
 
-You don't need to migrate an existing rule's `scan()` to add `evaluate()` — most rules can leave `scan()` exactly as-is.
+  For opt-in policy tags, `policy_exemption()` in `scanner/rules/_storage_common.py` returns the right `NOT_APPLICABLE` reason code, or `None` when the policy applies. A resource returned without an ID gets `missing_resource_id()` from `scanner/evaluation.py`.
+
+- `resource_id` must be a real, non-empty identifier. For a subscription-level result with no single resource to blame, use `subscription_scope_id(subscription_id)`, never `""`.
+- `UNKNOWN`, `ERROR`, and `NOT_APPLICABLE` require a `reason_code`.
+- Every `FAIL` must attach `finding=` with the same dict shape `scan()` returns; a `FAIL` without one is dropped from the findings list.
+- Record the value(s) you checked in `evidence`, for example `{"enable_rbac_authorization": False}`.
+- Register a fixture with at least one compliant and one non-compliant resource in `tests/test_rule_evaluation_contract.py`. That test runs against every rule exposing `evaluate()` and checks the failure, empty, finding-shape and `scan()` parity rules above.
 
 ---
 

scanner/azure_client.py

@@ -111,14 +111,23 @@ def parse_resource_id(resource_id: str) -> Dict[str, str]:
     # Storage                                                               #
     # ------------------------------------------------------------------ #
 
-    def get_storage_accounts(self) -> List[Any]:
-        """List all storage accounts in the subscription."""
+    def list_storage_accounts(self) -> Optional[List[Any]]:
+        """List all storage accounts in the subscription.
+
+        Returns ``None`` when the list call fails, so evaluate() can report
+        ERROR instead of mistaking a permissions or network failure for a
+        subscription with no storage accounts.
+        """
         try:
             client = StorageManagementClient(self.credential, self.subscription_id)
             return list(client.storage_accounts.list())
         except Exception as exc:
-            logger.error("get_storage_accounts failed: %s", exc)
-            return []
+            logger.error("list_storage_accounts failed: %s", exc)
+            return None
+
+    def get_storage_accounts(self) -> List[Any]:
+        """List all storage accounts, or ``[]`` on failure (legacy scan() callers)."""
+        return self.list_storage_accounts() or []
 
     def get_storage_lifecycle_policy(self, resource_group: str, account_name: str) -> Optional[bool]:
         """Check whether a storage account has a lifecycle management policy.
@@ -1152,14 +1161,21 @@ def get_managed_caches(self) -> Optional[List[Any]]:
     # Key Vault                                                             #
     # ------------------------------------------------------------------ #
 
-    def get_key_vaults(self) -> List[Any]:
-        """List all Key Vaults in the subscription with full properties."""
+    def list_key_vaults(self) -> Optional[List[Any]]:
+        """List all Key Vaults in the subscription with full properties.
+
+        Returns ``None`` when the list call fails; see list_storage_accounts().
+        """
         try:
             client = KeyVaultManagementClient(self.credential, self.subscription_id)
             return list(client.vaults.list_by_subscription())
         except Exception as exc:
-            logger.error("get_key_vaults failed: %s", exc)
-            return []
+            logger.error("list_key_vaults failed: %s", exc)
+            return None
+
+    def get_key_vaults(self) -> List[Any]:
+        """List all Key Vaults, or ``[]`` on failure (legacy scan() callers)."""
+        return self.list_key_vaults() or []
 
     def get_key_vault_certificates(self, vault_name: str) -> List[Any]:
         """List all certificates in a Key Vault using the Key Vault data plane API."""