app.py

import gevent
import gevent.monkey
gevent.monkey.patch_all()  # Must be called at the very top of your application file

import os
import pty
import select
import struct
import fcntl
import termios
import signal
import subprocess
from flask import Flask, render_template, request
from flask_socketio import SocketIO

app = Flask(__name__)
app.config["SECRET_KEY"] = os.environ.get("SECRET_KEY", "fallback-secret-key")
socketio = SocketIO(app, cors_allowed_origins="*")

# Keep track of active terminal sessions by connection ID (sid)
terminals = {}

# Verify if Bubblewrap sandboxing is supported by the host kernel on startup
HAS_BWRAP = False
try:
    # A lightweight test to check if unprivileged namespace sandboxing is allowed
    result = subprocess.run(
        ["bwrap", "--dev", "/dev", "--proc", "/proc", "--tmpfs", "/tmp", "true"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        timeout=2
    )
    if result.returncode == 0:
        HAS_BWRAP = True
        print("[System Check] Bubblewrap sandbox environment successfully verified and active.")
    else:
        print("[System Check] Bubblewrap is installed, but kernel namespaces are restricted. Defaulting to fallback shell.")
except Exception as e:
    print(f"[System Check] Bubblewrap test failed ({str(e)}). Defaulting to fallback shell.")

def set_winsize(fd, row, col, xpix=0, ypix=0):
    winsize = struct.pack("HHHH", row, col, xpix, ypix)
    fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)

def read_and_forward_pty_output(sid):
    max_read_bytes = 1024 * 20
    while sid in terminals:
        socketio.sleep(0.01)
        term_data = terminals.get(sid)
        if not term_data:
            break
        
        fd = term_data['fd']
        (data_ready, _, _) = select.select([fd], [], [], 0)
        if data_ready:
            try:
                output = os.read(fd, max_read_bytes)
                if len(output) == 0:
                    cleanup_terminal(sid)
                    break
                socketio.emit("pty_output", {"output": output.decode("utf-8", errors="replace")}, room=sid)
            except Exception:
                cleanup_terminal(sid)
                break

def cleanup_terminal(sid):
    if sid in terminals:
        term_data = terminals.pop(sid)
        try:
            os.close(term_data['fd'])
        except Exception:
            pass
        try:
            os.kill(term_data['pid'], signal.SIGKILL)
            os.waitpid(term_data['pid'], 0)
        except Exception:
            pass

@app.route("/")
def index():
    return render_template("index.html")

@socketio.on("connect")
def connect():
    sid = request.sid
    print(f"Client connected: {sid}")
    
    child_pid, fd = pty.fork()
    if child_pid == 0:
        if HAS_BWRAP:
            # Construct a Bubblewrap sandbox:
            # - Bind important system libraries and binaries as read-only
            # - Keep application directories (/app) completely hidden
            cmd = [
                "bwrap",
                "--ro-bind", "/usr", "/usr",
                "--ro-bind", "/bin", "/bin",
                "--ro-bind", "/lib", "/lib",
            ]
            
            if os.path.exists("/lib64"):
                cmd += ["--ro-bind", "/lib64", "/lib64"]
            if os.path.exists("/sbin"):
                cmd += ["--ro-bind", "/sbin", "/sbin"]
            
            cmd += [
                "--ro-bind", "/etc", "/etc",
                "--proc", "/proc",
                "--dev", "/dev",
                "--tmpfs", "/tmp",       # Empty memory-backed /tmp
                "--tmpfs", "/root",      # Empty memory-backed /root (discards files upon exit)
                "--unshare-pid",         # Isolate process visibility
                "--unshare-ipc",         # Isolate IPC mechanisms
                "--unshare-uts",         # Isolate hostname changes
                # Note: We do NOT include '--unshare-net' to allow standard outbound network access (e.g. curl/wget)
                "--setenv", "HOME", "/root",
                "--setenv", "TERM", "xterm-256color",
                "bash", "--login"
            ]
            
            try:
                os.execvp("bwrap", cmd)
            except Exception:
                pass  # Fall through to default if execution failed
            
        # Fallback shell if sandbox environment could not be created
        os.environ["TERM"] = "xterm-256color"
        os.environ["HOME"] = "/root"
        os.execlp("bash", "bash", "--login")
    else:
        terminals[sid] = {"fd": fd, "pid": child_pid}
        socketio.start_background_task(read_and_forward_pty_output, sid)

@socketio.on("disconnect")
def disconnect():
    sid = request.sid
    print(f"Client disconnected: {sid}")
    cleanup_terminal(sid)

@socketio.on("pty_input")
def pty_input(data):
    sid = request.sid
    if sid in terminals:
        fd = terminals[sid]['fd']
        try:
            os.write(fd, data["input"].encode())
        except Exception:
            cleanup_terminal(sid)

@socketio.on("resize")
def resize(data):
    sid = request.sid
    if sid in terminals:
        fd = terminals[sid]['fd']
        try:
            set_winsize(fd, data["rows"], data["cols"])
        except Exception:
            pass