sandbox.py

import sys
import subprocess
import json
import tempfile
import os

EXECUTION_TEMPLATE = """
import sys
import json
import traceback

# --- User Code Start ---
{user_code}
# --- User Code End ---

def run_tests():
    test_cases = {test_cases_json}
    results = []
    
    # Locate entry point function
    func = None
    if 'solve' in globals() and callable(globals()['solve']):
        func = globals()['solve']
    else:
        for name, obj in list(globals().items()):
            if callable(obj) and not name.startswith('_') and name not in ('run_tests', 'sys', 'json', 'traceback', 'copy'):
                func = obj
                break
                
    if not func:
        print(json.dumps([ {{"status": "error", "error": "No callable function found in your code. Please define a function (e.g., 'def solve(...)')."}} ]))
        return

    for idx, tc in enumerate(test_cases):
        inputs = tc.get("input", [])
        expected = tc.get("expected")
        if not isinstance(inputs, list):
            inputs = [inputs]
        try:
            import copy
            args = copy.deepcopy(inputs)
            res = func(*args)
            
            # Comparison
            is_match = (res == expected)
            results.append({{
                "index": idx,
                "status": "passed" if is_match else "failed",
                "input": inputs,
                "expected": expected,
                "got": res
            }})
        except Exception as e:
            results.append({{
                "index": idx,
                "status": "error",
                "input": inputs,
                "expected": expected,
                "error": str(e),
                "traceback": traceback.format_exc().splitlines()[-2:]
            }})
            
    print(json.dumps(results))

if __name__ == '__main__':
    run_tests()
"""


def run_python_sandbox(user_code: str, test_cases: list, timeout: float = 4.0) -> dict:
    """
    Executes user code against test cases in an isolated sandbox.
    Attempts Docker execution first, then falls back to restricted subprocess.
    """
    test_cases_json = json.dumps(test_cases)
    full_script = EXECUTION_TEMPLATE.format(
        user_code=user_code,
        test_cases_json=test_cases_json
    )
    
    # Try Docker execution
    try:
        # Run python:3.11-slim container with stdin, no network, 128MB memory, 0.5 CPU
        # On Windows, we pipe the script directly to python via stdin to avoid path mounts
        cmd = [
            "docker", "run", "--rm", "-i",
            "--network", "none",
            "--memory", "128m",
            "--cpus", "0.5",
            "python:3.11-slim",
            "python"
        ]
        
        proc = subprocess.Popen(
            cmd,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
        
        stdout, stderr = proc.communicate(input=full_script, timeout=timeout)
        
        if proc.returncode == 137:
            return {
                "success": False,
                "error": "Memory limit exceeded (OOM) or container killed.",
                "sandbox_type": "docker"
            }
        
        if proc.returncode != 0:
            return {
                "success": False,
                "error": stderr or f"Execution failed with code {proc.returncode}",
                "sandbox_type": "docker"
            }
            
        try:
            results = json.loads(stdout.strip())
            return {
                "success": True,
                "results": results,
                "sandbox_type": "docker"
            }
        except json.JSONDecodeError:
            return {
                "success": False,
                "error": f"Invalid output from execution: {stdout}\nErrors: {stderr}",
                "sandbox_type": "docker"
            }
            
    except (subprocess.TimeoutExpired, TimeoutError):
        # Clean up process
        try:
            proc.kill()
        except Exception:
            pass
        return {
            "success": False,
            "error": f"Execution timed out. Code exceeded max runtime of {timeout}s.",
            "sandbox_type": "docker"
        }
    except Exception as docker_err:
        # Fall back to native local subprocess execution with limited environment
        return run_local_fallback(full_script, timeout, str(docker_err))

def run_local_fallback(script_content: str, timeout: float, debug_msg: str) -> dict:
    """
    Fallback execution using current python environment.
    """
    try:
        proc = subprocess.Popen(
            [sys.executable],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
        stdout, stderr = proc.communicate(input=script_content, timeout=timeout)
        
        if proc.returncode != 0:
            return {
                "success": False,
                "error": stderr or f"Local execution failed with code {proc.returncode}",
                "sandbox_type": "local_fallback",
                "sandbox_warning": f"Docker isolation unavailable: {debug_msg}"
            }
            
        try:
            results = json.loads(stdout.strip())
            return {
                "success": True,
                "results": results,
                "sandbox_type": "local_fallback",
                "sandbox_warning": f"Docker isolation unavailable: {debug_msg}"
            }
        except json.JSONDecodeError:
            return {
                "success": False,
                "error": f"Invalid output from local runner: {stdout}",
                "sandbox_type": "local_fallback"
            }
    except subprocess.TimeoutExpired:
        try:
            proc.kill()
        except Exception:
            pass
        return {
            "success": False,
            "error": f"Local execution timed out. Code exceeded {timeout}s.",
            "sandbox_type": "local_fallback"
        }
    except Exception as e:
        return {
            "success": False,
            "error": f"Failed to execute local fallback runner: {str(e)}",
            "sandbox_type": "none"
        }