violin/benchmark/run.py

#!/usr/bin/env python3
"""run.py — Automated Hermes Profile Benchmark Runner with OpenRouter integration.

Executes Hermes non-interactively using the target profile against a benchmark lab target,
manages engagement state, and scores evidence automatically via score.py.
"""

import argparse
import copy
import ipaddress
import json
import os
import secrets
import shutil
import subprocess
import sys
import tempfile
from datetime import UTC, date, datetime
from pathlib import Path

import yaml
from yarl import URL

# Ensure repo root is on sys.path
REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
if hasattr(sys.stderr, "reconfigure"):
    sys.stderr.reconfigure(encoding="utf-8", errors="replace")

from benchmark.score import generate_markdown_summary, print_result, score_engagement  # noqa: E402
from plugins.violin_guard.command import validate_scope  # noqa: E402
from plugins.violin_guard.receipt_integrity import RECEIPT_KEY_ENV  # noqa: E402

_DEFAULT_HERMES_MAX_TOKENS = 12_000
_CLOSEOUT_TIMEOUT_SECONDS = 600


def _engagement_needs_closeout(eng_dir: Path) -> bool:
    """Return whether Hermes stopped before producing a complete closeout."""
    ptt_path = eng_dir / "state" / "ptt.md"
    if not ptt_path.exists():
        return True
    ptt_text = ptt_path.read_text(encoding="utf-8", errors="replace")
    if any(marker in ptt_text for marker in ("| [ ] |", "| [~] |")):
        return True
    findings_dir = eng_dir / "evidence" / "findings"
    if not findings_dir.exists() or not any(findings_dir.glob("FIND-*.md")):
        return True
    return not all(
        (eng_dir / relative).is_file()
        for relative in ("reporting/report.md", "retrospective/retrospective.md")
    )


def _closeout_command(args: argparse.Namespace, eng_dir: Path) -> list[str]:
    """Build a continuation goal that only closes an existing engagement."""
    command = [
        "hermes",
        "chat",
        "-p",
        "violin",
        "--provider",
        args.provider,
        "--yolo",
        "-q",
        f"/goal Continue the existing authorized engagement in {eng_dir.resolve()} and close it out following the pentest skill's closeout procedure. "
        "Never inspect benchmark answer keys, scorer code, challenge inventories, walkthroughs, or target-specific solution material.",
    ]
    if args.model:
        command.extend(["-m", args.model])
    return command


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Automated Hermes Profile Benchmark Runner with OpenRouter support."
    )
    parser.add_argument(
        "--eng-dir",
        type=Path,
        default=None,
        help="Engagement directory for benchmark execution (default: unique benchmark run directory)",
    )
    parser.add_argument(
        "--model",
        type=str,
        default="deepseek/deepseek-v4-flash-0731",
        help="Optional LLM model ID or openrouter/model-name (default: deepseek/deepseek-v4-flash-latest)",
    )
    parser.add_argument(
        "--skill",
        type=str,
        default="",
        help="Optional skill to preload (e.g. pentest)",
    )
    parser.add_argument(
        "--provider",
        type=str,
        default="openrouter",
        help="Hermes LLM provider (e.g. openrouter, openai, custom) (default: openrouter)",
    )
    parser.add_argument(
        "--api-base",
        type=str,
        default="https://openrouter.ai/api/v1",
        help="OpenAI-compatible API base URL (default: https://openrouter.ai/api/v1)",
    )
    parser.add_argument(
        "--target",
        type=str,
        default="https://duck-store.escape.tech",
        help="Target host or URL for the benchmark run (default: https://duck-store.escape.tech)",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Initialize engagement directory and print commands without executing Hermes CLI",
    )
    parser.add_argument(
        "--json-out",
        type=Path,
        help="Optional path to write benchmark results JSON",
    )
    parser.add_argument(
        "--markdown-out",
        type=Path,
        help="Optional path to write markdown summary",
    )
    return parser.parse_args()


