0.1.0-pre.004
This commit is contained in:
184
scripts/audit_khadhroony_workspace_rules.py
Executable file
184
scripts/audit_khadhroony_workspace_rules.py
Executable file
@@ -0,0 +1,184 @@
|
||||
#!/usr/bin/env python3
|
||||
# file: scripts/audit_khadhroony_workspace_rules.py
|
||||
# version: 2
|
||||
|
||||
"""Audit mechanically verifiable rules specific to khadhroony-bot3."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import dataclasses
|
||||
import pathlib
|
||||
import re
|
||||
import sys
|
||||
import tomllib
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class Violation:
|
||||
"""One project-specific workspace rule violation."""
|
||||
|
||||
code: str
|
||||
path: str
|
||||
line: int
|
||||
message: str
|
||||
|
||||
|
||||
def package_name(crate_dir: pathlib.Path) -> str | None:
|
||||
"""Return a crate package name."""
|
||||
|
||||
cargo = crate_dir / "Cargo.toml"
|
||||
if not cargo.exists():
|
||||
return None
|
||||
value = tomllib.loads(cargo.read_text(encoding="utf-8")).get("package", {}).get("name")
|
||||
return value if isinstance(value, str) else None
|
||||
|
||||
|
||||
def relevant_tracing_crate(crate_dir: pathlib.Path) -> bool:
|
||||
"""Return whether the crate owns a tracing target contract."""
|
||||
|
||||
cargo = crate_dir / "Cargo.toml"
|
||||
data = tomllib.loads(cargo.read_text(encoding="utf-8"))
|
||||
if "tracing" in data.get("dependencies", {}):
|
||||
return True
|
||||
return any("TRACING_TARGET" in path.read_text(encoding="utf-8") for path in crate_dir.glob("src/**/*.rs"))
|
||||
|
||||
|
||||
def audit_tracing(root: pathlib.Path) -> list[Violation]:
|
||||
"""Audit the project tracing target contract."""
|
||||
|
||||
violations: list[Violation] = []
|
||||
for cargo in sorted(root.glob("*/Cargo.toml")):
|
||||
crate = cargo.parent
|
||||
if crate.name == "kb-lib":
|
||||
continue
|
||||
if not relevant_tracing_crate(crate):
|
||||
continue
|
||||
name = package_name(crate)
|
||||
if name is None:
|
||||
continue
|
||||
constants = crate / "src/constants.rs"
|
||||
expected = f'pub(crate) const TRACING_TARGET: &str = "{name}";'
|
||||
relative_constants = constants.relative_to(root).as_posix()
|
||||
if not constants.exists() or expected not in constants.read_text(encoding="utf-8"):
|
||||
violations.append(Violation("KH_TRACE001", relative_constants, 1, f"expected `{expected}`"))
|
||||
crate_root = crate / "src/lib.rs"
|
||||
if not crate_root.exists():
|
||||
crate_root = crate / "src/main.rs"
|
||||
relative_root = crate_root.relative_to(root).as_posix()
|
||||
if not crate_root.exists() or "pub(crate) use crate::constants::TRACING_TARGET;" not in crate_root.read_text(encoding="utf-8"):
|
||||
violations.append(Violation("KH_TRACE002", relative_root, 1, "missing crate-root TRACING_TARGET re-export"))
|
||||
for path in crate.glob("src/**/*.rs"):
|
||||
relative = path.relative_to(root).as_posix()
|
||||
for index, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
||||
if "khbot." in line:
|
||||
violations.append(Violation("KH_TRACE003", relative, index, "legacy khbot tracing target is forbidden"))
|
||||
if "crate::constants::TRACING_TARGET" in line and path.name not in {"lib.rs", "main.rs"}:
|
||||
violations.append(Violation("KH_TRACE004", relative, index, "use crate::TRACING_TARGET outside crate root"))
|
||||
if re.search(r"(?:crate::[A-Za-z0-9_]+::|(?<!crate::))TRACING_TARGET", line) and "const TRACING_TARGET" not in line and "use crate::constants::TRACING_TARGET" not in line:
|
||||
if "crate::TRACING_TARGET" not in line:
|
||||
violations.append(Violation("KH_TRACE005", relative, index, "non-canonical TRACING_TARGET path"))
|
||||
return violations
|
||||
|
||||
|
||||
def audit_kb_lib_tracing(root: pathlib.Path) -> list[Violation]:
|
||||
"""Audit hierarchical tracing targets owned by operational kb-lib components."""
|
||||
|
||||
violations: list[Violation] = []
|
||||
crate = root / "kb-lib"
|
||||
crate_root = crate / "src/lib.rs"
|
||||
if not crate_root.exists():
|
||||
return violations
|
||||
root_text = crate_root.read_text(encoding="utf-8")
|
||||
aliases = set(
|
||||
re.findall(
|
||||
r"^pub\(crate\) use crate::[A-Za-z0-9_:]+::([A-Z][A-Z0-9_]+_TRACING_TARGET);$",
|
||||
root_text,
|
||||
re.MULTILINE,
|
||||
)
|
||||
)
|
||||
target_pattern = re.compile(
|
||||
r'^pub\(crate\) const TRACING_TARGET: &str = "([^"]+)";$',
|
||||
re.MULTILINE,
|
||||
)
|
||||
for constants in sorted((crate / "src").rglob("constants.rs")):
|
||||
text = constants.read_text(encoding="utf-8")
|
||||
match = target_pattern.search(text)
|
||||
if match is None:
|
||||
continue
|
||||
component = constants.parent.relative_to(crate / "src")
|
||||
expected = "kb-lib." + ".".join(component.parts)
|
||||
relative = constants.relative_to(root).as_posix()
|
||||
if match.group(1) != expected:
|
||||
violations.append(
|
||||
Violation(
|
||||
"KH_TRACE101",
|
||||
relative,
|
||||
text[: match.start()].count("\n") + 1,
|
||||
f"expected hierarchical target `{expected}`",
|
||||
)
|
||||
)
|
||||
macro_pattern = re.compile(
|
||||
r"tracing::(?:debug|error|info|trace|warn)!\(\s*target:\s*crate::([A-Z][A-Z0-9_]+_TRACING_TARGET)"
|
||||
)
|
||||
for path in sorted((crate / "src").rglob("*.rs")):
|
||||
relative = path.relative_to(root).as_posix()
|
||||
text = path.read_text(encoding="utf-8")
|
||||
for index, line in enumerate(text.splitlines(), 1):
|
||||
if "let _target" in line:
|
||||
violations.append(
|
||||
Violation(
|
||||
"KH_TRACE102",
|
||||
relative,
|
||||
index,
|
||||
"fake tracing target consumption is forbidden",
|
||||
)
|
||||
)
|
||||
for match in macro_pattern.finditer(text):
|
||||
if match.group(1) not in aliases:
|
||||
violations.append(
|
||||
Violation(
|
||||
"KH_TRACE103",
|
||||
relative,
|
||||
text[: match.start()].count("\n") + 1,
|
||||
f"missing crate-root re-export for `{match.group(1)}`",
|
||||
)
|
||||
)
|
||||
return violations
|
||||
|
||||
|
||||
def audit_solana_types(root: pathlib.Path) -> list[Violation]:
|
||||
"""Reject non-canonical Solana address types in workspace Rust code."""
|
||||
|
||||
violations: list[Violation] = []
|
||||
for path in sorted(root.rglob("*.rs")):
|
||||
if "target" in path.parts or ".git" in path.parts:
|
||||
continue
|
||||
relative = path.relative_to(root).as_posix()
|
||||
for index, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
||||
if "solana_address::Address" in line:
|
||||
violations.append(Violation("KH_SOL001", relative, index, "use solana_pubkey::Pubkey instead of solana_address::Address"))
|
||||
return violations
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Run the khadhroony-specific 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 = audit_tracing(root) + audit_kb_lib_tracing(root) + audit_solana_types(root)
|
||||
violations.sort(key=lambda item: (item.code, item.path, item.line, item.message))
|
||||
if not violations:
|
||||
sys.stdout.write("Khadhroony workspace rule audit: clean\n")
|
||||
return 0
|
||||
sys.stdout.write(f"Khadhroony workspace rule audit: {len(violations)} violation(s)\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())
|
||||
174
scripts/audit_rust_export_completeness.py
Executable file
174
scripts/audit_rust_export_completeness.py
Executable file
@@ -0,0 +1,174 @@
|
||||
#!/usr/bin/env python3
|
||||
# file: scripts/audit_rust_export_completeness.py
|
||||
# version: 2
|
||||
|
||||
"""Report missing crate-root exports and replaceable long internal paths.
|
||||
|
||||
The audit distinguishes declarations from their crate-root aliases. Crate-root
|
||||
files are excluded from long-path findings because re-export declarations must
|
||||
name their source module. Findings remain advisory until the affected crate is
|
||||
compiled.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import dataclasses
|
||||
import pathlib
|
||||
import re
|
||||
import sys
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class Candidate:
|
||||
"""One export-completeness candidate."""
|
||||
|
||||
code: str
|
||||
path: str
|
||||
line: int
|
||||
message: str
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class Declaration:
|
||||
"""One module-level public or crate-public declaration."""
|
||||
|
||||
module: str
|
||||
name: str
|
||||
visibility: str
|
||||
path: pathlib.Path
|
||||
line: int
|
||||
|
||||
|
||||
def crate_roots(root: pathlib.Path) -> list[pathlib.Path]:
|
||||
"""Return workspace crate directories that contain a Rust crate root."""
|
||||
|
||||
crates: list[pathlib.Path] = []
|
||||
for manifest in root.glob("*/Cargo.toml"):
|
||||
crate = manifest.parent
|
||||
if (crate / "src/lib.rs").is_file() or (crate / "src/main.rs").is_file():
|
||||
crates.append(crate)
|
||||
return sorted(crates)
|
||||
|
||||
|
||||
def module_path(crate: pathlib.Path, path: pathlib.Path) -> str:
|
||||
"""Return the Rust module path represented by one source file."""
|
||||
|
||||
relative = path.relative_to(crate / "src").with_suffix("")
|
||||
return "::".join(relative.parts)
|
||||
|
||||
|
||||
def declaration_candidates(crate: pathlib.Path, path: pathlib.Path) -> list[Declaration]:
|
||||
"""Return module-level public declarations from one Rust module."""
|
||||
|
||||
text = path.read_text(encoding="utf-8")
|
||||
depth = 0
|
||||
found: list[Declaration] = []
|
||||
pattern = re.compile(
|
||||
r"^\s*(pub(?:\(crate\))?)\s+(?:(?:async|unsafe|const)\s+)*(?:const|static|type|struct|enum|trait|union|fn)\s+([A-Za-z_][A-Za-z0-9_]*)"
|
||||
)
|
||||
for number, line in enumerate(text.splitlines(), 1):
|
||||
if depth == 0:
|
||||
match = pattern.match(line)
|
||||
if match is not None:
|
||||
found.append(
|
||||
Declaration(
|
||||
module_path(crate, path),
|
||||
match.group(2),
|
||||
match.group(1),
|
||||
path,
|
||||
number,
|
||||
)
|
||||
)
|
||||
depth += line.count("{") - line.count("}")
|
||||
if depth < 0:
|
||||
depth = 0
|
||||
return found
|
||||
|
||||
|
||||
def root_exports(crate_root: pathlib.Path) -> dict[tuple[str, str], str]:
|
||||
"""Return source module/name pairs and their crate-root aliases."""
|
||||
|
||||
exports: dict[tuple[str, str], str] = {}
|
||||
pattern = re.compile(
|
||||
r"^\s*pub(?:\(crate\))?\s+use\s+crate::([A-Za-z_][A-Za-z0-9_:]*)::([A-Za-z_][A-Za-z0-9_]*)(?:\s+as\s+([A-Za-z_][A-Za-z0-9_]*))?\s*;",
|
||||
re.MULTILINE,
|
||||
)
|
||||
for match in pattern.finditer(crate_root.read_text(encoding="utf-8")):
|
||||
exports[(match.group(1), match.group(2))] = match.group(3) or match.group(2)
|
||||
return exports
|
||||
|
||||
|
||||
def audit_crate(workspace: pathlib.Path, crate: pathlib.Path) -> list[Candidate]:
|
||||
"""Audit one crate and return advisory candidates."""
|
||||
|
||||
crate_root = crate / "src/lib.rs"
|
||||
if not crate_root.is_file():
|
||||
crate_root = crate / "src/main.rs"
|
||||
exports = root_exports(crate_root)
|
||||
candidates: list[Candidate] = []
|
||||
declarations: dict[tuple[str, str], Declaration] = {}
|
||||
for path in sorted((crate / "src").rglob("*.rs")):
|
||||
if path == crate_root:
|
||||
continue
|
||||
for declaration in declaration_candidates(crate, path):
|
||||
key = (declaration.module, declaration.name)
|
||||
declarations[key] = declaration
|
||||
if key not in exports:
|
||||
relative = path.relative_to(workspace).as_posix()
|
||||
candidates.append(
|
||||
Candidate(
|
||||
"EXPORT001",
|
||||
relative,
|
||||
declaration.line,
|
||||
f"{declaration.visibility} `{declaration.name}` has no crate-root re-export",
|
||||
)
|
||||
)
|
||||
long_path = re.compile(
|
||||
r"\bcrate::([A-Za-z_][A-Za-z0-9_:]*)::([A-Za-z_][A-Za-z0-9_]*)"
|
||||
)
|
||||
for path in sorted((crate / "src").rglob("*.rs")):
|
||||
if path == crate_root:
|
||||
continue
|
||||
for number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
||||
for match in long_path.finditer(line):
|
||||
key = (match.group(1), match.group(2))
|
||||
alias = exports.get(key)
|
||||
if alias is None:
|
||||
continue
|
||||
relative = path.relative_to(workspace).as_posix()
|
||||
candidates.append(
|
||||
Candidate(
|
||||
"EXPORT002",
|
||||
relative,
|
||||
number,
|
||||
f"long internal path `crate::{match.group(1)}::{match.group(2)}` must use `crate::{alias}`",
|
||||
)
|
||||
)
|
||||
return candidates
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Run the advisory export-completeness audit."""
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--root", default=".")
|
||||
parser.add_argument("--summary-only", action="store_true")
|
||||
arguments = parser.parse_args()
|
||||
workspace = pathlib.Path(arguments.root).resolve()
|
||||
candidates = [item for crate in crate_roots(workspace) for item in audit_crate(workspace, crate)]
|
||||
candidates.sort(key=lambda item: (item.code, item.path, item.line, item.message))
|
||||
counts: dict[str, int] = {}
|
||||
for item in candidates:
|
||||
counts[item.code] = counts.get(item.code, 0) + 1
|
||||
sys.stdout.write(f"Rust export completeness audit: {len(candidates)} candidate(s)\n")
|
||||
for code in sorted(counts):
|
||||
sys.stdout.write(f"{code}: {counts[code]}\n")
|
||||
if not arguments.summary_only:
|
||||
for item in candidates:
|
||||
sys.stdout.write(f"{item.code} {item.path}:{item.line}: {item.message}\n")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
102
scripts/audit_rust_general_rules.py
Executable file
102
scripts/audit_rust_general_rules.py
Executable file
@@ -0,0 +1,102 @@
|
||||
#!/usr/bin/env python3
|
||||
# file: scripts/audit_rust_general_rules.py
|
||||
# version: 3
|
||||
|
||||
"""Audit mechanically verifiable Rust rules shared by all Rust projects."""
|
||||
|
||||
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 Rust sources outside generated build directories."""
|
||||
|
||||
return sorted(
|
||||
path
|
||||
for path in root.rglob("*.rs")
|
||||
if "target" not in path.parts
|
||||
and ".git" not in path.parts
|
||||
and "mnt" not in path.relative_to(root).parts
|
||||
and path.relative_to(root).parts[:2]
|
||||
!= ("migration", "khadhroony-bot2-reference")
|
||||
and not any(part.startswith("pre035_") for part in path.relative_to(root).parts)
|
||||
)
|
||||
|
||||
|
||||
def audit_file(root: pathlib.Path, path: pathlib.Path) -> list[Violation]:
|
||||
"""Audit one Rust source against general rules."""
|
||||
|
||||
relative = path.relative_to(root).as_posix()
|
||||
text = path.read_text(encoding="utf-8")
|
||||
lines = text.splitlines()
|
||||
violations: list[Violation] = []
|
||||
expected_header = f"// file: {relative}"
|
||||
if not lines or lines[0] != expected_header:
|
||||
violations.append(Violation("RUST001", 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("RUST002", relative, 2, "missing positive file version"))
|
||||
if not text.endswith("\n") or text.endswith("\n\n"):
|
||||
violations.append(Violation("RUST003", relative, max(len(lines), 1), "file must end with exactly one newline"))
|
||||
for index, line in enumerate(lines, 1):
|
||||
stripped = line.strip()
|
||||
if re.match(r"^(?:pub(?:\(crate\))?\s+)?use\s+", stripped) is None:
|
||||
continue
|
||||
is_export = stripped.startswith("pub use ") or stripped.startswith("pub(crate) use ")
|
||||
justified_trait = "rust-rules: trait-import" in stripped
|
||||
justified_derive = "rust-rules: derive-import" in stripped
|
||||
if not is_export and not justified_trait and not justified_derive:
|
||||
violations.append(Violation("RUST010", relative, index, "ordinary use requires a trait or derive import justification"))
|
||||
if "::*" in stripped:
|
||||
violations.append(Violation("RUST012", relative, index, "glob imports are forbidden"))
|
||||
if "{" in stripped or "}" in stripped:
|
||||
code = "RUST011" if is_export else "RUST013"
|
||||
violations.append(Violation(code, relative, index, "grouped use declarations are forbidden"))
|
||||
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[:20]:
|
||||
violations.append(Violation("RUST020", relative, 1, f"missing `{attribute}`"))
|
||||
for index, line in enumerate(lines, 1):
|
||||
if not (line.startswith("pub use ") or line.startswith("pub(crate) use ")):
|
||||
continue
|
||||
previous = lines[index - 2].strip() if index >= 2 else ""
|
||||
if not previous.startswith("///"):
|
||||
violations.append(Violation("RUST021", relative, index, "crate-root re-export requires adjacent rustdoc"))
|
||||
return violations
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Run the general Rust 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")
|
||||
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())
|
||||
33
scripts/audit_rust_workspace_rules.py
Executable file
33
scripts/audit_rust_workspace_rules.py
Executable file
@@ -0,0 +1,33 @@
|
||||
#!/usr/bin/env python3
|
||||
# file: scripts/audit_rust_workspace_rules.py
|
||||
# version: 2
|
||||
|
||||
"""Run both the general Rust and khadhroony-specific workspace audits."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import pathlib
|
||||
import subprocess
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Run both independent audit scripts."""
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--root", default=".")
|
||||
parser.add_argument("--report-only", action="store_true")
|
||||
arguments = parser.parse_args()
|
||||
script_dir = pathlib.Path(__file__).resolve().parent
|
||||
commands = [
|
||||
["python3", str(script_dir / "audit_rust_general_rules.py"), "--root", arguments.root],
|
||||
["python3", str(script_dir / "audit_khadhroony_workspace_rules.py"), "--root", arguments.root],
|
||||
]
|
||||
if arguments.report_only:
|
||||
for command in commands:
|
||||
command.append("--report-only")
|
||||
return max(subprocess.run(command, check=False).returncode for command in commands)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user