mirror of
https://github.com/Strategic-Automation/violin.git
synced 2026-08-14 12:33:37 +02:00
357 lines
13 KiB
Python
357 lines
13 KiB
Python
"""Release gate checks for the Violin plugin.
|
|
|
|
The checker is a REAL gate: it runs an isolated plugin import, compares the
|
|
manifest's provides_tools against the tools actually registered, and (unless
|
|
disabled) shells out to ruff and pytest. Failures surface as errors and cause
|
|
a non-zero exit code — so CI cannot pass a broken tree.
|
|
|
|
Heavy checks (ruff/pytest) are gated behind VIOLIN_CHECK_RELEASE_SKIP_HEAVY=1
|
|
(default: run them).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import tomllib
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from .results import GuardResult
|
|
from .skill_policy import catalog_snapshot, validate_catalog
|
|
|
|
__all__ = [
|
|
"GuardResult",
|
|
"ReleaseCheckResult",
|
|
"check_release",
|
|
"main",
|
|
"resolve_reference",
|
|
]
|
|
|
|
|
|
def _project_python(repo_root: Path) -> str:
|
|
"""Prefer the repository virtualenv when a profile runtime invokes the CLI."""
|
|
|
|
candidates = (
|
|
repo_root / ".venv" / "Scripts" / "python.exe",
|
|
repo_root / ".venv" / "bin" / "python",
|
|
)
|
|
for candidate in candidates:
|
|
if candidate.exists():
|
|
return str(candidate)
|
|
return sys.executable
|
|
|
|
|
|
@dataclass
|
|
class ReleaseCheckResult(GuardResult):
|
|
pass
|
|
|
|
|
|
def _plugin_root() -> Path:
|
|
# plugins/violin_guard/release.py -> plugins/violin_guard
|
|
return Path(__file__).resolve().parent
|
|
|
|
|
|
def _pytest_basetemp(repo_path: Path) -> str:
|
|
"""Create pytest's private temp directory in the ignored engagement tree."""
|
|
|
|
engagement_root = repo_path / "engagements"
|
|
engagement_root.mkdir(parents=True, exist_ok=True)
|
|
return tempfile.mkdtemp(prefix=".pytest-release-", dir=engagement_root)
|
|
|
|
|
|
def resolve_reference(source: Path, reference: str) -> Path:
|
|
"""Resolve a pentest skill reference from the skill package root."""
|
|
source = source.resolve()
|
|
for parent in (source.parent, *source.parents):
|
|
if parent.name == "pentest" and parent.parent.name == "skills":
|
|
return (parent / reference).resolve()
|
|
raise ValueError(f"source is not inside skills/pentest: {source}")
|
|
|
|
|
|
def _check_manifest_and_changelog(root: Path, result: ReleaseCheckResult) -> list[str]:
|
|
"""Require exact SemVer equality across every published version surface."""
|
|
plugin_yaml = root / "plugin.yaml"
|
|
provides_tools: list[str] = []
|
|
versions: dict[str, str] = {}
|
|
if plugin_yaml.exists():
|
|
import yaml
|
|
|
|
data = yaml.safe_load(plugin_yaml.read_text(encoding="utf-8"))
|
|
version = str(data.get("version", "0.0.0"))
|
|
versions["plugin manifest"] = version
|
|
provides_tools = list(data.get("provides_tools", []) or [])
|
|
if not re.fullmatch(r"\d+\.\d+\.\d+", version):
|
|
result.add_error(f"plugin.yaml version '{version}' is not a valid semver")
|
|
else:
|
|
result.add_error("plugin.yaml not found")
|
|
|
|
repo = root.parent.parent
|
|
pyproject = repo / "pyproject.toml"
|
|
if pyproject.exists():
|
|
project = tomllib.loads(pyproject.read_text(encoding="utf-8"))["project"]
|
|
versions["project"] = str(project.get("version", ""))
|
|
dependencies = [str(value).lower() for value in project.get("dependencies", [])]
|
|
if not any(value.startswith("pyyaml") for value in dependencies):
|
|
result.add_error("PyYAML must be a runtime project dependency")
|
|
else:
|
|
result.add_info("PyYAML is declared as a runtime dependency")
|
|
else:
|
|
result.add_error("pyproject.toml not found")
|
|
|
|
distribution = repo / "distribution.yaml"
|
|
if distribution.exists():
|
|
import yaml
|
|
|
|
versions["distribution"] = str(
|
|
yaml.safe_load(distribution.read_text(encoding="utf-8")).get("version", "")
|
|
)
|
|
else:
|
|
result.add_error("distribution.yaml not found")
|
|
|
|
changelog = repo / "CHANGELOG.md"
|
|
if changelog.exists():
|
|
match = re.search(
|
|
r"(?m)^## (\d+\.\d+\.\d+)(?: \(Unreleased\))?\s*$",
|
|
changelog.read_text(encoding="utf-8"),
|
|
)
|
|
if match:
|
|
versions["changelog"] = match.group(1)
|
|
else:
|
|
result.add_error("CHANGELOG.md has no top SemVer entry")
|
|
else:
|
|
result.add_error("CHANGELOG.md not found")
|
|
|
|
invalid = {
|
|
name: value for name, value in versions.items() if not re.fullmatch(r"\d+\.\d+\.\d+", value)
|
|
}
|
|
for name, value in invalid.items():
|
|
result.add_error(f"{name} version '{value}' is not exact SemVer")
|
|
if versions and len(set(versions.values())) != 1:
|
|
result.add_error(
|
|
"version mismatch: " + ", ".join(f"{name}={value}" for name, value in versions.items())
|
|
)
|
|
elif versions:
|
|
result.add_info(f"all version surfaces match {next(iter(versions.values()))}")
|
|
|
|
return provides_tools
|
|
|
|
|
|
def _check_isolated_import_and_tools(
|
|
root: Path, provides_tools: list[str], result: ReleaseCheckResult
|
|
) -> None:
|
|
"""Verify isolated plugin import and manifest vs registered tools match."""
|
|
module_name = "violin_guard_release_check"
|
|
old_module = sys.modules.get(module_name)
|
|
mod = None
|
|
try:
|
|
sys.path.insert(0, str(root.parent))
|
|
spec = importlib.util.spec_from_file_location(
|
|
module_name,
|
|
root / "__init__.py",
|
|
submodule_search_locations=[str(root)],
|
|
)
|
|
if spec is None or spec.loader is None:
|
|
raise ImportError("could not build plugin import specification")
|
|
mod = importlib.util.module_from_spec(spec)
|
|
sys.modules[module_name] = mod
|
|
spec.loader.exec_module(mod)
|
|
result.add_info("isolated plugin import OK")
|
|
except Exception as exc: # noqa: BLE001
|
|
result.add_error(f"plugin import failed: {type(exc).__name__}: {exc}")
|
|
finally:
|
|
sys.path.pop(0)
|
|
if old_module is None:
|
|
sys.modules.pop(module_name, None)
|
|
else:
|
|
sys.modules[module_name] = old_module
|
|
|
|
if mod is not None:
|
|
registered = sorted(getattr(mod, "REGISTERED_TOOLS", []) or [])
|
|
if not registered:
|
|
result.add_warning("plugin exposes no REGISTERED_TOOLS list")
|
|
elif sorted(provides_tools) != registered:
|
|
result.add_error(
|
|
"provides_tools mismatch: manifest="
|
|
f"{sorted(provides_tools)} registered={registered}"
|
|
)
|
|
else:
|
|
result.add_info("provides_tools matches registered tools")
|
|
definitions = list(getattr(mod, "TOOL_DEFINITIONS", []) or [])
|
|
for definition in definitions:
|
|
expected = mod.schemas.to_tool_schema(definition.model)["parameters"]
|
|
if definition.schema.get("parameters") != expected:
|
|
result.add_error(f"tool schema drift for {definition.name}")
|
|
if len(definitions) == len(registered):
|
|
result.add_info("tool registry models and exported schemas are consistent")
|
|
|
|
|
|
def _check_skill_snapshot(root: Path, result: ReleaseCheckResult) -> None:
|
|
"""Verify checked-in external-skill dependency manifest against approved catalog."""
|
|
snapshot_path = root.parent.parent / "skills.snapshot.json"
|
|
catalog_errors = validate_catalog()
|
|
if catalog_errors:
|
|
result.errors.extend(catalog_errors)
|
|
elif not snapshot_path.exists():
|
|
result.add_error("skills.snapshot.json not found")
|
|
else:
|
|
import json
|
|
|
|
try:
|
|
checked_in = json.loads(snapshot_path.read_text(encoding="utf-8"))
|
|
expected = catalog_snapshot(root.parent.parent)
|
|
for entry in expected["skills"]:
|
|
entry.pop("path", None)
|
|
if checked_in != expected:
|
|
result.add_error("skills.snapshot.json does not match the approved skill catalog")
|
|
else:
|
|
result.add_info("external skill dependency snapshot matches catalog")
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
result.add_error(f"skills.snapshot.json is invalid: {exc}")
|
|
|
|
|
|
def _check_heavy_linter_and_tests(root: Path, result: ReleaseCheckResult) -> None:
|
|
"""Execute ruff and pytest heavy release gates unless opted out via env."""
|
|
if os.environ.get("VIOLIN_CHECK_RELEASE_SKIP_HEAVY") == "1":
|
|
result.add_info("heavy checks skipped (VIOLIN_CHECK_RELEASE_SKIP_HEAVY=1)")
|
|
return
|
|
|
|
repo_path = root.parent.parent
|
|
repo_root = str(repo_path)
|
|
python = _project_python(repo_path)
|
|
try:
|
|
ruff = subprocess.run(
|
|
[python, "-m", "ruff", "check", "."],
|
|
cwd=repo_root,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if ruff.returncode != 0:
|
|
result.add_error("ruff check failed:\n" + (ruff.stdout or ruff.stderr).strip()[:2000])
|
|
else:
|
|
result.add_info("ruff check passed")
|
|
except FileNotFoundError:
|
|
result.add_warning("ruff not installed; skipped")
|
|
|
|
basetemp = _pytest_basetemp(repo_path)
|
|
try:
|
|
pytest = subprocess.run(
|
|
[
|
|
python,
|
|
"-m",
|
|
"pytest",
|
|
"-q",
|
|
"-p",
|
|
"no:cacheprovider",
|
|
"--basetemp",
|
|
basetemp,
|
|
],
|
|
cwd=repo_root,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if pytest.returncode != 0:
|
|
result.add_error(
|
|
"test suite failed:\n" + (pytest.stdout or pytest.stderr).strip()[:2000]
|
|
)
|
|
else:
|
|
result.add_info("test suite passed")
|
|
except FileNotFoundError:
|
|
result.add_warning("pytest not installed; skipped")
|
|
finally:
|
|
shutil.rmtree(basetemp, ignore_errors=True)
|
|
|
|
|
|
def _check_stale_skill_docs(root: Path, result: ReleaseCheckResult) -> None:
|
|
"""Scan skill documentation for stale or deprecated guard surface references."""
|
|
profile_root = root.parent.parent
|
|
skills_root = profile_root / "skills"
|
|
forbidden = {
|
|
"scripts/guard/": "removed legacy guard package",
|
|
"hypothesis_guard.py": "removed hypothesis wrapper",
|
|
"session_search": "unavailable session-search tool",
|
|
"violin_record_history": "removed executor-owned history tool",
|
|
"violin_message_tick": "removed model-visible message tool",
|
|
"violin_guard.py close": "nonexistent close subcommand",
|
|
"check-closeout": "nonexistent closeout subcommand",
|
|
"sync-clear": "nonexistent sync-clear subcommand",
|
|
"validate_scope_data": "private legacy scope validator",
|
|
}
|
|
if not skills_root.exists():
|
|
return
|
|
|
|
docs = [*skills_root.rglob("*.md"), *skills_root.rglob("*.yaml")]
|
|
for doc in docs:
|
|
try:
|
|
text = doc.read_text(encoding="utf-8")
|
|
except Exception:
|
|
continue
|
|
for token, reason in forbidden.items():
|
|
if token in text:
|
|
result.add_error(
|
|
f"stale skill reference in {doc.relative_to(profile_root)}: "
|
|
f"{token!r} ({reason})"
|
|
)
|
|
if not any("stale skill reference" in e for e in result.errors):
|
|
result.add_info("skill documentation matches the current guard surface")
|
|
|
|
|
|
def _check_active_documentation_contracts(root: Path, result: ReleaseCheckResult) -> None:
|
|
repo = root.parent.parent
|
|
docs = [repo / "README.md", *sorted((repo / "skills").rglob("*.md"))]
|
|
stale_credit = re.compile(
|
|
r"(?:Recon|RECON):?\s*5|(?:Exploitation|EXPLOITATION):?\s*10", re.IGNORECASE
|
|
)
|
|
stale_findings = re.compile(r"\|\s*Findings\s*\|[^\n]*findings\.yaml", re.IGNORECASE)
|
|
for path in docs:
|
|
text = path.read_text(encoding="utf-8")
|
|
if stale_credit.search(text):
|
|
result.add_error(f"stale sync-credit contract in {path.relative_to(repo)}")
|
|
if stale_findings.search(text):
|
|
result.add_error(f"stale canonical-finding contract in {path.relative_to(repo)}")
|
|
if not any("contract in" in error for error in result.errors):
|
|
result.add_info("active documentation contracts are current")
|
|
|
|
|
|
def check_release() -> ReleaseCheckResult:
|
|
"""Run all release gate checks. This is a REAL gate — failures add errors
|
|
and cause a non-zero exit code (see CLI cmd_check_release)."""
|
|
result = ReleaseCheckResult()
|
|
root = _plugin_root()
|
|
|
|
provides_tools = _check_manifest_and_changelog(root, result)
|
|
_check_isolated_import_and_tools(root, provides_tools, result)
|
|
_check_skill_snapshot(root, result)
|
|
_check_heavy_linter_and_tests(root, result)
|
|
|
|
tests_dir = root.parent.parent / "tests"
|
|
if tests_dir.exists():
|
|
result.add_info(f"tests directory found: {tests_dir}")
|
|
else:
|
|
result.add_warning("tests directory not found")
|
|
|
|
_check_stale_skill_docs(root, result)
|
|
_check_active_documentation_contracts(root, result)
|
|
return result
|
|
|
|
|
|
def main() -> int:
|
|
"""Print release diagnostics and return nonzero only when errors exist."""
|
|
result = check_release()
|
|
for item in result.errors:
|
|
print(f"ERROR: {item}")
|
|
for item in result.warnings:
|
|
print(f"WARN: {item}")
|
|
for item in result.infos:
|
|
print(f"OK: {item}")
|
|
return 1 if result.errors else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|