#!/usr/bin/env python3 # file: scripts/audit_rust_general_rules.py # version: 1 """Audit mechanically verifiable Rust normalization rules used by KSP.""" from __future__ import annotations import argparse import dataclasses import pathlib import re import sys @dataclasses.dataclass(frozen=True) class Violation: """One mechanically detected general Rust rule violation.""" code: str path: str line: int message: str def rust_files(root: pathlib.Path) -> list[pathlib.Path]: """Return tracked-style Rust sources outside generated directories.""" crates = root / "crates" return sorted(path for path in crates.rglob("*.rs") if "target" not in path.parts and ".git" not in path.parts) def mask_rust_source(text: str) -> str: """Mask comments and literals while preserving braces, newlines and offsets.""" output = list(text) index = 0 state = "code" block_depth = 0 raw_hashes = 0 while index < len(text): char = text[index] nxt = text[index + 1] if index + 1 < len(text) else "" if state == "code": if char == "/" and nxt == "/": output[index] = " " output[index + 1] = " " index += 2 state = "line_comment" continue if char == "/" and nxt == "*": output[index] = " " output[index + 1] = " " index += 2 state = "block_comment" block_depth = 1 continue if char == '"': output[index] = " " index += 1 state = "string" continue if char == "'": # Treat a quote as a char literal only when it can close shortly; # lifetimes such as `'a` remain code. probe = index + 1 escaped = False found = False while probe < min(len(text), index + 8) and text[probe] != "\n": if not escaped and text[probe] == "'": found = True break if not escaped and text[probe] == "\\": escaped = True else: escaped = False probe += 1 if found: output[index] = " " index += 1 state = "char" continue if char == "r": probe = index + 1 hashes = 0 while probe < len(text) and text[probe] == "#": hashes += 1 probe += 1 if probe < len(text) and text[probe] == '"': for masked in range(index, probe + 1): output[masked] = " " index = probe + 1 raw_hashes = hashes state = "raw_string" continue index += 1 continue if state == "line_comment": if char == "\n": state = "code" else: output[index] = " " index += 1 continue if state == "block_comment": if char == "/" and nxt == "*": output[index] = " " output[index + 1] = " " block_depth += 1 index += 2 continue if char == "*" and nxt == "/": output[index] = " " output[index + 1] = " " block_depth -= 1 index += 2 if block_depth == 0: state = "code" continue if char != "\n": output[index] = " " index += 1 continue if state == "string": if char == "\\" and index + 1 < len(text): output[index] = " " if text[index + 1] != "\n": output[index + 1] = " " index += 2 continue if char == '"': output[index] = " " state = "code" elif char != "\n": output[index] = " " index += 1 continue if state == "char": if char == "\\" and index + 1 < len(text): output[index] = " " if text[index + 1] != "\n": output[index + 1] = " " index += 2 continue if char == "'": output[index] = " " state = "code" elif char != "\n": output[index] = " " index += 1 continue if state == "raw_string": if char == '"' and text.startswith("#" * raw_hashes, index + 1): output[index] = " " for masked in range(index + 1, index + 1 + raw_hashes): output[masked] = " " index += 1 + raw_hashes state = "code" continue if char != "\n": output[index] = " " index += 1 continue return "".join(output) def line_depths(masked: str) -> list[int]: """Return brace depth at the start of every line.""" depths: list[int] = [] depth = 0 for line in masked.splitlines(): depths.append(depth) depth += line.count("{") - line.count("}") if depth < 0: depth = 0 return depths def preceding_doc_line(lines: list[str], index: int) -> bool: """Return whether one declaration has adjacent useful line rustdoc.""" cursor = index - 2 while cursor >= 0 and lines[cursor].strip().startswith("#["): cursor -= 1 return cursor >= 0 and lines[cursor].lstrip().startswith("///") def export_symbol(line: str) -> str: """Return the re-exported symbol name from one ungrouped re-export.""" path = line.split(" use ", 1)[1].split(";", 1)[0].strip() return path.rsplit("::", 1)[-1] def natural_key(value: str) -> tuple[tuple[int, object], ...]: """Return a natural case-sensitive ordering key.""" tokens = re.findall(r"[A-Za-z_]+|[0-9]+", value) return tuple((1, int(token)) if token.isdigit() else (0, token) for token in tokens) def public_declaration(stripped: str) -> re.Match[str] | None: """Match a public or crate-public item or associated item declaration.""" return re.match( r"^pub(?:\(crate\))?\s+(?:(?:async|const|unsafe)\s+)*(?:const|static|type|struct|enum|trait|union|fn)\s+([A-Za-z_][A-Za-z0-9_]*)", stripped, ) def public_field(stripped: str) -> re.Match[str] | None: """Match a public or crate-public named field declaration.""" return re.match(r"^pub(?:\(crate\))?\s+([A-Za-z_][A-Za-z0-9_]*)\s*:", stripped) def declaration_start(stripped: str, kind: str) -> bool: """Return whether a line starts a declaration of one requested kind.""" prefixes = r"(?:(?:pub|pub\(crate\))\s+)?(?:(?:async|const|unsafe)\s+)*" return re.match(rf"^{prefixes}{kind}\b", stripped) is not None def audit_blank_lines_in_bodies(relative: str, lines: list[str], masked_lines: list[str], depths: list[int]) -> list[Violation]: """Reject blank lines inside functions/methods and struct/enum bodies.""" violations: list[Violation] = [] pending_kind: str | None = None pending_depth = 0 body_kind: str | None = None body_depth = 0 for idx, (line, masked_line) in enumerate(zip(lines, masked_lines, strict=True), 1): stripped = line.strip() if body_kind is not None: if not stripped and depths[idx - 1] >= body_depth: violations.append(Violation("RUST-FMT-101", relative, idx, f"blank line inside {body_kind} body")) if depths[idx - 1] < body_depth: body_kind = None if body_kind is None and pending_kind is None: if declaration_start(stripped, "fn"): pending_kind = "function/method" pending_depth = depths[idx - 1] elif declaration_start(stripped, "struct"): pending_kind = "struct" pending_depth = depths[idx - 1] elif declaration_start(stripped, "enum"): pending_kind = "enum" pending_depth = depths[idx - 1] if pending_kind is not None and "{" in masked_line: body_kind = pending_kind body_depth = pending_depth + 1 pending_kind = None elif pending_kind is not None and ";" in masked_line: pending_kind = None return violations def audit_top_level_const_blocks(relative: str, lines: list[str], depths: list[int]) -> list[Violation]: """Audit visibility/order/spacing for mechanically homogeneous const blocks.""" violations: list[Violation] = [] const_pattern = re.compile(r"^(pub\s+|pub\(crate\)\s+)?const\s+([A-Za-z_][A-Za-z0-9_]*)") previous: tuple[int, int, str] | None = None # rank: public, crate-public, private for idx, line in enumerate(lines, 1): if depths[idx - 1] != 0: continue match = const_pattern.match(line.strip()) if match is None: # Docs and attributes belong to the surrounding block and do not break it. if line.strip() and not line.lstrip().startswith(("///", "#[")): previous = None continue visibility = match.group(1) or "" rank = 0 if visibility == "pub " else 1 if visibility == "pub(crate) " else 2 name = match.group(2) if previous is not None: previous_line, previous_rank, previous_name = previous between = lines[previous_line:idx - 1] if any(not candidate.strip() for candidate in between): violations.append(Violation("RUST-FMT-102", relative, idx, "blank line inside one top-level const block")) if rank < previous_rank: violations.append(Violation("RUST-FMT-103", relative, idx, "const visibility order must be pub, pub(crate), then private")) if rank == previous_rank and natural_key(name) < natural_key(previous_name): violations.append(Violation("RUST-FMT-104", relative, idx, "const block is not alphabetically ordered")) previous = (idx, rank, name) return violations def audit_module_order(relative: str, lines: list[str], depths: list[int]) -> list[Violation]: """Audit top-level module declaration ordering with test modules last.""" violations: list[Violation] = [] modules: list[tuple[int, bool, str]] = [] for idx, line in enumerate(lines, 1): if depths[idx - 1] != 0: continue match = re.match(r"^mod\s+([A-Za-z_][A-Za-z0-9_]*)\s*(?:;|\{)", line.strip()) if match is None: continue cursor = idx - 2 is_test = False while cursor >= 0 and lines[cursor].strip().startswith("#["): if "cfg(test)" in lines[cursor].replace(" ", ""): is_test = True cursor -= 1 modules.append((idx, is_test, match.group(1))) production = [(idx, name) for idx, is_test, name in modules if not is_test] tests = [(idx, name) for idx, is_test, name in modules if is_test] if tests and production and min(idx for idx, _ in tests) < max(idx for idx, _ in production): first_test = min(idx for idx, _ in tests) violations.append(Violation("RUST-FMT-108", relative, first_test, "test modules must follow production modules")) for group, label in ((production, "production"), (tests, "test")): names = [name for _, name in group] if names != sorted(names, key=natural_key): for idx, name in group: expected = sorted(names, key=natural_key) if names.index(name) != expected.index(name): violations.append(Violation("RUST-FMT-109", relative, idx, f"{label} module declarations are not alphabetically ordered")) break return violations def audit_file(root: pathlib.Path, path: pathlib.Path) -> list[Violation]: """Audit one Rust source against general normalization rules.""" relative = path.relative_to(root).as_posix() text = path.read_text(encoding="utf-8") lines = text.splitlines() masked = mask_rust_source(text) masked_lines = masked.splitlines() depths = line_depths(masked) violations: list[Violation] = [] expected_header = f"// file: {relative}" if not lines or lines[0] != expected_header: violations.append(Violation("RUST-BASE-101", relative, 1, f"expected `{expected_header}`")) if len(lines) < 2 or re.fullmatch(r"// version: [1-9][0-9]*", lines[1]) is None: violations.append(Violation("RUST-BASE-102", relative, 2, "missing positive file version")) if not text.endswith("\n") or text.endswith("\n\n"): violations.append(Violation("RUST-FMT-100", relative, max(len(lines), 1), "file must end with exactly one newline")) local_modules = {match.group(1) for line in lines if (match := re.fullmatch(r"\s*mod\s+([A-Za-z_][A-Za-z0-9_]*)\s*;", line)) is not None} seen_declaration_before_use = False use_rows: list[tuple[int, str]] = [] for idx, line in enumerate(lines, 1): stripped = line.strip() depth = depths[idx - 1] if idx - 1 < len(depths) else 0 if re.match(r"^(?:pub\s+)?extern\s+crate\s+[A-Za-z_][A-Za-z0-9_]*\s+as\s+", stripped): violations.append(Violation("RUST-IMPORT-104", relative, idx, "extern crate aliases are forbidden")) if "pub(in " in stripped or "pub(super)" in stripped: violations.append(Violation("RUST-API-102", relative, idx, "restricted visibility is forbidden; use private or pub(crate) with crate-root re-export")) if stripped.startswith("pub mod "): violations.append(Violation("RUST-API-101", relative, idx, "modules must remain private and APIs use explicit re-exports")) if depth == 0 and re.match(r"^(?:pub(?:\(crate\))?\s+)?(?:const|static|type|struct|enum|trait|union|fn|impl)\b", stripped): seen_declaration_before_use = True use_match = re.match(r"^(pub(?:\(crate\))?\s+)?use\s+", stripped) if use_match is not None: is_export = stripped.startswith("pub use ") or stripped.startswith("pub(crate) use ") if depth != 0: violations.append(Violation("RUST-IMPORT-111", relative, idx, "use declarations must be at module scope, never inside a function/method/block")) if not is_export: use_rows.append((idx, stripped)) if seen_declaration_before_use: violations.append(Violation("RUST-IMPORT-112", relative, idx, "trait imports must remain at the beginning of the module before declarations")) if "rust-rules: trait-import" not in stripped: violations.append(Violation("RUST-IMPORT-101", relative, idx, "ordinary use requires an explicit trait import justification")) if "::*" in stripped: violations.append(Violation("RUST-IMPORT-102", relative, idx, "glob imports are forbidden")) if "{" in stripped or "}" in stripped: violations.append(Violation("RUST-IMPORT-103", relative, idx, "grouped use/re-export declarations are forbidden")) if re.search(r"\s+as\s+[A-Za-z_][A-Za-z0-9_]*", stripped): violations.append(Violation("RUST-IMPORT-104", relative, idx, "use/re-export aliases are forbidden")) if is_export and re.match(r"^pub(?:\(crate\))?\s+use\s+crate::", stripped): violations.append(Violation("RUST-IMPORT-105", relative, idx, "internal re-export must start with self::")) declaration = public_declaration(stripped) field = public_field(stripped) if declaration is not None or field is not None: if not preceding_doc_line(lines, idx): name = (declaration or field).group(1) violations.append(Violation("RUST-DOC-101", relative, idx, f"public/crate-public `{name}` requires adjacent rustdoc")) # The ordinary module import block is contiguous and alphabetic. for (previous_idx, previous_line), (current_idx, current_line) in zip(use_rows, use_rows[1:]): between = lines[previous_idx:current_idx - 1] if any(not candidate.strip() for candidate in between): violations.append(Violation("RUST-IMPORT-113", relative, current_idx, "module use block contains an empty line")) previous_path = previous_line.split("use ", 1)[1].split(";", 1)[0] current_path = current_line.split("use ", 1)[1].split(";", 1)[0] if natural_key(current_path) < natural_key(previous_path): violations.append(Violation("RUST-IMPORT-114", relative, current_idx, "module use block is not alphabetically ordered")) # Crate façade constraints. if path.name in {"lib.rs", "main.rs"}: for attribute in ("#![warn(missing_docs)]", "#![deny(unreachable_pub)]", "#![forbid(unsafe_code)]"): if attribute not in lines[:24]: violations.append(Violation("RUST-BASE-103", relative, 1, f"missing `{attribute}`")) exports: list[tuple[int, str, str]] = [] for idx, line in enumerate(lines, 1): stripped = line.strip() if stripped.startswith("pub use ") or stripped.startswith("pub(crate) use "): visibility = "pub(crate)" if stripped.startswith("pub(crate) use ") else "pub" exports.append((idx, visibility, export_symbol(stripped))) if not preceding_doc_line(lines, idx): violations.append(Violation("RUST-DOC-102", relative, idx, "crate-root re-export requires adjacent rustdoc")) source_match = re.match(r"^pub(?:\(crate\))?\s+use\s+(?:self::)?([A-Za-z_][A-Za-z0-9_]*)::", stripped) if source_match is not None and source_match.group(1) in local_modules and not re.match(r"^pub(?:\(crate\))?\s+use\s+self::", stripped): violations.append(Violation("RUST-IMPORT-106", relative, idx, "crate-root internal re-export must start with self::")) crate_seen = False previous_by_visibility: dict[str, tuple[int, str]] = {} for idx, visibility, symbol in exports: if visibility == "pub(crate)": crate_seen = True elif crate_seen: violations.append(Violation("RUST-FMT-105", relative, idx, "pub use block must precede pub(crate) use block")) previous = previous_by_visibility.get(visibility) if previous is not None: previous_idx, previous_symbol = previous # One homogeneous block may contain rustdocs but no blank line. between = lines[previous_idx:idx - 1] if any(not candidate.strip() for candidate in between): violations.append(Violation("RUST-FMT-106", relative, idx, f"{visibility} use block contains an empty line")) if natural_key(symbol) < natural_key(previous_symbol): violations.append(Violation("RUST-FMT-107", relative, idx, f"{visibility} use block is not alphabetically ordered by exported symbol")) previous_by_visibility[visibility] = (idx, symbol) violations.extend(audit_blank_lines_in_bodies(relative, lines, masked_lines, depths)) violations.extend(audit_module_order(relative, lines, depths)) violations.extend(audit_top_level_const_blocks(relative, lines, depths)) return violations def main() -> int: """Run the general Rust normalization audit.""" parser = argparse.ArgumentParser() parser.add_argument("--root", default=".") parser.add_argument("--report-only", action="store_true") arguments = parser.parse_args() root = pathlib.Path(arguments.root).resolve() violations = [item for path in rust_files(root) for item in audit_file(root, path)] violations.sort(key=lambda item: (item.code, item.path, item.line, item.message)) if not violations: sys.stdout.write("General Rust rule audit: clean\n") return 0 sys.stdout.write(f"General Rust rule audit: {len(violations)} violation(s)\n") counts: dict[str, int] = {} for item in violations: counts[item.code] = counts.get(item.code, 0) + 1 for code in sorted(counts): sys.stdout.write(f"{code}: {counts[code]}\n") for item in violations: sys.stdout.write(f"{item.code} {item.path}:{item.line}: {item.message}\n") return 0 if arguments.report_only else 1 if __name__ == "__main__": raise SystemExit(main())