diff --git a/CHANGELOG.md b/CHANGELOG.md index 196a0af8..44de247a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/api/models/finding.py b/api/models/finding.py index c68050a2..5c9c7602 100644 --- a/api/models/finding.py +++ b/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 = [] diff --git a/docs/adding-a-rule.md b/docs/adding-a-rule.md index 8780fb34..cbe62fba 100644 --- a/docs/adding-a-rule.md +++ b/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. --- diff --git a/scanner/azure_client.py b/scanner/azure_client.py index bf816694..315ac571 100644 --- a/scanner/azure_client.py +++ b/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.""" diff --git a/scanner/engine.py b/scanner/engine.py index 02137309..1fef83df 100644 --- a/scanner/engine.py +++ b/scanner/engine.py @@ -5,7 +5,7 @@ import uuid from datetime import datetime, timezone from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from api.observability import RULE_ERRORS_TOTAL from openshield.severity import CONTRACT_VERSION, SeverityContractError, normalize_severity, score_findings @@ -52,7 +52,9 @@ class ScanEngine: Each rule module must expose a ``scan(azure_client, subscription_id)`` function and the module-level constants ``RULE_ID``, ``RULE_NAME``, ``SEVERITY``, ``CATEGORY``, ``FRAMEWORKS``, ``DESCRIPTION``, - ``REMEDIATION``, and ``PLAYBOOK``. + ``REMEDIATION``, and ``PLAYBOOK``. A rule that also exposes + ``evaluate(azure_client, subscription_id)`` is run through evaluate() + only; its findings come from its FAIL evaluations. """ def __init__(self, subscription_id: str) -> None: @@ -104,7 +106,10 @@ def run_scan(self, scan_id: Optional[str] = None) -> Dict[str, Any]: Returns: dict with keys: scan_id, subscription_id, started_at, - completed_at, total_findings, findings. + completed_at, total_findings, findings, evaluations, and + failed_rule_ids (rules whose evaluator raised, returned malformed + data, or produced only ERROR evaluations; partial ERROR results + do not mark a rule failed when it also produced usable outcomes). """ scan_id = scan_id or str(uuid.uuid4()) started_at = datetime.now(timezone.utc).isoformat() @@ -123,13 +128,29 @@ def run_scan(self, scan_id: Optional[str] = None) -> Dict[str, Any]: # equivalent to "the rule ran and found nothing" - a caller scoring # PASS/FAIL from absence of findings (get_compliance_score()) must be # able to tell the two apart, or a crashed rule reads as a clean - # pass. Full per-resource evaluation persistence is issue #263; - # tracking which rules failed to complete at all is the minimum this - # scan result can honestly report without it. + # pass. A rule is listed here when it raised, returned malformed + # data, or produced only ERROR evaluations. Partial ERROR rows do not + # fail a rule when it also produced usable outcomes; if every outcome + # is ERROR, the rule supplied no usable coverage and is failed. This + # also covers inventory-wide failures, while preserving valid results + # from other inventories/resources. failed_rule_ids: List[str] = [] for rule in self.rules: rule_id = getattr(rule, "RULE_ID", "UNKNOWN") + if callable(getattr(rule, "evaluate", None)): + # evaluate() supersedes scan(): its FAIL evaluations carry the + # findings (collected below), so also calling scan() would only + # repeat the same Azure list calls and let the two paths drift. + rule_evaluations, completed = self._run_evaluate(rule, rule_id) + if not completed or ( + rule_evaluations and all(e.status == EvaluationStatus.ERROR for e in rule_evaluations) + ): + failed_rule_ids.append(rule_id) + evaluations.extend(rule_evaluations) + logger.info("Rule %s produced %d evaluation(s)", rule_id, len(rule_evaluations)) + continue + try: rule_findings = rule.scan(self.client, self.subscription_id) if not isinstance(rule_findings, list): @@ -159,14 +180,23 @@ def run_scan(self, scan_id: Optional[str] = None) -> Dict[str, Any]: logger.error("Rule %s raised an exception: %s", rule_id, exc, exc_info=True) failed_rule_ids.append(rule_id) - evaluations.extend(self._evaluate_rule(rule, rule_id)) + evaluations.append(self._legacy_placeholder(rule_id)) - # A FAIL evaluation contributes its own finding only if scan() hasn't - # already reported the same (rule_id, resource_id) violation, so a - # rule implementing both scan() and evaluate() never double-counts. + # Every FAIL evaluation contributes its attached finding once per + # (rule_id, resource_id), so repeated FAILs for one resource never + # double-count. existing_keys = {(f.get("rule_id"), f.get("resource_id")) for f in findings} for rule_evaluation in evaluations: - if rule_evaluation.status != EvaluationStatus.FAIL or not rule_evaluation.finding: + if rule_evaluation.status != EvaluationStatus.FAIL: + continue + if not rule_evaluation.finding: + # A FAIL without a finding would drop a real violation from + # the findings list while the coverage row still says FAIL. + logger.warning( + "Rule %s reported FAIL for %s without a finding", + rule_evaluation.rule_id, + rule_evaluation.resource_id, + ) continue key = (rule_evaluation.rule_id, rule_evaluation.resource_id) if key in existing_keys: @@ -201,42 +231,61 @@ def run_scan(self, scan_id: Optional[str] = None) -> Dict[str, Any]: return make_serializable(result) - def _evaluate_rule(self, rule: Any, rule_id: str) -> List[RuleEvaluation]: - """Return this rule's coverage statements for the current scan. + def _legacy_placeholder(self, rule_id: str) -> RuleEvaluation: + """Coverage for a rule that only has scan(). - A rule that exposes ``evaluate()`` reports its own PASS/FAIL/UNKNOWN - results. A rule that only has ``scan()`` has never stated what it - looked at, so its coverage is recorded as UNKNOWN rather than - inferred as PASS from the absence of a finding. + Such a rule has never stated what it looked at, so its coverage is + recorded as UNKNOWN rather than inferred as PASS from the absence of + a finding. """ - evaluate_fn = getattr(rule, "evaluate", None) - if not callable(evaluate_fn): - return [ - RuleEvaluation( - rule_id=rule_id, - resource_id=subscription_scope_id(self.subscription_id), - resource_type="", - status=EvaluationStatus.UNKNOWN, - reason_code="LEGACY_RULE_NOT_MIGRATED", - reason="This rule has not been migrated to the evaluate() coverage contract yet.", - ) - ] + return RuleEvaluation( + rule_id=rule_id, + resource_id=subscription_scope_id(self.subscription_id), + resource_type="", + status=EvaluationStatus.UNKNOWN, + reason_code="LEGACY_RULE_NOT_MIGRATED", + reason="This rule has not been migrated to the evaluate() coverage contract yet.", + ) + + def _run_evaluate(self, rule: Any, rule_id: str) -> Tuple[List[RuleEvaluation], bool]: + """Run a rule's evaluate() and report whether it completed. + A raised exception or a non-list return is recorded as a single ERROR + at the subscription scope. Items that are not RuleEvaluation objects + are dropped (they must not reach result serialisation, where they + would fail the whole scan), the valid evaluations are kept so their + FAIL findings still surface, and one ERROR naming the malformed + positions is added so the rule cannot read as clean. + """ try: - rule_evaluations = evaluate_fn(self.client, self.subscription_id) + rule_evaluations = rule.evaluate(self.client, self.subscription_id) if not isinstance(rule_evaluations, list): raise TypeError(f"evaluate() must return a list, got {type(rule_evaluations)}") - return rule_evaluations except Exception as exc: RULE_ERRORS_TOTAL.labels(rule_id=rule_id).inc() logger.error("Rule %s evaluate() raised an exception: %s", rule_id, exc, exc_info=True) - return [ - RuleEvaluation( - rule_id=rule_id, - resource_id=subscription_scope_id(self.subscription_id), - resource_type="", - status=EvaluationStatus.ERROR, - reason_code="EVALUATOR_EXCEPTION", - reason=str(exc), - ) - ] + return [self._evaluator_error(rule_id, "EVALUATOR_EXCEPTION", str(exc))], False + + valid = [item for item in rule_evaluations if isinstance(item, RuleEvaluation)] + malformed = [ + f"#{index} ({type(item).__name__})" + for index, item in enumerate(rule_evaluations) + if not isinstance(item, RuleEvaluation) + ] + if not malformed: + return rule_evaluations, True + + RULE_ERRORS_TOTAL.labels(rule_id=rule_id).inc() + reason = f"evaluate() returned {len(malformed)} item(s) that are not RuleEvaluation: {', '.join(malformed)}" + logger.error("Rule %s %s", rule_id, reason) + return valid + [self._evaluator_error(rule_id, "MALFORMED_EVALUATION", reason)], False + + def _evaluator_error(self, rule_id: str, reason_code: str, reason: str) -> RuleEvaluation: + return RuleEvaluation( + rule_id=rule_id, + resource_id=subscription_scope_id(self.subscription_id), + resource_type="", + status=EvaluationStatus.ERROR, + reason_code=reason_code, + reason=reason, + ) diff --git a/scanner/evaluation.py b/scanner/evaluation.py index 4b89fd1c..13b0bd75 100644 --- a/scanner/evaluation.py +++ b/scanner/evaluation.py @@ -8,13 +8,16 @@ def evaluate(azure_client, subscription_id) -> List[RuleEvaluation] which reports a status for every resource (or the subscription itself) it -looked at, PASS included. Rules that don't expose ``evaluate`` keep working -exactly as before via ``scan()``; the engine records their coverage as +looked at, PASS included. When a rule exposes ``evaluate``, the engine runs +it instead of ``scan()`` and takes findings from its FAIL evaluations; a +migrated rule keeps ``scan()`` as a thin wrapper over ``fail_findings()``. +Rules that don't expose ``evaluate`` keep working exactly as before via +``scan()``; the engine records their coverage as UNKNOWN/LEGACY_RULE_NOT_MIGRATED instead of inventing a PASS. """ from dataclasses import dataclass, field -from typing import Any, Dict, Iterable, Optional +from typing import Any, Dict, Iterable, List, Optional # Conservative rank: worse coverage information must never be hidden by # better information rolled up from a different resource under the same rule. @@ -47,6 +50,29 @@ def subscription_scope_id(subscription_id: str) -> str: return f"/subscriptions/{subscription_id}" +# Standard reason codes shared across rule families, so the same situation +# reads the same way in every compliance report: +# +# INVENTORY_UNAVAILABLE ERROR the resource list call failed +# NO_RESOURCES_FOUND NOT_APPLICABLE the list call succeeded but was empty +# EVIDENCE_UNAVAILABLE UNKNOWN a per-resource lookup returned None +# MISSING_PROPERTIES UNKNOWN the resource lacked a required field +# POLICY_NOT_REQUIRED NOT_APPLICABLE opt-in policy tag not set +# APPROVED_EXCEPTION NOT_APPLICABLE approved exception tag set +# +# ERROR indicates unavailable evaluation evidence. The scan-level +# ``failed_rule_ids`` list is stricter: partial ERROR rows do not fail a rule +# if it also returned usable outcomes, but a rule with only ERROR rows has no +# usable coverage and is listed as failed. A per-resource evidence gap should +# normally be UNKNOWN, not ERROR. +INVENTORY_UNAVAILABLE = "INVENTORY_UNAVAILABLE" +NO_RESOURCES_FOUND = "NO_RESOURCES_FOUND" +EVIDENCE_UNAVAILABLE = "EVIDENCE_UNAVAILABLE" +MISSING_PROPERTIES = "MISSING_PROPERTIES" +POLICY_NOT_REQUIRED = "POLICY_NOT_REQUIRED" +APPROVED_EXCEPTION = "APPROVED_EXCEPTION" + + @dataclass class RuleEvaluation: """One rule's coverage statement about one resource (or the subscription).""" @@ -94,3 +120,52 @@ def aggregate_status(statuses: Iterable[str]) -> str: if best is None: raise ValueError("aggregate_status requires at least one status") return best + + +def inventory_unavailable(rule_id: str, resource_type: str, subscription_id: str) -> "RuleEvaluation": + """ERROR for a rule whose resource list call failed (inventory returned None). + + A failed list must never read as PASS or NOT_APPLICABLE: the rule did not + get to look at anything. + """ + return RuleEvaluation( + rule_id=rule_id, + resource_id=subscription_scope_id(subscription_id), + resource_type=resource_type, + status=EvaluationStatus.ERROR, + reason_code=INVENTORY_UNAVAILABLE, + reason=f"The {resource_type} inventory could not be read for this subscription.", + ) + + +def no_resources_found(rule_id: str, resource_type: str, subscription_id: str) -> "RuleEvaluation": + """NOT_APPLICABLE for a rule whose list call succeeded but returned nothing.""" + return RuleEvaluation( + rule_id=rule_id, + resource_id=subscription_scope_id(subscription_id), + resource_type=resource_type, + status=EvaluationStatus.NOT_APPLICABLE, + reason_code=NO_RESOURCES_FOUND, + reason=f"No {resource_type} resources were returned for this subscription.", + ) + + +def missing_resource_id(rule_id: str, resource_type: str, subscription_id: str) -> "RuleEvaluation": + """UNKNOWN for a resource returned without an ID. + + It cannot be evaluated or remediated, and a finding with an empty + resource_id would collide with every other one in persistence. + """ + return RuleEvaluation( + rule_id=rule_id, + resource_id=subscription_scope_id(subscription_id), + resource_type=resource_type, + status=EvaluationStatus.UNKNOWN, + reason_code=MISSING_PROPERTIES, + reason=f"A {resource_type} resource was returned without a resource ID.", + ) + + +def fail_findings(evaluations: Iterable["RuleEvaluation"]) -> List[Dict[str, Any]]: + """Findings attached to FAIL evaluations, for a migrated rule's scan() wrapper.""" + return [e.finding for e in evaluations if e.status == EvaluationStatus.FAIL and e.finding] diff --git a/scanner/rules/_storage_common.py b/scanner/rules/_storage_common.py index 42a532ce..7c14a819 100644 --- a/scanner/rules/_storage_common.py +++ b/scanner/rules/_storage_common.py @@ -1,6 +1,8 @@ """Shared opt-in policy metadata for enterprise storage controls.""" -from typing import Any, Mapping +from typing import Any, Mapping, Optional + +from scanner.evaluation import APPROVED_EXCEPTION, POLICY_NOT_REQUIRED def tags_for(resource: Any) -> Mapping[str, Any]: @@ -19,6 +21,18 @@ def approved_exception(resource: Any) -> bool: return tag_true(resource, "oshield:exception-approved") +def policy_exemption(resource: Any, requirement_tag: str) -> Optional[str]: + """Return why an opt-in policy does not apply, or ``None`` when it does. + + The reason is the NOT_APPLICABLE reason_code evaluate() reports. + """ + if not tag_true(resource, requirement_tag): + return POLICY_NOT_REQUIRED + if approved_exception(resource): + return APPROVED_EXCEPTION + return None + + def policy_required(resource: Any, requirement_tag: str) -> bool: """Use explicit opt-in metadata to avoid assuming every account is critical.""" - return tag_true(resource, requirement_tag) and not approved_exception(resource) + return policy_exemption(resource, requirement_tag) is None diff --git a/scanner/rules/az_kv_006.py b/scanner/rules/az_kv_006.py index b4c755b6..43681e38 100644 --- a/scanner/rules/az_kv_006.py +++ b/scanner/rules/az_kv_006.py @@ -2,7 +2,14 @@ from typing import Any, Dict, List -from scanner.evaluation import EvaluationStatus, RuleEvaluation, subscription_scope_id +from scanner.evaluation import ( + MISSING_PROPERTIES, + EvaluationStatus, + RuleEvaluation, + fail_findings, + inventory_unavailable, + no_resources_found, +) RULE_ID = "AZ-KV-006" RULE_NAME = "Key Vault Using Legacy Access Policies Instead of Azure RBAC" @@ -22,6 +29,7 @@ "Note: switching to RBAC does not delete existing access policies, but they stop being enforced." ) PLAYBOOK = "playbooks/cli/fix_az_kv_006.sh" +RESOURCE_TYPE = "Microsoft.KeyVault/vaults" def _finding(azure_client: Any, vault: Any) -> Dict[str, Any]: @@ -33,7 +41,7 @@ def _finding(azure_client: Any, vault: Any) -> Dict[str, Any]: "category": CATEGORY, "resource_id": vault.id, "resource_name": vault.name, - "resource_type": "Microsoft.KeyVault/vaults", + "resource_type": RESOURCE_TYPE, "description": DESCRIPTION, "remediation": REMEDIATION, "playbook": PLAYBOOK, @@ -47,41 +55,17 @@ def _finding(azure_client: Any, vault: Any) -> Dict[str, Any]: def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: """Detect Key Vaults where enable_rbac_authorization is False or None.""" - findings: List[Dict[str, Any]] = [] - - for vault in azure_client.get_key_vaults(): - props = getattr(vault, "properties", None) - if props is None: - continue - - # Access policies are the legacy default; a vault must opt into RBAC. - rbac_enabled = getattr(props, "enable_rbac_authorization", False) - if not rbac_enabled: - findings.append(_finding(azure_client, vault)) - - return findings + return fail_findings(evaluate(azure_client, subscription_id)) def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: """Report this rule's coverage: a status for every vault it looked at, - PASS included, instead of only reporting violations via scan().""" - vaults = azure_client.get_key_vaults() + PASS included, instead of only reporting violations.""" + vaults = azure_client.list_key_vaults() + if vaults is None: + return [inventory_unavailable(RULE_ID, RESOURCE_TYPE, subscription_id)] if not vaults: - # AzureClient.get_key_vaults() returns [] both when there genuinely - # are no vaults and when the list call itself failed — evaluate() - # can't tell those apart on its own, so it reports NOT_APPLICABLE - # rather than claiming a PASS it can't actually back up. Closing - # that ambiguity with Azure Resource Graph is tracked separately. - return [ - RuleEvaluation( - rule_id=RULE_ID, - resource_id=subscription_scope_id(subscription_id), - resource_type="Microsoft.KeyVault/vaults", - status=EvaluationStatus.NOT_APPLICABLE, - reason_code="NO_RESOURCES_FOUND", - reason="No Key Vaults were returned for this subscription.", - ) - ] + return [no_resources_found(RULE_ID, RESOURCE_TYPE, subscription_id)] evaluations: List[RuleEvaluation] = [] for vault in vaults: @@ -91,22 +75,25 @@ def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: RuleEvaluation( rule_id=RULE_ID, resource_id=vault.id, - resource_type="Microsoft.KeyVault/vaults", + resource_type=RESOURCE_TYPE, status=EvaluationStatus.UNKNOWN, - reason_code="MISSING_PROPERTIES", + reason_code=MISSING_PROPERTIES, reason="Key Vault was returned without a properties payload.", ) ) continue + # Access policies are the legacy default; a vault must opt into RBAC. rbac_enabled = getattr(props, "enable_rbac_authorization", False) + evidence = {"enable_rbac_authorization": rbac_enabled} if rbac_enabled: evaluations.append( RuleEvaluation( rule_id=RULE_ID, resource_id=vault.id, - resource_type="Microsoft.KeyVault/vaults", + resource_type=RESOURCE_TYPE, status=EvaluationStatus.PASS, + evidence=evidence, ) ) else: @@ -114,8 +101,9 @@ def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: RuleEvaluation( rule_id=RULE_ID, resource_id=vault.id, - resource_type="Microsoft.KeyVault/vaults", + resource_type=RESOURCE_TYPE, status=EvaluationStatus.FAIL, + evidence=evidence, finding=_finding(azure_client, vault), ) ) diff --git a/scanner/rules/az_stor_001.py b/scanner/rules/az_stor_001.py index fbe298df..ba948e0d 100644 --- a/scanner/rules/az_stor_001.py +++ b/scanner/rules/az_stor_001.py @@ -2,6 +2,16 @@ from typing import Any, Dict, List +from scanner.evaluation import ( + MISSING_PROPERTIES, + EvaluationStatus, + RuleEvaluation, + fail_findings, + inventory_unavailable, + missing_resource_id, + no_resources_found, +) + RULE_ID = "AZ-STOR-001" RULE_NAME = "Public Blob Access Enabled on Storage Account" SEVERITY = "HIGH" @@ -17,28 +27,65 @@ "Navigate to Storage Account > Configuration > Blob public access and set it to Disabled." ) PLAYBOOK = "playbooks/cli/fix_az_stor_001.sh" +RESOURCE_TYPE = "Microsoft.Storage/storageAccounts" -def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: - """Detect storage accounts with allow_blob_public_access set to True.""" - findings: List[Dict[str, Any]] = [] - - for account in azure_client.get_storage_accounts(): - if getattr(account, "allow_blob_public_access", False): - findings.append( - { - "rule_id": RULE_ID, - "rule_name": RULE_NAME, - "severity": SEVERITY, - "category": CATEGORY, - "resource_id": account.id, - "resource_name": account.name, - "resource_type": "Microsoft.Storage/storageAccounts", - "description": DESCRIPTION, - "remediation": REMEDIATION, - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - } +def _finding(account: Any) -> Dict[str, Any]: + return { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": account.id, + "resource_name": account.name, + "resource_type": "Microsoft.Storage/storageAccounts", + "description": DESCRIPTION, + "remediation": REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + } + + +def _evaluate_account(azure_client: Any, account: Any, subscription_id: str) -> List[RuleEvaluation]: + if not getattr(account, "id", None): + return [missing_resource_id(RULE_ID, RESOURCE_TYPE, subscription_id)] + value = getattr(account, "allow_blob_public_access", None) + if value is None: + # Older API versions defaulted an unset value to allowed, so an + # unset value is not evidence of compliance. + return [ + _evaluation( + account.id, + EvaluationStatus.UNKNOWN, + reason_code=MISSING_PROPERTIES, + reason="allow_blob_public_access is not set on the storage account.", ) + ] + evidence = {"allow_blob_public_access": value} + if value: + return [_evaluation(account.id, EvaluationStatus.FAIL, evidence=evidence, finding=_finding(account))] + return [_evaluation(account.id, EvaluationStatus.PASS, evidence=evidence)] + + +def _evaluation(resource_id: str, status: str, **kwargs: Any) -> RuleEvaluation: + return RuleEvaluation( + rule_id=RULE_ID, resource_id=resource_id, resource_type=RESOURCE_TYPE, status=status, **kwargs + ) + + +def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: + """Return the findings attached to this rule's FAIL evaluations.""" + return fail_findings(evaluate(azure_client, subscription_id)) + - return findings +def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: + """Report a status for every storage account, PASS included.""" + accounts = azure_client.list_storage_accounts() + if accounts is None: + return [inventory_unavailable(RULE_ID, RESOURCE_TYPE, subscription_id)] + if not accounts: + return [no_resources_found(RULE_ID, RESOURCE_TYPE, subscription_id)] + evaluations: List[RuleEvaluation] = [] + for account in accounts: + evaluations.extend(_evaluate_account(azure_client, account, subscription_id)) + return evaluations diff --git a/scanner/rules/az_stor_002.py b/scanner/rules/az_stor_002.py index cda97ee9..4f4f3c35 100644 --- a/scanner/rules/az_stor_002.py +++ b/scanner/rules/az_stor_002.py @@ -2,6 +2,16 @@ from typing import Any, Dict, List +from scanner.evaluation import ( + MISSING_PROPERTIES, + EvaluationStatus, + RuleEvaluation, + fail_findings, + inventory_unavailable, + missing_resource_id, + no_resources_found, +) + RULE_ID = "AZ-STOR-002" RULE_NAME = "Storage Account Allows HTTP Traffic (Not HTTPS-Only)" SEVERITY = "HIGH" @@ -17,28 +27,65 @@ "Navigate to Storage Account > Configuration > Secure transfer required and enable it." ) PLAYBOOK = "playbooks/cli/fix_az_stor_002.sh" +RESOURCE_TYPE = "Microsoft.Storage/storageAccounts" +_UNSET = object() -def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: - """Detect storage accounts where enable_https_traffic_only is False.""" - findings: List[Dict[str, Any]] = [] - - for account in azure_client.get_storage_accounts(): - if not getattr(account, "enable_https_traffic_only", True): - findings.append( - { - "rule_id": RULE_ID, - "rule_name": RULE_NAME, - "severity": SEVERITY, - "category": CATEGORY, - "resource_id": account.id, - "resource_name": account.name, - "resource_type": "Microsoft.Storage/storageAccounts", - "description": DESCRIPTION, - "remediation": REMEDIATION, - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - } + +def _finding(account: Any) -> Dict[str, Any]: + return { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": account.id, + "resource_name": account.name, + "resource_type": "Microsoft.Storage/storageAccounts", + "description": DESCRIPTION, + "remediation": REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + } + + +def _evaluate_account(azure_client: Any, account: Any, subscription_id: str) -> List[RuleEvaluation]: + if not getattr(account, "id", None): + return [missing_resource_id(RULE_ID, RESOURCE_TYPE, subscription_id)] + value = getattr(account, "enable_https_traffic_only", _UNSET) + if value is _UNSET: + return [ + _evaluation( + account.id, + EvaluationStatus.UNKNOWN, + reason_code=MISSING_PROPERTIES, + reason="enable_https_traffic_only was not returned for the storage account.", ) + ] + evidence = {"enable_https_traffic_only": value} + if not value: + return [_evaluation(account.id, EvaluationStatus.FAIL, evidence=evidence, finding=_finding(account))] + return [_evaluation(account.id, EvaluationStatus.PASS, evidence=evidence)] + + +def _evaluation(resource_id: str, status: str, **kwargs: Any) -> RuleEvaluation: + return RuleEvaluation( + rule_id=RULE_ID, resource_id=resource_id, resource_type=RESOURCE_TYPE, status=status, **kwargs + ) + + +def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: + """Return the findings attached to this rule's FAIL evaluations.""" + return fail_findings(evaluate(azure_client, subscription_id)) + - return findings +def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: + """Report a status for every storage account, PASS included.""" + accounts = azure_client.list_storage_accounts() + if accounts is None: + return [inventory_unavailable(RULE_ID, RESOURCE_TYPE, subscription_id)] + if not accounts: + return [no_resources_found(RULE_ID, RESOURCE_TYPE, subscription_id)] + evaluations: List[RuleEvaluation] = [] + for account in accounts: + evaluations.extend(_evaluate_account(azure_client, account, subscription_id)) + return evaluations diff --git a/scanner/rules/az_stor_003.py b/scanner/rules/az_stor_003.py index 7227d54d..6c7ebfea 100644 --- a/scanner/rules/az_stor_003.py +++ b/scanner/rules/az_stor_003.py @@ -3,6 +3,17 @@ import logging from typing import Any, Dict, List, Optional +from scanner.evaluation import ( + EVIDENCE_UNAVAILABLE, + MISSING_PROPERTIES, + EvaluationStatus, + RuleEvaluation, + fail_findings, + inventory_unavailable, + no_resources_found, + subscription_scope_id, +) + logger = logging.getLogger(__name__) # subscription_id is received by scan() and passed to AzureClient methods @@ -35,13 +46,31 @@ "Data management > Lifecycle management > Add a rule." ) PLAYBOOK = "playbooks/cli/fix_az_stor_003.sh" - - -# ── Required scan function ─────────────────────────────────────────────────── - - -def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: - """Detect storage accounts with no lifecycle management policy. +RESOURCE_TYPE = "Microsoft.Storage/storageAccounts" + + +def _finding(resource_id: str, account_name: str, resource_group: str, location: str) -> Dict[str, Any]: + return { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": resource_id, + "resource_name": account_name, + "resource_type": "Microsoft.Storage/storageAccounts", + "description": DESCRIPTION, + "remediation": REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": { + "resource_group": resource_group, + "location": location, + }, + } + + +def _evaluate_account(azure_client: Any, account: Any, subscription_id: str) -> List[RuleEvaluation]: + """Evaluate one account's lifecycle management policy. The Azure Storage Management SDK exposes lifecycle policies via ``management_policies.get(resource_group, account_name)``. @@ -49,70 +78,85 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: the condition we flag as MEDIUM severity. Three-state return from get_storage_lifecycle_policy(): - True — policy exists and has rules → skip (compliant) - False — no policy exists → create finding - None — permissions error or unexpected failure → skip with warning - to avoid false positives - - Args: - azure_client: An AzureClient instance with all SDK clients - pre-configured. - subscription_id: The Azure subscription ID being scanned. - - Returns: - A list of finding dicts — one per storage account that has no - lifecycle policy. Accounts that could not be checked are skipped - and logged as warnings. + True — policy exists and has rules → PASS + False — no policy exists → FAIL with a finding + None — permissions error or unexpected failure → UNKNOWN, never + a finding, to avoid false positives """ - findings: List[Dict[str, Any]] = [] - - for account in azure_client.get_storage_accounts(): - resource_id = getattr(account, "id", "") - account_name = getattr(account, "name", "") - location = getattr(account, "location", "") - - if not resource_id or not account_name: - continue - - parsed = azure_client.parse_resource_id(resource_id) - resource_group = parsed.get("resource_group", "") - if not resource_group: - continue - - # True = compliant, False = no policy, None = could not determine - policy_status: Optional[bool] = azure_client.get_storage_lifecycle_policy(resource_group, account_name) - - if policy_status is None: - # Permissions error or unexpected SDK failure. - # Skip rather than flag — never create false positives. - logger.warning( - "AZ-STOR-003: Could not determine lifecycle policy for %s " - "— skipping. Ensure the service principal has " - "Microsoft.Storage/storageAccounts/managementPolicies/read " - "permission.", - account_name, + resource_id = getattr(account, "id", "") + account_name = getattr(account, "name", "") + location = getattr(account, "location", "") + + if not resource_id or not account_name: + return [ + _evaluation( + resource_id or subscription_scope_id(subscription_id), + EvaluationStatus.UNKNOWN, + reason_code=MISSING_PROPERTIES, + reason="Storage account was returned without an ID or name.", ) - continue - - if policy_status is False: - findings.append( - { - "rule_id": RULE_ID, - "rule_name": RULE_NAME, - "severity": SEVERITY, - "category": CATEGORY, - "resource_id": resource_id, - "resource_name": account_name, - "resource_type": "Microsoft.Storage/storageAccounts", - "description": DESCRIPTION, - "remediation": REMEDIATION, - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - "metadata": { - "resource_group": resource_group, - "location": location, - }, - } + ] + + parsed = azure_client.parse_resource_id(resource_id) + resource_group = parsed.get("resource_group", "") + if not resource_group: + return [ + _evaluation( + resource_id, + EvaluationStatus.UNKNOWN, + reason_code=MISSING_PROPERTIES, + reason="Resource group could not be parsed from the storage account ID.", ) + ] + + # True = compliant, False = no policy, None = could not determine + policy_status: Optional[bool] = azure_client.get_storage_lifecycle_policy(resource_group, account_name) + evidence = {"lifecycle_policy": policy_status} + + if policy_status is False: + finding = _finding(resource_id, account_name, resource_group, location) + return [_evaluation(resource_id, EvaluationStatus.FAIL, evidence=evidence, finding=finding)] + if policy_status is True: + return [_evaluation(resource_id, EvaluationStatus.PASS, evidence=evidence)] + + # Permissions error or unexpected SDK failure: report it rather than + # guess either way. + logger.warning( + "AZ-STOR-003: Could not determine lifecycle policy for %s " + "— reporting UNKNOWN. Ensure the service principal has " + "Microsoft.Storage/storageAccounts/managementPolicies/read " + "permission.", + account_name, + ) + return [ + _evaluation( + resource_id, + EvaluationStatus.UNKNOWN, + reason_code=EVIDENCE_UNAVAILABLE, + reason="Lifecycle management policy could not be read.", + ) + ] + + +def _evaluation(resource_id: str, status: str, **kwargs: Any) -> RuleEvaluation: + return RuleEvaluation( + rule_id=RULE_ID, resource_id=resource_id, resource_type=RESOURCE_TYPE, status=status, **kwargs + ) - return findings + +def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: + """Return the findings attached to this rule's FAIL evaluations.""" + return fail_findings(evaluate(azure_client, subscription_id)) + + +def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: + """Report a status for every storage account, PASS included.""" + accounts = azure_client.list_storage_accounts() + if accounts is None: + return [inventory_unavailable(RULE_ID, RESOURCE_TYPE, subscription_id)] + if not accounts: + return [no_resources_found(RULE_ID, RESOURCE_TYPE, subscription_id)] + evaluations: List[RuleEvaluation] = [] + for account in accounts: + evaluations.extend(_evaluate_account(azure_client, account, subscription_id)) + return evaluations diff --git a/scanner/rules/az_stor_004.py b/scanner/rules/az_stor_004.py index 71398731..bb83627b 100644 --- a/scanner/rules/az_stor_004.py +++ b/scanner/rules/az_stor_004.py @@ -3,6 +3,17 @@ import logging from typing import Any, Dict, List, Optional, Tuple +from scanner.evaluation import ( + EVIDENCE_UNAVAILABLE, + MISSING_PROPERTIES, + EvaluationStatus, + RuleEvaluation, + fail_findings, + inventory_unavailable, + no_resources_found, + subscription_scope_id, +) + logger = logging.getLogger(__name__) # ── Required module-level constants ───────────────────────────────────────── @@ -34,6 +45,7 @@ "StorageRead, StorageWrite, and StorageDelete." ) PLAYBOOK = "playbooks/cli/fix_az_stor_004.sh" +RESOURCE_TYPE = "Microsoft.Storage/storageAccounts" # Maps service key → (sub-resource path segment, resource_type) _SERVICES: Dict[str, Tuple[str, str]] = { @@ -43,83 +55,128 @@ } -# ── Required scan function ─────────────────────────────────────────────────── - - -def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: - """Detect storage account services with incomplete diagnostic logging. - - For each storage account, all three sub-services (blob, queue, table) are - checked independently. A separate finding is emitted for each service that - does not have StorageRead, StorageWrite, and StorageDelete all enabled. +def _service_evaluation(resource_id: str, resource_type: str, status: str, **kwargs: Any) -> RuleEvaluation: + return RuleEvaluation( + rule_id=RULE_ID, resource_id=resource_id, resource_type=resource_type, status=status, **kwargs + ) + + +def _finding(resource_id: str, account_name: str, resource_group: str, location: str, service: str) -> Dict[str, Any]: + svc_path, resource_type = _SERVICES[service] + return { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": f"{resource_id}/{svc_path}/default", + "resource_name": f"{account_name}/{svc_path}", + "resource_type": resource_type, + "description": DESCRIPTION.format(service=service), + "remediation": REMEDIATION.format(service=service), + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": { + "resource_group": resource_group, + "location": location, + "service": service, + }, + } + + +def _evaluate_account(azure_client: Any, account: Any, subscription_id: str) -> List[RuleEvaluation]: + """Evaluate diagnostic logging for each of an account's sub-services. + + All three sub-services (blob, queue, table) are evaluated independently, + each against its own sub-resource ID, because each needs StorageRead, + StorageWrite, and StorageDelete enabled. Three-state return from get_storage_service_logging(): - True — all three log categories enabled → skip (compliant) - False — one or more categories missing → create finding - None — permissions error or unexpected failure → skip with warning - to avoid false positives - - Args: - azure_client: An AzureClient instance with all SDK clients - pre-configured. - subscription_id: The Azure subscription ID being scanned. - - Returns: - A list of finding dicts — one per storage service sub-resource that - does not have full diagnostic logging. Services that could not be - checked are skipped and logged as warnings. + True — all three log categories enabled → PASS + False — one or more categories missing → FAIL with a finding + None — permissions error or unexpected failure → UNKNOWN, never + a finding, to avoid false positives """ - findings: List[Dict[str, Any]] = [] - - for account in azure_client.get_storage_accounts(): - resource_id = getattr(account, "id", "") - account_name = getattr(account, "name", "") - location = getattr(account, "location", "") - - if not resource_id or not account_name: - continue - - parsed = azure_client.parse_resource_id(resource_id) - resource_group = parsed.get("resource_group", "") - if not resource_group: - continue - - for service, (svc_path, resource_type) in _SERVICES.items(): - # True = compliant, False = logging incomplete, None = could not determine - logging_status: Optional[bool] = azure_client.get_storage_service_logging( - resource_group, account_name, service + resource_id = getattr(account, "id", "") + account_name = getattr(account, "name", "") + location = getattr(account, "location", "") + + if not resource_id or not account_name: + return [ + _evaluation( + resource_id or subscription_scope_id(subscription_id), + EvaluationStatus.UNKNOWN, + reason_code=MISSING_PROPERTIES, + reason="Storage account was returned without an ID or name.", ) - - if logging_status is None: - logger.warning( - "AZ-STOR-004: Could not determine %s logging status for %s " - "— skipping. Ensure the service principal has " - "microsoft.insights/diagnosticSettings/read permission.", - service, - account_name, + ] + + parsed = azure_client.parse_resource_id(resource_id) + resource_group = parsed.get("resource_group", "") + if not resource_group: + return [ + _evaluation( + resource_id, + EvaluationStatus.UNKNOWN, + reason_code=MISSING_PROPERTIES, + reason="Resource group could not be parsed from the storage account ID.", + ) + ] + + evaluations: List[RuleEvaluation] = [] + for service, (svc_path, resource_type) in _SERVICES.items(): + service_id = f"{resource_id}/{svc_path}/default" + # True = compliant, False = logging incomplete, None = could not determine + logging_status: Optional[bool] = azure_client.get_storage_service_logging(resource_group, account_name, service) + evidence = {"service": service, "logging_complete": logging_status} + + if logging_status is False: + finding = _finding(resource_id, account_name, resource_group, location, service) + evaluations.append( + _service_evaluation( + service_id, resource_type, EvaluationStatus.FAIL, evidence=evidence, finding=finding ) - continue - - if logging_status is False: - findings.append( - { - "rule_id": RULE_ID, - "rule_name": RULE_NAME, - "severity": SEVERITY, - "category": CATEGORY, - "resource_id": f"{resource_id}/{svc_path}/default", - "resource_name": f"{account_name}/{svc_path}", - "resource_type": resource_type, - "description": DESCRIPTION.format(service=service), - "remediation": REMEDIATION.format(service=service), - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - "metadata": { - "resource_group": resource_group, - "location": location, - "service": service, - }, - } + ) + elif logging_status is True: + evaluations.append(_service_evaluation(service_id, resource_type, EvaluationStatus.PASS, evidence=evidence)) + else: + logger.warning( + "AZ-STOR-004: Could not determine %s logging status for %s " + "— reporting UNKNOWN. Ensure the service principal has " + "microsoft.insights/diagnosticSettings/read permission.", + service, + account_name, + ) + evaluations.append( + _service_evaluation( + service_id, + resource_type, + EvaluationStatus.UNKNOWN, + reason_code=EVIDENCE_UNAVAILABLE, + reason=f"Diagnostic settings for the {service} service could not be read.", ) + ) + return evaluations - return findings + +def _evaluation(resource_id: str, status: str, **kwargs: Any) -> RuleEvaluation: + return RuleEvaluation( + rule_id=RULE_ID, resource_id=resource_id, resource_type=RESOURCE_TYPE, status=status, **kwargs + ) + + +def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: + """Return the findings attached to this rule's FAIL evaluations.""" + return fail_findings(evaluate(azure_client, subscription_id)) + + +def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: + """Report a status for every storage account, PASS included.""" + accounts = azure_client.list_storage_accounts() + if accounts is None: + return [inventory_unavailable(RULE_ID, RESOURCE_TYPE, subscription_id)] + if not accounts: + return [no_resources_found(RULE_ID, RESOURCE_TYPE, subscription_id)] + evaluations: List[RuleEvaluation] = [] + for account in accounts: + evaluations.extend(_evaluate_account(azure_client, account, subscription_id)) + return evaluations diff --git a/scanner/rules/az_stor_005.py b/scanner/rules/az_stor_005.py index 2683c747..59a65c2b 100644 --- a/scanner/rules/az_stor_005.py +++ b/scanner/rules/az_stor_005.py @@ -3,6 +3,16 @@ import logging from typing import Any, Dict, List +from scanner.evaluation import ( + MISSING_PROPERTIES, + EvaluationStatus, + RuleEvaluation, + fail_findings, + inventory_unavailable, + no_resources_found, + subscription_scope_id, +) + logger = logging.getLogger(__name__) RULE_ID = "AZ-STOR-005" @@ -31,6 +41,7 @@ "playbook." ) PLAYBOOK = "playbooks/cli/fix_az_stor_005.sh" +RESOURCE_TYPE = "Microsoft.Storage/storageAccounts" _GEO_REDUNDANT_SKUS = { "Standard_GRS", @@ -42,54 +53,88 @@ } -def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: - """Detect storage accounts not configured with geo-redundant replication.""" - findings: List[Dict[str, Any]] = [] - - for account in azure_client.get_storage_accounts(): - resource_id = getattr(account, "id", "") - account_name = getattr(account, "name", "") - location = getattr(account, "location", "") +def _finding(azure_client: Any, resource_id: str, account_name: str, location: str, sku_name: str) -> Dict[str, Any]: + parsed = azure_client.parse_resource_id(resource_id) + resource_group = parsed.get("resource_group", "") + return { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": resource_id, + "resource_name": account_name, + "resource_type": "Microsoft.Storage/storageAccounts", + "description": DESCRIPTION.format(sku_name=sku_name), + "remediation": REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": { + "resource_group": resource_group, + "location": location, + "current_sku": sku_name, + "recommended_sku": "Standard_GRS", + }, + } - if not resource_id or not account_name: - continue - sku = getattr(account, "sku", None) - sku_name = getattr(sku, "name", "") if sku else "" +def _evaluate_account(azure_client: Any, account: Any, subscription_id: str) -> List[RuleEvaluation]: + resource_id = getattr(account, "id", "") + account_name = getattr(account, "name", "") + location = getattr(account, "location", "") - if not sku_name: - logger.warning( - "AZ-STOR-005: Could not determine SKU for %s — skipping.", - account_name, + if not resource_id or not account_name: + return [ + _evaluation( + resource_id or subscription_scope_id(subscription_id), + EvaluationStatus.UNKNOWN, + reason_code=MISSING_PROPERTIES, + reason="Storage account was returned without an ID or name.", ) - continue - - if sku_name in _GEO_REDUNDANT_SKUS: - continue - - parsed = azure_client.parse_resource_id(resource_id) - resource_group = parsed.get("resource_group", "") - - findings.append( - { - "rule_id": RULE_ID, - "rule_name": RULE_NAME, - "severity": SEVERITY, - "category": CATEGORY, - "resource_id": resource_id, - "resource_name": account_name, - "resource_type": "Microsoft.Storage/storageAccounts", - "description": DESCRIPTION.format(sku_name=sku_name), - "remediation": REMEDIATION, - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - "metadata": { - "resource_group": resource_group, - "location": location, - "current_sku": sku_name, - "recommended_sku": "Standard_GRS", - }, - } + ] + + sku = getattr(account, "sku", None) + sku_name = getattr(sku, "name", "") if sku else "" + + if not sku_name: + logger.warning( + "AZ-STOR-005: Could not determine SKU for %s — reporting UNKNOWN.", + account_name, ) + return [ + _evaluation( + resource_id, + EvaluationStatus.UNKNOWN, + reason_code=MISSING_PROPERTIES, + reason="Storage account SKU was not returned.", + ) + ] + + evidence = {"sku": sku_name} + if sku_name in _GEO_REDUNDANT_SKUS: + return [_evaluation(resource_id, EvaluationStatus.PASS, evidence=evidence)] + finding = _finding(azure_client, resource_id, account_name, location, sku_name) + return [_evaluation(resource_id, EvaluationStatus.FAIL, evidence=evidence, finding=finding)] + + +def _evaluation(resource_id: str, status: str, **kwargs: Any) -> RuleEvaluation: + return RuleEvaluation( + rule_id=RULE_ID, resource_id=resource_id, resource_type=RESOURCE_TYPE, status=status, **kwargs + ) + + +def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: + """Return the findings attached to this rule's FAIL evaluations.""" + return fail_findings(evaluate(azure_client, subscription_id)) + - return findings +def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: + """Report a status for every storage account, PASS included.""" + accounts = azure_client.list_storage_accounts() + if accounts is None: + return [inventory_unavailable(RULE_ID, RESOURCE_TYPE, subscription_id)] + if not accounts: + return [no_resources_found(RULE_ID, RESOURCE_TYPE, subscription_id)] + evaluations: List[RuleEvaluation] = [] + for account in accounts: + evaluations.extend(_evaluate_account(azure_client, account, subscription_id)) + return evaluations diff --git a/scanner/rules/az_stor_006.py b/scanner/rules/az_stor_006.py index 2e16d38d..b6970276 100644 --- a/scanner/rules/az_stor_006.py +++ b/scanner/rules/az_stor_006.py @@ -3,6 +3,15 @@ import logging from typing import Any, Dict, List +from scanner.evaluation import ( + EvaluationStatus, + RuleEvaluation, + fail_findings, + inventory_unavailable, + missing_resource_id, + no_resources_found, +) + logger = logging.getLogger(__name__) RULE_ID = "AZ-STOR-006" @@ -26,34 +35,61 @@ "> Configuration > Allow storage account key access and set it to Disabled." ) PLAYBOOK = "playbooks/cli/fix_az_stor_006.sh" +RESOURCE_TYPE = "Microsoft.Storage/storageAccounts" -def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: - """Flag accounts where shared-key access is enabled or unset. +def _finding(account: Any, state: Any) -> Dict[str, Any]: + return { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": account.id, + "resource_name": account.name, + "resource_type": "Microsoft.Storage/storageAccounts", + "description": DESCRIPTION, + "remediation": REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": {"allow_shared_key_access": state}, + } + + +def _evaluate_account(azure_client: Any, account: Any, subscription_id: str) -> List[RuleEvaluation]: + """Fail accounts where shared-key access is enabled or unset. Azure documents allow_shared_key_access=None as equivalent to True: an account with no explicit setting permits Shared Key authorization. Only False (explicitly disabled) is compliant. """ - findings: List[Dict[str, Any]] = [] - for account in azure_client.get_storage_accounts(): - state = getattr(account, "allow_shared_key_access", None) - if state is False: - continue - findings.append( - { - "rule_id": RULE_ID, - "rule_name": RULE_NAME, - "severity": SEVERITY, - "category": CATEGORY, - "resource_id": account.id, - "resource_name": account.name, - "resource_type": "Microsoft.Storage/storageAccounts", - "description": DESCRIPTION, - "remediation": REMEDIATION, - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - "metadata": {"allow_shared_key_access": state}, - } - ) - return findings + if not getattr(account, "id", None): + return [missing_resource_id(RULE_ID, RESOURCE_TYPE, subscription_id)] + state = getattr(account, "allow_shared_key_access", None) + evidence = {"allow_shared_key_access": state} + if state is False: + return [_evaluation(account.id, EvaluationStatus.PASS, evidence=evidence)] + return [_evaluation(account.id, EvaluationStatus.FAIL, evidence=evidence, finding=_finding(account, state))] + + +def _evaluation(resource_id: str, status: str, **kwargs: Any) -> RuleEvaluation: + return RuleEvaluation( + rule_id=RULE_ID, resource_id=resource_id, resource_type=RESOURCE_TYPE, status=status, **kwargs + ) + + +def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: + """Return the findings attached to this rule's FAIL evaluations.""" + return fail_findings(evaluate(azure_client, subscription_id)) + + +def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: + """Report a status for every storage account, PASS included.""" + accounts = azure_client.list_storage_accounts() + if accounts is None: + return [inventory_unavailable(RULE_ID, RESOURCE_TYPE, subscription_id)] + if not accounts: + return [no_resources_found(RULE_ID, RESOURCE_TYPE, subscription_id)] + evaluations: List[RuleEvaluation] = [] + for account in accounts: + evaluations.extend(_evaluate_account(azure_client, account, subscription_id)) + return evaluations diff --git a/scanner/rules/az_stor_007.py b/scanner/rules/az_stor_007.py index 23c3673c..4c2c0dd1 100644 --- a/scanner/rules/az_stor_007.py +++ b/scanner/rules/az_stor_007.py @@ -4,6 +4,14 @@ from typing import Any, Dict, List from scanner.azure_client import enum_str +from scanner.evaluation import ( + EvaluationStatus, + RuleEvaluation, + fail_findings, + inventory_unavailable, + missing_resource_id, + no_resources_found, +) logger = logging.getLogger(__name__) @@ -26,46 +34,83 @@ "to Storage Account > Configuration > Minimum TLS version." ) PLAYBOOK = "playbooks/cli/fix_az_stor_007.sh" +RESOURCE_TYPE = "Microsoft.Storage/storageAccounts" _INSECURE_TLS = {"TLS1_0", "TLS1_1", "1.0", "1.1"} _SECURE_TLS = {"TLS1_2", "TLS1_3", "1.2", "1.3"} -def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: - """Flag accounts with TLS below 1.2 or with an unset minimum TLS version. +def _finding(account: Any, value: Any) -> Dict[str, Any]: + return { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": account.id, + "resource_name": account.name, + "resource_type": "Microsoft.Storage/storageAccounts", + "description": DESCRIPTION, + "remediation": REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": { + "minimum_tls_version": str(value), + "approved_minimum": "TLS1_2", + }, + } + + +def _evaluate_account(azure_client: Any, account: Any, subscription_id: str) -> List[RuleEvaluation]: + """Fail accounts with TLS below 1.2 or with an unset minimum TLS version. Azure documents an unset minimum_tls_version as TLS 1.0, so None is treated as insecure. enum_str() is used to handle SDK enum objects (e.g. MinimumTlsVersion.TLS1_0) so they compare correctly against the known-insecure set instead of producing a string like - 'MinimumTlsVersion.TLS1_0'. + 'MinimumTlsVersion.TLS1_0'. A value in neither set is UNKNOWN. """ - findings: List[Dict[str, Any]] = [] - for account in azure_client.get_storage_accounts(): - value = getattr(account, "minimum_tls_version", None) - if value is None: - normalized = "TLS1_0" - else: - normalized = enum_str(value).strip().upper() - if normalized not in _INSECURE_TLS: - continue - findings.append( - { - "rule_id": RULE_ID, - "rule_name": RULE_NAME, - "severity": SEVERITY, - "category": CATEGORY, - "resource_id": account.id, - "resource_name": account.name, - "resource_type": "Microsoft.Storage/storageAccounts", - "description": DESCRIPTION, - "remediation": REMEDIATION, - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - "metadata": { - "minimum_tls_version": str(value), - "approved_minimum": "TLS1_2", - }, - } + if not getattr(account, "id", None): + return [missing_resource_id(RULE_ID, RESOURCE_TYPE, subscription_id)] + value = getattr(account, "minimum_tls_version", None) + if value is None: + normalized = "TLS1_0" + else: + normalized = enum_str(value).strip().upper() + evidence = {"minimum_tls_version": str(value), "normalized": normalized} + if normalized in _INSECURE_TLS: + return [_evaluation(account.id, EvaluationStatus.FAIL, evidence=evidence, finding=_finding(account, value))] + if normalized in _SECURE_TLS: + return [_evaluation(account.id, EvaluationStatus.PASS, evidence=evidence)] + return [ + _evaluation( + account.id, + EvaluationStatus.UNKNOWN, + reason_code="UNRECOGNIZED_TLS_VERSION", + reason=f"Unrecognized minimum_tls_version {normalized!r}.", + evidence=evidence, ) - return findings + ] + + +def _evaluation(resource_id: str, status: str, **kwargs: Any) -> RuleEvaluation: + return RuleEvaluation( + rule_id=RULE_ID, resource_id=resource_id, resource_type=RESOURCE_TYPE, status=status, **kwargs + ) + + +def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: + """Return the findings attached to this rule's FAIL evaluations.""" + return fail_findings(evaluate(azure_client, subscription_id)) + + +def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: + """Report a status for every storage account, PASS included.""" + accounts = azure_client.list_storage_accounts() + if accounts is None: + return [inventory_unavailable(RULE_ID, RESOURCE_TYPE, subscription_id)] + if not accounts: + return [no_resources_found(RULE_ID, RESOURCE_TYPE, subscription_id)] + evaluations: List[RuleEvaluation] = [] + for account in accounts: + evaluations.extend(_evaluate_account(azure_client, account, subscription_id)) + return evaluations diff --git a/scanner/rules/az_stor_008.py b/scanner/rules/az_stor_008.py index 34f04d0c..6637d04e 100644 --- a/scanner/rules/az_stor_008.py +++ b/scanner/rules/az_stor_008.py @@ -3,7 +3,16 @@ import logging from typing import Any, Dict, List -from scanner.rules._storage_common import policy_required +from scanner.evaluation import ( + MISSING_PROPERTIES, + EvaluationStatus, + RuleEvaluation, + fail_findings, + inventory_unavailable, + missing_resource_id, + no_resources_found, +) +from scanner.rules._storage_common import policy_exemption logger = logging.getLogger(__name__) @@ -18,38 +27,90 @@ "after validating key access and rotation ownership." ) PLAYBOOK = "playbooks/cli/fix_az_stor_008.sh" +RESOURCE_TYPE = "Microsoft.Storage/storageAccounts" + +_REQUIREMENT_TAG = "oshield:cmk-required" + + +def _finding(account: Any, key_source: Any) -> Dict[str, Any]: + return { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": account.id, + "resource_name": account.name, + "resource_type": "Microsoft.Storage/storageAccounts", + "description": DESCRIPTION, + "remediation": REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": {"key_source": str(key_source), "requirement_tag": "oshield:cmk-required"}, + } + + +def _evaluate_account(azure_client: Any, account: Any, subscription_id: str) -> List[RuleEvaluation]: + if not getattr(account, "id", None): + return [missing_resource_id(RULE_ID, RESOURCE_TYPE, subscription_id)] + exemption = policy_exemption(account, _REQUIREMENT_TAG) + if exemption: + return [ + _evaluation( + account.id, + EvaluationStatus.NOT_APPLICABLE, + reason_code=exemption, + reason=f"Customer-managed keys are not required by the {_REQUIREMENT_TAG} tag.", + ) + ] + encryption = getattr(account, "encryption", None) + key_source = getattr(encryption, "key_source", None) if encryption else None + if key_source is None: + logger.warning("%s: encryption source unavailable; reporting UNKNOWN", RULE_ID) + return [ + _evaluation( + account.id, + EvaluationStatus.UNKNOWN, + reason_code=MISSING_PROPERTIES, + reason="Storage account encryption key source was not returned.", + ) + ] + evidence = {"key_source": str(key_source)} + source = str(key_source).strip().lower() + if source in {"microsoft.keyvault", "keyvault", "microsoft.keyvaultmanaged"}: + return [_evaluation(account.id, EvaluationStatus.PASS, evidence=evidence)] + if source != "microsoft.storage": + logger.warning("%s: unknown encryption source %r; reporting UNKNOWN", RULE_ID, key_source) + return [ + _evaluation( + account.id, + EvaluationStatus.UNKNOWN, + reason_code="UNRECOGNIZED_KEY_SOURCE", + reason=f"Unrecognized encryption key source {str(key_source)!r}.", + evidence=evidence, + ) + ] + return [_evaluation(account.id, EvaluationStatus.FAIL, evidence=evidence, finding=_finding(account, key_source))] + + +def _evaluation(resource_id: str, status: str, **kwargs: Any) -> RuleEvaluation: + return RuleEvaluation( + rule_id=RULE_ID, resource_id=resource_id, resource_type=RESOURCE_TYPE, status=status, **kwargs + ) def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: - findings: List[Dict[str, Any]] = [] - for account in azure_client.get_storage_accounts(): - if not policy_required(account, "oshield:cmk-required"): - continue - encryption = getattr(account, "encryption", None) - key_source = getattr(encryption, "key_source", None) if encryption else None - if key_source is None: - logger.warning("%s: encryption source unavailable; skipping", RULE_ID) - continue - source = str(key_source).strip().lower() - if source in {"microsoft.keyvault", "keyvault", "microsoft.keyvaultmanaged"}: - continue - if source != "microsoft.storage": - logger.warning("%s: unknown encryption source %r; skipping", RULE_ID, key_source) - continue - findings.append( - { - "rule_id": RULE_ID, - "rule_name": RULE_NAME, - "severity": SEVERITY, - "category": CATEGORY, - "resource_id": account.id, - "resource_name": account.name, - "resource_type": "Microsoft.Storage/storageAccounts", - "description": DESCRIPTION, - "remediation": REMEDIATION, - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - "metadata": {"key_source": str(key_source), "requirement_tag": "oshield:cmk-required"}, - } - ) - return findings + """Return the findings attached to this rule's FAIL evaluations.""" + return fail_findings(evaluate(azure_client, subscription_id)) + + +def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: + """Report a status for every storage account, PASS included.""" + accounts = azure_client.list_storage_accounts() + if accounts is None: + return [inventory_unavailable(RULE_ID, RESOURCE_TYPE, subscription_id)] + if not accounts: + return [no_resources_found(RULE_ID, RESOURCE_TYPE, subscription_id)] + evaluations: List[RuleEvaluation] = [] + for account in accounts: + evaluations.extend(_evaluate_account(azure_client, account, subscription_id)) + return evaluations diff --git a/scanner/rules/az_stor_009.py b/scanner/rules/az_stor_009.py index 457f427c..f852d4f3 100644 --- a/scanner/rules/az_stor_009.py +++ b/scanner/rules/az_stor_009.py @@ -4,7 +4,20 @@ from typing import Any, Dict, List from scanner.azure_client import enum_str -from scanner.rules._storage_common import policy_required +from scanner.evaluation import ( + APPROVED_EXCEPTION, + EVIDENCE_UNAVAILABLE, + MISSING_PROPERTIES, + NO_RESOURCES_FOUND, + POLICY_NOT_REQUIRED, + EvaluationStatus, + RuleEvaluation, + fail_findings, + inventory_unavailable, + no_resources_found, + subscription_scope_id, +) +from scanner.rules._storage_common import policy_exemption logger = logging.getLogger(__name__) @@ -19,6 +32,9 @@ "container after confirming retention requirements." ) PLAYBOOK = "playbooks/cli/fix_az_stor_009.sh" +RESOURCE_TYPE = "Microsoft.Storage/storageAccounts" +CONTAINER_RESOURCE_TYPE = "Microsoft.Storage/storageAccounts/blobServices/containers" +_REQUIREMENT_TAG = "oshield:immutability-required" def _container_properties(container: Any) -> Any: @@ -35,41 +51,131 @@ def _has_immutability(container: Any) -> bool: return state in {"locked", "unlocked"} and isinstance(retention, int) and retention > 0 -def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: - findings: List[Dict[str, Any]] = [] - for account in azure_client.get_storage_accounts(): - resource_id = getattr(account, "id", "") - account_name = getattr(account, "name", "") - parsed = azure_client.parse_resource_id(resource_id) - resource_group = parsed.get("resource_group", "") - if not resource_group or not account_name: - continue - containers = azure_client.get_blob_containers(resource_group, account_name) - if containers is None: - logger.warning("%s: blob containers unavailable for %s; skipping", RULE_ID, account_name) +def _container_evaluation(resource_id: str, status: str, **kwargs: Any) -> RuleEvaluation: + return RuleEvaluation( + rule_id=RULE_ID, resource_id=resource_id, resource_type=CONTAINER_RESOURCE_TYPE, status=status, **kwargs + ) + + +def _finding(resource_id: str, account_name: str, name: str) -> Dict[str, Any]: + return { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": f"{resource_id}/blobServices/default/containers/{name}", + "resource_name": f"{account_name}/{name}", + "resource_type": "Microsoft.Storage/storageAccounts/blobServices/containers", + "description": DESCRIPTION, + "remediation": REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": {"requirement_tag": "oshield:immutability-required"}, + } + + +def _immutability_evidence(container: Any) -> Dict[str, Any]: + props = _container_properties(container) + policy = getattr(container, "immutability_policy", None) or getattr(props, "immutability_policy", None) + if policy is None: + return {"immutability_policy": None} + return { + "state": enum_str(getattr(policy, "state", None)), + "retention_days": getattr(policy, "immutability_period_since_creation_in_days", None), + } + + +def _evaluate_account(azure_client: Any, account: Any, subscription_id: str) -> List[RuleEvaluation]: + """Evaluate every blob container of an account against the opt-in policy. + + The requirement tag may sit on the account (all containers) or on a + single container. Containers neither requires are NOT_APPLICABLE. + """ + resource_id = getattr(account, "id", "") + account_name = getattr(account, "name", "") + parsed = azure_client.parse_resource_id(resource_id) + resource_group = parsed.get("resource_group", "") + if not resource_group or not account_name: + return [ + _evaluation( + resource_id or subscription_scope_id(subscription_id), + EvaluationStatus.UNKNOWN, + reason_code=MISSING_PROPERTIES, + reason="Storage account was returned without a resource group or name.", + ) + ] + containers = azure_client.get_blob_containers(resource_group, account_name) + if containers is None: + logger.warning("%s: blob containers unavailable for %s; reporting UNKNOWN", RULE_ID, account_name) + return [ + _evaluation( + resource_id, + EvaluationStatus.UNKNOWN, + reason_code=EVIDENCE_UNAVAILABLE, + reason="Blob containers could not be listed for the storage account.", + ) + ] + if not containers: + return [ + _evaluation( + resource_id, + EvaluationStatus.NOT_APPLICABLE, + reason_code=NO_RESOURCES_FOUND, + reason="The storage account has no blob containers.", + ) + ] + + account_exemption = policy_exemption(account, _REQUIREMENT_TAG) + evaluations: List[RuleEvaluation] = [] + for container in containers: + name = getattr(container, "name", "") + container_id = f"{resource_id}/blobServices/default/containers/{name}" + container_exemption = policy_exemption(container, _REQUIREMENT_TAG) + if account_exemption and container_exemption: + exemption = ( + APPROVED_EXCEPTION + if APPROVED_EXCEPTION in {account_exemption, container_exemption} + else POLICY_NOT_REQUIRED + ) + evaluations.append( + _container_evaluation( + container_id, + EvaluationStatus.NOT_APPLICABLE, + reason_code=exemption, + reason=f"Immutability is not required by the {_REQUIREMENT_TAG} tag.", + ) + ) continue - account_required = policy_required(account, "oshield:immutability-required") - for container in containers: - container_required = policy_required(container, "oshield:immutability-required") - if not account_required and not container_required: - continue - if _has_immutability(container): - continue - name = getattr(container, "name", "") - findings.append( - { - "rule_id": RULE_ID, - "rule_name": RULE_NAME, - "severity": SEVERITY, - "category": CATEGORY, - "resource_id": f"{resource_id}/blobServices/default/containers/{name}", - "resource_name": f"{account_name}/{name}", - "resource_type": "Microsoft.Storage/storageAccounts/blobServices/containers", - "description": DESCRIPTION, - "remediation": REMEDIATION, - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - "metadata": {"requirement_tag": "oshield:immutability-required"}, - } + evidence = _immutability_evidence(container) + if _has_immutability(container): + evaluations.append(_container_evaluation(container_id, EvaluationStatus.PASS, evidence=evidence)) + else: + finding = _finding(resource_id, account_name, name) + evaluations.append( + _container_evaluation(container_id, EvaluationStatus.FAIL, evidence=evidence, finding=finding) ) - return findings + return evaluations + + +def _evaluation(resource_id: str, status: str, **kwargs: Any) -> RuleEvaluation: + return RuleEvaluation( + rule_id=RULE_ID, resource_id=resource_id, resource_type=RESOURCE_TYPE, status=status, **kwargs + ) + + +def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: + """Return the findings attached to this rule's FAIL evaluations.""" + return fail_findings(evaluate(azure_client, subscription_id)) + + +def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: + """Report a status for every storage account, PASS included.""" + accounts = azure_client.list_storage_accounts() + if accounts is None: + return [inventory_unavailable(RULE_ID, RESOURCE_TYPE, subscription_id)] + if not accounts: + return [no_resources_found(RULE_ID, RESOURCE_TYPE, subscription_id)] + evaluations: List[RuleEvaluation] = [] + for account in accounts: + evaluations.extend(_evaluate_account(azure_client, account, subscription_id)) + return evaluations diff --git a/scanner/rules/az_stor_010.py b/scanner/rules/az_stor_010.py index a6ea8998..57aeb070 100644 --- a/scanner/rules/az_stor_010.py +++ b/scanner/rules/az_stor_010.py @@ -4,6 +4,15 @@ from typing import Any, Dict, List from scanner.azure_client import enum_str +from scanner.evaluation import ( + EVIDENCE_UNAVAILABLE, + EvaluationStatus, + RuleEvaluation, + fail_findings, + inventory_unavailable, + missing_resource_id, + no_resources_found, +) logger = logging.getLogger(__name__) @@ -25,6 +34,7 @@ "IP inside the VNet." ) PLAYBOOK = "playbooks/cli/fix_az_stor_010.sh" +RESOURCE_TYPE = "Microsoft.Storage/storageAccounts" def _has_approved_private_endpoint(connections: Any) -> bool: @@ -36,54 +46,94 @@ def _has_approved_private_endpoint(connections: Any) -> bool: return False -def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: - """Flag publicly reachable storage accounts with no approved Private Endpoint. +def _finding(account: Any, connections: Any) -> Dict[str, Any]: + return { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": getattr(account, "id", ""), + "resource_name": getattr(account, "name", ""), + "resource_type": "Microsoft.Storage/storageAccounts", + "description": DESCRIPTION, + "remediation": REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": { + "public_network_access": enum_str(getattr(account, "public_network_access", None)) or "unspecified", + "private_endpoint_connections": len(connections), + }, + } + + +def _evaluate_account(azure_client: Any, account: Any, subscription_id: str) -> List[RuleEvaluation]: + """Fail publicly reachable storage accounts with no approved Private Endpoint. A storage account whose ``public_network_access`` is already ``Disabled`` is - network-isolated regardless of private endpoints and is treated as - NOT_APPLICABLE, so the rule does not raise a false finding against an account - that is closed to the public network by another means. + network-isolated regardless of private endpoints and is NOT_APPLICABLE, so + the rule does not raise a false finding against an account that is closed + to the public network by another means. ``private_endpoint_connections`` of ``None`` means the evidence is unavailable (the field was not populated / could not be read), not a - confirmed absence, so the account is skipped as indeterminate rather than - flagged. Only a genuine empty list (or connections with none Approved) is a - finding. + confirmed absence, so the account is UNKNOWN rather than failed. Only a + genuine empty list (or connections with none Approved) is a FAIL. """ - findings: List[Dict[str, Any]] = [] - - for account in azure_client.get_storage_accounts(): - if enum_str(getattr(account, "public_network_access", None)).lower() == "disabled": - continue - - connections = getattr(account, "private_endpoint_connections", None) - if connections is None: - logger.warning( - "AZ-STOR-010: private endpoint connections unavailable for %s — skipping (indeterminate)", - getattr(account, "name", ""), + if not getattr(account, "id", None): + return [missing_resource_id(RULE_ID, RESOURCE_TYPE, subscription_id)] + public_access = enum_str(getattr(account, "public_network_access", None)) + if public_access.lower() == "disabled": + return [ + _evaluation( + account.id, + EvaluationStatus.NOT_APPLICABLE, + reason_code="PUBLIC_ACCESS_DISABLED", + reason="Public network access is disabled on the storage account.", ) - continue - if _has_approved_private_endpoint(connections): - continue - - findings.append( - { - "rule_id": RULE_ID, - "rule_name": RULE_NAME, - "severity": SEVERITY, - "category": CATEGORY, - "resource_id": getattr(account, "id", ""), - "resource_name": getattr(account, "name", ""), - "resource_type": "Microsoft.Storage/storageAccounts", - "description": DESCRIPTION, - "remediation": REMEDIATION, - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - "metadata": { - "public_network_access": enum_str(getattr(account, "public_network_access", None)) or "unspecified", - "private_endpoint_connections": len(connections), - }, - } + ] + + connections = getattr(account, "private_endpoint_connections", None) + if connections is None: + logger.warning( + "AZ-STOR-010: private endpoint connections unavailable for %s — reporting UNKNOWN", + getattr(account, "name", ""), ) + return [ + _evaluation( + account.id, + EvaluationStatus.UNKNOWN, + reason_code=EVIDENCE_UNAVAILABLE, + reason="Private endpoint connections were not returned for the storage account.", + ) + ] + evidence = { + "public_network_access": public_access or "unspecified", + "private_endpoint_connections": len(connections), + } + if _has_approved_private_endpoint(connections): + return [_evaluation(account.id, EvaluationStatus.PASS, evidence=evidence)] + return [_evaluation(account.id, EvaluationStatus.FAIL, evidence=evidence, finding=_finding(account, connections))] + + +def _evaluation(resource_id: str, status: str, **kwargs: Any) -> RuleEvaluation: + return RuleEvaluation( + rule_id=RULE_ID, resource_id=resource_id, resource_type=RESOURCE_TYPE, status=status, **kwargs + ) + + +def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: + """Return the findings attached to this rule's FAIL evaluations.""" + return fail_findings(evaluate(azure_client, subscription_id)) + - return findings +def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: + """Report a status for every storage account, PASS included.""" + accounts = azure_client.list_storage_accounts() + if accounts is None: + return [inventory_unavailable(RULE_ID, RESOURCE_TYPE, subscription_id)] + if not accounts: + return [no_resources_found(RULE_ID, RESOURCE_TYPE, subscription_id)] + evaluations: List[RuleEvaluation] = [] + for account in accounts: + evaluations.extend(_evaluate_account(azure_client, account, subscription_id)) + return evaluations diff --git a/tests/helpers/mock_azure.py b/tests/helpers/mock_azure.py index 65d3c64b..69c6c95e 100644 --- a/tests/helpers/mock_azure.py +++ b/tests/helpers/mock_azure.py @@ -46,12 +46,12 @@ class MockAzureClient: """Drop-in replacement for AzureClient that returns configured fake data.""" def __init__(self) -> None: - self._storage_accounts: List[Any] = [] + self._storage_accounts: Optional[List[Any]] = [] self._network_security_groups: List[Any] = [] self._express_route_ports: Optional[List[Any]] = [] self._virtual_machines: List[Any] = [] self._virtual_machine_scale_sets: List[Any] = [] - self._key_vaults: List[Any] = [] + self._key_vaults: Optional[List[Any]] = [] self._sql_servers: List[Any] = [] self._service_principals: List[Any] = [] self._sql_firewall_rules: Dict[Tuple[str, str], List[Any]] = {} @@ -113,7 +113,8 @@ def __init__(self) -> None: # SDK management client inside scan() (e.g. AZ-NET-007..010). self.subscription_id = "00000000-0000-0000-0000-000000000001" - def set_storage_accounts(self, accounts: List[Any]) -> "MockAzureClient": + def set_storage_accounts(self, accounts: Optional[List[Any]]) -> "MockAzureClient": + """``None`` simulates a failed list call, as AzureClient.list_storage_accounts() reports it.""" self._storage_accounts = accounts return self @@ -216,7 +217,8 @@ def set_virtual_machine_scale_sets(self, scale_sets: List[Any]) -> "MockAzureCli self._virtual_machine_scale_sets = scale_sets return self - def set_key_vaults(self, vaults: List[Any]) -> "MockAzureClient": + def set_key_vaults(self, vaults: Optional[List[Any]]) -> "MockAzureClient": + """``None`` simulates a failed list call, as AzureClient.list_key_vaults() reports it.""" self._key_vaults = vaults return self @@ -234,9 +236,12 @@ def set_sql_server_firewall_rules( self._sql_firewall_rules[(resource_group, server_name)] = rules return self - def get_storage_accounts(self) -> List[Any]: + def list_storage_accounts(self) -> Optional[List[Any]]: return self._storage_accounts + def get_storage_accounts(self) -> List[Any]: + return self._storage_accounts or [] + def get_network_security_groups(self) -> List[Any]: return self._network_security_groups @@ -246,9 +251,12 @@ def get_virtual_machines(self) -> List[Any]: def get_virtual_machine_scale_sets(self) -> List[Any]: return self._virtual_machine_scale_sets - def get_key_vaults(self) -> List[Any]: + def list_key_vaults(self) -> Optional[List[Any]]: return self._key_vaults + def get_key_vaults(self) -> List[Any]: + return self._key_vaults or [] + def get_sql_servers(self) -> List[Any]: return self._sql_servers diff --git a/tests/test_azure_client_management.py b/tests/test_azure_client_management.py index 964f4018..0f3301c0 100644 --- a/tests/test_azure_client_management.py +++ b/tests/test_azure_client_management.py @@ -57,6 +57,33 @@ def test_list_wrappers_return_results_and_fail_closed(client, method, patch_targ assert getattr(client, method)() == [] +@pytest.mark.parametrize( + ("method", "patch_target", "operation"), + [ + ("list_storage_accounts", "scanner.azure_client.StorageManagementClient", "storage_accounts.list"), + ("list_key_vaults", "scanner.azure_client.KeyVaultManagementClient", "vaults.list_by_subscription"), + ], +) +def test_fallible_list_wrappers_distinguish_empty_from_failure(client, method, patch_target, operation): + """evaluate() relies on None meaning "the list call failed" and [] meaning + "the subscription has none"; collapsing the two hides permission errors.""" + with patch(patch_target) as constructor: + target = constructor.return_value + parts = operation.split(".") + for part in parts[:-1]: + target = getattr(target, part) + call = getattr(target, parts[-1]) + + call.return_value = [SimpleNamespace(name="one")] + assert [item.name for item in getattr(client, method)()] == ["one"] + + call.return_value = [] + assert getattr(client, method)() == [] + + call.side_effect = RuntimeError("Azure unavailable") + assert getattr(client, method)() is None + + def test_single_resource_and_policy_wrappers(client): with patch("scanner.azure_client.NetworkManagementClient") as constructor: constructor.return_value.network_interfaces.get.return_value = SimpleNamespace(name="nic") diff --git a/tests/test_compliance_scoring.py b/tests/test_compliance_scoring.py index 2d231f81..9e5e76db 100644 --- a/tests/test_compliance_scoring.py +++ b/tests/test_compliance_scoring.py @@ -392,6 +392,31 @@ def test_rule_that_did_not_complete_is_error_not_pass(tmp_path, monkeypatch): assert result["score_percent"] == 33 +def test_partial_error_evaluations_roll_up_to_error_without_failed_rule_marker(tmp_path, monkeypatch): + """Evaluation rows determine partial-coverage scoring; failed_rule_ids is + reserved for evaluator failures or rules with no usable outcomes.""" + rule_id = "AZ-TEST-001" + controls = {rule_id: _control("1.1", "direct")} + scan_row = {"scan_id": "scan-1", "compliance_mapping_snapshot": {"_scan_rule_outcomes": {"failed_rule_ids": []}}} + db, conn, framework_file = _patched_db_with_framework( + tmp_path, + controls, + scan_row, + [], + evaluation_rows=[(rule_id, "PASS"), (rule_id, "ERROR")], + ) + + monkeypatch.setattr(finding_module, "FRAMEWORKS_DIR", tmp_path) + monkeypatch.setitem(finding_module.FRAMEWORK_FILE_MAP, "testfw", framework_file) + + with patch.object(db, "_get_conn", return_value=conn): + result = db.get_compliance_score("testfw") + + assert result["controls"][0]["status"] == "ERROR" + assert result["error"] == 1 + assert result["score_percent"] == 0 + + def test_failed_rule_ids_from_an_older_scan_do_not_leak_into_a_later_ones_scoring(tmp_path, monkeypatch): """_scan_rule_outcomes is read from the latest scan's own snapshot only - a rule that failed on a previous scan but has a clean PASS evaluation on diff --git a/tests/test_rule_evaluation_contract.py b/tests/test_rule_evaluation_contract.py new file mode 100644 index 00000000..ecb882ef --- /dev/null +++ b/tests/test_rule_evaluation_contract.py @@ -0,0 +1,186 @@ +"""Contract every rule exposing evaluate() must satisfy (#369). + +Rules are discovered from ``scanner/rules/az_*.py`` exactly as the engine +loads them, so each family migration is checked as soon as it adds +``evaluate()``. A migrated rule must also register a fixture in +``FIXTURES`` below: a client holding at least one compliant and one +non-compliant resource, so the PASS/FAIL and scan() parity checks have +something real to compare. +""" + +import importlib +from pathlib import Path +from typing import Any, Callable, Dict + +import pytest + +from scanner.evaluation import EvaluationStatus, RuleEvaluation, fail_findings +from tests.helpers.mock_azure import MockAzureClient, make_resource + +_SUB = "00000000-0000-0000-0000-000000000001" +_RULES_DIR = Path(__file__).resolve().parent.parent / "scanner" / "rules" + + +def _migrated_rules(): + modules = [] + for path in sorted(_RULES_DIR.glob("az_*.py")): + module = importlib.import_module(f"scanner.rules.{path.stem}") + if callable(getattr(module, "evaluate", None)): + modules.append(module) + return modules + + +MIGRATED = _migrated_rules() + + +class _UniformInventoryClient(MockAzureClient): + """Every inventory and lookup call returns the same value. + + ``None`` simulates Azure failing every call; ``[]`` simulates a + subscription that has none of anything. + """ + + def __init__(self, value: Any) -> None: + super().__init__() + self._value = value + + def __getattribute__(self, name: str) -> Any: + if name.startswith(("list_", "get_")): + value = object.__getattribute__(self, "_value") + return lambda *args, **kwargs: value + return object.__getattribute__(self, name) + + +# ── Per-rule fixtures: at least one PASS and one FAIL ─────────────────────── + + +def _kv_006_fixture() -> MockAzureClient: + def vault(name: str, **props: Any) -> Any: + return make_resource( + id=f"/subscriptions/{_SUB}/resourceGroups/rg/providers/Microsoft.KeyVault/vaults/{name}", + name=name, + location="eastus", + properties=make_resource(**props), + ) + + return MockAzureClient().set_key_vaults( + [vault("kv-rbac", enable_rbac_authorization=True), vault("kv-policies", enable_rbac_authorization=False)] + ) + + +def _storage_fixture() -> MockAzureClient: + """One compliant and one non-compliant account for every AZ-STOR rule.""" + + def account_id(name: str) -> str: + return f"/subscriptions/{_SUB}/resourceGroups/rg/providers/Microsoft.Storage/storageAccounts/{name}" + + def endpoint(status: str) -> Any: + return make_resource(private_link_service_connection_state=make_resource(status=status)) + + required = {"oshield:cmk-required": "true", "oshield:immutability-required": "true"} + good = make_resource( + id=account_id("good"), + name="good", + location="eastus", + allow_blob_public_access=False, + enable_https_traffic_only=True, + allow_shared_key_access=False, + minimum_tls_version="TLS1_2", + sku=make_resource(name="Standard_GRS"), + encryption=make_resource(key_source="Microsoft.Keyvault"), + tags=required, + public_network_access="Enabled", + private_endpoint_connections=[endpoint("Approved")], + ) + bad = make_resource( + id=account_id("bad"), + name="bad", + location="eastus", + allow_blob_public_access=True, + enable_https_traffic_only=False, + allow_shared_key_access=True, + minimum_tls_version="TLS1_0", + sku=make_resource(name="Standard_LRS"), + encryption=make_resource(key_source="Microsoft.Storage"), + tags=required, + public_network_access="Enabled", + private_endpoint_connections=[], + ) + client = MockAzureClient().set_storage_accounts([good, bad]) + client.set_storage_lifecycle_policy("rg", "good", True) + client.set_storage_lifecycle_policy("rg", "bad", False) + for service in ("blob", "queue", "table"): + client.set_storage_service_logging("rg", "good", service, True) + client.set_storage_service_logging("rg", "bad", service, False) + locked = make_resource( + name="locked", + immutability_policy=make_resource(state="Locked", immutability_period_since_creation_in_days=30), + ) + client.set_blob_containers("rg", "good", [locked]) + client.set_blob_containers("rg", "bad", [make_resource(name="open", immutability_policy=None)]) + return client + + +FIXTURES: Dict[str, Callable[[], MockAzureClient]] = { + "AZ-KV-006": _kv_006_fixture, + **{f"AZ-STOR-{n:03d}": _storage_fixture for n in range(1, 11)}, +} + + +def _ids(module: Any) -> str: + return module.RULE_ID + + +def test_at_least_one_rule_is_migrated(): + assert MIGRATED, "no rule exposes evaluate(); the contract tests below would silently check nothing" + + +@pytest.mark.parametrize("rule", MIGRATED, ids=_ids) +def test_failed_inventory_is_error_never_pass_or_not_applicable(rule): + evaluations = rule.evaluate(_UniformInventoryClient(None), _SUB) + + assert evaluations, f"{rule.RULE_ID} reported no coverage when Azure failed" + statuses = {e.status for e in evaluations} + assert EvaluationStatus.ERROR in statuses + assert EvaluationStatus.PASS not in statuses + assert EvaluationStatus.NOT_APPLICABLE not in statuses + + +@pytest.mark.parametrize("rule", MIGRATED, ids=_ids) +def test_empty_inventory_is_not_applicable(rule): + evaluations = rule.evaluate(_UniformInventoryClient([]), _SUB) + + assert evaluations, f"{rule.RULE_ID} reported no coverage for an empty subscription" + assert {e.status for e in evaluations} == {EvaluationStatus.NOT_APPLICABLE} + + +@pytest.mark.parametrize("rule", MIGRATED, ids=_ids) +def test_rule_registers_a_contract_fixture(rule): + assert rule.RULE_ID in FIXTURES, f"add a {rule.RULE_ID} fixture to FIXTURES in {Path(__file__).name}" + + +@pytest.mark.parametrize("rule", [r for r in MIGRATED if r.RULE_ID in FIXTURES], ids=_ids) +def test_evaluations_are_well_formed_and_fail_carries_the_rule_finding(rule): + evaluations = rule.evaluate(FIXTURES[rule.RULE_ID](), _SUB) + + statuses = {e.status for e in evaluations} + assert {EvaluationStatus.PASS, EvaluationStatus.FAIL} <= statuses, "fixture must produce a PASS and a FAIL" + for evaluation in evaluations: + assert isinstance(evaluation, RuleEvaluation) + assert evaluation.rule_id == rule.RULE_ID + if evaluation.status == EvaluationStatus.FAIL: + finding = evaluation.finding + assert finding, f"{rule.RULE_ID} FAIL for {evaluation.resource_id} has no finding" + assert finding["rule_id"] == rule.RULE_ID + assert finding["resource_id"] == evaluation.resource_id + assert finding["severity"] == rule.SEVERITY + assert finding["frameworks"] == rule.FRAMEWORKS + else: + assert evaluation.finding is None + + +@pytest.mark.parametrize("rule", [r for r in MIGRATED if r.RULE_ID in FIXTURES], ids=_ids) +def test_scan_returns_exactly_the_fail_findings(rule): + client = FIXTURES[rule.RULE_ID]() + + assert rule.scan(client, _SUB) == fail_findings(rule.evaluate(client, _SUB)) diff --git a/tests/test_rule_evaluations.py b/tests/test_rule_evaluations.py index 1712a28b..aee2a496 100644 --- a/tests/test_rule_evaluations.py +++ b/tests/test_rule_evaluations.py @@ -1,7 +1,8 @@ """Regression tests for the rule evaluation coverage contract (#263). Covers: the EvaluationStatus/RuleEvaluation contract itself, engine wiring -(legacy rules, evaluator exceptions, FAIL-finding dedup against scan()), and +(legacy rules, evaluator exceptions, evaluate() superseding scan()), the +shared evaluation helpers, and DatabaseManager persistence of rule_evaluations in the same transaction as findings. """ @@ -14,7 +15,18 @@ import scanner.engine as engine_mod from api.models.finding import DatabaseManager from scanner.engine import ScanEngine -from scanner.evaluation import EvaluationStatus, RuleEvaluation, aggregate_status, subscription_scope_id +from scanner.evaluation import ( + EVIDENCE_UNAVAILABLE, + INVENTORY_UNAVAILABLE, + NO_RESOURCES_FOUND, + EvaluationStatus, + RuleEvaluation, + aggregate_status, + fail_findings, + inventory_unavailable, + no_resources_found, + subscription_scope_id, +) _SUB = "00000000-0000-0000-0000-000000000001" _SCAN_ID = "00000000-0000-0000-0000-000000000000" @@ -61,6 +73,31 @@ def test_aggregate_status_requires_at_least_one(): aggregate_status([]) +def test_inventory_unavailable_is_error_at_subscription_scope(): + evaluation = inventory_unavailable("AZ-TEST-010", "Microsoft.Test/things", _SUB) + assert evaluation.status == EvaluationStatus.ERROR + assert evaluation.reason_code == INVENTORY_UNAVAILABLE + assert evaluation.resource_id == subscription_scope_id(_SUB) + assert evaluation.resource_type == "Microsoft.Test/things" + + +def test_no_resources_found_is_not_applicable_at_subscription_scope(): + evaluation = no_resources_found("AZ-TEST-011", "Microsoft.Test/things", _SUB) + assert evaluation.status == EvaluationStatus.NOT_APPLICABLE + assert evaluation.reason_code == NO_RESOURCES_FOUND + assert evaluation.resource_id == subscription_scope_id(_SUB) + + +def test_fail_findings_returns_only_findings_attached_to_fail(): + finding = {"rule_id": "AZ-TEST-012", "resource_id": "/r/1"} + evaluations = [ + RuleEvaluation(rule_id="AZ-TEST-012", resource_id="/r/1", resource_type="t", status="FAIL", finding=finding), + RuleEvaluation(rule_id="AZ-TEST-012", resource_id="/r/2", resource_type="t", status="PASS"), + RuleEvaluation(rule_id="AZ-TEST-012", resource_id="/r/3", resource_type="t", status="UNKNOWN", reason_code="X"), + ] + assert fail_findings(evaluations) == [finding] + + # ── Engine wiring ──────────────────────────────────────────────────────────── @@ -167,9 +204,9 @@ def test_fail_evaluation_contributes_finding_when_scan_did_not_already_report_it assert result["findings"][0]["resource_id"] == finding["resource_id"] -def test_fail_evaluation_does_not_duplicate_a_finding_scan_already_reported(monkeypatch): - """A rule implementing both scan() and evaluate() must not double-count a - violation both already agree on.""" +def test_evaluate_supersedes_scan_so_a_violation_is_reported_once(monkeypatch): + """A rule implementing both scan() and evaluate() is run through evaluate() + only, so a violation both would report is counted once.""" _patch_engine_client(monkeypatch, MagicMock()) eng = ScanEngine.__new__(ScanEngine) eng.subscription_id = _SUB @@ -210,6 +247,263 @@ def test_fail_evaluation_does_not_duplicate_a_finding_scan_already_reported(monk assert result["total_findings"] == 1 +def _finding(rule_id: str, resource_id: str) -> dict: + return { + "rule_id": rule_id, + "rule_name": "Test", + "severity": "HIGH", + "category": "Test", + "resource_id": resource_id, + "resource_name": "r", + "resource_type": "Microsoft.Test/resources", + "description": "d", + "remediation": "r", + "playbook": "playbooks/cli/fix.sh", + "frameworks": {}, + "metadata": {}, + } + + +def _engine(rules) -> ScanEngine: + eng = ScanEngine.__new__(ScanEngine) + eng.subscription_id = _SUB + eng.client = MagicMock() + eng.rules = rules + return eng + + +def test_engine_does_not_call_scan_for_a_rule_with_evaluate(monkeypatch): + """evaluate() supersedes scan(); calling both repeats every Azure list call.""" + _patch_engine_client(monkeypatch, MagicMock()) + scan = MagicMock(return_value=[]) + rule = SimpleNamespace( + RULE_ID="AZ-TEST-020", + scan=scan, + evaluate=lambda *_: [no_resources_found("AZ-TEST-020", "Microsoft.Test/resources", _SUB)], + ) + + result = _engine([rule]).run_scan() + + scan.assert_not_called() + assert result["evaluations"][0]["status"] == EvaluationStatus.NOT_APPLICABLE + assert result["failed_rule_ids"] == [] + + +def test_engine_records_a_crashed_evaluate_as_a_failed_rule(monkeypatch): + """With scan() no longer run, an evaluate() crash is the rule failing to + complete and must reach failed_rule_ids, not just an ERROR row.""" + _patch_engine_client(monkeypatch, MagicMock()) + + def _boom(*_args): + raise RuntimeError("evaluator blew up") + + result = _engine([SimpleNamespace(RULE_ID="AZ-TEST-021", scan=lambda *_: [], evaluate=_boom)]).run_scan() + + assert result["failed_rule_ids"] == ["AZ-TEST-021"] + + +def test_engine_records_an_error_evaluation_as_a_failed_rule(monkeypatch): + """A rule whose inventory call failed inspected nothing; failed_rule_ids + must agree with its ERROR evaluation instead of reading as completed.""" + _patch_engine_client(monkeypatch, MagicMock()) + rule = SimpleNamespace( + RULE_ID="AZ-TEST-025", + scan=lambda *_: [], + evaluate=lambda *_: [inventory_unavailable("AZ-TEST-025", "Microsoft.Test/resources", _SUB)], + ) + + result = _engine([rule]).run_scan() + + assert result["evaluations"][0]["reason_code"] == INVENTORY_UNAVAILABLE + assert result["failed_rule_ids"] == ["AZ-TEST-025"] + + +def test_engine_does_not_fail_a_rule_for_partial_error_when_other_resources_passed(monkeypatch): + """A partial inventory failure does not make a rule with usable outcomes + a total failure; the ERROR and PASS rows are both preserved.""" + _patch_engine_client(monkeypatch, MagicMock()) + + def _evaluate(*_args): + return [ + RuleEvaluation(rule_id="AZ-TEST-026", resource_id="/r/1", resource_type="t", status=EvaluationStatus.PASS), + inventory_unavailable("AZ-TEST-026", "Microsoft.Test/second", _SUB), + ] + + result = _engine([SimpleNamespace(RULE_ID="AZ-TEST-026", scan=lambda *_: [], evaluate=_evaluate)]).run_scan() + + assert result["failed_rule_ids"] == [] + assert [e["status"] for e in result["evaluations"]] == [EvaluationStatus.PASS, EvaluationStatus.ERROR] + + +def test_engine_does_not_fail_a_rule_for_per_resource_error_when_another_passes(monkeypatch): + """A per-resource ERROR is retained but does not fail the rule if another + resource produced a usable result.""" + _patch_engine_client(monkeypatch, MagicMock()) + rule_id = "AZ-TEST-029" + + def _evaluate(*_args): + return [ + RuleEvaluation(rule_id=rule_id, resource_id="/r/1", resource_type="t", status=EvaluationStatus.PASS), + RuleEvaluation( + rule_id=rule_id, + resource_id="/r/2", + resource_type="t", + status=EvaluationStatus.ERROR, + reason_code="RESOURCE_EVALUATION_ERROR", + reason="Evidence lookup failed for this resource.", + ), + ] + + result = _engine([SimpleNamespace(RULE_ID=rule_id, evaluate=_evaluate)]).run_scan() + + assert [e["status"] for e in result["evaluations"]] == [EvaluationStatus.PASS, EvaluationStatus.ERROR] + assert result["failed_rule_ids"] == [] + + +def test_engine_fails_a_rule_when_every_per_resource_outcome_is_error(monkeypatch): + """All-error resource outcomes supply no usable coverage and count as a + total rule failure, even when the inventory itself was available.""" + _patch_engine_client(monkeypatch, MagicMock()) + rule_id = "AZ-TEST-030" + + def _evaluate(*_args): + return [ + RuleEvaluation( + rule_id=rule_id, + resource_id=f"/r/{index}", + resource_type="t", + status=EvaluationStatus.ERROR, + reason_code="RESOURCE_EVALUATION_ERROR", + reason="Evidence lookup failed for this resource.", + ) + for index in range(2) + ] + + result = _engine([SimpleNamespace(RULE_ID=rule_id, evaluate=_evaluate)]).run_scan() + + assert [e["status"] for e in result["evaluations"]] == [EvaluationStatus.ERROR, EvaluationStatus.ERROR] + assert result["failed_rule_ids"] == [rule_id] + + +def test_engine_does_not_list_a_rule_for_per_resource_unknown(monkeypatch): + """Evidence missing for some resources is UNKNOWN, not ERROR: a rule that + evaluated 8 resources and could not read 2 completed and is not listed.""" + _patch_engine_client(monkeypatch, MagicMock()) + + def _evaluate(*_args): + passed = [ + RuleEvaluation( + rule_id="AZ-TEST-027", resource_id=f"/r/{i}", resource_type="t", status=EvaluationStatus.PASS + ) + for i in range(8) + ] + unknown = [ + RuleEvaluation( + rule_id="AZ-TEST-027", + resource_id=f"/r/{i}", + resource_type="t", + status=EvaluationStatus.UNKNOWN, + reason_code=EVIDENCE_UNAVAILABLE, + ) + for i in range(8, 10) + ] + return passed + unknown + + result = _engine([SimpleNamespace(RULE_ID="AZ-TEST-027", scan=lambda *_: [], evaluate=_evaluate)]).run_scan() + + assert result["failed_rule_ids"] == [] + assert len(result["evaluations"]) == 10 + + +def test_engine_keeps_valid_evaluations_and_findings_when_some_items_are_malformed(monkeypatch): + """Dropping the whole list on one bad item would also drop real FAIL + findings; keep the valid evaluations and add one ERROR naming the bad items.""" + _patch_engine_client(monkeypatch, MagicMock()) + resource_id = "/subscriptions/x/resource/1" + + def _evaluate(*_args): + return [ + RuleEvaluation( + rule_id="AZ-TEST-028", + resource_id=resource_id, + resource_type="Microsoft.Test/resources", + status=EvaluationStatus.FAIL, + finding=_finding("AZ-TEST-028", resource_id), + ), + {"status": "PASS"}, + RuleEvaluation(rule_id="AZ-TEST-028", resource_id="/r/2", resource_type="t", status=EvaluationStatus.PASS), + None, + ] + + result = _engine([SimpleNamespace(RULE_ID="AZ-TEST-028", scan=lambda *_: [], evaluate=_evaluate)]).run_scan() + + statuses = [e["status"] for e in result["evaluations"]] + assert statuses == [EvaluationStatus.FAIL, EvaluationStatus.PASS, EvaluationStatus.ERROR] + error = result["evaluations"][-1] + assert error["reason_code"] == "MALFORMED_EVALUATION" + assert "#1 (dict)" in error["reason"] and "#3 (NoneType)" in error["reason"] + assert result["total_findings"] == 1 + assert result["findings"][0]["resource_id"] == resource_id + assert result["failed_rule_ids"] == ["AZ-TEST-028"] + + +def test_engine_rejects_evaluate_items_that_are_not_rule_evaluations(monkeypatch): + """A malformed item must become an ERROR for that rule instead of failing + the whole scan when results are serialised.""" + _patch_engine_client(monkeypatch, MagicMock()) + rule = SimpleNamespace(RULE_ID="AZ-TEST-022", scan=lambda *_: [], evaluate=lambda *_: [{"status": "PASS"}]) + + result = _engine([rule]).run_scan() + + assert len(result["evaluations"]) == 1 + assert result["evaluations"][0]["status"] == EvaluationStatus.ERROR + assert result["evaluations"][0]["reason_code"] == "MALFORMED_EVALUATION" + assert result["failed_rule_ids"] == ["AZ-TEST-022"] + + +def test_engine_reports_one_finding_per_resource_for_repeated_fail(monkeypatch): + _patch_engine_client(monkeypatch, MagicMock()) + resource_id = "/subscriptions/x/resource/1" + + def _evaluate(*_args): + return [ + RuleEvaluation( + rule_id="AZ-TEST-023", + resource_id=resource_id, + resource_type="Microsoft.Test/resources", + status=EvaluationStatus.FAIL, + finding=_finding("AZ-TEST-023", resource_id), + ) + for _ in range(2) + ] + + result = _engine([SimpleNamespace(RULE_ID="AZ-TEST-023", scan=lambda *_: [], evaluate=_evaluate)]).run_scan() + + assert result["total_findings"] == 1 + + +def test_engine_skips_a_fail_evaluation_without_a_finding(monkeypatch, caplog): + _patch_engine_client(monkeypatch, MagicMock()) + rule = SimpleNamespace( + RULE_ID="AZ-TEST-024", + scan=lambda *_: [], + evaluate=lambda *_: [ + RuleEvaluation( + rule_id="AZ-TEST-024", + resource_id="/subscriptions/x/resource/1", + resource_type="Microsoft.Test/resources", + status=EvaluationStatus.FAIL, + ) + ], + ) + + result = _engine([rule]).run_scan() + + assert result["total_findings"] == 0 + assert result["evaluations"][0]["status"] == EvaluationStatus.FAIL + assert "without a finding" in caplog.text + + # ── Persistence: rule_evaluations written in the same transaction ────────── diff --git a/tests/test_rules_keyvault.py b/tests/test_rules_keyvault.py index 4dad316d..599b98f8 100644 --- a/tests/test_rules_keyvault.py +++ b/tests/test_rules_keyvault.py @@ -241,15 +241,38 @@ def test_kv_006_evaluate_missing_properties_is_unknown_not_pass(mock_azure, subs def test_kv_006_evaluate_no_vaults_is_not_applicable(mock_azure, subscription_id): - """An empty vault list can't be told apart from a failed list call, so it - must not be reported as a PASS.""" + """A successful but empty vault list is NOT_APPLICABLE, never a PASS.""" mock_azure.set_key_vaults([]) evaluations = az_kv_006.evaluate(mock_azure, subscription_id) assert len(evaluations) == 1 assert evaluations[0].status == EvaluationStatus.NOT_APPLICABLE + assert evaluations[0].reason_code == "NO_RESOURCES_FOUND" assert evaluations[0].resource_id == f"/subscriptions/{subscription_id}" +def test_kv_006_evaluate_failed_vault_list_is_error_not_not_applicable(mock_azure, subscription_id): + """A failed list call (permissions, throttling) means nothing was checked.""" + mock_azure.set_key_vaults(None) + evaluations = az_kv_006.evaluate(mock_azure, subscription_id) + assert len(evaluations) == 1 + assert evaluations[0].status == EvaluationStatus.ERROR + assert evaluations[0].reason_code == "INVENTORY_UNAVAILABLE" + assert evaluations[0].resource_id == f"/subscriptions/{subscription_id}" + assert az_kv_006.scan(mock_azure, subscription_id) == [] + + +def test_kv_006_evaluate_records_rbac_setting_as_evidence(mock_azure, subscription_id): + mock_azure.set_key_vaults( + [ + _vault_with_props("kv-a", enable_rbac_authorization=True), + _vault_with_props("kv-b", enable_rbac_authorization=False), + ] + ) + evidence = {e.resource_id: e.evidence for e in az_kv_006.evaluate(mock_azure, subscription_id)} + assert evidence[_kv_id("kv-a")] == {"enable_rbac_authorization": True} + assert evidence[_kv_id("kv-b")] == {"enable_rbac_authorization": False} + + def test_kv_006_evaluate_reports_one_status_per_vault(mock_azure, subscription_id): mock_azure.set_key_vaults( [ diff --git a/tests/test_rules_storage.py b/tests/test_rules_storage.py index 95d06db0..3ad08a96 100644 --- a/tests/test_rules_storage.py +++ b/tests/test_rules_storage.py @@ -14,6 +14,7 @@ import scanner.rules.az_stor_008 as az_stor_008 import scanner.rules.az_stor_009 as az_stor_009 import scanner.rules.az_stor_010 as az_stor_010 +from scanner.evaluation import EvaluationStatus from tests.helpers.mock_azure import make_resource _REQUIRED_FIELDS = { @@ -375,3 +376,205 @@ def test_stor_010_unavailable_evidence_is_indeterminate_not_flagged(mock_azure, ) mock_azure.set_storage_accounts([account]) assert az_stor_010.scan(mock_azure, subscription_id) == [] + + +# ── evaluate(): coverage contract (#370) ──────────────────────────────────── +# +# Inventory failure, empty inventory, finding shape and scan() parity are +# covered for every rule by tests/test_rule_evaluation_contract.py. These +# tests pin the rule-specific branches that scan() used to skip silently. + + +def _status(evaluations, resource_id): + matches = [e for e in evaluations if e.resource_id == resource_id] + assert len(matches) == 1, f"expected one evaluation for {resource_id}, got {len(matches)}" + return matches[0].status, matches[0].reason_code + + +def test_stor_001_unset_public_access_is_unknown_not_pass(mock_azure, subscription_id): + mock_azure.set_storage_accounts( + [make_resource(id=_storage_id("sa-unset"), name="sa-unset", allow_blob_public_access=None)] + ) + evaluations = az_stor_001.evaluate(mock_azure, subscription_id) + assert _status(evaluations, _storage_id("sa-unset")) == (EvaluationStatus.UNKNOWN, "MISSING_PROPERTIES") + + +def test_stor_002_missing_field_is_unknown_and_none_still_fails(mock_azure, subscription_id): + mock_azure.set_storage_accounts( + [ + make_resource(id=_storage_id("sa-missing"), name="sa-missing"), + make_resource(id=_storage_id("sa-none"), name="sa-none", enable_https_traffic_only=None), + ] + ) + evaluations = az_stor_002.evaluate(mock_azure, subscription_id) + assert _status(evaluations, _storage_id("sa-missing")) == (EvaluationStatus.UNKNOWN, "MISSING_PROPERTIES") + assert _status(evaluations, _storage_id("sa-none")) == (EvaluationStatus.FAIL, None) + + +def test_stor_003_unreadable_policy_and_missing_resource_group_are_unknown(mock_azure, subscription_id): + no_rg_id = f"/subscriptions/{_SUB}/providers/Microsoft.Storage/storageAccounts/sa-norg" + mock_azure.set_storage_accounts( + [ + make_resource(id=_storage_id("sa-unread"), name="sa-unread", location="eastus"), + make_resource(id=no_rg_id, name="sa-norg", location="eastus"), + ] + ) + mock_azure.set_storage_lifecycle_policy(_RG, "sa-unread", None) + evaluations = az_stor_003.evaluate(mock_azure, subscription_id) + assert _status(evaluations, _storage_id("sa-unread")) == (EvaluationStatus.UNKNOWN, "EVIDENCE_UNAVAILABLE") + assert _status(evaluations, no_rg_id) == (EvaluationStatus.UNKNOWN, "MISSING_PROPERTIES") + + +def test_stor_004_reports_each_service_separately(mock_azure, subscription_id): + mock_azure.set_storage_accounts([make_resource(id=_storage_id("sa-logs"), name="sa-logs", location="eastus")]) + mock_azure.set_storage_service_logging(_RG, "sa-logs", "blob", True) + mock_azure.set_storage_service_logging(_RG, "sa-logs", "queue", False) + mock_azure.set_storage_service_logging(_RG, "sa-logs", "table", None) + evaluations = az_stor_004.evaluate(mock_azure, subscription_id) + base = _storage_id("sa-logs") + assert _status(evaluations, f"{base}/blobServices/default") == (EvaluationStatus.PASS, None) + assert _status(evaluations, f"{base}/queueServices/default") == (EvaluationStatus.FAIL, None) + assert _status(evaluations, f"{base}/tableServices/default") == ( + EvaluationStatus.UNKNOWN, + "EVIDENCE_UNAVAILABLE", + ) + assert {e.resource_type for e in evaluations} == { + "Microsoft.Storage/storageAccounts/blobServices", + "Microsoft.Storage/storageAccounts/queueServices", + "Microsoft.Storage/storageAccounts/tableServices", + } + + +def test_stor_005_missing_sku_is_unknown(mock_azure, subscription_id): + mock_azure.set_storage_accounts([make_resource(id=_storage_id("sa-nosku"), name="sa-nosku", sku=None)]) + evaluations = az_stor_005.evaluate(mock_azure, subscription_id) + assert _status(evaluations, _storage_id("sa-nosku")) == (EvaluationStatus.UNKNOWN, "MISSING_PROPERTIES") + + +def test_stor_006_account_without_id_is_unknown_instead_of_an_empty_id_finding(mock_azure, subscription_id): + """scan() used to emit a finding with resource_id "" that could not be + remediated and collided with every other ID-less finding.""" + mock_azure.set_storage_accounts([make_resource(id="", name="sa-noid", allow_shared_key_access=True)]) + evaluations = az_stor_006.evaluate(mock_azure, subscription_id) + assert _status(evaluations, f"/subscriptions/{subscription_id}") == ( + EvaluationStatus.UNKNOWN, + "MISSING_PROPERTIES", + ) + assert az_stor_006.scan(mock_azure, subscription_id) == [] + + +def test_stor_007_unrecognized_tls_value_is_unknown(mock_azure, subscription_id): + mock_azure.set_storage_accounts( + [make_resource(id=_storage_id("sa-weird"), name="sa-weird", minimum_tls_version="TLS9_9")] + ) + evaluations = az_stor_007.evaluate(mock_azure, subscription_id) + assert _status(evaluations, _storage_id("sa-weird")) == (EvaluationStatus.UNKNOWN, "UNRECOGNIZED_TLS_VERSION") + + +def test_stor_008_policy_exemptions_and_unknown_sources(mock_azure, subscription_id): + mock_azure.set_storage_accounts( + [ + make_resource(id=_storage_id("sa-untagged"), name="sa-untagged", tags={}), + make_resource( + id=_storage_id("sa-exempt"), + name="sa-exempt", + tags={"oshield:cmk-required": "true", "oshield:exception-approved": "true"}, + ), + make_resource( + id=_storage_id("sa-nosource"), + name="sa-nosource", + tags={"oshield:cmk-required": "true"}, + encryption=None, + ), + make_resource( + id=_storage_id("sa-other"), + name="sa-other", + tags={"oshield:cmk-required": "true"}, + encryption=make_resource(key_source="Other"), + ), + ] + ) + evaluations = az_stor_008.evaluate(mock_azure, subscription_id) + assert _status(evaluations, _storage_id("sa-untagged")) == (EvaluationStatus.NOT_APPLICABLE, "POLICY_NOT_REQUIRED") + assert _status(evaluations, _storage_id("sa-exempt")) == (EvaluationStatus.NOT_APPLICABLE, "APPROVED_EXCEPTION") + assert _status(evaluations, _storage_id("sa-nosource")) == (EvaluationStatus.UNKNOWN, "MISSING_PROPERTIES") + assert _status(evaluations, _storage_id("sa-other")) == (EvaluationStatus.UNKNOWN, "UNRECOGNIZED_KEY_SOURCE") + + +def test_stor_009_reports_each_container_and_account_level_gaps(mock_azure, subscription_id): + locked = make_resource(state="Locked", immutability_period_since_creation_in_days=30) + mock_azure.set_storage_accounts( + [ + make_resource(id=_storage_id("sa-mixed"), name="sa-mixed"), + make_resource(id=_storage_id("sa-empty"), name="sa-empty"), + make_resource(id=_storage_id("sa-unlisted"), name="sa-unlisted"), + ] + ) + mock_azure.set_blob_containers( + _RG, + "sa-mixed", + [ + make_resource(name="plain", tags={}, immutability_policy=None), + make_resource(name="kept", tags={"oshield:immutability-required": "true"}, immutability_policy=locked), + make_resource(name="open", tags={"oshield:immutability-required": "true"}, immutability_policy=None), + make_resource( + name="waived", + tags={"oshield:immutability-required": "true", "oshield:exception-approved": "true"}, + immutability_policy=None, + ), + ], + ) + mock_azure.set_blob_containers(_RG, "sa-empty", []) + mock_azure.set_blob_containers(_RG, "sa-unlisted", None) + evaluations = az_stor_009.evaluate(mock_azure, subscription_id) + containers = _storage_id("sa-mixed") + "/blobServices/default/containers" + assert _status(evaluations, f"{containers}/plain") == (EvaluationStatus.NOT_APPLICABLE, "POLICY_NOT_REQUIRED") + assert _status(evaluations, f"{containers}/kept") == (EvaluationStatus.PASS, None) + assert _status(evaluations, f"{containers}/open") == (EvaluationStatus.FAIL, None) + assert _status(evaluations, f"{containers}/waived") == (EvaluationStatus.NOT_APPLICABLE, "APPROVED_EXCEPTION") + assert _status(evaluations, _storage_id("sa-empty")) == (EvaluationStatus.NOT_APPLICABLE, "NO_RESOURCES_FOUND") + assert _status(evaluations, _storage_id("sa-unlisted")) == (EvaluationStatus.UNKNOWN, "EVIDENCE_UNAVAILABLE") + + +def test_stor_010_public_access_disabled_and_unavailable_connections(mock_azure, subscription_id): + mock_azure.set_storage_accounts( + [ + make_resource(id=_storage_id("sa-closed"), name="sa-closed", public_network_access="Disabled"), + make_resource( + id=_storage_id("sa-unread"), + name="sa-unread", + public_network_access="Enabled", + private_endpoint_connections=None, + ), + make_resource( + id=_storage_id("sa-private"), + name="sa-private", + public_network_access="Enabled", + private_endpoint_connections=[_private_endpoint("Approved")], + ), + ] + ) + evaluations = az_stor_010.evaluate(mock_azure, subscription_id) + assert _status(evaluations, _storage_id("sa-closed")) == (EvaluationStatus.NOT_APPLICABLE, "PUBLIC_ACCESS_DISABLED") + assert _status(evaluations, _storage_id("sa-unread")) == (EvaluationStatus.UNKNOWN, "EVIDENCE_UNAVAILABLE") + assert _status(evaluations, _storage_id("sa-private")) == (EvaluationStatus.PASS, None) + + +def test_storage_evaluations_record_the_checked_value_as_evidence(mock_azure, subscription_id): + mock_azure.set_storage_accounts( + [ + make_resource( + id=_storage_id("sa-ev"), + name="sa-ev", + allow_blob_public_access=True, + allow_shared_key_access=False, + minimum_tls_version="TLS1_2", + ) + ] + ) + assert az_stor_001.evaluate(mock_azure, subscription_id)[0].evidence == {"allow_blob_public_access": True} + assert az_stor_006.evaluate(mock_azure, subscription_id)[0].evidence == {"allow_shared_key_access": False} + assert az_stor_007.evaluate(mock_azure, subscription_id)[0].evidence == { + "minimum_tls_version": "TLS1_2", + "normalized": "TLS1_2", + }