explore of sanboxes concept

.gitignore

@@ -21,3 +21,6 @@ mini_elixir-*.tar
 
 # Temporary files, for example, from tests.
 /tmp/
+
+# CodeDB analyzer snapshot (code intelligence database)
+/codedb.snapshot

config/config.exs

@@ -0,0 +1,50 @@
+import Config
+
+config :pythonx, :uv_init,
+  pyproject_toml: """
+  [project]
+  name = "mini-elixir-python"
+  version = "0.0.0"
+  requires-python = "==3.13.*"
+  dependencies = [
+    "scikit-learn>=1.5.0",
+    "numpy>=1.26.0"
+  ]
+  """
+
+# MiniElixir Sandbox Configuration
+config :mini_elixir, :sandbox,
+  # Pool settings
+  min_warm: 5,
+  max_sandboxes: 100,
+  sandbox_timeout_ms: 30_000,
+  language: :python,
+
+  # VM resources
+  vm_memory_mb: 256,
+  vm_vcpu: 1
+
+# Unikraft/libvirt Configuration
+config :mini_elixir, :unikraft,
+  libvirt_uri: "qemu:///system",
+  python_image: "org/python-sandbox:latest",
+  elixir_image: "org/elixir-sandbox:latest"
+
+# Kuasar Sandbox Configuration
+config :mini_elixir, :kuasar,
+  # Sandboxer type: :vmm, :wasm, :quark, :runc
+  type: :vmm,
+  # API endpoint
+  api_url: "http://localhost:8080",
+  socket_path: "/run/vmm-sandboxer.sock",
+  # Pool settings
+  min_warm: 3,
+  max_sandboxes: 50,
+  sandbox_timeout_ms: 30_000,
+  language: :python,
+  # Images
+  vmm_python_image: "kuasar.io/python-sandbox:latest",
+  vmm_elixir_image: "kuasar.io/elixir-sandbox:latest",
+  wasm_python_image: "kuasar.io/wasm-python-sandbox:latest",
+  quark_python_image: "kuasar.io/quark-python-sandbox:latest",
+  runc_python_image: "kuasar.io/runc-python-sandbox:latest"

docker/sandbox_python/Dockerfile

@@ -0,0 +1,11 @@
+FROM unikraft/app-python3:stable
+
+# Copy sandbox files
+COPY server.py /server.py
+COPY restricted_builtins.py /restricted_builtins.py
+
+# Expose HTTP port
+EXPOSE 8080
+
+# Start the sandbox server
+CMD ["python3", "/server.py"]

docker/sandbox_python/restricted_builtins.py

@@ -0,0 +1,148 @@
+"""
+Restricted Python builtins for sandboxed execution.
+
+Provides a safe subset of Python builtins and import controls.
+"""
+
+import builtins
+import sys
+from typing import Any, Dict, Optional
+
+# Allowed builtins (safe subset)
+ALLOWED_BUILTINS = {
+    # Type constructors
+    "bool", "int", "float", "str", "list", "dict", "tuple", "set", "frozenset",
+    "complex", "bytes", "bytearray",
+
+    # Type checking
+    "isinstance", "issubclass", "type",
+
+    # Math
+    "abs", "divmod", "max", "min", "pow", "round", "sum",
+
+    # Iteration
+    "enumerate", "filter", "map", "reversed", "sorted", "zip",
+
+    # Sequence operations
+    "len", "slice", "range",
+
+    # Conversion
+    "chr", "hex", "oct", "ord", "bin", "int",
+
+    # I/O (limited)
+    "print", "repr", "format",
+
+    # Other safe builtins
+    "all", "any", "callable", "hash", "id", "iter", "next",
+    "object", "property", "staticmethod", "classmethod",
+    "NotImplemented", "Ellipsis",
+}
+
+# Blocked imports (dangerous modules)
+BLOCKED_IMPORTS = {
+    # OS/System
+    "os", "sys", "subprocess", "signal", "ctypes",
+
+    # File system
+    "pathlib", "shutil", "tempfile",
+
+    # Network
+    "socket", "http", "urllib", "requests", "aiohttp",
+
+    # Import system
+    "importlib",
+
+    # Threading/Processes
+    "threading", "multiprocessing", "concurrent",
+
+    # Code introspection (restricted)
+    "code", "codeop",
+
+    # Other dangerous modules
+    "pickle", "shelve", "dbm",
+
+    # Internal modules
+    "_thread", "_io", "_socket",
+}
+
+
+def get_restricted_globals() -> Dict[str, Any]:
+    """
+    Get restricted globals dictionary for code execution.
+
+    Returns a dict with safe builtins and restricted imports.
+    """
+    # Start with empty globals
+    globals_dict = {
+        "__builtins__": {},
+        "__name__": "__sandbox__",
+        "__doc__": None,
+        "__spec__": None,
+    }
+
+    # Add allowed builtins
+    for name in ALLOWED_BUILTINS:
+        if hasattr(builtins, name):
+            globals_dict["__builtins__"][name] = getattr(builtins, name)
+
+    # Add restricted __import__ function
+    globals_dict["__builtins__"]["__import__"] = restricted_import
+
+    return globals_dict
+
+
+def restricted_import(name: str, *args, **kwargs) -> Any:
+    """
+    Restricted import function that only allows safe modules.
+
+    Args:
+        name: Module name to import
+        *args: Additional import arguments
+        **kwargs: Additional import keyword arguments
+
+    Returns:
+        The imported module if allowed
+
+    Raises:
+        ImportError: If the module is not allowed
+    """
+    # Check if module is blocked
+    if name in BLOCKED_IMPORTS:
+        raise ImportError(f"Import of '{name}' is not allowed in sandbox")
+
+    # Check for sub-imports (e.g., os.path)
+    top_level = name.split(".")[0]
+    if top_level in BLOCKED_IMPORTS:
+        raise ImportError(f"Import of '{name}' is not allowed in sandbox")
+
+    # Allow the import
+    return __builtins__["__import__"](name, *args, **kwargs)
+
+
+def is_import_allowed(code: str) -> Optional[str]:
+    """
+    Check if code contains forbidden imports.
+
+    Args:
+        code: Python code to check
+
+    Returns:
+        Error message if forbidden import found, None otherwise
+    """
+    import re
+
+    # Patterns to match imports
+    patterns = [
+        r"^\s*import\s+([\w\.]+)",
+        r"^\s*from\s+([\w\.]+)\s+import",
+    ]
+
+    for pattern in patterns:
+        for match in re.finditer(pattern, code, re.MULTILINE):
+            module = match.group(1)
+            top_level = module.split(".")[0]
+
+            if top_level in BLOCKED_IMPORTS:
+                return f"Import of '{module}' is not allowed"
+
+    return None

