feat(engine): add rule evaluation coverage contract (#263)

alembic/versions/3f59f83a5253_rule_evaluations.py

@@ -0,0 +1,84 @@
+"""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")

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,
         }

docs/adding-a-rule.md

@@ -149,6 +149,31 @@ 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/`:

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),
+                )
+            ]