def _scope_for_target(target: str) -> dict:
    """Render the canonical benchmark fixture for one URL, domain, or IP target."""
    fixture_path = REPO_ROOT / "benchmark" / "targets" / "duck-store" / "scope.yaml"
    scope = copy.deepcopy(yaml.safe_load(fixture_path.read_text(encoding="utf-8")))
    raw_target = target.strip()
    parsed = URL(raw_target if "://" in raw_target else f"https://{raw_target}")
    if parsed.scheme not in {"http", "https"} or not parsed.host:
        raise ValueError("benchmark target must be an HTTP(S) URL, domain, or IP address")
    host = parsed.host
    try:
        ipaddress.ip_address(host)
        addresses = [host]
        domains: list[str] = []
    except ValueError:
        addresses = []
        domains = [host]
    base_url = str(parsed.with_path("").with_query(None).with_fragment(None)).rstrip("/")
    scope["engagement"]["date"] = date.today().isoformat()
    scope["targets"] = {
        "ip_addresses": addresses,
        "domains": domains,
        "urls": [str(parsed)],
        "in_scope_urls": [str(parsed)],
    }
    excluded_paths = list((scope.get("exclusions") or {}).get("paths") or [])
    scope.setdefault("exclusions", {})["urls"] = [
        f"{base_url}/{path.lstrip('/')}" for path in excluded_paths
    ]
    benchmark = scope.get("benchmark")
    if isinstance(benchmark, dict):
        benchmark.pop("openapi_spec", None)
    return scope


def init_benchmark_engagement(eng_dir: Path, target: str) -> None:
    """Initialize benchmark engagement directory structure cleanly (idempotent reset)."""
    resolved_eng = eng_dir.resolve()
    resolved_engagements = (REPO_ROOT / "engagements").resolve()
    temporary_root = Path(tempfile.gettempdir()).resolve()
    pytest_root = (REPO_ROOT / ".pytest-tmp").resolve()
    if not (
        resolved_eng == resolved_engagements
        or resolved_engagements in resolved_eng.parents
        or temporary_root in resolved_eng.parents
        or pytest_root in resolved_eng.parents
    ):
        raise ValueError(
            f"Safety error: engagement directory {resolved_eng} must be inside {resolved_engagements}"
        )

    if eng_dir.exists():
        for item in eng_dir.iterdir():
            try:
                if item.is_dir():
                    shutil.rmtree(item, ignore_errors=True)
                else:
                    item.unlink(missing_ok=True)
            except Exception:
                pass

    eng_dir.mkdir(parents=True, exist_ok=True)
    (eng_dir / "scope").mkdir(parents=True, exist_ok=True)
    (eng_dir / "state").mkdir(parents=True, exist_ok=True)
    (eng_dir / "evidence").mkdir(parents=True, exist_ok=True)
    for phase_dir in (
        "recon",
        "recon/active",
        "recon/passive",
        "vuln-research",
        "exploitation",
        "post-exploitation",
        "privesc",
        "flags",
        "reporting",
        "retrospective",
        "executions",
    ):
        (eng_dir / "evidence" / phase_dir).mkdir(parents=True, exist_ok=True)
    (eng_dir / "exploits").mkdir(parents=True, exist_ok=True)

    scope_yaml = eng_dir / "scope" / "scope.yaml"
    scope_yaml.write_text(
        yaml.safe_dump(_scope_for_target(target), sort_keys=False), encoding="utf-8"
    )
    validation = validate_scope(scope_yaml)
    if validation.errors:
        raise ValueError("generated benchmark scope is invalid: " + "; ".join(validation.errors))

    ptt_md = eng_dir / "state" / "ptt.md"
    ptt_md.write_text(
        """# Pentesting Task Tree (PTT)

## Phase: RECON

| ID | Status | Task | Notes |
|---|---|---|---|
| PT-101 | [~] | Reconnaissance & Tech Detection | benchmark bootstrap; record exact request/response evidence |

## Phase: VULN_RESEARCH

| ID | Status | Task | Notes |
|---|---|---|---|
| PT-102 | [ ] | Vulnerability Assessment | test and disposition every mapped route, method, parameter, and role boundary |

## Phase: EXPLOITATION

| ID | Status | Task | Notes |
|---|---|---|---|
| PT-103 | [ ] | Exploitation & Proof Verification | validate only in-scope, non-destructive findings |
""",
        encoding="utf-8",
    )

    hyp_md = eng_dir / "hypotheses.md"
    hyp_md.write_text(
        """# Hypothesis Board
### H-001: Initial Target Assessment
**Status:** Candidate
**Linked challenges:**
""",
        encoding="utf-8",
    )

    feedback_md = eng_dir / "state" / "framework_feedback.md"
    feedback_md.write_text(
        """# Violin Framework Feedback & Friction Log
If you encounter tool friction, missing binaries/CLI tools, guard pathing errors, or framework bugs in Violin during this assessment, record them below.

CRITICAL: Whenever you inspect or read the guard codebase (e.g. plugins/violin_guard/ or guard scripts) to understand how a tool works, why a command was blocked, or how state is managed, log the occurrence below with:
1. Category: Guard Code Inspection
2. Issue Description: Which guard file/code you read and what you were trying to understand
3. Impact / Workaround: What confusion or error forced you to read the code
4. Prevention Suggestion: What specifically would prevent needing to read guard code in the future (e.g. clearer error messages, self-documenting CLI parameters, tool docs, or explicit return details)

| Timestamp | Category | Issue Description | Impact / Workaround | Prevention Suggestion |
|---|---|---|---|---|
""",
        encoding="utf-8",
    )

    coverage = {
        "coverage": {
            name: {"status": "pending", "evidence_or_reason": ""}
            for name in (
                "routes",
                "methods",
                "parameters",
                "role_boundaries",
                "authentication_flows",
                "business_logic",
                "redirects",
                "rate_limits",
                "ssrf",
                "injection",
                "object_authorization",
            )
        }
    }
    (eng_dir / "state" / "coverage-matrix.yaml").write_text(
        yaml.safe_dump(coverage, sort_keys=False), encoding="utf-8"
    )

    hist_md = eng_dir / "state" / "history.md"
    hist_md.write_text(
        "# Command History Log\n# Format: TIMESTAMP | PHASE | TARGET | CMD\n",
        encoding="utf-8",
    )


