modules/cypher_code_as_tool.py

"""CYPHER V12 M37 — Code-as-Tool Generator.

CYPHER generates Python code as an ACTION (not as a text response). Code is
exec'd in M30 sandbox, output captured, fed back as ground truth.

Use cases:
  - Compute CVSS sub-scores from base vector
  - Parse complex logs
  - Validate IP ranges
  - Hash/encode operations
  - Custom one-off analysis

Pipeline:
  1. Detect prompt requiring computation
  2. CYPHER generates "code:" block
  3. Extract and validate code (AST parse, no imports beyond whitelist)
  4. Exec in sandbox (M30 ops or custom)
  5. Return result as part of response
"""
from __future__ import annotations

import ast
import logging
import re
from typing import Callable, Any

logger = logging.getLogger(__name__)


# Whitelisted modules (limited stdlib)
_ALLOWED_IMPORTS = {
    "re", "json", "base64", "hashlib", "math", "statistics",
    "datetime", "collections", "itertools", "functools",
    "ipaddress", "string", "binascii", "urllib.parse",
}

# Whitelisted built-ins
_ALLOWED_BUILTINS = {
    "abs", "all", "any", "ascii", "bin", "bool", "bytes", "chr",
    "dict", "divmod", "enumerate", "filter", "float", "format",
    "frozenset", "hex", "int", "isinstance", "issubclass", "iter",
    "len", "list", "map", "max", "min", "next", "oct", "ord",
    "pow", "print", "range", "repr", "reversed", "round", "set",
    "slice", "sorted", "str", "sum", "tuple", "type", "zip",
}


def detect_computation_request(prompt: str) -> bool:
    p = prompt.lower()
    triggers = (
        "compute", "calculate", "calcule", "calculer", "extract",
        "parse this", "analyze this", "count how many",
        "convert", "decode", "encode this", "hash this",
        "match this regex", "validate this", "what's the cvss",
    )
    return any(t in p for t in triggers)


def extract_code_block(text: str) -> str | None:
    """Extract Python code from text (markdown ``` or 'code:' prefix)."""
    # Markdown
    m = re.search(r"```(?:python)?\n(.*?)```", text, re.DOTALL)
    if m:
        return m.group(1).strip()
    # "code:" prefix
    m = re.search(r"code:\s*\n([\s\S]+?)(?:\n\n|\Z)", text, re.IGNORECASE)
    if m:
        return m.group(1).strip()
    return None


def validate_code(code: str) -> dict:
    """AST validation + import whitelist."""
    if not code:
        return {"valid": False, "error": "empty"}
    try:
        tree = ast.parse(code)
    except SyntaxError as e:
        return {"valid": False, "error": f"syntax: {e}"}
    issues: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for n in node.names:
                if n.name not in _ALLOWED_IMPORTS:
                    issues.append(f"disallowed_import:{n.name}")
        if isinstance(node, ast.ImportFrom):
            if node.module not in _ALLOWED_IMPORTS:
                issues.append(f"disallowed_from:{node.module}")
        if isinstance(node, ast.Call):
            if isinstance(node.func, ast.Name):
                if node.func.id in {"exec", "eval", "compile", "__import__", "open"}:
                    issues.append(f"disallowed_call:{node.func.id}")
            if isinstance(node.func, ast.Attribute):
                if node.func.attr in {"system", "popen", "spawn", "fork"}:
                    issues.append(f"disallowed_attr:{node.func.attr}")
    return {"valid": not issues, "issues": issues, "n_lines": len(code.splitlines())}


def run_code_via_sandbox(code: str, sandbox=None, timeout_sec: int = 8) -> dict:
    """Wrap and exec via M30 sandbox."""
    if sandbox is None:
        try:
            from cypher_sandbox_reasoning import exec_sandboxed
            # Use a generic op: we need to add a custom op or use existing
            # For now: use regex_match as a no-op carrier and emit result via print
            return {"ok": False, "error": "sandbox bridging not yet wired; use exec_sandboxed directly"}
        except ImportError:
            return {"ok": False, "error": "no sandbox available"}
    return sandbox.exec_custom(code, timeout_sec=timeout_sec)


class CodeAsTool:
    """High-level orchestrator."""

    def __init__(
        self,
        generate_fn: Callable[[str, int, float], str] | None = None,
        sandbox=None,
    ):
        self.generate_fn = generate_fn
        self.sandbox = sandbox

    def code_prompt(self, original_prompt: str) -> str:
        """Build a prompt instructing CYPHER to generate code."""
        return (
            f"You will solve this by writing a small Python snippet that prints "
            f"the answer as JSON. Use only: re, json, hashlib, base64, math, "
            f"ipaddress, urllib.parse, datetime. "
            f"Output code only (no explanation).\n\n"
            f"Task: {original_prompt}\n\n"
            f"```python\n"
        )

    def execute(self, prompt: str) -> dict:
        if not self.generate_fn:
            return {"ok": False, "error": "no generate_fn"}
        if not detect_computation_request(prompt):
            return {"ok": False, "error": "not a computation request",
                    "should_use": False}
        cprompt = self.code_prompt(prompt)
        try:
            generated = self.generate_fn(cprompt, 400, 0.20)
        except Exception as e:
            return {"ok": False, "error": f"gen: {e}"}
        code = extract_code_block(generated) or generated
        validation = validate_code(code)
        if not validation["valid"]:
            return {"ok": False, "error": "code invalid",
                    "validation": validation, "code": code[:300]}
        # Exec (placeholder — would call sandbox in production)
        return {
            "ok": True,
            "code": code,
            "validation": validation,
            "exec_status": "skipped_smoke",
        }


__all__ = ["CodeAsTool", "detect_computation_request", "extract_code_block", "validate_code"]


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    print("=== M37 cypher_code_as_tool SMOKE ===")
    # Detection
    tests = [
        ("Compute SHA256 of 'hello'", True),
        ("Hello, who are you?", False),
        ("Parse this Apache log line: ...", True),
        ("Extract IOCs from this dump", True),
    ]
    for p, expected in tests:
        got = detect_computation_request(p)
        mark = "✓" if got == expected else "✗"
        print(f"  {mark} '{p[:50]}' detect={got}")

    # Code extraction + validation
    text_with_code = """Here is the solution:
```python
import hashlib, json
h = hashlib.sha256(b'hello').hexdigest()
print(json.dumps({'hash': h}))
```
"""
    code = extract_code_block(text_with_code)
    print(f"\nExtracted code: {code[:100]!r}")
    v = validate_code(code)
    print(f"Validation: {v}")

    # Bad code (exec call)
    bad = "import os\nos.system('rm -rf /')"
    print(f"\nBad code validation: {validate_code(bad)}")

    print("\n=== SMOKE PASS ===")