rustfs/scripts/check_security_coverage.py

#!/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())