b0b51f1e71
- fail closed on MCP input and compile database policy - add shared scan catalogs, finding queries, and tiered gates
252 lines
8.9 KiB
Python
252 lines
8.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Compare normalized lint diagnostics with a reviewed accepted-risk baseline."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
SCHEMA_VERSION = 1
|
|
DIAGNOSTIC_RE = re.compile(
|
|
r"^(.+):(\d+):(\d+):\s+(warning|error):\s+(.+?)\s+\[([^\]]+)\]\s*$"
|
|
)
|
|
ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")
|
|
|
|
|
|
def normalize_path(value: str, root: Path, path_base: Path) -> str:
|
|
path = Path(value)
|
|
resolved = path.resolve() if path.is_absolute() else (path_base / path).resolve()
|
|
try:
|
|
return resolved.relative_to(root.resolve()).as_posix()
|
|
except ValueError:
|
|
return resolved.as_posix()
|
|
|
|
|
|
def parse_location_number(value: str) -> int:
|
|
try:
|
|
return int(value)
|
|
except ValueError as error:
|
|
raise ValueError(f"invalid diagnostic location: {value!r}") from error
|
|
|
|
|
|
def parse_diagnostics(
|
|
text: str, root: Path, path_base: Path | None = None
|
|
) -> list[dict[str, Any]]:
|
|
diagnostics: dict[tuple[str, int, int, str, str, str], dict[str, Any]] = {}
|
|
for raw_line in text.splitlines():
|
|
line = ANSI_RE.sub("", raw_line)
|
|
match = DIAGNOSTIC_RE.match(line)
|
|
if match is None:
|
|
continue
|
|
path, line_number, column, severity, message, rule = match.groups()
|
|
normalized = normalize_path(path, root, path_base or root)
|
|
if Path(normalized).is_absolute():
|
|
continue
|
|
record = {
|
|
"path": normalized,
|
|
"line": parse_location_number(line_number),
|
|
"column": parse_location_number(column),
|
|
"severity": severity,
|
|
"rule": rule,
|
|
"message": message,
|
|
}
|
|
identity = (
|
|
record["path"],
|
|
record["line"],
|
|
record["column"],
|
|
severity,
|
|
rule,
|
|
message,
|
|
)
|
|
diagnostics[identity] = record
|
|
return sorted(
|
|
diagnostics.values(),
|
|
key=lambda item: (
|
|
item["path"],
|
|
item["rule"],
|
|
item["message"],
|
|
item["line"],
|
|
item["column"],
|
|
),
|
|
)
|
|
|
|
|
|
def diagnostic_identity(record: dict[str, Any]) -> tuple[str, str, str, str]:
|
|
return (
|
|
str(record.get("path", "")),
|
|
str(record.get("severity", "")),
|
|
str(record.get("rule", "")),
|
|
str(record.get("message", "")),
|
|
)
|
|
|
|
|
|
def validate_baseline(document: Any, tool: str) -> list[str]:
|
|
errors: list[str] = []
|
|
if not isinstance(document, dict):
|
|
return ["baseline must be a JSON object"]
|
|
if document.get("schema_version") != SCHEMA_VERSION:
|
|
errors.append(f"unsupported schema_version: {document.get('schema_version')}")
|
|
if document.get("tool") != tool:
|
|
errors.append(f"baseline tool must be {tool!r}")
|
|
for field in ("reviewed_at", "reviewer"):
|
|
if not isinstance(document.get(field), str) or not document[field].strip():
|
|
errors.append(f"{field} must be a non-empty string")
|
|
rationales = document.get("rationales", {})
|
|
if not isinstance(rationales, dict) or not all(
|
|
isinstance(key, str)
|
|
and key.strip()
|
|
and isinstance(value, str)
|
|
and value.strip()
|
|
for key, value in rationales.items()
|
|
):
|
|
errors.append("rationales must map non-empty names to non-empty text")
|
|
rationales = {}
|
|
findings = document.get("findings")
|
|
if not isinstance(findings, list):
|
|
errors.append("findings must be an array")
|
|
return errors
|
|
seen: set[tuple[str, str, str, str]] = set()
|
|
for index, finding in enumerate(findings):
|
|
if not isinstance(finding, dict):
|
|
errors.append(f"findings[{index}] must be an object")
|
|
continue
|
|
for field in ("path", "severity", "rule", "message"):
|
|
if not isinstance(finding.get(field), str) or not finding[field].strip():
|
|
errors.append(f"findings[{index}].{field} must be a non-empty string")
|
|
rationale = finding.get("rationale")
|
|
rationale_ref = finding.get("rationale_ref")
|
|
if not (
|
|
isinstance(rationale, str)
|
|
and rationale.strip()
|
|
or isinstance(rationale_ref, str)
|
|
and rationale_ref in rationales
|
|
):
|
|
errors.append(
|
|
f"findings[{index}] requires a non-empty rationale or valid rationale_ref"
|
|
)
|
|
if finding.get("status") != "accepted-risk":
|
|
errors.append(f"findings[{index}].status must be 'accepted-risk'")
|
|
identity = diagnostic_identity(finding)
|
|
if identity in seen:
|
|
errors.append(f"duplicate baseline diagnostic at findings[{index}]")
|
|
seen.add(identity)
|
|
return errors
|
|
|
|
|
|
def compare_diagnostics(
|
|
current: list[dict[str, Any]], baseline: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
current_by_id: dict[tuple[str, str, str, str], list[dict[str, Any]]] = {}
|
|
for item in current:
|
|
current_by_id.setdefault(diagnostic_identity(item), []).append(item)
|
|
reviewed_by_id: dict[tuple[str, str, str, str], list[dict[str, Any]]] = {}
|
|
for item in baseline.get("findings", []):
|
|
reviewed_by_id.setdefault(diagnostic_identity(item), []).append(item)
|
|
|
|
new: list[dict[str, Any]] = []
|
|
persisting: list[dict[str, Any]] = []
|
|
resolved: list[dict[str, Any]] = []
|
|
for identity in sorted(set(current_by_id) | set(reviewed_by_id)):
|
|
current_items = current_by_id.get(identity, [])
|
|
reviewed_items = reviewed_by_id.get(identity, [])
|
|
matched = min(len(current_items), len(reviewed_items))
|
|
persisting.extend(current_items[:matched])
|
|
new.extend(current_items[matched:])
|
|
resolved.extend(reviewed_items[matched:])
|
|
return {
|
|
"schema_version": 1,
|
|
"tool": baseline["tool"],
|
|
"status": "failed" if new else "ok",
|
|
"counts": {
|
|
"current": len(current),
|
|
"reviewed": sum(len(items) for items in reviewed_by_id.values()),
|
|
"new": len(new),
|
|
"persisting": len(persisting),
|
|
"resolved": len(resolved),
|
|
},
|
|
"new": new,
|
|
"persisting": persisting,
|
|
"resolved": resolved,
|
|
"baseline": {
|
|
"reviewed_at": baseline["reviewed_at"],
|
|
"reviewer": baseline["reviewer"],
|
|
},
|
|
}
|
|
|
|
|
|
def read_exit_status(path: Path) -> int:
|
|
try:
|
|
value = path.read_text(encoding="utf-8").strip()
|
|
return int(value)
|
|
except OSError as error:
|
|
raise ValueError(
|
|
f"failed to read analyzer exit status {path}: {error}"
|
|
) from error
|
|
except ValueError as error:
|
|
raise ValueError(f"invalid analyzer exit status in {path}") from error
|
|
|
|
|
|
def load_json(path: Path) -> Any:
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except OSError as error:
|
|
raise ValueError(f"failed to read {path}: {error}") from error
|
|
except json.JSONDecodeError as error:
|
|
raise ValueError(f"invalid JSON in {path}: {error}") from error
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--tool", choices=("clang-tidy", "shellcheck"), required=True)
|
|
parser.add_argument("--baseline", type=Path, required=True)
|
|
parser.add_argument("--input", type=Path, required=True)
|
|
parser.add_argument("--analyzer-exit-file", type=Path)
|
|
parser.add_argument("--root", type=Path, default=Path.cwd())
|
|
parser.add_argument(
|
|
"--path-base",
|
|
type=Path,
|
|
help="directory used to resolve relative diagnostic paths (defaults to --root)",
|
|
)
|
|
parser.add_argument("--output", type=Path)
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
baseline = load_json(args.baseline)
|
|
errors = validate_baseline(baseline, args.tool)
|
|
if errors:
|
|
raise ValueError("; ".join(errors))
|
|
text = args.input.read_text(encoding="utf-8", errors="replace")
|
|
current = parse_diagnostics(text, args.root, args.path_base)
|
|
summary = compare_diagnostics(current, baseline)
|
|
analyzer_exit = (
|
|
read_exit_status(args.analyzer_exit_file)
|
|
if args.analyzer_exit_file is not None
|
|
else 0
|
|
)
|
|
accepted_exits = {0, 1} if args.tool == "shellcheck" else {0}
|
|
summary["analyzer_exit"] = analyzer_exit
|
|
if analyzer_exit not in accepted_exits:
|
|
summary["status"] = "failed"
|
|
summary["harness_error"] = (
|
|
f"{args.tool} exited {analyzer_exit}; inspect the complete log"
|
|
)
|
|
except (OSError, ValueError) as error:
|
|
print(f"lint baseline error: {error}", file=sys.stderr)
|
|
return 2
|
|
|
|
rendered = json.dumps(summary, indent=2, sort_keys=True) + "\n"
|
|
if args.output is not None:
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(rendered, encoding="utf-8")
|
|
print(rendered, end="")
|
|
if "harness_error" in summary:
|
|
return 2
|
|
return 1 if summary["status"] == "failed" else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|