#!/usr/bin/env python3 from __future__ import annotations import csv import re import sys from dataclasses import dataclass from pathlib import Path from generate_status_summary import END_MARKER, START_MARKER, format_gaps ROOT = Path(__file__).resolve().parents[1] STATUS_MATRIX = ROOT / "artifacts" / "status-matrix.csv" GAP_ANALYSIS = ROOT / "artifacts" / "gap-analysis.md" CREDENTIAL_SCOPE = ROOT / "artifacts" / "credential-scope-assessment.md" ALLOWED_SCOPE_TYPES = {"phase", "category", "gap", "artifact", "dataset"} ALLOWED_STATUSES = {"closed", "partial", "open", "blocked"} ALLOWED_PRIORITIES = {"", "low", "medium", "high", "critical"} ALLOWED_BLOCKED_REASONS = { "", "passcode_protection", "encryption", "ethical_restriction", "argocd_masking", "dns_unresolvable", } STATUS_MAP_RU = { "ЗАКРЫТО": "closed", "ЧАСТИЧНО": "partial", "ОТКРЫТО": "open", } @dataclass(frozen=True) class MatrixRow: scope_type: str scope_id: str parent_scope: str title: str status: str priority: str blocked_reason: str source_path: str source_line: int last_reviewed: str notes: str def load_rows(path: Path) -> list[MatrixRow]: with path.open(newline="") as handle: reader = csv.DictReader(handle) expected = [ "scope_type", "scope_id", "parent_scope", "title", "status", "priority", "blocked_reason", "source_path", "source_line", "last_reviewed", "notes", ] if reader.fieldnames != expected: raise ValueError(f"unexpected CSV headers: {reader.fieldnames}") rows = [] for raw in reader: rows.append( MatrixRow( scope_type=raw["scope_type"], scope_id=raw["scope_id"], parent_scope=raw["parent_scope"], title=raw["title"], status=raw["status"], priority=raw["priority"], blocked_reason=raw["blocked_reason"], source_path=raw["source_path"], source_line=int(raw["source_line"]), last_reviewed=raw["last_reviewed"], notes=raw["notes"], ) ) return rows def narrative_status_map(path: Path) -> dict[str, str]: lines = path.read_text().splitlines() result: dict[str, str] = {} current_title: str | None = None for line in lines: line = line.strip() if line.startswith("#### "): current_title = line[5:] continue match = re.match(r"\*\*Статус:\s+([А-ЯЁ]+)\*\*", line) if match and current_title: mapped = STATUS_MAP_RU.get(match.group(1)) if mapped: result[current_title] = mapped return result def validate_rows(rows: list[MatrixRow]) -> list[str]: errors: list[str] = [] seen_ids: set[str] = set() row_ids = {row.scope_id for row in rows} for row in rows: if row.scope_type not in ALLOWED_SCOPE_TYPES: errors.append(f"{row.scope_id}: invalid scope_type={row.scope_type}") if row.status not in ALLOWED_STATUSES: errors.append(f"{row.scope_id}: invalid status={row.status}") if row.priority not in ALLOWED_PRIORITIES: errors.append(f"{row.scope_id}: invalid priority={row.priority}") if row.blocked_reason not in ALLOWED_BLOCKED_REASONS: errors.append(f"{row.scope_id}: invalid blocked_reason={row.blocked_reason}") if row.scope_id in seen_ids: errors.append(f"{row.scope_id}: duplicate scope_id") seen_ids.add(row.scope_id) if row.parent_scope and row.parent_scope not in row_ids: errors.append(f"{row.scope_id}: unknown parent_scope={row.parent_scope}") source = ROOT / row.source_path if not source.exists(): errors.append(f"{row.scope_id}: missing source_path={row.source_path}") continue line_count = source.read_text(errors="ignore").count("\n") + 1 if row.source_line < 1 or row.source_line > line_count: errors.append( f"{row.scope_id}: source_line={row.source_line} out of range for {row.source_path} ({line_count} lines)" ) if row.status == "blocked" and not row.blocked_reason: errors.append(f"{row.scope_id}: blocked status requires blocked_reason") if row.status != "blocked" and row.blocked_reason: errors.append(f"{row.scope_id}: blocked_reason should be empty when status={row.status}") return errors def validate_narrative_sync(rows: list[MatrixRow]) -> list[str]: errors: list[str] = [] gap_statuses = narrative_status_map(GAP_ANALYSIS) phase_titles = { "PH1": "Phase 1 — Log Intake & Classification", "PH2": "Phase 2 — Credential Scope Assessment", "PH3": "Phase 3 — Session Exposure Analysis", "PH4": "Phase 4 — Digital Asset Impact", "PH5": "Phase 5 — Remediation Priority Matrix", } category_titles = { "EC1": "EC1 — Credential Exposure (T1078)", "EC2": "EC2 — Session Token Exposure (T1539)", "EC3": "EC3 — Financial Asset Exposure", "EC4": "EC4 — Identity Cascade Risk", "EC5": "EC5 — Infrastructure Exposure (T1528)", } by_id = {row.scope_id: row for row in rows} for scope_id, title in phase_titles.items(): expected = by_id[scope_id].status actual = gap_statuses.get(title) if actual != expected: errors.append(f"{scope_id}: gap-analysis status mismatch for '{title}': expected={expected}, actual={actual}") for scope_id, title in category_titles.items(): expected = by_id[scope_id].status actual = gap_statuses.get(title) if actual != expected: errors.append(f"{scope_id}: gap-analysis status mismatch for '{title}': expected={expected}, actual={actual}") scope_text = CREDENTIAL_SCOPE.read_text() if "status-matrix.csv" not in scope_text: errors.append("credential-scope-assessment.md: missing reference to status-matrix.csv") gap_text = GAP_ANALYSIS.read_text() if "status-matrix.csv" not in gap_text: errors.append("gap-analysis.md: missing reference to status-matrix.csv") pattern = re.compile( re.escape(START_MARKER) + r".*?" + re.escape(END_MARKER), re.DOTALL, ) match = pattern.search(gap_text) if not match: errors.append("gap-analysis.md: managed status summary block not found") else: expected_block = format_gaps( [ { "scope_type": row.scope_type, "scope_id": row.scope_id, "parent_scope": row.parent_scope, "title": row.title, "status": row.status, "priority": row.priority, "blocked_reason": row.blocked_reason, "source_path": row.source_path, "source_line": str(row.source_line), "last_reviewed": row.last_reviewed, "notes": row.notes, } for row in rows ] ) actual_block = match.group(0) if actual_block.strip() != expected_block.strip(): errors.append("gap-analysis.md: managed status summary block is out of date; run scripts/generate_status_summary.py") return errors def main() -> int: try: rows = load_rows(STATUS_MATRIX) except Exception as exc: print(f"ERROR: failed to load {STATUS_MATRIX}: {exc}") return 2 errors = [] errors.extend(validate_rows(rows)) errors.extend(validate_narrative_sync(rows)) if errors: print("status-matrix validation failed") for error in errors: print(f"- {error}") return 1 print(f"status-matrix validation OK: {len(rows)} rows checked") return 0 if __name__ == "__main__": raise SystemExit(main())