"""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 ===")