From ac27bcf47f58b4f4fce2e1fc21a45b77028b1567 Mon Sep 17 00:00:00 2001 From: Dipesh Ray Date: Sat, 29 Aug 2026 01:59:00 +0100 Subject: [PATCH 1/2] feat(engine): add rule evaluation coverage contract (#263) scan() only ever reports violations, so a scan with no findings for a rule is indistinguishable from compliant, not-applicable, and never evaluated. get_compliance_score() inferred PASS from the absence of a finding, silently treating errored or unmigrated rules as passing. Rules can now additionally expose evaluate(azure_client, subscription_id) -> List[RuleEvaluation], reporting a PASS/FAIL/ UNKNOWN/ERROR/NOT_APPLICABLE status per resource instead of only per violation. This is additive: scan() is unchanged, and a rule without evaluate() still runs, its coverage recorded as UNKNOWN/ LEGACY_RULE_NOT_MIGRATED rather than assumed to be a pass. - New rule_evaluations table (migration chained after #308's head), with CHECK constraints on the five statuses, a non-empty canonical resource_id, and a required reason_code for UNKNOWN/ERROR/ NOT_APPLICABLE. FAIL evaluations get a nullable finding_id FK, backfilled in the same transaction as the finding insert. - Engine wiring: evaluator exceptions produce an ERROR at a canonical rule/subscription scope rather than vanishing; a FAIL evaluation contributes its own finding only when scan() hasn't already reported the same (rule_id, resource_id), so a rule implementing both never double-counts. - get_compliance_score() now derives PASS/FAIL/UNKNOWN/ERROR/ NOT_APPLICABLE from rule_evaluations instead of inferring PASS from missing findings, and returns evaluated/passed/failed/unknown/error/ not_applicable counts separately. UNKNOWN and ERROR never improve the score. - AZ-KV-006 migrated as the reference evaluate() implementation. Single Alembic head and migration-chain checks, plus populated-database CHECK constraint tests gated on DATABASE_URL (CI already runs these against a live migrated Postgres). Signed-off-by: Dipesh Ray --- .../versions/3f59f83a5253_rule_evaluations.py | 85 +++++ api/models/finding.py | 86 ++++- docs/adding-a-rule.md | 24 ++ scanner/engine.py | 62 ++++ scanner/evaluation.py | 96 +++++ scanner/rules/az_kv_006.py | 105 ++++-- tests/test_alembic_migrations.py | 154 ++++++++ tests/test_clean_scan.py | 64 +++- tests/test_rule_evaluations.py | 333 ++++++++++++++++++ tests/test_rules_keyvault.py | 56 +++ tests/test_severity_contract.py | 4 +- 11 files changed, 1031 insertions(+), 38 deletions(-) create mode 100644 alembic/versions/3f59f83a5253_rule_evaluations.py create mode 100644 scanner/evaluation.py create mode 100644 tests/test_alembic_migrations.py create mode 100644 tests/test_rule_evaluations.py diff --git a/alembic/versions/3f59f83a5253_rule_evaluations.py b/alembic/versions/3f59f83a5253_rule_evaluations.py new file mode 100644 index 00000000..28dce658 --- /dev/null +++ b/alembic/versions/3f59f83a5253_rule_evaluations.py @@ -0,0 +1,85 @@ +"""Add rule_evaluations: per-resource coverage, not just findings (#263). + +Revision ID: 3f59f83a5253 +Revises: d8e4f6a1b2c3 +Create Date: 2026-08-29 00:00:00.000000 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +# Revision identifiers, used by Alembic. +revision: str = "3f59f83a5253" +down_revision: Union[str, Sequence[str], None] = "d8e4f6a1b2c3" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +_STATUS_CONSTRAINT = "ck_rule_evaluations_status_v1" +_SCOPE_CONSTRAINT = "ck_rule_evaluations_resource_id_not_empty" +_REASON_CONSTRAINT = "ck_rule_evaluations_reason_code_required" + + +def upgrade() -> None: + op.create_table( + "rule_evaluations", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("scan_id", postgresql.UUID(), nullable=False), + sa.Column("rule_id", sa.Text(), nullable=False), + sa.Column("resource_id", sa.Text(), nullable=False), + sa.Column("resource_type", sa.Text(), nullable=False, server_default=sa.text("''")), + sa.Column("status", sa.Text(), nullable=False), + sa.Column("reason_code", sa.Text(), nullable=True), + sa.Column("reason", sa.Text(), nullable=True), + sa.Column("evidence", postgresql.JSONB(), server_default=sa.text("'{}'::jsonb"), nullable=True), + # Nullable: only set for FAIL evaluations, and only once the finding + # row exists. Populated in the same transaction as the finding insert + # (see DatabaseManager.save_scan), never inferred after the fact. + sa.Column("finding_id", sa.Integer(), nullable=True), + sa.Column("evaluated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["scan_id"], ["scans.scan_id"], name="rule_evaluations_scan_id_fkey"), + sa.ForeignKeyConstraint( + ["finding_id"], ["findings.id"], name="rule_evaluations_finding_id_fkey", ondelete="SET NULL" + ), + sa.PrimaryKeyConstraint("id", name="rule_evaluations_pkey"), + # One coverage statement per rule per resource per scan. Also gives + # the persistence-layer FAIL -> finding_id backfill a stable join key. + sa.UniqueConstraint("scan_id", "rule_id", "resource_id", name="uq_rule_evaluations_scan_rule_resource"), + ) + + op.create_index("idx_rule_evaluations_scan_id", "rule_evaluations", ["scan_id"], unique=False) + op.create_index("idx_rule_evaluations_rule_id", "rule_evaluations", ["rule_id"], unique=False) + op.create_index("idx_rule_evaluations_status", "rule_evaluations", ["status"], unique=False) + + op.create_check_constraint( + _STATUS_CONSTRAINT, + "rule_evaluations", + "status IN ('PASS', 'FAIL', 'UNKNOWN', 'ERROR', 'NOT_APPLICABLE')", + ) + # A canonical scope identifier is required — never an empty string standing + # in for "no specific resource" (that collides across rules/subscriptions). + op.create_check_constraint( + _SCOPE_CONSTRAINT, + "rule_evaluations", + "resource_id <> ''", + ) + # UNKNOWN/ERROR/NOT_APPLICABLE must always explain themselves; only PASS + # and FAIL are self-evident from the status alone. + op.create_check_constraint( + _REASON_CONSTRAINT, + "rule_evaluations", + "status NOT IN ('UNKNOWN', 'ERROR', 'NOT_APPLICABLE') " + "OR (reason_code IS NOT NULL AND reason_code <> '')", + ) + + +def downgrade() -> None: + op.drop_constraint(_REASON_CONSTRAINT, "rule_evaluations", type_="check") + op.drop_constraint(_SCOPE_CONSTRAINT, "rule_evaluations", type_="check") + op.drop_constraint(_STATUS_CONSTRAINT, "rule_evaluations", type_="check") + op.drop_index("idx_rule_evaluations_status", table_name="rule_evaluations") + op.drop_index("idx_rule_evaluations_rule_id", table_name="rule_evaluations") + op.drop_index("idx_rule_evaluations_scan_id", table_name="rule_evaluations") + op.drop_table("rule_evaluations") diff --git a/api/models/finding.py b/api/models/finding.py index 0f366924..1e0df604 100644 --- a/api/models/finding.py +++ b/api/models/finding.py @@ -19,6 +19,7 @@ score_findings, severity_rank, ) +from scanner.evaluation import EvaluationStatus, aggregate_status logger = logging.getLogger(__name__) @@ -213,6 +214,8 @@ def save_scan(self, scan_result: Dict[str, Any]) -> None: # keeps the scan header, child rows, and recomputed score in # agreement instead of duplicating findings on every attempt. cur.execute("DELETE FROM findings WHERE scan_id = %s", (scan_result["scan_id"],)) + cur.execute("DELETE FROM rule_evaluations WHERE scan_id = %s", (scan_result["scan_id"],)) + finding_id_by_key: Dict[Any, int] = {} for f in findings: cur.execute( """ @@ -223,6 +226,7 @@ def save_scan(self, scan_result: Dict[str, Any]) -> None: frameworks, metadata, cve_references, cvss_score, exploit_available, detected_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) + RETURNING id """, ( # The parent scan owns every child in this batch. @@ -246,6 +250,39 @@ def save_scan(self, scan_result: Dict[str, Any]) -> None: f.get("detected_at"), ), ) + finding_id_by_key[(f.get("rule_id"), f.get("resource_id"))] = cur.fetchone()[0] + + # Coverage rows (#263): a status for every resource a migrated + # rule looked at, not just its violations. A FAIL evaluation + # is durably linked to the finding row it corresponds to + # right here, in the same transaction, instead of leaving + # callers to infer the relationship from rule_id/resource_id. + evaluated_at = completed_at + for evaluation in scan_result.get("evaluations", []): + status = evaluation.get("status") + finding_id = None + if status == EvaluationStatus.FAIL: + finding_id = finding_id_by_key.get((evaluation.get("rule_id"), evaluation.get("resource_id"))) + cur.execute( + """ + INSERT INTO rule_evaluations + (scan_id, rule_id, resource_id, resource_type, status, + reason_code, reason, evidence, finding_id, evaluated_at) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) + """, + ( + scan_result["scan_id"], + evaluation.get("rule_id"), + evaluation.get("resource_id"), + evaluation.get("resource_type") or "", + status, + evaluation.get("reason_code"), + evaluation.get("reason"), + json.dumps(evaluation.get("evidence", {})), + finding_id, + evaluated_at, + ), + ) conn.commit() except Exception: # psycopg2 connections remain in an aborted transaction after any @@ -584,9 +621,11 @@ def get_compliance_score(self, framework: str) -> Dict[str, Any]: controls = framework_data.get("controls", {}) - # Get failure detail from the latest completed scan only. This does - # not change the legacy absence-implies-PASS behavior tracked by #263; - # it prevents the frontend from inventing MEDIUM for failed controls. + # Finding detail (severity/category/resource count) still comes from + # findings — evaluations don't carry severity. Pass/fail/unknown/error + # status comes from rule_evaluations, so a rule that was never run, + # errored, or hasn't been migrated to evaluate() yet is never silently + # reported as PASS just because it produced no findings (#263). conn = self._get_conn() with conn.cursor() as cur: cur.execute( @@ -601,6 +640,17 @@ def get_compliance_score(self, framework: str) -> Dict[str, Any]: ) finding_rows = cur.fetchall() + cur.execute( + """ + SELECT rule_id, status + FROM rule_evaluations + WHERE scan_id = ( + SELECT scan_id FROM scans WHERE status = 'completed' ORDER BY started_at DESC LIMIT 1 + ) + """ + ) + evaluation_rows = cur.fetchall() + failures: Dict[str, Dict[str, Any]] = {} for rule_id, raw_severity, category, resource_count in finding_rows: severity = normalize_severity(raw_severity) @@ -617,10 +667,18 @@ def get_compliance_score(self, framework: str) -> Dict[str, Any]: current["severity"] = severity current["category"] = category + statuses_by_rule: Dict[str, List[str]] = {} + for rule_id, status in evaluation_rows: + statuses_by_rule.setdefault(rule_id, []).append(status) + aggregated_status = {rule_id: aggregate_status(statuses) for rule_id, statuses in statuses_by_rule.items()} + results = [] for rule_id, control in controls.items(): failure = failures.get(rule_id) - status = "FAIL" if failure else "PASS" + # No evaluation row at all means this rule was never run against + # this scan (predates rule_evaluations, or was skipped) — report + # UNKNOWN rather than defaulting to PASS or inferring from findings. + status = aggregated_status.get(rule_id, EvaluationStatus.UNKNOWN) results.append( { "rule_id": rule_id, @@ -634,16 +692,26 @@ def get_compliance_score(self, framework: str) -> Dict[str, Any]: ) total = len(results) - passed = sum(1 for r in results if r["status"] == "PASS") - failed = total - passed - score_pct = round((passed / total) * 100) if total else 0 + counts = { + "passed": sum(1 for r in results if r["status"] == EvaluationStatus.PASS), + "failed": sum(1 for r in results if r["status"] == EvaluationStatus.FAIL), + "unknown": sum(1 for r in results if r["status"] == EvaluationStatus.UNKNOWN), + "error": sum(1 for r in results if r["status"] == EvaluationStatus.ERROR), + "not_applicable": sum(1 for r in results if r["status"] == EvaluationStatus.NOT_APPLICABLE), + } + # UNKNOWN/ERROR must never improve the score: they count against the + # denominator (evaluated coverage) without counting as a pass. + # NOT_APPLICABLE controls fall outside the denominator entirely. + evaluated = total - counts["not_applicable"] + score_pct = round((counts["passed"] / evaluated) * 100) if evaluated else 0 return { "framework": framework_data.get("framework"), "version": framework_data.get("version"), + "contract_version": "2", "total_controls": total, - "passed": passed, - "failed": failed, + "evaluated": evaluated, + **counts, "score_percent": score_pct, "controls": results, } diff --git a/docs/adding-a-rule.md b/docs/adding-a-rule.md index b93ab33e..4bf0eed4 100644 --- a/docs/adding-a-rule.md +++ b/docs/adding-a-rule.md @@ -149,6 +149,30 @@ When a helper returns `None`, skip the resource and log a warning. Never create --- +## Optional: 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: + +```python +from scanner.evaluation import EvaluationStatus, RuleEvaluation, subscription_scope_id + +def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: + """Report a status for every resource this rule looked at, PASS included.""" +``` + +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. + +Rules of the contract (see `scanner/evaluation.py` and `scanner/rules/az_kv_006.py` for the reference implementation): + +- `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`. + +You don't need to migrate an existing rule's `scan()` to add `evaluate()` — most rules can leave `scan()` exactly as-is. + +--- + ## Write the Remediation Playbook Create a matching bash script in `playbooks/cli/`: diff --git a/scanner/engine.py b/scanner/engine.py index f7af9aa2..2ecf7468 100644 --- a/scanner/engine.py +++ b/scanner/engine.py @@ -10,6 +10,7 @@ from api.observability import RULE_ERRORS_TOTAL from openshield.severity import CONTRACT_VERSION, SeverityContractError, normalize_severity, score_findings from scanner.azure_client import AzureClient +from scanner.evaluation import EvaluationStatus, RuleEvaluation, subscription_scope_id logger = logging.getLogger(__name__) @@ -108,6 +109,7 @@ def run_scan(self, scan_id: Optional[str] = None) -> Dict[str, Any]: scan_id = scan_id or str(uuid.uuid4()) started_at = datetime.now(timezone.utc).isoformat() findings: List[Dict[str, Any]] = [] + evaluations: List[RuleEvaluation] = [] detected_at = datetime.now(timezone.utc).isoformat() logger.info( @@ -146,6 +148,25 @@ def run_scan(self, scan_id: Optional[str] = None) -> Dict[str, Any]: RULE_ERRORS_TOTAL.labels(rule_id=rule_id).inc() logger.error("Rule %s raised an exception: %s", rule_id, exc, exc_info=True) + evaluations.extend(self._evaluate_rule(rule, 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. + 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: + continue + key = (rule_evaluation.rule_id, rule_evaluation.resource_id) + if key in existing_keys: + continue + finding = dict(rule_evaluation.finding) + finding["severity"] = normalize_severity(finding.get("severity")) + finding.setdefault("detected_at", detected_at) + finding.setdefault("scan_id", scan_id) + findings.append(finding) + existing_keys.add(key) + completed_at = datetime.now(timezone.utc).isoformat() score = score_findings(findings) @@ -161,8 +182,49 @@ def run_scan(self, scan_id: Optional[str] = None) -> Dict[str, Any]: "score": score, "severity_contract_version": CONTRACT_VERSION, "findings": findings, + "evaluations": [e.to_dict() for e in evaluations], } logger.info("Scan %s complete — %d total finding(s). Normalising results...", scan_id, len(findings)) 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. + + 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. + """ + 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.", + ) + ] + + try: + rule_evaluations = evaluate_fn(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), + ) + ] diff --git a/scanner/evaluation.py b/scanner/evaluation.py new file mode 100644 index 00000000..4b89fd1c --- /dev/null +++ b/scanner/evaluation.py @@ -0,0 +1,96 @@ +"""Rule evaluation contract (issue #263): per-resource coverage, not just findings. + +``scan()`` only ever reports violations, so the absence of a finding is +indistinguishable from "compliant" and "never evaluated" — a rule that +errors out or hasn't been migrated yet silently reads as a pass. A rule +opts into this contract by additionally exposing:: + + 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 +UNKNOWN/LEGACY_RULE_NOT_MIGRATED instead of inventing a PASS. +""" + +from dataclasses import dataclass, field +from typing import Any, Dict, Iterable, Optional + +# Conservative rank: worse coverage information must never be hidden by +# better information rolled up from a different resource under the same rule. +_AGGREGATE_ORDER = ("FAIL", "ERROR", "UNKNOWN", "PASS", "NOT_APPLICABLE") +_RANK = {status: i for i, status in enumerate(_AGGREGATE_ORDER)} + +STATUSES = frozenset(_AGGREGATE_ORDER) +_REASON_REQUIRED = frozenset({"UNKNOWN", "ERROR", "NOT_APPLICABLE"}) + + +class EvaluationStatus: + """Canonical evaluation outcomes. Plain string constants, not an enum + class, so a status can be stored/compared as the same string Postgres's + CHECK constraint enforces.""" + + PASS = "PASS" + FAIL = "FAIL" + UNKNOWN = "UNKNOWN" + ERROR = "ERROR" + NOT_APPLICABLE = "NOT_APPLICABLE" + + +def subscription_scope_id(subscription_id: str) -> str: + """Canonical non-empty resource_id for a subscription/rule-level result. + + Used whenever an evaluation isn't about one specific resource (a legacy + rule's placeholder, an evaluator exception with no resource to blame). + Never an empty string — that would collide across rules/subscriptions. + """ + return f"/subscriptions/{subscription_id}" + + +@dataclass +class RuleEvaluation: + """One rule's coverage statement about one resource (or the subscription).""" + + rule_id: str + resource_id: str + resource_type: str + status: str + reason_code: Optional[str] = None + reason: Optional[str] = None + evidence: Dict[str, Any] = field(default_factory=dict) + finding: Optional[Dict[str, Any]] = None + + def __post_init__(self) -> None: + if self.status not in STATUSES: + raise ValueError(f"unsupported evaluation status: {self.status!r}") + if not self.resource_id: + raise ValueError("RuleEvaluation.resource_id must be a non-empty canonical identifier") + if self.status in _REASON_REQUIRED and not self.reason_code: + raise ValueError(f"status {self.status} requires a reason_code") + + def to_dict(self) -> Dict[str, Any]: + return { + "rule_id": self.rule_id, + "resource_id": self.resource_id, + "resource_type": self.resource_type, + "status": self.status, + "reason_code": self.reason_code, + "reason": self.reason, + "evidence": self.evidence, + } + + +def aggregate_status(statuses: Iterable[str]) -> str: + """Roll up several resource-level statuses for one rule into one status. + + FAIL beats ERROR beats UNKNOWN beats PASS beats NOT_APPLICABLE, so a + single bad resource (or a single evaluator failure) can never be + outvoted by resources that happened to pass. + """ + best = None + for status in statuses: + if best is None or _RANK[status] < _RANK[best]: + best = status + if best is None: + raise ValueError("aggregate_status requires at least one status") + return best diff --git a/scanner/rules/az_kv_006.py b/scanner/rules/az_kv_006.py index 1e329338..3dd01e40 100644 --- a/scanner/rules/az_kv_006.py +++ b/scanner/rules/az_kv_006.py @@ -2,6 +2,8 @@ from typing import Any, Dict, List +from scanner.evaluation import EvaluationStatus, RuleEvaluation, subscription_scope_id + RULE_ID = "AZ-KV-006" RULE_NAME = "Key Vault Using Legacy Access Policies Instead of Azure RBAC" SEVERITY = "MEDIUM" @@ -22,6 +24,27 @@ PLAYBOOK = "playbooks/cli/fix_az_kv_006.sh" +def _finding(azure_client: Any, vault: Any) -> Dict[str, Any]: + parsed = azure_client.parse_resource_id(vault.id) + return { + "rule_id": RULE_ID, + "rule_name": RULE_NAME, + "severity": SEVERITY, + "category": CATEGORY, + "resource_id": vault.id, + "resource_name": vault.name, + "resource_type": "Microsoft.KeyVault/vaults", + "description": DESCRIPTION, + "remediation": REMEDIATION, + "playbook": PLAYBOOK, + "frameworks": FRAMEWORKS, + "metadata": { + "resource_group": parsed.get("resource_group", ""), + "location": getattr(vault, "location", ""), + }, + } + + 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]] = [] @@ -34,25 +57,67 @@ def scan(azure_client: Any, subscription_id: str) -> List[Dict[str, Any]]: # Access policies are the legacy default; a vault must opt into RBAC. rbac_enabled = getattr(props, "enable_rbac_authorization", False) if not rbac_enabled: - parsed = azure_client.parse_resource_id(vault.id) - findings.append( - { - "rule_id": RULE_ID, - "rule_name": RULE_NAME, - "severity": SEVERITY, - "category": CATEGORY, - "resource_id": vault.id, - "resource_name": vault.name, - "resource_type": "Microsoft.KeyVault/vaults", - "description": DESCRIPTION, - "remediation": REMEDIATION, - "playbook": PLAYBOOK, - "frameworks": FRAMEWORKS, - "metadata": { - "resource_group": parsed.get("resource_group", ""), - "location": getattr(vault, "location", ""), - }, - } - ) + findings.append(_finding(azure_client, vault)) return findings + + +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() + 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.", + ) + ] + + evaluations: List[RuleEvaluation] = [] + for vault in vaults: + props = getattr(vault, "properties", None) + if props is None: + evaluations.append( + RuleEvaluation( + rule_id=RULE_ID, + resource_id=vault.id, + resource_type="Microsoft.KeyVault/vaults", + status=EvaluationStatus.UNKNOWN, + reason_code="MISSING_PROPERTIES", + reason="Key Vault was returned without a properties payload.", + ) + ) + continue + + rbac_enabled = getattr(props, "enable_rbac_authorization", False) + if rbac_enabled: + evaluations.append( + RuleEvaluation( + rule_id=RULE_ID, + resource_id=vault.id, + resource_type="Microsoft.KeyVault/vaults", + status=EvaluationStatus.PASS, + ) + ) + else: + evaluations.append( + RuleEvaluation( + rule_id=RULE_ID, + resource_id=vault.id, + resource_type="Microsoft.KeyVault/vaults", + status=EvaluationStatus.FAIL, + finding=_finding(azure_client, vault), + ) + ) + + return evaluations diff --git a/tests/test_alembic_migrations.py b/tests/test_alembic_migrations.py new file mode 100644 index 00000000..1817bb29 --- /dev/null +++ b/tests/test_alembic_migrations.py @@ -0,0 +1,154 @@ +"""Alembic migration-chain and populated-database checks for rule_evaluations (#263). + +The head/chain checks are static (parse revision/down_revision out of every +migration file) and always run. The CHECK-constraint checks are populated- +database checks: CI applies `alembic upgrade head` against a real Postgres +before pytest runs (see .github/workflows/ci.yml), so DATABASE_URL points at +an already-migrated database there. They're skipped when no DATABASE_URL is +set (e.g. a local run with no Postgres available). +""" + +import os +import re +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +VERSIONS_DIR = ROOT / "alembic" / "versions" + +_REV_RE = re.compile(r"^revision:.*=\s*[\"']([a-f0-9]+)[\"']", re.MULTILINE) +_DOWN_RE = re.compile(r"^down_revision.*=\s*[\"']([a-f0-9]+)[\"']", re.MULTILINE) + + +def _revision_graph(): + graph = {} + for path in VERSIONS_DIR.glob("*.py"): + text = path.read_text(encoding="utf-8") + rev = _REV_RE.search(text) + down = _DOWN_RE.search(text) + assert rev, f"{path.name}: no revision id found" + graph[rev.group(1)] = (path.name, down.group(1) if down else None) + return graph + + +def test_single_alembic_head(): + """Exactly one migration must have no other migration pointing at it as + its parent — two heads means a broken/forked migration history.""" + graph = _revision_graph() + downs = {down for _, down in graph.values() if down} + heads = [rev for rev in graph if rev not in downs] + assert len(heads) == 1, f"expected exactly one Alembic head, found {heads}" + + +def test_rule_evaluations_migration_chains_after_severity_contract_v1(): + graph = _revision_graph() + _, down = graph["3f59f83a5253"] + assert down == "d8e4f6a1b2c3" + + +def test_rule_evaluations_migration_defines_all_five_statuses(): + migration = (VERSIONS_DIR / "3f59f83a5253_rule_evaluations.py").read_text(encoding="utf-8") + for status in ("PASS", "FAIL", "UNKNOWN", "ERROR", "NOT_APPLICABLE"): + assert f"'{status}'" in migration + + +# ── Populated-database checks (require a live migrated Postgres) ─────────── + +DATABASE_URL = os.environ.get("DATABASE_URL") +pytestmark_db = pytest.mark.skipif(not DATABASE_URL, reason="requires a live migrated database (DATABASE_URL)") + + +def _conn(): + import psycopg2 + + conn = psycopg2.connect(DATABASE_URL) + conn.autocommit = False + return conn + + +def _seed_scan(cur, scan_id): + cur.execute( + "INSERT INTO scans (scan_id, subscription_id, started_at, status) " + "VALUES (%s, %s, now(), 'completed') ON CONFLICT (scan_id) DO NOTHING", + (scan_id, "00000000-0000-0000-0000-000000000001"), + ) + + +@pytestmark_db +def test_rule_evaluations_check_constraint_rejects_unsupported_status(): + import psycopg2 + + conn = _conn() + try: + with conn.cursor() as cur: + _seed_scan(cur, "10000000-0000-0000-0000-000000000001") + with pytest.raises(psycopg2.errors.CheckViolation): + cur.execute( + "INSERT INTO rule_evaluations " + "(scan_id, rule_id, resource_id, resource_type, status, evaluated_at) " + "VALUES (%s, 'AZ-TEST-001', '/subscriptions/x', '', 'BOGUS', now())", + ("10000000-0000-0000-0000-000000000001",), + ) + finally: + conn.rollback() + conn.close() + + +@pytestmark_db +def test_rule_evaluations_check_constraint_rejects_empty_resource_id(): + import psycopg2 + + conn = _conn() + try: + with conn.cursor() as cur: + _seed_scan(cur, "10000000-0000-0000-0000-000000000002") + with pytest.raises(psycopg2.errors.CheckViolation): + cur.execute( + "INSERT INTO rule_evaluations " + "(scan_id, rule_id, resource_id, resource_type, status, evaluated_at) " + "VALUES (%s, 'AZ-TEST-001', '', '', 'PASS', now())", + ("10000000-0000-0000-0000-000000000002",), + ) + finally: + conn.rollback() + conn.close() + + +@pytestmark_db +def test_rule_evaluations_check_constraint_requires_reason_code_for_unknown(): + import psycopg2 + + conn = _conn() + try: + with conn.cursor() as cur: + _seed_scan(cur, "10000000-0000-0000-0000-000000000003") + with pytest.raises(psycopg2.errors.CheckViolation): + cur.execute( + "INSERT INTO rule_evaluations " + "(scan_id, rule_id, resource_id, resource_type, status, reason_code, evaluated_at) " + "VALUES (%s, 'AZ-TEST-001', '/subscriptions/x', '', 'UNKNOWN', NULL, now())", + ("10000000-0000-0000-0000-000000000003",), + ) + finally: + conn.rollback() + conn.close() + + +@pytestmark_db +def test_rule_evaluations_accepts_a_valid_pass_row_and_finding_id_is_nullable(): + conn = _conn() + try: + with conn.cursor() as cur: + _seed_scan(cur, "10000000-0000-0000-0000-000000000004") + cur.execute( + "INSERT INTO rule_evaluations " + "(scan_id, rule_id, resource_id, resource_type, status, evaluated_at) " + "VALUES (%s, 'AZ-TEST-001', '/subscriptions/x', 'Microsoft.Test/resources', 'PASS', now()) " + "RETURNING finding_id", + ("10000000-0000-0000-0000-000000000004",), + ) + assert cur.fetchone()[0] is None + finally: + conn.rollback() + conn.close() diff --git a/tests/test_clean_scan.py b/tests/test_clean_scan.py index 55bbfbf5..ae7e8b8a 100644 --- a/tests/test_clean_scan.py +++ b/tests/test_clean_scan.py @@ -17,12 +17,20 @@ def _db() -> DatabaseManager: return db -def _mock_cursor(rows): - """Return a context-manager cursor mock that yields *rows* on fetchall().""" +def _mock_cursor(rows, evaluation_rows=None): + """Return a context-manager cursor mock that yields *rows* on fetchall(). + + get_compliance_score() issues two SELECT/fetchall pairs (findings, then + rule_evaluations); pass *evaluation_rows* to give the second call its own + result instead of reusing *rows*. + """ cur = MagicMock() cur.__enter__ = lambda s: s cur.__exit__ = MagicMock(return_value=False) - cur.fetchall.return_value = rows + if evaluation_rows is not None: + cur.fetchall.side_effect = [rows, evaluation_rows] + else: + cur.fetchall.return_value = rows cur.fetchone.return_value = rows[0] if rows else None return cur @@ -137,10 +145,14 @@ def test_get_compliance_score_scopes_to_latest_scan(): def test_get_compliance_score_all_pass_after_clean_scan(): - """All controls must show PASS when the latest completed scan has no findings.""" + """All controls show PASS when the latest scan recorded a PASS evaluation + for every rule and produced no findings.""" db = _db() conn = MagicMock() - cur = _mock_cursor([]) + cur = _mock_cursor( + [], + evaluation_rows=[("AZ-STOR-001", "PASS"), ("AZ-NET-001", "PASS")], + ) conn.cursor.return_value = cur import json @@ -171,11 +183,46 @@ def test_get_compliance_score_all_pass_after_clean_scan(): assert statuses["AZ-NET-001"] == "PASS" +def test_get_compliance_score_no_evaluation_rows_is_unknown_not_pass(): + """A clean (zero-finding) scan with no rule_evaluations rows at all — e.g. + a scan that predates #263, or a rule that was never evaluated — must + report UNKNOWN, never silently default to PASS (the bug #263 fixes).""" + db = _db() + conn = MagicMock() + cur = _mock_cursor([], evaluation_rows=[]) + conn.cursor.return_value = cur + + import json + import io + from pathlib import Path + + fake_framework = json.dumps( + { + "framework": "CIS Azure", + "version": "2.0", + "controls": { + "AZ-STOR-001": {"control_id": "3.1", "control_name": "No public blobs"}, + }, + } + ) + + with patch.object(db, "_get_conn", return_value=conn): + with patch("builtins.open", return_value=io.StringIO(fake_framework)): + with patch.object(Path, "exists", return_value=True): + result = db.get_compliance_score("cis") + + assert result["controls"][0]["status"] == "UNKNOWN" + assert result["passed"] == 0 + assert result["unknown"] == 1 + assert result["score_percent"] == 0 + + def test_get_compliance_score_remediated_rule_shows_pass(): - """A rule that fired in scan-1 but not scan-2 (clean) must show PASS.""" + """A rule that failed in scan-1 but recorded a PASS evaluation in scan-2 + (clean, remediated) must show PASS.""" db = _db() conn = MagicMock() - cur = _mock_cursor([]) + cur = _mock_cursor([], evaluation_rows=[("AZ-STOR-001", "PASS")]) conn.cursor.return_value = cur import json @@ -207,7 +254,8 @@ def test_get_compliance_score_reports_worst_critical_failure_without_inventing_p [ ("AZ-STOR-001", "HIGH", "Storage", 1), ("AZ-STOR-001", "CRITICAL", "Storage", 2), - ] + ], + evaluation_rows=[("AZ-STOR-001", "FAIL"), ("AZ-NET-001", "PASS")], ) conn.cursor.return_value = cur diff --git a/tests/test_rule_evaluations.py b/tests/test_rule_evaluations.py new file mode 100644 index 00000000..793cbffb --- /dev/null +++ b/tests/test_rule_evaluations.py @@ -0,0 +1,333 @@ +"""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 +DatabaseManager persistence of rule_evaluations in the same transaction as +findings. +""" + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +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 + +_SUB = "00000000-0000-0000-0000-000000000001" + + +# ── RuleEvaluation / EvaluationStatus contract ────────────────────────────── + + +def test_rule_evaluation_rejects_unknown_status(): + with pytest.raises(ValueError, match="unsupported evaluation status"): + RuleEvaluation(rule_id="AZ-TEST-001", resource_id="/subscriptions/x", resource_type="", status="BOGUS") + + +def test_rule_evaluation_rejects_empty_resource_id(): + with pytest.raises(ValueError, match="non-empty canonical identifier"): + RuleEvaluation(rule_id="AZ-TEST-001", resource_id="", resource_type="", status=EvaluationStatus.PASS) + + +@pytest.mark.parametrize("status", [EvaluationStatus.UNKNOWN, EvaluationStatus.ERROR, EvaluationStatus.NOT_APPLICABLE]) +def test_rule_evaluation_requires_reason_code_for_non_terminal_statuses(status): + with pytest.raises(ValueError, match="requires a reason_code"): + RuleEvaluation(rule_id="AZ-TEST-001", resource_id="/subscriptions/x", resource_type="", status=status) + + +def test_subscription_scope_id_is_non_empty_and_stable(): + scope = subscription_scope_id(_SUB) + assert scope == f"/subscriptions/{_SUB}" + + +def test_aggregate_status_conservative_order(): + # FAIL > ERROR > UNKNOWN > PASS > NOT_APPLICABLE regardless of input order. + assert aggregate_status(["PASS", "FAIL", "UNKNOWN"]) == "FAIL" + assert aggregate_status(["PASS", "ERROR"]) == "ERROR" + assert aggregate_status(["PASS", "UNKNOWN"]) == "UNKNOWN" + assert aggregate_status(["NOT_APPLICABLE", "PASS"]) == "PASS" + assert aggregate_status(["NOT_APPLICABLE"]) == "NOT_APPLICABLE" + + +def test_aggregate_status_requires_at_least_one(): + with pytest.raises(ValueError): + aggregate_status([]) + + +# ── Engine wiring ──────────────────────────────────────────────────────────── + + +def _patch_engine_client(monkeypatch, client): + monkeypatch.setattr(engine_mod, "AzureClient", lambda subscription_id: client) + + +def test_legacy_rule_without_evaluate_is_recorded_as_unknown(monkeypatch): + """A rule with only scan() must never contribute a PASS — its coverage is + UNKNOWN/LEGACY_RULE_NOT_MIGRATED, not silently absent.""" + _patch_engine_client(monkeypatch, MagicMock()) + eng = ScanEngine.__new__(ScanEngine) + eng.subscription_id = _SUB + eng.client = MagicMock() + eng.rules = [SimpleNamespace(RULE_ID="AZ-TEST-001", scan=lambda *_: [])] + + result = eng.run_scan() + + evaluations = result["evaluations"] + assert len(evaluations) == 1 + assert evaluations[0]["status"] == EvaluationStatus.UNKNOWN + assert evaluations[0]["reason_code"] == "LEGACY_RULE_NOT_MIGRATED" + assert evaluations[0]["resource_id"] == subscription_scope_id(_SUB) + + +def test_evaluate_exception_produces_error_at_canonical_scope(monkeypatch): + """A rule whose evaluate() raises must still produce a coverage row (ERROR), + not silently vanish the way a bare scan() exception does.""" + _patch_engine_client(monkeypatch, MagicMock()) + eng = ScanEngine.__new__(ScanEngine) + eng.subscription_id = _SUB + eng.client = MagicMock() + + def _boom(*_args, **_kwargs): + raise RuntimeError("evaluator blew up") + + eng.rules = [SimpleNamespace(RULE_ID="AZ-TEST-002", scan=lambda *_: [], evaluate=_boom)] + + result = eng.run_scan() # must not raise + + evaluations = result["evaluations"] + assert len(evaluations) == 1 + assert evaluations[0]["status"] == EvaluationStatus.ERROR + assert evaluations[0]["reason_code"] == "EVALUATOR_EXCEPTION" + assert evaluations[0]["resource_id"] == subscription_scope_id(_SUB) + + +def test_evaluate_must_return_a_list(monkeypatch): + """A non-list return from evaluate() is treated the same as a raised + exception (ERROR), not silently accepted or crashed on.""" + _patch_engine_client(monkeypatch, MagicMock()) + eng = ScanEngine.__new__(ScanEngine) + eng.subscription_id = _SUB + eng.client = MagicMock() + eng.rules = [SimpleNamespace(RULE_ID="AZ-TEST-003", scan=lambda *_: [], evaluate=lambda *_: "not-a-list")] + + result = eng.run_scan() + + assert result["evaluations"][0]["status"] == EvaluationStatus.ERROR + assert result["evaluations"][0]["reason_code"] == "EVALUATOR_EXCEPTION" + + +def test_fail_evaluation_contributes_finding_when_scan_did_not_already_report_it(monkeypatch): + """A rule that is evaluate()-only (no matching scan() finding) must still + have its FAIL surfaced as a real finding, not just a status row.""" + _patch_engine_client(monkeypatch, MagicMock()) + eng = ScanEngine.__new__(ScanEngine) + eng.subscription_id = _SUB + eng.client = MagicMock() + + finding = { + "rule_id": "AZ-TEST-004", + "rule_name": "Test", + "severity": "HIGH", + "category": "Test", + "resource_id": "/subscriptions/x/resource/1", + "resource_name": "r1", + "resource_type": "Microsoft.Test/resources", + "description": "d", + "remediation": "r", + "playbook": "playbooks/cli/fix_az_test_004.sh", + "frameworks": {}, + "metadata": {}, + } + eng.rules = [ + SimpleNamespace( + RULE_ID="AZ-TEST-004", + scan=lambda *_: [], + evaluate=lambda *_: [ + RuleEvaluation( + rule_id="AZ-TEST-004", + resource_id=finding["resource_id"], + resource_type=finding["resource_type"], + status=EvaluationStatus.FAIL, + finding=finding, + ) + ], + ) + ] + + result = eng.run_scan() + + assert result["total_findings"] == 1 + 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.""" + _patch_engine_client(monkeypatch, MagicMock()) + eng = ScanEngine.__new__(ScanEngine) + eng.subscription_id = _SUB + eng.client = MagicMock() + + shared = { + "rule_id": "AZ-TEST-005", + "rule_name": "Test", + "severity": "HIGH", + "category": "Test", + "resource_id": "/subscriptions/x/resource/1", + "resource_name": "r1", + "resource_type": "Microsoft.Test/resources", + "description": "d", + "remediation": "r", + "playbook": "playbooks/cli/fix_az_test_005.sh", + "frameworks": {}, + "metadata": {}, + } + eng.rules = [ + SimpleNamespace( + RULE_ID="AZ-TEST-005", + scan=lambda *_: [dict(shared)], + evaluate=lambda *_: [ + RuleEvaluation( + rule_id="AZ-TEST-005", + resource_id=shared["resource_id"], + resource_type=shared["resource_type"], + status=EvaluationStatus.FAIL, + finding=dict(shared), + ) + ], + ) + ] + + result = eng.run_scan() + + assert result["total_findings"] == 1 + + +# ── Persistence: rule_evaluations written in the same transaction ────────── + + +def _db() -> DatabaseManager: + db = DatabaseManager.__new__(DatabaseManager) + db.dsn = "postgresql://mock/mock" + db.conn = None + return db + + +def _cursor(): + cur = MagicMock() + cur.__enter__ = lambda s: s + cur.__exit__ = MagicMock(return_value=False) + # Every INSERT ... RETURNING id call returns an incrementing fake id. + ids = iter(range(1, 10_000)) + cur.fetchone.side_effect = lambda: (next(ids),) + return cur + + +def test_save_scan_persists_evaluations_and_links_fail_finding_id(): + db = _db() + cursor = _cursor() + conn = MagicMock() + conn.cursor.return_value = cursor + + finding = { + "rule_id": "AZ-TEST-006", + "rule_name": "Test", + "severity": "HIGH", + "category": "Test", + "resource_id": "/subscriptions/x/resource/1", + "resource_name": "r1", + "resource_type": "Microsoft.Test/resources", + "description": "d", + "remediation": "r", + "playbook": "playbooks/cli/fix_az_test_006.sh", + "frameworks": {}, + "metadata": {}, + "detected_at": "2026-08-29T00:00:00+00:00", + } + result = { + "scan_id": "00000000-0000-0000-0000-000000000000", + "subscription_id": _SUB, + "started_at": "2026-08-29T00:00:00+00:00", + "findings": [finding], + "evaluations": [ + { + "rule_id": "AZ-TEST-006", + "resource_id": finding["resource_id"], + "resource_type": finding["resource_type"], + "status": "FAIL", + "reason_code": None, + "reason": None, + "evidence": {}, + }, + { + "rule_id": "AZ-TEST-007", + "resource_id": subscription_scope_id(_SUB), + "resource_type": "", + "status": "UNKNOWN", + "reason_code": "LEGACY_RULE_NOT_MIGRATED", + "reason": "not migrated", + "evidence": {}, + }, + ], + } + + with patch.object(db, "_get_conn", return_value=conn): + db.save_scan(result) + + insert_calls = [c for c in cursor.execute.call_args_list if "INSERT INTO rule_evaluations" in c.args[0]] + assert len(insert_calls) == 2 + + fail_call_params = insert_calls[0].args[1] + # (scan_id, rule_id, resource_id, resource_type, status, reason_code, reason, evidence, finding_id, evaluated_at) + assert fail_call_params[1] == "AZ-TEST-006" + assert fail_call_params[4] == "FAIL" + assert fail_call_params[8] == 1 # linked to the finding's returned id + + unknown_call_params = insert_calls[1].args[1] + assert unknown_call_params[4] == "UNKNOWN" + assert unknown_call_params[8] is None # never linked to a finding + + +def test_save_scan_deletes_prior_evaluations_before_reinsert(): + """A worker retry must replace prior rule_evaluations atomically, exactly + like it already does for findings.""" + db = _db() + cursor = _cursor() + conn = MagicMock() + conn.cursor.return_value = cursor + + result = { + "scan_id": "00000000-0000-0000-0000-000000000000", + "subscription_id": _SUB, + "started_at": "2026-08-29T00:00:00+00:00", + "findings": [], + "evaluations": [], + } + + with patch.object(db, "_get_conn", return_value=conn): + db.save_scan(result) + + delete_sql = [c.args[0] for c in cursor.execute.call_args_list if c.args[0].strip().startswith("DELETE")] + assert any("rule_evaluations" in sql for sql in delete_sql) + + +def test_save_scan_evaluations_default_to_empty_list_for_backward_compatible_callers(): + """A caller that doesn't pass 'evaluations' (pre-#263 code paths, existing + tests) must not crash save_scan.""" + db = _db() + cursor = _cursor() + conn = MagicMock() + conn.cursor.return_value = cursor + + result = { + "scan_id": "00000000-0000-0000-0000-000000000000", + "subscription_id": _SUB, + "started_at": "2026-08-29T00:00:00+00:00", + "findings": [], + } + + with patch.object(db, "_get_conn", return_value=conn): + db.save_scan(result) # must not raise diff --git a/tests/test_rules_keyvault.py b/tests/test_rules_keyvault.py index 67704dfa..4dad316d 100644 --- a/tests/test_rules_keyvault.py +++ b/tests/test_rules_keyvault.py @@ -12,6 +12,7 @@ import scanner.rules.az_kv_004 as az_kv_004 import scanner.rules.az_kv_005 as az_kv_005 import scanner.rules.az_kv_006 as az_kv_006 +from scanner.evaluation import EvaluationStatus from tests.helpers.mock_azure import make_resource _REQUIRED_FIELDS = { @@ -205,3 +206,58 @@ def test_kv_006_missing_property_defaults_to_noncompliant(mock_azure, subscripti findings = az_kv_006.scan(mock_azure, subscription_id) assert len(findings) == 1 assert findings[0]["rule_id"] == "AZ-KV-006" + + +# ── AZ-KV-006 evaluate(): coverage contract (#263) ────────────────────────── + + +def test_kv_006_evaluate_compliant_vault_is_pass(mock_azure, subscription_id): + mock_azure.set_key_vaults([_vault_with_props("kv-rbac-on", enable_rbac_authorization=True)]) + evaluations = az_kv_006.evaluate(mock_azure, subscription_id) + assert len(evaluations) == 1 + assert evaluations[0].status == EvaluationStatus.PASS + assert evaluations[0].resource_id == _kv_id("kv-rbac-on") + assert evaluations[0].finding is None + + +def test_kv_006_evaluate_noncompliant_vault_is_fail_with_embedded_finding(mock_azure, subscription_id): + mock_azure.set_key_vaults([_vault_with_props("kv-rbac-off", enable_rbac_authorization=False)]) + evaluations = az_kv_006.evaluate(mock_azure, subscription_id) + assert len(evaluations) == 1 + assert evaluations[0].status == EvaluationStatus.FAIL + assert evaluations[0].finding["rule_id"] == "AZ-KV-006" + assert evaluations[0].finding["resource_id"] == _kv_id("kv-rbac-off") + + +def test_kv_006_evaluate_missing_properties_is_unknown_not_pass(mock_azure, subscription_id): + """A vault returned without a properties payload must not be silently + skipped (as scan() does) — evaluate() must state it couldn't be checked.""" + vault = make_resource(id=_kv_id("kv-no-props"), name="kv-no-props", location="eastus", properties=None) + mock_azure.set_key_vaults([vault]) + evaluations = az_kv_006.evaluate(mock_azure, subscription_id) + assert len(evaluations) == 1 + assert evaluations[0].status == EvaluationStatus.UNKNOWN + assert evaluations[0].reason_code == "MISSING_PROPERTIES" + + +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.""" + 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].resource_id == f"/subscriptions/{subscription_id}" + + +def test_kv_006_evaluate_reports_one_status_per_vault(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), + ] + ) + evaluations = az_kv_006.evaluate(mock_azure, subscription_id) + statuses = {e.resource_id: e.status for e in evaluations} + assert statuses[_kv_id("kv-a")] == EvaluationStatus.PASS + assert statuses[_kv_id("kv-b")] == EvaluationStatus.FAIL diff --git a/tests/test_severity_contract.py b/tests/test_severity_contract.py index 097ce33b..b127eee9 100644 --- a/tests/test_severity_contract.py +++ b/tests/test_severity_contract.py @@ -176,7 +176,9 @@ def test_persistence_canonicalizes_alias_and_records_contract_version(): scan_parameters = cursor.execute.call_args_list[0].args[1] delete_parameters = cursor.execute.call_args_list[1].args[1] - finding_parameters = cursor.execute.call_args_list[2].args[1] + # index 2 is the DELETE FROM rule_evaluations that accompanies the + # findings replacement (see DatabaseManager.save_scan). + finding_parameters = cursor.execute.call_args_list[3].args[1] assert scan_parameters[4] == 1 assert scan_parameters[5] == 100 assert scan_parameters[10] == CONTRACT_VERSION From 4754c1a8f08f765e33f2dff63d2b8ecca422bc2f Mon Sep 17 00:00:00 2001 From: Dipesh Ray Date: Sat, 29 Aug 2026 02:02:16 +0100 Subject: [PATCH 2/2] fix: satisfy ruff format --check - Join the two-line CHECK constraint SQL string in the rule_evaluations migration onto one line, under the 120-char limit. - Add the blank line ruff format wants before a top-level def in the evaluate() code sample, and drop an em dash from the surrounding text. Signed-off-by: Dipesh Ray --- alembic/versions/3f59f83a5253_rule_evaluations.py | 3 +-- docs/adding-a-rule.md | 3 ++- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/alembic/versions/3f59f83a5253_rule_evaluations.py b/alembic/versions/3f59f83a5253_rule_evaluations.py index 28dce658..a68d81a2 100644 --- a/alembic/versions/3f59f83a5253_rule_evaluations.py +++ b/alembic/versions/3f59f83a5253_rule_evaluations.py @@ -70,8 +70,7 @@ def upgrade() -> None: op.create_check_constraint( _REASON_CONSTRAINT, "rule_evaluations", - "status NOT IN ('UNKNOWN', 'ERROR', 'NOT_APPLICABLE') " - "OR (reason_code IS NOT NULL AND reason_code <> '')", + "status NOT IN ('UNKNOWN', 'ERROR', 'NOT_APPLICABLE') OR (reason_code IS NOT NULL AND reason_code <> '')", ) diff --git a/docs/adding-a-rule.md b/docs/adding-a-rule.md index 4bf0eed4..5da5d29b 100644 --- a/docs/adding-a-rule.md +++ b/docs/adding-a-rule.md @@ -156,11 +156,12 @@ When a helper returns `None`, skip the resource and log a warning. Never create ```python from scanner.evaluation import EvaluationStatus, RuleEvaluation, subscription_scope_id + def evaluate(azure_client: Any, subscription_id: str) -> List[RuleEvaluation]: """Report a status for every resource this rule looked at, PASS included.""" ``` -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. +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. Rules of the contract (see `scanner/evaluation.py` and `scanner/rules/az_kv_006.py` for the reference implementation):