def main() -> int:
    args = parse_args()

    timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
    if args.eng_dir is not None:
        eng_dir = args.eng_dir
    else:
        eng_dir = REPO_ROOT / "engagements" / f"benchmark-run-{timestamp}"

    print("=== HERMES PROFILE BENCHMARK RUNNER ===")
    print(f"Engagement Dir : {eng_dir}")
    print(f"Model          : {args.model}")
    print(f"Provider       : {args.provider}")
    print(f"API Base       : {args.api_base}")
    print(f"Target         : {args.target}")

    init_benchmark_engagement(eng_dir, args.target)

    receipt_key = secrets.token_bytes(32)
    env = os.environ.copy()
    env["ENG_DIR"] = str(eng_dir.resolve())
    env[RECEIPT_KEY_ENV] = receipt_key.hex()
    env["OPENAI_API_BASE"] = args.api_base
    env["OPENAI_BASE_URL"] = args.api_base
    env["CUSTOM_BASE_URL"] = args.api_base
    # OpenRouter rejects requests whose declared output ceiling exceeds the
    # account's affordable budget, even when the model would stop earlier.
    # Keep the benchmark provider-safe while allowing an explicit override for
    # accounts with a larger entitlement.
    configured_max_tokens = os.environ.get("VIOLIN_BENCHMARK_MAX_TOKENS", "").strip()
    env["HERMES_MAX_TOKENS"] = configured_max_tokens or str(_DEFAULT_HERMES_MAX_TOKENS)
    if "NOUS_API_KEY" in env and "OPENAI_API_KEY" not in env:
        env["OPENAI_API_KEY"] = env["NOUS_API_KEY"]
    if "OPENROUTER_API_KEY" in env and "OPENAI_API_KEY" not in env:
        env["OPENAI_API_KEY"] = env["OPENROUTER_API_KEY"]
    if "OPENAI_API_KEY" not in env:
        env["OPENAI_API_KEY"] = "not-needed"

    venv_scripts = str(REPO_ROOT / ".venv" / "Scripts")
    venv_bin = str(REPO_ROOT / ".venv" / "bin")
    current_path = env.get("PATH", "")
    env["PATH"] = os.pathsep.join([path for path in (venv_scripts, venv_bin, current_path) if path])
    env["PYTHONPATH"] = str(REPO_ROOT)

    cmd = [
        "hermes",
        "chat",
        "-p",
        "violin",
        "--provider",
        args.provider,
        "--yolo",
        "-q",
        f"/goal Perform an exhaustive authorized security assessment of {args.target} using the engagement in {eng_dir.resolve()}. "
        "Never inspect benchmark answer keys, scorer code, challenge inventories, walkthroughs, or target-specific solution material.",
    ]
    if args.skill:
        cmd.extend(["-s", args.skill])
    if args.model:
        cmd.extend(["-m", args.model])

    print(f"\nExecution Command: {' '.join(cmd)}")

    started = datetime.now(UTC)
    runner = {
        "status": "dry_run" if args.dry_run else "not_started",
        "valid": False,
        "exit_code": None,
        "provider": args.provider,
        "model": args.model,
        "started_at": started.isoformat(),
        "completed_at": None,
        "duration_seconds": None,
        "failure_reason": None,
    }
    if args.dry_run:
        print("[DRY-RUN] Benchmark engagement structure prepared. Skipping Hermes execution.")
    else:
        hermes_bin = shutil.which("hermes")
        if not hermes_bin:
            runner.update(
                status="failed_to_start",
                valid=False,
                failure_reason="hermes binary not found on PATH",
            )
            print("[ERROR] 'hermes' binary not found on PATH.")
        else:
            try:
                completed = subprocess.run(cmd, env=env, cwd=eng_dir, check=False)
                runner["exit_code"] = completed.returncode
                runner["valid"] = completed.returncode == 0
                runner["status"] = "completed" if completed.returncode == 0 else "failed"
                if completed.returncode != 0:
                    runner["failure_reason"] = f"Hermes exited with status {completed.returncode}"
            except Exception as exc:  # noqa: BLE001
                runner.update(status="failed_to_start", valid=False, failure_reason=str(exc))
                print(f"[ERROR] Hermes execution failed to start: {exc}")

            # Hermes can return successfully after the substantive assessment
            # while leaving PTT closeout/reporting work unfinished. Run a
            # bounded continuation pass so scoring never treats that partial
            # state as a completed engagement. Closeout is best-effort: a
            # timeout or nonzero exit must NOT invalidate a run whose
            # substantive evidence already exists (scoring proceeds), it is
            # reported as a soft warning instead.
            runner["closeout_attempts"] = []
            runner["closeout_complete"] = False
            if runner["valid"]:
                for _attempt in range(2):
                    if not _engagement_needs_closeout(eng_dir):
                        break
                    try:
                        closeout = subprocess.run(
                            _closeout_command(args, eng_dir),
                            env=env,
                            cwd=eng_dir,
                            check=False,
                            timeout=_CLOSEOUT_TIMEOUT_SECONDS,
                        )
                    except subprocess.TimeoutExpired:
                        runner["closeout_attempts"].append("timeout")
                        runner["closeout_warning"] = (
                            f"Hermes closeout exceeded {_CLOSEOUT_TIMEOUT_SECONDS} seconds; "
                            "scoring proceeds on substantive evidence"
                        )
                        break
                    runner["closeout_attempts"].append(closeout.returncode)
                    if closeout.returncode != 0:
                        runner["closeout_warning"] = (
                            f"Hermes closeout exited with status {closeout.returncode}"
                        )
                        break
                runner["closeout_complete"] = not _engagement_needs_closeout(eng_dir)

            history_path = eng_dir / "state" / "history.md"
            execution_dir = eng_dir / "evidence" / "executions"
            has_commands = history_path.exists() and any(
                line.strip().startswith("-")
                for line in history_path.read_text(encoding="utf-8", errors="replace").splitlines()
            )
            has_receipts = execution_dir.exists() and any(execution_dir.glob("*.json"))
            if runner["valid"] and not (has_commands and has_receipts):
                runner.update(
                    status="failed",
                    valid=False,
                    failure_reason="Hermes returned without producing benchmark execution evidence",
                )

    finished = datetime.now(UTC)
    runner["completed_at"] = finished.isoformat()
    runner["duration_seconds"] = round((finished - started).total_seconds(), 3)

    # Score engagement results
    print("\n=== SCORING ENGAGEMENT RESULTS ===")
    results = score_engagement(eng_dir, receipt_key=receipt_key)
    results["runner"] = runner
    results["score_benchmark_pass"] = results["benchmark_pass"]
    results["valid"] = runner["valid"]
    results["benchmark_pass"] = bool(runner["valid"] and results["benchmark_pass"])
    print_result(results)

    # Always write unique results into eng_dir
    (eng_dir / "results.json").write_text(json.dumps(results, indent=2), encoding="utf-8")
    (eng_dir / "results.md").write_text(generate_markdown_summary(results), encoding="utf-8")
    print(f"Wrote benchmark results to {eng_dir / 'results.json'} and {eng_dir / 'results.md'}")

    if args.json_out:
        args.json_out.parent.mkdir(parents=True, exist_ok=True)
        args.json_out.write_text(json.dumps(results, indent=2), encoding="utf-8")
        print(f"Wrote JSON results to {args.json_out}")

    if args.markdown_out:
        args.markdown_out.parent.mkdir(parents=True, exist_ok=True)
        md_summary = generate_markdown_summary(results)
        args.markdown_out.write_text(md_summary, encoding="utf-8")
        print(f"Wrote Markdown summary to {args.markdown_out}")
    if args.dry_run:
        return 0
    return 0 if runner["valid"] and results["benchmark_pass"] else 1


if __name__ == "__main__":
    sys.exit(main())