ci(coverage): add security ratchet calibration

scripts/check_security_coverage.py

@@ -0,0 +1,174 @@
+#!/usr/bin/env python3
+# Copyright 2024 RustFS Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Compare security-critical crate line coverage with the report-only baseline."""
+
+import argparse
+import json
+import os
+import sys
+import tempfile
+import tomllib
+from pathlib import Path
+
+from coverage_per_crate import fmt_pct, load_coverage
+
+
+def load_baselines(path: str) -> tuple[float, dict[str, tuple[int, int]]]:
+    with open(path, "rb") as fh:
+        config = tomllib.load(fh)
+
+    if config.get("phase") != "report-only":
+        raise ValueError("coverage baseline phase must be report-only")
+
+    allowed_drop = float(config["allowed_drop_percentage_points"])
+    if allowed_drop < 0:
+        raise ValueError("allowed_drop_percentage_points must be non-negative")
+
+    baselines: dict[str, tuple[int, int]] = {}
+    for crate, values in config["crates"].items():
+        covered = int(values["covered"])
+        count = int(values["count"])
+        if covered < 0 or count <= 0 or covered > count:
+            raise ValueError(f"invalid baseline for {crate}: {covered}/{count}")
+        baselines[crate] = (covered, count)
+    if not baselines:
+        raise ValueError("coverage baseline has no crates")
+    return allowed_drop, baselines
+
+
+def compare(
+    current: dict[str, list[int]],
+    baselines: dict[str, tuple[int, int]],
+    allowed_drop: float,
+) -> list[tuple[str, int, int, int, int, float, bool]]:
+    rows = []
+    for crate, (baseline_covered, baseline_count) in baselines.items():
+        if crate not in current:
+            raise ValueError(f"coverage report is missing {crate}")
+        covered, count = current[crate]
+        if covered < 0 or count <= 0 or covered > count:
+            raise ValueError(f"invalid coverage for {crate}: {covered}/{count}")
+        current_pct = 100.0 * covered / count
+        baseline_pct = 100.0 * baseline_covered / baseline_count
+        delta = current_pct - baseline_pct
+        rows.append((crate, covered, count, baseline_covered, baseline_count, delta, delta < -allowed_drop))
+    return rows
+
+
+def print_report(rows: list[tuple[str, int, int, int, int, float, bool]], allowed_drop: float) -> None:
+    print("## Security-critical coverage ratchet (report-only)")
+    print()
+    print(f"Calibration threshold: a drop greater than {allowed_drop:.2f} percentage points is reported as a regression.")
+    print()
+    print("| Crate | Current | Baseline | Delta | Status |")
+    print("|---|---:|---:|---:|---|")
+    for crate, covered, count, baseline_covered, baseline_count, delta, regressed in rows:
+        status = "REGRESSION (report-only)" if regressed else "OK"
+        print(
+            f"| `{crate}` | {fmt_pct(covered, count)} ({covered}/{count}) "
+            f"| {fmt_pct(baseline_covered, baseline_count)} ({baseline_covered}/{baseline_count}) "
+            f"| {delta:+.2f} pp | {status} |"
+        )
+    print()
+    print("This calibration phase records regressions without failing the job; malformed or incomplete evidence still fails closed.")
+
+
+def self_test() -> None:
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        coverage = root / "coverage.json"
+        baseline = root / "baseline.toml"
+        coverage.write_text(
+            json.dumps(
+                {
+                    "data": [
+                        {
+                            "files": [
+                                {
+                                    "filename": str(root / "crates/iam/src/lib.rs"),
+                                    "summary": {"lines": {"covered": 80, "count": 100}},
+                                },
+                                {
+                                    "filename": str(root / "crates/kms/src/lib.rs"),
+                                    "summary": {"lines": {"covered": 90, "count": 100}},
+                                },
+                            ],
+                            "totals": {"lines": {"covered": 170, "count": 200}},
+                        }
+                    ]
+                }
+            ),
+            encoding="utf-8",
+        )
+        baseline.write_text(
+            """phase = "report-only"
+allowed_drop_percentage_points = 1.0
+[crates."crates/iam"]
+covered = 90
+count = 100
+[crates."crates/kms"]
+covered = 85
+count = 100
+""",
+            encoding="utf-8",
+        )
+        current, _ = load_coverage(str(coverage), str(root))
+        allowed_drop, baselines = load_baselines(str(baseline))
+        rows = compare(current, baselines, allowed_drop)
+        assert [row[-1] for row in rows] == [True, False]
+        try:
+            compare({"crates/iam": current["crates/iam"]}, baselines, allowed_drop)
+        except ValueError as error:
+            assert str(error) == "coverage report is missing crates/kms"
+        else:
+            raise AssertionError("missing crate must fail closed")
+        try:
+            compare({**current, "crates/iam": [101, 100]}, baselines, allowed_drop)
+        except ValueError as error:
+            assert str(error) == "invalid coverage for crates/iam: 101/100"
+        else:
+            raise AssertionError("invalid coverage must fail closed")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("coverage_json", nargs="?")
+    parser.add_argument("--baseline", default=".config/coverage-baselines.toml")
+    parser.add_argument("--repo-root", default=os.getcwd())
+    parser.add_argument("--self-test", action="store_true")
+    args = parser.parse_args()
+
+    if args.self_test:
+        self_test()
+        print("security coverage self-test passed")
+        return 0
+    if not args.coverage_json:
+        parser.error("coverage_json is required unless --self-test is used")
+
+    try:
+        current, _ = load_coverage(args.coverage_json, os.path.abspath(args.repo_root))
+        allowed_drop, baselines = load_baselines(args.baseline)
+        rows = compare(current, baselines, allowed_drop)
+    except (OSError, ValueError, KeyError, IndexError, json.JSONDecodeError, tomllib.TOMLDecodeError) as error:
+        print(f"error: {error}", file=sys.stderr)
+        return 1
+
+    print_report(rows, allowed_drop)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())