docker/sandbox_python/server.py

@@ -0,0 +1,160 @@
+"""
+MiniElixir Python Sandbox Server
+
+A lightweight HTTP server that executes Python code in a restricted environment.
+Designed to run inside Unikraft microVMs for isolated code execution.
+"""
+
+import json
+import sys
+import traceback
+from http.server import HTTPServer, BaseHTTPRequestHandler
+from io import StringIO
+from restricted_builtins import get_restricted_globals, is_import_allowed
+
+
+class SandboxHandler(BaseHTTPRequestHandler):
+    """HTTP handler for code execution requests."""
+
+    def do_POST(self):
+        """Handle POST requests for code execution."""
+        if self.path == "/execute":
+            self._handle_execute()
+        elif self.path == "/health":
+            self._handle_health()
+        else:
+            self._send_response(404, {"error": "Not found"})
+
+    def do_GET(self):
+        """Handle GET requests for health checks."""
+        if self.path == "/health":
+            self._handle_health()
+        else:
+            self._send_response(404, {"error": "Not found"})
+
+    def _handle_execute(self):
+        """Execute code in sandbox."""
+        try:
+            content_length = int(self.headers.get("Content-Length", 0))
+            body = self.rfile.read(content_length)
+            request = json.loads(body)
+
+            code = request.get("code", "")
+            timeout = request.get("timeout", 30)
+
+            # Validate code before execution
+            validation_error = validate_code(code)
+            if validation_error:
+                self._send_response(400, {
+                    "status": "error",
+                    "error": validation_error
+                })
+                return
+
+            # Execute code
+            result = execute_code(code, timeout)
+            self._send_response(200, result)
+
+        except json.JSONDecodeError:
+            self._send_response(400, {"status": "error", "error": "Invalid JSON"})
+        except Exception as e:
+            self._send_response(500, {"status": "error", "error": str(e)})
+
+    def _handle_health(self):
+        """Health check endpoint."""
+        self._send_response(200, {"status": "ok", "language": "python"})
+
+    def _send_response(self, status_code, data):
+        """Send JSON response."""
+        self.send_response(status_code)
+        self.send_header("Content-Type", "application/json")
+        self.end_headers()
+        self.wfile.write(json.dumps(data).encode())
+
+    def log_message(self, format, *args):
+        """Suppress default logging."""
+        pass
+
+
+def validate_code(code):
+    """Validate code before execution. Returns error message or None."""
+    if not code or not code.strip():
+        return "Empty code"
+
+    if len(code) > 10000:
+        return "Code too long (max 10KB)"
+
+    # Check for blocked imports
+    error = is_import_allowed(code)
+    if error:
+        return error
+
+    # Check for blocked builtins
+    blocked = ["__import__", "exec(", "eval(", "compile(", "open(",
+               "globals(", "locals(", "getattr(", "setattr(", "delattr("]
+    for b in blocked:
+        if b in code:
+            return f"Forbidden: {b}"
+
+    return None
+
+
+def execute_code(code, timeout=30):
+    """Execute code in restricted environment."""
+    old_stdout = sys.stdout
+    old_stderr = sys.stderr
+    redirected_output = StringIO()
+    redirected_error = StringIO()
+
+    try:
+        sys.stdout = redirected_output
+        sys.stderr = redirected_error
+
+        # Get restricted globals
+        restricted_globals = get_restricted_globals()
+
+        # Capture result
+        restricted_globals["__result__"] = None
+
+        # Wrap code to capture last expression result
+        wrapped_code = f"""
+__result__ = None
+{code}
+"""
+        exec(wrapped_code, restricted_globals)
+
+        # Try to get result
+        result = restricted_globals.get("__result__")
+        stdout = redirected_output.getvalue()
+        stderr = redirected_error.getvalue()
+
+        return {
+            "status": "ok",
+            "result": result,
+            "stdout": stdout if stdout else None,
+            "stderr": stderr if stderr else None
+        }
+
+    except Exception as e:
+        return {
+            "status": "error",
+            "error": f"{type(e).__name__}: {str(e)}",
+            "traceback": traceback.format_exc(),
+            "stdout": redirected_output.getvalue(),
+            "stderr": redirected_error.getvalue()
+        }
+    finally:
+        sys.stdout = old_stdout
+        sys.stderr = old_stderr
+
+
+def main():
+    """Start the sandbox server."""
+    port = int(sys.argv[1]) if len(sys.argv) > 1 else 8080
+    server = HTTPServer(("0.0.0.0", port), SandboxHandler)
+    print(f"Sandbox server running on port {port}")
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()