harness_factory.py

"""
Rhodawk AI — Ethical PoC Harness Factory
=========================================
Generates minimal proof-of-concept test harnesses that exercise identified
assumption gaps LOCALLY in an isolated sandbox.

ETHICAL CONSTRAINTS (hard-coded, not configurable):
  - Generated harnesses target only locally cloned source code
  - No network calls from within generated harnesses
  - Execution is time-limited (default 30 s)
  - All secrets stripped from sandbox environment
  - Harness code is shown to operator BEFORE execution — never auto-run
  - Output is PoC-grade only: demonstrates behaviour, not weaponised

Orchestrated by Nous Hermes 3 via OpenRouter.
"""

from __future__ import annotations

import os
import re
import subprocess
import tempfile
import time
from pathlib import Path
from typing import Optional

import requests

OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY", "")
HERMES_MODEL = os.getenv(
    "RHODAWK_RESEARCH_MODEL",
    "nousresearch/hermes-3-llama-3.1-405b:free",
)
HARNESS_TIMEOUT = int(os.getenv("RHODAWK_HARNESS_TIMEOUT", "30"))

_SECRETS = [
    "OPENROUTER_API_KEY", "GITHUB_TOKEN", "GITHUB_PERSONAL_ACCESS_TOKEN",
    "TELEGRAM_BOT_TOKEN", "SLACK_WEBHOOK_URL", "RHODAWK_WEBHOOK_SECRET",
]

_HARNESS_PREAMBLE = """\
# ETHICAL POC — FOR RESPONSIBLE DISCLOSURE ONLY
# Generated by Rhodawk AI Ethical Security Research Platform
# This harness tests local code only — no network calls, no privilege escalation
# REVIEW CAREFULLY before authorising sandbox execution
"""


def _hermes(system: str, user: str) -> str:
    headers = {
        "Authorization": f"Bearer {OPENROUTER_API_KEY}",
        "Content-Type": "application/json",
        "HTTP-Referer": "https://rhodawk.ai",
    }
    payload = {
        "model": HERMES_MODEL,
        "messages": [
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
        "max_tokens": 4096,
        "temperature": 0.1,
    }
    resp = requests.post(
        "https://openrouter.ai/api/v1/chat/completions",
        headers=headers, json=payload, timeout=120,
    )
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]


def generate_poc_harness(assumption_gap: dict, repo_dir: str) -> dict:
    """
    Ask Hermes to generate a minimal PoC harness for a specific assumption gap.

    The harness is returned for HUMAN REVIEW — it is NOT executed here.
    Status is always 'PENDING_HUMAN_REVIEW' until the operator explicitly
    calls run_harness_in_sandbox().
    """
    gap_id   = assumption_gap.get("id", "unknown")
    gap_file = assumption_gap.get("file", "unknown")
    gap_desc = assumption_gap.get("description", "")
    gap_input = assumption_gap.get("untrusted_input", "")
    gap_check = assumption_gap.get("bypassed_check", "")
    gap_impact = assumption_gap.get("potential_impact", "")

    src_snippet = ""
    if gap_file and gap_file != "unknown":
        full_path = os.path.join(repo_dir, gap_file)
        if os.path.exists(full_path):
            try:
                lines = Path(full_path).read_text(
                    encoding="utf-8", errors="replace"
                ).splitlines()[:120]
                src_snippet = "\n".join(lines)
            except Exception:
                pass

    system = (
        "You are a responsible security researcher generating a minimal proof-of-concept. "
        "RULES:\n"
        "1. The harness imports and tests LOCAL code only — absolutely no network calls, "
        "   no socket connections, no subprocess calls to external tools.\n"
        "2. It must run in under 30 seconds.\n"
        "3. Last line must print exactly: TRIGGERED: True  OR  TRIGGERED: False\n"
        "4. It must NOT write files outside /tmp, escalate privileges, or exfiltrate data.\n"
        "5. PoC-grade only — demonstrates the behaviour, is not a weapon.\n"
        "6. Begin with the ethical preamble comment block shown in the user message."
    )

    user = (
        f"Generate a minimal Python PoC harness for this assumption gap.\n\n"
        f"Gap ID: {gap_id}\n"
        f"File: {gap_file}\n"
        f"Description: {gap_desc}\n"
        f"Untrusted input path: {gap_input}\n"
        f"Bypassed check: {gap_check}\n"
        f"Theoretical impact: {gap_impact}\n"
        f"Repo path: {repo_dir}\n\n"
        f"Relevant source snippet:\n```\n{src_snippet[:3000]}\n```\n\n"
        f"Start the file with this exact preamble:\n{_HARNESS_PREAMBLE}\n"
        f"Output ONLY the Python harness code — no explanation, no markdown fences."
    )

    try:
        raw = _hermes(system, user)
        code = re.sub(r"^```(?:python)?\n?", "", raw.strip(), flags=re.MULTILINE)
        code = re.sub(r"\n?```$", "", code.strip())

        if _HARNESS_PREAMBLE.strip().splitlines()[0] not in code:
            code = _HARNESS_PREAMBLE + "\n" + code

        return {
            "gap_id": gap_id,
            "file": gap_file,
            "harness_code": code,
            "status": "PENDING_HUMAN_REVIEW",
            "executed": False,
            "result": None,
            "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        }
    except Exception as e:
        return {
            "gap_id": gap_id,
            "error": str(e),
            "status": "GENERATION_FAILED",
        }


def run_harness_in_sandbox(
    harness_code: str,
    repo_dir: str,
    venv_dir: str = "/data/target_venv",
) -> dict:
    """
    Execute a HUMAN-REVIEWED harness in an isolated local sandbox.

    This function must only be called after the operator has:
      1. Read the harness code
      2. Clicked "I have reviewed this code" in the UI

    No network access, secrets stripped, time-limited.
    """
    if not harness_code.strip():
        return {"error": "Empty harness code.", "triggered": False}

    python_bin = (
        os.path.join(venv_dir, "bin", "python")
        if venv_dir and os.path.exists(os.path.join(venv_dir, "bin", "python"))
        else "python3"
    )

    fd, harness_path = tempfile.mkstemp(
        prefix="rhodawk_poc_", suffix=".py", dir="/tmp"
    )
    try:
        with os.fdopen(fd, "w") as f:
            f.write(harness_code)

        env = os.environ.copy()
        for secret in _SECRETS:
            env.pop(secret, None)
        env["PYTHONPATH"] = repo_dir
        env["PYTHONDONTWRITEBYTECODE"] = "1"

        proc = subprocess.run(
            [python_bin, harness_path],
            capture_output=True,
            text=True,
            timeout=HARNESS_TIMEOUT,
            cwd=repo_dir,
            env=env,
        )

        stdout = proc.stdout[:3000]
        triggered = "TRIGGERED: True" in stdout

        return {
            "exit_code": proc.returncode,
            "stdout": stdout,
            "stderr": proc.stderr[:1000],
            "triggered": triggered,
            "timed_out": False,
            "executed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        }

    except subprocess.TimeoutExpired:
        return {
            "triggered": False,
            "timed_out": True,
            "error": f"Harness timed out after {HARNESS_TIMEOUT}s — gap may not be triggerable.",
        }
    except Exception as e:
        return {"triggered": False, "error": str(e)}
    finally:
        try:
            os.unlink(harness_path)
        except OSError:
            pass