449 lines
17 KiB
Python
449 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import fnmatch
|
|
import json
|
|
import pathlib
|
|
import re
|
|
import sys
|
|
from collections.abc import Iterable, Mapping
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
REPORT_BASENAME = "lifecycle-status-guard-scan"
|
|
DEFAULT_EVIDENCE_WINDOW_CHARS = 240
|
|
|
|
|
|
def as_list(value: Any) -> list[str]:
|
|
if value is None:
|
|
return []
|
|
if isinstance(value, str):
|
|
return [value]
|
|
if isinstance(value, Iterable):
|
|
return [str(item) for item in value if item is not None]
|
|
return [str(value)]
|
|
|
|
|
|
def lower_set(values: Iterable[str]) -> set[str]:
|
|
return {value.casefold() for value in values}
|
|
|
|
|
|
def normalize_config(raw_config: Mapping[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"watched_paths": as_list(raw_config.get("watched_paths")) or ["**/*"],
|
|
"status_fields": as_list(raw_config.get("status_fields")),
|
|
"forbidden_status_values": lower_set(as_list(raw_config.get("forbidden_status_values"))),
|
|
"warning_status_values": lower_set(as_list(raw_config.get("warning_status_values"))),
|
|
"required_evidence_markers": as_list(raw_config.get("required_evidence_markers")),
|
|
"approved_phrases": as_list(raw_config.get("approved_phrases")),
|
|
"forbidden_phrases": as_list(raw_config.get("forbidden_phrases")),
|
|
"allowed_contexts": as_list(raw_config.get("allowed_contexts")),
|
|
"evidence_window_chars": int(
|
|
raw_config.get("evidence_window_chars", DEFAULT_EVIDENCE_WINDOW_CHARS)
|
|
),
|
|
}
|
|
|
|
|
|
def load_config(path: pathlib.Path) -> dict[str, Any]:
|
|
text = path.read_text(encoding="utf-8")
|
|
if path.suffix.lower() == ".json":
|
|
parsed = json.loads(text)
|
|
else:
|
|
parsed = yaml.safe_load(text)
|
|
if not isinstance(parsed, Mapping):
|
|
raise ValueError("Config must be a JSON or YAML object.")
|
|
return normalize_config(parsed)
|
|
|
|
|
|
def rel_path(path: pathlib.Path, root: pathlib.Path) -> str:
|
|
return path.relative_to(root).as_posix()
|
|
|
|
|
|
def discover_files(scan_root: pathlib.Path, watched_paths: list[str]) -> list[pathlib.Path]:
|
|
files: dict[pathlib.Path, None] = {}
|
|
for pattern in watched_paths:
|
|
for candidate in scan_root.glob(pattern):
|
|
if candidate.is_file():
|
|
files[candidate.resolve()] = None
|
|
return sorted(files)
|
|
|
|
|
|
def parse_markdown_front_matter(text: str) -> Any | None:
|
|
if not text.startswith("---"):
|
|
return None
|
|
lines = text.splitlines()
|
|
if not lines or lines[0].strip() != "---":
|
|
return None
|
|
for index in range(1, len(lines)):
|
|
if lines[index].strip() == "---":
|
|
front_matter = "\n".join(lines[1:index])
|
|
parsed = yaml.safe_load(front_matter) if front_matter.strip() else {}
|
|
return parsed if isinstance(parsed, Mapping) else None
|
|
return None
|
|
|
|
|
|
def parse_structured_content(path: pathlib.Path, text: str) -> Any | None:
|
|
suffix = path.suffix.lower()
|
|
try:
|
|
if suffix == ".json":
|
|
return json.loads(text)
|
|
if suffix in {".yaml", ".yml"}:
|
|
return yaml.safe_load(text)
|
|
if suffix in {".md", ".markdown"}:
|
|
return parse_markdown_front_matter(text)
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def scalar_values(value: Any) -> list[str]:
|
|
if isinstance(value, (str, int, float, bool)):
|
|
return [str(value)]
|
|
if isinstance(value, list):
|
|
values: list[str] = []
|
|
for item in value:
|
|
values.extend(scalar_values(item))
|
|
return values
|
|
return []
|
|
|
|
|
|
def object_contains_marker(value: Any, markers: list[str]) -> bool:
|
|
if not markers:
|
|
return False
|
|
marker_needles = [marker.casefold() for marker in markers]
|
|
|
|
def walk(current: Any) -> bool:
|
|
if isinstance(current, Mapping):
|
|
for key, nested in current.items():
|
|
key_text = str(key).casefold()
|
|
if any(marker in key_text for marker in marker_needles):
|
|
return True
|
|
if walk(nested):
|
|
return True
|
|
return False
|
|
if isinstance(current, list):
|
|
return any(walk(item) for item in current)
|
|
text = str(current).casefold()
|
|
return any(marker in text for marker in marker_needles)
|
|
|
|
return walk(value)
|
|
|
|
|
|
def line_col(text: str, index: int) -> tuple[int, int]:
|
|
line = text.count("\n", 0, index) + 1
|
|
last_newline = text.rfind("\n", 0, index)
|
|
column = index + 1 if last_newline == -1 else index - last_newline
|
|
return line, column
|
|
|
|
|
|
def make_context(text: str, index: int, length: int, window: int) -> str:
|
|
start = max(0, index - window)
|
|
end = min(len(text), index + length + window)
|
|
return re.sub(r"\s+", " ", text[start:end]).strip()
|
|
|
|
|
|
def context_has_any(context: str, values: list[str]) -> bool:
|
|
folded = context.casefold()
|
|
return any(value.casefold() in folded for value in values)
|
|
|
|
|
|
def find_phrase_matches(text: str, phrase: str) -> Iterable[int]:
|
|
if not phrase:
|
|
return []
|
|
folded_text = text.casefold()
|
|
folded_phrase = phrase.casefold()
|
|
matches: list[int] = []
|
|
start = 0
|
|
while True:
|
|
index = folded_text.find(folded_phrase, start)
|
|
if index == -1:
|
|
break
|
|
matches.append(index)
|
|
start = index + max(len(folded_phrase), 1)
|
|
return matches
|
|
|
|
|
|
def path_matches(field_path: str, key: str, status_fields: list[str]) -> bool:
|
|
if not status_fields:
|
|
return False
|
|
candidates = {field_path, key}
|
|
return any(
|
|
field in candidates or fnmatch.fnmatch(field_path, field) or fnmatch.fnmatch(key, field)
|
|
for field in status_fields
|
|
)
|
|
|
|
|
|
def child_path(parent_path: str, key: str) -> str:
|
|
return f"{parent_path}.{key}" if parent_path else key
|
|
|
|
|
|
def list_path(parent_path: str, index: int) -> str:
|
|
return f"{parent_path}[{index}]" if parent_path else f"[{index}]"
|
|
|
|
|
|
def scan_structured(
|
|
data: Any,
|
|
file_rel: str,
|
|
config: dict[str, Any],
|
|
report: dict[str, Any],
|
|
) -> None:
|
|
status_fields = config["status_fields"]
|
|
forbidden_values = config["forbidden_status_values"]
|
|
warning_values = config["warning_status_values"]
|
|
evidence_markers = config["required_evidence_markers"]
|
|
|
|
def walk(current: Any, path: str, parent: Any) -> None:
|
|
if isinstance(current, Mapping):
|
|
for key_obj, value in current.items():
|
|
key = str(key_obj)
|
|
current_path = child_path(path, key)
|
|
if path_matches(current_path, key, status_fields):
|
|
for scalar in scalar_values(value):
|
|
value_key = scalar.casefold()
|
|
if value_key in forbidden_values:
|
|
evidence_present = object_contains_marker(parent, evidence_markers)
|
|
record = {
|
|
"file": file_rel,
|
|
"source": "structured",
|
|
"path": current_path,
|
|
"field": key,
|
|
"value": scalar,
|
|
"severity": "blocking" if not evidence_present else "observation",
|
|
"evidence_present": evidence_present,
|
|
"message": (
|
|
"Forbidden lifecycle/status value lacks local evidence marker."
|
|
if not evidence_present
|
|
else "Lifecycle/status value has a local evidence marker; approval validity is not decided by this scan."
|
|
),
|
|
}
|
|
if evidence_present:
|
|
observation = dict(record)
|
|
observation["kind"] = "field_evidence_present"
|
|
report["observations"].append(observation)
|
|
else:
|
|
report["field_level_findings"].append(record)
|
|
report["blocking_errors"].append(record["message"])
|
|
report["missing_evidence_markers"].append(
|
|
{
|
|
"file": file_rel,
|
|
"path": current_path,
|
|
"required_evidence_markers": evidence_markers,
|
|
}
|
|
)
|
|
elif value_key in warning_values:
|
|
report["warnings"].append(
|
|
{
|
|
"file": file_rel,
|
|
"source": "structured",
|
|
"path": current_path,
|
|
"field": key,
|
|
"value": scalar,
|
|
"severity": "warning",
|
|
"message": "Warning lifecycle/status value matched configured policy.",
|
|
}
|
|
)
|
|
walk(value, current_path, value if isinstance(value, Mapping) else current)
|
|
elif isinstance(current, list):
|
|
for index, item in enumerate(current):
|
|
walk(item, list_path(path, index), item if isinstance(item, Mapping) else parent)
|
|
|
|
walk(data, "", data)
|
|
|
|
|
|
def scan_phrases(
|
|
text: str,
|
|
file_rel: str,
|
|
config: dict[str, Any],
|
|
report: dict[str, Any],
|
|
) -> None:
|
|
evidence_markers = config["required_evidence_markers"]
|
|
allowed_contexts = config["allowed_contexts"]
|
|
window = config["evidence_window_chars"]
|
|
|
|
for phrase in config["approved_phrases"]:
|
|
for index in find_phrase_matches(text, phrase):
|
|
context = make_context(text, index, len(phrase), window)
|
|
if context_has_any(context, allowed_contexts):
|
|
continue
|
|
evidence_present = context_has_any(context, evidence_markers)
|
|
line, column = line_col(text, index)
|
|
record = {
|
|
"file": file_rel,
|
|
"source": "text",
|
|
"phrase": phrase,
|
|
"line": line,
|
|
"column": column,
|
|
"severity": "blocking" if not evidence_present else "observation",
|
|
"evidence_present": evidence_present,
|
|
"context": context,
|
|
"message": (
|
|
"Approval claim phrase lacks nearby evidence marker."
|
|
if not evidence_present
|
|
else "Approval claim phrase has nearby evidence marker; approval validity is not decided by this scan."
|
|
),
|
|
}
|
|
if evidence_present:
|
|
observation = dict(record)
|
|
observation["kind"] = "evidence_present_claim"
|
|
report["observations"].append(observation)
|
|
else:
|
|
report["phrase_level_findings"].append(record)
|
|
report["blocking_errors"].append(record["message"])
|
|
report["missing_evidence_markers"].append(
|
|
{
|
|
"file": file_rel,
|
|
"line": line,
|
|
"phrase": phrase,
|
|
"required_evidence_markers": evidence_markers,
|
|
}
|
|
)
|
|
|
|
for phrase in config["forbidden_phrases"]:
|
|
for index in find_phrase_matches(text, phrase):
|
|
context = make_context(text, index, len(phrase), window)
|
|
if context_has_any(context, allowed_contexts):
|
|
continue
|
|
line, column = line_col(text, index)
|
|
record = {
|
|
"file": file_rel,
|
|
"source": "text",
|
|
"phrase": phrase,
|
|
"line": line,
|
|
"column": column,
|
|
"severity": "blocking",
|
|
"evidence_present": context_has_any(context, evidence_markers),
|
|
"context": context,
|
|
"message": "Forbidden phrase matched configured policy.",
|
|
}
|
|
report["phrase_level_findings"].append(record)
|
|
report["blocking_errors"].append(record["message"])
|
|
|
|
|
|
def empty_report(scan_root: pathlib.Path, config_path: pathlib.Path) -> dict[str, Any]:
|
|
return {
|
|
"scan_root": str(scan_root),
|
|
"config_path": str(config_path),
|
|
"files_scanned": [],
|
|
"possible_overclaims": [],
|
|
"field_level_findings": [],
|
|
"phrase_level_findings": [],
|
|
"missing_evidence_markers": [],
|
|
"warnings": [],
|
|
"blocking_errors": [],
|
|
"observations": [],
|
|
"machine_readable_summary": {},
|
|
}
|
|
|
|
|
|
def finalize_report(report: dict[str, Any]) -> None:
|
|
report["possible_overclaims"] = [
|
|
*report["field_level_findings"],
|
|
*report["phrase_level_findings"],
|
|
]
|
|
report["machine_readable_summary"] = {
|
|
"status": "FAIL" if report["blocking_errors"] else "PASS",
|
|
"files_scanned_count": len(report["files_scanned"]),
|
|
"field_finding_count": len(report["field_level_findings"]),
|
|
"phrase_finding_count": len(report["phrase_level_findings"]),
|
|
"missing_evidence_count": len(report["missing_evidence_markers"]),
|
|
"warning_count": len(report["warnings"]),
|
|
"blocking_count": len(report["blocking_errors"]),
|
|
"observation_count": len(report["observations"]),
|
|
}
|
|
|
|
|
|
def write_markdown_report(report: dict[str, Any], output_path: pathlib.Path) -> None:
|
|
summary = report["machine_readable_summary"]
|
|
lines = [
|
|
"# Lifecycle Status Guard Scan",
|
|
"",
|
|
"## Summary",
|
|
"",
|
|
f"- Status: `{summary['status']}`",
|
|
f"- Files scanned: {summary['files_scanned_count']}",
|
|
f"- Blocking findings: {summary['blocking_count']}",
|
|
f"- Warnings: {summary['warning_count']}",
|
|
f"- Evidence-present observations: {summary['observation_count']}",
|
|
"",
|
|
"## Blocking Findings",
|
|
"",
|
|
]
|
|
if report["possible_overclaims"]:
|
|
for finding in report["possible_overclaims"]:
|
|
location = finding.get("path") or f"line {finding.get('line')}"
|
|
claim = finding.get("value") or finding.get("phrase")
|
|
lines.append(f"- `{finding['file']}` {location}: `{claim}` - {finding['message']}")
|
|
else:
|
|
lines.append("- None")
|
|
lines.extend(["", "## Warnings", ""])
|
|
if report["warnings"]:
|
|
for warning in report["warnings"]:
|
|
lines.append(
|
|
f"- `{warning['file']}` {warning.get('path', '')}: `{warning.get('value', '')}` - {warning['message']}"
|
|
)
|
|
else:
|
|
lines.append("- None")
|
|
lines.extend(["", "## Observations", ""])
|
|
if report["observations"]:
|
|
for observation in report["observations"]:
|
|
location = observation.get("path") or f"line {observation.get('line')}"
|
|
claim = observation.get("value") or observation.get("phrase")
|
|
lines.append(
|
|
f"- `{observation['file']}` {location}: `{claim}` - evidence marker present; approval validity not decided."
|
|
)
|
|
else:
|
|
lines.append("- None")
|
|
lines.extend(["", "## Files Scanned", ""])
|
|
for file_name in report["files_scanned"]:
|
|
lines.append(f"- `{file_name}`")
|
|
output_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
|
def run(scan_root: pathlib.Path, config_path: pathlib.Path, output_dir: pathlib.Path) -> int:
|
|
scan_root = scan_root.resolve()
|
|
config_path = config_path.resolve()
|
|
if not scan_root.is_dir():
|
|
raise ValueError(f"Scan root is not a directory: {scan_root}")
|
|
if not config_path.is_file():
|
|
raise ValueError(f"Config file does not exist: {config_path}")
|
|
|
|
config = load_config(config_path)
|
|
report = empty_report(scan_root, config_path)
|
|
|
|
for path in discover_files(scan_root, config["watched_paths"]):
|
|
file_rel = rel_path(path, scan_root)
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
report["files_scanned"].append(file_rel)
|
|
structured = parse_structured_content(path, text)
|
|
if isinstance(structured, (Mapping, list)):
|
|
scan_structured(structured, file_rel, config, report)
|
|
scan_phrases(text, file_rel, config, report)
|
|
|
|
finalize_report(report)
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
json_path = output_dir / f"{REPORT_BASENAME}.json"
|
|
md_path = output_dir / f"{REPORT_BASENAME}.md"
|
|
json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
write_markdown_report(report, md_path)
|
|
return 1 if report["blocking_errors"] else 0
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Scan configured lifecycle/status claims.")
|
|
parser.add_argument("--scan-root", required=True, type=pathlib.Path)
|
|
parser.add_argument("--config", required=True, type=pathlib.Path)
|
|
parser.add_argument("--output-dir", required=True, type=pathlib.Path)
|
|
args = parser.parse_args(argv)
|
|
|
|
try:
|
|
return run(args.scan_root, args.config, args.output_dir)
|
|
except Exception as exc:
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|