256 lines
9.6 KiB
Python
Executable File
256 lines
9.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# file: scripts/audit_khadhroony_workspace_rules.py
|
|
# version: 4
|
|
|
|
"""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()
|
|
alias_pattern = re.compile(
|
|
r"^pub\(crate\) use crate::[A-Za-z0-9_:]+::"
|
|
r"([A-Z][A-Z0-9_]+)(?: as ([A-Z][A-Z0-9_]+))?;$",
|
|
re.MULTILINE,
|
|
)
|
|
for match in alias_pattern.finditer(root_text):
|
|
exported_name = match.group(2) if match.group(2) is not None else match.group(1)
|
|
if exported_name.endswith("_TRACING_TARGET"):
|
|
aliases.add(exported_name)
|
|
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 audit_wincode_resolution(root: pathlib.Path) -> list[Violation]:
|
|
"""Require the currently compatible Solana wincode dependency family."""
|
|
|
|
violations: list[Violation] = []
|
|
cargo_path = root / "Cargo.toml"
|
|
lock_path = root / "Cargo.lock"
|
|
if not cargo_path.exists():
|
|
return violations
|
|
cargo = tomllib.loads(cargo_path.read_text(encoding="utf-8"))
|
|
workspace_wincode = cargo.get("workspace", {}).get("dependencies", {}).get("wincode")
|
|
version = workspace_wincode.get("version") if isinstance(workspace_wincode, dict) else None
|
|
if version != "^0.5":
|
|
violations.append(
|
|
Violation(
|
|
"KH_DEP001",
|
|
"Cargo.toml",
|
|
1,
|
|
"workspace wincode must remain constrained to `^0.5`",
|
|
)
|
|
)
|
|
if not lock_path.exists():
|
|
violations.append(
|
|
Violation(
|
|
"KH_DEP002",
|
|
"Cargo.lock",
|
|
1,
|
|
"versioned application workspace lockfile is required",
|
|
)
|
|
)
|
|
return violations
|
|
lock = tomllib.loads(lock_path.read_text(encoding="utf-8"))
|
|
packages = lock.get("package", [])
|
|
resolved_wincode = [
|
|
package.get("version")
|
|
for package in packages
|
|
if package.get("name") == "wincode"
|
|
]
|
|
resolved_varint = [
|
|
package.get("version")
|
|
for package in packages
|
|
if package.get("name") == "solana-wincode-varint"
|
|
]
|
|
if resolved_wincode != ["0.5.5"]:
|
|
violations.append(
|
|
Violation(
|
|
"KH_DEP003",
|
|
"Cargo.lock",
|
|
1,
|
|
f"expected only wincode 0.5.5, resolved {resolved_wincode}",
|
|
)
|
|
)
|
|
if resolved_varint != ["1.0.0"]:
|
|
violations.append(
|
|
Violation(
|
|
"KH_DEP004",
|
|
"Cargo.lock",
|
|
1,
|
|
f"expected only solana-wincode-varint 1.0.0, resolved {resolved_varint}",
|
|
)
|
|
)
|
|
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)
|
|
+ audit_wincode_resolution(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())
|