Update app.py

app.py

@@ -5,43 +5,37 @@ import resource
 import multiprocessing
 import signal
 import time
-import threading
 import shutil
 import gradio as gr
+import threading
 
 # -------------------------
-# Store running sandboxes
+# Sandbox storage
 # -------------------------
 sandboxes = {}
-SANDBOX_TTL = 2 * 3600  # 2 hours
-MAX_RAM = 4 * 1024**3  # 4GB
+SANDBOX_TTL = 2*3600  # 2 hours
 
 # -------------------------
-# Sandbox worker function
+# Worker to run bare-metal process
 # -------------------------
-def sandbox_worker(code: str, requirements: str, sandbox_id: str):
+def sandbox_worker(code: str, sandbox_id: str):
     user_dir = f"/tmp/sandbox_{sandbox_id}"
     os.makedirs(user_dir, exist_ok=True)
     os.chdir(user_dir)
 
-    # Save requirements.txt
-    req_file = os.path.join(user_dir, "requirements.txt")
-    with open(req_file, "w") as f:
-        f.write(requirements)
-
     # Resource limits
-    resource.setrlimit(resource.RLIMIT_AS, (MAX_RAM, MAX_RAM))
-    resource.setrlimit(resource.RLIMIT_CPU, (SANDBOX_TTL, SANDBOX_TTL))
+    resource.setrlimit(resource.RLIMIT_AS, (4*1024**3, 4*1024**3))  # 4GB RAM
+    resource.setrlimit(resource.RLIMIT_CPU, (SANDBOX_TTL, SANDBOX_TTL))  # 2h CPU
 
-    # Save user code
-    code_file = os.path.join(user_dir, "user_code.py")
+    # Save code as a script
+    code_file = os.path.join(user_dir, "user_code.sh")
     with open(code_file, "w") as f:
         f.write(code)
+    os.chmod(code_file, 0o755)  # make executable
 
-    # Run code
     try:
         result = subprocess.run(
-            ["python3", code_file],
+            ["/bin/bash", code_file],
             capture_output=True,
             text=True,
             timeout=SANDBOX_TTL
@@ -53,10 +47,10 @@ def sandbox_worker(code: str, requirements: str, sandbox_id: str):
             f.write(result.stderr)
     except subprocess.TimeoutExpired:
         with open(os.path.join(user_dir, "stderr.log"), "w") as f:
-            f.write("Execution timed out (2h limit reached).")
+            f.write("Execution timed out (2h limit).")
 
 # -------------------------
-# Cleanup thread
+# Auto-cleanup thread
 # -------------------------
 def cleanup_sandboxes():
     while True:
@@ -76,11 +70,11 @@ def cleanup_sandboxes():
 threading.Thread(target=cleanup_sandboxes, daemon=True).start()
 
 # -------------------------
-# Gradio interface functions
+# Gradio API functions
 # -------------------------
-def launch_sandbox(code, requirements):
+def launch_sandbox(code):
     sandbox_id = str(uuid.uuid4())[:8]
-    p = multiprocessing.Process(target=sandbox_worker, args=(code, requirements, sandbox_id))
+    p = multiprocessing.Process(target=sandbox_worker, args=(code, sandbox_id))
     p.start()
     sandboxes[sandbox_id] = {"pid": p.pid, "start_time": time.time()}
     return f"Sandbox launched! ID: {sandbox_id}", sandbox_id
@@ -89,40 +83,36 @@ def fetch_logs(sandbox_id):
     user_dir = f"/tmp/sandbox_{sandbox_id}"
     if not os.path.exists(user_dir):
         return "Sandbox not found."
-    stdout_file = os.path.join(user_dir, "stdout.log")
-    stderr_file = os.path.join(user_dir, "stderr.log")
-    stdout = open(stdout_file).read() if os.path.exists(stdout_file) else ""
-    stderr = open(stderr_file).read() if os.path.exists(stderr_file) else ""
+    stdout = open(os.path.join(user_dir, "stdout.log")).read() if os.path.exists(os.path.join(user_dir, "stdout.log")) else ""
+    stderr = open(os.path.join(user_dir, "stderr.log")).read() if os.path.exists(os.path.join(user_dir, "stderr.log")) else ""
     return f"STDOUT:\n{stdout}\n\nSTDERR:\n{stderr}"
 
 def kill_sandbox(sandbox_id):
     if sandbox_id not in sandboxes:
         return "Sandbox not found."
-    pid = sandboxes[sandbox_id]["pid"]
     try:
-        os.kill(pid, signal.SIGKILL)
+        os.kill(sandboxes[sandbox_id]["pid"], signal.SIGKILL)
         folder = f"/tmp/sandbox_{sandbox_id}"
         if os.path.exists(folder):
             shutil.rmtree(folder)
         del sandboxes[sandbox_id]
         return f"Sandbox {sandbox_id} killed."
     except Exception as e:
-        return f"Error killing sandbox: {e}"
+        return f"Error: {e}"
 
 def status_sandbox(sandbox_id):
     if sandbox_id not in sandboxes:
         return "Sandbox not found."
-    elapsed = time.time() - sandboxes[sandbox_id]["start_time"]
-    return f"Running for {int(elapsed)} seconds"
+    elapsed = int(time.time() - sandboxes[sandbox_id]["start_time"])
+    return f"Running for {elapsed} seconds"
 
 # -------------------------
 # Gradio UI
 # -------------------------
 with gr.Blocks() as demo:
-    gr.Markdown("# Sandbox Backend with requirements.txt support")
+    gr.Markdown("# 🛡 Bare-Metal Sandbox")
 
-    code_input = gr.Textbox(label="Python Code", lines=15, placeholder="Write your code here...")
-    req_input = gr.Textbox(label="requirements.txt", lines=5, placeholder="Add dependencies here...")
+    code_input = gr.Textbox(label="Shell Script / Commands", lines=15, placeholder="#!/bin/bash\nls -la")
     launch_btn = gr.Button("Launch Sandbox")
     sandbox_id_output = gr.Textbox(label="Sandbox ID")
 
@@ -135,27 +125,15 @@ with gr.Blocks() as demo:
     status_btn = gr.Button("Check Status")
     status_output = gr.Textbox(label="Sandbox Status")
 
-    sandbox_id_storage = gr.State(value=None)
+    sandbox_state = gr.State(value=None)
 
     launch_btn.click(
         launch_sandbox,
-        inputs=[code_input, req_input],
-        outputs=[sandbox_id_output, sandbox_id_storage]
-    )
-    logs_btn.click(
-        fetch_logs,
-        inputs=sandbox_id_storage,
-        outputs=logs_output
-    )
-    kill_btn.click(
-        kill_sandbox,
-        inputs=sandbox_id_storage,
-        outputs=kill_output
-    )
-    status_btn.click(
-        status_sandbox,
-        inputs=sandbox_id_storage,
-        outputs=status_output
+        inputs=code_input,
+        outputs=[sandbox_id_output, sandbox_state]
     )
+    logs_btn.click(fetch_logs, inputs=sandbox_state, outputs=logs_output)
+    kill_btn.click(kill_sandbox, inputs=sandbox_state, outputs=kill_output)
+    status_btn.click(status_sandbox, inputs=sandbox_state, outputs=status_output)
 
 demo.launch()